[core] New implementation for the logging/formatting system

apps/logsupport.cpp

@@ -173,7 +173,7 @@ set<srt_logging::LogFA> SrtParseLogFA(string fa, set<string>* punknown)
 
 void ParseLogFASpec(const vector<string>& speclist, string& w_on, string& w_off)
 {
-    std::ostringstream son, soff;
+    fmt::obufstream son, soff;
 
     for (auto& s: speclist)
     {

apps/uriparser.cpp

@@ -64,7 +64,7 @@ string UriParser::makeUri()
         prefix = m_proto + "://";
     }
 
-    std::ostringstream out;
+    fmt::obufstream out;
 
     out << prefix << m_host;
     if ((m_port == "" || m_port == "0") && m_expect == EXPECT_FILE)

srtcore/api.cpp

@@ -231,7 +231,7 @@ string srt::CUDTUnited::CONID(SRTSOCKET sock)
     if (sock == 0)
         return "";
 
-    std::ostringstream os;
+    fmt::obufstream os;
     os << "@" << sock << ":";
     return os.str();
 }
@@ -3249,7 +3249,7 @@ bool srt::CUDTUnited::updateListenerMux(CUDTSocket* s, const CUDTSocket* ls)
             CMultiplexer& m = i->second;
 
 #if ENABLE_HEAVY_LOGGING
-            ostringstream that_muxer;
+            fmt::obufstream that_muxer;
             that_muxer << "id=" << m.m_iID << " port=" << m.m_iPort
                        << " ip=" << (m.m_iIPversion == AF_INET ? "v4" : "v6");
 #endif

srtcore/common.cpp

@@ -277,15 +277,12 @@ void srt::CIPAddress::pton(sockaddr_any& w_addr, const uint32_t ip[4], const soc
     }
     else
     {
-        LOGC(inlog.Error, log << "pton: IPE or net error: can't determine IPv4 carryover format: " << std::hex
-                << peeraddr16[0] << ":"
-                << peeraddr16[1] << ":"
-                << peeraddr16[2] << ":"
-                << peeraddr16[3] << ":"
-                << peeraddr16[4] << ":"
-                << peeraddr16[5] << ":"
-                << peeraddr16[6] << ":"
-                << peeraddr16[7] << std::dec);
+        fmt::obufstream peeraddr_form;
+        peeraddr_form << fmt::sfmt(peeraddr16[0], "04x");
+        for (int i = 1; i < 8; ++i)
+            peeraddr_form << ":" << fmt::sfmt(peeraddr16[i], "04x");
+
+        LOGC(inlog.Error, log << "pton: IPE or net error: can't determine IPv4 carryover format: " << peeraddr_form);
         *target_ipv4_addr = 0;
         if (peer.family() != AF_INET)
         {

srtcore/congctl.cpp

@@ -595,7 +595,7 @@ class FileCC : public SrtCongestionControlBase
             {
                 m_dPktSndPeriod = m_dCWndSize / (m_parent->SRTT() + m_iRCInterval);
                 HLOGC(cclog.Debug, log << "FileCC: CHKTIMER, SLOWSTART:OFF, sndperiod=" << m_dPktSndPeriod << "us AS wndsize/(RTT+RCIV) (wndsize="
-                    << setprecision(6) << m_dCWndSize << " RTT=" << m_parent->SRTT() << " RCIV=" << m_iRCInterval << ")");
+                    << fmt::sfmt(m_dCWndSize, ".6") << " RTT=" << m_parent->SRTT() << " RCIV=" << m_iRCInterval << ")");
             }
         }
         else

srtcore/core.cpp

@@ -88,10 +88,14 @@ using namespace std;
 using namespace srt;
 using namespace srt::sync;
 using namespace srt_logging;
+using fmt::sfmt;
+using fmt::sfmc;
 
 const SRTSOCKET UDT::INVALID_SOCK = srt::CUDT::INVALID_SOCK;
 const int       UDT::ERROR        = srt::CUDT::ERROR;
 
