For cpp ws srv long file core dump (#833)

zhaomingwork

2023-08-11 dd29390ef0fdfc0ad1c13f0246b3184323eedb93

For cpp ws srv long file core dump (#833)

* for cpp ws srv long wav core dump

* cpp wssrv run one thread to free resources

* little change for close situation

* change return to continue

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

@@ -15,7 +15,7 @@
#include <thread>

#include <utility>

#include <vector>



#include <chrono>

context_ptr WebSocketServer::on_tls_init(tls_mode mode,

                                         websocketpp::connection_hdl hdl,

                                         std::string& s_certfile,

@@ -83,8 +83,10 @@
  scoped_lock guard(thread_lock);

  if(!tpass_online_handle){

      LOG(INFO) << "tpass_online_handle  is free, return";

      msg["access_num"]=(int)msg["access_num"]-1;

      return;

  }

  try {

  FUNASR_RESULT Result = nullptr;

  int asr_mode_ = 2;

  if (msg.contains("mode")) {

@@ -102,10 +104,10 @@
    asr_mode_ = 2;

  }



  try {



    // loop to send chunk_size 800*2 data to asr engine.   TODO: chunk_size need

    // get from client

    while (buffer.size() >= 800 * 2) {

    while (buffer.size() >= 800 * 2 && !msg["is_eof"]) {

      std::vector<char> subvector = {buffer.begin(), buffer.begin() + 800 * 2};

      buffer.erase(buffer.begin(), buffer.begin() + 800 * 2);



@@ -117,10 +119,13 @@
                                       msg["wav_format"], (ASR_TYPE)asr_mode_);



        } else {

          msg["access_num"]=(int)msg["access_num"]-1;

          return;

        }

      } catch (std::exception const& e) {

        LOG(ERROR) << e.what();

        msg["access_num"]=(int)msg["access_num"]-1;

        return;

      }

      if (Result) {

        websocketpp::lib::error_code ec;

@@ -147,10 +152,14 @@
                                       is_final, msg["audio_fs"],

                                       msg["wav_format"], (ASR_TYPE)asr_mode_);

        } else {

          msg["access_num"]=(int)msg["access_num"]-1;

	 
          return;

        }

      } catch (std::exception const& e) {

        LOG(ERROR) << e.what();

        msg["access_num"]=(int)msg["access_num"]-1;

        return;

      }

      if(punc_cache.size()>0){

        for (auto& vec : punc_cache) {

@@ -175,16 +184,17 @@


  } catch (std::exception const& e) {

    std::cerr << "Error: " << e.what() << std::endl;



  }

  msg["access_num"]=(int)msg["access_num"]-1;

 
}



void WebSocketServer::on_open(websocketpp::connection_hdl hdl) {

  scoped_lock guard(m_lock);     // for threads safty

  try{

    check_and_clean_connection();  // remove closed connection 
  }catch (std::exception const& e) {

    std::cerr << "Error: " << e.what() << std::endl;

  }

 




  std::shared_ptr<FUNASR_MESSAGE> data_msg =

      std::make_shared<FUNASR_MESSAGE>();  // put a new data vector for new

@@ -195,6 +205,8 @@
  data_msg->msg = nlohmann::json::parse("{}");

  data_msg->msg["wav_format"] = "pcm";

  data_msg->msg["audio_fs"] = 16000;

  data_msg->msg["access_num"] = 0; // the number of access for this object, when it is 0, we can free it saftly

  data_msg->msg["is_eof"]=false; // if this connection is closed

  data_msg->punc_cache =

      std::make_shared<std::vector<std::vector<std::string>>>(2);

  // std::vector<int> chunk_size = {5, 10, 5};  //TODO, need get from client

@@ -203,6 +215,9 @@
  // data_msg->tpass_online_handle = tpass_online_handle;

  data_map.emplace(hdl, data_msg);

  LOG(INFO) << "on_open, active connections: " << data_map.size();

    }catch (std::exception const& e) {

    std::cerr << "Error: " << e.what() << std::endl;

  }

}



void remove_hdl(

@@ -220,30 +235,45 @@
  // 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) {

  if (data_msg->msg["access_num"]==0 && data_msg->msg["is_eof"]==true) {

    LOG(INFO) << "----------------FunTpassOnlineUninit----------------------";

    FunTpassOnlineUninit(data_msg->tpass_online_handle);

    data_msg->tpass_online_handle = nullptr;

    data_map.erase(hdl);

  }

  

 

  guard_decoder.unlock();

  data_map.erase(hdl);  // remove data vector when  connection is closed

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

  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 {

    return;

  }

  unique_lock guard_decoder(*(data_msg->thread_lock));

  data_msg->msg["is_eof"]=true;

  guard_decoder.unlock();

  //LOG(INFO) << "on_close, active connections: " << data_map.size();

}



 
// remove closed connection

void WebSocketServer::check_and_clean_connection() {

   LOG(INFO)<<"***********begin check_and_clean_connection ****************";

 while(true){



  std::this_thread::sleep_for(std::chrono::milliseconds(5000));

  LOG(INFO) << "run check_and_clean_connection" <<", active connections: " << data_map.size();;

  std::vector<websocketpp::connection_hdl> to_remove;  // remove list

  auto iter = data_map.begin();

  while (iter != data_map.end()) {  // loop to find closed connection

    websocketpp::connection_hdl hdl = iter->first;



    try{

    if (is_ssl) {

      wss_server::connection_ptr con = wss_server_->get_con_from_hdl(hdl);

      if (con->get_state() != 1) {  // session::state::open ==1

@@ -255,13 +285,34 @@
        to_remove.push_back(hdl);

      }

    }



    }

    catch (std::exception const &e)

   
    {

         // if connection is close, we set is_eof = true

          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 {

              continue;

            }

            unique_lock guard_decoder(*(data_msg->thread_lock));

            data_msg->msg["is_eof"]=true;

            guard_decoder.unlock();

            //remove_hdl(hdl, data_map);

            to_remove.push_back(hdl);

            LOG(INFO)<<"connection is closed: "<<e.what();

			
    }

    iter++;

  }

  for (auto hdl : to_remove) {

    remove_hdl(hdl, data_map);

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

  }



 }

}

void WebSocketServer::on_message(websocketpp::connection_hdl hdl,

                                 message_ptr msg) {

@@ -291,7 +342,7 @@
  }



  const std::string& payload = msg->get_payload();  // get msg type



  unique_lock guard_decoder(*(thread_lock_p)); // mutex for one connection

  switch (msg->get_opcode()) {

    case websocketpp::frame::opcode::text: {

      nlohmann::json jsonresult = nlohmann::json::parse(payload);

@@ -327,6 +378,7 @@


        // if it is in final message, post the sample_data to decode

        try{

		  
          asio::post(

              io_decoder_,

              std::bind(&WebSocketServer::do_decoder, this,

@@ -335,10 +387,12 @@
                        std::ref(*thread_lock_p), std::move(true),

                        msg_data->msg["wav_name"],

                        std::ref(msg_data->tpass_online_handle)));

          msg_data->msg["access_num"]=(int)(msg_data->msg["access_num"])+1;

        }

        catch (std::exception const &e)

        {

            LOG(ERROR)<<e.what();

			
        }

      }

      break;

@@ -374,6 +428,7 @@
                                std::ref(*thread_lock_p), std::move(false),

                                msg_data->msg["wav_name"],

                                std::ref(msg_data->tpass_online_handle)));

            msg_data->msg["access_num"]=(int)(msg_data->msg["access_num"])+1;

          }

          catch (std::exception const &e)

          {

@@ -389,6 +444,7 @@
    default:

      break;

  }

  guard_decoder.unlock();

}



// init asr model

@@ -400,6 +456,10 @@
      LOG(ERROR) << "FunTpassInit init failed";

      exit(-1);

    }

    LOG(INFO) << "initAsr run check_and_clean_connection";

    std::thread clean_thread(&WebSocketServer::check_and_clean_connection,this);  
    clean_thread.detach();

    LOG(INFO) << "initAsr run check_and_clean_connection finished";



  } catch (const std::exception& e) {

    LOG(INFO) << e.what();