Add lock to avoid race condition and add more log messages

Change-Id: I52f7022d9e9ec2df82b09f45e32a7fd8b63aa041

diff --git a/plugins/wakeup-manager/inc/wakeup_engine_manager.h b/plugins/wakeup-manager/inc/wakeup_engine_manager.h
index 68993cfd8e44b5bb2d82c0a8a72de22864fc2d92..4f659ed052c36d5b975b41569d5ec044c055e1aa 100644 (file)
--- a/plugins/wakeup-manager/inc/wakeup_engine_manager.h
+++ b/plugins/wakeup-manager/inc/wakeup_engine_manager.h
@@ -213,6 +213,14 @@ private:
        wakeup_manager_state_e mWakeupManagerState{WAKEUP_MANAGER_STATE_INACTIVE};
 
        bool mWakeWordAudioRequired{false};
+
+       typedef struct {
+               atomic_llong last_audio_fetched{0};
+               atomic_llong last_count_fetched{0};
+               atomic_llong last_audio_sent{0};
+       } StreamingHistory;
+
+       StreamingHistory mStreamingHistory;
 };
 
 } // wakeup

diff --git a/plugins/wakeup-manager/inc/wakeup_manager.h b/plugins/wakeup-manager/inc/wakeup_manager.h
index 657c3ec5f8e3237037766b726cc4445df261af1d..21f8ccdf030e9ea13061115923000c020dbbba73 100644 (file)
--- a/plugins/wakeup-manager/inc/wakeup_manager.h
+++ b/plugins/wakeup-manager/inc/wakeup_manager.h
@@ -28,6 +28,7 @@
 
 #include <memory>
 #include <map>
+#include <mutex>
 
 namespace multiassistant
 {
@@ -241,6 +242,8 @@ private:
        wakeup_manager_state_e mWakeupManagerState{WAKEUP_MANAGER_STATE_INACTIVE};
 
        mas_wakeup_event_info mLastWakeupEventInfo;
+
+       mutex mMutex;
 };
 
 } // wakeup

diff --git a/plugins/wakeup-manager/src/wakeup_engine_manager.cpp b/plugins/wakeup-manager/src/wakeup_engine_manager.cpp
index 2d439d0a4223df7ce15c74b99830ed8de086d992..307ff3c8ec325902cbf54d26d1b866a321301ae5 100644 (file)
--- a/plugins/wakeup-manager/src/wakeup_engine_manager.cpp
+++ b/plugins/wakeup-manager/src/wakeup_engine_manager.cpp
@@ -36,6 +36,16 @@ static auto contains(const C& v, const T& x) -> decltype(end(v), true)
        return end(v) != find(begin(v), end(v), x);
 }
 
+static long long get_current_milliseconds_after_epoch()
+{
+       auto now = chrono::steady_clock::now();
+       auto now_ms = chrono::time_point_cast<chrono::milliseconds>(now);
+       /* number of milliseconds since the epoch of system_clock */
+       auto value = now_ms.time_since_epoch();
+
+       return value.count();
+}
+
 CWakeupEngineManager::CWakeupEngineManager()
 {
 }
@@ -380,6 +390,8 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                /* get feedback data */
                if (interface && interface->get_utterance_data) {
                        ret = interface->get_utterance_data(index, &speech_data);
+                       mStreamingHistory.last_audio_fetched.store(get_current_milliseconds_after_epoch());
+
                        if (0 != ret) {
                                /* empty queue */
                                MWR_LOGD("[DEBUG] No feedback data. Waiting mode : %d", ret);
@@ -392,6 +404,7 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                                                MWR_LOGD("[INFO] Resume thread");
                                                break;
                                        }
+                                       mStreamingHistory.last_count_fetched.store(get_current_milliseconds_after_epoch());
                                        if (g_speech_pcm_wait_count < cnt) {
                                                unsigned char final_buffer[2] = {'\0', };
                                                MWR_LOGE("[ERROR] Wrong request, there's no pcm data");
@@ -413,6 +426,7 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                                                                        MAS_SPEECH_STREAMING_EVENT_FINISH, final_buffer, sizeof(final_buffer))) {
                                                                        LOGE("[Recorder WARNING] One of the observer returned false");
                                                                }
