/*
 * This module provides non-transparent DECnet connections to DECThreads
 * modules.
 *
 * int dnet_initialize();
 * int dnet_connect ( char *taskname, *dhandle );
 * int dnet_write ( dnhandle fptr, char *buffer, int bufsize, int *written );
 * int dnet_read ( dnhandle fptr, char *buffer, int bufsize, int *read );
 * int dnet_read_streamed ( dnhandle fptr, char *buffer, int bufsize, int *read );
 * int dnet_disconnect ( dnhandle fptr );
 * int dnet_format_error ( int code, char *buffer, int bufsize );
 * int dnet_set_time_limit ( dnhandle fptr, int seconds );
 *
 * Modified: 25-JAN-1996		support mbxnet.
 * Modified:  1-FEB-1996
 * Modified:  8-FEB-1996		Write EOF on mbxnet close.
 * Modified: 15-MAR-1996		cmbdef.h fix
 * Modified:  7-MAY-1997		Add time_limit.
 * Modified: 18-MAY-1997		Kill stalled mbxnet servers after 2 sec.
 * Modified: 14-JAN-1998		Fix bug in dnet_disconnect's handling
 *					of aborted mbxnet connections.
 * Modified: 3-MAR-1998			Change var to static in once routine.
 * Modified: 8-MAR-1998			add disown and claim routines.
 * Modified: 5-DEC-1998			Add dnet_read_streamed.
 * Modifed:  2-OCT-1999			Fix bugs in timeout checks not
 *					releasing mutex.
 */
#include "pthread_1c_np.h"
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <descrip.h>
#include <dvidef.h>
#include <iodef.h>
#include <jpidef.h>
#ifndef __DECC
#ifndef CMB$M_READONLY
#define CMB$M_READONLY 1
#define CMB$M_WRITEONLY 2
#endif
#else
#include <cmbdef.h>
#endif
#include <ssdef.h>
/* The ssdef.h may be out of date, hard code new values */
#ifndef SS$_NOREADER
#define SS$_NOREADER 0x024c4
#define SS$_NOWRITER 0x024cc
#endif
#ifndef IO$M_READERCHECK
/*
 * Out-of-date iodef.h, hard code the values for new mailbox driver options.
 */
#define IO$M_READERCHECK 0x0100
#define IO$M_WRITERCHECK 0x0200
#endif

#include "mbxnet.h"
#include "tutil.h"
#include "decnet_access.h"	/* validate prototypes against actual */
#define DISABLE_ASTS pthread_mutex_lock(&ctx_list); /* disable AST delivery */
#define ENABLE_ASTS pthread_mutex_unlock(&ctx_list); /* enable AST delivier */
int SYS$ASSIGN(), SYS$DASSGN(), SYS$QIO(), SYS$CREMBX(), SYS$CANCEL(),
	SYS$DELPRC();
/*
 * Global (module-wide) variables, Initialized by dnet_initialize.
 */
static $DESCRIPTOR(net_device,"_NET:");
static $DESCRIPTOR(mbxnet_device, "WWW_MBXNET_REQUEST");
static short mbxnet_request, mbxnet_response;
static int mbxnet_rsp_unit;
static netmbx_req_message response_message;
static long mbxnet_acppid;		/* 0 means use DECnet */
static long parent_pid;

struct dnet_iosb { short status; short length; long pid; };

struct connect_context {
    struct connect_context *flink, *blink;	/* Open connection list */
    int status;
    struct dnet_iosb iosb;
    short chan, mbxnet_state;
    int time_limit;			/* 0-none, 1-pending, 2-expired */
    struct timespec expiration;		/* Expiration time for timed I/O */
    pthread_cond_t io_done;
    long partner_pid;			/* server process */
};
typedef struct connect_context cnx_struct, *connection;

static int dnet_ef;
static connection free_connections;		/* Cache of free contexts. */
static pthread_mutex_t ctx_list;		/* protects connection list */
#define MBXNET_MAPSIZE 512
static connection mbxnet_active[MBXNET_MAPSIZE];
static mbxnet_active_count;		/* Max index in use */
static pthread_mutex_t dnet_io;			/* mutex for I/O cond. */
static pthread_key_t cnx_key;			/* Handle for pre-thread data */
/******************************************************************************/
/*
 * The following two routines are used to enabale a thread perform I/O 
 * without blocking the execution of other threads.  The thread must do
 * the following:
 *	1. Acquire io_done mutex.
 *	2  call SYS$QIO function specifying dnet_synch_ast as the
 *	   AST argument and the connection context as the parameter.
 *	3. call dnet_synch_io to get status and release the mutex.
 */
