[core] Fix a data race by adding a shared mutex

srtcore/api.cpp

@@ -541,6 +541,7 @@ int srt::CUDTUnited::newConnection(const SRTSOCKET     listen,
 
     try
     {
+        ScopedLock col(ls->core().m_ConnectionLock);
         ns = new CUDTSocket(*ls);
         // No need to check the peer, this is the address from which the request has come.
         ns->m_PeerAddr = peer;

srtcore/queue.cpp

@@ -1130,7 +1130,7 @@ srt::CRcvQueue::CRcvQueue()
     , m_szPayloadSize()
     , m_bClosing(false)
     , m_LSLock()
-    , m_pListener(NULL)
+    , m_pListener() 
     , m_pRendezvousQueue(NULL)
     , m_vNewEntry()
     , m_IDLock()
@@ -1405,10 +1405,11 @@ srt::EConnectStatus srt::CRcvQueue::worker_ProcessConnectionRequest(CUnit* unit,
     bool have_listener = false;
     {
         ScopedLock cg(m_LSLock);
-        if (m_pListener)
+        m_pListener.lockRead();
+        if (m_pListener.udt)
         {
-            LOGC(cnlog.Debug, log << "PASSING request from: " << addr.str() << " to listener:" << m_pListener->socketID());
-            listener_ret = m_pListener->processConnectRequest(addr, unit->m_Packet);
+            LOGC(cnlog.Debug, log << "PASSING request from: " << addr.str() << " to listener:" << m_pListener.udt->socketID());
+            listener_ret = m_pListener.udt->processConnectRequest(addr, unit->m_Packet);
 
             // This function does return a code, but it's hard to say as to whether
             // anything can be done about it. In case when it's stated possible, the
@@ -1418,6 +1419,7 @@ srt::EConnectStatus srt::CRcvQueue::worker_ProcessConnectionRequest(CUnit* unit,
 
             have_listener = true;
         }
+        m_pListener.unlockRead();
     }
 
     // NOTE: Rendezvous sockets do bind(), but not listen(). It means that the socket is
@@ -1690,21 +1692,29 @@ int srt::CRcvQueue::recvfrom(int32_t id, CPacket& w_packet)
 
 int srt::CRcvQueue::setListener(CUDT* u)
 {
-    ScopedLock lslock(m_LSLock);
-
-    if (NULL != m_pListener)
+    m_pListener.lockWrite();
+    if (NULL != m_pListener.udt)
+    {
+        m_pListener.unlockWrite();
         return -1;
+    }
+
+    m_pListener.udt = u;
+    m_pListener.unlockWrite();
 
-    m_pListener = u;
     return 0;
 }
 
 void srt::CRcvQueue::removeListener(const CUDT* u)
 {
     ScopedLock lslock(m_LSLock);
-
-    if (u == m_pListener)
-        m_pListener = NULL;
+    m_pListener.lockWrite();
+    if (u == m_pListener.udt)
+    {
+        m_pListener.udt = NULL;
+        m_pListener.unlockWrite();
+    }
+    m_pListener.unlockWrite();
 }
 
 void srt::CRcvQueue::registerConnector(const SRTSOCKET&                id,

srtcore/queue.h

@@ -67,6 +67,36 @@ namespace srt
 {
 class CChannel;
 class CUDT;
+class CUDTWrapper;
+
+class CUDTWrapper {
+public:
+    CUDT *udt;
+    sync::SharedMutex mut;
+
+public:
+    CUDTWrapper()
+    :udt(NULL)
+    ,mut()
+    {
+}
+void lockRead()
+{
+    return mut.lockRead();
+}
+void lockWrite()
+{
+    return mut.lockWrite();
+}
+void unlockRead()
+{
+    return mut.unlockRead();
+
+}
+void unlockWrite(){
+    return mut.unlockWrite();
+}
+};
 
 struct CUnit
 {
@@ -555,7 +585,7 @@ class CRcvQueue
 
 private:
     sync::Mutex       m_LSLock;
-    CUDT*             m_pListener;        // pointer to the (unique, if any) listening UDT entity
+    CUDTWrapper       m_pListener;        // pointer to the (unique, if any) listening UDT entity
     CRendezvousQueue* m_pRendezvousQueue; // The list of sockets in rendezvous mode
 
     std::vector<CUDT*> m_vNewEntry; // newly added entries, to be inserted

srtcore/sync.h

@@ -943,9 +943,74 @@ CUDTException& GetThreadLocalError();
 /// @param[in] maxVal maximum allowed value of the resulting random number.
 int genRandomInt(int minVal, int maxVal);
 
+class SharedMutex
+{
+    private:
+    Condition  m_pLockWriteCond;
+    Condition  m_pLockReadCond;
+
+    Mutex m_pMutex;
+
+    int m_pCountRead;
+    bool m_pWriterLocked;
+
+
+    public:
+    SharedMutex()
+    :m_pLockWriteCond()
+    ,m_pLockReadCond()
+    ,m_pMutex()
+    ,m_pCountRead(0)
+    ,m_pWriterLocked(false)
+    {
+        m_pCountRead = 0;
+        m_pWriterLocked = false;
+
+    }
+
+    void lockWrite()
+    {
+        UniqueLock l1(m_pMutex);
+        if(m_pWriterLocked)
+            m_pLockWriteCond.wait(l1);
+        m_pWriterLocked = true;
+        if(m_pCountRead)
+            m_pLockReadCond.wait(l1);
+    }
+
+    void unlockWrite()
+    {
+        UniqueLock l2(m_pMutex);
+        m_pWriterLocked = false;
+        m_pLockWriteCond.notify_all();
+    }
+
+    void lockRead()
+    {
+        UniqueLock l3(m_pMutex);
+        if(m_pWriterLocked)
+            m_pLockWriteCond.wait(l3);
+
+        m_pCountRead++;
+    }
+
+    void unlockRead()
+    {
+        UniqueLock l4(m_pMutex);
+        m_pCountRead--;
+        if(m_pWriterLocked)
+            m_pLockReadCond.notify_one();
+        else if (m_pCountRead > 0)
+            m_pLockWriteCond.notify_one();
+
+    }
+
+};
+
 } // namespace sync
 } // namespace srt
 
+
 #include "atomic_clock.h"
 
 #endif // INC_SRT_SYNC_H