15578322680

Committed 11 Jun 2025 07:06AM UTC coverage: 96.87% (-0.05%) from 96.923%

Build # 15578322680

Build Type

push

github

Committed by

realFlowControl

Commit Message

bump

Run Details

3002 of 3099 relevant lines covered (96.87%)

5740.73 hits per line

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

97.89

/src/parallel.c

/*
  +----------------------------------------------------------------------+
  | parallel                                                             |
  +----------------------------------------------------------------------+
  | Copyright (c) Joe Watkins 2019-2024                                  |
  +----------------------------------------------------------------------+
  | 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                                                      |
  +----------------------------------------------------------------------+
 */
#ifndef HAVE_PARALLEL_PARALLEL
#define HAVE_PARALLEL_PARALLEL

#include "parallel.h"

/* {{{ */
TSRM_TLS HashTable php_parallel_runtimes;

static struct {
    pthread_mutex_t     mutex;
    zend_string        *bootstrap;
    volatile zend_ulong running;
} php_parallel_globals;

#define PCG(e) php_parallel_globals.e
/* }}} */

/* {{{ */
typedef int (*php_sapi_deactivate_t)(void);
typedef size_t (*php_sapi_output_t)(const char*, size_t);

static php_sapi_deactivate_t php_sapi_deactivate_function;
static php_sapi_output_t     php_sapi_output_function;

static pthread_mutex_t php_parallel_output_mutex = PTHREAD_MUTEX_INITIALIZER;

static size_t php_parallel_output_function(const char *str, size_t len) {
    size_t result;

    pthread_mutex_lock(&php_parallel_output_mutex);
    result =
        php_sapi_output_function(str, len);
    pthread_mutex_unlock(&php_parallel_output_mutex);

    return result;
} /* }}} */

/* {{{ */
ZEND_BEGIN_ARG_INFO_EX(php_parallel_bootstrap_arginfo, 0, 0, 1)
    ZEND_ARG_TYPE_INFO(0, file, IS_STRING, 0)
ZEND_END_ARG_INFO()

static PHP_NAMED_FUNCTION(php_parallel_bootstrap)
{
    zend_string *bootstrap;

    ZEND_PARSE_PARAMETERS_START(1, 1)
        Z_PARAM_PATH_STR(bootstrap)
    ZEND_PARSE_PARAMETERS_END();

    pthread_mutex_lock(&PCG(mutex));

    if (PCG(bootstrap)) {
        php_parallel_exception_ex(
            php_parallel_runtime_error_bootstrap_ce,
            "\\parallel\\bootstrap already set to %s",
            ZSTR_VAL(PCG(bootstrap)));
        pthread_mutex_unlock(&PCG(mutex));
        return;
    }

    if (PCG(running)) {
        php_parallel_exception_ex(
            php_parallel_runtime_error_bootstrap_ce,
            "\\parallel\\bootstrap should be called once, "
            "before any calls to \\parallel\\run");
        pthread_mutex_unlock(&PCG(mutex));
        return;
    }

    PCG(bootstrap) =
        php_parallel_copy_string_interned(bootstrap);
    pthread_mutex_unlock(&PCG(mutex));
} /* }}} */

// This will return an idle runtime (aka thread) or spawn a new one, bootstap it
// and return that. This method is solely called by the userlands
// `\parallel\run()` function call.
/* {{{ */
static zend_always_inline php_parallel_runtime_t* php_parallel_runtimes_fetch() {
    php_parallel_runtime_t *runtime;

    ZEND_HASH_FOREACH_PTR(&php_parallel_runtimes, runtime) {
        if (!php_parallel_scheduler_busy(runtime)) {
            return runtime;
        }
    } ZEND_HASH_FOREACH_END();

    if (!(runtime = php_parallel_runtime_construct(PCG(bootstrap)))) {
        return NULL;
    }

    pthread_mutex_lock(&PCG(mutex));
    PCG(running)++;
    pthread_mutex_unlock(&PCG(mutex));

    return zend_hash_next_index_insert_ptr(&php_parallel_runtimes, runtime);
}