static int dnet_synch_ast ( connection ctx )
{
    int status;
    if ( ctx->status == 0 ) return 0;		/* on free list */
    /*
     * Signal any waiting threads.
     */
    status = pthread_cond_signal_int_np ( &ctx->io_done );
    return status;
}

static int dnet_synch_io ( int status, connection ctx )
{
    /*
     * Make sure operation is pending (success status).
     */
    if ( (status&1) == 1 ) {
	/*
	 * Loop until predicaate (ctx->iosb.status) is non-zero.
	 */
	do {
	    if ( ctx->time_limit ) {
		if ( pthread_cond_timedwait ( &ctx->io_done, &dnet_io,
			&ctx->expiration ) != 0 ) {
		    /*
		     * Timeout occurred, cancel I/O and wait for it to finish.
		     */
		    ctx->time_limit = 2;
		    if ( !ctx->iosb.status ) SYS$CANCEL ( ctx->chan );
		    while ( !ctx->iosb.status ) {
			if (pthread_cond_wait(&ctx->io_done, &dnet_io)) break;
		    }
		    break;
		}
	    } else {
	        pthread_cond_wait ( &ctx->io_done, &dnet_io );
	    }
	} while ( ctx->iosb.status == 0 );
	status = (unsigned) ctx->iosb.status;
    }
    /*
     * Condition satisfied, unlock mutex.
     */
    pthread_mutex_unlock ( &dnet_io );
    return status;
}
/***************************************************************************/
/* Start routine for private thread to listen for messages sent by
 * mbxnet ACP to our response mailbox.
 */
