diff --git a/NServer-0.9.1/checks/cReplyBudget.cc b/NServer-0.9.1/checks/cReplyBudget.cc index 455ff34..577f0c2 100644 --- a/NServer-0.9.1/checks/cReplyBudget.cc +++ b/NServer-0.9.1/checks/cReplyBudget.cc @@ -1,9 +1,12 @@ #include #include #include +#include #include #include #include +#include +#include #include #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; } diff --git a/NServer-0.9.1/sockstream.cc b/NServer-0.9.1/sockstream.cc index a8be565..1ce5baa 100644 --- a/NServer-0.9.1/sockstream.cc +++ b/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)); diff --git a/NServer-0.9.1/sockstream.h b/NServer-0.9.1/sockstream.h index e1da7bd..6290f62 100644 --- a/NServer-0.9.1/sockstream.h +++ b/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; diff --git a/changes/fix-stalled-reader-test-from-spool.md b/changes/fix-stalled-reader-test-from-spool.md new file mode 100644 index 0000000..eaf6ff7 --- /dev/null +++ b/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). diff --git a/tests/e2e/daemon_test.go b/tests/e2e/daemon_test.go index 87057b1..9df0f0e 100644 --- a/tests/e2e/daemon_test.go +++ b/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