/* $Id: nws_memory.c,v 1.176 2004/11/05 21:40:02 graziano Exp $ */

#include "config_nws.h"

#include <errno.h>           /* errno */
#include <stdio.h>           /* sprintf() */
#include <unistd.h>          /* getopt() stat() */
#include <stdlib.h>
#include <string.h>          /* strchr() strstr() */

#include "protocol.h"
#include "nws_state.h"
#include "host_protocol.h"
#include "diagnostic.h"
#include "strutil.h"
#include "dnsutil.h"
#include "osutil.h"
#include "messages.h"
#include "nws_memory.h"
#include "skills.h"
#include "nws_db.h"

/*
 * This program implements the NWS memory host.  See nws_memory.h for supported
 * messages.
 */

#define KEEP_A_LONG_TIME 315360000.0

/* we keep a list of series we know about, so that we can register them
 * with the nameserver. We have 2: one for the orphaned series and one
 * for the registered series: the orphaned will disappears when the new
 * series kicks in. */
static registrations *mySeries;		/* where we keep series registrations */
static registrations *orphanedSeries;	/* where we keep series registrations */

/* performance data */
static double fetchTime;		/* time spent serving FETCH */
static double oldStoreTime;		/* time spent serving STORE */
static double storeTime;		/* time spent serving the new STORE */
static double autoFetchTime;		/* time spent serving autofetching */
static unsigned long fetchQuery;
static unsigned long oldStoreQuery;
static unsigned long storeQuery;

/*
 * Information about a registered auto-fetch request. The list of series
 * the client is interested in and the socket is kept here.
 */
typedef struct {
	registrations *reg;
	Socket clientSock;
} AutoFetchInfo;


/*
 * Module globals.  #autoFetches# is the list of registered auto-fetch requests
 * that we're still servicing; #autoFetchCount# is the length of this list.
 * #fileSize# contains the maximum number of records allowed in files.
 */
AutoFetchInfo *autoFetches = NULL;
static size_t autoFetchCount = 0;
static size_t fileSize;

/* select which backend to use to store data: so far localdir is the most
 * stable and tested one. */
typedef enum {LOCALDIR, NETLOGGER, POSTGRES} memType;
static memType backType = LOCALDIR;

/*
 * Called when an operation fails.  Cleans up any bad connections and, if we've
 * run out, frees up a connection so that others may connect.
 */
static void
CheckConnections(void) {
	if(CloseDisconnections() == 0) {
		CloseConnections(0, 1, 1);
	}
}

/*
 * this function goes through a string which contains states separated
 * by '\n' and check if they are valid registrations. It starts from the
 * end of the string, check if the name is in #good# (the good list of
 * registrations) otherwise it adds to #orphan#. Checks only #howMany#
 * entries at a time. Returns 1 if more entries need to be processed.
 */