static void *dnet_mbxnet_dispatcher ( pthread_cond_t *ready )
{
    int status, i, code, LIB$GETJPI();
    struct dnet_iosb iosb;
    struct dsc$descriptor server_mbx;
    char mbxname[16];
    pthread_cond_t io_done;
    netmbx_req_message message;
    /*
     * Init condition used to synchronize decthreads with QIO.
     */
    INITIALIZE_CONDITION ( &io_done  );
    SET_COND_NAME(&io_done,"OSU DECnet io_done")
    parent_pid = 0;
    code = JPI$_OWNER;
    status = LIB$GETJPI ( &code, 0, 0, &parent_pid, 0, 0 );
    /*
     * Register ourselves with the ACP, primarily to get it's PID
     * Number in parenthesis is options flags.
     */
    message.gen.type = MSG$_NODEACC;
    message.gen.unit = mbxnet_rsp_unit;
    tu_strcpy ( message.gen.info, "HTTP server, flags=(1)" );
    pthread_mutex_lock ( &dnet_io );
    status = SYS$QIO ( dnet_ef, mbxnet_request, 
	IO$_WRITEVBLK|IO$M_READERCHECK, &iosb,
	pthread_cond_signal_int_np, &io_done,
	&message, 8 + tu_strlen(message.gen.info), 0, 0, 0, 0 );
    if ( (status&1) == 1 ) {
	while ( iosb.status == 0 ) pthread_cond_wait ( &io_done, &dnet_io );
	status = iosb.status;
    }
    if ( (status&1) == 1 ) {
	mbxnet_acppid = iosb.pid;
        /*
         * Write second message to synchronize update.
         */
        status = SYS$QIO ( dnet_ef, mbxnet_request, IO$_WRITEOF, &iosb,
		pthread_cond_signal_int_np, &io_done,
		&message, 8 + tu_strlen(message.gen.info), 0, 0, 0, 0 );
    	if ( (status&1) == 1 ) {
	    while ( iosb.status == 0 ) pthread_cond_wait ( &io_done, &dnet_io );
	    status = iosb.status;
	}
    } else mbxnet_acppid = 0;
    pthread_cond_signal ( ready );
    /*
     * Main loop, continue to read as long as mailbox valid.  Note that
     * we hold dnet_io mutex entire time we process message (released
     * during cond_wait calls).
     */
    while ( mbxnet_acppid ) {
	/*
	 * Read message, ignore errors except for nowriters.
	 */
	status = SYS$QIO ( dnet_ef, mbxnet_response,
		IO$_READVBLK|IO$M_WRITERCHECK, &iosb,
		pthread_cond_signal_int_np, &io_done,
		&message, sizeof(message), 0, 0, 0, 0 );
        if ( (status&1) == 1 ) {
	    while ( iosb.status == 0 ) pthread_cond_wait ( &io_done, &dnet_io );
	    status = iosb.status;
        }
#ifdef DEBUG
    printf(
	"Dispatcher read loop qio comp: %d iosb.size: %d, pid: %x/%x, unit: %d\n",
	 status,  iosb.length, iosb.pid, message.gen.pid, message.gen.unit );
#endif
	if ( (status&1) == 0 ) {
	    if ( status == SS$_NOWRITER ) {
		LOCK_C_RTL
#ifdef DEBUG
		printf("ACP died%s\n", parent_pid == mbxnet_acppid ? 
			" ACP is master" : "" );
#endif
		UNLOCK_C_RTL
		if ( parent_pid && (mbxnet_acppid == parent_pid) )
			exit(SS$_SHUT);
	    }
	    continue;
	}
	if ( iosb.pid != mbxnet_acppid ) continue;	/* wrong sender */
	/*
	 * Dispatch on code.
	 */
	for ( i = 1; i <= mbxnet_active_count; i++ ) {
	    if ( message.gen.pid == mbxnet_active[i]->partner_pid ) {
		break;
	    }
	}
	if ( message.gen.type == MSG$_CONFIRM ) {
	     /*
	      * PID field is partner process and unit is link mailbox.
	      */
	    if ( i <= mbxnet_active_count ) {
		/* existing connection that wasn't rundown */
		mbxnet_active[i]->mbxnet_state = 2;
	        status = SYS$CANCEL ( mbxnet_active[i]->chan );
		pthread_cond_signal ( &mbxnet_active[i]->io_done );
	    }
	    /*
	     * Search for next ctx on active list wanting a connection.
	     */
	    for ( i = 1; i <= mbxnet_active_count; i++ ) {
		if ( mbxnet_active[i]->partner_pid == 0 &&
			mbxnet_active[i]->mbxnet_state == 1 ) {
		    break;
		}
	    }
	    if ( i > mbxnet_active_count ) continue;	/* spurious confirm */
	    /*
	     * Assign channel to mailbox unit named.
	     */
	    tu_strcpy ( mbxname, "_MBA" );
	    tu_strint ( message.gen.unit, &mbxname[4] );
	    server_mbx.dsc$b_dtype = DSC$K_DTYPE_T;		/* text data */
	    server_mbx.dsc$b_class = DSC$K_CLASS_S;		/* fixed (Static) */
	    server_mbx.dsc$w_length = tu_strlen(mbxname);
	    server_mbx.dsc$a_pointer = mbxname;

	    status = SYS$ASSIGN ( &server_mbx, &mbxnet_active[i]->chan,
		0, 0, 0 );
	    if ( (status&1) == 0 ) {
		/*
		 * assign failed, disconnect.
		 */
		mbxnet_active[i]->mbxnet_state = 2;
	    }
	    mbxnet_active[i]->partner_pid = message.gen.pid;
	    if ( !message.gen.pid ) {
		mbxnet_active[i]->mbxnet_state = 2;
	        status = SYS$CANCEL ( mbxnet_active[i]->chan );
	    }
	    status = pthread_cond_signal ( &mbxnet_active[i]->io_done );

	} else if ( message.gen.type == MSG$_DISCON ) {
	    if ( i > mbxnet_active_count ) {
		/*
		 * PID not matched, see if connection in progress.
		 */
		if ( message.gen.unit != 0 ) continue;	/* spurious discon */
		for ( i = 1; i <= mbxnet_active_count; i++ ) {
		    if ( mbxnet_active[i]->partner_pid == 0 &&
			mbxnet_active[i]->mbxnet_state == 1 ) {
		        break;
		    }
		}
		if ( i > mbxnet_active_count ) continue;
	    }
	    if ( mbxnet_active[i]->mbxnet_state == 1 ) { 
		mbxnet_active[i]->mbxnet_state = 2;	/* close connection */
		status = SYS$CANCEL ( mbxnet_active[i]->chan );
		status = pthread_cond_signal ( &mbxnet_active[i]->io_done );
	    }
	    mbxnet_active[i] = mbxnet_active[mbxnet_active_count];
	    --mbxnet_active_count;
	}
    }
    /*
     * Getting to this point (loop exited) means ACP died.  Scan for active 
     * users and disconnect them.
     */
    for ( i = 1; i <= mbxnet_active_count; i++ ) {
	if ( mbxnet_active[i]->mbxnet_state == 1 ) { 
	    mbxnet_active[i]->mbxnet_state = 2;	/* close connection */
	    status = SYS$CANCEL ( mbxnet_active[i]->chan );
	    pthread_cond_signal ( &mbxnet_active[i]->io_done );
	}
    }
    pthread_mutex_unlock ( &dnet_io );
    return (void *) 0;
}
/***************************************************************************/
/* Cleanup any decnet connection contexts associated with a thread.
 * The ctx argument is the first context in the list of contexts allocated
 * by the exiting thread.
 */
