Subject: Re: innxmit dumps core on exit
From: Mike Bird <[email-masked]>
To: Dave Barr <[email-masked]>
Date: Fri, 24 May 1996 20:30:09 -0700

Dave,

The patch following this message is a combination of the 03/15 version of
the robust/fast streaming patch together with the 05/20 bug fix.  It is
relative to unoff4 which already includes Jerry's enhanced streaming patch.
I'll be sending the patch directly from [email-masked] to
avoid line-wrap problems in Eudora.

It does not include the higher priority for handing off readers in the
control loop.  I'm hoping someone will post that soon.  If not I may try my
hand at it.

--Mike
==================================================
Here is the 03/15 inn1.4unoff4 version of the innxmit
robust/fast streaming patch as amended 05/20 to fix a
benign bug which could result in occasional spurious
"stalloc: internal error" messages.

This patch works around problems with some TCP implementations
(e.g. Linux 1.2.X) which cause innxmit to hang when streaming.
Even if the TCP stream freezes solid the -T timeout works.

It is also speeds up Jerry Aguirre's protocol somewhat by
increasing I/O overlap on the NNTP stream.

A small fix to prevent memory corruption with quoted printable
is included.

NOTE: Needs to be able to fopen STNBUF articles instead of
STNBUF/2 so you may need to reduce STNBUF.  (This is not a
problem for Linux 1.2.X with STNBUF=32.)

NOTE: Consider running this with a pre-sorting NNTPSEND
for maximum performance.  >8) 

--Mike
----------------------------------------------------------
 Mike Bird          Tel: 209-742-5000   FAX: 209-966-3177
 President          POP: 209-742-5333   NOC: 209-742-9979 
 Iron Mtn Systems      http://www.ims.mariposa.ca.us/

Date: Fri, 24 May 1996 20:24:43 -0700
From: root <[email-masked]>
To: [email-masked]
Subject: Robust/Fast Streaming Patch
Cc: [email-masked]

--- innxmit.c.SAFE	Fri May 24 20:05:56 1996
+++ innxmit.c	Fri May 24 20:06:07 1996
@@ -2,6 +2,7 @@
 **
 **  Transmit articles to remote site.
 **  Modified for NNTP streaming: 3Jan96 Jerry Aguirre
+**  Modified for "non-blocking" NNTP streaming: 8Feb96 Mike Bird
 */
 #include "configdata.h"
 #include <stdio.h>
@@ -46,6 +47,7 @@
 */
 
 /* max number of articles that can be streamed ahead */
+/* must be able to fopen this many articles */
 #define STNBUF 32
 
 /* Send "takethis" without "check" if this many articles were
@@ -53,13 +55,11 @@
 */
 #define STNC 16
 
-/* typical number of articles to stream  */
-/* must be able to fopen this many articles */
-#define STNBUFL (STNBUF/2)
-
 /* number of retries before requeueing to disk */
 #define STNRETRY 5
 
+enum stbufmode { Empty=0, Idle, In_Flight, Do_Check, Do_Takethis };
+
 struct stbufs {		/* for each article we are procesing */
 	char *st_fname;		/* file name */
 	char *st_id;		/* message ID */
@@ -67,9 +67,11 @@
 	int   st_age;		/* age count */
 	QIOSTATE *st_qp;	/* IO to read article contents */
 	int   st_hash;		/* hash value to speed searches */
+	enum stbufmode st_mode; /* mode (see above) */
 };
 static struct stbufs stbuf[STNBUF]; /* we keep track of this many articles */
 static int stnq;	/* current number of active entries in stbuf */
+static int stnflight;   /* current number of entries in-flight */
 static long stnofail;	/* Count of consecutive successful sends */
 
 static int TryStream = TRUE;	/* Should attempt stream negotation? */
@@ -214,12 +216,16 @@
 STATIC BOOL
 REMflush()
 {
-    int		i;
+    STATIC BOOL stream_write(char *p, int i);
+    BOOL	b;
 
     if (REMbuffptr == REMbuffer) return TRUE; /* nothing buffered */
-    i = xwrite(ToServer, REMbuffer, (int)(REMbuffptr - REMbuffer));
+    if (CanStream)
+      b = stream_write(REMbuffer, (int)(REMbuffptr - REMbuffer));
+    else
+      b = (xwrite(ToServer, REMbuffer, (int)(REMbuffptr - REMbuffer)) >= 0);
     REMbuffptr = REMbuffer;
-    return i < 0 ? FALSE : TRUE;
+    return b;
 }
 
 /*
@@ -235,7 +241,7 @@
     register int i;
 
     for (i = 0; i < STNBUF; i++) { /* linear search for ID */
