aremmell / libsir / 536

Committed 07 Sep 2023 10:52PM UTC coverage: 94.481% (-0.2%) from 94.711%

Build # 536

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Coverity: Fix for unchecked return values (#261)

* Coverity: Fix for unchecked return values

---------

Signed-off-by: Jeffrey H. Johnson <trnsz@pobox.com>
Signed-off-by: Ryan M. Lederman <lederman@gmail.com>
Co-authored-by: Ryan M. Lederman <lederman@gmail.com>

Run Details

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

3133 of 3316 relevant lines covered (94.48%)

600260.38 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	#else /* __WIN__ */
84	(*pool)->threads[n] = (HANDLE)_beginthreadex(NULL, 0, &thread_pool_proc,
85	*pool, 0, NULL);
86	if (!(*pool)->threads[n]) {
87	thrd_err = errno;
88	thrd_create = false;
89	break;
90	}
91	#endif
92	}
93
94	#if !defined(__WIN__)
95	op = pthread_attr_destroy(&attr);	22✔
96	SIR_ASSERT_UNUSED(0 == op, op);	21✔
97	#endif
98
99	if (!thrd_create) {	22✔
100	bool destroy = _sir_threadpool_destroy(pool);	1✔
101	SIR_ASSERT_UNUSED(destroy, destroy);	1✔
102	return _sir_handleerr(thrd_err);	1✔
103	}
104
105	return !!*pool;	21✔
106	}
107
108	bool _sir_threadpool_add_job(sir_threadpool* pool, sir_threadpool_job* job) {	630✔
109	bool retval = false;	540✔
110
111	if (pool && pool->jobs && job && job->fn && job->data) {	630✔
112	bool locked = _sir_mutexlock(&pool->mutex);	630✔
113	SIR_ASSERT(locked);	600✔
114
115	if (locked) {	570✔
116	if (_sir_queue_push(pool->jobs, job)) {	630✔
117	retval = _sir_condbroadcast(&pool->cond);	630✔
118	_sir_selflog("added job; new size: %zu", _sir_queue_size(pool->jobs));	630✔
119	}
120
121	bool unlocked = _sir_mutexunlock(&pool->mutex);	630✔
122	SIR_ASSERT_UNUSED(unlocked, unlocked);	600✔
123	}
124	}
125
126	return retval;	630✔
127	}
128
129	bool _sir_threadpool_destroy(sir_threadpool** pool) {	22✔
130	if (!pool \|\| !*pool)	22✔
131	return _sir_seterror(_SIR_E_INVALID);	×
132
133	bool locked = _sir_mutexlock(&(*pool)->mutex);	22✔
134	SIR_ASSERT(locked);	21✔
135
136	if (locked) {	20✔
137	_sir_selflog("broadcasting signal to condition var...");	21✔
138	(*pool)->cancel = true;	22✔
139
140	bool bcast = _sir_condbroadcast(&(*pool)->cond);	22✔
141	SIR_ASSERT_UNUSED(bcast, bcast);	21✔
142
143	bool unlock = _sir_mutexunlock(&(*pool)->mutex);	22✔
144	SIR_ASSERT_UNUSED(unlock, unlock);	21✔
145	}
146
147	bool destroy = true;	19✔
148	for (size_t n = 0; n < (*pool)->num_threads; n++) {	110✔
149	SIR_ASSERT(0 != (*pool)->threads[n]);	84✔
150	if (0 == (*pool)->threads[n])	88✔
151	continue;	4✔
152	_sir_selflog("joining thread %zu of %zu...", n + 1, (*pool)->num_threads);	80✔
153	#if !defined(__WIN__)
154	int join = pthread_join((*pool)->threads[n], NULL);	84✔
155	SIR_ASSERT(0 == join);	80✔
156	_sir_eqland(destroy, 0 == join);	84✔
157	#else /* __WIN__ */
158	DWORD join = WaitForSingleObject((*pool)->threads[n], INFINITE);
159	SIR_ASSERT(WAIT_OBJECT_0 == join);
160	_sir_eqland(destroy, WAIT_OBJECT_0 == join);
161	#endif
162	}
163
164	_sir_eqland(destroy, _sir_queue_destroy(&(*pool)->jobs));	22✔
165	SIR_ASSERT(destroy);	21✔
166
167	_sir_eqland(destroy, _sir_conddestroy(&(*pool)->cond));	22✔
168	SIR_ASSERT(destroy);	21✔
169
170	_sir_eqland(destroy, _sir_mutexdestroy(&(*pool)->mutex));	22✔
171	SIR_ASSERT(destroy);	21✔
172
173	_sir_safefree(&(*pool)->threads);	22✔
174	_sir_safefree(pool);	22✔
175
176	return destroy;	22✔
177	}
178
179	#if !defined(__WIN__)
180	static void* thread_pool_proc(void* arg)	84✔
181	#else
182	static unsigned __stdcall thread_pool_proc(void* arg)
183	#endif
184	{
185	sir_threadpool* pool = (sir_threadpool*)arg;	72✔
186	while (true) {	138✔
187	bool locked = _sir_mutexlock(&pool->mutex);	222✔
188	SIR_ASSERT_UNUSED(locked, locked);	210✔
189
190	while (_sir_queue_isempty(pool->jobs) && !pool->cancel) {	342✔
191	#if !defined(__WIN__)
192	/* seconds; absolute fixed time. */
193	sir_wait wait = {time(NULL) + 2, 0};	120✔
194	#else
195	/* msec; relative from now. */
196	sir_wait wait = 2000;
197	#endif
198	(void)_sir_condwait_timeout(&pool->cond, &pool->mutex, &wait);	120✔
199	}
200
201	if (!pool->cancel) {	222✔
202	sir_threadpool_job* job = NULL;	138✔
203	bool job_popped = _sir_queue_pop(pool->jobs, (void**)&job);	138✔
204
205	bool unlocked = _sir_mutexunlock(&pool->mutex);	138✔
206	SIR_ASSERT_UNUSED(unlocked, unlocked);	130✔
207
208	if (job_popped) {	138✔
209	_sir_selflog("picked up job (fn: %"PRIxPTR", data: %p)",	138✔
210	(uintptr_t)job->fn, job->data);
211	job->fn(job->data);	138✔
212	_sir_safefree(&job);	138✔
213	}
214	} else {
215	_sir_selflog("cancel flag is set; exiting");	80✔
216	bool unlocked = _sir_mutexunlock(&pool->mutex);	84✔
217	SIR_ASSERT_UNUSED(unlocked, unlocked);	80✔
218	break;	72✔
219	}
220	}
221
222	#if !defined(__WIN__)
223	return NULL;	84✔
224	#else /* __WIN__ */
225	return 0U;
226	#endif
227	}

aremmell / libsir / 536

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