/*
 * This program runs as a standalone process for managing an adpative
 * pool of HPSS application processes as worker sub-processes.  After creating
 * the initial set of workers, we periodically check their activity and
 * dynamically increase the number to a maximum if client requests are
 * stalling waiting to get serviced.
 *
 * Usage:
 *    hpss_pool_master min[:max[:start]] mailbox-name worker-command...
 *
 * where:
 *    min		Minimum number of worker processes to keep active.
 *
 *    max		Maximum number of worker process to create,
 *			default is min.
 *
 *    start		Initial number to create (improves startup time if
 *			min is large).  If min is more than start, processes
 *			will gradually be added.
 *
 *    mailbox-name	HPSS server name.
 *
 *    worker-command... DCL command to use to run worker process, including
 *			any parameters.
 *
 * Environment variables:
 *    hpss_pool_debug		If defined, print debug information.
 *    hpss_pool_adjust_interval	Seconds to wait between activity checks.
 *    
 *
 * Author:	David Jones
 * Date:	20-JUL-2000
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include <ssdef.h>		/* VMS condition codes */
#include <descrip.h>
int SYS$SCHDWK(), SYS$HIBER(), LIB$SPAWN(), SYS$SETAST(), LIB$EMUL();
int SYS$WAKE();

#include "hpss_share.h"
static char *usage_summary = 
   "Usage: pool_master min:[max[:initial]] mbx-name worker-cmd";
static int debug_flag = 0;
/*
 * Define structure for tracking the worker processes we spawn.
 * 
 */
struct worker_blk {
    struct worker_blk *next;
    long pid;			/* process ID of worker */
    int state;			/* Updated at AST level */
    int exitstatus;
};

struct worker_list {
    int count;			/* number of list elements */
    struct worker_blk *first;
    struct worker_blk *last;
};

static struct {
    struct worker_blk *free;	/* lookaside list */
    struct worker_list active;
    struct worker_list dead;
} wrkr;
/*
 * Scan the command arguments and pull out arguments an construct
 * command to give to spawned workers.
 */
static int process_command_line ( int argc, char **argv, int *min_workers, 
	int *max_workers, int *initial_workers,
	char **mbx_name, char **worker_boot )
{
    int i, j, limit, len;
    char arg1[120], *tok, *cmd;

    if ( argc < 4 ) {
	fprintf(stderr, "%s\n", usage_summary );
	return SS$_BADPARAM;
    } else if ( strlen(argv[1]) > sizeof(arg1) ) {
	return SS$_BADPARAM;
    }
    /*
     * Parse first argument into 3 numbers separated by colons:
     */
    strcpy ( arg1, argv[1] );
    tok = strtok ( arg1, ":" );		/* Min-workers */
    if ( tok ) {
	*min_workers = atoi(tok);
	*max_workers = *min_workers;
	*initial_workers = *min_workers;

	tok = strtok ( (char *) 0, ":" );	/* max-workers */
    }
    if ( tok ) {
	*max_workers = atoi ( tok );
	tok = strtok ( (char *) 0, ":" );	/* inital-workers */
    }
    if ( tok ) *initial_workers = atoi ( tok );
    /*
     * Construct mailbox name and argv[2] with "/POOL=MASTER" appended.
     */
    *mbx_name = malloc ( strlen(argv[2]) + 16 );
    sprintf ( *mbx_name, "%s/POOL=MASTER", argv[2] );
    /*
     * Build command line.  Scan argv[3..argc-1] to determine max string
     * size, then contstruct command.
     */
    for ( limit=0, i=3; i < argc; i++ ) {
	limit = limit + (strlen(argv[i])) + 1;	/* include space */
    }
    cmd = malloc ( limit );
    strcpy ( cmd, argv[3] );
    len = strlen ( cmd );
    for ( i = 4; i < argc; i++ ) {
	cmd[len++] = ' ';
	strcpy ( &cmd[len], argv[i] );
	len += strlen(argv[i]);
    }
    cmd[len] = '\0';
    if ( debug_flag > 0 ) printf("Worker boot command: '%s'\n", cmd );
    *worker_boot = cmd;

    return 1;
}
/***************************************************************************/
/*  Routines for managing the worker list, including creating of worker
 * processes via lib$spawn.
 */
