1153

Committed 25 Jul 2023 04:30AM UTC coverage: 80.118% (-14.7%) from 94.801%

Build # 1153

Build Type

push

travis-ci

Committed by

aremmell

Commit Message

fix ternary

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

2571 of 3209 relevant lines covered (80.12%)

24530.87 hits per line

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

81.4

/src/sirthreadpool.c

/*
 * sirthreadpool.c
 *
 * Author:    Ryan M. Lederman <lederman@gmail.com>
 * Copyright: Copyright (c) 2018-2023
 * Version:   2.2.1
 * 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 thread_pool_proc(void* arg);
#endif

bool _sir_threadpool_create(sir_threadpool** pool, size_t num_threads) {
    if (!pool || !num_threads)
        return false;

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

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

    (*pool)->num_threads = num_threads;

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

    for (size_t n = 0; n < num_threads; n++) {
#if !defined(__WIN__)
        pthread_attr_t attr;
        pthread_attr_init(&attr);
        int op = pthread_create(&(*pool)->threads[n], &attr, &thread_pool_proc, *pool);
        if (0 != op) {
            (*pool)->threads[n] = 0;
            _sir_handleerr(op);
            _sir_threadpool_destroy(pool);
            return false;
        }

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

    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 false;

    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);
    }

    for (size_t n = 0; n < (*pool)->num_threads; n++) {
        if ((*pool)->threads[n]) {
            _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_UNUSED(0 == join, join);
#else /* __WIN__ */
            DWORD join = WaitForSingleObject((*pool)->threads[n], INFINITE);
            SIR_ASSERT_UNUSED(WAIT_OBJECT_0 == join, join);
#endif
        }
    }

    bool destroy = _sir_queue_destroy(&(*pool)->jobs);
    SIR_ASSERT_UNUSED(destroy, destroy);

    destroy = _sir_conddestroy(&(*pool)->cond);
    SIR_ASSERT_UNUSED(destroy, destroy);

    destroy = _sir_mutexdestroy(&(*pool)->mutex);
    SIR_ASSERT_UNUSED(destroy, destroy);

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

    return true;
}

#if !defined(__WIN__)
static void* thread_pool_proc(void* arg)
#else
static unsigned 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 0;
#endif
}

