debug onnxruntime multithread bugs

funasr/runtime/grpc/paraformer_server.cc

@@ -35,37 +35,25 @@ ASRServicer::ASRServicer(const char* model_path, int thread_num, bool quantize)
     init_flag = 0;
 }
 
-void ASRServicer::clear_states(const std::string& user) {
-    clear_buffers(user);
-    clear_transcriptions(user);
-}
-
-void ASRServicer::clear_buffers(const std::string& user) {
-    if (client_buffers.count(user)) {
-        client_buffers.erase(user);
-    }
-}
-
-void ASRServicer::clear_transcriptions(const std::string& user) {
-    if (client_transcription.count(user)) {
-        client_transcription.erase(user);
-    }
-}
-
-void ASRServicer::disconnect(const std::string& user) {
-    clear_states(user);
-    std::cout << "Disconnecting user: " << user << std::endl;
-}
-
 grpc::Status ASRServicer::Recognize(
     grpc::ServerContext* context,
     grpc::ServerReaderWriter<Response, Request>* stream) {
 
     Request req;
+    std::unordered_map<std::string, std::string> client_buffers;
+    std::unordered_map<std::string, std::string> client_transcription;
+    
     while (stream->Read(&req)) {
         if (req.isend()) {
             std::cout << "asr end" << std::endl;
-            disconnect(req.user());
+            // disconnect 
+            if (client_buffers.count(req.user())) {
+                client_buffers.erase(req.user());
+            }
+            if (client_transcription.count(req.user())) {
+                client_transcription.erase(req.user());
+            }
+
             Response res;
             res.set_sentence(
                 R"({"success": true, "detail": "asr end"})"
@@ -103,8 +91,14 @@ grpc::Status ASRServicer::Recognize(
                   auto& buf = client_buffers[req.user()];
                   buf.insert(buf.end(), req.audio_data().begin(), req.audio_data().end());
                 }
-                std::string tmp_data = this->client_buffers[req.user()];
+                std::string tmp_data = client_buffers[req.user()];
-                this->clear_states(req.user());
+                // clear_states
+                if (client_buffers.count(req.user())) {
+                    client_buffers.erase(req.user());
+                }
+                if (client_transcription.count(req.user())) {
+                    client_transcription.erase(req.user());
+                }
 
                 Response res;
                 res.set_sentence(
@@ -133,9 +127,6 @@ grpc::Status ASRServicer::Recognize(
                     res.set_user(req.user());
                     res.set_action("finish");
                     res.set_language(req.language());
-
-
-
                     stream->Write(res);
                 }
                 else {
@@ -155,7 +146,6 @@ grpc::Status ASRServicer::Recognize(
                     res.set_action("finish");
                     res.set_language(req.language());
 
-
                     stream->Write(res);
                 }
             }

funasr/runtime/grpc/paraformer_server.h

@@ -41,15 +41,9 @@ typedef struct
 class ASRServicer final : public ASR::Service {
   private:
     int init_flag;
-    std::unordered_map<std::string, std::string> client_buffers;
-    std::unordered_map<std::string, std::string> client_transcription;
 
   public:
     ASRServicer(const char* model_path, int thread_num, bool quantize);
-    void clear_states(const std::string& user);
-    void clear_buffers(const std::string& user);
-    void clear_transcriptions(const std::string& user);
-    void disconnect(const std::string& user);
     grpc::Status Recognize(grpc::ServerContext* context, grpc::ServerReaderWriter<Response, Request>* stream);
     RPASR_HANDLE AsrHanlde;
 	

funasr/runtime/onnxruntime/src/FeatureExtract.cpp

@@ -12,7 +12,9 @@ FeatureExtract::~FeatureExtract()
 {
     fftwf_free(fft_input);
     fftwf_free(fft_out);
-    fftwf_destroy_plan(p);
+    if(p){
+        fftwf_destroy_plan(p);
+    }
 }
 
 void FeatureExtract::reset()

funasr/runtime/onnxruntime/src/librapidasrapi.cpp

@@ -30,7 +30,7 @@ extern "C" {
 		int nStep = 0;
 		int nTotal = audio.get_queue_size();
 		while (audio.fetch(buff, len, flag) > 0) {
-			pRecogObj->reset();
+			//pRecogObj->reset();
 			string msg = pRecogObj->forward(buff, len, flag);
 			pResult->msg += msg;
 			nStep++;
@@ -60,7 +60,7 @@ extern "C" {
 		int nStep = 0;
 		int nTotal = audio.get_queue_size();
 		while (audio.fetch(buff, len, flag) > 0) {
-			pRecogObj->reset();
+			//pRecogObj->reset();
 			string msg = pRecogObj->forward(buff, len, flag);
 			pResult->msg += msg;
 			nStep++;
@@ -90,7 +90,7 @@ extern "C" {
 		int nStep = 0;
 		int nTotal = audio.get_queue_size();
 		while (audio.fetch(buff, len, flag) > 0) {
-			pRecogObj->reset();
+			//pRecogObj->reset();
 			string msg = pRecogObj->forward(buff, len, flag);
 			pResult->msg += msg;
 			nStep++;
@@ -120,7 +120,7 @@ extern "C" {
 		RPASR_RECOG_RESULT* pResult = new RPASR_RECOG_RESULT;
 		pResult->snippet_time = audio.get_time_len();
 		while (audio.fetch(buff, len, flag) > 0) {
-			pRecogObj->reset();
+			//pRecogObj->reset();
 			string msg = pRecogObj->forward(buff, len, flag);
 			pResult->msg+= msg;
 			nStep++;

funasr/runtime/onnxruntime/src/paraformer_onnx.cpp

@@ -18,7 +18,7 @@ ModelImp::ModelImp(const char* path,int nNumThread, bool quantize)
     cmvn_path = pathAppend(path, "am.mvn");
     config_path = pathAppend(path, "config.yaml");
 
-    fe = new FeatureExtract(3);
+    //fe = new FeatureExtract(3);
 
     //sessionOptions.SetInterOpNumThreads(1);
     sessionOptions.SetIntraOpNumThreads(nNumThread);
@@ -52,8 +52,8 @@ ModelImp::ModelImp(const char* path,int nNumThread, bool quantize)
 
 ModelImp::~ModelImp()
 {
-    if(fe)
+    //if(fe)
-        delete fe;
+    //    delete fe;
     if (m_session)
     {
         delete m_session;
@@ -65,7 +65,8 @@ ModelImp::~ModelImp()
 
 void ModelImp::reset()
 {
-    fe->reset();
+    //fe->reset();
+    printf("Not Imp!!!!!!\n");
 }
 
 void ModelImp::apply_lfr(Tensor<float>*& din)
@@ -159,8 +160,9 @@ string ModelImp::greedy_search(float * in, int nLen )
 
 string ModelImp::forward(float* din, int len, int flag)
 {
-
     Tensor<float>* in;
+    FeatureExtract* fe = new FeatureExtract(3);
+    fe->reset();
     fe->insert(din, len, flag);
     fe->fetch(in);
     apply_lfr(in);
@@ -192,7 +194,6 @@ string ModelImp::forward(float* din, int len, int flag)
         auto outputTensor = m_session->Run(run_option, m_szInputNames.data(), input_onnx.data(), m_szInputNames.size(), m_szOutputNames.data(), m_szOutputNames.size());
         std::vector<int64_t> outputShape = outputTensor[0].GetTensorTypeAndShapeInfo().GetShape();
 
-
         int64_t outputCount = std::accumulate(outputShape.begin(), outputShape.end(), 1, std::multiplies<int64_t>());
         float* floatData = outputTensor[0].GetTensorMutableData<float>();
         auto encoder_out_lens = outputTensor[1].GetTensorMutableData<int64_t>();
@@ -203,9 +204,14 @@ string ModelImp::forward(float* din, int len, int flag)
         result = "";
     }
 
-
+    if(in){
-    if(in)
         delete in;
+        in = nullptr;
+    }
+    if(fe){
+        delete fe;
+        fe = nullptr;
+    }
 
     return result;
 }

funasr/runtime/onnxruntime/src/paraformer_onnx.h

@@ -8,7 +8,7 @@ namespace paraformer {
 
     class ModelImp : public Model {
     private:
-        FeatureExtract* fe;
+        //FeatureExtract* fe;
 
         Vocab* vocab;
         vector<float> means_list;
@@ -34,8 +34,6 @@ namespace paraformer {
         vector<string> m_strInputNames, m_strOutputNames;
         vector<const char*> m_szInputNames;
         vector<const char*> m_szOutputNames;
-        //string m_strInputName, m_strInputNameLen;
-        //string m_strOutputName, m_strOutputNameLen;
 
     public:
         ModelImp(const char* path, int nNumThread=0, bool quantize=false);

funasr/runtime/onnxruntime/tester/tester.cpp

@@ -50,8 +50,7 @@ int main(int argc, char *argv[])
     {
         string msg = RapidAsrGetResult(Result, 0);
         setbuf(stdout, NULL);
-        cout << "Result: \"";
+        printf("Result: %s \n", msg.c_str());
-        cout << msg << "\"." << endl;
         snippet_time = RapidAsrGetRetSnippetTime(Result);
         RapidAsrFreeResult(Result);
     }
@@ -60,45 +59,6 @@ int main(int argc, char *argv[])
         cout <<"no return data!";
     }
  
-    //char* buff = nullptr;
-    //int len = 0;
-    //ifstream ifs(argv[2], std::ios::binary | std::ios::in);
-    //if (ifs.is_open())
-    //{
-    //    ifs.seekg(0, std::ios::end);
-    //    len = ifs.tellg();
-    //    ifs.seekg(0, std::ios::beg);
-
-    //    buff = new char[len];
-
-    //    ifs.read(buff, len);
-
-
-    //    //RPASR_RESULT Result = RapidAsrRecogPCMFile(AsrHanlde, argv[2], RASR_NONE, NULL);
-
-    //    RPASR_RESULT Result=RapidAsrRecogPCMBuffer(AsrHanlde, buff,len, RASR_NONE, NULL);
-    //    //RPASR_RESULT Result = RapidAsrRecogPCMFile(AsrHanlde, argv[2], RASR_NONE, NULL);
-    //    gettimeofday(&end, NULL);
-    //   
-    //    if (Result)
-    //    {
-    //        string msg = RapidAsrGetResult(Result, 0);
-    //        setbuf(stdout, NULL);
-    //        cout << "Result: \"";
-    //        cout << msg << endl;
-    //        cout << "\"." << endl;
-    //        snippet_time = RapidAsrGetRetSnippetTime(Result);
-    //        RapidAsrFreeResult(Result);
-    //    }
-    //    else
-    //    {
-    //        cout <<"no return data!";
-    //    }
-  
-    //   
-    //delete[]buff;
-    //}
- 
     printf("Audio length %lfs.\n", (double)snippet_time);
     seconds = (end.tv_sec - start.tv_sec);
     long taking_micros = ((seconds * 1000000) + end.tv_usec) - (start.tv_usec);

funasr/runtime/onnxruntime/tester/tester_rtf.cpp

@@ -11,8 +11,25 @@
 #include <fstream>
 #include <sstream>
 #include <vector>
+#include <thread>
 using namespace std;
 
+void runReg(vector<string> wav_list, RPASR_HANDLE AsrHanlde)
+{
+    for (size_t i = 0; i < wav_list.size(); i++)
+    {
+        RPASR_RESULT Result=RapidAsrRecogFile(AsrHanlde, wav_list[i].c_str(), RASR_NONE, NULL);
+
+        if(Result){
+            string msg = RapidAsrGetResult(Result, 0);
+            printf("Result: %s \n", msg.c_str());
+            RapidAsrFreeResult(Result);
+        }else{
+            cout <<"No return data!";
+        }
+    }
+}
+
 int main(int argc, char *argv[])
 {
 
@@ -53,46 +70,14 @@ int main(int argc, char *argv[])
         printf("Cannot load ASR Model from: %s, there must be files model.onnx and vocab.txt", argv[1]);
         exit(-1);
     }
-    gettimeofday(&end, NULL);
-    long seconds = (end.tv_sec - start.tv_sec);
-    long modle_init_micros = ((seconds * 1000000) + end.tv_usec) - (start.tv_usec);
-    printf("Model initialization takes %lfs.\n", (double)modle_init_micros / 1000000);
-
-    // warm up
-    for (size_t i = 0; i < 30; i++)
-    {
-        RPASR_RESULT Result=RapidAsrRecogFile(AsrHanlde, wav_list[0].c_str(), RASR_NONE, NULL);
-    }
-
-    // forward
-    float snippet_time = 0.0f;
-    float total_length = 0.0f;
-    long total_time = 0.0f;
     
-    for (size_t i = 0; i < wav_list.size(); i++)
+    std::thread t1(runReg, wav_list, AsrHanlde);
-    {
+    std::thread t2(runReg, wav_list, AsrHanlde);
-        gettimeofday(&start, NULL);
-        RPASR_RESULT Result=RapidAsrRecogFile(AsrHanlde, wav_list[i].c_str(), RASR_NONE, NULL);
-        gettimeofday(&end, NULL);
-        seconds = (end.tv_sec - start.tv_sec);
-        long taking_micros = ((seconds * 1000000) + end.tv_usec) - (start.tv_usec);
-        total_time += taking_micros;
 
-        if(Result){
+    t1.join();
-            string msg = RapidAsrGetResult(Result, 0);
+    t2.join();
-            printf("Result: %s \n", msg);
 
-            snippet_time = RapidAsrGetRetSnippetTime(Result);
+    //runReg(wav_list, AsrHanlde);
-            total_length += snippet_time;
-            RapidAsrFreeResult(Result);
-        }else{
-            cout <<"No return data!";
-        }
-    }
-
-    printf("total_time_wav %ld ms.\n", (long)(total_length * 1000));
-    printf("total_time_comput %ld ms.\n", total_time / 1000);
-    printf("total_rtf %05lf .\n", (double)total_time/ (total_length*1000000));
 
     RapidAsrUninit(AsrHanlde);
     return 0;

funasr/runtime/python/grpc/grpc_main_client.py

@@ -0,0 +1,62 @@
+import grpc
+import json
+import time
+import asyncio
+import soundfile as sf
+import argparse
+
+from grpc_client import transcribe_audio_bytes
+from paraformer_pb2_grpc import ASRStub
+
+# send the audio data once
+async def grpc_rec(wav_scp, grpc_uri, asr_user, language):
+    with grpc.insecure_channel(grpc_uri) as channel:
+        stub = ASRStub(channel)
+        for line in wav_scp:
+            wav_file = line.split()[1]
+            wav, _ = sf.read(wav_file, dtype='int16')
+            
+            b = time.time()
+            response = transcribe_audio_bytes(stub, wav.tobytes(), user=asr_user, language=language, speaking=False, isEnd=False)
+            resp = response.next()
+            text = ''
+            if 'decoding' == resp.action:
+                resp = response.next()
+                if 'finish' == resp.action:
+                    text = json.loads(resp.sentence)['text']
+            response = transcribe_audio_bytes(stub, None, user=asr_user, language=language, speaking=False, isEnd=True)
+            res= {'text': text, 'time': time.time() - b}
+            print(res)
+
+async def test(args):
+    wav_scp = open(args.wav_scp, "r").readlines()
+    uri = '{}:{}'.format(args.host, args.port)
+    res = await grpc_rec(wav_scp, uri, args.user_allowed, language = 'zh-CN')
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--host",
+                        type=str,
+                        default="127.0.0.1",
+                        required=False,
+                        help="grpc server host ip")
+    parser.add_argument("--port",
+                        type=int,
+                        default=10108,
+                        required=False,
+                        help="grpc server port")              
+    parser.add_argument("--user_allowed",
+                        type=str,
+                        default="project1_user1",
+                        help="allowed user for grpc client")
+    parser.add_argument("--sample_rate",
+                        type=int,
+                        default=16000,
+                        help="audio sample_rate from client") 
+    parser.add_argument("--wav_scp",
+                        type=str,
+                        required=True,
+                        help="audio wav scp")                    
+    args = parser.parse_args()
+    
+    asyncio.run(test(args))