for cpp websocket core dump (#828)

zhaomingwork

2023-08-10 30f0c7ff2941ab08edb8cb257eb6cef74be42ec7

for cpp websocket core dump (#828)

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

@@ -12,18 +12,20 @@
//                     [--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 <glog/logging.h>

#include <atomic>
#include <fstream>
#include <iostream>
#include <sstream>
#include <thread>
#include <websocketpp/client.hpp>
#include <websocketpp/common/thread.hpp>
#include <websocketpp/config/asio_client.hpp>
#include <iostream>
#include <fstream>
#include <sstream>
#include <atomic>
#include <thread>
#include <glog/logging.h>

#include "audio.h"
#include "nlohmann/json.hpp"
@@ -51,7 +53,8 @@
}

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;
@@ -109,20 +112,26 @@
        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;
        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(m_hdl, websocketpp::close::status::going_away, "", ec);
                //     if (ec){
                //         LOG(ERROR)<< "Error closing connection " << ec.message();
                //     }
                // }
        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) {
  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 =
@@ -143,17 +152,12 @@
        // 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);
        }

    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
@@ -183,7 +187,8 @@
        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) {
  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;

@@ -192,15 +197,12 @@
        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 ;
      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 ;
      if (!audio.LoadPcmwav(wav_path.c_str(), &sampling_rate)) return;
        }else{
            wav_format = "others";
            if (!audio.LoadOthers2Char(wav_path.c_str()))
                return ;
      if (!audio.LoadOthers2Char(wav_path.c_str())) return;
        }

        float* buff;
@@ -221,6 +223,7 @@
                  break;
                }
            }

            if (wait) {
                // LOG(INFO) << "wait.." << m_open;
                WaitABit();
@@ -326,24 +329,30 @@
};

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)", 
  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<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");
      "", "is-ssl",
      "is-ssl is 1 means use wss connection, or use ws connection", false, 1,
      "int");

    cmd.add(server_ip_);
    cmd.add(port_);
@@ -375,7 +384,6 @@
    int threads_num = thread_num_.getValue();
    int is_ssl = is_ssl_.getValue();

    std::vector<websocketpp::lib::thread> client_threads;
    std::string uri = "";
    if (is_ssl == 1) {
        uri = "wss://" + server_ip + ":" + port;
@@ -394,8 +402,7 @@
            return 0;
        }
        string line;
        while(getline(in, line))
        {
    while (getline(in, line)) {
            istringstream iss(line);
            string column1, column2;
            iss >> column1 >> column2;
@@ -408,18 +415,30 @@
        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 (wav_i + i >= wav_list.size()) {
        break;
      }
      std::vector<string> tmp_wav_list;
      std::vector<string> tmp_wav_ids;

      tmp_wav_list.emplace_back(wav_list[wav_i + i]);
      tmp_wav_ids.emplace_back(wav_ids[wav_i + i]);

      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.m_client.set_tls_init_handler(bind(&OnTlsInit, ::_1));

            c.run(uri, wav_list, wav_ids, asr_mode, chunk_size);
              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, wav_list, wav_ids, asr_mode, chunk_size);
              c.run(uri, tmp_wav_list, tmp_wav_ids, asr_mode, chunk_size);
          }
        });
    }
@@ -428,3 +447,4 @@
        t.join();
    }
}
}

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

@@ -54,7 +54,6 @@
}



nlohmann::json handle_result(FUNASR_RESULT result) {



    websocketpp::lib::error_code ec;

    nlohmann::json jsonresult;

    jsonresult["text"]="";

@@ -78,9 +77,8 @@
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;

@@ -101,25 +99,28 @@
  }

 

  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)

      {

        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"] != "") {

@@ -133,16 +134,18 @@
        }

        FunASRFreeResult(Result);

      }



    }

    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_);

      }catch (std::exception const &e)

      {

                                       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){

@@ -150,8 +153,7 @@
      }

      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) {

@@ -191,24 +193,36 @@
  // 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);

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

  {

  } else {

    return;

  }

  scoped_lock guard_decoder(*(data_msg->thread_lock));  //wait for do_decoder finished and avoid access freed 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

}



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();

}



@@ -234,7 +248,7 @@
    iter++;

  }

  for (auto hdl : to_remove) {

    data_map.erase(hdl);

    remove_hdl(hdl, data_map);

    LOG(INFO) << "remove one connection ";

  }

}

@@ -248,6 +262,9 @@
  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;

@@ -282,15 +299,17 @@
      }

      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----------------------";

          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";

@@ -315,7 +334,8 @@
      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 
        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);