Added balancing groups and common group receiver

apps/socketoptions.hpp

@@ -250,6 +250,7 @@ const SocketOption srt_options [] {
 #if ENABLE_BONDING
     { "groupconnect", 0, SRTO_GROUPCONNECT, SocketOption::PRE, SocketOption::INT, nullptr},
     { "groupminstabletimeo", 0, SRTO_GROUPMINSTABLETIMEO, SocketOption::PRE, SocketOption::INT, nullptr},
+    { "groupconfig", 0, SRTO_GROUPCONFIG, SocketOption::PRE, SocketOption::STRING, nullptr},
 #endif
 #ifdef SRT_ENABLE_BINDTODEVICE
     { "bindtodevice", 0, SRTO_BINDTODEVICE, SocketOption::PRE, SocketOption::STRING, nullptr},

srtcore/api.cpp

@@ -148,6 +148,17 @@ void srt::CUDTSocket::setBrokenClosed()
 
 bool srt::CUDTSocket::readReady()
 {
+#if ENABLE_BONDING
+
+    // If this is a group member socket, then reading happens exclusively from
+    // the group and the socket is only used as a connection point, packet
+    // dispatching and single link management. Data buffering and hence ability
+    // to deliver a packet through API is exclusively the matter of group,
+    // therefore a single socket is never "read ready".
+
+    if (m_GroupOf)
+        return false;
+#endif
     if (m_UDT.m_bConnected && m_UDT.isRcvBufferReady())
         return true;
 
@@ -716,35 +727,17 @@ int srt::CUDTUnited::newConnection(const SRTSOCKET     listen,
                 g->m_bConnected = true;
             }
 
-            // XXX PROLBEM!!! These events are subscribed here so that this is done once, lazily,
-            // but groupwise connections could be accepted from multiple listeners for the same group!
-            // m_listener MUST BE A CONTAINER, NOT POINTER!!!
-            // ALSO: Maybe checking "the same listener" is not necessary as subscruption may be done
-            // multiple times anyway?
-            if (!g->m_listener)
-            {
-                // Newly created group from the listener, which hasn't yet
-                // the listener set.
-                g->m_listener = ls;
-
-                // Listen on both first connected socket and continued sockets.
-                // This might help with jump-over situations, and in regular continued
-                // sockets the IN event won't be reported anyway.
-                int listener_modes = SRT_EPOLL_ACCEPT | SRT_EPOLL_UPDATE;
-                epoll_add_usock_INTERNAL(g->m_RcvEID, ls, &listener_modes);
-
-                // This listening should be done always when a first connected socket
-                // appears as accepted off the listener. This is for the sake of swait() calls
-                // inside the group receiving and sending functions so that they get
-                // interrupted when a new socket is connected.
-            }
+            // In the new recvbuffer mode (and common receiver buffer) there's no waiting for reception
+            // on a socket and no reading from a socket directly is being done; instead the reading API
+            // is directly bound to the group and reading happens directly from the group's buffer.
+            // This includes also a situation of a newly connected socket, which will be delivering packets
+            // into the same common receiver buffer for the group, so readable will be the group itself
+            // when it has its own common buffer read-ready, by whatever reason. Packets to the buffer
+            // will be delivered by the sockets' receiver threads, so all these things happen strictly
+            // in the background.
 
-            // Add also per-direction subscription for the about-to-be-accepted socket.
-            // Both first accepted socket that makes the group-accept and every next
-            // socket that adds a new link.
-            int read_modes  = SRT_EPOLL_IN | SRT_EPOLL_ERR;
+            // Keep per-socket sender ready EID.
             int write_modes = SRT_EPOLL_OUT | SRT_EPOLL_ERR;
-            epoll_add_usock_INTERNAL(g->m_RcvEID, ns, &read_modes);
             epoll_add_usock_INTERNAL(g->m_SndEID, ns, &write_modes);
 
             // With app reader, do not set groupPacketArrival (block the
@@ -1439,7 +1432,7 @@ int srt::CUDTUnited::groupConnect(CUDTGroup* pg, SRT_SOCKGROUPCONFIG* targets, i
         // Do it after setting all stored options, as some of them may
         // influence some group data.
 
-        srt::groups::SocketData data = srt::groups::prepareSocketData(ns);
+        srt::groups::SocketData data = srt::groups::prepareSocketData(ns, g.type());
         if (targets[tii].token != -1)
         {
             // Reuse the token, if specified by the caller
@@ -1530,7 +1523,6 @@ int srt::CUDTUnited::groupConnect(CUDTGroup* pg, SRT_SOCKGROUPCONFIG* targets, i
         // connection succeeded or failed and whether the new socket is
         // ready to use or needs to be closed.
         epoll_add_usock_INTERNAL(g.m_SndEID, ns, &connect_modes);
-        epoll_add_usock_INTERNAL(g.m_RcvEID, ns, &connect_modes);
 
         // Adding a socket on which we need to block to BOTH these tracking EIDs
         // and the blocker EID. We'll simply remove from them later all sockets that
@@ -1652,7 +1644,6 @@ int srt::CUDTUnited::groupConnect(CUDTGroup* pg, SRT_SOCKGROUPCONFIG* targets, i
                 f->sndstate = SRT_GST_BROKEN;
                 f->rcvstate = SRT_GST_BROKEN;
                 epoll_remove_socket_INTERNAL(g.m_SndEID, ns);
-                epoll_remove_socket_INTERNAL(g.m_RcvEID, ns);
             }
             else
             {
@@ -1738,7 +1729,6 @@ int srt::CUDTUnited::groupConnect(CUDTGroup* pg, SRT_SOCKGROUPCONFIG* targets, i
 
                 epoll_remove_socket_INTERNAL(eid, y->second);
                 epoll_remove_socket_INTERNAL(g.m_SndEID, y->second);
-                epoll_remove_socket_INTERNAL(g.m_RcvEID, y->second);
             }
         }
 
@@ -1778,7 +1768,6 @@ int srt::CUDTUnited::groupConnect(CUDTGroup* pg, SRT_SOCKGROUPCONFIG* targets, i
 
                 epoll_remove_socket_INTERNAL(eid, s);
                 epoll_remove_socket_INTERNAL(g.m_SndEID, s);
-                epoll_remove_socket_INTERNAL(g.m_RcvEID, s);
 
                 continue;
             }
@@ -2287,6 +2276,8 @@ int srt::CUDTUnited::select(UDT::UDSET* readfds, UDT::UDSET* writefds, UDT::UDSE
     return count;
 }
 
+// XXX This may crash when a member socket is added to selectEx.
+// Consider revising to prevent a member socket from being used.
 int srt::CUDTUnited::selectEx(const vector<SRTSOCKET>& fds,
                               vector<SRTSOCKET>*       readfds,
                               vector<SRTSOCKET>*       writefds,
@@ -2622,6 +2613,12 @@ void srt::CUDTUnited::checkBrokenSockets()
                 continue;
         }
         else
+
+        // Additional note on group receiver: with the new group
+        // receiver m_pRcvBuffer in the socket core is NULL always,
+        // but that's not a problem - you can close the member socket
+        // safely without worrying about reading data because they are
+        // in the group anyway.
         {
             CUDT& u = s->core();
 
@@ -3186,7 +3183,7 @@ void srt::CUDTUnited::updateMux(CUDTSocket* s, const sockaddr_any& reqaddr, cons
         m.m_pSndQueue = new CSndQueue;
         m.m_pSndQueue->init(m.m_pChannel, m.m_pTimer);
         m.m_pRcvQueue = new CRcvQueue;
-        m.m_pRcvQueue->init(128, s->core().maxPayloadSize(), m.m_iIPversion, 1024, m.m_pChannel, m.m_pTimer);
+        m.m_pRcvQueue->init(128, s->core().maxPayloadSize(), m.m_iIPversion, 1024, m.m_pChannel, m.m_pTimer, s->m_SocketID);
 
         // Rewrite the port here, as it might be only known upon return
         // from CChannel::open.

srtcore/api.h

@@ -125,7 +125,8 @@ class CUDTSocket
 
     // XXX Controversial as to whether it should be guarded by this lock.
     // It is used in many places without the lock, and it is also atomic.
-    SRT_ATTR_GUARDED_BY(m_ControlLock)
+    // (blocked because it makes more ado than actual help)
+    //SRT_ATTR_GUARDED_BY(m_ControlLock)
     sync::atomic<SRT_SOCKSTATUS> m_Status; //< current socket state
 
     /// Time when the socket is closed.
@@ -486,6 +487,8 @@ class CUDTUnited
 #endif
 
     void checkBrokenSockets();
+
+    SRT_ATTR_REQUIRES(m_GlobControlLock)
     void removeSocket(const SRTSOCKET u);
 
     CEPoll m_EPoll; // handling epoll data structures and events

srtcore/buffer_rcv.cpp

@@ -99,10 +99,9 @@ namespace {
  *    m_iMaxPosOff:     none? (modified on add and ack
  */
 
-CRcvBuffer::CRcvBuffer(int initSeqNo, size_t size, CUnitQueue* unitqueue, bool bMessageAPI)
+CRcvBuffer::CRcvBuffer(int initSeqNo, size_t size, /*CUnitQueue* unitqueue, */ bool bMessageAPI)
     : m_entries(size)
     , m_szSize(size) // TODO: maybe just use m_entries.size()
-    , m_pUnitQueue(unitqueue)
     , m_iStartSeqNo(initSeqNo) // NOTE: SRT_SEQNO_NONE is allowed here.
     , m_iStartPos(0)
     , m_iEndOff(0)
@@ -129,7 +128,7 @@ CRcvBuffer::~CRcvBuffer()
         if (!it->pUnit)
             continue;
 
-        m_pUnitQueue->makeUnitFree(it->pUnit);
+        it->pUnit->m_pParentQueue->makeUnitFree(it->pUnit);
         it->pUnit = NULL;
     }
 }
@@ -201,7 +200,8 @@ CRcvBuffer::InsertInfo CRcvBuffer::insert(CUnit* unit)
     }
     SRT_ASSERT(m_entries[newpktpos].pUnit == NULL);
 
-    m_pUnitQueue->makeUnitTaken(unit);
+    CUnitQueue* q = unit->m_pParentQueue;
+    q->makeUnitTaken(unit);
     m_entries[newpktpos].pUnit  = unit;
     m_entries[newpktpos].status = EntryState_Avail;
     countBytes(1, (int)unit->m_Packet.getLength());
@@ -1133,7 +1133,7 @@ void CRcvBuffer::releaseUnitInPos(CPos pos)
     CUnit* tmp = m_entries[pos].pUnit;
     m_entries[pos] = Entry(); // pUnit = NULL; status = Empty
     if (tmp != NULL)
-        m_pUnitQueue->makeUnitFree(tmp);
+        tmp->m_pParentQueue->makeUnitFree(tmp);
 }
 
 bool CRcvBuffer::dropUnitInPos(CPos pos)
@@ -1631,5 +1631,29 @@ int32_t CRcvBuffer::getFirstLossSeq(int32_t fromseq, int32_t* pw_end)
     return ret_seq;
 }
 
+void CRcvBuffer::getUnitSeriesInfo(int32_t fromseq, size_t maxsize, std::vector<SRTSOCKET>& w_sources)
+{
+    const int offset = CSeqNo(fromseq) - m_iStartSeqNo;
+
+    // Check if it's still inside the buffer
+    if (offset < 0 || offset >= m_iMaxPosOff)
+        return;
+
+    // All you need to do is to check if there's a valid packet
+    // at given position
+    size_t pass = 0;
+    for (int off = offset; off < m_iMaxPosOff; ++off)
+    {
+        int pos = incPos(m_iStartPos, off);
+        if (m_entries[pos].pUnit)
+        {
+            w_sources.push_back(m_entries[pos].pUnit->m_pParentQueue->ownerID());
+            ++pass;
+            if (pass == maxsize)
+                break;
+        }
+    }
+}
+
 
 } // namespace srt

srtcore/buffer_rcv.h

@@ -406,7 +406,7 @@ class CRcvBuffer
     typedef sync::steady_clock::duration   duration;
 
 public:
-    CRcvBuffer(int initSeqNo, size_t size, CUnitQueue* unitqueue, bool bMessageAPI);
+    CRcvBuffer(int initSeqNo, size_t size, /*CUnitQueue* unitqueue, */ bool bMessageAPI);
 
     ~CRcvBuffer();
 
@@ -619,6 +619,7 @@ class CRcvBuffer
     std::pair<int, int> getAvailablePacketsRange() const;
 
     int32_t getFirstLossSeq(int32_t fromseq, int32_t* opt_end = NULL);
+    void getUnitSeriesInfo(int32_t fromseq, size_t maxsize, std::vector<SRTSOCKET>& w_sources);
 
     bool empty() const
     {
@@ -826,7 +827,11 @@ class CRcvBuffer
     entries_t m_entries;
 
     const size_t m_szSize;     // size of the array of units (buffer)
-    CUnitQueue*  m_pUnitQueue; // the shared unit queue
+
+    //XXX removed. In this buffer the units may come from various different
+    // queues, and the unit has a pointer pointing to the queue from which
+    // it comes, and it should be returned to the same queue.
+    //CUnitQueue*  m_pUnitQueue; // the shared unit queue
 
     CSeqNo m_iStartSeqNo;
     CPos m_iStartPos;        // the head position for I/O (inclusive)

srtcore/buffer_snd.cpp

@@ -550,12 +550,20 @@ void CSndBuffer::ackData(int offset)
 {
     ScopedLock bufferguard(m_BufLock);
 
+    // It is illegal to call this function without having first checked
+    // that the offset is within the range between 0 and the current count.
+    SRT_ASSERT(offset <= m_iCount);
+
     bool move = false;
     for (int i = 0; i < offset; ++i)
     {
         m_iBytesCount -= m_pFirstBlock->m_iLength;
         if (m_pFirstBlock == m_pCurrBlock)
             move = true;
+
+        // Sanity check to see if signing off for removal didn't
+        // exceed the last position of the used space.
+        SRT_ASSERT(m_pFirstBlock != m_pLastBlock);
         m_pFirstBlock = m_pFirstBlock->m_pNext;
     }
     if (move)
@@ -566,6 +574,18 @@ void CSndBuffer::ackData(int offset)
     updAvgBufSize(steady_clock::now());
 }
 
+void CSndBuffer::clear()
+{
+    ScopedLock bufferguard(m_BufLock);
+
+    // Keep the m_pLastBlock intact and adjust the other
+    // fields to it. Blocks are still linked in circle.
+
+    m_pCurrBlock = m_pFirstBlock = m_pLastBlock;
+    m_iCount = 0;
+    m_iBytesCount = 0;
+}
+
 int CSndBuffer::getCurrBufSize() const
 {
     return m_iCount;

srtcore/buffer_snd.h

@@ -171,6 +171,8 @@ class CSndBuffer
 
     void ackData(int offset);
 
+    void clear();
+
     /// Read size of data still in the sending list.
     /// @return Current size of the data in the sending list.
     int getCurrBufSize() const;
@@ -241,12 +243,30 @@ class CSndBuffer
             return m_iMsgNoBitset & MSGNO_SEQ::mask;
         }
 
-    } * m_pBlock, *m_pFirstBlock, *m_pCurrBlock, *m_pLastBlock;
-
-    // m_pBlock:         The head pointer
-    // m_pFirstBlock:    The first block
-    // m_pCurrBlock:	 The current block
-    // m_pLastBlock:     The last block (if first == last, buffer is empty)
+    }
+    /// The head pointer. Only used in constructor to roll over the initial set of blocks.
+        * m_pBlock,
+    /// The edge of the buffer at the reading end, pointing to the oldest ever stored
+    /// block that can still be read (although it's not readable if it is equal to m_pLastBlock).
+        * m_pFirstBlock,
+    /// The pointer to the next block to be read when extracting packets to send over the
+    /// network as "unique" (first time sent). Not used when extracting a packet for
+    /// retransmission, although this field must be updated every time any block removal
+    /// happens in case it would become stale.
+        * m_pCurrBlock,
+    /// Points to the block considered past-the-end of the used block space, simultaneously
+    /// the edge of the buffer at the writing end. The distance between m_pFirstBlock and
+    /// m_pLastBlock in the order of linked objects through the m_pNext field should be
+    /// always equal to m_iCount.
+        * m_pLastBlock;
+
+    // NOTE: The actual reserved space size for the buffer is saved in m_iSize and this is
+    // the true number of allocated blocks. There is no pointer to point directly to the end
+    // of reserved space - this can only be determined by the series of buffers that are
+    // linked to one another through the m_pNext field, which is NULL in the last one. Note
+    // also that Blocks are linked in circle, while Buffers are linked up to the last one,
+    // and Buffers are the true holders of the pieces of memory, while a block operates on
+    // its fragment only, which's size is defined by m_iBlockLen.
 
     struct Buffer
     {