+static inline char fmt_onoff(bool val) { return val ? '+' : '-'; }
+
 //#define SRT_CMD_HSREQ       1           /* SRT Handshake Request (sender) */
 #define SRT_CMD_HSREQ_MINSZ 8 /* Minumum Compatible (1.x.x) packet size (bytes) */
 #define SRT_CMD_HSREQ_SZ 12   /* Current version packet size */
@@ -1773,8 +1777,11 @@ bool srt::CUDT::createSrtHandshake(
             m_RejectReason = SRT_REJ_IPE;
             LOGC(cnlog.Error,
                  log << CONID() << "createSrtHandshake: IPE: need to send KM, but CryptoControl does not exist."
-                     << " Socket state: connected=" << boolalpha << m_bConnected << ", connecting=" << m_bConnecting
-                     << ", broken=" << m_bBroken << ", closing=" << m_bClosing << ".");
+                     << " Socket state: "
+                     << fmt_onoff(m_bConnected) << "connected, "
+                     << fmt_onoff(m_bConnecting) << "connecting, "
+                     << fmt_onoff(m_bBroken) << "broken, "
+                     << fmt_onoff(m_bClosing) << "closing.");
             return false;
         }
 
@@ -2102,8 +2109,9 @@ int srt::CUDT::processSrtMsg_HSREQ(const uint32_t *srtdata, size_t bytelen, uint
     }
 
     LOGC(cnlog.Debug, log << "HSREQ/rcv: cmd=" << SRT_CMD_HSREQ << "(HSREQ) len=" << bytelen
-                          << hex << " vers=0x" << srtdata[SRT_HS_VERSION] << " opts=0x" << srtdata[SRT_HS_FLAGS]
-                          << dec << " delay=" << SRT_HS_LATENCY_RCV::unwrap(srtdata[SRT_HS_LATENCY]));
+                          << " vers=0x" << sfmt(srtdata[SRT_HS_VERSION], "x")
+                          << " opts=0x" << sfmt(srtdata[SRT_HS_FLAGS], "x")
+                          << " delay=" << SRT_HS_LATENCY_RCV::unwrap(srtdata[SRT_HS_LATENCY]));
 
     m_uPeerSrtVersion = srtdata[SRT_HS_VERSION];
     m_uPeerSrtFlags   = srtdata[SRT_HS_FLAGS];
@@ -2337,8 +2345,8 @@ int srt::CUDT::processSrtMsg_HSRSP(const uint32_t *srtdata, size_t bytelen, uint
     m_uPeerSrtFlags   = srtdata[SRT_HS_FLAGS];
 
     HLOGC(cnlog.Debug, log << "HSRSP/rcv: Version: " << SrtVersionString(m_uPeerSrtVersion)
-                           << " Flags: SND:" << setw(8) << setfill('0') << hex << m_uPeerSrtFlags
-                           << setw(0) << " (" << SrtFlagString(m_uPeerSrtFlags) << ")");
+                           << " Flags: SND:" << sfmt(m_uPeerSrtFlags, "08x")
+                           << " (" << SrtFlagString(m_uPeerSrtFlags) << ")");
     // Basic version check
     if (m_uPeerSrtVersion < m_config.uMinimumPeerSrtVersion)
     {
@@ -4101,8 +4109,12 @@ EConnectStatus srt::CUDT::craftKmResponse(uint32_t* aw_kmdata, size_t& w_kmdatas
             m_RejectReason = SRT_REJ_IPE;
             LOGC(cnlog.Error,
                  log << CONID() << "IPE: craftKmResponse needs to send KM, but CryptoControl does not exist."
-                     << " Socket state: connected=" << boolalpha << m_bConnected << ", connecting=" << m_bConnecting
-                     << ", broken=" << m_bBroken << ", opened " << m_bOpened << ", closing=" << m_bClosing << ".");
+                     << " Socket state: "
+                     << fmt_onoff(m_bConnected) << "connected, "
+                     << fmt_onoff(m_bConnecting) << "connecting, "
+                     << fmt_onoff(m_bBroken) << "broken, "
+                     << fmt_onoff(m_bOpened) << "opened, "
+                     << fmt_onoff(m_bClosing) << "closing.");
             return CONN_REJECT;
         }
         // This is a periodic handshake update, so you need to extract the KM data from the