static struct worker_blk *alloc_active_worker ( )
{
    struct worker_blk *blk;
    /*
     * Allocate worker block.
     */
    blk = wrkr.free;
    if ( wrkr.free ) {
	wrkr.free = blk->next;
    } else {
	blk = malloc ( sizeof(struct worker_blk) );
	if ( !blk ) return blk;
    }
    blk->next = (struct worker_blk *) 0;
    blk->state = 0;
    blk->pid = 0;
    SYS$SETAST(0);
    wrkr.active.count++;
    if ( wrkr.active.first ) wrkr.active.last->next = blk;
    else wrkr.active.first = blk;
    wrkr.active.last = blk;
    SYS$SETAST(1);
    return blk;
}
static void worker_exit_ast ( struct worker_blk *blk )
{
    struct worker_blk *cur, *prev;
    /*
     * Search active list and remove block of exiting process.
     */
    if ( blk == wrkr.active.first ) {
	wrkr.active.first = blk->next;
    } else if ( wrkr.active.first ) {
	prev = wrkr.active.first;
	for ( cur = prev->next; (cur != blk); cur = cur->next ) {
	    if ( !cur ) {
		fprintf(stderr,"Worker block lost for %x\n", blk->pid );
		return;
	    }
	    prev = cur;
	}
	prev->next = cur->next;
    } else {
	fprintf(stderr, "Worker block lost for %x (list empty)\n", blk->pid);
	return;
    }
    wrkr.active.count--;
    if ( wrkr.active.count <= 0 ) {
	/*
	 * Bad news, all workers are dead, kick the main loop to
	 * rejuvenate the pool.
	 */
	SYS$WAKE(0,0);
    }
    /*
     * Place block on dead list.
     */
    blk->next = (struct worker_blk *) 0;
    blk->state = 2;
    wrkr.dead.count++;
    if ( wrkr.dead.first ) wrkr.dead.last->next = blk;
    else wrkr.dead.first = blk;
    wrkr.dead.last = blk;
}
/*
 * Move blocks from dead list to free list, inspecting exitstatus if needed.
 */
static int validate_worker_list( )
{
   struct worker_blk *first, *blk, *next;
    int count;
    /*
     * Remove entire dead list for processing at non-AST level.
     */
    SYS$SETAST(0);
    count = wrkr.active.count;
    first = wrkr.dead.first;
    wrkr.dead.count = 0;
    wrkr.dead.first = (struct worker_blk *) 0;
    SYS$SETAST(1);

    for ( blk= first; blk; blk = next ) {
	next = blk->next;
	blk->next = wrkr.free;
	wrkr.free = blk;
    }
    return count;
}
/*
 * Spawn additional worker processes and update worker list.
 */
