24243026519

Committed 10 Apr 2026 12:31PM UTC coverage: 82.48% (-2.5%) from 85.021%

Build # 24243026519

Build Type

push

github

Committed by

scokmen

Commit Message

ci: fixed ubuntu based analyzer errors

Coverage Stats

8 of 28 new or added lines in 6 files covered. (28.57%)

67 existing lines in 7 files now uncovered.

612 of 742 relevant lines covered (82.48%)

5562.42 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.0

/src/worker.c

#include <getopt.h>
#include <jp_command.h>
#include <jp_encoder.h>
#include <jp_errno.h>
#include <jp_field.h>
#include <jp_memory.h>
#include <jp_queue.h>
#include <jp_reader.h>
#include <jp_worker.h>
#include <jp_writer.h>
#include <limits.h>
#include <signal.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <unistd.h>

typedef struct {
    int input_stream;
    bool dry_run;
    size_t chunk_size;
    size_t buffer_size;
    jp_queue_policy_t policy;
    jp_queue_t* queue;
    jp_field_set_t* fields;
    const char* out_dir;
} worker_ctx_t;

static jp_errno_t display_help(void) {
    JP_LOG_INFO("Usage: jpipe run [options]");
    JP_LOG_INFO("\nExecute the data processing engine with the following configurations:");
    JP_LOG_INFO("\nOptions:");
    JP_LOG_INFO("  -c, --chunk-size  <size>     Chunk size (e.g., 16kb, 64kb). Range: 1kb-128kb  (default: 16kb).");
    JP_LOG_INFO("  -b, --buffer-size <count>    Max pending operations. Range: 1-1024 (default: 64).");
    JP_LOG_INFO("  -p, --policy      <type>     Overflow policy: 'wait' or 'drop' (default: wait).");
    JP_LOG_INFO("  -o, --output      <path>     Output directory (default: current dir).");
    JP_LOG_INFO("  -f, --field       key=value  Additional field to the JSON output. Can be used multiple times.");
    JP_LOG_INFO("  -n, --dry-run                Dry run.");
    JP_LOG_INFO("  -h, --help                   Show this help message.");
    JP_LOG_INFO("\nField Options:");
    JP_LOG_INFO("  -f, --field \"key=value\"   Add a field to the JSON output.");
    JP_LOG_INFO("\n  Key Rules:");
    JP_LOG_INFO("    - Must contain only: 'a-z', 'A-Z', '0-9', '_' and '-'.");
    JP_LOG_INFO("    - Maximum length: 64 characters.");
    JP_LOG_INFO("\n  Value Type Inference:");
    JP_LOG_INFO("    - key=123        -> Number  (no quotes in JSON).");
    JP_LOG_INFO("    - key=1e+10      -> Number  (no quotes in JSON).");
    JP_LOG_INFO("    - key=true|false -> Boolean (no quotes in JSON).");
    JP_LOG_INFO("    - key=string     -> String.");
    JP_LOG_INFO("    - key=\"123\"      -> Forced string.");
    JP_LOG_INFO("    - key=\"true\"     -> Forced string.");
    JP_LOG_INFO("\n  Example:");
    JP_LOG_INFO("    Input : jpipe -f \"id=101\" -f \"name=app\" -f \"active=true\" -f \"ver=1.2.0\"");
    JP_LOG_INFO("    Output: {\"id\": 101, \"name\": \"app\", \"active\": true, \"ver\": \"1.2.0\"}");
    return 0;
}

static void display_summary(worker_ctx_t* ctx) {
    const double estimated_mem_usage =
        (double) ctx->chunk_size * (double) ctx->buffer_size / (BYTES_IN_KB * BYTES_IN_KB);

    JP_LOG_MSG("Application is starting...\n");
    JP_LOG_MSG("[Runtime Parameters]");
    JP_LOG_MSG("• Chunk Size   (-c) : %zu KB", (ctx->chunk_size / BYTES_IN_KB));
    JP_LOG_MSG("• Buffer Size  (-b) : %zu", ctx->buffer_size);
    JP_LOG_MSG("• Output Dir   (-o) : %s", ctx->out_dir);
    JP_LOG_MSG("• Policy       (-p) : %s", ctx->policy == JP_QUEUE_POLICY_WAIT ? "WAIT" : "DROP");
    if (ctx->fields->len > 0) {
        JP_LOG_MSG("• Fields       (-f) :");
        for (size_t i = 0; i < ctx->fields->len; i++) {
            JP_LOG_MSG("     %zu. %-32.*s= %-32.*s",
                       i + 1,
                       (int) ctx->fields->fields[i]->key_len,
                       ctx->fields->fields[i]->key,
                       (int) ctx->fields->fields[i]->val_len,
                       (char*) ctx->fields->fields[i]->val);
        }
    }
    JP_LOG_MSG("\n[Resource Utilization]");
    JP_LOG_MSG("• Memory Usage      :  ~%.2f MB", estimated_mem_usage);
    JP_LOG_MSG("\n* These values are based on user-provided parameters.");
    JP_LOG_MSG(
        "* Memory usage is an approximation;"
        "operating system overhead and thread stack allocations are not included.\n");
}