ZEND_BEGIN_ARG_WITH_RETURN_OBJ_INFO_EX(php_parallel_run_arginfo, 0, 1, \\parallel\\Future, 1)
    ZEND_ARG_OBJ_INFO(0, task, Closure, 0)
    ZEND_ARG_TYPE_INFO(0, argv, IS_ARRAY, 0)
ZEND_END_ARG_INFO()

static PHP_NAMED_FUNCTION(php_parallel_run)
{
    php_parallel_runtime_t *runtime;
    zval *closure = NULL;
    zval *argv = NULL;

    ZEND_PARSE_PARAMETERS_START(1, 2)
        Z_PARAM_OBJECT_OF_CLASS(closure, zend_ce_closure)
        Z_PARAM_OPTIONAL
        Z_PARAM_ARRAY(argv)
    ZEND_PARSE_PARAMETERS_END();

    runtime = php_parallel_runtimes_fetch();

    if (!EG(exception)) {
        php_parallel_scheduler_push(
            runtime, closure, argv, return_value);
    }
} /* }}} */

#ifdef ZEND_DEBUG
ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(php_parallel_count_arginfo, 0, 0, IS_LONG, 0)
ZEND_END_ARG_INFO()

PHP_NAMED_FUNCTION(php_parallel_count)
{
    RETURN_LONG(zend_hash_num_elements(&php_parallel_runtimes));
}
#endif

/* {{{ */
zend_function_entry php_parallel_functions[] = {
        ZEND_NS_FENTRY("parallel", bootstrap, php_parallel_bootstrap, php_parallel_bootstrap_arginfo, 0)
        ZEND_NS_FENTRY("parallel", run,       php_parallel_run,       php_parallel_run_arginfo, 0)
#ifdef ZEND_DEBUG
        ZEND_NS_FENTRY("parallel", count,     php_parallel_count,     php_parallel_count_arginfo, 0)
#endif
    PHP_FE_END
}; /* }}} */