-	if ((stbuf[i].st_id) && (stbuf[i].st_id[0])
+	if (stbuf[i].st_mode != Empty
 	 && (stbuf[i].st_hash == hash)) {
 	    register int n;
 
@@ -282,7 +288,7 @@
     register int i;
 
     for (i = 0; i < STNBUF; i++) {
-	if ((!stbuf[i].st_fname) || (stbuf[i].st_fname[0] == '\0')) break;
+	if (stbuf[i].st_mode == Empty) break;
     }
     if (i >= STNBUF) { /* stnq says not full but can not find unused */
 	syslog(L_ERROR, "stalloc: Internal error");
@@ -292,18 +298,13 @@
 	syslog(L_ERROR, "stalloc: filename longer than %d", SPOOLNAMEBUFF);
 	return (-1);
     }
-    /* allocate buffers on first use.
-    ** If filename ever is longer than SPOOLNAMEBUFF then code will abort.
-    ** If ID is ever longer than NNTP_STRLEN then other code would break.
-    */
-    if (!stbuf[i].st_fname) stbuf[i].st_fname = NEW(char, SPOOLNAMEBUFF);
-    if (!stbuf[i].st_id) stbuf[i].st_id = NEW(char, NNTP_STRLEN);
     (void)strcpy(stbuf[i].st_fname, Article);
     (void)strcpy(stbuf[i].st_id, MessageID);
     stbuf[i].st_qp = qp;
     stbuf[i].st_hash = hash;
     stbuf[i].st_retry = 0;
     stbuf[i].st_age = 0;
+    stbuf[i].st_mode = Idle;
     stnq++;
     return i;
 }
@@ -317,8 +318,7 @@
 	    QIOclose(stbuf[i].st_qp);
 	    stbuf[i].st_qp = 0;
 	}
-	if (stbuf[i].st_id) stbuf[i].st_id[0] = '\0';
-	if (stbuf[i].st_fname) stbuf[i].st_fname[0] = '\0';
+	stbuf[i].st_mode = Empty;
 	stnq--;
 }
 
@@ -391,12 +391,12 @@
     register int	prev;
 
     /* Buffer too full? */