static jp_errno_t set_out_dir(const char* arg, worker_ctx_t* ctx) {
    size_t len = 0;
    JP_FREE(ctx->out_dir);

    if (arg == NULL) {
        return JP_ERRNO_RAISE(JP_EOUT_DIR);
    }

    len = strlen(arg);
    if (len == 0) {
        return JP_ERRNO_RAISE(JP_EOUT_DIR);
    }
    if (len > JP_PATH_MAX) {
        return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Path is too long. Maximum allowed length: %d", JP_PATH_MAX);
    }

    ctx->out_dir = jp_mem_strdup(arg);
    return 0;
}

static jp_errno_t set_field(const char* arg, worker_ctx_t* ctx) {
    if (arg == NULL) {
        return JP_ERRNO_RAISE(JP_EINV_FIELD_KEY);
    }

    JP_ATTR_ASSUME(ctx->fields != NULL);

    const jp_errno_t err = jp_field_set_add(ctx->fields, arg);
    if (err == 0) {
        return 0;
    }
    return err == JP_ETOO_MANY_FIELD ? JP_ERRNO_RAISE(err) : JP_ERRNO_RAISEF(err, "Field is invalid: \"%s\"", arg);
}

static jp_errno_t set_chunk_size(const char* arg, worker_ctx_t* ctx) {
    char* end_ptr;
    size_t chunk_size        = 0;
    unsigned long long param = 0;

    if (arg == NULL) {
        return JP_ERRNO_RAISE(JP_ECHUNK_SIZE);
    }

    errno = 0;
    param = strtoull(arg, &end_ptr, 10);

    if (errno == ERANGE || errno == EINVAL) {
        return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);
    }

    if (end_ptr == arg) {
        return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);
    }

    if (*end_ptr != '\0' && !strcmp(end_ptr, "kb") && param <= JP_CONF_CHUNK_SIZE_MAX / BYTES_IN_KB) {
        chunk_size += param * BYTES_IN_KB;
    } else {
        return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);
    }

    if (chunk_size < JP_CONF_CHUNK_SIZE_MIN || chunk_size > JP_CONF_CHUNK_SIZE_MAX) {
        return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);
    }
    ctx->chunk_size = chunk_size;
    return 0;
}

static jp_errno_t set_buffer_size(const char* arg, worker_ctx_t* ctx) {
    char* end_ptr;
    unsigned long long param = 0;

    if (arg == NULL) {
        return JP_ERRNO_RAISE(JP_EBUFFER_SIZE);
    }

    errno = 0;
    param = strtoull(arg, &end_ptr, 10);

    if (errno == ERANGE || errno == EINVAL || end_ptr == arg || *end_ptr != '\0') {
        return JP_ERRNO_RAISEF(JP_EBUFFER_SIZE, "Size is invalid: \"%.32s\"", arg);
    }

    if (param < JP_CONF_BUFFER_SIZE_MIN || param > JP_CONF_BUFFER_SIZE_MAX) {
        return JP_ERRNO_RAISEF(JP_EBUFFER_SIZE, "Size is invalid: \"%.32s\"", arg);
    }

    ctx->buffer_size = (size_t) param;
    return 0;
}

static jp_errno_t set_policy(const char* arg, worker_ctx_t* ctx) {
    if (arg == NULL) {
        return JP_ERRNO_RAISE(JP_EOVERFLOW_POLICY);
    }

    if (!strcmp(arg, "wait")) {
        ctx->policy = JP_QUEUE_POLICY_WAIT;
        return 0;
    }
    if (!strcmp(arg, "drop")) {
        ctx->policy = JP_QUEUE_POLICY_DROP;
        return 0;
    }

    return JP_ERRNO_RAISEF(JP_EOVERFLOW_POLICY, "Policy is invalid: \"%.32s\"", arg);
}

static jp_errno_t handle_unknown_argument(const char* cmd) {
    return JP_ERRNO_RAISEF(JP_EUNKNOWN_RUN_CMD, "Invalid or incomplete command: \"%.32s\"", cmd);
}

static jp_errno_t create_and_normalize_out_dir(worker_ctx_t* ctx) {
    char tmp[JP_PATH_MAX]           = {0};
    char absolute_path[JP_PATH_MAX] = {0};
    char* p                         = NULL;
    struct stat st;
    const char* output = ctx->out_dir != NULL ? ctx->out_dir : JP_CONF_OUTDIR_DEF;

    size_t path_len = strlen(output);
    strncpy(tmp, output, sizeof(tmp));

    while (path_len > 1 && tmp[path_len - 1] == '/') {
        tmp[path_len - 1] = '\0';
        path_len--;
    }

    for (p = tmp + 1; *p; p++) {
        if (*p == '/') {
            *p = '\0';
            if (mkdir(tmp, 0755) != 0 && errno != EEXIST) {
                return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Could not create the output directory: \"%128s\"", tmp);
            }
            *p = '/';
        }
    }

    if (mkdir(tmp, 0755) != 0 && errno != EEXIST) {
        return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Could not create the directory: \"%128s\"", tmp);
    }

    if (stat(tmp, &st) != 0 || !S_ISDIR(st.st_mode)) {
        return JP_ERRNO_RAISEF(JP_EOUT_DIR, "The target is not a directory: \"%128s\"", tmp);
    }

    if (access(tmp, W_OK) != 0) {
        return JP_ERRNO_RAISEF(JP_EOUT_DIR, "The target is inaccessible: \"%128s\"", tmp);
    }

    if (realpath(tmp, absolute_path) == NULL) {
        return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Could not resolve absolute path: \"%128s\"", tmp);
    }

    JP_FREE(ctx->out_dir);
    ctx->out_dir = jp_mem_strdup(absolute_path);
    return 0;
}

