30916206264

Committed 04 Aug 2026 01:54PM UTC coverage: 94.962% (-0.08%) from 95.041%

Build # 30916206264

Build Type

Pull #392

github

Committed by

web-flow

Commit Message

Merge 04e87b26a into e0b2788fa

Pull Request Pull Request #392: Replace `Events` polling with notifications

Coverage Stats

156 of 171 new or added lines in 4 files covered. (91.23%)

20 existing lines in 1 file now uncovered.

3035 of 3196 relevant lines covered (94.96%)

4863.22 hits per line

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

93.44

/src/poll.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_EVENTS_POLL
#define HAVE_PARALLEL_EVENTS_POLL

#include "parallel.h"

#if PHP_VERSION_ID >= 80300
#include "Zend/zend_hrtime.h"
#endif

#include <errno.h>
#include <time.h>

#ifdef _WIN32
#if PHP_VERSION_ID < 80300
#include <windows.h>
#endif
#else
#include <sys/select.h>
#endif

#define PHP_PARALLEL_EVENTS_MICRO_IN_SEC 1000000ULL

#if PHP_VERSION_ID >= 80400
#include "ext/random/php_random.h"
#else
#include "ext/standard/php_mt_rand.h"
#endif

typedef struct _php_parallel_events_poll_t {
        uint32_t try;
        uint64_t stop;
        bool     native;
        struct {
                zend_fcall_info       fci;
                zend_fcall_info_cache fcc;
                zval                  ival;
        } block;
        php_parallel_events_state_t state;
} php_parallel_events_poll_t;

static zend_always_inline uint64_t php_parallel_events_poll_now(void)
{
#if PHP_VERSION_ID >= 80300
        return zend_hrtime() / 1000;
#elif defined(_WIN32)
        LARGE_INTEGER now, frequency;

        QueryPerformanceCounter(&now);
        QueryPerformanceFrequency(&frequency);

        return (now.QuadPart / frequency.QuadPart) * PHP_PARALLEL_EVENTS_MICRO_IN_SEC +
               (now.QuadPart % frequency.QuadPart) * PHP_PARALLEL_EVENTS_MICRO_IN_SEC / frequency.QuadPart;
#else
        struct timespec now;

        clock_gettime(CLOCK_MONOTONIC, &now);

        return (uint64_t)now.tv_sec * PHP_PARALLEL_EVENTS_MICRO_IN_SEC + now.tv_nsec / 1000;
#endif
}

static zend_always_inline php_parallel_events_poll_t *php_parallel_events_poll_init(php_parallel_events_t *events)
{
        php_parallel_events_poll_t *poll;

        if (zend_hash_num_elements(&events->targets) == 0) {
                return NULL;
        }

        poll = (php_parallel_events_poll_t *)pecalloc(1, sizeof(php_parallel_events_poll_t), 1);
#ifndef _WIN32
        poll->native = true;
#endif

        if (events->timeout > -1) {
                poll->stop = php_parallel_events_poll_now() + (uint64_t)events->timeout;
        }

        if (!Z_ISUNDEF(events->blocker)) {
                zend_fcall_info_init(&events->blocker, 0, &poll->block.fci, &poll->block.fcc, NULL, NULL);
                poll->block.fci.retval = &poll->block.ival;
        } else {
                memset(&poll->block, 0, sizeof(poll->block));
        }

        return poll;
}

static zend_always_inline void php_parallel_events_poll_free(php_parallel_events_poll_t *poll) { pefree(poll, 1); }

static zend_always_inline void php_parallel_events_poll_unlock(php_parallel_events_poll_t *poll)
{
        if (poll->state.type == PHP_PARALLEL_EVENTS_LINK) {
                php_parallel_channel_t *channel = php_parallel_channel_fetch(poll->state.object);

                php_parallel_link_unlock(channel->link);
        } else {
                php_parallel_future_t *future = php_parallel_future_fetch(poll->state.object);

                php_parallel_future_unlock(future);
        }
}

