/*
 * DECnet object for coordinated merging of access log records from
 * from multipe servers.  Server configuration specifies ddecnet task
 * string rather than a filename (i.e. accesslog 0::"0=WWWLOG").
 *
 * Usage:
 *    $ run cluster_log
 *
 * Required logical names:
 *    CLUSTER_LOG_ACCESS	Multi-valued logical name, each equivalence name
 *				is of form NODE::USERNAME.
 *
 * Required environment variables:
 *    CLUSTER_LOG_FILE		Access  filename.
 *    CLUSTER_LOG_OBJECT	Decnet object (e.g. WWWLOG) to declare.  Server
 *				sets accesslog to node::"0=WWWLOG".
 *
 * Required privileges:
 *    OPER
 *    SYSNAM
 */
#include "pthread_np.h"
#include <stdio.h>
#ifdef __DECC
/* Work around problems with fsync() not being ANSI standard */
#ifndef fsync
#define fsync decc$fsync
int decc$fsync(int fd);
#endif
#endif
#include <stdlib.h>
#include "tutil.h"
#include "tserver_decnet.h"

#define TID task_id(pthread_self())
int task_id();
/*
 * Lines to be written to file are staged to a global list of buffers drawn from
 * a pool.
 */
struct log_buf {
    struct log_buf *next;	/* Chain to next buffer in list */
    int type;			/* 1 - data, 2 - sync, 3 - newfile, 4 - exit */
    int length;			/* Line length used */
    char source[64];		/* node::user,task */
    char line[4096];
};

static struct log_buf *free_lbuf, *lbuf_queue_head, *lbuf_queue_tail;
static int lbuf_pending;		/* number waiting on free buffer */
static pthread_mutex_t lbuf_lock;
static pthread_cond_t lbuf_free, lbuf_flush;

static int new_client 
	( ts_client_ctx ctx, struct ncbdef *ncb, int ndx, int available );

/******************************************************************************/
/* Initialize program globals that manage buffer pool
 */
static void init_log_buffers ()
{
   int i;
   pthread_mutex_init ( &lbuf_lock, pthread_mutexattr_default );
   pthread_cond_init ( &lbuf_free, pthread_condattr_default );
   i = pthread_cond_init ( &lbuf_flush, pthread_condattr_default );
   if ( i ) printf("status of cond_init: %d\n", i );

   free_lbuf = lbuf_queue_head = lbuf_queue_tail = (struct log_buf *) 0;
   lbuf_pending = 0;

   for ( i = 0; i < 1000; i++ ) {
	struct log_buf *new;
	new = (struct log_buf *) malloc ( sizeof(struct log_buf) );
	new->next = free_lbuf;
	free_lbuf = new;
   }
}

/******************************************************************************/
/* Allocate buffer from free list, waiting if none available.
 */
static struct log_buf *alloc_log_buffer()
{
    struct log_buf *new;
    pthread_mutex_lock ( &lbuf_lock );
    while ( !free_lbuf ) {
	lbuf_pending++;
	pthread_cond_wait ( &lbuf_free, &lbuf_lock );
	--lbuf_pending;
    }
    new = free_lbuf;
    free_lbuf = new->next;
    pthread_mutex_unlock ( &lbuf_lock );
    return new;
}

/******************************************************************************/
/* Queue allocated  buffer to flusher thread (type 1-4) or place on
 * free list (type 0).  Type argument is saved in type field of buffer.
 */
static void dispose_log_buffer ( struct log_buf *buffer, int type )
{
    pthread_mutex_lock ( &lbuf_lock );
    if ( type > 0 && type < 5) {
	buffer->next = (struct log_buf *) 0;
	buffer->type = type;
	if ( !lbuf_queue_tail ) {
	    lbuf_queue_head = buffer;
	    pthread_cond_signal ( &lbuf_flush );
	} else lbuf_queue_tail->next = buffer;

	lbuf_queue_tail = buffer;
    } else {
	buffer->next = free_lbuf;
	free_lbuf = buffer;
	if ( lbuf_pending ) pthread_cond_signal ( &lbuf_free );
    }
    pthread_mutex_unlock ( &lbuf_lock );
}

