From 62cdc592c090616bf9ab1c513d7350295d7c73d8 Mon Sep 17 00:00:00 2001 From: Cory McWilliams Date: Sun, 8 Oct 2023 15:14:42 +0000 Subject: [PATCH] Move sending history streams to the worker threads. git-svn-id: https://www.unprompted.com/svn/projects/tildefriends/trunk@4499 ed5197a5-7fde-0310-b194-c3ffbd925b24 --- src/ssb.c | 69 +++++++++++++++++++++++++++++++++++++++++++++++- src/ssb.h | 5 ++++ src/ssb.rpc.c | 72 +++++++++++++++++++++++++++++++++++---------------- 3 files changed, 123 insertions(+), 23 deletions(-) diff --git a/src/ssb.c b/src/ssb.c index 7d7c79fe..fa432f64 100644 --- a/src/ssb.c +++ b/src/ssb.c @@ -257,6 +257,7 @@ typedef struct _tf_ssb_connection_t uv_tcp_t tcp; uv_connect_t connect; uv_async_t async; + bool closing; tf_ssb_connection_t* tunnel_connection; int32_t tunnel_request_number; @@ -318,6 +319,7 @@ typedef struct _tf_ssb_connection_t int scheduled_count; tf_ssb_debug_message_t* debug_messages[k_debug_close_message_count]; + int ref_count; } tf_ssb_connection_t; static JSClassID _connection_class_id; @@ -1767,6 +1769,7 @@ static void _tf_ssb_connection_dispatch_scheduled(tf_ssb_connection_t* connectio static void _tf_ssb_connection_destroy(tf_ssb_connection_t* connection, const char* reason) { tf_ssb_t* ssb = connection->ssb; + connection->closing = true; if (!connection->destroy_reason) { connection->destroy_reason = reason; @@ -1851,7 +1854,8 @@ static void _tf_ssb_connection_destroy(tf_ssb_connection_t* connection, const ch if (JS_IsUndefined(connection->object) && !connection->async.data && !connection->tcp.data && - !connection->connect.data) + !connection->connect.data && + connection->ref_count == 0) { JS_FreeValue(ssb->context, connection->ebt_send_clock); connection->ebt_send_clock = JS_UNDEFINED; @@ -3646,3 +3650,66 @@ void tf_ssb_set_main_thread(tf_ssb_t* ssb) { ssb->thread_self = uv_thread_self(); } + +typedef struct _connection_work_t +{ + uv_work_t work; + tf_ssb_connection_t* connection; + void (*work_callback)(tf_ssb_connection_t* connection, void* user_data); + void (*after_work_callback)(tf_ssb_connection_t* connection, int result, void* user_data); + void* user_data; +} connection_work_t; + +static void _tf_ssb_connection_work_callback(uv_work_t* work) +{ + connection_work_t* data = work->data; + tf_ssb_record_thread_busy(data->connection->ssb, true); + if (data->work_callback) + { + data->work_callback(data->connection, data->user_data); + } + tf_ssb_record_thread_busy(data->connection->ssb, false); +} + +static void _tf_ssb_connection_after_work_callback(uv_work_t* work, int status) +{ + connection_work_t* data = work->data; + if (data->after_work_callback) + { + data->after_work_callback(data->connection, status, data->user_data); + } + data->connection->ref_count--; + if (data->connection->ref_count == 0 && + data->connection->closing) + { + _tf_ssb_connection_destroy(data->connection, "work completed"); + } + tf_free(data); +} + +void tf_ssb_connection_run_work( + tf_ssb_connection_t* connection, + void (*work_callback)(tf_ssb_connection_t* connection, void* user_data), + void (*after_work_callback)(tf_ssb_connection_t* connection, int result, void* user_data), + void* user_data) +{ + connection_work_t* work = tf_malloc(sizeof(connection_work_t)); + *work = (connection_work_t) + { + .work = + { + .data = work, + }, + .connection = connection, + .work_callback = work_callback, + .after_work_callback = after_work_callback, + .user_data = user_data, + }; + + connection->ref_count++; + int result = uv_queue_work(connection->ssb->loop, &work->work, _tf_ssb_connection_work_callback, _tf_ssb_connection_after_work_callback); + if (result) + { + _tf_ssb_connection_after_work_callback(&work->work, result); + } +} diff --git a/src/ssb.h b/src/ssb.h index 87603ad2..72561ee1 100644 --- a/src/ssb.h +++ b/src/ssb.h @@ -173,6 +173,11 @@ void tf_ssb_connection_remove_request(tf_ssb_connection_t* connection, int32_t r typedef void (tf_ssb_scheduled_callback_t)(tf_ssb_connection_t* connection, void* user_data); void tf_ssb_connection_schedule_idle(tf_ssb_connection_t* connection, tf_ssb_scheduled_callback_t* callback, void* user_data); +void tf_ssb_connection_run_work( + tf_ssb_connection_t* connection, + void (*work_callback)(tf_ssb_connection_t* connection, void* user_data), + void (*after_work_callback)(tf_ssb_connection_t* connection, int result, void* user_data), + void* user_data); void tf_ssb_connection_add_new_message_request(tf_ssb_connection_t* connection, const char* author, int32_t request_number, bool keys); void tf_ssb_connection_remove_new_message_request(tf_ssb_connection_t* connection, const char* author); diff --git a/src/ssb.rpc.c b/src/ssb.rpc.c index 87c1575b..9428867e 100644 --- a/src/ssb.rpc.c +++ b/src/ssb.rpc.c @@ -19,7 +19,6 @@ #endif static void _tf_ssb_connection_send_history_stream(tf_ssb_connection_t* connection, int32_t request_number, const char* author, int64_t sequence, bool keys, bool live); -static void _tf_ssb_connection_send_history_stream_internal(tf_ssb_connection_t* connection, int32_t request_number, const char* author, int64_t sequence, bool keys, bool live); static void _tf_ssb_rpc_start_delete_blobs(tf_ssb_t* ssb, int delay_ms); static int64_t _get_global_setting_int64(tf_ssb_t* ssb, const char* name, int64_t default_value) @@ -765,36 +764,39 @@ typedef struct _tf_ssb_connection_send_history_stream_t int64_t sequence; bool keys; bool live; + + bool out_finished; + int64_t out_max_sequence_seen; + char** out_messages; + int out_messages_count; } tf_ssb_connection_send_history_stream_t; -static void _tf_ssb_connection_send_history_stream_callback(tf_ssb_connection_t* connection, void* user_data) +static void _tf_ssb_connection_send_history_stream_work(tf_ssb_connection_t* connection, void* user_data) { tf_ssb_connection_send_history_stream_t* request = user_data; - if (tf_ssb_connection_is_connected(connection)) + if (!tf_ssb_connection_is_connected(connection)) { - _tf_ssb_connection_send_history_stream_internal(connection, request->request_number, request->author, request->sequence, request->keys, request->live); + return; } - tf_free(request); -} - -static void _tf_ssb_connection_send_history_stream_internal(tf_ssb_connection_t* connection, int32_t request_number, const char* author, int64_t sequence, bool keys, bool live) -{ tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection); - JSContext* context = tf_ssb_get_context(ssb); sqlite3* db = tf_ssb_acquire_db_reader(ssb); sqlite3_stmt* statement; const int k_max = 32; - int64_t max_sequence_seen = 0; if (sqlite3_prepare(db, "SELECT previous, author, id, sequence, timestamp, hash, content, signature, sequence_before_author FROM messages WHERE author = ?1 AND sequence >= ?2 AND sequence < ?3 ORDER BY sequence", -1, &statement, NULL) == SQLITE_OK) { - if (sqlite3_bind_text(statement, 1, author, -1, NULL) == SQLITE_OK && - sqlite3_bind_int64(statement, 2, sequence) == SQLITE_OK && - sqlite3_bind_int64(statement, 3, sequence + k_max) == SQLITE_OK) + if (sqlite3_bind_text(statement, 1, request->author, -1, NULL) == SQLITE_OK && + sqlite3_bind_int64(statement, 2, request->sequence) == SQLITE_OK && + sqlite3_bind_int64(statement, 3, request->sequence + k_max) == SQLITE_OK) { + JSMallocFunctions funcs = { 0 }; + tf_get_js_malloc_functions(&funcs); + JSRuntime* runtime = JS_NewRuntime2(&funcs, NULL); + JSContext* context = JS_NewContext(runtime); + while (sqlite3_step(statement) == SQLITE_ROW) { JSValue message = JS_UNDEFINED; - max_sequence_seen = sqlite3_column_int64(statement, 3); + request->out_max_sequence_seen = sqlite3_column_int64(statement, 3); JSValue formatted = tf_ssb_format_message( context, @@ -806,7 +808,7 @@ static void _tf_ssb_connection_send_history_stream_internal(tf_ssb_connection_t* (const char*)sqlite3_column_text(statement, 6), (const char*)sqlite3_column_text(statement, 7), sqlite3_column_int(statement, 8)); - if (keys) + if (request->keys) { message = JS_NewObject(context); JS_SetPropertyStr(context, message, "key", JS_NewString(context, (const char*)sqlite3_column_text(statement, 2))); @@ -817,9 +819,22 @@ static void _tf_ssb_connection_send_history_stream_internal(tf_ssb_connection_t* { message = formatted; } - tf_ssb_connection_rpc_send_json(connection, k_ssb_rpc_flag_stream, request_number, message, NULL, NULL, NULL); + JSValue json = JS_JSONStringify(context, message, JS_NULL, JS_NULL); + size_t size = 0; + const char* string = JS_ToCStringLen(context, &size, json); + + request->out_messages = tf_resize_vec(request->out_messages, sizeof(char*) * (request->out_messages_count + 1)); + char* copy = tf_malloc(size + 1); + memcpy(copy, string, size + 1); + JS_FreeCString(context, string); + request->out_messages[request->out_messages_count++] = copy; + + JS_FreeValue(context, json); JS_FreeValue(context, message); } + + JS_FreeContext(context); + JS_FreeRuntime(runtime); } sqlite3_finalize(statement); } @@ -828,23 +843,36 @@ static void _tf_ssb_connection_send_history_stream_internal(tf_ssb_connection_t* tf_printf("prepare failed: %s\n", sqlite3_errmsg(db)); } tf_ssb_release_db_reader(ssb, db); + request->out_finished = request->out_max_sequence_seen != request->sequence + k_max - 1; +} - if (max_sequence_seen == sequence + k_max - 1) +static void _tf_ssb_connection_send_history_stream_after_work(tf_ssb_connection_t* connection, int result, void* user_data) +{ + tf_ssb_connection_send_history_stream_t* request = user_data; + for (int i = 0; i < request->out_messages_count; i++) { - _tf_ssb_connection_send_history_stream(connection, request_number, author, max_sequence_seen + 1, keys, live); + tf_ssb_connection_rpc_send(connection, k_ssb_rpc_flag_stream | k_ssb_rpc_flag_json, request->request_number, (const uint8_t*)request->out_messages[i], strlen(request->out_messages[i]), NULL, NULL, NULL); + tf_free(request->out_messages[i]); } - else if (!live) + tf_free(request->out_messages); + + if (!request->out_finished) + { + _tf_ssb_connection_send_history_stream(connection, request->request_number, request->author, request->out_max_sequence_seen + 1, request->keys, request->live); + } + else if (!request->live) { tf_ssb_connection_rpc_send( connection, k_ssb_rpc_flag_json, - request_number, + request->request_number, (const uint8_t*)"false", strlen("false"), NULL, NULL, NULL); } + tf_free(request); } static void _tf_ssb_connection_send_history_stream(tf_ssb_connection_t* connection, int32_t request_number, const char* author, int64_t sequence, bool keys, bool live) @@ -858,7 +886,7 @@ static void _tf_ssb_connection_send_history_stream(tf_ssb_connection_t* connecti .live = live, }; snprintf(async->author, sizeof(async->author), "%s", author); - tf_ssb_connection_schedule_idle(connection, _tf_ssb_connection_send_history_stream_callback, async); + tf_ssb_connection_run_work(connection, _tf_ssb_connection_send_history_stream_work, _tf_ssb_connection_send_history_stream_after_work, async); } static void _tf_ssb_rpc_createHistoryStream(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)