Plain sockets: a send cut short by the send timeout ends the write; TestClientStalledReader measures the session's end

NServer-0.9.1/checks/cReplyBudget.cc

@@ -1,9 +1,12 @@
 #include <iostream>
 #include <sstream>
 #include <fcntl.h>
+#include <string.h>
 #include <signal.h>
 #include <sys/socket.h>
 #include <sys/wait.h>
+#include <netinet/in.h>
+#include <arpa/inet.h>
 #include <unistd.h>
 #include "Logger.h"
 #include "ReplyBudget.h"
@@ -97,6 +100,70 @@ static void client_connection_gone()
 	close(fd[1]);
 }
 
+// When the send timeout runs out after send() has taken part of the data,
+// Linux and FreeBSD return that part instead of EAGAIN. The rest must not
+// wait for a second timeout: the write to a peer that reads nothing fails
+// after one (it took two, and a stalled client's session two
+// ClientTimeouts).
+static void short_send_ends_write()
+{
+	int l = socket(AF_INET, SOCK_STREAM, 0);
+	struct sockaddr_in sa;
+	socklen_t salen = sizeof(sa);
+	memset(&sa, 0, sizeof(sa));
+	sa.sin_family = AF_INET;
+	sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+	int small = 4096;
+	// (a small window, inherited by the accepted socket)
+	setsockopt(l, SOL_SOCKET, SO_RCVBUF, &small, sizeof(small));
+	if (l < 0 || bind(l, (struct sockaddr *) &sa, sizeof(sa)) != 0 ||
+	    listen(l, 1) != 0 ||
+	    getsockname(l, (struct sockaddr *) &sa, &salen) != 0) {
+		std::cerr << "short_send_ends_write: no loopback socket, skipped\n";
+		if (l >= 0)
+			close(l);
+		return;
+	}
+	// (on a failure no accept(): it would wait for good)
+	int c = socket(AF_INET, SOCK_STREAM, 0);
+	if (c < 0 || connect(c, (struct sockaddr *) &sa, sizeof(sa)) != 0) {
+		CHECK(!"connect to the loopback listener");
+		if (c >= 0)
+			close(c);
+		close(l);
+		return;
+	}
+	int a = accept(l, NULL, NULL);
+	if (a < 0) {
+		CHECK(!"accept");
+		close(c);
+		close(l);
+		return;
+	}
+	{
+		sockstream stream;
+		stream.attach(c);
+		stream.set_io_timeout(1);	// SO_SNDTIMEO 1 s
+		const std::string chunk(1000, 'x');
+		struct timespec before, after;
+		clock_gettime(CLOCK_MONOTONIC, &before);
+		for (int i = 0; i < 1000000 && stream.good(); ++i)
+			stream << chunk;
+		stream.flush();
+		clock_gettime(CLOCK_MONOTONIC, &after);
+		double elapsed = after.tv_sec - before.tv_sec +
+		    (after.tv_nsec - before.tv_nsec) / 1000000000.0;
+		std::cerr << "short_send_ends_write: the write failed after "
+		    << elapsed << " s\n";
+		CHECK(!stream.good());
+		// one send timeout (1 s), not two
+		CHECK(elapsed < 1.5);
+		stream.abort();
+	}
+	close(a);
+	close(l);
+}
+
 int main()
 {
 #if defined(NEWSCACHE_TEST_SHORT_REPLY_LIMITS)
@@ -154,5 +221,6 @@ int main()
 	CHECK(line == "abc" && wire == 3 && partial.eof());
 	partial_line_deadline();
 	client_connection_gone();
+	short_send_ends_write();
 	return failures ? 1 : 0;
 }

NServer-0.9.1/sockstream.cc

@@ -568,6 +568,39 @@ int sockstream::sockstreambuf::overflow(int c)
   return c;
 }
 