static int
CheckOldNames(	char *names,
		registrations *good,
		registrations *orphan,
		int howMany) {
	char *where, *tmp;
	Object obj;
	int nskill, i, k, done;
	MeasuredResources m;
	KnownSkills t;

	/* sanity check */
	if (orphan == NULL || good == NULL) {
		ERROR("CheckOldNames: NULL parameter\n");
		return 0;
	}
	if (howMany < 0) {
		howMany = 10;
	}
	if (names == NULL) {
		/* we are done */
		return 0;
	}

	for (where = names + strlen(names); where > names; *(--where) = '\0') {
		/* look for the beginning of the filename */
		while (*where != '\n' && where > names) {
			where--;
		}
		where++;

		/* let's check if we have just a newline */
		if (strlen(where) <= 0) {
			continue;
		}

		/* let's check the filename/table */
		switch (backType) {
		case LOCALDIR:
			if (!CheckFileName(where)) {
				INFO1("CheckOldNames: %s is not a valid NWS filename\n", where);
				continue;
			}
			break;

		case POSTGRES:
			if (!CheckTable(where)) {
				INFO1("CheckOldNames: %s is not a valid NWS table\n", where);
				continue;
			}
			break;
		
		case NETLOGGER:
			INFO("CheckOldNames: not implemented for netlogger\n");
			return 0;
			break;
		}

		/* if the series is already registered in the good lists,
		 * we skip it */
		if (SearchForName(good, where, 1, &i)) {
			continue;
		}

		/* if we already have it in the orphaned series we skip: 
		 * we need it anyway to find where to insert the object */
		if (SearchForName(orphan, where, 1, &i)) {
			continue;
		}

		/* we need to insert it: let's create the object first
		 * then add it */
		obj = NewObject();
		AddNwsAttribute(&obj, "name", where);
		AddNwsAttribute(&obj, "memory", EstablishedRegistration());
		AddNwsAttribute(&obj, "objectclass", "nwsSeries");

		/* let's find which resource we are dealing with */
		for (m = 0; m < RESOURCE_COUNT; m++) {
			if (strstr(where, ResourceName(m)) != NULL) {
				break;
			}
		}
		if (m >= RESOURCE_COUNT) {
			INFO1("CheckOldNames: unknown resource in %s\n", where);
			continue;
		}

		/* now, we are checking if the resource as a target
		 * options: if so the target is at the end of the name,
		 * after the options. This is from Ye  code. Thanks Ye! */ 
		/* to do that let's find the skill responsible for the
		 * resource */
		done = 0;
		for (t = 0; t < SKILL_COUNT; t++) {
			const MeasuredResources *resources;

			if (!SkillResources(t, &resources, &nskill)) {
				INFO("CheckOldNames: failed to get info on skill\n");
				continue;
			}
			for (k = 0; k < nskill; k++) {
				if (resources[k] == m) {
					/* we found the right skill:
					 * let's get out of here */
					done = 1;
					break;
				}
			}
			if (done) {
				break;
			}
		}

		/* if we didn't find the skill, it's a very bad news. */
		if (t >= SKILL_COUNT) {
			WARN1("CheckOldNames: unable to find skill for resource %s\n", ResourceName(m));
			continue;
		}

		/* now let's look if we have a target option: clique
		 * control adds it. */
		if (SkillAvailableForControl(t, "", CLIQUE_CONTROL)) {
			const char *options;
			options = SkillSupportedOptions(t);
			if (options == NULL) {
				/* I guess we don't have the target */
				continue;
			}
			/* let's count how many options we have (the are
			 * comma separated so there are 1 + # of commas) */
			for (done = 1; *options != '\0'; options++) {
				if (*options == ',') {
					done++;
				}
			}

			/* now let's get the target: it is done commas
			 * away from the resource name */
			tmp = strstr(where, ResourceName(m));
			/* we have an extra '.' (the one after
			 * the resource name */
			for (done++; *tmp != '\0'; tmp++) {
				if (*tmp == '.') {
					done--;
				}
				if (done == 0) {
					/* we got it: let's pass the dot */
					tmp++;
					AddNwsAttribute(&obj, "target", tmp);
					break;
				}
			}
		}
		
		/* now let's add the resource and the host */
		AddNwsAttribute(&obj, "resource", ResourceName(m));

		/* if we have the resource we can take an educate
		 * guess on what the host is */
		tmp = strstr(where, ResourceName(m));
		if (tmp != NULL) {
			tmp--;
			if (*tmp == '.') {
				*tmp = '\0';
				AddNwsAttribute(&obj, "host", where);
			}
		}

		InsertRegistration(orphan, obj, 10, i);
		FreeObject(&obj);

		/* done with this name */
		*where = '\0';

		if (--howMany <= 0) {
			/* done for this round */
			break;
		}
	}

	return (howMany <= 0);
}


/* 
 * this function looks for series that needs to be registered (expiration
 * will tell) with the nameserver. We register only #hoeMany# of them at a
 * time. Returns 1 if more registration work needs to be done.
 */
static int
RegisterSeries(	registrations *r, 
		int howMany) {
	int i, j;
	unsigned long now;
	
	/* sanity check */
	if (r->howMany <= 0) {
		return 0;
	}

	now = (unsigned long)CurrentTime();

	for (i = 0, j = 0; j < r->howMany; j++) {
		if (r->expirations[j] < now) {
			/* need to be registered */
			if (RegisterWithNameServer(r->vals[j], DEFAULT_HOST_BEAT * 10)) {
				r->expirations[j] = now + DEFAULT_HOST_BEAT * 5;
				/* we register only howMany series at a time */
				if (i++ >= howMany) 
					break;
			}
		}
	}

	return  (j < r->howMany);
}

/* 
 * This is somewhat hopeless, but we tried to unregister all the series!!
 */
static void
UnregisterSeries(	registrations *r) {
	int j;
	char *name, filter[255];
	const char *myName;
	
	myName = EstablishedRegistration();
	for (j = 0; j < r->howMany; j++) {
		/* let's get the name and unregister the beast */
		name = NwsAttributeValue_r(FindNwsAttribute(r->vals[j], "name"));
		if (name == NULL) {
			WARN("UnregisterSeries: object with no name??\n");
			continue;
		}
		snprintf(filter, 255, "%s)(memory=%s", name, myName);
		FREE(name);

		UnregisterObject(filter);
	}

	return;
}
/* todo list when we exit */
static int
MyExit(void) {
	LOG("MyExit: unregistering series, it may take sometime!\n");
	UnregisterSeries(mySeries);

	return 1;
}


/*
 * this functions gets a #registration# of a Series in input and looks for
 * it in the lists of Series names. If not found it adds it. 
 */