1	/*
2	* sirthreadpool.c
3	*
4	* Author: Ryan M. Lederman <lederman@gmail.com>
5	* Copyright: Copyright (c) 2018-2023
6	* Version: 2.2.1
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 thread_pool_proc(void* arg);
36	#endif
37
38	bool _sir_threadpool_create(sir_threadpool** pool, size_t num_threads) {	3✔
39	if (!pool \|\| !num_threads)	3✔
40	return false;	×
41
42	*pool = calloc(1, sizeof(sir_threadpool));	3✔
43	if (!*pool) {	3✔
44	_sir_handleerr(errno);	×
45	return false;	×
46	}
47
48	(*pool)->threads = calloc(num_threads, sizeof(sir_thread));	3✔
49	if (!(*pool)->threads) {	3✔
50	_sir_handleerr(errno);	×
51	_sir_threadpool_destroy(pool);	×
52	return false;	×
53	}
54
55	(*pool)->num_threads = num_threads;	3✔
56
57	if (!_sir_queue_create(&(pool)->jobs) \|\| !_sir_condcreate(&(pool)->cond) \|\|	6✔
58	!_sir_mutexcreate(&(*pool)->mutex)) {	3✔
59	_sir_threadpool_destroy(pool);	×
60	return false;	×
61	}
62
63	for (size_t n = 0; n < num_threads; n++) {	15✔
64	#if !defined(__WIN__)
65	pthread_attr_t attr;
66	pthread_attr_init(&attr);	12✔
67	int op = pthread_create(&(pool)->threads[n], &attr, &thread_pool_proc, pool);	12✔
68	if (0 != op) {	12✔
69	(*pool)->threads[n] = 0;	×
70	_sir_handleerr(op);	×
71	_sir_threadpool_destroy(pool);	×
72	return false;	×
73	}
74
75	#else /* __WIN__ */
76	(*pool)->threads[n] = (HANDLE)_beginthreadex(NULL, 0, &thread_pool_proc,
77	*pool, 0, NULL);
78	if (!(*pool)->threads[n]) {
79	_sir_handleerr(errno);
80	_sir_threadpool_destroy(pool);
81	return false;
82	}
83	#endif
84	}
85
86	return !!*pool;	3✔
87	}
88
89	bool _sir_threadpool_add_job(sir_threadpool* pool, sir_threadpool_job* job) {	90✔
90	bool retval = false;	×
91
92	if (pool && pool->jobs && job && job->fn && job->data) {	90✔
93	bool locked = _sir_mutexlock(&pool->mutex);	90✔
94	SIR_ASSERT(locked);	60✔
95
96	if (locked) {	90✔
97	if (_sir_queue_push(pool->jobs, job)) {	90✔
98	retval = _sir_condbroadcast(&pool->cond);	90✔
99	_sir_selflog("added job; new size: %zu", _sir_queue_size(pool->jobs));	90✔
100	}
101
102	bool unlocked = _sir_mutexunlock(&pool->mutex);	90✔
103	SIR_ASSERT_UNUSED(unlocked, unlocked);	60✔
104	}
105	}
106
107	return retval;	90✔
108	}
109
110	bool _sir_threadpool_destroy(sir_threadpool** pool) {	3✔
111	if (!pool \|\| !*pool)	3✔
112	return false;	×
113
114	bool locked = _sir_mutexlock(&(*pool)->mutex);	3✔
115	SIR_ASSERT(locked);	2✔
116
117	if (locked) {	3✔
118	_sir_selflog("broadcasting signal to condition var...");	2✔
119	(*pool)->cancel = true;	3✔
120
121	bool bcast = _sir_condbroadcast(&(*pool)->cond);	3✔
122	SIR_ASSERT_UNUSED(bcast, bcast);	2✔
123
124	bool unlock = _sir_mutexunlock(&(*pool)->mutex);	3✔
125	SIR_ASSERT_UNUSED(unlock, unlock);	2✔
126	}
127
128	for (size_t n = 0; n < (*pool)->num_threads; n++) {	15✔
129	if ((*pool)->threads[n]) {	12✔
130	_sir_selflog("joining thread %zu of %zu...", n + 1, (*pool)->num_threads);	8✔
131	#if !defined(__WIN__)
132	int join = pthread_join((*pool)->threads[n], NULL);	12✔
133	SIR_ASSERT_UNUSED(0 == join, join);	8✔
134	#else /* __WIN__ */
135	DWORD join = WaitForSingleObject((*pool)->threads[n], INFINITE);
136	SIR_ASSERT_UNUSED(WAIT_OBJECT_0 == join, join);
137	#endif
138	}
139	}
140
141	bool destroy = _sir_queue_destroy(&(*pool)->jobs);	3✔
142	SIR_ASSERT_UNUSED(destroy, destroy);	2✔
143
144	destroy = _sir_conddestroy(&(*pool)->cond);	3✔
145	SIR_ASSERT_UNUSED(destroy, destroy);	2✔
146
147	destroy = _sir_mutexdestroy(&(*pool)->mutex);	3✔
148	SIR_ASSERT_UNUSED(destroy, destroy);	2✔
149
150	_sir_safefree(&(*pool)->threads);	3✔
151	_sir_safefree(pool);	3✔
152
153	return true;	3✔
154	}
155
156	#if !defined(__WIN__)
157	static void* thread_pool_proc(void* arg)	12✔
158	#else
159	static unsigned thread_pool_proc(void* arg)
160	#endif
161	{
162	sir_threadpool* pool = (sir_threadpool*)arg;	×
163	while (true) {	21✔
164	bool locked = _sir_mutexlock(&pool->mutex);	33✔
165	SIR_ASSERT_UNUSED(locked, locked);	22✔
166
167	while (_sir_queue_isempty(pool->jobs) && !pool->cancel) {	43✔
168	#if !defined(__WIN__)
169	/* seconds; absolute fixed time. */
170	sir_wait wait = {time(NULL) + 2, 0};	10✔
171	#else
172	/* msec; relative from now. */
173	sir_wait wait = 2000;
174	#endif
175	(void)_sir_condwait_timeout(&pool->cond, &pool->mutex, &wait);	10✔
176	}
177
178	if (!pool->cancel) {	33✔
179	sir_threadpool_job* job = NULL;	21✔
180	bool job_popped = _sir_queue_pop(pool->jobs, (void**)&job);	21✔
181
182	bool unlocked = _sir_mutexunlock(&pool->mutex);	21✔
183	SIR_ASSERT_UNUSED(unlocked, unlocked);	14✔
184
185	if (job_popped) {	21✔
186	_sir_selflog("picked up job (fn: %"PRIxPTR", data: %p)",	21✔
187	(uintptr_t)job->fn, job->data);
188	job->fn(job->data);	21✔
189	_sir_safefree(&job);	19✔
190	}
191	} else {
192	_sir_selflog("cancel flag is set; exiting");	8✔
193	bool unlocked = _sir_mutexunlock(&pool->mutex);	12✔
194	SIR_ASSERT_UNUSED(unlocked, unlocked);	8✔
195	break;	×
196	}
197	}
198
199	#if !defined(__WIN__)
200	return NULL;	12✔
201	#else /* __WIN__ */
202	return 0;
203	#endif
204	}

aremmell / libsir / 1153

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