limit worker_connections

http/server/HttpHandler.cpp

@@ -127,9 +127,9 @@ bool HttpHandler::SwitchWebSocket(hio_t* io) {
             break;
         }
     };
-    // NOTE: cancel keepalive timer, judge alive by heartbeat.
-    ws_channel->setKeepaliveTimeout(0);
     if (ws_service && ws_service->ping_interval > 0) {
+        // NOTE: cancel keepalive timer, judge alive by heartbeat.
+        ws_channel->setKeepaliveTimeout(0);
         int ping_interval = MAX(ws_service->ping_interval, 1000);
         ws_channel->setHeartbeat(ping_interval, [this](){
             if (last_recv_pong_time < last_send_ping_time) {

http/server/HttpServer.cpp

@@ -203,6 +203,11 @@ static void on_close(hio_t* io) {
         }
     }
 
+    EventLoop* loop = currentThreadEventLoop;
+    if (loop) {
+        --loop->connectionNum;
+    }
+
     hevent_set_userdata(io, NULL);
     delete handler;
 }
@@ -219,6 +224,14 @@ static void on_accept(hio_t* io) {
             SOCKADDR_STR(hio_peeraddr(io), peeraddrstr));
     */
 
+    EventLoop* loop = currentThreadEventLoop;
+    if (loop->connectionNum >= server->worker_connections) {
+        hlogw("over worker_connections");
+        hio_close(io);
+        return;
+    }
+    ++loop->connectionNum;
+
     hio_setcb_close(io, on_close);
     hio_setcb_read(io, on_recv);
     hio_read(io);

http/server/HttpServer.h

@@ -17,6 +17,7 @@ typedef struct http_server_s {
     int http_version;
     int worker_processes;
     int worker_threads;
+    uint32_t worker_connections; // max_connections = workers * worker_connections
     HttpService* service; // http service
     WebSocketService* ws; // websocket service
     void* userdata;
@@ -35,6 +36,7 @@ typedef struct http_server_s {
         http_version = 1;
         worker_processes = 0;
         worker_threads = 0;
+        worker_connections = 1024;
         service = NULL;
         ws = NULL;
         listenfd[0] = listenfd[1] = -1;