static int
StoreSeriesName(char *registration) {
	int ind;
	Object obj;
	char *name;

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

	/* we'll need the name later */
	name = NwsAttributeValue_r(FindNwsAttribute(registration, "name"));
	if (name == NULL) {
		ERROR1("StoreSeriesName: object without a name (%s)!\n", registration);
		return 0;
	}

	/* before doing anything else, let's set straight who is the
	 * memory here */
	obj = strdup(registration);
	DeleteNwsAttribute(&obj, "memory");
	AddNwsAttribute(&obj, "memory", EstablishedRegistration());

	/* we need to see if there is similar registration in the
	 * orphaned list: if so remove it and add it to the good list */
	if (SearchForName(orphanedSeries, name, 1, &ind)) {
		/* we found it: remove it from orphaned ... */
		DeleteRegistration(orphanedSeries, ind);
		/* .. and from the nameserver */
		UnregisterObject(name);
	}
	FREE(name);

	/* let's look for the registration in the good list */
	if (!SearchForObject(mySeries, obj, &ind)) {
		/* expiration is bogus so that we'll register the series
		 * right away */
		InsertRegistration(mySeries, obj, 10, ind);
	}
	FREE(obj);

	return 1;
}

/*
 * A "local" function of ProcessRequest().  Implements the MEMORY_CLEAN service
 * by deleting all files in the memory directory that have not been accessed
 * within the past #idle# seconds.  Returns 1 if successful, else 0.
 */
static int
DoClean(unsigned long idle) {
	int ret;

	switch (backType) {
	case LOCALDIR:
		ret = CleanLocalStates(idle);
		break;

	case POSTGRES:
		INFO("DoClean: no cleaning implemented when using database\n");
		ret = 0;
		break;
	
	case NETLOGGER:
		INFO("DoClean: no cleaning implemented when using netlooger\n");
		ret = 0;
		break;
	}

	return ret;
}


/*
 * Removes from the #auotFetchInfo# module variable all information for the
 * client connected to #sock#.
 */
static void
EndAutoFetch(Socket sock) {
	int i;

	/* sanity check */
	if (sock == NO_SOCKET) {
		return;
	}

	for(i = 0; i < autoFetchCount; i++) {
		if(autoFetches[i].clientSock == sock) {
			/* feedback */
			INFO1("EndAutoFetch: cleaning after socket %d\n", sock);

			/* remove all the registrations */
			while(autoFetches[i].reg->howMany > 0) {
				DeleteRegistration(autoFetches[i].reg, 0);
			}
			autoFetches[i].clientSock = NO_SOCKET;
		}
	}
}

/* Used in ProcessRequest to store the data.
 * Stores #data# in #directory# with the attributes indicated by #s#.
 * Fails if the record size and count in #s# yield more than #len#
 * total bytes.  Returns 1 if successful, else 0.
 */
static int
KeepState(	const struct nws_memory_state *s,
		const char *data,
		size_t len) {
	const char *curr;
	int i, ret;
	size_t recordSize;

	/* sanity check */
	if (s == NULL || data == NULL) {
		ERROR("KeepState: NULL parameter\n");
		return 0;
	}
	if(s->rec_count > fileSize) {
		FAIL("KeepState: rec count too big\n");
	}
	if(s->rec_size > MAX_RECORD_SIZE) {
		WARN("KeepState: state record too big.\n");
		recordSize = MAX_RECORD_SIZE;
	} else {
		recordSize = (size_t)s->rec_size;
	}
	if(s->rec_count * recordSize > len) {
		FAIL1("KeepState: too much data %d\n", s->rec_count * recordSize);
	}

	/* all right, we are clear to store the data */
	ret = 1;
	for(curr = data, i = 0; i < s->rec_count; curr += recordSize, i++) {
		switch (backType) {
		case LOCALDIR:
			ret = WriteState(s->id,
					fileSize,
					s->time_out,
					s->seq_no,
					curr,
					recordSize);
			if (!ret) {
				/* as of 2.8.2 we disabled the journal */
				/*|| !EnterInJournal(s->id, s->seq_no)) */
				if(errno == EMFILE) {
					CheckConnections();
				}
			}
			break;

		case POSTGRES:
			/* let's go to the database */
			ret = WriteNwsDB(s->id, 
					s->time_out, 
					s->seq_no, 
					curr, 
					recordSize);
			break;

		case NETLOGGER:
#ifdef WITH_NETLOGGER
			ret = WriteStateNL(logLoc->path,
					s->id,
					fileSize,
					s->time_out,
					s->seq_no,
					curr,
					recordSize);
#endif
			break;
		}
		/* let's see how did it go */
		if (ret == 0) {
			ERROR("KeepState: write failed\n");
			break;
		}
	}

	return ret;
}


static void
NotifyFetchers(	const struct nws_memory_state *s,
		const char *data,
		size_t len) {
	DataDescriptor contentsDescriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	int i, j;
	Socket sock;

	/* set the size of the experiments string to be sent */
	contentsDescriptor.repetitions = len;

	/* forward to the autofetch clients */
	for (i = 0; i < autoFetchCount; i++) {
		if (autoFetches[i].clientSock == NO_SOCKET) {
			/* empty slot */
			continue;
		}

		if (SearchForName(autoFetches[i].reg, s->id, 1, &j)) {
			if (!SendMessageAndDatas(autoFetches[i].clientSock, STATE_FETCHED, s, stateDescriptor, stateDescriptorLength, data, &contentsDescriptor, 1, -1)) {
				/* some feedback */
				LOG1("NotifyFetchers: failed for socket %d\n", autoFetches[i].clientSock);
	
				/* now let's remove traces for this guy */
				sock = autoFetches[i].clientSock;
				EndAutoFetch(autoFetches[i].clientSock);
				DROP_SOCKET(&sock);
			}
		}
	}
}

