22283118179

Committed 22 Feb 2026 06:48PM UTC coverage: 89.017% (+0.8%) from 88.176%

Build # 22283118179

Build Type

push

github

Committed by

scokmen

Commit Message

feat: added async queue and threadsanitizer tests

Run Details

47 of 50 new or added lines in 1 file covered. (94.0%)

308 of 346 relevant lines covered (89.02%)

25.36 hits per line

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

94.0

/src/queue.c

#include "jp_queue.h"
#include <stdlib.h>
#include <string.h>

jp_queue_t *jp_queue_create(size_t capacity, size_t chunk_size) {
    jp_queue_t *queue;

    size_t blocks_offset = sizeof(jp_queue_t);
    size_t area_offset = blocks_offset + (capacity * sizeof(jp_block_t));
    size_t total_size = area_offset + (capacity * chunk_size);

    JP_ALLOC_OR_RET(queue, malloc(total_size), NULL);

    queue->capacity = capacity;
    queue->chunk_size = chunk_size;
    queue->head = 0;
    queue->tail = 0;
    queue->length = 0;
    queue->blocks = (jp_block_t *) ((unsigned char *) queue + blocks_offset);
    queue->area = (unsigned char *) queue + area_offset;

    for (size_t i = 0; i < capacity; i++) {
        queue->blocks[i].data = queue->area + (i * chunk_size);
        queue->blocks[i].length = 0;
    }

    pthread_mutex_init(&queue->lock, NULL);
    pthread_cond_init(&queue->not_empty, NULL);
    pthread_cond_init(&queue->not_full, NULL);

    return queue;
}

int jp_queue_push(jp_queue_t *queue, const void *src, size_t len) {
    pthread_mutex_lock(&queue->lock);

    while (queue->length == queue->capacity) {
        pthread_cond_wait(&queue->not_full, &queue->lock);
    }

    size_t block_size = (len > queue->chunk_size) ? queue->chunk_size : len;
    memcpy(queue->blocks[queue->tail].data, src, block_size);
    queue->blocks[queue->tail].length = block_size;
    queue->tail = (queue->tail + 1) % queue->capacity;
    queue->length++;

    pthread_cond_signal(&queue->not_empty);
    pthread_mutex_unlock(&queue->lock);
    return 0;
}

int jp_queue_pop(jp_queue_t *queue, unsigned char *dest_buffer, size_t *out_len) {
    pthread_mutex_lock(&queue->lock);

    while (queue->length == 0) {
        pthread_cond_wait(&queue->not_empty, &queue->lock);
    }

    size_t block_len = queue->blocks[queue->head].length;
    memcpy(dest_buffer, queue->blocks[queue->head].data, block_len);
    *out_len = block_len;

    queue->head = (queue->head + 1) % queue->capacity;
    queue->length--;

    pthread_cond_signal(&queue->not_full);
    pthread_mutex_unlock(&queue->lock);

    return 0;
}

void jp_queue_destroy(jp_queue_t *queue) {
    if (queue == NULL) {
        return;
    }

    pthread_mutex_destroy(&queue->lock);
    pthread_cond_destroy(&queue->not_empty);
    pthread_cond_destroy(&queue->not_full);
    JP_FREE_IF_ALLOC(queue);
}

1	#include "jp_queue.h"
2	#include <stdlib.h>
3	#include <string.h>
4
5	jp_queue_t *jp_queue_create(size_t capacity, size_t chunk_size) {	1✔
6	jp_queue_t *queue;
7
8	size_t blocks_offset = sizeof(jp_queue_t);	1✔
9	size_t area_offset = blocks_offset + (capacity * sizeof(jp_block_t));	1✔
10	size_t total_size = area_offset + (capacity * chunk_size);	1✔
11
12	JP_ALLOC_OR_RET(queue, malloc(total_size), NULL);	1✔
13
14	queue->capacity = capacity;	1✔
15	queue->chunk_size = chunk_size;	1✔
16	queue->head = 0;	1✔
17	queue->tail = 0;	1✔
18	queue->length = 0;	1✔
19	queue->blocks = (jp_block_t ) ((unsigned char ) queue + blocks_offset);	1✔
20	queue->area = (unsigned char *) queue + area_offset;	1✔
21
22	for (size_t i = 0; i < capacity; i++) {	5✔
23	queue->blocks[i].data = queue->area + (i * chunk_size);	4✔
24	queue->blocks[i].length = 0;	4✔
25	}
26
27	pthread_mutex_init(&queue->lock, NULL);	1✔
28	pthread_cond_init(&queue->not_empty, NULL);	1✔
29	pthread_cond_init(&queue->not_full, NULL);	1✔
30
31	return queue;	1✔
32	}
33
34	int jp_queue_push(jp_queue_t queue, const void src, size_t len) {	4✔
35	pthread_mutex_lock(&queue->lock);	4✔
36
37	while (queue->length == queue->capacity) {	4✔
NEW 38	pthread_cond_wait(&queue->not_full, &queue->lock);	×
39	}
40
41	size_t block_size = (len > queue->chunk_size) ? queue->chunk_size : len;	4✔
42	memcpy(queue->blocks[queue->tail].data, src, block_size);	4✔
43	queue->blocks[queue->tail].length = block_size;	4✔
44	queue->tail = (queue->tail + 1) % queue->capacity;	4✔
45	queue->length++;	4✔
46
47	pthread_cond_signal(&queue->not_empty);	4✔
48	pthread_mutex_unlock(&queue->lock);	4✔
49	return 0;	4✔
50	}
51
52	int jp_queue_pop(jp_queue_t queue, unsigned char dest_buffer, size_t *out_len) {	4✔
53	pthread_mutex_lock(&queue->lock);	4✔
54
55	while (queue->length == 0) {	4✔
NEW 56	pthread_cond_wait(&queue->not_empty, &queue->lock);	×
57	}
58
59	size_t block_len = queue->blocks[queue->head].length;	4✔
60	memcpy(dest_buffer, queue->blocks[queue->head].data, block_len);	4✔
61	*out_len = block_len;	4✔
62
63	queue->head = (queue->head + 1) % queue->capacity;	4✔
64	queue->length--;	4✔
65
66	pthread_cond_signal(&queue->not_full);	4✔
67	pthread_mutex_unlock(&queue->lock);	4✔
68
69	return 0;	4✔
70	}
71
72	void jp_queue_destroy(jp_queue_t *queue) {	1✔
73	if (queue == NULL) {	1✔
NEW 74	return;	×
75	}
76
77	pthread_mutex_destroy(&queue->lock);	1✔
78	pthread_cond_destroy(&queue->not_empty);	1✔
79	pthread_cond_destroy(&queue->not_full);	1✔
80	JP_FREE_IF_ALLOC(queue);	1✔
81	}

scokmen / jpipe / 22283118179

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