static void dnet_rundown ( connection ctx )
{
    connection cur, next;

    for ( cur = ctx; cur; cur = next ) {
	/*
	 * Deassign channel
	 */
	LOCK_C_RTL
	printf("/decnet/ running down dangling connection, ctx->status=%d\n",
		cur->status );
	UNLOCK_C_RTL
	next = cur->flink;
	SYS$DASSGN ( cur->chan );
	cur->chan = 0;
	if ( mbxnet_acppid ) {
	    /* Remove from active table */
	    int i;
	    pthread_mutex_lock ( &dnet_io );
	    for ( i = 1; i <= mbxnet_active_count; i++ ) {
		if ( mbxnet_active[i] == ctx ) {
		    mbxnet_active[i] = mbxnet_active[mbxnet_active_count];
		    --mbxnet_active_count;
		    break;
		}
	    }
	    pthread_mutex_unlock ( &dnet_io );
	}
	/*
	 * Place block on free list.
	 */
	DISABLE_ASTS
	cur->flink = free_connections;
	free_connections = cur;
	ENABLE_ASTS
    }
}
/****************************************************************************/
/*  Routine to peform 1-time initilization at program startup.
 */
static int dnet_init_status;
static void dnet_init_once()
{
    int status, mstatus, flags, LIB$GET_EF();
    pthread_t dispatcher;
    /*
     * Initialize decthreads objects.
     */
    free_connections = (connection) NULL;
    status = INITIALIZE_MUTEX ( &ctx_list );
    SET_MUTEX_NAME(&ctx_list,"OSU DECnet ctx_list")
    if ( status != 0 ) { dnet_init_status = 0; return; }
    status = INITIALIZE_MUTEX ( &dnet_io );
    SET_MUTEX_NAME(&dnet_io,"OSU DECnet dnet_io")
    status = CREATE_KEY (&cnx_key, (pthread_destructor_t) dnet_rundown);
    if ( status != 0 ) perror ( "Error creating file access context key" );
    SET_KEY_NAME(cnx_key,"OSU DECnet rundown")

    LOCK_VMS_RTL
    status = LIB$GET_EF ( &dnet_ef );
    UNLOCK_VMS_RTL
    /*
     * Check for substitute DECnet acp.  flags will specify readonly.
     */
    mbxnet_acppid = 0;
    mbxnet_active_count = 0;
    flags = CMB$M_WRITEONLY;
    mstatus = SYS$ASSIGN ( &mbxnet_device, &mbxnet_request, 0, 0, flags );
    if ( (mstatus&1) == 1 ) {
	int LIB$GETDVI(), code;
	static pthread_cond_t ready;
	/*
	 * Create mailbox to get response and find its unit number.
	 */
	flags = CMB$M_READONLY;
	code = DVI$_UNIT;
	status = SYS$CREMBX ( 0, &mbxnet_response, 256, 512, 0xff00, 0,
		0, flags );
	mbxnet_rsp_unit = 0;
	if ( (status&1) == 1 ) status = LIB$GETDVI ( &code,
		&mbxnet_response, 0, &mbxnet_rsp_unit );
	/*
	 * Create thread to read the response mailbox.
	 */
	INITIALIZE_CONDITION ( &ready );
	SET_COND_NAME(&ready,"OSU DECNET disp. init")
	mbxnet_acppid = -1;
	pthread_mutex_lock ( &dnet_io );
#ifdef PTHREAD_USE_D4
	status = pthread_create ( &dispatcher, pthread_attr_default,
		(pthread_startroutine_t) dnet_mbxnet_dispatcher, 
		(pthread_addr_t) &ready );
#else
	status = pthread_create ( &dispatcher, NULL, 
		(pthread_startroutine_t) dnet_mbxnet_dispatcher, 
		(void *) &ready );
#endif
	if ( status == 0 ) {
	    status = 1;
	    while ( mbxnet_acppid == -1 ) 
		pthread_cond_wait ( &ready, &dnet_io );
	} else status = 0;
	pthread_mutex_unlock ( &dnet_io );
	
    }
    dnet_init_status = status;
}
/*
 * External routine to initialize module.
 */
