!530 修复websocket限流可能引起卡死

Merge pull request !530 from JingWoo/master

src/websocket/service/ws_server.cc

@@ -288,18 +288,8 @@ int WebsocketServer::Wswrite(struct lws *wsi, void *in, size_t len)
     return 0;
 }
 
-void WebsocketServer::Receive(struct lws *wsi, void *user, void *in, size_t len)
+void WebsocketServer::Receive(struct lws *wsi, void *in, size_t len)
 {
-    if (user != nullptr) {
-        struct per_session_data__echo *pss = (struct per_session_data__echo *)user;
-        pss->final = lws_is_final_fragment(wsi);
-        pss->binary = lws_frame_is_binary(wsi);
-
-        (void)memcpy(&pss->buf[LWS_PRE], in, len);
-        pss->len = (unsigned int)len;
-        pss->rx += len;
-        lws_rx_flow_control(wsi, 0);
-    }
     if (m_wsis.find(wsi) == m_wsis.end()) {
         ERROR("invailed websocket session!");
         return;
@@ -353,13 +343,11 @@ int WebsocketServer::Callback(struct lws *wsi, enum lws_callback_reasons reason,
                     return -1;
                 }
                 WebsocketServer::GetInstance()->SetLwsSendedFlag(wsi, true);
-                lws_rx_flow_control(wsi, 1);
             }
             break;
         case LWS_CALLBACK_RECEIVE: {
                 std::lock_guard<std::mutex> lock(m_mutex);
-                WebsocketServer::GetInstance()->Receive(wsi, nullptr, (char *)in, len);
-                lws_rx_flow_control(wsi, 0);
+                WebsocketServer::GetInstance()->Receive(wsi, (char *)in, len);
             }
             break;
         case LWS_CALLBACK_CLOSED: {

src/websocket/service/ws_server.h

@@ -27,7 +27,7 @@
 #include "url.h"
 #include "errors.h"
 
-#define MAX_ECHO_PAYLOAD 1024
+#define MAX_ECHO_PAYLOAD 4096
 #define MAX_ARRAY_LEN 2
 #define MAX_BUF_LEN 256
 #define MAX_PROTOCOL_NUM 2
@@ -37,16 +37,6 @@
 #define BUF_BASE_SIZE 1024
 #define LWS_TIMEOUT 50
 
-struct per_session_data__echo {
-    size_t rx, tx;
-    unsigned char buf[LWS_PRE + MAX_ECHO_PAYLOAD + 1];
-    unsigned int len;
-    unsigned int index;
-    int final;
-    int continuation;
-    int binary;
-};
-
 enum WebsocketChannel {
     STDINCHANNEL = 0,
     STDOUTCHANNEL,
@@ -95,7 +85,7 @@ private:
     std::vector<std::string> split(std::string str, char r);
     static void EmitLog(int level, const char *line);
     int CreateContext();
-    inline void Receive(struct lws *client, void *user, void *in, size_t len);
+    inline void Receive(struct lws *client, void *in, size_t len);
     int  Wswrite(struct lws *wsi, void *in, size_t len);
     inline int DumpHandshakeInfo(struct lws *wsi) noexcept;
     static int Callback(struct lws *wsi, enum lws_callback_reasons reason,
@@ -110,7 +100,7 @@ private:
     volatile int m_force_exit = 0;
     std::thread m_pthread_service;
     const struct lws_protocols m_protocols[MAX_PROTOCOL_NUM] = {
-        {  "channel.k8s.io", Callback, sizeof(struct per_session_data__echo), MAX_ECHO_PAYLOAD, },
+        {  "channel.k8s.io", Callback, 0, MAX_ECHO_PAYLOAD, },
         { NULL, NULL, 0, 0 }
     };
     RouteCallbackRegister m_handler;