Start refactoring backoff.

kinyoklion · kinyoklion · commit 6b8d925045bb · 2025-10-15T18:23:44.000-07:00
diff --git a/libs/server-sent-events/src/CMakeLists.txt b/libs/server-sent-events/src/CMakeLists.txt
@@ -12,12 +12,14 @@ set(SSE_SOURCES
         event.cpp
         error.cpp
         backoff_detail.cpp
-        backoff.cpp)
+        backoff.cpp
+        curl_client.cpp
+        backoff_timer.cpp)
 
 if (LD_CURL_NETWORKING)
     message(STATUS "LaunchDarkly SSE: CURL networking enabled")
     find_package(CURL REQUIRED)
-    list(APPEND SSE_SOURCES curl_client.cpp)
+    list(APPEND SSE_SOURCES)
 endif()
 
 add_library(${LIBNAME} OBJECT ${SSE_SOURCES})
diff --git a/libs/server-sent-events/src/backoff_timer.cpp b/libs/server-sent-events/src/backoff_timer.cpp
@@ -0,0 +1,125 @@
+#include "backoff_timer.hpp"
+
+#include <boost/asio/post.hpp>
+
+namespace launchdarkly::sse {
+
+BackoffTimer::BackoffTimer(boost::asio::any_io_executor executor)
+    : executor_(std::move(executor)) {
+    std::thread t([this]() { timer_thread_func(); });
+    t.detach();
+}
+
+BackoffTimer::~BackoffTimer() {
+    // Signal shutdown and wake up the timer thread
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        shutdown_ = true;
+        cancelled_ = true;
+    }
+    cv_.notify_one();
+
+    // Wait for the timer thread to complete
+    // if (timer_thread_.joinable()) {
+    //     timer_thread_.join();
+    // }
+}
+
+void BackoffTimer::expires_after(std::chrono::milliseconds duration,
+                                 std::function<void(bool cancelled)> callback) {
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+
+        // Cancel any existing timer
+        cancelled_ = timer_active_;
+
+        // Set up the new timer
+        expiry_time_ = std::chrono::steady_clock::now() + duration;
+        callback_ = std::move(callback);
+        timer_active_ = true;
+        cancelled_ = false;
+    }
+
+    // Wake up the timer thread
+    cv_.notify_one();
+}
+
+void BackoffTimer::cancel() {
+    {
+        std::lock_guard<std::mutex> lock(mutex_);
+        if (timer_active_) {
+            cancelled_ = true;
+        }
+    }
+
+    // Wake up the timer thread
+    cv_.notify_one();
+}
+
+boost::asio::any_io_executor BackoffTimer::get_executor() const {
+    return executor_;
+}
+
+void BackoffTimer::timer_thread_func() {
+    while (true) {
+        std::function<void(bool cancelled)> callback_to_invoke;
+        bool should_invoke = false;
+        bool was_cancelled = false;
+
+        {
+            std::unique_lock<std::mutex> lock(mutex_);
+
+            // Wait for either:
+            // 1. A timer to be set (timer_active_ becomes true)
+            // 2. Shutdown signal
+            while (!timer_active_ && !shutdown_) {
+                cv_.wait(lock);
+            }
+
+            // Check if we're shutting down
+            if (shutdown_) {
+                return;
+            }
+
+            // Wait until the timer expires or is cancelled
+            while (timer_active_ && !cancelled_ && !shutdown_) {
+                auto now = std::chrono::steady_clock::now();
+                if (now >= expiry_time_) {
+                    // Timer expired
+                    should_invoke = true;
+                    was_cancelled = false;
+                    callback_to_invoke = std::move(callback_);
+                    timer_active_ = false;
+                    break;
+                }
+
+                // Wait until expiry time or until notified
+                cv_.wait_until(lock, expiry_time_);
+            }
+
+            // Check if timer was cancelled
+            if (cancelled_ && timer_active_) {
+                should_invoke = true;
+                was_cancelled = true;
+                callback_to_invoke = std::move(callback_);
+                timer_active_ = false;
+                cancelled_ = false;
+            }
+
+            // Check if we're shutting down
+            if (shutdown_) {
+                return;
+            }
+        }
+
+        // Invoke callback outside of lock by posting to the executor
+        if (should_invoke && callback_to_invoke) {
+            boost::asio::post(executor_, [callback = std::move(callback_to_invoke),
+                                         was_cancelled]() {
+                callback(was_cancelled);
+            });
+        }
+    }
+}
+
+} // namespace launchdarkly::sse
diff --git a/libs/server-sent-events/src/backoff_timer.hpp b/libs/server-sent-events/src/backoff_timer.hpp
@@ -0,0 +1,88 @@
+#pragma once
+
+#include <boost/asio/any_io_executor.hpp>
+
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <functional>
+#include <memory>
+#include <mutex>
+#include <thread>
+
+namespace launchdarkly::sse {
+
+/**
+ * A thread-based timer that waits for a specified duration and then posts
+ * a callback to an ASIO executor. The timer can be cancelled, making it
+ * suitable for use in backoff scenarios where cleanup is required during
+ * destruction.
+ *
+ * The timer uses a dedicated thread for waiting (via condition_variable)
+ * and posts the callback to the provided ASIO executor when the timer expires.
+ * This avoids blocking the ASIO thread pool during backoff periods.
+ */
+class BackoffTimer {
+public:
+    /**
+     * Construct a BackoffTimer with the given ASIO executor.
+     * @param executor The ASIO executor to post callbacks to when timer expires.
+     */
+    explicit BackoffTimer(boost::asio::any_io_executor executor);
+
+    /**
+     * Destructor. Cancels any pending timer and waits for the timer thread
+     * to complete.
+     */
+    ~BackoffTimer();
+
+    // Non-copyable and non-movable
+    BackoffTimer(const BackoffTimer&) = delete;
+    BackoffTimer& operator=(const BackoffTimer&) = delete;
+    BackoffTimer(BackoffTimer&&) = delete;
+    BackoffTimer& operator=(BackoffTimer&&) = delete;
+
+    /**
+     * Start an asynchronous wait. When the duration expires, the callback
+     * will be posted to the executor provided in the constructor.
+     *
+     * If a timer is already running, it will be cancelled before starting
+     * the new timer.
+     *
+     * @param duration The duration to wait before invoking the callback.
+     * @param callback The callback to invoke when the timer expires.
+     *                 The callback receives a boolean indicating whether
+     *                 the timer was cancelled (true) or expired normally (false).
+     */
+    void expires_after(std::chrono::milliseconds duration,
+                      std::function<void(bool cancelled)> callback);
+
+    /**
+     * Cancel any pending timer. If a timer is running, the callback will
+     * be invoked with cancelled=true.
+     */
+    void cancel();
+
+    /**
+     * Get the executor used by this timer.
+     */
+    boost::asio::any_io_executor get_executor() const;
+
+private:
+    void timer_thread_func();
+
+    boost::asio::any_io_executor executor_;
+
+    std::mutex mutex_;
+    std::condition_variable cv_;
+    std::atomic<bool> shutdown_{false};
+    std::atomic<bool> cancelled_{false};
+
+    std::chrono::steady_clock::time_point expiry_time_;
+    std::function<void(bool cancelled)> callback_;
+    bool timer_active_{false};
+
+    std::thread timer_thread_;
+};
+
+} // namespace launchdarkly::sse
diff --git a/libs/server-sent-events/src/curl_client.cpp b/libs/server-sent-events/src/curl_client.cpp
@@ -62,32 +62,12 @@ CurlClient::CurlClient(boost::asio::any_io_executor executor,
         write_timeout,
         std::move(custom_ca_file),
         std::move(proxy_url),
-        skip_verify_peer,
-        [this](const std::string& message) {
-            boost::asio::post(backoff_timer_.get_executor(), [this, message]() {
-                async_backoff(message);
-            });
-        },
-        [this](const Event& event) {
-            boost::asio::post(backoff_timer_.get_executor(), [this, event]() {
-                event_receiver_(event);
-            });
-        },
-        [this](const Error& error) {
-            boost::asio::post(backoff_timer_.get_executor(), [this, error]() {
-                report_error(error);
-            });
-        },
-        [this]() {
-            boost::asio::post(backoff_timer_.get_executor(), [this]() {
-                backoff_.succeed();
-            });
-        }, [this](const std::string& message) {
-            log_message(message);
-        });
+        skip_verify_peer);
+    std::cout << "Curl client created: " << host << std::endl;
 }
 
 CurlClient::~CurlClient() {
+    std::cout << "Curl client destructing: " << request_context_->url << std::endl;
     request_context_->shutdown();
     backoff_timer_.cancel();
 }
