[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

@@ -1377,7 +1377,7 @@ SRTSOCKET CUDTUnited::accept(const SRTSOCKET listen, sockaddr* pw_addr, int* pw_
     }
 
     CUDTSocket* ls;
-    SocketKeeper keep_ls;
+    SocketKeeper keep_ls = CUDT::keep();
 
     // We keep the mutex locked for the whole time of instant checks.
     // Once they pass, extend the life for the scope by SocketKeeper.
@@ -1462,7 +1462,7 @@ SRTSOCKET CUDTUnited::accept(const SRTSOCKET listen, sockaddr* pw_addr, int* pw_
 
     // NOTE: release() locks m_GlobControlLock.
     // Once we extracted the accepted socket, we don't need to keep ls busy.
-    keep_ls.release(*this);
+    keep_ls.release();
     ls = NULL; // NOT USABLE ANYMORE!
 
     if (!accepted) // The loop was interrupted
@@ -2387,7 +2387,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());
 
@@ -2397,7 +2398,7 @@ SRTSTATUS CUDTUnited::close(const SRTSOCKET u, int reason)
     // Releasing under the global lock to avoid even theoretical
     // data race.
 
-    k.release(*this);
+    k.release();
     return cstatus;
 }
 
@@ -2525,7 +2526,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;
@@ -4443,7 +4444,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);
@@ -4617,7 +4618,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)
@@ -4740,13 +4741,13 @@ SRTSTATUS CUDT::getsockopt(SRTSOCKET u, int, SRT_SOCKOPT optname, void* pw_optva
 #if SRT_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;
     }
@@ -4771,13 +4772,13 @@ SRTSTATUS CUDT::setsockopt(SRTSOCKET u, int, SRT_SOCKOPT optname, const void* op
 #if SRT_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;
     }
@@ -4816,12 +4817,12 @@ int CUDT::sendmsg2(SRTSOCKET u, const char* buf, int len, SRT_MSGCTRL& w_m)
 #if SRT_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)
     {
@@ -4860,12 +4861,12 @@ int CUDT::recvmsg2(SRTSOCKET u, char* buf, int len, SRT_MSGCTRL& w_m)
 #if SRT_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)
     {
@@ -4882,7 +4883,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)
@@ -4904,7 +4905,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)
     {
@@ -5209,7 +5210,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;
     }
@@ -5229,7 +5230,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;
     }
@@ -5251,7 +5252,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)
     {
@@ -5273,7 +5274,7 @@ SRT_SOCKSTATUS CUDT::getsockstate(SRTSOCKET u)
 #if SRT_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

@@ -276,12 +276,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.
@@ -568,10 +562,14 @@ class CUDTUnited
     };
 #endif
 
+public:
+
     CUDTSocket* locateAcquireSocket(SRTSOCKET u, ErrorHandling erh = ERH_RETURN);
     bool acquireSocket(CUDTSocket* s);
     void releaseSocket(CUDTSocket* s);
 
+private:
+
     SRT_TSA_NEEDS_LOCKED(m_InitLock)
     bool startGarbageCollector();
 
@@ -584,64 +582,6 @@ class CUDTUnited
     void cleanupAllSockets();
     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);
-        }
-
-        void acquire_LOCKED(CUDTSocket* s)
-        {
-            socket = s;
-            s->apiAcquire();
-        }
-
-        // 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;
-        }
-
-        bool release(CUDTUnited& glob)
-        {
-            if (!socket)
-                return false;
-
-            glob.releaseSocket(socket);
-            socket = NULL;
-            return true;
-        }
-
-        ~SocketKeeper()
-        {
-            if (socket)
-            {
-                SRT_ASSERT(socket->isStillBusy() > 0);
-                socket->apiRelease();
-            }
-        }
-    };
 
 private:
 

srtcore/buffer_snd.cpp

@@ -349,6 +349,58 @@ int32_t CSndBuffer::getMsgNoAtSeq(const int32_t seqno)
     return m_Packets[offset].getMsgSeq();
 }
 
+bool CSndBuffer::getPacketRangeSize(int32_t seqlo, int32_t seqhi, int& w_packets, int& w_bytes)
+{
+    ScopedLock bufferguard(m_BufLock);
+    int offset_lo = CSeqNo::seqoff(m_iSndLastDataAck, seqlo);
+    int offset_hi = CSeqNo::seqoff(m_iSndLastDataAck, seqhi);
+
+    w_packets = 0;
+    w_bytes = 0;
+
+    if (offset_hi < 0 || offset_hi < offset_lo)
+    {
+        HLOGC(bslog.Debug, log << "getPacketRangeSize: invalid seq range %(" << seqlo << "-" << seqhi
+                << ") map to off=(" << offset_lo << ", " << offset_hi << ")");
+        return false;
+    }
+
+    // Rule out empty packets case, not sure if possible, but still
+    if (m_Packets.empty())
+    {
+        // Treat this as false because if the buffer is empty,
+        // the sequence numbers definitely don't refer to any existing
+        // packets in the buffer.
+        return false;
+    }
+
+    bool full_range = true;
+    if (offset_lo < 0)
+    {
+        offset_lo = 0;
+        full_range = false;
+    }
+
+    if (offset_hi >= int(m_Packets.size()))
+    {
+        offset_hi = m_Packets.size() - 1;
+        full_range = false;
+    }
+
+    int npackets = offset_hi - offset_lo + 1,
+        nbytes = 0;
+
+    for (int i = offset_lo; i <= offset_hi; ++i)
+    {
+        nbytes += m_Packets[i].m_iLength;
+    }
+
+    w_packets = npackets;
+    w_bytes = nbytes;
+
+    return full_range;
+}
+
 // XXX Likely unused (left for the use by tests)
 int CSndBuffer::readOldPacket(int32_t seqno, CSndPacket& w_sndpkt, steady_clock::time_point& w_srctime, DropRange& w_drop)
 {

srtcore/buffer_snd.h

@@ -453,6 +453,7 @@ class CSndBuffer
 
     int  getAvgBufSize(int& bytes, int& timespan);
 
+    bool getPacketRangeSize(int32_t seqlo, int32_t seqhi, int& w_packets, int& w_bytes);
     int getCurrBufSize(int& bytes, int& timespan) const
     {
         sync::ScopedLock lk (m_BufLock);

srtcore/common.cpp

@@ -69,6 +69,7 @@ modified by
  #include <ifaddrs.h>
 #endif
 
+#include "api.h"
 #include "md5.h"
 #include "common.h"
 #include "netinet_any.h"
@@ -554,6 +555,7 @@ vector<LocalInterface> GetLocalInterfaces()
     return locals;
 }
 
+SRTSOCKET SocketKeeper::id() const { return socket ? socket->id() : SRT_INVALID_SOCK; }
 
 
 // Value display utilities