path: root/instr-daemon.c

#include "socket-common.h"
#include "response.h"
#include "lcd.h"
#include "flash.h"
#include "globals.h"
#include "bus.h"
#include "device-functions.h"
#include "nicutils.h"
#include "monitor.h"
#include "menus.h"
#include <stdlib.h>
#include <ctype.h>
#include <glib.h>

#define STDIN_BUF_SIZE 1024

int port=3333; //port to listen
int connections=0;
int maxConn=16; //max connections - 16

guint signalMyCb; //signal to register which is used in cbClientInput(), step 10 from requirements
GAsyncQueue** stdinQueue=NULL;
GThread **peers; //actual connected peers


void send_message(gchar* message)
{
	if(NULL == message) {
		return;
	}
	int count=0;
	gssize size = strlen(message);
	if(!size) {
		return;
	}

	//send the final buffer to the queue
	for(count=0; count<maxConn; count++) {
		//allocate on the heap and deallocate in response.c, cbClientOutput()
		char *toSend = realloc(NULL, size+1);
		memcpy(toSend, message, size+1);

		//make sure we send to only valid peers
		if(peers[count] != 0) {
			g_async_queue_push(stdinQueue[count], toSend);
		}
	}
}

/** pulls an integer position from the vector which contains the
	position to use for the stdinQueue vector
	@param id - the thread id
*/
static int pullIndex(GThread* id)
{
	int i=0,ret=-1;
	static GStaticMutex mutex = G_STATIC_MUTEX_INIT;
	g_static_mutex_lock (&mutex);
	for(i=0; i<maxConn; i++)
		if(peers[i] == id) {
			peers[i] = (GThread*)0;
			ret=i;
			break;
		}
	g_static_mutex_unlock (&mutex);
	return ret;
}

/**
 * pushed in the peers vector the thread id for future reference
 * @param id
 */
static int pushIndex(GThread* id)
{
	int i=0,ret=-1;
	static GStaticMutex mutex = G_STATIC_MUTEX_INIT;
	g_static_mutex_lock (&mutex);
	for(i=0; i<maxConn; i++)
		if(peers[i] == 0) {
			peers[i] = id;
			ret=i;
			break;
		}

	g_static_mutex_unlock (&mutex);
	return ret;
}

//handler for incoming connection, closes the connection if maxCon connections are already handled
static gboolean
incomingConnection (GSocketService *service,
                    GSocketConnection      *connection,
                    GSocketListener        *listener,
                    gpointer                user_data)
{
	if(connections +1 > maxConn) {
		g_print_debug("Connection closed. Max reached\n");
		return TRUE;
	}
	connections++;
	g_print_debug("Incoming connection\n");
	return FALSE;
}


//thread connection handler
static gboolean
handler (GThreadedSocketService *service,
         GSocketConnection      *connection,
         GSocketListener        *listener,
         gpointer                user_data)
{


	GOutputStream *out;
	GInputStream *in;

	out = g_io_stream_get_output_stream (G_IO_STREAM (connection));
	in = g_io_stream_get_input_stream (G_IO_STREAM (connection));

	g_print_debug("Handling, connections: %d\n", connections);

	//register ourselves in the peers vector, use the index obtained in the stdinQueue
	//should not get -1
	GThread* self = g_thread_self();
	int index=pushIndex(self);

	GSource *outSource = NULL;
	GSource *inSource = g_pollable_input_stream_create_source((struct GPollableInputStream*)in, NULL);

	//get a reference to the async queue
	GAsyncQueue *queue = g_async_queue_ref(stdinQueue[index]);

	//input from client
	g_source_set_callback(inSource, cbClientInput, (GPollableOutputStream*)out, NULL);
	g_source_attach(inSource, NULL);

	//keep thread alive, every 1000 microseconds
	while(g_source_is_destroyed(inSource)==FALSE) {
		g_usleep(1000);
		gint elems;

		//verify our queue length and activate/deactivate the "out" Source for this connection
		//if we don't do this, the out Source will be scheduled frequently and will busy loop
		if((elems=g_async_queue_length(queue))>0) {
			if(outSource == NULL) {

				outSource = g_pollable_output_stream_create_source((struct GPollableOutputStream*)out, NULL);
				g_source_set_callback(outSource, cbClientOutput, queue, NULL);
				g_source_attach(outSource, NULL);
			} else {
				g_source_destroy(outSource);
				g_source_unref(outSource);
				outSource = NULL;
			}

		} else {
			if(outSource!= NULL && !g_source_is_destroyed(outSource)) {
				g_print_debug("Destroy source\n");
				g_source_destroy(outSource);
				g_source_unref(outSource);
				outSource = NULL; //added
			}
		}
		//end of activate/deactivate
	}

	//reached the end of the thread
	if (g_output_stream_close(out, NULL, NULL) == FALSE) {
		g_print_debug("out not closed\n");
	}
	if (g_input_stream_close(in, NULL, NULL) == FALSE) {
		g_print_debug("in not closed\n");
	}

	g_print_debug("Thread end\n");
	connections--; //keep track of connections
	g_async_queue_unref(queue); //unreference the queue
	pullIndex(g_thread_self()); //unregister from the peers vector
	return TRUE;
}

