24945903988

Committed 26 Apr 2026 02:02AM UTC coverage: 85.127% (-0.002%) from 85.129%

Build # 24945903988

Build Type

push

github

Committed by

saitoha

Commit Message

tests: avoid helper name collision in amalgamated test runner

Coverage Stats

46896 of 116668 branches covered (40.2%)

2 of 8 new or added lines in 1 file covered. (25.0%)

13 existing lines in 6 files now uncovered.

136927 of 160850 relevant lines covered (85.13%)

7827879.67 hits per line

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

83.83

/src/threadpool.c

/*
 * SPDX-License-Identifier: MIT
 *
 * Copyright (c) 2025 libsixel developers. See `AUTHORS`.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of
 * this software and associated documentation files (the "Software"), to deal in
 * the Software without restriction, including without limitation the rights to
 * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
 * of the Software, and to permit persons to whom the Software is furnished to do
 * so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

#if defined(HAVE_CONFIG_H)
#include "config.h"
#endif

#if SIXEL_ENABLE_THREADS

#include <errno.h>
#include <stdlib.h>
#include <string.h>

#include "threadpool.h"
#include "threading.h"

typedef struct threadpool_worker threadpool_worker_t;

struct threadpool_worker {
    threadpool_t *pool;
    sixel_thread_t thread;
    void *workspace;
    int started;
    int index;
    int pinned;
};

struct threadpool {
    int nthreads;
    int qsize;
    size_t workspace_size;
    tp_workspace_cleanup_fn workspace_cleanup;
    tp_worker_fn worker;
    void *userdata;
    tp_job_t *jobs;
    int head;
    int tail;
    int count;
    int running;
    int shutting_down;
    int joined;
    int error;
    int threads_started;
    int worker_capacity;
    int pin_threads;
    int hw_threads;
    sixel_mutex_t mutex;
    sixel_cond_t cond_not_empty;
    sixel_cond_t cond_not_full;
    sixel_cond_t cond_drained;
    int mutex_ready;
    int cond_not_empty_ready;
    int cond_not_full_ready;
    int cond_drained_ready;
    threadpool_worker_t **workers; /* owned worker slots (stable addresses) */
};

static void threadpool_free(threadpool_t *pool);
static int threadpool_worker_main(void *arg);
static int threadpool_spawn_worker(threadpool_t *pool,
                                   threadpool_worker_t *worker);

/*
 * Release every dynamically allocated component of the pool. Callers must
 * ensure that worker threads have already terminated before invoking this
 * helper; otherwise joining would operate on freed memory.
 */
static void
threadpool_free(threadpool_t *pool)
{
    int i;

    if (pool == NULL) {
        return;
    }
    if (pool->workers != NULL) {
        for (i = 0; i < pool->worker_capacity; ++i) {
            threadpool_worker_t *worker;
            worker = pool->workers[i];
            if (worker == NULL) {
                continue;
            }
            if (worker->workspace != NULL) {
                if (pool->workspace_cleanup != NULL) {
                    pool->workspace_cleanup(worker->workspace);
                }
                free(worker->workspace);
            }
            free(worker);
        }
        free(pool->workers);
    }
    if (pool->jobs != NULL) {
        free(pool->jobs);
    }
    if (pool->cond_drained_ready) {
        sixel_cond_destroy(&pool->cond_drained);
    }
    if (pool->cond_not_full_ready) {
        sixel_cond_destroy(&pool->cond_not_full);
    }
    if (pool->cond_not_empty_ready) {
        sixel_cond_destroy(&pool->cond_not_empty);
    }
    if (pool->mutex_ready) {
        sixel_mutex_destroy(&pool->mutex);
    }
    free(pool);
}

/*
 * Worker threads pull jobs from the ring buffer, execute the supplied callback
 * outside the critical section, and record the first failure code. All
 * synchronization is delegated to the mutex/condition helpers provided by the
 * threading abstraction.
 */
static int
threadpool_worker_main(void *arg)
{
    threadpool_worker_t *worker;
    threadpool_t *pool;
    tp_job_t job;
    int rc;

    worker = (threadpool_worker_t *)arg;
    pool = worker->pool;
    for (;;) {
        sixel_mutex_lock(&pool->mutex);
        while (pool->count == 0 && !pool->shutting_down) {
            sixel_cond_wait(&pool->cond_not_empty, &pool->mutex);
        }
        if (pool->count == 0 && pool->shutting_down) {
            sixel_mutex_unlock(&pool->mutex);
            break;
        }
        job = pool->jobs[pool->head];
        pool->head = (pool->head + 1) % pool->qsize;
        pool->count -= 1;
        pool->running += 1;
        sixel_cond_signal(&pool->cond_not_full);
        sixel_mutex_unlock(&pool->mutex);

        if (pool->pin_threads && !worker->pinned && pool->hw_threads > 0) {
            int cpu_index;

            cpu_index = worker->index % pool->hw_threads;
            (void)sixel_thread_pin_self(cpu_index);
            worker->pinned = 1;
        }

        rc = pool->worker(job, pool->userdata, worker->workspace);

        sixel_mutex_lock(&pool->mutex);
        pool->running -= 1;
        if (rc != SIXEL_OK && pool->error == SIXEL_OK) {
            pool->error = rc;
        }
        if (pool->count == 0 && pool->running == 0) {
            sixel_cond_broadcast(&pool->cond_drained);
        }
        sixel_mutex_unlock(&pool->mutex);
    }
    return SIXEL_OK;
}

