/* $Id: periodic.c,v 1.64 2004/11/13 08:53:31 graziano Exp $ */

#include "config_nws.h"
#include <unistd.h>     /* getppid() */
#include <signal.h>     /* kill() */
#include <stdio.h>      /* sprintf() */
#include <stdlib.h>
#include <string.h>     /* strcmp() strlen() */
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>

#include "experiments.h"   /* experiment storage and registration */
#include "host_protocol.h"  /* MakeHostCookie() RegisterObject() */
#include "diagnostic.h"     /* ABORT() FAIL() WARN() */
#include "messages.h"       /* message facilities */
#include "osutil.h"         /* CurrentTime() */
#include "skills.h"         /* skill invocation */
#include "strutil.h"        /* GETTOK() SAFESTRCPY() */
#include "periodic.h"
#include "nws_sensor.h"


#define DEFAULT_PERIOD "10"

typedef struct {
	char name[63 + 1];			/* registration name */
	int period;				/* period of the activity */
	pid_t pid;				/* pid of the child */
	KnownSkills skill;			/* exercised skill */
	char skillOptions[EXP_LIST_SIZE];	/* options for this instance */
	unsigned long nextRun;			/* next time we need to run */
} PeriodicActivity;


static void *lock = NULL;		/* local lock */

/* Module globals.  */
static PeriodicActivity *activities = NULL;
static int activityCount = 0;
static int doFork = 1;			/* fork by deafult */

/*
 * Handles a #header#.message periodic control message arrived on #sd#
 * accompanied by #header#.
 */
static void
ProcessRequest(Socket *sd,
               MessageHeader header) {

	char *data;
	DataDescriptor descriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	DataDescriptor measDescriptor = SIMPLE_DATA(DOUBLE_TYPE, 3);
	double d[3];
	NWSAPI_Measurement toStore;

	switch(header.message) {
	case PERIODIC_EXPERIMENT:
		/* the first data are the measurement/value pair and the
		 * lenght of the series registration */
		if(!RecvData(*sd, d, &measDescriptor, 1, -1)) {
			ERROR("ProcessRequest: data receive failed\n");
			break;
		} else {
			/* series registration will be here */
			data = (char *)MALLOC((int)d[2] + 1);
			if(data == NULL) {
				ERROR("ProcessRequest: out of memory\n");
				break;
			} 
			descriptor.repetitions = (int)d[2];
			if(!RecvData(*sd, data, &descriptor, 1, -1)) {
				ERROR("ProcessRequest: data receive failed\n");
			} else {
				/* terminate the string */
				data[(int)d[2]] = '\0';

				/* got them: set the Experiment */
				toStore.timeStamp = d[0];
				toStore.measurement = d[1];

				/* register the experiment */
				RegisterExperiment(data, &toStore);
			}
			FREE(data);
		}

		break;
			
	default:
		DROP_SOCKET(sd);
		ERROR1("ProcessRequest: unknown message %d\n", header.message);
		break;
	}

	/* done with the socket */
	if (*sd != NO_SOCKET) {
		SocketIsAvailable(*sd);
	}
}


/*
** Registers #periodic# with the name server.
*/
static void
RegisterActivity(const PeriodicActivity *periodic) {
	int i, length;
	char option[EXP_LIST_SIZE], res[EXP_LIST_SIZE], skill[EXP_NAME_SIZE];
	const MeasuredResources *resources;
	Object toRegister;

	/* sanity check */
	if (periodic == NULL) {
		ERROR("RegisterActivity: NULL parameter\n");
		return;
	}

	/* let's fill in the activity */
	res[0] = '\0';
	if (!SkillResources(periodic->skill, &resources, &length)) {
		ERROR("RegisterActivity: failed to get info on skill\n");
		return;
	}
	for(i = 0; i < length; i++) {
		if(i > 0) {
			strcat(res, "\t");
		}
		strcat(res, ResourceName(resources[i]));
	}
	SAFESTRCPY(skill, SkillName(periodic->skill));
	sprintf(option, "period:%d", periodic->period);
	if(periodic->skillOptions[0] != '\0') {
		strcat(option, "\t");
		strcat(option, periodic->skillOptions);
	}
	toRegister = CreateActivityObject(periodic->name, 
			PERIODIC_CONTROL_NAME,
			EstablishedRegistration(),
			option,
			res,
			skill);
	RegisterObject(toRegister);
	FreeObject(&toRegister);
}


