Fix race conditions by minimizing lock usages.

Author: jianjunz

source/agent/addons/quic/QuicTransportConnection.cc

@@ -93,11 +93,14 @@ NAUV_WORK_CB(QuicTransportConnection::onStreamCallback)
     if (obj == nullptr || obj->m_streamsToBeNotified.empty()) {
         return;
     }
-    std::lock_guard<std::mutex> lock(obj->m_streamQueueMutex);
     while (!obj->m_streamsToBeNotified.empty()) {
-        v8::Local<v8::Object> streamObject = QuicTransportStream::newInstance(obj->m_streamsToBeNotified.front());
+        obj->m_streamQueueMutex.lock();
+        auto quicStream=obj->m_streamsToBeNotified.front();
+        obj->m_streamsToBeNotified.pop();
+        obj->m_streamQueueMutex.unlock();
+        v8::Local<v8::Object> streamObject = QuicTransportStream::newInstance(quicStream);
         QuicTransportStream* stream = Nan::ObjectWrap::Unwrap<QuicTransportStream>(streamObject);
-        obj->m_streamsToBeNotified.front()->SetVisitor(stream);
+        quicStream->SetVisitor(stream);
         Nan::MaybeLocal<v8::Value> onEvent = Nan::Get(obj->handle(), Nan::New<v8::String>("onincomingstream").ToLocalChecked());
         if (!onEvent.IsEmpty()) {
             v8::Local<v8::Value> onEventLocal = onEvent.ToLocalChecked();
@@ -110,7 +113,6 @@ NAUV_WORK_CB(QuicTransportConnection::onStreamCallback)
         } else {
             ELOG_DEBUG("onEvent is empty");
         }
-        obj->m_streamsToBeNotified.pop();
     }
 }
 

source/agent/addons/quic/QuicTransportServer.h

@@ -13,7 +13,6 @@
 #include <mutex>
 #include <nan.h>
 #include <string>
-#include <thread>
 #include <unordered_map>
 
 class QuicTransportServer : public Nan::ObjectWrap, owt::quic::QuicTransportServerInterface::Visitor, QuicTransportConnection::Visitor {

source/agent/addons/quic/QuicTransportStream.cc

@@ -97,8 +97,8 @@ NAN_METHOD(QuicTransportStream::write)
     QuicTransportStream* obj = Nan::ObjectWrap::Unwrap<QuicTransportStream>(info.Holder());
     uint8_t* buffer = (uint8_t*)node::Buffer::Data(info[0]->ToObject());
     unsigned int length = info[1]->Uint32Value();
-    obj->m_stream->Write(buffer, length);
-    info.GetReturnValue().Set(info.This());
+    auto written = obj->m_stream->Write(buffer, length);
+    info.GetReturnValue().Set(Nan::New(static_cast<int>(written)));
 }
 
 NAN_METHOD(QuicTransportStream::addDestination)
@@ -131,7 +131,7 @@ v8::Local<v8::Object> QuicTransportStream::newInstance(owt::quic::QuicTransportS
 
 void QuicTransportStream::MaybeReadContentSessionId()
 {
-    if (!m_receivedContentSessionId) {
+    if (!m_receivedContentSessionId && m_stream->ReadableBytes() > 0) {
         // Match to a content session.
         if (m_stream->ReadableBytes() > 0 && m_stream->ReadableBytes() < uuidSizeInBytes) {
             ELOG_ERROR("No enough data to get content session ID.");

source/agent/quic/index.js

@@ -39,6 +39,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         new Map();  // Key is publication ID, value is stream pipeline.
     const outgoingStreamPipelines =
         new Map();  // Key is subscription ID, value is stream pipeline.
+    let quicTransportServer;
 
     const notifyStatus = (controller, sessionId, direction, status) => {
         rpcClient.remoteCast(controller, 'onSessionProgress', [sessionId, direction, status]);
@@ -53,7 +54,7 @@ module.exports = function (rpcClient, selfRpcId, parentRpcId, clusterWorkerIP) {
         return;
       }
       log.info('path is '+path.resolve(global.config.quic.keystorePath));
-      const quicTransportServer = new QuicTransportServer(
+      quicTransportServer = new QuicTransportServer(
           addon, global.config.quic.port, path.resolve(global.config.quic.keystorePath),
           password);
       quicTransportServer.start();