#module NNTP_MCMUTEK "V1.2"

/*

NNTP_TCPCMU_M - a multi-threaded nntp server interface for CMU-TEK IP/TCP 6.5.
Version 1.1

Some points to note:

1) It would appear (although I haven't checked) that when a detached process is
created, ie. a WKS process, SYS$SCRATCH is not defined. Consequently, the
routine sys_remote_send in NEWSDIST.C will fail and so will any attempt to
distribute NNTP POSTed beyond the server's database. I've modified NEWSDIST.C
so that all references to SYS$SCRATCH (one of them) have been changed to
NEWS_MANAGER. This is reasonable since file creation at these points is done
with SYSPRV enabled.

2) The server interface 'simulates' the 'declaration' of the process as a
network object (c.f. DECnet and PSI) by always maintaining a passive listen on
port 119. On successful completion of the passive listen, the channel is handed
to thread code and another passive listen is created. It is conceivable that
another call could be made during the short period when a listen ISN'T in
progress - this will cause the IP_ACP to fire-up another server process. I
haven't added any code to check for multiple occurances of such processes (such
as using locks, etc).

3) Passive listens are maintained when MAXTHREADS is achieved. Calls are
accepted so that an NNTP service failure message can be sent. The call is then
closed and the passive listen re-established. This behaviour could be changed
so that when MAXTHREADS is achieved, further passive listens are disabled,
enabling the IP_ACP to fire-off more server processes.

4) Idle timeouts are used (IO$_CREATE parameter) to clear forgotten
connections. The only problem with this is that an idle connection is closed
without being able to send an NNTP service failure message. So far, I haven't
encountered any problems with this - rtass and rrn (the clients mostly used at
our site) re-establish contact without problems. Also, clients are often just
as rude, rather than sending a QUIT, they often tend to just abort the
connection.

5) The server interface buffers an entire transaction for each active thread
because of the presumed limitations of mixing calls to malloc and free in AST
context (ie, I couldn't get it to work). Consequently, the stream of data
following a POST command will be buffered on the heap until "." is received and
only then will the next_function for that thread be called. Similarly, commands
such as LIST will have their outputs buffered on the heap until their function
completes. Just make sure that the process has enough PGFLQUOTA.

6) This code is free of debug and trace facilities. To debug this I tend to
rely entirely upon VAX DEBUG and VMXED. The code works well as far as I can
tell, errors are handled by abandoning the threads - no in-depth analysis of
exception codes is performed and no recovery is attempted within the confines
of the current thread.

7) The server code is compiled and linked as for the single-threaded version
(NNTP_TCPCMU.C).

8) This is the line we have in our INTERNET.CONFIG for the multi-threaded
server. It appears to suffice:

WKS:119:NEWSRV:TCP$NEWSRV:NETWRK:NETMBX,TMPMBX,PHY_IO,SYSPRV,SYSLCK:\
BYTLM=65535,BIOLM=32767,DIOLM=32767,ASTLM=200,ENQLM=100:\
SYS$NULL:SYS$NULL:SYS$NULL:4:5

where TCP$NEWSRV is defined as NEWS_IMAGE:NNTP_TCPCMU_M.EXE
and   SYS$NULL is defined as NLA0:
Note that SYSLCK privilege is needed due to rather strange behaviour on the
part of one of our mail protocol handlers - you probably won't need this.

I'd be very grateful if users of this code would send any fixes, improvements,
etc. to me as well as to the net or wherever.

Keith
--
Keith Halewood           Janet: keith@uk.ac.liv.cs.and
Dept. Computer Science,  Internet: keith@and.cs.liv.ac.uk
Liverpool University.
"If they're the only survivors of a nuclear holocaust then they can't be in
very good shape." - Beverly Crusher, ST:TNG

 *  9-Aug-1993  Charles Bailey  bailey@genetics.upenn.edu
 *   - added close_net() as part of revising News memory management
*/


/* Multithreaded NNTP server for ANU_NEWS V6.x and CMU-TEK TCP/IP V6.5
 * By Keith Halewood. The 'getremhost' routine has been stolen from the
 * original single-thread server interface.
 *
 * Changes V1.0 -> V1.1
 * 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).
 */

#define _NNTP_SERVER_DRIVER_C

#include "nntpinclude.h"

#define PORT       119     /* Should be 119, use 8888 for testing. */
#define AUTHPORT   113     /* Authentication WKS */
#define MAXTHREADS 16      /* Maximum number of services - Make sure
                            * process quotas are capable of handling 
                            * the number chosen.
                            */
