diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index bf37dd18a..2a27b7707 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 9d397b042..cef8f7538 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,10 +18,10 @@ 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" -#include "transports/sentry_disk_transport.h" } #ifdef __GNUC__ @@ -167,13 +165,10 @@ 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(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); if (should_handle) { bool capture_screenshot = options->attach_screenshot; @@ -246,33 +241,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 61a44a170..3e71aa88f 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 @@ -26,7 +25,6 @@ extern "C" { #endif #include "sentry_utils.h" #include "sentry_uuid.h" -#include "transports/sentry_disk_transport.h" } #include @@ -440,13 +438,10 @@ 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(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); should_dump = !sentry_value_is_null(crash_event); @@ -461,13 +456,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)) { @@ -480,15 +469,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 5004f547f..1faa00109 100644 --- a/src/backends/sentry_backend_inproc.c +++ b/src/backends/sentry_backend_inproc.c @@ -8,8 +8,6 @@ #include "sentry_database.h" #include "sentry_envelope.h" #include "sentry_logger.h" -#include "sentry_logs.h" -#include "sentry_metrics.h" #include "sentry_options.h" #if defined(SENTRY_PLATFORM_WINDOWS) # include "sentry_os.h" @@ -18,10 +16,10 @@ #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" -#include "transports/sentry_disk_transport.h" #include #include #ifdef SENTRY_PLATFORM_UNIX @@ -1127,13 +1125,10 @@ 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(); - } - 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; @@ -1176,33 +1171,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 010444388..844306457 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,10 +35,10 @@ #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" -#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 @@ -977,13 +975,10 @@ 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(); - } - if (options->enable_metrics) { - sentry__metrics_flush_crash_safe(); - } + sentry__telemetry_flush_crash_safe(); // Write crash marker sentry__write_crash_marker(options); @@ -1075,32 +1070,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_batcher.c b/src/sentry_batcher.c index 4572ab666..db13db795 100644 --- a/src/sentry_batcher.c +++ b/src/sentry_batcher.c @@ -20,8 +20,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) { @@ -31,6 +31,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; @@ -67,15 +68,13 @@ sentry__batcher_release(sentry_batcher_t *batcher) 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 * @@ -128,6 +127,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 @@ -154,27 +164,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 @@ -226,22 +459,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... @@ -251,7 +469,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; } @@ -321,6 +539,13 @@ SENTRY_THREAD_FN batcher_thread_func(void *data) { sentry_batcher_t *batcher = data; + const char *thread_name = "sentry-batcher"; + if (batcher->data_category == SENTRY_DATA_CATEGORY_LOG_ITEM) { + thread_name = "sentry-logs"; + } else if (batcher->data_category == SENTRY_DATA_CATEGORY_TRACE_METRIC) { + thread_name = "sentry-metrics"; + } + sentry__thread_setname(sentry__current_thread(), thread_name); SENTRY_DEBUG("Starting batching thread"); // Transition from STARTING to RUNNING using compare-and-swap @@ -433,14 +658,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; } @@ -453,6 +681,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 @@ -471,6 +700,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 50e1bc6e7..68695dc7c 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 sentry_dsn_t *dsn; @@ -62,8 +68,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); /** * Acquires a reference to the batcher behind `ref`, atomically incrementing diff --git a/src/sentry_core.c b/src/sentry_core.c index ce47a38d4..cb53ab3a7 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" @@ -161,6 +160,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 +202,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; @@ -278,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); @@ -309,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; @@ -341,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 7c5991682..1c4bf6c44 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; @@ -648,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_logs.h b/src/sentry_logs.h index 062508028..30a4be0a0 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 033af6b46..cec746ef9 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; @@ -152,24 +154,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_metrics.h b/src/sentry_metrics.h index 04053351f..2ebd741da 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_options.c b/src/sentry_options.c index 7ea042851..46de8c3a0 100644 --- a/src/sentry_options.c +++ b/src/sentry_options.c @@ -194,7 +194,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_sync.c b/src/sentry_sync.c index de51acc4a..9d7a92b74 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 @@ -101,6 +102,280 @@ thread_setname(sentry_threadid_t thread_id, const char *thread_name) } #endif +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); +} + +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 adc904fb2..08b3e3829 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 @@ -369,6 +370,9 @@ sentry__cond_wait_timeout( } #endif +int sentry__thread_setname( + sentry_threadid_t thread_id, const char *thread_name); + static inline long sentry__atomic_fetch_and_add(volatile long *val, long diff) { @@ -431,6 +435,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 @@ -460,8 +493,26 @@ sentry__atomic_fetch_u64(uint64_t *val) 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/src/sentry_telemetry.c b/src/sentry_telemetry.c new file mode 100644 index 000000000..b170ebfc2 --- /dev/null +++ b/src/sentry_telemetry.c @@ -0,0 +1,84 @@ +#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, g_telemetry_pool); + } + if (options->enable_metrics) { + 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 +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 000000000..6e76b07e1 --- /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 diff --git a/src/sentry_transport.c b/src/sentry_transport.c index 03c6a5fbf..09d292755 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 a4ce8197e..4c4dd0ef0 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/CMakeLists.txt b/tests/unit/CMakeLists.txt index b3689ba7d..f1093cea3 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 000000000..ae03a4778 --- /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 e97b90901..d9cac2d81 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/test_concurrency.c b/tests/unit/test_concurrency.c index 1129ff7f5..cdc3c739a 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 6e8cb1081..169d0fb24 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_sync.c b/tests/unit/test_sync.c index 98c5eeef9..3bc95c096 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/test_uninit.c b/tests/unit/test_uninit.c index 03ff3209d..478096f2c 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 d078e4d7c..bf59376b5 100644 --- a/tests/unit/tests.inc +++ b/tests/unit/tests.inc @@ -47,6 +47,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) @@ -354,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) @@ -368,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) @@ -377,6 +384,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)