static int
AddAutoFetcher(	Socket *sd,
		char *stateNames) {
	int i, ret, j;
	const char *word;
	char name[255 + 1];
	AutoFetchInfo *expandedAutoFetches;
	Object obj;

	/* sanity check */
	if (*sd == NO_SOCKET || stateNames == NULL) {
		WARN("AddAutoFetcher: NULL parameter(s)\n");
		return 0;
	}

	/* we have a new list of series to autoFetch: is
	 * this an old client? */
	for(j = -1, i = 0; i < autoFetchCount; i++) {
		if (autoFetches[i].clientSock == NO_SOCKET && j == -1) {
			/* remember empty slot */
			j = i;
		}
		if (autoFetches[i].clientSock == *sd) {
			/* old client: remove the old list */
			EndAutoFetch(*sd);
		}
	}

	/* let's make room for the new autofetcher */
	if (j == -1) {
		expandedAutoFetches = REALLOC(autoFetches, (i+1)*sizeof(AutoFetchInfo));
		if (expandedAutoFetches == NULL) {
			ERROR("AddAutoFetcher: out of memory\n");
			return 0;
		}
		autoFetchCount++;
		autoFetches = expandedAutoFetches;
		j = i;

		/* let's set the new registrations */
		if (!InitRegistrations(&autoFetches[j].reg)) {
			ERROR("AddAutoFetcher: failed to init structure\n");
			EndAutoFetch(*sd);
			return 0;
		}
	}
	autoFetches[j].clientSock = *sd;

	/* let's get the registrations one by one */
	for (word = stateNames; GETWORD(name, word, &word); ) {
		/* let's get the right spot top add it */
		if (SearchForName(autoFetches[j].reg, name, 1, &ret)) {
			/* we already have it */
			LOG1("AddAutoFetcher: duplicate series (%s)\n", name);
			continue;
		}
					
		/* set the object */
		obj = NewObject();
		AddNwsAttribute(&obj, "name", name);

		if (!InsertRegistration(autoFetches[j].reg,
				obj,
				0,
				ret)) {
			ABORT("AddAutoFetcher: failed to insert series!\n");
		}
		FreeObject(&obj);
	}

	INFO1("AddAutoFetcher: we have %d autofetchers\n", autoFetchCount);

	return 1;
}


/* wrap around the reading functions */
static int
ReadRecord(	char *where,
		int maxSize,
		struct nws_memory_state *stateDesc) {
	int ret = 0;

	/* let's see which backend are we using */
	switch (backType) {
	case LOCALDIR:
		ret = ReadState(	stateDesc->id,
					where,
					stateDesc->rec_count,
					stateDesc->rec_count * maxSize,
					stateDesc->seq_no,
					&stateDesc->time_out,
					&stateDesc->seq_no,
					&stateDesc->rec_count,
					&stateDesc->rec_size);
		break;

	case POSTGRES:
		ret = ReadNwsDB(	stateDesc->id,
					where,
					stateDesc->rec_count,
					stateDesc->rec_count * maxSize,
					stateDesc->seq_no,
					&stateDesc->time_out,
					&stateDesc->seq_no,
					&stateDesc->rec_count,
					&stateDesc->rec_size);
		break;

	case NETLOGGER:
		ERROR("HELP!\n");
		break;
	}

	return ret;
}


/*
 * A "local" function of main().  Handles a #header#.message message arrived on
 * #sd# accompanied by #header#.dataSize bytes of data.
 */