#define TIMEOUT    1800    /* Connection inactivty timeout in seconds;
                            * causes an SS$_ABORT on reads and closes
                            * the connection. Not exactly in the spirit
                            * of the NNTP protocol, but most NNTP clients
                            * are just as ill-mannered.
                            */
#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.
                            */

/* 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 PORT
 */

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

/* I/O queue definition.
 */

struct line_block
  {
    struct line_block *next;
    char *line;
  };

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

volatile static struct thread_struct
  {
    int active;
    unsigned short channel;
    unsigned short write_iosb[4], read_iosb[4];
    char write_buffer[BUFSIZE-1], read_buffer[BUFSIZE-1];
    char read_residue[LINESIZE-1];
    struct line_block *read_head, *read_tail, *write_head, *write_tail;
    int write_in_progress;
    int next_type;
    int (*next_function)();
    char remhost[255];
    char remuser[32];
  } threads[MAXTHREADS+1];

$DESCRIPTOR(ipdsc,"INET$DEVICE");

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

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

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

#define MAXQUEUE 255                   /* Arbitrary but large */

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

static int enqueue(thread,event)
int thread, event;
{
  int result = 1;
  (void)sys$setast(0);
  if (queue_count == MAXQUEUE)
  {
    result = 0;
  }
  else
  {
    queue_count++;
    queue_thread_array[queue_head] = thread;
    queue_event_array[queue_head++] = event;
    if (queue_head==MAXQUEUE) queue_head = 0;
  }
  (void)sys$setast(1);
  return(result);
}

static int dequeue(thread,event)
int *thread, *event;
{
  int result = 1;
  (void)sys$setast(0);
  if (queue_count > 0)
  {
    *thread = queue_thread_array[queue_tail];
    *event = queue_event_array[queue_tail++];
    queue_count--;
    if (queue_tail==MAXQUEUE) queue_tail = 0;
  } else
  {
    result = 0;
  }
  (void)sys$setast(1);
  return(result);
}

/* Convenient way of duplicating a string
 */

char *strdup(s)
char *s;
{
  char *t = (char *)news_malloc(sizeof(char)*(strlen(s)+1));
  strcpy(t,s);
  return(t);
}

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

void ast_signal_open_complete(dummy)
int dummy;
{
  (void)enqueue(0,(listen_iosb[0]&1) ? EV_OPEN_COMPLETED : EV_OPEN_ABORTED);
  (void)sys$wake(0,0);
}

void ast_signal_read_complete(thread)
int thread;
{
  if (threads[thread].active==ACT_HEALTHY)
  {
    (void)enqueue(thread, (threads[thread].read_iosb[0]&1) ?
                          EV_READ_COMPLETED : EV_READ_ABORTED);
    (void)sys$wake(0,0);
  }
}

void ast_signal_write_complete(thread)
int thread;
{
  if (threads[thread].active==ACT_HEALTHY)
  {
    (void)enqueue(thread, (threads[thread].write_iosb[0]&1) ?
                          EV_WRITE_COMPLETED : EV_WRITE_ABORTED );
    (void)sys$wake(0,0);
  }
}

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

static void read_thread(thread)
int thread;
{
  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.
   */
  status = sys$qio(0,t->channel,IO$_READVBLK,t->read_iosb,
                   ast_signal_read_complete,thread,
                   t->read_buffer,BUFSIZE,0,0,0,0);
  if (!(status&1)) (void)enqueue(thread,EV_READ_ABORTED);
}

static void write_thread(thread)
int thread;
{
  struct thread_struct *t = &threads[thread];
  struct line_block *b = t->write_head;
  int status, iosize, s;
  if ((t->active==ACT_HEALTHY) && (t->write_in_progress==0))
  {
    if (b!=NULL)
    {
      iosize = 0;
      t->write_buffer[0] = '\0';
      while (b)
      {
        s = strlen(b->line);
        if (s+iosize > BUFSIZE-2) break;
        strcat(t->write_buffer,b->line);
        iosize = iosize + s;
        t->write_head = b->next;
        news_free(b->line);
        news_free(b);
        b = t->write_head;
      }
      /* IO$_WRITEVBLK writes a block of data (length in P2) to an
       * open channel as in I/O users guide.
       */
      status = sys$qio(0,t->channel,IO$_WRITEVBLK,t->write_iosb,
                       ast_signal_write_complete,thread,
                       t->write_buffer,iosize,0,0,0,0);
      t->write_in_progress = status&1;
      if (!(status&1)) (void)enqueue(thread,EV_WRITE_ABORTED);
    }
  }
}

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