int dnet_initialize()
{
    static pthread_once_t setup = PTHREAD_ONCE_INIT;

    pthread_once ( &setup, dnet_init_once );
    return dnet_init_status;
}
/****************************************************************************/
/*
 * Establish connection to designated task.  Taskname must be valid
 * DECnet task specification:
 *
 *	node["user [password]"]::"[0=task|object]"
 */
int dnet_connect ( char *taskname, void **dptr )
{
    connection ctx;
    int i, status, SYS$ASSIGN(), length;
    struct { int length; char *data; } ncb_desc;
    netmbx_req_message message;
    char *ncb;
    /*
     * Allocate a connection context block.
     */
    *dptr = (void *) NULL;
    ncb = message.gen.info;
    DISABLE_ASTS
    ctx = free_connections;
    if ( ctx ) free_connections = ctx->flink;
    ENABLE_ASTS

    if ( ctx == (connection) NULL ) {
	LOCK_C_RTL
	ctx = (connection) malloc ( sizeof(cnx_struct) );
	UNLOCK_C_RTL
	if ( !ctx ) return 0;
	INITIALIZE_CONDITION ( &ctx->io_done );
        SET_COND_NAME(&ctx->io_done,"OSU DECnet cnx io_done")
	ctx->flink = ctx->blink = (connection) NULL;
	ctx->status = 2;
	ctx->chan = 0;
    }
    ctx->time_limit = 0;
    /*
     * Build NCB and descriptor.  Replace final quote with
     * ncb connect/opt-data structure.
     */
    tu_strnzcpy ( ncb, taskname, 80 );
    length = tu_strlen ( ncb );
    if ( length > 0 ) ncb[length-1] = '/';
    ncb[length++] = 0;
    ncb[length++] = 0;			/* zero word (new connect */
    for ( i = 0; i < 17; i++ ) ncb[length++] = 0; 	/* null opt. data */
    ncb[length++] = '"';		/* closing quote */

    ncb_desc.length = length;
    ncb_desc.data = ncb;
    /*
     * See if using surrogate DECnet.
     */
    if ( mbxnet_acppid ) {
	/*
	 * Send connect request to acp process.
	 */
	ctx->mbxnet_state = 1;			/* open */
	ctx->partner_pid = 0;			/* no connection */
	ctx->chan = 0;
        status = pthread_mutex_lock ( &dnet_io );
	mbxnet_active_count += 1;
	mbxnet_active[mbxnet_active_count] = ctx;
	message.gen.type = MSG$_CONNECT;	/* request connection */
	message.gen.unit = mbxnet_rsp_unit;
	status = SYS$QIO ( dnet_ef, mbxnet_request,
		IO$_WRITEVBLK|IO$M_READERCHECK, &ctx->iosb,
		dnet_synch_ast, ctx, &message, length + 8, 0, 0, 0, 0 );
	status = dnet_synch_io ( status, ctx );
	if ( (status&1) == 0 ) ctx->mbxnet_state = 2;
	else if ( ctx->iosb.pid != mbxnet_acppid ) {
	    ctx->mbxnet_state = 2;
	}
	/*
	 * wait for dispatcher thread to fill in the partner PID.
	 * (or set mbxnet_state to closed status on failure).  Dispatcher
	 * assigns channel to link mailbox before signaling.
	 */
	pthread_mutex_lock ( &dnet_io );
	while ( !ctx->partner_pid && (ctx->mbxnet_state==1) ) {
		pthread_cond_wait ( &ctx->io_done, &dnet_io );
	}
	if ( ctx->mbxnet_state != 1 ) {
	    status = SS$_ENDOFFILE;
	    pthread_mutex_unlock ( &dnet_io );
	} else {
	    /*
	     * send connect message to partner as first message.
	     */
	    ctx->status = 1;
	    status = SYS$QIO ( dnet_ef, ctx->chan,
		IO$_WRITEVBLK, &ctx->iosb,
		dnet_synch_ast, ctx, &message, length + 8, 0, 0, 0, 0 );
	    status = dnet_synch_io ( status, ctx );
        }
    } else {
	/*
	 * Attempt connection to decnet object.
         */
	ctx->mbxnet_state = 0;			/* using DECnet */
        status = SYS$ASSIGN ( &net_device, &ctx->chan, 0, 0 );
        if ( (status&1) == 0 ) {
	    DISABLE_ASTS 
	    ctx->flink = free_connections; free_connections = ctx;
	    ENABLE_ASTS
	    return 0;
        }

        status = pthread_mutex_lock ( &dnet_io );
        ctx->status = 1;
        status = SYS$QIO ( dnet_ef, ctx->chan, IO$_ACCESS, &ctx->iosb,
		dnet_synch_ast, ctx,
		0, &ncb_desc, 0, 0, 0, 0 );
        status = dnet_synch_io ( status, ctx );
    }
    if ( (status&1) == 1 ) {
	/*
	 * Add to context list for this thread.
	 */
	connection cur;
	GET_SPECIFIC(cnx_key, cur)
	ctx->blink = (connection) NULL;
	if ( cur ) cur->blink = ctx;
	ctx->flink = cur;
	pthread_setspecific ( cnx_key, ctx );
    } else {
	/* Failure */
	ctx->status = 0;
	if ( !mbxnet_acppid ) SYS$DASSGN ( ctx->chan );
	DISABLE_ASTS
	ctx->flink = free_connections; free_connections = ctx;
	ENABLE_ASTS
	ctx = (connection) NULL;
    }
    /*
     * return status and context poitner to caller.
     */
    *dptr = (void *) ctx;
    return status;
}
/****************************************************************************/
int dnet_write ( void *dptr, char *buffer, int bufsize, int *written )
{
    int status;
    connection ctx;

    ctx = (connection) dptr;
    status = pthread_mutex_lock ( &dnet_io );
    if ( ctx->mbxnet_state ) {
	/*
	 * Check for closed connection.
	 */
	if ( ctx->mbxnet_state == 2 ) {
	    status = SS$_LINKDISCON;
	    *written = 0;
	    pthread_mutex_unlock ( &dnet_io );
	    return status;
	}
    }
    if ( ctx->time_limit == 2 ) {
	status = 556;
	ctx->iosb.length = 0;
    } else {
	status = SYS$QIO ( dnet_ef, ctx->chan, IO$_WRITEVBLK, &ctx->iosb,
		dnet_synch_ast, ctx,
		buffer, bufsize, 0, 0, 0, 0 );
    }
    status = dnet_synch_io ( status, ctx );

    *written = (unsigned) ctx->iosb.length;
    return status;
}
/****************************************************************************/
/* Read record from DECnet mailbox.
 */
