fix: curl_exec() in a coroutine resumes after curl_pause(CONT)

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]);

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--
+<?php
+if (!class_exists('Fiber')) die('skip requires Fibers');
+if (!function_exists('proc_open')) die('skip requires proc_open');
+if (trim((string) shell_exec(escapeshellarg(PHP_BINARY) . ' -r "echo function_exists(\'curl_pause\') ? 1 : 0;"')) !== '1') die('skip requires ext/curl');
+?>
+--FILE--
+<?php
+// A server in its own process sending a 4 MiB body.
+$server = '
+$s = stream_socket_server("tcp://127.0.0.1:0");
+$n = stream_socket_get_name($s, false); echo substr($n, strrpos($n, ":") + 1), "\n"; flush();
+while ($c = @stream_socket_accept($s, 20)) {
+    fread($c, 8192);
+    $len = 4 << 20;
+    fwrite($c, "HTTP/1.0 200 OK\r\nContent-Length: $len\r\nConnection: close\r\n\r\n");
+    for ($i = 0; $i < $len; $i += 65536) fwrite($c, str_repeat("x", 65536));
+    fclose($c);
+}';
+$srv = proc_open([PHP_BINARY, '-n', '-r', $server], [1 => ['pipe', 'w']], $pipes);
+$port = trim(fgets($pipes[1]));
+
+$child = tempnam(sys_get_temp_dir(), 'phasync_cp_') . '.php';
+file_put_contents($child, '<?php $port = ' . (int) $port . '; require ' . var_export(__DIR__ . '/loop.inc', true) . ';' . <<<'CHILD'
+
+// A buffers in its WRITEFUNCTION and pauses above 256 KiB; B drains the buffer
+// and resumes the transfer with curl_pause(CONT).
+$ch = curl_init("http://127.0.0.1:$port/");
+$buf = ''; $total = 0; $paused = false; $pauses = 0; $done = false; $r = null;
+curl_setopt($ch, CURLOPT_WRITEFUNCTION, function ($ch, $data) use (&$buf, &$paused, &$pauses) {
+    if (strlen($buf) >= 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)

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);