/*
 * Define thread-friendly wrappers for basic  TCP/IP client operations.
 * This version is implemented using the UCX QIO interface, to compile with
 * multinet, define MULTINET.
 *
 * Revised:	8-MAY-2002	Fix bug in kernel-thread version of
 *				QIO_AND_WAIT macro.
 */
#include <stdio.h>
#include <stdlib.h>
#include <descrip.h>
#include <iodef.h>
#include <ssdef.h>
#include "tclient_tcp.h"
#include "pthread_1c_np.h"
#include "tutil.h"

#ifdef MULTINET
#include "MULTINET_ROOT:[MULTINET.INCLUDE.SYS]TYPES.H"
#include "MULTINET_ROOT:[MULTINET.INCLUDE.NETINET]IN.H"
#include "MULTINET_ROOT:[MULTINET.INCLUDE.SYS]SOCKET.H"
#include "MULTINET_ROOT:[MULTINET.INCLUDE.VMS]INETIODEF.H"

#define SOCKADDRIN sockaddr_in
#define SIN$W_FAMILY sin_family
#define SIN$W_PORT sin_port
#define SIN$L_ADDR sin_addr.s_addr
#define UCX$C_TCP IPPROTO_TCP
#define UCX$C_STREAM SOCK_STREAM
#define UCX$C_AF_INET AF_INET
#define UCX$C_SOCKOPT 1
#define UCX$C_LINGER SO_LINGER
#define UCX$C_REUSEADDR SO_REUSEADDR
#define INETACP$C_TRANS 2
#define INETACP_FUNC$C_GETHOSTBYNAME 1
#else
#include <UCX$INETDEF.H>
#define sockaddr_in SOCKADDRIN
#ifdef TCPIP$INETDEF_LOADED
/* TCPIP 5.0 changed the definitions in ucx$inetdef. */
#define SOCKADDRIN _SOCKADDRIN
#ifndef UCX$C_AF_INET
#define UCX$C_AF_INET TCPIP$C_AF_INET
#endif
#ifndef UCX$C_SOCKOPT
#define UCX$C_SOCKOPT TCPIP$C_SOCKOPT
#endif
#ifndef UCX$C_REUSEADDR
#define UCX$C_REUSEADDR TCPIP$C_REUSEADDR
#endif
#ifndef UCX$C_KEEPALIVE
#define UCX$C_KEEPALIVE TCPIP$C_KEEPALIVE
#endif
#ifndef UCX$C_TCP
#define UCX$C_TCP TCPIP$C_TCP
#endif
#ifndef UCX$C_STREAM
#define UCX$C_STREAM TCPIP$C_STREAM
#endif
#ifndef UCX$C_LINGER
#define UCX$C_LINGER TCPIP$C_LINGER
#endif
#endif
#endif

static $DESCRIPTOR(ucx_device,"UCX$DEVICE");

static pthread_once_t setup = PTHREAD_ONCE_INIT;
static pthread_mutex_t client_io;	/* lock for I/O synchronization */
static pthread_mutex_t client_ctl;	/* lock for global client list */
static pthread_key_t client_key;	/* hook for context */

struct client_context {
    struct client_context *next;
    int chan;			/* VMS I/O channel */
    int cond_valid;
    struct { unsigned short status, count; 
	unsigned long devsts; } iosb;
    pthread_cond_t io_done;
    struct timespec expiration;		/* expriation time for timed I/O */
    int time_limit;			/* 0-none, 1-pending, 2-expired */
    /*
     * Asynch control.
     */
    int completed_flags;		/* bit set indicates done */
    int pending_flags;			/* bit set indicates stalled. */
    int bytes_read;
    int bytes_written;
    int rbuf_size;
    int wbuf_size;
    struct { unsigned short status, count; 
	unsigned long devsts; } wiosb, riosb;
    char *rbuf;
    char *wbuf;
};
typedef struct client_context *client_ctx;
static client_ctx free_ctx;

int http_log_level;
int os_kernel_threads();
int SYS$DASSGN(), SYS$QIO(), SYS$ASSIGN(), SYS$CANCEL();

#define HTONS(_x) (((_x)/256)&255) | (((_x)&255)<<8);
/*************************************************************************/
#ifdef PTHREAD_USE_D4
/*
 * Draft 4 library lacks upcalls so always synchronize via ASTs signalling
 * a condition variable.
 */
