keep the entire packet alive (attached to GC root)

Author: splitice

src/quic/quic.cc

@@ -225,6 +225,7 @@ bool PreferredAddress::Resolve(
 
 void Packet::MemoryInfo(MemoryTracker* tracker) const {
   tracker->TrackFieldWithSize("allocated", ptr_ != data_ ? len_ : 0);
+  tracker->TrackField("retained", retained);
 }
 
 Path::Path(

src/quic/quic.h

@@ -398,11 +398,16 @@ class Packet final : public MemoryRetainer {
   SET_MEMORY_INFO_NAME(Packet);
   SET_SELF_SIZE(Packet);
 
+  inline void AddRetained(BaseObjectPtr<BaseObject> retain){
+    retained.push_back(retain);
+  }
+
  private:
   uint8_t data_[kDefaultMaxPacketLength];
   uint8_t* ptr_ = nullptr;
   size_t len_ = kDefaultMaxPacketLength;
   const char* diagnostic_label_ = nullptr;
+  std::vector<BaseObjectPtr<BaseObject>> retained;
 };
 
 // A utility class that wraps ngtcp2_path to adapt it to work with SocketAddress

src/quic/session.cc

@@ -3241,20 +3241,21 @@ bool Session::Application::SendPendingData() {
       return false;
     }
 
-    // If stream_data.id is -1, then we're not serializing any data for any
-    // specific stream. We still need to process QUIC session packets tho.
-    if (stream_data.id > -1)
-      Debug(session(), "Serializing packets for stream id %" PRId64,
-            stream_data.id);
-    else
-      Debug(session(), "Serializing session packets");
-
     // If the packet was sent previously, then packet will have been reset.
-    if (!packet) {
+    if (!pos) {
       packet = CreateStreamDataPacket();
       pos = packet->data();
     }
 
+    // If stream_data.id is -1, then we're not serializing any data for any
+    // specific stream. We still need to process QUIC session packets tho.
+    if (stream_data.id > -1) {
+      Debug(session(), "Serializing packets for stream id %" PRId64,
+            stream_data.id);
+      packet->AddRetained(stream_data.stream->GetOutboundSource());
+    } else
+      Debug(session(), "Requesting serialized packet flush");
+
     ssize_t nwrite = WriteVStream(&path, pos, &ndatalen, stream_data);
     if (stream_data.id >= 0) {
       Debug(session(),
@@ -3312,25 +3313,26 @@ bool Session::Application::SendPendingData() {
           continue;
       }
 
-      if(nwrite != 0){
+      if(nwrite != 0){ // -ve response i.e error
         packet.reset();
         session()->set_last_error(kQuicInternalError);
         return false;
-      } else {
-          if (stream_data.id >= 0)
-            ResumeStream(stream_data.id);
-
-          // We are either congestion limited or done.
-          if (pos - packet->data()) {
-            // Some data was serialized into the packet. We need to send it.
-            packet->set_length(pos - packet->data());
-            Debug(session(), "Congestion limited, but %" PRIu64 " bytes pending",
-                  packet->length());
-            if (!session()->SendPacket(std::move(packet), path))
-              return false;
-          }
-          return true;
       }
+      
+      // 0 bytes in this sending operation
+      if (stream_data.id >= 0)
+        ResumeStream(stream_data.id);
+
+      // We are either congestion limited or done.
+      if (pos - packet->data()) {
+        // Some data was serialized into the packet. We need to send it.
+        packet->set_length(pos - packet->data());
+        Debug(session(), "Congestion limited, but %" PRIu64 " bytes pending",
+              packet->length());
+        if (!session()->SendPacket(std::move(packet), path))
+          return false;
+      }
+      return true;
     }
 
     pos += nwrite;
@@ -3346,7 +3348,6 @@ bool Session::Application::SendPendingData() {
       Debug(session(), "-- Failed to send packet");
       return false;
     }
-    packet.reset();
     pos = nullptr;
     if (++packets_sent == kMaxPackets) {
       Debug(session(), "-- Max packets sent");

src/quic/stream.cc

@@ -336,6 +336,10 @@ void Stream::Commit(size_t amount) {
   CHECK_LE(actual, amount);
 }
 
+BaseObjectPtr<BaseObject> Stream::GetOutboundSource() const {
+  return outbound_source_strong_ptr_;
+}
+
 int Stream::DoPull(
     bob::Next<ngtcp2_vec> next,
     int options,

src/quic/stream.h

@@ -187,6 +187,8 @@ class Stream final : public AsyncWrap,
   // Attaches an outbound Buffer::Source
   void AttachOutboundSource(Buffer::Source* source);
 
+  BaseObjectPtr<BaseObject> GetOutboundSource() const;
+
   // Signals the beginning of a new block of headers.
   void BeginHeaders(HeadersKind kind);