X-Git-Url: https://git.exim.org/exim.git/blobdiff_plain/f3ebb786e451da973560f1c9d8cdb151d25108b5..2791749f220602482c2cce772e6520c54218c0dd:/src/src/transport.c

diff --git a/src/src/transport.c b/src/src/transport.c
index ce02adbf2..c4f374509 100644
--- a/src/src/transport.c
+++ b/src/src/transport.c
@@ -172,6 +172,20 @@ for (transport_instance * t = transports; t; t = t->next)
 *             Write block of data                *
 *************************************************/
 
+static int
+tpt_write(int fd, uschar * block, int len, BOOL more, int options)
+{
+return
+#ifndef DISABLE_TLS
+  tls_out.active.sock == fd
+    ? tls_write(tls_out.active.tls_ctx, block, len, more) :
+#endif
+#ifdef MSG_MORE
+  more && !(options & topt_not_socket) ? send(fd, block, len, MSG_MORE) :
+#endif
+  write(fd, block, len);
+}
+
 /* Subroutine called by write_chunk() and at the end of the message actually
 to write a data block. Also called directly by some transports to write
 additional data to the file descriptor (e.g. prefix, suffix).
@@ -215,10 +229,11 @@ Returns:    TRUE on success, FALSE on failure (with errno preserved);
 */
 
 static BOOL
-transport_write_block_fd(transport_ctx * tctx, uschar *block, int len, BOOL more)
+transport_write_block_fd(transport_ctx * tctx, uschar * block, int len, BOOL more)
 {
 int rc, save_errno;
 int local_timeout = transport_write_timeout;
+int connretry = 1;
 int fd = tctx->u.fd;
 
 /* This loop is for handling incomplete writes and other retries. In most
@@ -230,48 +245,42 @@ for (int i = 0; i < 100; i++)
     debug_printf("writing data block fd=%d size=%d timeout=%d%s\n",
       fd, len, local_timeout, more ? " (more expected)" : "");
 
-  /* This code makes use of alarm() in order to implement the timeout. This
-  isn't a very tidy way of doing things. Using non-blocking I/O with select()
-  provides a neater approach. However, I don't know how to do this when TLS is
-  in use. */
+  /* When doing TCP Fast Open we may get this far before the 3-way handshake
+  is complete, and write returns ENOTCONN.  Detect that, wait for the socket
+  to become writable, and retry once only. */
 
-  if (transport_write_timeout <= 0)   /* No timeout wanted */
+  for(;;)
     {
-    rc =
-#ifndef DISABLE_TLS
-	tls_out.active.sock == fd ? tls_write(tls_out.active.tls_ctx, block, len, more) :
-#endif
-#ifdef MSG_MORE
-	more && !(tctx->options & topt_not_socket)
-	  ? send(fd, block, len, MSG_MORE) :
-#endif
-	write(fd, block, len);
-    save_errno = errno;
-    }
-
-  /* Timeout wanted. */
+    fd_set fds;
+    /* This code makes use of alarm() in order to implement the timeout. This
+    isn't a very tidy way of doing things. Using non-blocking I/O with select()
+    provides a neater approach. However, I don't know how to do this when TLS is
+    in use. */
 
-  else
-    {
-    ALARM(local_timeout);
-
-    rc =
-#ifndef DISABLE_TLS
-	tls_out.active.sock == fd ? tls_write(tls_out.active.tls_ctx, block, len, more) :
-#endif
-#ifdef MSG_MORE
-	more && !(tctx->options & topt_not_socket)
-	  ? send(fd, block, len, MSG_MORE) :
-#endif
-	write(fd, block, len);
-
-    save_errno = errno;
-    local_timeout = ALARM_CLR(0);
-    if (sigalrm_seen)
+    if (transport_write_timeout <= 0)   /* No timeout wanted */
       {
-      errno = ETIMEDOUT;
-      return FALSE;
+      rc = tpt_write(fd, block, len, more, tctx->options);
+      save_errno = errno;
+      }
+    else				/* Timeout wanted. */
+      {
+      ALARM(local_timeout);
+	rc = tpt_write(fd, block, len, more, tctx->options);
+	save_errno = errno;
+      local_timeout = ALARM_CLR(0);
+      if (sigalrm_seen)
+	{
+	errno = ETIMEDOUT;
+	return FALSE;
+	}
       }
+
+    if (rc >= 0 || errno != ENOTCONN || connretry <= 0)
+      break;
+
+    FD_ZERO(&fds); FD_SET(fd, &fds);
+    select(fd+1, NULL, &fds, NULL, NULL);	/* could set timout? */
+    connretry--;
     }
 
   /* Hopefully, the most common case is success, so test that first. */
@@ -1245,16 +1254,24 @@ if ((write_pid = fork()) == 0)
   rc = internal_transport_write_message(tctx, size_limit);
 
   save_errno = errno;
-  if (  write(pfd[pipe_write], (void *)&rc, sizeof(BOOL))
-        != sizeof(BOOL)
-     || write(pfd[pipe_write], (void *)&save_errno, sizeof(int))
-        != sizeof(int)
-     || write(pfd[pipe_write], (void *)&tctx->addr->more_errno, sizeof(int))
-        != sizeof(int)
-     || write(pfd[pipe_write], (void *)&tctx->addr->delivery_usec, sizeof(int))
-        != sizeof(int)
-     )
-    rc = FALSE;	/* compiler quietening */
+  errno = 0;
+  for (int retries = 10;;)
+    {
+    if (  os_pipe_write(pfd[pipe_write], (void *)&rc, sizeof(BOOL))
+	  != sizeof(BOOL)
+       || os_pipe_write(pfd[pipe_write], (void *)&save_errno, sizeof(int))
+	  != sizeof(int)
+       || os_pipe_write(pfd[pipe_write], (void *)&tctx->addr->more_errno, sizeof(int))
+	  != sizeof(int)
+       || os_pipe_write(pfd[pipe_write], (void *)&tctx->addr->delivery_usec, sizeof(int))
+	  != sizeof(int)
+       )
+      if (errno == EINTR && --retries > 0)
+	continue;
+      else
+	rc = FALSE;	/* compiler quietening */
+    break;
+    }
   exim_underbar_exit(0);
   }
 save_errno = errno;
