/* Multithreaded NNTP server for ANU_NEWS V6.x and VMS UCX 1.3 and UCX 2.0,
 *   (runs also under MULTINET).
 * Written by Steve Bour (jsbour@ualr.edu).
 * taken from the NNTP server for CMU-TEK TCP/IP V6.5 by Keith Halewood
 * The 'getremhost' routine has been stolen from the original single-thread
 * server interface.
 *
 * Changes made CMU-TEK version -> VMS UCX version (by Steve Bour)
 * 1. Changed QIOs from CMU-TEK to UCX (minimal hacking required, some
 *    required no changes at all).
 * 2. Added in NNTP inactivity timeout functionality missing from UCX
 *    by use of an "inactivity timer tick", forcing a SYS$CANCEL on any
 *    thread that is inactive (no reads or writes) for INACTIVITY_TICK_MAX
 *    ticks.  Currently the ticks are 900 seconds (15 minutes), and the
 *    SYS$CANCEL is executed on any channel that's inactive for 10 ticks
 *    (~ 2.5 hours).  It would have been better, perhaps, to have seperate
 *    AST timers on each active channel, but my VMS system call experience
 *    is weak (read: too lazy to figure it out :-).  The UCX driver doesn't
 *    appear to support the CMU-TEK style timeout parameter in the QIOs, thus
 *    this kludge.
 * 3. Added code to log low-level driver activity to a file; some of this is
 *    commented out and when I'm feeling really good about *my* part of the
 *    code (if that's possible :-) it'll probably all be ripped out.
 * 4. I didn't implement RFC931 authentication -- if anyone needs it,
 *    harass me and I might put it in.
 * 5. DCL command to start up server (executed from the News Manager account):
 *
 *               $ run/detach -
 *                     /privileges=(NOSAME,SYSPRV,PRMMBX,NETMBX,SYSLCK)-
 *                     /output = NLA0: -
 *                     /error = NLA0: -
 *                     /input = NLA0: -
 *                     /process_name = "UCX/NNTP server" -
 *                     NEWS_MANAGER_DEV:[NEWS_DIST]NNTP_TCPUCXM
 *
 * 6. Many thanks to Keith Halewood for the base code.
 *    
 * Changes V1.0 -> V1.1 (CMU-TEK version, by Keith Halewood)
 * 1. Server exits when the passive listen times out and there are no
 *    active connections. Previously, the server exited immediately on
 *    the termination of the last active connection.
 * 2. As a consequence of the above, a bug preventing threads from
 *    being freed correctly has now been fixed.
 * 3. Rearranged manage_open so that a new passive listen is queued
 *    before the new thread is initialised to minimise the chances of
 *    of missing closely-spaced incoming connect requests.
 * 4. getremhost() has been modified to perform RFC931 authentication
 *    checking. Provided that a daemon is running on the client host,
 *    NNTP can now log the user as well as the host (a la DECnet).
 *    It's the result of 10 mins work so the code isn't pretty.
 * 5. Modified write_thread so that multiple lines are blocked into one
 *    QIO operation saving lots of CPU and increasing I/O throughput.
 * 6. log_to_file() is now called (see nntp_server source code).
 * 7. Added call to already_running which uses the channel count on
 *    a permanent mailbox to determine whether a nntp_server is already
 *    running. It is currently programmed to terminate if one is
 *    already active.
 * 8. Fixed read buffer overflow into read residue for each thread. -1
 *    should be +1 in array declarations.
 * 8. Manage_read_completion more efficient.
 *
 *  MODIFICATION HISTORY:
 *      V1.1-1   9-Aug-1993     Charles Bailey  bailey@genetics.upenn.edu
 *      - added close_net() as part of revising News memory management
 *
 *      V1.2    14-Sep-1993     Mark Martinec   mark.martinec@ijs.si
 *      - added checking of status returned from system and RMS calls
 *
 *      V1.3    19-Dec-1993     Mark Martinec   mark.martinec@ijs.si
 *      - added routine purge_queue() as part of a thread rundown,
 *        to avoid accessing an already closed channel during
 *        abnormal thread closure (which used to crash server
 *        after status checking has been tightened up with V1.2).
 *
 *      V1.4    17-Jan-1994	Mark Martinec   mark.martinec@ijs.si
 *      - cleanup/rewrite the buffer size administration and fix the bug
 *        where very long lines caused data corruption;
 *      - replace all strcpy/strcat/strlen C-style string operations
 *        with more efficient memcpy/memmove and explicit string length
 *        (which was already in the data structure, but mostly unused).
 *        This avoids numerous counting and recounting of the already
 *        known string sizes;  This module is now '\0'-clean, except
 *        for the interface routine 'write_net'.
 *      - write_thread: more tightly packed net transfer buffers, composed of
 *        partial line_blocks if necessary;  this also avoids infinite loop
 *        when writing lines longer than BUFSIZE;
 *      - additional checks on buffer sizes (e.g. in read_net)
 *      - introduce 'sessionlog' (conditionally compiled) - writes net dialog
 *        to a file;
 *      - minor optimization: when line_block.line_len is 0
 *        (empty lines in text) avoid mallocating empty string and leave
 *        line_block.line as NULL.
 *
 *      V1.5    16-Mar-1994	Mark Martinec   mark.martinec@ijs.si
 *	- ignore SS$_ENDOFFILE in ast_signal_read_complete()
 * 	- make sure it runs under MULTINET (in UCX compatibility mode);
 *	  (by now I have much confidence in TCPUCXM (after weeding out
 *	  many errors) that I prefer to run this server even after we replaced
 *	  UCX with MULTINET)
 *
 *      V1.6     6-May-1994 	Mark Martinec   mark.martinec@ijs.si
 *	- define macro EXPECTED_IO_ERROR() and unify checking for
 *	  expected i/o errors.
 *	- write out a list of required privileges if $crembx fails
 *	- make it compile (under MULTINET) without UCX$INETDEF.H
 *
 *      V1.7    11-May-1994 	Mark Martinec   mark.martinec@ijs.si
 *	- check status after fopen(drvrlogfile)
 *	- drvrlogfile may now be a logical name
 *	- write server version to log file when started
 *	  (this is also a test to check if we can write to a log file)
 *      V6.1b9  23-May-1994 	Alan Greig  ccdarg@zippy.dct.ac.uk
 *	- change #include errno to #include <errno.h> under DECC, because
 *	  DECC chokes on the former under some (unclear) circumstances.
**	V6.1b9	17-Aug-1994     Mark Martinec   mark.martinec@ijs.si
**	  - code cleanup to make it compile under gcc 2.6.0 with full
**	    warnings reporting turned on - with no or very few harmless warnings
**	V6.1b9	17-Sep-1994     Mark Martinec   mark.martinec@ijs.si
**	  - code cleanup to preserve the read-only nature of string literals
**	    (strategically placed 'const' attribute to string parameters)
**	V6.1b9	17-Feb-1995	Mark Martinec   mark.martinec@ijs.si
**	  - rearanged getremhost() to support other TCP/IP transports
**/