static void init_thread(thread,channel)
int thread;
unsigned short channel;
{
  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;
  server_init_unit(thread);
  read_thread(thread);
  write_thread(thread);
}

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

static void init_listener()
{
  int status;
  active_listener = 0;
  status = sys$assign(&ipdsc,&listen_channel,0,0);
  if (!(status&1)) return;
  /* IO$_CREATE in this instance opens a passive listen on PORT
   * the listen as well as the future established connection on this
   * channel will timeout after TIMEOUT second's worth of inactivity
   * and send an SS$_ABORT to any queued I/O operations on the channel.
   */
  status = sys$qio(0,listen_channel,IO$_CREATE,listen_iosb,
           ast_signal_open_complete,0,0,0,PORT,0,0,TIMEOUT);
  if (!(status&1))
  {
    (void)enqueue(0,EV_OPEN_ABORTED);
  }
  else
  {
    active_listener = 1;
  }
}

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

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

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

static void enqueue_reader(thread,string)
int thread;
char *string;
{
  struct thread_struct *t = &threads[thread];
  struct line_block *b;
  if (t->active==ACT_HEALTHY)
  {
    b = (struct line_block *) news_malloc(sizeof(struct line_block));
    b->next = NULL;
    b->line = strdup(string);
    if (t->read_head!=NULL)
    {
      t->read_tail->next = b;
      t->read_tail = b;
    }
    else
    {
      t->read_tail = b;
      t->read_head = b;
    }
    if ((t->next_type==TYP_CMD_INPUT) ||
        ((t->next_type==TYP_FILE_INPUT) && !strcmp(string,".")))
    {
      (t->next_function)(thread);
      write_thread(thread);
    }
  }
}

static void abort_thread(thread)
int thread;
{
  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;
      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;
      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,".");
    if (t->next_type==TYP_CMD_INPUT) enqueue_reader(thread,"QUIT");
    log_to_file(thread);
    (void)sys$dassgn(t->channel);
    active_count--;
    /* if (active_count<=0) exit(1); */
  }
}

static void close_thread(thread)
int thread;
{
  struct thread_struct *t = &threads[thread];
  /* IO$_DELETE closes the connection after ensuring that all data
   * transfer in progress has reached its destinations.
   */
  t->active = ACT_UNASSIGNED;
  log_to_file(thread);
  (void)sys$qiow(0,t->channel,IO$_DELETE,t->write_iosb,0,0,0,0,0,0,0,0);
  (void)sys$dassgn(t->channel);
  active_count--;
  /* if (active_count<=0) exit(1); */
}


static void manage_open()
{
  int thread, status;
  unsigned short new_channel;
  char resource_msg[] = "400 Server resource limits exceeded.\r\n";
  active_listener = 0;
  new_channel = listen_channel;
  init_listener();
  thread = find_clear_thread();
  if (thread>0)
  {
    init_thread(thread,new_channel);
    active_count++;
  }
  else
  {
    status = sys$qiow(0,new_channel,IO$_WRITEVBLK,new_iosb,0,0,
                      resource_msg,strlen(resource_msg),0,0,0,0);
    status = sys$qiow(0,new_channel,IO$_DELETE,listen_iosb,0,0,0,0,0,0,0,0);
    (void)sys$dassgn(new_channel);
  }
}

static void manage_failed_open()
{
  (void)sys$dassgn(listen_channel);
  active_listener = 0;
  if (active_count<=0) exit(1);
  init_listener();
}

static void manage_read_completion(thread)
int thread;
{
  struct thread_struct *t = &threads[thread];
  char *cp;
  t->read_buffer[t->read_iosb[1]]='\0';
  strcat(t->read_residue,t->read_buffer);
  while (t->next_type!=TYP_NO_INPUT && 
           (cp = strchr(t->read_residue,'\n'))!=NULL)
  {
    *cp='\0'; if (cp[-1]=='\r') cp[-1]='\0'; cp++;
    enqueue_reader(thread,t->read_residue);
    strcpy(t->read_residue,cp);
  }
  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;
{
  struct thread_struct *t = &threads[thread];
  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()
{
  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;
    }
  }
}

main()
{
  init();
  for (;;)
  {
    drain_queue();
    (void)sys$hiber();
  }
}

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