int dnet_read ( void *dptr, char *buffer, int bufsize, int *read )
{
    int status;
    connection ctx;

    ctx = (connection) dptr;
    status = pthread_mutex_lock ( &dnet_io );
    if ( ctx->mbxnet_state ) {
	/*
	 * Check for closed connection.
	 */
	if ( ctx->mbxnet_state == 2 ) {
	    status = SS$_LINKDISCON;
	    *read = 0;
	    pthread_mutex_unlock ( &dnet_io );
	    return status;
	}
    }
    if ( ctx->time_limit == 2 ) {	/* timed out */
	status = 556;
	ctx->iosb.length = 0;
    } else { 
	status = SYS$QIO ( dnet_ef, ctx->chan, IO$_READVBLK, &ctx->iosb,
		dnet_synch_ast, ctx,
		buffer, bufsize, 0, 0, 0, 0 );
    }
    status = dnet_synch_io ( status, ctx );
    *read = (unsigned) ctx->iosb.length;
    return status;
}
/****************************************************************************/
/* Read bytes from DECnet mailbox, ignoring record boundaries (requires
 * 6.2).  Note that in stream mode, zero-length bufsize is not allowed.
 */
int dnet_read_streamed ( void *dptr, char *buffer, int bufsize, int *read )
{
    int status, func;
    connection ctx;

    ctx = (connection) dptr;
    status = pthread_mutex_lock ( &dnet_io );
    /*
     * Select proper function code for QIO call.
     */
    if ( ctx->mbxnet_state ) {
#ifndef IO$M_STREAM
#define IO$M_STREAM 0
#endif
	func = IO$_READVBLK | IO$M_STREAM;
	/*
	 * Check for closed connection.
	 */
	if ( ctx->mbxnet_state == 2 ) {
	    status = SS$_LINKDISCON;
	    *read = 0;
	    pthread_mutex_unlock ( &dnet_io );
	    return status;
	}
    }
    else func = IO$_READVBLK | IO$M_MULTIPLE;
    /*
     * Check for timeout and issue QIO.
     */
    if ( ctx->time_limit == 2 ) {
	status = 556;
	ctx->iosb.length = 0;
    } else { status = SYS$QIO ( dnet_ef, ctx->chan, func, &ctx->iosb,
		dnet_synch_ast, ctx,
		buffer, bufsize, 0, 0, 0, 0 );
    }
    /*
     * Wait for completion and convert status if using DECnet driver.
     */
    status = dnet_synch_io ( status, ctx );
    if ( (status == SS$_BUFFEROVF) && !ctx->mbxnet_state ) status = SS$_NORMAL;
    *read = (unsigned) ctx->iosb.length;
    return status;
}
/****************************************************************************/
/* Perform a synchronous disconnect operation.
 */