#ifdef vaxc
#module NNTP_TCPUCXM "V6.2"
#endif

#define version      "V6.2, 16-May-1997"

#define _NNTP_SERVER_DRIVER_C
#define _NNTP_TCPUCXM_C
#define module_name "NNTP_TCPUCXM"

#include "nntpinclude.h"

#ifndef UCX
#define INET$C_TCP		IPPROTO_TCP
#define INET$C_AF_INET		AF_INET
#define INET$C_INADDR_ANY	INADDR_ANY
#define INET_PROTYP$C_STREAM	SOCK_STREAM
#endif
  
#define NNTP_PORT   119    /* Should be 119, use 8888 for testing. */
#define MAXTHREADS   16    /* Maximum number of services - Make sure
                            * process quotas are capable of handling 
                            * the number chosen.
                            */
#define AUTHTIMEOUT  30    /* Authentication timeout.
                            * Some Un*x auth servers are very slow, but
                            * none should be slower than this.
                            */
#define BUFSIZE    1024    /* Constants for QIO buffer lengths and
                            */
#define LINESIZE   2048    /* the read line assembly buffer, > BUFSIZE
                            */
#define NNTP_STRLEN 2048   /* should be less or equal to the
                              parameter string_max_size in read_net()
                           */

#define EXPECTED_IO_ERROR(s) \
	((s) == SS$_CONNECFAIL || \
	 (s) == SS$_LINKABORT  || \
	 (s) == SS$_LINKDISCON || \
	 (s) == SS$_UNREACHABLE|| \
	 (s) == SS$_REJECT  || \
	 (s) == SS$_SHUT    || \
	 (s) == SS$_CANCEL  || \
	 (s) == SS$_TIMEOUT || \
	 (s) == SS$_ENDOFFILE)

/* Event numbers used by AST completion routines. The are placed,
 * along with a thread number into the attention queue.
 */

#define EV_NULL                     0
#define EV_OPEN_COMPLETED           1
#define EV_OPEN_ABORTED             2
#define EV_READ_COMPLETED           3
#define EV_READ_ABORTED             4
#define EV_WRITE_COMPLETED          5
#define EV_WRITE_ABORTED            6

/* Is the thread in use (healthy) or unassigned. Unassigned is also
 * a state used during shutdown to control the discarding of entries
 * in the I/O queues.
 */

#define ACT_UNASSIGNED              0
#define ACT_HEALTHY                 1

/* 'Next function' return codes, determines whether the 'next function' call
 * for a thread is activated on receipt of a line (command) or on receipt of
 * the file terminator (a single full-stop) or whether the thread is to be
 * closed down.
 */

#define TYP_NO_INPUT                0
#define TYP_CMD_INPUT               1
#define TYP_FILE_INPUT              2

/* Definitions for the passive open (listen) on NNTP_PORT
 */

unsigned short listen_channel, clone_channel, listen_iosb[4], new_iosb[4];
int active_count = 0;
int active_listener = 0;
int status;

/* I/O queue definition.
 */

struct line_block
  {
    struct line_block *next;
    char *line;    /* this is NOT a '\0' terminated string ! */
    int line_len;  /* actual line length */
  };

/* Thread structures. [0] isn't used.
 */

