From 2e8cd07df898afbc48c92c81244034c24516b65a Mon Sep 17 00:00:00 2001 From: J-P Nurmi Date: Wed, 22 Jul 2026 13:42:31 +0200 Subject: [PATCH 1/5] ref(transport): move crash-time disk dumping into transport Let crash backends suspend the transport's network submission and automatically route crash-time envelopes through its disk-dump path. This removes the need for allocating fallback disk transports in the backends, and prepares for the upcoming thread-pool telemetry module, which cannot safely drain the transport queue from a crash handler. --- src/backends/sentry_backend_breakpad.cpp | 23 ++---- src/backends/sentry_backend_crashpad.cpp | 22 ++---- src/backends/sentry_backend_inproc.c | 23 ++---- src/backends/sentry_backend_native.c | 41 +++++------ src/sentry_core.c | 11 ++- src/sentry_options.c | 2 +- src/sentry_transport.c | 36 ++++++++++ src/sentry_transport.h | 6 ++ tests/unit/test_concurrency.c | 91 ++++++++++++++++++++++++ tests/unit/test_failures.c | 2 +- tests/unit/test_uninit.c | 13 ++++ tests/unit/tests.inc | 1 + 12 files changed, 192 insertions(+), 79 deletions(-) diff --git a/src/backends/sentry_backend_breakpad.cpp b/src/backends/sentry_backend_breakpad.cpp index 9d397b0428..f180259e3e 100644 --- a/src/backends/sentry_backend_breakpad.cpp +++ b/src/backends/sentry_backend_breakpad.cpp @@ -23,7 +23,6 @@ extern "C" { #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_unix_pageallocator.h" -#include "transports/sentry_disk_transport.h" } #ifdef __GNUC__ @@ -167,6 +166,8 @@ breakpad_backend_callback(const google_breakpad::MinidumpDescriptor &descriptor, should_handle = !sentry_value_is_null(result); } + sentry__transport_suspend(options->transport); + // Flush logs and metrics in a crash-safe manner before crash handling if (options->enable_logs) { sentry__logs_flush_crash_safe(); @@ -246,33 +247,21 @@ breakpad_backend_callback(const google_breakpad::MinidumpDescriptor &descriptor, if (envelope && sentry__session_replay_has_pending(options)) { sentry_value_t crash_event = sentry_envelope_get_event(envelope); - sentry_transport_t *replay_transport - = sentry_new_disk_transport(options->run); - if (replay_transport) { - sentry__session_replay_flush_pending( - options, replay_transport, crash_event); - sentry__transport_dump_queue( - replay_transport, options->run); - sentry_transport_free(replay_transport); - } + sentry__session_replay_flush_pending( + options, options->transport, crash_event); } if (!sentry__launch_external_crash_reporter(options, envelope)) { - // capture the envelopes with the disk transport - sentry_transport_t *disk_transport - = sentry_new_disk_transport(options->run); if (!sentry_value_is_null(transaction)) { sentry_envelope_t *tx_envelope = sentry__prepare_transaction( options, transaction, nullptr); if (tx_envelope) { sentry__capture_envelope( - disk_transport, tx_envelope, options); + options->transport, tx_envelope, options); } } - sentry__capture_envelope(disk_transport, envelope, options); - sentry__transport_dump_queue(disk_transport, options->run); - sentry_transport_free(disk_transport); + sentry__capture_envelope(options->transport, envelope, options); } else { sentry_value_decref(transaction); } diff --git a/src/backends/sentry_backend_crashpad.cpp b/src/backends/sentry_backend_crashpad.cpp index 71fabc3d73..a4d6eca9cf 100644 --- a/src/backends/sentry_backend_crashpad.cpp +++ b/src/backends/sentry_backend_crashpad.cpp @@ -26,7 +26,6 @@ extern "C" { #endif #include "sentry_utils.h" #include "sentry_uuid.h" -#include "transports/sentry_disk_transport.h" } #include @@ -477,6 +476,8 @@ crashpad_handler(int signum, siginfo_t *info, ucontext_t *user_context) crash_event, nullptr, options->before_send_data); } + sentry__transport_suspend(options->transport); + // Flush logs and metrics in a crash-safe manner before crash handling if (options->enable_logs) { sentry__logs_flush_crash_safe(); @@ -498,13 +499,7 @@ crashpad_handler(int signum, siginfo_t *info, ucontext_t *user_context) if (session) { sentry_envelope_t *envelope = sentry__envelope_new(); sentry__envelope_add_session(envelope, session); - - // capture the envelope with the disk transport - sentry_transport_t *disk_transport - = sentry_new_disk_transport(options->run); - sentry__capture_envelope(disk_transport, envelope, options); - sentry__transport_dump_queue(disk_transport, options->run); - sentry_transport_free(disk_transport); + sentry__capture_envelope(options->transport, envelope, options); } if (sentry__session_replay_has_pending(options)) { @@ -517,15 +512,8 @@ crashpad_handler(int signum, siginfo_t *info, ucontext_t *user_context) sentry__ringbuffer_to_list(scope->breadcrumbs)); } - sentry_transport_t *replay_transport - = sentry_new_disk_transport(options->run); - if (replay_transport) { - sentry__session_replay_flush_pending( - options, replay_transport, crash_event); - sentry__transport_dump_queue( - replay_transport, options->run); - sentry_transport_free(replay_transport); - } + sentry__session_replay_flush_pending( + options, options->transport, crash_event); } sentry_value_decref(crash_event); } else { diff --git a/src/backends/sentry_backend_inproc.c b/src/backends/sentry_backend_inproc.c index 05cf02e1ba..2b1db89967 100644 --- a/src/backends/sentry_backend_inproc.c +++ b/src/backends/sentry_backend_inproc.c @@ -19,7 +19,6 @@ #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_unix_pageallocator.h" -#include "transports/sentry_disk_transport.h" #include #include #ifdef SENTRY_PLATFORM_UNIX @@ -1086,6 +1085,8 @@ process_ucontext_deferred(const sentry_ucontext_t *uctx, SENTRY_DEBUG("skipping `on_crash` hook due to recursive crash"); } + sentry__transport_suspend(options->transport); + // Flush logs in a crash-safe manner before crash handling if (options->enable_logs) { sentry__logs_flush_crash_safe(); @@ -1135,33 +1136,21 @@ process_ucontext_deferred(const sentry_ucontext_t *uctx, if (envelope && sentry__session_replay_has_pending(options)) { sentry_value_t crash_event = sentry_envelope_get_event(envelope); - sentry_transport_t *replay_transport - = sentry_new_disk_transport(options->run); - if (replay_transport) { - sentry__session_replay_flush_pending( - options, replay_transport, crash_event); - sentry__transport_dump_queue( - replay_transport, options->run); - sentry_transport_free(replay_transport); - } + sentry__session_replay_flush_pending( + options, options->transport, crash_event); } if (!sentry__launch_external_crash_reporter(options, envelope)) { - // capture the envelopes with the disk transport - sentry_transport_t *disk_transport - = sentry_new_disk_transport(options->run); if (!sentry_value_is_null(transaction)) { sentry_envelope_t *tx_envelope = sentry__prepare_transaction( options, transaction, NULL); if (tx_envelope) { sentry__capture_envelope( - disk_transport, tx_envelope, options); + options->transport, tx_envelope, options); } } - sentry__capture_envelope(disk_transport, envelope, options); - sentry__transport_dump_queue(disk_transport, options->run); - sentry_transport_free(disk_transport); + sentry__capture_envelope(options->transport, envelope, options); } else { sentry_value_decref(transaction); } diff --git a/src/backends/sentry_backend_native.c b/src/backends/sentry_backend_native.c index 3a5627975b..bf8c898fe0 100644 --- a/src/backends/sentry_backend_native.c +++ b/src/backends/sentry_backend_native.c @@ -40,7 +40,6 @@ #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_value.h" -#include "transports/sentry_disk_transport.h" // Global process-wide synchronization for IPC and shared memory access // This lives for the entire backend lifetime and is shared across all threads @@ -989,6 +988,8 @@ native_backend_except(sentry_backend_t *backend, const sentry_ucontext_t *uctx) SENTRY_DEBUG("handling native backend exception"); + sentry__transport_suspend(options->transport); + // Flush logs and metrics in a crash-safe manner before crash handling if (options->enable_logs) { sentry__logs_flush_crash_safe(); @@ -1087,32 +1088,22 @@ native_backend_except(sentry_backend_t *backend, const sentry_ucontext_t *uctx) SENTRY_SESSION_STATUS_CRASHED); if (session || !sentry_value_is_null(transaction)) { - sentry_transport_t *disk_transport - = sentry_new_disk_transport(options->run); - if (disk_transport) { - if (!sentry_value_is_null(transaction)) { - sentry_envelope_t *tx_envelope - = sentry__prepare_transaction( - options, transaction, NULL); - if (tx_envelope) { - sentry__capture_envelope( - disk_transport, tx_envelope, options); - } + if (!sentry_value_is_null(transaction)) { + sentry_envelope_t *tx_envelope + = sentry__prepare_transaction( + options, transaction, NULL); + if (tx_envelope) { + sentry__capture_envelope( + options->transport, tx_envelope, options); } - if (session) { - sentry_envelope_t *envelope - = sentry__envelope_new(); - if (envelope) { - sentry__envelope_add_session(envelope, session); - sentry__capture_envelope( - disk_transport, envelope, options); - } + } + if (session) { + sentry_envelope_t *envelope = sentry__envelope_new(); + if (envelope) { + sentry__envelope_add_session(envelope, session); + sentry__capture_envelope( + options->transport, envelope, options); } - sentry__transport_dump_queue( - disk_transport, options->run); - sentry_transport_free(disk_transport); - } else { - sentry_value_decref(transaction); } } diff --git a/src/sentry_core.c b/src/sentry_core.c index 7aaa65ad4f..ec17f915fa 100644 --- a/src/sentry_core.c +++ b/src/sentry_core.c @@ -161,6 +161,10 @@ sentry_init(sentry_options_t *options) } transport = options->transport; + if (!transport) { + transport = sentry__transport_new_null(); + options->transport = transport; + } sentry_path_t *database_path = options->database_path; options->database_path = sentry__path_absolute(database_path); if (options->database_path) { @@ -199,7 +203,12 @@ sentry_init(sentry_options_t *options) // Also, we want to continue - crash capture doesn't need transport. sentry__transport_shutdown(transport, 0); sentry_options_set_transport(options, NULL); - transport = NULL; + transport = options->transport; + if (!transport + || sentry__transport_startup(transport, options) != 0) { + SENTRY_WARN("failed to initialize fallback transport"); + goto fail; + } #else SENTRY_WARN("failed to initialize transport"); goto fail; diff --git a/src/sentry_options.c b/src/sentry_options.c index ee18a5e6c5..2408f77a91 100644 --- a/src/sentry_options.c +++ b/src/sentry_options.c @@ -197,7 +197,7 @@ sentry_options_set_transport( sentry_options_t *opts, sentry_transport_t *transport) { sentry_transport_free(opts->transport); - opts->transport = transport; + opts->transport = transport ? transport : sentry__transport_new_null(); } #ifdef SENTRY_PLATFORM_NX diff --git a/src/sentry_transport.c b/src/sentry_transport.c index 03c6a5fbf6..09d2927552 100644 --- a/src/sentry_transport.c +++ b/src/sentry_transport.c @@ -2,6 +2,7 @@ #include "sentry_alloc.h" #include "sentry_envelope.h" #include "sentry_options.h" +#include "sentry_sync.h" struct sentry_transport_s { void (*send_envelope_func)(sentry_envelope_t *envelope, void *state); @@ -13,9 +14,18 @@ struct sentry_transport_s { void (*retry_func)(void *state); void (*cleanup_func)(const sentry_options_t *options, void *state); void *state; + sentry_run_t *run; + long suspended; bool running; }; +static void +send_null_envelope(sentry_envelope_t *envelope, void *state) +{ + (void)state; + sentry_envelope_free(envelope); +} + sentry_transport_t * sentry_transport_new( void (*send_func)(sentry_envelope_t *envelope, void *state)) @@ -28,6 +38,12 @@ sentry_transport_new( return transport; } + +sentry_transport_t * +sentry__transport_new_null(void) +{ + return sentry_transport_new(send_null_envelope); +} void sentry_transport_set_state(sentry_transport_t *transport, void *state) { @@ -73,14 +89,24 @@ sentry__transport_send_envelope( sentry_envelope_free(envelope); return; } + if (sentry__atomic_fetch(&transport->suspended)) { + sentry__run_write_envelope(transport->run, envelope); + sentry_envelope_free(envelope); + return; + } SENTRY_DEBUG("sending envelope"); transport->send_envelope_func(envelope, transport->state); + if (sentry__atomic_fetch(&transport->suspended)) { + sentry__transport_dump_queue(transport, transport->run); + } } int sentry__transport_startup( sentry_transport_t *transport, const sentry_options_t *options) { + sentry__run_free(transport->run); + transport->run = sentry__run_incref(options->run); if (transport->startup_func) { SENTRY_DEBUG("starting transport"); int rv = transport->startup_func(options, transport->state); @@ -131,6 +157,15 @@ sentry__transport_dump_queue(sentry_transport_t *transport, sentry_run_t *run) return dumped; } +void +sentry__transport_suspend(sentry_transport_t *transport) +{ + if (!transport) { + return; + } + sentry__atomic_store(&transport->suspended, 1); +} + void sentry_transport_free(sentry_transport_t *transport) { @@ -140,6 +175,7 @@ sentry_transport_free(sentry_transport_t *transport) if (transport->free_func) { transport->free_func(transport->state); } + sentry__run_free(transport->run); sentry_free(transport); } diff --git a/src/sentry_transport.h b/src/sentry_transport.h index a4ce8197ed..4c4dd0ef01 100644 --- a/src/sentry_transport.h +++ b/src/sentry_transport.h @@ -55,6 +55,12 @@ sentry_transport_t *sentry__transport_new_default(void); size_t sentry__transport_dump_queue( sentry_transport_t *transport, sentry_run_t *run); +/** Creates an internal transport that discards envelopes. */ +sentry_transport_t *sentry__transport_new_null(void); + +/** Sets the run used for automatic queue dumps. */ +void sentry__transport_suspend(sentry_transport_t *transport); + void *sentry__transport_get_state(sentry_transport_t *transport); void sentry__transport_set_retry_func( diff --git a/tests/unit/test_concurrency.c b/tests/unit/test_concurrency.c index 1129ff7f53..cdc3c739aa 100644 --- a/tests/unit/test_concurrency.c +++ b/tests/unit/test_concurrency.c @@ -1,5 +1,10 @@ #include "sentry_core.h" +#include "sentry_database.h" +#include "sentry_envelope.h" +#include "sentry_options.h" +#include "sentry_path.h" #include "sentry_testsupport.h" +#include "sentry_transport.h" #include @@ -76,6 +81,92 @@ SENTRY_TEST(multiple_inits) TEST_CHECK_INT_EQUAL(called, 4); } +typedef struct { + sentry_envelope_t *queued; + long send_started; + long release_send; + long send_count; + long dump_count; +} suspend_transport_state_t; + +static void +suspend_send(sentry_envelope_t *envelope, void *data) +{ + suspend_transport_state_t *state = data; + sentry__atomic_store(&state->send_started, 1); + while (!sentry__atomic_fetch(&state->release_send)) { + sentry__thread_yield(); + } + state->queued = envelope; + sentry__atomic_fetch_and_add(&state->send_count, 1); +} + +static size_t +suspend_dump(sentry_run_t *run, void *data) +{ + (void)run; + suspend_transport_state_t *state = data; + sentry__atomic_fetch_and_add(&state->dump_count, 1); + if (!state->queued) { + return 0; + } + sentry_envelope_free(state->queued); + state->queued = NULL; + return 1; +} + +SENTRY_THREAD_FN +suspend_send_thread(void *data) +{ + sentry__transport_send_envelope(data, sentry__envelope_new()); + return 0; +} + +SENTRY_TEST(transport_suspend) +{ + sentry_path_t *database_path = sentry__path_from_str( + SENTRY_TEST_PATH_PREFIX ".transport-crash-mode"); + TEST_ASSERT(!!database_path); + sentry__path_remove_all(database_path); + TEST_ASSERT(!sentry__path_create_dir_all(database_path)); + sentry_run_t *run = sentry__run_new(database_path); + TEST_ASSERT(!!run); + + suspend_transport_state_t state = { 0 }; + sentry_transport_t *transport = sentry_transport_new(suspend_send); + TEST_ASSERT(!!transport); + sentry_transport_set_state(transport, &state); + sentry__transport_set_dump_func(transport, suspend_dump); + sentry_options_t startup_options = { 0 }; + startup_options.run = run; + TEST_CHECK(!sentry__transport_startup(transport, &startup_options)); + + sentry_threadid_t thread; + sentry__thread_init(&thread); + TEST_ASSERT(!sentry__thread_spawn(&thread, suspend_send_thread, transport)); + while (!sentry__atomic_fetch(&state.send_started)) { + sentry__thread_yield(); + } + + sentry__transport_suspend(transport); + sentry__atomic_store(&state.release_send, 1); + sentry__thread_join(thread); + + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.send_count), 1); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.dump_count), 1); + TEST_CHECK(!state.queued); + + sentry__transport_send_envelope(transport, sentry__envelope_new()); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&state.send_count), 1); + TEST_CHECK(sentry__atomic_fetch(&run->retain)); + + sentry_transport_free(transport); + sentry__run_clean(run, true); + sentry__run_free(run); + sentry__path_remove_all(database_path); + sentry__path_free(database_path); +} + SENTRY_THREAD_FN thread_worker(void *called) { diff --git a/tests/unit/test_failures.c b/tests/unit/test_failures.c index 6e8cb1081f..169d0fb247 100644 --- a/tests/unit/test_failures.c +++ b/tests/unit/test_failures.c @@ -30,7 +30,7 @@ SENTRY_TEST(init_failure) // On NX a failing transport must not fail initialization. TEST_CHECK(rv == 0); SENTRY_WITH_OPTIONS (runtime_options) { - TEST_CHECK(runtime_options->transport == NULL); + TEST_CHECK(runtime_options->transport != NULL); } #else TEST_CHECK(rv != 0); diff --git a/tests/unit/test_uninit.c b/tests/unit/test_uninit.c index 03ff3209df..478096f2c4 100644 --- a/tests/unit/test_uninit.c +++ b/tests/unit/test_uninit.c @@ -1,4 +1,9 @@ +#include "sentry_core.h" +#include "sentry_envelope.h" +#include "sentry_options.h" +#include "sentry_sync.h" #include "sentry_testsupport.h" +#include "sentry_transport.h" SENTRY_TEST(uninitialized) { @@ -45,6 +50,14 @@ SENTRY_TEST(empty_transport) sentry_uuid_t id = sentry_capture_event(event); TEST_CHECK(!sentry_uuid_is_nil(&id)); + SENTRY_WITH_OPTIONS (runtime_options) { + TEST_ASSERT(!!runtime_options->transport); + sentry__transport_suspend(runtime_options->transport); + sentry__transport_send_envelope( + runtime_options->transport, sentry__envelope_new()); + TEST_CHECK(sentry__atomic_fetch(&runtime_options->run->retain)); + } + sentry_close(); } diff --git a/tests/unit/tests.inc b/tests/unit/tests.inc index 5c0fbc7d0c..eb76115057 100644 --- a/tests/unit/tests.inc +++ b/tests/unit/tests.inc @@ -382,6 +382,7 @@ XX(transactions_skip_before_send) XX(transport_retry) XX(transport_sampling_transactions) XX(transport_sampling_transactions_set_trace) +XX(transport_suspend) XX(tus_file_attachment_preserves_original) XX(tus_placeholder_uses_raw_location) XX(tus_request_preparation) From db9964f1815d1e691e44faa841d96c77bcfdd9d8 Mon Sep 17 00:00:00 2001 From: J-P Nurmi Date: Fri, 24 Jul 2026 15:51:30 +0200 Subject: [PATCH 2/5] ref(batcher): name dedicated batcher threads Give the logs and metrics batchers distinct OS-level thread names while keeping the batcher implementation generic. Named dedicated threads make profiling and tracing easier because background work can be identified directly in debuggers, crash dumps, and trace timelines instead of showing up as anonymous worker activity. --- src/sentry_batcher.c | 18 +++++++++++++++++- src/sentry_batcher.h | 3 +++ src/sentry_logs.c | 1 + src/sentry_metrics.c | 1 + src/sentry_sync.c | 6 ++++++ src/sentry_sync.h | 11 +++++++++++ tests/test_integration_logger.py | 5 +++-- 7 files changed, 42 insertions(+), 3 deletions(-) diff --git a/src/sentry_batcher.c b/src/sentry_batcher.c index 4572ab6660..d0b797c901 100644 --- a/src/sentry_batcher.c +++ b/src/sentry_batcher.c @@ -2,12 +2,14 @@ #include "sentry_alloc.h" #include "sentry_cpu_relax.h" #include "sentry_options.h" +#include "sentry_string.h" #include "sentry_utils.h" // The batcher thread sleeps for this interval between flush cycles. // When the timer fires and there are items in the buffer, they are flushed // regardless of how recently they were enqueued. #define SENTRY_BATCHER_FLUSH_INTERVAL_MS 5000 +#define SENTRY_BATCHER_THREAD_NAME "sentry-batcher" #ifdef SENTRY_UNITTEST # ifdef SENTRY_PLATFORM_WINDOWS @@ -59,11 +61,22 @@ sentry__batcher_release(sentry_batcher_t *batcher) for (long i = 0; i < SENTRY_BATCHER_BUFFER_COUNT; i++) { buffer_drain(&batcher->buffers[i]); } + sentry_free(batcher->thread_name); sentry__dsn_decref(batcher->dsn); sentry__thread_free(&batcher->batching_thread); sentry_free(batcher); } +void +sentry__batcher_setname(sentry_batcher_t *batcher, const char *thread_name) +{ + if (!batcher) { + return; + } + sentry_free(batcher->thread_name); + batcher->thread_name = sentry__string_clone(thread_name); +} + static inline void lock_ref(sentry_batcher_ref_t *ref) { @@ -321,7 +334,10 @@ SENTRY_THREAD_FN batcher_thread_func(void *data) { sentry_batcher_t *batcher = data; - SENTRY_DEBUG("Starting batching thread"); + const char *thread_name = batcher->thread_name ? batcher->thread_name + : SENTRY_BATCHER_THREAD_NAME; + sentry__thread_setname(sentry__current_thread(), thread_name); + SENTRY_DEBUGF("Starting %s thread", thread_name); // Transition from STARTING to RUNNING using compare-and-swap // CAS ensures atomic state verification: only succeeds if state is STARTING diff --git a/src/sentry_batcher.h b/src/sentry_batcher.h index 50e1bc6e71..af04b9fa9c 100644 --- a/src/sentry_batcher.h +++ b/src/sentry_batcher.h @@ -50,6 +50,7 @@ typedef struct { sentry_threadid_t batching_thread; // the batching thread sentry_batch_func_t batch_func; // function to add items to envelope sentry_data_category_t data_category; // for client report discard tracking + char *thread_name; sentry_dsn_t *dsn; sentry_transport_t *transport; sentry_run_t *run; @@ -64,6 +65,8 @@ typedef struct { sentry_batcher_t *sentry__batcher_new( sentry_batch_func_t batch_func, sentry_data_category_t data_category); +void sentry__batcher_setname( + sentry_batcher_t *batcher, const char *thread_name); /** * Acquires a reference to the batcher behind `ref`, atomically incrementing diff --git a/src/sentry_logs.c b/src/sentry_logs.c index 7c59916829..9edcc85594 100644 --- a/src/sentry_logs.c +++ b/src/sentry_logs.c @@ -636,6 +636,7 @@ sentry__logs_startup(const sentry_options_t *options) return; } + sentry__batcher_setname(batcher, "sentry-logs"); sentry__batcher_startup(batcher, options); sentry_batcher_t *old = sentry__batcher_swap(&g_batcher, batcher); diff --git a/src/sentry_metrics.c b/src/sentry_metrics.c index 033af6b467..99685a1f7a 100644 --- a/src/sentry_metrics.c +++ b/src/sentry_metrics.c @@ -140,6 +140,7 @@ sentry__metrics_startup(const sentry_options_t *options) return; } + sentry__batcher_setname(batcher, "sentry-metrics"); sentry__batcher_startup(batcher, options); sentry_batcher_t *old = sentry__batcher_swap(&g_batcher, batcher); diff --git a/src/sentry_sync.c b/src/sentry_sync.c index de51acc4a8..b63698396a 100644 --- a/src/sentry_sync.c +++ b/src/sentry_sync.c @@ -101,6 +101,12 @@ thread_setname(sentry_threadid_t thread_id, const char *thread_name) } #endif +int +sentry__thread_setname(sentry_threadid_t thread_id, const char *thread_name) +{ + return thread_setname(thread_id, thread_name); +} + /** * Queue operations, locking and Reference counting: * diff --git a/src/sentry_sync.h b/src/sentry_sync.h index adc904fb28..62028777d4 100644 --- a/src/sentry_sync.h +++ b/src/sentry_sync.h @@ -457,6 +457,17 @@ sentry__atomic_fetch_u64(uint64_t *val) #endif } +/** + * Sets the OS-level name for a thread. + * + * On macOS, only the current thread can be named. + * + * Returns 0 on success, or a non-zero value if the platform reports that + * setting the thread name failed. + */ +int sentry__thread_setname( + sentry_threadid_t thread_id, const char *thread_name); + struct sentry_bgworker_s; typedef struct sentry_bgworker_s sentry_bgworker_t; diff --git a/tests/test_integration_logger.py b/tests/test_integration_logger.py index 43fe801493..f504ea8093 100644 --- a/tests/test_integration_logger.py +++ b/tests/test_integration_logger.py @@ -1,6 +1,7 @@ import shutil import subprocess import sys +import re import pytest @@ -67,7 +68,7 @@ def parse_logger_output(output): # Background thread startup messages (e.g. from the metrics/logs batcher) can # race with the pre-crash marker and are not crash-time logs. # (from batcher_thread_func in sentry_batcher.c) - background_thread_messages = {"Starting batching thread"} + background_thread_message = re.compile(r"^Starting sentry-(logs|metrics) thread$") lines = output.split("\n") @@ -89,7 +90,7 @@ def parse_logger_output(output): # Track logs that occur after the pre-crash marker, # excluding background thread noise - if pre_crash_completed and log_message not in background_thread_messages: + if pre_crash_completed and not background_thread_message.match(log_message): parsed_data["logs_after_pre_crash"].append(log_message) return parsed_data From 3dc8feab40aaefd4de2b2f67c458d060ed961ace Mon Sep 17 00:00:00 2001 From: J-P Nurmi Date: Fri, 17 Jul 2026 16:51:35 +0200 Subject: [PATCH 3/5] feat: add thread pool Add a bounded thread pool that runs tasks in parallel and invokes completion callbacks in submission order. Add unit coverage for ordered parallel execution. --- src/sentry_sync.c | 269 +++++++++++++++++++++++++++++++++++++++++ src/sentry_sync.h | 48 ++++++++ tests/unit/test_sync.c | 111 +++++++++++++++++ tests/unit/tests.inc | 2 + 4 files changed, 430 insertions(+) diff --git a/src/sentry_sync.c b/src/sentry_sync.c index b63698396a..acb33b1697 100644 --- a/src/sentry_sync.c +++ b/src/sentry_sync.c @@ -1,6 +1,7 @@ #include "sentry_sync.h" #include "sentry_alloc.h" #include "sentry_core.h" +#include "sentry_cpu_relax.h" #include "sentry_string.h" #include "sentry_utils.h" #include @@ -107,6 +108,274 @@ sentry__thread_setname(sentry_threadid_t thread_id, const char *thread_name) return thread_setname(thread_id, thread_name); } +typedef struct sentry_threadpool_task_s { + struct sentry_threadpool_task_s *next; + void (*exec_func)(void *task_data); + void (*complete_func)(void *task_data); + void (*cleanup_func)(void *task_data); + void *task_data; + bool done; +} sentry_threadpool_task_t; + +struct sentry_threadpool_s { + sentry_threadid_t *threads; + char *thread_name; + size_t thread_count; + size_t started_threads; + sentry_mutex_t lock; + sentry_cond_t work_signal; + sentry_cond_t state_signal; + sentry_threadpool_task_t *first_task; + sentry_threadpool_task_t *last_task; + sentry_threadpool_task_t *next_task; + long pending; + long index; + bool running; + bool stopping; + bool committing; +}; + +static void +threadpool_task_free(sentry_threadpool_task_t *task) +{ + if (task->cleanup_func) { + task->cleanup_func(task->task_data); + } + sentry_free(task); +} + +static void +threadpool_wake_all(sentry_threadpool_t *pool) +{ + for (size_t i = 0; i < pool->started_threads; i++) { + sentry__cond_wake(&pool->work_signal); + } +} + +static void +threadpool_commit_ready(sentry_threadpool_t *pool) +{ + if (pool->committing) { + return; + } + pool->committing = true; + + while (pool->first_task && pool->first_task->done) { + sentry_threadpool_task_t *task = pool->first_task; + pool->first_task = task->next; + if (!pool->first_task) { + pool->last_task = NULL; + } + + sentry__mutex_unlock(&pool->lock); + if (task->complete_func) { + task->complete_func(task->task_data); + } + threadpool_task_free(task); + sentry__mutex_lock(&pool->lock); + + sentry__atomic_fetch_and_add(&pool->pending, -1); + sentry__cond_wake(&pool->state_signal); + } + + pool->committing = false; + if (sentry__atomic_fetch(&pool->pending) == 0) { + threadpool_wake_all(pool); + } +} + +SENTRY_THREAD_FN +threadpool_worker(void *data) +{ + sentry_threadpool_t *pool = data; + if (pool->thread_name) { + const long index = sentry__atomic_fetch_and_add(&pool->index, 1); + char thread_name[16]; + snprintf(thread_name, sizeof(thread_name), "%s-%ld", pool->thread_name, + index); + sentry__thread_setname(sentry__current_thread(), thread_name); + } + + while (true) { + sentry__mutex_lock(&pool->lock); + sentry_threadpool_task_t *task = pool->next_task; + while (!task) { + if (pool->stopping && sentry__atomic_fetch(&pool->pending) == 0) { + sentry__mutex_unlock(&pool->lock); + return 0; + } + sentry__cond_wait(&pool->work_signal, &pool->lock); + task = pool->next_task; + } + pool->next_task = task->next; + sentry__mutex_unlock(&pool->lock); + + task->exec_func(task->task_data); + + sentry__mutex_lock(&pool->lock); + task->done = true; + threadpool_commit_ready(pool); + sentry__cond_wake(&pool->work_signal); + sentry__mutex_unlock(&pool->lock); + } +} + +sentry_threadpool_t * +sentry__threadpool_new(size_t thread_count) +{ + if (thread_count == 0) { + return NULL; + } + sentry_threadpool_t *pool = SENTRY_MAKE(sentry_threadpool_t); + if (!pool) { + return NULL; + } + pool->threads = sentry__calloc(thread_count, sizeof(sentry_threadid_t)); + if (!pool->threads) { + sentry_free(pool); + return NULL; + } + pool->thread_count = thread_count; + sentry__mutex_init(&pool->lock); + sentry__cond_init(&pool->work_signal); + sentry__cond_init(&pool->state_signal); + for (size_t i = 0; i < thread_count; i++) { + sentry__thread_init(&pool->threads[i]); + } + return pool; +} + +void +sentry__threadpool_setname(sentry_threadpool_t *pool, const char *thread_name) +{ + if (!pool) { + return; + } + sentry_free(pool->thread_name); + pool->thread_name = sentry__string_clone(thread_name); +} + +int +sentry__threadpool_start(sentry_threadpool_t *pool) +{ + if (!pool || pool->running) { + return pool ? 0 : 1; + } + pool->running = true; + pool->stopping = false; + for (size_t i = 0; i < pool->thread_count; i++) { + if (sentry__thread_spawn(&pool->threads[i], threadpool_worker, pool) + != 0) { + sentry__mutex_lock(&pool->lock); + pool->stopping = true; + threadpool_wake_all(pool); + sentry__mutex_unlock(&pool->lock); + for (size_t j = 0; j < pool->started_threads; j++) { + sentry__thread_join(pool->threads[j]); + } + pool->started_threads = 0; + pool->running = false; + return 1; + } + pool->started_threads++; + } + return 0; +} + +int +sentry__threadpool_submit(sentry_threadpool_t *pool, + void (*exec_func)(void *task_data), void (*complete_func)(void *task_data), + void (*cleanup_func)(void *task_data), void *task_data) +{ + if (!pool || !exec_func) { + return 1; + } + sentry_threadpool_task_t *task = SENTRY_MAKE(sentry_threadpool_task_t); + if (!task) { + return 1; + } + task->exec_func = exec_func; + task->complete_func = complete_func; + task->cleanup_func = cleanup_func; + task->task_data = task_data; + + sentry__mutex_lock(&pool->lock); + if (!pool->running || pool->stopping) { + sentry__mutex_unlock(&pool->lock); + sentry_free(task); + return 1; + } + + if (pool->last_task) { + pool->last_task->next = task; + } else { + pool->first_task = task; + } + pool->last_task = task; + if (!pool->next_task) { + pool->next_task = task; + } + sentry__atomic_fetch_and_add(&pool->pending, 1); + sentry__cond_wake(&pool->work_signal); + sentry__mutex_unlock(&pool->lock); + return 0; +} + +void +sentry__threadpool_flush(sentry_threadpool_t *pool) +{ + if (!pool || !pool->running) { + return; + } + sentry__mutex_lock(&pool->lock); + while (sentry__atomic_fetch(&pool->pending) > 0) { + sentry__cond_wait(&pool->state_signal, &pool->lock); + } + sentry__mutex_unlock(&pool->lock); +} + +void +sentry__threadpool_shutdown(sentry_threadpool_t *pool) +{ + if (!pool || !pool->running) { + return; + } + sentry__mutex_lock(&pool->lock); + pool->stopping = true; + threadpool_wake_all(pool); + sentry__cond_wake(&pool->state_signal); + sentry__mutex_unlock(&pool->lock); + + for (size_t i = 0; i < pool->started_threads; i++) { + sentry__thread_join(pool->threads[i]); + } + pool->started_threads = 0; + pool->running = false; + pool->index = 0; +} + +void +sentry__threadpool_free(sentry_threadpool_t *pool) +{ + if (!pool) { + return; + } + sentry__threadpool_shutdown(pool); + sentry_threadpool_task_t *task = pool->first_task; + while (task) { + sentry_threadpool_task_t *next = task->next; + threadpool_task_free(task); + task = next; + } + for (size_t i = 0; i < pool->thread_count; i++) { + sentry__thread_free(&pool->threads[i]); + } + sentry_free(pool->thread_name); + sentry__mutex_free(&pool->lock); + sentry_free(pool->threads); + sentry_free(pool); +} + /** * Queue operations, locking and Reference counting: * diff --git a/src/sentry_sync.h b/src/sentry_sync.h index 62028777d4..a2d4cb0f30 100644 --- a/src/sentry_sync.h +++ b/src/sentry_sync.h @@ -3,6 +3,7 @@ #include "sentry_boot.h" #include "sentry_core.h" +#include "sentry_cpu_relax.h" #include #include @@ -431,6 +432,35 @@ sentry__atomic_compare_swap(volatile long *val, long expected, long desired) #endif } +typedef bool (*sentry_spin_wait_func_t)(int attempt, void *data); + +static inline void +sentry__spin_lock(volatile long *lock) +{ + while (!sentry__atomic_compare_swap(lock, 0, 1)) { + sentry__cpu_relax(); + } +} + +static inline bool +sentry__spin_lock_wait( + volatile long *lock, sentry_spin_wait_func_t wait_func, void *data) +{ + int attempts = 0; + while (!sentry__atomic_compare_swap(lock, 0, 1)) { + if (!wait_func || !wait_func(++attempts, data)) { + return false; + } + } + return true; +} + +static inline void +sentry__spin_unlock(volatile long *lock) +{ + sentry__atomic_store(lock, 0); +} + /** * 64-bit variants of the atomic helpers above. The `long`-based helpers are * only 32 bits wide on Windows and 32-bit POSIX targets, so callers that need @@ -471,8 +501,26 @@ int sentry__thread_setname( struct sentry_bgworker_s; typedef struct sentry_bgworker_s sentry_bgworker_t; +struct sentry_threadpool_s; +typedef struct sentry_threadpool_s sentry_threadpool_t; + typedef void (*sentry_task_exec_func_t)(void *task_data, void *state); +/** + * Creates a thread pool. Tasks execute in parallel, while completion callbacks + * run in submission order. + */ +sentry_threadpool_t *sentry__threadpool_new(size_t thread_count); +void sentry__threadpool_setname( + sentry_threadpool_t *pool, const char *thread_name); +int sentry__threadpool_start(sentry_threadpool_t *pool); +int sentry__threadpool_submit(sentry_threadpool_t *pool, + void (*exec_func)(void *task_data), void (*complete_func)(void *task_data), + void (*cleanup_func)(void *task_data), void *task_data); +void sentry__threadpool_flush(sentry_threadpool_t *pool); +void sentry__threadpool_shutdown(sentry_threadpool_t *pool); +void sentry__threadpool_free(sentry_threadpool_t *pool); + /** * Creates a new background worker thread. * diff --git a/tests/unit/test_sync.c b/tests/unit/test_sync.c index 98c5eeef91..3bc95c0964 100644 --- a/tests/unit/test_sync.c +++ b/tests/unit/test_sync.c @@ -581,6 +581,117 @@ SENTRY_TEST(bgworker_delayed_shutdown) sentry__bgworker_decref(bgw); } +struct threadpool_test_state { + volatile long first_started; + volatile long release_first; + volatile long second_ran; + bool first_timed_out; + int completion_order[2]; + int completion_count; + int cleanup_count; +}; + +struct threadpool_test_task { + struct threadpool_test_state *state; + int id; +}; + +static void +threadpool_test_exec(void *data) +{ + struct threadpool_test_task *task = data; + struct threadpool_test_state *state = task->state; + const uint64_t deadline = sentry__monotonic_time() + 1000; + + if (task->id == 0) { + sentry__atomic_store(&state->first_started, 1); + while (!sentry__atomic_fetch(&state->release_first)) { + if (sentry__monotonic_time() >= deadline) { + state->first_timed_out = true; + break; + } + sleep_ms(1); + } + } else { + while (!sentry__atomic_fetch(&state->first_started) + && sentry__monotonic_time() < deadline) { + sleep_ms(1); + } + sentry__atomic_store(&state->second_ran, 1); + sentry__atomic_store(&state->release_first, 1); + } +} + +static void +threadpool_test_complete(void *data) +{ + struct threadpool_test_task *task = data; + struct threadpool_test_state *state = task->state; + state->completion_order[state->completion_count++] = task->id; +} + +static void +threadpool_test_cleanup(void *data) +{ + struct threadpool_test_task *task = data; + task->state->cleanup_count++; +} + +SENTRY_TEST(threadpool_ordered_parallel) +{ + struct threadpool_test_state state = { 0 }; + struct threadpool_test_task tasks[] = { + { &state, 0 }, + { &state, 1 }, + }; + sentry_threadpool_t *pool = sentry__threadpool_new(2); + TEST_ASSERT(!!pool); + TEST_ASSERT(sentry__threadpool_start(pool) == 0); + + for (size_t i = 0; i < 2; i++) { + TEST_ASSERT( + sentry__threadpool_submit(pool, threadpool_test_exec, + threadpool_test_complete, threadpool_test_cleanup, &tasks[i]) + == 0); + } + sentry__threadpool_flush(pool); + + TEST_CHECK(sentry__atomic_fetch(&state.second_ran)); + TEST_CHECK(!state.first_timed_out); + TEST_CHECK_INT_EQUAL(state.completion_count, 2); + TEST_CHECK_INT_EQUAL(state.completion_order[0], 0); + TEST_CHECK_INT_EQUAL(state.completion_order[1], 1); + TEST_CHECK_INT_EQUAL(state.cleanup_count, 2); + + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); +} + +static long g_spin_waits = 0; + +static bool +spin_wait(int UNUSED(attempt), void *UNUSED(data)) +{ + sentry__atomic_fetch_and_add(&g_spin_waits, 1); + return sentry__atomic_fetch(&g_spin_waits) < 2; +} + +SENTRY_TEST(spin_lock) +{ + long lock = 0; + sentry__atomic_store(&g_spin_waits, 0); + + sentry__spin_lock(&lock); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&lock), 1); + TEST_CHECK(!sentry__spin_lock_wait(&lock, spin_wait, NULL)); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&g_spin_waits), 2); + sentry__spin_unlock(&lock); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&lock), 0); + TEST_CHECK(sentry__spin_lock_wait(&lock, spin_wait, NULL)); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&g_spin_waits), 2); + sentry__spin_unlock(&lock); +} + SENTRY_TEST(cond_wait_timeout_overflow) { #if !(defined(SENTRY_PLATFORM_MACOS) \ diff --git a/tests/unit/tests.inc b/tests/unit/tests.inc index eb76115057..3d6d4c60e1 100644 --- a/tests/unit/tests.inc +++ b/tests/unit/tests.inc @@ -359,6 +359,7 @@ XX(span_data_n) XX(span_tagging) XX(span_tagging_n) XX(spans_on_scope) +XX(spin_lock) XX(stack_guarantee) XX(stack_guarantee_auto_init) XX(strict_continuation_asymmetric_lenient_continues) @@ -373,6 +374,7 @@ XX(stringbuilder_reserve_overflow) XX(symbolizer) XX(task_queue) XX(thread_without_name_still_valid) +XX(threadpool_ordered_parallel) XX(trace_continuation_truth_table) XX(trace_finish) XX(traceparent_header_disabled_by_default) From 95134a3934f8afdc2e619f3d22a87096802460bb Mon Sep 17 00:00:00 2001 From: J-P Nurmi Date: Fri, 17 Jul 2026 16:53:00 +0200 Subject: [PATCH 4/5] ref: add sentry_telemetry module Add an internal telemetry lifecycle layer for logs and metrics startup, shutdown, force flush, and crash-safe flush. Keep the existing logs and metrics behavior behind the shared coordination API. --- src/CMakeLists.txt | 2 ++ src/backends/sentry_backend_breakpad.cpp | 10 ++---- src/backends/sentry_backend_crashpad.cpp | 10 ++---- src/backends/sentry_backend_inproc.c | 10 ++---- src/backends/sentry_backend_native.c | 10 ++---- src/sentry_core.c | 34 +++---------------- src/sentry_logs.c | 4 --- src/sentry_metrics.c | 4 --- src/sentry_telemetry.c | 43 ++++++++++++++++++++++++ src/sentry_telemetry.h | 11 ++++++ 10 files changed, 69 insertions(+), 69 deletions(-) create mode 100644 src/sentry_telemetry.c create mode 100644 src/sentry_telemetry.h diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index bf37dd18a3..2a27b77071 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -59,6 +59,8 @@ sentry_target_sources_cwd(sentry sentry_symbolizer.h sentry_sync.c sentry_sync.h + sentry_telemetry.c + sentry_telemetry.h sentry_thread_stackwalk.h sentry_transport.c sentry_transport.h diff --git a/src/backends/sentry_backend_breakpad.cpp b/src/backends/sentry_backend_breakpad.cpp index f180259e3e..cef8f75383 100644 --- a/src/backends/sentry_backend_breakpad.cpp +++ b/src/backends/sentry_backend_breakpad.cpp @@ -9,8 +9,6 @@ extern "C" { #include "sentry_database.h" #include "sentry_envelope.h" #include "sentry_logger.h" -#include "sentry_logs.h" -#include "sentry_metrics.h" #include "sentry_options.h" #ifdef SENTRY_PLATFORM_WINDOWS # include "sentry_os.h" @@ -20,6 +18,7 @@ extern "C" { #include "sentry_session_replay.h" #include "sentry_string.h" #include "sentry_sync.h" +#include "sentry_telemetry.h" #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_unix_pageallocator.h" @@ -169,12 +168,7 @@ breakpad_backend_callback(const google_breakpad::MinidumpDescriptor &descriptor, sentry__transport_suspend(options->transport); // Flush logs and metrics in a crash-safe manner before crash handling - if (options->enable_logs) { - sentry__logs_flush_crash_safe(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); if (should_handle) { bool capture_screenshot = options->attach_screenshot; diff --git a/src/backends/sentry_backend_crashpad.cpp b/src/backends/sentry_backend_crashpad.cpp index a4d6eca9cf..712177957b 100644 --- a/src/backends/sentry_backend_crashpad.cpp +++ b/src/backends/sentry_backend_crashpad.cpp @@ -9,8 +9,6 @@ extern "C" { #include "sentry_database.h" #include "sentry_envelope.h" #include "sentry_logger.h" -#include "sentry_logs.h" -#include "sentry_metrics.h" #include "sentry_options.h" #ifdef SENTRY_PLATFORM_WINDOWS # include "sentry_os.h" @@ -19,6 +17,7 @@ extern "C" { #include "sentry_screenshot.h" #include "sentry_session_replay.h" #include "sentry_sync.h" +#include "sentry_telemetry.h" #include "sentry_transport.h" #include "sentry_value.h" #ifdef SENTRY_PLATFORM_LINUX @@ -479,12 +478,7 @@ crashpad_handler(int signum, siginfo_t *info, ucontext_t *user_context) sentry__transport_suspend(options->transport); // Flush logs and metrics in a crash-safe manner before crash handling - if (options->enable_logs) { - sentry__logs_flush_crash_safe(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); should_dump = !sentry_value_is_null(crash_event); diff --git a/src/backends/sentry_backend_inproc.c b/src/backends/sentry_backend_inproc.c index 2b1db89967..86854ef6dc 100644 --- a/src/backends/sentry_backend_inproc.c +++ b/src/backends/sentry_backend_inproc.c @@ -8,14 +8,13 @@ #include "sentry_database.h" #include "sentry_envelope.h" #include "sentry_logger.h" -#include "sentry_logs.h" -#include "sentry_metrics.h" #include "sentry_options.h" #include "sentry_os.h" #include "sentry_scope.h" #include "sentry_screenshot.h" #include "sentry_session_replay.h" #include "sentry_sync.h" +#include "sentry_telemetry.h" #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_unix_pageallocator.h" @@ -1088,12 +1087,7 @@ process_ucontext_deferred(const sentry_ucontext_t *uctx, sentry__transport_suspend(options->transport); // Flush logs in a crash-safe manner before crash handling - if (options->enable_logs) { - sentry__logs_flush_crash_safe(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); TEST_CRASH_POINT("before_capture"); if (should_handle) { bool capture_screenshot = options->attach_screenshot; diff --git a/src/backends/sentry_backend_native.c b/src/backends/sentry_backend_native.c index bf8c898fe0..ffc14c2182 100644 --- a/src/backends/sentry_backend_native.c +++ b/src/backends/sentry_backend_native.c @@ -28,8 +28,6 @@ #include "sentry_envelope.h" #include "sentry_json.h" #include "sentry_logger.h" -#include "sentry_logs.h" -#include "sentry_metrics.h" #include "sentry_options.h" #include "sentry_os.h" #include "sentry_path.h" @@ -37,6 +35,7 @@ #include "sentry_scope.h" #include "sentry_session.h" #include "sentry_sync.h" +#include "sentry_telemetry.h" #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_value.h" @@ -991,12 +990,7 @@ native_backend_except(sentry_backend_t *backend, const sentry_ucontext_t *uctx) sentry__transport_suspend(options->transport); // Flush logs and metrics in a crash-safe manner before crash handling - if (options->enable_logs) { - sentry__logs_flush_crash_safe(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); // Write crash marker sentry__write_crash_marker(options); diff --git a/src/sentry_core.c b/src/sentry_core.c index ec17f915fa..999f3c835e 100644 --- a/src/sentry_core.c +++ b/src/sentry_core.c @@ -12,8 +12,6 @@ #include "sentry_database.h" #include "sentry_envelope.h" #include "sentry_hint.h" -#include "sentry_logs.h" -#include "sentry_metrics.h" #include "sentry_options.h" #include "sentry_path.h" #include "sentry_process.h" @@ -22,6 +20,7 @@ #include "sentry_session.h" #include "sentry_string.h" #include "sentry_sync.h" +#include "sentry_telemetry.h" #include "sentry_tracing.h" #include "sentry_transport.h" #include "sentry_tsan.h" @@ -287,13 +286,8 @@ sentry_init(sentry_options_t *options) sentry_start_session(); } - if (options->enable_logs) { - sentry__logs_startup(options); - } - - if (options->enable_metrics) { - sentry__metrics_startup(options); - } + sentry__telemetry_shutdown(0); + sentry__telemetry_startup(options); if (options->enable_app_hang_tracking) { sentry__app_hang_monitor_start(options); @@ -318,20 +312,7 @@ sentry_flush(uint64_t timeout) int rv = 0; SENTRY_WITH_OPTIONS (options) { // flush logs and metrics in parallel - uintptr_t ltoken = 0; - uintptr_t mtoken = 0; - if (options->enable_logs) { - ltoken = sentry__logs_force_flush_begin(); - } - if (options->enable_metrics) { - mtoken = sentry__metrics_force_flush_begin(); - } - if (ltoken) { - sentry__logs_force_flush_wait(ltoken); - } - if (mtoken) { - sentry__metrics_force_flush_wait(mtoken); - } + sentry__telemetry_force_flush(); rv = sentry__transport_flush(options->transport, timeout); } return rv; @@ -350,12 +331,7 @@ sentry_close(void) } } - if (options->enable_logs) { - sentry__logs_shutdown(options->shutdown_timeout); - } - if (options->enable_metrics) { - sentry__metrics_shutdown(options->shutdown_timeout); - } + sentry__telemetry_shutdown(options->shutdown_timeout); } // Stop it before locking options to prevent a deadlock. The watchdog diff --git a/src/sentry_logs.c b/src/sentry_logs.c index 9edcc85594..37e6cc28b6 100644 --- a/src/sentry_logs.c +++ b/src/sentry_logs.c @@ -649,24 +649,20 @@ sentry__logs_startup(const sentry_options_t *options) void sentry__logs_shutdown(uint64_t timeout) { - SENTRY_DEBUG("shutting down logs system"); sentry_batcher_t *batcher = sentry__batcher_swap(&g_batcher, NULL); if (batcher) { sentry__batcher_shutdown(batcher, timeout); sentry__batcher_release(batcher); } - SENTRY_DEBUG("logs system shutdown complete"); } void sentry__logs_flush_crash_safe(void) { - SENTRY_SIGNAL_SAFE_LOG("DEBUG crash-safe logs flush"); sentry_batcher_t *batcher = sentry__batcher_peek(&g_batcher); if (batcher) { sentry__batcher_flush_crash_safe(batcher); } - SENTRY_SIGNAL_SAFE_LOG("DEBUG crash-safe logs flush complete"); } uintptr_t diff --git a/src/sentry_metrics.c b/src/sentry_metrics.c index 99685a1f7a..6d0a8fa8bd 100644 --- a/src/sentry_metrics.c +++ b/src/sentry_metrics.c @@ -153,24 +153,20 @@ sentry__metrics_startup(const sentry_options_t *options) void sentry__metrics_shutdown(uint64_t timeout) { - SENTRY_DEBUG("shutting down metrics system"); sentry_batcher_t *batcher = sentry__batcher_swap(&g_batcher, NULL); if (batcher) { sentry__batcher_shutdown(batcher, timeout); sentry__batcher_release(batcher); } - SENTRY_DEBUG("metrics system shutdown complete"); } void sentry__metrics_flush_crash_safe(void) { - SENTRY_SIGNAL_SAFE_LOG("DEBUG crash-safe metrics flush"); sentry_batcher_t *batcher = sentry__batcher_peek(&g_batcher); if (batcher) { sentry__batcher_flush_crash_safe(batcher); } - SENTRY_SIGNAL_SAFE_LOG("DEBUG crash-safe metrics flush complete"); } uintptr_t diff --git a/src/sentry_telemetry.c b/src/sentry_telemetry.c new file mode 100644 index 0000000000..387c3881f8 --- /dev/null +++ b/src/sentry_telemetry.c @@ -0,0 +1,43 @@ +#include "sentry_telemetry.h" +#include "sentry_logger.h" +#include "sentry_logs.h" +#include "sentry_metrics.h" +#include "sentry_options.h" + +void +sentry__telemetry_startup(const sentry_options_t *options) +{ + if (options->enable_logs) { + sentry__logs_startup(options); + } + if (options->enable_metrics) { + sentry__metrics_startup(options); + } +} + +void +sentry__telemetry_shutdown(uint64_t timeout) +{ + SENTRY_DEBUG("shutting down telemetry"); + sentry__logs_shutdown(timeout); + sentry__metrics_shutdown(timeout); + SENTRY_DEBUG("telemetry shutdown complete"); +} + +void +sentry__telemetry_force_flush(void) +{ + uintptr_t ltoken = sentry__logs_force_flush_begin(); + uintptr_t mtoken = sentry__metrics_force_flush_begin(); + sentry__logs_force_flush_wait(ltoken); + sentry__metrics_force_flush_wait(mtoken); +} + +void +sentry__telemetry_flush_crash_safe(void) +{ + SENTRY_SIGNAL_SAFE_LOG("DEBUG crash-safe telemetry flush"); + sentry__logs_flush_crash_safe(); + sentry__metrics_flush_crash_safe(); + SENTRY_SIGNAL_SAFE_LOG("DEBUG crash-safe telemetry flush complete"); +} diff --git a/src/sentry_telemetry.h b/src/sentry_telemetry.h new file mode 100644 index 0000000000..6e76b07e15 --- /dev/null +++ b/src/sentry_telemetry.h @@ -0,0 +1,11 @@ +#ifndef SENTRY_TELEMETRY_H_INCLUDED +#define SENTRY_TELEMETRY_H_INCLUDED + +#include "sentry_boot.h" + +void sentry__telemetry_startup(const sentry_options_t *options); +void sentry__telemetry_shutdown(uint64_t timeout); +void sentry__telemetry_force_flush(void); +void sentry__telemetry_flush_crash_safe(void); + +#endif From d1a77fe5eb0e4f6fd4f1e9c3d6512b3552c6ac4c Mon Sep 17 00:00:00 2001 From: J-P Nurmi Date: Fri, 17 Jul 2026 16:53:32 +0200 Subject: [PATCH 5/5] perf: serialize telemetry batches on shared thread pool Let the telemetry lifecycle own a shared serialization pool for enabled telemetry batchers. Use the pool for log and metric batch serialization while keeping completion ordered through the batcher flush lifecycle. --- src/sentry_batcher.c | 303 +++++++++++++++++++++++++++----- src/sentry_batcher.h | 10 +- src/sentry_logs.c | 5 +- src/sentry_logs.h | 4 +- src/sentry_metrics.c | 8 +- src/sentry_metrics.h | 4 +- src/sentry_telemetry.c | 45 ++++- tests/unit/CMakeLists.txt | 1 + tests/unit/test_batcher.c | 207 ++++++++++++++++++++++ tests/unit/test_client_report.c | 5 +- tests/unit/tests.inc | 5 + 11 files changed, 543 insertions(+), 54 deletions(-) create mode 100644 tests/unit/test_batcher.c diff --git a/src/sentry_batcher.c b/src/sentry_batcher.c index d0b797c901..89b7522ab5 100644 --- a/src/sentry_batcher.c +++ b/src/sentry_batcher.c @@ -22,8 +22,8 @@ #endif sentry_batcher_t * -sentry__batcher_new( - sentry_batch_func_t batch_func, sentry_data_category_t data_category) +sentry__batcher_new(sentry_batch_func_t batch_func, + sentry_data_category_t data_category, sentry_threadpool_t *threadpool) { sentry_batcher_t *batcher = SENTRY_MAKE(sentry_batcher_t); if (!batcher) { @@ -33,6 +33,7 @@ sentry__batcher_new( batcher->batch_func = batch_func; batcher->data_category = data_category; batcher->thread_state = (long)SENTRY_BATCHER_THREAD_STOPPED; + batcher->threadpool = threadpool; sentry__waitable_flag_init(&batcher->request_flush); sentry__thread_init(&batcher->batching_thread); return batcher; @@ -80,15 +81,13 @@ sentry__batcher_setname(sentry_batcher_t *batcher, const char *thread_name) static inline void lock_ref(sentry_batcher_ref_t *ref) { - while (!sentry__atomic_compare_swap(&ref->lock, 0, 1)) { - sentry__cpu_relax(); - } + sentry__spin_lock(&ref->lock); } static inline void unlock_ref(sentry_batcher_ref_t *ref) { - sentry__atomic_store(&ref->lock, 0); + sentry__spin_unlock(&ref->lock); } sentry_batcher_t * @@ -141,6 +140,17 @@ crash_safe_sleep_ms(uint64_t delay_ms) } } +static bool +crash_safe_spin_wait(int attempt, void *UNUSED(data)) +{ + if (attempt > 200) { + return false; + } + const uint32_t sleep_time = (attempt < 10) ? 1 : (attempt < 100) ? 5 : 10; + crash_safe_sleep_ms(sleep_time); + return true; +} + // Rotate the active buffer so producers can continue while the consumer is // busy. Producers and the consumer may both rotate, so use CAS. static bool @@ -167,27 +177,250 @@ rotate_buffer(sentry_batcher_t *batcher, long old_idx) return true; } +typedef enum { + SENTRY_BATCH_TASK_PENDING = 0, + SENTRY_BATCH_TASK_RUNNING = 1, + SENTRY_BATCH_TASK_READY = 2, + SENTRY_BATCH_TASK_DUMPED = 3, +} sentry_batch_task_state_t; + +typedef struct sentry_batch_task_s { + struct sentry_batch_task_s *next; + sentry_batcher_t *batcher; + sentry_envelope_t *envelope; + sentry_value_t items; + long state; +} sentry_batch_task_t; + +static void +lock_tasks(sentry_batcher_t *batcher) +{ + sentry__spin_lock(&batcher->task_lock); +} + +static bool +lock_tasks_crash_safe(sentry_batcher_t *batcher) +{ + return sentry__spin_lock_wait( + &batcher->task_lock, crash_safe_spin_wait, NULL); +} + +static void +unlock_tasks(sentry_batcher_t *batcher) +{ + sentry__spin_unlock(&batcher->task_lock); +} + +static void +batch_task_link(sentry_batch_task_t *task) +{ + sentry_batcher_t *batcher = task->batcher; + lock_tasks(batcher); + task->next = batcher->tasks; + batcher->tasks = task; + unlock_tasks(batcher); +} + +static void +batch_task_unlink_locked(sentry_batch_task_t *task) +{ + sentry_batch_task_t *prev = NULL; + sentry_batcher_t *batcher = task->batcher; + sentry_batch_task_t *cur = batcher->tasks; + while (cur) { + if (cur == task) { + if (prev) { + prev->next = cur->next; + } else { + batcher->tasks = cur->next; + } + task->next = NULL; + return; + } + prev = cur; + cur = cur->next; + } +} + +static void +batch_task_unlink(sentry_batch_task_t *task) +{ + sentry_batcher_t *batcher = task->batcher; + lock_tasks(batcher); + batch_task_unlink_locked(task); + unlock_tasks(batcher); +} + +static void +batch_task_exec(void *task_data) +{ + sentry_batch_task_t *task = task_data; + sentry_batcher_t *batcher = task->batcher; + + lock_tasks(batcher); + if (!sentry__atomic_compare_swap(&task->state, SENTRY_BATCH_TASK_PENDING, + SENTRY_BATCH_TASK_RUNNING)) { + unlock_tasks(batcher); + return; + } + unlock_tasks(batcher); + + task->batcher->batch_func(task->envelope, task->items); + sentry_value_decref(task->items); + task->items = sentry_value_new_null(); + lock_tasks(batcher); + sentry__atomic_store(&task->state, SENTRY_BATCH_TASK_READY); + unlock_tasks(batcher); +} + +static void +batch_task_complete(void *task_data) +{ + sentry_batch_task_t *task = task_data; + sentry_batcher_t *batcher = task->batcher; + + lock_tasks(batcher); + const long state = sentry__atomic_fetch(&task->state); + if (state == SENTRY_BATCH_TASK_DUMPED) { + batch_task_unlink_locked(task); + unlock_tasks(batcher); + return; + } + batch_task_unlink_locked(task); + unlock_tasks(batcher); + + if (sentry__atomic_fetch(&task->batcher->crash_flush)) { + sentry__run_write_envelope(task->batcher->run, task->envelope); + sentry_envelope_free(task->envelope); + } else if (!sentry__run_should_skip_upload(task->batcher->run)) { + // Normal operation: use transport for HTTP transmission + sentry__transport_send_envelope( + task->batcher->transport, task->envelope); + } else { + sentry_envelope_free(task->envelope); + } + task->envelope = NULL; +} + +static void +batch_task_cleanup(void *task_data) +{ + sentry_batch_task_t *task = task_data; + batch_task_unlink(task); + sentry_value_decref(task->items); + sentry_envelope_free(task->envelope); + sentry_free(task); +} + +static void +batch_task_dump_pending(sentry_batch_task_t *task) +{ + sentry_batcher_t *batcher = task->batcher; + const long state = sentry__atomic_fetch(&task->state); + if (state == SENTRY_BATCH_TASK_PENDING) { + sentry__atomic_store(&task->state, SENTRY_BATCH_TASK_DUMPED); + batcher->batch_func(task->envelope, task->items); + sentry_value_decref(task->items); + task->items = sentry_value_new_null(); + } else if (state == SENTRY_BATCH_TASK_READY) { + sentry__atomic_store(&task->state, SENTRY_BATCH_TASK_DUMPED); + } else { + return; + } + + sentry__run_write_envelope(batcher->run, task->envelope); + sentry_envelope_free(task->envelope); + task->envelope = NULL; +} + +static void +batch_task_dump_pending_all(sentry_batcher_t *batcher) +{ + int attempts = 0; + while (true) { + bool has_running = false; + if (!lock_tasks_crash_safe(batcher)) { + return; + } + for (sentry_batch_task_t *task = batcher->tasks; task; + task = task->next) { + batch_task_dump_pending(task); + has_running = has_running + || sentry__atomic_fetch(&task->state) + == SENTRY_BATCH_TASK_RUNNING; + } + unlock_tasks(batcher); + + if (!has_running) { + return; + } + + const int max_attempts = 200; + if (++attempts > max_attempts) { + return; + } + const uint32_t sleep_time = (attempts < 10) ? 1 + : (attempts < 100) ? 5 + : 10; + crash_safe_sleep_ms(sleep_time); + } +} + +static void +process_batch(sentry_batcher_t *batcher, sentry_value_t items, bool crash_safe) +{ + sentry_envelope_t *envelope = sentry__envelope_new_with_dsn(batcher->dsn); + if (crash_safe) { + // Write directly to disk to avoid transport queuing during crash + batcher->batch_func(envelope, items); + sentry__run_write_envelope(batcher->run, envelope); + sentry_envelope_free(envelope); + sentry_value_decref(items); + return; + } + + sentry_batch_task_t *task = SENTRY_MAKE(sentry_batch_task_t); + if (!task) { + sentry_value_decref(items); + sentry_envelope_free(envelope); + return; + } + + task->batcher = batcher; + task->envelope = envelope; + task->items = items; + task->state = SENTRY_BATCH_TASK_PENDING; + batch_task_link(task); + + if (!batcher->threadpool) { + batch_task_exec(task); + batch_task_complete(task); + batch_task_cleanup(task); + return; + } + + if (sentry__threadpool_submit(batcher->threadpool, batch_task_exec, + batch_task_complete, batch_task_cleanup, task) + != 0) { + SENTRY_WARN( + "dropping telemetry batch: serialization pool unavailable or out " + "of memory"); + batch_task_unlink(task); + batch_task_cleanup(task); + return; + } +} + bool sentry__batcher_flush(sentry_batcher_t *batcher, bool crash_safe) { if (crash_safe) { - // In crash-safe mode, spin lock with timeout and backoff - int attempts = 0; - while (!sentry__atomic_compare_swap(&batcher->flushing, 0, 1)) { - const int max_attempts = 200; - if (++attempts > max_attempts) { - SENTRY_SIGNAL_SAFE_LOG( - "WARN sentry__batcher_flush: timeout waiting for " - "flushing lock in crash-safe mode"); - return false; - } - - // backoff max-wait with max_attempts = 200 based sleep slots: - // 9ms + 450ms + 1010ms = 1500ish ms - const uint32_t sleep_time = (attempts < 10) ? 1 - : (attempts < 100) ? 5 - : 10; - crash_safe_sleep_ms(sleep_time); + if (!sentry__spin_lock_wait( + &batcher->flushing, crash_safe_spin_wait, NULL)) { + SENTRY_SIGNAL_SAFE_LOG( + "WARN sentry__batcher_flush: timeout waiting for " + "flushing lock in crash-safe mode"); + return false; } } else { // Normal mode: try once and return if already flushing @@ -239,22 +472,7 @@ sentry__batcher_flush(sentry_batcher_t *batcher, bool crash_safe) } sentry_value_set_by_key(logs, "items", log_items); - sentry_envelope_t *envelope - = sentry__envelope_new_with_dsn(batcher->dsn); - batcher->batch_func(envelope, logs); - - if (crash_safe) { - // Write directly to disk to avoid transport queuing during - // crash - sentry__run_write_envelope(batcher->run, envelope); - sentry_envelope_free(envelope); - } else if (!sentry__run_should_skip_upload(batcher->run)) { - // Normal operation: use transport for HTTP transmission - sentry__transport_send_envelope(batcher->transport, envelope); - } else { - sentry_envelope_free(envelope); - } - sentry_value_decref(logs); + process_batch(batcher, logs, crash_safe); } // Reset the drained buffer... @@ -264,7 +482,7 @@ sentry__batcher_flush(sentry_batcher_t *batcher, bool crash_safe) (old_buf_idx + 1) % SENTRY_BATCHER_BUFFER_COUNT); } - sentry__atomic_store(&batcher->flushing, 0); + sentry__spin_unlock(&batcher->flushing); return true; } @@ -449,14 +667,17 @@ sentry__batcher_shutdown(sentry_batcher_t *batcher, uint64_t timeout) // Perform final flush to ensure any remaining items are sent sentry__batcher_flush(batcher, false); + sentry__threadpool_flush(batcher->threadpool); } void sentry__batcher_flush_crash_safe(sentry_batcher_t *batcher) { // Check if batcher is initialized + sentry__atomic_store(&batcher->crash_flush, 1); const long state = sentry__atomic_fetch(&batcher->thread_state); if (state == SENTRY_BATCHER_THREAD_STOPPED) { + batch_task_dump_pending_all(batcher); return; } @@ -469,6 +690,7 @@ sentry__batcher_flush_crash_safe(sentry_batcher_t *batcher) // This is safe because we're in a crash scenario and the main thread // is likely dead or dying anyway sentry__batcher_flush(batcher, true); + batch_task_dump_pending_all(batcher); } void @@ -487,6 +709,7 @@ sentry__batcher_force_flush_wait(sentry_batcher_t *batcher) } // retry if the batcher thread (woken by _begin) wins the race } while (!sentry__batcher_flush(batcher, false)); + sentry__threadpool_flush(batcher->threadpool); } #ifdef SENTRY_UNITTEST diff --git a/src/sentry_batcher.h b/src/sentry_batcher.h index af04b9fa9c..653c8238fe 100644 --- a/src/sentry_batcher.h +++ b/src/sentry_batcher.h @@ -39,15 +39,21 @@ typedef struct { typedef sentry_envelope_item_t *(*sentry_batch_func_t)( sentry_envelope_t *envelope, sentry_value_t items); +struct sentry_batch_task_s; + typedef struct { long refcount; // (atomic) reference count sentry_batcher_buffer_t buffers[SENTRY_BATCHER_BUFFER_COUNT]; long active_idx; // (atomic) index to the active buffer long drain_idx; // (atomic) index to the oldest buffer to drain long flushing; // (atomic) reentrancy guard to the flusher + long crash_flush; // (atomic) write completed batch work to disk + long task_lock; // (atomic) protects in-flight batch tasks + struct sentry_batch_task_s *tasks; // in-flight batch tasks long thread_state; // (atomic) sentry_batcher_thread_state_t sentry_waitable_flag_t request_flush; // level-triggered flush flag sentry_threadid_t batching_thread; // the batching thread + sentry_threadpool_t *threadpool; // thread pool for batch work sentry_batch_func_t batch_func; // function to add items to envelope sentry_data_category_t data_category; // for client report discard tracking char *thread_name; @@ -63,8 +69,8 @@ typedef struct { #define SENTRY_BATCHER_REF_INIT { NULL, 0 } -sentry_batcher_t *sentry__batcher_new( - sentry_batch_func_t batch_func, sentry_data_category_t data_category); +sentry_batcher_t *sentry__batcher_new(sentry_batch_func_t batch_func, + sentry_data_category_t data_category, sentry_threadpool_t *threadpool); void sentry__batcher_setname( sentry_batcher_t *batcher, const char *thread_name); diff --git a/src/sentry_logs.c b/src/sentry_logs.c index 37e6cc28b6..3a32820546 100644 --- a/src/sentry_logs.c +++ b/src/sentry_logs.c @@ -627,10 +627,11 @@ sentry_scope_capture_log(sentry_scope_t *scope, sentry_level_t level, } void -sentry__logs_startup(const sentry_options_t *options) +sentry__logs_startup( + const sentry_options_t *options, sentry_threadpool_t *threadpool) { sentry_batcher_t *batcher = sentry__batcher_new( - sentry__envelope_add_logs, SENTRY_DATA_CATEGORY_LOG_ITEM); + sentry__envelope_add_logs, SENTRY_DATA_CATEGORY_LOG_ITEM, threadpool); if (!batcher) { SENTRY_WARN("failed to allocate logs batcher"); return; diff --git a/src/sentry_logs.h b/src/sentry_logs.h index 062508028e..30a4be0a08 100644 --- a/src/sentry_logs.h +++ b/src/sentry_logs.h @@ -2,6 +2,7 @@ #define SENTRY_LOGS_H_INCLUDED #include "sentry_boot.h" +#include "sentry_sync.h" log_return_value_t sentry__logs_log( sentry_level_t level, const char *message, va_list args); @@ -9,7 +10,8 @@ log_return_value_t sentry__logs_log( /** * Sets up the logs timer/flush thread */ -void sentry__logs_startup(const sentry_options_t *options); +void sentry__logs_startup( + const sentry_options_t *options, sentry_threadpool_t *threadpool); /** * Shuts down the logs timer/flush thread. diff --git a/src/sentry_metrics.c b/src/sentry_metrics.c index 6d0a8fa8bd..c669edaf91 100644 --- a/src/sentry_metrics.c +++ b/src/sentry_metrics.c @@ -131,10 +131,12 @@ sentry_metrics_distribution( } void -sentry__metrics_startup(const sentry_options_t *options) +sentry__metrics_startup( + const sentry_options_t *options, sentry_threadpool_t *threadpool) { - sentry_batcher_t *batcher = sentry__batcher_new( - sentry__envelope_add_metrics, SENTRY_DATA_CATEGORY_TRACE_METRIC); + sentry_batcher_t *batcher + = sentry__batcher_new(sentry__envelope_add_metrics, + SENTRY_DATA_CATEGORY_TRACE_METRIC, threadpool); if (!batcher) { SENTRY_WARN("failed to allocate metrics batcher"); return; diff --git a/src/sentry_metrics.h b/src/sentry_metrics.h index 04053351f4..2ebd741da2 100644 --- a/src/sentry_metrics.h +++ b/src/sentry_metrics.h @@ -2,11 +2,13 @@ #define SENTRY_METRICS_H_INCLUDED #include "sentry_boot.h" +#include "sentry_sync.h" /** * Sets up the metrics timer/flush thread */ -void sentry__metrics_startup(const sentry_options_t *options); +void sentry__metrics_startup( + const sentry_options_t *options, sentry_threadpool_t *threadpool); /** * Shuts down the metrics timer/flush thread. diff --git a/src/sentry_telemetry.c b/src/sentry_telemetry.c index 387c3881f8..b170ebfc23 100644 --- a/src/sentry_telemetry.c +++ b/src/sentry_telemetry.c @@ -1,36 +1,77 @@ #include "sentry_telemetry.h" +#include "sentry_batcher.h" #include "sentry_logger.h" #include "sentry_logs.h" #include "sentry_metrics.h" #include "sentry_options.h" +#include "sentry_sync.h" + +static sentry_threadpool_t *g_telemetry_pool = NULL; +#ifdef SENTRY__MUTEX_INIT_DYN +SENTRY__MUTEX_INIT_DYN(g_telemetry_lock) +#else +static sentry_mutex_t g_telemetry_lock = SENTRY__MUTEX_INIT; +#endif void sentry__telemetry_startup(const sentry_options_t *options) { + SENTRY__MUTEX_INIT_DYN_ONCE(g_telemetry_lock); + sentry__mutex_lock(&g_telemetry_lock); + + if (!options->enable_logs && !options->enable_metrics) { + sentry__mutex_unlock(&g_telemetry_lock); + return; + } + + // use two workers for serializing telemetry batches off the batcher threads + g_telemetry_pool = sentry__threadpool_new(2); + sentry__threadpool_setname(g_telemetry_pool, "sentry-tele"); + if (!g_telemetry_pool || sentry__threadpool_start(g_telemetry_pool) != 0) { + sentry__threadpool_free(g_telemetry_pool); + g_telemetry_pool = NULL; + SENTRY_WARN( + "telemetry pool unavailable; serializing in batcher thread"); + } + if (options->enable_logs) { - sentry__logs_startup(options); + sentry__logs_startup(options, g_telemetry_pool); } if (options->enable_metrics) { - sentry__metrics_startup(options); + sentry__metrics_startup(options, g_telemetry_pool); } + sentry__mutex_unlock(&g_telemetry_lock); } void sentry__telemetry_shutdown(uint64_t timeout) { + SENTRY__MUTEX_INIT_DYN_ONCE(g_telemetry_lock); + sentry__mutex_lock(&g_telemetry_lock); + SENTRY_DEBUG("shutting down telemetry"); sentry__logs_shutdown(timeout); sentry__metrics_shutdown(timeout); + sentry__threadpool_shutdown(g_telemetry_pool); + sentry__threadpool_free(g_telemetry_pool); + g_telemetry_pool = NULL; SENTRY_DEBUG("telemetry shutdown complete"); + + sentry__mutex_unlock(&g_telemetry_lock); } void sentry__telemetry_force_flush(void) { + SENTRY__MUTEX_INIT_DYN_ONCE(g_telemetry_lock); + sentry__mutex_lock(&g_telemetry_lock); + uintptr_t ltoken = sentry__logs_force_flush_begin(); uintptr_t mtoken = sentry__metrics_force_flush_begin(); sentry__logs_force_flush_wait(ltoken); sentry__metrics_force_flush_wait(mtoken); + + sentry__mutex_unlock(&g_telemetry_lock); } void diff --git a/tests/unit/CMakeLists.txt b/tests/unit/CMakeLists.txt index b3689ba7d7..f1093cea34 100644 --- a/tests/unit/CMakeLists.txt +++ b/tests/unit/CMakeLists.txt @@ -21,6 +21,7 @@ add_executable(sentry_test_unit ${SENTRY_SOURCES} main.c sentry_testsupport.h + test_batcher.c test_app_hang.c test_attachments.c test_basic.c diff --git a/tests/unit/test_batcher.c b/tests/unit/test_batcher.c new file mode 100644 index 0000000000..ae03a47780 --- /dev/null +++ b/tests/unit/test_batcher.c @@ -0,0 +1,207 @@ +#include "sentry_batcher.h" +#include "sentry_database.h" +#include "sentry_envelope.h" +#include "sentry_path.h" +#include "sentry_sync.h" +#include "sentry_testsupport.h" +#include "sentry_transport.h" +#include "sentry_value.h" + +typedef struct { + long started; + long release; +} blocking_task_t; + +static void +blocking_task_exec(void *data) +{ + blocking_task_t *task = data; + sentry__atomic_store(&task->started, 1); + while (!sentry__atomic_fetch(&task->release)) { + sentry__thread_yield(); + } +} + +static sentry_envelope_item_t * +pending_batch_func(sentry_envelope_t *envelope, sentry_value_t items) +{ + (void)items; + return sentry__envelope_add_from_buffer(envelope, "{}", 2, "event"); +} + +static void +counting_transport_send(sentry_envelope_t *envelope, void *data) +{ + sentry__atomic_fetch_and_add((long *)data, 1); + sentry_envelope_free(envelope); +} + +static sentry_run_t * +new_test_run(const char *name, sentry_path_t **database_path) +{ + *database_path = sentry__path_from_str(name); + TEST_ASSERT(!!*database_path); + sentry__path_remove_all(*database_path); + TEST_ASSERT(!sentry__path_create_dir_all(*database_path)); + sentry_run_t *run = sentry__run_new(*database_path); + TEST_ASSERT(!!run); + return run; +} + +static void +free_test_run(sentry_run_t *run, sentry_path_t *database_path) +{ + sentry__run_clean(run, true); + sentry__run_free(run); + sentry__path_remove_all(database_path); + sentry__path_free(database_path); +} + +SENTRY_TEST(batcher_sync_flush_sends) +{ + sentry_batcher_t *batcher = sentry__batcher_new( + pending_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, NULL); + TEST_ASSERT(!!batcher); + sentry_path_t *database_path = NULL; + sentry_run_t *run = new_test_run( + SENTRY_TEST_PATH_PREFIX ".batcher-sync-flush", &database_path); + long sent = 0; + sentry_transport_t *transport + = sentry_transport_new(counting_transport_send); + TEST_ASSERT(!!transport); + sentry_transport_set_state(transport, &sent); + batcher->run = run; + batcher->transport = transport; + + TEST_CHECK(sentry__batcher_enqueue(batcher, sentry_value_new_null())); + TEST_CHECK(sentry__batcher_flush(batcher, false)); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&sent), 1); + + sentry__batcher_release(batcher); + sentry_transport_free(transport); + free_test_run(run, database_path); +} + +SENTRY_TEST(batcher_enqueue_overflow) +{ + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + sentry_batcher_t *batcher = sentry__batcher_new( + pending_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, pool); + TEST_ASSERT(!!batcher); + + for (int i = 0; + i < SENTRY_BATCHER_BUFFER_COUNT * SENTRY_BATCHER_QUEUE_LENGTH; i++) { + TEST_CHECK(sentry__batcher_enqueue(batcher, sentry_value_new_null())); + } + TEST_CHECK(!sentry__batcher_enqueue(batcher, sentry_value_new_null())); + + sentry__batcher_release(batcher); + sentry__threadpool_free(pool); +} + +SENTRY_TEST(batcher_force_flush_sends) +{ + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + TEST_ASSERT(!sentry__threadpool_start(pool)); + sentry_path_t *database_path = NULL; + sentry_run_t *run = new_test_run( + SENTRY_TEST_PATH_PREFIX ".batcher-force-flush", &database_path); + long sent = 0; + sentry_transport_t *transport + = sentry_transport_new(counting_transport_send); + TEST_ASSERT(!!transport); + sentry_transport_set_state(transport, &sent); + sentry_batcher_t *batcher = sentry__batcher_new( + pending_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, pool); + TEST_ASSERT(!!batcher); + batcher->run = run; + batcher->transport = transport; + + TEST_CHECK(sentry__batcher_enqueue(batcher, sentry_value_new_null())); + sentry__batcher_force_flush_begin(batcher); + sentry__batcher_force_flush_wait(batcher); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&sent), 1); + + sentry__batcher_release(batcher); + sentry_transport_free(transport); + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); + free_test_run(run, database_path); +} + +SENTRY_TEST(batcher_crash_flush_buffers) +{ + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + sentry_path_t *database_path = NULL; + sentry_run_t *run = new_test_run( + SENTRY_TEST_PATH_PREFIX ".batcher-crash-flush", &database_path); + sentry_batcher_t *batcher = sentry__batcher_new( + pending_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, pool); + TEST_ASSERT(!!batcher); + batcher->run = run; + sentry__atomic_store( + &batcher->thread_state, (long)SENTRY_BATCHER_THREAD_RUNNING); + + TEST_CHECK(sentry__batcher_enqueue(batcher, sentry_value_new_null())); + sentry__batcher_flush_crash_safe(batcher); + TEST_CHECK(sentry__atomic_fetch(&run->retain)); + + sentry__batcher_release(batcher); + sentry__threadpool_free(pool); + free_test_run(run, database_path); +} + +SENTRY_TEST(batcher_crash_flush_pending) +{ + sentry_path_t *database_path = sentry__path_from_str( + SENTRY_TEST_PATH_PREFIX ".batcher-pending-crash-flush"); + TEST_ASSERT(!!database_path); + sentry__path_remove_all(database_path); + TEST_ASSERT(!sentry__path_create_dir_all(database_path)); + sentry_run_t *run = sentry__run_new(database_path); + TEST_ASSERT(!!run); + + sentry_threadpool_t *pool = sentry__threadpool_new(1); + TEST_ASSERT(!!pool); + TEST_ASSERT(!sentry__threadpool_start(pool)); + + blocking_task_t blocking_task = { 0 }; + TEST_ASSERT(!sentry__threadpool_submit( + pool, blocking_task_exec, NULL, NULL, &blocking_task)); + while (!sentry__atomic_fetch(&blocking_task.started)) { + sentry__thread_yield(); + } + + sentry_batcher_t *batcher = sentry__batcher_new( + pending_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, pool); + TEST_ASSERT(!!batcher); + long sent = 0; + sentry_transport_t *transport + = sentry_transport_new(counting_transport_send); + TEST_ASSERT(!!transport); + sentry_transport_set_state(transport, &sent); + batcher->run = run; + batcher->transport = transport; + + TEST_CHECK(sentry__batcher_enqueue(batcher, sentry_value_new_null())); + TEST_CHECK(sentry__batcher_flush(batcher, false)); + TEST_CHECK(!sentry__atomic_fetch(&run->retain)); + + sentry__batcher_flush_crash_safe(batcher); + TEST_CHECK(sentry__atomic_fetch(&run->retain)); + + sentry__atomic_store(&blocking_task.release, 1); + sentry__threadpool_flush(pool); + TEST_CHECK_INT_EQUAL(sentry__atomic_fetch(&sent), 0); + sentry__threadpool_shutdown(pool); + sentry__threadpool_free(pool); + sentry__batcher_release(batcher); + sentry_transport_free(transport); + sentry__run_clean(run, true); + sentry__run_free(run); + sentry__path_remove_all(database_path); + sentry__path_free(database_path); +} diff --git a/tests/unit/test_client_report.c b/tests/unit/test_client_report.c index e97b90901e..d9cac2d81a 100644 --- a/tests/unit/test_client_report.c +++ b/tests/unit/test_client_report.c @@ -5,7 +5,6 @@ #include "sentry_options.h" #include "sentry_path.h" #include "sentry_ratelimiter.h" -#include "sentry_sync.h" #include "sentry_testsupport.h" #include "sentry_uuid.h" #include "sentry_value.h" @@ -390,8 +389,8 @@ SENTRY_TEST(client_report_queue_overflow) SENTRY_TEST_OPTIONS_NEW(options); sentry_init(options); - sentry_batcher_t *batcher - = sentry__batcher_new(dummy_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM); + sentry_batcher_t *batcher = sentry__batcher_new( + dummy_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, NULL); TEST_CHECK(!!batcher); // Fill all buffers (SENTRY_BATCHER_QUEUE_LENGTH is 5 in unit tests) diff --git a/tests/unit/tests.inc b/tests/unit/tests.inc index 3d6d4c60e1..f806dfa2f9 100644 --- a/tests/unit/tests.inc +++ b/tests/unit/tests.inc @@ -48,6 +48,11 @@ XX(basic_tracing_context) XX(basic_transaction) XX(basic_transport_thread_name) XX(basic_write_envelope_to_file) +XX(batcher_crash_flush_buffers) +XX(batcher_crash_flush_pending) +XX(batcher_enqueue_overflow) +XX(batcher_force_flush_sends) +XX(batcher_sync_flush_sends) XX(before_breadcrumb_discard) XX(before_breadcrumb_modify) XX(before_breadcrumb_passthrough)