@@ -5460,14 +5472,14 @@ void * srt::CUDT::tsbpd(void* param)
                 HLOGC(tslog.Debug,
                     log << self->CONID() << "tsbpd: DROPSEQ: up to seqno %" << CSeqNo::decseq(info.seqno) << " ("
                     << iDropCnt << " packets) playable at " << FormatTime(info.tsbpd_time) << " delayed "
-                    << (timediff_us / 1000) << "." << std::setw(3) << std::setfill('0') << (timediff_us % 1000) << " ms");
+                    << (timediff_us / 1000) << "." << sfmt(timediff_us % 1000, "03") << " ms");
 #endif
                 string why;
                 if (self->frequentLogAllowed(FREQLOGFA_RCV_DROPPED, tnow, (why)))
                 {
                     LOGC(brlog.Warn, log << self->CONID() << "RCV-DROPPED " << iDropCnt << " packet(s). Packet seqno %" << info.seqno
-                            << " delayed for " << (timediff_us / 1000) << "." << std::setw(3) << std::setfill('0')
-                            << (timediff_us % 1000) << " ms " << why);
+                            << " delayed for " << (timediff_us / 1000) << "."
+                            << sfmt(timediff_us % 1000, "03") << " ms " << why);
                 }
 #if SRT_ENABLE_FREQUENT_LOG_TRACE
                 else
@@ -7767,7 +7779,7 @@ bool srt::CUDT::updateCC(ETransmissionEvent evt, const EventVariant arg)
         HLOGC(rslog.Debug,
               log << CONID() << "updateCC: updated values from congctl: interval=" << FormatDuration<DUNIT_US>(m_tdSendInterval)
                   << " (cfg:" << m_CongCtl->pktSndPeriod_us() << "us) cgwindow="
-                  << std::setprecision(3) << cgwindow);
+                  << sfmt(cgwindow, sfmc().precision(3)));
 #endif
     }
 
@@ -8408,7 +8420,7 @@ void srt::CUDT::processCtrlAck(const CPacket &ctrlpkt, const steady_clock::time_
         // included, but it also triggers for any other kind of invalid value.
         // This check MUST BE DONE before making any operation on this number.
         LOGC(inlog.Error, log << CONID() << "ACK: IPE/EPE: received invalid ACK value: " << ackdata_seqno
-                << " " << std::hex << ackdata_seqno << " (IGNORED)");
+                << " " << sfmt(ackdata_seqno, "x") << " (IGNORED)");
         return;
     }
 
@@ -10002,12 +10014,11 @@ int srt::CUDT::checkLazySpawnTsbPdThread()
 
         HLOGP(qrlog.Debug, "Spawning Socket TSBPD thread");
 #if ENABLE_HEAVY_LOGGING
-        std::ostringstream tns1, tns2;
+        fmt::obufstream buf;
         // Take the last 2 ciphers from the socket ID.
-        tns1 << setfill('0') << setw(2) << m_SocketID;
-        std::string s = tns1.str();
-        tns2 << "SRT:TsbPd:@" << s.substr(s.size()-2, 2);
-        const string thname = tns2.str();
+        string s = fmt::sfmts(m_SocketID, "02");
+        buf << "SRT:TsbPd:@" << s.substr(s.size()-2, 2);
+        const string thname = buf.str();
 #else
         const string thname = "SRT:TsbPd";
 #endif

srtcore/group.cpp