@@ -1275,7 +1292,7 @@ if (write_pid < 0)
 
 /* When testing, let the subprocess get going */
 
-if (f.running_in_test_harness) millisleep(250);
+testharness_pause_ms(250);
 
 DEBUG(D_transport)
   debug_printf("process %d writing to transport filter\n", (int)write_pid);
@@ -1367,7 +1384,7 @@ if (write_pid > 0)
     if (rc == 0)
       {
       BOOL ok;
-      if (read(pfd[pipe_read], (void *)&ok, sizeof(BOOL)) != sizeof(BOOL))
+      if (os_pipe_read(pfd[pipe_read], (void *)&ok, sizeof(BOOL)) != sizeof(BOOL))
 	{
 	DEBUG(D_transport)
 	  debug_printf("pipe read from writing process: %s\n", strerror(errno));
@@ -1376,9 +1393,9 @@ if (write_pid > 0)
 	}
       else if (!ok)
         {
-	int dummy = read(pfd[pipe_read], (void *)&save_errno, sizeof(int));
-        dummy = read(pfd[pipe_read], (void *)&tctx->addr->more_errno, sizeof(int));
-        dummy = read(pfd[pipe_read], (void *)&tctx->addr->delivery_usec, sizeof(int));
+	int dummy = os_pipe_read(pfd[pipe_read], (void *)&save_errno, sizeof(int));
+        dummy = os_pipe_read(pfd[pipe_read], (void *)&tctx->addr->more_errno, sizeof(int));
+        dummy = os_pipe_read(pfd[pipe_read], (void *)&tctx->addr->delivery_usec, sizeof(int));
 	dummy = dummy;		/* compiler quietening */
         yield = FALSE;
         }
@@ -1722,16 +1739,14 @@ while (1)
   for (i = msgq_count - 1; i >= 0; --i) if (msgq[i].bKeep)
     {
     uschar subdir[2];
+    uschar * mid = msgq[i].message_id;
 
-    subdir[0] = split_spool_directory ? msgq[i].message_id[5] : 0;
-    subdir[1] = 0;
-
-    if (Ustat(spool_fname(US"input", subdir, msgq[i].message_id, US"-D"),
-	      &statbuf) != 0)
+    set_subdir_str(subdir, mid, 0);
+    if (Ustat(spool_fname(US"input", subdir, mid, US"-D"), &statbuf) != 0)
       msgq[i].bKeep = FALSE;
-    else if (!oicf_func || oicf_func(msgq[i].message_id, oicf_data))
+    else if (!oicf_func || oicf_func(mid, oicf_data))
       {
-      Ustrcpy_nt(new_message_id, msgq[i].message_id);
+      Ustrcpy_nt(new_message_id, mid);
       msgq[i].bKeep = FALSE;
       bFound = TRUE;
       break;
@@ -1946,7 +1961,7 @@ if ((pid = fork()) == 0)
     DEBUG(D_transport) debug_printf("transport_pass_socket succeeded (final-pid %d)\n", pid);
     _exit(EXIT_SUCCESS);
     }
-  if (f.running_in_test_harness) sleep(1);
+  testharness_pause_ms(1000);
 
   transport_do_pass_socket(transport_name, hostname, hostaddress,
     id, socket_fd);