int SYS$QIO();
#define QIO_AND_WAIT(cond,chan,func,iosb,p1,p2,p3,p4,p5,p6) \
     (pthread_mutex_lock(&client_io),\
      synch_io(SYS$QIO(1,chan,func,iosb,pthread_cond_signal_int_np,cond,\
		p1,p2,p3,p4,p5,p6),cond, (short *) iosb))
#else
/*
 * With standard V7 library, select synchronization method based upon whether
 * kernel thread upcalls enabled.
 */
int SYS$QIOW(), SYS$QIO(), os_kernel_threads();
static int kernel_threads = 0;
#ifdef __ALPHA
#define ENF 128
#else
#define ENF 1
#endif
#define QIO_AND_WAIT(cond,chan,func,iosb,p1,p2,p3,p4,p5,p6) \
 (kernel_threads ? \
     tc_ios(SYS$QIOW(ENF,chan,func,iosb,0,0,p1,p2,p3,p4,p5,p6), \
	(short *) iosb ) : \
     (pthread_mutex_lock(&client_io),\
     synch_io(SYS$QIO(ENF,chan,func,iosb,pthread_cond_signal_int_np,\
		cond,p1,p2,p3,p4,p5,p6),cond, (short *) iosb)) )
#endif

static int tc_ios ( int status, short *iosb )
{ 
    if ( (status&1) == 1 ) status = *iosb; 
    return status; 
}
static int synch_io ( int status, pthread_cond_t *condition, short *iosb )
{
    /*
     * Make sure operation is pending (success status).
     */
    if ( (status&1) == 1 ) {
	/*
	 * Loop until predicaate (iosb.status) is non-zero.
	 */
	do {
	    pthread_cond_wait ( condition, &client_io );
	} while ( iosb[0] == 0 );
	status = (unsigned short) iosb[0];
    }
    /*
     * Condition satisfied, unlock mutex.
     */
    pthread_mutex_unlock ( &client_io );
    return status;
}
/***************************************************************************/
static int synch_client ( int status, client_ctx ctx )
{
    /*
     * Only do wait if system service status OK.
     */
    if ( (status&1) == 1 ) {
	/*
	 * wait is either timed or open.
	 */
	if ( ctx->time_limit ) {
	    do {
	        status = pthread_cond_timedwait ( &ctx->io_done, &client_io,
			&ctx->expiration );
		if ( status != 0 ) {
		     /*
		      * Timeout occurred, cancel I/O and wait on it to finish.
		      */
		    ctx->time_limit = 2;
		    if ( ctx->iosb.status==0 ) SYS$CANCEL ( ctx->chan );
		    while ( !ctx->iosb.status ) {
			if (pthread_cond_wait(&ctx->io_done, &client_io)) break;
		    }
		    break;
		}
            } while ( ctx->iosb.status == 0 );
	} else {
            /*
             * Loop until predicate (iosb nozero) is true.
             */
	    do {
	        status = pthread_cond_wait ( &ctx->io_done, &client_io );
	        if ( status != 0 ) break;
            } while ( ctx->iosb.status == 0 );

	}
	status = ctx->iosb.status;
    }
    pthread_mutex_unlock ( &client_io );
    return status;
}
/************************************************************************/
/* Define context key rundown routine that closes dangling files.
 * Return value 0 for success.
 */
static void tc_rundown ( client_ctx ctx )
{
    client_ctx cur;
    if ( !ctx ) return;			/* null context value */
    /*
     * Close open connections.
     */
    for ( cur = ctx; cur; cur = cur->next ) {
	SYS$DASSGN ( cur->chan );
    }
    /*
     * Place blocks onto free-list.
     */
    for ( cur=ctx; cur->next; cur = cur->next );
    pthread_mutex_lock ( &client_ctl );
    cur->next = free_ctx;
    free_ctx = ctx;
    pthread_mutex_unlock ( &client_ctl );
}
/***************************************************************************/
/* Initialization routine, called once.
 */
