Don't write data after write side closed. (#60)

Stream could be invalid when read side and write side are closed, or
connection is closed.

Author: jianjunz

web_transport/docs/threading_model.md

@@ -2,14 +2,27 @@
 
 OWT QUIC SDK provides thread safe APIs, but object's creation and deletion must be called on the same thread. Objects are usually created by `WebTransportFactory`.
 
+All calls to QUIC SDK are proxied to an internal IO thread as described below. All callbacks from QUIC SDK are called from this internal IO thread as well. It is recommended to do lightweight tasks only in callbacks.
+
+## Deadlock
+
+Because your thread is blocked when it calls QUIC SDK, please be cautious for deadlocks. An example to cause deadlock is:
+
+1. An external thread calls an API of QUIC SDK.
+1. QUIC SDK fires an event while the origin call is not finished.
+1. Event handler calls the same external thread.
+
+Deadlock happens because the external thread is waiting for the internal IO thread, and internal IO thread is waiting for the external thread.
+
 ## Internals
 
-There are two threads maintained internally:
+There is a thread maintained internally:
 
 - `io_thread`: Calls to Chromium's QUIC implementation are delegated to this thread.
-- `event_thread`: Callbacks and events are fired on this thread. We may move to sequence later.
 
-These two threads are owned by `WebTransportFactory`. Basically, all objects created by `WebTransportFactory` re-use the same task queues based on the two threads above.
+This thread are owned by `WebTransportFactory`. Basically, all objects created by `WebTransportFactory` re-use the same task queue based on the thread above.
+
+Another thread `event_thread` was used for callbacks. But it will be removed.
 
 ## Reference
 

web_transport/sdk/impl/web_transport_server_session.cc