PHP_MINIT_FUNCTION(PARALLEL_CORE)
{
    if (strncmp(sapi_module.name, "cli", sizeof("cli")-1) == SUCCESS) {
        php_sapi_deactivate_function = sapi_module.deactivate;

        sapi_module.deactivate = NULL;
    }

    memset(&php_parallel_globals, 0, sizeof(php_parallel_globals));

    php_sapi_output_function = sapi_module.ub_write;

    sapi_module.ub_write = php_parallel_output_function;

    PHP_MINIT(PARALLEL_HANDLERS)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MINIT(PARALLEL_EXCEPTIONS)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MINIT(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MINIT(PARALLEL_SCHEDULER)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MINIT(PARALLEL_CHANNEL)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MINIT(PARALLEL_EVENTS)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MINIT(PARALLEL_SYNC)(INIT_FUNC_ARGS_PASSTHRU);

    php_parallel_mutex_init(&PCG(mutex), 1);
    PCG(running) = 0;
    PCG(bootstrap) = NULL;

    return SUCCESS;
}

PHP_MSHUTDOWN_FUNCTION(PARALLEL_CORE)
{
    PHP_MSHUTDOWN(PARALLEL_SYNC)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MSHUTDOWN(PARALLEL_EVENTS)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MSHUTDOWN(PARALLEL_CHANNEL)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MSHUTDOWN(PARALLEL_SCHEDULER)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MSHUTDOWN(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MSHUTDOWN(PARALLEL_EXCEPTIONS)(INIT_FUNC_ARGS_PASSTHRU);
    PHP_MSHUTDOWN(PARALLEL_HANDLERS)(INIT_FUNC_ARGS_PASSTHRU);

    php_parallel_mutex_destroy(&PCG(mutex));

    if (strncmp(sapi_module.name, "cli", sizeof("cli")-1) == SUCCESS) {
        sapi_module.deactivate = php_sapi_deactivate_function;
    }

    sapi_module.ub_write = php_sapi_output_function;

    return SUCCESS;
}

static void php_parallel_runtimes_release(zval *zv) {
    php_parallel_runtime_t *runtime =
        (php_parallel_runtime_t*) Z_PTR_P(zv);

    OBJ_RELEASE(&runtime->std);
    pthread_mutex_lock(&PCG(mutex));
    PCG(running)--;
    pthread_mutex_unlock(&PCG(mutex));
}

PHP_RINIT_FUNCTION(PARALLEL_CORE)
{
    PHP_RINIT(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);

    zend_hash_init(
        &php_parallel_runtimes, 16, NULL, php_parallel_runtimes_release, 0);

        return SUCCESS;
}

PHP_RSHUTDOWN_FUNCTION(PARALLEL_CORE)
{
    zend_hash_destroy(&php_parallel_runtimes);

    PHP_RSHUTDOWN(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);

    // In case of a `zend_bailout()` this mutex could still be locked, so we
    // unlock it just in case.
    // See https://github.com/krakjoe/parallel/issues/313 for more details
    if (UNEXPECTED(CG(unclean_shutdown) == 1)) {
        pthread_mutex_unlock(&php_parallel_output_mutex);
    }

        return SUCCESS;
}
#endif

1	/*
2	+----------------------------------------------------------------------+
3	\| parallel \|
4	+----------------------------------------------------------------------+
5	\| Copyright (c) Joe Watkins 2019-2024 \|
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	#ifndef HAVE_PARALLEL_PARALLEL
19	#define HAVE_PARALLEL_PARALLEL
20
21	#include "parallel.h"
22
23	/* {{{ */
24	TSRM_TLS HashTable php_parallel_runtimes;
25
26	static struct {
27	pthread_mutex_t mutex;
28	zend_string *bootstrap;
29	volatile zend_ulong running;
30	} php_parallel_globals;
31
32	#define PCG(e) php_parallel_globals.e
33	/* }}} */
34
35	/* {{{ */
36	typedef int (*php_sapi_deactivate_t)(void);
37	typedef size_t (php_sapi_output_t)(const char, size_t);
38
39	static php_sapi_deactivate_t php_sapi_deactivate_function;
40	static php_sapi_output_t php_sapi_output_function;
41
42	static pthread_mutex_t php_parallel_output_mutex = PTHREAD_MUTEX_INITIALIZER;
43
44	static size_t php_parallel_output_function(const char *str, size_t len) {	21,084✔
45	size_t result;	21,084✔
46
47	pthread_mutex_lock(&php_parallel_output_mutex);	21,084✔
48	result =	21,084✔
49	php_sapi_output_function(str, len);	21,084✔
50	pthread_mutex_unlock(&php_parallel_output_mutex);	21,084✔
51
52	return result;	21,084✔
53	} /* }}} */
54
55	/* {{{ */
56	ZEND_BEGIN_ARG_INFO_EX(php_parallel_bootstrap_arginfo, 0, 0, 1)
57	ZEND_ARG_TYPE_INFO(0, file, IS_STRING, 0)
58	ZEND_END_ARG_INFO()
59
60	static PHP_NAMED_FUNCTION(php_parallel_bootstrap)	135✔
61	{
62	zend_string *bootstrap;	135✔
63
64	ZEND_PARSE_PARAMETERS_START(1, 1)	135✔
65	Z_PARAM_PATH_STR(bootstrap)	270✔
66	ZEND_PARSE_PARAMETERS_END();	180✔
67
68	pthread_mutex_lock(&PCG(mutex));	135✔
69
70	if (PCG(bootstrap)) {	135✔
71	php_parallel_exception_ex(	15✔
72	php_parallel_runtime_error_bootstrap_ce,
73	"\\parallel\\bootstrap already set to %s",
74	ZSTR_VAL(PCG(bootstrap)));
75	pthread_mutex_unlock(&PCG(mutex));	15✔
76	return;	15✔
77	}
78
79	if (PCG(running)) {	120✔
80	php_parallel_exception_ex(	30✔
81	php_parallel_runtime_error_bootstrap_ce,
82	"\\parallel\\bootstrap should be called once, "
83	"before any calls to \\parallel\\run");
84	pthread_mutex_unlock(&PCG(mutex));	30✔
85	return;	30✔
86	}
87
88	PCG(bootstrap) =	180✔
89	php_parallel_copy_string_interned(bootstrap);	90✔
90	pthread_mutex_unlock(&PCG(mutex));	90✔
91	} /* }}} */
92
93	// This will return an idle runtime (aka thread) or spawn a new one, bootstap it
94	// and return that. This method is solely called by the userlands
95	// `\parallel\run()` function call.
96	/* {{{ */
97	static zend_always_inline php_parallel_runtime_t* php_parallel_runtimes_fetch() {	405✔
98	php_parallel_runtime_t *runtime;	405✔
99
100	ZEND_HASH_FOREACH_PTR(&php_parallel_runtimes, runtime) {	1,125✔
101	if (!php_parallel_scheduler_busy(runtime)) {	735✔
102	return runtime;
103	}
104	} ZEND_HASH_FOREACH_END();
105
106	if (!(runtime = php_parallel_runtime_construct(PCG(bootstrap)))) {	390✔
107	return NULL;
108	}
109
110	pthread_mutex_lock(&PCG(mutex));	390✔
111	PCG(running)++;	390✔
112	pthread_mutex_unlock(&PCG(mutex));	390✔
113
114	return zend_hash_next_index_insert_ptr(&php_parallel_runtimes, runtime);	780✔
115	}
116
117	ZEND_BEGIN_ARG_WITH_RETURN_OBJ_INFO_EX(php_parallel_run_arginfo, 0, 1, \\parallel\\Future, 1)
118	ZEND_ARG_OBJ_INFO(0, task, Closure, 0)
119	ZEND_ARG_TYPE_INFO(0, argv, IS_ARRAY, 0)
120	ZEND_END_ARG_INFO()
121
122	static PHP_NAMED_FUNCTION(php_parallel_run)	405✔
123	{
124	php_parallel_runtime_t *runtime;	405✔
125	zval *closure = NULL;	405✔
126	zval *argv = NULL;	405✔
127
128	ZEND_PARSE_PARAMETERS_START(1, 2)	405✔
129	Z_PARAM_OBJECT_OF_CLASS(closure, zend_ce_closure)	810✔
130	Z_PARAM_OPTIONAL	405✔
131	Z_PARAM_ARRAY(argv)	645✔
132	ZEND_PARSE_PARAMETERS_END();	405✔
133
134	runtime = php_parallel_runtimes_fetch();	405✔
135
136	if (!EG(exception)) {	405✔
137	php_parallel_scheduler_push(	405✔
138	runtime, closure, argv, return_value);
139	}
140	} /* }}} */
141
142	#ifdef ZEND_DEBUG
143	ZEND_BEGIN_ARG_WITH_RETURN_TYPE_INFO_EX(php_parallel_count_arginfo, 0, 0, IS_LONG, 0)
144	ZEND_END_ARG_INFO()
145
146	PHP_NAMED_FUNCTION(php_parallel_count)	×
147	{
148	RETURN_LONG(zend_hash_num_elements(&php_parallel_runtimes));	×
149	}
150	#endif
151
152	/* {{{ */
153	zend_function_entry php_parallel_functions[] = {
154	ZEND_NS_FENTRY("parallel", bootstrap, php_parallel_bootstrap, php_parallel_bootstrap_arginfo, 0)
155	ZEND_NS_FENTRY("parallel", run, php_parallel_run, php_parallel_run_arginfo, 0)
156	#ifdef ZEND_DEBUG
157	ZEND_NS_FENTRY("parallel", count, php_parallel_count, php_parallel_count_arginfo, 0)
158	#endif
159	PHP_FE_END
160	}; /* }}} */
161
162	PHP_MINIT_FUNCTION(PARALLEL_CORE)	2,684✔
163	{
164	if (strncmp(sapi_module.name, "cli", sizeof("cli")-1) == SUCCESS) {	2,684✔
165	php_sapi_deactivate_function = sapi_module.deactivate;	2,684✔
166
167	sapi_module.deactivate = NULL;	2,684✔
168	}
169
170	memset(&php_parallel_globals, 0, sizeof(php_parallel_globals));	2,684✔
171
172	php_sapi_output_function = sapi_module.ub_write;	2,684✔
173
174	sapi_module.ub_write = php_parallel_output_function;	2,684✔
175
176	PHP_MINIT(PARALLEL_HANDLERS)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
177	PHP_MINIT(PARALLEL_EXCEPTIONS)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
178	PHP_MINIT(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
179	PHP_MINIT(PARALLEL_SCHEDULER)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
180	PHP_MINIT(PARALLEL_CHANNEL)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
181	PHP_MINIT(PARALLEL_EVENTS)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
182	PHP_MINIT(PARALLEL_SYNC)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
183
184	php_parallel_mutex_init(&PCG(mutex), 1);	2,684✔
185	PCG(running) = 0;	2,684✔
186	PCG(bootstrap) = NULL;	2,684✔
187
188	return SUCCESS;	2,684✔
189	}
190
191	PHP_MSHUTDOWN_FUNCTION(PARALLEL_CORE)	2,684✔
192	{
193	PHP_MSHUTDOWN(PARALLEL_SYNC)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
194	PHP_MSHUTDOWN(PARALLEL_EVENTS)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
195	PHP_MSHUTDOWN(PARALLEL_CHANNEL)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
196	PHP_MSHUTDOWN(PARALLEL_SCHEDULER)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
197	PHP_MSHUTDOWN(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
198	PHP_MSHUTDOWN(PARALLEL_EXCEPTIONS)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
199	PHP_MSHUTDOWN(PARALLEL_HANDLERS)(INIT_FUNC_ARGS_PASSTHRU);	2,684✔
200
201	php_parallel_mutex_destroy(&PCG(mutex));	2,684✔
202
203	if (strncmp(sapi_module.name, "cli", sizeof("cli")-1) == SUCCESS) {	2,684✔
204	sapi_module.deactivate = php_sapi_deactivate_function;	2,684✔
205	}
206
207	sapi_module.ub_write = php_sapi_output_function;	2,684✔
208
209	return SUCCESS;	2,684✔
210	}
211
212	static void php_parallel_runtimes_release(zval *zv) {	390✔
213	php_parallel_runtime_t *runtime =	390✔
214	(php_parallel_runtime_t*) Z_PTR_P(zv);
215
216	OBJ_RELEASE(&runtime->std);	390✔
217	pthread_mutex_lock(&PCG(mutex));	390✔
218	PCG(running)--;	390✔
219	pthread_mutex_unlock(&PCG(mutex));	390✔
220	}	390✔
221
222	PHP_RINIT_FUNCTION(PARALLEL_CORE)	4,487✔
223	{
224	PHP_RINIT(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);	4,487✔
225
226	zend_hash_init(	4,487✔
227	&php_parallel_runtimes, 16, NULL, php_parallel_runtimes_release, 0);
228
229	return SUCCESS;	4,487✔
230	}
231
232	PHP_RSHUTDOWN_FUNCTION(PARALLEL_CORE)	4,487✔
233	{
234	zend_hash_destroy(&php_parallel_runtimes);	4,487✔
235
236	PHP_RSHUTDOWN(PARALLEL_COPY)(INIT_FUNC_ARGS_PASSTHRU);	4,487✔
237
238	// In case of a `zend_bailout()` this mutex could still be locked, so we
239	// unlock it just in case.
240	// See https://github.com/krakjoe/parallel/issues/313 for more details
241	if (UNEXPECTED(CG(unclean_shutdown) == 1)) {	4,487✔
242	pthread_mutex_unlock(&php_parallel_output_mutex);	90✔
243	}
244
245	return SUCCESS;	4,487✔
246	}
247	#endif

krakjoe / parallel / 15578322680

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