diff --git a/io.c b/io.c index 1fbb72f..94e43ad 100644 --- a/io.c +++ b/io.c @@ -2209,6 +2209,27 @@ static bool phasync_call_named(const char *name, zval *retval, uint32_t argc, zv return ok; } +/* Header and body bytes the easy handle has received plus sent so far: whether + * the transfer moved. */ +static zend_long phasync_curl_bytes(zval *ch) +{ + static const char *names[] = { "CURLINFO_SIZE_DOWNLOAD_T", "CURLINFO_SIZE_UPLOAD_T", "CURLINFO_HEADER_SIZE" }; + zend_long sum = 0; + zval args[2], ret, *c; + + for (int i = 0; i < 3; i++) { + if ((c = zend_get_constant_str(names[i], strlen(names[i])))) { + ZVAL_COPY_VALUE(&args[0], ch); + ZVAL_COPY_VALUE(&args[1], c); + if (phasync_call_named("curl_getinfo", &ret, 2, args)) { + sum += zval_get_long(&ret); + } + zval_ptr_dtor(&ret); + } + } + return sum; +} + /* curl_exec() inside a scope, in a fiber: the same transfer on a private * curl_multi handle, waiting in the cooperative curl_multi_select(), so option * callbacks run on PHP's thread, in the coroutine. The result is native: @@ -2221,8 +2242,8 @@ static bool phasync_call_named(const char *name, zval *retval, uint32_t argc, zv static ZEND_NAMED_FUNCTION(phasync_curl_exec_override) { zval mh, args[2], ret, running, info; - zend_long result = 0; - bool added = false; + zend_long result = 0, bytes, last_bytes = -1, usec = 1000, stuck = 0; + bool added = false, ready = false; if (!phasync_reading() || EG(active_fiber) == NULL || phasync_from_c() || ZEND_NUM_ARGS() != 1 || Z_TYPE_P(ZEND_CALL_ARG(execute_data, 1)) != IS_OBJECT @@ -2256,11 +2277,29 @@ static ZEND_NAMED_FUNCTION(phasync_curl_exec_override) if (Z_TYPE_P(r) != IS_LONG || Z_LVAL_P(r) == 0) { break; } + /* A paused transfer (a callback returned CURL_*FUNC_PAUSE) keeps its + * socket ready, so the select reports activity at once and nothing + * moves: sleep through the loop (1ms, doubling to 20ms) until it does, + * so the coroutine that calls curl_pause(CONT) gets to run. A connect or + * TLS handshake also has ready rounds with no bytes counted, so only + * several in a row count as stuck. */ + bytes = phasync_curl_bytes(ZEND_CALL_ARG(execute_data, 1)); + stuck = ready && bytes == last_bytes ? stuck + 1 : 0; + if (stuck >= 4) { + if (phasync_call_sleep(phasync_sleep_handler(), usec) < 0) { + goto cancelled; + } + usec = MIN(usec * 2, 20000); + } else { + usec = 1000; + } + last_bytes = bytes; ZVAL_UNDEF(&ret); phasync_curl_select(&mh, 1.0, &ret); if (EG(exception)) { goto cancelled; } + ready = Z_TYPE(ret) == IS_LONG && Z_LVAL(ret) > 0; zval_ptr_dtor(&ret); } zval_ptr_dtor(&args[1]); diff --git a/tests/curl_exec_pause_resume.phpt b/tests/curl_exec_pause_resume.phpt new file mode 100644 index 0000000..360d9f2 --- /dev/null +++ b/tests/curl_exec_pause_resume.phpt @@ -0,0 +1,72 @@ +--TEST-- +curl_exec() in a coroutine: a WRITEFUNCTION that pauses (CURL_WRITEFUNC_PAUSE) resumes when another coroutine calls curl_pause(CONT) +--EXTENSIONS-- +phasync +--SKIPIF-- + +--FILE-- + ['pipe', 'w']], $pipes); +$port = trim(fgets($pipes[1])); + +$child = tempnam(sys_get_temp_dir(), 'phasync_cp_') . '.php'; +file_put_contents($child, '= 262144) { $paused = true; $pauses++; return CURL_WRITEFUNC_PAUSE; } + $buf .= $data; + return strlen($data); +}); +(new Loop)->runAll( + function () use ($ch, &$r, &$done) { $r = curl_exec($ch); $done = true; }, + function () use ($ch, &$buf, &$total, &$paused, &$done) { + while (!$done) { + usleep(5000); + if ($paused) { $total += strlen($buf); $buf = ''; $paused = false; curl_pause($ch, CURLPAUSE_CONT); } + } + }, +); +$total += strlen($buf); +var_dump($r, curl_errno($ch), $total, $pauses > 0); +CHILD); +$so = realpath(ini_get('extension_dir') . '/phasync.so'); +// The child is killed after 15s, so a hang fails instead of blocking the run. +$p = proc_open([PHP_BINARY, '-d', 'extension=' . $so, $child], [1 => ['pipe', 'w'], 2 => ['redirect', 1]], $cp); +stream_set_blocking($cp[1], false); +$out = ''; $deadline = microtime(true) + 15; +while (!feof($cp[1]) && microtime(true) < $deadline) { + $rd = [$cp[1]]; $w = $e = null; + if (stream_select($rd, $w, $e, 0, 200000)) $out .= fread($cp[1], 65536); +} +if (!feof($cp[1])) { proc_terminate($p, 9); $out .= "HUNG\n"; } +echo $out; +fclose($cp[1]); proc_close($p); +unlink($child); +proc_terminate($srv); +fclose($pipes[1]); +proc_close($srv); +?> +--EXPECT-- +bool(true) +int(0) +int(4194304) +bool(true) diff --git a/tests/preempt_basic.inc b/tests/preempt_basic.inc index 0323f16..4c078d4 100644 --- a/tests/preempt_basic.inc +++ b/tests/preempt_basic.inc @@ -2,6 +2,11 @@ use function phasync\ext\set_preempt_function; function threads(): int { return count(scandir('/proc/self/task')) - 2; } +// A joined thread's /proc/self/task entry can linger briefly: wait up to 200ms. +function threadsBecome(int $want): bool { + for ($i = 0; threads() !== $want && $i < 200; $i++) usleep(1000); + return threads() === $want; +} echo "-- none set\n"; $base = threads(); @@ -29,7 +34,7 @@ $b = function () {}; var_dump(set_preempt_function($b) === $a); var_dump(set_preempt_function(null) === $b); var_dump(set_preempt_function(null)); -var_dump(threads() === $base); +var_dump(threadsBecome($base)); $n = 0; $t = microtime(true); while (microtime(true) - $t < 0.3);