16253909665

Committed 13 Jul 2025 10:04PM UTC coverage: 93.07% (-0.3%) from 93.378%

Build # 16253909665

Build Type

push

github

Committed by

krakjoe

Commit Message

gcov is not re-entrant

Run Details

5144 of 5527 relevant lines covered (93.07%)

68741.08 hits per line

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

93.29

/src/maths/pool.c

/*
  +----------------------------------------------------------------------+
  | ort                                                                  |
  +----------------------------------------------------------------------+
  | Copyright (c) Joe Watkins 2025                                       |
  +----------------------------------------------------------------------+
  | This source file is subject to version 3.01 of the PHP license,      |
  | that is bundled with this package in the file LICENSE, and is        |
  | available through the world-wide-web at the following url:           |
  | http://www.php.net/license/3_01.txt                                  |
  | If you did not receive a copy of the PHP license and are unable to   |
  | obtain it through the world-wide-web, please send a note to          |
  | license@php.net so we can mail you a copy immediately.               |
  +----------------------------------------------------------------------+
  | Author: krakjoe                                                      |
  +----------------------------------------------------------------------+
 */

#include "maths/cast.h"
#include "maths/pool.h"
#include "maths/result.h"

#if defined(_WIN32)
# include <windows.h>
# define ORT_POOL_LOCAL __declspec(thread)
  typedef HANDLE ort_thread_t;
  typedef CRITICAL_SECTION ort_mutex_t;
  typedef CONDITION_VARIABLE ort_cond_t;
#define ort_pool_mutex_init(mutex) \
    InitializeCriticalSection(mutex)
#define ort_pool_mutex_lock(mutex) \
    EnterCriticalSection(mutex)
#define ort_pool_mutex_unlock(mutex) \
    LeaveCriticalSection(mutex)
#define ort_pool_mutex_destroy(mutex) \
    DeleteCriticalSection(mutex)
#define ort_pool_cond_init(cond) \
    InitializeConditionVariable(cond)
#define ort_pool_cond_wait(cond, mutex) \
    SleepConditionVariableCS(cond, mutex, INFINITE)
#define ort_pool_cond_signal(cond) \
    WakeConditionVariable(cond)
#define ort_pool_cond_broadcast(cond) \
    WakeAllConditionVariable(cond)
#define ort_pool_cond_destroy(cond)
#else
# include <pthread.h>
# include <unistd.h>
# define ORT_POOL_LOCAL __thread
  typedef pthread_t ort_thread_t;
  typedef pthread_mutex_t ort_mutex_t;
  typedef pthread_cond_t ort_cond_t;
#define ort_pool_mutex_init(mutex) \
    pthread_mutex_init(mutex, NULL)
#define ort_pool_mutex_lock(mutex) \
    pthread_mutex_lock(mutex)
#define ort_pool_mutex_unlock(mutex) \
    pthread_mutex_unlock(mutex)
#define ort_pool_mutex_destroy(mutex) \
    pthread_mutex_destroy(mutex)
#define ort_pool_cond_init(cond) \
    pthread_cond_init(cond, NULL)
#define ort_pool_cond_wait(cond, mutex) \
    pthread_cond_wait(cond, mutex)
#define ort_pool_cond_signal(cond) \
    pthread_cond_signal(cond)
#define ort_pool_cond_broadcast(cond) \
    pthread_cond_broadcast(cond)
#define ort_pool_cond_destroy(cond) \
    pthread_cond_destroy(cond)
#endif

typedef struct _ort_task_t {
    ort_task_func_t func;
    void *arg;
    size_t count;
    volatile size_t next;
    volatile size_t completed;
    ort_mutex_t mutex;
    ort_cond_t cond;
    volatile int done;
    volatile int refcount;
} ort_task_t;

typedef struct _ort_pool_t {
    ort_thread_t *threads;
    size_t     size;
    ort_task_t *task;
    ort_mutex_t mutex;
    ort_cond_t cond;
    int stop;
} ort_pool_t;