SIXELAPI threadpool_t *
threadpool_create(int nthreads,
                  int qsize,
                  size_t workspace_size,
                  tp_worker_fn worker,
                  void *userdata,
                  tp_workspace_cleanup_fn workspace_cleanup)
{
    threadpool_t *pool;
    int i;
    int rc;

    if (nthreads <= 0 || qsize <= 0 || worker == NULL) {
        return NULL;
    }
    pool = (threadpool_t *)calloc(1, sizeof(threadpool_t));
    if (pool == NULL) {
        return NULL;
    }
    pool->nthreads = nthreads;
    pool->qsize = qsize;
    pool->workspace_size = workspace_size;
    pool->worker = worker;
    pool->userdata = userdata;
    pool->jobs = NULL;
    pool->head = 0;
    pool->tail = 0;
    pool->count = 0;
    pool->running = 0;
    pool->shutting_down = 0;
    pool->joined = 0;
    pool->error = SIXEL_OK;
    pool->threads_started = 0;
    pool->mutex_ready = 0;
    pool->cond_not_empty_ready = 0;
    pool->cond_not_full_ready = 0;
    pool->cond_drained_ready = 0;
    pool->pin_threads = 0;
    pool->hw_threads = 0;
    pool->workers = NULL;
    pool->workspace_cleanup = workspace_cleanup;

    rc = sixel_mutex_init(&pool->mutex);
    if (rc != SIXEL_OK) {
        errno = EINVAL;
        threadpool_free(pool);
        return NULL;
    }
    pool->mutex_ready = 1;

    rc = sixel_cond_init(&pool->cond_not_empty);
    if (rc != SIXEL_OK) {
        errno = EINVAL;
        threadpool_free(pool);
        return NULL;
    }
    pool->cond_not_empty_ready = 1;

    rc = sixel_cond_init(&pool->cond_not_full);
    if (rc != SIXEL_OK) {
        errno = EINVAL;
        threadpool_free(pool);
        return NULL;
    }
    pool->cond_not_full_ready = 1;

    rc = sixel_cond_init(&pool->cond_drained);
    if (rc != SIXEL_OK) {
        errno = EINVAL;
        threadpool_free(pool);
        return NULL;
    }
    pool->cond_drained_ready = 1;

    pool->jobs = (tp_job_t *)malloc(sizeof(tp_job_t) * (size_t)qsize);
    if (pool->jobs == NULL) {
        threadpool_free(pool);
        return NULL;
    }

    pool->worker_capacity = nthreads;
    pool->workers = (threadpool_worker_t **)calloc((size_t)nthreads,
            sizeof(threadpool_worker_t *));
    if (pool->workers == NULL) {
        threadpool_free(pool);
        return NULL;
    }

    for (i = 0; i < nthreads; ++i) {
        pool->workers[i] = (threadpool_worker_t *)
            calloc(1, sizeof(threadpool_worker_t));
        if (pool->workers[i] == NULL) {
            pool->shutting_down = 1;
            sixel_cond_broadcast(&pool->cond_not_empty);
            break;
        }
        pool->workers[i]->pool = pool;
        pool->workers[i]->workspace = NULL;
        pool->workers[i]->started = 0;
        pool->workers[i]->index = i;
        pool->workers[i]->pinned = 0;
        if (workspace_size > 0) {
            /*
             * Zero-initialize the per-thread workspace so that structures like
             * `sixel_parallel_worker_state_t` start with predictable values.
             * The worker initialization logic assumes fields such as
             * `initialized` are cleared before the first job.
             */
            pool->workers[i]->workspace = calloc(1, workspace_size);
            if (pool->workers[i]->workspace == NULL) {
                pool->shutting_down = 1;
                sixel_cond_broadcast(&pool->cond_not_empty);
                break;
            }
        }
        rc = threadpool_spawn_worker(pool, pool->workers[i]);
        if (rc != SIXEL_OK) {
            break;
        }
    }

    if (pool->threads_started != nthreads) {
        int started;

        started = pool->threads_started;
        for (i = 0; i < started; ++i) {
            sixel_cond_broadcast(&pool->cond_not_empty);
            sixel_thread_join(&pool->workers[i]->thread);
        }
        threadpool_free(pool);
        return NULL;
    }

    return pool;
}