+// After a send that began at t0 returned short: waits until the socket
+// can take more, for what is left of its send timeout (SO_SNDTIMEO:
+// ClientTimeout for a client, set on its socket, or the I/O timeout).
+// False when nothing is left or the socket stays full: the client isn't
+// reading. Without a send timeout, true (the next send waits as before).
+bool sockstream::sockstreambuf::wait_rest_of_send_timeout(const struct timespec &t0)
+{
+  struct timeval tv;
+  socklen_t len = sizeof(tv);
+  if (getsockopt(_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, &len) != 0 ||
+      (tv.tv_sec == 0 && tv.tv_usec == 0))
+    return true;
+  const long long limit = (long long) tv.tv_sec * 1000 + tv.tv_usec / 1000;
+  int rc;
+  do
+  {
+    struct timespec t1;
+    if (interrupted() || clock_gettime(CLOCK_MONOTONIC, &t1) != 0)
+      return false;
+    const long long left = limit -
+      (((long long) t1.tv_sec - t0.tv_sec) * 1000 +
+       ((long long) t1.tv_nsec - t0.tv_nsec) / 1000000);
+    if (left <= 0)
+      return false;
+    struct pollfd p;
+    p.fd = _fd;
+    p.events = POLLOUT;
+    p.revents = 0;
+    rc = poll(&p, 1, left > INT_MAX ? INT_MAX : (int) left);
+  } while (rc < 0 && errno == EINTR);
+  return rc > 0;
+}
+
 int sockstream::sockstreambuf::sync()
 {
   size_t off = 0;
@@ -582,9 +615,13 @@ int sockstream::sockstreambuf::sync()
     {
       return -1;
     }
+    const size_t want = pptr() - pbase() - off;
+    struct timespec t0;
+    bool timed;
     do
     {
-      rc_send = ::send(_fd, pbase() + off, pptr() - pbase() - off, 0);
+      timed = clock_gettime(CLOCK_MONOTONIC, &t0) == 0;
+      rc_send = ::send(_fd, pbase() + off, want, 0);
     } while ((rc_send < 0) && (errno == EINTR) &&
 	     !interrupted());
 
@@ -598,6 +635,22 @@ int sockstream::sockstreambuf::sync()
     }
 
     off += rc_send;
+
+    // A blocking send returns less than asked only when it was cut
+    // short: by a signal, or by the send timeout (SO_SNDTIMEO) after it
+    // had taken part of the data, which FreeBSD returns as a short count
+    // instead of EAGAIN. The rest is sent only if the socket can take
+    // more within what is left of the timeout; after the whole timeout
+    // the client isn't reading, and that is final too. Sending the rest
+    // at once waited for a second timeout: a session stalled for two
+    // ClientTimeouts instead of one. (Not during a reply deadline: the
+    // socket is non-blocking then, and a short count comes at once.)
+    if ((size_t) rc_send < want && timed && !_reply_deadline_active &&
+	!wait_rest_of_send_timeout(t0))
+    {
+      setp(_pbuf, _pbuf + sizeof(_pbuf));
+      return -1;
+    }
   }
 
   setp(_pbuf, _pbuf + sizeof(_pbuf));

NServer-0.9.1/sockstream.h

@@ -109,6 +109,7 @@ class sockstream
     char _pbuf[4096];
 
     void apply_io_timeout();
+    bool wait_rest_of_send_timeout(const struct timespec &t0);
 
     // seconds wait_io() waits for reading / writing, 0 = _io_timeout
     unsigned int _wait_read, _wait_write;

changes/fix-stalled-reader-test-from-spool.md

@@ -0,0 +1,20 @@
+Bug fixes:
+* The session of a client that stopped reading in the middle of a reply
+  could end after two ClientTimeouts instead of one (seen on FreeBSD):
+  when the send timeout runs out after send() has taken part of the
+  data, Linux and FreeBSD return that part instead of an error, and the
+  rest was sent with a second wait. After a short send, the rest now
+  goes out only if the socket can take more within what is left of the
+  send timeout; otherwise the write fails, like a failed send. The same
+  applies to connections to the upstream servers (ReadTimeout).
+
+Other changes:
+* TestClientStalledReader checks again that the session of a client
+  that stops reading in the middle of OVER or XOVER ends after one
+  ClientTimeout: it now fills the cache first, so that the reply comes
+  from the spool, and wants the session gone within 1.5 ClientTimeouts
+  of the request (3.0 to 3.3 s, also under -race). It only checked that
+  another client was still served, which since the records are written
+  with the group's lock released passed even when the session never
+  ended (3cefac9 reverted) or ended a ClientTimeout late (763126c
+  reverted: 6.0 s).

tests/e2e/daemon_test.go

