/*
 *	Copyright 1990 by Rayan S. Zachariassen, all rights reserved.
 *	This will be free software, but only when it is finished.
 */

#include "hostenv.h"
#include <stdio.h>
#include "scheduler.h"

struct vertex *curitem = NULL;

time_t qipcretry = 0;
time_t sweepretry = 0;

#define SALARM(N)  (verbose ? printf("alarm(%d) = %d\n", (N),alarm(N)):alarm(N))

void
disentangle(vp)
	struct vertex *vp;
{
	struct web *wp;

	wp = vp->orig[L_HOST];

	if (wp->fifohead == vp)
		wp->fifohead = vp->nextitem;
	else if (wp->fifotail == vp)
		wp->fifotail = vp->previtem;

	wp = vp->orig[L_CHANNEL];

	if (wp->fifohead == vp)
		wp->fifohead = vp->nextitem;
	else if (wp->fifotail == vp)
		wp->fifotail = vp->previtem;

	if (curitem == vp)
		curitem = vp->nextitem;
	if (vp->proc != NULL) {
		if (vp->proc->vertices == vp) {
			vp->proc->vertices = vp->nextitem;
			if (vp->proc->vertices != NULL)
				vp->proc->vertices->proc = vp->proc;
		}
		vp->proc = NULL;
	}
	if (vp->previtem != NULL)
		vp->previtem->nextitem = vp->nextitem;
	if (vp->nextitem != NULL)
		vp->nextitem->previtem = vp->previtem;
}

void
reschedule(vp, factor, index)
	struct vertex *vp;
	int factor, index;
{
	time_t now;
	int skew;
	struct vertex *ap, *pap;
	extern time_t time();
	extern int verbose, sweepinterval;
	extern int ranny();
	extern void expire();

	/* find out when to retry */
	now = time((time_t *)0);

	if (verbose)
		(void) printf("reschedule %x now %d factor %d inum %d (%s/%s: %s)\n",
				   vp, now, factor, vp->cfp->id,
				   vp->orig[L_CHANNEL]->name,
				   vp->orig[L_HOST]->name,
				   vp->cfp->mid);
	/* if we are already scheduled for the future, don't reschedule */
	if (vp->wakeup > now) {
		if (verbose)
			printf("prescheduled\n");
		return;
	}

	if (vp->ce.nretries <= 0) {
		if (verbose)
			printf("ce.retries = %d\n", vp->ce.nretries);
		return;
	}

	if (vp->ce.expiry > 0 && vp->ce.expiry <= now && vp->attempts > 0) {
		if (verbose)
			printf("ce.expiry = %d, %d attempts\n",
					  vp->ce.expiry, vp->attempts);
		expire(vp, index);
		return;
	}

	/* unlink from list of scheduled vertices */
	disentangle(vp);

	if (vp->wakeup == 0)
		vp->wakeup = now;
	if (factor == -1) {
		if (vp->retryindex >= vp->ce.nretries)
			vp->retryindex = ranny(vp->ce.nretries-1);

		/*
		 * clamp retry time to a predictable interval so we
		 * eventually bunch up deliveries.
		 */
		skew = vp->wakeup % vp->ce.interval;
		if (skew <= vp->ce.interval / 2)
			skew = - (skew + (vp->ce.skew - 1));
		else
			skew = skew + (vp->ce.skew - 1);
		skew = skew / vp->ce.skew; /* want int div truncation */

		vp->wakeup += skew
			+ vp->ce.retries[vp->retryindex] * vp->ce.interval;
		vp->retryindex++;
	} else if (factor < -1) {
		vp->wakeup = -factor;
	} else
		vp->wakeup += factor * vp->ce.interval;

	/* XX: change this to a mod expression */
	if (vp->wakeup < now)
		vp->wakeup = (((now - vp->wakeup) / vp->ce.interval)+1)
					* vp->ce.interval;

	if (vp->ce.expiry > 0 && vp->ce.expiry <= vp->wakeup
	    && vp->attempts > 0) {
		if (verbose)
			printf("ce.expiry = %d, %d attempts\n",
					  vp->ce.expiry, vp->attempts);
		expire(vp, index);
		return;
	}

	if (verbose)
		printf("wakeup %d pending %d\n", vp->wakeup, vp->ce.pending);

	/* link it back in at the right spot */
	for (ap = curitem, pap = NULL ; ap != NULL; ap = ap->nextitem) {
		if (ap->wakeup > vp->wakeup) {
			vp->previtem = ap->previtem;
			if (ap->previtem != NULL) {
				ap->previtem->nextitem = vp;
			}
			ap->previtem = vp;
			vp->nextitem = ap;
			break;
		}
		pap = ap;
	}
	if (verbose)
		printf("ap %x pap %x curitem %x\n", ap, pap, curitem);

	if (ap == NULL && pap != NULL) {
		/* append to list */
		pap->nextitem = vp;
		vp->previtem = pap;
		vp->nextitem = NULL;
	} else if (ap == curitem) {
		/* either there is no list or we inserted at head */
		vp->nextitem = curitem;
		vp->previtem = NULL;
		curitem = vp;
		if (curitem->wakeup - now > 0
		    && curitem->wakeup - now <= sweepinterval)
			SALARM((u_int)(curitem->wakeup - now));
		else
			SALARM(sweepinterval/2);
	}
}

