Ensure download thread liveness

* Don't wait forever for the client to remove data from the
  buffer. This does mean that the buffer can grow without bounds
  (e.g. when downloading is faster than writing to disk), but meh.

* Don't hold the state lock while calling the sink. The sink could
  take any amount of time to process the data (in particular when it's
  actually a coroutine), so we don't want to block the download
  thread.

1 files changed, 26 insertions, 19 deletions

diff --git a/src/libstore/download.cc b/src/libstore/download.cc
index 13913d031daa..f44f1836b31e 100644
--- a/src/libstore/download.cc
+++ b/src/libstore/download.cc
@@ -710,11 +710,12 @@ void Downloader::download(DownloadRequest && request, Sink & sink)
 
         /* If the buffer is full, then go to sleep until the calling
            thread wakes us up (i.e. when it has removed data from the
-           buffer). Note: this does stall the download thread. */
-        while (state->data.size() > 1024 * 1024) {
-            if (state->quit) return;
+           buffer). We don't wait forever to prevent stalling the
+           download thread. (Hopefully sleeping will throttle the
+           sender.) */
+        if (state->data.size() > 1024 * 1024) {
             debug("download buffer is full; going to sleep");
-            state.wait(state->request);
+            state.wait_for(state->request, std::chrono::seconds(10));
         }
 
         /* Append data to the buffer and wake up the calling
@@ -736,30 +737,36 @@ void Downloader::download(DownloadRequest && request, Sink & sink)
             state->request.notify_one();
         }});
 
-    auto state(_state->lock());
-
     while (true) {
         checkInterrupt();
 
-        /* If no data is available, then wait for the download thread
-           to wake us up. */
-        if (state->data.empty()) {
+        std::string chunk;
+
+        /* Grab data if available, otherwise wait for the download
+           thread to wake us up. */
+        {
+            auto state(_state->lock());
+
+            while (state->data.empty()) {
 
-            if (state->quit) {
-                if (state->exc) std::rethrow_exception(state->exc);
-                break;
+                if (state->quit) {
+                    if (state->exc) std::rethrow_exception(state->exc);
+                    return;
+                }
+
+                state.wait(state->avail);
             }
 
-            state.wait(state->avail);
-        }
+            chunk = std::move(state->data);
 
-        /* If data is available, then flush it to the sink and wake up
-           the download thread if it's blocked on a full buffer. */
-        if (!state->data.empty()) {
-            sink((unsigned char *) state->data.data(), state->data.size());
-            state->data.clear();
             state->request.notify_one();
         }
+
+        /* Flush the data to the sink and wake up the download thread
+           if it's blocked on a full buffer. We don't hold the state
+           lock while doing this to prevent blocking the download
+           thread if sink() takes a long time. */
+        sink((unsigned char *) chunk.data(), chunk.size());
     }
 }