update websocket-server-2pass

雾聪

2023-08-11 b83536869d3728b61a01535b101205b4811f6f71

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

@@ -143,7 +143,7 @@
        FunASRFreeResult(Result);

      }

    }

    if (is_final) {

    if (is_final && !msg["is_eof"]) {

      try {

        if (tpass_online_handle) {

          Result = FunTpassInferBuffer(tpass_handle, tpass_online_handle,

@@ -203,12 +203,8 @@
    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

    // FUNASR_HANDLE tpass_online_handle =

    //     FunTpassOnlineInit(tpass_handle, chunk_size);

    // data_msg->tpass_online_handle = tpass_online_handle;

    data_map.emplace(hdl, data_msg);

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

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

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

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

  }

@@ -230,14 +226,13 @@
  // finished and avoid access freed tpass_online_handle

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

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

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

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

    FunTpassOnlineUninit(data_msg->tpass_online_handle);

    data_msg->tpass_online_handle = nullptr;

     data_map.erase(hdl);

  }

 

  guard_decoder.unlock();

    // remove data vector when  connection is closed

}



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

@@ -257,10 +252,10 @@
 

// remove closed connection

void WebSocketServer::check_and_clean_connection() {

   LOG(INFO)<<"***********begin 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();;

    // 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

@@ -291,7 +286,6 @@
        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();

        

@@ -300,7 +294,7 @@
    }

    for (auto hdl : to_remove) {

      remove_hdl(hdl, data_map);

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

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

    }

  }

}

@@ -353,8 +347,6 @@
        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;

@@ -368,7 +360,6 @@


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

        try{

		  
          asio::post(

              io_decoder_,

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

@@ -382,7 +373,6 @@
        catch (std::exception const &e)

        {

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

			
        }

      }

      break;