keep the entire packet alive (attached to GC root)

splitice · splitice · commit 60d9bbcc7174 · 2022-01-31T17:41:47.000+11:00
diff --git a/src/quic/quic.cc b/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(
diff --git a/src/quic/quic.h b/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
diff --git a/src/quic/session.cc b/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");
diff --git a/src/quic/stream.cc b/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,
diff --git a/src/quic/stream.h b/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);
 

Original file line number	Diff line number	Diff line change
`@@ -225,6 +225,7 @@ bool PreferredAddress::Resolve(`
`225`	`225`
`226`	`226`	`void Packet::MemoryInfo(MemoryTracker* tracker) const {`
`227`	`227`	`tracker->TrackFieldWithSize("allocated", ptr_ != data_ ? len_ : 0);`
	`228`	`+ tracker->TrackField("retained", retained);`
`228`	`229`	`}`
`229`	`230`
`230`	`231`	`Path::Path(`