ORT_POOL_LOCAL ort_pool_t ort_maths_pool;

void ort_pool_binary_worker(void *arg, size_t index, size_t count) {
    ort_pool_binary_ctx_t *ctx =
        (ort_pool_binary_ctx_t *)arg;

    size_t chunk = ctx->layout.chunk;
    size_t start = index * chunk;
    size_t end   = start + chunk;

    if (end > ctx->layout.total)
        end = ctx->layout.total;
    
    size_t n = end - start;
    if (n > 0) {
        ctx->op(
            (char*)ctx->result    + start * ctx->layout.element,
            (char*)ctx->a         + start * ctx->layout.element,
            (char*)ctx->b         + start * ctx->layout.element,
            n
        );
    }
}

void ort_pool_unary_worker(void *arg, size_t index, size_t count) {
    ort_pool_unary_ctx_t *ctx =
        (ort_pool_unary_ctx_t *)arg;

    size_t chunk = ctx->layout.chunk;
    size_t start = index * chunk;
    size_t end   = start + chunk;

    if (end > ctx->layout.total)
        end = ctx->layout.total;
    
    size_t n = end - start;

    if (n > 0) {
        ctx->op(
            (char*)ctx->result    + start * ctx->layout.element,
            (char*)ctx->a         + start * ctx->layout.element,
            n
        );
    }
}

void ort_pool_scalar_worker(void *arg, size_t index, size_t count) {
    ort_pool_scalar_ctx_t *ctx =
        (ort_pool_scalar_ctx_t *)arg;
    size_t chunk = ctx->layout.chunk;
    size_t start = index * chunk;
    size_t end   = start + chunk;

    if (end > ctx->layout.total)
        end = ctx->layout.total;
    
    size_t n = end - start;

    if (n > 0) {
        ctx->op(
            (char*)ctx->result    + start * ctx->layout.element,
            (char*)ctx->a         + start * ctx->layout.element,
            ctx->b,
            n
        );
    }
}

void ort_pool_reduce_tensor_worker(void *arg, size_t index, size_t count) {
    ort_pool_reduce_tensor_ctx_t *ctx =
        (ort_pool_reduce_tensor_ctx_t *)arg;
    size_t chunk = ctx->layout.chunk;
    size_t start = index * chunk;
    size_t end = start + chunk;

    if (end > ctx->layout.total)
        end = ctx->layout.total;
    
    size_t n = end - start;

    if (n > 0) {
        // Each thread reduces a chunk of the input, but for full reduction to scalar,
        // only one thread should do the reduction (to avoid race conditions).
        // For now, only thread 0 does the reduction.
        if (start == 0) {
            ctx->op(
                ctx->result, ctx->a, ctx->elements);
        }
    }
}

void ort_pool_reduce_axis_worker(void *arg, size_t index, size_t count) {
    ort_pool_reduce_axis_ctx_t *ctx =
        (ort_pool_reduce_axis_ctx_t *)arg;

    size_t chunk     = ctx->layout.chunk;
    size_t outer     = ctx->layout.outer;
    size_t inner     = ctx->layout.inner;
    size_t axis_size = ctx->layout.axis_size;
    size_t element   = ctx->layout.element;

    size_t start = index * chunk;
    size_t end = start + chunk;
    if (end > outer)
        end = outer;

    if (start < end) {
        void *result_ptr =
            (char*)ctx->result +
                start * inner * element;
        const void *a_ptr =
            (const char*)ctx->a + 
                start * axis_size * inner * element;
        ctx->op(
            result_ptr,
            a_ptr,
            end - start,
            axis_size,
            inner);
    }
}

