diff --git a/dump_pcm_data/src/dump_writer.c b/dump_pcm_data/src/dump_writer.c index fdbe0a5..bdafde6 100644 --- a/dump_pcm_data/src/dump_writer.c +++ b/dump_pcm_data/src/dump_writer.c @@ -86,25 +86,21 @@ void *writer_thread_func(void *arg) { session = session_find_by_generation(&g_ctx.sessions, block.key1, block.key2, block.generation); if (session != NULL) { bool is_new = false; + bool tracked = session->fp != NULL && file_mgr_is_tracking(&g_ctx.file_mgr, session->fp); + session_file_action_t action = session_file_action(session, tracked, get_file_size_limit_bytes()); - /* The file manager closes the stalest handle when MAX_OPEN_FILES is - * reached, without telling the owning session. Reopen instead of - * writing through a closed FILE*. */ - if (session->fp != NULL && !file_mgr_is_tracking(&g_ctx.file_mgr, session->fp)) { - session->fp = NULL; - } - - if (session->fp != NULL && session->file_size >= get_file_size_limit_bytes()) { - if (file_mgr_is_tracking(&g_ctx.file_mgr, session->fp)) { + if (action == SESSION_FILE_ROLLOVER) { + if (tracked) { file_mgr_close(&g_ctx.file_mgr, session->fp); } session->split_index++; session->fp = NULL; session->file_path[0] = '\0'; session->file_size = 0; + action = SESSION_FILE_REOPEN; } - if (session->fp == NULL) { + if (action == SESSION_FILE_REOPEN) { session->fp = file_mgr_get_handle(&g_ctx.file_mgr, session->key1, session->key2, session->generation, session->split_index, session->file_path, diff --git a/dump_pcm_data/src/session.c b/dump_pcm_data/src/session.c index 02d6473..9ac7c86 100644 --- a/dump_pcm_data/src/session.c +++ b/dump_pcm_data/src/session.c @@ -185,3 +185,18 @@ uint32_t session_live_count(session_manager_t *mgr) { return count; } + +session_file_action_t session_file_action(const session_info_t *session, bool handle_tracked, + uint64_t size_limit) { + if (session == NULL || session->fp == NULL) { + return SESSION_FILE_REOPEN; + } + if (!handle_tracked) { + return SESSION_FILE_REOPEN; + } + if (size_limit > 0 && session->file_size >= size_limit) { + return SESSION_FILE_ROLLOVER; + } + + return SESSION_FILE_USE; +} diff --git a/dump_pcm_data/src/session.h b/dump_pcm_data/src/session.h index 516865c..ab2435a 100644 --- a/dump_pcm_data/src/session.h +++ b/dump_pcm_data/src/session.h @@ -16,6 +16,14 @@ typedef enum { SESSION_STOPPING = 1 } session_state_t; +/* What the writer should do with a session's file handle for the next block. + * Pure decision so it can be tested without threads or timing. */ +typedef enum { + SESSION_FILE_USE = 0, + SESSION_FILE_REOPEN = 1, + SESSION_FILE_ROLLOVER = 2 +} session_file_action_t; + typedef struct { char key1[KEY1_MAX_LEN + 1]; char key2[KEY2_MAX_LEN + 1]; @@ -48,4 +56,10 @@ session_info_t *session_get_or_create(session_manager_t *mgr, const char *key1, uint32_t session_count(session_manager_t *mgr); uint32_t session_live_count(session_manager_t *mgr); +/* handle_tracked reports whether the file manager still owns session->fp. + * A handle can be closed behind the session's back when MAX_OPEN_FILES is + * reached, and writing through it would touch a closed FILE*. */ +session_file_action_t session_file_action(const session_info_t *session, bool handle_tracked, + uint64_t size_limit); + #endif diff --git a/dump_pcm_data/test/main.c b/dump_pcm_data/test/main.c index 71b3fbc..5f7c57a 100644 --- a/dump_pcm_data/test/main.c +++ b/dump_pcm_data/test/main.c @@ -1175,6 +1175,39 @@ static void test_session_highest_generation_lookup(void) { pass(name); } +static void test_session_file_action_policy(void) { + const char *name = "session file action policy"; + session_info_t session = {0}; + + if (session_file_action(NULL, false, 1024) != SESSION_FILE_REOPEN || + session_file_action(&session, false, 1024) != SESSION_FILE_REOPEN) { + fail(name, "missing handle must reopen"); + return; + } + + session.fp = stdout; + if (session_file_action(&session, false, 1024) != SESSION_FILE_REOPEN) { + fail(name, "untracked handle must reopen"); + return; + } + if (session_file_action(&session, true, 1024) != SESSION_FILE_USE) { + fail(name, "tracked handle below the limit must be reused"); + return; + } + + session.file_size = 1024; + if (session_file_action(&session, true, 1024) != SESSION_FILE_ROLLOVER) { + fail(name, "size limit must roll over"); + return; + } + if (session_file_action(&session, false, 1024) != SESSION_FILE_REOPEN) { + fail(name, "evicted handle wins over the size limit"); + return; + } + + pass(name); +} + static void test_queue_allocation_failures(void) { const char *name = "queue allocation failures"; queue_t queue; @@ -2667,6 +2700,7 @@ int main(void) { test_queue_empty_pop_and_byte_limit(); test_session_direct(); test_session_highest_generation_lookup(); + test_session_file_action_policy(); test_queue_allocation_failures(); test_session_allocation_failures(); test_public_write_allocation_failures();