static void module_init() {
    int status, i;
    /*
     * Initialize pthread structures.
     */
    status = INITIALIZE_MUTEX ( &client_io );
    SET_MUTEX_NAME ( &client_io, "OSU tclient IO" )
    status = INITIALIZE_MUTEX ( &client_ctl );
    SET_MUTEX_NAME ( &client_io, "OSU tclient control" )
    status = CREATE_KEY ( &client_key, (pthread_destructor_t) tc_rundown );
    SET_KEY_NAME ( client_key, "OSU tclient rundown" )
    /*
     * Initialize free list of contexts with 10 free blocks.
     */
    LOCK_C_RTL
    free_ctx = (client_ctx) malloc ( sizeof(struct client_context) * 10 );
    UNLOCK_C_RTL
    for ( i = 0; i < 10; i++ ) {
	free_ctx[i].next = &free_ctx[i+1];
	free_ctx[i].cond_valid = 0;		/* cond var not initted */
    }
    free_ctx[9].next = (client_ctx) 0;
#ifndef PTHREAD_USE_D4
    /*
     * See if kernel threads available so we can use 'W' form of systems
     * services.
     */
    kernel_threads = os_kernel_threads();
#endif
}
/***************************************************************************/
/* Allocate and initialized client context.
 */
static client_ctx allocate_context()
{
    int status, i;
    client_ctx ctx;
    pthread_mutex_lock ( &client_ctl );
    ctx = free_ctx;
    if ( !ctx ) {
	/*
	 * Nothing on free list, expand it.
	 */
	LOCK_C_RTL
	free_ctx = (client_ctx) malloc ( sizeof(struct client_context) * 10 );
	UNLOCK_C_RTL
	for ( i = 0; i < 10; i++ ) {
	    free_ctx[i].next = &free_ctx[i+1];
	    free_ctx[i].cond_valid = 0;		/* cond var not initted */
	}
	free_ctx[9].next = (client_ctx) 0;
	
    	ctx = free_ctx;
    }
    free_ctx = ctx->next;
    pthread_mutex_unlock ( &client_ctl );
    /*
     * initialize for use.
     */
    if ( !ctx->cond_valid ) {
	ctx->cond_valid = 1;
	INITIALIZE_CONDITION ( &ctx->io_done );
	SET_COND_NAME ( &ctx->io_done, "OSU tclient io_done" )
    }
    ctx->next = (client_ctx) 0;
    ctx->chan = 0;
    ctx->time_limit = 0;
    ctx->completed_flags = ctx->pending_flags = 0;
    return ctx;
}
/***************************************************************************/
/* Convert host name to IP addresses.  Ctx argument is assumed to have a
 * valid channel assigned.
 */