static void *ort_pool_worker(void *arg) {
    ort_pool_t *pool = (ort_pool_t *)arg;
    while (1) {
        ort_pool_mutex_lock(&pool->mutex);
        while (!pool->task && !pool->stop) {
            ort_pool_cond_wait(&pool->cond, &pool->mutex);
        }
        if (pool->stop) {
            ort_pool_mutex_unlock(&pool->mutex);
            break;
        }
        ort_task_t *task = pool->task;
        php_ort_atomic_addref((uint32_t*)&task->refcount);
        ort_pool_mutex_unlock(&pool->mutex);

        while (1) {
#if defined(_WIN32)
            size_t index = InterlockedIncrement64((volatile LONGLONG*)&task->next) - 1;
#else
            size_t index = __sync_fetch_and_add(&task->next, 1);
#endif
            if (index >= task->count)
                break;
            task->func(task->arg, index, task->count);
#if defined(_WIN32)
            InterlockedIncrement64((volatile LONGLONG*)&task->completed);
#else
            __sync_fetch_and_add(&task->completed, 1);
#endif
        }

        ort_pool_mutex_lock(&task->mutex);
        if (task->completed == task->count) {
            ort_pool_cond_signal(&task->cond);
            // Wait for main thread to set done
            while (!task->done) {
                ort_pool_cond_wait(&task->cond, &task->mutex);
            }
        }
        ort_pool_mutex_unlock(&task->mutex);
        // Drop worker's reference
        if (php_ort_atomic_delref((uint32_t*)&task->refcount) == 0) {
#if defined(_WIN32)
            DeleteCriticalSection(&task->mutex);
            free(task);
#else
            pthread_mutex_destroy(&task->mutex);
            pthread_cond_destroy(&task->cond);
            free(task);
#endif
        }
    }
#if defined(_WIN32)
    return 0;
#else
    return NULL;
#endif
}

static inline size_t ort_pool_cores_env() {
    const char *env = getenv("ORT_POOL_CORES");
    if (env) {
        char *endptr;
        long n = strtol(env, &endptr, 10);
        if (*endptr == '\0' && n > 0) {
            return (size_t)n;
        }
    }
    return 0; // No valid environment variable set
}

#if defined(_WIN32)
size_t ort_pool_cores(void) {
    size_t env = ort_pool_cores_env();
    if (env > 0) {
        return env;
    }

    SYSTEM_INFO sysinfo;
    GetSystemInfo(&sysinfo);

    return (size_t)sysinfo.dwNumberOfProcessors;
}
#else
size_t ort_pool_cores(void) {
    size_t env = ort_pool_cores_env();
    if (env > 0) {
        return env;
    }

    long n = sysconf(
        _SC_NPROCESSORS_ONLN);
    return (n > 0) ? (size_t)n : 1;
}
#endif

int ort_pool_init(size_t size) {
    memset(&ort_maths_pool, 0, sizeof(ort_maths_pool));
    if (size == 0)
        size = ort_pool_cores();
    ort_maths_pool.size = size;
    ort_maths_pool.threads =
        (ort_thread_t*)calloc(size, sizeof(ort_thread_t));
    if (!ort_maths_pool.threads) {
        ort_maths_pool.size = 0;
        return FAILURE;
    }
    ort_pool_mutex_init(&ort_maths_pool.mutex);
    ort_pool_cond_init(&ort_maths_pool.cond);
    ort_maths_pool.stop = 0;
    for (size_t i = 0; i < size; ++i) {
#if defined(_WIN32)
        ort_maths_pool.threads[i] = CreateThread(
            NULL, 0, 
            ort_pool_worker, 
            &ort_maths_pool, 0, NULL);
#else
        pthread_create(
            &ort_maths_pool.threads[i], 
            NULL, 
            ort_pool_worker, 
            &ort_maths_pool);
#endif
    }
    return SUCCESS;
}

