quic: bring in a related performance optimizations en route

Author: pimterry

src/quic/application.h

@@ -146,6 +146,9 @@ class Session::Application : public MemoryRetainer {
   virtual void ReceiveStreamClose(Stream* stream,
                                   QuicError&& error = QuicError());
 
+  // Notify the Application that this stream is about to be removed
+  virtual void StreamRemoved(stream_id id) {}
+
   // Notifies the Application that the identified stream has been reset.
   virtual void ReceiveStreamReset(Stream* stream,
                                   uint64_t final_size,

src/quic/endpoint.cc

@@ -466,14 +466,24 @@ class Endpoint::UDP::Impl final : public HandleWrap {
     // UV_UDP_MMSG_CHUNK is set for each packet in a recvmmsg batch.
     // Processing is the same as for a single-message receive — ngtcp2
     // copies what it needs synchronously from the buf slice.
+    uint64_t now;
+    if (flags & UV_UDP_MMSG_CHUNK) {
+      if (impl->recv_batch_ts_ == 0) impl->recv_batch_ts_ = uv_hrtime();
+      now = impl->recv_batch_ts_;
+    } else {
+      now = uv_hrtime();
+    }
     impl->endpoint_->Receive(reinterpret_cast<const uint8_t*>(buf->base),
                              static_cast<size_t>(nread),
-                             SocketAddress(addr));
+                             SocketAddress(addr),
+                             now);
   }
 
   uv_udp_t handle_;
   Endpoint* endpoint_;
 
+  uint64_t recv_batch_ts_ = 0;
+
   friend class UDP;
 };
 