SIXELAPI void
threadpool_set_affinity(threadpool_t *pool, int pin_threads)
{
    if (pool == NULL) {
        return;
    }

    sixel_mutex_lock(&pool->mutex);
    pool->pin_threads = (pin_threads != 0) ? 1 : 0;
    if (pool->pin_threads != 0) {
        pool->hw_threads = sixel_get_hw_threads();
        if (pool->hw_threads < 1) {
            pool->pin_threads = 0;
        }
    } else {
        pool->hw_threads = 0;
    }
    sixel_mutex_unlock(&pool->mutex);
}

static int
threadpool_spawn_worker(threadpool_t *pool, threadpool_worker_t *worker)
{
    int rc;

    if (pool == NULL || worker == NULL) {
        return SIXEL_BAD_ARGUMENT;
    }
    rc = sixel_thread_create(&worker->thread,
                             threadpool_worker_main,
                             worker);
    if (rc != SIXEL_OK) {
        sixel_mutex_lock(&pool->mutex);
        pool->shutting_down = 1;
        sixel_cond_broadcast(&pool->cond_not_empty);
        sixel_mutex_unlock(&pool->mutex);
        return rc;
    }
    worker->started = 1;
    pool->threads_started += 1;
    return SIXEL_OK;
}

SIXELAPI void
threadpool_destroy(threadpool_t *pool)
{
    if (pool == NULL) {
        return;
    }
    threadpool_finish(pool);
    threadpool_free(pool);
}

SIXELAPI void
threadpool_push(threadpool_t *pool, tp_job_t job)
{
    if (pool == NULL) {
        return;
    }
    sixel_mutex_lock(&pool->mutex);
    if (pool->shutting_down) {
        sixel_mutex_unlock(&pool->mutex);
        return;
    }
    while (pool->count == pool->qsize && !pool->shutting_down) {
        sixel_cond_wait(&pool->cond_not_full, &pool->mutex);
    }
    if (pool->shutting_down) {
        sixel_mutex_unlock(&pool->mutex);
        return;
    }
    pool->jobs[pool->tail] = job;
    pool->tail = (pool->tail + 1) % pool->qsize;
    pool->count += 1;
    sixel_cond_signal(&pool->cond_not_empty);
    sixel_mutex_unlock(&pool->mutex);
}

SIXELAPI void
threadpool_finish(threadpool_t *pool)
{
    int i;

    if (pool == NULL) {
        return;
    }
    sixel_mutex_lock(&pool->mutex);
    if (pool->joined) {
        sixel_mutex_unlock(&pool->mutex);
        return;
    }
    pool->shutting_down = 1;
    sixel_cond_broadcast(&pool->cond_not_empty);
    sixel_cond_broadcast(&pool->cond_not_full);
    while (pool->count > 0 || pool->running > 0) {
        sixel_cond_wait(&pool->cond_drained, &pool->mutex);
    }
    sixel_mutex_unlock(&pool->mutex);

    for (i = 0; i < pool->threads_started; ++i) {
        if (pool->workers[i] != NULL && pool->workers[i]->started) {
            sixel_thread_join(&pool->workers[i]->thread);
            pool->workers[i]->started = 0;
        }
    }

    sixel_mutex_lock(&pool->mutex);
    pool->joined = 1;
    sixel_mutex_unlock(&pool->mutex);
}

SIXELAPI int
threadpool_grow(threadpool_t *pool, int additional_threads)
{
    threadpool_worker_t **expanded;
    int new_target;
    int started_new;
    int i;
    int rc;

    if (pool == NULL || additional_threads <= 0) {
        return SIXEL_OK;
    }

    sixel_mutex_lock(&pool->mutex);
    if (pool->shutting_down) {
        sixel_mutex_unlock(&pool->mutex);
        return SIXEL_RUNTIME_ERROR;
    }
    new_target = pool->nthreads + additional_threads;
    /*
     * Worker structs stay heap-allocated per slot so pointer-table growth
     * never invalidates addresses already held by running threads.
     */
    if (new_target > pool->worker_capacity) {
        expanded = (threadpool_worker_t **)realloc(
            pool->workers,
            (size_t)new_target * sizeof(threadpool_worker_t *));
        if (expanded == NULL) {
            sixel_mutex_unlock(&pool->mutex);
            return SIXEL_BAD_ALLOCATION;
        }
        memset(expanded + pool->worker_capacity,
               0,
               (size_t)(new_target - pool->worker_capacity)
                   * sizeof(threadpool_worker_t *));
        pool->workers = expanded;
        pool->worker_capacity = new_target;
    }
    sixel_mutex_unlock(&pool->mutex);

    started_new = 0;
    rc = SIXEL_OK;
    for (i = pool->nthreads; i < new_target; ++i) {
        pool->workers[i] = (threadpool_worker_t *)
            calloc(1, sizeof(threadpool_worker_t));
        if (pool->workers[i] == NULL) {
            rc = SIXEL_BAD_ALLOCATION;
            break;
        }
        pool->workers[i]->pool = pool;
        pool->workers[i]->workspace = NULL;
        pool->workers[i]->started = 0;
        pool->workers[i]->index = i;
        pool->workers[i]->pinned = 0;
        if (pool->workspace_size > 0) {
            pool->workers[i]->workspace =
                calloc(1, pool->workspace_size);
            if (pool->workers[i]->workspace == NULL) {
                rc = SIXEL_BAD_ALLOCATION;
                break;
            }
        }

        rc = threadpool_spawn_worker(pool, pool->workers[i]);
        if (rc != SIXEL_OK) {
            break;
        }
        started_new += 1;
    }

    if (rc != SIXEL_OK) {
        int j;

        for (j = i; j < new_target; ++j) {
            if (pool->workers[j] != NULL) {
                if (pool->workers[j]->workspace != NULL) {
                    free(pool->workers[j]->workspace);
                }
                free(pool->workers[j]);
                pool->workers[j] = NULL;
            }
        }
    }

    sixel_mutex_lock(&pool->mutex);
    pool->nthreads = pool->nthreads + started_new;
    sixel_mutex_unlock(&pool->mutex);

    return rc;
}