static int translate_hostname ( client_ctx ctx, char *name, 
	unsigned long *list, int list_size, int *count )
{
    int i, j, length, status;
    struct { unsigned short length, type; void *ptr; } p1, p2, p4;
    unsigned char subfunc[4];
    /*
     * Scan name and see if name is 'dot' format numeric.
     */
    for ( i = 0; name[i]; i++ ) if ( name[i] != '.' &&
	(name[i] < '0' || name[i] > '9') ) break;
    if ( !name[i] ) {
	unsigned long addr[4];
	addr[0] = addr[1] = addr[2] = addr[3] = 0;
	for ( i = j = 0; name[i]; i++ ) {
	    if ( name[i] == '.' ) {
		j++;  if ( j >= 4 ) break;
	    } else {
		addr[j] = addr[j]*10 + (name[i]-'0');
	    }
	}
	if ( !name[i] )	{
	    *count = 1;
	    list[0] = addr[0] | (addr[1]<<8) | (addr[2]<<16) | (addr[3]<<24);
	    return 1;
	}
    }
    /*
     * Build arguments for QIO interface.
     */
    subfunc[0] = INETACP_FUNC$C_GETHOSTBYNAME;
    subfunc[1] = INETACP$C_TRANS;
    subfunc[2] = subfunc[3] = 0;
    p1.type = 0; p1.length = 4; p1.ptr = subfunc;
    p2.type = 0; p2.length = tu_strlen(name); p2.ptr = name;
    if ( p2.length <= 0 ) return 20;
    p4.type = 0; p4.length = list_size*4; p4.ptr = list;

    length = 0;
    status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, IO$_ACPCONTROL, 
	&ctx->iosb, &p1, &p2, &length, &p4, 0, 0 );
    if ( status&1 ) *count = length / sizeof(unsigned long);
    else count = 0;
    return status;
}
/***************************************************************************/
static int create_socket ( int attr_count, char **attributes, int port,
	int backlog, pthread_cond_t *cond, int *chan, char *err_msg )
{
    int status, i, j, addr_count;
    int flag;
    struct sockname { long length; void *va; int *retadr; long list_end;
	struct SOCKADDRIN a; };
    struct sockname local;
    struct sockchar { short protocol; char ptype, domain; } sockchar;
    struct { short length, type; void *ptr; short len1, code1;
	void *addr1; int linger_onoff, linger_time; } sockopt;
    struct  itlst { short length, code;
	  union { struct itlst *lst; int *val; } addr; } olst[12];
    struct { unsigned short sts, count, d1, d2; } iosb;

    status = SYS$ASSIGN ( &ucx_device, chan, 0, 0, 0 );
    if ( (status&1) == 0 ) {
	/*
	 * ASSIGN failure, deallocate and abort.
	 */
	tu_strcpy ( err_msg, "Failed to assign to UCX device" );
	return status;
    }
    /*
     * Set up structures needed for socket creation with default values.
     */
    sockchar.protocol = UCX$C_TCP;
    sockchar.ptype =  UCX$C_STREAM;
    sockchar.domain = UCX$C_AF_INET;

    local.length = sizeof(struct SOCKADDRIN);
    local.va = &local.a;
    local.retadr = 0;
    local.list_end = 0;
    local.a.SIN$W_FAMILY = UCX$C_AF_INET;
    local.a.SIN$W_PORT = HTONS(port);
    local.a.SIN$L_ADDR = 0;

    sockopt.length = 8; sockopt.type = UCX$C_SOCKOPT;
    sockopt.ptr = &sockopt.len1;
    sockopt.len1 = 8;
    sockopt.code1 = UCX$C_LINGER;
    sockopt.addr1 = &sockopt.linger_onoff;
    sockopt.linger_onoff = 1;
    sockopt.linger_time = 40;

    /*
     * Scan attributes and make modifier list for socket create and bind.
     */
    olst[0].code = UCX$C_SOCKOPT; olst[0].length = 0;
    olst[0].addr.lst = &olst[1];
    for ( j = i = 0; i < attr_count; i++ ) {
	if ( tu_strncmp ( attributes[i], "reuseaddr", 10 ) == 0 ) {
	    j++;
	    olst[0].length += sizeof(struct itlst);
	    olst[j].length = 4; olst[j].code = UCX$C_REUSEADDR;
	    olst[j].addr.val = &flag; flag = 1;

	} else if ( tu_strncmp ( attributes[i], "local_host=", 11 ) == 0 ) {
	} else if ( tu_strncmp ( attributes[i], "local_port=", 11 ) == 0 ) {
	    LOCK_C_RTL
	    local.a.SIN$W_PORT = HTONS(atoi(&attributes[i][11]));
	    UNLOCK_C_RTL
	} else if ( tu_strncmp ( attributes[i], "keepalive", 11 ) == 0 ) {
	    j++;
	    olst[0].length += sizeof(struct itlst);
	    olst[j].length = 4; olst[j].code = UCX$C_KEEPALIVE;
	    olst[j].addr.val = &flag; flag = 1;
	}
    }

    status = QIO_AND_WAIT ( cond, *chan, IO$_SETMODE, &iosb,
	&sockchar, 0, 0, 0, 0, 0 );

    if ( (status&1) && j > 0 ) {
	status = QIO_AND_WAIT ( cond, *chan, IO$_SETMODE, &iosb,
		0, 0, 0, 0, olst, 0 );
    }

    if ( (status&1) ) {
	status = QIO_AND_WAIT ( cond, *chan, IO$_SETMODE, &iosb,
		0, 0, &local, backlog, &sockopt, 0 );
    }
    return status;
}
/***************************************************************************/
int tc_open_tcp_socket ( 
    char *target_host, 			/* Remote host to connect to */
    int target_port, 			/* Report port, (local port 'ANY') */
    int attr_count,			/* Optional attributes */
    char **attributes, 			/* Attribute specifcations, see below */
    tc_socket *cnx, 			/* Handle to connection */
    char err_msg[256] )			/* Description of error code, if any */
{
    int status, i, addr_count;
    client_ctx ctx, ctx_list;
    unsigned long addr_list[128];
    int flag;
    struct sockname { long length; void *va; int *retadr; long list_end;
	struct SOCKADDRIN a; };
    struct sockname remote;
    /*
     * Do one-time setup.
     */
    pthread_once ( &setup, module_init );
    /*
     * Allocate context and assign channel.  Bind into per-thread context to
     * ensure cleanup on thread exit.
     */
    ctx = allocate_context();
    if ( !ctx ) {
	tu_strnzcpy ( err_msg, "Failed to allocate context", 255 );
	return 0;
    }
    status = create_socket ( attr_count, attributes, 0, 0,
		&ctx->io_done, &ctx->chan, err_msg );
    if ( (status&1) == 0 ) {
	pthread_mutex_lock ( &client_ctl );
	ctx->next = free_ctx;
	free_ctx = ctx;
	pthread_mutex_unlock ( &client_ctl );
	return status;
    } else {
	/*
	 * Link block into per-thread list to ensure rundown.
	 */
	GET_SPECIFIC(client_key, ctx_list);
	ctx->next = ctx_list;
	pthread_setspecific ( client_key, ctx );
    }
    /*
     * Convert target host name to address.
     */
    status = translate_hostname(ctx, target_host, addr_list, 12, &addr_count);
    if ( (status&1) == 0 ) {
	tu_strcpy ( err_msg, "Host lookup failure" );
	SYS$DASSGN ( ctx->chan );
	pthread_setspecific ( client_key, ctx->next );
	pthread_mutex_lock ( &client_ctl );
	ctx->next = free_ctx;
	free_ctx = ctx;
	pthread_mutex_unlock ( &client_ctl );
	return status;
    }
    /* printf("/tclient/ address count: %d, addr[0]: %x, port: %d\n", addr_count, 
	addr_list[0], target_port ); */
    /*
     * Attempt connection to all addresses in list.
     */
     if ( (status&1) == 1 ) for ( i = 0; i < addr_count; i++ ) {
	/*
	 * Attempt to connect to next address.
	 */
	remote.length = sizeof(struct SOCKADDRIN);
	remote.va = &remote.a;
	remote.retadr = 0;
        remote.list_end = 0;
	remote.a.SIN$W_FAMILY = UCX$C_AF_INET;
	remote.a.SIN$W_PORT = HTONS(target_port);
	remote.a.SIN$L_ADDR = addr_list[i];

	status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, IO$_ACCESS,
		&ctx->iosb, 0, 0, &remote, 0, 0, 0 );
	if ( (status&1) == 1 ) break;	/* Success! quit trying */
	tu_strcpy ( err_msg, "Connect failure" );
     } else {
        tu_strnzcpy ( err_msg, "Unimplemented", 255 );
     }
    /*
     * If connect failed, unwind from per-thread context to keep a brain-dead
     * thread that retries failed connects from consuming resources.
     */
    if ( (status&1) == 0 ) {
	pthread_setspecific ( client_key, ctx->next );
	SYS$DASSGN ( ctx->chan );
	pthread_mutex_lock ( &client_ctl );
	ctx->next = free_ctx;
	free_ctx = ctx;
	pthread_mutex_unlock ( &client_ctl );
    }
    *cnx = (void *) ctx;
    return status;
}
/****************************************************************************/
/*
 * Close connection.  Channel is deassigned regardless of deaccess status.
 */