/* member alignment is required to assure atomic volatile access */
#ifdef __DECC
#pragma member_alignment save
#pragma member_alignment
#endif
static volatile struct thread_struct
  {
    struct line_block *read_head, *read_tail;  /* each block one line, no \n  */
    struct line_block *write_head, *write_tail;/* stream, blocks contain \r\n */
    void (*next_function)(int);
    int active;
    int inactivity_ticks;
    int read_buffer_len;  /* actual read_buffer length */
    int read_residue_len; /* actual read_residue length */
    int write_in_progress;
    int next_type;
    unsigned short write_iosb[4], read_iosb[4];
    unsigned short channel;
    unsigned short dummy1;  /* alignment padding */
    char write_buffer[BUFSIZE];  /* this is NOT a '\0' terminated string ! */
    char read_buffer[BUFSIZE];   /* this is NOT a '\0' terminated string ! */
    char read_residue[LINESIZE]; /* this is NOT a '\0' terminated string ! */
    char remuser[32];
    char remhost[255];
    char dummy2;   /* alignment padding */
  } threads[MAXTHREADS+1];
#ifdef __DECC
#pragma member_alignment restore
#endif

$DESCRIPTOR_CONST(ipdsc,"UCX$DEVICE");
/* $DESCRIPTOR_CONST(ipdsc,"BG:"); */
/* I couldn't seem to get UCX$DEVICE to work; not only that, but BG: is
   what *DIGITAL* uses in their example code.  Hmmmm.............. jsb */
/* UCX$DEVICE works for me as supposed.  Could have been a problem
   with data corruption, that is fixed by now.   mark.martinec@ijs.si  */

/* NNTP inactivity timeout definitions
 */

#define INACTIVITY_TICK_LENGTH 900	  /* how many seconds between ticks */
#define INACTIVITY_TICK_MAX 10		  /* how many ticks before abort */
					  /* 900 * 10 = 9000, or 2.5 hours */
					  /* (minus up to 15 minutes) */
static volatile int
  inactivity_tick_trap_flag = 0;          /* gets set to 1 when tick occurs */

/* code for logging server driver events
 */
FILE *drvrlog = NULL;

/* driver log file name (possibly a logical name) */
char drvrlogfile[] = "NNTP_SERVER_DRVR";

/* driver log file default name */
char drvrlogfile_def[] = "dna=NEWS_MANAGER_DEV:[NEWS_LOG].LOG";

/* if 'sessionlogfile' is defined, the code for writing session dialog will be
 * compiled into code, otherwise not.  It is intended for testing only.
 */
/*
#define sessionlogfile	"NEWS_MANAGER_DEV:[NEWS_LOG]NNTP_SERVER_SESSION.LOG"
*/

#ifdef sessionlogfile 
FILE *sessionlog;
#endif

/* Definition for time of day routines
 */

time_t currtime;

/* stuff for listener initialization
 */

struct mysockaddr 
  {
    short inet_family;
    short inet_port;
    unsigned char adrs[4];
    char blkb[8];
  };

struct mysockaddr2
  {
    short   inet_family;
    short   inet_port;
    int     adrs;
    char    blkb[8];
  };

struct itlst
  {
    int lgth;
    struct mysockaddr2 *hst;
  };

struct itlst_3
  {
    int lgth;
    struct mysockaddr *hst;
    int *retlth;
  };

struct mysockaddr remote_hostaddr;

int retlen;
int beenhere=0;
struct mysockaddr2 local_host;
struct itlst lhst_adrs;
struct itlst_3 rhst_adrs;
short sck_parm[2] = { INET$C_TCP, INET_PROTYP$C_STREAM };

/******************************************************************
 * Exteral routines from NNTP_SERVER                              *
 ******************************************************************/

int server_init();
void server_init_unit();
void log_to_file();
int server_condition_handler();

/*
 * The following routines should retain parameter volatile attributes.
 * Actually they do not, but just serve as a reminder of a possible problem.
 */
#define strcpyv(s1,s2)		strcpy(s1,s2)
#define memcpyv(s1,s2,n)	memcpy(s1,s2,n)
#define memmovev(s1,s2,n)	memmove(s1, (char *)s2, n)
#define memchrv(s,c,n)		memchr((char *)s, c, n)


/******************************************************************
 * Thread event queue routines - ast interlocked (ie. safe)       *
 ******************************************************************/

#define MAXQUEUE 255                   /* Arbitrary but large */

static volatile int queue_thread_array[MAXQUEUE];
static volatile int queue_event_array[MAXQUEUE];
static volatile int queue_count = 0;
static volatile int queue_head = 0;
static volatile int queue_tail = 0;

void flush_sessionlog(thread,msg)
  int thread;
  char *msg;
{
#ifdef sessionlogfile
  fprintf(sessionlog, "%s(%d): ------------\n",msg,thread);
  if (fclose(sessionlog)) _ck_close(sessionlog);
  sessionlog = fopen(sessionlogfile,"a+");
  _ck_open_a(sessionlog,sessionlogfile);
#endif
}

static int enqueue(thread,event)
int thread, event;
{
  int status;     /* in AST it must be a *local* variable for use by _c$cks() */
  int result = 1;
  _c$cks(sys$setast(0));
  if (queue_count >= MAXQUEUE) result = 0;
  else {
    queue_thread_array[queue_head] = thread;
    queue_event_array[queue_head] = event;
    if (++queue_head >= MAXQUEUE) queue_head = 0;
    queue_count++;
  }
  _c$cks(sys$setast(1));
  return(result);
}

