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 "gpib.h"
#include "parser.h"

#include <stdlib.h>
#include <ctype.h>
#include <glib.h>
#include <fcntl.h>

#define STDIN_BUF_SIZE 1024


static gboolean periodic_poll (void);
static gboolean finish_boot (void);


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

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(globals.Remote.connections +1 > maxConn) {
		g_print_debug("Connection closed. Max reached\n");
		return TRUE;
	}
	globals.Remote.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 %d connections\n", globals.Remote.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");
	globals.Remote.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 ();
	g_thread_init (NULL);

	bus_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);

	GPIB_initialize();

	IO_Setup_RS232(	globals.Flash.baud,
	                globals.Flash.hardhand,
	                FALSE);

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

	globals.Timers.startup_timer_value = sec_timer ();

	Main_Rst();

	//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);

	GMainLoop *loop = g_main_loop_new (NULL, FALSE);

	g_timeout_add (20, (GSourceFunc) periodic_poll, NULL);
	g_timeout_add (100, (GSourceFunc) finish_boot, NULL);

	g_main_loop_run (loop);

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

	return 0;

}


static gboolean finish_boot (void)
{

#define MIN_STARTUP_DELAY 2
#define MAX_STARTUP_DELAY 120

	long on_delay = (long)globals.Flash.turn_on_dly;
	if (on_delay < MIN_STARTUP_DELAY) {
		on_delay = MIN_STARTUP_DELAY;
	}
	if (on_delay > MAX_STARTUP_DELAY) {
		on_delay = MAX_STARTUP_DELAY;
	}

	long timer_count;

	if ((timer_count=sec_timer()-globals.Timers.startup_timer_value) < on_delay) {
		gchar *message = g_strdup_printf ("%ld ", on_delay - timer_count);
		LCD_write(3,27,message);
		g_free (message);
		return TRUE;	// exit and call by timeout again
	}

	LCD_write(3,27,"OK");

	I2C_Setup_Monitor();

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

	globals.Sys.startup_complete = 1;

	Show_Main_Menu();

	return FALSE;	// no more calls to this function are needed
}


static gboolean periodic_poll (void)
{
	if (globals.HWDetect.beaglebone && bus_getpin (POWER_FAIL)) {

		globals.Sys.shutdown_started = TRUE;

		while (globals.Sys.flash_write_in_progress) {
			g_usleep(1000);
		}

		LCD_clear();
		//	       0123456789012345678901234567890123456789
                LCD_write(0,0,"PWR: Wait 10s. If still frozen, turn");
                LCD_write(1,0,"off 1min & retry.");

		// use plain old open to avoid any buffering etc
		int enablefd = open("/proc/sys/kernel/sysrq", O_SYNC | O_RDWR);
		int trgfd = open("/proc/sysrq-trigger", O_SYNC | O_RDWR);

		// enable sysrq
		write(enablefd, "1\n", 2);
		close(enablefd);

		// sync disks
		write(trgfd, "s\n", 2);

		// remount ro
		write(trgfd, "u\n", 2);
		close(trgfd);

		// poweroff
		system ("/usr/bin/systemctl poweroff -f");

		exit(0);
	}

	if (globals.Sys.startup_complete) {
		int i, output_on_time_so_far;

		for (i=0; i<(globals.Flash.ChanKey_output_state?globals.Flash.channels:1); ++i) {
			output_on_time_so_far = (int) (sec_timer()-globals.Timers.last_activity_at[i]);
			if (	(globals.Flash.output_timer[i]>0) &&
			                (globals.Timers.last_activity_at[i]>0) &&
			                (output_on_time_so_far > globals.Flash.output_timer[i])) {
				Set_Output_State(i,output_off);
				Show_Main_Menu();
			}
		}

		Update_Main_Menu_If_Visible();

		Menu_Check_Buttons ();

		for (i=0; i<max_channels; i++) {
			if (globals.Flash.monitor_enabled[i]) {
				I2C_Check_Monitors();
			}
		}

		Update_Main_Menu_If_Visible();

		GPIB_check_for_device_clear_signal();
		if (GPIB_check_for_messages(globals.Registers.gpib_buffer)) {
			if (GPIB_handle_new_input(globals.Registers.gpib_buffer)) {
				Parser_main(globals.Registers.gpib_buffer, 0, GPIB_send_query_response, NULL);
			}
		}
		GPIB_check_for_device_clear_signal();

		Update_Main_Menu_If_Visible();
	}

	return TRUE;
}