@@ -132,7 +132,7 @@ WebTransportStreamInterface*
 WebTransportServerSession::CreateBidirectionalStreamOnCurrentThread() {
   ::quic::WebTransportStream* wt_stream =
       session_->OpenOutgoingBidirectionalStream();
-  std::unique_ptr<WebTransportStreamInterface> stream =
+  std::unique_ptr<WebTransportStreamImpl> stream =
       std::make_unique<WebTransportStreamImpl>(
           wt_stream,
           http3_session_->GetOrCreateStream(wt_stream->GetStreamId()),
@@ -205,15 +205,17 @@ void WebTransportServerSession::OnIncomingUnidirectionalStreamAvailable() {
 void WebTransportServerSession::OnSessionClosed(
     ::quic::WebTransportSessionError error_code,
     const std::string& error_message) {
+  for (auto& stream : streams_) {
+    stream->OnSessionClosed();
+  }
   if (visitor_) {
     visitor_->OnConnectionClosed();
   }
 }
 
 void WebTransportServerSession::AcceptIncomingStream(
     ::quic::WebTransportStream* stream) {
-  LOG(INFO) << "Accept incoming stream.";
-  std::unique_ptr<WebTransportStreamInterface> wt_stream =
+  std::unique_ptr<WebTransportStreamImpl> wt_stream =
       std::make_unique<WebTransportStreamImpl>(
           stream, http3_session_->GetOrCreateStream(stream->GetStreamId()),
           io_runner_, event_runner_);

web_transport/sdk/impl/web_transport_server_session.h

@@ -19,6 +19,8 @@
 namespace owt {
 namespace quic {
 
+class WebTransportStreamImpl;
+
 // A proxy of ::quic::WebTransportHttp3. WebTransport over HTTP/2 is not
 // supported.
 class WebTransportServerSession : public WebTransportSessionInterface,
@@ -66,7 +68,7 @@ class WebTransportServerSession : public WebTransportSessionInterface,
   ::quic::QuicSpdySession* http3_session_;
   base::SingleThreadTaskRunner* io_runner_;
   base::SingleThreadTaskRunner* event_runner_;
-  std::vector<std::unique_ptr<WebTransportStreamInterface>> streams_;
+  std::vector<std::unique_ptr<WebTransportStreamImpl>> streams_;
   WebTransportSessionInterface::Visitor* visitor_;
   ConnectionStats stats_;
 };

web_transport/sdk/impl/web_transport_stream_impl.cc

@@ -28,8 +28,17 @@ class WebTransportStreamVisitorAdapter
       : visitor_(visitor) {}
   void OnCanRead() override { visitor_->OnCanRead(); }
   void OnCanWrite() override { visitor_->OnCanWrite(); }
-  void OnResetStreamReceived(::quic::WebTransportStreamError error) override {}
-  void OnStopSendingReceived(::quic::WebTransportStreamError error) override {}
+  void OnResetStreamReceived(::quic::WebTransportStreamError error) override {
+    LOG(INFO)<<"OnResetStream received.";
+    if (visitor_) {
+      visitor_->OnResetStreamReceived(error);
+    }
+  }
+  void OnStopSendingReceived(::quic::WebTransportStreamError error) override {
+    if (visitor_) {
+      visitor_->OnStopSendingReceived(error);
+    }
+  }
   void OnWriteSideInDataRecvdState() override {}
 
  private:
@@ -45,7 +54,8 @@ WebTransportStreamImpl::WebTransportStreamImpl(
       quic_stream_(quic_stream),
       io_runner_(io_runner),
       event_runner_(event_runner),
-      visitor_(nullptr) {
+      visitor_(nullptr),
+      write_side_closed_(false) {
   CHECK(stream_);
   CHECK(quic_stream_);
   CHECK(io_runner_);
@@ -63,17 +73,24 @@ size_t WebTransportStreamImpl::Write(const uint8_t* data, size_t length) {
   DCHECK_EQ(sizeof(uint8_t), sizeof(char));
   CHECK(io_runner_);
   if (io_runner_->BelongsToCurrentThread()) {
+    if (write_side_closed_) {
+      return 0;
+    }
     return stream_->Write(
         absl::string_view(reinterpret_cast<const char*>(data), length));
   }
-  bool result;
+  bool result = false;
   base::WaitableEvent done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
   io_runner_->PostTask(
       FROM_HERE,
       base::BindOnce(
-          [](WebTransportStreamImpl* stream, const uint8_t* data, size_t& length,
-             bool& result, base::WaitableEvent* event) {
+          [](base::WeakPtr<WebTransportStreamImpl> stream, const uint8_t* data,
+             size_t& length, bool& result, base::WaitableEvent* event) {
+            if (!stream || stream->write_side_closed_) {
+              event->Signal();
+              return;
+            }
             if (stream->stream_->CanWrite()) {
               result = stream->stream_->Write(absl::string_view(
                   reinterpret_cast<const char*>(data), length));
@@ -82,7 +99,7 @@ size_t WebTransportStreamImpl::Write(const uint8_t* data, size_t length) {
             }
             event->Signal();
           },
-          base::Unretained(this), base::Unretained(data), std::ref(length),
+          weak_factory_.GetWeakPtr(), base::Unretained(data), std::ref(length),
           std::ref(result), base::Unretained(&done)));
   done.Wait();
   return result ? length : 0;
@@ -100,21 +117,25 @@ size_t WebTransportStreamImpl::Read(uint8_t* data, size_t length) {
     // TODO: FIN is not handled.
     return read_result.bytes_read;
   }
-  size_t result;
+  size_t result = 0;
   base::WaitableEvent done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
   io_runner_->PostTask(
       FROM_HERE,
       base::BindOnce(
-          [](WebTransportStreamImpl* stream, uint8_t* data, size_t& length,
-             size_t& result, base::WaitableEvent* event) {
+          [](base::WeakPtr<WebTransportStreamImpl> stream, uint8_t* data,
+             size_t& length, size_t& result, base::WaitableEvent* event) {
+            if (!stream) {
+              event->Signal();
+              return;
+            }
             auto read_result =
                 stream->stream_->Read(reinterpret_cast<char*>(data), length);
             // TODO: FIN is not handled.
             result = read_result.bytes_read;
             event->Signal();
           },
-          base::Unretained(this), base::Unretained(data), std::ref(length),
+          weak_factory_.GetWeakPtr(), base::Unretained(data), std::ref(length),
           std::ref(result), base::Unretained(&done)));
   done.Wait();
   return result;
@@ -124,18 +145,22 @@ size_t WebTransportStreamImpl::ReadableBytes() const {
   if (io_runner_->BelongsToCurrentThread()) {
     return stream_->ReadableBytes();
   }
-  size_t result;
+  size_t result = 0;
   base::WaitableEvent done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
-  io_runner_->PostTask(
-      FROM_HERE,
-      base::BindOnce(
-          [](WebTransportStreamImpl const* stream, size_t& result,
-             base::WaitableEvent* event) {
-            result = stream->stream_->ReadableBytes();
-            event->Signal();
-          },
-          base::Unretained(this), std::ref(result), base::Unretained(&done)));
+  io_runner_->PostTask(FROM_HERE,
+                       base::BindOnce(
+                           [](base::WeakPtr<WebTransportStreamImpl> stream,
+                              size_t& result, base::WaitableEvent* event) {
+                             if (!stream) {
+                               event->Signal();
+                               return;
+                             }
+                             result = stream->stream_->ReadableBytes();
+                             event->Signal();
+                           },
+                           weak_factory_.GetWeakPtr(), std::ref(result),
+                           base::Unretained(&done)));
   done.Wait();
   return result;
 }
@@ -150,34 +175,42 @@ void WebTransportStreamImpl::Close() {
   base::WaitableEvent done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
   io_runner_->PostTask(
-      FROM_HERE,
-      base::BindOnce(
-          [](WebTransportStreamImpl* stream, base::WaitableEvent* event) {
-            if (!stream->stream_->SendFin()) {
-              LOG(ERROR) << "Failed to send FIN.";
-            }
-            event->Signal();
-          },
-          base::Unretained(this), base::Unretained(&done)));
+      FROM_HERE, base::BindOnce(
+                     [](base::WeakPtr<WebTransportStreamImpl> stream,
+                        base::WaitableEvent* event) {
+                       if (!stream) {
+                         event->Signal();
+                         return;
+                       }
+                       if (!stream->stream_->SendFin()) {
+                         LOG(ERROR) << "Failed to send FIN.";
+                       }
+                       event->Signal();
+                     },
+                     weak_factory_.GetWeakPtr(), base::Unretained(&done)));
   done.Wait();
 }
 
 uint64_t WebTransportStreamImpl::BufferedDataBytes() const {
   if (io_runner_->BelongsToCurrentThread()) {
     return quic_stream_->BufferedDataBytes();
   }
-  uint64_t result;
+  uint64_t result = 0;
   base::WaitableEvent done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
-  io_runner_->PostTask(
-      FROM_HERE,
-      base::BindOnce(
-          [](WebTransportStreamImpl const* stream, uint64_t& result,
-             base::WaitableEvent* event) {
-            result = stream->quic_stream_->BufferedDataBytes();
-            event->Signal();
-          },
-          base::Unretained(this), std::ref(result), base::Unretained(&done)));
+  io_runner_->PostTask(FROM_HERE,
+                       base::BindOnce(
+                           [](base::WeakPtr<WebTransportStreamImpl> stream,
+                              uint64_t& result, base::WaitableEvent* event) {
+                             if (!stream) {
+                               event->Signal();
+                               return;
+                             }
+                             result = stream->quic_stream_->BufferedDataBytes();
+                             event->Signal();
+                           },
+                           weak_factory_.GetWeakPtr(), std::ref(result),
+                           base::Unretained(&done)));
   done.Wait();
   return result;
 }
@@ -186,47 +219,49 @@ bool WebTransportStreamImpl::CanWrite() const {
   if (io_runner_->BelongsToCurrentThread()) {
     return stream_->CanWrite();
   }
-  bool result;
+  bool result = false;
   base::WaitableEvent done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
                            base::WaitableEvent::InitialState::NOT_SIGNALED);
-  io_runner_->PostTask(
-      FROM_HERE,
-      base::BindOnce(
-          [](WebTransportStreamImpl const* stream, bool& result,
-             base::WaitableEvent* event) {
-            result = stream->stream_->CanWrite();
-            event->Signal();
-          },
-          base::Unretained(this), std::ref(result), base::Unretained(&done)));
+  io_runner_->PostTask(FROM_HERE,
+                       base::BindOnce(
+                           [](base::WeakPtr<WebTransportStreamImpl> stream,
+                              bool& result, base::WaitableEvent* event) {
+                             if (!stream) {
+                               event->Signal();
+                               return;
+                             }
+                             result = stream->stream_->CanWrite();
+                             event->Signal();
+                           },
+                           weak_factory_.GetWeakPtr(), std::ref(result),
+                           base::Unretained(&done)));
   done.Wait();
   return result;
 }
 
 void WebTransportStreamImpl::OnCanRead() {
-  event_runner_->PostTask(
-      FROM_HERE,
-      base::BindOnce(&WebTransportStreamImpl::OnCanReadOnCurrentThread,
-                     weak_factory_.GetWeakPtr()));
-}
-
-void WebTransportStreamImpl::OnCanWrite() {
-  event_runner_->PostTask(
-      FROM_HERE,
-      base::BindOnce(&WebTransportStreamImpl::OnCanWriteOnCurrentThread,
-                     weak_factory_.GetWeakPtr()));
-}
-
-void WebTransportStreamImpl::OnCanReadOnCurrentThread() {
   if (visitor_) {
     visitor_->OnCanRead();
   }
 }
 
-void WebTransportStreamImpl::OnCanWriteOnCurrentThread() {
+void WebTransportStreamImpl::OnCanWrite() {
   if (visitor_) {
     visitor_->OnCanWrite();
   }
 }
 
+void WebTransportStreamImpl::OnResetStreamReceived(
+    ::quic::WebTransportStreamError error) {
+  write_side_closed_ = true;
+}
+
+void WebTransportStreamImpl::OnStopSendingReceived(
+    ::quic::WebTransportStreamError error) {}
+
+void WebTransportStreamImpl::OnSessionClosed() {
+  write_side_closed_ = true;
+}
+
 }  // namespace quic
 }  // namespace owt

web_transport/sdk/impl/web_transport_stream_impl.h

@@ -48,16 +48,17 @@ class WebTransportStreamImpl : public WebTransportStreamInterface,
   uint64_t BufferedDataBytes() const override;
   bool CanWrite() const override;
 
+  void OnSessionClosed();
+
   // Overrides ::quic::WebTransportStreamVisitor.
   void OnCanRead() override;
   void OnCanWrite() override;
-  void OnResetStreamReceived(::quic::WebTransportStreamError error) override {}
-  void OnStopSendingReceived(::quic::WebTransportStreamError error) override {}
+  void OnResetStreamReceived(::quic::WebTransportStreamError error) override;
+  void OnStopSendingReceived(::quic::WebTransportStreamError error) override;
   void OnWriteSideInDataRecvdState() override {}
 
  private:
   void OnCanReadOnCurrentThread();
-  void OnFinReadOnCurrentThread();
   void OnCanWriteOnCurrentThread();
 
   ::quic::WebTransportStream* stream_;
@@ -68,6 +69,7 @@ class WebTransportStreamImpl : public WebTransportStreamInterface,
   base::SingleThreadTaskRunner* io_runner_;
   base::SingleThreadTaskRunner* event_runner_;
   owt::quic::WebTransportStreamInterface::Visitor* visitor_;
+  bool write_side_closed_;
   base::WeakPtrFactory<WebTransportStreamImpl> weak_factory_{this};
 };
 }  // namespace quic