Merge branch 'master' into version-4.7.1

swoole · Aug 13, 2021 · 49e774c · 49e774c
2 parents fd03650 + 0ebeef2
commit 49e774c
Show file tree

Hide file tree

Showing 12 changed files with 153 additions and 31 deletions.
diff --git a/core-tests/src/network/dns.cpp b/core-tests/src/network/dns.cpp
@@ -119,7 +119,7 @@ TEST(dns, gethosts) {
     file << "       127.0.0.1 bbb.com               ccc.com      #ddd.com\n";
     file.close();
 
-    swoole::coroutine::swoole_set_hosts_path(hosts_file);
+    swoole_set_hosts_path(hosts_file);
 
     std::string ip = swoole::coroutine::get_ip_by_hosts("localhost");
     ASSERT_EQ(ip, "127.0.0.1");

diff --git a/ext-src/php_swoole.cc b/ext-src/php_swoole.cc
@@ -54,8 +54,8 @@ END_EXTERN_C()
 #include <ares.h>
 #endif
 
-using swoole::network::Socket;
 using swoole::Server;
+using swoole::network::Socket;
 
 ZEND_DECLARE_MODULE_GLOBALS(swoole)
 
@@ -385,6 +385,9 @@ void php_swoole_set_global_option(HashTable *vht) {
     if (php_swoole_array_get_value(vht, "socket_timeout", ztmp)) {
         Socket::default_read_timeout = Socket::default_write_timeout = timeout_format(ztmp);
     }
+    if (php_swoole_array_get_value(vht, "max_concurrency", ztmp)) {
+        SwooleG.max_concurrency = (uint32_t) SW_MAX(0, zval_get_long(ztmp));
+    }
 }
 
 void php_swoole_register_rshutdown_callback(swoole::Callback cb, void *private_data) {

diff --git a/ext-src/php_swoole_library.h b/ext-src/php_swoole_library.h
@@ -2,7 +2,7 @@
  * Generated by build-library.php, Please DO NOT modify!
  */
 
-/* $Id: 05b8161677541803bf1644f9a57057863c67937b */
+/* $Id: 1b421d51331d70d14a1b971c1a20c31c14b523d7 */
 
 static const char* swoole_library_source_constants =
     "\n"
@@ -3014,7 +3014,7 @@ static const char* swoole_library_source_core_database_pdo_statement_proxy =
     "        return $this->__object->setFetchMode(...$this->setFetchModeContext);\n"
     "    }\n"
     "\n"
-    "    public function bindParam($parameter, &$variable, $data_type = PDO::PARAM_STR, $length = null, $driver_options = null): bool\n"
+    "    public function bindParam($parameter, &$variable, $data_type = PDO::PARAM_STR, $length = 0, $driver_options = null): bool\n"
     "    {\n"
     "        $this->bindParamContext[$parameter] = [$variable, $data_type, $length, $driver_options];\n"
     "        return $this->__object->bindParam($parameter, $variable, $data_type, $length, $driver_options);\n"
@@ -3564,6 +3564,8 @@ static const char* swoole_library_source_core_curl_handler =
     "    /** @var callable */\n"
     "    private $writeFunction;\n"
     "\n"
+    "    private $noProgress = true;\n"
+    "\n"
     "    /** @var callable */\n"
     "    private $progressFunction;\n"
     "\n"
@@ -4072,6 +4074,9 @@ static const char* swoole_library_source_core_curl_handler =
     "            case CURLOPT_WRITEFUNCTION:\n"
     "                $this->writeFunction = $value;\n"
     "                break;\n"
+    "            case CURLOPT_NOPROGRESS:\n"
+    "                $this->noProgress = $value;\n"
+    "                break;\n"
     "            case CURLOPT_PROGRESSFUNCTION:\n"
     "                $this->progressFunction = $value;\n"
     "                break;\n"

diff --git a/ext-src/swoole_coroutine.cc b/ext-src/swoole_coroutine.cc
@@ -597,9 +597,9 @@ void PHPCoroutine::on_resume(void *arg) {
     restore_task(task);
     record_last_msec(task);
     swoole_trace_log(SW_TRACE_COROUTINE,
-               "php_coro_resume from cid=%ld to cid=%ld",
-               Coroutine::get_current_cid(),
-               task->co->get_cid());
+                     "php_coro_resume from cid=%ld to cid=%ld",
+                     Coroutine::get_current_cid(),
+                     task->co->get_cid());
 }
 
 void PHPCoroutine::on_close(void *arg) {
@@ -631,16 +631,20 @@ void PHPCoroutine::on_close(void *arg) {
         efree(task->array_walk_fci);
     }
 #endif
+
+    if (SwooleG.max_concurrency > 0 && task->pcid == -1) {
+        SwooleWG.worker_concurrency--;
+    }
     vm_stack_destroy();
     restore_task(origin_task);
 
     swoole_trace_log(SW_TRACE_COROUTINE,
-               "coro close cid=%ld and resume to %ld, %zu remained. usage size: %zu. malloc size: %zu",
-               cid,
-               origin_cid,
-               (uintmax_t) Coroutine::count() - 1,
-               (uintmax_t) zend_memory_usage(0),
-               (uintmax_t) zend_memory_usage(1));
+                     "coro close cid=%ld and resume to %ld, %zu remained. usage size: %zu. malloc size: %zu",
+                     cid,
+                     origin_cid,
+                     (uintmax_t) Coroutine::count() - 1,
+                     (uintmax_t) zend_memory_usage(0),
+                     (uintmax_t) zend_memory_usage(1));
 }
 
 void PHPCoroutine::main_func(void *arg) {
@@ -732,11 +736,31 @@ void PHPCoroutine::main_func(void *arg) {
         record_last_msec(task);
 
         swoole_trace_log(SW_TRACE_COROUTINE,
-                   "Create coro id: %ld, origin cid: %ld, coro total count: %zu, heap size: %zu",
-                   task->co->get_cid(),
-                   task->co->get_origin_cid(),
-                   (uintmax_t) Coroutine::count(),
-                   (uintmax_t) zend_memory_usage(0));
+                         "Create coro id: %ld, origin cid: %ld, coro total count: %zu, heap size: %zu",
+                         task->co->get_cid(),
+                         task->co->get_origin_cid(),
+                         (uintmax_t) Coroutine::count(),
+                         (uintmax_t) zend_memory_usage(0));
+
+        if (SwooleG.max_concurrency > 0 && task->pcid == -1) {
+            // wait until concurrency slots are available
+            while (SwooleWG.worker_concurrency > SwooleG.max_concurrency - 1) {
+                swoole_trace_log(SW_TRACE_COROUTINE,
+                                 "php_coro cid=%ld waiting for concurrency slots: max: %d, used: %d",
+                                 task->co->get_cid(),
+                                 SwooleG.max_concurrency,
+                                 SwooleWG.worker_concurrency);
+
+                swoole_event_defer(
+                    [](void *data) {
+                        Coroutine *co = (Coroutine *) data;
+                        co->resume();
+                    },
+                    (void *) task->co);
+                task->co->yield();
+            }
+            SwooleWG.worker_concurrency++;
+        }
 
         if (SwooleG.hooks[SW_GLOBAL_HOOK_ON_CORO_START]) {
             swoole_call_hook(SW_GLOBAL_HOOK_ON_CORO_START, task);

diff --git a/include/swoole.h b/include/swoole.h
@@ -645,6 +645,7 @@ struct Global {
     uint32_t pagesize;
     struct utsname uname;
     uint32_t max_sockets;
+    uint32_t max_concurrency;
     //-----------------------[Memory]--------------------------
     MemoryPool *memory_pool;
     Allocator std_allocator;
@@ -655,6 +656,7 @@ struct Global {
     double dns_cache_refresh_time;
     int dns_tries;
     std::string dns_resolvconf_path;
+    std::string dns_hosts_path;
     //-----------------------[AIO]--------------------------
     uint32_t aio_core_worker_num;
     uint32_t aio_worker_num;
@@ -701,7 +703,8 @@ SW_API const char *swoole_strerror(int code);
 SW_API void swoole_throw_error(int code);
 SW_API void swoole_set_log_level(int level);
 SW_API void swoole_set_trace_flags(int flags);
-SW_API void swoole_set_dns_server(const std::string server);
+SW_API void swoole_set_dns_server(const std::string &server);
+SW_API void swoole_set_hosts_path(const std::string &hosts_file);
 SW_API std::pair<std::string, int> swoole_get_dns_server();
 SW_API bool swoole_load_resolv_conf();
 

diff --git a/include/swoole_c_api.h b/include/swoole_c_api.h
@@ -35,6 +35,8 @@ enum swGlobalHookType {
     SW_GLOBAL_HOOK_AFTER_SERVER_SHUTDOWN,
     SW_GLOBAL_HOOK_BEFORE_WORKER_STOP,
     SW_GLOBAL_HOOK_ON_REACTOR_DESTROY,
+    SW_GLOBAL_HOOK_BEFORE_SERVER_CREATE,
+    SW_GLOBAL_HOOK_AFTER_SERVER_CREATE,
     SW_GLOBAL_HOOK_END  = SW_MAX_HOOK_TYPE - 1,
 };
 

diff --git a/include/swoole_coroutine_socket.h b/include/swoole_coroutine_socket.h
@@ -610,7 +610,6 @@ std::vector<std::string> dns_lookup_impl_with_socket(const char *domain, int fam
 std::vector<std::string> dns_lookup_impl_with_cares(const char *domain, int family, double timeout);
 #endif
 std::string get_ip_by_hosts(const std::string &domain);
-void swoole_set_hosts_path(char *hosts_file);
 //-------------------------------------------------------------------------------
 }  // namespace coroutine
 }  // namespace swoole

diff --git a/include/swoole_process_pool.h b/include/swoole_process_pool.h
@@ -93,6 +93,7 @@ struct WorkerGlobal {
     String **output_buffer;
     Worker *worker;
     time_t exit_time;
+    uint32_t worker_concurrency = 0;
 };
 
 struct Worker {

diff --git a/src/core/base.cc b/src/core/base.cc
@@ -148,6 +148,7 @@ void swoole_init(void) {
     // init global shared memory
     SwooleG.memory_pool = new swoole::GlobalMemory(SW_GLOBAL_MEMORY_PAGESIZE, true);
     SwooleG.max_sockets = SW_MAX_SOCKETS_DEFAULT;
+    SwooleG.max_concurrency = 0;
     struct rlimit rlmt;
     if (getrlimit(RLIMIT_NOFILE, &rlmt) < 0) {
         swoole_sys_warning("getrlimit() failed");
@@ -250,7 +251,7 @@ SW_API void swoole_set_trace_flags(int flags) {
     SwooleG.trace_flags = flags;
 }
 
-SW_API void swoole_set_dns_server(const std::string server) {
+SW_API void swoole_set_dns_server(const std::string &server) {
     char *_port;
     int dns_server_port = SW_DNS_SERVER_PORT;
     char dns_server_host[32];

diff --git a/src/network/dns.cc b/src/network/dns.cc
@@ -57,16 +57,20 @@ bool swoole_load_resolv_conf() {
     return true;
 }
 
+void swoole_set_hosts_path(const std::string &hosts_file) {
+    SwooleG.dns_hosts_path = hosts_file;
+}
+
 namespace swoole {
 namespace coroutine {
 
-enum swDNS_type {
+enum RecordType {
     SW_DNS_A_RECORD = 0x01,     // Lookup IPv4 address
     SW_DNS_AAAA_RECORD = 0x1c,  // Lookup IPv6 address
     SW_DNS_MX_RECORD = 0x0f     // Lookup mail server for domain
 };
 
-enum swDNS_error {
+enum DNSError {
     SW_DNS_NOT_EXIST,  // Error: address does not exist
     SW_DNS_TIMEOUT,    // Lookup time expired
     SW_DNS_ERROR       // No memory or other error
@@ -104,14 +108,13 @@ struct RR_FLAGS {
 };
 
 static uint16_t dns_request_id = 1;
-char *swoole_hosts = nullptr;
 
 static int domain_encode(const char *src, int n, char *dest);
 static void domain_decode(char *str);
 static std::string parse_ip_address(void *vaddr, int type);
 
 std::string get_ip_by_hosts(const std::string &search_domain) {
-    std::ifstream file(swoole_hosts ? swoole_hosts : SW_PATH_HOSTS);
+    std::ifstream file(SwooleG.dns_hosts_path.empty() ? SW_PATH_HOSTS : SwooleG.dns_hosts_path);
     if (!file.is_open()) {
         return "";
     }
@@ -159,10 +162,6 @@ std::string get_ip_by_hosts(const std::string &search_domain) {
     return "";
 }
 
-void swoole_set_hosts_path(char *hosts_file) {
-    swoole_hosts = hosts_file;
-}
-
 static std::string parse_ip_address(void *vaddr, int type) {
     auto addr = reinterpret_cast<unsigned char *>(vaddr);
     std::string ip_addr;

diff --git a/src/server/master.cc b/src/server/master.cc
@@ -668,6 +668,10 @@ int Server::create() {
         return SW_ERR;
     }
 
+    if (swoole_isset_hook(SW_GLOBAL_HOOK_BEFORE_SERVER_CREATE)) {
+        swoole_call_hook(SW_GLOBAL_HOOK_BEFORE_SERVER_CREATE, this);
+    }
+
     session_list = (Session *) sw_shm_calloc(SW_SESSION_LIST_SIZE, sizeof(Session));
     if (session_list == nullptr) {
         swoole_error("sw_shm_calloc(%ld) for session_list failed", SW_SESSION_LIST_SIZE * sizeof(Session));
@@ -734,13 +738,20 @@ int Server::create() {
         return SW_ERR;
     }
 
+    int retval;
     if (is_base_mode()) {
         factory = new BaseFactory(this);
-        return create_reactor_processes();
+        retval = create_reactor_processes();
     } else {
         factory = new ProcessFactory(this);
-        return create_reactor_threads();
+        retval = create_reactor_threads();
     }
+
+    if (swoole_isset_hook(SW_GLOBAL_HOOK_AFTER_SERVER_CREATE)) {
+        swoole_call_hook(SW_GLOBAL_HOOK_AFTER_SERVER_CREATE, this);
+    }
+
+    return retval;
 }
 
 void Server::clear_timer() {

diff --git a/tests/swoole_server/max_concurrency.phpt b/tests/swoole_server/max_concurrency.phpt
@@ -0,0 +1,74 @@
+--TEST--
+swoole_server: max_concurrency
+--SKIPIF--
+<?php require __DIR__ . '/../include/skipif.inc'; ?>
+--FILE--
+<?php
+require __DIR__ . '/../include/bootstrap.php';
+
+use Swoole\Coroutine\Client;
+use Swoole\Timer;
+use Swoole\Event;
+use Swoole\Server;
+
+$pm = new SwooleTest\ProcessManager;
+
+$pm->parentFunc = function ($pid) use ($pm) {
+    for ($i=0; $i < 5; $i++) { 
+        go(function () use ($pm, $i) {
+            $client = new Client(SWOOLE_SOCK_TCP);
+            $client->set([
+                "open_eof_check" => true,
+                "open_eof_split" => true,
+                "package_eof" => "\r\n\r\n",
+            ]);
+            $r = $client->connect('127.0.0.1', $pm->getFreePort(), -1);
+            $data = "$i\r\n\r\n";
+            $client->send($data);
+            $ret = $client->recv();
+            var_dump(trim($ret));
+            $client->close();
+        });
+    }
+
+    Event::wait();
+
+    swoole_process::kill($pid);
+};
+
+$pm->childFunc = function () use ($pm)
+{
+    Co::set(['max_concurrency' => 1]);
+    $serv = new Server('127.0.0.1', $pm->getFreePort(), SWOOLE_PROCESS);
+    $serv->set([
+        'worker_num' => 1,
+        'dispatch_mode' => 1,
+        'open_eof_split' => true,
+        'package_eof' => "\r\n\r\n",
+        'log_file' => '/dev/null',
+    ]);
+    $serv->on("WorkerStart", function (Server $serv)  use ($pm)
+    {
+        $pm->wakeup();
+    });
+    $serv->on("Receive", function (Server $serv, $fd, $reactorId, $data)
+    {
+        global $count;
+        $count = 0;
+        co::sleep(0.05);
+        $count += 1;
+        $serv->send($fd, "$count\r\n\r\n");
+    });
+    $serv->start();
+};
+
+$pm->childFirst();
+$pm->run();
+
+?>
+--EXPECT--
+string(1) "1"
+string(1) "1"
+string(1) "1"
+string(1) "1"
+string(1) "1"