+                                                               mStreamingHistory.last_audio_sent.store(get_current_milliseconds_after_epoch());
                                                        }
                                                }
                                                return;
@@ -443,8 +457,8 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                        const int max_burst_count = 3;
                        if (++burst_count >= max_burst_count) {
                                burst_count = 0;
-                               this_thread::sleep_for(chrono::milliseconds(sleep_duration_in_millis));
                                MWR_LOGI("[INFO] Streaming data burst transmission detected, forcing sleep");
+                               this_thread::sleep_for(chrono::milliseconds(sleep_duration_in_millis));
                        }
                        for (const auto& observer : mObservers) {
                                if (observer) {
@@ -452,6 +466,7 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                                                speech_data.event, speech_data.buffer, speech_data.len)) {
                                                LOGE("[Recorder WARNING] One of the observer returned false");
                                        }
+                                       mStreamingHistory.last_audio_sent.store(get_current_milliseconds_after_epoch());
                                }
                        }
 
@@ -464,6 +479,7 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                        index++;
                }
        }
+       MWR_LOGE("[INFO] Streaming loop exit");
 
        if (true != finish_event_sent) {
                unsigned char final_buffer[2] = {'\0', };
@@ -474,6 +490,7 @@ void CWakeupEngineManager::streaming_speech_data_thread_func()
                                        MAS_SPEECH_STREAMING_EVENT_FINISH, final_buffer, sizeof(final_buffer))) {
                                        LOGE("[Recorder WARNING] One of the observer returned false");
                                }
+                               mStreamingHistory.last_audio_sent.store(get_current_milliseconds_after_epoch());
                        }
                }
 #ifdef BUF_SAVE_MODE
@@ -503,7 +520,11 @@ void CWakeupEngineManager::stop_streaming_current_utterance_data()
 {
        MWR_LOGI("[ENTER]");
        if (mStreamingThread.joinable()) {
-               MWR_LOGD("mStreamingThread is joinable, trying join()");
+               MWR_LOGE("mStreamingThread is joinable, trying join() %lld %lld %lld %lld",
+                       get_current_milliseconds_after_epoch(),
+                       mStreamingHistory.last_audio_fetched.load(),
+                       mStreamingHistory.last_count_fetched.load(),
+                       mStreamingHistory.last_audio_sent.load());
                mStopStreamingThread.store(true);
                mStreamingThread.join();
        }

diff --git a/plugins/wakeup-manager/src/wakeup_manager.cpp b/plugins/wakeup-manager/src/wakeup_manager.cpp
index c38667a7de291a68a5a71bdd106cd4c6bc6d0887..31eb2b9c9bb174cb6ffaedded283c3e22b702ba6 100644 (file)
--- a/plugins/wakeup-manager/src/wakeup_manager.cpp
+++ b/plugins/wakeup-manager/src/wakeup_manager.cpp
@@ -401,6 +401,7 @@ STREAMING_MODE CWakeupManager::get_streaming_mode()
 
 bool CWakeupManager::set_streaming_mode(STREAMING_MODE mode)
 {
+       lock_guard<mutex> lock(mMutex);
        mStreamingMode = mode;
        return true;
 }
@@ -682,6 +683,7 @@ bool CWakeupManager::start_streaming_utterance_data()
 bool CWakeupManager::stop_streaming_utterance_data()
 {
        MWR_LOGI("[ENTER]");
+
        if (STREAMING_MODE::UTTERANCE != mStreamingMode) return false;
 
        mAudioManager.stop_streaming_current_utterance_data();