From 8bcce71aa3420d4786e04978a1327da5ca3daab0 Mon Sep 17 00:00:00 2001 From: Richard Fuchs Date: Mon, 27 Jun 2022 13:41:46 -0400 Subject: [PATCH] TT#182200 delay Redis entry deletion until after MySQL transaction The MySQL INSERT statements to move processed Redis acc records from Redis to the respective backup/trash MySQL tables are always issued within a MySQL transaction (med_handler via medmysql_batch_start), but the deletions from Redis were done immediately. Therefore if mediator were to abort within a processing loop, the MySQL transaction would be rolled back after the entries had already been deleted from Redis, therefore losing the acc entries. Solve this by using an internal queue for Redis entries to hold the lists of entries to be deleted until the MySQL transaction is commited. Change-Id: Ib41d0e2ca722c66f9e078ca31f7e5ca2b9d9fe2d (cherry picked from commit bbbe2c9a7bade2e42c7edd07e5a37c18bca4dc43) --- cdr.c | 8 ++++--- cdr.h | 2 +- mediator.c | 6 +++-- medredis.c | 66 +++++++++++++++++++++++++++++++++++++++++++++--------- medredis.h | 5 +++-- 5 files changed, 69 insertions(+), 18 deletions(-) diff --git a/cdr.c b/cdr.c index d0cfcc7..947b4ac 100644 --- a/cdr.c +++ b/cdr.c @@ -62,9 +62,11 @@ static void free_cdrs(cdr_entry_t **cdrs, uint64_t cdr_count) { } -int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_count, +int cdr_process_records(med_entry_t **records_p, uint64_t *count_p, uint64_t *ext_count, struct medmysql_batches *batches, int do_intermediate) { + med_entry_t *records = *records_p; + uint64_t count = *count_p; int ret = 0; uint8_t trash = 0; uint64_t i; @@ -162,7 +164,7 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun { if (has_redis) { - if(medredis_backup_entries(records, count) != 0) + if(medredis_backup_entries(records_p, count_p) != 0) goto error; } if (has_mysql) @@ -210,7 +212,7 @@ int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *ext_coun { if (has_redis) { - if(medredis_trash_entries(records, count) != 0) + if(medredis_trash_entries(records_p, count_p) != 0) goto error; } if (has_mysql) diff --git a/cdr.h b/cdr.h index ade2397..db65a7c 100644 --- a/cdr.h +++ b/cdr.h @@ -59,7 +59,7 @@ typedef struct { #undef F #undef FA -int cdr_process_records(med_entry_t *records, uint64_t count, uint64_t *cdr_count, struct medmysql_batches *, +int cdr_process_records(med_entry_t **records, uint64_t *count, uint64_t *cdr_count, struct medmysql_batches *, int do_intermediate); void cdr_fix_accids(med_entry_t *records, uint64_t count); int cdr_fill_record(cdr_entry_t *cdr); diff --git a/mediator.c b/mediator.c index 3ff146c..aee975f 100644 --- a/mediator.c +++ b/mediator.c @@ -370,7 +370,7 @@ int main(int argc, char **argv) continue; } - if(cdr_process_records(mysql_records, mysql_rec_count, &cdr_count, batches, do_intermediate) != 0) + if(cdr_process_records(&mysql_records, &mysql_rec_count, &cdr_count, batches, do_intermediate) != 0) goto out; if(mysql_rec_count > 0) @@ -436,7 +436,7 @@ int main(int argc, char **argv) L_DEBUG("process cdr with cid '%s' and %"PRIu64" records\n", redis_callids[i].value, redis_rec_count); if (redis_rec_count) { - if(cdr_process_records(redis_records, redis_rec_count, &cdr_count, batches, do_intermediate) != 0) { + if(cdr_process_records(&redis_records, &redis_rec_count, &cdr_count, batches, do_intermediate) != 0) { free(redis_records); goto out; } @@ -460,6 +460,8 @@ int main(int argc, char **argv) if (medmysql_batch_end(batches)) break; + if (medredis_batch_end()) + break; gettimeofday(&loop_tv_stop, NULL); loop_runtime = mediator_calc_runtime(&loop_tv_start, &loop_tv_stop); diff --git a/medredis.c b/medredis.c index a47382e..d2af9a4 100644 --- a/medredis.c +++ b/medredis.c @@ -44,15 +44,21 @@ typedef struct { uint64_t index; } medredis_cidlist_t; +typedef struct { + med_entry_t *records; + uint64_t count; +} medredis_records_list_t; + typedef struct { const char *name; + GQueue record_lists; } medredis_table_t; static medredis_con_t *con = NULL; static char medredis_srem_key_lua[41]; // sha-1 hex string -static medredis_table_t medredis_table_trash = { .name = "trash" }; -static medredis_table_t medredis_table_backup = { .name = "backup" }; +static medredis_table_t medredis_table_trash = { .name = "trash", .record_lists = G_QUEUE_INIT }; +static medredis_table_t medredis_table_backup = { .name = "backup", .record_lists = G_QUEUE_INIT }; /**********************************************************************/ static void medredis_free_reply(redisReply **reply) { @@ -829,16 +835,34 @@ err: } /**********************************************************************/ -static int medredis_cleanup_entries(med_entry_t *records, uint64_t count, medredis_table_t *table) { - char buffer[512]; - - if (medmysql_insert_records(records, count, table->name) != 0) { +static int medredis_cleanup_entries(med_entry_t **records, uint64_t *count, medredis_table_t *table) { + if (medmysql_insert_records(*records, *count, table->name) != 0) { L_CRITICAL("Failed to cleanup redis records\n"); goto err; } - for (uint64_t i = 0; i < count; ++i) { - med_entry_t *e = &(records[i]); + // take ownership of the contents of the `records` array and swap out contents with + // an empty array + medredis_records_list_t *list = g_slice_alloc(sizeof(*list)); + list->records = *records; + list->count = *count; + *count = 0; + *records = NULL; + + g_queue_push_tail(&table->record_lists, list); + + return 0; + +err: + return -1; +} + +/**********************************************************************/ +static int medredis_batch_end_records(medredis_records_list_t *list) { + char buffer[512]; + + for (uint64_t i = 0; i < list->count; ++i) { + med_entry_t *e = &(list->records[i]); if (!e->redis) continue; @@ -880,11 +904,33 @@ err: } /**********************************************************************/ -int medredis_trash_entries(med_entry_t *records, uint64_t count) { +static int medredis_batch_end_table(medredis_table_t *table) { + while (table->record_lists.length) { + medredis_records_list_t *list = g_queue_pop_head(&table->record_lists); + int ret = medredis_batch_end_records(list); + free(list->records); + g_slice_free1(sizeof(*list), list); + if (ret) + return -1; + } + return 0; +} + +/**********************************************************************/ +int medredis_batch_end(void) { + if (medredis_batch_end_table(&medredis_table_trash)) + return -1; + if (medredis_batch_end_table(&medredis_table_backup)) + return -1; + return 0; +} + +/**********************************************************************/ +int medredis_trash_entries(med_entry_t **records, uint64_t *count) { return medredis_cleanup_entries(records, count, &medredis_table_trash); } /**********************************************************************/ -int medredis_backup_entries(med_entry_t *records, uint64_t count) { +int medredis_backup_entries(med_entry_t **records, uint64_t *count) { return medredis_cleanup_entries(records, count, &medredis_table_backup); } diff --git a/medredis.h b/medredis.h index 3d39e44..9d2a485 100644 --- a/medredis.h +++ b/medredis.h @@ -10,7 +10,8 @@ int medredis_init(void); void medredis_cleanup(void); med_callid_t *medredis_fetch_callids(uint64_t *count); int medredis_fetch_records(med_callid_t *callid, med_entry_t **entries, uint64_t *count); -int medredis_trash_entries(med_entry_t *records, uint64_t count); -int medredis_backup_entries(med_entry_t *records, uint64_t count); +int medredis_trash_entries(med_entry_t **records, uint64_t *count); +int medredis_backup_entries(med_entry_t **records, uint64_t *count); +int medredis_batch_end(void); #endif /* _MED_REDIS_H */