@@ -14,6 +14,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"runtime"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -1072,7 +1073,38 @@ func testClientStalledReader(t *testing.T, command string) {
 	e := newEnv(t)
 	e.ClientTimeout = 3
 	e.Store.AddGenerated(group, "", fakeserver.Generator{Low: 1, High: 40000})
-	e.standalone()
+	mode, err := logMode(e.Bin)
+	if err != nil {
+		t.Fatal(err)
+	}
+	haveLog := mode != "syslog"
+	// A's session end is seen in the daemon's children: /proc, or pgrep -P
+	// without /proc (FreeBSD, macOS)
+	if _, ok := daemonChildren(os.Getpid()); !ok {
+		t.Skip("can't list a process's children (no /proc, no pgrep) to see the session end")
+	}
+	p := e.standalone()
+	// standalone()'s readiness check is a session of its own
+	waitForNoLinuxChildren(t, p.Pid(), 10*time.Second)
+
+	// Fill the cache first, so that A's reply below comes from the spool
+	// and its socket fills at once: the time to the session's end must not
+	// count how fast the fake upstream, which runs in this process (slow
+	// under -race), serves 40000 records.
+	w := e.dial()
+	w.Deadline = 60 * time.Second
+	if code, line := w.Cmd("GROUP %s", group); code != 211 {
+		t.Fatalf("warming GROUP: %q", line)
+	}
+	if code, line := w.Cmd("%s 1-40000", command); code != 224 {
+		t.Fatalf("warming %s: %q", command, line)
+	}
+	if n := len(w.ReadDotLines()); n != 40000 {
+		t.Fatalf("warming %s returned %d records, want 40000", command, n)
+	}
+	w.Cmd("QUIT")
+	waitForNoLinuxChildren(t, p.Pid(), 10*time.Second)
+
 	a, err := net.Dial("tcp", e.Listen)
 	if err != nil {
 		t.Fatal(err)
@@ -1086,6 +1118,34 @@ func testClientStalledReader(t *testing.T, command string) {
 	if l, err := ra.ReadString('\n'); err != nil || !strings.HasPrefix(l, "211") {
 		t.Fatalf("A: GROUP: %q %v", l, err)
 	}
+	// The daemon's only child is A's session.
+	kids := linuxProcChildren(t, p.Pid())
+	if len(kids) != 1 {
+		t.Fatalf("daemon children before the stall = %v, want A's session", kids)
+	}
+	aPid := kids[0]
+	stallStart := time.Now()
+	// When A's session ends, seen from the request: checked at the end,
+	// after B, which a busy machine slows down.
+	// (Not through linuxProcChildren: no t.Fatal from this goroutine,
+	// and it stops with the test when the session doesn't end.)
+	sessionEnd := make(chan time.Duration, 1)
+	stop := make(chan struct{})
+	defer close(stop)
+	go func() {
+		for {
+			kids, ok := daemonChildren(p.Pid())
+			if ok && !slices.Contains(kids, aPid) {
+				sessionEnd <- time.Since(stallStart)
+				return
+			}
+			select {
+			case <-stop:
+				return
+			case <-time.After(20 * time.Millisecond):
+			}
+		}
+	}()
 	io.WriteString(a, command+" 1-40000\r\n")
 	// A reads the status line and a little, then nothing more
 	if l, err := ra.ReadString('\n'); err != nil || !strings.HasPrefix(l, "224") {
@@ -1104,6 +1164,38 @@ func testClientStalledReader(t *testing.T, command string) {
 	}
 	b.ReadDotLines()
 	t.Logf("B served after %v", time.Since(start).Round(100*time.Millisecond))
+
+	// A's session must end one ClientTimeout after the request (the send
+	// timeout): 3.0 to 3.1 s with ClientTimeout 3, 3.3 s under -race. It
+	// used to stay blocked in send() for good (3cefac9), and a failed send
+	// that left its data in the buffer made the next flush wait for the
+	// timeout again: 6.0 s here (763126c). The bound is 1.5 ClientTimeouts.
+	endBound := time.Duration(e.ClientTimeout) * 1500 * time.Millisecond
+	select {
+	case d := <-sessionEnd:
+		t.Logf("the stalled session ended %v after the request", d.Round(time.Millisecond))
+		if d > endBound {
+			t.Errorf("stalled reader's session %s ended %v after the request, want within %v", aPid, d.Round(time.Millisecond), endBound)
+		}
+	case <-time.After(endBound - time.Since(stallStart) + 10*time.Second):
+		t.Errorf("stalled reader's session %s still running %v after the request, want it gone within %v", aPid, time.Since(stallStart).Round(time.Millisecond), endBound)
+	}
+	// With a log file: the session ended because the send failed. (Not
+	// the time of its end: that line comes at the first failed write.)
+	if !haveLog {
+		return
+	}
+	for {
+		log := e.logFile()
+		if strings.Contains(log, "can't be written to, session ended") {
+			break
+		}
+		if time.Since(stallStart) > endBound+10*time.Second {
+			t.Errorf("stalled reader's session not logged as ended %v after the request", time.Since(stallStart).Round(time.Millisecond))
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
 }
 
 // TestClientPostStall: a client that sends POST, part of the article and