1873

Committed 06 Mar 2025 06:19PM UTC coverage: 95.5% (+0.001%) from 95.499%

Build # 1873

Build Type

Pull #449

gitlab-ci

Committed by

johnsonjh

Commit Message

Add z/OS _sir_gettid implementation

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

Pull Request Pull Request #449: Portability improvements

Run Details

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

1 existing line in 1 file now uncovered.

3693 of 3867 relevant lines covered (95.5%)

355104.5 hits per line

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

89.42

/src/sirthreadpool.c

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

aremmell / libsir / 1873

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