int tc_close ( tc_socket cnx )		/* Synchronous close */
{
    int status;
    client_ctx ctx, ctx_list;
    /*
     * Cleanly close connection.
     */
    ctx = (client_ctx) cnx;
    status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, IO$_DEACCESS, &ctx->iosb,
	0, 0, 0, 0, 0, 0 );
    SYS$DASSGN ( ctx->chan );
    /*
     * Remove contxt block from per-thread list of open connections and 
     * place on free list.
     */
    tc_detach ( cnx );
    pthread_mutex_lock ( &client_ctl );
    ctx->next = free_ctx;
    free_ctx = ctx;
    pthread_mutex_unlock ( &client_ctl );
    return status;
}

int tc_attach ( tc_socket cnx ) {
    int status;
    client_ctx ctx, ctx_list;
    /*
     * Link context into per-thread chain, assum connection was detached.
     */
    ctx = (client_ctx) cnx;
    GET_SPECIFIC(client_key, ctx->next );
    pthread_setspecific ( client_key, ctx );
    return 1;
}


int tc_detach ( tc_socket cnx ) {
    int status;
    client_ctx ctx, ctx_list;
    /*
     * Unlink context block.
     */
    status = 1;
    ctx = (client_ctx) cnx;
    GET_SPECIFIC(client_key, ctx_list);
    if ( ctx_list == ctx ) {	/* head of list */
	ctx_list = ctx->next;
	pthread_setspecific ( client_key, ctx_list );
    } else {
	while ( ctx_list->next && (ctx_list->next != ctx) ) {
	    ctx_list = ctx_list->next;
	}
	if ( !ctx_list->next ) {
		/* Error, context block not on list */
	    status = 0;
	} else {
	    /* Remove from chain */
	    ctx_list->next = ctx->next;
	}
    }
    return status;
}
/*
 * Time limit only affects reads and writes.
 */