static int dequeue(thread,event)
int *thread, *event;
{
  int status;     /* in AST it must be a *local* variable for use by _c$cks() */
  int result = 1;
  _c$cks(sys$setast(0));
  if (queue_count <= 0) result = 0;
  else {
    *thread = queue_thread_array[queue_tail];
    *event = queue_event_array[queue_tail];
    if (++queue_tail >= MAXQUEUE) queue_tail = 0;
    queue_count--;
  }
  _c$cks(sys$setast(1));
  return(result);
}

/* Discard queue elements belonging to the given thread.
 * Typically should be called during thread rundown.
 */
static void purge_queue(thread)
int thread;
{
  int status;     /* in AST it must be a *local* variable for use by _c$cks() */
  int cnt,purge_cnt,ind_src,ind_dst;

  purge_cnt = 0; drvrlog = NULL;
/* exclusive lock on queue_head, queue_tail, queue_count */
  _c$cks(sys$setast(0));
  ind_src = ind_dst = queue_tail;
  for (cnt=1; cnt<=queue_count; cnt++)
  {
    if (queue_thread_array[ind_src] == thread) {
      if (purge_cnt <= 0) {
        currtime = time(NULL);
        drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
        if (!_ck_open_a(drvrlog,drvrlogfile)) drvrlog = NULL;
        if (drvrlog)
          fprintf(drvrlog,"%.24s: Rundown of thread %d, events purged: %d",
                  asctime(localtime(&currtime)), thread, active_count);
      }
      if (drvrlog) fprintf(drvrlog," %d",queue_event_array[ind_src]);
      purge_cnt++;
    }
    else {
      if (ind_src != ind_dst) {                      /* don't copy needlessly */
        queue_thread_array[ind_dst] = queue_thread_array[ind_src];
        queue_event_array[ind_dst]  = queue_event_array[ind_src];
      }
      if (++ind_dst >= MAXQUEUE) ind_dst = 0;
    }
    if (++ind_src >= MAXQUEUE) ind_src = 0;
  }
  queue_head = ind_dst; queue_count-= purge_cnt;
  _c$cks(sys$setast(1));
  if (purge_cnt && drvrlog) {
    fputc('\n',drvrlog);
    if (fclose(drvrlog)) _ck_close(drvrlog);
  }
  return;
}

/***************************************************************
 * AST handlers                                                *
 ***************************************************************/

void ast_signal_open_complete(dummy)
int dummy;
{
  int status;     /* in AST it must be a *local* variable for use by _c$cks() */
  int s = listen_iosb[0];
  if (s&1) ;
  else if (EXPECTED_IO_ERROR(s)) ;  /* no panic, expected errors */
  else _c$cks(s);
  (void)enqueue(0,(s&1) ? EV_OPEN_COMPLETED : EV_OPEN_ABORTED);
  _c$cks(sys$wake(0,0));
}

void ast_signal_read_complete(thread)
int thread;
{
  int status;     /* in AST it must be a *local* variable for use by _c$cks() */
  if (threads[thread].active==ACT_HEALTHY)
  {
    int s = threads[thread].read_iosb[0];
    int len = (s&1) ? threads[thread].read_iosb[1] : 0;
    if (s&1) ;
    else if (EXPECTED_IO_ERROR(s)) ;  /* no panic, expected errors */
    else _c$cks(s);
    threads[thread].read_buffer_len = len;
    (void)enqueue(thread, (s&1) ? EV_READ_COMPLETED : EV_READ_ABORTED);
    _c$cks(sys$wake(0,0));
  }
}

void ast_signal_write_complete(thread)
int thread;
{
  int status;     /* in AST it must be a *local* variable for use by _c$cks() */
  if (threads[thread].active==ACT_HEALTHY)
  {
    int s = threads[thread].write_iosb[0];
    if (s&1) ;
    else if (EXPECTED_IO_ERROR(s)) ;  /* no panic, expected errors */
    else _c$cks(s);
    (void)enqueue(thread, (s&1) ? EV_WRITE_COMPLETED : EV_WRITE_ABORTED );
    _c$cks(sys$wake(0,0));
  }
}

void inactivity_tick_trap(sig)
  int sig;
{
  inactivity_tick_trap_flag = 1;
}

/***************************************************************
 * Read/Write queueing routines                                *
 ***************************************************************/

static void read_thread(thread)
int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  int status;
  /* IO$_READVBLK reads a block of data (upto BUFSIZE) from an
   * open channel as in I/O users guide.
   */
  if (!_c$cks(sys$qio(0,t->channel,IO$_READVBLK,t->read_iosb,
                      ast_signal_read_complete,thread,t->read_buffer,BUFSIZE,
                      0,0,0,0))) {
    t->read_buffer_len = 0;
    (void)enqueue(thread,EV_READ_ABORTED);
  }
}

