aremmell / libsir / 523

Committed 07 Sep 2023 09:58PM UTC coverage: 94.624% (-0.09%) from 94.711%

Build # 523

Build Type

Pull #263

gitlab-ci

Committed by

web-flow

Commit Message

Merge branch 'master' into johnsonjh/20230906/cbmc1

Signed-off-by: Jeffrey H. Johnson <trnsz@pobox.com>

Pull Request Pull Request #263: Thread pool fixes

Run Details

50 of 50 new or added lines in 2 files covered. (100.0%)

3133 of 3311 relevant lines covered (94.62%)

600587.25 hits per line

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

89.32

/src/sirthreadpool.c

/*
 * sirthreadpool.c
 *
 * Author:    Ryan M. Lederman <lederman@gmail.com>
 * Copyright: Copyright (c) 2018-2023
 * Version:   2.2.3
 * License:   The MIT License (MIT)
 *
 * 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.
 */
#include "sir/condition.h"
#include "sir/threadpool.h"
#include "sir/internal.h"
#include "sir/queue.h"
#include "sir/mutex.h"

#if !defined(__WIN__)
static void* thread_pool_proc(void* arg);
#else
static unsigned __stdcall thread_pool_proc(void* arg);
#endif

bool _sir_threadpool_create(sir_threadpool** pool, size_t num_threads) {
    if (!pool || !num_threads || num_threads > SIR_THREADPOOL_MAX_THREADS)
        return _sir_seterror(_SIR_E_INVALID);

    *pool = calloc(1, sizeof(sir_threadpool));
    if (!*pool)
        return _sir_handleerr(errno);

    (*pool)->threads = calloc(num_threads, sizeof(sir_thread));
    if (!(*pool)->threads) {
        int err = errno;
        _sir_safefree(pool);
        return _sir_handleerr(err);
    }

    (*pool)->num_threads = num_threads;

    if (!_sir_queue_create(&(*pool)->jobs) || !_sir_condcreate(&(*pool)->cond) ||
        !_sir_mutexcreate(&(*pool)->mutex)) {
        bool destroy = _sir_threadpool_destroy(pool);
        SIR_ASSERT_UNUSED(destroy, destroy);
        return false;
    }

#if !defined(__WIN__)
    pthread_attr_t attr = {0};
    int op = pthread_attr_init(&attr);
    if (0 != op) {
        bool destroy = _sir_threadpool_destroy(pool);
        SIR_ASSERT_UNUSED(destroy, destroy);
        return _sir_handleerr(op);
    }
#endif

    int thrd_err     = 0;
    bool thrd_create = true;
    for (size_t n = 0; n < num_threads; n++) {
#if !defined(__WIN__)
        op = pthread_create(&(*pool)->threads[n], &attr, &thread_pool_proc, *pool);
        if (0 != op) {
            (*pool)->threads[n] = 0;
            thrd_err    = op;
            thrd_create = false;
            break;
        }

#else /* __WIN__ */
        (*pool)->threads[n] = (HANDLE)_beginthreadex(NULL, 0, &thread_pool_proc,
            *pool, 0, NULL);
        if (!(*pool)->threads[n]) {
            thrd_err    = errno;
            thrd_create = false;
            break;
        }
#endif
    }

#if !defined(__WIN__)
    op = pthread_attr_destroy(&attr);
    SIR_ASSERT_UNUSED(0 == op, op);
#endif

    if (!thrd_create) {
        bool destroy = _sir_threadpool_destroy(pool);
        SIR_ASSERT_UNUSED(destroy, destroy);
        return _sir_handleerr(thrd_err);
    }

    return !!*pool;
}

bool _sir_threadpool_add_job(sir_threadpool* pool, sir_threadpool_job* job) {
    bool retval = false;

    if (pool && pool->jobs && job && job->fn && job->data) {
        bool locked = _sir_mutexlock(&pool->mutex);
        SIR_ASSERT(locked);

        if (locked) {
            if (_sir_queue_push(pool->jobs, job)) {
                retval = _sir_condbroadcast(&pool->cond);
                _sir_selflog("added job; new size: %zu", _sir_queue_size(pool->jobs));
            }

            bool unlocked = _sir_mutexunlock(&pool->mutex);
            SIR_ASSERT_UNUSED(unlocked, unlocked);
        }
    }

    return retval;
}

