Mirrors/miller
1
0
Fork 0
mirror of https://github.com/johnkerl/miller.git synced 2026-07-21 02:23:20 +00:00
Code Issues Projects Releases Wiki Activity

join iterate

Browse source
This commit is contained in:
John Kerl 2015-06-15 22:54:07 -04:00
parent 2b97065e91
commit befe197ec6
13 changed files with 58 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

6
c/Makefile
View file

@ -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 \

29
c/input/file_reader_mmap.c
View file

@ -5,34 +5,35 @@
#include <unistd.h>
#include <sys/stat.h>
#include <sys/mman.h>
#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);
}

5
c/input/file_reader_mmap.h
View file

@ -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

4
c/input/lrec_reader.h
View file

@ -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;

3
c/input/lrec_reader_mmap_csv.c
View file

@ -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;

2
c/input/lrec_reader_mmap_dkvp.c
View file

@ -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;

2
c/input/lrec_reader_mmap_nidx.c
View file

@ -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;

2
c/input/lrec_reader_mmap_xtab.c
View file

@ -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;

3
c/input/lrec_reader_stdio_csv.c
View file

@ -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;

3
c/input/lrec_reader_stdio_dkvp.c
View file

@ -1,6 +1,7 @@
#include <stdio.h>
#include <stdlib.h>
#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;

3
c/input/lrec_reader_stdio_nidx.c
View file

@ -1,5 +1,6 @@
#include <stdlib.h>
#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;

3
c/input/lrec_reader_stdio_xtab.c
View file

@ -1,6 +1,7 @@
#include <stdio.h>
#include <stdlib.h>
#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;

51
c/stream/stream.c
View file

@ -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;
}