SIXELAPI int
threadpool_get_error(threadpool_t *pool)
{
    int error;

    if (pool == NULL) {
        return SIXEL_BAD_ARGUMENT;
    }
    sixel_mutex_lock(&pool->mutex);
    error = pool->error;
    sixel_mutex_unlock(&pool->mutex);
    return error;
}

#endif  /* SIXEL_ENABLE_THREADS */

/* emacs Local Variables:      */
/* emacs mode: c               */
/* emacs tab-width: 4          */
/* emacs indent-tabs-mode: nil */
/* emacs c-basic-offset: 4     */
/* emacs End:                  */
/* vim: set expandtab ts=4 sts=4 sw=4 : */
/* EOF */

1	/*
2	* SPDX-License-Identifier: MIT
3	*
4	* Copyright (c) 2025 libsixel developers. See `AUTHORS`.
5	*
6	* Permission is hereby granted, free of charge, to any person obtaining a copy of
7	* this software and associated documentation files (the "Software"), to deal in
8	* the Software without restriction, including without limitation the rights to
9	* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
10	* of the Software, and to permit persons to whom the Software is furnished to do
11	* so, subject to the following conditions:
12	*
13	* The above copyright notice and this permission notice shall be included in all
14	* copies or substantial portions of the Software.
15	*
16	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17	* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18	* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19	* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20	* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21	* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
22	* SOFTWARE.
23	*/
24
25	#if defined(HAVE_CONFIG_H)
26	#include "config.h"
27	#endif
28
29	#if SIXEL_ENABLE_THREADS
30
31	#include <errno.h>
32	#include <stdlib.h>
33	#include <string.h>
34
35	#include "threadpool.h"
36	#include "threading.h"
37
38	typedef struct threadpool_worker threadpool_worker_t;
39
40	struct threadpool_worker {
41	threadpool_t *pool;
42	sixel_thread_t thread;
43	void *workspace;
44	int started;
45	int index;
46	int pinned;
47	};
48
49	struct threadpool {
50	int nthreads;
51	int qsize;
52	size_t workspace_size;
53	tp_workspace_cleanup_fn workspace_cleanup;
54	tp_worker_fn worker;
55	void *userdata;
56	tp_job_t *jobs;
57	int head;
58	int tail;
59	int count;
60	int running;
61	int shutting_down;
62	int joined;
63	int error;
64	int threads_started;
65	int worker_capacity;
66	int pin_threads;
67	int hw_threads;
68	sixel_mutex_t mutex;
69	sixel_cond_t cond_not_empty;
70	sixel_cond_t cond_not_full;
71	sixel_cond_t cond_drained;
72	int mutex_ready;
73	int cond_not_empty_ready;
74	int cond_not_full_ready;
75	int cond_drained_ready;
76	threadpool_worker_t *workers; / owned worker slots (stable addresses) */
77	};
78
79	static void threadpool_free(threadpool_t *pool);
80	static int threadpool_worker_main(void *arg);
81	static int threadpool_spawn_worker(threadpool_t *pool,
82	threadpool_worker_t *worker);
83
84	/*
85	* Release every dynamically allocated component of the pool. Callers must
86	* ensure that worker threads have already terminated before invoking this
87	* helper; otherwise joining would operate on freed memory.
88	*/
89	static void
90	threadpool_free(threadpool_t *pool)	45,013✔
91	{
92	int i;	23,546✔
93
94	if (pool == NULL) {	45,013✔
95	return;
96	}
97	if (pool->workers != NULL) {	45,013✔
98	for (i = 0; i < pool->worker_capacity; ++i) {	165,626!
99	threadpool_worker_t *worker;	63,193✔
100	worker = pool->workers[i];	120,613✔
101	if (worker == NULL) {	120,613✔
102	continue;	×
103	}
104	if (worker->workspace != NULL) {	120,613✔
105	if (pool->workspace_cleanup != NULL) {	104,127✔
106	pool->workspace_cleanup(worker->workspace);	14,270✔
107	}	2,234✔
108	free(worker->workspace);	104,127✔
109	}	53,854✔
110	free(worker);	120,613✔
111	}	57,072!
112	free(pool->workers);	45,013✔
113	}	20,916✔
114	if (pool->jobs != NULL) {	45,013✔
115	free(pool->jobs);	45,013✔
116	}	20,916✔
117	if (pool->cond_drained_ready) {	45,013✔
118	sixel_cond_destroy(&pool->cond_drained);	45,013✔
119	}	20,916✔
120	if (pool->cond_not_full_ready) {	45,013✔
121	sixel_cond_destroy(&pool->cond_not_full);	45,013✔
122	}	20,916✔
123	if (pool->cond_not_empty_ready) {	45,013✔
124	sixel_cond_destroy(&pool->cond_not_empty);	45,013✔
125	}	20,916✔
126	if (pool->mutex_ready) {	45,013✔
127	sixel_mutex_destroy(&pool->mutex);	45,013✔
128	}	20,916✔
129	free(pool);	45,013✔
130	}	20,916✔
131
132	/*
133	* Worker threads pull jobs from the ring buffer, execute the supplied callback
134	* outside the critical section, and record the first failure code. All
135	* synchronization is delegated to the mutex/condition helpers provided by the
136	* threading abstraction.
137	*/
138	static int
139	threadpool_worker_main(void *arg)	120,608✔
140	{
141	threadpool_worker_t *worker;	63,192✔
142	threadpool_t *pool;	63,192✔
143	tp_job_t job;	63,192✔
144	int rc;	63,192✔
145
146	worker = (threadpool_worker_t *)arg;	120,608✔
147	pool = worker->pool;	120,608✔
148	for (;;) {	402,899✔
149	sixel_mutex_lock(&pool->mutex);	434,068✔
150	while (pool->count == 0 && !pool->shutting_down) {	588,242!
151	sixel_cond_wait(&pool->cond_not_empty, &pool->mutex);	154,169✔
152	}
153	if (pool->count == 0 && pool->shutting_down) {	434,078!
154	sixel_mutex_unlock(&pool->mutex);	120,613✔
155	break;	120,613✔
156	}
157	job = pool->jobs[pool->head];	310,816✔
158	pool->head = (pool->head + 1) % pool->qsize;	310,816✔
159	pool->count -= 1;	310,816✔
160	pool->running += 1;	310,816✔
161	sixel_cond_signal(&pool->cond_not_full);	310,816✔
162	sixel_mutex_unlock(&pool->mutex);	310,817✔
163
164	if (pool->pin_threads && !worker->pinned && pool->hw_threads > 0) {	310,817!
165	int cpu_index;	39,308✔
166
167	cpu_index = worker->index % pool->hw_threads;	77,553✔
168	(void)sixel_thread_pin_self(cpu_index);	77,553✔
169	worker->pinned = 1;	77,551✔
170	}	31,816✔
171
172	rc = pool->worker(job, pool->userdata, worker->workspace);	304,228✔
173
174	sixel_mutex_lock(&pool->mutex);	304,121✔
175	pool->running -= 1;	304,229✔
176	if (rc != SIXEL_OK && pool->error == SIXEL_OK) {	304,229!
177	pool->error = rc;	×
178	}
179	if (pool->count == 0 && pool->running == 0) {	304,872!
180	sixel_cond_broadcast(&pool->cond_drained);	96,543✔
181	}	51,706✔
182	sixel_mutex_unlock(&pool->mutex);	313,466✔
183	}
184	return SIXEL_OK;	120,613✔
185	}	31,305✔
186
187	SIXELAPI threadpool_t *
188	threadpool_create(int nthreads,	45,012✔
189	int qsize,
190	size_t workspace_size,
191	tp_worker_fn worker,
192	void *userdata,
193	tp_workspace_cleanup_fn workspace_cleanup)
194	{
195	threadpool_t *pool;	23,546✔
196	int i;	23,546✔
197	int rc;	23,546✔
198
199	if (nthreads <= 0 \|\| qsize <= 0 \|\| worker == NULL) {	45,012!
200	return NULL;
201	}
202	pool = (threadpool_t *)calloc(1, sizeof(threadpool_t));	45,012✔
203	if (pool == NULL) {	45,012✔
204	return NULL;
205	}
206	pool->nthreads = nthreads;	45,012✔
207	pool->qsize = qsize;	45,012✔
208	pool->workspace_size = workspace_size;	45,012✔
209	pool->worker = worker;	45,012✔
210	pool->userdata = userdata;	45,012✔
211	pool->jobs = NULL;	45,012✔
212	pool->head = 0;	45,012✔
213	pool->tail = 0;	45,012✔
214	pool->count = 0;	45,012✔
215	pool->running = 0;	45,012✔
216	pool->shutting_down = 0;	45,012✔
217	pool->joined = 0;	45,012✔
218	pool->error = SIXEL_OK;	45,012✔
219	pool->threads_started = 0;	45,012✔
220	pool->mutex_ready = 0;	45,012✔
221	pool->cond_not_empty_ready = 0;	45,012✔
222	pool->cond_not_full_ready = 0;	45,012✔
223	pool->cond_drained_ready = 0;	45,012✔
224	pool->pin_threads = 0;	45,012✔
225	pool->hw_threads = 0;	45,012✔
226	pool->workers = NULL;	45,012✔
227	pool->workspace_cleanup = workspace_cleanup;	45,012✔
228
229	rc = sixel_mutex_init(&pool->mutex);	45,012✔
230	if (rc != SIXEL_OK) {	45,013!
231	errno = EINVAL;	×
232	threadpool_free(pool);	×
233	return NULL;	×
234	}
235	pool->mutex_ready = 1;	45,013✔
236
237	rc = sixel_cond_init(&pool->cond_not_empty);	45,013✔
238	if (rc != SIXEL_OK) {	45,013!
239	errno = EINVAL;	×
240	threadpool_free(pool);	×
241	return NULL;	×
242	}
243	pool->cond_not_empty_ready = 1;	45,013✔
244
245	rc = sixel_cond_init(&pool->cond_not_full);	45,013✔
246	if (rc != SIXEL_OK) {	45,013!
247	errno = EINVAL;	×
248	threadpool_free(pool);	×
249	return NULL;	×
250	}
251	pool->cond_not_full_ready = 1;	45,013✔
252
253	rc = sixel_cond_init(&pool->cond_drained);	45,013✔
254	if (rc != SIXEL_OK) {	45,013!
255	errno = EINVAL;	×
256	threadpool_free(pool);	×
257	return NULL;	×
258	}
259	pool->cond_drained_ready = 1;	45,013✔
260
261	pool->jobs = (tp_job_t )malloc(sizeof(tp_job_t) (size_t)qsize);	45,013✔
262	if (pool->jobs == NULL) {	45,013✔
263	threadpool_free(pool);	×
264	return NULL;	×
265	}
266
267	pool->worker_capacity = nthreads;	45,013✔
268	pool->workers = (threadpool_worker_t **)calloc((size_t)nthreads,	45,013✔
269	sizeof(threadpool_worker_t *));
270	if (pool->workers == NULL) {	45,013✔
271	threadpool_free(pool);	×
272	return NULL;	×
273	}
274
275	for (i = 0; i < nthreads; ++i) {	141,453!
276	pool->workers[i] = (threadpool_worker_t *)	96,441✔
277	calloc(1, sizeof(threadpool_worker_t));	96,441✔
278	if (pool->workers[i] == NULL) {	96,441✔
279	pool->shutting_down = 1;	×
280	sixel_cond_broadcast(&pool->cond_not_empty);	×
281	break;	×
282	}
283	pool->workers[i]->pool = pool;	96,441✔
284	pool->workers[i]->workspace = NULL;	96,441✔
285	pool->workers[i]->started = 0;	96,441✔
286	pool->workers[i]->index = i;	96,441✔
287	pool->workers[i]->pinned = 0;	96,441✔
288	if (workspace_size > 0) {	96,441✔
289	/*
290	* Zero-initialize the per-thread workspace so that structures like
291	* `sixel_parallel_worker_state_t` start with predictable values.
292	* The worker initialization logic assumes fields such as
293	* `initialized` are cleared before the first job.
294	*/
295	pool->workers[i]->workspace = calloc(1, workspace_size);	79,957✔
296	if (pool->workers[i]->workspace == NULL) {	79,957✔
297	pool->shutting_down = 1;	×
298	sixel_cond_broadcast(&pool->cond_not_empty);	×
299	break;	×
300	}
301	}	41,290✔
302	rc = threadpool_spawn_worker(pool, pool->workers[i]);	96,441✔
303	if (rc != SIXEL_OK) {	96,440!
304	break;
305	}
306	}	44,508✔
307
308	if (pool->threads_started != nthreads) {	45,012!
309	int started;
310
311	started = pool->threads_started;
312	for (i = 0; i < started; ++i) {	×
313	sixel_cond_broadcast(&pool->cond_not_empty);	×
314	sixel_thread_join(&pool->workers[i]->thread);	×
315	}
316	threadpool_free(pool);	×
317	return NULL;	×
318	}
319
320	return pool;	32,925✔
321	}	20,916✔
322
323	SIXELAPI void
324	threadpool_set_affinity(threadpool_t *pool, int pin_threads)	44,507✔
325	{
326	if (pool == NULL) {	44,507✔
327	return;
328	}
329
330	sixel_mutex_lock(&pool->mutex);	44,507✔
331	pool->pin_threads = (pin_threads != 0) ? 1 : 0;	44,507!
332	if (pool->pin_threads != 0) {	44,507✔
333	pool->hw_threads = sixel_get_hw_threads();	43,481✔
334	if (pool->hw_threads < 1) {	43,481!
335	pool->pin_threads = 0;	×
336	}
337	} else {	20,312✔
338	pool->hw_threads = 0;	1,026✔
339	}
340	sixel_mutex_unlock(&pool->mutex);	44,507✔
341	}	20,618✔
342
343	static int
344	threadpool_spawn_worker(threadpool_t pool, threadpool_worker_t worker)	120,613✔
345	{
346	int rc;	63,193✔
347
348	if (pool == NULL \|\| worker == NULL) {	120,613!
349	return SIXEL_BAD_ARGUMENT;
350	}
351	rc = sixel_thread_create(&worker->thread,	177,685✔
352	threadpool_worker_main,
353	worker);	57,072✔
354	if (rc != SIXEL_OK) {	120,611!
355	sixel_mutex_lock(&pool->mutex);	×
356	pool->shutting_down = 1;	×
357	sixel_cond_broadcast(&pool->cond_not_empty);	×
358	sixel_mutex_unlock(&pool->mutex);	×
359	return rc;	×
360	}
361	worker->started = 1;	120,611✔
362	pool->threads_started += 1;	120,611✔
363	return SIXEL_OK;	120,611✔
364	}	57,072✔
365
366	SIXELAPI void
367	threadpool_destroy(threadpool_t *pool)	45,013✔
368	{
369	if (pool == NULL) {	45,013✔
370	return;
371	}
372	threadpool_finish(pool);	45,013✔
373	threadpool_free(pool);	45,013✔
374	}	20,916✔
375
376	SIXELAPI void
377	threadpool_push(threadpool_t *pool, tp_job_t job)	313,458✔
378	{
379	if (pool == NULL) {	313,458✔
380	return;
381	}
382	sixel_mutex_lock(&pool->mutex);	313,458✔
383	if (pool->shutting_down) {	313,464!
UNCOV 384	sixel_mutex_unlock(&pool->mutex);	×
385	return;	×
386	}
387	while (pool->count == pool->qsize && !pool->shutting_down) {	334,758!
388	sixel_cond_wait(&pool->cond_not_full, &pool->mutex);	21,292✔
389	}
390	if (pool->shutting_down) {	313,466!
391	sixel_mutex_unlock(&pool->mutex);	×
392	return;	×
393	}
394	pool->jobs[pool->tail] = job;	313,466✔
395	pool->tail = (pool->tail + 1) % pool->qsize;	313,466✔
396	pool->count += 1;	313,466✔
397	sixel_cond_signal(&pool->cond_not_empty);	313,466✔
398	sixel_mutex_unlock(&pool->mutex);	313,463✔
399	}	144,305✔
400
401	SIXELAPI void
402	threadpool_finish(threadpool_t *pool)	90,026✔
403	{
404	int i;	47,092✔
405
406	if (pool == NULL) {	90,026✔
407	return;
408	}
409	sixel_mutex_lock(&pool->mutex);	90,026✔
410	if (pool->joined) {	90,026✔
411	sixel_mutex_unlock(&pool->mutex);	45,013✔
412	return;	45,013✔
413	}
414	pool->shutting_down = 1;	45,013✔
415	sixel_cond_broadcast(&pool->cond_not_empty);	45,013✔
416	sixel_cond_broadcast(&pool->cond_not_full);	45,013✔
417	while (pool->count > 0 \|\| pool->running > 0) {	79,883!
418	sixel_cond_wait(&pool->cond_drained, &pool->mutex);	34,870✔
419	}
420	sixel_mutex_unlock(&pool->mutex);	45,013✔
421
422	for (i = 0; i < pool->threads_started; ++i) {	177,713!
423	if (pool->workers[i] != NULL && pool->workers[i]->started) {	120,613!
424	sixel_thread_join(&pool->workers[i]->thread);	120,613✔
425	pool->workers[i]->started = 0;	120,613✔
426	}	57,072✔
427	}	57,072✔
428
429	sixel_mutex_lock(&pool->mutex);	45,013✔
430	pool->joined = 1;	45,013✔
431	sixel_mutex_unlock(&pool->mutex);	45,013✔
432	}	41,832!
433
434	SIXELAPI int
435	threadpool_grow(threadpool_t *pool, int additional_threads)	20,289✔
436	{
437	threadpool_worker_t **expanded;	11,049✔
438	int new_target;	11,049✔
439	int started_new;	11,049✔
440	int i;	11,049✔
441	int rc;	11,049✔
442
443	if (pool == NULL \|\| additional_threads <= 0) {	20,289!
444	return SIXEL_OK;
445	}
446
447	sixel_mutex_lock(&pool->mutex);	20,289✔
448	if (pool->shutting_down) {	20,289!
449	sixel_mutex_unlock(&pool->mutex);	×
450	return SIXEL_RUNTIME_ERROR;	×
451	}
452	new_target = pool->nthreads + additional_threads;	20,289✔
453	/*
454	* Worker structs stay heap-allocated per slot so pointer-table growth
455	* never invalidates addresses already held by running threads.
456	*/
457	if (new_target > pool->worker_capacity) {	20,289!
458	expanded = (threadpool_worker_t **)realloc(	20,289✔
459	pool->workers,	20,289✔
460	(size_t)new_target * sizeof(threadpool_worker_t *));	20,289✔
461	if (expanded == NULL) {	20,289✔
462	sixel_mutex_unlock(&pool->mutex);	×
463	return SIXEL_BAD_ALLOCATION;	×
464	}
465	memset(expanded + pool->worker_capacity,	21,330✔
466	0,
467	(size_t)(new_target - pool->worker_capacity)	9,322✔
468	* sizeof(threadpool_worker_t *));	8,281✔
469	pool->workers = expanded;	20,289✔
470	pool->worker_capacity = new_target;	20,289✔
471	}	12,008✔
472	sixel_mutex_unlock(&pool->mutex);	20,289✔
473
474	started_new = 0;	20,289✔
475	rc = SIXEL_OK;	20,289✔
476	for (i = pool->nthreads; i < new_target; ++i) {	44,459!
477	pool->workers[i] = (threadpool_worker_t *)	24,170✔
478	calloc(1, sizeof(threadpool_worker_t));	24,170✔
479	if (pool->workers[i] == NULL) {	24,170✔
480	rc = SIXEL_BAD_ALLOCATION;
481	break;
482	}
483	pool->workers[i]->pool = pool;	24,170✔
484	pool->workers[i]->workspace = NULL;	24,170✔
485	pool->workers[i]->started = 0;	24,170✔
486	pool->workers[i]->index = i;	24,170✔
487	pool->workers[i]->pinned = 0;	24,170✔
488	if (pool->workspace_size > 0) {	24,170!
489	pool->workers[i]->workspace =	24,170✔
490	calloc(1, pool->workspace_size);	24,170✔
491	if (pool->workers[i]->workspace == NULL) {	24,170✔
492	rc = SIXEL_BAD_ALLOCATION;
493	break;
494	}
495	}	12,564✔
496
497	rc = threadpool_spawn_worker(pool, pool->workers[i]);	24,170✔
498	if (rc != SIXEL_OK) {	24,170!
499	break;
500	}
501	started_new += 1;	24,170✔
502	}	12,564✔
503
504	if (rc != SIXEL_OK) {	18,765!
505	int j;
506
507	for (j = i; j < new_target; ++j) {	×
508	if (pool->workers[j] != NULL) {	×
509	if (pool->workers[j]->workspace != NULL) {	×
510	free(pool->workers[j]->workspace);	×
511	}
512	free(pool->workers[j]);	×
513	pool->workers[j] = NULL;	×
514	}
515	}
516	}
517
518	sixel_mutex_lock(&pool->mutex);	20,289✔
519	pool->nthreads = pool->nthreads + started_new;	20,289✔
520	sixel_mutex_unlock(&pool->mutex);	20,289✔
521
522	return rc;	20,289✔
523	}	12,008✔
524
525	SIXELAPI int
526	threadpool_get_error(threadpool_t *pool)	45,013✔
527	{
528	int error;	23,546✔
529
530	if (pool == NULL) {	45,013✔
531	return SIXEL_BAD_ARGUMENT;
532	}
533	sixel_mutex_lock(&pool->mutex);	45,013✔
534	error = pool->error;	45,013✔
535	sixel_mutex_unlock(&pool->mutex);	45,013✔
536	return error;	45,013✔
537	}	20,916✔
538
539	#endif /* SIXEL_ENABLE_THREADS */
540
541	/* emacs Local Variables: */
542	/* emacs mode: c */
543	/* emacs tab-width: 4 */
544	/* emacs indent-tabs-mode: nil */
545	/* emacs c-basic-offset: 4 */
546	/* emacs End: */
547	/* vim: set expandtab ts=4 sts=4 sw=4 : */
548	/* EOF */

saitoha / libsixel / 24945903988

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