Refactor curl native impl (#5743)

* Optimize the implementation of curl coroutines to support KeepAlive, reduce memory allocation, and align the logic of Multi::exec() with that of Multi::select(). Refer: GH-5474 & GH-5101

* Copy handle->sockets to a new unordered_map to avoid memory errors caused by deleting elements during iteration.

* fix --filter=[unit]

* fix 2 --filter=[unit]

Author: matyhtf

ext-src/php_swoole_curl.h

@@ -53,11 +53,15 @@ struct HandleSocket {
 struct Handle {
     CURL *cp;
     Multi *multi;
+    // This is only for the swoole_curl_easy_perform function, and it has a one-to-one relationship with the curl
+    // handle. It must be destroyed when the curl handle is released.
+    Multi *easy_multi;
     std::unordered_map<int, HandleSocket *> sockets;
 
     Handle(CURL *_cp) {
         cp = _cp;
         multi = nullptr;
+        easy_multi = nullptr;
     }
 
     HandleSocket *create_socket(curl_socket_t sockfd);
@@ -79,17 +83,17 @@ class Multi {
     long timeout_ms_ = 0;
     Coroutine *co = nullptr;
     int running_handles_ = 0;
-    int last_sockfd;
     int event_count_ = 0;
     bool defer_callback = false;
-    std::unique_ptr<Selector> selector;
+    Selector selector;
 
     CURLcode read_info();
 
     HandleSocket *create_socket(Handle *handle, curl_socket_t sockfd);
 
     void set_event(CURL *cp, void *socket_ptr, curl_socket_t sockfd, int action);
     void del_event(CURL *cp, void *socket_ptr, curl_socket_t sockfd);
+    void selector_finish();
 
     void add_timer(long timeout_ms) {
         if (timer && swoole_timer_is_available()) {
@@ -139,10 +143,6 @@ class Multi {
         return running_handles_;
     }
 
-    void set_selector(Selector *_selector) {
-        selector.reset(_selector);
-    }
-
     CURLMcode add_handle(Handle *handle);
     CURLMcode remove_handle(Handle *handle);
 

ext-src/swoole_curl.cc

@@ -47,6 +47,11 @@ void destroy_handle(CURL *cp) {
     }
     auto handle = iter->second;
     handle_buckets.erase(iter);
+
+    if (handle->easy_multi) {
+        delete handle->easy_multi;
+    }
+
     delete handle;
     swoole_trace_log(SW_TRACE_CO_CURL, SW_ECHO_RED " handle=%p, curl=%p", "[DESTROY]", handle, cp);
 }
@@ -191,10 +196,6 @@ void Multi::set_event(CURL *cp, void *socket_ptr, curl_socket_t sockfd, int acti
 }
 
 CURLMcode Multi::add_handle(Handle *handle) {
-    if (handle == nullptr) {
-        php_swoole_fatal_error(E_WARNING, "The given handle is not initialized in coroutine");
-        return CURLM_INTERNAL_ERROR;
-    }
     auto retval = curl_multi_add_handle(multi_handle_, handle->cp);
     if (retval == CURLM_OK) {
         handle->multi = this;
@@ -242,21 +243,8 @@ CURLcode Multi::exec(Handle *handle) {
             break;
         }
 
-        int sockfd = last_sockfd;
-        int bitmask = 0;
-        if (sockfd >= 0) {
-            auto it = handle->sockets.find(sockfd);
-            if (it != handle->sockets.end()) {
-                curl_socket = it->second;
-                bitmask = curl_socket->event_bitmask;
-                if (!curl_socket->socket->removed && swoole_event_del(curl_socket->socket) == SW_OK) {
-                    event_count_--;
-                }
-            }
-        }
-        del_timer();
+        selector_finish();
 
-        curl_multi_socket_action(multi_handle_, sockfd, bitmask, &running_handles_);
         swoole_trace_log(SW_TRACE_CO_CURL,
                          "curl_multi_socket_action: handle=%p, sockfd=%d, bitmask=%d, running_handles_=%d",
                          handle,
@@ -267,38 +255,17 @@ CURLcode Multi::exec(Handle *handle) {
             break;
         }
         set_timer();
-        if (sockfd >= 0) {
-            auto it = handle->sockets.find(sockfd);
-            if (it != handle->sockets.end()) {
-                curl_socket = it->second;
-                if (curl_socket->socket && curl_socket->socket->removed) {
-                    if (swoole_event_add(curl_socket->socket, get_event(curl_socket->action)) == SW_OK) {
-                        event_count_++;
-                    }
-                }
-            }
-        }
+    }
 
-        if (!timer) {
-            bool removed = true;
-            for (auto it = handle->sockets.begin(); it != handle->sockets.end();) {
-                curl_socket = it->second;
-                if (curl_socket->socket) {
-                    if (curl_socket->socket->removed) {
-                        it = handle->sockets.erase(it);
-                        delete curl_socket;
-                        continue;
-                    } else {
-                        removed = false;
-                    }
-                }
-                ++it;
-            }
-            if (removed) {
-                break;
+    for (auto it : handle->sockets) {
+        curl_socket = it.second;
+        if (curl_socket->socket && !curl_socket->socket->removed) {
+            if (swoole_event_del(curl_socket->socket) == SW_OK) {
+                event_count_--;
             }
         }
     }
+    del_timer();
 
     CURLcode retval = read_info();
     remove_handle(handle);
@@ -347,6 +314,33 @@ int Multi::handle_timeout(CURLM *mh, long timeout_ms, void *userp) {
     return 0;
 }
 
+void Multi::selector_finish() {
+    del_timer();
+
+    if (selector.timer_callback) {
+        selector.timer_callback = false;
+        curl_multi_socket_action(multi_handle_, CURL_SOCKET_TIMEOUT, 0, &running_handles_);
+        swoole_trace_log(SW_TRACE_CO_CURL, "socket_action[timer], running_handles=%d", running_handles_);
+    }
+
+    for (auto handle : selector.active_handles) {
+        /**
+         * In `curl_multi_socket_action`, `Handle::destroy_socket()` may be invoked,
+         * which will remove entries from the `unordered_map`.
+         * In C++, removing elements during iteration can render the iterator invalid; hence,
+         * it's necessary to copy `handle->sockets` into a new `unordered_map`.
+         */
+        auto sockets = handle->sockets;
+        for (auto it : sockets) {
+            HandleSocket *handle_socket = it.second;
+            curl_multi_socket_action(
+                multi_handle_, handle_socket->event_fd, handle_socket->event_bitmask, &running_handles_);
+            swoole_trace_log(SW_TRACE_CO_CURL, "socket_action[socket], running_handles=%d", running_handles_);
+        }
+    }
+    selector.active_handles.clear();
+}
+
 long Multi::select(php_curlm *mh, double timeout) {
     if (zend_llist_count(&mh->easyh) == 0) {
         return 0;
@@ -399,7 +393,7 @@ long Multi::select(php_curlm *mh, double timeout) {
 
     swoole_trace_log(SW_TRACE_CO_CURL, "yield timeout, count=%lu", zend_llist_count(&mh->easyh));
 
-    auto count = selector->active_handles.size();
+    auto count = selector.active_handles.size();
 
     for (zend_llist_element *element = mh->easyh.head; element; element = element->next) {
         zval *z_ch = (zval *) element->data;
@@ -419,44 +413,17 @@ long Multi::select(php_curlm *mh, double timeout) {
             }
         }
     }
-    del_timer();
 
-    if (selector->timer_callback) {
-        selector->timer_callback = false;
-        curl_multi_socket_action(multi_handle_, CURL_SOCKET_TIMEOUT, 0, &running_handles_);
-        swoole_trace_log(SW_TRACE_CO_CURL, "socket_action[timer], running_handles=%d", running_handles_);
-    }
-
-    for (auto iter = selector->active_handles.begin(); iter != selector->active_handles.end(); iter++) {
-        Handle *handle = *iter;
-        if (handle) {
-            for (auto it = handle->sockets.begin(); it != handle->sockets.end();) {
-                HandleSocket *handle_socket = it->second;
-                it++;
-                curl_multi_socket_action(
-                    multi_handle_, handle_socket->event_fd, handle_socket->event_bitmask, &running_handles_);
-                swoole_trace_log(SW_TRACE_CO_CURL, "socket_action[socket], running_handles=%d", running_handles_);
-            }
-        }
-    }
-
-    selector->active_handles.clear();
+    selector_finish();
 
     return count;
 }
 
 void Multi::callback(Handle *handle, int event_bitmask, int sockfd) {
     swoole_trace_log(
         SW_TRACE_CO_CURL, "handle=%p, event_bitmask=%d, co=%p, sockfd=%d", handle, event_bitmask, co, sockfd);
-    if (handle) {
-        last_sockfd = sockfd;
-    } else {
-        last_sockfd = -1;
-    }
-    if (selector.get()) {
-        if (!handle) {
-            selector->timer_callback = true;
-        }
+    if (!handle) {
+        selector.timer_callback = true;
     }
     if (!co) {
         if (handle) {
@@ -470,8 +437,8 @@ void Multi::callback(Handle *handle, int event_bitmask, int sockfd) {
         }
         return;
     }
-    if (selector.get() && handle) {
-        selector->active_handles.insert(handle);
+    if (handle) {
+        selector.active_handles.insert(handle);
     }
     if (defer_callback) {
         return;
@@ -490,10 +457,11 @@ void Multi::callback(Handle *handle, int event_bitmask, int sockfd) {
 }  // namespace swoole
 
 CURLcode swoole_curl_easy_perform(CURL *cp) {
-    Multi *multi = new Multi();
-    CURLcode error = multi->exec(swoole::curl::get_handle(cp));
-    delete multi;
-    return error;
+    auto handle = swoole::curl::get_handle(cp);
+    if (!handle->easy_multi) {
+        handle->easy_multi = new Multi();
+    }
+    return handle->easy_multi->exec(handle);
 }
 
 php_curl *swoole_curl_get_handle(zval *zid, bool exclusive, bool required) {
@@ -503,6 +471,10 @@ php_curl *swoole_curl_get_handle(zval *zid, bool exclusive, bool required) {
     }
     if (exclusive && swoole_coroutine_is_in()) {
         auto handle = swoole::curl::get_handle(ch->cp);
+        if (required && !handle) {
+            php_swoole_fatal_error(E_WARNING, "The given handle is not initialized in coroutine");
+            return nullptr;
+        }
         if (handle && handle->multi && handle->multi->check_bound_co() == nullptr) {
             return nullptr;
         }

tests/swoole_curl/keepalive.phpt

@@ -0,0 +1,75 @@
+--TEST--
+swoole_curl: keepalive
+--SKIPIF--
+<?php
+require __DIR__ . '/../include/skipif.inc';
+?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+use Swoole\Http\Request;
+use Swoole\Http\Response;
+use Swoole\Runtime;
+
+use function Swoole\Coroutine\run;
+$pm = new SwooleTest\ProcessManager;
+
+$pm->parentFunc = function () use ($pm) {
+    Runtime::enableCoroutine(SWOOLE_HOOK_NATIVE_CURL);
+    run(function () use ($pm) {
+        $ch = curl_init();
+        $code = uniqid('swoole_');
+
+        $execFn = function () use ($ch, $code, $pm) {
+            $url = "http://127.0.0.1:" . $pm->getFreePort() . "/?code=" . urlencode($code);
+            curl_setopt($ch, CURLOPT_URL, $url);
+            curl_setopt($ch, CURLOPT_RETURNTRANSFER, 1);
+            curl_setopt($ch, CURLOPT_HEADER, 0);
+            curl_setopt($ch, CURLOPT_HEADERFUNCTION, function ($ch, $strHeader) {
+                return strlen($strHeader);
+            });
+            $output = curl_exec($ch);
+            $info = curl_getinfo($ch);
+            Assert::eq($output, "Hello World\n" . $code);
+            if ($output === false) {
+                exit("CURL Error:" . curl_error($ch));
+            }
+            return $info;
+        };
+
+        echo "co 1 exec\n";
+        $info1 = $execFn();
+
+        Co::sleep(0.1);
+
+        echo "co 2 exec\n";
+        $info2 = $execFn();
+
+        Assert::eq($info1['local_port'], $info2['local_port']);
+
+        curl_close($ch);
+    });
+    $pm->kill();
+    echo "Done\n";
+};
+$pm->childFunc = function () use ($pm) {
+    $http = new Swoole\Http\Server("127.0.0.1", $pm->getFreePort(), SWOOLE_PROCESS);
+    $http->set(['worker_num' => 2, 'log_file' => '/dev/null']);
+
+    $http->on("start", function ($server) use ($pm) {
+        $pm->wakeup();
+    });
+
+    $http->on("request", function (Request $request, Response $response) {
+        $response->end("Hello World\n" . $request->get['code']);
+    });
+    $http->start();
+};
+$pm->childFirst();
+$pm->run();
+?>
+--EXPECTF--
+co 1 exec
+co 2 exec
+Done

tests/swoole_curl/multi/add_after_easy_exec.phpt

@@ -0,0 +1,31 @@
+--TEST--
+swoole_curl/multi: add handle after easy exec
+--SKIPIF--
+<?php require __DIR__ . '/../../include/skipif.inc'; ?>
+<?php
+if (!extension_loaded("curl")) exit("skip curl extension not loaded");
+?>
+--FILE--
+<?php
+require __DIR__ . '/../../include/bootstrap.php';
+use Swoole\Runtime;
+
+use function Swoole\Coroutine\run;
+
+Runtime::enableCoroutine(SWOOLE_HOOK_NATIVE_CURL);
+
+use SwooleTest\CurlManager;
+
+$cm = new CurlManager();
+$cm->run(function ($host) {
+    $ch1 = curl_init();
+    curl_setopt($ch1, CURLOPT_URL, "{$host}/get.php?test=get");
+    curl_setopt($ch1, CURLOPT_RETURNTRANSFER, 1);
+    $rs = curl_exec($ch1);
+    Assert::eq($rs, "Hello World!\nHello World!");
+
+    $mh1 = curl_multi_init();
+    Assert::eq(curl_multi_add_handle($mh1, $ch1), 0);
+});
+?>
+--EXPECT--