static jp_errno_t collect_cli_args(int argc, char* argv[], worker_ctx_t* ctx) {
    int option;
    opterr = 0;
    optind = 1;

    static struct option long_options[] = {{"chunk-size", required_argument, 0, 'c'},
                                           {"buffer-size", required_argument, 0, 'b'},
                                           {"policy", required_argument, 0, 'p'},
                                           {"field", required_argument, 0, 'f'},
                                           {"output", required_argument, 0, 'o'},
                                           {"help", no_argument, 0, 'h'},
                                           {"dry-run", no_argument, 0, 'n'},
                                           {"quiet", no_argument, 0, 'q'},
                                           {"no-color", no_argument, 0, 'C'},
                                           {0, 0, 0, 0}};

    while ((option = getopt_long(argc, argv, ":c:b:p:o:f:hnqC", long_options, NULL)) != -1) {
        switch (option) {
            case 'c':
                JP_VERIFY(set_chunk_size(optarg, ctx));
                break;
            case 'b':
                JP_VERIFY(set_buffer_size(optarg, ctx));
                break;
            case 'p':
                JP_VERIFY(set_policy(optarg, ctx));
                break;
            case 'o':
                JP_VERIFY(set_out_dir(optarg, ctx));
                break;
            case 'n':
                ctx->dry_run = true;
                break;
            case 'f':
                JP_VERIFY(set_field(optarg, ctx));
                break;
            case 'q':
                JP_ATTR_FALLTHROUGH;
            case 'C':
                JP_ATTR_FALLTHROUGH;
            case 'h':
                break;
            case ':':
                JP_ATTR_FALLTHROUGH;
            case '?':
                JP_VERIFY(handle_unknown_argument((optind < argc) ? argv[optind] : argv[argc - 1]));
                break;
            default: {
            }
        }
    }

    if (optind < argc) {
        JP_VERIFY(handle_unknown_argument(argv[optind]));
    }

    return 0;
}

static jp_errno_t finalize_worker_args(worker_ctx_t* ctx) {
    jp_errno_t err = 0;
    JP_VERIFY(create_and_normalize_out_dir(ctx));
    ctx->queue = jp_queue_create(ctx->buffer_size, ctx->chunk_size, ctx->policy, &err);
    if (err || ctx->queue == NULL) {
        return JP_ERRNO_RAISE(err ? err : JP_ESYS_ERR);
    }
    return 0;
}

static void thread_cleanup(void* data) {
    const worker_ctx_t* args = data;
    jp_queue_finalize(args->queue);
    JP_ERRNO_DUMP();
    JP_ERRNO_RESET();
}

static void* consumer_thread_init(void* data) {
    const worker_ctx_t* args         = data;
    const jp_reader_ctx_t reader_ctx = {
        .chunk_size   = args->chunk_size,
        .queue        = args->queue,
        .input_stream = args->input_stream,
    };

    pthread_cleanup_push(thread_cleanup, data);

    jp_errno_t* result = jp_mem_malloc(sizeof(jp_errno_t));
    *result            = jp_reader_consume(reader_ctx);
    pthread_exit(result);

    pthread_cleanup_pop(1);
}

static void* producer_thread_init(void* data) {
    const worker_ctx_t* args         = data;
    const jp_writer_ctx_t writer_ctx = {.chunk_size = args->chunk_size,
                                        .queue      = args->queue,
                                        .output_dir = args->out_dir,
                                        .fields     = args->fields,
                                        .encoder    = jp_encoder_json};

    pthread_cleanup_push(thread_cleanup, data);

    jp_errno_t* result = jp_mem_malloc(sizeof(jp_errno_t));
    *result            = jp_writer_produce(writer_ctx);
    pthread_exit(result);

    pthread_cleanup_pop(1);
}

static void* watcher_thread_init(void* data) {
    int sig;
    sigset_t set;
    const worker_ctx_t* args = data;

    sigemptyset(&set);
    sigaddset(&set, SIGINT);
    sigaddset(&set, SIGTERM);
    if (sigwait(&set, &sig) == 0) {
        JP_LOG_DEBUG("[WATCHER]: Termination signal (%s) was received. Shutting down...",
                     sig == SIGINT ? "SIGINT" : "SIGTERM");
        jp_queue_finalize(args->queue);
    }
    pthread_exit(NULL);
}

static jp_errno_t join_worker_thread(const pthread_t thread_id) {
    void* thread_result;
    const int err = pthread_join(thread_id, &thread_result);
    if (err) {
        return JP_ERRNO_RAISE_POSIX(JP_ESYS_ERR, err);
    }
    if (thread_result == NULL || thread_result == PTHREAD_CANCELED) {
        return 0;
    }
    const jp_errno_t thread_value = *(jp_errno_t*) thread_result;
    JP_FREE(thread_result);
    return thread_value;
}

