/*
 * Sit on a particular server.
 *
 * Wait for a command, send it to the server, wait for it to complete,
 * report completion, and start over.
 */

#include <stdio.h>
#include <fcntl.h>

#include <r/xdr.h>
#include <r/r.h>
#include <lwp.h>

#include "../butler/butler.h"
#include "../butler/mreg.h"

#include "bmake.h"

extern char **environ;
extern char *PathLookup();
extern char *getenv();
extern char *malloc();

int machineType;
int netSide;
char *output_pfx;
int dying = 0;

struct server *server_queue;
int server_alloc_failed;
int niniting;
int nprocs;
int nrunning;
struct server *done_queue;
extern int job_slots;

void bmake_init()
{
    int e;
    int i;
    
    PathInit();
    e = bclient_RunInit();
    if (e<0) {
	error("Couldn't init butler client\n");
	exit(1);
    }
    machineType = MR_MachineType();
    netSide = MR_NetSide();
    output_pfx = getenv("BMAKEOUTPUTDIR");
    if (!output_pfx)
	output_pfx = "/afs/athena/user/w/wesommer/public/bmake";

    /* grab the compute servers now.. */
    for (i=0; i<job_slots;i++)
	new_server();
    
}

/*
 * this server has failed.
 */

void server_fail (sp, why)
    struct server *sp;
    char *why;
{
    if (sp->next != 0)
	abort();
    printf("call to %s(%d) for %s failed\n", sp->name, sp->unique, why);
    bclient_Release(sp->conn);
    /* XXX should flush other stuff here */
    free(sp);
}

/*
 * wait for next command.
 */

int server_waitcmd(sp)
    struct server *sp;
{
    if (sp->next)
	abort();

#ifdef DEBUG 
    printf("%s(%d): queueing on server queue\n", sp->name, sp->unique);
#endif
    sp->next = server_queue;
    server_queue = sp;

    LWP_NoYieldSignal(&server_queue);

    while (!sp->argv && !dying) {
#ifdef DEBUG
	printf("%s(%d): still waiting for command\n", sp->name, sp->unique);
#endif
	LWP_WaitProcess(sp);
    }	

    if (dying)
	return -1;
    
    return 0;
}

void server_finish(sp)
    struct server *sp;
{
    (void) bclient_Release(sp->conn);
    IOMGR_Sleep(2);
}

void server_done (sp, status)
    struct server *sp;
    int status;
{
    sp->status = status;

    if (sp->next != 0)
	abort();
#ifdef DEBUG
    printf("%s(%d): queueing on done queue...\n", sp->name, sp->unique);
#endif
    sp->argv = 0;

    sp->next = done_queue;
    done_queue = sp;
    LWP_NoYieldSignal(&done_queue);

    while (sp->unique != 0) {
#ifdef DEBUG
	printf("%s(%d): still waiting for done...\n", sp->name, sp->unique);
#endif 
	LWP_WaitProcess(sp);
    }
#ifdef DEBUG
    printf("%s(%d): we're now undone\n", sp->name, sp->unique);
#endif
}

void do_server(sp)
    struct server *sp;
{
    int e;
    int nbuilds = 0;

    nprocs++;
    if (sp->state != server_init)
	abort();
    niniting--;
    
    for (;;) {
	struct Run_FD fds[3];
	char *program;
	int pid, status;
	FILE *f;
	
	sp->state = server_cmd_wait;
#ifdef DEBUG
	printf("%s(%d): waiting for a command..\n", sp->name, sp->unique);
#endif
	e = server_waitcmd(sp);
	if (e == -1) {
	    server_finish(sp);
	    printf("%d build%s on %s\n", nbuilds,
		   (nbuilds==1?"":"s"),
		   sp->name);
	    goto out;
	}
	sp->state = server_running;
	nrunning++;
#ifdef VERBOSE
	printf("\n%s(%d): starting build of %s\n", sp->name, sp->unique, sp->target);
#else
	printf("%s(%d): %s\n", sp->name, sp->unique, sp->target);	
#endif
	
	if ((program = PathLookup (sp->argv[0])) == NULL) {
	    error("%s(%d): Command not found", sp->argv[0]);
	    server_done(sp, 1);
	    continue;
	}

	fds[0].what = RUN_FDOPEN;
	fds[0].name = "/dev/null";
	fds[0].flags = O_RDONLY;
	fds[0].mode = 0;

	fds[1].what = RUN_FDOPEN;
	fds[1].name = sp->output_file;
	fds[1].flags = O_RDWR|O_CREAT|O_TRUNC;
	fds[1].mode = 0644;

	fds[2].what = RUN_FDDUP;
	fds[2].fd = 1;
	
	pid = bclient_Run (sp->conn, program, sp->argv, fds, 3);
	if (pid < 0) {
#if 0
	    server_fail(sp, "run");
#endif
	    server_done(sp, 255);
	    nrunning--;
	    goto out;
	}
#ifdef DEBUG
	printf("%s(%d): running...\n", sp->name, sp->unique);
#endif
	e = bclient_Wait(sp->conn, pid, &status);
	if (e < 0) {
#if 0
	    server_fail(sp, "wait");
#endif
	    server_done (sp, 255);
	    nrunning--;
	    goto out;
	}
#ifdef DEBUG
	printf("%s(%d): done...\n", sp->name, sp->unique);
#endif
#ifdef VERBOSE
	printf("\n%s done:\n",sp->name, sp->unique);
#endif
	f = fopen(sp->output_file, "r");
	/* copy output to stdout */
	if (f != NULL) {
	    char buf[1024];
	    while ((fgets(buf, sizeof(buf), f)) != NULL)
		printf("%s(%d): %s", sp->name, sp->unique, buf);

	    fclose(f);
	}
	(void) unlink(sp->output_file);
	/* end output */
	nrunning--;
	nbuilds++;
	server_done(sp, status);
    }
out:
    nprocs--;
    LWP_SignalProcess(&nprocs);
    return;
}