/******************************************************************************/
static void *ticker ( int interval )
{
    struct timespec delta, abstime;
    pthread_cond_t timeout;
    struct log_buf *msg;
    int status;

    status = pthread_cond_init ( &timeout, pthread_condattr_default );

    delta.tv_sec = interval;
    delta.tv_nsec = 0;

    for ( ; ; ) {
	status = pthread_get_expiration_np ( &delta, &abstime );
	if ( status != 0 ) {
	    printf("Error computing expiration time\n");
	    break;
	}

        pthread_mutex_lock ( &lbuf_lock );
	while ( 0 == pthread_cond_timedwait 
		( &timeout, &lbuf_lock, &abstime ) ) {
	}
	pthread_mutex_unlock ( &lbuf_lock );

	msg = alloc_log_buffer ();
	dispose_log_buffer ( msg, 2 );
    }
    return (void *) 0;
}
/**************************************************************************/
/* The flusher routine opens the log file and receives messages to
 * process via the lbuf queue.
 */
static void *flusher ( char *log_fname )
{
    struct log_buf *cur;
    FILE *log;
    /*
     * Open log file.
     */
    log = fopen ( log_fname, "a", "dna=sys$disk:[].log", "mrs=900", "shr=upd" );
    if ( !log ) {
	fprintf(stderr,"error opening log file '%s'\n", log_fname );
	log = stdout;
    }
    /*
     * Main loop.
     */
    for ( ; ; ) {
	/*
	 * De-queue next buffer from queue, waiting when queue empty.
	 */
	pthread_mutex_lock ( &lbuf_lock );
	while (!lbuf_queue_head) pthread_cond_wait (&lbuf_flush, &lbuf_lock);
	cur = lbuf_queue_head;
	lbuf_queue_head = cur->next;
	if ( !lbuf_queue_head ) lbuf_queue_tail = lbuf_queue_head;
	pthread_mutex_unlock ( &lbuf_lock );
	/*
	 * Take action depending upon the buffer type.
	 */
	/* printf("%d: Dequeued buffer type %d\n", TID, cur->type ); */
	if ( cur->type == 1 ) { int status;
	    /*
	     * Data for log file, ensure newline present.
	     */
	    if ( cur->length > 0 ) if ( cur->line[cur->length-1] != '\n' ) {
	        if ( cur->length < 4095 ) cur->line[cur->length++] = '\n';
	    }
	    status = fwrite ( cur->line, cur->length, 1, log );
	    if ( status != 1 ) printf("fwrite length: %d status: %d\n", 
			cur->length, status );
	} else if ( cur->type == 2 ) {
	    /*
	     * Do periodic processing (sync file to disk)
	     */
	    fflush ( log );
	    fsync ( fileno(log) );
	} else if ( cur->type == 3 ) {
	    /*
	     * Create new version of log file.
	     */
	    fclose ( log );
	    log = fopen ( log_fname, "w", "dna=sys$disk:[].log", "mrs=900",
			"shr=upd" );
	    if ( !log ) {
		log = stdout;
	    }
	} else if ( cur->type == 4 ) {
	    /* 
	     * Shutdown message, used as marker to ensure all pending type 1
	     * message processed.  Buffer is auto variable of initial thread,
	     * so we don't move it to another queue.
	     */
	    break;
	}
	dispose_log_buffer ( cur, 0 );
    }
    /*
     * Initial thread is waiting for us to acknowlege  shutdown message by
     * clearing the type field in the shutdown message buffer.
     */
    pthread_mutex_lock ( &lbuf_lock );
    cur->type = 0;
    pthread_cond_signal ( &lbuf_free );
    pthread_mutex_unlock ( &lbuf_lock );

    return (void *) 0;
}

