path: root/src/worker.c

#include "worker.h"

#include "nanomsg/nn.h"
#include "nanomsg/pipeline.h"

#include "port.h"

#include "pthread.h"

#include <glib.h>

#include <stdio.h>
#include <string.h>
#include <stdlib.h>

#include <signal.h>

#include "json-c/json.h"

const char *worker_exection_mode_str( worker_execution_mode_t mode )
{
	switch( mode ) {
		case WORKER_EXECUTION_DISABLED: return "disabled";
		case WORKER_EXECUTION_DIRECT: return "direct";
		default: return "<unknown>";
	}
}

const char *worker_state_str( worker_state_t state )
{
	switch( state ) {
		case WORKER_STATE_STOPPED: return "stopped";
		case WORKER_STATE_RUNNING: return "running";
		default: return "<unknown>";
	}
}

worker_execution_mode_t worker_execution_mode_from_str( const char *s )
{
	if( strcasecmp( s, "disabled" ) == 0 ) {
		return WORKER_EXECUTION_DISABLED;
	} else if( strcasecmp( s, "direct" ) == 0 ) {
		return WORKER_EXECUTION_DIRECT;
	} else {
		fprintf( stderr, "Warning: unknown worker execution mode '%s'!\n", s );
		return WORKER_EXECUTION_DISABLED;
	}
}

worker_state_t worker_state_from_str( const char *s )
{
	if( strcasecmp( s, "stopped" ) == 0 ) {
		return WORKER_STATE_STOPPED;
	} else if( strcasecmp( s, "running" ) == 0 ) {
		return WORKER_STATE_RUNNING;
	} else {
		fprintf( stderr, "Warning: unknown worker state '%s' (assuming 'stopped')!\n", s );
		return WORKER_STATE_STOPPED;
	}
}		

typedef struct {
	GPid pid;
	gint out, err;
	GIOChannel *out_channel;
	GIOChannel *err_channel;
	worker_t *worker;
	pthread_t thread;
	GMainLoop *main_loop;
	int sock;
} direct_glib_execution_worker_data_t;