@@ -15,6 +15,8 @@ extern const int32_t SRT_DEF_VERSION;
 
 namespace srt {
 
+static inline char fmt_onoff(bool val) { return val ? '+' : '-'; }
+
 int32_t CUDTGroup::s_tokenGen = 0;
 
 // [[using locked(this->m_GroupLock)]];
@@ -2190,8 +2192,8 @@ int CUDTGroup::recv(char* buf, int len, SRT_MSGCTRL& w_mc)
         if (!m_bOpened || !m_bConnected)
         {
             LOGC(grlog.Error,
-                 log << boolalpha << "grp/recv: $" << id() << ": ABANDONING: opened=" << m_bOpened
-                     << " connected=" << m_bConnected);
+                 log << "grp/recv: $" << id() << ": ABANDONING: opened" << fmt_onoff(m_bOpened)
+                     << " connected" << fmt_onoff(m_bConnected));
             throw CUDTException(MJ_CONNECTION, MN_NOCONN, 0);
         }
 

srtcore/logging.cpp

@@ -43,7 +43,7 @@ LogDispatcher::Proxy LogDispatcher::operator()()
     return Proxy(*this);
 }
 
-void LogDispatcher::CreateLogLinePrefix(std::ostringstream& serr)
+void LogDispatcher::CreateLogLinePrefix(fmt::obufstream& serr)
 {
     using namespace std;
     using namespace srt;
@@ -60,7 +60,7 @@ void LogDispatcher::CreateLogLinePrefix(std::ostringstream& serr)
 
         if (strftime(tmp_buf, sizeof(tmp_buf), "%X.", &tm))
         {
-            serr << tmp_buf << setw(6) << setfill('0') << tv.tv_usec;
+            serr << tmp_buf << fmt::sfmt(tv.tv_usec, "06");
         }
     }
 

srtcore/logging.h

@@ -20,7 +20,6 @@ written by
 #include <iostream>
 #include <iomanip>
 #include <set>
-#include <sstream>
 #include <cstdarg>
 #ifdef _WIN32
 #include "win/wintime.h"
@@ -29,6 +28,8 @@ written by
 #include <sys/time.h>
 #endif
 
+#include "sfmt.h"
+
 #include "srt.h"
 #include "utilities.h"
 #include "threadname.h"
@@ -192,7 +193,7 @@ struct SRT_API LogDispatcher
 
     bool CheckEnabled();
 
-    void CreateLogLinePrefix(std::ostringstream&);
+    void CreateLogLinePrefix(fmt::obufstream&);
     void SendLogLine(const char* file, int line, const std::string& area, const std::string& sl);
 
     // log.Debug("This is the ", nth, " time");  <--- C++11 only.
@@ -285,7 +286,7 @@ struct LogDispatcher::Proxy
 {
     LogDispatcher& that;
 
-    std::ostringstream os;
+    fmt::obufstream os;
 
     // Cache the 'enabled' state in the beginning. If the logging
     // becomes enabled or disabled in the middle of the log, we don't
@@ -336,12 +337,49 @@ struct LogDispatcher::Proxy
         return *this;
     }
 
