(ws) Add a new ws sub-command, push_server. This command runs a server which sends many messages in a loop to a websocket client. We can receive above 200,000 messages per second (cf #235).

bsergean · bsergean · commit 6e47c62c066d · 2020-08-02T12:41:34.000-07:00
diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
@@ -1,9 +1,14 @@
 # Changelog
+
 All changes to this project will be documented in this file.
 
+## [10.1.5] - 2020-08-02
+
+(ws) Add a new ws sub-command, push_server. This command runs a server which sends many messages in a loop to a websocket client. We can receive above 200,000 messages per second (cf #235).
+
 ## [10.1.4] - 2020-08-02
 
-(ws) Add a new ws sub-command, echo_client. This command send a message to an echo server, and send back to a server whatever message it does receive. When connecting to a local ws echo_server, on my MacBook Pro 2015 I can send/receive around 30,000 messages per second. (cf #235)
+(ws) Add a new ws sub-command, echo_client. This command sends a message to an echo server, and send back to a server whatever message it does receive. When connecting to a local ws echo_server, on my MacBook Pro 2015 I can send/receive around 30,000 messages per second. (cf #235)
 
 ## [10.1.3] - 2020-08-02
 
diff --git a/docs/performance.md b/docs/performance.md
@@ -0,0 +1,37 @@
+
+## WebSocket Client performance
+
+We will run a client and a server on the same machine, connecting to localhost. This bench is run on a MacBook Pro from 2015. We can receive over 200,000 (small) messages per second, another way to put it is that it takes 5 micro-second to receive and process one message. This is an indication about the minimal latency to receive messages.
+
+### Receiving messages
+
+By using the push_server ws sub-command, the server will send the same message in a loop to any connected client.
+
+```
+ws push_server -q --send_msg 'yo'
+```
+
+By using the echo_client ws sub-command, with the -m (mute or no_send), we will display statistics on how many messages we can receive per second.
+
+```
+$ ws echo_client -m ws://localhost:8008
+[2020-08-02 12:31:17.284] [info] ws_echo_client: connected
+[2020-08-02 12:31:17.284] [info] Uri: /
+[2020-08-02 12:31:17.284] [info] Headers:
+[2020-08-02 12:31:17.284] [info] Connection: Upgrade
+[2020-08-02 12:31:17.284] [info] Sec-WebSocket-Accept: byy/pMK2d0PtRwExaaiOnXJTQHo=
+[2020-08-02 12:31:17.284] [info] Server: ixwebsocket/10.1.4 macos ssl/SecureTransport zlib 1.2.11
+[2020-08-02 12:31:17.284] [info] Upgrade: websocket
+[2020-08-02 12:31:17.663] [info] messages received: 0 per second 2595307 total
+[2020-08-02 12:31:18.668] [info] messages received: 79679 per second 2674986 total
+[2020-08-02 12:31:19.668] [info] messages received: 207438 per second 2882424 total
+[2020-08-02 12:31:20.673] [info] messages received: 209207 per second 3091631 total
+[2020-08-02 12:31:21.676] [info] messages received: 216056 per second 3307687 total
+[2020-08-02 12:31:22.680] [info] messages received: 214927 per second 3522614 total
+[2020-08-02 12:31:23.684] [info] messages received: 216960 per second 3739574 total
+[2020-08-02 12:31:24.688] [info] messages received: 215232 per second 3954806 total
+[2020-08-02 12:31:25.691] [info] messages received: 212300 per second 4167106 total
+[2020-08-02 12:31:26.694] [info] messages received: 212501 per second 4379607 total
+[2020-08-02 12:31:27.699] [info] messages received: 212330 per second 4591937 total
+[2020-08-02 12:31:28.702] [info] messages received: 216511 per second 4808448 total
+```
diff --git a/ixwebsocket/IXWebSocketVersion.h b/ixwebsocket/IXWebSocketVersion.h
@@ -6,4 +6,4 @@
 
 #pragma once
 
-#define IX_WEBSOCKET_VERSION "10.1.4"
+#define IX_WEBSOCKET_VERSION "10.1.5"
diff --git a/ws/CMakeLists.txt b/ws/CMakeLists.txt
@@ -50,6 +50,7 @@ add_executable(ws
   ws_http_client.cpp
   ws_ping_pong.cpp
   ws_broadcast_server.cpp
+  ws_push_server.cpp
   ws_echo_server.cpp
   ws_echo_client.cpp
   ws_chat.cpp
diff --git a/ws/ws.cpp b/ws/ws.cpp
@@ -148,6 +148,7 @@ int main(int argc, char** argv)
     bool version = false;
     bool verifyNone = false;
     bool disablePong = false;
+    bool noSend = false;
     int port = 8008;
     int redisPort = 6379;
     int statsdPort = 8125;
@@ -254,6 +255,7 @@ int main(int argc, char** argv)
         "--ping_interval", pingIntervalSecs, "Interval between sending pings");
     echoClientApp->add_option("--subprotocol", subprotocol, "Subprotocol");
     echoClientApp->add_option("--send_msg", sendMsg, "Send message");
+    echoClientApp->add_flag("-m", noSend, "Do not send messages, only receive messages");
     addTLSOptions(echoClientApp);
 
     CLI::App* chatApp = app.add_subcommand("chat", "Group chat");
@@ -272,6 +274,18 @@ int main(int argc, char** argv)
     echoServerApp->add_flag("-p", disablePong, "Disable sending PONG in response to PING");
     addTLSOptions(echoServerApp);
 
+    CLI::App* pushServerApp = app.add_subcommand("push_server", "Push server");
+    pushServerApp->fallthrough();
+    pushServerApp->add_option("--port", port, "Port");
+    pushServerApp->add_option("--host", hostname, "Hostname");
+    pushServerApp->add_flag("-q", quiet, "Quiet / only display warnings and errors");
+    pushServerApp->add_flag("-g", greetings, "Greet");
+    pushServerApp->add_flag("-6", ipv6, "IpV6");
+    pushServerApp->add_flag("-x", disablePerMessageDeflate, "Disable per message deflate");
+    pushServerApp->add_flag("-p", disablePong, "Disable sending PONG in response to PING");
+    pushServerApp->add_option("--send_msg", sendMsg, "Send message");
+    addTLSOptions(pushServerApp);
+
     CLI::App* broadcastServerApp = app.add_subcommand("broadcast_server", "Broadcasting server");
     broadcastServerApp->fallthrough();
     broadcastServerApp->add_option("--port", port, "Port");
@@ -525,13 +539,25 @@ int main(int argc, char** argv)
                                  tlsOptions,
                                  subprotocol,
                                  pingIntervalSecs,
-                                 sendMsg);
+                                 sendMsg,
+                                 noSend);
     }
     else if (app.got_subcommand("echo_server"))
     {
         ret = ix::ws_echo_server_main(
             port, greetings, hostname, tlsOptions, ipv6, disablePerMessageDeflate, disablePong);
     }
+    else if (app.got_subcommand("push_server"))
+    {
+        ret = ix::ws_push_server(port,
+                                 greetings,
+                                 hostname,
+                                 tlsOptions,
+                                 ipv6,
+                                 disablePerMessageDeflate,
+                                 disablePong,
+                                 sendMsg);
+    }
     else if (app.got_subcommand("transfer"))
     {
         ret = ix::ws_transfer_main(port, hostname, tlsOptions);
diff --git a/ws/ws.h b/ws/ws.h
@@ -35,6 +35,15 @@ namespace ix
                             bool disablePerMessageDeflate,
                             bool disablePong);
 
+    int ws_push_server(int port,
+                       bool greetings,
+                       const std::string& hostname,
+                       const ix::SocketTLSOptions& tlsOptions,
+                       bool ipv6,
+                       bool disablePerMessageDeflate,
+                       bool disablePong,
+                       const std::string& sendMsg);
+
     int ws_broadcast_server_main(int port,
                                  const std::string& hostname,
                                  const ix::SocketTLSOptions& tlsOptions);
@@ -60,7 +69,8 @@ namespace ix
                        const ix::SocketTLSOptions& tlsOptions,
                        const std::string& subprotocol,
                        int pingIntervalSecs,
-                       const std::string& sendMsg);
+                       const std::string& sendMsg,
+                       bool noSend);
 
     int ws_receive_main(const std::string& url,
                         bool enablePerMessageDeflate,
diff --git a/ws/ws_echo_client.cpp b/ws/ws_echo_client.cpp
@@ -21,7 +21,8 @@ namespace ix
                        const ix::SocketTLSOptions& tlsOptions,
                        const std::string& subprotocol,
                        int pingIntervalSecs,
-                       const std::string& sendMsg)
+                       const std::string& sendMsg,
+                       bool noSend)
     {
         // Our websocket object
         ix::WebSocket webSocket;
@@ -46,26 +47,33 @@ namespace ix
 
         // Setup a callback to be fired (in a background thread, watch out for race conditions !)
         // when a message or an event (open, close, error) is received
-        webSocket.setOnMessageCallback(
-            [&webSocket, &receivedCount, &sendMsg, binaryMode](const ix::WebSocketMessagePtr& msg) {
-                if (msg->type == ix::WebSocketMessageType::Message)
+        webSocket.setOnMessageCallback([&webSocket, &receivedCount, &sendMsg, noSend, binaryMode](
+                                           const ix::WebSocketMessagePtr& msg) {
+            if (msg->type == ix::WebSocketMessageType::Message)
+            {
+                if (!noSend)
                 {
                     webSocket.send(msg->str, msg->binary);
-                    receivedCount++;
                 }
-                else if (msg->type == ix::WebSocketMessageType::Open)
+                receivedCount++;
+            }
+            else if (msg->type == ix::WebSocketMessageType::Open)
+            {
+                spdlog::info("ws_echo_client: connected");
+                spdlog::info("Uri: {}", msg->openInfo.uri);
+                spdlog::info("Headers:");
+                for (auto it : msg->openInfo.headers)
                 {
-                    spdlog::info("ws_echo_client: connected");
-                    spdlog::info("Uri: {}", msg->openInfo.uri);
-                    spdlog::info("Headers:");
-                    for (auto it : msg->openInfo.headers)
-                    {
-                        spdlog::info("{}: {}", it.first, it.second);
-                    }
-
-                    webSocket.send(sendMsg, binaryMode);
+                    spdlog::info("{}: {}", it.first, it.second);
                 }
-            });
+
+                webSocket.send(sendMsg, binaryMode);
+            }
+            else if (msg->type == ix::WebSocketMessageType::Pong)
+            {
+                spdlog::info("Received pong {}", msg->str);
+            }
+        });
 
         auto timer = [&receivedCount, &receivedCountTotal, &receivedCountPerSecs] {
             setThreadName("Timer");
diff --git a/ws/ws_push_server.cpp b/ws/ws_push_server.cpp
@@ -0,0 +1,108 @@
+/*
+ *  ws_push_server.cpp
+ *  Author: Benjamin Sergeant
+ *  Copyright (c) 2020 Machine Zone, Inc. All rights reserved.
+ */
+
+#include <ixwebsocket/IXNetSystem.h>
+#include <ixwebsocket/IXWebSocketServer.h>
+#include <spdlog/spdlog.h>
+#include <sstream>
+
+namespace ix
+{
+    int ws_push_server(int port,
+                       bool greetings,
+                       const std::string& hostname,
+                       const ix::SocketTLSOptions& tlsOptions,
+                       bool ipv6,
+                       bool disablePerMessageDeflate,
+                       bool disablePong,
+                       const std::string& sendMsg)
+    {
+        spdlog::info("Listening on {}:{}", hostname, port);
+
+        ix::WebSocketServer server(port,
+                                   hostname,
+                                   SocketServer::kDefaultTcpBacklog,
+                                   SocketServer::kDefaultMaxConnections,
+                                   WebSocketServer::kDefaultHandShakeTimeoutSecs,
+                                   (ipv6) ? AF_INET6 : AF_INET);
+
+        server.setTLSOptions(tlsOptions);
+
+        if (disablePerMessageDeflate)
+        {
+            spdlog::info("Disable per message deflate");
+            server.disablePerMessageDeflate();
+        }
+
+        if (disablePong)
+        {
+            spdlog::info("Disable responding to PING messages with PONG");
+            server.disablePong();
+        }
+
+        server.setOnClientMessageCallback(
+            [greetings, &sendMsg](std::shared_ptr<ConnectionState> connectionState,
+                                  ConnectionInfo& connectionInfo,
+                                  WebSocket& webSocket,
+                                  const WebSocketMessagePtr& msg) {
+                auto remoteIp = connectionInfo.remoteIp;
+                if (msg->type == ix::WebSocketMessageType::Open)
+                {
+                    spdlog::info("New connection");
+                    spdlog::info("remote ip: {}", remoteIp);
+                    spdlog::info("id: {}", connectionState->getId());
+                    spdlog::info("Uri: {}", msg->openInfo.uri);
+                    spdlog::info("Headers:");
+                    for (auto it : msg->openInfo.headers)
+                    {
+                        spdlog::info("{}: {}", it.first, it.second);
+                    }
+
+                    if (greetings)
+                    {
+                        webSocket.sendText("Welcome !");
+                    }
+
+                    bool binary = false;
+                    while (true)
+                    {
+                        webSocket.send(sendMsg, binary);
+                    }
+                }
+                else if (msg->type == ix::WebSocketMessageType::Close)
+                {
+                    spdlog::info("Closed connection: client id {} code {} reason {}",
+                                 connectionState->getId(),
+                                 msg->closeInfo.code,
+                                 msg->closeInfo.reason);
+                }
+                else if (msg->type == ix::WebSocketMessageType::Error)
+                {
+                    spdlog::error("Connection error: {}", msg->errorInfo.reason);
+                    spdlog::error("#retries: {}", msg->errorInfo.retries);
+                    spdlog::error("Wait time(ms): {}", msg->errorInfo.wait_time);
+                    spdlog::error("HTTP Status: {}", msg->errorInfo.http_status);
+                }
+                else if (msg->type == ix::WebSocketMessageType::Message)
+                {
+                    spdlog::info("Received {} bytes", msg->wireSize);
+                    webSocket.send(msg->str, msg->binary);
+                }
+            });
+
+        auto res = server.listen();
+        if (!res.first)
+        {
+            spdlog::error(res.second);
+            return 1;
+        }
+
+        server.start();
+        server.wait();
+
+        return 0;
+    }
+} // namespace ix

Original file line number	Diff line number	Diff line change
`@@ -6,4 +6,4 @@`
`6`	`6`
`7`	`7`	`#pragma once`
`8`	`8`
`9`		`-#define IX_WEBSOCKET_VERSION "10.1.4"`
	`9`	`+#define IX_WEBSOCKET_VERSION "10.1.5"`