Fix memory leak

[xiaozhihong]

No description provided.

[maxsharabayko]

The fix is correct. A couple of suggestions to improve the code.

TODO

• CPacket::clone() must set m_data_owned to true. Preferably use CPacket::allocate(size_t) instead of manually handling memory allocation.
• In this PR and in CRcvQueue::recvfrom(..) rely on CPacket destructor instead of delete[] newpkt->m_pcData;.
• CRcvQueue::storePkt(int32_t id, CPacket* pkt) expects the packet will be owned, but provides no clue to this. What if it clones the packet inside, and the packet is given by const reference? storPkt -> storeClonePkt?

[ethouris]

Just a side note, should that help understand this code.

This storage is used for the initial period, when the case of "confused connection state" happens. This situation happens in case when you have a caller-listener arrangement, the LISTENER side is about to send data, and the flow is:

1. induction handshake exchange, listener receives conclusion handshake request from the caller
2. Listener: send conclusion handshake response - this packet gets LOST
3. Listener: considers itself connected so it starts sending the data
4. Caller: still waiting for conclusion handshake response, but instead received a data packet

In this case Caller will repeat sending the conclusion handshake request and wait for the listener's response, but in the meantime it must do something about these data packets. Therefore these packets are stored in a special spare storage, from which they will be delivered directly to the receiver buffer once the caller received finally the handshake conclusion response packet.

I don't know what the state of this feature really is (if anyone fixed it, then I missed it), but it exists since UDT codebase and it didn't work there. That is, the storage was being filled with incoming packets, but I didn't see any code that re-delivers these packets. Instead, all packets between ISN and the last received one before reception of the final handshake packet are simply considered lost and get retransmitted.

EDIT:

I took a look at the code and the only place where the CRcvQueue::recvfrom function is called is CUDT::startConnect, and only in blocking mode. It seems to be then a legitimate intermediate place where CONTROL (not DATA) packets are to be kept to be next picked up by the blocking-mode connection function for handling the handshake roundtrip. Might be that it was intended to serve both purposes and only one remained.

Ah, and as per possible refactoring there: note that this feature is older than CPacket::m_data_owned field, which I introduced once in order to handle packet objects with "its own" memory that it should delete by itself (unlike packets that get memory "borrowed" from CUnitQueue). That's why "manual deletion" is in CRcvQueue::recvfrom, CRcvQueue::~CRcvQueue and likely is also required in the place you delivered the fix for. It would be definitely nice to use this field to clean things up, but I'm afraid there's here much more to do than just ordering up the allocation/deletion things.

[maxsharabayko]

More fixes after my changes are required.

diff --git a/srtcore/packet.cpp b/srtcore/packet.cpp
index 33555e7b..3a17c44b 100644
--- a/srtcore/packet.cpp
+++ b/srtcore/packet.cpp
@@ -221,6 +221,7 @@ void CPacket::deallocate()
     if (m_data_owned)
         delete[](char*) m_PacketVector[PV_DATA].data();
     m_PacketVector[PV_DATA].set(NULL, 0);
+    m_data_owned = false;
 }
 
 char* CPacket::release()
@@ -241,8 +242,7 @@ CPacket::~CPacket()
 {
     // PV_HEADER is always owned, PV_DATA may use a "borrowed" buffer.
     // Delete the internal buffer only if it was declared as owned.
-    if (m_data_owned)
-        delete[](char*) m_PacketVector[PV_DATA].data();
+    deallocate();
 }
 
 size_t CPacket::getLength() const