struct server *get_server()
{
    struct server *sp = 0;
    static int id = 10000;
    
#ifdef DEBUG
    printf("(get_server) start\n");
#endif
    
#if 0
    if (! server_queue  && !server_alloc_failed) 
	new_server();
#endif
    
#ifdef DEBUG
    printf("(get_server) waiting for server ...\n");
#endif
    while (! server_queue && niniting) {
	LWP_WaitProcess(&server_queue);
#ifdef DEBUG
	printf("(get_server) woke up...\n");
#endif
    }
    
    if (server_queue) {
#ifdef DEBUG
	printf("(get_server) got %s\n", server_queue->name);
#endif
	sp = server_queue;
	server_queue = sp->next;
	sp->next = 0;
	sp->unique = id++;
    }
    return sp;
}

start_command(sp)
    struct server *sp;
{
    /*
     * fire off the command
     */
    
    if (sp->next != 0)
	abort();

    LWP_SignalProcess(sp);
}

/*
 */

static void gen_output(sp)
    struct server *sp;
{
    int len = strlen(output_pfx) + strlen(sp->name) + 2;
    char *cp = malloc(len);

    strcpy(cp, output_pfx);
    strcat(cp, "/");
    strcat(cp, sp->name);
    sp->output_file = cp;
}

/*
 * start a new server connection and associated LWP.
 */

new_server()
{
    struct server *sp = (struct server *)malloc (sizeof(*sp));
    u_long host;
    int e;
    char *name = malloc (64);

    struct MR_DescArray desc;
    struct MR_Description attrs[2];

#ifdef DEBUG
    printf("(new_server) started\n");
#endif
    desc.attrs = attrs;
    desc.nAttrs = 2;
    attrs[0].attr = mr_machineType;
    attrs[0].value = machineType;
    attrs[1].attr = mr_netSide;
    attrs[1].value = netSide;

    sp->state = server_init;
    e = bclient_GetMachine(desc, 0, &sp->conn, &host, name);

    if (e < 0) {
	printf("couldn't get machine..\n");
	return;
    }
#ifdef DEBUG
    printf("(new_server) Got Machine %s...\n", name);
#endif
    sp->name = name;
    sp->argv = 0;
    sp->target = 0;
    gen_output(sp);
    e = bclient_SetContext(sp->conn, environ, (char *)NULL);
    if (e < 0) {
	printf(sp, "couldn't set context");
	return;
    }
    niniting++;
#ifdef DEBUG
    printf("(new_server) Creating LWP %s...\n", name);
#endif
    
    LWP_CreateProcess(do_server, 16000, LWP_NORMAL_PRIORITY, sp, sp->name, &sp->pid);
#ifdef DEBUG
    printf("(new_server) Created LWP %s\n", name);
#endif
}

remote_cleanup()
{
    struct server *sp;
    dying = 1;

#ifdef DEBUG
    printf("remote_cleanup...");
#endif
    while (nprocs > 0) {
	for (sp = server_queue; sp; sp = sp->next)
	    LWP_NoYieldSignal(sp);
#ifdef DEBUG
	printf("%d...", nprocs);
	fflush(stdout);
#endif
	if (nprocs > 0)
	    LWP_WaitProcess(&nprocs);
    }
#ifdef DEBUG
    printf("done\n");
#endif
}