basic & pipelined transmit testcases

4 files changed, 145 insertions, 65 deletions

diff --git a/src/src/receive.c b/src/src/receive.c
index 9ff339d39..3b048252d 100644
--- a/src/src/receive.c
+++ b/src/src/receive.c
@@ -682,7 +682,8 @@ while ((ch = (receive_getc)()) != EOF)
     case 1:                         /* After written "\n" */
     if (ch == '.') { ch_state = 3; continue; }
     if (ch == '\r') { ch_state = 2; continue; }
-    if (ch != '\n') ch_state = 0; else linelength = -1;
+    if (ch == '\n') { body_linecount++; linelength = -1; }
+    else ch_state = 0;
     break;
 
     case 2:
diff --git a/src/src/structs.h b/src/src/structs.h
index 2a6ca68ab..e378a16a2 100644
--- a/src/src/structs.h
+++ b/src/src/structs.h
@@ -225,9 +225,13 @@ typedef struct transport_info {
 
 /* smtp transport datachunk callback */
 
+#define tc_reap_prev	BIT(0)	/* Flags: reap previous SMTP cmd responses */
+#define tc_reap_one	BIT(1)	/* reap one SMTP response */
+#define tc_chunk_last	BIT(2)	/* annotate chunk SMTP cmd as LAST */
+
 struct transport_context;
 typedef int (*tpt_chunk_cmd_cb)(int fd, struct transport_context * tctx,
-				unsigned len, BOOL last);
+				unsigned len, unsigned flags);
 
 /* Structure for information about a delivery-in-progress */
 
@@ -246,8 +250,11 @@ typedef struct transport_context {
   struct address_item	* first_addr;
   struct address_item	**sync_addr;
   BOOL			  pending_MAIL;
+  BOOL			  pending_BDAT;
+  BOOL			  good_RCPT;
   BOOL			* completed_address;
   int			  cmd_count;
+  uschar		* buffer;
 } transport_ctx;
 
 
diff --git a/src/src/transport.c b/src/src/transport.c
index 8c81e8a9e..e55e81f2f 100644
--- a/src/src/transport.c
+++ b/src/src/transport.c
@@ -423,7 +423,7 @@ for (ptr = start; ptr < end; ptr++)
     from previous SMTP commands. */
 
     if (tctx &&  tctx->options & topt_use_bdat  &&  tctx->chunk_cb)
-      if (tctx->chunk_cb(fd, tctx, (unsigned)len, FALSE) != OK)
+      if (tctx->chunk_cb(fd, tctx, (unsigned)len, tc_reap_prev|tc_reap_one) != OK)
 	return FALSE;
 
     if (!transport_write_block(fd, deliver_out_buffer, len))