+    // Special case for atomics, as passing them to snprintf() call
+    // requires unpacking the real underlying value.
+    template <class T>
+    Proxy& operator<<(const srt::sync::atomic<T>& arg)
+    {
+        if (that_enabled)
+        {
+            os << arg.load();
+        }
+        return *this;
+    }
+
+
+#if HAVE_CXX11
+
+    void dispatch() {}
+
+    template<typename Arg1, typename... Args>
+    void dispatch(const Arg1& a1, const Args&... others)
+    {
+        *this << a1;
+        dispatch(others...);
+    }
+
+    // Special dispatching for atomics must be provided here.
+    // By some reason, "*this << a1" expression gets dispatched
+    // to the general version of operator<<, not the overload for
+    // atomic. Even though the compiler shows Arg1 type as atomic.
+    template<typename Arg1, typename... Args>
+    void dispatch(const srt::sync::atomic<Arg1>& a1, const Args&... others)
+    {
+        *this << a1.load();
+        dispatch(others...);
+    }
+
+#endif
+
     ~Proxy()
     {
-        if ( that_enabled )
+        if (that_enabled)
         {
-            if ( (flags & SRT_LOGF_DISABLE_EOL) == 0 )
-                os << std::endl;
+            if ((flags & SRT_LOGF_DISABLE_EOL) == 0)
+                os << fmt::seol;
             that.SendLogLine(i_file, i_line, area, os.str());
         }
         // Needed in destructor?
@@ -435,28 +473,28 @@ inline bool LogDispatcher::CheckEnabled()
 
 //extern std::mutex Debug_mutex;
 
-inline void PrintArgs(std::ostream&) {}
+inline void PrintArgs(fmt::obufstream&) {}
 
 template <class Arg1, class... Args>
-inline void PrintArgs(std::ostream& serr, Arg1&& arg1, Args&&... args)
+inline void PrintArgs(fmt::obufstream& serr, Arg1&& arg1, Args&&... args)
 {
     serr << std::forward<Arg1>(arg1);
     PrintArgs(serr, args...);
 }
 
+// Add exceptional handling for sync::atomic
+template <class Arg1, class... Args>
+inline void PrintArgs(fmt::obufstream& serr, const srt::sync::atomic<Arg1>& arg1, Args&&... args)
+{
+    serr << arg1.load();
+    PrintArgs(serr, args...);
+}
+
 template <class... Args>
 inline void LogDispatcher::PrintLogLine(const char* file SRT_ATR_UNUSED, int line SRT_ATR_UNUSED, const std::string& area SRT_ATR_UNUSED, Args&&... args SRT_ATR_UNUSED)
 {
 #ifdef ENABLE_LOGGING
-    std::ostringstream serr;
-    CreateLogLinePrefix(serr);
-    PrintArgs(serr, args...);
-
-    if ( !isset(SRT_LOGF_DISABLE_EOL) )
-        serr << std::endl;
-
-    // Not sure, but it wasn't ever used.
-    SendLogLine(file, line, area, serr.str());
+    Proxy(*this).dispatch(args...);
 #endif
 }
 
@@ -466,15 +504,7 @@ template <class Arg>
 inline void LogDispatcher::PrintLogLine(const char* file SRT_ATR_UNUSED, int line SRT_ATR_UNUSED, const std::string& area SRT_ATR_UNUSED, const Arg& arg SRT_ATR_UNUSED)
 {
 #ifdef ENABLE_LOGGING
-    std::ostringstream serr;
-    CreateLogLinePrefix(serr);
-    serr << arg;
-
-    if ( !isset(SRT_LOGF_DISABLE_EOL) )
-        serr << std::endl;
-
-    // Not sure, but it wasn't ever used.
-    SendLogLine(file, line, area, serr.str());
+    Proxy(*this) << arg;
 #endif
 }
 

srtcore/queue.cpp

@@ -1078,8 +1078,8 @@ bool srt::CRendezvousQueue::qualifyToHandle(EReadStatus    rst,
         else
         {
             HLOGC(cnlog.Debug,
-                  log << "RID: socket @" << i->m_iID << " still active (remaining " << std::fixed
-                      << (count_microseconds(i->m_tsTTL - tsNow) / 1000000.0) << "s of TTL)...");
+                  log << "RID: socket @" << i->m_iID << " still active (remaining "
+                      << fmt::sfmt(count_microseconds(i->m_tsTTL - tsNow) / 1000000.0, "f") << "s of TTL)...");
         }
 
         const steady_clock::time_point tsLastReq = i->m_pUDT->m_tsLastReqTime;

srtcore/queue.h

@@ -592,6 +592,10 @@ struct CMultiplexer
         , m_pRcvQueue(NULL)
         , m_pChannel(NULL)
         , m_pTimer(NULL)
+        , m_iPort(0)
+        , m_iIPversion(0)
+        , m_iRefCount(1)
+        , m_iID(-1)
     {
     }