static void write_thread(thread)
int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  struct line_block *b = t->write_head;
  int status, len, write_buffer_len;

  if ((t->active==ACT_HEALTHY) && (t->write_in_progress==0)) {
    for (write_buffer_len=0; b && write_buffer_len<BUFSIZE; ) {
      len = min(b->line_len, BUFSIZE-write_buffer_len);
      if (len>0)
        memcpyv(t->write_buffer + write_buffer_len, b->line, len);
      write_buffer_len += len;
      if (len >= b->line_len) {
        /* the whole string was taken */
        t->write_head = b->next;
        if (b->line) news_free(b->line);
        news_free(b);
        b = t->write_head;
        }
      else {
        /* partial string was taken, shift the remaining characters */
        memmove(b->line, b->line + len, b->line_len - len);
        b->line_len -= len;
      }
    }
#ifdef sessionlogfile
    { int j; char c;
      fprintf(sessionlog,"wthr(%d): ",thread);
      for (j=0; j<write_buffer_len; j++) {
        c = (t->write_buffer)[j];
        if      (c == '\n') fprintf(sessionlog,"<LF>");
        else if (c == '\r') fprintf(sessionlog,"<CR>");
        else if (c<' ' || c>=127) fprintf(sessionlog,"<%d>",c);
        else fputc(c,sessionlog);
      }
      fputc('\n',sessionlog);
    }
#endif
    /* IO$_WRITEVBLK writes a block of data (length in P2) */
    if (! (t->write_in_progress =
           _c$cks(sys$qio(0,t->channel,IO$_WRITEVBLK,t->write_iosb,
                    ast_signal_write_complete,thread,
                    t->write_buffer,write_buffer_len,
                    0,0,0,0))))   (void)enqueue(thread,EV_WRITE_ABORTED);
  }
}

/***********************************************************************
 * Inits                                                               *
 ***********************************************************************/

static void init_thread(thread,channel)
int thread;
unsigned short channel;
{
  volatile struct thread_struct *t = &threads[thread];
  t->active = ACT_HEALTHY;
  t->channel = channel;
  t->read_head = NULL; t->read_tail = NULL;
  t->write_head = NULL; t->write_tail = NULL;
  t->write_in_progress = 0;
  t->read_residue_len = 0;
  t->read_buffer_len = 0;
  t->inactivity_ticks = 0;
  server_init_unit(thread);
  read_thread(thread);
  write_thread(thread);
}

static int find_clear_thread(void)
{
  int i;
  for (i=1;i<=MAXTHREADS;i++)
    if (threads[i].active==ACT_UNASSIGNED) return(i);
  return(0);
}

static void init_listener(void)
{
  int status;

  active_listener = 0;
  if (!beenhere)
  {
    lhst_adrs.lgth = sizeof local_host;
    lhst_adrs.hst = &local_host;
    local_host.inet_port = htons(NNTP_PORT);
    local_host.inet_family = INET$C_AF_INET; /* INET family */
    local_host.adrs = INET$C_INADDR_ANY; /* Network/subnetwork*/
    local_host.blkb[0] = 0;
    local_host.blkb[1] = 0;
    local_host.blkb[2] = 0;
    local_host.blkb[3] = 0;
    local_host.blkb[4] = 0;
    local_host.blkb[5] = 0;
    local_host.blkb[6] = 0;
    local_host.blkb[7] = 0;

    rhst_adrs.lgth = 16;
    rhst_adrs.hst = &remote_hostaddr;
    rhst_adrs.retlth = &retlen;
    retlen=0;

    /* set up listener channel */
    if (!_ck_sys(NNTP$_ASGNLIS,0, sys$assign(&ipdsc,&listen_channel,0,0) ))
      return;  /* Failed to assign Listener channel */
    if (!_ck_sys(NNTP$_BINDLIS,0, sys$qiow(0,listen_channel,IO$_SETMODE,listen_iosb,
                 0,0,&sck_parm,0,&lhst_adrs,3,0,0) ))
      return;  /* Failed to bind Listener channel */
    if (!_ck_sys(NNTP$_BINDLIS,0, listen_iosb[0]))
      return;  /* Failed to bind Listener channel */
    beenhere=1;
  }
  if (!_ck_sys(NNTP$_ASGNCLO,0, sys$assign(&ipdsc,&clone_channel,0,0) ))
    (void)enqueue(0,EV_OPEN_ABORTED);  /* Failed to assign Clone channel */
  if (!_ck_sys(NNTP$_BINDCLO,0, sys$qio(0,listen_channel,IO$_ACCESS|IO$M_ACCEPT,listen_iosb,
                                        ast_signal_open_complete,0,0,0,
                                        &rhst_adrs,&clone_channel,0,0) ))
    (void)enqueue(0,EV_OPEN_ABORTED);  /* Failed to bind Clone channel */
  else
    active_listener = 1;
}

static void init_threads(void)
{
  volatile struct thread_struct *t;
  int c;
  active_count = 0;
  for (t=threads,c=0; c<MAXTHREADS; t++,c++) t->active = ACT_UNASSIGNED;
}

void init_inactivity_tick(void)
{
  signal(SIGALRM,inactivity_tick_trap);
  alarm(INACTIVITY_TICK_LENGTH);
}

static int already_running(void)
{
  $DESCRIPTOR_CONST(mbxdesc,"NNTP_ANCHOR");
  static unsigned short mbxchan;
  int status;
  int chncnt = 0;
  struct il
  { unsigned short buflen, itemcode;
    int *bufadr, *retlenadr; } itmlst[2];
  status = sys$crembx(1,&mbxchan,0,0,0xff,0,&mbxdesc);
  if (status == SS$_NOPRIV) {
    report_sys_error_routine(NEWS$_NOPRIV,
      (int)(c$dsc("SYSPRV,SYSNAM,SYSLCK,PRMMBX")),0,0,0);
    return(0);
    }
  else if (!_c$cks(status)) return(0);
  if (!_c$cks(sys$delmbx(mbxchan))) return(0);
  itmlst[0].buflen = 4;
  itmlst[0].itemcode = DVI$_REFCNT;
  itmlst[0].bufadr = &chncnt;
  itmlst[0].retlenadr = 0;
  itmlst[1].buflen = 0; itmlst[1].itemcode = 0;
  if (!_c$cks(sys$getdviw(0,mbxchan,0,itmlst,0,0,0,0))) return(0);
  return(chncnt - 1);
}