bool _sir_threadpool_destroy(sir_threadpool** pool) {
    if (!pool || !*pool)
        return _sir_seterror(_SIR_E_INVALID);

    bool locked = _sir_mutexlock(&(*pool)->mutex);
    SIR_ASSERT(locked);

    if (locked) {
        _sir_selflog("broadcasting signal to condition var...");
        (*pool)->cancel = true;

        bool bcast = _sir_condbroadcast(&(*pool)->cond);
        SIR_ASSERT_UNUSED(bcast, bcast);

        bool unlock = _sir_mutexunlock(&(*pool)->mutex);
        SIR_ASSERT_UNUSED(unlock, unlock);
    }

    bool destroy = true;
    for (size_t n = 0; n < (*pool)->num_threads; n++) {
        SIR_ASSERT(0 != (*pool)->threads[n]);
        if (0 == (*pool)->threads[n])
            continue;
        _sir_selflog("joining thread %zu of %zu...", n + 1, (*pool)->num_threads);
#if !defined(__WIN__)
        int join = pthread_join((*pool)->threads[n], NULL);
        SIR_ASSERT(0 == join);
        _sir_eqland(destroy, 0 == join);
#else /* __WIN__ */
        DWORD join = WaitForSingleObject((*pool)->threads[n], INFINITE);
        SIR_ASSERT(WAIT_OBJECT_0 == join);
        _sir_eqland(destroy, WAIT_OBJECT_0 == join);
#endif
    }

    _sir_eqland(destroy, _sir_queue_destroy(&(*pool)->jobs));
    SIR_ASSERT(destroy);

    _sir_eqland(destroy, _sir_conddestroy(&(*pool)->cond));
    SIR_ASSERT(destroy);

    _sir_eqland(destroy, _sir_mutexdestroy(&(*pool)->mutex));
    SIR_ASSERT(destroy);

    _sir_safefree(&(*pool)->threads);
    _sir_safefree(pool);

    return destroy;
}

#if !defined(__WIN__)
static void* thread_pool_proc(void* arg)
#else
static unsigned __stdcall thread_pool_proc(void* arg)
#endif
{
    sir_threadpool* pool = (sir_threadpool*)arg;
    while (true) {
        bool locked = _sir_mutexlock(&pool->mutex);
        SIR_ASSERT_UNUSED(locked, locked);

        while (_sir_queue_isempty(pool->jobs) && !pool->cancel) {
#if !defined(__WIN__)
            /* seconds; absolute fixed time. */
            sir_wait wait = {time(NULL) + 2, 0};
#else
            /* msec; relative from now. */
            sir_wait wait = 2000;
#endif
            (void)_sir_condwait_timeout(&pool->cond, &pool->mutex, &wait);
        }

        if (!pool->cancel) {
            sir_threadpool_job* job = NULL;
            bool job_popped         = _sir_queue_pop(pool->jobs, (void**)&job);

            bool unlocked = _sir_mutexunlock(&pool->mutex);
            SIR_ASSERT_UNUSED(unlocked, unlocked);

            if (job_popped) {
                _sir_selflog("picked up job (fn: %"PRIxPTR", data: %p)",
                    (uintptr_t)job->fn, job->data);
                job->fn(job->data);
                _sir_safefree(&job);
            }
        } else {
            _sir_selflog("cancel flag is set; exiting");
            bool unlocked = _sir_mutexunlock(&pool->mutex);
            SIR_ASSERT_UNUSED(unlocked, unlocked);
            break;
        }
    }

#if !defined(__WIN__)
    return NULL;
#else /* __WIN__ */
    return 0U;
#endif
}