@@ -843,8 +843,6 @@ static BOOL
 internal_transport_write_message(int fd, transport_ctx * tctx, int size_limit)
 {
 int len;
-off_t fsize;
-int size;
 
 /* Initialize pointer in output buffer. */
 
@@ -935,27 +933,51 @@ last BDAT, consisting of the current write_chunk() output buffer fill
 (optimally, all of the headers - but it does not matter if we already had to
 flush that buffer with non-last BDAT prependix) plus the amount of body data
 (as expanded for CRLF lines).  Then create and write the BDAT, and ensure
-that further use of write_chunk() will not prepend BDATs.  */
+that further use of write_chunk() will not prepend BDATs.
+The first BDAT written will also first flush any outstanding MAIL and RCPT
+commands which were buffered thans to PIPELINING.
+Commands go out (using a send()) from a different buffer to data (using a
+write()).  They might not end up in the same TCP segment, which is
+suboptimal. */
 
 if (tctx->options & topt_use_bdat)
   {
-  if ((size = chunk_ptr - deliver_out_buffer) < 0)
-    size = 0;
+  off_t fsize;
+  int hsize, size;
+
+  if ((hsize = chunk_ptr - deliver_out_buffer) < 0)
+    hsize = 0;
   if (!(tctx->options & topt_no_body))
     {
     if ((fsize = lseek(deliver_datafile, 0, SEEK_END)) < 0) return FALSE;
     fsize -= SPOOL_DATA_START_OFFSET;
     if (size_limit > 0  &&  fsize > size_limit)
       fsize = size_limit;
-    size += fsize;
+    size = hsize + fsize;
     if (tctx->options & topt_use_crlf)
       size += body_linecount;	/* account for CRLF-expansion */
     }
 
-  /*XXX CHUNKING:
-  Emit a LAST datachunk command. */
+  /* If the message is large, emit first a non-LAST chunk with just the
+  headers, and reap the command responses.  This lets us error out early
+  on RCPT rejects rather than sending megabytes of data.  Include headers
+  on the assumption they are cheap enough and some clever implementations
+  might errorcheck them too, on-the-fly, and reject that chunk. */
+
+  if (size > DELIVER_OUT_BUFFER_SIZE && hsize > 0)
+    {
+    if (  tctx->chunk_cb(fd, tctx, hsize, 0) != OK
+       || !transport_write_block(fd, deliver_out_buffer, hsize)
+       || tctx->chunk_cb(fd, tctx, 0, tc_reap_prev) != OK
+       )
+      return FALSE;
+    chunk_ptr = deliver_out_buffer;
+    size -= hsize;
+    }
+
+  /* Emit a LAST datachunk command. */
 
-  if (tctx->chunk_cb(fd, tctx, size, TRUE) != OK)
+  if (tctx->chunk_cb(fd, tctx, size, tc_chunk_last) != OK)
     return FALSE;
 
   tctx->options &= ~topt_use_bdat;
@@ -969,14 +991,19 @@ it, applying the size limit if required. */
 
 if (!(tctx->options & topt_no_body))
   {
+  int size = size_limit;
+
   nl_check_length = abs(nl_check_length);
   nl_partial_match = 0;
   if (lseek(deliver_datafile, SPOOL_DATA_START_OFFSET, SEEK_SET) < 0)
     return FALSE;
   while (  (len = MAX(DELIVER_IN_BUFFER_SIZE, size)) > 0
 	&& (len = read(deliver_datafile, deliver_in_buffer, len)) > 0)
+    {
     if (!write_chunk(fd, tctx, deliver_in_buffer, len))
       return FALSE;
+    size -= len;
+    }
 
   /* A read error on the body will have left len == -1 and errno set. */
 
diff --git a/src/src/transports/smtp.c b/src/src/transports/smtp.c
index 58a59433d..00274656c 100644
--- a/src/src/transports/smtp.c
+++ b/src/src/transports/smtp.c
@@ -282,6 +282,7 @@ static uschar *rf_names[] = { US"NEVER", US"SUCCESS", US"FAILURE", US"DELAY" };
 static uschar *smtp_command;   /* Points to last cmd for error messages */
 static uschar *mail_command;   /* Points to MAIL cmd for error messages */
 static BOOL    update_waiting; /* TRUE to update the "wait" database */
+static BOOL    pipelining_active; /* current transaction is in pipe mode */
 
 
 /*************************************************
@@ -510,13 +511,7 @@ static BOOL
 check_response(host_item *host, int *errno_value, int more_errno,
   uschar *buffer, int *yield, uschar **message, BOOL *pass_message)
 {
-uschar *pl = US"";
-
-if (smtp_use_pipelining &&
-    (Ustrcmp(smtp_command, "MAIL") == 0 ||
-     Ustrcmp(smtp_command, "RCPT") == 0 ||
-     Ustrcmp(smtp_command, "DATA") == 0))
-  pl = US"pipelined ";
+uschar * pl = pipelining_active ? US"pipelined " : US"";
 
 *yield = '4';    /* Default setting is to give a temporary error */
 
@@ -786,6 +781,7 @@ if (pending_MAIL)
   count--;
   if (!smtp_read_response(inblock, buffer, buffsize, '2', timeout))
     {
+    DEBUG(D_transport) debug_printf("bad response for MAIL\n");
     Ustrcpy(big_buffer, mail_command);  /* Fits, because it came from there! */
     if (errno == 0 && buffer[0] != 0)
       {
@@ -1363,57 +1359,95 @@ return checks;
 
 
 /* Callback for emitting a BDAT data chunk header.
-Flush any buffered SMTP commands first.
-Reap SMTP command responses if not the BDAT LAST.
 
-A nonlast request that is size zero is special-cased to only flush the
-command buffer and reap all outstanding responses.
+If given a nonzero size, first flush any buffered SMTP commands
+then emit the command.
+
+Reap previous SMTP command responses if requested.
+Reap one SMTP command response if requested.
 
 Returns:	OK or ERROR
 */
 
 static int
 smtp_chunk_cmd_callback(int fd, transport_ctx * tctx,
-  unsigned chunk_size, BOOL chunk_last)
+  unsigned chunk_size, unsigned flags)
 {
 smtp_transport_options_block * ob =
   (smtp_transport_options_block *)(tctx->tblock->options_block);
-uschar buffer[128];
+int cmd_count = 0;
+int prev_cmd_count;
+uschar * buffer = tctx->buffer;
 
-if (  (tctx->cmd_count = chunk_size == 0 && !chunk_last
 
-	/* Handle flush request */
-	?  smtp_write_command(tctx->outblock, FALSE, NULL)
+/* Write SMTP chunk header command */
 
-	/* Write SMTP chunk header command */
-	: smtp_write_command(tctx->outblock, FALSE, "BDAT %u%s\r\n",
-			      chunk_size, chunk_last ? " LAST" : "")
-      )
-    < 0)
-  return ERROR;
+if (chunk_size > 0)
+  if((cmd_count = smtp_write_command(tctx->outblock, FALSE, "BDAT %u%s\r\n",
+			      chunk_size,
+			      flags & tc_chunk_last ? " LAST" : "")
+     ) < 0) return ERROR;
+
+prev_cmd_count = cmd_count += tctx->cmd_count;
 
-if (chunk_last)
-  return OK;
+/* Reap responses for any previous, but not one we just emitted */
 
-/* Reap responses for this and any previous, and error out on failure */
-debug_printf("(look for %d responses)\n", tctx->cmd_count);
+if (chunk_size > 0)
+  prev_cmd_count--;
+if (tctx->pending_BDAT)
+  prev_cmd_count--;
 
-switch(sync_responses(tctx->first_addr, tctx->tblock->rcpt_include_affixes,
-	tctx->sync_addr, tctx->host, tctx->cmd_count,
-	ob->address_retry_include_sender,
-	tctx->pending_MAIL, 0,
-	tctx->inblock,
-	ob->command_timeout,
-	buffer, sizeof(buffer)))
+if (flags & tc_reap_prev  &&  prev_cmd_count > 0)
   {
-  case 1:				/* 2xx (only) => OK */
-  case 3: 				/* 2xx & 5xx => OK & progress made */
-  case 2: *tctx->completed_address = TRUE; /* 5xx (only) => progress made */
-  case 0: return OK;			/* No 2xx or 5xx, but no probs */
 
-  case -1:				/* Timeout on RCPT */
-  default: return ERROR;		/* I/O error, or any MAIL/DATA error */
+  switch(sync_responses(tctx->first_addr, tctx->tblock->rcpt_include_affixes,
+	  tctx->sync_addr, tctx->host, prev_cmd_count,
+	  ob->address_retry_include_sender,
+	  tctx->pending_MAIL, 0,
+	  tctx->inblock,
+	  ob->command_timeout,
+	  buffer, 4096))
+/*XXX buffer size! */
+    {
+    case 1:				/* 2xx (only) => OK */
+    case 3: tctx->good_RCPT = TRUE;	/* 2xx & 5xx => OK & progress made */
+    case 2: *tctx->completed_address = TRUE; /* 5xx (only) => progress made */
+    case 0: break;			/* No 2xx or 5xx, but no probs */
+
+    case -1:				/* Timeout on RCPT */
+    default: return ERROR;		/* I/O error, or any MAIL/DATA error */
+    }
+  cmd_count = 1;
+  if (!tctx->pending_BDAT)
+    pipelining_active = FALSE;
   }
+
+/* Reap response for the cmd we just emitted, or an outstanding BDAT */
+
+if (flags & tc_reap_one  ||  tctx->pending_BDAT)
+  {
+/*XXX buffer size! */
+  if (!smtp_read_response(tctx->inblock, buffer, 4096, '2',
+       ob->command_timeout))
+    {
+    if (errno == 0 && buffer[0] == '4')
+      {
+      errno = ERRNO_DATA4XX;	/*XXX does this actually get used? */
+      tctx->first_addr->more_errno |=
+	((buffer[1] - '0')*10 + buffer[2] - '0') << 8;
+      }
+    return ERROR;
+    }
+  cmd_count--;
+  tctx->pending_BDAT = FALSE;
+  pipelining_active = FALSE;
+  }
+else if (chunk_size > 0)
+  tctx->pending_BDAT = TRUE;
+
+
+tctx->cmd_count = cmd_count;
+return OK;
 }
 
 
@@ -2043,6 +2077,7 @@ if (continue_hostname == NULL
     case FAIL:		goto RESPONSE_FAILED;
     }
   }
+pipelining_active = smtp_use_pipelining;
 
 /* The setting up of the SMTP call is now complete. Any subsequent errors are
 message-specific. */
@@ -2399,6 +2434,7 @@ if (  !(peer_offered & PEER_OFFERED_CHUNKING)
     case -1: goto END_OFF;               /* Timeout on RCPT */
     default: goto RESPONSE_FAILED;       /* I/O error, or any MAIL/DATA error */
     }
+  pipelining_active = FALSE;
   }
 
 /* If there were no good recipients (but otherwise there have been no
@@ -2445,7 +2481,11 @@ else
     tctx.first_addr = first_addr;
     tctx.sync_addr = &sync_addr;
     tctx.pending_MAIL = pending_MAIL;
+    tctx.pending_BDAT = FALSE;
+    tctx.good_RCPT = ok;
     tctx.completed_address = &completed_address;
+    tctx.cmd_count = 0;
+    tctx.buffer = buffer;
     }
   else
     tctx.options |= topt_end_dot;
@@ -2453,6 +2493,11 @@ else
   /* Save the first address of the next batch. */
   first_addr = addr;
 
+  /* Responses from CHUNKING commands go in buffer.  Otherwise,
+  there has not been a response. */
+
+  buffer[0] = 0;
+
   sigalrm_seen = FALSE;
   transport_write_timeout = ob->data_timeout;
   smtp_command = US"sending data block";   /* For error messages */
@@ -2477,13 +2522,11 @@ else
   transport_write_timeout = 0;   /* for subsequent transports */
 
   /* Failure can either be some kind of I/O disaster (including timeout),
-  or the failure of a transport filter or the expansion of added headers. */
+  or the failure of a transport filter or the expansion of added headers.
+  Or, when CHUNKING, it can be a protocol-detected failure. */
 
   if (!ok)
-    {
-    buffer[0] = 0;              /* There hasn't been a response */
     goto RESPONSE_FAILED;
-    }
 
   /* We used to send the terminating "." explicitly here, but because of
   buffering effects at both ends of TCP/IP connections, you don't gain
@@ -2523,16 +2566,16 @@ else
     {
     ok = smtp_read_response(&inblock, buffer, sizeof(buffer), '3',
       ob->final_timeout);
-    if (!ok && errno == 0)
-      switch(buffer[0])
-        {
-	case '2': prdr_active = FALSE;
-	          ok = TRUE;
-		  break;
-	case '4': errno = ERRNO_DATA4XX;
-                  addrlist->more_errno |= ((buffer[1] - '0')*10 + buffer[2] - '0') << 8;
-		  break;
-        }
+    if (!ok && errno == 0) switch(buffer[0])
+      {
+      case '2': prdr_active = FALSE;
+		ok = TRUE;
+		break;
+      case '4': errno = ERRNO_DATA4XX;
+		addrlist->more_errno |=
+		  ((buffer[1] - '0')*10 + buffer[2] - '0') << 8;
+		break;
+      }
     }
   else
 #endif
@@ -2569,7 +2612,9 @@ else
     int delivery_time = (int)(time(NULL) - start_delivery_time);
     int len;
     uschar *conf = NULL;
+
     send_rset = FALSE;
+    pipelining_active = FALSE;
 
     /* Set up confirmation if needed - applies only to SMTP */