added stopping of worker

1 files changed, 89 insertions, 0 deletions

diff --git a/src/coordinator.c b/src/coordinator.c
index 30afb62..1366d88 100644
--- a/src/coordinator.c
+++ b/src/coordinator.c
@@ -61,6 +61,39 @@ static char *create_worker_started_answer( const char *name, bool worker_found )
 	return res;
 }
 
+static char *create_worker_stopped_answer( const char *name, bool worker_found )
+{
+	json_object *obj = json_object_new_object( );
+	json_object *op = json_object_new_string( "stopped" );
+	json_object_object_add( obj, "op", op );
+
+	/* we are a coordinator */
+	json_object *role = json_object_new_string( "coordinator" );
+	json_object_object_add( obj, "role", role );
+
+	/* return hostname for unique identification (maybe better
+	 * or additionally the MAC of the first interface or hostid?)
+	 */
+	char hostname[100];
+	gethostname( hostname, sizeof( hostname ) );
+	json_object *host = json_object_new_string( hostname );
+	json_object_object_add( obj, "host", host );
+	
+	json_object *worker = json_object_new_string( name );
+	json_object_object_add( obj, "worker", worker );
+	
+	json_object *found = json_object_new_boolean( worker_found );
+	json_object_object_add( obj, "found", found );
+
+	/* produce message as string, caller must free it */
+	const char *msg = json_object_to_json_string( obj );
+	char *res = strdup( msg );
+	
+	json_object_put( obj );
+	
+	return res;
+}
+
 static char *create_register_answer( )
 {
 	json_object *obj = json_object_new_object( );
@@ -162,6 +195,32 @@ static int coordinator_start_worker( const char *name, bool *found )
 	return 0;
 }
 
+static int coordinator_stop_worker( const char *name, bool *found )
+{
+	worker_t *w = NULL;
+	worker_t *wfound = NULL;
+	*found = false;
+	for( int i = 0; i < nof_workers; i++ ) {
+		w = &worker[i];
+		if( strcmp( name, w->name ) == 0 ) {
+			*found = true;
+			wfound = w;
+			break;
+		}
+	}
+	
+	if( !( *found ) ) {
+		return 0;
+	}
+	
+	printf( "STOPPING WORKER '%s'\n", wfound->name );
+	
+	worker_terminate( wfound );
+		
+	return 0;
+}
+
+
 static void *coordinator_func( void *thread_data )
 {
 	char *control = (char *)thread_data;
@@ -240,6 +299,36 @@ static void *coordinator_func( void *thread_data )
 				}
 				
 				free( msg );
+			} else if( strcmp( op, "stop" ) == 0 ) {
+				
+				json_object *worker_obj;
+				json_object_object_get_ex( recv_obj, "worker", &worker_obj );
+				const char *worker = json_object_get_string( worker_obj );
+
+				bool found;
+				/*int res = */
+				(void)coordinator_stop_worker( worker, &found );
+				
+				char *msg = create_worker_stopped_answer( worker, found );
+				size_t msg_size = strlen( msg ) + 1;
+		
+				// TODO: tons of copy-paste code here, think!
+				printf( "coordinator send: %s\n", msg );
+				bytes = nn_send( coordinator_sock, msg, msg_size, 0 );
+				if( bytes < 0 ) {
+					if( errno == ETERM ) {
+						coordinator_must_terminate = 1;
+						continue;
+					} else {
+						fprintf( stderr, "ERROR: nn_send returned %d: %s (%d)\n",
+							bytes, nn_strerror( errno ), errno );
+					}
+				}
+				if( bytes != msg_size ) {
+					fprintf( stderr, "ERROR: truncated message!" );
+				}
+				
+				free( msg );
 			}
 
 			json_object_put( recv_obj );