int main(int argc, char **argv)
{
	GSocketService *service = NULL;
	GError *error = NULL;
	GIOChannel* stdinChannel = NULL;

	g_type_init ();

	LCD_initialize();
	LCD_write(0,0,"Starting...");

	initFlash (&globals.Flash, FALSE, 0);

	gchar *message = g_strdup_printf ("%s, S/N %s", globals.Flash.model_num, globals.Flash.serial_num);
	LCD_display_extended_message (message, FALSE, FALSE);
	g_free (message);

	int i;
	for (i=0; i<8; i++) {
		set_dac(i,globals.Flash.initial_dac_settings[i]);
	}

	fixFlash(&globals.Flash);

	// count startups
	if (globals.Flash.self_cal) {
		++globals.Flash.self_cal_startups;
		int eprom_loc = (char *) &(globals.Flash.self_cal_startups) - (char *) &(globals.Flash.flash_start);
		writeUserBlock(&globals.Flash, eprom_loc, sizeof(globals.Flash.self_cal_startups));
	}

	nicinfo info;
	strcpy(info.ip,"-");
	nicutils_infofordefaultroute(&info);
	message = g_strdup_printf ("GPIB addr %d, Eth addr %s", globals.Flash.gpib_address,info.ip);
	LCD_write(2,0,message);
	g_free (message);

	// FIXME - init RS232

	/* start-up delay */
	LCD_write(3,0,"Warming up, please wait... ");

	globals.Timers.startup_timer_value = sec_timer ();
	long timer_count;
	while ((timer_count=sec_timer()-globals.Timers.startup_timer_value) < (long)globals.Flash.turn_on_dly) {
		message = g_strdup_printf ("%ld ", (long) globals.Flash.turn_on_dly - timer_count);
		LCD_write(3,27,message);
		g_free (message);
	}
	LCD_write(3,27,"OK");

	bus_init();

	I2C_Setup_Monitor();

	Main_Rst();
	Menu_Update_Display(); // FIXME?

	// FIXME - self-cal here - or in thread beside user sessions?

	// FIXME - need auto-timer output-off somewhere - launch event from set_output?

	// FIXME - need main loops, checks monitors too

	//register stdin channel
	stdinChannel = g_io_channel_unix_new(0);
	if(stdinChannel == NULL) {
		g_printerr("No io channel\n");
		exit(-1);
	}

	int idx=0;

	//allocate a maxConn queue on the heap
	stdinQueue = malloc(sizeof(struct GAsyncQueue*) * maxConn);
	//allocate peers vector on the heap
	peers = malloc(sizeof(GThread*) * maxConn);
	for(idx=0; idx<maxConn; idx++) {
		peers[idx] = 0;
		stdinQueue[idx] = g_async_queue_new();
	}

	//create a threaded service
	service = g_threaded_socket_service_new (maxConn+1);
	if (!g_socket_listener_add_inet_port (G_SOCKET_LISTENER (service),
	                                      port,
	                                      NULL,
	                                      &error)) {
		g_printerr ("%s: %s\n", argv[0], error->message);
		free(stdinQueue);
		free(peers);
		return 1;
	}

	//init the signal signalMyCb
	initSignals(&signalMyCb);

	g_print_debug("Server listening on port %d\n", port);
	g_signal_connect (service, "run", G_CALLBACK (handler), NULL);
	g_signal_connect (service, "incoming", G_CALLBACK(incomingConnection), NULL);

	g_main_loop_run (g_main_loop_new (NULL, FALSE));

	bus_shutdown();
	free(stdinQueue);
	free(peers);

	return 0;

}