void
PeriodicChildDeath(int pid) {
	int i;
	char *name, *opts;
	KnownSkills skill;


	/* something funk cold have happened, like the child unexpectedly
	 * died: we try to restart it here in that case. We are working
	 * on global variables so we need to lock everything up */

	/* find the record for this child */
	opts = name = NULL;
	skill = 0;
	GetNWSLock(&lock);
	for(i = 0; i < activityCount; i++) {
		if(activities[i].pid != pid) {
			continue;
		}

		/* let's check we have a name */
		if (activities[i].name == NULL) {
			ERROR("PeriodicChildDeath: activity with no name\n");
		} else {
			LOG1("PeriodicChildDeath: %s was dead: restarting\n", activities[i].name);

			/* save old value: we'll stop and restart the
			 * activity */
			name = strdup(activities[i].name);
			skill = activities[i].skill;
			opts = strdup(activities[i].skillOptions);
			if (name == NULL || opts == NULL) {
				ERROR("PeriodicChildDeath: out of memory\n");
				break;
			}
		}
	}
	ReleaseNWSLock(&lock);

	/* let restart the activity if we need to */
	if (name != NULL) {
		StopPeriodicActivity(name);
		StartPeriodicActivity(name, SkillName(skill), opts);
		FREE(name);
		FREE(opts);
	}
}


int
PeriodicInit(const char *options) {
	static int initialized = 0;
	int i;
	KnownSkills skill;
	Object toRegister;
	char *tmp, skills[EXP_LIST_SIZE], opts[EXP_LIST_SIZE];

	/* initialize only once */
	GetNWSLock(&lock);
	if (initialized) {
		ReleaseNWSLock(&lock);
		ERROR("PeriodicInit: you can initialize only once!\n");
		return 0;
	}
	initialized = 1;
	ReleaseNWSLock(&lock);

	/* we need to remember if we are forking */
	tmp = GetOptionValue(options, "fork", NULL);
	if (tmp == NULL || strncmp(tmp, "yes", (strlen(tmp) > 3 ? 3 : strlen(tmp))) == 0) {
		doFork = 1;
	} else {
		doFork = 0;
	}
	FREE(tmp);

	/* Register the periodic control. */
	skills[0] = '\0';
	for(i = 0; i < SKILL_COUNT; i++) {
		skill = (KnownSkills) i;
		if(!SkillAvailableForControl(skill, options, PERIODIC_CONTROL))
			continue;

		if (skills[0] != '\0')
			strcat(skills, "\t");
		strcat(skills, SkillName(skill));
	}
	SAFESTRCPY(opts, "period:0_to_1_int");
	toRegister = CreateControlObject(NULL, 
			PERIODIC_CONTROL_NAME,
			EstablishedRegistration(),
			opts,
			skills);
	RegisterObject(toRegister);
	FreeObject(&toRegister);

	RegisterListener(PERIODIC_EXPERIMENT, "PERIODIC_EXPERIMENT", &ProcessRequest);

	return 1;
}

double
NextPeriodicWork() {
	int i;
	double next;

	next = 0.0;

	GetNWSLock(&lock);
	for (i = 0; i < activityCount; i++) {
		/* if there is a child working on it, skip it */
		if (activities[i].pid != 0) {
			continue;
		}

		if (next == 0.0 || next >= activities[i].nextRun) {
			next = activities[i].nextRun;
		}
	}
	ReleaseNWSLock(&lock);

	return next;
}