static void cancel_worker_thread(const pthread_t thread_id) {
    const int err = pthread_cancel(thread_id);
    if (err) {
        JP_ERRNO_RAISE_POSIX(JP_ESYS_ERR, err);
    }
}

static jp_errno_t orchestrate_threads(worker_ctx_t* ctx) {
    int err = 0;
    sigset_t set;
    pthread_attr_t attr;
    struct sched_param param;
    pthread_t consumer_thread, producer_thread, watcher_thread;
    uint8_t flags = 0x0;

    sigemptyset(&set);
    sigaddset(&set, SIGTERM);
    sigaddset(&set, SIGINT);
    pthread_sigmask(SIG_BLOCK, &set, NULL);

    pthread_attr_init(&attr);

    err = pthread_create(&consumer_thread, &attr, consumer_thread_init, ctx);
    if (err) {
        goto clean_up;
    }
    flags = flags | 0x1;

    err = pthread_create(&producer_thread, &attr, producer_thread_init, ctx);
    if (err) {
        goto clean_up;
    }
    flags = flags | 0x2;

    param.sched_priority = 0;
    pthread_attr_setstacksize(&attr, PTHREAD_STACK_MIN);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    pthread_attr_setschedpolicy(&attr, SCHED_OTHER);
    pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);
    pthread_attr_setschedparam(&attr, &param);

    err = pthread_create(&watcher_thread, &attr, watcher_thread_init, ctx);

clean_up:
    if (err) {
        JP_ERRNO_RAISE_POSIX(JP_ESYS_ERR, err);
        if (flags & 0x01) {
            cancel_worker_thread(consumer_thread);
        }
        if (flags & 0x02) {
            cancel_worker_thread(producer_thread);
        }
    }
    if (flags & 0x1) {
        err += (int) join_worker_thread(consumer_thread);
    }
    if (flags & 0x2) {
        err += (int) join_worker_thread(producer_thread);
    }

    jp_queue_finalize(ctx->queue);
    pthread_attr_destroy(&attr);
    return err ? JP_ERUN_FAILED : 0;
}

jp_errno_t jp_wrk_exec(int argc, char* argv[]) {
    jp_errno_t err   = 0;
    worker_ctx_t ctx = {.input_stream = STDIN_FILENO,
                        .buffer_size  = JP_CONF_BUFFER_SIZE_DEF,
                        .chunk_size   = JP_CONF_CHUNK_SIZE_DEF,
                        .out_dir      = NULL};

    if (jp_cmd_count(argc, argv, "-h", "--help") > 0) {
        return display_help();
    }

    ctx.fields = jp_field_set_create(JP_CONF_FIELDS_MAX);
    err        = collect_cli_args(argc, argv, &ctx);
    if (err) {
        goto clean_up;
    }

    err = finalize_worker_args(&ctx);
    if (err) {
        goto clean_up;
    }

    display_summary(&ctx);
    if (!ctx.dry_run) {
        err = orchestrate_threads(&ctx);
    }

clean_up:
    JP_FREE(ctx.out_dir);
    jp_field_set_destroy(ctx.fields);
    jp_queue_destroy(ctx.queue);
    return err;
}

