diff --git a/src/ssb.c b/src/ssb.c index e520cf268..29c4a91d2 100644 --- a/src/ssb.c +++ b/src/ssb.c @@ -261,6 +261,7 @@ typedef struct _tf_ssb_connection_message_request_t typedef struct _tf_ssb_connection_scheduled_t { + char key[k_id_base64_len]; tf_ssb_scheduled_callback_t* callback; void* user_data; } tf_ssb_connection_scheduled_t; @@ -595,24 +596,55 @@ static void _tf_ssb_connection_dispatch_scheduled(tf_ssb_connection_t* connectio { while (((connection->active_write_count == 0 && connection->read_back_pressure == 0) || connection->closing) && connection->scheduled_count && connection->scheduled) { - tf_ssb_connection_scheduled_t scheduled = connection->scheduled[0]; - memmove(connection->scheduled, connection->scheduled + 1, sizeof(tf_ssb_connection_scheduled_t) * (connection->scheduled_count - 1)); + int index = uv_hrtime() % connection->scheduled_count; + tf_ssb_connection_scheduled_t scheduled = connection->scheduled[index]; + if (index != connection->scheduled_count - 1) + { + memmove(connection->scheduled + index, connection->scheduled + index + 1, sizeof(tf_ssb_connection_scheduled_t) * (connection->scheduled_count - index - 1)); + } connection->scheduled_count--; tf_trace_begin(connection->ssb->trace, "scheduled callback"); PRE_CALLBACK(connection->ssb, scheduled.callback); - scheduled.callback(connection, scheduled.user_data); + scheduled.callback(connection, false, scheduled.user_data); POST_CALLBACK(connection->ssb, scheduled.callback); tf_trace_end(connection->ssb->trace); } } -void tf_ssb_connection_schedule_idle(tf_ssb_connection_t* connection, tf_ssb_scheduled_callback_t* callback, void* user_data) +static int _tf_ssb_connection_scheduled_compare(const void* a, const void* b) { - connection->scheduled = tf_resize_vec(connection->scheduled, sizeof(tf_ssb_connection_scheduled_t) * (connection->scheduled_count + 1)); - connection->scheduled[connection->scheduled_count++] = (tf_ssb_connection_scheduled_t) { - .callback = callback, - .user_data = user_data, - }; + const char* key = a; + const tf_ssb_connection_scheduled_t* scheduled = b; + return strcmp(key, scheduled->key); +} + +void tf_ssb_connection_schedule_idle(tf_ssb_connection_t* connection, const char* key, tf_ssb_scheduled_callback_t* callback, void* user_data) +{ + int index = tf_util_insert_index(key, connection->scheduled, connection->scheduled_count, sizeof(tf_ssb_connection_scheduled_t), _tf_ssb_connection_scheduled_compare); + if (index != connection->scheduled_count && strcmp(key, connection->scheduled[index].key) == 0) + { + tf_ssb_connection_scheduled_t scheduled = connection->scheduled[index]; + connection->scheduled[index].callback = callback; + connection->scheduled[index].user_data = user_data; + + tf_trace_begin(connection->ssb->trace, "scheduled callback (skip)"); + PRE_CALLBACK(connection->ssb, scheduled.callback); + scheduled.callback(connection, true, scheduled.user_data); + POST_CALLBACK(connection->ssb, scheduled.callback); + tf_trace_end(connection->ssb->trace); + } + else + { + connection->scheduled = tf_resize_vec(connection->scheduled, sizeof(tf_ssb_connection_scheduled_t) * (connection->scheduled_count + 1)); + memmove(connection->scheduled + index + 1, connection->scheduled + index, sizeof(tf_ssb_connection_scheduled_t) * (connection->scheduled_count - index)); + connection->scheduled[index] = (tf_ssb_connection_scheduled_t) { + .callback = callback, + .user_data = user_data, + }; + snprintf(connection->scheduled[index].key, sizeof(connection->scheduled[index].key), "%s", key); + connection->scheduled_count++; + } + uv_async_send(&connection->scheduled_async); } diff --git a/src/ssb.h b/src/ssb.h index 2e780441c..0c45dbfb0 100644 --- a/src/ssb.h +++ b/src/ssb.h @@ -799,17 +799,19 @@ JSValue tf_ssb_connection_requests_to_object(tf_ssb_connection_t* connection); /** ** A function scheduled to be run later. ** @param connection The owning connection. +** @param skip Whether the work ought to be skipped, because it is being replaced. ** @param user_data User data registered with the callback. */ -typedef void(tf_ssb_scheduled_callback_t)(tf_ssb_connection_t* connection, void* user_data); +typedef void(tf_ssb_scheduled_callback_t)(tf_ssb_connection_t* connection, bool skip, void* user_data); /** ** Schedule work to be run when the connection is next idle. ** @param connection The owning connection. +** @param key A key identifying the work. If work by the same key already exists, it will be replaced. ** @param callback The callback to call. ** @param user_data User data to pass to the callback. */ -void tf_ssb_connection_schedule_idle(tf_ssb_connection_t* connection, tf_ssb_scheduled_callback_t* callback, void* user_data); +void tf_ssb_connection_schedule_idle(tf_ssb_connection_t* connection, const char* key, tf_ssb_scheduled_callback_t* callback, void* user_data); /** ** Schedule work to run on a worker thread. diff --git a/src/ssb.rpc.c b/src/ssb.rpc.c index a3a56d74e..f9e76e638 100644 --- a/src/ssb.rpc.c +++ b/src/ssb.rpc.c @@ -855,6 +855,16 @@ static void _tf_ssb_connection_send_history_stream_work(tf_ssb_connection_t* con request->out_finished = request->out_max_sequence_seen != request->sequence + k_max - 1; } +static void _tf_ssb_connection_send_history_stream_destroy(tf_ssb_connection_send_history_stream_t* request) +{ + for (int i = 0; i < request->out_messages_count; i++) + { + tf_free(request->out_messages[i]); + } + tf_free(request->out_messages); + tf_free(request); +} + 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; @@ -879,24 +889,19 @@ static void _tf_ssb_connection_send_history_stream_after_work(tf_ssb_connection_ tf_ssb_connection_rpc_send(connection, k_ssb_rpc_flag_json, request->request_number, NULL, (const uint8_t*)"false", strlen("false"), NULL, NULL, NULL); } } - for (int i = 0; i < request->out_messages_count; i++) - { - tf_free(request->out_messages[i]); - } - tf_free(request->out_messages); - tf_free(request); + _tf_ssb_connection_send_history_stream_destroy(request); } -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_callback(tf_ssb_connection_t* connection, bool skip, void* user_data) { tf_ssb_connection_adjust_write_count(connection, 1); - if (tf_ssb_connection_is_connected(connection) && !tf_ssb_is_shutting_down(tf_ssb_connection_get_ssb(connection))) + if (!skip && tf_ssb_connection_is_connected(connection) && !tf_ssb_is_shutting_down(tf_ssb_connection_get_ssb(connection))) { tf_ssb_connection_run_work(connection, _tf_ssb_connection_send_history_stream_work, _tf_ssb_connection_send_history_stream_after_work, user_data); } else { - _tf_ssb_connection_send_history_stream_after_work(connection, -1, user_data); + _tf_ssb_connection_send_history_stream_destroy(user_data); } } @@ -914,7 +919,7 @@ static void _tf_ssb_connection_send_history_stream( .end_request = end_request, }; 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_schedule_idle(connection, author, _tf_ssb_connection_send_history_stream_callback, async); } } @@ -1166,11 +1171,14 @@ typedef struct _resend_clock_t int32_t request_number; } resend_clock_t; -static void _tf_ssb_rpc_ebt_replicate_resend_clock(tf_ssb_connection_t* connection, void* user_data) +static void _tf_ssb_rpc_ebt_replicate_resend_clock(tf_ssb_connection_t* connection, bool skip, void* user_data) { resend_clock_t* resend = user_data; - _tf_ssb_rpc_ebt_replicate_send_clock(resend->connection, resend->request_number, JS_UNDEFINED); - tf_ssb_connection_set_sent_clock(resend->connection, true); + if (!skip) + { + _tf_ssb_rpc_ebt_replicate_send_clock(resend->connection, resend->request_number, JS_UNDEFINED); + tf_ssb_connection_set_sent_clock(resend->connection, true); + } tf_free(user_data); } @@ -1208,7 +1216,7 @@ static void _tf_ssb_rpc_ebt_replicate(tf_ssb_connection_t* connection, uint8_t f .connection = connection, .request_number = request_number, }; - tf_ssb_connection_schedule_idle(connection, _tf_ssb_rpc_ebt_replicate_resend_clock, resend); + tf_ssb_connection_schedule_idle(connection, "ebt.clock", _tf_ssb_rpc_ebt_replicate_resend_clock, resend); } } else