void
doagenda()
{
	time_t now;
	int didsomething;
	struct vertex *vp, *ncuritem;
	extern int verbose, gotalarm, sweepinterval;
	extern time_t time();
	extern void unravel(), queryipcinit();

	ncuritem = curitem;
	curitem = NULL;
	do {
		now = time((time_t *)0);
		didsomething = 0;
		if (verbose)
			(void) printf("curitem %x curitem->wakeup %d now %d\n",
				ncuritem,
				ncuritem != NULL ? ncuritem->wakeup : 0,
				now);
		while (ncuritem != NULL && ncuritem->wakeup <= now) {
			for (vp = ncuritem; vp != NULL; vp = vp->nextitem)
				if (vp->wakeup > now)
					break;
			didsomething = 1;
			if (vp != NULL && vp->previtem != NULL) {
				vp->previtem->nextitem = NULL;
				vp->previtem = NULL;
			}
			unravel(ncuritem);
			ncuritem = vp;
		}
	} while (didsomething);
	for (vp = curitem ; vp != NULL ; vp = vp->nextitem)
		if (vp->nextitem == NULL)
			break;
	if (vp != NULL) {
		vp->nextitem = ncuritem;
		if (ncuritem != NULL) {
			ncuritem->previtem = vp;
		}
	} else
		curitem = ncuritem;
	if (verbose)
		printf("alarmed %d\n", now);
	if (qipcretry > 0 && qipcretry <= now) {
		qipcretry = 0;
		queryipcinit();
		/*
		 * If qipcretry is set here, the value will be ignored, but
		 * that's ok since sweepretry is active by now
		 */
	}
	if (sweepretry > 0 && sweepretry <= now)
		sweepretry = now + sweepinterval;
	if (ncuritem != NULL && ncuritem->wakeup < sweepretry)
		SALARM((u_int)(ncuritem->wakeup - now));
	else if (sweepretry > now)
		SALARM(sweepretry - now);
	gotalarm = 0;
}

void
unravel(head)
	struct vertex *head;
{
	register struct vertex *vp, *vp2;
	register int n, i, ni;
	struct vertex *thread, *nvp, *nvp2, *tail;
	static u_int ur_size = 0;
	static struct vertex **ur_arr;
	extern int verbose;
	extern void spool();
	extern int ranny();

	/* randomize order of nodes */
	for (vp = head, n = 0; vp != NULL; vp = vp->nextitem)
		++n;
	if (ur_size == 0) {
		ur_size = 100;
		ur_arr = (struct vertex **)
				emalloc(ur_size * sizeof (struct vertex *));
	}
	while (ur_size < n) {
		ur_size *= 2;
		ur_arr = (struct vertex **)realloc((char *)ur_arr,
				ur_size * sizeof (struct vertex *));
	}
	for (i = 0, vp = head; i < n; ++i, vp = vp->nextitem)
		ur_arr[i] = vp;
	for (i = 0; i < n; ++i) {
		ni = ranny(n-1);
		vp = ur_arr[i];
		ur_arr[i] = ur_arr[ni];
		ur_arr[ni] = vp;
	}
	for (i = 0; i < n; ++i) {
		if (i == 0)
			ur_arr[i]->previtem = NULL;
		else
			ur_arr[i]->previtem = ur_arr[i-1];
		if (i == n-1)
			ur_arr[i]->nextitem = NULL;
		else
			ur_arr[i]->nextitem = ur_arr[i+1];
	}
	head = ur_arr[0];

	/* process */
	if (verbose) {
		(void) printf("unravel: %x", head);
		for (vp = head->nextitem ; vp != NULL ; vp = vp->nextitem)
			(void) printf("->%x", vp);
		(void) printf("\n");
	}
	for (vp = head; vp != NULL ; vp = nvp) {
		nvp = vp->nextitem;
		thread = tail = NULL;
		for (vp2 = vp; vp2 != NULL ; vp2 = nvp2) {
			nvp2 = vp2->nextitem;
			if (vp2->ce.command == vp->ce.command
			    && vp2->ce.uid == vp->ce.uid
			    && vp2->ce.gid == vp->ce.gid
			    && (vp->ce.bychannel == 0
				|| vp->orig[L_CHANNEL] == vp2->orig[L_CHANNEL])
			    && (vp->ce.byhost == 0
				|| vp->orig[L_HOST] == vp2->orig[L_HOST])) {
				if (vp2 == nvp)
					nvp = nvp->nextitem;
				if (vp2->nextitem != NULL)
					vp2->nextitem->previtem = vp2->previtem;
				if (vp2->previtem != NULL)
					vp2->previtem->nextitem = vp2->nextitem;
				if (thread == NULL) {
					thread = vp2;
					vp2->nextitem = vp2->previtem = NULL;
				} else {
					tail->nextitem = vp2;
					vp2->previtem = tail;
					vp2->nextitem = NULL;
				}
				tail = vp2;
			}
		}
		if (verbose) {
			(void) printf("thread: %x", thread);
			for (vp2 = thread->nextitem ; vp2 != NULL ; vp2 = vp2->nextitem)
				(void) printf("->%x", vp2);
			(void) printf("\n");
		}
		spool(thread);
	}
}