int tc_set_time_limit ( tc_socket cnx,  int limit )
{
    client_ctx ctx;
    ctx = cnx;
    if ( limit != 0 ) {
	/*
	 * Compute timeout time.
	 */
	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 {
	    return 0;
	}
    } else ctx->time_limit = 0;		/* no expiration */
    return 1;
}
int tc_write ( tc_socket cnx, void *buffer, int bufsize, int *written )
{
    int status;
    client_ctx ctx;

    ctx = (client_ctx) cnx;
    if ( ctx->time_limit ) {
	if ( ctx->time_limit == 2 ) return 556;		/* timed out */
	pthread_mutex_lock ( &client_io );
	status = SYS$QIO ( 1, ctx->chan, IO$_WRITEVBLK, &ctx->iosb,
	    pthread_cond_signal_int_np, &ctx->io_done, buffer, bufsize,
		0, 0, 0, 0 );
	status = synch_client ( status, ctx );
    } else {
        status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, IO$_WRITEVBLK, 
		&ctx->iosb, buffer, bufsize, 0, 0, 0, 0 );
    }
    if ( (status&1) == 0 ) *written = 0;
    else *written = ctx->iosb.count;

    return status;
}

int tc_read ( tc_socket cnx, void *buffer, int bufsize, int *written )
{
    int status;
    client_ctx ctx;
    /*
     * Check if time_limit to reads.
     */
    ctx = (client_ctx) cnx;
    if ( ctx->time_limit ) {
	if ( ctx->time_limit == 2 ) return 556;		/* timed out */
	pthread_mutex_lock ( &client_io );
	status = SYS$QIO ( 1, ctx->chan, IO$_READVBLK, &ctx->iosb,
	    pthread_cond_signal_int_np, &ctx->io_done, buffer, bufsize,
		0, 0, 0, 0 );
	status = synch_client ( status, ctx );     } else {
        status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, IO$_READVBLK, 
		&ctx->iosb, buffer, bufsize, 0, 0, 0, 0 );
    }
    if ( (status&1) == 0 ) *written = 0;
    else *written = ctx->iosb.count;
    return status;
}
/***************************************************************************/
static void asynch_write ( client_ctx ctx )
{
    int status;
    status = SYS$QIO ( 2, ctx->chan, IO$_WRITEVBLK|IO$M_NOWAIT, &ctx->wiosb,
	pthread_cond_signal_int_np, &ctx->io_done, 
	ctx->wbuf, ctx->wbuf_size, 0, 0, 0, 0 );
}
static void asynch_read ( client_ctx ctx )
{
    int status;
    status = SYS$QIO ( 2, ctx->chan, IO$_READVBLK|IO$M_NOWAIT, &ctx->riosb,
	pthread_cond_signal_int_np, &ctx->io_done, 
	ctx->rbuf, ctx->rbuf_size, 0, 0, 0, 0 );
}
/***************************************************************************/
/* Asynch (full duplex) interface.
 */