@@ -561,10 +561,8 @@ CPacket* CPacket::clone() const
 {
     CPacket* pkt = new CPacket;
     memcpy((pkt->m_nHeader), m_nHeader, HDR_SIZE);
-    pkt->m_pcData = new char[m_PacketVector[PV_DATA].size()];
+    pkt->allocate(m_PacketVector[PV_DATA].size());
     memcpy((pkt->m_pcData), m_pcData, m_PacketVector[PV_DATA].size());
-    pkt->m_PacketVector[PV_DATA].setLength(m_PacketVector[PV_DATA].size());
-
     pkt->m_DestAddr = m_DestAddr;
 
     return pkt;
diff --git a/srtcore/queue.cpp b/srtcore/queue.cpp
index 4282965b..8db669f7 100644
--- a/srtcore/queue.cpp
+++ b/srtcore/queue.cpp
@@ -1162,7 +1162,6 @@ srt::CRcvQueue::~CRcvQueue()
         while (!i->second.empty())
         {
             CPacket* pkt = i->second.front();
-            delete[] pkt->m_pcData;
             delete pkt;
             i->second.pop();
         }
@@ -1365,14 +1364,12 @@ srt::EReadStatus srt::CRcvQueue::worker_RetrieveUnit(int32_t& w_id, CUnit*& w_un
     {
         // no space, skip this packet
         CPacket temp;
-        temp.m_pcData = new char[m_szPayloadSize];
-        temp.setLength(m_szPayloadSize);
+        temp.allocate(m_szPayloadSize);
         THREAD_PAUSED();
         EReadStatus rst = m_pChannel->recvfrom((w_addr), (temp));
         THREAD_RESUMED();
         // Note: this will print nothing about the packet details unless heavy logging is on.
         LOGC(qrlog.Error, log << CONID() << "LOCAL STORAGE DEPLETED. Dropping 1 packet: " << temp.Info());
-        delete[] temp.m_pcData;
 
         // Be transparent for RST_ERROR, but ignore the correct
         // data read and fake that the packet was dropped.
@@ -1680,7 +1677,6 @@ int srt::CRcvQueue::recvfrom(int32_t id, CPacket& w_packet)
     w_packet.setLength(newpkt->getLength());
     w_packet.m_DestAddr = newpkt->m_DestAddr;
 
-    delete[] newpkt->m_pcData;
     delete newpkt;
 
     // remove this message from queue,
@@ -1735,7 +1731,6 @@ void srt::CRcvQueue::removeConnector(const SRTSOCKET& id)
               log << "removeConnector: ... and its packet queue with " << i->second.size() << " packets collected");
         while (!i->second.empty())
         {
-            delete[] i->second.front()->m_pcData;
             delete i->second.front();
             i->second.pop();
         }
@@ -1783,7 +1778,10 @@ void srt::CRcvQueue::storePkt(int32_t id, CPacket* pkt)
     {
         // avoid storing too many packets, in case of malfunction or attack
         if (i->second.size() > 16)
+        {
+            delete pkt;
             return;
+        }
 
         i->second.push(pkt);
     }

[maxsharabayko]

Reverted my changes. Too much refactoring hides the bugfix itself. The refactoring is to be moved to a separate PR.

diff --git a/srtcore/packet.cpp b/srtcore/packet.cpp
index 33555e7..3a17c44 100644
--- a/srtcore/packet.cpp
+++ b/srtcore/packet.cpp
@@ -221,6 +221,7 @@ void CPacket::deallocate()
     if (m_data_owned)
         delete[](char*) m_PacketVector[PV_DATA].data();
     m_PacketVector[PV_DATA].set(NULL, 0);
+    m_data_owned = false;
 }
 
 char* CPacket::release()
@@ -241,8 +242,7 @@ CPacket::~CPacket()
 {
     // PV_HEADER is always owned, PV_DATA may use a "borrowed" buffer.
     // Delete the internal buffer only if it was declared as owned.
-    if (m_data_owned)
-        delete[](char*) m_PacketVector[PV_DATA].data();
+    deallocate();
 }
 
 size_t CPacket::getLength() const
@@ -561,10 +561,8 @@ CPacket* CPacket::clone() const
 {
     CPacket* pkt = new CPacket;
     memcpy((pkt->m_nHeader), m_nHeader, HDR_SIZE);
-    pkt->m_pcData = new char[m_PacketVector[PV_DATA].size()];
+    pkt->allocate(m_PacketVector[PV_DATA].size());
     memcpy((pkt->m_pcData), m_pcData, m_PacketVector[PV_DATA].size());
-    pkt->m_PacketVector[PV_DATA].setLength(m_PacketVector[PV_DATA].size());
-
     pkt->m_DestAddr = m_DestAddr;
 
     return pkt;
diff --git a/srtcore/queue.cpp b/srtcore/queue.cpp
index 9b13a62..b8e3f07 100644
--- a/srtcore/queue.cpp
+++ b/srtcore/queue.cpp
@@ -1162,7 +1162,6 @@ srt::CRcvQueue::~CRcvQueue()
         while (!i->second.empty())
         {
             CPacket* pkt = i->second.front();
-            delete[] pkt->m_pcData;
             delete pkt;
             i->second.pop();
         }
@@ -1365,14 +1364,12 @@ srt::EReadStatus srt::CRcvQueue::worker_RetrieveUnit(int32_t& w_id, CUnit*& w_un
     {
         // no space, skip this packet
         CPacket temp;
-        temp.m_pcData = new char[m_szPayloadSize];
-        temp.setLength(m_szPayloadSize);
+        temp.allocate(m_szPayloadSize);
         THREAD_PAUSED();
         EReadStatus rst = m_pChannel->recvfrom((w_addr), (temp));
         THREAD_RESUMED();
         // Note: this will print nothing about the packet details unless heavy logging is on.
         LOGC(qrlog.Error, log << CONID() << "LOCAL STORAGE DEPLETED. Dropping 1 packet: " << temp.Info());
-        delete[] temp.m_pcData;
 
         // Be transparent for RST_ERROR, but ignore the correct
         // data read and fake that the packet was dropped.
@@ -1680,7 +1677,6 @@ int srt::CRcvQueue::recvfrom(int32_t id, CPacket& w_packet)
     w_packet.setLength(newpkt->getLength());
     w_packet.m_DestAddr = newpkt->m_DestAddr;
 
-    delete[] newpkt->m_pcData;
     delete newpkt;
 
     // remove this message from queue,
@@ -1735,7 +1731,6 @@ void srt::CRcvQueue::removeConnector(const SRTSOCKET& id)
               log << "removeConnector: ... and its packet queue with " << i->second.size() << " packets collected");
         while (!i->second.empty())
         {
-            delete[] i->second.front()->m_pcData;
             delete i->second.front();
             i->second.pop();
         }
@@ -1781,10 +1776,9 @@ void srt::CRcvQueue::storePkt(int32_t id, CPacket* pkt)
     }
     else
     {
-        // avoid storing too many packets, in case of malfunction or attack
+        // Avoid storing too many packets, in case of malfunction or attack.
         if (i->second.size() > 16)
         {
-            delete[] pkt->m_pcData;
             delete pkt;
             return;
         }
@@ -1795,7 +1789,7 @@ void srt::CRcvQueue::storePkt(int32_t id, CPacket* pkt)
 
 void srt::CMultiplexer::destroy()
 {
-    // Reverse order of the assigned
+    // Reverse order of the assigned.
     delete m_pRcvQueue;
     delete m_pSndQueue;
     delete m_pTimer;