int main ( int argc, char **argv )
{
    int status, *final_status, pthread_create();
    char *object, *log_fname, *arg;
    pthread_attr_t attr;
    pthread_t listener, writer, clock;
    struct log_buf shutdown;
    /*
     * Get environment variables.
     */
    arg = getenv ( "CLUSTER_LOG_FILE" );
    if ( arg ) {
	log_fname = malloc ( tu_strlen(arg) + 1 );
	tu_strcpy ( log_fname, arg );
    } else {
	fprintf(stderr, "CLUSTER_LOG_FILE environment variable not defined\n");
	log_fname = (char *) 0;
    }
    arg = getenv ( "CLUSTER_LOG_OBJECT" );
    if ( arg ) {
	object = malloc ( tu_strlen(arg) + 1 );
	tu_strcpy ( object, arg );
    } else {
	fprintf(stderr, 
		"CLUSTER_LOG_OBJECT environment variable not defined\n");
	object = (char *) 0;
    }
    if ( !log_fname || !object ) exit ( 1 );
    /*
     * Initialize global variables.
     */
    init_log_buffers();
    status = ts_set_access ( "CLUSTER_LOG_ACCESS" );
    if ( (status&1) == 0 ) {
	fprintf(stderr,"Error setting access list\n");
	exit ( status );
    }
    /*
     * Create threads:
     *   Writer thread manages output file, writes coordinated through it.
     *   Clock thread periodically tells writer to flush buffers.
     *   Listener thread listens for client connects, creating new client
     *        threads (client limit set to 32).
     */
    pthread_attr_create ( &attr );
    pthread_attr_setstacksize ( &attr, 100000 );

    status = pthread_create ( &writer, attr, flusher, log_fname );
    if ( status != 0 ) printf("Thread create failure\n" );
    status = pthread_create ( &clock, attr, ticker, 60 );
    if ( status != 0 ) printf("Thread create failure\n" );
    status = ts_declare_decnet_object 
		( object, 32, &attr, &listener, new_client );
    if ( (status&1) == 0 ) {
	fprintf(stderr, "Error declaring object %s: %d\n", object, status );
	exit(status);
    }
    /*
     * Wait for listener thread exit.
     */
    pthread_join ( listener, (void *)  &final_status );

    dispose_log_buffer ( &shutdown, 4 );
    pthread_mutex_lock ( &lbuf_lock );
    while ( shutdown.type ) pthread_cond_wait ( &lbuf_free, &lbuf_lock );
    pthread_mutex_unlock ( &lbuf_lock );
    exit ( *final_status );
}

static int new_client ( ts_client_ctx ctx, 
	struct ncbdef *ncb, 
	int ndx, 
	int available )
{
    int status, length, i, j;
    struct log_buf *buf;
    char taskname[256], remote_node[256], remote_user[64];
    char source[64];

    ts_decnet_info ( taskname, remote_node, remote_user );
/*
    printf("%d: Connect to %s from %s at %s\n", TID, taskname, remote_user,remote_node);
*/
    tu_strcpy ( source, remote_node );
    tu_strcpy ( &source[tu_strlen(source)], remote_user );

    for ( ; ; ) {
	/*
	 * Get buffer to hold next line.
	 */
	buf = alloc_log_buffer ( );
	tu_strcpy ( buf->source, source );
	/*
	 * Fill buffer with next line of data from network link.
	 */
	for ( buf->length = 0; buf->length <= 0; ) {
	    status = ts_decnet_read ( ctx, buf->line, sizeof(buf->line)-1,
		&buf->length );
	    if ( (status&1) == 0 ) break;
	}
	/*
	 * Queue buffer to writer thread.
	 */
	if ( (status&1) == 1 ) {
	    if ( buf->length == 7 && 
		    0 == tu_strncmp ( buf->line, "/NEWLOG", 7 ) ) 
	        dispose_log_buffer ( buf, 3 );
	    else
	        dispose_log_buffer ( buf, 1 );
	} else {
	    dispose_log_buffer ( buf, 0 );	/* put back on free list */
	    break;
	}
    }
    return status;
}
int task_id(long addr, long tid ) { return (tid&0x0ffff); }
