diff --git a/olp-cpp-sdk-core/src/http/DefaultNetwork.cpp b/olp-cpp-sdk-core/src/http/DefaultNetwork.cpp index 4f6841528..22578fb77 100644 --- a/olp-cpp-sdk-core/src/http/DefaultNetwork.cpp +++ b/olp-cpp-sdk-core/src/http/DefaultNetwork.cpp @@ -27,7 +27,9 @@ namespace olp { namespace http { DefaultNetwork::DefaultNetwork(std::shared_ptr network) - : current_statistics_bucket_{0}, network_{std::move(network)} {} + : current_statistics_bucket_{0}, + buckets_{std::make_shared>()}, + network_{std::move(network)} {} DefaultNetwork::~DefaultNetwork() = default; @@ -43,10 +45,13 @@ SendOutcome DefaultNetwork::Send(NetworkRequest request, Payload payload, AppendDefaultHeaders(request_headers); } + const auto buckets = buckets_; const auto bucket_id = current_statistics_bucket_.load(); - auto user_callback = [=](NetworkResponse response) { - LockStatistics(bucket_id, [&](Statistics& stats) { + auto user_callback = [buckets, bucket_id, + callback](NetworkResponse response) { + buckets->locked([&](BucketsContainer& container) { + auto& stats = container[bucket_id]; const auto status = response.GetStatus(); if (status < HttpStatusCode::OK || status >= HttpStatusCode::BAD_REQUEST) { @@ -81,10 +86,9 @@ void DefaultNetwork::SetCurrentBucket(uint8_t bucket_id) { } DefaultNetwork::Statistics DefaultNetwork::GetStatistics(uint8_t bucket_id) { - Statistics result; - LockStatistics(bucket_id, - [&](Statistics& statistics) { result = statistics; }); - return result; + return buckets_->locked([bucket_id](BucketsContainer& container) { + return container[bucket_id]; + }); } void DefaultNetwork::AppendUserAgent(Headers& request_headers) const { @@ -110,11 +114,5 @@ void DefaultNetwork::AppendDefaultHeaders(Headers& request_headers) const { default_headers_.end()); } -void DefaultNetwork::LockStatistics(uint8_t bucket_id, - std::function callback) { - buckets_.locked( - [&](BucketsContainer& container) { callback(container[bucket_id]); }); -} - } // namespace http } // namespace olp diff --git a/olp-cpp-sdk-core/src/http/DefaultNetwork.h b/olp-cpp-sdk-core/src/http/DefaultNetwork.h index 4739f30ff..60cc6f3b2 100644 --- a/olp-cpp-sdk-core/src/http/DefaultNetwork.h +++ b/olp-cpp-sdk-core/src/http/DefaultNetwork.h @@ -68,13 +68,10 @@ class DefaultNetwork final : public Network { void AppendUserAgent(Headers& request_headers) const; void AppendDefaultHeaders(Headers& request_headers) const; - void LockStatistics(uint8_t bucket_id, - std::function callback); - std::atomic current_statistics_bucket_; using BucketsContainer = std::unordered_map; - thread::Atomic buckets_; + std::shared_ptr> buckets_; std::mutex default_headers_mutex_; Headers default_headers_; diff --git a/olp-cpp-sdk-core/src/http/curl/NetworkCurl.cpp b/olp-cpp-sdk-core/src/http/curl/NetworkCurl.cpp index 956baabe3..7fc324782 100644 --- a/olp-cpp-sdk-core/src/http/curl/NetworkCurl.cpp +++ b/olp-cpp-sdk-core/src/http/curl/NetworkCurl.cpp @@ -410,10 +410,9 @@ void WithDiagnostics(NetworkResponse& response, CURL* handle) { } // anonymous namespace NetworkCurl::NetworkCurl(NetworkInitializationSettings settings) - : handles_(settings.max_requests_count), - static_handle_count_( - std::max(static_cast(1u), settings.max_requests_count / 4u)), - certificate_settings_(std::move(settings.certificate_settings)), + : impl_(std::make_shared(settings)), + certificate_settings_(std::make_shared( + std::move(settings.certificate_settings))), max_transfer_bytes_per_second_(settings.max_transfer_bytes_per_second) { OLP_SDK_LOG_TRACE(kLogTag, "Created NetworkCurl with address=" << this << ", handles_count=" @@ -472,7 +471,8 @@ NetworkCurl::NetworkCurl(NetworkInitializationSettings settings) version_data->ssl_version ? version_data->ssl_version : ""); if (settings.diagnostic_output_path) { - stderr_ = fopen(settings.diagnostic_output_path->c_str(), "a"); + stderr_ = std::shared_ptr( + fopen(settings.diagnostic_output_path->c_str(), "a"), fclose); if (!stderr_) { const auto* path_error = strerror(errno); OLP_SDK_LOG_ERROR_F( @@ -481,20 +481,19 @@ NetworkCurl::NetworkCurl(NetworkInitializationSettings settings) } else { OLP_SDK_LOG_INFO_F(kLogTag, "Using diagnostic output file, path=%s", settings.diagnostic_output_path->c_str()); - verbose_ = true; } } } +NetworkCurl::Impl::Impl(const NetworkInitializationSettings& settings) + : handles_(settings.max_requests_count) {} + NetworkCurl::~NetworkCurl() { OLP_SDK_LOG_TRACE(kLogTag, "Destroyed NetworkCurl object, this=" << this); Deinitialize(); if (curl_initialized_) { curl_global_cleanup(); } - if (stderr_) { - fclose(stderr_); - } } bool NetworkCurl::Initialize() { @@ -504,11 +503,27 @@ bool NetworkCurl::Initialize() { return false; } - if (state_ != WorkerState::STOPPED) { + if (*impl_->state_ != WorkerState::STOPPED) { OLP_SDK_LOG_DEBUG(kLogTag, "Already initialized, this=" << this); return true; } + if (!impl_->Initialize()) { + return false; + } + + std::unique_lock lock(impl_->event_mutex_); + // start worker thread + // Keep `impl_` alive until the thread is finished. + auto impl = impl_; + thread_ = std::thread([impl] { impl->Run(); }); + + impl_->event_condition_.wait(lock, [this] { return Initialized(); }); + + return true; +} + +bool NetworkCurl::Impl::Initialize() { #ifdef OLP_SDK_NETWORK_HAS_PIPE2 if (pipe2(pipe_, O_NONBLOCK)) { OLP_SDK_LOG_ERROR(kLogTag, "pipe2 failed, this=" << this); @@ -553,13 +568,6 @@ bool NetworkCurl::Initialize() { const auto connects_cache_size = handles_.size() * 4; curl_multi_setopt(curl_, CURLMOPT_MAXCONNECTS, connects_cache_size); - std::unique_lock lock(event_mutex_); - // start worker thread - thread_ = std::thread(&NetworkCurl::Run, this); - - event_condition_.wait(lock, - [this] { return state_ == WorkerState::STARTED; }); - return true; } @@ -575,37 +583,28 @@ void NetworkCurl::Deinitialize() { OLP_SDK_LOG_TRACE(kLogTag, "Deinitialize NetworkCurl, this=" << this); { - std::lock_guard lock(event_mutex_); - state_ = WorkerState::STOPPING; + std::lock_guard lock(impl_->event_mutex_); + *impl_->state_ = WorkerState::STOPPING; } - // We should not destroy this thread from itself if (thread_.get_id() != std::this_thread::get_id()) { - event_condition_.notify_all(); -#if defined(OLP_SDK_NETWORK_HAS_PIPE) || defined(OLP_SDK_NETWORK_HAS_PIPE2) - char tmp = 1; - if (write(pipe_[1], &tmp, 1) < 0) { - OLP_SDK_LOG_INFO(kLogTag, __PRETTY_FUNCTION__ - << ". Failed to write pipe. Error " - << errno); - } -#endif + impl_->NotifyEvent(); thread_.join(); - -#if (defined OLP_SDK_NETWORK_HAS_PIPE) || (defined OLP_SDK_NETWORK_HAS_PIPE2) - close(pipe_[0]); - close(pipe_[1]); -#endif } else { - // We are trying to stop the very thread we are in. This is not recommended, - // but we try to handle it gracefully. This could happen by calling from one - // of the static functions (rxFunction or headerFunction) that was passed to - // the cURL as callbacks. + // We are trying to stop the very thread we are in. This could happen if the + // last instance of NetworkCurl is destroyed from one of the callbacks. For + // example, if caller made a callback to capture a shared_ptr to NetworkCurl + // and then released its own instance of NetworkCurl. That way, the captured + // shared_ptr will be the last one and will be destroyed in the callback + // from the worker thread. + // Since this is the last cleanup action to perform here, and the thread + // itself does the teardown to release resources, we can safely detach the + // thread and let it finish its work. thread_.detach(); } } -void NetworkCurl::Teardown() { +void NetworkCurl::Impl::Teardown() { std::vector > completed_messages; { std::lock_guard lock(event_mutex_); @@ -621,7 +620,6 @@ void NetworkCurl::Teardown() { } handle.curl_handle = nullptr; } - handle.self.reset(); } // cURL teardown @@ -638,28 +636,20 @@ void NetworkCurl::Teardown() { .WithError("Offline: network is deinitialized")); } } -} -bool NetworkCurl::IsStarted() const { return state_ == WorkerState::STARTED; } +#if (defined OLP_SDK_NETWORK_HAS_PIPE) || (defined OLP_SDK_NETWORK_HAS_PIPE2) + close(pipe_[0]); + close(pipe_[1]); +#endif +} -bool NetworkCurl::Initialized() const { return IsStarted(); } +bool NetworkCurl::IsStarted() const { return impl_->IsStarted(); } -bool NetworkCurl::Ready() { - if (!IsStarted()) { - return false; - } - std::lock_guard lock(event_mutex_); - return std::any_of( - std::begin(handles_), std::end(handles_), - [](const RequestHandle& handle) { return !handle.in_use; }); +bool NetworkCurl::Impl::IsStarted() const { + return *state_ == WorkerState::STARTED; } -size_t NetworkCurl::AmountPending() { - std::lock_guard lock(event_mutex_); - return static_cast( - std::count_if(std::begin(handles_), std::end(handles_), - [](const RequestHandle& handle) { return handle.in_use; })); -} +bool NetworkCurl::Initialized() const { return IsStarted(); } SendOutcome NetworkCurl::Send(NetworkRequest request, std::shared_ptr payload, @@ -675,7 +665,7 @@ SendOutcome NetworkCurl::Send(NetworkRequest request, RequestId request_id{}; { - std::lock_guard lock(event_mutex_); + std::lock_guard lock(impl_->event_mutex_); request_id = request_id_counter_; if (request_id_counter_ == @@ -712,9 +702,9 @@ ErrorCode NetworkCurl::SendImplementation( const auto& config = request.GetSettings(); RequestHandle* handle = [&] { - std::lock_guard lock(event_mutex_); + std::lock_guard lock(impl_->event_mutex_); - auto* request_handle = InitRequestHandleUnsafe(); + auto* request_handle = impl_->InitRequestHandleUnsafe(); if (request_handle) { request_handle->id = id; @@ -724,6 +714,8 @@ ErrorCode NetworkCurl::SendImplementation( request_handle->out_data_stream = payload; request_handle->request_body = request.GetBody(); request_handle->request_headers = SetupHeaders(request.GetHeaders()); + request_handle->log_file = stderr_; + request_handle->certificate_settings = certificate_settings_; } return request_handle; @@ -742,11 +734,9 @@ ErrorCode NetworkCurl::SendImplementation( curl_easy_setopt(curl_handle, CURLOPT_NOSIGNAL, 1L); - if (verbose_) { + if (stderr_ != nullptr) { curl_easy_setopt(curl_handle, CURLOPT_VERBOSE, 1L); - if (stderr_ != nullptr) { - curl_easy_setopt(curl_handle, CURLOPT_STDERR, stderr_); - } + curl_easy_setopt(curl_handle, CURLOPT_STDERR, stderr_.get()); } else { curl_easy_setopt(curl_handle, CURLOPT_VERBOSE, 0L); } @@ -800,7 +790,7 @@ ErrorCode NetworkCurl::SendImplementation( // "unable to set private key file: '(memory blob)' type PEM". // Therefore, skip the entire blob-based cert setup and let InjectEnvelopeKey // set both the client certificate and the EVP_PKEY via SSL_CTX_*. - if (ssl_certificates_blobs_ && !certificate_settings_.pkey_handle) { + if (ssl_certificates_blobs_ && !certificate_settings_->pkey_handle) { #else if (ssl_certificates_blobs_) { #endif @@ -829,7 +819,7 @@ ErrorCode NetworkCurl::SendImplementation( curl_easy_setopt(curl_handle, CURLOPT_SSL_VERIFYHOST, 2L); #ifdef OLP_SDK_ENABLE_ENVELOPE_PKEY - if (certificate_settings_.pkey_handle) { + if (certificate_settings_->pkey_handle) { curl_easy_setopt(curl_handle, CURLOPT_SSL_CTX_FUNCTION, &NetworkCurl::InjectEnvelopeKey); curl_easy_setopt(curl_handle, CURLOPT_SSL_CTX_DATA, handle); @@ -853,11 +843,9 @@ ErrorCode NetworkCurl::SendImplementation( curl_easy_setopt(curl_handle, CURLOPT_CONNECTTIMEOUT_MS, connect_timeout_ms); curl_easy_setopt(curl_handle, CURLOPT_TIMEOUT_MS, timeout_ms); - curl_easy_setopt(curl_handle, CURLOPT_WRITEFUNCTION, - &NetworkCurl::RxFunction); + curl_easy_setopt(curl_handle, CURLOPT_WRITEFUNCTION, &Impl::RxFunction); curl_easy_setopt(curl_handle, CURLOPT_WRITEDATA, handle); - curl_easy_setopt(curl_handle, CURLOPT_HEADERFUNCTION, - &NetworkCurl::HeaderFunction); + curl_easy_setopt(curl_handle, CURLOPT_HEADERFUNCTION, &Impl::HeaderFunction); curl_easy_setopt(curl_handle, CURLOPT_HEADERDATA, handle); curl_easy_setopt(curl_handle, CURLOPT_FAILONERROR, 0L); if (stderr_ == nullptr) { @@ -900,8 +888,8 @@ ErrorCode NetworkCurl::SendImplementation( #endif { - std::lock_guard lock(event_mutex_); - AddEvent(EventInfo::Type::SEND_EVENT, handle); + std::lock_guard lock(impl_->event_mutex_); + impl_->AddEvent(EventInfo::Type::SEND_EVENT, handle); } return ErrorCode::SUCCESS; // NetworkProtocol::ErrorNone; } // namespace http @@ -911,11 +899,11 @@ void NetworkCurl::Cancel(RequestId id) { OLP_SDK_LOG_ERROR(kLogTag, "Cancel failed - network is offline, id=" << id); return; } - std::lock_guard lock(event_mutex_); - for (auto& handle : handles_) { + std::lock_guard lock(impl_->event_mutex_); + for (auto& handle : impl_->handles_) { if (handle.in_use && (handle.id == id)) { handle.is_cancelled = true; - AddEvent(EventInfo::Type::CANCEL_EVENT, &handle); + impl_->AddEvent(EventInfo::Type::CANCEL_EVENT, &handle); OLP_SDK_LOG_DEBUG(kLogTag, "Cancel request with id=" << id); return; @@ -924,8 +912,12 @@ void NetworkCurl::Cancel(RequestId id) { OLP_SDK_LOG_WARNING(kLogTag, "Cancel non-existing request with id=" << id); } -void NetworkCurl::AddEvent(EventInfo::Type type, RequestHandle* handle) { +void NetworkCurl::Impl::AddEvent(EventInfo::Type type, RequestHandle* handle) { events_.emplace_back(type, handle); + NotifyEvent(); +} + +void NetworkCurl::Impl::NotifyEvent() { event_condition_.notify_all(); #if (defined OLP_SDK_NETWORK_HAS_PIPE) || (defined OLP_SDK_NETWORK_HAS_PIPE2) @@ -933,16 +925,16 @@ void NetworkCurl::AddEvent(EventInfo::Type type, RequestHandle* handle) { // the network thread is currently blocked there. char tmp = 1; if (write(pipe_[1], &tmp, 1) < 0) { - OLP_SDK_LOG_WARNING(kLogTag, "AddEvent - failed for id=" - << handle->id << ", err=" << errno); + OLP_SDK_LOG_WARNING(kLogTag, __PRETTY_FUNCTION__ + << " - failed to write pipe, errno=" + << errno); } #else - OLP_SDK_LOG_WARNING(kLogTag, - "AddEvent for id=" << handle->id << " - no pipe"); + OLP_SDK_LOG_WARNING(kLogTag, __PRETTY_FUNCTION__ << " - no pipe available"); #endif } -NetworkCurl::RequestHandle* NetworkCurl::InitRequestHandleUnsafe() { +NetworkCurl::RequestHandle* NetworkCurl::Impl::InitRequestHandleUnsafe() { const auto unused_handle_it = std::find_if(handles_.begin(), handles_.end(), [](const RequestHandle& request_handle) { @@ -962,15 +954,15 @@ NetworkCurl::RequestHandle* NetworkCurl::InitRequestHandleUnsafe() { } unused_handle_it->in_use = true; - unused_handle_it->self = shared_from_this(); unused_handle_it->send_time = std::chrono::steady_clock::now(); unused_handle_it->log_context = logging::GetContext(); + unused_handle_it->network_state = state_; return &*unused_handle_it; } -void NetworkCurl::ReleaseHandleUnlocked(RequestHandle* handle, - bool cleanup_easy_handle) { +void NetworkCurl::Impl::ReleaseHandleUnlocked(RequestHandle* handle, + bool cleanup_easy_handle) { // Reset the RequestHandle to default, but keep the curl_handle. std::shared_ptr curl_handle; std::swap(curl_handle, handle->curl_handle); @@ -995,19 +987,14 @@ void NetworkCurl::ReleaseHandleUnlocked(RequestHandle* handle, OLP_SDK_CORE_UNUSED(cleanup_easy_handle); } -size_t NetworkCurl::RxFunction(void* ptr, size_t size, size_t nmemb, - RequestHandle* handle) { +size_t NetworkCurl::Impl::RxFunction(void* ptr, size_t size, size_t nmemb, + RequestHandle* handle) { const size_t len = size * nmemb; OLP_SDK_LOG_TRACE(kLogTag, "Received " << len << " bytes for id=" << handle->id); - std::shared_ptr that = handle->self.lock(); - if (!that) { - return len; - } - - if (that->IsStarted() && !handle->is_cancelled) { + if (*handle->network_state == WorkerState::STARTED && !handle->is_cancelled) { if (handle->out_data_callback) { handle->out_data_callback(static_cast(ptr), handle->bytes_received, len); @@ -1032,29 +1019,30 @@ size_t NetworkCurl::RxFunction(void* ptr, size_t size, size_t nmemb, } // In case we have curl verbose and stderr enabled to log the error content - if (that->stderr_) { + if (handle->log_file) { long http_status = 0L; curl_easy_getinfo(handle->curl_handle.get(), CURLINFO_RESPONSE_CODE, &http_status); if (http_status >= http::HttpStatusCode::BAD_REQUEST) { // Log the error content to help troubleshooting - fprintf(that->stderr_, "\n---ERRORCONTENT BEGIN HANDLE=%p BLOCKSIZE=%u\n", - handle, static_cast(size * nmemb)); - fwrite(ptr, size, nmemb, that->stderr_); - fprintf(that->stderr_, "\n---ERRORCONTENT END HANDLE=%p BLOCKSIZE=%u\n", - handle, static_cast(size * nmemb)); + fprintf(handle->log_file.get(), + "\n---ERRORCONTENT BEGIN HANDLE=%p BLOCKSIZE=%u\n", handle, + static_cast(size * nmemb)); + fwrite(ptr, size, nmemb, handle->log_file.get()); + fprintf(handle->log_file.get(), + "\n---ERRORCONTENT END HANDLE=%p BLOCKSIZE=%u\n", handle, + static_cast(size * nmemb)); } } return len; } -size_t NetworkCurl::HeaderFunction(char* ptr, size_t size, size_t nitems, - RequestHandle* handle) { +size_t NetworkCurl::Impl::HeaderFunction(char* ptr, size_t size, size_t nitems, + RequestHandle* handle) { const size_t len = size * nitems; - std::shared_ptr that = handle->self.lock(); - if (!that || !that->IsStarted() || handle->is_cancelled) { + if (*handle->network_state != WorkerState::STARTED || handle->is_cancelled) { return len; } @@ -1084,12 +1072,12 @@ size_t NetworkCurl::HeaderFunction(char* ptr, size_t size, size_t nitems, } // Callback with header key+value - handle->out_header_callback(key, value); + handle->out_header_callback(std::move(key), std::move(value)); return len; } -void NetworkCurl::CompleteMessage(CURL* curl_handle, CURLcode result) { +void NetworkCurl::Impl::CompleteMessage(CURL* curl_handle, CURLcode result) { std::unique_lock lock(event_mutex_); // When curl returns an error of the handle, it is possible that error @@ -1182,7 +1170,8 @@ void NetworkCurl::CompleteMessage(CURL* curl_handle, CURLcode result) { callback(response); } -NetworkCurl::RequestHandle* NetworkCurl::FindRequestHandle(const CURL* handle) { +NetworkCurl::RequestHandle* NetworkCurl::Impl::FindRequestHandle( + const CURL* handle) { for (auto& request_handle : handles_) { if (request_handle.in_use && request_handle.curl_handle.get() == handle) { return &request_handle; @@ -1191,12 +1180,12 @@ NetworkCurl::RequestHandle* NetworkCurl::FindRequestHandle(const CURL* handle) { return nullptr; } -void NetworkCurl::Run() { +void NetworkCurl::Impl::Run() { olp::utils::Thread::SetCurrentThreadName(kCurlThreadName); { std::lock_guard lock(event_mutex_); - state_ = WorkerState::STARTED; + *state_ = WorkerState::STARTED; event_condition_.notify_one(); } @@ -1406,24 +1395,21 @@ void NetworkCurl::Run() { } Teardown(); - { - std::lock_guard lock(event_mutex_); - state_ = WorkerState::STOPPED; - } + *state_ = WorkerState::STOPPED; OLP_SDK_LOG_DEBUG(kLogTag, "Thread exit, this=" << this); } #ifdef OLP_SDK_CURL_HAS_SUPPORT_SSL_BLOBS void NetworkCurl::SetupCertificateBlobs() { - if (certificate_settings_.client_cert_file_blob.empty() && - certificate_settings_.client_key_file_blob.empty() && - certificate_settings_.cert_file_blob.empty()) { + if (certificate_settings_->client_cert_file_blob.empty() && + certificate_settings_->client_key_file_blob.empty() && + certificate_settings_->cert_file_blob.empty()) { OLP_SDK_LOG_INFO(kLogTag, "No certificate blobs provided"); return; } auto setup_blob = [](SslCertificateBlobs::OptionalBlob& blob, - std::string& src) { + const std::string& src) { if (src.empty()) { blob.reset(); return; @@ -1437,11 +1423,11 @@ void NetworkCurl::SetupCertificateBlobs() { ssl_certificates_blobs_ = SslCertificateBlobs{}; setup_blob(ssl_certificates_blobs_->ssl_cert_blob, - certificate_settings_.client_cert_file_blob); + certificate_settings_->client_cert_file_blob); setup_blob(ssl_certificates_blobs_->ssl_key_blob, - certificate_settings_.client_key_file_blob); + certificate_settings_->client_key_file_blob); setup_blob(ssl_certificates_blobs_->ca_info_blob, - certificate_settings_.cert_file_blob); + certificate_settings_->cert_file_blob); auto to_log_str = [](const SslCertificateBlobs::OptionalBlob& blob) { return blob ? "" : ""; @@ -1460,9 +1446,9 @@ void NetworkCurl::SetupCertificateBlobs() { #ifdef OLP_SDK_USE_MD5_CERT_LOOKUP CURLcode NetworkCurl::AddMd5LookupMethod(CURL*, SSL_CTX* ssl_ctx, RequestHandle* handle) { - auto self = handle->self.lock(); - if (!self) { - OLP_SDK_LOG_ERROR(kLogTag, "Unable to lock cURL handle"); + if (*handle->network_state != WorkerState::STARTED || handle->is_cancelled) { + OLP_SDK_LOG_ERROR(kLogTag, + "Network is not started or request is cancelled"); return CURLE_ABORTED_BY_CALLBACK; } @@ -1491,13 +1477,7 @@ CURLcode NetworkCurl::AddMd5LookupMethod(CURL*, SSL_CTX* ssl_ctx, #ifdef OLP_SDK_ENABLE_ENVELOPE_PKEY CURLcode NetworkCurl::InjectEnvelopeKey(CURL*, SSL_CTX* ssl_ctx, RequestHandle* handle) { - auto self = handle->self.lock(); - if (!self) { - OLP_SDK_LOG_ERROR(kLogTag, "Unable to lock cURL handle"); - return CURLE_ABORTED_BY_CALLBACK; - } - - const auto& cert_blob = self->certificate_settings_.client_cert_file_blob; + const auto& cert_blob = handle->certificate_settings->client_cert_file_blob; if (!cert_blob.empty()) { BIO* bio = BIO_new_mem_buf(cert_blob.data(), static_cast(cert_blob.size())); @@ -1523,7 +1503,7 @@ CURLcode NetworkCurl::InjectEnvelopeKey(CURL*, SSL_CTX* ssl_ctx, } } - const auto& ca_blob = self->certificate_settings_.cert_file_blob; + const auto& ca_blob = handle->certificate_settings->cert_file_blob; if (!ca_blob.empty()) { BIO* bio = BIO_new_mem_buf(ca_blob.data(), static_cast(ca_blob.size())); @@ -1563,7 +1543,7 @@ CURLcode NetworkCurl::InjectEnvelopeKey(CURL*, SSL_CTX* ssl_ctx, // ERR_get_error() in the error path below reflects the actual key error ERR_clear_error(); if (SSL_CTX_use_PrivateKey(ssl_ctx, - self->certificate_settings_.pkey_handle) != 1) { + handle->certificate_settings->pkey_handle) != 1) { OLP_SDK_LOG_ERROR( kLogTag, "Failed to use provided EVP_PKEY, error=" << ERR_error_string( ERR_get_error(), nullptr)); diff --git a/olp-cpp-sdk-core/src/http/curl/NetworkCurl.h b/olp-cpp-sdk-core/src/http/curl/NetworkCurl.h index 9503ce90f..c41327563 100644 --- a/olp-cpp-sdk-core/src/http/curl/NetworkCurl.h +++ b/olp-cpp-sdk-core/src/http/curl/NetworkCurl.h @@ -69,8 +69,7 @@ namespace http { /** * @brief The implementation of Network based on cURL. */ -class NetworkCurl : public Network, - public std::enable_shared_from_this { +class NetworkCurl : public Network { public: /** * @brief NetworkCurl constructor. @@ -115,6 +114,15 @@ class NetworkCurl : public Network, void Cancel(RequestId id) override; private: + /** + * @brief @copydoc NetworkCurl::Impl::state_ + */ + enum class WorkerState { + STOPPED, ///< The worker thread is not started. + STARTED, ///< The worker thread is running. + STOPPING, ///< The worker thread will be stopped soon. + }; + /** * @brief Context of each network request. */ @@ -129,7 +137,6 @@ class NetworkCurl : public Network, std::uint64_t bytes_received{0}; std::chrono::steady_clock::time_point send_time{}; - std::weak_ptr self{}; std::shared_ptr curl_handle; RequestId id{}; @@ -138,6 +145,9 @@ class NetworkCurl : public Network, bool is_cancelled{false}; char error_text[CURL_ERROR_SIZE]{}; std::shared_ptr log_context; + std::shared_ptr log_file; + std::shared_ptr> network_state; + std::shared_ptr certificate_settings; }; /** @@ -221,85 +231,11 @@ class NetworkCurl : public Network, */ bool Initialized() const; - /** - * @brief. Check whether NetworkCurl has resources to handle more requests. - * @return @c true if curl network has free network connections, - * @c false otherwise. - */ - bool Ready(); - - /** - * @brief Return count of pending network requests. - * @return Number of pending network requests. - */ - size_t AmountPending(); - - /** - * @brief Find a handle in handles_ by curl handle. - * @param[in] handle CURL handle. - * @return Pointer to the RequestHandle. - */ - RequestHandle* FindRequestHandle(const CURL* handle); - - /** - * @brief Allocate new handle RequestHandle. - * @note Must be protected by event_mutex_ - * - * @return Pointer to the allocated RequestHandle. - */ - RequestHandle* InitRequestHandleUnsafe(); - - /** - * @brief Reset the handle after network request is done. - * @param[in] handle Request handle. - * @param[in] cleanup_handle If true then handle is completelly release. - * Otherwise, a handle is reset, which preserves DNS cache, Session ID cache, - * cookies, and so on. - */ - static void ReleaseHandleUnlocked(RequestHandle* handle, bool cleanup_handle); - - /** - * @brief Routine that is called when the last bit of response is received. - * - * @param[in] curl_handle CURL handle associated with request. - * @param[in] result CURL return code. - */ - void CompleteMessage(CURL* curl_handle, CURLcode result); - - /** - * @brief CURL read callback. - */ - static size_t RxFunction(void* ptr, size_t size, size_t nmemb, - RequestHandle* handle); - - /** - * @brief CURL header callback. - */ - static size_t HeaderFunction(char* ptr, size_t size, size_t nmemb, - RequestHandle* handle); - - /** - * @brief The worker thread's main method. - */ - void Run(); - - /** - * @brief Free resources after the thread terminates. - */ - void Teardown(); - - /** - * @brief Notify worker thread on some event. - * @param[in] type Event type. - * @param[in] handle Related RequestHandle. - */ - void AddEvent(EventInfo::Type type, RequestHandle* handle); - /** * @brief Checks whether the worker thread is started. * @return @c true if the thread is started, @c false otherwise. */ - inline bool IsStarted() const; + bool IsStarted() const; #ifdef OLP_SDK_CURL_HAS_SUPPORT_SSL_BLOBS /** @@ -334,17 +270,115 @@ class NetworkCurl : public Network, RequestHandle* handle); #endif - /// Contexts for every network request. - std::vector handles_; + struct Impl { + Impl(const NetworkInitializationSettings& settings); + + /** + * @brief Initialize internal data structures. + * @return @c true if initialized successfully, @c false otherwise. + */ + bool Initialize(); + + /** + * @brief Find a handle in handles_ by curl handle. + * @param[in] handle CURL handle. + * @return Pointer to the RequestHandle. + */ + RequestHandle* FindRequestHandle(const CURL* handle); + + /** + * @brief Allocate new handle RequestHandle. + * @note Must be protected by event_mutex_ + * + * @return Pointer to the allocated RequestHandle. + */ + RequestHandle* InitRequestHandleUnsafe(); + + /** + * @brief Reset the handle after network request is done. + * @param[in] handle Request handle. + * @param[in] cleanup_handle If true then handle is completelly release. + * Otherwise, a handle is reset, which preserves DNS cache, Session ID + * cache, cookies, and so on. + */ + static void ReleaseHandleUnlocked(RequestHandle* handle, + bool cleanup_handle); + + /** + * @brief Routine that is called when the last bit of response is received. + * + * @param[in] curl_handle CURL handle associated with request. + * @param[in] result CURL return code. + */ + void CompleteMessage(CURL* curl_handle, CURLcode result); + + /** + * @brief CURL read callback. + */ + static size_t RxFunction(void* ptr, size_t size, size_t nmemb, + RequestHandle* handle); + + /** + * @brief CURL header callback. + */ + static size_t HeaderFunction(char* ptr, size_t size, size_t nmemb, + RequestHandle* handle); + + /** + * @brief The worker thread's main method. + */ + void Run(); + + /** + * @brief Free resources after the thread terminates. + */ + void Teardown(); + + /** + * @brief Notify worker thread on some event. + * @param[in] type Event type. + * @param[in] handle Related RequestHandle. + */ + void AddEvent(EventInfo::Type type, RequestHandle* handle); + + /** + * @brief Notify worker thread to wake up on some event. + */ + void NotifyEvent(); + + /** + * @brief Checks whether the worker thread is started. + * @return @c true if the thread is started, @c false otherwise. + */ + bool IsStarted() const; + + /// Contexts for every network request. + std::vector handles_; + + /// Condition variable used to notify worker thread on event. + std::condition_variable event_condition_; - /// Number of CURL easy handles that are always opened. - const size_t static_handle_count_; + /// Synchronization mutex used during event processing. + std::mutex event_mutex_; - /// Condition variable used to notify worker thread on event. - std::condition_variable event_condition_; + /// The state of the worker thread. + std::shared_ptr> state_{ + std::make_shared>(WorkerState::STOPPED)}; - /// Synchronization mutex used during event processing. - std::mutex event_mutex_; + private: + /// Queue of events passed to worker thread. + std::deque events_{}; + + /// CURL multi handle. Shared among all network requests. + CURLM* curl_{nullptr}; + +#if (defined OLP_SDK_NETWORK_HAS_PIPE) || (defined OLP_SDK_NETWORK_HAS_PIPE2) + /// UNIX Pipe used to notify sleeping worker thread during select() call. + int pipe_[2]{}; +#endif + }; + + std::shared_ptr impl_; /// Synchronization mutex prevents parallel initialization of network. std::mutex init_mutex_; @@ -356,39 +390,15 @@ class NetworkCurl : public Network, RequestId request_id_counter_{ static_cast(RequestIdConstants::RequestIdMin)}; - /** - * @brief @copydoc NetworkCurl::state_ - */ - enum class WorkerState { - STOPPED, ///< The worker thread is not started. - STARTED, ///< The worker thread is running. - STOPPING, ///< The worker thread will be stopped soon. - }; - - /// The state of the worker thread. - std::atomic state_{WorkerState::STOPPED}; - - /// Queue of events passed to worker thread. - std::deque events_{}; - - /// CURL multi handle. Shared among all network requests. - CURLM* curl_{nullptr}; - - /// Turn on and off verbose mode for CURL. - bool verbose_{false}; - /// Set custom stderr for CURL. - FILE* stderr_{nullptr}; - - /// UNIX Pipe used to notify sleeping worker thread during select() call. - int pipe_[2]{}; + std::shared_ptr stderr_; /// Stores value if `curl_global_init()` was successful on construction. bool curl_initialized_; /// Store original certificate setting in order to reference them in the SSL /// blobs so cURL does not need to copy them. - CertificateSettings certificate_settings_; + std::shared_ptr certificate_settings_; /// Maximum transfer rate in bytes per second applied per connection (0 = /// unlimited). diff --git a/olp-cpp-sdk-core/tests/CMakeLists.txt b/olp-cpp-sdk-core/tests/CMakeLists.txt index fc13fc82e..ed29027bc 100644 --- a/olp-cpp-sdk-core/tests/CMakeLists.txt +++ b/olp-cpp-sdk-core/tests/CMakeLists.txt @@ -71,6 +71,9 @@ set(OLP_CPP_SDK_CORE_TESTS_SOURCES ./thread/ThreadPoolTaskSchedulerTest.cpp ./http/NetworkSettingsTest.cpp + ./http/LoopbackHttpServer.cpp + ./http/LoopbackHttpServer.h + ./http/NetworkTest.cpp ./http/NetworkUtils.cpp ./utils/JsonTest.cpp @@ -131,4 +134,8 @@ else() ../src/cache ) + if (WIN32) + target_link_libraries(olp-cpp-sdk-core-tests PRIVATE ws2_32) + endif() + endif() diff --git a/olp-cpp-sdk-core/tests/http/LoopbackHttpServer.cpp b/olp-cpp-sdk-core/tests/http/LoopbackHttpServer.cpp new file mode 100644 index 000000000..0376f0afa --- /dev/null +++ b/olp-cpp-sdk-core/tests/http/LoopbackHttpServer.cpp @@ -0,0 +1,257 @@ +/* + * Copyright (C) 2026 HERE Europe B.V. + * + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * License-Filename: LICENSE + */ + +#include "LoopbackHttpServer.h" + +#include +#include +#include + +#ifdef _WIN32 +#include +#include +#include +#else +#include +#include +#include +#include +#endif + +#include + +namespace olp { +namespace http { +namespace test { + +namespace { + +#ifdef _WIN32 +using SocketHandle = SOCKET; +constexpr SocketHandle kInvalidSocket = INVALID_SOCKET; + +bool EnsureWinSockInitialized() { + static std::once_flag once; + static bool initialized = false; + std::call_once(once, []() { + WSADATA wsadata; + initialized = (WSAStartup(MAKEWORD(2, 2), &wsadata) == 0); + }); + + return initialized; +} + +void CloseSocket(SocketHandle socket) { + if (socket != kInvalidSocket) { + (void)::closesocket(socket); + } +} + +#else +using SocketHandle = int; +constexpr SocketHandle kInvalidSocket = -1; + +void CloseSocket(SocketHandle socket) { + if (socket != kInvalidSocket) { + (void)::close(socket); + } +} +#endif + +bool IsValidSocket(SocketHandle socket) { return socket != kInvalidSocket; } + +} // namespace + +LoopbackHttpServer::LoopbackHttpServer() { EXPECT_TRUE(Start()); } + +LoopbackHttpServer::~LoopbackHttpServer() { Stop(); } + +std::uint16_t LoopbackHttpServer::Port() const { return port_; } + +std::string LoopbackHttpServer::CreateUrl() const { + return std::string("http://127.0.0.1:") + std::to_string(port_) + "/"; +} + +bool LoopbackHttpServer::Start() { +#ifdef _WIN32 + if (!EnsureWinSockInitialized()) { + ADD_FAILURE() << "Failed to initialize WinSock"; + return false; + } +#endif + + SocketHandle listen_socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (!IsValidSocket(listen_socket)) { + ADD_FAILURE() << "Failed to create loopback test socket"; + return false; + } + + int reuse = 1; + const int setopt_result = + ::setsockopt(listen_socket, SOL_SOCKET, SO_REUSEADDR, + reinterpret_cast(&reuse), sizeof(reuse)); + if (setopt_result != 0) { + CloseSocket(listen_socket); + ADD_FAILURE() << "Failed to set SO_REUSEADDR on loopback socket"; + return false; + } + + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = htons(0); + if (::bind(listen_socket, reinterpret_cast(&address), + sizeof(address)) != 0) { + CloseSocket(listen_socket); + ADD_FAILURE() << "Failed to bind loopback socket"; + return false; + } + + socklen_t address_length = sizeof(address); + if (::getsockname(listen_socket, reinterpret_cast(&address), + &address_length) != 0) { + CloseSocket(listen_socket); + ADD_FAILURE() << "Failed to read loopback socket port"; + return false; + } + port_ = ntohs(address.sin_port); + + if (::listen(listen_socket, 8) != 0) { + CloseSocket(listen_socket); + ADD_FAILURE() << "Failed to listen on loopback socket"; + return false; + } + +#ifdef _WIN32 + listen_socket_ = static_cast(listen_socket); +#else + listen_socket_ = listen_socket; +#endif + + running_.store(true); + server_thread_ = std::thread([this]() { ServeLoop(); }); + return true; +} + +void LoopbackHttpServer::Stop() { + if (!running_.exchange(false)) { + return; + } + +#ifdef _WIN32 + const SocketHandle listen_socket = static_cast(listen_socket_); +#else + const SocketHandle listen_socket = listen_socket_; +#endif + + // Unblock accept() so ServeLoop can stop quickly. + const SocketHandle wake_socket = ::socket(AF_INET, SOCK_STREAM, 0); + if (IsValidSocket(wake_socket)) { + sockaddr_in wake_address{}; + wake_address.sin_family = AF_INET; + wake_address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + wake_address.sin_port = htons(port_); + const int connect_result = + ::connect(wake_socket, reinterpret_cast(&wake_address), + sizeof(wake_address)); + if (connect_result == -1) { + // Best-effort wakeup only; close and continue shutdown. + } + CloseSocket(wake_socket); + } + + if (server_thread_.joinable()) { + server_thread_.join(); + } + + CloseSocket(listen_socket); +#ifdef _WIN32 + listen_socket_ = ~static_cast(0); +#else + listen_socket_ = kInvalidSocket; +#endif +} + +void LoopbackHttpServer::ServeLoop() { +#ifdef _WIN32 + const SocketHandle listen_socket = static_cast(listen_socket_); +#else + const SocketHandle listen_socket = listen_socket_; +#endif + + while (running_.load()) { + const SocketHandle client_socket = + ::accept(listen_socket, nullptr, nullptr); + if (!IsValidSocket(client_socket)) { + continue; + } + + std::string request; + char request_buffer[1024]; + for (;;) { + const int received = + ::recv(client_socket, request_buffer, sizeof(request_buffer), 0); + if (received <= 0) { + break; + } + + request.append(request_buffer, static_cast(received)); + if (request.find("\r\n\r\n") != std::string::npos) { + break; + } + } + + std::string body(64 * 1024, 'a'); + std::string response_headers = + "HTTP/1.1 200 OK\r\n" + "Content-Type: text/plain\r\n" + "Connection: close\r\n" + "Content-Length: " + + std::to_string(body.size()) + "\r\n\r\n"; + std::string response = response_headers + body; + + std::size_t written = 0; + while (written < response.size()) { + const std::size_t chunk_size = + std::min(response.size() - written, 16 * 1024); + const int sent = ::send(client_socket, response.data() + written, + static_cast(chunk_size), 0); + if (sent <= 0) { + break; + } + written += static_cast(sent); + } + +#ifdef _WIN32 + (void)::shutdown(client_socket, SD_BOTH); +#else + (void)::shutdown(client_socket, SHUT_RDWR); +#endif + CloseSocket(client_socket); + } +} + +std::string CreateLoopbackUrl() { + static LoopbackHttpServer server; + return server.CreateUrl(); +} + +} // namespace test +} // namespace http +} // namespace olp diff --git a/olp-cpp-sdk-core/tests/http/LoopbackHttpServer.h b/olp-cpp-sdk-core/tests/http/LoopbackHttpServer.h new file mode 100644 index 000000000..4885915b6 --- /dev/null +++ b/olp-cpp-sdk-core/tests/http/LoopbackHttpServer.h @@ -0,0 +1,59 @@ +/* + * Copyright (C) 2026 HERE Europe B.V. + * + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * License-Filename: LICENSE + */ + +#pragma once + +#include +#include +#include +#include + +namespace olp { +namespace http { +namespace test { + +class LoopbackHttpServer final { + public: + LoopbackHttpServer(); + ~LoopbackHttpServer(); + + std::uint16_t Port() const; + std::string CreateUrl() const; + + private: + bool Start(); + void Stop(); + void ServeLoop(); + + private: + std::atomic running_{false}; +#ifdef _WIN32 + std::uintptr_t listen_socket_{~static_cast(0)}; +#else + int listen_socket_{-1}; +#endif + std::uint16_t port_{0}; + std::thread server_thread_; +}; + +std::string CreateLoopbackUrl(); + +} // namespace test +} // namespace http +} // namespace olp diff --git a/olp-cpp-sdk-core/tests/http/NetworkTest.cpp b/olp-cpp-sdk-core/tests/http/NetworkTest.cpp new file mode 100644 index 000000000..3255a311f --- /dev/null +++ b/olp-cpp-sdk-core/tests/http/NetworkTest.cpp @@ -0,0 +1,221 @@ +/* + * Copyright (C) 2026 HERE Europe B.V. + * + * 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. + * + * SPDX-License-Identifier: Apache-2.0 + * License-Filename: LICENSE + */ + +#include + +#include +#include +#include +#include +#include +#include +#include + +#include + +#include "LoopbackHttpServer.h" + +namespace olp { +namespace http { +namespace test { + +namespace { + +enum class NetworkDestroyContext { Synchronous, Asynchronous }; + +using NetworkLifetimeTest = ::testing::TestWithParam; + +TEST_P(NetworkLifetimeTest, + CallbackReleasesNetworkObjectShouldStayAliveDuringCallback) { + auto network = CreateDefaultNetwork({}); + + std::promise promise; + auto future = promise.get_future(); + + NetworkRequest req(CreateLoopbackUrl()); + + auto callback = [&network, &promise](NetworkResponse /*resp*/) { + auto network_reset = [&network, &promise]() { + network.reset(); + promise.set_value(); + }; + switch (GetParam()) { + case NetworkDestroyContext::Synchronous: + network_reset(); + break; + case NetworkDestroyContext::Asynchronous: + std::thread(std::move(network_reset)).detach(); + break; + } + }; + + auto outcome = network->Send(req, nullptr, callback, nullptr, nullptr); + ASSERT_TRUE(outcome.IsSuccessful()) << "Send failed before test could run"; + + // Wait for callback to be executed (timeout to avoid hangs in CI). + auto status = future.wait_for(std::chrono::seconds(1)); + ASSERT_EQ(status, std::future_status::ready) + << "Callback did not run in time"; +} + +TEST_P(NetworkLifetimeTest, NoCrashIfNetworkDestroyedDuringCallback) { + auto network = CreateDefaultNetwork({}); + + std::promise promise; + auto future = promise.get_future(); + + NetworkRequest req(CreateLoopbackUrl()); + + auto callback = [&network, &promise](NetworkResponse /*resp*/) { + auto network_reset = [&network, &promise]() { + network.reset(); + promise.set_value(); + }; + switch (GetParam()) { + case NetworkDestroyContext::Synchronous: + network_reset(); + break; + case NetworkDestroyContext::Asynchronous: + std::thread(std::move(network_reset)).detach(); + break; + } + }; + + auto outcome = network->Send(req, nullptr, callback, nullptr, nullptr); + ASSERT_TRUE(outcome.IsSuccessful()) << "Send failed before test could run"; + + // Wait for callback to be executed (timeout to avoid hangs in CI). + auto status = future.wait_for(std::chrono::seconds(2)); + ASSERT_EQ(status, std::future_status::ready) + << "Callback did not run in time"; +} + +TEST_P(NetworkLifetimeTest, NoCrashIfNetworkDestroyedDuringHeaderCallback) { + auto network = CreateDefaultNetwork({}); + + std::promise promise; + auto future = promise.get_future(); + auto fired = std::make_shared>(false); + + NetworkRequest req(CreateLoopbackUrl()); + + auto header_callback = [&network, &promise, fired](std::string /*key*/, + std::string /*value*/) { + if (fired->exchange(true)) { + return; + } + auto network_reset = [&network, &promise]() { + network.reset(); + promise.set_value(); + }; + switch (GetParam()) { + case NetworkDestroyContext::Synchronous: + network_reset(); + break; + case NetworkDestroyContext::Asynchronous: + std::thread(std::move(network_reset)).detach(); + break; + } + }; + + auto outcome = network->Send(req, nullptr, [](NetworkResponse /*resp*/) {}, + header_callback, nullptr); + ASSERT_TRUE(outcome.IsSuccessful()) << "Send failed before test could run"; + + // Wait for callback to be executed (timeout to avoid hangs in CI). + auto status = future.wait_for(std::chrono::seconds(2)); + ASSERT_EQ(status, std::future_status::ready) + << "Callback did not run in time"; +} + +TEST_P(NetworkLifetimeTest, NoCrashIfNetworkDestroyedDuringDataCallback) { + auto network = CreateDefaultNetwork({}); + + std::promise promise; + auto future = promise.get_future(); + auto fired = std::make_shared>(false); + + NetworkRequest req(CreateLoopbackUrl()); + + auto data_callback = [&network, &promise, fired](const std::uint8_t* /*data*/, + std::uint64_t /*offset*/, + std::size_t /*length*/) { + if (fired->exchange(true)) { + return; + } + auto network_reset = [&network, &promise]() { + network.reset(); + promise.set_value(); + }; + switch (GetParam()) { + case NetworkDestroyContext::Synchronous: + network_reset(); + break; + case NetworkDestroyContext::Asynchronous: + std::thread(std::move(network_reset)).detach(); + break; + } + }; + + auto outcome = network->Send(req, nullptr, [](NetworkResponse /*resp*/) {}, + nullptr, data_callback); + ASSERT_TRUE(outcome.IsSuccessful()) << "Send failed before test could run"; + + // Wait for callback to be executed (timeout to avoid hangs in CI). + auto status = future.wait_for(std::chrono::seconds(2)); + ASSERT_EQ(status, std::future_status::ready) + << "Callback did not run in time"; +} + +TEST_P(NetworkLifetimeTest, NoCrashIfNetworkDestroyedImmediatelyAfterSend) { + auto network = CreateDefaultNetwork({}); + + NetworkRequest req(CreateLoopbackUrl()); + + auto outcome = + network->Send(req, nullptr, [](NetworkResponse /*resp*/) {}, + [](std::string /*key*/, std::string /*value*/) {}, + [](const std::uint8_t* /*data*/, std::uint64_t /*offset*/, + std::size_t /*length*/) {}); + ASSERT_TRUE(outcome.IsSuccessful()) << "Send failed before test could run"; + + // Destroy network immediately after send. + switch (GetParam()) { + case NetworkDestroyContext::Synchronous: + network.reset(); + break; + case NetworkDestroyContext::Asynchronous: { + auto network_reset = [network]() mutable { network.reset(); }; + network.reset(); + std::thread(std::move(network_reset)).join(); + break; + } + } +} + +INSTANTIATE_TEST_SUITE_P( + , NetworkLifetimeTest, + ::testing::Values(NetworkDestroyContext::Synchronous, + NetworkDestroyContext::Asynchronous)); + +} // namespace + +} // namespace test +} // namespace http +} // namespace olp