Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Appearance settings

Commit 829189b

Browse filesBrowse files
jpnurmijpnurmi
committed
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.
1 parent f6edfe7 commit 829189b
Copy full SHA for 829189b

8 files changed

+131-32Lines changed: 131 additions & 32 deletions

File tree

Expand file treeCollapse file tree
Open diff view settings
Filter options
Expand file treeCollapse file tree
Open diff view settings
Collapse file

‎src/sentry_batcher.c‎

Copy file name to clipboardExpand all lines: src/sentry_batcher.c
+83-19Lines changed: 83 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -20,17 +20,19 @@
2020
#endif
2121

2222
sentry_batcher_t *
23-
sentry__batcher_new(
24-
sentry_batch_func_t batch_func, sentry_data_category_t data_category)
23+
sentry__batcher_new(sentry_batch_func_t batch_func,
24+
sentry_data_category_t data_category, sentry_threadpool_t *threadpool)
2525
{
2626
sentry_batcher_t *batcher = SENTRY_MAKE(sentry_batcher_t);
27-
if (!batcher) {
27+
if (!batcher || !threadpool) {
28+
sentry_free(batcher);
2829
return NULL;
2930
}
3031
batcher->refcount = 1;
3132
batcher->batch_func = batch_func;
3233
batcher->data_category = data_category;
3334
batcher->thread_state = (long)SENTRY_BATCHER_THREAD_STOPPED;
35+
batcher->threadpool = threadpool;
3436
sentry__waitable_flag_init(&batcher->request_flush);
3537
sentry__thread_init(&batcher->batching_thread);
3638
return batcher;
@@ -154,6 +156,80 @@ rotate_buffer(sentry_batcher_t *batcher, long old_idx)
154156
return true;
155157
}
156158

159+
typedef struct {
160+
sentry_batcher_t *batcher;
161+
sentry_envelope_t *envelope;
162+
sentry_value_t items;
163+
} sentry_batch_task_t;
164+
165+
static void
166+
batch_task_exec(void *task_data)
167+
{
168+
sentry_batch_task_t *task = task_data;
169+
task->batcher->batch_func(task->envelope, task->items);
170+
sentry_value_decref(task->items);
171+
task->items = sentry_value_new_null();
172+
}
173+
174+
static void
175+
batch_task_complete(void *task_data)
176+
{
177+
sentry_batch_task_t *task = task_data;
178+
if (sentry__atomic_fetch(&task->batcher->crash_flush)) {
179+
sentry__run_write_envelope(task->batcher->run, task->envelope);
180+
sentry_envelope_free(task->envelope);
181+
} else if (!sentry__run_should_skip_upload(task->batcher->run)) {
182+
// Normal operation: use transport for HTTP transmission
183+
sentry__transport_send_envelope(
184+
task->batcher->transport, task->envelope);
185+
} else {
186+
sentry_envelope_free(task->envelope);
187+
}
188+
task->envelope = NULL;
189+
}
190+
191+
static void
192+
batch_task_cleanup(void *task_data)
193+
{
194+
sentry_batch_task_t *task = task_data;
195+
sentry_value_decref(task->items);
196+
sentry_envelope_free(task->envelope);
197+
sentry_free(task);
198+
}
199+
200+
static void
201+
process_batch(sentry_batcher_t *batcher, sentry_value_t items, bool crash_safe)
202+
{
203+
sentry_envelope_t *envelope = sentry__envelope_new_with_dsn(batcher->dsn);
204+
if (crash_safe) {
205+
// Write directly to disk to avoid transport queuing during crash
206+
batcher->batch_func(envelope, items);
207+
sentry__run_write_envelope(batcher->run, envelope);
208+
sentry_envelope_free(envelope);
209+
sentry_value_decref(items);
210+
return;
211+
}
212+
213+
sentry_batch_task_t *task = SENTRY_MAKE(sentry_batch_task_t);
214+
if (!task) {
215+
sentry_value_decref(items);
216+
sentry_envelope_free(envelope);
217+
return;
218+
}
219+
220+
task->batcher = batcher;
221+
task->envelope = envelope;
222+
task->items = items;
223+
if (sentry__threadpool_submit(batcher->threadpool, batch_task_exec,
224+
batch_task_complete, batch_task_cleanup, task)
225+
!= 0) {
226+
sentry_value_decref(items);
227+
sentry_envelope_free(envelope);
228+
sentry_free(task);
229+
return;
230+
}
231+
}
232+
157233
bool
158234
sentry__batcher_flush(sentry_batcher_t *batcher, bool crash_safe)
159235
{
@@ -226,22 +302,7 @@ sentry__batcher_flush(sentry_batcher_t *batcher, bool crash_safe)
226302
}
227303
sentry_value_set_by_key(logs, "items", log_items);
228304

229-
sentry_envelope_t *envelope
230-
= sentry__envelope_new_with_dsn(batcher->dsn);
231-
batcher->batch_func(envelope, logs);
232-
233-
if (crash_safe) {
234-
// Write directly to disk to avoid transport queuing during
235-
// crash
236-
sentry__run_write_envelope(batcher->run, envelope);
237-
sentry_envelope_free(envelope);
238-
} else if (!sentry__run_should_skip_upload(batcher->run)) {
239-
// Normal operation: use transport for HTTP transmission
240-
sentry__transport_send_envelope(batcher->transport, envelope);
241-
} else {
242-
sentry_envelope_free(envelope);
243-
}
244-
sentry_value_decref(logs);
305+
process_batch(batcher, logs, crash_safe);
245306
}
246307

