nghttpx: Implement stream level timeout using ev_timer

src/shrpx_downstream.cc

@@ -824,120 +824,116 @@ bool Downstream::response_pseudo_header_allowed(int token) const {
   return http2::check_http2_response_pseudo_header(response_hdidx_, token);
 }
 
-void Downstream::init_upstream_timer() {
+namespace {
-  // auto evbase = upstream_->get_client_handler()->get_evbase();
+void reset_timer(struct ev_loop *loop, ev_timer *w) { ev_timer_again(loop, w); }
+} // namespace
 
-  // if (get_config()->stream_read_timeout.tv_sec > 0) {
+namespace {
-  //   upstream_rtimerev_ = init_timer(evbase, upstream_rtimeoutcb, this);
+void try_reset_timer(struct ev_loop *loop, ev_timer *w) {
-  // }
+  if (!ev_is_active(w)) {
-
+    return;
-  // if (get_config()->stream_write_timeout.tv_sec > 0) {
+  }
-  //   upstream_wtimerev_ = init_timer(evbase, upstream_wtimeoutcb, this);
+  ev_timer_again(loop, w);
-  // }
 }
+} // namespace
 
-// namespace {
+namespace {
-// void reset_timer(event *timer, const timeval *timeout) {
+void ensure_timer(struct ev_loop *loop, ev_timer *w) {
-//   if (!timer) {
+  if (ev_is_active(w)) {
-//     return;
+    return;
-//   }
+  }
+  ev_timer_again(loop, w);
+}
+} // namespace
 
-//   event_add(timer, timeout);
+namespace {
-// }
+void disable_timer(struct ev_loop *loop, ev_timer *w) {
-// } // namespace
+  ev_timer_stop(loop, w);
-
+}
-// namespace {
+} // namespace
-// void try_reset_timer(event *timer, const timeval *timeout) {
-// if (!timer) {
-//   return;
-// }
-
-// if (!evtimer_pending(timer, nullptr)) {
-//   return;
-// }
-
-// event_add(timer, timeout);
-// }
-// } // namespace
-
-// namespace {
-// void ensure_timer(event *timer, const timeval *timeout) {
-// if (!timer) {
-//   return;
-// }
-
-// if (evtimer_pending(timer, nullptr)) {
-//   return;
-// }
-
-// event_add(timer, timeout);
-// }
-// } // namespace
-
-// namespace {
-// void disable_timer(event *timer) {
-// if (!timer) {
-//   return;
-// }
-
-// event_del(timer);
-// }
-// } // namespace
 
 void Downstream::reset_upstream_rtimer() {
-  // reset_timer(upstream_rtimerev_, &get_config()->stream_read_timeout);
+  if (get_config()->stream_read_timeout == 0.) {
-  // try_reset_timer(upstream_wtimerev_, &get_config()->stream_write_timeout);
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  reset_timer(loop, &upstream_rtimer_);
 }
 
 void Downstream::reset_upstream_wtimer() {
-  // reset_timer(upstream_wtimerev_, &get_config()->stream_write_timeout);
+  auto loop = upstream_->get_client_handler()->get_loop();
-  // try_reset_timer(upstream_rtimerev_, &get_config()->stream_read_timeout);
+  if (get_config()->stream_write_timeout != 0.) {
+    reset_timer(loop, &upstream_wtimer_);
+  }
+  if (get_config()->stream_read_timeout != 0.) {
+    try_reset_timer(loop, &upstream_rtimer_);
+  }
 }
 
 void Downstream::ensure_upstream_wtimer() {
-  // ensure_timer(upstream_wtimerev_, &get_config()->stream_write_timeout);
+  if (get_config()->stream_write_timeout == 0.) {
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  ensure_timer(loop, &upstream_wtimer_);
 }
 
 void Downstream::disable_upstream_rtimer() {
-  // disable_timer(upstream_rtimerev_);
+  if (get_config()->stream_read_timeout == 0.) {
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  disable_timer(loop, &upstream_rtimer_);
 }
 
 void Downstream::disable_upstream_wtimer() {
-  // disable_timer(upstream_wtimerev_);
+  if (get_config()->stream_write_timeout == 0.) {
-}
+    return;
-
+  }
-void Downstream::init_downstream_timer() {
+  auto loop = upstream_->get_client_handler()->get_loop();
-  // auto evbase = upstream_->get_client_handler()->get_evbase();
+  disable_timer(loop, &upstream_wtimer_);
-
-  // if (get_config()->stream_read_timeout.tv_sec > 0) {
-  //   downstream_rtimerev_ = init_timer(evbase, downstream_rtimeoutcb, this);
-  // }
-
-  // if (get_config()->stream_write_timeout.tv_sec > 0) {
-  //   downstream_wtimerev_ = init_timer(evbase, downstream_wtimeoutcb, this);
-  // }
 }
 
 void Downstream::reset_downstream_rtimer() {
-  // reset_timer(downstream_rtimerev_, &get_config()->stream_read_timeout);
+  if (get_config()->stream_read_timeout == 0.) {
-  // try_reset_timer(downstream_wtimerev_, &get_config()->stream_write_timeout);
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  reset_timer(loop, &downstream_rtimer_);
 }
 
 void Downstream::reset_downstream_wtimer() {
-  // reset_timer(downstream_wtimerev_, &get_config()->stream_write_timeout);
+  auto loop = upstream_->get_client_handler()->get_loop();
-  // try_reset_timer(downstream_rtimerev_, &get_config()->stream_read_timeout);
+  if (get_config()->stream_write_timeout != 0.) {
+    reset_timer(loop, &downstream_wtimer_);
+  }
+  if (get_config()->stream_read_timeout != 0.) {
+    try_reset_timer(loop, &downstream_rtimer_);
+  }
 }
 
 void Downstream::ensure_downstream_wtimer() {
-  // ensure_timer(downstream_wtimerev_, &get_config()->stream_write_timeout);
+  if (get_config()->stream_write_timeout == 0.) {
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  ensure_timer(loop, &downstream_wtimer_);
 }
 
 void Downstream::disable_downstream_rtimer() {
-  // disable_timer(downstream_rtimerev_);
+  if (get_config()->stream_read_timeout == 0.) {
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  disable_timer(loop, &downstream_rtimer_);
 }
 
 void Downstream::disable_downstream_wtimer() {
-  // disable_timer(downstream_wtimerev_);
+  if (get_config()->stream_write_timeout == 0.) {
+    return;
+  }
+  auto loop = upstream_->get_client_handler()->get_loop();
+  disable_timer(loop, &downstream_wtimer_);
 }
 
 bool Downstream::accesslog_ready() const { return response_http_status_ > 0; }

src/shrpx_downstream.h

@@ -246,18 +246,15 @@ public:
   bool get_rst_stream_after_end_stream() const;
   void set_rst_stream_after_end_stream(bool f);
 
-  // Initializes upstream timers, but they are not pending.
+  // Resets upstream read timer.  If it is active, timeout value is
-  void init_upstream_timer();
+  // reset.  If it is not active, timer will be started.
-  // Makes upstream read timer pending.  If it is already pending,
-  // timeout value is reset.  This function also resets write timer if
-  // it is already pending.
   void reset_upstream_rtimer();
-  // Makes upstream write timer pending.  If it is already pending,
+  // Resets upstream write timer. If it is active, timeout value is
-  // timeout value is reset.  This function also resets read timer if
+  // reset.  If it is not active, timer will be started.  This
-  // it is already pending.
+  // function also resets read timer if it has been started.
   void reset_upstream_wtimer();
-  // Makes upstream write timer pending.  If it is already pending, do
+  // Makes sure that upstream write timer is started.  If it has been
-  // nothing.
+  // started, do nothing.  Otherwise, write timer will be started.
   void ensure_upstream_wtimer();
   // Disables upstream read timer.
   void disable_upstream_rtimer();
@@ -266,7 +263,6 @@ public:
 
   // Downstream timer functions.  They works in a similar way just
   // like the upstream timer function.
-  void init_downstream_timer();
   void reset_downstream_rtimer();
   void reset_downstream_wtimer();
   void ensure_downstream_wtimer();
@@ -355,8 +351,6 @@ private:
   bool response_connection_close_;
   bool response_header_key_prev_;
   bool expect_final_response_;
-
-  bool use_timer_;
 };
 
 } // namespace shrpx

src/shrpx_http2_downstream_connection.cc

@@ -105,8 +105,7 @@ int Http2DownstreamConnection::attach_downstream(Downstream *downstream) {
   }
 
   downstream_ = downstream;
-
+  downstream_->reset_downstream_rtimer();
-  downstream_->init_downstream_timer();
 
   return 0;
 }

src/shrpx_http2_upstream.cc

@@ -123,7 +123,6 @@ int Http2Upstream::upgrade_upstream(HttpsUpstream *http) {
   auto downstream = http->pop_downstream();
   downstream->reset_upstream(this);
   downstream->set_stream_id(1);
-  downstream->init_upstream_timer();
   downstream->reset_upstream_rtimer();
   downstream->set_stream_id(1);
   downstream->set_priority(0);
@@ -222,7 +221,6 @@ int on_begin_headers_callback(nghttp2_session *session,
   auto downstream =
       util::make_unique<Downstream>(upstream, frame->hd.stream_id, 0);
 
-  downstream->init_upstream_timer();
   downstream->reset_upstream_rtimer();
 
   // Although, we deprecated minor version from HTTP/2, we supply

src/shrpx_spdy_upstream.cc

@@ -152,7 +152,6 @@ void on_ctrl_recv_callback(spdylay_session *session, spdylay_frame_type type,
     auto downstream = upstream->add_pending_downstream(
         frame->syn_stream.stream_id, frame->syn_stream.pri);
 
-    downstream->init_upstream_timer();
     downstream->reset_upstream_rtimer();
 
     auto nv = frame->syn_stream.nv;