static zend_always_inline void php_parallel_events_poll_end(php_parallel_events_poll_t *poll)
{
        php_parallel_events_poll_unlock(poll);
        php_parallel_events_poll_free(poll);
}

static zend_always_inline bool php_parallel_events_poll_expired(php_parallel_events_poll_t *poll,
                                                                php_parallel_events_t      *events)
{
        if (events->timeout > -1 && php_parallel_events_poll_now() >= poll->stop) {
                php_parallel_exception_ex(php_parallel_events_error_timeout_ce, "timeout occured");
                return 1;
        }

        return 0;
}

static zend_always_inline bool php_parallel_events_poll_busy(php_parallel_events_poll_t *poll,
                                                             php_parallel_events_t      *events)
{
        if ((poll->try++ % 10) == 0) {
                usleep(1);
        }

        return !php_parallel_events_poll_expired(poll, events);
}

#ifndef _WIN32
static zend_always_inline bool php_parallel_events_poll_descriptors(php_parallel_events_t *events, fd_set *readfds,
                                                                    int *maximum)
{
        uint32_t index;

        FD_ZERO(readfds);
        *maximum = -1;

        for (index = 0; index < events->targets.nNumUsed; index++) {
                Bucket      *bucket = &events->targets.arData[index];
                zend_object *object;
                int          descriptor;

                if (Z_ISUNDEF(bucket->val)) {
                        continue;
                }

                object = Z_OBJ(bucket->val);

                if (instanceof_function(object->ce, php_parallel_channel_ce)) {
                        php_parallel_channel_t *channel = php_parallel_channel_fetch(object);
                        bool                    writable = php_parallel_events_input_exists(&events->input, bucket->key);

                        php_parallel_link_lock(channel->link);
                        descriptor = php_parallel_link_notify(channel->link, writable);
                        php_parallel_link_unlock(channel->link);
                } else {
                        php_parallel_future_t *future = php_parallel_future_fetch(object);

                        php_parallel_future_lock(future);
                        descriptor = php_parallel_monitor_notify(future->monitor);
                        php_parallel_future_unlock(future);
                }

                if (descriptor < 0 || descriptor >= FD_SETSIZE) {
                        return false;
                }

                FD_SET(descriptor, readfds);
                if (descriptor > *maximum) {
                        *maximum = descriptor;
                }
        }

        return *maximum >= 0;
}

static zend_always_inline bool php_parallel_events_poll_native(php_parallel_events_poll_t *poll,
                                                               php_parallel_events_t      *events)
{
        fd_set          readfds;
        int             maximum;
        int             result;
        struct timeval  timeout;
        struct timeval *timeout_pointer = NULL;

        if (!php_parallel_events_poll_descriptors(events, &readfds, &maximum)) {
                poll->native = false;
                return php_parallel_events_poll_busy(poll, events);
        }

        if (events->timeout > -1) {
                uint64_t now = php_parallel_events_poll_now();
                uint64_t remaining;

                if (now >= poll->stop) {
                        return !php_parallel_events_poll_expired(poll, events);
                }

                remaining = poll->stop - now;
                timeout.tv_sec = (long)(remaining / PHP_PARALLEL_EVENTS_MICRO_IN_SEC);
                timeout.tv_usec = (long)(remaining % PHP_PARALLEL_EVENTS_MICRO_IN_SEC);
                timeout_pointer = &timeout;
        }

        result = select(maximum + 1, &readfds, NULL, NULL, timeout_pointer);

        if (result >= 0) {
                return result > 0 || !php_parallel_events_poll_expired(poll, events);
        }

        if (errno == EINTR) {
                return true;
        }

        poll->native = false;
        return php_parallel_events_poll_busy(poll, events);
}
#endif

static zend_always_inline bool php_parallel_events_poll_wait(php_parallel_events_poll_t *poll,
                                                             php_parallel_events_t      *events)
{
#ifndef _WIN32
        if (poll->native) {
                return php_parallel_events_poll_native(poll, events);
        }
#endif

        return php_parallel_events_poll_busy(poll, events);
}