int dnet_disconnect ( void *dptr )
{
    int status;
    connection ctx, first, cur;
    /*
     * Perform the disconnect.
     */
    ctx = (connection) dptr;
    if ( ctx->status == 0 ) return 20;
    status = pthread_mutex_lock ( &dnet_io );
    if ( ctx->mbxnet_state ) {
	/* Write EOF to kill kills doing final read */
	if ( ctx->mbxnet_state == 1 ) {
	    dnet_set_time_limit ( dptr, 2 );	/* give 2 seconds */
	    status = SYS$QIO ( dnet_ef, ctx->chan, IO$_WRITEOF,
		&ctx->iosb, dnet_synch_ast, ctx,
		0, 0, 0, 0, 0, 0 );
	    status = dnet_synch_io ( status, ctx );
	    if ( (status==SS$_ABORT) && ctx->partner_pid ) {
		/* We timed out, kill the process */
		printf("Deleting stalled process: %x\n", ctx->partner_pid );
		SYS$DELPRC ( &ctx->partner_pid, 0 );
	    }
	} else {
	    /*  Normally, dnet_synch_io unlocks mutex, do it explicitly
	     *  since we aren't calling it.
	     */
	    pthread_mutex_unlock ( &dnet_io );
	    status = 0;
	}
    } else {
	ctx->time_limit = 0;	/* disable time limit */
        status = SYS$QIO ( dnet_ef, ctx->chan, IO$_DEACCESS|IO$M_SYNCH, 
		&ctx->iosb, dnet_synch_ast, ctx,
		0, 0, 0, 0, 0, 0 );
	status = dnet_synch_io ( status, ctx );
    }
    ctx->status = 0;
    SYS$DASSGN ( ctx->chan );
    /*
     * Verify the connection belongs to thread.
     */
    GET_SPECIFIC(cnx_key, cur)
    for ( ; cur; cur = cur->flink ) {
	if ( cur == ctx ) break;
    }
    if ( cur != ctx ) {
	LOCK_C_RTL
	printf("Disconnect of unowned dnet connection\n");
	UNLOCK_C_RTL
    }
    /*
     * Remove from thread's connection list.
     */
    if ( ctx->flink ) ctx->flink->blink = ctx->blink;
    if ( ctx->blink ) ctx->blink->flink = ctx->flink;
    else pthread_setspecific ( cnx_key, ctx->flink );
    if ( mbxnet_acppid ) {
	/* Remove from active table */
	int i;
	pthread_mutex_lock ( &dnet_io );
	for ( i = 1; i <= mbxnet_active_count; i++ ) {
	    if ( mbxnet_active[i] == ctx ) {
		    mbxnet_active[i] = mbxnet_active[mbxnet_active_count];
		    --mbxnet_active_count;
		    break;
	    }
	}
	pthread_mutex_unlock ( &dnet_io );
    }
    /*
     * Place control block onto free list.
     */
    DISABLE_ASTS
    ctx->flink = free_connections; free_connections = ctx;
    ENABLE_ASTS

    return status;
}
/****************************************************************************/
/* Disassociate connection block from thread's context so it won't be
 * disconnected on rundown.  Returns 0 on error, 1 on success.
 */