int tc_start_io ( tc_socket cnx, 
	int direction, 			/* 0-write, 1-read */
	void *buffer, int bufsize )
{
    int status, func, mask, *count;
    client_ctx ctx;
    ctx = (client_ctx) cnx;

    if ( direction == 0 ) {
	mask = 1;
	if ( ctx->pending_flags & mask ) return SS$_DEVACTIVE;
	func = IO$_WRITEVBLK | IO$M_NOWAIT;
	ctx->wbuf_size = bufsize;
	ctx->wbuf = buffer;
	count = &ctx->bytes_written;
    } else {
	mask = 2;
	if ( ctx->pending_flags & mask ) return SS$_DEVACTIVE;
	func = IO$_READVBLK | IO$M_NOWAIT;
	ctx->rbuf_size = bufsize;
	ctx->rbuf = buffer;
	count = &ctx->bytes_read;
    }
    /*
     * Try operation, nonblocking.
     */
    status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, func, &ctx->iosb,
	buffer, bufsize, 0, 0, 0, 0 );
    if ( status == SS$_SUSPENDED ) {
	/*
	 * Set attention AST, clear IOSB now since QIO that would normally
	 * clear it doesn't get called until later.
	 */
	if ( direction == 0 ) {
	    func = IO$_SETMODE|IO$M_WRTATTN;
	    ctx->wiosb.status = 0;
	} else {
	    func = IO$_SETMODE|IO$M_READATTN;
	    ctx->riosb.status = 0;
	}
	status = QIO_AND_WAIT ( &ctx->io_done, ctx->chan, func, &ctx->iosb,
		(direction==0)?asynch_write:asynch_read, ctx, 0, 0, 0, 0 );
	if ( status&1 ) ctx->pending_flags |= mask;
    } else {
	/*
	 * I/O completed, update context.
	 */
	*count = ctx->iosb.count;
	pthread_mutex_lock ( &client_io );
	ctx->completed_flags |= mask;
	if ( (status&1) == 0 ) ctx->completed_flags |= 4;
	pthread_mutex_unlock ( &client_io );
    }
    return 1;
}
int tc_io_stall ( tc_socket cnx, 
	int *flags, 			/*bits: 0-write, 1-read, 2-err, 3-attn*/
	int *bytes_read, int *bytes_written )
{
    int status;
    client_ctx ctx;
    ctx = (client_ctx) cnx;

    pthread_mutex_lock ( &client_io );
    while ( ctx->completed_flags == 0 ) {
        /*
         * First mark any completed asynchronous I/O.
         */
        if ( ctx->pending_flags ) {
	    if ( (ctx->pending_flags&1) && (ctx->wiosb.status != 0) ) {
	        if ( ctx->wiosb.status&1 == 0 ) ctx->completed_flags |= 4;
	        ctx->completed_flags |= 1;
	        ctx->pending_flags ^= 1;
	        ctx->bytes_written = ctx->wiosb.count;
	    }
	    if ( (ctx->pending_flags&2) && (ctx->riosb.status != 0) ) {
	        if ( ctx->riosb.status&1 == 0 ) ctx->completed_flags |= 4;
	        ctx->completed_flags |= 2;
	        ctx->pending_flags ^= 2;
	        ctx->bytes_read = ctx->riosb.count;
	    }
        }
	/*
	 * Issue wait if nothing completed.
	 */
	if ( ctx->completed_flags == 0 ) {
	    status = pthread_cond_wait ( &ctx->io_done, &client_io );
	}
    }
    /*
     * return completed info, and clear flags.
     */
    *flags = ctx->completed_flags;

    if ( ctx->completed_flags&1 ) *bytes_written = ctx->bytes_written;
    if ( ctx->completed_flags&2 ) *bytes_read = ctx->bytes_read;
    ctx->completed_flags = 0;

    pthread_mutex_unlock ( &client_io );
	
    return 1;
}

int tc_attention ( tc_socket cnx )
{
    client_ctx ctx;
    ctx = (client_ctx) cnx;
    pthread_mutex_lock ( &client_io );
    if ( (ctx->completed_flags & 8) == 0 ) {
	ctx->completed_flags |= 8;
	pthread_cond_signal ( &ctx->io_done );
    }
    pthread_mutex_unlock ( &client_io );
    return 1;
}
/*****************************************************************************/
/* Listener operations:
 */
struct listen_context {
    struct client_context *next;
    int chan;			/* VMS I/O channel for listen socket */
    struct { unsigned short status, count; 
	unsigned long devsts; } iosb;
    pthread_cond_t io_done;
    pthread_mutex_t lock;
    pthread_cond_t idle;
    int active;
    int acceptors_pending;
};
typedef struct listen_context *listen_ctx;

int tc_create_tcp_listener ( 
	int port, 		/* Port number to listen on */
	int backlog, 		/* Listen queue depth */
	int attr_count,		/* number of optional attributes */
	char **attributes, 	/* Options for listen */
	tc_listener *lstn, 	/* Result */
	char err_msg[256] )	/* Error message if return status != 1 */
{
    int status, i;
    listen_ctx ctx;
    struct sockname { long length; void *va; int *retadr; long list_end;
	struct SOCKADDRIN a; };
    struct sockname local, remote;
    struct sockchar { short protocol; char ptype, domain; } sockchar;
    struct { short length, type; void *ptr; short len1, code1;
	void *addr1; int linger_onoff, linger_time; } sockopt;
    /*
     * Do one-time setup.
     */
    pthread_once ( &setup, module_init );
    /*
     * Allocate and initialize structure.
     */
    LOCK_C_RTL
    ctx = (listen_ctx) malloc ( sizeof(struct listen_context) );
    UNLOCK_C_RTL
    ctx->chan = 0;
    INITIALIZE_CONDITION ( &ctx->io_done );
    SET_COND_NAME ( &ctx->io_done, "OSU tclient listen io_done" )
    INITIALIZE_MUTEX ( &ctx->lock );
    SET_MUTEX_NAME ( &ctx->lock, "OSU tclient listener" )
    INITIALIZE_CONDITION ( &ctx->idle );
    SET_COND_NAME ( &ctx->idle, "OSU tclient idle" )
    ctx->active = 0;
    ctx->acceptors_pending = 0;
    /*
     * Create socket.
     */
    status = create_socket ( attr_count, attributes, port, backlog,
	&ctx->io_done, &ctx->chan, err_msg );

    *lstn = (tc_listener) ctx;
    if ( (status&1) == 0 ) tc_destroy_listener ( lstn );
    return status;
}

