From ffb05b9ae7eccc47416e9e7fae9dea54d400a245 Mon Sep 17 00:00:00 2001
From: 雾聪 <wucong.lyb@alibaba-inc.com>
Date: 星期四, 10 八月 2023 19:05:51 +0800
Subject: [PATCH] Merge branch 'main' of https://github.com/alibaba-damo-academy/FunASR into main
---
funasr/runtime/websocket/websocket-server-2pass.cpp | 242 ++++++++++++++++++++++++++++--------------------
1 files changed, 141 insertions(+), 101 deletions(-)
diff --git a/funasr/runtime/websocket/websocket-server-2pass.cpp b/funasr/runtime/websocket/websocket-server-2pass.cpp
index 7df6341..50f0edc 100644
--- a/funasr/runtime/websocket/websocket-server-2pass.cpp
+++ b/funasr/runtime/websocket/websocket-server-2pass.cpp
@@ -54,35 +54,37 @@
}
nlohmann::json handle_result(FUNASR_RESULT result) {
+ websocketpp::lib::error_code ec;
+ nlohmann::json jsonresult;
+ jsonresult["text"] = "";
- 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";
+ }
- 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);
+ if(!tpass_online_handle){
+ LOG(INFO) << "tpass_online_handle is free, return";
+ return;
+ }
FUNASR_RESULT Result = nullptr;
int asr_mode_ = 2;
if (msg.contains("mode")) {
@@ -99,30 +101,33 @@
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 +138,28 @@
}
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){
- vec.clear();
+ if(punc_cache.size()>0){
+ 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) {
@@ -178,8 +186,8 @@
std::make_shared<FUNASR_MESSAGE>(); // put a new data vector for new
// connection
data_msg->samples = std::make_shared<std::vector<char>>();
- data_msg->thread_lock = new websocketpp::lib::mutex();
-
+ data_msg->thread_lock = std::make_shared<websocketpp::lib::mutex>();
+
data_msg->msg = nlohmann::json::parse("{}");
data_msg->msg["wav_format"] = "pcm";
data_msg->msg["audio_fs"] = 16000;
@@ -191,25 +199,38 @@
// data_msg->tpass_online_handle = tpass_online_handle;
data_map.emplace(hdl, data_msg);
LOG(INFO) << "on_open, active connections: " << data_map.size();
+}
+
+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) {
+ 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;
+ }
+ // 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();
+ data_map.erase(hdl); // remove data vector when connection is closed
}
void WebSocketServer::on_close(websocketpp::connection_hdl hdl) {
scoped_lock guard(m_lock);
- 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;
- }
- 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);
- data_map.erase(hdl); // remove data vector when connection is closed
- LOG(INFO) << "on_close, active connections: "<< data_map.size();
+ remove_hdl(hdl, data_map);
+ LOG(INFO) << "on_close, active connections: " << data_map.size();
}
// remove closed connection
@@ -234,7 +255,7 @@
iter++;
}
for (auto hdl : to_remove) {
- data_map.erase(hdl);
+ remove_hdl(hdl, data_map);
LOG(INFO) << "remove one connection ";
}
}
@@ -248,20 +269,23 @@
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;
-
+ std::shared_ptr<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,30 +304,38 @@
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";
// if it is in final message, post the sample_data to decode
- asio::post(
- io_decoder_,
- std::bind(&WebSocketServer::do_decoder, this,
- std::move(*(sample_data_p.get())), std::move(hdl),
- std::ref(msg_data->msg), std::ref(*(punc_cache_p.get())),
- std::ref(*thread_lock_p), std::move(true),
- msg_data->msg["wav_name"],
- std::ref(msg_data->tpass_online_handle)));
+ try{
+ asio::post(
+ io_decoder_,
+ std::bind(&WebSocketServer::do_decoder, this,
+ std::move(*(sample_data_p.get())), std::move(hdl),
+ std::ref(msg_data->msg), std::ref(*(punc_cache_p.get())),
+ std::ref(*thread_lock_p), std::move(true),
+ msg_data->msg["wav_name"],
+ std::ref(msg_data->tpass_online_handle)));
+ }
+ catch (std::exception const &e)
+ {
+ LOG(ERROR)<<e.what();
+ }
}
break;
}
@@ -311,30 +343,38 @@
// 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);
+ 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);
+
+ try{
// 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)));
+ 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)));
+ }
+ catch (std::exception const &e)
+ {
+ LOG(ERROR)<<e.what();
+ }
}
} else {
sample_data_p->insert(sample_data_p->end(), pcm_data,
--
Gitblit v1.9.1