static zend_always_inline bool php_parallel_events_poll_begin_link(php_parallel_events_t       *events,
                                                                   php_parallel_events_state_t *state,
                                                                   zend_string *name, zend_object *object)
{
        php_parallel_channel_t *channel = php_parallel_channel_fetch(object);

        php_parallel_link_lock(channel->link);

        if (php_parallel_link_closed(channel->link)) {
                state->closed = 1;
        } else {
                if (php_parallel_events_input_exists(&events->input, name)) {
                        state->writable = php_parallel_link_writable(channel->link);
                } else {
                        state->readable = php_parallel_link_readable(channel->link);
                }
        }

        if (state->readable || state->writable || state->closed) {
                state->type = PHP_PARALLEL_EVENTS_LINK;
                state->name = name;
                state->object = object;

                return 1;
        }

        php_parallel_link_unlock(channel->link);
        return 0;
}

static zend_always_inline bool php_parallel_events_poll_begin_future(php_parallel_events_t       *events,
                                                                     php_parallel_events_state_t *state,
                                                                     zend_string *name, zend_object *object)
{
        php_parallel_future_t *future = php_parallel_future_fetch(object);

        php_parallel_future_lock(future);

        state->readable = php_parallel_future_readable(future);

        if (state->readable) {
                state->type = PHP_PARALLEL_EVENTS_FUTURE;
                state->name = name;
                state->object = object;

                return 1;
        }

        php_parallel_future_unlock(future);
        return 0;
}

static zend_always_inline bool php_parallel_events_poll_begin(php_parallel_events_t       *events,
                                                              php_parallel_events_state_t *state)
{
        uint32_t size = events->targets.nNumUsed;
        uint32_t index = (uint32_t)php_mt_rand_range(0, (zend_long)size - 1);
        uint32_t scanned;

        for (scanned = 0; scanned < size; scanned++) {
                Bucket      *bucket = &events->targets.arData[index];
                zend_object *object;

                if (++index == size) {
                        index = 0;
                }

                if (Z_ISUNDEF(bucket->val)) {
                        continue;
                }

                memset(state, 0, sizeof(php_parallel_events_state_t));
                object = Z_OBJ(bucket->val);

                if (instanceof_function(object->ce, php_parallel_channel_ce)) {
                        if (php_parallel_events_poll_begin_link(events, state, bucket->key, object)) {
                                return 1;
                        }
                } else if (php_parallel_events_poll_begin_future(events, state, bucket->key, object)) {
                        return 1;
                }
        }

        return 0;
}

static zend_always_inline bool php_parallel_events_poll_link(php_parallel_events_t       *events,
                                                             php_parallel_events_state_t *state, zval *retval)
{
        php_parallel_channel_t *channel = php_parallel_channel_fetch(state->object);
        zval                   *input;

        if (state->closed) {
                php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_CLOSE, state->name, state->object, NULL,
                                                    retval);

                return 1;
        } else {
                if ((input = php_parallel_events_input_find(&events->input, state->name))) {

                        if (state->writable) {

                                if (php_parallel_link_send_event(channel->link, input)) {

                                        php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_WRITE, state->name,
                                                                            state->object, NULL, retval);

                                        return 1;
                                }
                        }
                } else {
                        if (state->readable) {
                                zval read;

                                if (php_parallel_link_recv(channel->link, &read)) {

                                        php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_READ, state->name,
                                                                            state->object, &read, retval);

                                        return 1;
                                }
                        }
                }
        }

        return 0;
}

