path: root/src/master.c

#include "master.h"

#include "pthread.h"

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

#include "json-c/json.h"

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

static pthread_t master_thread;
static int master_sock;
static int master_must_terminate = 0;

coordinator_t coordinator[MAX_COORDINATORS];

static void recompute_coordinator_states( void )
{
	time_t now;
	time( &now );
	for( int pos = 0; pos < MAX_COORDINATORS; pos++ ) {
		coordinator_t *c = &coordinator[pos];
		if( c->used ) {
			double d = difftime( now, c->lastAlive );
			fprintf( stderr, "%s %lld %lld %g\n", c->host, (long long)c->lastAlive, (long long)now, d );
			if( difftime( now, c->lastAlive ) > MAX_COORDINATOR_AGE ) {
				c->alive = false;
			}
		}
	}
}

static int register_coordinator( json_object *obj )
{
	json_object *host_obj;
	json_object_object_get_ex( obj, "host", &host_obj );
	const char *host = json_object_get_string( host_obj );

	json_object *os_obj;
	json_object_object_get_ex( obj, "os", &os_obj );
	const char *os = json_object_get_string( os_obj );

	json_object *arch_obj;
	json_object_object_get_ex( obj, "arch", &arch_obj );
	const char *arch = json_object_get_string( arch_obj );
	
	json_object *cpus_obj;
	json_object_object_get_ex( obj, "cpus", &cpus_obj );
	unsigned int cpus = json_object_get_int( cpus_obj );
	
	int pos;
	for( pos = 0; pos < MAX_COORDINATORS; pos++ ) {
		coordinator_t *coord = &coordinator[pos];
		if( coord->used &&
			strcmp( coord->host, host ) == 0 ) {
				time( &coord->lastAlive );
				coord->alive = true;
				fprintf( stderr, "master: already registered coordinator from host '%s' (%d)\n", host, pos );
			return 0;
		}
	}
	
	for( pos = 0; pos < MAX_COORDINATORS && coordinator[pos].used; pos++ );
		
	if( pos == MAX_COORDINATORS ) {
		fprintf( stderr, "Can't accept more coordinators, limit reached!\n" );
		return -1;
	}
	coordinator_t *coord = &coordinator[pos];

	printf( "master: registering coordinator from host '%s' (%d)\n", host, pos );
	
	coord->host = strdup( host );
	coord->os = strdup( os );
	coord->arch = strdup( arch );
	coord->cpus = cpus;
	coord->alive = true;
	time( &coord->lastAlive );
	coord->used = true;
	
	json_object *worker_array;
	json_object_object_get_ex( obj, "workers", &worker_array );	
	coord->nof_workers = json_object_array_length( worker_array );

	if( coord->nof_workers >= MAX_WORKERS ) {
		fprintf( stderr, "Can't define more workers per coordinator, limit reached!\n" );		
		return -1;
	}
	
	for( int i = 0; i < coord->nof_workers; i++ ) {
		json_object *worker_obj;
		worker_obj= json_object_array_get_idx( worker_array, i );
		
		json_object *name_obj;
		json_object_object_get_ex( worker_obj, "name", &name_obj );
		const char *name = json_object_get_string( name_obj );

		json_object *mode_obj;
		json_object_object_get_ex( worker_obj, "mode", &mode_obj );
		const char *mode = json_object_get_string( mode_obj );

		json_object *command_obj;
		json_object_object_get_ex( worker_obj, "command", &command_obj );
		const char *command = json_object_get_string( command_obj );
		
		worker_t *w = &coord->worker[i];
		
		w->name = strdup( name );
		w->mode = worker_execution_mode_from_str( mode );
		if( command != NULL ) {
			w->command = strdup( command );
		} else {
			w->mode = WORKER_EXECUTION_DISABLED;
		}
	}
	
	return 0;
}

static char *alloc_message( const char *msg )
{
	char *buf = strdup( msg );
	
	// crashes in survey mode!
	//~ size_t len = strlen( msg );
	//~ char *buf = (char *)nn_allocmsg( len + 1, 0 );
	//~ strncpy( buf, msg, len );

	return buf;
}

static char *create_discover_request( )
{
	json_object *obj = json_object_new_object( );
	json_object *op = json_object_new_string( "discover" );
	json_object_object_add( obj, "op", op );
	json_object *role = json_object_new_string( "master" );
	json_object_object_add( obj, "role", role );

	/* produce message as string, caller must free it */
	const char *msg = json_object_to_json_string( obj );
	char *res = alloc_message( msg );
	
	json_object_put( obj );
	
	return res;
}

static char *create_start_request( const char *name )
{
	json_object *obj = json_object_new_object( );
	json_object *op = json_object_new_string( "start" );
	json_object_object_add( obj, "op", op );
	json_object *role = json_object_new_string( "master" );
	json_object_object_add( obj, "role", role );
	json_object *worker = json_object_new_string( name );
	json_object_object_add( obj, "worker", worker );

	/* produce message as string, caller must free it */
	const char *msg = json_object_to_json_string( obj );
	char *res = alloc_message( msg );
           
	json_object_put( obj );
	
	return res;
}

#define MAX_NOF_COMMANDS 12

static int nof_queued_commands = 0;

char *queued_command[MAX_NOF_COMMANDS];

static pthread_mutex_t queued_command_mutex;