/* local function wrapper to UseSkill */
static int
DoWork(		const PeriodicActivity *activity,
		Socket childToParent) {
	char *opts, *tmp;
	int length, i, ret;
	SkillResult *results;
	double d[3];
	NWSAPI_Measurement toStore;
	Object toRegister;
	DataDescriptor nameDescriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	DataDescriptor measDescriptor = SIMPLE_DATA(DOUBLE_TYPE, 3);
	
	if (activity->skillOptions != NULL) {
		opts = strdup(activity->skillOptions);
	} else {
		opts = strdup("");
	}
	if (opts == NULL) {
		ERROR("DoWork: out of memory\n");
		return 0;
	}
	/* add fork option if not already present */
	tmp = GetOptionValue(activity->skillOptions, "fork", NULL);
	if (tmp == NULL) {
		opts = (char *)REALLOC(opts, strlen(opts) + 10);
		if (opts == NULL) {
			ERROR("DoWork: out of memory\n");
			return 0;
		}
		if (opts[0] != '\0') {
			strcat(opts, "\t");
		}
		sprintf(opts + strlen(opts), "fork:%s", (doFork) ? "yes" : "no");
	}
	FREE(tmp);

	UseSkill(activity->skill,
			opts,
			activity->period,
			&results,
		  	&length);
	FREE(opts);

	/* let's fill in the fixed fields */
	toStore.timeStamp = d[0] = CurrentTime();

	/* pass the data to dad */
	ret = 1;
	for(i = 0; i < length && ret; i++) {
		if(!results[i].succeeded) {
			ret = 0;
			WARN1("DoWork: %s failed\n", activity->name);
			continue;
		}

		/* we need the generate the series registration */
		toRegister = CreateSeriesObject(opts, 
				EstablishedRegistration(),
				ResourceLabel(results[i].resource),
				"noname",
				results[i].options,
				ResourceName(results[i].resource),
				activity->name);

		/* if we forked, we need to send a message to the parent ... */
		if (doFork) {
			/* let's tell the parent to store the
			 * experiment: the parent knows which
			 * memory is active right now, so it will
			 * do the right thing */
			d[1] = results[i].measurement;

			/* this is a trick: to keep the code
			 * simple we put the lenght of the
			 * series registration as 3 values here */
			d[2] = strlen(toRegister);
			nameDescriptor.repetitions = (int)d[2];
			if (!SendMessageAndDatas(childToParent,
						PERIODIC_EXPERIMENT,
						d,
						&measDescriptor,
						1,
						toRegister,
						&nameDescriptor,
						1,
						0)) {
				/* if we cannot talk to our
				 * parent there is no point on
				 * going ahead (we don't save the
				 * results directly) */
				ret = 0;
				ERROR("DoWork: failed to talk to parent\n");
			}
		} else {
			/* ... otherwise we register ourself */
			toStore.measurement = results[i].measurement;
			RegisterExperiment(toRegister, &toStore);
		}
		/* free the memory */
		FreeObject(&toRegister);
	}
	FreeSkillResults(length, &results);

	return ret; 
}


void
PeriodicWork(void) {
	int i;
	double now;
	PeriodicActivity tmp;
	
	now = CurrentTime();

	GetNWSLock(&lock);
	for (i = 0; i < activityCount; i++) {
		/* if there is a child working on it, skip it */
		if (activities[i].pid != 0) {
			continue;
		}
		/* if it is no time yet, skip it */
		if (activities[i].nextRun > now) {
			continue;
		}
		
		/* given that activities is a global array, we need to
		 * copy into a local temp */
		tmp = activities[i];
		ReleaseNWSLock(&lock);

		tmp.nextRun = now + tmp.period;

		/* let's try to do the work */
		DoWork(&tmp, NO_SOCKET);

		/* get the lock and keep it around the for to protect
		 * access to activityCount */
		GetNWSLock(&lock);
		activities[i].nextRun = now + activities[i].period;
	}
	ReleaseNWSLock(&lock);
}


int
RecognizedPeriodicActivity(const char *name) {
	int i, ret;

	/* sanity check */
	if (name == NULL) {
		ERROR("RecognizedPeriodicActivity: NULL parameter\n");
		return 0;
	}

	ret = 0;
	GetNWSLock(&lock);
	for(i = 0; i < activityCount && !ret; i++) {
		ret = strcmp(activities[i].name, name) == 0;
	}
	ReleaseNWSLock(&lock);

	return ret;
}