static int add_workers ( int delta, char *worker_cmd )
{
    static $DESCRIPTOR(cmd_dx,"");
    int status, i, flags;

    cmd_dx.dsc$w_length = strlen(worker_cmd);
    cmd_dx.dsc$a_pointer = worker_cmd;
    flags = 1;		/* nowait */

    for ( i = 0; i < delta; i++ ) {
	struct worker_blk *blk;
	blk = alloc_active_worker();
	if ( !blk ) break;
	status = LIB$SPAWN ( &cmd_dx, 0, 0, &flags, 0,
		&blk->pid, &blk->exitstatus, 0, worker_exit_ast,
		blk, 0, 0, 0 );
	if ( debug_flag > 0 ) printf("status of spawn: %d\n", status );
	SYS$SETAST(0);
	if ( (status&1) == 0 ) {
	    /*
	     * Spawn failed, manually invoke exit handler.
	     */
	    blk->exitstatus = status;
	    worker_exit_ast ( blk );
	} else {
	    if ( blk->state == 0 ) blk->state = 1;
	}
	SYS$SETAST(1);
    }
    return 1;
}
/***************************************************************************/
int main ( int argc, char **argv )
{
    int status, context, interval[2];
    int min_workers, max_workers, initial_workers, worker_count, delta;
    int adjust_interval, ticks_per_sec, offset;
    char *mbx_name, *worker_boot, *var;
    /*
     * get command line parameters and read environment variables.
     */
    status = process_command_line ( argc, argv, &min_workers,
		&max_workers, &initial_workers, &mbx_name, &worker_boot );
    if ( (status&1) == 0 ) return status;
    var = getenv ( "HPSS_POOL_DEBUG" );
    if ( var ) debug_flag = atoi ( var );
    var = getenv ( "HPSS_POOL_ADJUST_INTERVAL" );
    adjust_interval = (var) ? atoi(var) : 63;
    wrkr.free = wrkr.active.first = wrkr.dead.first = (struct worker_blk *) 0;
    wrkr.active.count = wrkr.dead.count = 0;
    /*
     * Initialize HPSS, mbx_name includes qualifier to make mailbox
     * read/write, which has 2 effects:
     *   1.  We can sent EOFs to the mailbox to trigger worker exits.
     *   2.  Clients will always think a reader is present.
     */
    status = hpss_initialize_c ( mbx_name, 0, &context );
    if ( (status&1) == 0 ) {
	fprintf ( stderr, "Error initializing HPPS: %d\n", status );
	return status;
    }
    /*
     * Schedule repeating wakeup at 1 minute (63 second) intervals.
     */
    ticks_per_sec = -10000000;		/* 100-nanosecond ticks */
    offset = -2000000;
    LIB$EMUL ( &adjust_interval, &ticks_per_sec, &offset, interval );
    status = SYS$SCHDWK ( 0, 0, interval, interval );
    /*
     * main loop, go forever.
     */
    do {
	int short_waits, short_wait_threshold, accepts, swing;
	/*
	 * check amount of HPSS activity while we slept.
	 */
	short_wait_threshold = 10000;		/* 1 seconds */
	status = hpss_reset_pool_lock ( &context, &short_wait_threshold,
		&short_waits, &accepts );
	/*
	 * Get current number of workers and determine number we
	 * should add or subtract this cycle.
	 */
	worker_count = validate_worker_list();
	delta = 0;
	swing = worker_count - min_workers;
	if ( swing < 0 ) {
	    /*
	     * Fell below hard limit, increase work count.  First time though
	     * loop will add initial_workers processes.  Afterwards we
	     * asymptotically appoach min_workers.
	     */
	    if ( worker_count == 0 ) delta = initial_workers;
	    else delta = 1 + (swing/4);

	} if ( (swing > 0) && (short_waits <= 0) ) {
	    /*
	     * see if excess workers can be layed off.
	     */
	    delta = (-1) - (swing/8);

	} else if ( (short_waits > 0) && (worker_count < max_workers) ) {
	    /*
	     * Clients are waiting for attention from workers, add some more.
	     */
	    delta = 1 + (max_workers-worker_count)/4;
	}
 	/*
	 * Update the worker pool.
	 */
	if ( delta > 0 ) {
	    status = add_workers ( delta, worker_boot );
	    if (debug_flag > 0) printf("status of add_workers: %d\n", status );
	} else if ( delta < 0 ) {
	    int count;
	    count = delta * (-1);

	    status = hpss_reduce_pool_workers ( &context, &count );
	    if (debug_flag > 0) printf (
		"Status of pool reduction (%d): %d\n", count, status );
	}
    } while ( 1&SYS$HIBER() );
    return status;
}