static int enqueue_request( char *cmd )
{
	pthread_mutex_lock( &queued_command_mutex );

	if( nof_queued_commands >= MAX_NOF_COMMANDS ) {
		fprintf( stderr, "ERROR: internal command queue overflow!\n" );
		pthread_mutex_unlock( &queued_command_mutex );
		return -1;
	}
	
	queued_command[nof_queued_commands] = cmd;
	nof_queued_commands++;

	pthread_mutex_unlock( &queued_command_mutex );
	
	return 0;
}

static char *get_next_request( )
{
	pthread_mutex_lock( &queued_command_mutex );
	char *cmd;
	if( nof_queued_commands > 0 ) {
		cmd = queued_command[0];
		--nof_queued_commands;
		memmove( &queued_command[0], &queued_command[1], nof_queued_commands * sizeof( char * ) );
		pthread_mutex_unlock( &queued_command_mutex );
		return cmd;
	} else {
		pthread_mutex_unlock( &queued_command_mutex );
		return create_discover_request( );
	}
}

static void *master_func( void *thread_data )
{
	char *control = (char *)thread_data;

	pthread_mutex_init( &queued_command_mutex, NULL );
	
	if( ( master_sock = nn_socket( AF_SP, NN_SURVEYOR ) ) < 0 ) {
		fprintf( stderr, "master, nn_socket( AF_SP, NN_SURVEYOR ) error: %s (%d)\n", strerror( errno ), errno );
		abort( );
	}
	
	int deadline = 5000;
	if( nn_setsockopt( master_sock, NN_SURVEYOR, NN_SURVEYOR_DEADLINE, &deadline, sizeof( deadline ) ) < 0 ) {
		fprintf( stderr, "master, nn_setsockopt error: %s (%d)\n", strerror( errno ), errno );
		abort( );
	}
	
	(void)nn_bind( master_sock, control );

	printf( "master connected to %s\n", control );

NEXT_COMMAND:
	;
	char *msg = get_next_request( );
	int msg_size = strlen( msg ) + 1;
		
	printf( "master send: %s\n", msg );
	int bytes = nn_send( master_sock, msg, msg_size /* NN_MSG */, 0 );
	if( bytes < 0 ) {
		if( errno == ETERM ) {
			master_must_terminate = 1;
		} else {
			fprintf( stderr, "ERROR: nn_send returned %d: %s (%d)\n",
				bytes, nn_strerror( errno ), errno );
		}
	} else if( bytes != msg_size ) {
		fprintf( stderr, "ERROR: truncated message!" );
	}
	// breaks in coordinator, so the memory is passed between threads?
	free( msg ); msg = NULL;

	while( !master_must_terminate ) {
		
		printf( "master idle: %d\n", master_must_terminate );
		
		recompute_coordinator_states( );
		
		char *answer = NULL;
		bytes = nn_recv( master_sock, &answer, NN_MSG, 0 );
		if( master_must_terminate ) continue;
		if( bytes >= 0 ) {
			json_object *recv_obj = json_tokener_parse( answer );
			const char *recv_obj_str = json_object_to_json_string( recv_obj );
			printf( "master received: %s\n", recv_obj_str );
			json_object *op_obj;
			json_object_object_get_ex( recv_obj, "op", &op_obj );
			const char *op = json_object_get_string( op_obj );

			if( strcmp( op, "register" ) == 0 ) {
				json_object *role_obj;
				json_object_object_get_ex( recv_obj, "role", &role_obj );
				const char *role = json_object_get_string( role_obj );
				if( strcmp( role, "coordinator" ) == 0 ) {
					if( register_coordinator( recv_obj ) < 0 ) {
						fprintf( stderr, "Terminating master because out of resources!\n" );
						master_must_terminate = 1;
						continue;
					}
				}
			}
			
			json_object_put( recv_obj );
			nn_freemsg( answer );
		}
		if( bytes < 0 ) {
			/* documentation says we should get a ETIMEDOUT, but in
			 * fact we get EFSM here, change if fixed in nanomsg */
			if( errno == EFSM /* ETIMEDOUT */ ) {
				goto NEXT_COMMAND;
			} else if( errno == EAGAIN || errno == EINTR ) {
				continue;
			} else if( errno == ETERM ) {
				master_must_terminate = 1;
				continue;
			} else {
				fprintf( stderr, "ERROR: nn_recv returned %d: %s (%d)\n",
					bytes, nn_strerror( errno ), errno );
			}
		}
	}

	(void)nn_shutdown( master_sock, 0 );

	puts( "master disconnected" );
	
	return NULL;
}

int master_init( const char *control )
{
	pthread_attr_t attr;
	int res;
	
	master_must_terminate = 0;

	memset( coordinator, 0, sizeof( coordinator_t * ) * MAX_COORDINATORS );

	res = pthread_attr_init( &attr );
	if( res != 0 ) {
		return 1;
	}
	
	res = pthread_create( &master_thread, &attr, master_func, (void *)control );
	if( res != 0 ) {
		return 1;
	}
		
	return 0;
}

void master_terminate( int terminate_nanomsg )
{
	master_must_terminate = 1;

	if( terminate_nanomsg ) {
		nn_term( );
	}
}

int master_free( )
{
	void *result;
	int res;
	
	res = pthread_join( master_thread, &result );
	if( res != 0 ) {
		return 1;
	}

	return 0;	
}

int master_start_worker( const char *name )
{
	char *msg = create_start_request( name );
	int res = enqueue_request( msg );

	return res;
}