void ort_pool_destroy() {
    ort_pool_mutex_lock(&ort_maths_pool.mutex);
    ort_maths_pool.stop = 1;
    ort_pool_cond_broadcast(&ort_maths_pool.cond);
    ort_pool_mutex_unlock(&ort_maths_pool.mutex);

    for (size_t i = 0; i < ort_maths_pool.size; ++i) {
#if defined(_WIN32)
        WaitForSingleObject(ort_maths_pool.threads[i], INFINITE);
        CloseHandle(ort_maths_pool.threads[i]);
#else
        pthread_join(ort_maths_pool.threads[i], NULL);
#endif
    }

    ort_pool_mutex_destroy(&ort_maths_pool.mutex);
    ort_pool_cond_destroy(&ort_maths_pool.cond);
    free(ort_maths_pool.threads);
}

void ort_pool_submit(ort_task_func_t func, void *arg, size_t count) {
    ort_task_t *task = (ort_task_t*)calloc(1, sizeof(ort_task_t));
    ort_pool_mutex_init(&task->mutex);
    ort_pool_cond_init(&task->cond);
    task->func = func;
    task->arg = arg;
    task->count = count;
    task->next = 0;
    task->completed = 0;
    task->done = 0;
    task->refcount = 1 + ort_maths_pool.size; // main thread + all workers

    ort_pool_mutex_lock(&ort_maths_pool.mutex);
    ort_maths_pool.task = task;
    ort_pool_cond_broadcast(&ort_maths_pool.cond);
    ort_pool_mutex_unlock(&ort_maths_pool.mutex);

    ort_pool_mutex_lock(&task->mutex);
    while (task->completed < task->count) {
        ort_pool_cond_wait(&task->cond, &task->mutex);
    }
    // Signal workers that it's safe to exit
    task->done = 1;
    ort_pool_cond_broadcast(&task->cond);
    ort_pool_mutex_unlock(&task->mutex);

    // Drop main thread's reference
    if (php_ort_atomic_delref((uint32_t*)&task->refcount) == 0) {
#if defined(_WIN32)
        DeleteCriticalSection(&task->mutex);
        free(task);
#else
        pthread_mutex_destroy(&task->mutex);
        pthread_cond_destroy(&task->cond);
        free(task);
#endif
    }

    ort_pool_mutex_lock(&ort_maths_pool.mutex);
    ort_maths_pool.task = NULL;
    ort_pool_mutex_unlock(&ort_maths_pool.mutex);
}