int tc_destroy_listener ( tc_listener lstn )
{
    listen_ctx ctx;
    int status;

    ctx = (listen_ctx) lstn;
    status = SYS$DASSGN ( ctx->chan );
    pthread_cond_destroy ( &ctx->idle );
    pthread_mutex_destroy ( &ctx->lock );
    pthread_cond_destroy ( &ctx->io_done );
    LOCK_C_RTL
    free ( ctx );
    UNLOCK_C_RTL
    return 1;
}

int tc_accept ( tc_listener lstn, tc_socket *cnx, int *remote_port, 
	unsigned char *remote_address, int ra_size, char *err_msg )
{
    struct listen_context *lctx;
    client_ctx ctx, ctx_list;
    unsigned char *ra;
    int status;
    struct sockname { long length; void *va; int *retadr; long list_end;
	struct SOCKADDRIN a; };
    struct sockname remote;
    struct { long length; struct SOCKADDRIN *va;
		int *retadr; long list_end; } sockname;

    lctx = (listen_ctx) lstn;
    /*
     * Allocate context and assign channel.
     */
    ctx = allocate_context();
    if ( !ctx ) {
	tu_strnzcpy ( err_msg, "Failed to allocate context", 255 );
	return 0;
    }
    status = SYS$ASSIGN ( &ucx_device, &ctx->chan, 0, 0, 0 );
    if ( (status&1) == 0 ) {
	/*
	 * ASSIGN failure, deallocate and abort.
	 */
	tu_strcpy ( err_msg, "Failed to assign to UCX device" );
	pthread_mutex_lock ( &client_ctl );
	ctx->next = free_ctx;
	free_ctx = ctx;
	pthread_mutex_unlock ( &client_ctl );
	return status;
    } else {
	/*
	 * Link block into per-thread list to ensure rundown.
	 */
	GET_SPECIFIC(client_key, ctx_list);
	ctx->next = ctx_list;
	pthread_setspecific ( client_key, ctx );
    }
    /*
     * Get exclusive access since lctx->iosb is shared.
     */
    pthread_mutex_lock ( &lctx->lock );
    while ( lctx->active != 0 ) {
	lctx->acceptors_pending++;
	if ( 0 != pthread_cond_wait ( &lctx->idle, &lctx->lock ) ) {
    	    pthread_mutex_unlock ( &lctx->lock );
	    return 0;
	}
	--lctx->acceptors_pending;
    }
    lctx->active = 1;
    pthread_mutex_unlock ( &lctx->lock );
    /*
     * Wait for connection.
     */
    sockname.length = sizeof(struct SOCKADDRIN);
    sockname.va = &remote.a;
    sockname.retadr = (int *) &remote.length;
    sockname.list_end = 0;
    remote.a.SIN$W_FAMILY = UCX$C_AF_INET;
    remote.a.SIN$W_PORT = 0;	/* any port */
    remote.a.SIN$L_ADDR = 0;

    status = QIO_AND_WAIT ( &lctx->io_done, lctx->chan, IO$_ACCESS|IO$M_ACCEPT, 
	&lctx->iosb, 0, 0, &sockname, &ctx->chan, 0, 0 );
    /*
     * Release lock so other acceptors may proceed.
     */
    pthread_mutex_lock ( &lctx->lock );
    lctx->active = 0;
    if ( lctx->acceptors_pending > 0 ) pthread_cond_signal ( &lctx->idle );
    pthread_mutex_unlock ( &lctx->lock );
    /*
     * Return connection and client into to caller.
     */
    ra = (unsigned char *) &remote.a.SIN$L_ADDR;
    remote_address[0] = ra[0];
    remote_address[1] = ra[1];
    remote_address[2] = ra[2];
    remote_address[3] = ra[3];
    *remote_port = HTONS ( remote.a.SIN$W_PORT );

    *cnx = (tc_socket) ctx;
    return 1;
}