int read_net(string,size,thread)
char *string;
int size, thread;
{
  struct thread_struct *t = &threads[thread];
  struct line_block *b = t->read_head;
  if (b)
  {
    strcpy(string,b->line); strcat(string,"\n");
    t->read_head = b->next;
    news_free(b->line);
    news_free(b);
    return(0);
  }
  return(1);
}

int write_net(string,thread)
char *string;
int thread;
{
  struct thread_struct *t = &threads[thread];
  struct line_block *b;
  if (t->active==ACT_HEALTHY)
  {
    b = (struct line_block *) news_malloc(sizeof(struct line_block));
    b->next = NULL;
    b->line = strdup(string);
    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.
 */

int close_net()
{
  int i;

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


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

void getremhost(remhost,remuser,thread)
char *remhost, *remuser;
int thread;
{
struct thread_struct *t = &threads[thread];
unsigned short auth_channel, auth_iosb[4];
char auth_line[1024];
char *cp, *dp;
struct ib {
    unsigned char fhost_len;
    unsigned char lhost_len;
    char fhost[128];
    short fill;
    unsigned short fport;
    short fill_1;
    char lhost[128];
    unsigned short lport;
    short fill_2;
    unsigned int linet_addr;
    unsigned int finet_addr;
    } _align(quadword) info_buffer;
  unsigned short iosb[4];
  int status;
  strcpy(remuser,"nntp"); strcpy(t->remuser,"nntp");
  strcpy(remhost,"unknown"); strcpy(t->remhost,"unknown");
  /* IO$_MODIFY retrieves connection information from the active channel
   * specified. See struct ib above for the format of the received message.
   */
  status = sys$qiow(0,threads[thread].channel,IO$_MODIFY,iosb,0,0,&info_buffer,
                    sizeof(info_buffer),
                    0,0,0,0);

  if (!(status&1)) return;
  if (!(iosb[0]&1)) return;
  strncpy(remhost,info_buffer.fhost,info_buffer.fhost_len);
  remhost[info_buffer.fhost_len]='\0';
  strncpy(t->remhost,info_buffer.fhost,info_buffer.fhost_len);
  t->remhost[info_buffer.fhost_len]='\0';
  if (!*remhost)
    sprintf(remhost,"%d.%d.%d.%d",
                (info_buffer.finet_addr) & 0xff,
                (info_buffer.finet_addr >> 8) & 0xff,
                (info_buffer.finet_addr >> 16) & 0xff,
                (info_buffer.finet_addr >> 24) & 0xff);
    strcpy(t->remhost,remhost);
  /* Attempt authentication */
  status = sys$assign(&ipdsc,&auth_channel,0,0);
  if (!(status&1)) return;
  status = sys$qiow(0,auth_channel,IO$_CREATE,auth_iosb,0,0,remhost,AUTHPORT,
                    0,1,0,AUTHTIMEOUT);
  if (!(status&1)) return;
  if (!(auth_iosb[0]&1)) return;
  sprintf(auth_line,"%d, %d\r\n",info_buffer.fport, info_buffer.lport);
  status = sys$qiow(0,auth_channel,IO$_WRITEVBLK,auth_iosb,0,0,
                    auth_line,strlen(auth_line),0,0,0,0);
  if (!((status&1) && (auth_iosb[0]&1)))
  {
    (void)sys$dassgn(auth_channel);
    return;
  }
  status = sys$qiow(0,auth_channel,IO$_READVBLK,auth_iosb,0,0,auth_line,1023,
                    0,0,0,0);
  if (!((status&1) && (auth_iosb[0]&1)))
  {
    (void)sys$dassgn(auth_channel);
    return;
  }
  auth_line[auth_iosb[1]]='\0';
  (void)sys$qiow(0,auth_channel,IO$_DELETE,auth_iosb,0,0,0,0,0,0,0,0);
  (void)sys$dassgn(auth_channel);
  cp = strchr(auth_line,':'); /* Really wonderful parsing here :-) */
  if (!cp) return;
  cp = strchr(++cp,':');
  if (!cp) return;
  cp = strchr(++cp,':');
  if (!cp) return;
  cp++;
  while (*cp==' ') cp++;
  dp = cp;
  while (*cp!=' ' && *cp!='\r' && *cp!='\n') cp++;
  *cp='\0';
  strcpy(remuser,dp);
  strcpy(t->remuser,dp);
}