static zend_always_inline bool php_parallel_events_poll_future(php_parallel_events_t       *events,
                                                               php_parallel_events_state_t *state, zval *retval)
{
        if (state->readable) {
                zval                             read;
                php_parallel_events_event_type_t type = PHP_PARALLEL_EVENTS_EVENT_READ;
                php_parallel_future_t           *future = php_parallel_future_fetch(state->object);

                ZVAL_NULL(&read);

                if (php_parallel_monitor_check(future->monitor, PHP_PARALLEL_KILLED)) {
                        type = PHP_PARALLEL_EVENTS_EVENT_KILL;
                } else if (php_parallel_monitor_check(future->monitor, PHP_PARALLEL_CANCELLED)) {
                        type = PHP_PARALLEL_EVENTS_EVENT_CANCEL;
                } else {
                        if (php_parallel_monitor_check(future->monitor, PHP_PARALLEL_ERROR)) {
                                type = PHP_PARALLEL_EVENTS_EVENT_ERROR;
                        }

                        php_parallel_future_value(future, &read);
                }

                php_parallel_events_event_construct(events, type, state->name, state->object, &read, retval);

                return 1;
        }

        return 0;
}

void php_parallel_events_poll(php_parallel_events_t *events, zval *retval)
{
        php_parallel_events_poll_t *poll;

        if (!(poll = php_parallel_events_poll_init(events))) {
                ZVAL_NULL(retval);
                return;
        }

        do {
                if (!php_parallel_events_poll_begin(events, &poll->state)) {
                        if (!events->blocking) {
                                php_parallel_events_poll_free(poll);
                                return;
                        }

                        if (poll->block.fci.size) {
                                zend_call_function(&poll->block.fci, &poll->block.fcc);

                                if (zend_is_true(&poll->block.ival)) {
                                        zval_ptr_dtor(&poll->block.ival);
                                        php_parallel_events_poll_free(poll);
                                        return;
                                }

                                zval_ptr_dtor(&poll->block.ival);

                                if (php_parallel_events_poll_expired(poll, events)) {
                                        php_parallel_events_poll_free(poll);
                                        return;
                                }

                                continue;
                        }

                        if (!php_parallel_events_poll_wait(poll, events)) {
                                php_parallel_events_poll_free(poll);
                                return;
                        }

                        continue;
                }

                if (poll->state.type == PHP_PARALLEL_EVENTS_LINK) {
                        if (php_parallel_events_poll_link(events, &poll->state, retval)) {
                                break;
                        }
                } else {
                        if (php_parallel_events_poll_future(events, &poll->state, retval)) {
                                break;
                        }
                }

                /* Unlock but don't free - we're continuing the loop */
                php_parallel_events_poll_unlock(poll);
        } while (1);

        php_parallel_events_poll_end(poll);
}
#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_EVENTS_POLL
19	#define HAVE_PARALLEL_EVENTS_POLL
20
21	#include "parallel.h"
22
23	#if PHP_VERSION_ID >= 80300
24	#include "Zend/zend_hrtime.h"
25	#endif
26
27	#include <errno.h>
28	#include <time.h>
29
30	#ifdef _WIN32
31	#if PHP_VERSION_ID < 80300
32	#include <windows.h>
33	#endif
34	#else
35	#include <sys/select.h>
36	#endif
37
38	#define PHP_PARALLEL_EVENTS_MICRO_IN_SEC 1000000ULL
39
40	#if PHP_VERSION_ID >= 80400
41	#include "ext/random/php_random.h"
42	#else
43	#include "ext/standard/php_mt_rand.h"
44	#endif
45
46	typedef struct _php_parallel_events_poll_t {
47	uint32_t try;
48	uint64_t stop;
49	bool native;
50	struct {
51	zend_fcall_info fci;
52	zend_fcall_info_cache fcc;
53	zval ival;
54	} block;
55	php_parallel_events_state_t state;
56	} php_parallel_events_poll_t;
57
58	static zend_always_inline uint64_t php_parallel_events_poll_now(void)	252✔
59	{
60	#if PHP_VERSION_ID >= 80300
61	return zend_hrtime() / 1000;	36✔
62	#elif defined(_WIN32)
63	LARGE_INTEGER now, frequency;
64
65	QueryPerformanceCounter(&now);
66	QueryPerformanceFrequency(&frequency);
67
68	return (now.QuadPart / frequency.QuadPart) * PHP_PARALLEL_EVENTS_MICRO_IN_SEC +
69	(now.QuadPart % frequency.QuadPart) * PHP_PARALLEL_EVENTS_MICRO_IN_SEC / frequency.QuadPart;
70	#else
71	struct timespec now;	126✔
72
73	clock_gettime(CLOCK_MONOTONIC, &now);	36✔
74
75	return (uint64_t)now.tv_sec * PHP_PARALLEL_EVENTS_MICRO_IN_SEC + now.tv_nsec / 1000;	126✔
76	#endif
77	}
78
79	static zend_always_inline php_parallel_events_poll_t php_parallel_events_poll_init(php_parallel_events_t events)	4,086✔
80	{
81	php_parallel_events_poll_t *poll;	4,086✔
82
83	if (zend_hash_num_elements(&events->targets) == 0) {	4,086✔
84	return NULL;
85	}
86
87	poll = (php_parallel_events_poll_t *)pecalloc(1, sizeof(php_parallel_events_poll_t), 1);	4,050✔
88	#ifndef _WIN32
89	poll->native = true;	4,050✔
90	#endif
91
92	if (events->timeout > -1) {	4,050✔
93	poll->stop = php_parallel_events_poll_now() + (uint64_t)events->timeout;	162✔
94	}
95
96	if (!Z_ISUNDEF(events->blocker)) {	4,050✔
97	zend_fcall_info_init(&events->blocker, 0, &poll->block.fci, &poll->block.fcc, NULL, NULL);	18✔
98	poll->block.fci.retval = &poll->block.ival;	18✔
99	} else {
100	memset(&poll->block, 0, sizeof(poll->block));	4,032✔
101	}
102
103	return poll;
104	}
105
106	static zend_always_inline void php_parallel_events_poll_free(php_parallel_events_poll_t *poll) { pefree(poll, 1); }	4,050✔
107
108	static zend_always_inline void php_parallel_events_poll_unlock(php_parallel_events_poll_t *poll)	3,942✔
109	{
110	if (poll->state.type == PHP_PARALLEL_EVENTS_LINK) {	3,942✔
111	php_parallel_channel_t *channel = php_parallel_channel_fetch(poll->state.object);	3,798✔
112
113	php_parallel_link_unlock(channel->link);	3,798✔
114	} else {
115	php_parallel_future_t *future = php_parallel_future_fetch(poll->state.object);	144✔
116
117	php_parallel_future_unlock(future);	144✔
118	}
119	}
120
121	static zend_always_inline void php_parallel_events_poll_end(php_parallel_events_poll_t *poll)	3,942✔
122	{
123	php_parallel_events_poll_unlock(poll);	7,884✔
124	php_parallel_events_poll_free(poll);	3,942✔
125	}	3,942✔
126
127	static zend_always_inline bool php_parallel_events_poll_expired(php_parallel_events_poll_t *poll,	48,948✔
128	php_parallel_events_t *events)
129	{
130	if (events->timeout > -1 && php_parallel_events_poll_now() >= poll->stop) {	144✔
NEW 131	php_parallel_exception_ex(php_parallel_events_error_timeout_ce, "timeout occured");	×
NEW 132	return 1;	×
133	}
134
135	return 0;
136	}
137
138	static zend_always_inline bool php_parallel_events_poll_busy(php_parallel_events_poll_t *poll,	48,876✔
139	php_parallel_events_t *events)
140	{
141	if ((poll->try++ % 10) == 0) {	48,876✔
142	usleep(1);	4,920✔
143	}
144
145	return !php_parallel_events_poll_expired(poll, events);	48,876✔
146	}
147
148	#ifndef _WIN32
149	static zend_always_inline bool php_parallel_events_poll_descriptors(php_parallel_events_t events, fd_set readfds,
150	int *maximum)
151	{
152	uint32_t index;
153
154	FD_ZERO(readfds);	60,128✔
155	*maximum = -1;
156
157	for (index = 0; index < events->targets.nNumUsed; index++) {	16,597✔
158	Bucket *bucket = &events->targets.arData[index];	12,875✔
159	zend_object *object;	12,875✔
160	int descriptor;	12,875✔
161
162	if (Z_ISUNDEF(bucket->val)) {	12,875✔
NEW 163	continue;	×
164	}
165
166	object = Z_OBJ(bucket->val);	12,875✔
167
168	if (instanceof_function(object->ce, php_parallel_channel_ce)) {	12,911✔
169	php_parallel_channel_t *channel = php_parallel_channel_fetch(object);	12,839✔
170	bool writable = php_parallel_events_input_exists(&events->input, bucket->key);	12,839✔
171
172	php_parallel_link_lock(channel->link);	12,839✔
173	descriptor = php_parallel_link_notify(channel->link, writable);	12,839✔
174	php_parallel_link_unlock(channel->link);	12,839✔
175	} else {
176	php_parallel_future_t *future = php_parallel_future_fetch(object);	36✔
177
178	php_parallel_future_lock(future);	36✔
179	descriptor = php_parallel_monitor_notify(future->monitor);	36✔
180	php_parallel_future_unlock(future);	36✔
181	}
182
183	if (descriptor < 0 \|\| descriptor >= FD_SETSIZE) {	12,875✔
184	return false;
185	}
186
187	FD_SET(descriptor, readfds);	12,839✔
188	if (descriptor > *maximum) {	12,839✔
189	*maximum = descriptor;
190	}
191	}
192
193	return *maximum >= 0;	3,722✔
194	}
195
196	static zend_always_inline bool php_parallel_events_poll_native(php_parallel_events_poll_t *poll,
197	php_parallel_events_t *events)
198	{
199	fd_set readfds;
200	int maximum;
201	int result;
202	struct timeval timeout;
203	struct timeval *timeout_pointer = NULL;	63,886✔
204
205	if (!php_parallel_events_poll_descriptors(events, &readfds, &maximum)) {	3,758✔
206	poll->native = false;	36✔
207	return php_parallel_events_poll_busy(poll, events);	72✔
208	}
209
210	if (events->timeout > -1) {	3,722✔
211	uint64_t now = php_parallel_events_poll_now();	72✔
212	uint64_t remaining;	72✔
213
214	if (now >= poll->stop) {	72✔
NEW 215	return !php_parallel_events_poll_expired(poll, events);	×
216	}
217
218	remaining = poll->stop - now;	72✔
219	timeout.tv_sec = (long)(remaining / PHP_PARALLEL_EVENTS_MICRO_IN_SEC);	72✔
220	timeout.tv_usec = (long)(remaining % PHP_PARALLEL_EVENTS_MICRO_IN_SEC);	72✔
221	timeout_pointer = &timeout;	72✔
222	}
223
224	result = select(maximum + 1, &readfds, NULL, NULL, timeout_pointer);	3,722✔
225
226	if (result >= 0) {	3,722✔
227	return result > 0 \|\| !php_parallel_events_poll_expired(poll, events);	3,794✔
228	}
229
NEW 230	if (errno == EINTR) {	×
231	return true;
232	}
233
NEW 234	poll->native = false;	×
NEW 235	return php_parallel_events_poll_busy(poll, events);	×
236	}
237	#endif
238
239	static zend_always_inline bool php_parallel_events_poll_wait(php_parallel_events_poll_t *poll,	52,598✔
240	php_parallel_events_t *events)
241	{
242	#ifndef _WIN32
243	if (poll->native) {	52,598✔
244	return php_parallel_events_poll_native(poll, events);	3,758✔
245	}
246	#endif
247
248	return php_parallel_events_poll_busy(poll, events);	97,680✔
249	}
250
251	static zend_always_inline bool php_parallel_events_poll_begin_link(php_parallel_events_t *events,	33,421✔
252	php_parallel_events_state_t *state,
253	zend_string name, zend_object object)
254	{
255	php_parallel_channel_t *channel = php_parallel_channel_fetch(object);	33,421✔
256
257	php_parallel_link_lock(channel->link);	33,421✔
258
259	if (php_parallel_link_closed(channel->link)) {	33,421✔
260	state->closed = 1;	54✔
261	} else {
262	if (php_parallel_events_input_exists(&events->input, name)) {	33,367✔
263	state->writable = php_parallel_link_writable(channel->link);	3,679✔
264	} else {
265	state->readable = php_parallel_link_readable(channel->link);	29,688✔
266	}
267	}
268
269	if (state->readable \|\| state->writable \|\| state->closed) {	33,421✔
270	state->type = PHP_PARALLEL_EVENTS_LINK;	3,798✔
271	state->name = name;	3,798✔
272	state->object = object;	3,798✔
273
274	return 1;	3,798✔
275	}
276
277	php_parallel_link_unlock(channel->link);	29,623✔
278	return 0;	29,623✔
279	}
280
281	static zend_always_inline bool php_parallel_events_poll_begin_future(php_parallel_events_t *events,	49,045✔
282	php_parallel_events_state_t *state,
283	zend_string name, zend_object object)
284	{
285	php_parallel_future_t *future = php_parallel_future_fetch(object);	49,045✔
286
287	php_parallel_future_lock(future);	49,045✔
288
289	state->readable = php_parallel_future_readable(future);	49,045✔
290
291	if (state->readable) {	49,045✔
292	state->type = PHP_PARALLEL_EVENTS_FUTURE;	144✔
293	state->name = name;	144✔
294	state->object = object;	144✔
295
296	return 1;	144✔
297	}
298
299	php_parallel_future_unlock(future);	48,901✔
300	return 0;	48,901✔
301	}
302
303	static zend_always_inline bool php_parallel_events_poll_begin(php_parallel_events_t *events,	56,576✔
304	php_parallel_events_state_t *state)
305	{
306	uint32_t size = events->targets.nNumUsed;	56,576✔
307	uint32_t index = (uint32_t)php_mt_rand_range(0, (zend_long)size - 1);	113,152✔
308	uint32_t scanned;	56,576✔
309
310	for (scanned = 0; scanned < size; scanned++) {	135,113✔
311	Bucket *bucket = &events->targets.arData[index];	82,479✔
312	zend_object *object;	82,479✔
313
314	if (++index == size) {	82,479✔
315	index = 0;	56,556✔
316	}
317
318	if (Z_ISUNDEF(bucket->val)) {	82,479✔
319	continue;	13✔
320	}
321
322	memset(state, 0, sizeof(php_parallel_events_state_t));	82,466✔
323	object = Z_OBJ(bucket->val);	82,466✔
324
325	if (instanceof_function(object->ce, php_parallel_channel_ce)) {	131,511✔
326	if (php_parallel_events_poll_begin_link(events, state, bucket->key, object)) {	63,044✔
327	return 1;	3,798✔
328	}
329	} else if (php_parallel_events_poll_begin_future(events, state, bucket->key, object)) {	97,946✔
330	return 1;	144✔
331	}
332	}
333
334	return 0;
335	}
336
337	static zend_always_inline bool php_parallel_events_poll_link(php_parallel_events_t *events,	3,798✔
338	php_parallel_events_state_t state, zval retval)
339	{
340	php_parallel_channel_t *channel = php_parallel_channel_fetch(state->object);	3,798✔
341	zval *input;	3,798✔
342
343	if (state->closed) {	3,798✔
344	php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_CLOSE, state->name, state->object, NULL,	54✔
345	retval);
346
347	return 1;	54✔
348	} else {
349	if ((input = php_parallel_events_input_find(&events->input, state->name))) {	3,744✔
350
351	if (state->writable) {	1,854✔
352
353	if (php_parallel_link_send_event(channel->link, input)) {	1,854✔
354
355	php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_WRITE, state->name,	1,854✔
356	state->object, NULL, retval);
357
358	return 1;	1,854✔
359	}
360	}
361	} else {
362	if (state->readable) {	1,890✔
363	zval read;	1,890✔
364
365	if (php_parallel_link_recv(channel->link, &read)) {	1,890✔
366
367	php_parallel_events_event_construct(events, PHP_PARALLEL_EVENTS_EVENT_READ, state->name,	1,890✔
368	state->object, &read, retval);
369
370	return 1;	1,890✔
371	}
372	}
373	}
374	}
375
376	return 0;
377	}
378
379	static zend_always_inline bool php_parallel_events_poll_future(php_parallel_events_t *events,	144✔
380	php_parallel_events_state_t state, zval retval)
381	{
382	if (state->readable) {	144✔
383	zval read;	144✔
384	php_parallel_events_event_type_t type = PHP_PARALLEL_EVENTS_EVENT_READ;	144✔
385	php_parallel_future_t *future = php_parallel_future_fetch(state->object);	144✔
386
387	ZVAL_NULL(&read);	144✔
388
389	if (php_parallel_monitor_check(future->monitor, PHP_PARALLEL_KILLED)) {	144✔
390	type = PHP_PARALLEL_EVENTS_EVENT_KILL;
391	} else if (php_parallel_monitor_check(future->monitor, PHP_PARALLEL_CANCELLED)) {	126✔
392	type = PHP_PARALLEL_EVENTS_EVENT_CANCEL;
393	} else {
394	if (php_parallel_monitor_check(future->monitor, PHP_PARALLEL_ERROR)) {	108✔
395	type = PHP_PARALLEL_EVENTS_EVENT_ERROR;	36✔
396	}
397
398	php_parallel_future_value(future, &read);	108✔
399	}
400
401	php_parallel_events_event_construct(events, type, state->name, state->object, &read, retval);	144✔
402
403	return 1;	144✔
404	}
405
406	return 0;
407	}
408
409	void php_parallel_events_poll(php_parallel_events_t events, zval retval)	4,086✔
410	{
411	php_parallel_events_poll_t *poll;	4,086✔
412
413	if (!(poll = php_parallel_events_poll_init(events))) {	4,086✔
414	ZVAL_NULL(retval);	36✔
415	return;	36✔
416	}
417
418	do {	56,576✔
419	if (!php_parallel_events_poll_begin(events, &poll->state)) {	3,942✔
420	if (!events->blocking) {	52,634✔
421	php_parallel_events_poll_free(poll);	18✔
422	return;	18✔
423	}
424
425	if (poll->block.fci.size) {	52,616✔
426	zend_call_function(&poll->block.fci, &poll->block.fcc);	18✔
427
428	if (zend_is_true(&poll->block.ival)) {	18✔
429	zval_ptr_dtor(&poll->block.ival);	18✔
430	php_parallel_events_poll_free(poll);	18✔
431	return;	18✔
432	}
433
434	zval_ptr_dtor(&poll->block.ival);	×
435
NEW 436	if (php_parallel_events_poll_expired(poll, events)) {	×
NEW 437	php_parallel_events_poll_free(poll);	×
NEW 438	return;	×
439	}
440
NEW 441	continue;	×
442	}
443
444	if (!php_parallel_events_poll_wait(poll, events)) {	105,196✔
445	php_parallel_events_poll_free(poll);	72✔
446	return;	72✔
447	}
448
449	continue;	52,526✔
450	}
451
452	if (poll->state.type == PHP_PARALLEL_EVENTS_LINK) {	3,942✔
453	if (php_parallel_events_poll_link(events, &poll->state, retval)) {	7,542✔
454	break;
455	}
456	} else {
457	if (php_parallel_events_poll_future(events, &poll->state, retval)) {	288✔
458	break;
459	}
460	}
461
462	/* Unlock but don't free - we're continuing the loop */
463	php_parallel_events_poll_unlock(poll);	56,576✔
464	} while (1);
465
466	php_parallel_events_poll_end(poll);	3,942✔
467	}
468	#endif

krakjoe / parallel / 30916206264

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