From befe197ec65909ecbf83b2c9034fc2aeced091ad Mon Sep 17 00:00:00 2001 From: John Kerl Date: Mon, 15 Jun 2015 22:54:07 -0400 Subject: [PATCH] join iterate --- c/Makefile | 6 ++-- c/input/file_reader_mmap.c | 29 +++++++++++------- c/input/file_reader_mmap.h | 5 +++- c/input/lrec_reader.h | 4 +++ c/input/lrec_reader_mmap_csv.c | 3 ++ c/input/lrec_reader_mmap_dkvp.c | 2 ++ c/input/lrec_reader_mmap_nidx.c | 2 ++ c/input/lrec_reader_mmap_xtab.c | 2 ++ c/input/lrec_reader_stdio_csv.c | 3 ++ c/input/lrec_reader_stdio_dkvp.c | 3 ++ c/input/lrec_reader_stdio_nidx.c | 3 ++ c/input/lrec_reader_stdio_xtab.c | 3 ++ c/stream/stream.c | 51 +++++--------------------------- 13 files changed, 58 insertions(+), 58 deletions(-) diff --git a/c/Makefile b/c/Makefile index ecaab77df..48d110b9c 100644 --- a/c/Makefile +++ b/c/Makefile @@ -12,10 +12,10 @@ NON_DSL_SRCS = *.c cli/*.c lib/*.c containers/*.c stream/*.c input/*.c mapping/* PDSL_OBJS = ./dsls/put_dsl_parse.o ./dsls/put_dsl_lexer.o ./dsls/put_dsl_wrapper.o FDSL_OBJS = ./dsls/filter_dsl_parse.o ./dsls/filter_dsl_lexer.o ./dsls/filter_dsl_wrapper.o -TEST_LREC_SRCS = lib/mlrutil.c containers/lrec.c containers/header_keeper.c containers/sllv.c \ +TEST_LREC_SRCS = lib/mlrutil.c lib/mlr_globals.c containers/lrec.c containers/header_keeper.c containers/sllv.c \ containers/slls.c containers/lhmslv.c \ -input/file_reader_mmap.c \ -input/lrec_reader_mmap_csv.c input/lrec_reader_stdio_csv.c \ +input/file_reader_mmap.c input/file_reader_stdio.c \ +input/lrec_reader_mmap_csv.c input/lrec_reader_stdio_csv.c \ input/lrec_reader_mmap_dkvp.c input/lrec_reader_stdio_dkvp.c \ input/lrec_reader_mmap_nidx.c input/lrec_reader_stdio_nidx.c \ input/lrec_reader_mmap_xtab.c input/lrec_reader_stdio_xtab.c \ diff --git a/c/input/file_reader_mmap.c b/c/input/file_reader_mmap.c index fe968c71e..5bf9aebf9 100644 --- a/c/input/file_reader_mmap.c +++ b/c/input/file_reader_mmap.c @@ -5,34 +5,35 @@ #include #include #include +#include "lib/mlrutil.h" #include "file_reader_mmap.h" static char empty_buf[1] = { 0 }; -file_reader_mmap_state_t file_reader_mmap_open(char* file_name) { - file_reader_mmap_state_t state; - state.fd = open(file_name, O_RDONLY); - if (state.fd < 0) { +file_reader_mmap_state_t* file_reader_mmap_open(char* file_name) { + file_reader_mmap_state_t* pstate = mlr_malloc_or_die(sizeof(file_reader_mmap_state_t)); + pstate->fd = open(file_name, O_RDONLY); + if (pstate->fd < 0) { perror("open"); exit(1); } struct stat stat; - if (fstat(state.fd, &stat) < 0) { + if (fstat(pstate->fd, &stat) < 0) { perror("fstat"); exit(1); } if (stat.st_size == 0) { // mmap doesn't allow us to map zero-length files but zero-length files do exist. - state.sol = &empty_buf[0]; + pstate->sol = &empty_buf[0]; } else { - state.sol = mmap(NULL, (size_t)stat.st_size, PROT_READ|PROT_WRITE, MAP_FILE|MAP_PRIVATE, state.fd, (off_t)0); - if (state.sol == MAP_FAILED) { + pstate->sol = mmap(NULL, (size_t)stat.st_size, PROT_READ|PROT_WRITE, MAP_FILE|MAP_PRIVATE, pstate->fd, (off_t)0); + if (pstate->sol == MAP_FAILED) { perror("mmap"); exit(1); } } - state.eof = state.sol + stat.st_size; - return state; + pstate->eof = pstate->sol + stat.st_size; + return pstate; } void file_reader_mmap_close(file_reader_mmap_state_t* pstate) { @@ -40,4 +41,12 @@ void file_reader_mmap_close(file_reader_mmap_state_t* pstate) { perror("close"); exit(1); } + free(pstate); +} + +void* file_reader_mmap_vopen(char* file_name) { + return file_reader_mmap_open(file_name); +} +void file_reader_mmap_vclose(void* pvhandle) { + file_reader_mmap_close(pvhandle); } diff --git a/c/input/file_reader_mmap.h b/c/input/file_reader_mmap.h index 2370fddea..648064efe 100644 --- a/c/input/file_reader_mmap.h +++ b/c/input/file_reader_mmap.h @@ -7,7 +7,10 @@ typedef struct _file_reader_mmap_state_t { int fd; } file_reader_mmap_state_t; -file_reader_mmap_state_t file_reader_mmap_open(char* file_name); +file_reader_mmap_state_t* file_reader_mmap_open(char* file_name); void file_reader_mmap_close(file_reader_mmap_state_t* pstate); +void* file_reader_mmap_vopen(char* file_name); +void file_reader_mmap_vclose(void* pvhandle); + #endif // FILE_READER_MMAP_H diff --git a/c/input/lrec_reader.h b/c/input/lrec_reader.h index f07318420..f8d457cbb 100644 --- a/c/input/lrec_reader.h +++ b/c/input/lrec_reader.h @@ -7,12 +7,16 @@ #include "input/file_reader_mmap.h" // The void* pvhandle is either FILE* for stdio readers or file_reader_mmap_state_t* for mmap readers. +typedef void* lrec_reader_open_func_t(char* filename); +typedef void lrec_reader_close_func_t(void* pvhandle); typedef lrec_t* lrec_reader_process_func_t(void* pvhandle, void* pvstate, context_t* pctx); typedef void lrec_reader_sof_func_t(void* pvstate); typedef void lrec_reader_free_func_t(void* pvstate); typedef struct _lrec_reader_t { void* pvstate; + lrec_reader_open_func_t* popen_func; + lrec_reader_close_func_t* pclose_func; lrec_reader_process_func_t* pprocess_func; lrec_reader_sof_func_t* psof_func; lrec_reader_free_func_t* pfree_func; diff --git a/c/input/lrec_reader_mmap_csv.c b/c/input/lrec_reader_mmap_csv.c index 7032627e4..d22077629 100644 --- a/c/input/lrec_reader_mmap_csv.c +++ b/c/input/lrec_reader_mmap_csv.c @@ -209,6 +209,9 @@ lrec_reader_t* lrec_reader_mmap_csv_alloc(char irs, char ifs, int allow_repeat_i pstate->pheader_keepers = lhmslv_alloc(); plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_mmap_vopen; + plrec_reader->pclose_func = &file_reader_mmap_vclose; + plrec_reader->pprocess_func = &lrec_reader_mmap_csv_process; plrec_reader->pprocess_func = &lrec_reader_mmap_csv_process; plrec_reader->psof_func = &lrec_reader_mmap_csv_sof; plrec_reader->pfree_func = NULL; diff --git a/c/input/lrec_reader_mmap_dkvp.c b/c/input/lrec_reader_mmap_dkvp.c index 2720eb724..ed1de4e0e 100644 --- a/c/input/lrec_reader_mmap_dkvp.c +++ b/c/input/lrec_reader_mmap_dkvp.c @@ -35,6 +35,8 @@ lrec_reader_t* lrec_reader_mmap_dkvp_alloc(char irs, char ifs, char ips, int all pstate->allow_repeat_ifs = allow_repeat_ifs; plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_mmap_vopen; + plrec_reader->pclose_func = &file_reader_mmap_vclose; plrec_reader->pprocess_func = &lrec_reader_mmap_dkvp_process; plrec_reader->psof_func = &lrec_reader_mmap_dkvp_sof; plrec_reader->pfree_func = NULL; diff --git a/c/input/lrec_reader_mmap_nidx.c b/c/input/lrec_reader_mmap_nidx.c index c6159fc43..ebdd15fe1 100644 --- a/c/input/lrec_reader_mmap_nidx.c +++ b/c/input/lrec_reader_mmap_nidx.c @@ -32,6 +32,8 @@ lrec_reader_t* lrec_reader_mmap_nidx_alloc(char irs, char ifs, int allow_repeat_ pstate->allow_repeat_ifs = allow_repeat_ifs; plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_mmap_vopen; + plrec_reader->pclose_func = &file_reader_mmap_vclose; plrec_reader->pprocess_func = &lrec_reader_mmap_nidx_process; plrec_reader->psof_func = &lrec_reader_mmap_nidx_sof; plrec_reader->pfree_func = NULL; diff --git a/c/input/lrec_reader_mmap_xtab.c b/c/input/lrec_reader_mmap_xtab.c index 992d086fe..340e497b4 100644 --- a/c/input/lrec_reader_mmap_xtab.c +++ b/c/input/lrec_reader_mmap_xtab.c @@ -35,6 +35,8 @@ lrec_reader_t* lrec_reader_mmap_xtab_alloc(char irs, char ips, int allow_repeat_ pstate->allow_repeat_ips = TRUE; plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_mmap_vopen; + plrec_reader->pclose_func = &file_reader_mmap_vclose; plrec_reader->pprocess_func = &lrec_reader_mmap_xtab_process; plrec_reader->psof_func = &lrec_reader_mmap_xtab_sof; plrec_reader->pfree_func = NULL; diff --git a/c/input/lrec_reader_stdio_csv.c b/c/input/lrec_reader_stdio_csv.c index 3bd4b89cc..b507454db 100644 --- a/c/input/lrec_reader_stdio_csv.c +++ b/c/input/lrec_reader_stdio_csv.c @@ -3,6 +3,7 @@ #include "lib/mlrutil.h" #include "containers/slls.h" #include "containers/lhmslv.h" +#include "input/file_reader_stdio.h" #include "input/lrec_readers.h" // Idea of pheader_keepers: each header_keeper object retains the input-line backing @@ -135,6 +136,8 @@ lrec_reader_t* lrec_reader_stdio_csv_alloc(char irs, char ifs, int allow_repeat_ pstate->pheader_keepers = lhmslv_alloc(); plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_stdio_vopen; + plrec_reader->pclose_func = &file_reader_stdio_vclose; plrec_reader->pprocess_func = &lrec_reader_stdio_csv_process; plrec_reader->psof_func = &lrec_reader_stdio_sof; plrec_reader->pfree_func = &lrec_reader_stdio_csv_free; diff --git a/c/input/lrec_reader_stdio_dkvp.c b/c/input/lrec_reader_stdio_dkvp.c index ce6c6dc0c..2badfa323 100644 --- a/c/input/lrec_reader_stdio_dkvp.c +++ b/c/input/lrec_reader_stdio_dkvp.c @@ -1,6 +1,7 @@ #include #include #include "lib/mlrutil.h" +#include "input/file_reader_stdio.h" #include "input/lrec_readers.h" typedef struct _lrec_reader_stdio_dkvp_state_t { @@ -41,6 +42,8 @@ lrec_reader_t* lrec_reader_stdio_dkvp_alloc(char irs, char ifs, char ips, int al pstate->allow_repeat_ifs = allow_repeat_ifs; plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_stdio_vopen; + plrec_reader->pclose_func = &file_reader_stdio_vclose; plrec_reader->pprocess_func = &lrec_reader_stdio_dkvp_process; plrec_reader->psof_func = &lrec_reader_csv_sof; plrec_reader->pfree_func = &lrec_reader_stdio_dkvp_free; diff --git a/c/input/lrec_reader_stdio_nidx.c b/c/input/lrec_reader_stdio_nidx.c index 5d6b86296..ae99a8187 100644 --- a/c/input/lrec_reader_stdio_nidx.c +++ b/c/input/lrec_reader_stdio_nidx.c @@ -1,5 +1,6 @@ #include #include "lib/mlrutil.h" +#include "input/file_reader_stdio.h" #include "input/lrec_readers.h" typedef struct _lrec_reader_stdio_nidx_state_t { @@ -35,6 +36,8 @@ lrec_reader_t* lrec_reader_stdio_nidx_alloc(char irs, char ifs, int allow_repeat pstate->allow_repeat_ifs = allow_repeat_ifs; plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_stdio_vopen; + plrec_reader->pclose_func = &file_reader_stdio_vclose; plrec_reader->pprocess_func = &lrec_reader_stdio_nidx_process; plrec_reader->psof_func = &lrec_reader_stdio_nidx_sof; plrec_reader->pfree_func = &lrec_reader_stdio_nidx_free; diff --git a/c/input/lrec_reader_stdio_xtab.c b/c/input/lrec_reader_stdio_xtab.c index 373998903..2f94a852c 100644 --- a/c/input/lrec_reader_stdio_xtab.c +++ b/c/input/lrec_reader_stdio_xtab.c @@ -1,6 +1,7 @@ #include #include #include "lib/mlrutil.h" +#include "input/file_reader_stdio.h" #include "input/lrec_readers.h" typedef struct _lrec_reader_stdio_xtab_state_t { @@ -59,6 +60,8 @@ lrec_reader_t* lrec_reader_stdio_xtab_alloc(char ips, int allow_repeat_ips) { pstate->at_eof = FALSE; plrec_reader->pvstate = (void*)pstate; + plrec_reader->popen_func = &file_reader_stdio_vopen; + plrec_reader->pclose_func = &file_reader_stdio_vclose; plrec_reader->pprocess_func = &lrec_reader_stdio_xtab_process; plrec_reader->psof_func = &lrec_reader_stdio_xtab_sof; plrec_reader->pfree_func = &lrec_reader_stdio_xtab_free; diff --git a/c/stream/stream.c b/c/stream/stream.c index 3ccbaeca7..abf849a96 100644 --- a/c/stream/stream.c +++ b/c/stream/stream.c @@ -12,8 +12,6 @@ static int do_file_chained(char* filename, context_t* pctx, lrec_reader_t* plrec_reader_stdio, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream); -static int do_file_chained_mmap(char* filename, context_t* pctx, - lrec_reader_t* plrec_reader_stdio, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream); static sllv_t* chain_map(lrec_t* pinrec, context_t* pctx, sllve_t* pmapper_list_head); @@ -39,14 +37,10 @@ int do_stream_chained(char** filenames, int use_file_reader_mmap, lrec_reader_t* ctx.filenum++; ctx.filename = *pfilename; ctx.fnr = 0; + lrec_reader_t* plrec_reader = use_file_reader_mmap ? plrec_reader_mmap : plrec_reader_stdio; // Start-of-file hook, e.g. expecting CSV headers on input. - if (use_file_reader_mmap) { - plrec_reader_mmap->psof_func(plrec_reader_mmap->pvstate); - ok = do_file_chained_mmap(*pfilename, &ctx, plrec_reader_mmap, pmapper_list, plrec_writer, output_stream) && ok; - } else { - plrec_reader_stdio->psof_func(plrec_reader_stdio->pvstate); - ok = do_file_chained(*pfilename, &ctx, plrec_reader_stdio, pmapper_list, plrec_writer, output_stream) && ok; - } + plrec_reader->psof_func(plrec_reader->pvstate); + ok = do_file_chained(*pfilename, &ctx, plrec_reader, pmapper_list, plrec_writer, output_stream) && ok; } } @@ -62,21 +56,12 @@ int do_stream_chained(char** filenames, int use_file_reader_mmap, lrec_reader_t* // ---------------------------------------------------------------- static int do_file_chained(char* filename, context_t* pctx, - lrec_reader_t* plrec_reader_stdio, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream) + lrec_reader_t* plrec_reader, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream) { - FILE* input_stream = stdin; - - if (!streq(filename, "-")) { - input_stream = fopen(filename, "r"); - if (input_stream == NULL) { - fprintf(stderr, "%s: Couldn't open \"%s\" for read.\n", MLR_GLOBALS.argv0, filename); - perror(filename); - return 0; - } - } + void* pvhandle = plrec_reader->popen_func(filename); while (1) { - lrec_t* pinrec = plrec_reader_stdio->pprocess_func(input_stream, plrec_reader_stdio->pvstate, pctx); + lrec_t* pinrec = plrec_reader->pprocess_func(pvhandle, plrec_reader->pvstate, pctx); if (pinrec == NULL) break; pctx->nr++; @@ -84,29 +69,7 @@ static int do_file_chained(char* filename, context_t* pctx, drive_lrec(pinrec, pctx, pmapper_list->phead, plrec_writer, output_stream); } - if (input_stream != stdin) - fclose(input_stream); - - return 1; -} - -// ---------------------------------------------------------------- -static int do_file_chained_mmap(char* filename, context_t* pctx, - lrec_reader_t* plrec_reader_stdio, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream) -{ - // xxx communicate error back from open, or rename it to ..._open_or_die - file_reader_mmap_state_t handle = file_reader_mmap_open(filename); - - while (1) { - lrec_t* pinrec = plrec_reader_stdio->pprocess_func(&handle, plrec_reader_stdio->pvstate, pctx); - if (pinrec == NULL) - break; - pctx->nr++; - pctx->fnr++; - drive_lrec(pinrec, pctx, pmapper_list->phead, plrec_writer, output_stream); - } - - file_reader_mmap_close(&handle); + plrec_reader->pclose_func(pvhandle); return 1; }