patchset 3

src/quic/buffer.cc

@@ -150,6 +150,7 @@ size_t Buffer::Chunk::Seek(size_t amount) {
 size_t Buffer::Chunk::Acknowledge(size_t amount) {
   amount = std::min(amount, unacknowledged_);
   unacknowledged_ -= amount;
+  CHECK_GE(unacknowledged_, 0);
   return amount;
 }
 

src/quic/endpoint.cc

@@ -496,6 +496,10 @@ Endpoint::SendWrap::SendWrap(
   MakeWeak();
 }
 
+Endpoint::SendWrap::~SendWrap(){
+  packet_.reset();
+}
+
 void Endpoint::SendWrap::MemoryInfo(MemoryTracker* tracker) const {
   tracker->TrackField("destination", destination_);
   tracker->TrackField("packet", packet_);
@@ -864,15 +868,14 @@ bool Endpoint::MaybeStatelessReset(
 }
 
 uv_buf_t Endpoint::OnAlloc(size_t suggested_size) {
-  return AllocatedBuffer::AllocateManaged(env(), suggested_size).release();
+  return env()->allocate_managed_buffer(suggested_size);
 }
 
 void Endpoint::OnReceive(
     size_t nread,
-    const uv_buf_t& buf,
+    std::shared_ptr<BackingStore> store,
     const std::shared_ptr<SocketAddress>& remote_address) {
 
-  AllocatedBuffer buffer(env(), buf);
   // When diagnostic packet loss is enabled, the packet will be randomly
   // dropped based on the rx_loss probability.
   if (UNLIKELY(is_diagnostic_packet_loss(config_.rx_loss))) {
@@ -888,8 +891,6 @@ void Endpoint::OnReceive(
   // }
 
   IncrementStat(&EndpointStats::bytes_received, nread);
-
-  std::shared_ptr<BackingStore> store = buffer.ReleaseBackingStore();
   if (UNLIKELY(!store)) {
     // TODO(@jasnell): Send immediate close?
     ProcessReceiveFailure(UV_ENOMEM);
@@ -1496,15 +1497,31 @@ void Endpoint::UDP::OnReceive(
     const uv_buf_t* buf,
     const sockaddr* addr,
     unsigned int flags) {
+      OnReceive(handle, nread, *buf, addr, flags);
+}
+
+void Endpoint::UDP::OnReceive(
+    uv_udp_t* handle,
+    ssize_t nread,
+    const uv_buf_t& buf,
+    const sockaddr* addr,
+    unsigned int flags) {
+
   UDP* udp = ContainerOf(&Endpoint::UDP::handle_, handle);
+
+  std::unique_ptr<BackingStore> store = udp->env()->release_managed_buffer(buf);
+
   if (nread < 0) {
     udp->endpoint_->ProcessReceiveFailure(static_cast<int>(nread));
     return;
+  } else if (nread == 0) {
+    // Nothing to do it in this case.
+    return;
+  } else {
+    CHECK_LE(static_cast<size_t>(nread), store->ByteLength());
+    store = BackingStore::Reallocate(udp->env()->isolate(), std::move(store), nread);
   }
 
-  // Nothing to do it in this case.
-  if (nread == 0) return;
-
   Debug(udp, "Receiving UDP packet. %llu bytes.", nread);
 
   CHECK_NOT_NULL(addr);
@@ -1516,7 +1533,7 @@ void Endpoint::UDP::OnReceive(
 
   udp->endpoint_->OnReceive(
       static_cast<size_t>(nread),
-      *buf,
+      std::move(store),
       std::make_shared<SocketAddress>(addr));
 }
 

src/quic/endpoint.h

@@ -303,6 +303,8 @@ class Endpoint final : public MemoryRetainer,
       std::unique_ptr<Packet> packet,
       BaseObjectPtr<EndpointWrap> endpoint);
 
+    ~SendWrap() override;
+
     inline void Attach(const BaseObjectPtr<BaseObject>& strong_ptr) {
       strong_ptr_ = strong_ptr;
     }
@@ -375,6 +377,12 @@ class Endpoint final : public MemoryRetainer,
         size_t suggested_size,
         uv_buf_t* buf);
 
+    static void OnReceive(
+        uv_udp_t* handle,
+        ssize_t nread,
+        const uv_buf_t& buf,
+        const sockaddr* addr,
+        unsigned int flags);
     static void OnReceive(
         uv_udp_t* handle,
         ssize_t nread,
@@ -663,7 +671,7 @@ class Endpoint final : public MemoryRetainer,
   uv_buf_t OnAlloc(size_t suggested_size);
   void OnReceive(
       size_t nread,
-      const uv_buf_t& buf,
+      std::shared_ptr<v8::BackingStore> store,
       const std::shared_ptr<SocketAddress>& address);
 
   void ProcessOutbound();

src/quic/session.cc

@@ -1655,7 +1655,7 @@ void Session::Destroy() {
     retransmit_.Stop();
   }
 
-  // The Session instances are kept alive usingBaseObjectPtr. The
+  // The Session instances are kept alive using BaseObjectPtr. The
   // only persistent BaseObjectPtr is the map in the associated
   // Endpoint. Removing the Session from the Endpoint will free
   // that pointer, allowing the Session to be deconstructed once
@@ -3313,6 +3313,7 @@ bool Session::Application::SendPendingData() {
           pos += ndatalen;
           continue;
       }
+      packet.reset();
       session()->set_last_error(kQuicInternalError);
       return false;
     }