static void init(void)
{
  if (already_running()) exit(1);
  if (!server_init(MAXTHREADS)) exit(1);
  init_threads();
  init_listener();
  init_inactivity_tick();
}

/***********************************************************************
 * Active routines                                                     *
 ***********************************************************************/

static void enqueue_reader(thread,string,len)
int thread;
char *string;  /* string of 'len' characters - N.B.: not terminated with \0 ! */
int len;
{
  volatile struct thread_struct *t = &threads[thread];
  struct line_block *b;
  if (t->active==ACT_HEALTHY)
  {
#ifdef sessionlogfile
    fprintf(sessionlog,"enqr(%d): ",thread);
    fwrite(string, sizeof(char), len, sessionlog);
    fputc('\n',sessionlog);
#endif
    b = (struct line_block *)news_malloc(sizeof(struct line_block));
    b->next = NULL;
    b->line = (len<=0) ? NULL : memcpy((char *)news_malloc(len), string, len);
    b->line_len = len;
    if (t->read_head == NULL) {
      t->read_tail = t->read_head = b;
    }
    else {
      news_assert(t->read_tail != NULL);
      t->read_tail->next = b;
      t->read_tail = b;
    }
    if ((t->next_type==TYP_CMD_INPUT) ||
        ((t->next_type==TYP_FILE_INPUT) && (len==1 && string[0]=='.') ))
    {
      (t->next_function)(thread);
      write_thread(thread);
    }
  }
}

static void abort_thread(thread)
int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  struct line_block *b;
  if (t->active != ACT_UNASSIGNED)
  {
    t->active = ACT_UNASSIGNED;
    b = t->read_head;
    while(b!=NULL)
    {
      struct line_block *c = b;
      b=b->next;
      if (c->line) news_free(c->line);
      news_free(c);
    }
    t->read_head = NULL; t->read_tail = NULL;
    b = t->write_head;
    while(b!=NULL)
    {
      struct line_block *c = b;
      b=b->next;
      if (c->line) news_free(c->line);
      news_free(c);
    }
    t->write_head = NULL; t->write_tail = NULL;
    if (t->next_type==TYP_FILE_INPUT) enqueue_reader(thread,".",1);
    if (t->next_type==TYP_CMD_INPUT) enqueue_reader(thread,"QUIT",4);
    log_to_file(thread);
    purge_queue(thread);
    _c$cks(sys$dassgn(t->channel));
    active_count--;
    currtime = time(NULL);
    drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
    if (_ck_open_a(drvrlog,drvrlogfile)) {
      fprintf(drvrlog,"%.24s: Thread %d aborted, %d active\n",
              asctime(localtime(&currtime)),
              thread,
              active_count);
      if (fclose(drvrlog)) _ck_close(drvrlog);
      }
    flush_sessionlog(thread,"abrt");
    /* if (active_count<=0) exit(1); */
  }
}

static void close_thread(thread)
int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  int s,status;
  /* deaccess channel, don't flush pending i/o
   */
  t->active = ACT_UNASSIGNED;
  log_to_file(thread);
  s = sys$qiow(0,t->channel,IO$_DEACCESS,t->write_iosb,0,0,0,0,0,0,0,0);
  if (s&1) s = t->write_iosb[0];
  if (s&1) ;
  else if (EXPECTED_IO_ERROR(s)) ;  /* no panic, expected errors */
  else _c$cks(s);
  purge_queue(thread);
  _c$cks(sys$dassgn(t->channel));
  active_count--;
  currtime = time(NULL);
  drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
  if (_ck_open_a(drvrlog,drvrlogfile)) {
    fprintf(drvrlog,"%.24s: Thread %d closed, %d active\n",
              asctime(localtime(&currtime)),
              thread,
              active_count);
    if (fclose(drvrlog)) _ck_close(drvrlog);
    }
  flush_sessionlog(thread,"clos");
/* if (active_count<=0) exit(1); */
}

static void manage_open(void)
{
  int thread, s, status;
  unsigned short new_channel;
  char resource_msg[] = "400 Server resource limits exceeded.\r\n";

  active_listener = 0;
  new_channel = clone_channel;
  init_listener();
  thread = find_clear_thread();
  if (thread>0)
  {
    init_thread(thread,new_channel);
    active_count++;
    currtime = time(NULL);
    drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
    if (_ck_open_a(drvrlog,drvrlogfile)) {
      fprintf(drvrlog,"%.24s: Thread %d opened, %d.%d.%d.%d %d, %d active\n",
              asctime(localtime(&currtime)),
              thread,
              remote_hostaddr.adrs[0],
              remote_hostaddr.adrs[1],
              remote_hostaddr.adrs[2],
              remote_hostaddr.adrs[3],
              ntohs(remote_hostaddr.inet_port),
              active_count);
      if (fclose(drvrlog)) _ck_close(drvrlog);
      }
}
  else
  {
    s = sys$qiow(0,new_channel,IO$_WRITEVBLK,new_iosb,0,0,
                 resource_msg,strlen(resource_msg),0,0,0,0);
    if (s&1) s = new_iosb[0];
    if (s&1) ;
    else if (EXPECTED_IO_ERROR(s)) ;  /* no panic, expected errors */
    else _c$cks(s);

    s = sys$qiow(0,new_channel,IO$_DEACCESS,listen_iosb,0,0,0,0,0,0,0,0);
    if (s&1) s = listen_iosb[0];
    if (s&1) ;
    else if (EXPECTED_IO_ERROR(s)) ;  /* no panic, expected errors */
    else _c$cks(s);
    _c$cks(sys$dassgn(new_channel));
  }
}

