From:	ADVAX::"mcnc!VM1.NoDak.EDU!keith%AND.CS.LIV.AC.UK" "Keith Halewood" 28-JUN-1991 22:46:43.93
To:	Multiple recipients of list ANU-NEWS <ANU-NEWS@NDSUVM1>
CC:	
Subj:	Improved multithreaded nntpserver for CMUTEK.

Received: by ADVAX.DECnet (utk-mail11 v1.5) ; Fri, 28 Jun 91 22:42:22 EDT
Received:  from mcnc by ge-dab.GE.COM (5.61/GE-DAB 1.15) with UUCP
	id AA23585 for ; Fri, 28 Jun 91 21:37:52 -0400
Received: from VM1.NoDak.EDU by mcnc.mcnc.org (5.59/MCNC/3-21-91)
	id AA24909; Tue, 25 Jun 91 02:32:08 -0400
	for ARISIA.DNET.ge.com!EVERHART
Message-Id: <9106250632.AA24909@mcnc.mcnc.org>
Received: from NDSUVM1.BITNET by VM1.NoDak.EDU (IBM VM SMTP R1.2.1MX) with BSMTP id 4710; Tue, 25 Jun 91 01:30:57 CDT
Received: from NDSUVM1.BITNET by NDSUVM1.BITNET (Mailer R2.07) with BSMTP id
 4279; Tue, 25 Jun 91 01:30:53 CDT
Date:         Mon, 24 Jun 1991 16:57:23 GMT
Reply-To: Keith Halewood <mcnc!VM1.NoDak.EDU!keith%AND.CS.LIV.AC.UK>
Sender: ANU-NEWS Discussion <ANU-NEWS@VM1.NoDak.EDU>
From: Keith Halewood <mcnc!VM1.NoDak.EDU!keith%AND.CS.LIV.AC.UK>
Subject:      Improved multithreaded nntpserver for CMUTEK.
To: Multiple recipients of list ANU-NEWS <ANU-NEWS@NDSUVM1>
 
Here's an improved version of the multithreaded NNTP server interface for
CMU-TEK. Improvements include server persistence, shameful bugs fixed, RFC931
authentication checking (discovers who - ie. userid - is on the client end,
like DECnet). Also included is the source code for an authentication server for
VAX/VMS.
 
Keith
 