static char *create_worker_output_answer( const char *name, const char *s, size_t len, bool is_stdout )
{
	json_object *obj = json_object_new_object( );
	json_object *op = json_object_new_string( "output" );
	json_object_object_add( obj, "op", op );

	/* we are a worker */
	json_object *role = json_object_new_string( "worker" );
	json_object_object_add( obj, "role", role );
	
	json_object *name_obj = json_object_new_string( name );
	json_object_object_add( obj, "worker", name_obj );
	
	json_object *msg_obj = json_object_new_string_len( s, len );
	json_object_object_add( obj, "msg", msg_obj );
	
	json_object *stream_obj = json_object_new_string( is_stdout ? "stdout" : "stderr" );
	json_object_object_add( obj, "stream", stream_obj );
	
	time_t ts;
	time( &ts );
	json_object *timestamp_obj = json_object_new_int( ts );
	json_object_object_add( obj, "timestamp", timestamp_obj );
		
	/* 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_worker_termination_answer( const char *name, GPid pid )
{
	json_object *obj = json_object_new_object( );
	json_object *op = json_object_new_string( "terminated" );
	json_object_object_add( obj, "op", op );

	/* we are a worker */
	json_object *role = json_object_new_string( "worker" );
	json_object_object_add( obj, "role", role );
	
	json_object *name_obj = json_object_new_string( name );
	json_object_object_add( obj, "worker", name_obj );
	
	char pid_str[64];
	snprintf( pid_str, sizeof( pid_str ), PRIgid, pid );
	json_object *pid_obj = json_object_new_string( pid_str );
	json_object_object_add( obj, "pid", pid_obj );
		
	time_t ts;
	time( &ts );
	json_object *timestamp_obj = json_object_new_int( ts );
	json_object_object_add( obj, "timestamp", timestamp_obj );
		
	/* 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_worker_created_answer( const char *name, GPid pid )
{
	json_object *obj = json_object_new_object( );
	json_object *op = json_object_new_string( "created" );
	json_object_object_add( obj, "op", op );

	/* we are a worker */
	json_object *role = json_object_new_string( "worker" );
	json_object_object_add( obj, "role", role );
	
	json_object *name_obj = json_object_new_string( name );
	json_object_object_add( obj, "worker", name_obj );
	
	char pid_str[64];
	snprintf( pid_str, sizeof( pid_str ), PRIgid, pid );
	json_object *pid_obj = json_object_new_string( pid_str );
	json_object_object_add( obj, "pid", pid_obj );
		
	time_t ts;
	time( &ts );
	json_object *timestamp_obj = json_object_new_int( ts );
	json_object_object_add( obj, "timestamp", timestamp_obj );
		
	/* 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 int worker_send_message( worker_t *worker, const char *msg )
{
	direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)worker->execution_data;

	size_t msg_size = strlen( msg ) + 1;
	
	printf( "worker send: %s\n", msg );

	int bytes = nn_send( wed->sock, msg, msg_size, 0 );
	if( bytes < 0 ) {
		fprintf( stderr, "ERROR: worker '%s' nn_send returned %d: %s (%d)\n",
			worker->name, bytes, nn_strerror( errno ), errno );
		return -1;
	}
	if( bytes != msg_size ) {
		fprintf( stderr, "ERROR: worker '%s' truncated message!",
			worker->name );
		return -1;
	}
	
	return 0;
}

static void watch_child( GPid pid, gint status, gpointer *data )
{
	direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)data;
	worker_t *worker = wed->worker;
	
	printf( "Worker child with PID " PRIgid " terminated.\n", wed->pid );

	char *msg = create_worker_termination_answer( worker->name, wed->pid );
	worker_send_message( worker, msg );	
	free( msg );

	g_spawn_close_pid( pid );
	
	g_main_loop_quit( wed->main_loop );
}

static gboolean watch_output( GIOChannel *channel, GIOCondition condition, gpointer *data )
{
	direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)data;
	worker_t *worker = wed->worker;

	bool is_stdout = ( wed->out_channel == channel );
	
	gchar *buf;
	gsize len;
	
	if( ( condition & G_IO_HUP ) == G_IO_HUP ) {
		g_io_channel_unref( channel );
		return FALSE;
	}
	
	g_io_channel_read_line( channel, &buf, &len, NULL, NULL );

	char *msg = create_worker_output_answer( worker->name, buf, len, is_stdout );
	worker_send_message( worker, msg );	
	free( msg );
	
	g_free( buf );

	return TRUE;
}

static void *worker_func( void *thread_data )
{
	direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)thread_data;
	worker_t *worker = wed->worker;

	gchar **args;
	gint nof_args;
	GError *error = NULL;
	if( !g_shell_parse_argv( worker->command, &nof_args, &args, &error ) ) {
		fprintf( stderr, "ERROR: Failed to parse command of worker '%s', '%s': %s\n",
			worker->name, worker->command, error->message );
		g_strfreev( args );
		return NULL;
	}
		
	wed->sock = nn_socket( AF_SP, NN_PUSH );
	
	(void)nn_connect( wed->sock, worker->data );

	printf( "worker connected to data channel '%s'\n", worker->data );	
	
	error = NULL;
	gboolean ret = g_spawn_async_with_pipes( NULL,
		args, NULL, G_SPAWN_DO_NOT_REAP_CHILD, NULL,
		NULL, &wed->pid, NULL, &wed->out, &wed->err, &error );
	if( !ret ) {
		fprintf( stderr, "ERROR: Starting worker failed: %s\n", error->message );
		g_strfreev( args );
		return NULL;
	}

	printf( "Worker child with PID " PRIgid " created: %s.\n",
		wed->pid, worker->command );

	char *msg = create_worker_created_answer( worker->name, wed->pid );
	worker_send_message( worker, msg );	
	free( msg );
	
	GMainContext *context = g_main_context_default( );
	wed->main_loop = g_main_loop_new( context, TRUE );
                                                                                             
	g_child_watch_add( wed->pid, (GChildWatchFunc)watch_child, (gpointer *)wed );


#ifdef G_OS_WIN32
	wed->out_channel = g_io_channel_win32_new_fd( wed->out );
	wed->err_channel = g_io_channel_win32_new_fd( wed->err );
#else
	wed->out_channel = g_io_channel_unix_new( wed->out );
	wed->err_channel = g_io_channel_unix_new( wed->err );
#endif

	GIOCondition cond = (GIOCondition)( G_IO_IN | G_IO_HUP );
	g_io_add_watch( wed->out_channel, cond, (GIOFunc)watch_output, (gpointer *)wed );
	g_io_add_watch( wed->err_channel, cond, (GIOFunc)watch_output, (gpointer *)wed );

	worker->state = WORKER_STATE_RUNNING;

	printf( "worker %s: entering glib worker main loop..\n", worker->name );
	g_main_loop_run( wed->main_loop );
	printf( "worker %s: leaving glib worker main loop..\n", worker->name );

	worker->state = WORKER_STATE_STOPPED;

	(void)nn_shutdown( wed->sock, 0 );

	puts( "worker disconnected" );

	free( worker->execution_data );
	worker->execution_data = NULL;

	g_strfreev( args );
	
	return NULL;
}
                    
int worker_init( worker_t *worker )
{
	switch( worker->state ) {
		case WORKER_STATE_RUNNING:
			fprintf( stderr, "worker %s is already running\n", worker->name );
			return 0;
			
		case WORKER_STATE_STOPPED:
			// execute, skip
			break;
	}
	
	switch( worker->mode ) {
		case WORKER_EXECUTION_DIRECT:
			// execute, skip
			break;
			
		case WORKER_EXECUTION_DISABLED:
			fprintf( stderr, "worker %s is disabled, won't execute\n", worker->name );
			return 0;
	}

	pthread_attr_t attr;
	int res;
	
	res = pthread_attr_init( &attr );
	if( res != 0 ) {
		return 1;
	}

	worker->execution_data = malloc( sizeof( direct_glib_execution_worker_data_t ) );
	direct_glib_execution_worker_data_t *wed = 
		(direct_glib_execution_worker_data_t *)worker->execution_data;
	wed->worker = worker;
	
	res = pthread_create( &wed->thread, &attr, worker_func, (void *)wed );
	if( res != 0 ) {
		return 1;
	}
	
	pthread_detach( wed->thread );

	return 0;
}

void worker_terminate( worker_t *worker )
{
	direct_glib_execution_worker_data_t *wed = (direct_glib_execution_worker_data_t *)worker->execution_data;

	switch( worker->state ) {
		case WORKER_STATE_RUNNING:
			break;
			
		case WORKER_STATE_STOPPED:
			fprintf( stderr, "worker %s is already stopped\n", worker->name );
			return;
	}

#ifndef _WIN32
	if( kill( wed->pid, SIGTERM ) < 0 ) {
		fprintf( stderr, "ERROR: worker %s (PID " PRIgid ") could not be killed: %s\n",
			worker->name, wed->pid, strerror( errno ) );
	}
#else
	if( !TerminateProcess( wed->pid, 0 ) ) {
		fprintf( stderr, "ERROR: worker %s (PID " PRIgid ") could not be killed\n",
			worker->name, wed->pid );
	}
#endif
}

int worker_free( worker_t *worker )
{
	return 0;
}