static void manage_failed_open(void)
{
  currtime = time(NULL);
  drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
  if (_ck_open_a(drvrlog,drvrlogfile)) {
    fprintf(drvrlog,"%.24s: manage_failed_open, %d total active threads\n",
                     asctime(localtime(&currtime)),
                     active_count);
    if (fclose(drvrlog)) _ck_close(drvrlog);
    }
  _c$cks(sys$dassgn(clone_channel));
  active_listener = 0;
  if (active_count<=0) exit(1);
  init_listener();
}

static void manage_read_completion(thread)
int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  volatile char *cs, *cp;
  int len, cs_l;

  t->inactivity_ticks = 0; /* reset inactivity tick counter */
  news_assert(t->read_residue_len + t->read_buffer_len <= LINESIZE);
  memcpyv(t->read_residue + t->read_residue_len,
          t->read_buffer, t->read_buffer_len);
  t->read_residue_len += t->read_buffer_len;
  
  for (cs=t->read_residue,cs_l=t->read_residue_len; cs_l>0; cs_l-=cp-cs,cs=cp) {
    cp = memchrv(cs,'\n',min(cs_l,NNTP_STRLEN-2));
    if (cp) {
      len = cp-cs;
      if (len>0 && cp[-1]=='\r') len--;
      cp++;
    } else if (LINESIZE-cs_l < BUFSIZE) {
      /* must force line break in order to guarantee sufficient space
         in read_residue for the next record */

      len = min(cs_l,NNTP_STRLEN-2);
           /* take all there is in the buffer - but no more
              than the NNTP_SERVER can handle;
              location of the line break is quite arbitrary,
              could take e.g. BUFSIZE or LINESIZE-BUFSIZE
           */
      cp = cs+len;
    } else break;  /* leave remaining characters as a residue */
    if (t->next_type != TYP_NO_INPUT) enqueue_reader(thread,cs,len);
  }
  news_assert(cs_l >= 0);
  if (cs_l>0 && cs != t->read_residue)
    memmovev(t->read_residue,cs,cs_l);
  t->read_residue_len = cs_l;
  if (t->next_type!=TYP_NO_INPUT) read_thread(thread);
}

static void manage_failed_read(thread)
int thread;
{
  abort_thread(thread);
}

static void manage_failed_write(thread)
int thread;
{
  abort_thread(thread);
}

static void manage_write_completion(thread)
int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  t->inactivity_ticks = 0; /* reset inactivity tick counter */
  t->write_in_progress = 0;
  if (t->write_head)
  {
    write_thread(thread);
  }
  else
  {
    if (t->next_type==TYP_NO_INPUT) close_thread(thread);
  }
}

static void drain_queue(void)
{
  int thread, event;
  while (dequeue(&thread,&event))
  {
    switch (event)
    {
       case EV_OPEN_COMPLETED:  manage_open(); break;
       case EV_OPEN_ABORTED:    manage_failed_open(); break;
       case EV_READ_COMPLETED:  manage_read_completion(thread); break;
       case EV_READ_ABORTED:    manage_failed_read(thread); break;
       case EV_WRITE_COMPLETED: manage_write_completion(thread); break;
       case EV_WRITE_ABORTED:   manage_failed_write(thread); break;
    }
  }
}

void check_for_inactive_threads(void)
{
  volatile struct thread_struct *t;
  int i;

  inactivity_tick_trap_flag=0;
  init_inactivity_tick();

  for(i=1;i<=MAXTHREADS;i++)
  {
    t = &threads[i];
    if (t->active == ACT_HEALTHY)
    {
      if(++(t->inactivity_ticks) > INACTIVITY_TICK_MAX)
      {
        currtime = time(NULL);
        drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
        if (_ck_open_a(drvrlog,drvrlogfile)) {
          fprintf(drvrlog,"%.24s: Thread %d timed out\n",
                  asctime(localtime(&currtime)),
                  i);
          if (fclose(drvrlog)) _ck_close(drvrlog);
          }
        flush_sessionlog(i,"timo");
        _c$cks(sys$cancel(t->channel));
      }
    }
  }
}

int main(void)
{
  establish_handler(server_condition_handler);
#ifdef sessionlogfile
  sessionlog = fopen(sessionlogfile,"w");
  _ck_open_w(sessionlog,sessionlogfile);
#endif
  currtime = time(NULL);
  drvrlog = fopen(drvrlogfile,"a+",drvrlogfile_def);
  if (_ck_open_a(drvrlog,drvrlogfile)) {
    fprintf(drvrlog,"%.24s: NNTP_TCPUCXM server started, version: %s\n",
            asctime(localtime(&currtime)),
            version);
    if (fclose(drvrlog)) _ck_close(drvrlog);
    }
  init();
  for (;;)
  {
    drain_queue();
    if (inactivity_tick_trap_flag)
	check_for_inactive_threads();
    _c$cks(sys$hiber());
  }
#ifdef sessionlogfile
  if (fclose(sessionlog)) _ck_close(sessionlog);
#endif
  return(1);
}

