diff --git a/src/client/client_comlib/client_api.c b/src/client/client_comlib/client_api.c index cec31051..07735226 100644 --- a/src/client/client_comlib/client_api.c +++ b/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); diff --git a/src/fe/startup/spindle_fe.cc b/src/fe/startup/spindle_fe.cc index a038e201..0d5c0817 100644 --- a/src/fe/startup/spindle_fe.cc +++ b/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; diff --git a/src/include/ldcs_api.h b/src/include/ldcs_api.h index 0bcd8f40..2b261a56 100644 --- a/src/include/ldcs_api.h +++ b/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; diff --git a/src/server/auditserver/ldcs_audit_server_handlers.c b/src/server/auditserver/ldcs_audit_server_handlers.c index 1afdf2e8..6b3a6b1c 100644 --- a/src/server/auditserver/ldcs_audit_server_handlers.c +++ b/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(" 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++; diff --git a/src/server/auditserver/ldcs_audit_server_handlers.h b/src/server/auditserver/ldcs_audit_server_handlers.h index 8702316c..790e52b3 100644 --- a/src/server/auditserver/ldcs_audit_server_handlers.h +++ b/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); diff --git a/src/server/auditserver/ldcs_audit_server_process.c b/src/server/auditserver/ldcs_audit_server_process.c index b6ecdbff..7e32337d 100644 --- a/src/server/auditserver/ldcs_audit_server_process.c +++ b/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; } diff --git a/src/server/comlib/ldcs_api_util.c b/src/server/comlib/ldcs_api_util.c index af101815..28a3d90c 100644 --- a/src/server/comlib/ldcs_api_util.c +++ b/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";