upload-pack: adapt keepalives based on buffering

The function `create_pack_file()` is responsible for sending the
packfile data to the client of git-upload-pack(1). As generating the
bytes may take significant computing resources we also have a mechanism
in place that optionally sends keepalive pktlines in case we haven't
sent out any data.

The keepalive logic is purely based poll(3p): we pass a timeout to that
syscall, and if the call times out we send out the keepalive pktline.
While reasonable, this logic isn't entirely sufficient: even if the call
to poll(3p) ends because we have received data on any of the file
descriptors we may not necessarily send data to the client.

The most important edge case here happens in `relay_pack_data()`. When
we haven't seen the initial "PACK" signature from git-pack-objects(1)
yet we buffer incoming data. So in the worst case, if each of the bytes
of that signature arrive shortly before the configured keepalive
timeout, then we may not send out any data for a time period that is
(almost) four times as long as the configured timeout.

This edge case is rather unlikely to matter in practice. But in a
subsequent commit we're going to adapt our buffering mechanism to become
more aggressive, which makes it more likely that we don't send any data
for an extended amount of time.

Adapt the logic so that instead of using a fixed timeout on every call
to poll(3p), we instead figure out how much time has passed since the
last-sent data.

Signed-off-by: Patrick Steinhardt <ps@pks.im>
Signed-off-by: Junio C Hamano <gitster@pobox.com>

diff --git a/upload-pack.c b/upload-pack.c
index c2643c0295bad8f0f9154e3db1a515b34200d8ba..04521e57c967f302af77735aab8db6b34109d8bb 100644 (file)
--- a/upload-pack.c
+++ b/upload-pack.c
@@ -29,6 +29,7 @@
 #include "commit-graph.h"
 #include "commit-reach.h"
 #include "shallow.h"
+#include "trace.h"
 #include "write-or-die.h"
 #include "json-writer.h"
 #include "strmap.h"
@@ -218,7 +219,8 @@ struct output_state {
 };
 
 static int relay_pack_data(int pack_objects_out, struct output_state *os,
-                          int use_sideband, int write_packfile_line)
+                          int use_sideband, int write_packfile_line,
+                          bool *did_send_data)
 {
        /*
         * We keep the last byte to ourselves
@@ -232,6 +234,8 @@ static int relay_pack_data(int pack_objects_out, struct output_state *os,
         */
        ssize_t readsz;
 
+       *did_send_data = false;
+
        readsz = xread(pack_objects_out, os->buffer + os->used,
                       sizeof(os->buffer) - os->used);
        if (readsz < 0) {
@@ -247,6 +251,7 @@ static int relay_pack_data(int pack_objects_out, struct output_state *os,
                                if (os->packfile_uris_started)
                                        packet_delim(1);
                                packet_write_fmt(1, "\1packfile\n");
+                               *did_send_data = true;
                        }
                        break;
                }
@@ -259,6 +264,7 @@ static int relay_pack_data(int pack_objects_out, struct output_state *os,
                        }
                        *p = '\0';
                        packet_write_fmt(1, "\1%s\n", os->buffer);
+                       *did_send_data = true;
 
                        os->used -= p - os->buffer + 1;
                        memmove(os->buffer, p + 1, os->used);
@@ -279,6 +285,7 @@ static int relay_pack_data(int pack_objects_out, struct output_state *os,
                os->used = 0;
        }
 
+       *did_send_data = true;
        return readsz;
 }
 
@@ -290,6 +297,7 @@ static void create_pack_file(struct upload_pack_data *pack_data,
        char progress[128];
        char abort_msg[] = "aborting due to possible repository "
                "corruption on the remote side.";
+       uint64_t last_sent_ms = 0;
        ssize_t sz;
        int i;
        FILE *pipe_fd;
@@ -365,10 +373,14 @@ static void create_pack_file(struct upload_pack_data *pack_data,
         */
 
        while (1) {
+               uint64_t now_ms = getnanotime() / 1000000;
                struct pollfd pfd[2];
-               int pe, pu, pollsize, polltimeout;
+               int pe, pu, pollsize, polltimeout_ms;
                int ret;
 
+               if (!last_sent_ms)
+                       last_sent_ms = now_ms;
+
                reset_timeout(pack_data->timeout);
 
                pollsize = 0;
@@ -390,11 +402,21 @@ static void create_pack_file(struct upload_pack_data *pack_data,
                if (!pollsize)
                        break;
 
-               polltimeout = pack_data->keepalive < 0
-                       ? -1
-                       : 1000 * pack_data->keepalive;
+               if (pack_data->keepalive < 0) {
+                       polltimeout_ms = -1;
+               } else {
+                       /*
+                        * The polling timeout needs to be adjusted based on
+                        * the time we have sent our last package. The longer
+                        * it's been in the past, the shorter the timeout
+                        * becomes until we eventually don't block at all.
+                        */
+                       polltimeout_ms = 1000 * pack_data->keepalive - (now_ms - last_sent_ms);
+                       if (polltimeout_ms < 0)
+                               polltimeout_ms = 0;
+               }
 
-               ret = poll(pfd, pollsize, polltimeout);
+               ret = poll(pfd, pollsize, polltimeout_ms);
 
                if (ret < 0) {
                        if (errno != EINTR) {
@@ -403,16 +425,18 @@ static void create_pack_file(struct upload_pack_data *pack_data,
                        }
                        continue;
                }
+
                if (0 <= pe && (pfd[pe].revents & (POLLIN|POLLHUP))) {
                        /* Status ready; we ship that in the side-band
                         * or dump to the standard error.
                         */
                        sz = xread(pack_objects.err, progress,
                                  sizeof(progress));
-                       if (0 < sz)
+                       if (0 < sz) {
                                send_client_data(2, progress, sz,
                                                 pack_data->use_sideband);
-                       else if (sz == 0) {
+                               last_sent_ms = now_ms;
+                       } else if (sz == 0) {
                                close(pack_objects.err);
                                pack_objects.err = -1;
                        }
@@ -421,11 +445,14 @@ static void create_pack_file(struct upload_pack_data *pack_data,
                        /* give priority to status messages */
                        continue;
                }
+
                if (0 <= pu && (pfd[pu].revents & (POLLIN|POLLHUP))) {
+                       bool did_send_data;
                        int result = relay_pack_data(pack_objects.out,
                                                     output_state,
                                                     pack_data->use_sideband,
-                                                    !!uri_protocols);
+                                                    !!uri_protocols,
+                                                    &did_send_data);
 
                        if (result == 0) {
                                close(pack_objects.out);
@@ -433,6 +460,9 @@ static void create_pack_file(struct upload_pack_data *pack_data,
                        } else if (result < 0) {
                                goto fail;
                        }
+
+                       if (did_send_data)
+                               last_sent_ms = now_ms;
                }
 
                /*
@@ -448,6 +478,7 @@ static void create_pack_file(struct upload_pack_data *pack_data,
                if (!ret && pack_data->use_sideband) {
                        static const char buf[] = "0005\1";
                        write_or_die(1, buf, 5);
+                       last_sent_ms = now_ms;
                }
        }