for cpp websocket core dump (#828)

funasr/runtime/websocket/funasr-wss-client-2pass.cpp

@@ -8,22 +8,24 @@
 // ./funasr-wss-client  --server-ip <string>
 //                     --port <string>
 //                     --wav-path <string>
-//                     [--thread-num <int>] 
+//                     [--thread-num <int>]
 //                     [--is-ssl <int>]  [--]
 //                     [--version] [-h]
 // example:
-// ./funasr-wss-client --server-ip 127.0.0.1 --port 10095 --wav-path test.wav --thread-num 1 --is-ssl 1
+// ./funasr-wss-client --server-ip 127.0.0.1 --port 10095 --wav-path test.wav
+// --thread-num 1 --is-ssl 1
 
 #define ASIO_STANDALONE 1
-#include <websocketpp/client.hpp>
-#include <websocketpp/common/thread.hpp>
-#include <websocketpp/config/asio_client.hpp>
-#include <iostream>
+#include <glog/logging.h>
+
+#include <atomic>
 #include <fstream>
+#include <iostream>
 #include <sstream>
-#include <atomic>
 #include <thread>
-#include <glog/logging.h>
+#include <websocketpp/client.hpp>
+#include <websocketpp/common/thread.hpp>
+#include <websocketpp/config/asio_client.hpp>
 
 #include "audio.h"
 #include "nlohmann/json.hpp"
@@ -33,398 +35,416 @@
  * Define a semi-cross platform helper method that waits/sleeps for a bit.
  */
 void WaitABit() {
-    #ifdef WIN32
-        Sleep(500);
-    #else
-        usleep(500);
-    #endif
+#ifdef WIN32
+  Sleep(500);
+#else
+  usleep(500);
+#endif
 }
 std::atomic<int> wav_index(0);
 
 bool IsTargetFile(const std::string& filename, const std::string target) {
-    std::size_t pos = filename.find_last_of(".");
-    if (pos == std::string::npos) {
-        return false;
-    }
-    std::string extension = filename.substr(pos + 1);
-    return (extension == target);
+  std::size_t pos = filename.find_last_of(".");
+  if (pos == std::string::npos) {
+    return false;
+  }
+  std::string extension = filename.substr(pos + 1);
+  return (extension == target);
 }
 
 typedef websocketpp::config::asio_client::message_type::ptr message_ptr;
-typedef websocketpp::lib::shared_ptr<websocketpp::lib::asio::ssl::context> context_ptr;
+typedef websocketpp::lib::shared_ptr<websocketpp::lib::asio::ssl::context>
+    context_ptr;
 using websocketpp::lib::bind;
 using websocketpp::lib::placeholders::_1;
 using websocketpp::lib::placeholders::_2;
 context_ptr OnTlsInit(websocketpp::connection_hdl) {
-    context_ptr ctx = websocketpp::lib::make_shared<asio::ssl::context>(
-        asio::ssl::context::sslv23);
-
-    try {
-        ctx->set_options(
-            asio::ssl::context::default_workarounds | asio::ssl::context::no_sslv2 |
-            asio::ssl::context::no_sslv3 | asio::ssl::context::single_dh_use);
-
-    } catch (std::exception& e) {
-        LOG(ERROR) << e.what();
-    }
-    return ctx;
+  context_ptr ctx = websocketpp::lib::make_shared<asio::ssl::context>(
+      asio::ssl::context::sslv23);
+
+  try {
+    ctx->set_options(
+        asio::ssl::context::default_workarounds | asio::ssl::context::no_sslv2 |
+        asio::ssl::context::no_sslv3 | asio::ssl::context::single_dh_use);
+
+  } catch (std::exception& e) {
+    LOG(ERROR) << e.what();
+  }
+  return ctx;
 }
 
 // template for tls or not config
 template <typename T>
 class WebsocketClient {
-  public:
-    // typedef websocketpp::client<T> client;
-    // typedef websocketpp::client<websocketpp::config::asio_tls_client>
-    // wss_client;
-    typedef websocketpp::lib::lock_guard<websocketpp::lib::mutex> scoped_lock;
-
-    WebsocketClient(int is_ssl) : m_open(false), m_done(false) {
-        // set up access channels to only log interesting things
-        m_client.clear_access_channels(websocketpp::log::alevel::all);
-        m_client.set_access_channels(websocketpp::log::alevel::connect);
-        m_client.set_access_channels(websocketpp::log::alevel::disconnect);
-        m_client.set_access_channels(websocketpp::log::alevel::app);
-
-        // Initialize the Asio transport policy
-        m_client.init_asio();
-
-        // Bind the handlers we are using
-        using websocketpp::lib::bind;
-        using websocketpp::lib::placeholders::_1;
-        m_client.set_open_handler(bind(&WebsocketClient::on_open, this, _1));
-        m_client.set_close_handler(bind(&WebsocketClient::on_close, this, _1));
-
-        m_client.set_message_handler(
-            [this](websocketpp::connection_hdl hdl, message_ptr msg) {
-              on_message(hdl, msg);
-            });
-
-        m_client.set_fail_handler(bind(&WebsocketClient::on_fail, this, _1));
-        m_client.clear_access_channels(websocketpp::log::alevel::all);
-    }
+ public:
+  // typedef websocketpp::client<T> client;
+  // typedef websocketpp::client<websocketpp::config::asio_tls_client>
+  // wss_client;
+  typedef websocketpp::lib::lock_guard<websocketpp::lib::mutex> scoped_lock;
+
+  WebsocketClient(int is_ssl) : m_open(false), m_done(false) {
+    // set up access channels to only log interesting things
+    m_client.clear_access_channels(websocketpp::log::alevel::all);
+    m_client.set_access_channels(websocketpp::log::alevel::connect);
+    m_client.set_access_channels(websocketpp::log::alevel::disconnect);
+    m_client.set_access_channels(websocketpp::log::alevel::app);
+
+    // Initialize the Asio transport policy
+    m_client.init_asio();
+
+    // Bind the handlers we are using
+    using websocketpp::lib::bind;
+    using websocketpp::lib::placeholders::_1;
+    m_client.set_open_handler(bind(&WebsocketClient::on_open, this, _1));
+    m_client.set_close_handler(bind(&WebsocketClient::on_close, this, _1));
+
+    m_client.set_message_handler(
+        [this](websocketpp::connection_hdl hdl, message_ptr msg) {
+          on_message(hdl, msg);
+        });
 
-    void on_message(websocketpp::connection_hdl hdl, message_ptr msg) {
-        const std::string& payload = msg->get_payload();
-        switch (msg->get_opcode()) {
-            case websocketpp::frame::opcode::text:
-                nlohmann::json jsonresult = nlohmann::json::parse(payload);
-                LOG(INFO)<< "Thread: " << this_thread::get_id() <<",on_message = " << payload;
-				
-                // if (jsonresult["is_final"] == true){
-				// 	websocketpp::lib::error_code ec;
-				// 	m_client.close(m_hdl, websocketpp::close::status::going_away, "", ec);
-				// 	if (ec){
-                //         LOG(ERROR)<< "Error closing connection " << ec.message();
-				// 	}
-                // }
+    m_client.set_fail_handler(bind(&WebsocketClient::on_fail, this, _1));
+    m_client.clear_access_channels(websocketpp::log::alevel::all);
+  }
+
+  void on_message(websocketpp::connection_hdl hdl, message_ptr msg) {
+    const std::string& payload = msg->get_payload();
+    switch (msg->get_opcode()) {
+      case websocketpp::frame::opcode::text:
+        nlohmann::json jsonresult = nlohmann::json::parse(payload);
+        LOG(INFO) << "Thread: " << this_thread::get_id()
+                  << ",on_message = " << payload << "jsonresult" << jsonresult;
+
+        if (jsonresult["is_final"] == true) {
+          websocketpp::lib::error_code ec;
+           
+          m_client.close(hdl, websocketpp::close::status::going_away, "", ec);
+
+          if (ec) {
+            LOG(ERROR) << "Error closing connection " << ec.message();
+          }
+       
         }
     }
-
-    // This method will block until the connection is complete  
-    void run(const std::string& uri, const std::vector<string>& wav_list, const std::vector<string>& wav_ids, std::string asr_mode, std::vector<int> chunk_size) {
-        // Create a new connection to the given URI
-        websocketpp::lib::error_code ec;
-        typename websocketpp::client<T>::connection_ptr con =
-            m_client.get_connection(uri, ec);
-        if (ec) {
-            m_client.get_alog().write(websocketpp::log::alevel::app,
-                                    "Get Connection Error: " + ec.message());
-            return;
-        }
-        // Grab a handle for this connection so we can talk to it in a thread
-        // safe manor after the event loop starts.
-        m_hdl = con->get_handle();
-
-        // Queue the connection. No DNS queries or network connections will be
-        // made until the io_service event loop is run.
-        m_client.connect(con);
-
-        // Create a thread to run the ASIO io_service event loop
-        websocketpp::lib::thread asio_thread(&websocketpp::client<T>::run,
-                                            &m_client);
-        while(true){
-            int i = wav_index.fetch_add(1);
-            if (i >= wav_list.size()) {
-                break;
-            }
-            send_wav_data(wav_list[i], wav_ids[i], asr_mode, chunk_size);
-        }
-        WaitABit(); 
-
-        asio_thread.join();
-
+  }
+
+  // This method will block until the connection is complete
+  void run(const std::string& uri, const std::vector<string>& wav_list,
+           const std::vector<string>& wav_ids, std::string asr_mode,
+           std::vector<int> chunk_size) {
+    // Create a new connection to the given URI
+    websocketpp::lib::error_code ec;
+    typename websocketpp::client<T>::connection_ptr con =
+        m_client.get_connection(uri, ec);
+    if (ec) {
+      m_client.get_alog().write(websocketpp::log::alevel::app,
+                                "Get Connection Error: " + ec.message());
+      return;
     }
-
-    // The open handler will signal that we are ready to start sending data
-    void on_open(websocketpp::connection_hdl) {
-        m_client.get_alog().write(websocketpp::log::alevel::app,
-                                "Connection opened, starting data!");
-
-        scoped_lock guard(m_lock);
-        m_open = true;
-    }
-
-    // The close handler will signal that we should stop sending data
-    void on_close(websocketpp::connection_hdl) {
-        m_client.get_alog().write(websocketpp::log::alevel::app,
-                                  "Connection closed, stopping data!");
-
-        scoped_lock guard(m_lock);
-        m_done = true;
+    // Grab a handle for this connection so we can talk to it in a thread
+    // safe manor after the event loop starts.
+    m_hdl = con->get_handle();
+
+    // Queue the connection. No DNS queries or network connections will be
+    // made until the io_service event loop is run.
+    m_client.connect(con);
+
+    // Create a thread to run the ASIO io_service event loop
+    websocketpp::lib::thread asio_thread(&websocketpp::client<T>::run,
+                                         &m_client);
+
+    send_wav_data(wav_list[0], wav_ids[0], asr_mode, chunk_size);
+
+    WaitABit();
+
+    asio_thread.join();
+  }
+
+  // The open handler will signal that we are ready to start sending data
+  void on_open(websocketpp::connection_hdl) {
+    m_client.get_alog().write(websocketpp::log::alevel::app,
+                              "Connection opened, starting data!");
+
+    scoped_lock guard(m_lock);
+    m_open = true;
+  }
+
+  // The close handler will signal that we should stop sending data
+  void on_close(websocketpp::connection_hdl) {
+    m_client.get_alog().write(websocketpp::log::alevel::app,
+                              "Connection closed, stopping data!");
+
+    scoped_lock guard(m_lock);
+    m_done = true;
+  }
+
+  // The fail handler will signal that we should stop sending data
+  void on_fail(websocketpp::connection_hdl) {
+    m_client.get_alog().write(websocketpp::log::alevel::app,
+                              "Connection failed, stopping data!");
+
+    scoped_lock guard(m_lock);
+    m_done = true;
+  }
+  // send wav to server
+  void send_wav_data(string wav_path, string wav_id, std::string asr_mode,
+                     std::vector<int> chunk_vector) {
+    uint64_t count = 0;
+    std::stringstream val;
+
+    funasr::Audio audio(1);
+    int32_t sampling_rate = 16000;
+    std::string wav_format = "pcm";
+    if (IsTargetFile(wav_path.c_str(), "wav")) {
+      int32_t sampling_rate = -1;
+      if (!audio.LoadWav(wav_path.c_str(), &sampling_rate)) return;
+    } else if (IsTargetFile(wav_path.c_str(), "pcm")) {
+      if (!audio.LoadPcmwav(wav_path.c_str(), &sampling_rate)) return;
+    } else {
+      wav_format = "others";
+      if (!audio.LoadOthers2Char(wav_path.c_str())) return;
     }
 
-    // The fail handler will signal that we should stop sending data
-    void on_fail(websocketpp::connection_hdl) {
-        m_client.get_alog().write(websocketpp::log::alevel::app,
-                                  "Connection failed, stopping data!");
-
+    float* buff;
+    int len;
+    int flag = 0;
+    bool wait = false;
+    while (1) {
+      {
         scoped_lock guard(m_lock);
-        m_done = true;
-    }
-    // send wav to server
-    void send_wav_data(string wav_path, string wav_id, std::string asr_mode, std::vector<int> chunk_vector) {
-        uint64_t count = 0;
-        std::stringstream val;
-
-		funasr::Audio audio(1);
-        int32_t sampling_rate = 16000;
-        std::string wav_format = "pcm";
-		if(IsTargetFile(wav_path.c_str(), "wav")){
-			int32_t sampling_rate = -1;
-			if(!audio.LoadWav(wav_path.c_str(), &sampling_rate))
-				return ;
-		}else if(IsTargetFile(wav_path.c_str(), "pcm")){
-			if (!audio.LoadPcmwav(wav_path.c_str(), &sampling_rate))
-				return ;
-		}else{
-			wav_format = "others";
-            if (!audio.LoadOthers2Char(wav_path.c_str()))
-				return ;
-		}
-
-        float* buff;
-        int len;
-        int flag = 0;
-        bool wait = false;
-        while (1) {
-            {
-                scoped_lock guard(m_lock);
-                // If the connection has been closed, stop generating data
-                if (m_done) {
-                  break;
-                }
-                // If the connection hasn't been opened yet wait a bit and retry
-                if (!m_open) {
-                  wait = true;
-                } else {
-                  break;
-                }
-            }
-            if (wait) {
-                // LOG(INFO) << "wait.." << m_open;
-                WaitABit();
-                continue;
-            }
+        // If the connection has been closed, stop generating data
+        if (m_done) {
+          break;
         }
-        websocketpp::lib::error_code ec;
-
-        nlohmann::json jsonbegin;
-        nlohmann::json chunk_size = nlohmann::json::array();
-        chunk_size.push_back(chunk_vector[0]);
-        chunk_size.push_back(chunk_vector[1]);
-        chunk_size.push_back(chunk_vector[2]);
-        jsonbegin["mode"] = asr_mode;
-        jsonbegin["chunk_size"] = chunk_size;
-        jsonbegin["wav_name"] = wav_id;
-        jsonbegin["wav_format"] = wav_format;
-        jsonbegin["is_speaking"] = true;
-        m_client.send(m_hdl, jsonbegin.dump(), websocketpp::frame::opcode::text,
-                      ec);
-
-        // fetch wav data use asr engine api
-        if(wav_format == "pcm"){
-            while (audio.Fetch(buff, len, flag) > 0) {
-                short* iArray = new short[len];
-                for (size_t i = 0; i < len; ++i) {
-                iArray[i] = (short)(buff[i]*32768);
-                }
-
-                // send data to server
-                int offset = 0;
-                int block_size = 102400;
-                while(offset < len){
-                    int send_block = 0;
-                    if (offset + block_size <= len){
-                        send_block = block_size;
-                    }else{
-                        send_block = len - offset;
-                    }
-                    m_client.send(m_hdl, iArray+offset, send_block * sizeof(short),
-                        websocketpp::frame::opcode::binary, ec);
-                    offset += send_block;
-                }
-
-                LOG(INFO) << "sended data len=" << len * sizeof(short);
-                // The most likely error that we will get is that the connection is
-                // not in the right state. Usually this means we tried to send a
-                // message to a connection that was closed or in the process of
-                // closing. While many errors here can be easily recovered from,
-                // in this simple example, we'll stop the data loop.
-                if (ec) {
-                m_client.get_alog().write(websocketpp::log::alevel::app,
-                                            "Send Error: " + ec.message());
-                break;
-                }
-                delete[] iArray;
-                // WaitABit();
-            }
-        }else{
-            int offset = 0;
-            int block_size = 204800;
-            len = audio.GetSpeechLen();
-            char* others_buff = audio.GetSpeechChar();
-
-            while(offset < len){
-                int send_block = 0;
-                if (offset + block_size <= len){
-                    send_block = block_size;
-                }else{
-                    send_block = len - offset;
-                }
-                m_client.send(m_hdl, others_buff+offset, send_block,
-                    websocketpp::frame::opcode::binary, ec);
-                offset += send_block;
-            }
-
-            LOG(INFO) << "sended data len=" << len;
-            // The most likely error that we will get is that the connection is
-            // not in the right state. Usually this means we tried to send a
-            // message to a connection that was closed or in the process of
-            // closing. While many errors here can be easily recovered from,
-            // in this simple example, we'll stop the data loop.
-            if (ec) {
-                m_client.get_alog().write(websocketpp::log::alevel::app,
-                                        "Send Error: " + ec.message());
-            }
+        // If the connection hasn't been opened yet wait a bit and retry
+        if (!m_open) {
+          wait = true;
+        } else {
+          break;
         }
+      }
 
-        nlohmann::json jsonresult;
-        jsonresult["is_speaking"] = false;
-        m_client.send(m_hdl, jsonresult.dump(), websocketpp::frame::opcode::text,
-                      ec);
+      if (wait) {
+        // LOG(INFO) << "wait.." << m_open;
         WaitABit();
+        continue;
+      }
     }
-    websocketpp::client<T> m_client;
-
-  private:
-    websocketpp::connection_hdl m_hdl;
-    websocketpp::lib::mutex m_lock;
-    bool m_open;
-    bool m_done;
-	int total_num=0;
-};
+    websocketpp::lib::error_code ec;
+
+    nlohmann::json jsonbegin;
+    nlohmann::json chunk_size = nlohmann::json::array();
+    chunk_size.push_back(chunk_vector[0]);
+    chunk_size.push_back(chunk_vector[1]);
+    chunk_size.push_back(chunk_vector[2]);
+    jsonbegin["mode"] = asr_mode;
+    jsonbegin["chunk_size"] = chunk_size;
+    jsonbegin["wav_name"] = wav_id;
+    jsonbegin["wav_format"] = wav_format;
+    jsonbegin["is_speaking"] = true;
+    m_client.send(m_hdl, jsonbegin.dump(), websocketpp::frame::opcode::text,
+                  ec);
+
+    // fetch wav data use asr engine api
+    if (wav_format == "pcm") {
+      while (audio.Fetch(buff, len, flag) > 0) {
+        short* iArray = new short[len];
+        for (size_t i = 0; i < len; ++i) {
+          iArray[i] = (short)(buff[i] * 32768);
+        }
 
-int main(int argc, char* argv[]) {
+        // send data to server
+        int offset = 0;
+        int block_size = 102400;
+        while (offset < len) {
+          int send_block = 0;
+          if (offset + block_size <= len) {
+            send_block = block_size;
+          } else {
+            send_block = len - offset;
+          }
+          m_client.send(m_hdl, iArray + offset, send_block * sizeof(short),
+                        websocketpp::frame::opcode::binary, ec);
+          offset += send_block;
+        }
 
-    google::InitGoogleLogging(argv[0]);
-    FLAGS_logtostderr = true;
-
-    TCLAP::CmdLine cmd("funasr-wss-client", ' ', "1.0");
-    TCLAP::ValueArg<std::string> server_ip_("", "server-ip", "server-ip", true,
-                                           "127.0.0.1", "string");
-    TCLAP::ValueArg<std::string> port_("", "port", "port", true, "10095", "string");
-    TCLAP::ValueArg<std::string> wav_path_("", "wav-path", 
-        "the input could be: wav_path, e.g.: asr_example.wav; pcm_path, e.g.: asr_example.pcm; wav.scp, kaldi style wav list (wav_id \t wav_path)", 
-        true, "", "string");
-    TCLAP::ValueArg<std::string>    asr_mode_("", ASR_MODE, "offline, online, 2pass", false, "2pass", "string");
-    TCLAP::ValueArg<std::string>    chunk_size_("", "chunk-size", "chunk_size: 5-10-5 or 5-12-5", false, "5-10-5", "string");
-    TCLAP::ValueArg<int> thread_num_("", "thread-num", "thread-num",
-                                       false, 1, "int");
-    TCLAP::ValueArg<int> is_ssl_(
-        "", "is-ssl", "is-ssl is 1 means use wss connection, or use ws connection", 
-        false, 1, "int");
-
-    cmd.add(server_ip_);
-    cmd.add(port_);
-    cmd.add(wav_path_);
-    cmd.add(asr_mode_);
-    cmd.add(chunk_size_);
-    cmd.add(thread_num_);
-    cmd.add(is_ssl_);
-    cmd.parse(argc, argv);
-
-    std::string server_ip = server_ip_.getValue();
-    std::string port = port_.getValue();
-    std::string wav_path = wav_path_.getValue();
-    std::string asr_mode = asr_mode_.getValue();
-    std::string chunk_size_str = chunk_size_.getValue();
-    // get chunk_size
-    std::vector<int> chunk_size;
-    std::stringstream ss(chunk_size_str);
-    std::string item;   
-    while (std::getline(ss, item, '-')) {
-        try {
-            chunk_size.push_back(stoi(item));
-        } catch (const invalid_argument&) {
-            LOG(ERROR) << "Invalid argument: " << item;
-            exit(-1);
+        LOG(INFO) << "sended data len=" << len * sizeof(short);
+        // The most likely error that we will get is that the connection is
+        // not in the right state. Usually this means we tried to send a
+        // message to a connection that was closed or in the process of
+        // closing. While many errors here can be easily recovered from,
+        // in this simple example, we'll stop the data loop.
+        if (ec) {
+          m_client.get_alog().write(websocketpp::log::alevel::app,
+                                    "Send Error: " + ec.message());
+          break;
+        }
+        delete[] iArray;
+        // WaitABit();
+      }
+    } else {
+      int offset = 0;
+      int block_size = 204800;
+      len = audio.GetSpeechLen();
+      char* others_buff = audio.GetSpeechChar();
+
+      while (offset < len) {
+        int send_block = 0;
+        if (offset + block_size <= len) {
+          send_block = block_size;
+        } else {
+          send_block = len - offset;
         }
+        m_client.send(m_hdl, others_buff + offset, send_block,
+                      websocketpp::frame::opcode::binary, ec);
+        offset += send_block;
+      }
+
+      LOG(INFO) << "sended data len=" << len;
+      // The most likely error that we will get is that the connection is
+      // not in the right state. Usually this means we tried to send a
+      // message to a connection that was closed or in the process of
+      // closing. While many errors here can be easily recovered from,
+      // in this simple example, we'll stop the data loop.
+      if (ec) {
+        m_client.get_alog().write(websocketpp::log::alevel::app,
+                                  "Send Error: " + ec.message());
+      }
     }
 
-    int threads_num = thread_num_.getValue();
-    int is_ssl = is_ssl_.getValue();
+    nlohmann::json jsonresult;
+    jsonresult["is_speaking"] = false;
+    m_client.send(m_hdl, jsonresult.dump(), websocketpp::frame::opcode::text,
+                  ec);
+    WaitABit();
+  }
+  websocketpp::client<T> m_client;
+
+ private:
+  websocketpp::connection_hdl m_hdl;
+  websocketpp::lib::mutex m_lock;
+  bool m_open;
+  bool m_done;
+  int total_num = 0;
+};
 
-    std::vector<websocketpp::lib::thread> client_threads;
-    std::string uri = "";
-    if (is_ssl == 1) {
-        uri = "wss://" + server_ip + ":" + port;
-    } else {
-        uri = "ws://" + server_ip + ":" + port;
+int main(int argc, char* argv[]) {
+  google::InitGoogleLogging(argv[0]);
+  FLAGS_logtostderr = true;
+
+  TCLAP::CmdLine cmd("funasr-wss-client", ' ', "1.0");
+  TCLAP::ValueArg<std::string> server_ip_("", "server-ip", "server-ip", true,
+                                          "127.0.0.1", "string");
+  TCLAP::ValueArg<std::string> port_("", "port", "port", true, "10095",
+                                     "string");
+  TCLAP::ValueArg<std::string> wav_path_(
+      "", "wav-path",
+      "the input could be: wav_path, e.g.: asr_example.wav; pcm_path, e.g.: "
+      "asr_example.pcm; wav.scp, kaldi style wav list (wav_id \t wav_path)",
+      true, "", "string");
+  TCLAP::ValueArg<std::string> asr_mode_("", ASR_MODE, "offline, online, 2pass",
+                                         false, "2pass", "string");
+  TCLAP::ValueArg<std::string> chunk_size_("", "chunk-size",
+                                           "chunk_size: 5-10-5 or 5-12-5",
+                                           false, "5-10-5", "string");
+  TCLAP::ValueArg<int> thread_num_("", "thread-num", "thread-num", false, 1,
+                                   "int");
+  TCLAP::ValueArg<int> is_ssl_(
+      "", "is-ssl",
+      "is-ssl is 1 means use wss connection, or use ws connection", false, 1,
+      "int");
+
+  cmd.add(server_ip_);
+  cmd.add(port_);
+  cmd.add(wav_path_);
+  cmd.add(asr_mode_);
+  cmd.add(chunk_size_);
+  cmd.add(thread_num_);
+  cmd.add(is_ssl_);
+  cmd.parse(argc, argv);
+
+  std::string server_ip = server_ip_.getValue();
+  std::string port = port_.getValue();
+  std::string wav_path = wav_path_.getValue();
+  std::string asr_mode = asr_mode_.getValue();
+  std::string chunk_size_str = chunk_size_.getValue();
+  // get chunk_size
+  std::vector<int> chunk_size;
+  std::stringstream ss(chunk_size_str);
+  std::string item;
+  while (std::getline(ss, item, '-')) {
+    try {
+      chunk_size.push_back(stoi(item));
+    } catch (const invalid_argument&) {
+      LOG(ERROR) << "Invalid argument: " << item;
+      exit(-1);
     }
-
-    // read wav_path
-    std::vector<string> wav_list;
-    std::vector<string> wav_ids;
-    string default_id = "wav_default_id";
-    if(IsTargetFile(wav_path, "scp")){
-        ifstream in(wav_path);
-        if (!in.is_open()) {
-            printf("Failed to open scp file");
-            return 0;
-        }
-        string line;
-        while(getline(in, line))
-        {
-            istringstream iss(line);
-            string column1, column2;
-            iss >> column1 >> column2;
-            wav_list.emplace_back(column2);
-            wav_ids.emplace_back(column1);
-        }
-        in.close();
-    }else{
-        wav_list.emplace_back(wav_path);
-        wav_ids.emplace_back(default_id);
+  }
+
+  int threads_num = thread_num_.getValue();
+  int is_ssl = is_ssl_.getValue();
+
+  std::string uri = "";
+  if (is_ssl == 1) {
+    uri = "wss://" + server_ip + ":" + port;
+  } else {
+    uri = "ws://" + server_ip + ":" + port;
+  }
+
+  // read wav_path
+  std::vector<string> wav_list;
+  std::vector<string> wav_ids;
+  string default_id = "wav_default_id";
+  if (IsTargetFile(wav_path, "scp")) {
+    ifstream in(wav_path);
+    if (!in.is_open()) {
+      printf("Failed to open scp file");
+      return 0;
     }
-    
+    string line;
+    while (getline(in, line)) {
+      istringstream iss(line);
+      string column1, column2;
+      iss >> column1 >> column2;
+      wav_list.emplace_back(column2);
+      wav_ids.emplace_back(column1);
+    }
+    in.close();
+  } else {
+    wav_list.emplace_back(wav_path);
+    wav_ids.emplace_back(default_id);
+  }
+
+  for (size_t wav_i = 0; wav_i < wav_list.size(); wav_i = wav_i + threads_num) {
+    std::vector<websocketpp::lib::thread> client_threads;
     for (size_t i = 0; i < threads_num; i++) {
-        client_threads.emplace_back([uri, wav_list, wav_ids, asr_mode, chunk_size, is_ssl]() {
-          if (is_ssl == 1) {
-            WebsocketClient<websocketpp::config::asio_tls_client> c(is_ssl);
+      if (wav_i + i >= wav_list.size()) {
+        break;
+      }
+      std::vector<string> tmp_wav_list;
+      std::vector<string> tmp_wav_ids;
 
-            c.m_client.set_tls_init_handler(bind(&OnTlsInit, ::_1));
+      tmp_wav_list.emplace_back(wav_list[wav_i + i]);
+      tmp_wav_ids.emplace_back(wav_ids[wav_i + i]);
 
-            c.run(uri, wav_list, wav_ids, asr_mode, chunk_size);
-          } else {
-            WebsocketClient<websocketpp::config::asio_client> c(is_ssl);
+      client_threads.emplace_back(
+          [uri, tmp_wav_list, tmp_wav_ids, asr_mode, chunk_size, is_ssl]() {
+            if (is_ssl == 1) {
+              WebsocketClient<websocketpp::config::asio_tls_client> c(is_ssl);
 
-            c.run(uri, wav_list, wav_ids, asr_mode, chunk_size);
-          }
-        });
+              c.m_client.set_tls_init_handler(bind(&OnTlsInit, ::_1));
+
+              c.run(uri, tmp_wav_list, tmp_wav_ids, asr_mode, chunk_size);
+            } else {
+              WebsocketClient<websocketpp::config::asio_client> c(is_ssl);
+
+              c.run(uri, tmp_wav_list, tmp_wav_ids, asr_mode, chunk_size);
+            }
+          });
     }
 
     for (auto& t : client_threads) {
-        t.join();
+      t.join();
     }
+  }
 }

funasr/runtime/websocket/websocket-server-2pass.cpp

@@ -54,33 +54,31 @@ context_ptr WebSocketServer::on_tls_init(tls_mode mode,
 }
 
 nlohmann::json handle_result(FUNASR_RESULT result) {
+  websocketpp::lib::error_code ec;
+  nlohmann::json jsonresult;
+  jsonresult["text"] = "";
+
+  std::string tmp_online_msg = FunASRGetResult(result, 0);
+  if (tmp_online_msg != "") {
+    LOG(INFO) << "online_res :" << tmp_online_msg;
+    jsonresult["text"] = tmp_online_msg;
+    jsonresult["mode"] = "2pass-online";
+  }
+  std::string tmp_tpass_msg = FunASRGetTpassResult(result, 0);
+  if (tmp_tpass_msg != "") {
+    LOG(INFO) << "offline results : " << tmp_tpass_msg;
+    jsonresult["text"] = tmp_tpass_msg;
+    jsonresult["mode"] = "2pass-offline";
+  }
 
-    websocketpp::lib::error_code ec;
-    nlohmann::json jsonresult;
-    jsonresult["text"]="";
-
-    std::string tmp_online_msg = FunASRGetResult(result, 0);
-    if (tmp_online_msg != "") {
-      LOG(INFO) << "online_res :" << tmp_online_msg;
-      jsonresult["text"] = tmp_online_msg; 
-      jsonresult["mode"] = "2pass-online";
-    }
-    std::string tmp_tpass_msg = FunASRGetTpassResult(result, 0);
-    if (tmp_tpass_msg != "") {
-      LOG(INFO) << "offline results : " << tmp_tpass_msg;
-      jsonresult["text"] = tmp_tpass_msg; 
-      jsonresult["mode"] = "2pass-offline";    
-    }
-
-    return jsonresult;
+  return jsonresult;
 }
 // feed buffer to asr engine for decoder
 void WebSocketServer::do_decoder(
     std::vector<char>& buffer, websocketpp::connection_hdl& hdl,
     nlohmann::json& msg, std::vector<std::vector<std::string>>& punc_cache,
-    websocketpp::lib::mutex& thread_lock, bool& is_final,
-    std::string wav_name, FUNASR_HANDLE& tpass_online_handle) {
- 
+    websocketpp::lib::mutex& thread_lock, bool& is_final, std::string wav_name,
+    FUNASR_HANDLE& tpass_online_handle) {
   // lock for each connection
   scoped_lock guard(thread_lock);
   FUNASR_RESULT Result = nullptr;
@@ -99,30 +97,33 @@ void WebSocketServer::do_decoder(
     msg["mode"] = "2pass";
     asr_mode_ = 2;
   }
- 
+
   try {
-    // loop to send chunk_size 800*2 data to asr engine.   TODO: chunk_size need get from client 
+    // loop to send chunk_size 800*2 data to asr engine.   TODO: chunk_size need
+    // get from client
     while (buffer.size() >= 800 * 2) {
-      std::vector<char> subvector = {buffer.begin(),
-                                      buffer.begin() + 800 * 2};
+      std::vector<char> subvector = {buffer.begin(), buffer.begin() + 800 * 2};
       buffer.erase(buffer.begin(), buffer.begin() + 800 * 2);
 
-      try{
-        Result =
-            FunTpassInferBuffer(tpass_handle, tpass_online_handle,
-                                subvector.data(), subvector.size(), punc_cache,
-                                false, msg["audio_fs"], msg["wav_format"], (ASR_TYPE)asr_mode_);
-      }catch (std::exception const &e)
-      {
-          LOG(ERROR)<<e.what();
+      try {
+        if (tpass_online_handle) {
+          Result = FunTpassInferBuffer(tpass_handle, tpass_online_handle,
+                                       subvector.data(), subvector.size(),
+                                       punc_cache, false, msg["audio_fs"],
+                                       msg["wav_format"], (ASR_TYPE)asr_mode_);
+
+        } else {
+          return;
+        }
+      } catch (std::exception const& e) {
+        LOG(ERROR) << e.what();
       }
       if (Result) {
         websocketpp::lib::error_code ec;
-        nlohmann::json jsonresult =
-            handle_result(Result);
+        nlohmann::json jsonresult = handle_result(Result);
         jsonresult["wav_name"] = wav_name;
         jsonresult["is_final"] = false;
-        if(jsonresult["text"] != "") {
+        if (jsonresult["text"] != "") {
           if (is_ssl) {
             wss_server_->send(hdl, jsonresult.dump(),
                               websocketpp::frame::opcode::text, ec);
@@ -133,25 +134,26 @@ void WebSocketServer::do_decoder(
         }
         FunASRFreeResult(Result);
       }
-
     }
-    if(is_final){
-
-      try{
-        Result = FunTpassInferBuffer(tpass_handle, tpass_online_handle,
-                                      buffer.data(), buffer.size(), punc_cache,
-                                      is_final, msg["audio_fs"], msg["wav_format"], (ASR_TYPE)asr_mode_);
-      }catch (std::exception const &e)
-      {
-          LOG(ERROR)<<e.what();
+    if (is_final) {
+      try {
+        if (tpass_online_handle) {
+          Result = FunTpassInferBuffer(tpass_handle, tpass_online_handle,
+                                       buffer.data(), buffer.size(), punc_cache,
+                                       is_final, msg["audio_fs"],
+                                       msg["wav_format"], (ASR_TYPE)asr_mode_);
+        } else {
+          return;
+        }
+      } catch (std::exception const& e) {
+        LOG(ERROR) << e.what();
       }
-      for(auto &vec:punc_cache){
+      for (auto& vec : punc_cache) {
         vec.clear();
       }
       if (Result) {
         websocketpp::lib::error_code ec;
-        nlohmann::json jsonresult =
-            handle_result(Result);
+        nlohmann::json jsonresult = handle_result(Result);
         jsonresult["wav_name"] = wav_name;
         jsonresult["is_final"] = true;
         if (is_ssl) {
@@ -179,7 +181,7 @@ void WebSocketServer::on_open(websocketpp::connection_hdl hdl) {
                                            // connection
   data_msg->samples = std::make_shared<std::vector<char>>();
   data_msg->thread_lock = new websocketpp::lib::mutex();
- 
+
   data_msg->msg = nlohmann::json::parse("{}");
   data_msg->msg["wav_format"] = "pcm";
   data_msg->msg["audio_fs"] = 16000;
@@ -191,25 +193,37 @@ void WebSocketServer::on_open(websocketpp::connection_hdl hdl) {
   // data_msg->tpass_online_handle = tpass_online_handle;
   data_map.emplace(hdl, data_msg);
   LOG(INFO) << "on_open, active connections: " << data_map.size();
- 
 }
 
-void WebSocketServer::on_close(websocketpp::connection_hdl hdl) {
-  scoped_lock guard(m_lock);
-  std::shared_ptr<FUNASR_MESSAGE> data_msg  = nullptr;
+void remove_hdl(
+    websocketpp::connection_hdl hdl,
+    std::map<websocketpp::connection_hdl, std::shared_ptr<FUNASR_MESSAGE>,
+             std::owner_less<websocketpp::connection_hdl>>& data_map) {
+  // return;
+  std::shared_ptr<FUNASR_MESSAGE> data_msg = nullptr;
   auto it_data = data_map.find(hdl);
   if (it_data != data_map.end()) {
-    data_msg  = it_data->second;
-  }
-  else
-  {
+    data_msg = it_data->second;
+  } else {
     return;
   }
-  scoped_lock guard_decoder(*(data_msg->thread_lock));  //wait for do_decoder finished and avoid access freed tpass_online_handle 
-  LOG(INFO) << "----------------FunTpassOnlineUninit----------------------";
-  FunTpassOnlineUninit(data_msg->tpass_online_handle);
+  // scoped_lock guard_decoder(*(data_msg->thread_lock));  //wait for do_decoder
+  // finished and avoid access freed tpass_online_handle
+  unique_lock guard_decoder(*(data_msg->thread_lock));
+  if (data_msg->tpass_online_handle) {
+    LOG(INFO) << "----------------FunTpassOnlineUninit----------------------";
+    FunTpassOnlineUninit(data_msg->tpass_online_handle);
+    data_msg->tpass_online_handle = nullptr;
+  }
+  guard_decoder.unlock();
+  delete data_msg->thread_lock;
   data_map.erase(hdl);  // remove data vector when  connection is closed
-  LOG(INFO) << "on_close, active connections: "<< data_map.size();
+}
+
+void WebSocketServer::on_close(websocketpp::connection_hdl hdl) {
+  scoped_lock guard(m_lock);
+  remove_hdl(hdl, data_map);
+  LOG(INFO) << "on_close, active connections: " << data_map.size();
 }
 
 // remove closed connection
@@ -234,7 +248,7 @@ void WebSocketServer::check_and_clean_connection() {
     iter++;
   }
   for (auto hdl : to_remove) {
-    data_map.erase(hdl);
+    remove_hdl(hdl, data_map);
     LOG(INFO) << "remove one connection ";
   }
 }
@@ -248,20 +262,23 @@ void WebSocketServer::on_message(websocketpp::connection_hdl hdl,
   auto it_data = data_map.find(hdl);
   if (it_data != data_map.end()) {
     msg_data = it_data->second;
+  } else {
+    lock.unlock();
+    return;
   }
 
   std::shared_ptr<std::vector<char>> sample_data_p = msg_data->samples;
   std::shared_ptr<std::vector<std::vector<std::string>>> punc_cache_p =
       msg_data->punc_cache;
   websocketpp::lib::mutex* thread_lock_p = msg_data->thread_lock;
- 
+
   lock.unlock();
 
   if (sample_data_p == nullptr) {
     LOG(INFO) << "error when fetch sample data vector";
     return;
   }
- 
+
   const std::string& payload = msg->get_payload();  // get msg type
 
   switch (msg->get_opcode()) {
@@ -280,17 +297,19 @@ void WebSocketServer::on_message(websocketpp::connection_hdl hdl,
       if (jsonresult.contains("audio_fs")) {
         msg_data->msg["audio_fs"] = jsonresult["audio_fs"];
       }
-      if (jsonresult.contains("chunk_size")){
-        if(msg_data->tpass_online_handle == NULL){
-          std::vector<int> chunk_size_vec = jsonresult["chunk_size"].get<std::vector<int>>();
-          LOG(INFO) << "----------------FunTpassOnlineInit----------------------";
+      if (jsonresult.contains("chunk_size")) {
+        if (msg_data->tpass_online_handle == NULL) {
+          std::vector<int> chunk_size_vec =
+              jsonresult["chunk_size"].get<std::vector<int>>();
+          LOG(INFO)
+              << "----------------FunTpassOnlineInit----------------------";
           FUNASR_HANDLE tpass_online_handle =
               FunTpassOnlineInit(tpass_handle, chunk_size_vec);
           msg_data->tpass_online_handle = tpass_online_handle;
         }
       }
-      LOG(INFO) << "jsonresult=" << jsonresult << ", msg_data->msg="
-                << msg_data->msg;
+      LOG(INFO) << "jsonresult=" << jsonresult
+                << ", msg_data->msg=" << msg_data->msg;
       if (jsonresult["is_speaking"] == false ||
           jsonresult["is_finished"] == true) {
         LOG(INFO) << "client done";
@@ -311,30 +330,31 @@ void WebSocketServer::on_message(websocketpp::connection_hdl hdl,
       // recived binary data
       const auto* pcm_data = static_cast<const char*>(payload.data());
       int32_t num_samples = payload.size();
- 
+
       if (isonline) {
         sample_data_p->insert(sample_data_p->end(), pcm_data,
                               pcm_data + num_samples);
-        int setpsize = 800 * 2;  // TODO, need get from client 
-		    // if sample_data size > setpsize, we post data to decode
+        int setpsize =
+            800 * 2;  // TODO, need get from client
+                      // if sample_data size > setpsize, we post data to decode
         if (sample_data_p->size() > setpsize) {
-            int chunksize = floor(sample_data_p->size() / setpsize);
-            // make sure the subvector size is an integer multiple of setpsize
-            std::vector<char> subvector = {
-                  sample_data_p->begin(),
-                  sample_data_p->begin() + chunksize * setpsize};
-            // keep remain in sample_data
-            sample_data_p->erase(sample_data_p->begin(),
-                                  sample_data_p->begin() + chunksize * setpsize);
-            // post to decode
-            asio::post(io_decoder_,
-                        std::bind(&WebSocketServer::do_decoder, this,
-                                  std::move(subvector), std::move(hdl),
-                                  std::ref(msg_data->msg),
-                                  std::ref(*(punc_cache_p.get())),
-                                  std::ref(*thread_lock_p), std::move(false),
-                                  msg_data->msg["wav_name"],
-                                  std::ref(msg_data->tpass_online_handle)));
+          int chunksize = floor(sample_data_p->size() / setpsize);
+          // make sure the subvector size is an integer multiple of setpsize
+          std::vector<char> subvector = {
+              sample_data_p->begin(),
+              sample_data_p->begin() + chunksize * setpsize};
+          // keep remain in sample_data
+          sample_data_p->erase(sample_data_p->begin(),
+                               sample_data_p->begin() + chunksize * setpsize);
+          // post to decode
+          asio::post(io_decoder_,
+                     std::bind(&WebSocketServer::do_decoder, this,
+                               std::move(subvector), std::move(hdl),
+                               std::ref(msg_data->msg),
+                               std::ref(*(punc_cache_p.get())),
+                               std::ref(*thread_lock_p), std::move(false),
+                               msg_data->msg["wav_name"],
+                               std::ref(msg_data->tpass_online_handle)));
         }
       } else {
         sample_data_p->insert(sample_data_p->end(), pcm_data,