substitute: Cache and reuse connections while substituting.

That way, when fetching a series of substitutes from the same server(s),
the connection is reused instead of being closed/opened for each
substitutes, which saves on network round trips and TLS handshakes.

* guix/http-client.scm (http-fetch): Add #:keep-alive? and honor it.
* guix/progress.scm (progress-report-port): Add #:close? parameter and
honor it.
* guix/scripts/substitute.scm (at-most): Return the tail as a second
value.
(fetch): Add #:port and #:keep-alive? and honor them.
(%max-cached-connections): New variable.
(open-connection-for-uri/cached, call-with-cached-connection): New
procedures.
(with-cached-connection): New macro.
(process-substitution): Wrap 'fetch' call in 'with-cached-connection'.
Pass #:close? to 'progress-report-port'.

1 files changed, 18 insertions, 11 deletions

diff --git a/nix/libstore/build.cc b/nix/libstore/build.cc
index 50d300253db..6cfe7aba7e6 100644
--- a/nix/libstore/build.cc
+++ b/nix/libstore/build.cc
@@ -3114,17 +3114,24 @@ void SubstitutionGoal::handleChildOutput(int fd, const string & data)
     }
 
     if (fd == substituter->fromAgent.readSide) {
-        /* Trim whitespace to the right.  */
-        size_t end = data.find_last_not_of(" \t\n");
-        string trimmed = (end != string::npos) ? data.substr(0, end + 1) : data;
-
-        if (expectedHashStr == "") {
-            expectedHashStr = trimmed;
-        } else if (status == "") {
-            status = trimmed;
-            worker.wakeUp(shared_from_this());
-        } else {
-            printMsg(lvlError, format("unexpected substituter message '%1%'") % data);
+        /* DATA may consist of several lines.  Process them one by one.  */
+        string input = data;
+        while (!input.empty()) {
+            /* Process up to the first newline.  */
+            size_t end = input.find_first_of("\n");
+            string trimmed = (end != string::npos) ? input.substr(0, end) : input;
+
+            /* Update the goal's state accordingly.  */
+            if (expectedHashStr == "") {
+                expectedHashStr = trimmed;
+            } else if (status == "") {
+                status = trimmed;
+                worker.wakeUp(shared_from_this());
+            } else {
+                printMsg(lvlError, format("unexpected substituter message '%1%'") % input);
+            }
+
+            input = (end != string::npos) ? input.substr(end + 1) : "";
         }
     }
 }