Move cachpath consensus to be calculated at init time

src/client/client_comlib/client_api.c

@@ -40,42 +40,31 @@ static struct lock_t comm_lock;
 
 
 int send_cachepath_query( int fd, char **chosen_realized_cachepath, char **chosen_parsed_cachepath){
-   int retries = 0, max_retries = 1000, rc = 0;
-   struct timespec delay_between_retries = { .tv_sec = 0, .tv_nsec = 1000000 };
+   int rc = 0;
    ldcs_message_t message;
-   char buffer[MAX_PATH_LEN+1];
+   char buffer[2*(MAX_PATH_LEN+1)];
    buffer[MAX_PATH_LEN] = '\0';
 
 
-   do{
-       message.header.type = LDCS_MSG_CHOSEN_CACHEPATH_REQUEST;
-       message.header.len = 0;
-       message.data = buffer;
-
-       COMM_LOCK;
-
-       debug_printf3("sending message of type: CHOSEN_CACHEPATH_REQUEST.\n" );
-       rc = client_send_msg(fd, &message);
-       if( rc != 0 ){
-           return rc;
-       }
-       rc = client_recv_msg_static(fd, &message, LDCS_READ_BLOCK);
-       if( rc != 0 ){
-           return rc;
-       }
-
-       COMM_UNLOCK;
-
-       if( message.header.type == LDCS_MSG_NO_CACHEPATH_CONSENSUS_YET ){
-           if( retries++ >= max_retries ){
-               break;
-           }
-           nanosleep( &delay_between_retries, NULL );
-           continue;
-       }
-       break;
+   message.header.type = LDCS_MSG_CHOSEN_CACHEPATH_REQUEST;
+   message.header.len = 0;
+   message.data = buffer;
+
+   debug_printf3("sending message of type: CHOSEN_CACHEPATH_REQUEST.\n" );
+   COMM_LOCK;
 
-   }while( 1 );
+   rc = client_send_msg(fd, &message);
+   if( rc != 0 ){
+      COMM_UNLOCK;
+      return rc;
+   }
+   rc = client_recv_msg_static(fd, &message, LDCS_READ_BLOCK);
+   if( rc != 0 ){
+      COMM_UNLOCK;      
+      return rc;
+   }
+
+   COMM_UNLOCK;
 
    if (message.header.type != LDCS_MSG_CHOSEN_CACHEPATH || message.header.len > MAX_PATH_LEN) {
       err_printf("Got unexpected message of type %d\n", (int) message.header.type);

src/fe/startup/spindle_fe.cc

@@ -41,7 +41,6 @@ static const char *logging_file = NULL;
 #endif
 static const char spindle_bootstrap[] = LIBEXECDIR "/spindle_bootstrap";
 static bool sendAndWaitForAlive();
-static void determineCachepathConsensus();
 
 #define STARTUP_TIMEOUT 60
 
@@ -433,7 +432,6 @@ int spindleInitFE(const char **hosts, spindle_args_t *params)
 
    /* Wait for servers to indicate startup */
    sendAndWaitForAlive();
-   determineCachepathConsensus();
 
    return 0;   
 }
@@ -490,17 +488,6 @@ void markRSHPidReapedFE()
    clear_fe_rsh_pid();
 }
 