1	/*
2	* sirthreadpool.c
3	*
4	* Author: Ryan M. Lederman <lederman@gmail.com>
5	* Copyright: Copyright (c) 2018-2023
6	* Version: 2.2.3
7	* License: The MIT License (MIT)
8	*
9	* Permission is hereby granted, free of charge, to any person obtaining a copy of
10	* this software and associated documentation files (the "Software"), to deal in
11	* the Software without restriction, including without limitation the rights to
12	* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
13	* the Software, and to permit persons to whom the Software is furnished to do so,
14	* subject to the following conditions:
15	*
16	* The above copyright notice and this permission notice shall be included in all
17	* copies or substantial portions of the Software.
18	*
19	* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
20	* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
21	* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
22	* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
23	* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
24	* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
25	*/
26	#include "sir/condition.h"
27	#include "sir/threadpool.h"
28	#include "sir/internal.h"
29	#include "sir/queue.h"
30	#include "sir/mutex.h"
31
32	#if !defined(__WIN__)
33	static void* thread_pool_proc(void* arg);
34	#else
35	static unsigned __stdcall thread_pool_proc(void* arg);
36	#endif
37
38	bool _sir_threadpool_create(sir_threadpool** pool, size_t num_threads) {	23✔
39	if (!pool \|\| !num_threads \|\| num_threads > SIR_THREADPOOL_MAX_THREADS)	23✔
40	return _sir_seterror(_SIR_E_INVALID);	×
41
42	*pool = calloc(1, sizeof(sir_threadpool));	23✔
43	if (!*pool)	23✔
44	return _sir_handleerr(errno);	1✔
45
46	(*pool)->threads = calloc(num_threads, sizeof(sir_thread));	22✔
47	if (!(*pool)->threads) {	22✔
48	int err = errno;	×
49	_sir_safefree(pool);	×
50	return _sir_handleerr(err);	×
51	}
52
53	(*pool)->num_threads = num_threads;	22✔
54
55	if (!_sir_queue_create(&(pool)->jobs) \|\| !_sir_condcreate(&(pool)->cond) \|\|	25✔
56	!_sir_mutexcreate(&(*pool)->mutex)) {	22✔
57	bool destroy = _sir_threadpool_destroy(pool);	×
58	SIR_ASSERT_UNUSED(destroy, destroy);	×
59	return false;	×
60	}
61
62	#if !defined(__WIN__)
63	pthread_attr_t attr = {0};	22✔
64	int op = pthread_attr_init(&attr);	22✔
65	if (0 != op) {	22✔
66	bool destroy = _sir_threadpool_destroy(pool);	×
67	SIR_ASSERT_UNUSED(destroy, destroy);	×
68	return _sir_handleerr(op);	×
69	}
70	#endif
71
72	int thrd_err = 0;	19✔
73	bool thrd_create = true;	19✔
74	for (size_t n = 0; n < num_threads; n++) {	106✔
75	#if !defined(__WIN__)
76	op = pthread_create(&(pool)->threads[n], &attr, &thread_pool_proc, pool);	85✔
77	if (0 != op) {	85✔
78	(*pool)->threads[n] = 0;	1✔
79	thrd_err = op;	1✔
80	thrd_create = false;	1✔
81	break;	1✔
82	}
83
84	#else /* __WIN__ */
85	(*pool)->threads[n] = (HANDLE)_beginthreadex(NULL, 0, &thread_pool_proc,
86	*pool, 0, NULL);
87	if (!(*pool)->threads[n]) {
88	thrd_err = errno;
89	thrd_create = false;
90	break;
91	}
92	#endif
93	}
94
95	#if !defined(__WIN__)
96	op = pthread_attr_destroy(&attr);	22✔
97	SIR_ASSERT_UNUSED(0 == op, op);	21✔
98	#endif
99
100	if (!thrd_create) {	22✔
101	bool destroy = _sir_threadpool_destroy(pool);	1✔
102	SIR_ASSERT_UNUSED(destroy, destroy);	1✔
103	return _sir_handleerr(thrd_err);	1✔
104	}
105
106	return !!*pool;	21✔
107	}
108
109	bool _sir_threadpool_add_job(sir_threadpool* pool, sir_threadpool_job* job) {	630✔
110	bool retval = false;	540✔
111
112	if (pool && pool->jobs && job && job->fn && job->data) {	630✔
113	bool locked = _sir_mutexlock(&pool->mutex);	630✔
114	SIR_ASSERT(locked);	600✔
115
116	if (locked) {	570✔
117	if (_sir_queue_push(pool->jobs, job)) {	630✔
118	retval = _sir_condbroadcast(&pool->cond);	630✔
119	_sir_selflog("added job; new size: %zu", _sir_queue_size(pool->jobs));	630✔
120	}
121
122	bool unlocked = _sir_mutexunlock(&pool->mutex);	630✔
123	SIR_ASSERT_UNUSED(unlocked, unlocked);	600✔
124	}
125	}
126
127	return retval;	630✔
128	}
129
130	bool _sir_threadpool_destroy(sir_threadpool** pool) {	22✔
131	if (!pool \|\| !*pool)	22✔
132	return _sir_seterror(_SIR_E_INVALID);	×
133
134	bool locked = _sir_mutexlock(&(*pool)->mutex);	22✔
135	SIR_ASSERT(locked);	21✔
136
137	if (locked) {	20✔
138	_sir_selflog("broadcasting signal to condition var...");	21✔
139	(*pool)->cancel = true;	22✔
140
141	bool bcast = _sir_condbroadcast(&(*pool)->cond);	22✔
142	SIR_ASSERT_UNUSED(bcast, bcast);	21✔
143
144	bool unlock = _sir_mutexunlock(&(*pool)->mutex);	22✔
145	SIR_ASSERT_UNUSED(unlock, unlock);	21✔
146	}
147
148	bool destroy = true;	19✔
149	for (size_t n = 0; n < (*pool)->num_threads; n++) {	110✔
150	SIR_ASSERT(0 != (*pool)->threads[n]);	84✔
151	if (0 == (*pool)->threads[n])	88✔
152	continue;	4✔
153	_sir_selflog("joining thread %zu of %zu...", n + 1, (*pool)->num_threads);	80✔
154	#if !defined(__WIN__)
155	int join = pthread_join((*pool)->threads[n], NULL);	84✔
156	SIR_ASSERT(0 == join);	80✔
157	_sir_eqland(destroy, 0 == join);	84✔
158	#else /* __WIN__ */
159	DWORD join = WaitForSingleObject((*pool)->threads[n], INFINITE);
160	SIR_ASSERT(WAIT_OBJECT_0 == join);
161	_sir_eqland(destroy, WAIT_OBJECT_0 == join);
162	#endif
163	}
164
165	_sir_eqland(destroy, _sir_queue_destroy(&(*pool)->jobs));	22✔
166	SIR_ASSERT(destroy);	21✔
167
168	_sir_eqland(destroy, _sir_conddestroy(&(*pool)->cond));	22✔
169	SIR_ASSERT(destroy);	21✔
170
171	_sir_eqland(destroy, _sir_mutexdestroy(&(*pool)->mutex));	22✔
172	SIR_ASSERT(destroy);	21✔
173
174	_sir_safefree(&(*pool)->threads);	22✔
175	_sir_safefree(pool);	22✔
176
177	return destroy;	22✔
178	}
179
180	#if !defined(__WIN__)
181	static void* thread_pool_proc(void* arg)	84✔
182	#else
183	static unsigned __stdcall thread_pool_proc(void* arg)
184	#endif
185	{
186	sir_threadpool* pool = (sir_threadpool*)arg;	72✔
187	while (true) {	165✔
188	bool locked = _sir_mutexlock(&pool->mutex);	249✔
189	SIR_ASSERT_UNUSED(locked, locked);	237✔
190
191	while (_sir_queue_isempty(pool->jobs) && !pool->cancel) {	417✔
192	#if !defined(__WIN__)
193	/* seconds; absolute fixed time. */
194	sir_wait wait = {time(NULL) + 2, 0};	168✔
195	#else
196	/* msec; relative from now. */
197	sir_wait wait = 2000;
198	#endif
199	(void)_sir_condwait_timeout(&pool->cond, &pool->mutex, &wait);	168✔
200	}
201
202	if (!pool->cancel) {	249✔
203	sir_threadpool_job* job = NULL;	165✔
204	bool job_popped = _sir_queue_pop(pool->jobs, (void**)&job);	165✔
205
206	bool unlocked = _sir_mutexunlock(&pool->mutex);	165✔
207	SIR_ASSERT_UNUSED(unlocked, unlocked);	157✔
208
209	if (job_popped) {	165✔
210	_sir_selflog("picked up job (fn: %"PRIxPTR", data: %p)",	165✔
211	(uintptr_t)job->fn, job->data);
212	job->fn(job->data);	165✔
213	_sir_safefree(&job);	165✔
214	}
215	} else {
216	_sir_selflog("cancel flag is set; exiting");	80✔
217	bool unlocked = _sir_mutexunlock(&pool->mutex);	84✔
218	SIR_ASSERT_UNUSED(unlocked, unlocked);	80✔
219	break;	72✔
220	}
221	}
222
223	#if !defined(__WIN__)
224	return NULL;	84✔
225	#else /* __WIN__ */
226	return 0U;
227	#endif
228	}

aremmell / libsir / 523

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