1	/*
2	+----------------------------------------------------------------------+
3	\| ort \|
4	+----------------------------------------------------------------------+
5	\| Copyright (c) Joe Watkins 2025 \|
6	+----------------------------------------------------------------------+
7	\| This source file is subject to version 3.01 of the PHP license, \|
8	\| that is bundled with this package in the file LICENSE, and is \|
9	\| available through the world-wide-web at the following url: \|
10	\| http://www.php.net/license/3_01.txt \|
11	\| If you did not receive a copy of the PHP license and are unable to \|
12	\| obtain it through the world-wide-web, please send a note to \|
13	\| license@php.net so we can mail you a copy immediately. \|
14	+----------------------------------------------------------------------+
15	\| Author: krakjoe \|
16	+----------------------------------------------------------------------+
17	*/
18
19	#include "maths/cast.h"
20	#include "maths/pool.h"
21	#include "maths/result.h"
22
23	#if defined(_WIN32)
24	# include <windows.h>
25	# define ORT_POOL_LOCAL __declspec(thread)
26	typedef HANDLE ort_thread_t;
27	typedef CRITICAL_SECTION ort_mutex_t;
28	typedef CONDITION_VARIABLE ort_cond_t;
29	#define ort_pool_mutex_init(mutex) \
30	InitializeCriticalSection(mutex)
31	#define ort_pool_mutex_lock(mutex) \
32	EnterCriticalSection(mutex)
33	#define ort_pool_mutex_unlock(mutex) \
34	LeaveCriticalSection(mutex)
35	#define ort_pool_mutex_destroy(mutex) \
36	DeleteCriticalSection(mutex)
37	#define ort_pool_cond_init(cond) \
38	InitializeConditionVariable(cond)
39	#define ort_pool_cond_wait(cond, mutex) \
40	SleepConditionVariableCS(cond, mutex, INFINITE)
41	#define ort_pool_cond_signal(cond) \
42	WakeConditionVariable(cond)
43	#define ort_pool_cond_broadcast(cond) \
44	WakeAllConditionVariable(cond)
45	#define ort_pool_cond_destroy(cond)
46	#else
47	# include <pthread.h>
48	# include <unistd.h>
49	# define ORT_POOL_LOCAL __thread
50	typedef pthread_t ort_thread_t;
51	typedef pthread_mutex_t ort_mutex_t;
52	typedef pthread_cond_t ort_cond_t;
53	#define ort_pool_mutex_init(mutex) \
54	pthread_mutex_init(mutex, NULL)
55	#define ort_pool_mutex_lock(mutex) \
56	pthread_mutex_lock(mutex)
57	#define ort_pool_mutex_unlock(mutex) \
58	pthread_mutex_unlock(mutex)
59	#define ort_pool_mutex_destroy(mutex) \
60	pthread_mutex_destroy(mutex)
61	#define ort_pool_cond_init(cond) \
62	pthread_cond_init(cond, NULL)
63	#define ort_pool_cond_wait(cond, mutex) \
64	pthread_cond_wait(cond, mutex)
65	#define ort_pool_cond_signal(cond) \
66	pthread_cond_signal(cond)
67	#define ort_pool_cond_broadcast(cond) \
68	pthread_cond_broadcast(cond)
69	#define ort_pool_cond_destroy(cond) \
70	pthread_cond_destroy(cond)
71	#endif
72
73	typedef struct _ort_task_t {
74	ort_task_func_t func;
75	void *arg;
76	size_t count;
77	volatile size_t next;
78	volatile size_t completed;
79	ort_mutex_t mutex;
80	ort_cond_t cond;
81	volatile int done;
82	volatile int refcount;
83	} ort_task_t;
84
85	typedef struct _ort_pool_t {
86	ort_thread_t *threads;
87	size_t size;
88	ort_task_t *task;
89	ort_mutex_t mutex;
90	ort_cond_t cond;
91	int stop;
92	} ort_pool_t;
93
94	ORT_POOL_LOCAL ort_pool_t ort_maths_pool;
95
96	void ort_pool_binary_worker(void *arg, size_t index, size_t count) {	6,464✔
97	ort_pool_binary_ctx_t *ctx =	6,464✔
98	(ort_pool_binary_ctx_t *)arg;
99
100	size_t chunk = ctx->layout.chunk;	6,464✔
101	size_t start = index * chunk;	6,464✔
102	size_t end = start + chunk;	6,464✔
103
104	if (end > ctx->layout.total)	6,464✔
105	end = ctx->layout.total;
106
107	size_t n = end - start;	6,464✔
108	if (n > 0) {	6,464✔
109	ctx->op(	6,464✔
110	(char)ctx->result + start ctx->layout.element,	6,464✔
111	(char)ctx->a + start ctx->layout.element,	6,464✔
112	(char)ctx->b + start ctx->layout.element,	6,464✔
113	n
114	);
115	}
116	}	6,464✔
117
118	void ort_pool_unary_worker(void *arg, size_t index, size_t count) {	7,824✔
119	ort_pool_unary_ctx_t *ctx =	7,824✔
120	(ort_pool_unary_ctx_t *)arg;
121
122	size_t chunk = ctx->layout.chunk;	7,824✔
123	size_t start = index * chunk;	7,824✔
124	size_t end = start + chunk;	7,824✔
125
126	if (end > ctx->layout.total)	7,824✔
127	end = ctx->layout.total;
128
129	size_t n = end - start;	7,824✔
130
131	if (n > 0) {	7,824✔
132	ctx->op(	7,824✔
133	(char)ctx->result + start ctx->layout.element,	7,824✔
134	(char)ctx->a + start ctx->layout.element,	7,824✔
135	n
136	);
137	}
138	}	7,824✔
139
140	void ort_pool_scalar_worker(void *arg, size_t index, size_t count) {	5,792✔
141	ort_pool_scalar_ctx_t *ctx =	5,792✔
142	(ort_pool_scalar_ctx_t *)arg;
143	size_t chunk = ctx->layout.chunk;	5,792✔
144	size_t start = index * chunk;	5,792✔
145	size_t end = start + chunk;	5,792✔
146
147	if (end > ctx->layout.total)	5,792✔
148	end = ctx->layout.total;
149
150	size_t n = end - start;	5,792✔
151
152	if (n > 0) {	5,792✔
153	ctx->op(	5,792✔
154	(char)ctx->result + start ctx->layout.element,	5,792✔
155	(char)ctx->a + start ctx->layout.element,	5,792✔
156	ctx->b,	5,792✔
157	n
158	);
159	}
160	}	5,792✔
161
162	void ort_pool_reduce_tensor_worker(void *arg, size_t index, size_t count) {	592✔
163	ort_pool_reduce_tensor_ctx_t *ctx =	592✔
164	(ort_pool_reduce_tensor_ctx_t *)arg;
165	size_t chunk = ctx->layout.chunk;	592✔
166	size_t start = index * chunk;	592✔
167	size_t end = start + chunk;	592✔
168
169	if (end > ctx->layout.total)	592✔
170	end = ctx->layout.total;
171
172	size_t n = end - start;	592✔
173
174	if (n > 0) {	592✔
175	// Each thread reduces a chunk of the input, but for full reduction to scalar,
176	// only one thread should do the reduction (to avoid race conditions).
177	// For now, only thread 0 does the reduction.
178	if (start == 0) {	592✔
179	ctx->op(	592✔
180	ctx->result, ctx->a, ctx->elements);
181	}
182	}
183	}	592✔
184
185	void ort_pool_reduce_axis_worker(void *arg, size_t index, size_t count) {	2,480✔
186	ort_pool_reduce_axis_ctx_t *ctx =	2,480✔
187	(ort_pool_reduce_axis_ctx_t *)arg;
188
189	size_t chunk = ctx->layout.chunk;	2,480✔
190	size_t outer = ctx->layout.outer;	2,480✔
191	size_t inner = ctx->layout.inner;	2,480✔
192	size_t axis_size = ctx->layout.axis_size;	2,480✔
193	size_t element = ctx->layout.element;	2,480✔
194
195	size_t start = index * chunk;	2,480✔
196	size_t end = start + chunk;	2,480✔
197	if (end > outer)	2,480✔
198	end = outer;
199
200	if (start < end) {	2,480✔
201	void *result_ptr =	2,480✔
202	(char*)ctx->result +	2,480✔
203	start * inner * element;	2,480✔
204	const void *a_ptr =	2,480✔
205	(const char*)ctx->a +	2,480✔
206	start * axis_size * inner * element;	2,480✔
207	ctx->op(	2,480✔
208	result_ptr,
209	a_ptr,
210	end - start,
211	axis_size,
212	inner);
213	}
214	}	2,480✔
215
216	static void ort_pool_worker(void arg) {	3,152✔
217	ort_pool_t pool = (ort_pool_t )arg;	3,152✔
218	while (1) {	50,289✔
219	ort_pool_mutex_lock(&pool->mutex);	50,289✔
220	while (!pool->task && !pool->stop) {	71,311✔
221	ort_pool_cond_wait(&pool->cond, &pool->mutex);	21,022✔
222	}
223	if (pool->stop) {	50,289✔
224	ort_pool_mutex_unlock(&pool->mutex);	3,152✔
225	break;	3,152✔
226	}
227	ort_task_t *task = pool->task;	47,137✔
228	php_ort_atomic_addref((uint32_t*)&task->refcount);	47,137✔
229	ort_pool_mutex_unlock(&pool->mutex);	47,137✔
230
231	while (1) {	93,441✔
232	#if defined(_WIN32)
233	size_t index = InterlockedIncrement64((volatile LONGLONG*)&task->next) - 1;
234	#else
235	size_t index = __sync_fetch_and_add(&task->next, 1);	70,289✔
236	#endif
237	if (index >= task->count)	70,289✔
238	break;
239	task->func(task->arg, index, task->count);	23,152✔
240	#if defined(_WIN32)
241	InterlockedIncrement64((volatile LONGLONG*)&task->completed);
242	#else
243	__sync_fetch_and_add(&task->completed, 1);	23,152✔
244	#endif
245	}
246
247	ort_pool_mutex_lock(&task->mutex);	47,137✔
248	if (task->completed == task->count) {	47,137✔
249	ort_pool_cond_signal(&task->cond);	47,137✔
250	// Wait for main thread to set done
251	while (!task->done) {	70,288✔
252	ort_pool_cond_wait(&task->cond, &task->mutex);	23,151✔
253	}
254	}
255	ort_pool_mutex_unlock(&task->mutex);	47,137✔
256	// Drop worker's reference
257	if (php_ort_atomic_delref((uint32_t*)&task->refcount) == 0) {	47,137✔
258	#if defined(_WIN32)
259	DeleteCriticalSection(&task->mutex);
260	free(task);
261	#else
262	pthread_mutex_destroy(&task->mutex);	×
263	pthread_cond_destroy(&task->cond);	×
264	free(task);	×
265	#endif
266	}
267	}
268	#if defined(_WIN32)
269	return 0;
270	#else
271	return NULL;	3,152✔
272	#endif
273	}
274
275	static inline size_t ort_pool_cores_env() {	25,712✔
276	const char *env = getenv("ORT_POOL_CORES");	25,712✔
277	if (env) {	25,712✔
278	char *endptr;	25,712✔
279	long n = strtol(env, &endptr, 10);	25,712✔
280	if (*endptr == '\0' && n > 0) {	25,712✔
281	return (size_t)n;	25,712✔
282	}
283	}
284	return 0; // No valid environment variable set
285	}
286
287	#if defined(_WIN32)
288	size_t ort_pool_cores(void) {
289	size_t env = ort_pool_cores_env();
290	if (env > 0) {
291	return env;
292	}
293
294	SYSTEM_INFO sysinfo;
295	GetSystemInfo(&sysinfo);
296
297	return (size_t)sysinfo.dwNumberOfProcessors;
298	}
299	#else
300	size_t ort_pool_cores(void) {	25,712✔
301	size_t env = ort_pool_cores_env();	25,712✔
302	if (env > 0) {	25,712✔
303	return env;
304	}
305
306	long n = sysconf(	×
307	_SC_NPROCESSORS_ONLN);
308	return (n > 0) ? (size_t)n : 1;	×
309	}
310	#endif
311
312	int ort_pool_init(size_t size) {	3,136✔
313	memset(&ort_maths_pool, 0, sizeof(ort_maths_pool));	3,136✔
314	if (size == 0)	3,136✔
315	size = ort_pool_cores();	×
316	ort_maths_pool.size = size;	3,136✔
317	ort_maths_pool.threads =	3,136✔
318	(ort_thread_t*)calloc(size, sizeof(ort_thread_t));	3,136✔
319	if (!ort_maths_pool.threads) {	3,136✔
320	ort_maths_pool.size = 0;	×
321	return FAILURE;	×
322	}
323	ort_pool_mutex_init(&ort_maths_pool.mutex);	3,136✔
324	ort_pool_cond_init(&ort_maths_pool.cond);	3,136✔
325	ort_maths_pool.stop = 0;	3,136✔
326	for (size_t i = 0; i < size; ++i) {	6,288✔
327	#if defined(_WIN32)
328	ort_maths_pool.threads[i] = CreateThread(
329	NULL, 0,
330	ort_pool_worker,
331	&ort_maths_pool, 0, NULL);
332	#else
333	pthread_create(	3,152✔
334	&ort_maths_pool.threads[i],	3,152✔
335	NULL,
336	ort_pool_worker,
337	&ort_maths_pool);
338	#endif
339	}
340	return SUCCESS;
341	}
342
343	void ort_pool_destroy() {	3,136✔
344	ort_pool_mutex_lock(&ort_maths_pool.mutex);	3,136✔
345	ort_maths_pool.stop = 1;	3,136✔
346	ort_pool_cond_broadcast(&ort_maths_pool.cond);	3,136✔
347	ort_pool_mutex_unlock(&ort_maths_pool.mutex);	3,136✔
348
349	for (size_t i = 0; i < ort_maths_pool.size; ++i) {	6,288✔
350	#if defined(_WIN32)
351	WaitForSingleObject(ort_maths_pool.threads[i], INFINITE);
352	CloseHandle(ort_maths_pool.threads[i]);
353	#else
354	pthread_join(ort_maths_pool.threads[i], NULL);	3,152✔
355	#endif
356	}
357
358	ort_pool_mutex_destroy(&ort_maths_pool.mutex);	3,136✔
359	ort_pool_cond_destroy(&ort_maths_pool.cond);	3,136✔
360	free(ort_maths_pool.threads);	3,136✔
361	}	3,136✔
362
363	void ort_pool_submit(ort_task_func_t func, void *arg, size_t count) {	23,152✔
364	ort_task_t task = (ort_task_t)calloc(1, sizeof(ort_task_t));	23,152✔
365	ort_pool_mutex_init(&task->mutex);	23,152✔
366	ort_pool_cond_init(&task->cond);	23,152✔
367	task->func = func;	23,152✔
368	task->arg = arg;	23,152✔
369	task->count = count;	23,152✔
370	task->next = 0;	23,152✔
371	task->completed = 0;	23,152✔
372	task->done = 0;	23,152✔
373	task->refcount = 1 + ort_maths_pool.size; // main thread + all workers	23,152✔
374
375	ort_pool_mutex_lock(&ort_maths_pool.mutex);	23,152✔
376	ort_maths_pool.task = task;	23,152✔
377	ort_pool_cond_broadcast(&ort_maths_pool.cond);	23,152✔
378	ort_pool_mutex_unlock(&ort_maths_pool.mutex);	23,152✔
379
380	ort_pool_mutex_lock(&task->mutex);	23,152✔
381	while (task->completed < task->count) {	46,303✔
382	ort_pool_cond_wait(&task->cond, &task->mutex);	23,151✔
383	}
384	// Signal workers that it's safe to exit
385	task->done = 1;	23,152✔
386	ort_pool_cond_broadcast(&task->cond);	23,152✔
387	ort_pool_mutex_unlock(&task->mutex);	23,152✔
388
389	// Drop main thread's reference
390	if (php_ort_atomic_delref((uint32_t*)&task->refcount) == 0) {	23,152✔
391	#if defined(_WIN32)
392	DeleteCriticalSection(&task->mutex);
393	free(task);
394	#else
395	pthread_mutex_destroy(&task->mutex);	×
396	pthread_cond_destroy(&task->cond);	×
397	free(task);	×
398	#endif
399	}
400
401	ort_pool_mutex_lock(&ort_maths_pool.mutex);	23,152✔
402	ort_maths_pool.task = NULL;	23,152✔
403	ort_pool_mutex_unlock(&ort_maths_pool.mutex);	23,152✔
404	}	23,152✔

krakjoe / ort / 16253909665

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