diff --git a/srtcore/api.cpp b/srtcore/api.cpp index 7ec8ff570..94d813cae 100644 --- a/srtcore/api.cpp +++ b/srtcore/api.cpp @@ -117,14 +117,14 @@ SRT_SOCKSTATUS srt::CUDTSocket::getStatus() } // [[using locked(m_GlobControlLock)]] -void srt::CUDTSocket::breakSocket_LOCKED() +void srt::CUDTSocket::breakSocket_LOCKED(int reason) { // This function is intended to be called from GC, // under a lock of m_GlobControlLock. m_UDT.m_bBroken = true; m_UDT.m_iBrokenCounter = 0; HLOGC(smlog.Debug, log << "@" << m_SocketID << " CLOSING AS SOCKET"); - m_UDT.closeInternal(); + m_UDT.closeInternal(reason); setClosed(); } @@ -813,7 +813,7 @@ int srt::CUDTUnited::newConnection(const SRTSOCKET listen, #endif SRTSOCKET id = ns->m_SocketID; - ns->core().closeInternal(); + ns->core().closeInternal(SRT_CLS_LATE); ns->setClosed(); // The mapped socket should be now unmapped to preserve the situation that @@ -911,6 +911,37 @@ SRT_SOCKSTATUS srt::CUDTUnited::getStatus(const SRTSOCKET u) return i->second->getStatus(); } +int srt::CUDTUnited::getCloseReason(const SRTSOCKET u, SRT_CLOSE_INFO& info) +{ + // protects the m_Sockets structure + ScopedLock cg(m_GlobControlLock); + + // We need to search for the socket in: + // m_Sockets, if it is somehow still alive, + // m_ClosedSockets, if it's when it should be, + // m_ClosedDatabase, if it has been already garbage-collected and deleted. + + sockets_t::const_iterator i = m_Sockets.find(u); + if (i != m_Sockets.end()) + { + i->second->core().copyCloseInfo((info)); + return 0; + } + + i = m_ClosedSockets.find(u); + if (i != m_ClosedSockets.end()) + { + i->second->core().copyCloseInfo((info)); + } + + map::iterator c = m_ClosedDatabase.find(u); + if (c == m_ClosedDatabase.end()) + return -1; + + info = c->second.info; + return 0; +} + int srt::CUDTUnited::bind(CUDTSocket* s, const sockaddr_any& name) { ScopedLock cg(s->m_ControlLock); @@ -1811,7 +1842,7 @@ int srt::CUDTUnited::groupConnect(CUDTGroup* pg, SRT_SOCKGROUPCONFIG* targets, i continue; // This will also automatically remove it from the group and all eids - close(s); + close(s, SRT_CLS_INTERNAL); } // There's no possibility to report a problem on every connection @@ -1895,7 +1926,7 @@ int srt::CUDTUnited::connectIn(CUDTSocket* s, const sockaddr_any& target_addr, i return 0; } -int srt::CUDTUnited::close(const SRTSOCKET u) +int srt::CUDTUnited::close(const SRTSOCKET u, int reason) { #if ENABLE_BONDING if (u & SRTGROUP_MASK) @@ -1910,7 +1941,7 @@ int srt::CUDTUnited::close(const SRTSOCKET u) if (!s) throw CUDTException(MJ_NOTSUP, MN_SIDINVAL, 0); - return close(s); + return close(s, reason); } #if ENABLE_BONDING @@ -1961,7 +1992,54 @@ void srt::CUDTUnited::deleteGroup_LOCKED(CUDTGroup* g) } #endif -int srt::CUDTUnited::close(CUDTSocket* s) +// [[using locked(m_GlobControlLock)]] +void srt::CUDTUnited::recordCloseReason(CUDTSocket* s) +{ + CloseInfo ci; + ci.info.agent = SRT_CLOSE_REASON(s->core().m_AgentCloseReason.load()); + ci.info.peer = SRT_CLOSE_REASON(s->core().m_PeerCloseReason.load()); + ci.info.time = s->core().m_CloseTimeStamp.load().time_since_epoch().count(); + + m_ClosedDatabase[s->m_SocketID] = ci; + + // As a DOS attack prevention, do not allow to keep more than 10 records. + // In a normal functioning of the application this shouldn't be necessary, + // but it is still needed that a record of a dead socket is kept for + // 10 gc cycles more to ensure that the application can obtain it even after + // the socket has been physically removed. But if we don't limit the number + // of these records, this could be vulnerable for DOS attack if the user + // forces the application to create and close SRT sockets very quickly. + // Hence remove the oldest record, which can be recognized from the `time` + // field, if the number of records exceeds 10. + if (m_ClosedDatabase.size() > MAX_CLOSE_RECORD_SIZE) + { + // remove the oldest one + // This can only be done by collecting all time info + map which; + + for (map::iterator x = m_ClosedDatabase.begin(); + x != m_ClosedDatabase.end(); ++x) + { + which[x->second.info.time] = x->first; + } + + map::iterator y = which.begin(); + size_t ntodel = m_ClosedDatabase.size() - MAX_CLOSE_RECORD_SIZE; + for (size_t i = 0; i < ntodel; ++i) + { + // Sanity check - should never happen because it's unlikely + // that two different sockets were closed exactly at the same + // nanosecond time. + if (y == which.end()) + break; + + m_ClosedDatabase.erase(y->second); + ++y; + } + } +} + +int srt::CUDTUnited::close(CUDTSocket* s, int reason) { HLOGC(smlog.Debug, log << s->core().CONID() << "CLOSE. Acquiring control lock"); ScopedLock socket_cg(s->m_ControlLock); @@ -1994,6 +2072,8 @@ int srt::CUDTUnited::close(CUDTSocket* s) // broadcast all "accept" waiting CSync::lock_notify_all(s->m_AcceptCond, s->m_AcceptLock); + + s->core().setAgentCloseReason(reason); } else { @@ -2003,7 +2083,7 @@ int srt::CUDTUnited::close(CUDTSocket* s) // may block INDEFINITELY. As long as it's acceptable to block the // call to srt_close(), and all functions in all threads where this // very socket is used, this shall not block the central database. - s->core().closeInternal(); + s->core().closeInternal(reason); // synchronize with garbage collection. HLOGC(smlog.Debug, @@ -2038,6 +2118,8 @@ int srt::CUDTUnited::close(CUDTSocket* s) } #endif + recordCloseReason(s); + m_Sockets.erase(s->m_SocketID); m_ClosedSockets[s->m_SocketID] = s; HLOGC(smlog.Debug, log << "@" << u << "U::close: Socket MOVED TO CLOSED for collecting later."); @@ -2639,6 +2721,12 @@ void srt::CUDTUnited::checkBrokenSockets() HLOGC(smlog.Debug, log << "checkBrokenSockets: moving BROKEN socket to CLOSED: @" << i->first); + // Note that this will not override the value that has been already + // set by some other functionality, only set it when not yet set. + s->core().setAgentCloseReason(SRT_CLS_INTERNAL); + + recordCloseReason(s); + // close broken connections and start removal timer s->setClosed(); tbc.push_back(i->first); @@ -2757,7 +2845,7 @@ void srt::CUDTUnited::removeSocket(const SRTSOCKET u) CUDTSocket* as = si->second; - as->breakSocket_LOCKED(); + as->breakSocket_LOCKED(SRT_CLS_DEADLSN); m_ClosedSockets[*q] = as; m_Sockets.erase(*q); } @@ -2783,7 +2871,7 @@ void srt::CUDTUnited::removeSocket(const SRTSOCKET u) m_ClosedSockets.erase(i); HLOGC(smlog.Debug, log << "GC/removeSocket: closing associated UDT @" << u); - s->core().closeInternal(); + s->core().closeInternal(SRT_CLS_INTERNAL); HLOGC(smlog.Debug, log << "GC/removeSocket: DELETING SOCKET @" << u); delete s; HLOGC(smlog.Debug, log << "GC/removeSocket: socket @" << u << " DELETED. Checking muxer."); @@ -2823,6 +2911,31 @@ void srt::CUDTUnited::removeSocket(const SRTSOCKET u) } } +void srt::CUDTUnited::checkTemporaryDatabases() +{ + ScopedLock cg(m_GlobControlLock); + + // It's not very efficient to collect first the keys of all + // elements to remove and then remove from the map by key. + + // In C++20 this is possible by doing + // m_ClosedDatabase.erase_if([](auto& c) { return --c.generation <= 0; }); + // but nothing equivalent in the earlier standards. + + vector expired; + + for (map::iterator c = m_ClosedDatabase.begin(); + c != m_ClosedDatabase.end(); ++c) + { + --c->second.generation; + if (c->second.generation <= 0) + expired.push_back(c->first); + } + + for (vector::iterator i = expired.begin(); i != expired.end(); ++i) + m_ClosedDatabase.erase(*i); +} + void srt::CUDTUnited::configureMuxer(CMultiplexer& w_m, const CUDTSocket* s, int af) { w_m.m_mcfg = s->core().m_config; @@ -3278,14 +3391,20 @@ void* srt::CUDTUnited::garbageCollect(void* p) UniqueLock gclock(self->m_GCStopLock); + // START LIBRARY RUNNING LOOP while (!self->m_bClosing) { INCREMENT_THREAD_ITERATIONS(); self->checkBrokenSockets(); + self->checkTemporaryDatabases(); HLOGC(inlog.Debug, log << "GC: sleep 1 s"); self->m_GCStopCond.wait_for(gclock, seconds_from(1)); } + // END. + + // All the below code does the library cleanup, which should + // happen as a result of an application calling `srt_cleanup()`. // remove all sockets and multiplexers HLOGC(inlog.Debug, log << "GC: GLOBAL EXIT - releasing all pending sockets. Acquring control lock..."); @@ -3293,10 +3412,14 @@ void* srt::CUDTUnited::garbageCollect(void* p) { ScopedLock glock(self->m_GlobControlLock); + // Do not do generative expiry removal - there's no chance + // anyone can extract the close reason information since this point on. + self->m_ClosedDatabase.clear(); + for (sockets_t::iterator i = self->m_Sockets.begin(); i != self->m_Sockets.end(); ++i) { CUDTSocket* s = i->second; - s->breakSocket_LOCKED(); + s->breakSocket_LOCKED(SRT_CLS_CLEANUP); #if ENABLE_BONDING if (s->m_GroupOf) @@ -3705,11 +3828,11 @@ int srt::CUDT::connect(SRTSOCKET u, const sockaddr* name, int namelen, int32_t f } } -int srt::CUDT::close(SRTSOCKET u) +int srt::CUDT::close(SRTSOCKET u, int reason) { try { - return uglobal().close(u); + return uglobal().close(u, reason); } catch (const CUDTException& e) { @@ -4366,7 +4489,7 @@ int connect(SRTSOCKET u, const struct sockaddr* name, int namelen) int close(SRTSOCKET u) { - return srt::CUDT::close(u); + return srt::CUDT::close(u, SRT_CLS_API); } int getpeername(SRTSOCKET u, struct sockaddr* name, int* namelen) diff --git a/srtcore/api.h b/srtcore/api.h index 9ba77d23a..1441e7bd9 100644 --- a/srtcore/api.h +++ b/srtcore/api.h @@ -186,7 +186,7 @@ class CUDTSocket /// from within the GC thread only (that is, only when /// the socket should be no longer visible in the /// connection, including for sending remaining data). - void breakSocket_LOCKED(); + void breakSocket_LOCKED(int reason); /// This makes the socket no longer capable of performing any transmission /// operation, but continues to be responsive in the connection in order @@ -233,6 +233,8 @@ class CUDTUnited // Public constants static const int32_t MAX_SOCKET_VAL = SRTGROUP_MASK - 1; // maximum value for a regular socket + static const int MAX_CLOSE_RECORD_TTL = 10; + static const size_t MAX_CLOSE_RECORD_SIZE = 10; public: enum ErrorHandling @@ -295,8 +297,8 @@ class CUDTUnited int groupConnect(CUDTGroup* g, SRT_SOCKGROUPCONFIG targets[], int arraysize); int singleMemberConnect(CUDTGroup* g, SRT_SOCKGROUPCONFIG* target); #endif - int close(const SRTSOCKET u); - int close(CUDTSocket* s); + int close(const SRTSOCKET u, int reason); + int close(CUDTSocket* s, int reason); void getpeername(const SRTSOCKET u, sockaddr* name, int* namelen); void getsockname(const SRTSOCKET u, sockaddr* name, int* namelen); int select(UDT::UDSET* readfds, UDT::UDSET* writefds, UDT::UDSET* exceptfds, const timeval* timeout); @@ -488,6 +490,25 @@ class CUDTUnited CEPoll m_EPoll; // handling epoll data structures and events + struct CloseInfo + { + SRT_CLOSE_INFO info; + int generation; + + // The value here defines how many GC rolls it takes + // to remove the record. As GC rolls every 1 second, + // this is more-less the number of seconds this record + // will be alive AFTER you close the socket. + CloseInfo(): generation(MAX_CLOSE_RECORD_TTL) {} + }; + std::map m_ClosedDatabase; + + void checkTemporaryDatabases(); + void recordCloseReason(CUDTSocket* s); + +public: + int getCloseReason(const SRTSOCKET u, SRT_CLOSE_INFO& info); + private: CUDTUnited(const CUDTUnited&); CUDTUnited& operator=(const CUDTUnited&); diff --git a/srtcore/atomic_clock.h b/srtcore/atomic_clock.h index e01012313..840d35252 100644 --- a/srtcore/atomic_clock.h +++ b/srtcore/atomic_clock.h @@ -73,6 +73,12 @@ class AtomicClock dur.store(uint64_t(d.time_since_epoch().count())); } + void compare_exchange(const time_point_type& exp, const time_point_type& toset) + { + uint64_t val = exp.time_since_epoch().count(); + dur.compare_exchange(val, toset.time_since_epoch().count()); + } + AtomicClock& operator=(const time_point_type& s) { dur = s.time_since_epoch().count(); diff --git a/srtcore/core.cpp b/srtcore/core.cpp index 0e3cce0ee..17cbaeaab 100644 --- a/srtcore/core.cpp +++ b/srtcore/core.cpp @@ -3703,7 +3703,9 @@ void srt::CUDT::startConnect(const sockaddr_any& serv_addr, int32_t forced_isn) { HLOGC(cnlog.Debug, log << CONID() << "startConnect: REJECTED by processConnectResponse - sending SHUTDOWN"); - sendCtrl(UMSG_SHUTDOWN); + setAgentCloseReason(SRT_CLS_LATE); + uint32_t reason[1] = { SRT_CLS_LATE }; + sendCtrl(UMSG_SHUTDOWN, NULL, reason, sizeof reason); } if (cst != CONN_CONTINUE && cst != CONN_CONFUSED) @@ -6052,7 +6054,7 @@ void srt::CUDT::addressAndSend(CPacket& w_pkt) } // [[using maybe_locked(m_GlobControlLock, if called from GC)]] -bool srt::CUDT::closeInternal() +bool srt::CUDT::closeInternal(int reason) { // NOTE: this function is called from within the garbage collector thread. @@ -6104,6 +6106,14 @@ bool srt::CUDT::closeInternal() } } + // Some calls of closeInternal pass UNKNOWN here, which means + // that they don't want to change the code. It should have been + // set already somewhere else, however. + if (reason != SRT_CLS_UNKNOWN) + { + setAgentCloseReason(reason); + } + // remove this socket from the snd queue if (m_bConnected) m_pSndQueue->m_pSndUList->remove(this); @@ -6186,7 +6196,8 @@ bool srt::CUDT::closeInternal() if (!m_bShutdown) { HLOGC(smlog.Debug, log << CONID() << "CLOSING - sending SHUTDOWN to the peer @" << m_PeerID); - sendCtrl(UMSG_SHUTDOWN); + int32_t shdata[1] = { reason }; + sendCtrl(UMSG_SHUTDOWN, NULL, shdata, sizeof shdata); } // Store current connection information. @@ -7843,7 +7854,7 @@ void srt::CUDT::sendCtrl(UDTMessageType pkttype, const int32_t* lparam, void* rp case UMSG_SHUTDOWN: // 101 - Shutdown if (m_PeerID == 0) // Dont't send SHUTDOWN if we don't know peer ID. break; - ctrlpkt.pack(pkttype); + ctrlpkt.pack(pkttype, NULL, rparam, size); ctrlpkt.m_iID = m_PeerID; nbsent = m_pSndQueue->sendto(m_PeerAddr, ctrlpkt, m_SourceAddr); @@ -8333,6 +8344,7 @@ void srt::CUDT::processCtrlAck(const CPacket &ctrlpkt, const steady_clock::time_ << m_iSndCurrSeqNo << " by " << (CSeqNo::seqoff(m_iSndCurrSeqNo, ackdata_seqno) - 1) << "!"); m_bBroken = true; m_iBrokenCounter = 0; + setAgentCloseReason(SRT_CLS_IPE); return; } @@ -8742,6 +8754,7 @@ void srt::CUDT::processCtrlLossReport(const CPacket& ctrlpkt) // this should not happen: attack or bug m_bBroken = true; m_iBrokenCounter = 0; + setAgentCloseReason(SRT_CLS_ROGUE); return; } @@ -8921,8 +8934,31 @@ void srt::CUDT::processCtrlDropReq(const CPacket& ctrlpkt) } } -void srt::CUDT::processCtrlShutdown() +void srt::CUDT::processCtrlShutdown(const CPacket& ctrlpkt) { + const uint32_t* data = (const uint32_t*) ctrlpkt.m_pcData; + const size_t data_len = ctrlpkt.getLength() / 4; + + int reason = 0; + + // This condition should be ALWAYS satisfied, it's only + // a sanity check before reading the data. Versions that + // do not support close reason will simply send 0 here because + // it's the padding 0 that is provided in every command + // that is not expected to carry any "body". It is acceptable + // that the old versions simply send 0 here, but then you + // can't have the UNKNOWN value in any of close reason + // fields because it means that it wasn't set. + if (data_len > 0) + { + reason = data[0]; + } + + if (reason == 0) + { + setPeerCloseReason(SRT_CLS_FALLBACK); + } + m_bShutdown = true; m_bClosing = true; m_bBroken = true; @@ -9008,7 +9044,7 @@ void srt::CUDT::processCtrl(const CPacket &ctrlpkt) break; case UMSG_SHUTDOWN: // 101 - Shutdown - processCtrlShutdown(); + processCtrlShutdown(ctrlpkt); break; case UMSG_DROPREQ: // 111 - Msg drop request @@ -9697,10 +9733,12 @@ bool srt::CUDT::packUniqueData(CPacket& w_packet) return true; } -// This is a close request, but called from the +// This is a close request, but called from the handler of the +// buffer overflow in live mode. void srt::CUDT::processClose() { - sendCtrl(UMSG_SHUTDOWN); + uint32_t res[1] = { SRT_CLS_OVERFLOW }; + sendCtrl(UMSG_SHUTDOWN, NULL, res, sizeof res); m_bShutdown = true; m_bClosing = true; @@ -11265,6 +11303,7 @@ bool srt::CUDT::checkExpTimer(const steady_clock::time_point& currtime, int chec if (m_bBreakAsUnstable || ((m_iEXPCount > COMM_RESPONSE_MAX_EXP) && (currtime - last_rsp_time > microseconds_from(PEER_IDLE_TMO_US)))) { + setAgentCloseReason(SRT_CLS_PEERIDLE); // // Connection is broken. // UDT does not signal any information about this instead of to stop quietly. @@ -11755,3 +11794,34 @@ void srt::CUDT::processKeepalive(const CPacket& ctrlpkt, const time_point& tsArr if (m_config.bDriftTracer) m_pRcvBuffer->addRcvTsbPdDriftSample(ctrlpkt.getMsgTimeStamp(), tsArrival, -1); } + +// This function should be called when closing the socket internally. +void srt::CUDT::setAgentCloseReason(int reason) +{ + m_AgentCloseReason.compare_exchange(SRT_CLS_UNKNOWN, reason); + + // Do not touch m_PeerCloseReason, it should remain SRT_CLS_UNKNOWN. + // If this reason is already set to some value, then m_AgentCloseReason + // should have been already set to SRT_CLS_PEER. + + m_CloseTimeStamp.compare_exchange(time_point(), steady_clock::now()); +} + +// This function should be called in a handler of UMSG_SHUTDOWN. +void srt::CUDT::setPeerCloseReason(int reason) +{ + m_AgentCloseReason.compare_exchange(SRT_CLS_UNKNOWN, SRT_CLS_PEER); + if (m_AgentCloseReason == SRT_CLS_PEER) + { + m_PeerCloseReason.compare_exchange(SRT_CLS_UNKNOWN, reason); + + m_CloseTimeStamp.compare_exchange(time_point(), steady_clock::now()); + } +} + +void srt::CUDT::copyCloseInfo(SRT_CLOSE_INFO& info) +{ + info.agent = SRT_CLOSE_REASON(m_AgentCloseReason.load()); + info.peer = SRT_CLOSE_REASON(m_PeerCloseReason.load()); + info.time = m_CloseTimeStamp.load().time_since_epoch().count(); +} diff --git a/srtcore/core.h b/srtcore/core.h index e7ca57c50..7df4c1cdc 100644 --- a/srtcore/core.h +++ b/srtcore/core.h @@ -205,7 +205,7 @@ class CUDT #if ENABLE_BONDING static int connectLinks(SRTSOCKET grp, SRT_SOCKGROUPCONFIG links [], int arraysize); #endif - static int close(SRTSOCKET u); + static int close(SRTSOCKET u, int reason); static int getpeername(SRTSOCKET u, sockaddr* name, int* namelen); static int getsockname(SRTSOCKET u, sockaddr* name, int* namelen); static int getsockopt(SRTSOCKET u, int level, SRT_SOCKOPT optname, void* optval, int* optlen); @@ -432,7 +432,11 @@ class CUDT SRTU_PROPERTY_RR(sync::Condition*, recvTsbPdCond, &m_RcvTsbPdCond); /// @brief Request a socket to be broken due to too long instability (normally by a group). - void breakAsUnstable() { m_bBreakAsUnstable = true; } + void breakAsUnstable() + { + m_bBreakAsUnstable = true; + setAgentCloseReason(SRT_CLS_UNSTABLE); + } void ConnectSignal(ETransmissionEvent tev, EventSlot sl); void DisconnectSignal(ETransmissionEvent tev); @@ -573,10 +577,13 @@ class CUDT /// Close the opened UDT entity. - bool closeInternal(); + bool closeInternal(int reason); void updateBrokenConnection(); void completeBrokenConnectionDependencies(int errorcode); + void setAgentCloseReason(int reason); + void setPeerCloseReason(int reason); + /// Request UDT to send out a data block "data" with size of "len". /// @param data [in] The address of the application data to be sent. /// @param len [in] The size of the data block. @@ -793,6 +800,15 @@ class CUDT sync::atomic m_bBreakAsUnstable; // A flag indicating that the socket should become broken because it has been unstable for too long. sync::atomic m_bPeerHealth; // If the peer status is normal sync::atomic m_RejectReason; + + // If the socket was closed by some reason locally, the reason is + // in m_AgentCloseReason and the m_PeerCloseReason is then SRT_CLS_UNKNOWN. + // If the socket was closed due to reception of UMSG_SHUTDOWN, the reason + // exctracted from the message is written to m_PeerCloseReason and the + // m_AgentCloseReason == SRT_CLS_PEER. + sync::atomic m_AgentCloseReason; + sync::atomic m_PeerCloseReason; + atomic_time_point m_CloseTimeStamp; // Time when the close reason was first set bool m_bOpened; // If the UDT entity has been opened // A counter (number of GC checks happening every 1s) to let the GC tag this socket as closed. sync::atomic m_iBrokenCounter; // If a broken socket still has data in the receiver buffer, it is not marked closed until the counter is 0. @@ -1042,7 +1058,7 @@ class CUDT void processCtrlDropReq(const CPacket& ctrlpkt); /// @brief Process incoming shutdown control packet - void processCtrlShutdown(); + void processCtrlShutdown(const CPacket& ctrlpkt); /// @brief Process incoming user defined control packet /// @param ctrlpkt incoming user defined packet void processCtrlUserDefined(const CPacket& ctrlpkt); @@ -1145,6 +1161,8 @@ class CUDT static const int SEND_LITE_ACK = sizeof(int32_t); // special size for ack containing only ack seq static const int PACKETPAIR_MASK = 0xF; + void copyCloseInfo(SRT_CLOSE_INFO&); + private: // Timers functions #if ENABLE_BONDING time_point m_tsFreshActivation; // GROUPS: time of fresh activation of the link, or 0 if past the activation phase or idle diff --git a/srtcore/group.cpp b/srtcore/group.cpp index f4dfba1ba..835248968 100644 --- a/srtcore/group.cpp +++ b/srtcore/group.cpp @@ -944,7 +944,7 @@ void CUDTGroup::close() { try { - CUDT::uglobal().close(*i); + CUDT::uglobal().close(*i, SRT_CLS_INTERNAL); } catch (CUDTException&) { @@ -2231,6 +2231,7 @@ int CUDTGroup::recv(char* buf, int len, SRT_MSGCTRL& w_mc) LOGC(grlog.Error, log << "grp/recv: $" << id() << ": @" << ps->m_SocketID << ": SEQUENCE DISCREPANCY: base=%" << m_RcvBaseSeqNo << " vs pkt=%" << info.seqno << ", setting ESECFAIL"); + ps->core().setAgentCloseReason(SRT_CLS_ROGUE); ps->core().m_bBroken = true; broken.insert(ps); continue; @@ -3350,7 +3351,7 @@ void CUDTGroup::sendBackup_RetryWaitBlocked(SendBackupCtx& w_sendBackupCtx HLOGC(gslog.Debug, log << "grp/sendBackup: swait/ex on @" << (id) << " while waiting for any writable socket - CLOSING"); - CUDT::uglobal().close(s); // << LOCKS m_GlobControlLock, then GroupLock! + CUDT::uglobal().close(s, SRT_CLS_INTERNAL); // << LOCKS m_GlobControlLock, then GroupLock! } else { diff --git a/srtcore/packet.cpp b/srtcore/packet.cpp index 33555e7bb..c0be3aa2b 100644 --- a/srtcore/packet.cpp +++ b/srtcore/packet.cpp @@ -391,7 +391,7 @@ void CPacket::pack(UDTMessageType pkttype, const int32_t* lparam, void* rparam, case UMSG_SHUTDOWN: // 0101 - Shutdown // control info field should be none // but "writev" does not allow this - m_PacketVector[PV_DATA].set((void*)&m_extra_pad, 4); + m_PacketVector[PV_DATA].set(rparam, size); break; diff --git a/srtcore/queue.cpp b/srtcore/queue.cpp index 4282965b4..9f5b178ce 100644 --- a/srtcore/queue.cpp +++ b/srtcore/queue.cpp @@ -963,7 +963,8 @@ void srt::CRendezvousQueue::updateConnStatus(EReadStatus rst, EConnectStatus cst LinkStatusInfo fi = *i; fi.errorcode = SRT_ECONNREJ; toRemove.push_back(fi); - i->u->sendCtrl(UMSG_SHUTDOWN); + uint32_t res[1] = {SRT_CLS_DEADLSN}; + i->u->sendCtrl(UMSG_SHUTDOWN, NULL, res, sizeof res); } } diff --git a/srtcore/srt.h b/srtcore/srt.h index c30169f05..e36339d06 100644 --- a/srtcore/srt.h +++ b/srtcore/srt.h @@ -575,6 +575,37 @@ enum SRT_REJECT_REASON #define SRT_REJC_USERDEFINED 2000 // User defined error codes +enum SRT_CLOSE_REASON +{ + SRT_CLS_UNKNOWN, // Unset + SRT_CLS_INTERNAL, // Closed by internal reasons during connection attempt + SRT_CLS_PEER, // Received SHUTDOWN message from the peer + SRT_CLS_RESOURCE, // Problem with resource allocation + SRT_CLS_ROGUE, // Received wrong data in the packet + SRT_CLS_OVERFLOW, // Emergency close due to receiver buffer overflow + SRT_CLS_IPE, // Internal program error + SRT_CLS_API, // The application called srt_close(). + SRT_CLS_FALLBACK, // Used for peer that do not support close reason featur. + SRT_CLS_LATE, // Accepted-socket late-rejection or in-handshake rollback + SRT_CLS_CLEANUP, // All sockets are being closed due to srt_cleanup() call + SRT_CLS_DEADLSN, // This is an accepted socket off a dead listener + SRT_CLS_PEERIDLE, // Peer didn't send any packet for a time of SRTO_PEERIDLETIMEO + SRT_CLS_UNSTABLE, // Requested to be broken as unstable in Backup group + + SRT_CLS_E_SIZE +}; + +typedef struct SRT_CLOSE_INFO +{ + enum SRT_CLOSE_REASON agent; + enum SRT_CLOSE_REASON peer; + int64_t time; +} SRT_CLOSE_INFO; + +#define SRT_CLSC_INTERNAL 0 +#define SRT_CLSC_USER 100 + + // Logging API - specialization for SRT. // WARNING: This part is generated. @@ -783,6 +814,8 @@ SRT_API int srt_rendezvous (SRTSOCKET u, const struct sockaddr* local_na const struct sockaddr* remote_name, int remote_namelen); SRT_API int srt_close (SRTSOCKET u); +SRT_API int srt_close_withreason(SRTSOCKET u, int reason); +SRT_API int srt_close_getreason(SRTSOCKET u, SRT_CLOSE_INFO* info); SRT_API int srt_getpeername (SRTSOCKET u, struct sockaddr* name, int* namelen); SRT_API int srt_getsockname (SRTSOCKET u, struct sockaddr* name, int* namelen); SRT_API int srt_getsockopt (SRTSOCKET u, int level /*ignored*/, SRT_SOCKOPT optname, void* optval, int* optlen); diff --git a/srtcore/srt_c_api.cpp b/srtcore/srt_c_api.cpp index 885c80068..ac65aac9c 100644 --- a/srtcore/srt_c_api.cpp +++ b/srtcore/srt_c_api.cpp @@ -21,6 +21,7 @@ written by #include "common.h" #include "packet.h" #include "core.h" +#include "api.h" #include "utilities.h" using namespace std; @@ -154,7 +155,33 @@ int srt_close(SRTSOCKET u) return 0; } - return CUDT::close(u); + return CUDT::close(u, SRT_CLS_API); +} + +int srt_close_withreason(SRTSOCKET u, int reason) +{ + SRT_SOCKSTATUS st = srt_getsockstate(u); + + if ((st == SRTS_NONEXIST) || + (st == SRTS_CLOSED) || + (st == SRTS_CLOSING) ) + { + // It's closed already. Do nothing. + return 0; + } + + if (reason < SRT_CLSC_USER) + reason = SRT_CLS_API; + + return CUDT::close(u, reason); +} + +int srt_close_getreason(SRTSOCKET u, SRT_CLOSE_INFO* info) +{ + if (!info || u == SRT_INVALID_SOCK) + return -1; + + return CUDT::uglobal().getCloseReason(u, *info); } int srt_getpeername(SRTSOCKET u, struct sockaddr * name, int * namelen) { return CUDT::getpeername(u, name, namelen); } diff --git a/testing/testmedia.cpp b/testing/testmedia.cpp index 34d2bdc9b..505614144 100755 --- a/testing/testmedia.cpp +++ b/testing/testmedia.cpp @@ -62,6 +62,50 @@ bool transmit_use_sourcetime = false; int transmit_retry_connect = 0; bool transmit_retry_always = false; +struct CloseReasonMap +{ + map at; + + CloseReasonMap() + { + at[SRT_CLS_UNKNOWN] = "Unset"; + at[SRT_CLS_INTERNAL] = "Closed by internal reasons during connection attempt"; + at[SRT_CLS_PEER] = "Received SHUTDOWN message from the peer"; + at[SRT_CLS_RESOURCE] = "Problem with resource allocation"; + at[SRT_CLS_ROGUE] = "Received wrong data in the packet"; + at[SRT_CLS_OVERFLOW] = "Emergency close due to receiver buffer overflow"; + at[SRT_CLS_IPE] = "Internal program error"; + at[SRT_CLS_API] = "The application called srt_close()."; + at[SRT_CLS_FALLBACK] = "The peer doesn't support close reason feature."; + at[SRT_CLS_LATE] = "Accepted-socket late-rejection or in-handshake rollback"; + at[SRT_CLS_CLEANUP] = "All sockets are being closed due to srt_cleanup() call"; + at[SRT_CLS_DEADLSN] = "This was an accepted socket off a dead listener"; + at[SRT_CLS_PEERIDLE] = "Peer didn't send any packet for a time of SRTO_PEERIDLETIMEO"; + at[SRT_CLS_UNSTABLE] = "Requested to be broken as unstable in Backup group"; + } + + string operator[](SRT_CLOSE_REASON reason) + { + if (int(reason) >= SRT_CLSC_USER) + { + string extra; + if (reason == SRT_CLSC_USER) + extra = " - Application exit due to interrupted transmission"; + + if (reason == SRT_CLSC_USER + 1) + extra = " - Error during configuration, transmission not started"; + + return Sprint("User-defined reason #", reason - SRT_CLSC_USER, extra); + } + + auto p = at.find(reason); + if (p == at.end()) + return "UNDEFINED"; + return p->second; + } + +} g_close_reason; + // Do not unblock. Copy this to an app that uses applog and set appropriate name. //srt_logging::Logger applog(SRT_LOGFA_APP, srt_logger_config, "srt-test"); @@ -706,7 +750,7 @@ void SrtCommon::Init(string host, int port, string path, map par, if (m_bindsock != SRT_INVALID_SOCK) srt_close(m_bindsock); if (m_sock != SRT_INVALID_SOCK) - srt_close(m_sock); + srt_close_withreason(m_sock, SRT_CLSC_USER+1); m_sock = m_bindsock = SRT_INVALID_SOCK; throw; } @@ -1328,7 +1372,7 @@ void SrtCommon::ConnectClient(string host, int port) continue; } - srt_close(m_sock); + srt_close_withreason(m_sock, SRT_CLSC_USER+1); Error("srt_connect", reason); } break; @@ -1394,6 +1438,9 @@ void SrtCommon::ConnectClient(string host, int port) void SrtCommon::Error(string src, int reason, int force_result) { + SRT_CLOSE_INFO cli; + int cls = srt_close_getreason(m_sock, &cli); + int errnov = 0; const int result = force_result == 0 ? srt_getlasterror(&errnov) : force_result; if (result == SRT_SUCCESS) @@ -1416,9 +1463,23 @@ void SrtCommon::Error(string src, int reason, int force_result) else { if ( Verbose::on ) - Verb() << "FAILURE\n" << src << ": [" << result << "." << errnov << "] " << message; + Verb() << "FAILURE\n" << src << ": [" << result << "." << errnov << "] " << message; else - cerr << "\nERROR #" << result << "." << errnov << ": " << message << endl; + cerr << "\nERROR #" << result << "." << errnov << ": " << message << endl; + } + + if (cls == 0) + { + int64_t srt_now = srt_time_now(); + int64_t ago = srt_now - cli.time; + cerr << "CLOSE REASON: ->\n\tagent=" << cli.agent << " [" << g_close_reason[cli.agent] + << "]\n\tpeer=" << cli.peer << " [" << g_close_reason[cli.peer] + << "]\n\ttime=" << cli.time << " (" << fixed << (ago/1000000.0) << "s ago)" + << endl; + } + else + { + cerr << "(CLOSE REASON not found)\n"; } throw TransmissionError("error: " + src + ": " + message); @@ -1448,7 +1509,7 @@ void SrtCommon::SetupRendezvous(string adapter, string host, int port) int stat = srt_bind(m_sock, localsa.get(), localsa.size()); if (stat == SRT_ERROR) { - srt_close(m_sock); + srt_close_withreason(m_sock, SRT_CLSC_USER+1); Error("srt_bind"); } } @@ -1465,8 +1526,21 @@ void SrtCommon::Close() { Verb() << "SrtCommon: DESTROYING CONNECTION, closing socket (rt%" << m_sock << ")..."; srt_setsockflag(m_sock, SRTO_SNDSYN, &yes, sizeof yes); - srt_close(m_sock); + srt_close_withreason(m_sock, SRT_CLSC_USER); any = true; + + SRT_CLOSE_INFO cli; + int cls = srt_close_getreason(m_sock, &cli); + + if (cls == 0) + { + int64_t srt_now = srt_time_now(); + int64_t ago = srt_now - cli.time; + cerr << "POST-FACTUM CLOSE REASON: ->\n\tagent=" << cli.agent << " [" << g_close_reason[cli.agent] + << "]\n\tpeer=" << cli.peer << " [" << g_close_reason[cli.peer] + << "]\n\ttime=" << cli.time << " (" << fixed << (ago/1000000.0) << "s ago)" + << endl; + } } if (m_bindsock != SRT_INVALID_SOCK) @@ -1474,7 +1548,7 @@ void SrtCommon::Close() Verb() << "SrtCommon: DESTROYING SERVER, closing socket (ls%" << m_bindsock << ")..."; // Set sndsynchro to the socket to synch-close it. srt_setsockflag(m_bindsock, SRTO_SNDSYN, &yes, sizeof yes); - srt_close(m_bindsock); + srt_close_withreason(m_bindsock, SRT_CLSC_USER); any = true; }