feat: Copy curl queue and worker from B&A to common repo

This is part of a series of CLs to copy
the KV BYOS clients from B&A to common repo.

Bug: 411430242
Change-Id: I55de66cb464d93b69726d9aa8b73f02b4c5b7b3c
GitOrigin-RevId: b33aeef6c0e19112a9c55326db158fe2dfdc3292

Author: Privacy Sandbox Team

src/clients/http_fetcher_async/BUILD.bazel

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-load("@rules_cc//cc:defs.bzl", "cc_library")
+load("@rules_cc//cc:defs.bzl", "cc_library", "cc_test")
 
 cc_library(
     name = "http_fetcher_async",
@@ -47,6 +47,50 @@ cc_library(
     ],
 )
 
+cc_library(
+    name = "curl_request_queue",
+    srcs = [
+        "curl_request_queue.cc",
+    ],
+    hdrs = [
+        "curl_request_queue.h",
+    ],
+    deps = [
+        ":curl_request_data",
+        "//src/concurrent:executor",
+        "//src/core/event",
+        "//src/core/event:event_base",
+        "//src/util/containers:doubly_linked_list",
+        "@com_google_absl//absl/container:flat_hash_map",
+        "@com_google_absl//absl/log:check",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/time",
+    ],
+)
+
+cc_library(
+    name = "curl_request_worker",
+    srcs = [
+        "curl_request_worker.cc",
+    ],
+    hdrs = [
+        "curl_request_worker.h",
+    ],
+    deps = [
+        ":curl_request_data",
+        ":curl_request_queue",
+        ":multi_curl_request_manager",
+        "//src/concurrent:executor",
+        "@com_google_absl//absl/log:check",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/time",
+    ],
+)
+
 cc_library(
     name = "multi_curl_request_manager",
     srcs = [
@@ -70,3 +114,47 @@ cc_library(
         "@libevent//:event",
     ],
 )
+
+cc_test(
+    name = "curl_request_queue_test",
+    size = "small",
+    srcs = ["curl_request_queue_test.cc"],
+    deps = [
+        ":curl_request_data",
+        ":curl_request_queue",
+        "//src/concurrent:executor",
+        "//src/logger:request_context_logger",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest",
+        "@com_google_googletest//:gtest_main",
+        "@rapidjson",
+    ],
+)
+
+cc_test(
+    name = "curl_request_worker_test",
+    size = "small",
+    srcs = ["curl_request_worker_test.cc"],
+    tags = [
+        "notsan",
+    ],
+    deps = [
+        ":curl_request_data",
+        ":curl_request_queue",
+        ":curl_request_worker",
+        "//src/concurrent:executor",
+        "//src/logger:request_context_logger",
+        "@com_github_grpc_grpc//:grpc++",
+        "@com_google_absl//absl/status",
+        "@com_google_absl//absl/strings",
+        "@com_google_absl//absl/synchronization",
+        "@com_google_absl//absl/time",
+        "@com_google_googletest//:gtest",
+        "@com_google_googletest//:gtest_main",
+        "@rapidjson",
+    ],
+)

src/clients/http_fetcher_async/curl_request_queue.cc

@@ -0,0 +1,138 @@
+//  Copyright 2025 Google LLC
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+#include "src/clients/http_fetcher_async/curl_request_queue.h"
+
+#include <utility>
+
+#include "absl/log/check.h"
+#include "absl/synchronization/mutex.h"
+#include "src/core/event/event.h"
+#include "src/util/containers/doubly_linked_list.h"
+
+namespace privacy_sandbox::server_common::clients {
+namespace {
+// Timer used by the event on event loop to signal that event loop has
+// successfully started.
+inline static timeval OneMicrosecond = {0, 1};
+
+}  // namespace
+
+CurlRequestQueue::CurlRequestQueue(server_common::Executor* executor,
+                                   int capacity, absl::Duration max_wait_time)
+    : executor_(executor),
+      capacity_(capacity),
+      max_wait_(max_wait_time),
+      expiry_(absl::ToTimeval(max_wait_time)),
+      eventloop_started_event_(Event(event_base_.get(), /* fd= */ -1,
+                                     /* event_type= */ 0,
+                                     /* event_callback= */ StartedEventLoop,
+                                     /* arg= */ this,
+                                     /* priority= */ kNumEventPriorities / 2,
+                                     &OneMicrosecond)),
+      ticker_(Event(event_base_.get(), /*fd=*/-1,
+                    /*event_type=*/EV_PERSIST,
+                    /*event_callback=*/RemoveExpiredEntries,
+                    /*arg=*/this,
+                    /*priority=*/kNumEventPriorities / 2, &expiry_)) {
+  DCHECK(executor_ != nullptr);
+  // Start the event loop - used to remove expired request from the request
+  // queue.
+  executor_->Run([this]() { event_base_dispatch(event_base_.get()); });
+  eventloop_started_.WaitForNotification();
+}
+
+absl::Mutex& CurlRequestQueue::Mu() { return mu_; }
+
+bool CurlRequestQueue::Full() { return size_ == capacity_; }
+
+bool CurlRequestQueue::Empty() { return size_ == 0; }
+
+std::unique_ptr<CurlRequestData> CurlRequestQueue::Dequeue() {
+  if (Empty()) {
+    return nullptr;
+  }
+
+  --size_;
+  auto* node = dll_.Tail();
+  std::unique_ptr<CurlRequestData> curl_request_data =
+      std::move(node->data->key);
+  dll_.Remove(curl_handles_[curl_request_data.get()]);
+  curl_handles_.erase(curl_request_data.get());
+  return curl_request_data;
+}
+
+// static
+void CurlRequestQueue::StartedEventLoop(int fd, int16_t event_type, void* arg) {
+  auto* self = reinterpret_cast<CurlRequestQueue*>(arg);
+  PS_VLOG(5) << "Notifying event loop start";
+  self->eventloop_started_.Notify();
+}
+
+// static
+void CurlRequestQueue::RemoveExpiredEntries(int fd, int16_t event_type,
+                                            void* arg) {
+  auto* self = reinterpret_cast<CurlRequestQueue*>(arg);
+  if (self->shutting_down_.HasBeenNotified()) {
+    absl::MutexLock lock(&self->Mu());
+    while (auto request = self->Dequeue()) {
+      std::move(request->done_callback)(absl::InternalError("Shutting down."));
+    }
+    event_base_loopbreak(self->event_base_.get());
+    self->shutdown_completed_.Notify();
+    return;
+  }
+
+  absl::MutexLock lock(&self->mu_);
+  while (auto* node = self->dll_.Head()) {
+    auto* curl_request_data_ptr = node->data->key.get();
+    if (absl::Now() - curl_request_data_ptr->start_time < self->max_wait_) {
+      return;
+    }
+
+    --self->size_;
+    std::unique_ptr<CurlRequestData> curl_request_data =
+        std::move(node->data->key);
+    self->dll_.Remove(self->curl_handles_[curl_request_data_ptr]);
+    self->curl_handles_.erase(curl_request_data_ptr);
+
+    // Move the callback to a different thread.
+    self->executor_->Run([curl_request_data = std::move(curl_request_data)]() {
+      std::move(curl_request_data->done_callback)(
+          absl::InternalError("Request timed out waiting in the queue"));
+    });
+  }
+}
+
+void CurlRequestQueue::Enqueue(std::unique_ptr<CurlRequestData> request) {
+  if (shutting_down_.HasBeenNotified()) {
+    std::move(request->done_callback)(absl::InternalError("Shutting down."));
+    return;
+  }
+
+  auto* curl_request_data = request.get();
+  curl_request_data->start_time = absl::Now();
+  auto data = std::make_unique<DLLData>(DLLData{.key = std::move(request)});
+  auto* dll_node = dll_.InsertAtFront(std::move(data));
+  curl_handles_[curl_request_data] = dll_node;
+  ++size_;
+}
+
+CurlRequestQueue::~CurlRequestQueue() {
+  shutting_down_.Notify();
+  event_active(ticker_.get(), /*res=*/0, /*ncalls=*/0);
+  shutdown_completed_.WaitForNotification();
+}
+
+}  // namespace privacy_sandbox::server_common::clients

src/clients/http_fetcher_async/curl_request_queue.h

@@ -0,0 +1,128 @@
+//  Copyright 2025 Google LLC
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+#ifndef SRC_CLIENTS_HTTP_FETCHER_ASYNC_CURL_REQUEST_QUEUE_H_
+#define SRC_CLIENTS_HTTP_FETCHER_ASYNC_CURL_REQUEST_QUEUE_H_
+
+#include <memory>
+
+#include "absl/base/thread_annotations.h"
+#include "absl/container/flat_hash_map.h"
+#include "absl/synchronization/mutex.h"
+#include "absl/synchronization/notification.h"
+#include "absl/time/time.h"
+#include "src/clients/http_fetcher_async/curl_request_data.h"
+#include "src/concurrent/executor.h"
+#include "src/core/event/event.h"
+#include "src/core/event/event_base.h"
+#include "src/util/containers/doubly_linked_list.h"
+
+namespace privacy_sandbox::server_common::clients {
+
+// Request queue used to decouple the producers/reactors who want to make curl
+// calls from the executor (a thread that will run the curl transfers).
+// Idea is that queuing/dequeuing will be a fast operation and reactor threads
+// don't have to block (for long periods) to add requests even if the executor
+// is taking longer to process any request.
+//
+// Note: Mutex `Mu()` should be held before accessing any public interface of
+// this class.
+class CurlRequestQueue {
+ public:
+  explicit CurlRequestQueue(server_common::Executor* executor, int capacity,
+                            absl::Duration max_wait_time);
+  ~CurlRequestQueue();
+
+  // Callers can check the fullness before attempting to queue. This is
+  // helpful since otherwise callers will try to queue and lose the request
+  // data (including any callbacks), if the queue is full.
+  bool Full() ABSL_EXCLUSIVE_LOCKS_REQUIRED(Mu());
+
+  // Returns a bool to reflect emptiness of the queue. It helps the executor
+  // thread to setup a condition variable and wait for new requests.
+  bool Empty() ABSL_EXCLUSIVE_LOCKS_REQUIRED(Mu());
+
+  // Removes element from a queue. Returns nullptr if no element is present.
+  // Will be used by the executor thread to get tasks off the queue and
+  // timeout thread to remove tasks that have been on the queue for a
+  // (configurable) timeout period.
+  std::unique_ptr<CurlRequestData> Dequeue()
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(Mu());
+
+  // Adds an element to a queue. Caller must ensure that the queue is not
+  // full before calling this method.
+  void Enqueue(std::unique_ptr<CurlRequestData> request)
+      ABSL_EXCLUSIVE_LOCKS_REQUIRED(Mu());
+
+  // Controls concurrent access to the queue.
+  absl::Mutex& Mu();
+
+ private:
+  using DLLData = CacheHashData<std::unique_ptr<CurlRequestData>,
+                                /*unused=*/char>;
+  using NodeType = Node<std::unique_ptr<CurlRequestData>, /*unused=*/char>;
+
+  // Executor is used to run a thread for expiring entries from the queue.
+  server_common::Executor* executor_;
+
+  // Signals successful start of event loop.
+  static void StartedEventLoop(int fd, int16_t event_type, void* arg);
+
+  // Callback to remove expired entries.
+  static void RemoveExpiredEntries(int fd, int16_t event_type, void* arg);
+
+  // Maps from curl request to the node in the doubly linked list.
+  absl::flat_hash_map<CurlRequestData*, NodeType*> curl_handles_;
+
+  // Doubly linked list is used to queue/dequeue requests.
+  DoublyLinkedList<std::unique_ptr<CurlRequestData>, /*unused=*/char> dll_;
+
+  // Number of elements currently in the queue.
+  int size_ = 0;
+
+  // Maximum number of elements that the queue will hold.
+  const int capacity_;
+
+  // Mutex used to expire entries from the queue.
+  absl::Mutex mu_;
+
+  // Event base used to register the request expiry events.
+  EventBase event_base_ = EventBase(/*use_pthreads=*/true);
+
+  // Max time a request is allowed to wait in the queue before it is expired.
+  absl::Duration max_wait_;
+
+  // Maximum time to wait for the request to be dequeued before the request
+  // forcibly expired from the queue.
+  struct timeval expiry_;
+
+  // Activates on event loop start and signals that event loop has successfully
+  // started.
+  Event eventloop_started_event_;
+
+  // Callback executed every `max_wait_` time to remove expired entries off
+  // the queue.
+  Event ticker_;
+
+  // Notification to sync the caller with the event loop start.
+  absl::Notification eventloop_started_;
+
+  // Once set all the incoming enqueue requests will be rejected right away.
+  absl::Notification shutting_down_;
+  absl::Notification shutdown_completed_;
+};
+
+}  // namespace privacy_sandbox::server_common::clients
+
+#endif  // SRC_CLIENTS_HTTP_FETCHER_ASYNC_CURL_REQUEST_QUEUE_H_