-    if (REMbuffend - REMbuffptr < i + 3) {
+    if (REMbuffend - REMbuffptr < i*3 + 3) {
 	if (!REMflush())
 	    return FALSE;
-	if (REMbuffend - REMbuffer < i + 3) {
+	if (REMbuffend - REMbuffer < i*3 + 3) {
 	    /* Line too long -- grow buffer. */
-	    size = i * 2;
+	    size = i*3 * 2;
 	    RENEW(REMbuffer, char, size);
 	    REMbuffend = &REMbuffer[size];
 	}
@@ -449,6 +449,9 @@
     double		usertime;
     double		systime;
 
+    STATIC BOOL stream_term();
+    stream_term();
+
     if (!Purging) {
 	(void)REMwrite(QUIT, STRLEN(QUIT), FALSE);
 	(void)REMflush();
@@ -566,10 +569,11 @@
 	register int i;
 
 	for (i = 0; i < STNBUF; i++) {    /* requeue unacknowledged articles */
-	    if ((stbuf[i].st_fname) && (stbuf[i].st_fname[0] != '\0')) {
+	    if (stbuf[i].st_mode != Empty) {
 		if (Debug)
-		    (void)fprintf(stderr, "stbuf[%d]= %s, %s\n",
-			    i, stbuf[i].st_fname, stbuf[i].st_id);
+		    (void)fprintf(stderr, "stbuf[%d](%c)= %s, %s\n",
+			    i, "EIFCT"[stbuf[i].st_mode],
+			    stbuf[i].st_fname, stbuf[i].st_id);
 		Requeue(stbuf[i].st_fname, stbuf[i].st_id);
 		if (Article == stbuf[i].st_fname) Article = NULL;
 		strel(i); /* release entry */
@@ -955,6 +959,8 @@
 	return TRUE;
     }
     STAToffered++;
+    stbuf[i].st_mode = In_Flight;
+    stnflight++;
     if (Debug) {
 	if (stbuf[i].st_retry)
 	    (void)fprintf(stderr, "> %s (retry %d)\n", buff, stbuf[i].st_retry);
@@ -1005,6 +1011,8 @@
     if (!REMsendarticle(stbuf[i].st_fname, stbuf[i].st_id,
 	    stbuf[i].st_qp))
 	return TRUE;
+    stbuf[i].st_mode = In_Flight;
+    stnflight++;
     QIOclose(stbuf[i].st_qp);	/* should not need file again */
     stbuf[i].st_qp = 0;		/* so close to free descriptor */
     stbuf[i].st_age = 0;
@@ -1013,33 +1021,28 @@
 }

-/* listen for responses.  Process acknowledgments to remove items from
-** the queue.  Also sends the articles on request.  Returns TRUE on error.
+/* Process acknowledgments to remove items from
+** the queue.  Also sends the articles on request. 
 ** return TRUE on failure.
 */
 STATIC BOOL
-strlisten()
+strlisten(buff, writing_p)
+     char *buff;
+     BOOL writing_p;
 {
     int		resp;
-    int		i;
+    int		i, j;
     char	*id, *p;
-    char	buff[NNTP_STRLEN];
     int		hash;
 
-    while(TRUE) {
-	if (!REMread(buff, (int)sizeof buff)) {
-	    (void)fprintf(stderr, "No reply to check, %s\n", strerror(errno));
-	    return TRUE;
-	}
-	if (GotInterrupt)
-	    Interrupted((char *)0, (char *)0);
+    {
 	if (Debug)
 	    (void)fprintf(stderr, "< %s", buff);
 
 	/* Parse the reply. */
 	resp =  atoi(buff);
 	/* Skip the 1XX informational messages */
-	if ((resp >= 100) && (resp < 200)) continue;
+	if ((resp >= 100) && (resp < 200)) return FALSE;
 	switch (resp) { /* first time is to verify it */
 	case NNTP_ERR_GOTID_VAL:
 	case NNTP_OK_SENDID_VAL:
@@ -1055,6 +1058,8 @@
 		    syslog(L_NOTICE, CANT_FINDIT, REMhost, REMclean(buff));
 		    return (TRUE); /* can't find it! */
 		}
+		stbuf[i].st_mode = Idle;
+		--stnflight;
 	    } else {
 		syslog(L_NOTICE, CANT_PARSEIT, REMhost, REMclean(buff));
 		return (TRUE);
@@ -1076,7 +1081,10 @@
 	case NNTP_RESENDID_VAL:	/* remote wants it later */
 	    /* try again now because time has passed */
 	    if (stbuf[i].st_retry < STNRETRY) {
-		if (check(i)) return TRUE;
+	        if (writing_p)
+		  stbuf[i].st_mode = Do_Check;
+		else if (check(i))
+		  return TRUE;
 		stbuf[i].st_retry++;
 		stbuf[i].st_age = 0;
 	    } else { /* requeue to disk for later */
@@ -1091,7 +1099,10 @@
 	    break;
 		
 	case NNTP_OK_SENDID_VAL:	/* remote wants article */
-	    if (takethis(i)) return TRUE;
+	    if (writing_p)
+	      stbuf[i].st_mode = Do_Takethis;
+	    else if (takethis(i))
+	      return TRUE;
 	    stnofail++;
 	    break;
 
@@ -1106,8 +1117,9 @@
 	    stnofail = 0;
 	    break;
 	}
-	break;
     }
+    if (Debug)
+      (void)fprintf(stderr, " ... OK\n");
     return (FALSE);
 }
 
@@ -1147,6 +1159,9 @@
     unsigned int	TotalTimeout;
     extern char		*nntp_port;
 
+    STATIC BOOL stream_init();
+    STATIC BOOL stream_process(BOOL writing_p);
+
     /* Set defaults. */
     ConnectTimeout = 0;
     TotalTimeout = 0;
@@ -1338,11 +1353,18 @@
 		int i;
 
 		for (i = 0; i < STNBUF; i++) { /* reset buffers */
-		    stbuf[i].st_fname = 0;
-		    stbuf[i].st_id = 0;
+		    /*
+		    ** If filename ever is longer than SPOOLNAMEBUFF then code will abort.
+		    ** If ID is ever longer than NNTP_STRLEN then other code would break.
+		    */
+		    stbuf[i].st_fname = NEW(char, SPOOLNAMEBUFF);
+		    stbuf[i].st_id = NEW(char, NNTP_STRLEN);
 		    stbuf[i].st_qp = 0;
+		    stbuf[i].st_mode = Empty;
 		}
 		stnq = 0;
+		stnflight = 0;
+		stream_init();
 	    }
 	}
     }
@@ -1505,16 +1527,19 @@
 		QIOclose(qp);
 		continue;
 	    }
-	    /* This code tries to optimize by sending a burst of "check"
-	     * commands before flushing the buffer.  This should result
-	     * in several being sent in one packet reducing the network
-	     * overhead.
-	     */
-	    if (DoCheck && (stnofail < STNC)) lim = STNBUF;
-	    else                              lim = STNBUFL;
-	    if (stnq >= lim) { /* need to empty a buffer */
-		while (stnq >= STNBUFL) { /* or several */
-		    if (strlisten()) {
+	    /* 18May96 [email-masked] fixed bug identified by
+	       [email-masked] which caused stalloc:Internal error
+	       if alarm went off while all buffers were in use */
+	    while (stnq >= STNBUF) { /* need to empty a buffer */
+	        {
+		    if (GotAlarm) {
+		        (void)fprintf(stderr, "Timed out\n");
+		        /* Don't resend the current article. */
+		        RequeueRestAndExit((char *)NULL, (char *)NULL);
+		    }
+		    if (GotInterrupt)
+		        Interrupted(Article, MessageID);
+		    if (!stream_process(FALSE)) {
 			RequeueRestAndExit(Article, MessageID);
 		    }
 		}
@@ -1534,24 +1559,6 @@
 		    RequeueRestAndExit((char *)NULL, (char *)NULL);
 		}
 	    }
-	    /* check for need to resend any IDs */
-	    for (i = 0; i < STNBUF; i++) {
-		if ((stbuf[i].st_fname) && (stbuf[i].st_fname[0] != '\0')) {
-		    if (stbuf[i].st_age++ > stnq) {
-			/* This should not happen but just in case ... */
-			if (stbuf[i].st_retry < STNRETRY) {
-			    if (check(i)) /* resend check */
-				RequeueRestAndExit((char *)NULL, (char *)NULL);
-			    retries++;
-			    stbuf[i].st_retry++;
-			    stbuf[i].st_age = 0;
-			} else { /* requeue to disk for later */
-			    Requeue(stbuf[i].st_fname, stbuf[i].st_id);
-			    strel(i); /* release entry */
-			}
-		    }
-		}
-	    }
 	    continue; /* next article */
 	}
 	(void)sprintf(buff, "%s %s", Slavish ? "xreplic" : "ihave", MessageID);
@@ -1610,8 +1617,8 @@
 	QIOclose(qp);
     }
     if (CanStream) { /* need to wait for rest of ACKs */
-	while (stnq > 0) {
-	    if (strlisten()) {
+	while (stnq > 0 && !GotInterrupt && !GotAlarm) {
+	    if (!stream_process(FALSE)) {
 		RequeueRestAndExit((char *)NULL, (char *)NULL);
 	    }
 	}
@@ -1625,4 +1632,174 @@
 		BATCHtemp, strerror(errno));
     ExitWithStats(0);
     /* NOTREACHED */
+}
+
+/*
+ * MGB: Here is a "non-blocking" I/O interface for streaming
+ *      Beware that it can't write from within a write
+ */
+
+STATIC char stream_command_buf[BUFSIZ];
+
+STATIC char stream_read_buf[BUFSIZ];
+STATIC char *stream_read_ptr;
+STATIC int stream_read_left;
+
+STATIC char *stream_write_ptr;
+STATIC int stream_write_left;
+
+STATIC BOOL
+stream_init()
+{
+  if (SetNonBlocking(FromServer) < 0 || SetNonBlocking(ToServer) < 0)
+    return FALSE;
+  stream_read_ptr = stream_read_buf;
+  stream_read_left = BUFSIZ-2;
+  stream_write_left = 0;
+}
+
+STATIC void
+stream_term()
+{}
+
+STATIC BOOL
+stream_respond()
+{
+  BOOL busy_p;
+  do {
+    int i;
+    busy_p = FALSE;
+    for (i = 0; i < STNBUF; ++i) {
+      if (GotInterrupt || GotAlarm)
+	break;
+      switch (stbuf[i].st_mode) {
+      case Do_Check:
+	if (check(i))
+	  return FALSE;
+	busy_p = TRUE;
+	break;
+      case Do_Takethis:
+	if (takethis(i))
+	  return FALSE;
+	busy_p = TRUE;
+	break;
+      default: break;
+      }
+    }
+  } while (busy_p);
+  return TRUE;
+}
+
+STATIC BOOL
+stream_process(writing_p)
+     BOOL writing_p;
+{
+  int max_fd, i;
+  fd_set inp_set, out_set, exc_set;
+  struct timeval timeout;
+  if (!writing_p && (!stream_respond() || !REMflush()))
+    return FALSE;
+  max_fd = FromServer;
+  if (ToServer > max_fd)
+    max_fd = ToServer;
+  FD_ZERO(&inp_set);
+  FD_ZERO(&out_set);
+  FD_ZERO(&exc_set);
+  FD_SET(FromServer, &exc_set);
+  FD_SET(FromServer, &inp_set);
+  FD_SET(ToServer, &exc_set);
+  if (stream_write_left)
+    FD_SET(ToServer, &out_set);
+  timeout.tv_sec = 10;
+  timeout.tv_usec = 0;
+  if (Debug)
+    fprintf(stderr, "stream_process::select R%c, queued=%d, buffered=%dB, in-flight=%d\n",
+	    writing_p ? 'W' : 'O', stnq-stnflight, stream_write_left, stnflight);
+  if ((i = select(max_fd+1, &inp_set, &out_set, &exc_set, &timeout)) < 0)
+    return FALSE;
+
+  /* INTERRUPT OR TIMEOUT */
+      
+  if (i == 0 || GotInterrupt || GotAlarm)
+    return TRUE;
+    
+  /* PROBLEM CONNECTION? */
+      
+  if (FD_ISSET(FromServer, &exc_set) || FD_ISSET(ToServer, &exc_set))
+    return FALSE;
+    
+  /* INPUT AVAILABLE? */
+      
+  if (FD_ISSET(FromServer, &inp_set)) {
+    int i, m, n;
+    char *p, *q;
+    if (stream_read_left == 0) {
+      /* Overlong response - continually overwrite last character */
+      stream_read_ptr--;
+      stream_read_left = 1;
+    }
+    i = read(FromServer, stream_read_ptr, stream_read_left);
+    if (i < 0 && i != EAGAIN)
+      return FALSE;
+    if (i > 0) {
+      stream_read_ptr += i;
+      stream_read_left -= i;
+      *stream_read_ptr = '\0';
+      if (Debug)
+	fprintf(stderr, "read (%d) %s\n", i, "" /*stream_read_ptr-i*/);
+      while ((p = strchr(stream_read_buf, '\n')) != NULL) {
+	if (p > stream_read_buf && p[-1] == '\r')
+	  p[-1] = '\0';
+	*p = '\0';
+	m = (p+1)-stream_read_buf;
+	/* Handle the dot escape. */
+	if (stream_read_buf[0] == '.') {
+	  if (stream_read_buf[1] == '\0')
+	    /* EOF. */
+	    return FALSE;
+	  memmove(stream_command_buf, stream_read_buf+1, m-1);
+	} else
+	  memmove(stream_command_buf, stream_read_buf, m);
+	m = (p+1)-stream_read_buf;
+	n = BUFSIZ-m;
+	memmove(stream_read_buf, p+1, n);
+	stream_read_ptr -= m;
+	stream_read_left += m;
+	if (strlisten(stream_command_buf, writing_p))
+	  return FALSE;
+      }
+    }
+  }
+  
+    
+  /* OUTPUT POSSIBLE? */
+      
+  if (FD_ISSET(ToServer, &out_set) && stream_write_left) {
+    int i = write(ToServer, stream_write_ptr, stream_write_left);
+    if (i < 0 && errno != EAGAIN)
+      return FALSE;
+    if (i > 0) {
+      if (Debug)
+	fprintf(stderr, "wrote %d\n", i);
+      stream_write_ptr += i;
+      stream_write_left -= i;
+    }
+  }
+
+  return TRUE;
+}
+
+STATIC BOOL
+stream_write(p, i)
+  char	*p;
+  int	i;
+{
+  if (Debug)
+    fprintf(stderr, "stream_write (%d) %s\n", i, "" /*p*/);
+  stream_write_ptr = p;
+  stream_write_left = i;
+  while (stream_write_left && !GotInterrupt && !GotAlarm)
+    if (!stream_process(TRUE))
+      return FALSE;
+  return TRUE;
 }