libstore: use curl_multi_{poll,wakeup}

the previous solution to the wakeup problem (adding a pipe and passing
it as an additional fd to curl_multi_wait) worked, but there have been
builtin alternatives for this since 2020. not only do these save code,
they're also a lot more likely to work natively on windows when needed

Change-Id: Iab751b900997110a8d15de45ea3ab0c42f7e5973

1 files changed, 10 insertions, 23 deletions

diff --git a/src/libstore/filetransfer.cc b/src/libstore/filetransfer.cc
index 9ef220769..7d36796f4 100644
--- a/src/libstore/filetransfer.cc
+++ b/src/libstore/filetransfer.cc
@@ -504,11 +504,6 @@ struct curlFileTransfer : public FileTransfer
 
     Sync<State> state_;
 
-    /* We can't use a std::condition_variable to wake up the curl
-       thread, because it only monitors file descriptors. So use a
-       pipe instead. */
-    Pipe wakeupPipe;
-
     std::thread workerThread;
 
     curlFileTransfer()
@@ -523,9 +518,6 @@ struct curlFileTransfer : public FileTransfer
         curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS,
             fileTransferSettings.httpConnections.get());
 
-        wakeupPipe.create();
-        fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK);
-
         workerThread = std::thread([&]() { workerThreadEntry(); });
     }
 
@@ -538,6 +530,12 @@ struct curlFileTransfer : public FileTransfer
         if (curlm) curl_multi_cleanup(curlm);
     }
 
+    void wakeup()
+    {
+        if (auto mc = curl_multi_wakeup(curlm))
+            throw nix::Error("unexpected error from curl_multi_wakeup(): %s", curl_multi_strerror(mc));
+    }
+
     void stopWorkerThread()
     {
         /* Signal the worker thread to exit. */
@@ -545,7 +543,7 @@ struct curlFileTransfer : public FileTransfer
             auto state(state_.lock());
             state->quit = true;
         }
-        writeFull(wakeupPipe.writeSide.get(), " ", false);
+        wakeup();
     }
 
     void workerThreadMain()
@@ -587,32 +585,21 @@ struct curlFileTransfer : public FileTransfer
             }
 
             /* Wait for activity, including wakeup events. */
-            int numfds = 0;
-            struct curl_waitfd extraFDs[1];
-            extraFDs[0].fd = wakeupPipe.readSide.get();
-            extraFDs[0].events = CURL_WAIT_POLLIN;
-            extraFDs[0].revents = 0;
             long maxSleepTimeMs = items.empty() ? 10000 : 100;
             auto sleepTimeMs =
                 nextWakeup != std::chrono::steady_clock::time_point()
                 ? std::max(0, (int) std::chrono::duration_cast<std::chrono::milliseconds>(nextWakeup - std::chrono::steady_clock::now()).count())
                 : maxSleepTimeMs;
             vomit("download thread waiting for %d ms", sleepTimeMs);
-            mc = curl_multi_wait(curlm, extraFDs, 1, sleepTimeMs, &numfds);
+            mc = curl_multi_poll(curlm, nullptr, 0, sleepTimeMs, nullptr);
             if (mc != CURLM_OK)
-                throw nix::Error("unexpected error from curl_multi_wait(): %s", curl_multi_strerror(mc));
+                throw nix::Error("unexpected error from curl_multi_poll(): %s", curl_multi_strerror(mc));
 
             nextWakeup = std::chrono::steady_clock::time_point();
 
             /* Add new curl requests from the incoming requests queue,
                except for requests that are embargoed (waiting for a
                retry timeout to expire). */
-            if (extraFDs[0].revents & CURL_WAIT_POLLIN) {
-                char buf[1024];
-                auto res = read(extraFDs[0].fd, buf, sizeof(buf));
-                if (res == -1 && errno != EINTR)
-                    throw SysError("reading curl wakeup socket");
-            }
 
             std::vector<std::shared_ptr<TransferItem>> incoming;
             auto now = std::chrono::steady_clock::now();
@@ -675,7 +662,7 @@ struct curlFileTransfer : public FileTransfer
                 throw nix::Error("cannot enqueue download request because the download thread is shutting down");
             state->incoming.push(item);
         }
-        writeFull(wakeupPipe.writeSide.get(), " ");
+        wakeup();
     }
 
 #if ENABLE_S3