-static void determineCachepathConsensus( void ){
-   ldcs_message_t consensus_req_msg;
-   consensus_req_msg.header.type = LDCS_MSG_REQUEST_CACHEPATH_CONSENSUS;
-   consensus_req_msg.header.len = 0;
-   consensus_req_msg.data = NULL;
-   int result = ldcs_audit_server_fe_broadcast(&consensus_req_msg, NULL);
-   if (result == -1) {
-      debug_printf("Failure sending cachepath consensus message\n");
-   }
-}
-
 static bool sendAndWaitForAlive()
 {
    int result;

src/include/ldcs_api.h

@@ -85,10 +85,8 @@ typedef enum {
    LDCS_MSG_PICKONE_RESP,
    LDCS_MSG_ALIVE_REQ,
    LDCS_MSG_ALIVE_RESP,
-   LDCS_MSG_REQUEST_CACHEPATH_CONSENSUS,
    LDCS_MSG_CHOSEN_CACHEPATH_REQUEST,
    LDCS_MSG_CHOSEN_CACHEPATH,
-   LDCS_MSG_NO_CACHEPATH_CONSENSUS_YET,
    LDCS_MSG_UNKNOWN
 } ldcs_message_ids_t;
 

src/server/auditserver/ldcs_audit_server_handlers.c

@@ -181,7 +181,6 @@ static int handle_setup_alias(ldcs_process_data_t *procdata, char *pathname, cha
 static int handle_client_dirlists_req(ldcs_process_data_t *procdata, int nc);
 static int handle_close_client_query(ldcs_process_data_t *procdata, int nc);
 static int handle_alive_msg(ldcs_process_data_t *procdata, ldcs_message_t *msg);
-static int handle_cachepath_consensus(ldcs_process_data_t *procdata, ldcs_message_t *msg);
 static int handle_chosen_cachepath_request(ldcs_process_data_t *procdata, int nc);
 
 extern void getValidCachePathByIndex( uint64_t validBitIdx, char **realizedCachePath, char **parsedCachePath, char **symbolicCachePath );
@@ -1997,8 +1996,6 @@ int handle_server_message(ldcs_process_data_t *procdata, node_peer_t peer, ldcs_
       case LDCS_MSG_ALIVE_REQ:
       case LDCS_MSG_ALIVE_RESP:
          return handle_alive_msg(procdata, msg);
-      case LDCS_MSG_REQUEST_CACHEPATH_CONSENSUS:
-         return handle_cachepath_consensus(procdata, msg);
       default:
          err_printf("Received unexpected message from node: %d\n", (int) msg->header.type);
          assert(0);
@@ -2961,35 +2958,26 @@ static int handle_client_pickone_msg(ldcs_process_data_t *procdata, int nc, ldcs
 }
 
 /**
- * Handle LDCS_MSG_REQUEST_CACHEPATH_CONSENSUS to determine which cachepaths are
- * available across all of the servers.
+ * Determine which cachepaths are available across all of the servers.
  */
-static int cachepath_consensus_reached;
-static int handle_cachepath_consensus(ldcs_process_data_t *procdata, ldcs_message_t *msg){
-
+int handle_cachepath_consensus(ldcs_process_data_t *procdata)
+{
     int num_children = ldcs_audit_server_md_get_num_children(procdata);
 
-    debug_printf( "Processing REQUEST_CACHEPATH_CONSENSUS.\n" );
+    debug_printf( "Calculating cachepath consensus.\n" );
     debug_printf3( "  procdata->cachepath_bitidx = %#"PRIx64"\n", procdata->cachepath_bitidx );
     debug_printf3( "  procdata->cachepaths       = %s\n", procdata->cachepaths );
     debug_printf3( "  procdata->cachepath        = %s [should be null]\n", procdata->cachepath  );
     debug_printf3( "  procdata->commpath         = %s\n", procdata->commpath );
     debug_printf3( "  num_children               = %d\n", num_children );
 
-    if (num_children) {
-        spindle_broadcast(procdata, msg);
-        debug_printf3( "Successfully broadcast REQUEST_CACHEPATH_CONSENSUS\n" );
-        msgbundle_force_flush(procdata);
-        debug_printf3( "Successfully flushed the broadcast of REQUEST_CACHEPATH_CONSENSUS\n" );
-    }
-
     ldcs_audit_server_md_allreduce_AND( &procdata->cachepath_bitidx );
     debug_printf3( "The consensus value for procdata->cachepath_bitidx is:  %#"PRIx64"\n", procdata->cachepath_bitidx );
 
-    if( procdata->cachepath_bitidx == 0 ){
-       err_printf("No valid cachepath path available.  Falling back to \"commpath\" path (%s).\n", procdata->commpath);
+    if( procdata->cachepath_bitidx == 0 ) {
+       debug_printf("No valid cachepath path available.  Falling back to \"commpath\" path (%s).\n", procdata->commpath);
        procdata->cachepath = procdata->commpath;
-    }else{
+    } else {
         getValidCachePathByIndex( procdata->cachepath_bitidx,
                 &procdata->cachepath,
                 &procdata->parsed_cachepath,
@@ -3003,14 +2991,10 @@ static int handle_cachepath_consensus(ldcs_process_data_t *procdata, ldcs_messag
     ldcs_audit_server_filemngt_init(procdata->cachepath, procdata->commpath);
 
     test_printf("<internal> cachepath=%s\n", procdata->cachepath);
-    cachepath_consensus_reached = 1;
     return 0;
 }
 
-/**
- * Handle LDCS_MSG_CHOSEN_CACHEPATH_REQUEST
- */
-static int handle_chosen_cachepath_request(ldcs_process_data_t *procdata, int nc){
+static int handle_chosen_cachepath_request(ldcs_process_data_t *procdata, int nc) {
    ldcs_message_t msg;
    int connid;
    ldcs_client_t *client;
@@ -3022,18 +3006,12 @@ static int handle_chosen_cachepath_request(ldcs_process_data_t *procdata, int nc
       return 0;
 
 
-   if( cachepath_consensus_reached ){
-       msg.header.type = LDCS_MSG_CHOSEN_CACHEPATH;
-       msg.header.len = strlen(procdata->cachepath) + 1 + strlen(procdata->parsed_cachepath) + 1;
-       msg.data = calloc( 1, msg.header.len );
-       strcpy( msg.data, procdata->cachepath );
-       strcpy( &msg.data[ strlen(procdata->cachepath)+1 ], procdata->parsed_cachepath );
-   }else{
-       msg.header.type = LDCS_MSG_NO_CACHEPATH_CONSENSUS_YET;
-       msg.header.len = 0;
-       msg.data = NULL;
-   }
-
+   msg.header.type = LDCS_MSG_CHOSEN_CACHEPATH;
+   msg.header.len = strlen(procdata->cachepath) + 1 + strlen(procdata->parsed_cachepath) + 1;
+   msg.data = calloc( 1, msg.header.len );
+   strcpy( msg.data, procdata->cachepath );
+   strcpy( &msg.data[ strlen(procdata->cachepath)+1 ], procdata->parsed_cachepath );
+
    ldcs_send_msg(connid, &msg);
    free( msg.data );
    procdata->server_stat.clientmsg.cnt++;

src/server/auditserver/ldcs_audit_server_handlers.h

@@ -25,6 +25,7 @@ int handle_server_error(ldcs_process_data_t *procdata, node_peer_t peer);
 int handle_client_message(ldcs_process_data_t *procdata, int nc, ldcs_message_t *msg);
 int handle_client_start(ldcs_process_data_t *procdata, int nc);
 int handle_client_end(ldcs_process_data_t *procdata, int nc);
+int handle_cachepath_consensus(ldcs_process_data_t *procdata);
 int exit_note_cb(int infd, int serverid, void *data);
 
 

src/server/auditserver/ldcs_audit_server_process.c

@@ -137,7 +137,7 @@ void stopprofile()
 
 int ldcs_audit_server_process(spindle_args_t *args)
 {
-   int serverid, fd;
+   int serverid, fd, result;
 
    startprofile(args);
 
@@ -198,7 +198,20 @@ int ldcs_audit_server_process(spindle_args_t *args)
    if (ldcs_process_data.opts & OPT_PROCCLEAN)
       init_cleanup_proc(ldcs_process_data.cachepath, ldcs_process_data.commpath);
 
-   debug_printf3("Initializing connections for clients at %s and %lu\n",
+   /* Calculate location of cache */
+   debug_printf2("Calculating cache path location\n");
+   determineValidCachePaths(
+           &ldcs_process_data.cachepath_bitidx,
+           ldcs_process_data.cachepaths,
+           ldcs_process_data.number );
+   result = handle_cachepath_consensus(&ldcs_process_data);
+   if (result == -1) {
+      err_printf("Could not determine cachepath consensus\n");
+      return -1;
+   }   
+
+   /* Setup connections for clients to start connecting */
+   debug_printf2("Initializing connections for clients at %s and %lu\n",
                  ldcs_process_data.commpath, (unsigned long) ldcs_process_data.number);
    serverid = ldcs_create_server(ldcs_process_data.commpath, ldcs_process_data.number);
    if (serverid == -1) {
@@ -231,10 +244,7 @@ int ldcs_audit_server_process(spindle_args_t *args)
    if (fd != -1) {
       ldcs_listen_register_fd(fd, serverid, forceExitCB, (void *) &ldcs_process_data);
    }
-   determineValidCachePaths(
-           &ldcs_process_data.cachepath_bitidx,
-           ldcs_process_data.cachepaths,
-           ldcs_process_data.number );
+
    return 0;
 }  
 

src/server/comlib/ldcs_api_util.c

@@ -91,10 +91,8 @@ char* _message_type_to_str (ldcs_message_ids_t type) {
       STR_CASE(LDCS_MSG_PICKONE_RESP);
       STR_CASE(LDCS_MSG_ALIVE_REQ);
       STR_CASE(LDCS_MSG_ALIVE_RESP);      
-      STR_CASE(LDCS_MSG_REQUEST_CACHEPATH_CONSENSUS);
       STR_CASE(LDCS_MSG_CHOSEN_CACHEPATH_REQUEST);
       STR_CASE(LDCS_MSG_CHOSEN_CACHEPATH);
-      STR_CASE(LDCS_MSG_NO_CACHEPATH_CONSENSUS_YET);
       STR_CASE(LDCS_MSG_UNKNOWN);
    }
    return "unknown";