int
StartPeriodicActivity(	const char *registration,
			const char *skill,
			const char *options) {
	Socket childToParent;
	int i, ret;
	PeriodicActivity newActivity;
	double nextRunTime;
	struct timeval tv;
	char *tmp;

	/* sanity check */
	if (skill == NULL || options == NULL) {
		ERROR("StartPeriodicActivity: NULL parameters\n");
		return 0;
	}
	
	/* let's see the name of this activity */
	if (registration != NULL) {
		SAFESTRCPY(newActivity.name, registration);
	} else {
		tmp = NameOfActivity_r(EstablishedRegistration(),
				skill, 
				options);
		SAFESTRCPY(newActivity.name, tmp);
		free(tmp);
	}

	for(i = 0; i < SKILL_COUNT; i++) {
		KnownSkills s = (KnownSkills) i;
		if (!SkillAvailableForControl(s, options, PERIODIC_CONTROL)) {
			continue;
		}
		if(strcmp(skill, SkillName(s)) == 0) {
			newActivity.skill = s;
			break;
		}
	}

	/* let's check if we have the skill ability */
	if(i == SKILL_COUNT) {
		ERROR1("StartPeriodicActivity: Unsupported skill %s requested\n", skill);
		return 0;
	}


	tmp = GetOptionValue(options, "period", DEFAULT_PERIOD);
	newActivity.period = strtod(tmp, NULL);
	FREE(tmp);
	SkillOptions(newActivity.skill, options, newActivity.skillOptions, sizeof(newActivity.skillOptions));

	/* next service is now */
	newActivity.nextRun = 1;

	/* let's see if we need to fork or not */
	if (doFork) {
		if(!CreateLocalChild(&newActivity.pid, NULL, &childToParent)) {
			ERROR("StartPeriodicActivity: fork failed\n");
			return 0;
		}
	} else {
		/* non-forking: nothing to do here */
		newActivity.pid = 0;
	}
		

	if((doFork && newActivity.pid != 0) || !doFork) {  
		/* Parent process. */
		/* kill all childs already doing this activity */
		if (StopPeriodicActivity(newActivity.name)) {
			LOG("StartPeriodicActivity: stopped previous activity.\n");
		}

		GetNWSLock(&lock);
		activityCount++;
		activities = (PeriodicActivity *) REALLOC(activities, activityCount * sizeof(PeriodicActivity));
		if (activities == NULL) {
			ABORT("StartPeriodicActivity: out of memory\n");
		}
		activities[activityCount - 1] = newActivity;
		ReleaseNWSLock(&lock);
		RegisterActivity(&newActivity);
		return 1;
	}

	/* sanity check: we shouldn't get here if we don't want to fork */
	if (!doFork) {
		ERROR("StartPeriodicActivity: forking when asked not to!\n");
		return 0;
	} else {
		/* let's give some time for the parent to get ready to
		 * listen to us: got problems at least on Solaris 5.8 */
		nextRunTime = CurrentTime() + 1;
	}

	/* Run as long as the parent process runs. */
	for (ret = 1; ret ; ) {
		/* let's go till our parent is available */
		if (getppid() == 1) {
			/* init is our parent: we are done here */
			ret = 0;
			continue;
		}

		/* let's sleep till the next exepriment. */
		tv.tv_sec = nextRunTime - CurrentTime();
		if(tv.tv_sec > 0) {
			/* we use select instead of sleep because some
			 * system (ie condor) doesn't like us to use it */
			tv.tv_usec = 0;
			PortabilitySelect(0, NULL, NULL, NULL, &tv, NULL);
		}
		nextRunTime = CurrentTime() + newActivity.period;

		/* let's try to do the work */
		ret = DoWork(&newActivity, childToParent);
	}
	exit(0);
	return 0; 
}


int
StopPeriodicActivity(const char *registration) {
	int i, j, ret;

	/* sanity check */
	if (registration == NULL) {
		ERROR("StopPeriodicActivity: NULL parameter\n");
		return 0;
	}

	ret = 0;

	/* this is quite brutal: we lock the whole loop since we play
	 * with activities all along. Moreover if we are terminating and
	 * we are asked to stop we need to wait the lock! Let's by pass
	 * it when we are going to terminate */
	if (registration[0] != '\0') {
		GetNWSLock(&lock);
	}
	for(i = 0; i < activityCount; i++) {
		/* if the registration is empty stop everything,
		 * otherwise look for the right activity */
		if(registration[0] == '\0' || strcmp(activities[i].name, registration) == 0) {

			if (activities[i].pid) {
				/* nice way to stop the activity: we don't want
				 * the child to call TermHandler(). We
				 * don't care to check the outcome
				 * because we can be called from
				 * PeriodicChildDeath (pid
				 * non-existent). */
				kill(activities[i].pid, SIGKILL);
			} 

			/* remove traces of this activity */
			UnregisterObject(activities[i].name);

			/* remove the record */
			for(j = i; j < (activityCount -1); j++) {
				activities[j] = activities[j+1];
			}

			activities = REALLOC(activities, --activityCount * sizeof(PeriodicActivity));
			if (activities == NULL && activityCount > 0) {
				ABORT("StopPeriodicActivity: out of memory\n");
			}

			/* now there is one less activityCount and i
			 * points to the next activity: adjust */
			i--;

			/* we stopped an activity */
			ret = 1;
		}
	}
	if (registration[0] != '\0') {
		ReleaseNWSLock(&lock);
	}

	return ret;
}