@@ -97,12 +77,58 @@ void CurlClient::async_connect() {
                       [self = shared_from_this()]() { self->do_run(); });
 }
 
-void CurlClient::do_run() const {
+void CurlClient::do_run() {
     if (request_context_->is_shutting_down()) {
         return;
     }
 
     auto ctx = request_context_;
+    auto weak_self = weak_from_this();
+    ctx->set_callbacks(Callbacks([weak_self, ctx](const std::string& message) {
+                                    std::cout << "Backoff " << ctx->url << " " << message << std::endl;
+                                     if (auto self = weak_self.lock()) {
+                                         boost::asio::post(
+                                             self->backoff_timer_.
+                                                   get_executor(),
+                                             [self, message]() {
+                                                 self->async_backoff(message);
+                                             });
+                                     }
+                                 },
+                                 [weak_self, ctx](const Event& event) {
+                                     std::cout << "Event " << ctx->url << " " << event.data() << std::endl;
+                                     if (auto self = weak_self.lock()) {
+                                         boost::asio::post(
+                                             self->backoff_timer_.
+                                                   get_executor(),
+                                             [self, event]() {
+                                                 self->event_receiver_(event);
+                                             });
+                                     }
+                                 },
+                                 [weak_self, ctx](const Error& error) {
+                                     std::cout << "Error " << ctx->url << " " << error << std::endl;
+                                     if (const auto self = weak_self.lock()) {
+                                         // report_error does an asio post.
+                                         self->report_error(error);
+                                     }
+                                 },
+                                 [weak_self, ctx]() {
+                                     std::cout << "Reset backoff" << ctx->url << std::endl;
+                                     if (const auto self = weak_self.lock()) {
+                                         boost::asio::post(
+                                             self->backoff_timer_.
+                                                   get_executor(),
+                                             [self]() {
+                                                 self->backoff_.succeed();
+                                             });
+                                     }
+                                 }, [weak_self, ctx](const std::string& message) {
+                                     std::cout << "Log " << ctx->url << " " << message << std::endl;
+                                     if (const auto self = weak_self.lock()) {
+                                         self->log_message(message);
+                                     }
+                                 }));
     // Start request in a separate thread since CURL blocks
     // Capture only raw 'this' pointer, not shared_ptr
     std::thread t(
@@ -122,19 +148,21 @@ void CurlClient::async_backoff(std::string const& reason) {
 
     log_message(msg.str());
 
-    backoff_timer_.expires_after(backoff_.delay());
-    backoff_timer_.async_wait([self = shared_from_this()](
-        const boost::system::error_code& ec) {
-            self->on_backoff(ec);
+    // auto weak_self = weak_from_this();
+    backoff_timer_.expires_after(backoff_.delay(),
+        [this](bool cancelled) {
+            if (request_context_->is_shutting_down()) {
+                return;
+            }
+            // if (const auto self = weak_self.lock()) {
+                on_backoff(cancelled);
+            // }
         });
 }
 
-void CurlClient::on_backoff(const boost::system::error_code& ec) const {
-    {
-        if (ec == boost::asio::error::operation_aborted || request_context_->
-            is_shutting_down()) {
-            return;
-        }
+void CurlClient::on_backoff(bool cancelled) {
+    if (cancelled || request_context_->is_shutting_down()) {
+        return;
     }
     do_run();
 }
@@ -492,6 +520,7 @@ size_t CurlClient::HeaderCallback(const char* buffer,
 }
 
 void CurlClient::PerformRequest(std::shared_ptr<RequestContext> context) {
+    std::cout << "CurlClient::PerformRequest: " << context->url << std::endl;
     if (context->is_shutting_down()) {
         return;
     }
diff --git a/libs/server-sent-events/src/curl_client.hpp b/libs/server-sent-events/src/curl_client.hpp