@@ -1353,9 +1363,8 @@ void Endpoint::CloseGracefully() {
 
 void Endpoint::Receive(const uint8_t* data,
                        size_t len,
-                       const SocketAddress& remote_address) {
-  const uint64_t now = uv_hrtime();
-
+                       const SocketAddress& remote_address,
+                       uint64_t now) {
   // Block list filtering — applied before any packet processing to
   // minimize resource expenditure on blocked sources.
   if (options_.block_list) {

src/quic/endpoint.h

@@ -435,7 +435,10 @@ class Endpoint final : public AsyncWrap, public Packet::Listener {
   // Ref() causes a listening Endpoint to keep the event loop active.
   JS_METHOD(Ref);
 
-  void Receive(const uint8_t* data, size_t len, const SocketAddress& from);
+  void Receive(const uint8_t* data,
+               size_t len,
+               const SocketAddress& from,
+               uint64_t now);
 
   AliasedStruct<Stats> stats_;
   AliasedStruct<State> state_;

src/quic/http3.cc

@@ -514,8 +514,12 @@ class Http3ApplicationImpl final : public Session::Application {
           flags.fin,
           flags.early);
 
+    uint64_t ts = session().rx_packet_ts();
+    if (ts == 0) [[unlikely]] {
+      ts = uv_hrtime();
+    }
     auto nread = nghttp3_conn_read_stream2(
-        *this, id, data, datalen, flags.fin ? 1 : 0, uv_hrtime());
+        *this, id, data, datalen, flags.fin ? 1 : 0, ts);
 
     if (nread < 0) {
       Debug(&session(),
@@ -665,6 +669,11 @@ class Http3ApplicationImpl final : public Session::Application {
     session().Close();
   }
 
+  void StreamRemoved(stream_id id) override {
+    if (conn_) nghttp3_conn_set_stream_user_data(*this, id, nullptr);
+    header_state_.erase(id);
+  }
+
   bool SendHeaders(const Stream& stream,
                    HeadersKind kind,
                    const Local<Array>& headers,
@@ -944,7 +953,7 @@ class Http3ApplicationImpl final : public Session::Application {
   }
 
   void OnBeginHeaders(stream_id id) {
-    auto stream = FindOrCreateStream(conn_.get(), &session(), id);
+    auto stream = FindOrCreateStream(id);
     if (!stream) [[unlikely]]
       return;
     Debug(&session(),
@@ -997,7 +1006,7 @@ class Http3ApplicationImpl final : public Session::Application {
   }
 
   void OnBeginTrailers(stream_id id) {
-    auto stream = FindOrCreateStream(conn_.get(), &session(), id);
+    auto stream = FindOrCreateStream(id);
     if (!stream) [[unlikely]]
       return;
     Debug(&session(),
@@ -1208,13 +1217,19 @@ class Http3ApplicationImpl final : public Session::Application {
     return app;
   }
 
-  static BaseObjectWeakPtr<Stream> FindOrCreateStream(nghttp3_conn* conn,
-                                                      Session* session,
-                                                      stream_id id) {
-    if (auto stream = session->FindStream(id)) {
+  // Persist the Stream* in the stream user data, so we can look it
+  // up directly without a FindStream map lookup every time.
+  void BindStreamUserData(stream_id id, Stream* stream) {
+    if (conn_) nghttp3_conn_set_stream_user_data(*this, id, stream);
+  }
+
+  BaseObjectWeakPtr<Stream> FindOrCreateStream(stream_id id) {
+    if (auto stream = session().FindStream(id)) {
+      BindStreamUserData(id, stream.get());
       return stream;
     }
-    if (auto stream = session->CreateStream(id)) {
+    if (auto stream = session().CreateStream(id)) {
+      BindStreamUserData(id, stream.get());
       return stream;
     }
     return {};
@@ -1238,8 +1253,11 @@ class Http3ApplicationImpl final : public Session::Application {
     auto& app = *ptr;
     NgHttp3CallbackScope scope(&app.session());
 
-    auto stream = app.session().FindStream(id);
-    if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE;
+    BaseObjectPtr<Stream> stream(static_cast<Stream*>(stream_user_data));
+    if (!stream) [[unlikely]] {
+      stream = app.session().FindStream(id);
+      if (!stream) return NGHTTP3_ERR_CALLBACK_FAILURE;
+    }
 
     if (stream->is_eos()) {
       *pflags |= NGHTTP3_DATA_FLAG_EOF;
@@ -1320,7 +1338,11 @@ class Http3ApplicationImpl final : public Session::Application {
     auto ptr = From(conn, conn_user_data);
     CHECK_NOT_NULL(ptr);
     auto& app = *ptr;
-    if (auto stream = app.session().FindStream(id)) {
+    BaseObjectPtr<Stream> stream(static_cast<Stream*>(stream_user_data));
+    if (!stream) [[unlikely]] {
+      stream = app.session().FindStream(id);
+    }
+    if (stream) {
       stream->Acknowledge(static_cast<size_t>(datalen));
     }
     return NGTCP2_SUCCESS;
@@ -1351,12 +1373,16 @@ class Http3ApplicationImpl final : public Session::Application {
     if (app.is_control_stream(id)) [[unlikely]] {
       return NGHTTP3_ERR_CALLBACK_FAILURE;
     }
-    auto& session = app.session();
-    if (auto stream = FindOrCreateStream(conn, &session, id)) [[likely]] {
-      stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
-      return NGTCP2_SUCCESS;
+    BaseObjectPtr<Stream> stream(static_cast<Stream*>(stream_user_data));
+    if (!stream) [[unlikely]] {
+      if (auto created = app.FindOrCreateStream(id)) {
+        created->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
+        return NGTCP2_SUCCESS;
+      }
+      return NGHTTP3_ERR_CALLBACK_FAILURE;
     }
-    return NGHTTP3_ERR_CALLBACK_FAILURE;
+    stream->ReceiveData(data, datalen, Stream::ReceiveDataFlags{});
+    return NGTCP2_SUCCESS;
   }
 
   static int on_deferred_consume(nghttp3_conn* conn,

src/quic/session.cc

@@ -3041,7 +3041,9 @@ bool Session::ReadPacket(const uint8_t* data,
     // receive path caches a timestamp and passes it to all ReadPacket()
     // calls in the same I/O burst.
     if (ts == 0) ts = uv_hrtime();
+    rx_packet_ts_ = ts;
     err = ngtcp2_conn_read_pkt(*this, &path, pkt_info, data, len, ts);
+    rx_packet_ts_ = 0;
   }
   if (is_destroyed()) return false;
 
@@ -3538,6 +3540,8 @@ void Session::RemoveStream(stream_id id) {
 
   ngtcp2_conn_set_stream_user_data(*this, id, nullptr);
 
+  if (impl_->application_) application().StreamRemoved(id);
+
   // Note that removing the stream from the streams map likely releases
   // the last BaseObjectPtr holding onto the Stream instance, at which
   // point it will be freed. If there are other BaseObjectPtr instances

src/quic/session.h

@@ -407,6 +407,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
                   const PacketInfo& pkt_info = PacketInfo(),
                   uint64_t ts = 0);
 
+  uint64_t rx_packet_ts() const { return rx_packet_ts_; }
+
   // Called by BindingData's flush callback to trigger SendPendingData
   // on this session. Encapsulates the application() access so that
   // bindingdata.cc doesn't need the full Application type definition.
@@ -736,6 +738,8 @@ class Session final : public AsyncWrap, private SessionTicket::AppData::Source {
   };
   Flags flags_;
 
+  uint64_t rx_packet_ts_ = 0;
+
   bool hello_processed_ = false;
 
   bool active_ = false;