int dnet_disown_context ( void *dptr )
{
    int status;
    connection ctx, first, cur;
    /*
     * Verify the connection belongs to thread.
     */
    ctx = (connection) dptr;
    if ( ctx->status == 0 ) return 20;
    GET_SPECIFIC(cnx_key, cur)
    for ( ; cur; cur = cur->flink ) {
	if ( cur == ctx ) break;
    }
    if ( cur != ctx ) {
	LOCK_C_RTL
	printf("Disown of unowned dnet connection\n");
	UNLOCK_C_RTL
	return 0;
    }
    /*
     * Remove from thread's connection list.
     */
    ctx->time_limit = 0;			/* clear timeouts */
    if ( ctx->flink ) ctx->flink->blink = ctx->blink;
    if ( ctx->blink ) ctx->blink->flink = ctx->flink;
    else pthread_setspecific ( cnx_key, ctx->flink );

    return 1;
}
/****************************************************************************/
/* Move unowned context into connection chain for of current thread, 'claiming'
 * it.  No checks are performed for whether context is currently owned.
 */
int dnet_claim_context ( void *dptr )
{
    int status;
    connection ctx, first, cur;
    /*
     * Link into this thread's chain.
     */
    ctx = (connection) dptr;
    if ( ctx->status == 0 ) return 20;
    GET_SPECIFIC(cnx_key, cur)
    ctx->blink = (connection) NULL;
    if ( cur ) cur->blink = ctx;
    ctx->flink = cur;
    pthread_setspecific ( cnx_key, ctx );
    
    return 1;
}
/****************************************************************************/
int dnet_format_error ( int code, char *buffer, int bufsize )
{
    int flags, status, SYS$GETMSG(), msglen, info;
    struct dsc$descriptor buf;

    buf.dsc$b_dtype = DSC$K_DTYPE_T;		/* text data */
    buf.dsc$b_class = DSC$K_CLASS_S;		/* fixed (Static) */
    buf.dsc$w_length = bufsize - 1;
    buf.dsc$a_pointer = buffer;
    flags = 0;

    pthread_lock_global_np();
    msglen = 0;
    status = SYS$GETMSG ( code, &msglen, &buf, flags, &info );
    pthread_unlock_global_np();
    if ( (status&1) == 1 ) buffer[msglen] = '\0';
    else buffer[0] = '\0';
    return status;
}
/****************************************************************************/
/* Set connection so that I/O after limit seconds will fail.
 */
int dnet_set_time_limit ( void *dptr, int limit )
{
    int status;
    connection ctx, first, cur;
    /*
     * Convert handle to pointer.
     */
    ctx = (connection) dptr;
    if ( ctx->status == 0 ) return 20;

    if ( limit == 0 ) {
	ctx->time_limit = 0;		/* turn off time limit */
    } else {
	/*
	 * Compute absolute expriation time limit seconds from now.
	 */
	struct timespec delta;
	delta.tv_sec = limit;
	delta.tv_nsec = 0;
	if ( 0 == pthread_get_expiration_np ( &delta, &ctx->expiration ) ) {
	    ctx->time_limit = 1;	/* time limit now active */
	} else {
	    ctx->time_limit = 0;	/* no expiration */
	    return 0;			/* error */
	}
    }
    return 1;
}