void
spool(thread)
	struct vertex *thread;
{
	struct vertex *vp, *nvp;
	extern int numkids, verbose;
	extern void transport(), fifoput();

	for (vp = thread; vp != NULL ; vp = nvp) {
		nvp = vp->nextitem;
		if (numkids >= vp->ce.maxkids)
			vp->ce.pending = SIZE_L;
		else if (vp->orig[L_CHANNEL]->kids >= vp->ce.maxkidChannel)
			vp->ce.pending = L_CHANNEL;
		else if (vp->orig[L_HOST]->kids >= vp->ce.maxkidHost)
			vp->ce.pending = L_HOST;
		else
			vp->ce.pending = 0;

		if (vp->ce.pending) {
			if (verbose)
				(void) printf("%s: (%d %dC %dH) >= (%d %dC %dH)\n",
					    vp->ce.command,
					    numkids,
					    vp->orig[L_CHANNEL]->kids,
					    vp->orig[L_HOST]->kids,
					    vp->ce.maxkids,
					    vp->ce.maxkidChannel,
					    vp->ce.maxkidHost);
			/*
			 * Would go over limit.  Rescheduling for the next
			 * (single) interval works ok in many situation.
			 * However when the scheduler is very busy one can
			 * run into systemic problems with some set of messages
			 * blocking another set of messages.  The only way
			 * around that is a roundrobin scheme, implemented
			 * by the fifo rescheduling functions below.
			 */
			fifoput(vp);
		} else
			vp->cfp->mark = V_SELECT;
	}
	transport(thread, thread->orig[L_CHANNEL]->name,
			  thread->orig[L_HOST]->name);
}

void
fifoput(vp)
	struct vertex *vp;
{
	struct web *wp;
	extern int verbose;

	if (vp->ce.pending != L_CHANNEL && vp->ce.pending != L_HOST) {
		reschedule(vp, 0, -1);
		return;
	}

	disentangle(vp);

	wp = vp->orig[vp->ce.pending];
	if (wp->fifohead == NULL) {
		wp->fifohead = wp->fifotail = vp;
		vp->nextitem = vp->previtem = NULL;
	} else {
		vp->previtem = wp->fifotail;
		wp->fifotail->nextitem = vp;
		wp->fifotail = vp;
		vp->nextitem = NULL;
	}
	if (verbose)
		(void) printf("fifo append: %s/%s\n",
			     vp->orig[L_CHANNEL]->name,
			     vp->orig[L_HOST]->name);
}

/*
 * Called when a transport agent has terminated, to attempt to start another
 * transport agent instance with the same parameters.  The consistent (with
 * respect to the vertex at the head of the fifo queue) vertices underneath
 * wp are unlinked from the fifo queue and rescheduled.
 */

void
fifoschedule(wp)
	struct web *wp;
{
	struct vertex *vp, *nvp, *head;
	extern int verbose;

	if (wp->fifohead == NULL)
		return;
	
	head = wp->fifohead;
	for (vp = head; vp != NULL; vp = nvp) {
		nvp = vp->nextitem;
		if (vp->ce.command == head->ce.command
		    && vp->ce.uid == head->ce.uid
		    && vp->ce.gid == head->ce.gid
		    && (head->ce.bychannel == 0
			|| head->orig[L_CHANNEL] == vp->orig[L_CHANNEL])
		    && (head->ce.byhost == 0
			|| head->orig[L_HOST] == vp->orig[L_HOST])) {
			if (wp->fifohead == vp) {
				wp->fifohead = vp->nextitem;
				if (wp->fifohead != NULL)
					wp->fifohead->previtem = NULL;
			} else if (vp->previtem != NULL)
				vp->previtem->nextitem = vp->nextitem;
			if (wp->fifotail == vp) {
				wp->fifotail = vp->previtem;
				if (wp->fifotail != NULL)
					wp->fifotail->nextitem = NULL;
			} else if (vp->nextitem != NULL)
				vp->nextitem->previtem = vp->previtem;
			vp->previtem = vp->nextitem = NULL;
			vp->wakeup = 0;
			if (verbose)
				(void) printf("fifo schedule: %s/%s\n",
					     vp->orig[L_CHANNEL]->name,
					     vp->orig[L_HOST]->name);
			reschedule(vp, 0, -1);
		}
	}
}