/*************************************************************************
 * NNTP_SERVER interface routines                                        *
 *************************************************************************/

int read_net(string,string_max_size,thread)
char *string;
int string_max_size, thread;
{
  volatile struct thread_struct *t = &threads[thread];
  struct line_block *b = t->read_head;
  int b_len;
  if (b)
  {
    b_len = b->line_len;
    news_assert(b_len+2 <= string_max_size);
    if (b_len>0) memcpy(string,b->line,b_len);
    string[b_len] = '\n';
    string[b_len+1] = '\0';
    t->read_head = b->next;
    if (t->read_head == NULL) t->read_tail = NULL;
    if (b->line) news_free(b->line);
    news_free(b);
#ifdef sessionlogfile
    fprintf(sessionlog,"rd_n(%d): %s",thread,string);
#endif
    return(b_len+1);   /* +1 for the CR */
  }
  return(0);   /* was return(1); ?!! */
}

void write_net(string,thread)
  const char *string;
  int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  struct line_block *b;
  int string_len = strlen(string);

  if (t->active==ACT_HEALTHY)
  {
#ifdef sessionlogfile
    fprintf(sessionlog,"wr_n(%d): %s",thread,string);
#endif
    b = (struct line_block *) news_malloc(sizeof(struct line_block));
    b->next = NULL;
    b->line = (string_len<=0) ? NULL
                : memcpy((char *)news_malloc(string_len), string, string_len);
    b->line_len = string_len;
    if (t->write_head)
    {
      t->write_tail->next = b;
      t->write_tail = b;
    }
    else
    {
      t->write_tail = b;
      t->write_head = b;
    }
  }
}

/*
 * close_net - shut down all threads.  Called only by mem_fail when preparing
 * to exit due to memory allocation failure.
 */

void close_net()
{
  int i;

  for (i=1; i <= MAXTHREADS; i++)
    if (threads[i].active == ACT_HEALTHY) abort_thread(i);
}


/*
 *  next_call
 *
 *  Set the next function to be invoked by the server
 */

void next_call(thread,func,type)
    int thread;
    void (*func)(int);
    int type;
{
    threads[thread].next_function = func;
    threads[thread].next_type = type;
}

void getremhost(remhost,remuser,thread)
  char *remhost, *remuser;
  int thread;
{
  volatile struct thread_struct *t = &threads[thread];
  struct mysockaddr rmtsckadr;
  struct hostent *hostinfop;
  int retlen;
  int status;

  strcpyv((char *)t->remuser,"nntp");    if (remuser) strcpy(remuser,"nntp");
  strcpyv((char *)t->remhost,"unknown"); if (remhost) strcpy(remhost,"unknown");

  retlen = sizeof rmtsckadr;   /* on entry it must contain the size available */
#if defined(TCPWARE) /*  || defined(MULTINET)  */
/* this usage is appropriate for: TCPWARE, MULTINET */
  { unsigned short iosb[4];
    if (!_c$cks(sys$qiow(0,t->channel,IO$_GETPEERNAME,iosb,0,0,
                         &rmtsckadr,&retlen,0,0,0,0))) return;
    if (!(iosb[0] & 1)) { _ck_sys(NNTP$_GETPEERNAMEFAIL,0,iosb[0]); return; }
  }
#elif defined(UCX) || defined(MULTINET)
/* this usage is appropriate for: UCX, MULTINET(ucx compatibility) */
  { struct itlst_3 rsck_adrs;
    unsigned short iosb[4];
    retlen=0; rsck_adrs.retlth = &retlen;
    rsck_adrs.hst = &rmtsckadr; rsck_adrs.lgth = sizeof rmtsckadr;
    if (!_c$cks(sys$qiow(0,t->channel,IO$_SENSEMODE,iosb,0,0,
                         0,0,0,&rsck_adrs,0,0))) return;
    if (!(iosb[0] & 1)) { _ck_sys(NNTP$_GETPEERNAMEFAIL,0,iosb[0]); return; }
  }
#else
/* this usage is appropriate for: UCX, MULTINET */
  if ( (status=getpeername(t->channel,(struct sockaddr*)&rmtsckadr,&retlen)) ) {
#ifdef MULTINET
    set_errno(socket_errno); set_vaxc_errno(vmserrno);
#endif
    report_cio_error(NNTP$_GETPEERNAMEFAIL,0); return;
    }
#endif

  if ((hostinfop=gethostbyaddr((char*)rmtsckadr.adrs,4,AF_INET))!=NULL)
    strcpy((char *)t->remhost,hostinfop->h_name);
/* try twice, in case the nameserver is slow */
  else if ((hostinfop=gethostbyaddr((char*)rmtsckadr.adrs,4,AF_INET))!=NULL)
    strcpy((char *)t->remhost,hostinfop->h_name);
  else {
    sprintf((char *)t->remhost,"%d.%d.%d.%d",
      rmtsckadr.adrs[0],rmtsckadr.adrs[1],rmtsckadr.adrs[2],rmtsckadr.adrs[3]);
    }
  if (remhost) strcpyv(remhost,(char *)t->remhost);
}
