quic: update the packet generation logic

Author: jasnell

src/quic/application.cc

@@ -204,7 +204,7 @@ Packet* Session::Application::CreateStreamDataPacket() {
   return Packet::Create(env(),
                         session_->endpoint_.get(),
                         session_->remote_address_,
-                        ngtcp2_conn_get_max_tx_udp_payload_size(*session_),
+                        session_->max_packet_size(),
                         "stream data");
 }
 
@@ -238,37 +238,33 @@ void Session::Application::StreamReset(Stream* stream,
 void Session::Application::SendPendingData() {
   Debug(session_, "Application sending pending data");
   PathStorage path;
+  PathStorage prev_path;
 
   Packet* packet = nullptr;
   uint8_t* pos = nullptr;
+  uint8_t* begin = nullptr;
   int err = 0;
+  size_t last_nwrite = 0;
 
-  size_t maxPacketCount = std::min(static_cast<size_t>(64000),
-                                   ngtcp2_conn_get_send_quantum(*session_));
-  size_t packetSendCount = 0;
+  // The maximum size of packet to create.
+  const size_t max_packet_size = session_->max_packet_size();
 
-  const auto updateTimer = [&] {
-    Debug(session_, "Application updating the session timer");
-    ngtcp2_conn_update_pkt_tx_time(*session_, uv_hrtime());
-    session_->UpdateTimer();
-  };
-
-  const auto congestionLimited = [&](auto packet) {
-    auto len = pos - ngtcp2_vec(*packet).base;
-    // We are either congestion limited or done.
-    if (len) {
-      // Some data was serialized into the packet. We need to send it.
-      packet->Truncate(len);
-      session_->Send(std::move(packet), path);
-    }
+  // The maximum number of packets to send in this call to SendPendingData.
+  const size_t max_packet_count = std::min(
+      static_cast<size_t>(64000),
+      ngtcp2_conn_get_send_quantum(*session_) / max_packet_size);
 
-    updateTimer();
-  };
+  // The number of packets that have been sent in this call to SendPendingData.
+  size_t packet_send_count = 0;
 
   for (;;) {
+    // ndatalen is the amount of stream data that was accepted into the packet,
+    // if any.
     ssize_t ndatalen = 0;
-    StreamData stream_data;
 
+    // The stream_data is the next block of data from the application stream to
+    // send.
+    StreamData stream_data;
     err = GetStreamData(&stream_data);
 
     if (err < 0) {
@@ -278,115 +274,148 @@ void Session::Application::SendPendingData() {
       return session_->Close(Session::CloseMethod::SILENT);
     }
 
+    // If we got here, we were at least successful in checking for stream data.
+    // There might not be any stream data to send.
+    Debug(session_, "Application using stream data: %s", stream_data);
+
+    // Now let's make sure we have a packet to write data into.
     if (packet == nullptr) {
       packet = CreateStreamDataPacket();
       if (packet == nullptr) {
         session_->last_error_ = QuicError::ForNgtcp2Error(NGTCP2_ERR_INTERNAL);
         return session_->Close(Session::CloseMethod::SILENT);
       }
-      pos = ngtcp2_vec(*packet).base;
+      pos = begin = ngtcp2_vec(*packet).base;
     }
 
-    Debug(session_, "Application using stream data: %s", stream_data);
     DCHECK_NOT_NULL(pos);
+    DCHECK_NOT_NULL(begin);
     DCHECK_NOT_NULL(stream_data.buf);
-    ssize_t nwrite = WriteVStream(&path, pos, &ndatalen, stream_data);
-    Debug(session_, "Application accepted %zu bytes", ndatalen);
 
-    if (nwrite <= 0) {
+    // Awesome, let's write our packet!
+    ssize_t nwrite = WriteVStream(&path, pos, &ndatalen, max_packet_size, stream_data);
+    Debug(session_, "Application accepted %zu bytes into packet", ndatalen);
+
+    // A negative nwrite value indicates either an error or that there is more data
+    // to write into the packet.
+    if (nwrite < 0) {
       switch (nwrite) {
-        case 0: {
-          Debug(session_, "Congestion limited. Sending nothing.");
-          if (stream_data.id >= 0) ResumeStream(stream_data.id);
-          return congestionLimited(std::move(packet));
-        }
         case NGTCP2_ERR_STREAM_DATA_BLOCKED: {
-          session().StreamDataBlocked(stream_data.id);
-          if (session().max_data_left() == 0) {
-            if (stream_data.id >= 0) ResumeStream(stream_data.id);
-            return congestionLimited(std::move(packet));
-          }
-          CHECK_LE(ndatalen, 0);
+          // We could not write any data for this stream into the packet because
+          // the flow control for the stream itself indicates that the stream
+          // is blocked. We'll skip and move on to the next stream.
+          DCHECK_EQ(ndatalen, -1);
+          session_->StreamDataBlocked(stream_data.id);
           continue;
         }
         case NGTCP2_ERR_STREAM_SHUT_WR: {
-          Debug(session_, "Stream %" PRIi64 " is closed for writing",
-                stream_data.id);
-          // Indicates that the writable side of the stream has been closed
+          // Indicates that the writable side of the stream should be closed
           // locally or the stream is being reset. In either case, we can't send
           // any stream data!
-          CHECK_GE(stream_data.id, 0);
-          // We need to notify the stream that the writable side has been closed
-          // and no more outbound data can be sent.
-          CHECK_LE(ndatalen, 0);
-          auto stream = session_->FindStream(stream_data.id);
-          if (stream) stream->EndWritable();
+          Debug(session_, "Stream %" PRIi64 " should be closed for writing",
+                stream_data.id);
+          DCHECK_EQ(ndatalen, -1);
+          // If we got this response, then there should have been a stream associated
+          // with the stream_data, otherwise the response wouldn't make any sense.
+          DCHECK(stream_data.stream);
+          stream_data.stream->EndWritable();
           continue;
         }
         case NGTCP2_ERR_WRITE_MORE: {
           Debug(session_, "Application should write more to packet");
-          CHECK_GT(ndatalen, 0);
-          if (!StreamCommit(&stream_data, ndatalen)) return session_->Close();
-          pos += ndatalen;
+          DCHECK_GE(ndatalen, 0);
+          if (!StreamCommit(&stream_data, ndatalen)) {
+            packet->Done(UV_ECANCELED);
+            return session_->Close(CloseMethod::SILENT);
+          }
           continue;
         }
       }
 
+      // Some other type of error happened.
+      DCHECK_EQ(ndatalen, -1);
+      Debug(session_, "Application encountered error while writing packet: %s",
+            ngtcp2_strerror(nwrite));
       packet->Done(UV_ECANCELED);
-      session_->last_error_ = QuicError::ForNgtcp2Error(nwrite);
-      Debug(session_, "Application failed to write stream data with error %s",
-            session_->last_error_);
+      session_->SetLastError(QuicError::ForNgtcp2Error(nwrite));
       return session_->Close(Session::CloseMethod::SILENT);
+    } else if (ndatalen >= 0) {
+      // We wrote some data into the packet. We need to update the flow control
+      // by committing the data.
+      if (!StreamCommit(&stream_data, ndatalen)) {
+        packet->Done(UV_ECANCELED);
+        return session_->Close(CloseMethod::SILENT);
+      }
     }
 
-    pos += nwrite;
-    if (ndatalen > 0 && !StreamCommit(&stream_data, ndatalen)) {
-      // Since we are closing the session here, we don't worry about updating
-      // the pkt tx time. The failed StreamCommit should have updated the
-      // last_error_ appropriately.
-      packet->Done(UV_ECANCELED);
-      return session_->Close(Session::CloseMethod::SILENT);
+    // When nwrite is zero, it means we are congestion limited.
+    if (nwrite == 0) {
+      Debug(session_, "Congestion limited.");
+      // There might be a partial packet already prepared. If so, send it.
+      if (pos - begin) {
+        Debug(session_, "Packet has at least some data to send");
+        // At least some data had been written into the packet. We should send it.
+        packet->Truncate(pos - begin);
+        session_->Send(packet, path);
+      } else {
+        packet->Done(UV_ECANCELED);
+      }
+
+      session_->UpdatePacketTxTime();
+      return;
     }
 
-    if (stream_data.id >= 0 && ndatalen < 0) ResumeStream(stream_data.id);
+    pos += nwrite;
 
-    Debug(session_, "Packet ready to send with %zu bytes", nwrite);
-    packet->Truncate(nwrite);
-    session_->Send(packet, path);
-    packet = nullptr;
-    pos = nullptr;
+    if (packet_send_count == 0) {
+      path.CopyTo(&prev_path);
+      last_nwrite = nwrite;
+    } else if (prev_path != path ||
+               static_cast<size_t>(nwrite) > last_nwrite ||
+               (last_nwrite > max_packet_size && static_cast<size_t>(nwrite) != last_nwrite)) {
+      auto datalen = pos - begin - nwrite;
+      packet->Truncate(datalen);
+      session_->Send(packet->Clone(), prev_path);
+      session_->Send(packet, path);
+      session_->UpdatePacketTxTime();
+      packet = nullptr;
+      pos = nullptr;
+      begin = nullptr;
+      continue;
+    }
 
-    if (++packetSendCount == maxPacketCount) {
-      break;
+    if (++packet_send_count == max_packet_count || static_cast<size_t>(nwrite) < last_nwrite) {
+      auto datalen = pos - begin;
+      packet->Truncate(datalen);
+      session_->Send(packet, path);
+      session_->UpdatePacketTxTime();
+      return;
     }
   }
-
-  updateTimer();
 }
 
 ssize_t Session::Application::WriteVStream(PathStorage* path,
-                                           uint8_t* buf,
+                                           uint8_t* dest,
                                            ssize_t* ndatalen,
+                                           size_t max_packet_size,
                                            const StreamData& stream_data) {
-  CHECK_LE(stream_data.count, kMaxVectorCount);
-  uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_NONE;
-  if (stream_data.remaining > 0) flags |= NGTCP2_WRITE_STREAM_FLAG_MORE;
+  DCHECK_LE(stream_data.count, kMaxVectorCount);
+  uint32_t flags = NGTCP2_WRITE_STREAM_FLAG_MORE;
   if (stream_data.fin) flags |= NGTCP2_WRITE_STREAM_FLAG_FIN;
   ngtcp2_pkt_info pi;
-
-  ssize_t ret = ngtcp2_conn_writev_stream(
+  Debug(session_, "Writing max %zu bytes to packet", max_packet_size);
+  return ngtcp2_conn_writev_stream(
       *session_,
       &path->path,
       &pi,
-      buf,
-      ngtcp2_conn_get_max_tx_udp_payload_size(*session_),
+      dest,
+      max_packet_size,
       ndatalen,
       flags,
       stream_data.id,
       stream_data.buf,
       stream_data.count,
       uv_hrtime());
-  return ret;
 }
 
 // The DefaultApplication is the default implementation of Session::Application

src/quic/application.h

@@ -132,6 +132,7 @@ class Session::Application : public MemoryRetainer {
   ssize_t WriteVStream(PathStorage* path,
                        uint8_t* buf,
                        ssize_t* ndatalen,
+                       size_t max_packet_size,
                        const StreamData& stream_data);
 
  private:

src/quic/data.cc

@@ -47,13 +47,27 @@ std::string Path::ToString() const {
   return res;
 }
 
-PathStorage::PathStorage() {
-  ngtcp2_path_storage_zero(this);
-}
+PathStorage::PathStorage() { Reset(); }
 PathStorage::operator ngtcp2_path() {
   return path;
 }
 
+void PathStorage::Reset() {
+  ngtcp2_path_storage_zero(this);
+}
+
+void PathStorage::CopyTo(PathStorage* path) const {
+  ngtcp2_path_copy(&path->path, &this->path);
+}
+
+bool PathStorage::operator==(const PathStorage& other) const {
+  return ngtcp2_path_eq(&path, &other.path) != 0;
+}
+
+bool PathStorage::operator!=(const PathStorage& other) const {
+  return ngtcp2_path_eq(&path, &other.path) == 0;
+}
+
 // ============================================================================
 
 Store::Store(std::shared_ptr<v8::BackingStore> store,

src/quic/data.h

@@ -24,6 +24,12 @@ struct Path final : public ngtcp2_path {
 struct PathStorage final : public ngtcp2_path_storage {
   PathStorage();
   operator ngtcp2_path();
+
+  void Reset();
+  void CopyTo(PathStorage* path) const;
+
+  bool operator==(const PathStorage& other) const;
+  bool operator!=(const PathStorage& other) const;
 };
 
 class Store final : public MemoryRetainer {

src/quic/http3.cc

@@ -362,6 +362,9 @@ class Http3Application final : public Session::Application {
         return static_cast<int>(ret);
       } else {
         data->remaining = data->count = static_cast<size_t>(ret);
+        if (data->id > 0) {
+          data->stream = session().FindStream(data->id);
+        }
       }
     }
     return 0;

src/quic/packet.cc

@@ -161,7 +161,7 @@ Packet* Packet::FromFreeList(Environment* env,
   CHECK_NOT_NULL(packet);
   CHECK_EQ(env, packet->env());
   Debug(packet, "Reusing packet from freelist");
-  packet->data_ = data;
+  packet->data_ = std::move(data);
   packet->destination_ = destination;
   packet->listener_ = listener;
   return packet;

src/quic/session.cc

@@ -572,6 +572,10 @@ Session::~Session() {
   DCHECK(streams_.empty());
 }
 
+size_t Session::max_packet_size() const {
+  return ngtcp2_conn_get_max_tx_udp_payload_size(*this);
+}
+
 Session::operator ngtcp2_conn*() const {
   return connection_.get();
 }
@@ -862,6 +866,10 @@ void Session::Send(Packet* packet, const PathStorage& path) {
   Send(packet);
 }
 
+void Session::UpdatePacketTxTime() {
+  ngtcp2_conn_update_pkt_tx_time(*this, uv_hrtime());
+}
+
 uint64_t Session::SendDatagram(Store&& data) {
   auto tp = ngtcp2_conn_get_remote_transport_params(*this);
   uint64_t max_datagram_size = tp->max_datagram_frame_size;
@@ -1322,8 +1330,8 @@ void Session::OnTimeout() {
   if (is_destroyed()) return;
 
   int ret = ngtcp2_conn_handle_expiry(*this, uv_hrtime());
-  if (NGTCP2_OK(ret) && !is_in_closing_period() && !is_in_draining_period() &&
-      env()->can_call_into_js()) {
+  if (NGTCP2_OK(ret) && !is_in_closing_period() && !is_in_draining_period()) {
+    Debug(this, "Sending pending data after timr expiry");
     SendPendingDataScope send_scope(this);
     return;
   }
@@ -1346,6 +1354,7 @@ void Session::UpdateTimer() {
   }
 
   auto timeout = (expiry - now) / NGTCP2_MILLISECONDS;
+  Debug(this, "Updating timeout to %zu milliseconds", timeout);
 
   // If timeout is zero here, it means our timer is less than a millisecond
   // off from expiry. Let's bump the timer to 1.

src/quic/session.h

@@ -235,6 +235,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   bool is_destroyed() const;
   bool is_server() const;
 
+  size_t max_packet_size() const;
+
   void set_priority_supported(bool on = true);
 
   std::string diagnostic_name() const override;
@@ -246,6 +248,7 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
 
   TransportParams GetLocalTransportParams() const;
   TransportParams GetRemoteTransportParams() const;
+  void UpdatePacketTxTime();
 
   void MemoryInfo(MemoryTracker* tracker) const override;
   SET_MEMORY_INFO_NAME(Session)