static void
ProcessRequest(	Socket *sd,
		MessageHeader header) {
	char *contents;
	unsigned long expiration;
	DataDescriptor contentsDes = SIMPLE_DATA(CHAR_TYPE, 0);
	DataDescriptor expDescriptor = SIMPLE_DATA(UNSIGNED_LONG_TYPE, 1);
	DataDescriptor stateNamesDescriptor = SIMPLE_DATA(CHAR_TYPE, 0);
	int i, ret;
	struct nws_memory_state stateDesc;
	struct nws_memory_new_state newStateDesc;
	char *stateNames, *tmp;

	/* makes the compiler happy */
	contents = NULL;

	switch(header.message) {
	case FETCH_STATE:
		/* got an extra query */
		fetchQuery++;
		fetchTime -= MicroTime();

		if(!RecvData(*sd, &stateDesc, stateDescriptor, stateDescriptorLength, -1)) {
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: state receive failed\n");
		} else {
			/* we are cool */
			contents = (char *)MALLOC(stateDesc.rec_count * MAX_RECORD_SIZE);
			if(contents == NULL) {
				(void)SendMessage(*sd, MEMORY_FAILED, -1);
				ERROR("ProcessRequest: out of memory\n");
			} else {
				if(ReadRecord( contents,
						MAX_RECORD_SIZE,
						&stateDesc)) {
					if(stateDesc.rec_count > 0) {
						contentsDes.repetitions = stateDesc.rec_size * stateDesc.rec_count;
						(void)SendMessageAndDatas(*sd,
							STATE_FETCHED,
							&stateDesc,
							stateDescriptor,
							stateDescriptorLength,
							contents,
							&contentsDes,
							1,
							-1);
					} else {
						(void)SendMessageAndData(*sd,
							STATE_FETCHED,
							&stateDesc,
							stateDescriptor,
							stateDescriptorLength,
							-1);
					}
				} else {
					INFO1("ProcessRequest: couldn't read state %s\n", stateDesc.id);
					(void)SendMessage(*sd, MEMORY_FAILED, -1);
					if(errno == EMFILE) {
						CheckConnections();
					}
				}
				free(contents);
			}
		}

		fetchTime += MicroTime();	/* stop the timer */
		break;

	case STORE_STATE:
		/* we got an old store request */
		oldStoreQuery++;
		oldStoreTime -= MicroTime();

		ret = 0;

		if(!RecvData(*sd,
				&stateDesc,
				stateDescriptor,
				stateDescriptorLength,
				-1)) {
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: state receive failed\n");
		} else {
			/* we are cool */
			contentsDes.repetitions = stateDesc.rec_size * stateDesc.rec_count;
			contents = (char *)MALLOC(contentsDes.repetitions + 1);
			if(contents == NULL) {
				(void)SendMessage(*sd, MEMORY_FAILED, -1);
				ERROR("ProcessRequest: out of memory\n");
				break;
			}
			if(!RecvData(*sd, contents, &contentsDes, 1, -1)) {
				DROP_SOCKET(sd);
				ERROR("ProcessRequest: data receive failed\n");
				break;
			}
			contents[contentsDes.repetitions] = '\0';
	
			/* let's try to save the received experiment */
			ret = KeepState(&stateDesc, contents, contentsDes.repetitions);
			if (ret) {
				(void)SendMessage(*sd, STATE_STORED, -1);
			} else {
				(void)SendMessage(*sd, MEMORY_FAILED, -1);
			}
		}
		oldStoreTime += MicroTime();	/* stop the timer */

		/* notify the auto fetchters */
		autoFetchTime -= MicroTime();
		if (ret) {
			NotifyFetchers(	&stateDesc, 
					contents, 
					contentsDes.repetitions);
		}
		autoFetchTime += MicroTime();

		FREE(contents);
		break;

	/* new message since 2.9: we receive the full registration piggy
	 * backed into the packed experiment */
	case STORE_AND_REGISTER:
		/* we got a new query */
		storeQuery++;
		storeTime -= MicroTime();

		ret = 0;

		if(!RecvData(		*sd, 
					&newStateDesc,
					newStateDescriptor,
					newStateDescriptorLength,
					-1)) {
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: new state receive failed\n");
			return;
		} else {
			/* we are cool */
			contentsDes.repetitions = newStateDesc.rec_size * newStateDesc.rec_count + newStateDesc.id_len;
			contents = (char *)MALLOC(contentsDes.repetitions + 1);
			tmp = (char *)MALLOC(newStateDesc.id_len + 1);
			if(contents == NULL || tmp == NULL) {
				(void)SendMessage(*sd, MEMORY_FAILED, -1);
				ERROR("ProcessRequest: out of memory\n");
				break;
			}
			if(!RecvData(*sd, contents, &contentsDes, 1, -1)) {
				FREE(contents);
				FREE(tmp);
				DROP_SOCKET(sd);
				ERROR("ProcessRequest: data receive failed\n");
				break;
			}
			contents[contentsDes.repetitions] = '\0';

			/* let's try to save the received experiment */
			/* now is a trick: KeepState expects a stateDesc not a
			 * newStateDesc, so we just do it */
			stateDesc.rec_size = newStateDesc.rec_size;
			stateDesc.rec_count = newStateDesc.rec_count;
			stateDesc.seq_no = newStateDesc.seq_no;
			stateDesc.time_out = newStateDesc.time_out;

			/* let's extract the registration now: it's at the
			 * beginning of the content */
			memcpy(tmp, contents, newStateDesc.id_len);
			tmp[newStateDesc.id_len] = '\0';
	
			/* here only the name is needed */
			stateNames = NwsAttributeValue_r(FindNwsAttribute(tmp, "name"));
			if (stateNames == NULL) {
				ERROR("ProcessRequest: series without a name?!\n");
				DROP_SOCKET(sd);
				FREE(tmp);
				FREE(contents);
				break;
			}
			SAFESTRCPY(stateDesc.id, stateNames);
			FREE(stateNames);

			/* let's see how big is the record to be registered */
			i = strlen(contents + newStateDesc.id_len);
			ret = KeepState(&stateDesc, contents + newStateDesc.id_len, i);
			if (ret) {
				(void)SendMessage(*sd, STATE_STORED, -1);
			} else {
				(void)SendMessage(*sd, MEMORY_FAILED, -1);
			}

			/* last to be done for this is the REGISTER part: let's
			 * keep track of this registration */
			StoreSeriesName(tmp);
		}
		storeTime += MicroTime();	/* stop the timer */

		/* notify the auto fetchters */
		autoFetchTime -= MicroTime();
		if (ret) {
			NotifyFetchers(	&stateDesc, 
					contents + newStateDesc.id_len, 
					i);
		}
		autoFetchTime += MicroTime();

		FREE(contents);
		FREE(tmp);
		break;

	case AUTOFETCH_BEGIN:
		autoFetchTime -= MicroTime();

		stateNamesDescriptor.repetitions = header.dataSize;
		stateNames = (char *)MALLOC(header.dataSize);
		if(stateNames == NULL) {
			(void)SendMessage(*sd, MEMORY_FAILED, -1);
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: out of memory\n");
		} else if(!RecvData(*sd,
					stateNames,
					&stateNamesDescriptor,
					1,
					-1)) {
			(void)SendMessage(*sd, MEMORY_FAILED, -1);
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: data receive failed\n");
		} else if(*stateNames == '\0') {
			/* empty list: remove the client from the list of
			 * our autofetchers */
			EndAutoFetch(*sd);

			(void)SendMessage(*sd, AUTOFETCH_ACK, -1);
		} else {
			/* let's add the new autofetcher */
			if (!AddAutoFetcher(sd, stateNames)) {
				SendMessage(*sd, MEMORY_FAILED, -1);
				DROP_SOCKET(sd);
				ERROR("ProcessRequest: failed to add autoFetcher\n");
			} else {
				(void)SendMessage(*sd, AUTOFETCH_ACK, -1);
			}
		}
		FREE(stateNames);
		autoFetchTime += MicroTime();
		break;

	case MEMORY_CLEAN:
		if (!RecvData(*sd, &expiration, &expDescriptor, 1, -1)) {
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: data receive failed\n");
		} else {
			(void)SendMessage(*sd, MEMORY_CLEANED, -1);
			(void)DoClean(expiration);
		}
		break;

#ifdef WITH_NETLOGGER	  
static struct loglocation memLogLocation;
	case MEMORY_LOGDEST: /* config message contains log location */
		if (!RecvData(*sd,
				&memLogLocation,
				loglocationDescriptor,
				loglocationDescriptorLength,
				-1)) {
			DROP_SOCKET(sd);
			ERROR("ProcessRequest: loglocation receive failed\n");
			return;
		}else {
			(void)SendMessage(*sd, MEMORY_LOGDEST_ACK, -1);
		}
		LOG2("ProcessRequest: loglocation %d .%s.\n", memLogLocation.loc_type, memLogLocation.path);
		break;
#endif /* WITH_NETLOGGER */	  	

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

	/* socket is now available */
	SocketIsAvailable(*sd);
}

#define DEFAULT_MEMORY_DIR "."
#define DEFAULT_MEMORY_SIZE "2000"
#define DEFAULT_JOURNAL_SIZE "2000"
#define SWITCHES "C:a:d:n:e:l:N:Pp:s:v:Vi:B:"

static void usage() {
	printf("\nUsage: nws_memory [OPTIONS]\n");
	printf("Memory for the Network Weather Service\n");
	printf("\nOPTIONS can be:\n");
	printf("\t-e filename         write error messages to filename\n");
	printf("\t-l filename         write info/debug messages to filename\n");
	printf("\t-i filename         write pid to filename\n");
	printf("\t-a address          use this address as mine (ie multi-homed hosts)\n");
	printf("\t-n name             use name as my hostname\n");
	printf("\t-N nameserver       register with this nameserver\n");
	printf("\t-p port             bind to port instead of the default\n");
	printf("\t-s num              keep #num# measurement per experiment\n");
	printf("\t-d dir              save series files into directory dir\n");
	printf("\t-B db               use database #db# to store data\n");
	printf("\t-C #                # of caches to keep (0 to disable cache)\n");
	printf("\t-v level            verbose level (up to 5)\n");
	printf("\t-V                  print version\n");
	printf("Report bugs to <nws@nws.cs.ucsb.edu>.\n\n");
}

int
main(		int argc,
		char *argv[]) {
	IPAddress addresses[MAX_ADDRESSES], tmpAddress;
	unsigned int addressesCount;
	struct host_cookie memCookie,
		nsCookie;
	double nextBeatTime, wakeup;
	double now;
	int opt, 
		tmp,
		cacheEntries = 256,
		verbose;
	extern char *optarg;
	char tmpIP[127 + 1],
		password[127 + 1],
		memoryDir[255 + 1];
	FILE *logFD, *errFD;
	const char *c;
	char *toProcess, *dbname, *pidFile;

	/* Set up default values */
	verbose = 2;
	logFD = stdout;
	errFD = stderr;
	addressesCount = 0;
	toProcess = pidFile = NULL;

	dbname = GetEnvironmentValue("MEMORY_SIZE", "~", ".nwsrc", DEFAULT_MEMORY_SIZE);
	if (dbname == NULL) {
		fileSize = strtol(DEFAULT_MEMORY_SIZE, NULL, 10);
	} else {
		fileSize = strtol(dbname, NULL, 10);
		FREE(dbname);
	}
	dbname = GetEnvironmentValue("MEMORY_DIR", "~", ".nwsrc", DEFAULT_MEMORY_DIR);
	if (dbname == NULL) {
		SAFESTRCPY(memoryDir, DEFAULT_MEMORY_DIR);
	} else {
		SAFESTRCPY(memoryDir, dbname);
		FREE(dbname);
	}
	dbname = GetEnvironmentValue("NAME_SERVER", "~", ".nwsrc", "localhost");
	if (dbname == NULL) {
		SAFESTRCPY(nsCookie.name, "localhost");
	} else {
		SAFESTRCPY(nsCookie.name, dbname);
		FREE(dbname);
	}
	Host2Cookie(nsCookie.name, DefaultHostPort(NAME_SERVER_HOST), &nsCookie);
	password[0] = '\0';
	dbname = NULL;

	/* get my deafult name */
	Host2Cookie(MyMachineName(), DefaultHostPort(MEMORY_HOST), &memCookie);
	addressesCount = IPAddressValues(MyMachineName(),
			&addresses[0],
			MAX_ADDRESSES);
	if (addressesCount == 0) {
		ERROR1("Couldn't resolve my name (%s)\n", MyMachineName());
	}

	while((int)(opt = getopt(argc, argv, SWITCHES)) != EOF) {
		switch(opt) {
		case 'n':
			/* let's check we have a good name */
			if (!IPAddressValue(optarg, &tmpAddress)) {
				/* save the inet addresses */
				addressesCount += IPAddressValues(optarg, &addresses[addressesCount], MAX_ADDRESSES - addressesCount);
			} else {
				ERROR1("Unable to convert '%s': I'll do what you said but expect problems!\n", optarg);
			}

			/* overrride the name of the machine: we hope the
			 * user knows what is doing! */
			SAFESTRCPY(memCookie.name, optarg);

			break;

		case 'a':
			/* let's add this IPs to the list of my addresses */
			for (c = optarg; GETTOK(tmpIP, c, ",", &c); ) {
				tmp = IPAddressValues(tmpIP,
						&addresses[addressesCount],
						MAX_ADDRESSES - addressesCount);
				if (tmp == 0) {
					ERROR1("Unable to convert '%s' into an IP address\n", tmpIP);
				} else {
					addressesCount += tmp;
				}
			}
			break;

		case 'd':
			SAFESTRCPY(memoryDir, optarg);
			break;

		case 'e':
			/* open the error file */
			errFD = fopen(optarg, "w");
			if (errFD == NULL) {
				printf("Couldn't open %s!\n", optarg);
				exit(1);
			}
			break;

		case 'l':
			/* open the error file */
			logFD = fopen(optarg, "w");
			if (logFD == NULL) {
				printf("Couldn't open %s!\n", optarg);
				exit(1);
			}
			break;

		case 'i':
			/* write pid to file */
			pidFile = strdup(optarg);
			if (pidFile == NULL) {
				ABORT("out of memory\n");
			}
			break;

		case 'N':
			Host2Cookie(optarg, DefaultHostPort(NAME_SERVER_HOST), &nsCookie);
			break;

		case 'p':
			memCookie.port = strtol(optarg, NULL, 10);
			break;

		case 'P':
			fprintf(stdout, "Password? ");
			fscanf(stdin, "%s", password);
			break;

		case 's':
			fileSize = strtol(optarg, NULL, 10);
			break;

		case 'B':
			dbname = strdup(optarg);
			if (dbname == NULL) {
				fprintf(stderr, "out of memory\n");
			}
#ifdef NWS_WITH_DB
			backType = POSTGRES;
			cacheEntries = 0;
#else
			fprintf(stderr, "database is not enabled: ignoring -B\n");
#endif

		case 'C':
			cacheEntries = strtol(optarg, NULL, 10);
			break;

		case 'V':
			printf("nws_memory for NWS version %s", VERSION);
#ifdef HAVE_PTHREAD_H
			printf(", with thread support");
#endif
#ifdef WITH_DEBUG
			printf(", with debug support");
#endif
			printf("\n\n");
			exit(0);
			break;

		case 'v':
			verbose = (unsigned short)atol(optarg);
			break;

		case '?':
			if (optopt == 'v') {
				/* using the first level */
				verbose = 1;
				break;
			}

		default:
			usage();
			exit(1);
			break;

		}
	}

	/* initialize the performance counters */
	fetchTime = storeTime = oldStoreTime = autoFetchTime = 0;
	fetchQuery = storeQuery = oldStoreQuery = 0;

	/* let's set the verbose evel */
	SetDiagnosticLevel(verbose, errFD, logFD);

	/* now initializing the backend: if we use the directory, we need
	 * to initialize the caching schema */
	/* let's see if we are trying to use a database */
	switch (backType) {
	case LOCALDIR:
		/* WARNING: filesize s very important to have it right or
		 * cache and backing store could be inconsistent */
		InitStateModule(cacheEntries, fileSize, memoryDir);

		/* let's get the list of filename of possibly old states */
		toProcess = ReadOldStates();
		break;

	case POSTGRES:
		if (dbname == NULL || strlen(dbname) == 0) {
			ABORT("You need to specify a database name (-B)\n");
		}
		if (!ConnectToNwsDB(dbname)) {
			ABORT("Fail to use database try to remove -B\n");
		}

		/* let's get the list of tables of possibly old states */
		toProcess = GetTables();
		break;
	
	case NETLOGGER:
		WARN("we should initialize to use netlooger\n");
		break;
	}

	/* initialize the mySeries/orphanedSeries structures */
	if (!InitRegistrations(&orphanedSeries)) {
		ABORT("main: out of memory\n");
	}
	if (!InitRegistrations(&mySeries)) {
		ABORT("main: out of memory\n");
	}

	/* let's get the port and start serving */
	if(!EstablishHost(HostCImage(&memCookie),
                    MEMORY_HOST,
                    addresses,
                    addressesCount,
                    memCookie.port,
                    password,
                    &nsCookie,
		    NULL,		/* we are the memory ... */
                    &MyExit)) {
		ABORT("Unable to establish host: port already in use?\n");
	}

	/* now that we've got the port, we can print the pid into the
	 * pid file. We thus avoid to override pid files of running
	 * nameservers */
	if (pidFile != NULL) {
		FILE *pidfile = fopen(pidFile, "w");
		if (!pidfile) {
			ABORT1("Can't write pidfile %s\n", pidFile);
		}
		fprintf(pidfile, "%d", (int)getpid());
		fclose(pidfile);
		free(pidFile);
	}

	fclose(stdin);

	RegisterListener(STORE_STATE, "STORE_STATE", &ProcessRequest);
	RegisterListener(STORE_AND_REGISTER, "STORE_AND_REGISTER", &ProcessRequest);
	RegisterListener(FETCH_STATE, "FETCH_STATE", &ProcessRequest);
	RegisterListener(AUTOFETCH_BEGIN, "AUTOFETCH_BEGIN", &ProcessRequest);
	RegisterListener(MEMORY_CLEAN, "MEMORY_CLEAN", &ProcessRequest);
#ifdef WITH_NETLOGGER
	RegisterListener(MEMORY_LOGDEST, "MEMORY_LOGDEST", &ProcessRequest);
#endif
	if (!NotifyOnDisconnection(&EndAutoFetch)) {
		WARN("main: failed to register for disconnections\n");
	}

	nextBeatTime = 0;

	/* main service loop */
	while(1) {
		now = CurrentTime();
		if(now >= nextBeatTime) {
			RegisterHost(DEFAULT_HOST_BEAT * 2);
			nextBeatTime = now + (HostHealthy() ? DEFAULT_HOST_BEAT : SHORT_HOST_BEAT);

			/* print performance timing */
			LOG1("main: spent %.0fms in fetch\n", fetchTime/1000);
			LOG1("main: spent %.0fms in store\n", storeTime/1000);
			LOG1("main: spent %.0fms in old store\n", oldStoreTime/1000);
			LOG1("main: spent %.0fms for autofetching\n", autoFetchTime/1000);
			LOG3("main: got %d fetch, %d store and %d old store requests\n", fetchQuery, storeQuery, oldStoreQuery);
			/* resetting */
			fetchTime = storeTime = oldStoreTime = autoFetchTime = 0;
			fetchQuery = storeQuery = oldStoreQuery = 0;


			LOG2("main: we have %d series and %d orphaned series\n", mySeries->howMany, orphanedSeries->howMany);
		}

		/* we want to wake up at least once a minute */
		if ((now + 60) < nextBeatTime) {
			wakeup = 60;
		} else {
			wakeup = nextBeatTime - now;
		}

		/* let's convert filenames to orphaned series: if the
		 * directory is big it may take time to read it. */
		if (toProcess != NULL) {
			if (CheckOldNames(	toProcess,
						mySeries,
						orphanedSeries,
						1)) {
				wakeup = -1;
			} else {
				FREE(toProcess);
			}
		}

		/* let's register the series we are in charge of */
		if (RegisterSeries(orphanedSeries, 10)) {
			wakeup = -1;
		}
		if (RegisterSeries(mySeries, 10)) {
			wakeup = -1;
		}

		/* we want the memory to be repsonsive, so we flush all
		 * the waiting messages */
		while(ListenForMessages(-1)) {
			;
		}
		ListenForMessages((wakeup > 0) ? wakeup : -1);
	}

	/* return(0); Never reached */
}