cory/tildefriends
5
6
Fork 2
Code Issues 39 Pull Requests Actions Releases 34 Wiki Activity

An experiment in controlling memory usage when syncing. uv_read_stop when we have too active message/blob writes to the database and uv_read_start when we're back under control. #64

Browse Source
This commit is contained in:
Cory McWilliams
2024-05-22 19:53:33 -04:00
parent 1be94ae0be
commit d974a5e044
3 changed files with 72 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/ssb.rpc.c
View File

@ -404,6 +404,7 @@ typedef struct _blobs_get_t
bool done;
bool storing;
tf_ssb_t* ssb;
tf_ssb_connection_t* connection;
uint8_t buffer[];
} blobs_get_t;
@ -411,6 +412,7 @@ static void _tf_ssb_rpc_blob_store_callback(const char* id, bool is_new, void* u
{
blobs_get_t* get = user_data;
get->storing = false;
tf_ssb_connection_adjust_read_backpressure(get->connection, -1);
if (get->done)
{
tf_free(get);
@ -433,6 +435,7 @@ static void _tf_ssb_rpc_connection_blobs_get_callback(
if (JS_ToBool(context, args))
{
get->storing = true;
tf_ssb_connection_adjust_read_backpressure(connection, 1);
tf_ssb_db_blob_store_async(ssb, get->buffer, get->received, _tf_ssb_rpc_blob_store_callback, get);
}
/* TODO: Should we send the response in the callback? */
@ -455,7 +458,7 @@ static void _tf_ssb_rpc_connection_blobs_get_cleanup(tf_ssb_t* ssb, void* user_d
static void _tf_ssb_rpc_connection_blobs_get(tf_ssb_connection_t* connection, const char* blob_id, size_t size)
{
blobs_get_t* get = tf_malloc(sizeof(blobs_get_t) + size);
*get = (blobs_get_t) { .ssb = tf_ssb_connection_get_ssb(connection), .expected_size = size };
*get = (blobs_get_t) { .ssb = tf_ssb_connection_get_ssb(connection), .connection = connection, .expected_size = size };
snprintf(get->id, sizeof(get->id), "%s", blob_id);
memset(get->buffer, 0, size);
@ -1000,6 +1003,12 @@ static void _tf_ssb_rpc_ebt_replicate_send_messages(tf_ssb_connection_t* connect
}
}
static void _tf_ssb_rpc_ebt_replicate_store_callback(const char* id, bool verified, bool is_new, void* user_data)
{
tf_ssb_connection_t* connection = user_data;
tf_ssb_connection_adjust_read_backpressure(connection, -1);
}
static void _tf_ssb_rpc_ebt_replicate(tf_ssb_connection_t* connection, uint8_t flags, int32_t request_number, JSValue args, const uint8_t* message, size_t size, void* user_data)
{
tf_ssb_t* ssb = tf_ssb_connection_get_ssb(connection);
@ -1022,7 +1031,8 @@ static void _tf_ssb_rpc_ebt_replicate(tf_ssb_connection_t* connection, uint8_t f
if (!JS_IsUndefined(author))
{
/* Looks like a message. */
tf_ssb_verify_strip_and_store_message(ssb, args, NULL, NULL);
tf_ssb_connection_adjust_read_backpressure(connection, 1);
tf_ssb_verify_strip_and_store_message(ssb, args, _tf_ssb_rpc_ebt_replicate_store_callback, connection);
}
else
{