1	#include <getopt.h>
2	#include <jp_command.h>
3	#include <jp_encoder.h>
4	#include <jp_errno.h>
5	#include <jp_field.h>
6	#include <jp_memory.h>
7	#include <jp_queue.h>
8	#include <jp_reader.h>
9	#include <jp_worker.h>
10	#include <jp_writer.h>
11	#include <limits.h>
12	#include <signal.h>
13	#include <stdint.h>
14	#include <stdio.h>
15	#include <stdlib.h>
16	#include <string.h>
17	#include <sys/stat.h>
18	#include <unistd.h>
19
20	typedef struct {
21	int input_stream;
22	bool dry_run;
23	size_t chunk_size;
24	size_t buffer_size;
25	jp_queue_policy_t policy;
26	jp_queue_t* queue;
27	jp_field_set_t* fields;
28	const char* out_dir;
29	} worker_ctx_t;
30
31	static jp_errno_t display_help(void) {	2✔
32	JP_LOG_INFO("Usage: jpipe run [options]");	2✔
33	JP_LOG_INFO("\nExecute the data processing engine with the following configurations:");	2✔
34	JP_LOG_INFO("\nOptions:");	2✔
35	JP_LOG_INFO(" -c, --chunk-size <size> Chunk size (e.g., 16kb, 64kb). Range: 1kb-128kb (default: 16kb).");	2✔
36	JP_LOG_INFO(" -b, --buffer-size <count> Max pending operations. Range: 1-1024 (default: 64).");	2✔
37	JP_LOG_INFO(" -p, --policy <type> Overflow policy: 'wait' or 'drop' (default: wait).");	2✔
38	JP_LOG_INFO(" -o, --output <path> Output directory (default: current dir).");	2✔
39	JP_LOG_INFO(" -f, --field key=value Additional field to the JSON output. Can be used multiple times.");	2✔
40	JP_LOG_INFO(" -n, --dry-run Dry run.");	2✔
41	JP_LOG_INFO(" -h, --help Show this help message.");	2✔
42	JP_LOG_INFO("\nField Options:");	2✔
43	JP_LOG_INFO(" -f, --field \"key=value\" Add a field to the JSON output.");	2✔
44	JP_LOG_INFO("\n Key Rules:");	2✔
45	JP_LOG_INFO(" - Must contain only: 'a-z', 'A-Z', '0-9', '_' and '-'.");	2✔
46	JP_LOG_INFO(" - Maximum length: 64 characters.");	2✔
47	JP_LOG_INFO("\n Value Type Inference:");	2✔
48	JP_LOG_INFO(" - key=123 -> Number (no quotes in JSON).");	2✔
49	JP_LOG_INFO(" - key=1e+10 -> Number (no quotes in JSON).");	2✔
50	JP_LOG_INFO(" - key=true\|false -> Boolean (no quotes in JSON).");	2✔
51	JP_LOG_INFO(" - key=string -> String.");	2✔
52	JP_LOG_INFO(" - key=\"123\" -> Forced string.");	2✔
53	JP_LOG_INFO(" - key=\"true\" -> Forced string.");	2✔
54	JP_LOG_INFO("\n Example:");	2✔
55	JP_LOG_INFO(" Input : jpipe -f \"id=101\" -f \"name=app\" -f \"active=true\" -f \"ver=1.2.0\"");	2✔
56	JP_LOG_INFO(" Output: {\"id\": 101, \"name\": \"app\", \"active\": true, \"ver\": \"1.2.0\"}");	2✔
57	return 0;	2✔
58	}
59
60	static void display_summary(worker_ctx_t* ctx) {	24✔
61	const double estimated_mem_usage =	24✔
62	(double) ctx->chunk_size * (double) ctx->buffer_size / (BYTES_IN_KB * BYTES_IN_KB);	24✔
63
64	JP_LOG_MSG("Application is starting...\n");	24✔
65	JP_LOG_MSG("[Runtime Parameters]");	24✔
66	JP_LOG_MSG("• Chunk Size (-c) : %zu KB", (ctx->chunk_size / BYTES_IN_KB));	24✔
67	JP_LOG_MSG("• Buffer Size (-b) : %zu", ctx->buffer_size);	24✔
68	JP_LOG_MSG("• Output Dir (-o) : %s", ctx->out_dir);	24✔
69	JP_LOG_MSG("• Policy (-p) : %s", ctx->policy == JP_QUEUE_POLICY_WAIT ? "WAIT" : "DROP");	26✔
70	if (ctx->fields->len > 0) {	24✔
71	JP_LOG_MSG("• Fields (-f) :");	1✔
72	for (size_t i = 0; i < ctx->fields->len; i++) {	33✔
73	JP_LOG_MSG(" %zu. %-32.s= %-32.s",	32✔
74	i + 1,
75	(int) ctx->fields->fields[i]->key_len,
76	ctx->fields->fields[i]->key,
77	(int) ctx->fields->fields[i]->val_len,
78	(char*) ctx->fields->fields[i]->val);
79	}
80	}
81	JP_LOG_MSG("\n[Resource Utilization]");	24✔
82	JP_LOG_MSG("• Memory Usage : ~%.2f MB", estimated_mem_usage);	24✔
83	JP_LOG_MSG("\n* These values are based on user-provided parameters.");	24✔
84	JP_LOG_MSG(	24✔
85	"* Memory usage is an approximation;"
86	"operating system overhead and thread stack allocations are not included.\n");
87	}	24✔
88
89	static jp_errno_t set_out_dir(const char* arg, worker_ctx_t* ctx) {	10✔
90	size_t len = 0;	10✔
91	JP_FREE(ctx->out_dir);	10✔
92
93	if (arg == NULL) {	10✔
UNCOV 94	return JP_ERRNO_RAISE(JP_EOUT_DIR);	×
95	}
96
97	len = strlen(arg);	10✔
98	if (len == 0) {	10✔
99	return JP_ERRNO_RAISE(JP_EOUT_DIR);	2✔
100	}
101	if (len > JP_PATH_MAX) {	8✔
UNCOV 102	return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Path is too long. Maximum allowed length: %d", JP_PATH_MAX);	×
103	}
104
105	ctx->out_dir = jp_mem_strdup(arg);	8✔
106	return 0;	8✔
107	}
108
109	static jp_errno_t set_field(const char* arg, worker_ctx_t* ctx) {	65✔
110	if (arg == NULL) {	65✔
UNCOV 111	return JP_ERRNO_RAISE(JP_EINV_FIELD_KEY);	×
112	}
113
114	JP_ATTR_ASSUME(ctx->fields != NULL);	65✔
115
116	const jp_errno_t err = jp_field_set_add(ctx->fields, arg);	65✔
117	if (err == 0) {	65✔
118	return 0;
119	}
120	return err == JP_ETOO_MANY_FIELD ? JP_ERRNO_RAISE(err) : JP_ERRNO_RAISEF(err, "Field is invalid: \"%s\"", arg);	1✔
121	}
122
123	static jp_errno_t set_chunk_size(const char* arg, worker_ctx_t* ctx) {	28✔
124	char* end_ptr;	28✔
125	size_t chunk_size = 0;	28✔
126	unsigned long long param = 0;	28✔
127
128	if (arg == NULL) {	28✔
UNCOV 129	return JP_ERRNO_RAISE(JP_ECHUNK_SIZE);	×
130	}
131
132	errno = 0;	28✔
133	param = strtoull(arg, &end_ptr, 10);	28✔
134
135	if (errno == ERANGE \|\| errno == EINVAL) {	28✔
136	return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);	2✔
137	}
138
139	if (end_ptr == arg) {	26✔
140	return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);	4✔
141	}
142
143	if (*end_ptr != '\0' && !strcmp(end_ptr, "kb") && param <= JP_CONF_CHUNK_SIZE_MAX / BYTES_IN_KB) {	22✔
144	chunk_size += param * BYTES_IN_KB;	10✔
145	} else {
146	return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);	12✔
147	}
148
149	if (chunk_size < JP_CONF_CHUNK_SIZE_MIN \|\| chunk_size > JP_CONF_CHUNK_SIZE_MAX) {	10✔
150	return JP_ERRNO_RAISEF(JP_ECHUNK_SIZE, "Size is invalid: \"%.32s\"", arg);	2✔
151	}
152	ctx->chunk_size = chunk_size;	8✔
153	return 0;	8✔
154	}
155
156	static jp_errno_t set_buffer_size(const char* arg, worker_ctx_t* ctx) {	20✔
157	char* end_ptr;	20✔
158	unsigned long long param = 0;	20✔
159
160	if (arg == NULL) {	20✔
UNCOV 161	return JP_ERRNO_RAISE(JP_EBUFFER_SIZE);	×
162	}
163
164	errno = 0;	20✔
165	param = strtoull(arg, &end_ptr, 10);	20✔
166
167	if (errno == ERANGE \|\| errno == EINVAL \|\| end_ptr == arg \|\| *end_ptr != '\0') {	20✔
168	return JP_ERRNO_RAISEF(JP_EBUFFER_SIZE, "Size is invalid: \"%.32s\"", arg);	6✔
169	}
170
171	if (param < JP_CONF_BUFFER_SIZE_MIN \|\| param > JP_CONF_BUFFER_SIZE_MAX) {	14✔
172	return JP_ERRNO_RAISEF(JP_EBUFFER_SIZE, "Size is invalid: \"%.32s\"", arg);	6✔
173	}
174
175	ctx->buffer_size = (size_t) param;	8✔
176	return 0;	8✔
177	}
178
179	static jp_errno_t set_policy(const char* arg, worker_ctx_t* ctx) {	10✔
180	if (arg == NULL) {	10✔
UNCOV 181	return JP_ERRNO_RAISE(JP_EOVERFLOW_POLICY);	×
182	}
183
184	if (!strcmp(arg, "wait")) {	10✔
185	ctx->policy = JP_QUEUE_POLICY_WAIT;	2✔
186	return 0;	2✔
187	}
188	if (!strcmp(arg, "drop")) {	8✔
189	ctx->policy = JP_QUEUE_POLICY_DROP;	2✔
190	return 0;	2✔
191	}
192
193	return JP_ERRNO_RAISEF(JP_EOVERFLOW_POLICY, "Policy is invalid: \"%.32s\"", arg);	6✔
194	}
195
196	static jp_errno_t handle_unknown_argument(const char* cmd) {	4✔
197	return JP_ERRNO_RAISEF(JP_EUNKNOWN_RUN_CMD, "Invalid or incomplete command: \"%.32s\"", cmd);	4✔
198	}
199
200	static jp_errno_t create_and_normalize_out_dir(worker_ctx_t* ctx) {	28✔
201	char tmp[JP_PATH_MAX] = {0};	28✔
202	char absolute_path[JP_PATH_MAX] = {0};	28✔
203	char* p = NULL;	28✔
204	struct stat st;	28✔
205	const char* output = ctx->out_dir != NULL ? ctx->out_dir : JP_CONF_OUTDIR_DEF;	28✔
206
207	size_t path_len = strlen(output);	28✔
208	strncpy(tmp, output, sizeof(tmp));	28✔
209
210	while (path_len > 1 && tmp[path_len - 1] == '/') {	28✔
UNCOV 211	tmp[path_len - 1] = '\0';	×
UNCOV 212	path_len--;	×
213	}
214
215	for (p = tmp + 1; *p; p++) {	238✔
216	if (*p == '/') {	210✔
217	*p = '\0';	15✔
218	if (mkdir(tmp, 0755) != 0 && errno != EEXIST) {	15✔
UNCOV 219	return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Could not create the output directory: \"%128s\"", tmp);	×
220	}
221	*p = '/';	15✔
222	}
223	}
224
225	if (mkdir(tmp, 0755) != 0 && errno != EEXIST) {	28✔
226	return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Could not create the directory: \"%128s\"", tmp);	3✔
227	}
228
229	if (stat(tmp, &st) != 0 \|\| !S_ISDIR(st.st_mode)) {	25✔
230	return JP_ERRNO_RAISEF(JP_EOUT_DIR, "The target is not a directory: \"%128s\"", tmp);	1✔
231	}
232
233	if (access(tmp, W_OK) != 0) {	24✔
UNCOV 234	return JP_ERRNO_RAISEF(JP_EOUT_DIR, "The target is inaccessible: \"%128s\"", tmp);	×
235	}
236
237	if (realpath(tmp, absolute_path) == NULL) {	24✔
UNCOV 238	return JP_ERRNO_RAISEF(JP_EOUT_DIR, "Could not resolve absolute path: \"%128s\"", tmp);	×
239	}
240
241	JP_FREE(ctx->out_dir);	24✔
242	ctx->out_dir = jp_mem_strdup(absolute_path);	24✔
243	return 0;	24✔
244	}
245
246	static jp_errno_t collect_cli_args(int argc, char* argv[], worker_ctx_t* ctx) {	73✔
247	int option;	73✔
248	opterr = 0;	73✔
249	optind = 1;	73✔
250
251	static struct option long_options[] = {{"chunk-size", required_argument, 0, 'c'},	73✔
252	{"buffer-size", required_argument, 0, 'b'},
253	{"policy", required_argument, 0, 'p'},
254	{"field", required_argument, 0, 'f'},
255	{"output", required_argument, 0, 'o'},
256	{"help", no_argument, 0, 'h'},
257	{"dry-run", no_argument, 0, 'n'},
258	{"quiet", no_argument, 0, 'q'},
259	{"no-color", no_argument, 0, 'C'},
260	{0, 0, 0, 0}};
261
262	while ((option = getopt_long(argc, argv, ":c:b:p:o:f:hnqC", long_options, NULL)) != -1) {	310✔
263	switch (option) {	208✔
264	case 'c':	28✔
265	JP_VERIFY(set_chunk_size(optarg, ctx));	28✔
266	break;
267	case 'b':	20✔
268	JP_VERIFY(set_buffer_size(optarg, ctx));	20✔
269	break;
270	case 'p':	10✔
271	JP_VERIFY(set_policy(optarg, ctx));	10✔
272	break;
273	case 'o':	10✔
274	JP_VERIFY(set_out_dir(optarg, ctx));	10✔
275	break;
276	case 'n':	72✔
277	ctx->dry_run = true;	72✔
278	break;	72✔
279	case 'f':	65✔
280	JP_VERIFY(set_field(optarg, ctx));	65✔
281	break;
282	case 'q':
283	JP_ATTR_FALLTHROUGH;
284	case 'C':
285	JP_ATTR_FALLTHROUGH;
286	case 'h':
287	break;
288	case ':':	3✔
289	JP_ATTR_FALLTHROUGH;	3✔
290	case '?':
291	JP_VERIFY(handle_unknown_argument((optind < argc) ? argv[optind] : argv[argc - 1]));	3✔
292	break;
293	default: {	164✔
294	}	237✔
295	}
296	}
297
298	if (optind < argc) {	29✔
299	JP_VERIFY(handle_unknown_argument(argv[optind]));	1✔
300	}
301
302	return 0;
303	}
304
305	static jp_errno_t finalize_worker_args(worker_ctx_t* ctx) {	28✔
306	jp_errno_t err = 0;	28✔
307	JP_VERIFY(create_and_normalize_out_dir(ctx));	28✔
308	ctx->queue = jp_queue_create(ctx->buffer_size, ctx->chunk_size, ctx->policy, &err);	24✔
309	if (err \|\| ctx->queue == NULL) {	24✔
NEW 310	return JP_ERRNO_RAISE(err ? err : JP_ESYS_ERR);	×
311	}
312	return 0;
313	}
314
315	static void thread_cleanup(void* data) {	2✔
316	const worker_ctx_t* args = data;	2✔
317	jp_queue_finalize(args->queue);	2✔
318	JP_ERRNO_DUMP();	2✔
319	JP_ERRNO_RESET();	2✔
320	}	2✔
321
322	static void* consumer_thread_init(void* data) {	1✔
323	const worker_ctx_t* args = data;	1✔
324	const jp_reader_ctx_t reader_ctx = {	1✔
325	.chunk_size = args->chunk_size,	1✔
326	.queue = args->queue,	1✔
327	.input_stream = args->input_stream,	1✔
328	};
329
330	pthread_cleanup_push(thread_cleanup, data);	2✔
331
332	jp_errno_t* result = jp_mem_malloc(sizeof(jp_errno_t));	1✔
333	*result = jp_reader_consume(reader_ctx);	1✔
334	pthread_exit(result);	1✔
335
336	pthread_cleanup_pop(1);
337	}
338
339	static void* producer_thread_init(void* data) {	1✔
340	const worker_ctx_t* args = data;	1✔
341	const jp_writer_ctx_t writer_ctx = {.chunk_size = args->chunk_size,	1✔
342	.queue = args->queue,	1✔
343	.output_dir = args->out_dir,	1✔
344	.fields = args->fields,	1✔
345	.encoder = jp_encoder_json};
346
347	pthread_cleanup_push(thread_cleanup, data);	2✔
348
349	jp_errno_t* result = jp_mem_malloc(sizeof(jp_errno_t));	1✔
350	*result = jp_writer_produce(writer_ctx);	1✔
351	pthread_exit(result);	1✔
352
353	pthread_cleanup_pop(1);
354	}
355
356	static void* watcher_thread_init(void* data) {	1✔
357	int sig;	1✔
358	sigset_t set;	1✔
359	const worker_ctx_t* args = data;	1✔
360
361	sigemptyset(&set);	1✔
362	sigaddset(&set, SIGINT);	1✔
363	sigaddset(&set, SIGTERM);	1✔
364	if (sigwait(&set, &sig) == 0) {	1✔
UNCOV 365	JP_LOG_DEBUG("[WATCHER]: Termination signal (%s) was received. Shutting down...",	×
366	sig == SIGINT ? "SIGINT" : "SIGTERM");
UNCOV 367	jp_queue_finalize(args->queue);	×
368	}
UNCOV 369	pthread_exit(NULL);	×
370	}
371
372	static jp_errno_t join_worker_thread(const pthread_t thread_id) {	2✔
373	void* thread_result;	2✔
374	const int err = pthread_join(thread_id, &thread_result);	2✔
375	if (err) {	2✔
UNCOV 376	return JP_ERRNO_RAISE_POSIX(JP_ESYS_ERR, err);	×
377	}
378	if (thread_result == NULL \|\| thread_result == PTHREAD_CANCELED) {	2✔
379	return 0;
380	}
381	const jp_errno_t thread_value = (jp_errno_t) thread_result;	2✔
382	JP_FREE(thread_result);	2✔
383	return thread_value;
384	}
385
UNCOV 386	static void cancel_worker_thread(const pthread_t thread_id) {	×
UNCOV 387	const int err = pthread_cancel(thread_id);	×
UNCOV 388	if (err) {	×
UNCOV 389	JP_ERRNO_RAISE_POSIX(JP_ESYS_ERR, err);	×
390	}
UNCOV 391	}	×
392
393	static jp_errno_t orchestrate_threads(worker_ctx_t* ctx) {	1✔
394	int err = 0;	1✔
395	sigset_t set;	1✔
396	pthread_attr_t attr;	1✔
397	struct sched_param param;	1✔
398	pthread_t consumer_thread, producer_thread, watcher_thread;	1✔
399	uint8_t flags = 0x0;	1✔
400
401	sigemptyset(&set);	1✔
402	sigaddset(&set, SIGTERM);	1✔
403	sigaddset(&set, SIGINT);	1✔
404	pthread_sigmask(SIG_BLOCK, &set, NULL);	1✔
405
406	pthread_attr_init(&attr);	1✔
407
408	err = pthread_create(&consumer_thread, &attr, consumer_thread_init, ctx);	1✔
409	if (err) {	1✔
UNCOV 410	goto clean_up;	×
411	}
412	flags = flags \| 0x1;	1✔
413
414	err = pthread_create(&producer_thread, &attr, producer_thread_init, ctx);	1✔
415	if (err) {	1✔
416	goto clean_up;	×
417	}
418	flags = flags \| 0x2;	1✔
419
420	param.sched_priority = 0;	1✔
421	pthread_attr_setstacksize(&attr, PTHREAD_STACK_MIN);	1✔
422	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);	1✔
423	pthread_attr_setschedpolicy(&attr, SCHED_OTHER);	1✔
424	pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);	1✔
425	pthread_attr_setschedparam(&attr, &param);	1✔
426
427	err = pthread_create(&watcher_thread, &attr, watcher_thread_init, ctx);	1✔
428
429	clean_up:	1✔
430	if (err) {	1✔
UNCOV 431	JP_ERRNO_RAISE_POSIX(JP_ESYS_ERR, err);	×
UNCOV 432	if (flags & 0x01) {	×
UNCOV 433	cancel_worker_thread(consumer_thread);	×
434	}
UNCOV 435	if (flags & 0x02) {	×
UNCOV 436	cancel_worker_thread(producer_thread);	×
437	}
438	}
439	if (flags & 0x1) {	1✔
440	err += (int) join_worker_thread(consumer_thread);	1✔
441	}
442	if (flags & 0x2) {	1✔
443	err += (int) join_worker_thread(producer_thread);	1✔
444	}
445
446	jp_queue_finalize(ctx->queue);	1✔
447	pthread_attr_destroy(&attr);	1✔
448	return err ? JP_ERUN_FAILED : 0;	1✔
449	}
450
451	jp_errno_t jp_wrk_exec(int argc, char* argv[]) {	75✔
452	jp_errno_t err = 0;	75✔
453	worker_ctx_t ctx = {.input_stream = STDIN_FILENO,	75✔
454	.buffer_size = JP_CONF_BUFFER_SIZE_DEF,
455	.chunk_size = JP_CONF_CHUNK_SIZE_DEF,
456	.out_dir = NULL};
457
458	if (jp_cmd_count(argc, argv, "-h", "--help") > 0) {	75✔
459	return display_help();	2✔
460	}
461
462	ctx.fields = jp_field_set_create(JP_CONF_FIELDS_MAX);	73✔
463	err = collect_cli_args(argc, argv, &ctx);	73✔
464	if (err) {	73✔
465	goto clean_up;	45✔
466	}
467
468	err = finalize_worker_args(&ctx);	28✔
469	if (err) {	28✔
470	goto clean_up;	4✔
471	}
472
473	display_summary(&ctx);	24✔
474	if (!ctx.dry_run) {	24✔
475	err = orchestrate_threads(&ctx);	1✔
476	}
477
478	clean_up:	23✔
479	JP_FREE(ctx.out_dir);	73✔
480	jp_field_set_destroy(ctx.fields);	73✔
481	jp_queue_destroy(ctx.queue);	73✔
482	return err;	73✔
483	}

scokmen / jpipe / 24243026519

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous