[FR] Add the scheduled sending mode for live mode sending

apps/socketoptions.hpp

@@ -254,7 +254,8 @@ const SocketOption srt_options [] {
     { "bindtodevice", 0, SRTO_BINDTODEVICE, SocketOption::PRE, SocketOption::STRING, nullptr},
     { "retransmitalgo", 0, SRTO_RETRANSMITALGO, SocketOption::PRE, SocketOption::INT, nullptr },
     { "cryptomode", 0, SRTO_CRYPTOMODE, SocketOption::PRE, SocketOption::INT, nullptr },
-    { "maxrexmitbw", 0, SRTO_MAXREXMITBW, SocketOption::POST, SocketOption::INT64, nullptr }
+    { "maxrexmitbw", 0, SRTO_MAXREXMITBW, SocketOption::POST, SocketOption::INT64, nullptr },
+    { "sendmode", 0, SRTO_SENDMODE, SocketOption::PRE, SocketOption::INT, nullptr }
 };
 }
 

srtcore/api.cpp

@@ -2261,7 +2261,8 @@ SRTSTATUS CUDTUnited::close(const SRTSOCKET u, int reason)
     };
 #endif
 
-    SocketKeeper k(*this, u, ERH_THROW);
+    SocketKeeper k = SOCKET_KEEP(u, ERH_THROW);
+
     IF_HEAVY_LOGGING(ScopedExitLog slog(k.socket));
     HLOGC(smlog.Debug, log << "CUDTUnited::close/begin: @" << u << " busy=" << k.socket->isStillBusy());
 
@@ -2394,7 +2395,7 @@ void CUDTSocket::breakNonAcceptedSockets()
         HLOGC(smlog.Debug, log << "breakNonAcceptedSockets: found " << accepted.size() << " leaky accepted sockets");
         for (vector<SRTSOCKET>::iterator i = accepted.begin(); i != accepted.end(); ++i)
         {
-            CUDTUnited::SocketKeeper sk(m_UDT.uglobal(), *i);
+            SocketKeeper sk = SOCKET_KEEP(*i, ERH_RETURN);
             if (sk.socket)
             {
                 sk.socket->m_UDT.m_bBroken = true;
@@ -4212,7 +4213,7 @@ SRTSTATUS CUDT::getGroupData(SRTSOCKET groupid, SRT_SOCKGROUPDATA* pdata, size_t
         return APIError(MJ_NOTSUP, MN_INVAL, 0);
     }
 
-    CUDTUnited::GroupKeeper k(uglobal(), groupid, CUDTUnited::ERH_RETURN);
+    CUDTUnited::GroupKeeper k(uglobal(), groupid, ERH_RETURN);
     if (!k.group)
     {
         return APIError(MJ_NOTSUP, MN_INVAL, 0);
@@ -4386,7 +4387,7 @@ SRTSOCKET CUDT::connectLinks(SRTSOCKET grp, SRT_SOCKGROUPCONFIG targets[], int a
 
     try
     {
-        CUDTUnited::GroupKeeper k(uglobal(), grp, CUDTUnited::ERH_THROW);
+        CUDTUnited::GroupKeeper k(uglobal(), grp, ERH_THROW);
         return uglobal().groupConnect(k.group, targets, arraysize);
     }
     catch (CUDTException& e)
@@ -4509,13 +4510,13 @@ SRTSTATUS CUDT::getsockopt(SRTSOCKET u, int, SRT_SOCKOPT optname, void* pw_optva
 #if ENABLE_BONDING
         if (CUDT::isgroup(u))
         {
-            CUDTUnited::GroupKeeper k(uglobal(), u, CUDTUnited::ERH_THROW);
+            CUDTUnited::GroupKeeper k(uglobal(), u, ERH_THROW);
             k.group->getOpt(optname, (pw_optval), (*pw_optlen));
             return SRT_STATUS_OK;
         }
 #endif
 
-        CUDT& udt = uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core();
+        CUDT& udt = uglobal().locateSocket(u, ERH_THROW)->core();
         udt.getOpt(optname, (pw_optval), (*pw_optlen));
         return SRT_STATUS_OK;
     }
@@ -4540,13 +4541,13 @@ SRTSTATUS CUDT::setsockopt(SRTSOCKET u, int, SRT_SOCKOPT optname, const void* op
 #if ENABLE_BONDING
         if (CUDT::isgroup(u))
         {
-            CUDTUnited::GroupKeeper k(uglobal(), u, CUDTUnited::ERH_THROW);
+            CUDTUnited::GroupKeeper k(uglobal(), u, ERH_THROW);
             k.group->setOpt(optname, optval, optlen);
             return SRT_STATUS_OK;
         }
 #endif
 
-        CUDT& udt = uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core();
+        CUDT& udt = uglobal().locateSocket(u, ERH_THROW)->core();
         udt.setOpt(optname, optval, optlen);
         return SRT_STATUS_OK;
     }
@@ -4585,12 +4586,12 @@ int CUDT::sendmsg2(SRTSOCKET u, const char* buf, int len, SRT_MSGCTRL& w_m)
 #if ENABLE_BONDING
         if (CUDT::isgroup(u))
         {
-            CUDTUnited::GroupKeeper k(uglobal(), u, CUDTUnited::ERH_THROW);
+            CUDTUnited::GroupKeeper k(uglobal(), u, ERH_THROW);
             return k.group->send(buf, len, (w_m));
         }
 #endif
 
-        return uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core().sendmsg2(buf, len, (w_m));
+        return uglobal().locateSocket(u, ERH_THROW)->core().sendmsg2(buf, len, (w_m));
     }
     catch (const CUDTException& e)
     {
@@ -4629,12 +4630,12 @@ int CUDT::recvmsg2(SRTSOCKET u, char* buf, int len, SRT_MSGCTRL& w_m)
 #if ENABLE_BONDING
         if (CUDT::isgroup(u))
         {
-            CUDTUnited::GroupKeeper k(uglobal(), u, CUDTUnited::ERH_THROW);
+            CUDTUnited::GroupKeeper k(uglobal(), u, ERH_THROW);
             return k.group->recv(buf, len, (w_m));
         }
 #endif
 
-        return uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core().recvmsg2(buf, len, (w_m));
+        return uglobal().locateSocket(u, ERH_THROW)->core().recvmsg2(buf, len, (w_m));
     }
     catch (const CUDTException& e)
     {
@@ -4651,7 +4652,7 @@ int64_t CUDT::sendfile(SRTSOCKET u, fstream& ifs, int64_t& offset, int64_t size,
 {
     try
     {
-        CUDT& udt = uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core();
+        CUDT& udt = uglobal().locateSocket(u, ERH_THROW)->core();
         return udt.sendfile(ifs, offset, size, block);
     }
     catch (const CUDTException& e)
@@ -4673,7 +4674,7 @@ int64_t CUDT::recvfile(SRTSOCKET u, fstream& ofs, int64_t& offset, int64_t size,
 {
     try
     {
-        return uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core().recvfile(ofs, offset, size, block);
+        return uglobal().locateSocket(u, ERH_THROW)->core().recvfile(ofs, offset, size, block);
     }
     catch (const CUDTException& e)
     {
@@ -4978,7 +4979,7 @@ SRTSTATUS CUDT::bstats(SRTSOCKET u, CBytePerfMon* perf, bool clear, bool instant
 
     try
     {
-        CUDT& udt = uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core();
+        CUDT& udt = uglobal().locateSocket(u, ERH_THROW)->core();
         udt.bstats(perf, clear, instantaneous);
         return SRT_STATUS_OK;
     }
@@ -4998,7 +4999,7 @@ SRTSTATUS CUDT::groupsockbstats(SRTSOCKET u, CBytePerfMon* perf, bool clear)
 {
     try
     {
-        CUDTUnited::GroupKeeper k(uglobal(), u, CUDTUnited::ERH_THROW);
+        CUDTUnited::GroupKeeper k(uglobal(), u, ERH_THROW);
         k.group->bstatsSocket(perf, clear);
         return SRT_STATUS_OK;
     }
@@ -5020,7 +5021,7 @@ CUDT* CUDT::getUDTHandle(SRTSOCKET u)
 {
     try
     {
-        return &uglobal().locateSocket(u, CUDTUnited::ERH_THROW)->core();
+        return &uglobal().locateSocket(u, ERH_THROW)->core();
     }
     catch (const CUDTException& e)
     {
@@ -5042,7 +5043,7 @@ SRT_SOCKSTATUS CUDT::getsockstate(SRTSOCKET u)
 #if ENABLE_BONDING
         if (CUDT::isgroup(u))
         {
-            CUDTUnited::GroupKeeper k(uglobal(), u, CUDTUnited::ERH_THROW);
+            CUDTUnited::GroupKeeper k(uglobal(), u, ERH_THROW);
             return k.group->getStatus();
         }
 #endif

srtcore/api.h

@@ -254,12 +254,6 @@ class CUDTUnited
     static const size_t MAX_CLOSE_RECORD_SIZE = 10;
 
 public:
-    enum ErrorHandling
-    {
-        ERH_RETURN,
-        ERH_THROW,
-        ERH_ABORT
-    };
     static std::string CONID(SRTSOCKET sock);
 
     /// initialize the UDT library.
@@ -545,48 +539,6 @@ class CUDTUnited
 
     void closeAllSockets();
 
-public:
-    struct SocketKeeper
-    {
-        CUDTSocket* socket;
-
-        SocketKeeper(): socket(NULL) {}
-
-        // This is intended for API functions to lock the socket's existence
-        // for the lifetime of their call.
-        SocketKeeper(CUDTUnited& glob, SRTSOCKET id, ErrorHandling erh = ERH_RETURN) { socket = glob.locateAcquireSocket(id, erh); }
-
-        // This is intended for TSBPD thread that should lock the socket's
-        // existence until it exits.
-        SocketKeeper(CUDTUnited& glob, CUDTSocket* s)
-        {
-            acquire(glob, s);
-        }
-
-        // Note: acquire doesn't check if the keeper already keeps anything.
-        // This is only for a use together with an empty constructor.
-        bool acquire(CUDTUnited& glob, CUDTSocket* s)
-        {
-            if (s == NULL)
-            {
-                socket = NULL;
-                return false;
-            }
-
-            const bool caught = glob.acquireSocket(s);
-            socket = caught ? s : NULL;
-            return caught;
-        }
-
-        ~SocketKeeper()
-        {
-            if (socket)
-            {
-                SRT_ASSERT(socket->isStillBusy() > 0);
-                socket->apiRelease();
-            }
-        }
-    };
 
 private:
 

srtcore/buffer_snd.cpp

@@ -87,8 +87,8 @@ CSndBuffer::CSndBuffer(int ip_family, int size, int maxpld, int authtag)
     m_pBuffer->m_pNext  = NULL;
 
     // circular linked list for out bound packets
-    m_pBlock  = new Block;
-    Block* pb = m_pBlock;
+    m_pBlock  = new CSndBlock;
+    CSndBlock* pb = m_pBlock;
     char* pc  = m_pBuffer->m_pcData;
 
     for (int i = 0; i < m_iSize; ++i)
@@ -99,7 +99,7 @@ CSndBuffer::CSndBuffer(int ip_family, int size, int maxpld, int authtag)
 
         if (i < m_iSize - 1)
         {
-            pb->m_pNext        = new Block;
+            pb->m_pNext        = new CSndBlock;
             pb                 = pb->m_pNext;
         }
     }
@@ -112,10 +112,10 @@ CSndBuffer::CSndBuffer(int ip_family, int size, int maxpld, int authtag)
 
 CSndBuffer::~CSndBuffer()
 {
-    Block* pb = m_pBlock->m_pNext;
+    CSndBlock* pb = m_pBlock->m_pNext;
     while (pb != m_pBlock)
     {
-        Block* temp = pb;
+        CSndBlock* temp = pb;
         pb          = pb->m_pNext;
         delete temp;
     }
@@ -170,7 +170,7 @@ void CSndBuffer::addBuffer(const char* data, int len, SRT_MSGCTRL& w_mctrl)
     // If there's more than one packet, this function must increase it by itself
     // and then return the accordingly modified sequence number in the reference.
 
-    Block* s = m_pLastBlock;
+    CSndBlock* s = m_pLastBlock;
 
     if (w_msgno == SRT_MSGNO_NONE) // DEFAULT-UNCHANGED msgno supplied
     {
@@ -258,7 +258,7 @@ int CSndBuffer::addBufferFromFile(fstream& ifs, int len)
           log << CONID() << "addBufferFromFile: adding " << iPktLen << " packets (" << len
               << " bytes) to send, msgno=" << m_iNextMsgNo);
 
-    Block* s     = m_pLastBlock;
+    CSndBlock* s     = m_pLastBlock;
     int    total = 0;
     for (int i = 0; i < iNumBlocks; ++i)
     {
@@ -343,7 +343,7 @@ int CSndBuffer::readData(CPacket& w_packet, steady_clock::time_point& w_srctime,
             m_pCurrBlock->m_iMsgNoBitset |= MSGNO_ENCKEYSPEC::wrap(kflgs);
         }
 
-        Block* p = m_pCurrBlock;
+        CSndBlock* p = m_pCurrBlock;
         w_packet.set_msgflags(m_pCurrBlock->m_iMsgNoBitset);
         w_srctime = m_pCurrBlock->m_tsOriginTime;
         m_pCurrBlock = m_pCurrBlock->m_pNext;
@@ -381,7 +381,7 @@ int32_t CSndBuffer::getMsgNoAt(const int offset)
 {
     ScopedLock bufferguard(m_BufLock);
 
-    Block* p = m_pFirstBlock;
+    CSndBlock* p = m_pFirstBlock;
 
     if (p)
     {
@@ -401,7 +401,7 @@ int32_t CSndBuffer::getMsgNoAt(const int offset)
     // XXX Suboptimal procedure to keep the blocks identifiable
     // by sequence number. Consider using some circular buffer.
     int       i;
-    Block* ee SRT_ATR_UNUSED = 0;
+    CSndBlock* ee SRT_ATR_UNUSED = 0;
     for (i = 0; i < offset && p; ++i)
     {
         ee = p;
@@ -423,14 +423,56 @@ int32_t CSndBuffer::getMsgNoAt(const int offset)
     return p->getMsgSeq();
 }
 
+bool CSndBuffer::getPacketRangeSize(int32_t seqlo, int32_t seqhi, int& w_packets, int& w_bytes)
+{
+    ScopedLock bufferguard(m_BufLock);
+
+    int npackets = 0, nbytes = 0;
+
+    // XXX Suboptimal procedure to keep the blocks identifiable
+    // by sequence number. Consider using some circular buffer.
+    CSndBlock* p = m_pFirstBlock;
+    for ( ; p != m_pLastBlock; p = p->m_pNext)
+    {
+        if (p->m_iSeqNo == seqlo)
+            break;
+    }
+    if (p == m_pLastBlock)
+        return false;
+
+    for ( ; p != m_pLastBlock; p = p->m_pNext)
+    {
+        ++npackets;
+        nbytes += p->m_iLength;
+        if (p->m_iSeqNo == seqhi)
+            break;
+    }
+
+    w_packets = npackets;
+    w_bytes = nbytes;
+
+    if (p == m_pLastBlock)
+    {
+        if (p->m_iSeqNo == seqhi)
+            return true;
+
+        // This means it caught some initial packets, but didn't reach
+        // the end of range. This way it will have some packets, but
+        // this was an error
+        return false;
+    }
+
+    return true;
+}
+
 int CSndBuffer::readData(const int offset, CPacket& w_packet, steady_clock::time_point& w_srctime, DropRange& w_drop)
 {
     // NOTE: w_packet.m_iSeqNo is expected to be set to the value
     // of the sequence number with which this packet should be sent.
 
     ScopedLock bufferguard(m_BufLock);
 
-    Block* p = m_pFirstBlock;
+    CSndBlock* p = m_pFirstBlock;
 
     // XXX Suboptimal procedure to keep the blocks identifiable
     // by sequence number. Consider using some circular buffer.
@@ -533,7 +575,7 @@ int CSndBuffer::readData(const int offset, CPacket& w_packet, steady_clock::time
 sync::steady_clock::time_point CSndBuffer::getPacketRexmitTime(const int offset)
 {
     ScopedLock bufferguard(m_BufLock);
-    const Block* p = m_pFirstBlock;
+    const CSndBlock* p = m_pFirstBlock;
 
     // XXX Suboptimal procedure to keep the blocks identifiable
     // by sequence number. Consider using some circular buffer.
@@ -721,20 +763,20 @@ void CSndBuffer::increase()
     p->m_pNext = nbuf;
 
     // new packet blocks
-    Block* nblk = NULL;
+    CSndBlock* nblk = NULL;
     try
     {
-        nblk = new Block;
+        nblk = new CSndBlock;
     }
     catch (...)
     {
         delete nblk;
         throw CUDTException(MJ_SYSTEMRES, MN_MEMORY, 0);
     }
-    Block* pb = nblk;
+    CSndBlock* pb = nblk;
     for (int i = 1; i < unitsize; ++i)
     {
-        pb->m_pNext = new Block;
+        pb->m_pNext = new CSndBlock;
         pb          = pb->m_pNext;
     }