$! ------------------ CUT HERE -----------------------
$ v='f$verify(f$trnlnm("SHARE_VERIFY"))'
$!
$! This archive created by VMS_SHARE Version 7.2-007  22-FEB-1990
$!   On 24-JUN-1991 16:50:58.47   By user KEITH
$!
$! This VMS_SHARE Written by:
$!    Andy Harper, Kings College London UK
$!
$! Acknowledgements to:
$!    James Gray       - Original VMS_SHARE
$!    Michael Bednarek - Original Concept and implementation
$!
$! TO UNPACK THIS SHARE FILE, CONCATENATE ALL PARTS IN ORDER
$! AND EXECUTE AS A COMMAND PROCEDURE  (  @name  )
$!
$! THE FOLLOWING FILE(S) WILL BE CREATED AFTER UNPACKING:
$!       1. AUTHSRV.C;19
$!       2. NNTP_TCPCMU_M.C;26
$!       3. README.TXT;3
$!
$set="set"
$set symbol/scope=(nolocal,noglobal)
$f=f$parse("SHARE_TEMP","SYS$SCRATCH:.TMP_"+f$getjpi("","PID"))
$e="write sys$error  ""%UNPACK"", "
$w="write sys$output ""%UNPACK"", "
$ if f$trnlnm("SHARE_LOG") then $ w = "!"
$ ve=f$getsyi("version")
$ if ve-f$extract(0,1,ve) .ges. "4.4" then $ goto START
$ e "-E-OLDVER, Must run at least VMS 4.4"
$ v=f$verify(v)
$ exit 44
$UNPACK: SUBROUTINE ! P1=filename, P2=checksum
$ if f$search(P1) .eqs. "" then $ goto file_absent
$ e "-W-EXISTS, File ''P1' exists. Skipped."
$ delete 'f'*
$ exit
$file_absent:
$ if f$parse(P1) .nes. "" then $ goto dirok
$ dn=f$parse(P1,,,"DIRECTORY")
$ w "-I-CREDIR, Creating directory ''dn'."
$ create/dir 'dn'
$ if $status then $ goto dirok
$ e "-E-CREDIRFAIL, Unable to create ''dn'. File skipped."
$ delete 'f'*
$ exit
$dirok:
$ w "-I-PROCESS, Processing file ''P1'."
$ if .not. f$verify() then $ define/user sys$output nl:
$ EDIT/TPU/NOSEC/NODIS/COM=SYS$INPUT 'f'/OUT='P1'
PROCEDURE Unpacker ON_ERROR ENDON_ERROR;SET(FACILITY_NAME,"UNPACK");SET(
SUCCESS,OFF);SET(INFORMATIONAL,OFF);f:=GET_INFO(COMMAND_LINE,"file_name");b:=
CREATE_BUFFER(f,f);p:=SPAN(" ")@r&LINE_END;POSITION(BEGINNING_OF(b));
LOOP EXITIF SEARCH(p,FORWARD)=0;POSITION(r);ERASE(r);ENDLOOP;POSITION(
BEGINNING_OF(b));g:=0;LOOP EXITIF MARK(NONE)=END_OF(b);x:=ERASE_CHARACTER(1);
IF g=0 THEN IF x="X" THEN MOVE_VERTICAL(1);ENDIF;IF x="V" THEN APPEND_LINE;
MOVE_HORIZONTAL(-CURRENT_OFFSET);MOVE_VERTICAL(1);ENDIF;IF x="+" THEN g:=1;
ERASE_LINE;ENDIF;ELSE IF x="-" THEN IF INDEX(CURRENT_LINE,"+-+-+-+-+-+-+-+")=
1 THEN g:=0;ENDIF;ENDIF;ERASE_LINE;ENDIF;ENDLOOP;t:="0123456789ABCDEF";
POSITION(BEGINNING_OF(b));LOOP r:=SEARCH("`",FORWARD);EXITIF r=0;POSITION(r);
ERASE(r);x1:=INDEX(t,ERASE_CHARACTER(1))-1;x2:=INDEX(t,ERASE_CHARACTER(1))-1;
COPY_TEXT(ASCII(16*x1+x2));ENDLOOP;WRITE_FILE(b,GET_INFO(COMMAND_LINE,
"output_file"));ENDPROCEDURE;Unpacker;QUIT;
$ delete/nolog 'f'*
$ CHECKSUM 'P1'
$ IF CHECKSUM$CHECKSUM .eqs. P2 THEN $ EXIT
$ e "-E-CHKSMFAIL, Checksum of ''P1' failed."
$ ENDSUBROUTINE
$START:
$ create 'f'
X#module AUTHSRV "V1.0"
X
X/* Authentication server (RFC931) for VAX/VMS and CMUTEK V6.5.
X * Keith Halewood, 8-JUN-1991.
X *
X * Notes:
X * Compile and link. Define TCP$AUTHSRV as a pointer to the image, and
X * add the following to SYS$MANAGER:INTERNET.CONFIG:
X * WKS:113:AUTHSRV:TCP$AUTHSRV:NETWRK:NETMBX,TMPMBX,PHY_IO,WORLD,SYSPRV:\
X * :SYS$NULL:SYS$NULL:SYS$NULL:4:5
X *
X * Warning:
X * With regard to the error responses, this server doesn't exactly comply
X * with the RFC. 'ERROR: UNKNOWN-ERROR' has been made more informative.
X * Also, the op-sys field is VAXVMS. I haven't bothered to check if this
X * is a valid NIC-defined OS identifier - who cares anyway.
X */
X
X#include descrip
X#include iodef
X#include signal
X#include stdio
X#include jpidef
X
X#define PORT 113          /* WKS port for RFC931. */
X#define TIMEOUT 30        /* Back off after 30 seconds of inactivity */
X#define MAXTCB 128        /* IO$_UNLOAD operations are very touchy about
X                           * the size of this which was calculated carefully
X                           * (read 'by trial and error').
X                           */
X
Xunsigned short f;
Xvolatile unsigned short iosb`5B4`5D;
X
Xchar input_line`5B1024`5D, output_line`5B1024`5D;
X
Xint status;
Xint locport, forport;
X
Xint tcblist`5BMAXTCB`5D;
X
Xstruct tcbtype
X`7B
X  unsigned int addr, procid;
X  unsigned short tcbstate, tcblaststate;
X  unsigned int forhost, forport, lochost, locport;
X  unsigned int filler`5B22`5D;
X`7D tcbblock;
X
Xstruct item_desc
X`7B
X   unsigned short buff_len;
X   unsigned short item_code;
X   unsigned long buffer_addr;
X   unsigned long ret_len_addr;
X`7D;
X
Xchar userid`5B13`5D;
Xint useridlen;
X
Xstruct item_desc itmlst`5B2`5D = `7B `7B 12, JPI$_USERNAME, &userid, &userid
Vlen `7D,
X                               `7B 0,  0,             0,       0 `7D `7D;
X
Xint i;
X
Xmain()
X`7B
X  open_net();
X  status = sys$qiow(0,f,IO$_READVBLK,iosb,0,0,input_line,1023,0,0,0,0);
X  if (!(status&1)) exit(1);
X  if (!(iosb`5B0`5D&1)) exit(1);
X  input_line`5Biosb`5B1`5D`5D='\0';
X  if (sscanf(input_line,"%d,%d\r\n",&locport,&forport)!=2)
X  `7B
X    sprintf(output_line,"%d, %d : ERROR : Invalid syntax\r\n",0,0);
X  `7D
X  else
X  `7B
X    tcblist`5B0`5D=0;
X    status = sys$qiow(0,f,IO$_UNLOAD,iosb,0,0,tcblist,MAXTCB*4,3,0,0,0);
X    if (!((status&1) && (iosb`5B0`5D&1)))
X    `7B
X      sprintf(output_line,"%d, %d : ERROR : Cannot retrieve TCBlist\r\n",
X              locport,forport);
X    `7D
X    else
X    `7B
X      sprintf(output_line,"%d, %d : ERROR : No such connection\r\n",
X              locport,forport);
X      for (i=1;i<=tcblist`5B0`5D;i++)
X      `7B
X        status = sys$qiow(0,f,IO$_UNLOAD,iosb,0,0,&tcbblock,sizeof(tcbblock)
V,4,
X                          tcblist`5Bi`5D,0,0);
X        if (!((status&1) && (iosb`5B0`5D&1)))
X        `7B
X          sprintf(output_line,"%d, %d : ERROR : Cannot retrieve TCB\r\n",
X                  locport,forport);
X          break;
X        `7D
X        else if ((tcbblock.locport == locport) && (tcbblock.forport == forpo
Vrt))
X        `7B
X          status = sys$getjpiw(0,&tcbblock.procid,0,itmlst,0,0,0);
X          if (status&1)
X          `7B
X            userid`5Buseridlen`5D='\0';
X            sprintf(output_line,"%d, %d : USERID : VAXVMS : %s\r\n",locport,
X                                forport,userid);
X          `7D
X          else
X          `7B
X            sprintf(output_line,"%d, %d : ERROR : No such process\r\n",
X                    locport,forport);
X          `7D
X          break;
X        `7D
X      `7D
X    `7D
X  `7D
X  status = sys$qiow(0,f,IO$_WRITEVBLK,iosb,0,0,output_line,
X                    strlen(output_line),0,0,0,0);
X  close_net();
X`7D
X
Xopen_net()
X`7B
X  int status;
X  $DESCRIPTOR(ipdsc,"INET$DEVICE");
X  status = sys$assign(&ipdsc,&f,0,0);
X  if (!(status&1)) exit(1);
X  status = sys$qiow(0,f,IO$_CREATE,iosb,0,0,0,0,PORT,0,0,TIMEOUT);
X  if (!(status&1)) exit(1);
X  if (!(iosb`5B0`5D&1)) exit(1);
X  return(1);
X`7D
X
Xclose_net()
X`7B
X  (void)sys$cancel(f);
X  (void)sys$qiow(0,f,IO$_DELETE,iosb,0,0,0,0,0,0,0,0);
X  (void)sys$dassgn(f);
X`7D
$ CALL UNPACK AUTHSRV.C;19 1647606117
$ create 'f'
X#module NNTP_MCMUTEK "V1.1"
X
X/* Multithreaded NNTP server for ANU_NEWS V6.x and CMU-TEK TCP/IP V6.5
X * By Keith Halewood. The 'getremhost' routine has been stolen from the
X * original single-thread server interface.
X *
X * Changes V1.0 -> V1.1
X * 1. Server exits when the passive listen times out and there are no
X *    active connections. Previously, the server exited immediately on
X *    the termination of the last active connection.
X * 2. As a consequence of the above, a bug preventing threads from
X *    being freed correctly has now been fixed.
X * 3. Rearranged manage_open so that a new passive listen is queued
X *    before the new thread is initialised to minimise the chances of
X *    of missing closely-spaced incoming connect requests.
X * 4. getremhost() has been modified to perform RFC931 authentication
X *    checking. Provided that a daemon is running on the client host,
X *    NNTP can now log the user as well as the host (a la DECnet).
X *    It's the result of 10 mins work so the code isn't pretty.
X * 5. Modified write_thread so that multiple lines are blocked into one
X *    QIO operation saving lots of CPU and increasing I/O throughput.
X * 6. log_to_file() is now called (see nntp_server source code).
X */
X
X#include descrip
X#include iodef
X#include signal
X#include stdio
X
X#define PORT       119     /* Should be 119, use 8888 for testing. */
X#define AUTHPORT   113     /* Authentication WKS */
X#define MAXTHREADS 16      /* Maximum number of services - Make sure
X                            * process quotas are capable of handling`20
X                            * the number chosen.
X                            */
X#define TIMEOUT    1800    /* Connection inactivty timeout in seconds;
X                            * causes an SS$_ABORT on reads and closes
X                            * the connection. Not exactly in the spirit
X                            * of the NNTP protocol, but most NNTP clients
X                            * are just as ill-mannered.
X                            */
X#define AUTHTIMEOUT  30    /* Authentication timeout.
X                            * Some Un*x auth servers are very slow, but
X                            * none should be slower than this.
X                            */
X#define BUFSIZE    1024    /* Constants for QIO buffer lengths and
X                            */
X#define LINESIZE   2048    /* the read line assembly buffer.
X                            */
X
X/* Event numbers used by AST completion routines. The are placed,
X * along with a thread number into the attention queue.
X */
X
X#define EV_NULL                     0
X#define EV_OPEN_COMPLETED           1
X#define EV_OPEN_ABORTED             2
X#define EV_READ_COMPLETED           3
X#define EV_READ_ABORTED             4
X#define EV_WRITE_COMPLETED          5
X#define EV_WRITE_ABORTED            6
X
X/* Is the thread in use (healthy) or unassigned. Unassigned is also
X * a state used during shutdown to control the discarding of entries
X * in the I/O queues.
X */
X
X#define ACT_UNASSIGNED              0
X#define ACT_HEALTHY                 1
X
X/* 'Next function' return codes, determines whether the 'next function' call
X * for a thread is activated on receipt of a line (command) or on receipt of
X * the file terminator (a single full-stop) or whether the thread is to be
X * closed down.
X */
X
X#define TYP_NO_INPUT                0
X#define TYP_CMD_INPUT               1
X#define TYP_FILE_INPUT              2
X
X/* Definitions for the passive open (listen) on PORT
X */
X
Xunsigned short listen_channel, listen_iosb`5B4`5D, new_iosb`5B4`5D;
Xint active_count = 0;
Xint active_listener = 0;
X
X/* I/O queue definition.
X */
X
Xstruct line_block
X  `7B
X    struct line_block *next;
X    char *line;
X  `7D;
X
X/* Thread structures. `5B0`5D isn't used.
X */
X
Xvolatile static struct thread_struct
X  `7B
X    int active;
X    unsigned short channel;
X    unsigned short write_iosb`5B4`5D, read_iosb`5B4`5D;
X    char write_buffer`5BBUFSIZE-1`5D, read_buffer`5BBUFSIZE-1`5D;
X    char read_residue`5BLINESIZE-1`5D;
X    struct line_block *read_head, *read_tail, *write_head, *write_tail;
X    int write_in_progress;
X    int next_type;
X    int (*next_function)();
X    char remhost`5B255`5D;
X    char remuser`5B32`5D;
X  `7D threads`5BMAXTHREADS+1`5D;
X
X$DESCRIPTOR(ipdsc,"INET$DEVICE");
X
X/******************************************************************
X * Exteral routines from NNTP_SERVER                              *
X ******************************************************************/
X
Xint server_init();
Xvoid server_init_unit();
Xvoid log_to_file();
X
X/******************************************************************
X * Thread event queue routines - ast interlocked (ie. safe)       *
X ******************************************************************/
X
X#define MAXQUEUE 255                   /* Arbitrary but large */
X
Xvolatile static int queue_thread_array`5BMAXQUEUE`5D;
Xvolatile static int queue_event_array`5BMAXQUEUE`5D;
Xvolatile static int queue_count = 0;
Xvolatile static int queue_head = 0;
Xvolatile static int queue_tail = 0;
X
Xstatic int enqueue(thread,event)
Xint thread, event;
X`7B
X  int result = 1;
X  (void)sys$setast(0);
X  if (queue_count == MAXQUEUE)
X  `7B
X    result = 0;
X  `7D
X  else
X  `7B
X    queue_count++;
X    queue_thread_array`5Bqueue_head`5D = thread;
X    queue_event_array`5Bqueue_head++`5D = event;
X    if (queue_head==MAXQUEUE) queue_head = 0;
X  `7D
X  (void)sys$setast(1);
X  return(result);
X`7D
X
Xstatic int dequeue(thread,event)
Xint *thread, *event;
X`7B
X  int result = 1;
X  (void)sys$setast(0);
X  if (queue_count > 0)
X  `7B
X    *thread = queue_thread_array`5Bqueue_tail`5D;
X    *event = queue_event_array`5Bqueue_tail++`5D;
X    queue_count--;
X    if (queue_tail==MAXQUEUE) queue_tail = 0;
X  `7D else
X  `7B
X    result = 0;
X  `7D
X  (void)sys$setast(1);
X  return(result);
X`7D
X
X/* Convenient way of duplicating a string
X */
X
Xchar *strdup(s)
Xchar *s;
X`7B
X  char *t = (char *)malloc(sizeof(char)*(strlen(s)+1));
X  strcpy(t,s);
X  return(t);
X`7D
X
X/***************************************************************
X * AST handlers                                                *
X ***************************************************************/
X
Xvoid ast_signal_open_complete(dummy)
Xint dummy;
X`7B
X  (void)enqueue(0,(listen_iosb`5B0`5D&1) ? EV_OPEN_COMPLETED : EV_OPEN_ABORT
VED);
X  (void)sys$wake(0,0);
X`7D
X
Xvoid ast_signal_read_complete(thread)
Xint thread;
X`7B
X  if (threads`5Bthread`5D.active==ACT_HEALTHY)
X  `7B
X    (void)enqueue(thread, (threads`5Bthread`5D.read_iosb`5B0`5D&1) ?
X                          EV_READ_COMPLETED : EV_READ_ABORTED);
X    (void)sys$wake(0,0);
X  `7D
X`7D
X
Xvoid ast_signal_write_complete(thread)
Xint thread;
X`7B
X  if (threads`5Bthread`5D.active==ACT_HEALTHY)
X  `7B
X    (void)enqueue(thread, (threads`5Bthread`5D.write_iosb`5B0`5D&1) ?
X                          EV_WRITE_COMPLETED : EV_WRITE_ABORTED );
X    (void)sys$wake(0,0);
X  `7D
X`7D
X
X/***************************************************************
X * Read/Write queueing routines                                *
X ***************************************************************/
X
Xstatic void read_thread(thread)
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  int status;
X  /* IO$_READVBLK reads a block of data (upto BUFSIZE) from an
X   * open channel as in I/O users guide.
X   */
X  status = sys$qio(0,t->channel,IO$_READVBLK,t->read_iosb,
X                   ast_signal_read_complete,thread,
X                   t->read_buffer,BUFSIZE,0,0,0,0);
X  if (!(status&1)) (void)enqueue(thread,EV_READ_ABORTED);
X`7D
X
Xstatic void write_thread(thread)
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  struct line_block *b = t->write_head;
X  int status, iosize, s;
X  if ((t->active==ACT_HEALTHY) && (t->write_in_progress==0))
X  `7B
X    if (b!=NULL)
X    `7B
X      iosize = 0;
X      t->write_buffer`5B0`5D = '\0';
X      while (b)
X      `7B
X        s = strlen(b->line);
X        if (s+iosize > BUFSIZE-2) break;
X        strcat(t->write_buffer,b->line);
X        iosize = iosize + s;
X        t->write_head = b->next;
X        free(b->line);
X        free(b);
X        b = t->write_head;
X      `7D
X      /* IO$_WRITEVBLK writes a block of data (length in P2) to an
X       * open channel as in I/O users guide.
X       */
X      status = sys$qio(0,t->channel,IO$_WRITEVBLK,t->write_iosb,
X                       ast_signal_write_complete,thread,
X                       t->write_buffer,iosize,0,0,0,0);
X      t->write_in_progress = status&1;
X      if (!(status&1)) (void)enqueue(thread,EV_WRITE_ABORTED);
X    `7D
X  `7D
X`7D
X
X/***********************************************************************
X * Inits                                                               *
X ***********************************************************************/
X
Xstatic void init_thread(thread,channel)
Xint thread;
Xunsigned short channel;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  t->active = ACT_HEALTHY;
X  t->channel = channel;
X  t->read_head = NULL; t->read_tail = NULL;
X  t->write_head = NULL; t->write_tail = NULL;
X  t->write_in_progress = 0;
X  server_init_unit(thread);
X  read_thread(thread);
X  write_thread(thread);
X`7D
X
Xstatic int find_clear_thread()
X`7B
X  int i;
X  for (i=1;i<=MAXTHREADS;i++)
X    if (threads`5Bi`5D.active==ACT_UNASSIGNED) return(i);
X  return(0);
X`7D
X
Xstatic void init_listener()
X`7B
X  int status;
X  active_listener = 0;
X  status = sys$assign(&ipdsc,&listen_channel,0,0);
X  if (!(status&1)) return;
X  /* IO$_CREATE in this instance opens a passive listen on PORT
X   * the listen as well as the future established connection on this
X   * channel will timeout after TIMEOUT second's worth of inactivity
X   * and send an SS$_ABORT to any queued I/O operations on the channel.
X   */
X  status = sys$qio(0,listen_channel,IO$_CREATE,listen_iosb,
X           ast_signal_open_complete,0,0,0,PORT,0,0,TIMEOUT);
X  if (!(status&1))
X  `7B
X    (void)enqueue(0,EV_OPEN_ABORTED);
X  `7D
X  else
X  `7B
X    active_listener = 1;
X  `7D
X`7D
X
Xstatic void init_threads()
X`7B
X  struct thread_struct *t;
X  int c;
X  active_count = 0;
X  for (t=threads,c=0;c<MAXTHREADS;t++,c++) t->active = ACT_UNASSIGNED;
X`7D
X
Xstatic void init()
X`7B
X  if (!server_init(MAXTHREADS)) exit(1);
X  init_threads();
X  init_listener();
X`7D
X
X/***********************************************************************
X * Active routines                                                     *
X ***********************************************************************/
X
Xstatic void enqueue_reader(thread,string)
Xint thread;
Xchar *string;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  struct line_block *b;
X  if (t->active==ACT_HEALTHY)
X  `7B
X    b = (struct line_block *) malloc(sizeof(struct line_block));
X    b->next = NULL;
X    b->line = strdup(string);
X    if (t->read_head!=NULL)
X    `7B
X      t->read_tail->next = b;
X      t->read_tail = b;
X    `7D
X    else
X    `7B
X      t->read_tail = b;
X      t->read_head = b;
X    `7D
X    if ((t->next_type==TYP_CMD_INPUT) `7C`7C
X        ((t->next_type==TYP_FILE_INPUT) && !strcmp(string,".")))
X    `7B
X      (t->next_function)(thread);
X      write_thread(thread);
X    `7D
X  `7D
X`7D
X
Xstatic void abort_thread(thread)
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  struct line_block *b;
X  if (t->active != ACT_UNASSIGNED)
X  `7B
X    t->active = ACT_UNASSIGNED;
X    b = t->read_head;
X    while(b!=NULL)
X    `7B
X      struct line_block *c = b;
X      b=b->next;
X      free(c->line);
X      free(c);
X    `7D
X    t->read_head = NULL; t->read_tail = NULL;
X    b = t->write_head;
X    while(b!=NULL)
X    `7B
X      struct line_block *c = b;
X      b=b->next;
X      free(c->line);
X      free(c);
X    `7D
X    t->write_head = NULL; t->write_tail = NULL;
X    if (t->next_type==TYP_FILE_INPUT) enqueue_reader(thread,".");
X    if (t->next_type==TYP_CMD_INPUT) enqueue_reader(thread,"QUIT");
X    log_to_file(thread);
X    (void)sys$dassgn(t->channel);
X    active_count--;
X    /* if (active_count<=0) exit(1); */
X  `7D
X`7D
X
Xstatic void close_thread(thread)
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  /* IO$_DELETE closes the connection after ensuring that all data
X   * transfer in progress has reached its destinations.
X   */
X  t->active = ACT_UNASSIGNED;
X  log_to_file(thread);
X  (void)sys$qiow(0,t->channel,IO$_DELETE,t->write_iosb,0,0,0,0,0,0,0,0);
X  (void)sys$dassgn(t->channel);
X  active_count--;
X  /* if (active_count<=0) exit(1); */
X`7D
X
X
Xstatic void manage_open()
X`7B
X  int thread, status;
X  unsigned short new_channel;
X  char resource_msg`5B`5D = "400 Server resource limits exceeded.\r\n";
X  active_listener = 0;
X  new_channel = listen_channel;
X  init_listener();
X  thread = find_clear_thread();
X  if (thread>0)
X  `7B
X    init_thread(thread,new_channel);
X    active_count++;
X  `7D
X  else
X  `7B
X    status = sys$qiow(0,new_channel,IO$_WRITEVBLK,new_iosb,0,0,
X                      resource_msg,strlen(resource_msg),0,0,0,0);
X    status = sys$qiow(0,new_channel,IO$_DELETE,listen_iosb,0,0,0,0,0,0,0,0);
X    (void)sys$dassgn(new_channel);
X  `7D
X`7D
X
Xstatic void manage_failed_open()
X`7B
X  (void)sys$dassgn(listen_channel);
X  active_listener = 0;
X  if (active_count<=0) exit(1);
X  init_listener();
X`7D
X
Xstatic void manage_read_completion(thread)
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  char *cp;
X  t->read_buffer`5Bt->read_iosb`5B1`5D`5D='\0';
X  strcat(t->read_residue,t->read_buffer);
X  while (t->next_type!=TYP_NO_INPUT &&`20
X           (cp = strchr(t->read_residue,'\n'))!=NULL)
X  `7B
X    *cp='\0'; if (cp`5B-1`5D=='\r') cp`5B-1`5D='\0'; cp++;
X    enqueue_reader(thread,t->read_residue);
X    strcpy(t->read_residue,cp);
X  `7D
X  if (t->next_type!=TYP_NO_INPUT) read_thread(thread);
X`7D
X
Xstatic void manage_failed_read(thread)
Xint thread;
X`7B
X  abort_thread(thread);
X`7D
X
Xstatic void manage_failed_write(thread)
Xint thread;
X`7B
X  abort_thread(thread);
X`7D
X
Xstatic void manage_write_completion(thread)
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  t->write_in_progress = 0;
X  if (t->write_head)
X  `7B
X    write_thread(thread);
X  `7D
X  else
X  `7B
X    if (t->next_type==TYP_NO_INPUT) close_thread(thread);
X  `7D
X`7D
X
Xstatic void drain_queue()
X`7B
X  int thread, event;
X  while (dequeue(&thread,&event))
X  `7B
X    switch (event)
X    `7B
X       case EV_OPEN_COMPLETED:  manage_open(); break;
X       case EV_OPEN_ABORTED:    manage_failed_open(); break;
X       case EV_READ_COMPLETED:  manage_read_completion(thread); break;
X       case EV_READ_ABORTED:    manage_failed_read(thread); break;
X       case EV_WRITE_COMPLETED: manage_write_completion(thread); break;
X       case EV_WRITE_ABORTED:   manage_failed_write(thread); break;
X    `7D
X  `7D
X`7D
X
Xmain()
X`7B
X  init();
X  for (;;)
X  `7B
X    drain_queue();
X    (void)sys$hiber();
X  `7D
X`7D
X
X/*************************************************************************
X * NNTP_SERVER interface routines                                        *
X *************************************************************************/
X
Xint read_net(string,size,thread)
Xchar *string;
Xint size, thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  struct line_block *b = t->read_head;
X  if (b)
X  `7B
X    strcpy(string,b->line); strcat(string,"\n");
X    t->read_head = b->next;
X    free(b->line);
X    free(b);
X    return(0);
X  `7D
X  return(1);
X`7D
X
Xint write_net(string,thread)
Xchar *string;
Xint thread;
X`7B
X  struct thread_struct *t = &threads`5Bthread`5D;
X  struct line_block *b;
X  if (t->active==ACT_HEALTHY)
X  `7B
X    b = (struct line_block *) malloc(sizeof(struct line_block));
X    b->next = NULL;
X    b->line = strdup(string);
X    if (t->write_head)
X    `7B
X      t->write_tail->next = b;
X      t->write_tail = b;
X    `7D
X    else
X    `7B
X      t->write_tail = b;
X      t->write_head = b;
X    `7D
X  `7D
X`7D
X
Xint next_call(thread,func,type)
Xint thread;
Xint (*func)();
Xint type;
X`7B
X  threads`5Bthread`5D.next_function = func;
X  threads`5Bthread`5D.next_type = type;
X`7D
X
Xvoid getremhost(remhost,remuser,thread)
Xchar *remhost, *remuser;
Xint thread;
X`7B
Xstruct thread_struct *t = &threads`5Bthread`5D;
Xunsigned short auth_channel, auth_iosb`5B4`5D;
Xchar auth_line`5B1024`5D;
Xchar *cp, *dp;
Xstruct ib `7B
X    unsigned char fhost_len;
X    unsigned char lhost_len;
X    char fhost`5B128`5D;
X    short fill;
X    unsigned short fport;
X    short fill_1;
X    char lhost`5B128`5D;
X    unsigned short lport;
X    short fill_2;
X    unsigned int linet_addr;
X    unsigned int finet_addr;
X    `7D _align(quadword) info_buffer;
X  unsigned short iosb`5B4`5D;
X  int status;
X  strcpy(remuser,"nntp"); strcpy(t->remuser,"nntp");
X  strcpy(remhost,"unknown"); strcpy(t->remhost,"unknown");
X  /* IO$_MODIFY retrieves connection information from the active channel
X   * specified. See struct ib above for the format of the received message.
X   */
X  status = sys$qiow(0,threads`5Bthread`5D.channel,IO$_MODIFY,iosb,0,0,&info_
Vbuffer,
X                    sizeof(info_buffer),
X                    0,0,0,0);
X
X  if (!(status&1)) return;
X  if (!(iosb`5B0`5D&1)) return;
X  strncpy(remhost,info_buffer.fhost,info_buffer.fhost_len);
X  remhost`5Binfo_buffer.fhost_len`5D='\0';
X  strncpy(t->remhost,info_buffer.fhost,info_buffer.fhost_len);
X  t->remhost`5Binfo_buffer.fhost_len`5D='\0';
X  if (!*remhost)
X    sprintf(remhost,"%d.%d.%d.%d",
X                (info_buffer.finet_addr) & 0xff,
X                (info_buffer.finet_addr >> 8) & 0xff,
X                (info_buffer.finet_addr >> 16) & 0xff,
X                (info_buffer.finet_addr >> 24) & 0xff);
X    strcpy(t->remhost,remhost);
X  /* Attempt authentication */
X  status = sys$assign(&ipdsc,&auth_channel,0,0);
X  if (!(status&1)) return;
X  status = sys$qiow(0,auth_channel,IO$_CREATE,auth_iosb,0,0,remhost,AUTHPORT
V,
X                    0,1,0,AUTHTIMEOUT);
X  if (!(status&1)) return;
X  if (!(auth_iosb`5B0`5D&1)) return;
X  sprintf(auth_line,"%d, %d\r\n",info_buffer.fport, info_buffer.lport);
X  status = sys$qiow(0,auth_channel,IO$_WRITEVBLK,auth_iosb,0,0,
X                    auth_line,strlen(auth_line),0,0,0,0);
X  if (!((status&1) && (auth_iosb`5B0`5D&1)))
X  `7B
X    (void)sys$dassgn(auth_channel);
X    return;
X  `7D
X  status = sys$qiow(0,auth_channel,IO$_READVBLK,auth_iosb,0,0,auth_line,1023
V,
X                    0,0,0,0);
X  if (!((status&1) && (auth_iosb`5B0`5D&1)))
X  `7B
X    (void)sys$dassgn(auth_channel);
X    return;
X  `7D
X  auth_line`5Bauth_iosb`5B1`5D`5D='\0';
X  (void)sys$qiow(0,auth_channel,IO$_DELETE,auth_iosb,0,0,0,0,0,0,0,0);
X  (void)sys$dassgn(auth_channel);
X  cp = strchr(auth_line,':'); /* Really wonderful parsing here :-) */
X  if (!cp) return;
X  cp = strchr(++cp,':');
X  if (!cp) return;
X  cp = strchr(++cp,':');
X  if (!cp) return;
X  cp++;
X  while (*cp==' ') cp++;
X  dp = cp;
X  while (*cp!=' ' && *cp!='\r' && *cp!='\n') cp++;
X  *cp='\0';
X  strcpy(remuser,dp);
X  strcpy(t->remuser,dp);
X`7D
$ CALL UNPACK NNTP_TCPCMU_M.C;26 963328687
$ create 'f'
XNNTP_TCPCMU_M - a multi-threaded nntp server interface for CMU-TEK IP/TCP 6.
V5.
XVersion 1.1
X
XSome points to note:
X
X1) It would appear (although I haven't checked) that when a detached process
V is
Xcreated, ie. a WKS process, SYS$SCRATCH is not defined. Consequently, the
Xroutine sys_remote_send in NEWSDIST.C will fail and so will any attempt to
Xdistribute NNTP POSTed beyond the server's database. I've modified NEWSDIST.
VC
Xso that all references to SYS$SCRATCH (one of them) have been changed to
XNEWS_MANAGER. This is reasonable since file creation at these points is done
Xwith SYSPRV enabled.
X
X2) The server interface 'simulates' the 'declaration' of the process as a
Xnetwork object (c.f. DECnet and PSI) by always maintaining a passive listen
V on
Xport 119. On successful completion of the passive listen, the channel is han
Vded
Xto thread code and another passive listen is created. It is conceivable that
Xanother call could be made during the short period when a listen ISN'T in
Xprogress - this will cause the IP_ACP to fire-up another server process. I
Xhaven't added any code to check for multiple occurances of such processes (s
Vuch
Xas using locks, etc).
X
X3) Passive listens are maintained when MAXTHREADS is achieved. Calls are
Xaccepted so that an NNTP service failure message can be sent. The call is th
Ven
Xclosed and the passive listen re-established. This behaviour could be change
Vd
Xso that when MAXTHREADS is achieved, further passive listens are disabled,
Xenabling the IP_ACP to fire-off more server processes.
X
X4) Idle timeouts are used (IO$_CREATE parameter) to clear forgotten
Xconnections. The only problem with this is that an idle connection is closed
Xwithout being able to send an NNTP service failure message. So far, I haven'
Vt
Xencountered any problems with this - rtass and rrn (the clients mostly used
V at
Xour site) re-establish contact without problems. Also, clients are often jus
Vt
Xas rude, rather than sending a QUIT, they often tend to just abort the
Xconnection.
X
X5) The server interface buffers an entire transaction for each active thread
Xbecause of the presumed limitations of mixing calls to malloc and free in AS
VT
Xcontext (ie, I couldn't get it to work). Consequently, the stream of data
Xfollowing a POST command will be buffered on the heap until "." is received
V and
Xonly then will the next_function for that thread be called. Similarly, comma
Vnds
Xsuch as LIST will have their outputs buffered on the heap until their functi
Von
Xcompletes. Just make sure that the process has enough PGFLQUOTA.
X
X6) This code is free of debug and trace facilities. To debug this I tend to
Xrely entirely upon VAX DEBUG and VMXED. The code works well as far as I can
Xtell, errors are handled by abandoning the threads - no in-depth analysis of
Xexception codes is performed and no recovery is attempted within the confine
Vs
Xof the current thread.
X
X7) The server code is compiled and linked as for the single-threaded version
X(NNTP_TCPCMU.C).
X
X8) This is the line we have in our INTERNET.CONFIG for the multi-threaded
Xserver. It appears to suffice:
X
XWKS:119:NEWSRV:TCP$NEWSRV:NETWRK:NETMBX,TMPMBX,PHY_IO,SYSPRV,SYSLCK:\
XBYTLM=65535,BIOLM=32767,DIOLM=32767,ASTLM=200,ENQLM=100:\
XSYS$NULL:SYS$NULL:SYS$NULL:4:5
X
Xwhere TCP$NEWSRV is defined as NEWS_IMAGE:NNTP_TCPCMU_M.EXE
Xand   SYS$NULL is defined as NLA0:
XNote that SYSLCK privilege is needed due to rather strange behaviour on the
Xpart of one of our mail protocol handlers - you probably won't need this.
X
XI'd be very grateful if users of this code would send any fixes, improvement
Vs,
Xetc. to me as well as to the net or wherever.
X
XKeith
X--
XKeith Halewood           Janet: keith@uk.ac.liv.cs.and
XDept. Computer Science,  Internet: keith@and.cs.liv.ac.uk
XLiverpool University.
X"If they're the only survivors of a nuclear holocaust then they can't be in
Xvery good shape." - Beverly Crusher, ST:TNG
$ CALL UNPACK README.TXT;3 708847804
$ v=f$verify(v)
$ EXIT
--
Keith Halewood           Janet: keith@uk.ac.liv.cs.and
Dept. Computer Science,  Internet: keith%and.cs.liv.ac.uk@nsfnet-relay.ac.uk
Liverpool University.    UUCP: ..!mcsun!ukc!liv-cs!keith
"If they're the only survivors of a nuclear holocaust then they can't be in
very good shape." - Beverly Crusher, ST:TNG