247308
// Reset the drained buffer...
@@ -433,12 +494,14 @@ sentry__batcher_shutdown(sentry_batcher_t *batcher, uint64_t timeout)
433494

434495
// Perform final flush to ensure any remaining items are sent
435496
sentry__batcher_flush(batcher, false);
497+
sentry__threadpool_flush(batcher->threadpool);
436498
}
437499

438500
void
439501
sentry__batcher_flush_crash_safe(sentry_batcher_t *batcher)
440502
{
441503
// Check if batcher is initialized
504+
sentry__atomic_store(&batcher->crash_flush, 1);
442505
const long state = sentry__atomic_fetch(&batcher->thread_state);
443506
if (state == SENTRY_BATCHER_THREAD_STOPPED) {
444507
return;
@@ -471,6 +534,7 @@ sentry__batcher_force_flush_wait(sentry_batcher_t *batcher)
471534
}
472535
// retry if the batcher thread (woken by _begin) wins the race
473536
} while (!sentry__batcher_flush(batcher, false));
537+
sentry__threadpool_flush(batcher->threadpool);
474538
}
475539

476540
#ifdef SENTRY_UNITTEST
Collapse file

‎src/sentry_batcher.h‎

Copy file name to clipboardExpand all lines: src/sentry_batcher.h
+4-2Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -45,9 +45,11 @@ typedef struct {
4545
long active_idx; // (atomic) index to the active buffer
4646
long drain_idx; // (atomic) index to the oldest buffer to drain
4747
long flushing; // (atomic) reentrancy guard to the flusher
48+
long crash_flush; // (atomic) write completed batch work to disk
4849
long thread_state; // (atomic) sentry_batcher_thread_state_t
4950
sentry_waitable_flag_t request_flush; // level-triggered flush flag
5051
sentry_threadid_t batching_thread; // the batching thread
52+
sentry_threadpool_t *threadpool; // thread pool for batch work
5153
sentry_batch_func_t batch_func; // function to add items to envelope
5254
sentry_data_category_t data_category; // for client report discard tracking
5355
sentry_dsn_t *dsn;
@@ -62,8 +64,8 @@ typedef struct {
6264

6365
#define SENTRY_BATCHER_REF_INIT { NULL, 0 }
6466

65-
sentry_batcher_t *sentry__batcher_new(
66-
sentry_batch_func_t batch_func, sentry_data_category_t data_category);
67+
sentry_batcher_t *sentry__batcher_new(sentry_batch_func_t batch_func,
68+
sentry_data_category_t data_category, sentry_threadpool_t *threadpool);
6769

6870
/**
6971
* Acquires a reference to the batcher behind `ref`, atomically incrementing
Collapse file

‎src/sentry_logs.c‎

Copy file name to clipboardExpand all lines: src/sentry_logs.c
+3-2Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -627,10 +627,11 @@ sentry_scope_capture_log(sentry_scope_t *scope, sentry_level_t level,
627627
}
628628

629629
void
630-
sentry__logs_startup(const sentry_options_t *options)
630+
sentry__logs_startup(
631+
const sentry_options_t *options, sentry_threadpool_t *threadpool)
631632
{
632633
sentry_batcher_t *batcher = sentry__batcher_new(
633-
sentry__envelope_add_logs, SENTRY_DATA_CATEGORY_LOG_ITEM);
634+
sentry__envelope_add_logs, SENTRY_DATA_CATEGORY_LOG_ITEM, threadpool);
634635
if (!batcher) {
635636
SENTRY_WARN("failed to allocate logs batcher");
636637
return;
Collapse file

‎src/sentry_logs.h‎

Copy file name to clipboardExpand all lines: src/sentry_logs.h
+3-1Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,14 +2,16 @@
22
#define SENTRY_LOGS_H_INCLUDED
33

44
#include "sentry_boot.h"
5+
#include "sentry_sync.h"
56

67
log_return_value_t sentry__logs_log(
78
sentry_level_t level, const char *message, va_list args);
89

910
/**
1011
* Sets up the logs timer/flush thread
1112
*/
12-
void sentry__logs_startup(const sentry_options_t *options);
13+
void sentry__logs_startup(
14+
const sentry_options_t *options, sentry_threadpool_t *threadpool);
1315

1416
/**
1517
* Shuts down the logs timer/flush thread.
Collapse file

‎src/sentry_metrics.c‎

Copy file name to clipboardExpand all lines: src/sentry_metrics.c
+5-3Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -131,10 +131,12 @@ sentry_metrics_distribution(
131131
}
132132

133133
void
134-
sentry__metrics_startup(const sentry_options_t *options)
134+
sentry__metrics_startup(
135+
const sentry_options_t *options, sentry_threadpool_t *threadpool)
135136
{
136-
sentry_batcher_t *batcher = sentry__batcher_new(
137-
sentry__envelope_add_metrics, SENTRY_DATA_CATEGORY_TRACE_METRIC);
137+
sentry_batcher_t *batcher
138+
= sentry__batcher_new(sentry__envelope_add_metrics,
139+
SENTRY_DATA_CATEGORY_TRACE_METRIC, threadpool);
138140
if (!batcher) {
139141
SENTRY_WARN("failed to allocate metrics batcher");
140142
return;
Collapse file

‎src/sentry_metrics.h‎

Copy file name to clipboardExpand all lines: src/sentry_metrics.h
+3-1Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,13 @@
22
#define SENTRY_METRICS_H_INCLUDED
33

44
#include "sentry_boot.h"
5+
#include "sentry_sync.h"
56

67
/**
78
* Sets up the metrics timer/flush thread
89
*/
9-
void sentry__metrics_startup(const sentry_options_t *options);
10+
void sentry__metrics_startup(
11+
const sentry_options_t *options, sentry_threadpool_t *threadpool);
1012

1113
/**
1214
* Shuts down the metrics timer/flush thread.
Collapse file

‎src/sentry_telemetry.c‎

Copy file name to clipboardExpand all lines: src/sentry_telemetry.c
+28-2Lines changed: 28 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,38 @@
11
#include "sentry_telemetry.h"
2+
#include "sentry_batcher.h"
3+
#include "sentry_logger.h"
24
#include "sentry_logs.h"
35
#include "sentry_metrics.h"
46
#include "sentry_options.h"
7+
#include "sentry_sync.h"
8+
9+
static sentry_threadpool_t *g_telemetry_pool = NULL;
510

611
void
712
sentry__telemetry_startup(const sentry_options_t *options)
813
{
14+
if (!options->enable_logs && !options->enable_metrics) {
15+
return;
16+
}
17+
18+
// use two workers for serializing telemetry batches off the batcher threads
19+
g_telemetry_pool = sentry__threadpool_new(2);
20+
if (!g_telemetry_pool) {
21+
SENTRY_WARN("failed to allocate telemetry serialization pool");
22+
return;
23+
}
24+
if (sentry__threadpool_start(g_telemetry_pool) != 0) {
25+
SENTRY_WARN("failed to start telemetry serialization pool");
26+
sentry__threadpool_free(g_telemetry_pool);
27+
g_telemetry_pool = NULL;
28+
return;
29+
}
30+
931
if (options->enable_logs) {
10-
sentry__logs_startup(options);
32+
sentry__logs_startup(options, g_telemetry_pool);
1133
}
1234
if (options->enable_metrics) {
13-
sentry__metrics_startup(options);
35+
sentry__metrics_startup(options, g_telemetry_pool);
1436
}
1537
}
1638

@@ -19,6 +41,9 @@ sentry__telemetry_shutdown(uint64_t timeout)
1941
{
2042
sentry__logs_shutdown(timeout);
2143
sentry__metrics_shutdown(timeout);
44+
sentry__threadpool_shutdown(g_telemetry_pool);
45+
sentry__threadpool_free(g_telemetry_pool);
46+
g_telemetry_pool = NULL;
2247
}
2348

2449
void
@@ -35,4 +60,5 @@ sentry__telemetry_flush_crash_safe(void)
3560
{
3661
sentry__logs_flush_crash_safe();
3762
sentry__metrics_flush_crash_safe();
63+
sentry__threadpool_flush(g_telemetry_pool);
3864
}
Collapse file

‎tests/unit/test_client_report.c‎

Copy file name to clipboardExpand all lines: tests/unit/test_client_report.c
+2-2Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -390,8 +390,8 @@ SENTRY_TEST(client_report_queue_overflow)
390390
SENTRY_TEST_OPTIONS_NEW(options);
391391
sentry_init(options);
392392

393-
sentry_batcher_t *batcher
394-
= sentry__batcher_new(dummy_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM);
393+
sentry_batcher_t *batcher = sentry__batcher_new(
394+
dummy_batch_func, SENTRY_DATA_CATEGORY_LOG_ITEM, NULL);
395395
TEST_CHECK(!!batcher);
396396

397397
// Fill all buffers (SENTRY_BATCHER_QUEUE_LENGTH is 5 in unit tests)

0 commit comments

Comments
0 (0)
Morty Proxy This is a proxified and sanitized view of the page, visit original site.