Make HttpSM server reference a Transaction instead of a Session

iocore/net/libinknet_stub.cc

@@ -21,11 +21,11 @@
   limitations under the License.
  */
 
-#include "HttpSessionManager.h"
+class EThread;
+class Continuation;
 void
 initialize_thread_for_http_sessions(EThread *, int)
 {
-  ink_assert(false);
 }
 
 #include "P_UnixNet.h"

proxy/PoolableSession.h

@@ -33,9 +33,11 @@ class PoolableSession : public ProxySession
 public:
   enum PooledState {
     INIT,
-    SSN_IN_USE,  // actively in use
-    KA_RESERVED, // stuck to client
-    KA_POOLED,   // free for reuse
+    SSN_IN_USE,    // actively in use
+    KA_RESERVED,   // stuck to client
+    KA_POOLED,     // free for reuse
+    SSN_CLOSED,    // Session ready to be freed
+    SSN_TO_RELEASE // Session reaady to be released
   };
 
   /// Hash map descriptor class for IP map.
@@ -72,18 +74,30 @@ class PoolableSession : public ProxySession
   TSServerSessionSharingMatchMask sharing_match = TS_SERVER_SESSION_SHARING_MATCH_MASK_NONE;
   TSServerSessionSharingPoolType sharing_pool   = TS_SERVER_SESSION_SHARING_POOL_GLOBAL;
 
-  // Keep track of connection limiting and a pointer to the
-  // singleton that keeps track of the connection counts.
-  OutboundConnTrack::Group *conn_track_group = nullptr;
+  void enable_outbound_connection_tracking(OutboundConnTrack::Group *group);
+  void release_outbound_connection_tracking();
+
+  void attach_hostname(const char *hostname);
 
   void set_active();
   bool is_active();
   void set_private(bool new_private = true);
   bool is_private() const;
 
-  void set_netvc(NetVConnection *newvc);
+  virtual void set_netvc(NetVConnection *newvc);
 
-  virtual IOBufferReader *get_reader() = 0;
+  // Used to determine whether the session is for parent proxy
+  // it is session to origin server
+  // We need to determine whether a closed connection was to
+  // close parent proxy to update the
+  // proxy.process.http.current_parent_proxy_connections
+  bool to_parent_proxy = false;
+
+  // Keep track of connection limiting and a pointer to the
+  // singleton that keeps track of the connection counts.
+  OutboundConnTrack::Group *conn_track_group = nullptr;
+
+  virtual IOBufferReader *get_remote_reader() = 0;
 
 private:
   // Sessions become if authentication headers
@@ -192,3 +206,34 @@ PoolableSession::FQDNLinkage::equal(CryptoHash const &lhs, CryptoHash const &rhs
 {
   return lhs == rhs;
 }
+
+inline void
+PoolableSession::enable_outbound_connection_tracking(OutboundConnTrack::Group *group)
+{
+  ink_assert(nullptr == conn_track_group);
+  conn_track_group = group;
+}
+
+inline void
+PoolableSession::release_outbound_connection_tracking()
+{
+  // Update upstream connection tracking data if present.
+  if (conn_track_group) {
+    if (conn_track_group->_count >= 0) {
+      (conn_track_group->_count)--;
+      conn_track_group = nullptr;
+    } else {
+      // A bit dubious, as there's no guarantee it's still negative, but even that would be interesting to know.
+      Error("[http_ss] [%" PRId64 "] number of connections should be greater than or equal to zero: %u", con_id,
+            conn_track_group->_count.load());
+    }
+  }
+}
+
+inline void
+PoolableSession::attach_hostname(const char *hostname)
+{
+  if (CRYPTO_HASH_ZERO == hostname_hash) {
+    CryptoContext().hash_immediate(hostname_hash, (unsigned char *)hostname, strlen(hostname));
+  }
+}

proxy/ProxySession.h

@@ -107,7 +107,6 @@ class ProxySession : public VConnection, public PluginUserArgs<TS_USER_ARGS_SSN>
   virtual void hook_add(TSHttpHookID id, INKContInternal *cont);
 
   virtual bool is_chunked_encoding_supported() const;
-
   virtual void set_half_close_flag(bool flag);
   virtual bool get_half_close_flag() const;
 
@@ -137,7 +136,7 @@ class ProxySession : public VConnection, public PluginUserArgs<TS_USER_ARGS_SSN>
   void clear_session_active();
   bool is_active() const;
   bool is_draining() const;
-  bool is_client_closed() const;
+  bool is_peer_closed() const;
 
   int64_t connection_id() const;
   TSHttpHookID get_hookid() const;
@@ -157,6 +156,12 @@ class ProxySession : public VConnection, public PluginUserArgs<TS_USER_ARGS_SSN>
   void do_io_shutdown(ShutdownHowTo_t howto) override;
   void reenable(VIO *vio) override;
 
+  virtual ProxyTransaction *
+  new_transaction()
+  {
+    return nullptr;
+  }
+
   ////////////////////
   // Members
 
@@ -236,7 +241,7 @@ ProxySession::is_draining() const
 }
 
 inline bool
-ProxySession::is_client_closed() const
+ProxySession::is_peer_closed() const
 {
   return get_netvc() == nullptr;
 }

proxy/ProxyTransaction.cc

@@ -57,8 +57,8 @@ ProxyTransaction::new_transaction(bool from_early_data)
     }
   }
 
-  this->increment_client_transactions_stat();
-  _sm->attach_client_session(this, _reader);
+  this->increment_transactions_stat();
+  _sm->attach_client_session(this);
 }
 
 bool
@@ -181,7 +181,7 @@ void
 ProxyTransaction::transaction_done()
 {
   SCOPED_MUTEX_LOCK(lock, this->mutex, this_ethread());
-  this->decrement_client_transactions_stat();
+  this->decrement_transactions_stat();
 }
 
 // Implement VConnection interface.
@@ -221,6 +221,18 @@ ProxyTransaction::has_request_body(int64_t request_content_length, bool is_chunk
   return request_content_length > 0 || is_chunked;
 }
 
+void
+ProxyTransaction::attach_transaction(HttpSM *attach_sm)
+{
+  _sm = attach_sm;
+}
+
+HTTPVersion
+ProxyTransaction::get_version(HTTPHdr &hdr) const
+{
+  return hdr.version_get();
+}
+
 bool
 ProxyTransaction::allow_half_open() const
 {

proxy/ProxyTransaction.h

@@ -39,10 +39,11 @@ class ProxyTransaction : public VConnection
   /// Virtual Methods
   //
   virtual void new_transaction(bool from_early_data = false);
+  virtual void attach_transaction(HttpSM *attach_sm);
   virtual bool attach_server_session(PoolableSession *ssession, bool transaction_done = true);
   Action *adjust_thread(Continuation *cont, int event, void *data);
-  virtual void release(IOBufferReader *r) = 0;
-  virtual void transaction_done();
+  virtual void release()          = 0;
+  virtual void transaction_done() = 0;
 
   virtual void set_active_timeout(ink_hrtime timeout_in);
   virtual void set_inactivity_timeout(ink_hrtime timeout_in);
@@ -63,8 +64,9 @@ class ProxyTransaction : public VConnection
   virtual int get_transaction_priority_weight() const;
   virtual int get_transaction_priority_dependence() const;
   virtual bool allow_half_open() const;
-  virtual void increment_client_transactions_stat() = 0;
-  virtual void decrement_client_transactions_stat() = 0;
+
+  virtual void increment_transactions_stat() = 0;
+  virtual void decrement_transactions_stat() = 0;
 
   virtual NetVConnection *get_netvc() const;
   virtual bool is_first_transaction() const;
@@ -85,6 +87,10 @@ class ProxyTransaction : public VConnection
   // Returns true if there is a request body for this request
   virtual bool has_request_body(int64_t content_length, bool is_chunked_set) const;
 
+  sockaddr const *get_remote_addr() const;
+
+  virtual HTTPVersion get_version(HTTPHdr &hdr) const;
+
   /// Non-Virtual Methods
   //
   const char *get_protocol_string();
@@ -114,15 +120,17 @@ class ProxyTransaction : public VConnection
   // This function must return a non-negative number that is different for two in-progress transactions with the same proxy_ssn
   // session.
   //
-  void set_rx_error_code(ProxyError e);
-  void set_tx_error_code(ProxyError e);
+  virtual void set_rx_error_code(ProxyError e);
+  virtual void set_tx_error_code(ProxyError e);
 
   bool support_sni() const;
 
   /// Variables
   //
   HttpSessionAccept::Options upstream_outbound_options; // overwritable copy of options
 
+  IOBufferReader *get_remote_reader();
+
 protected:
   ProxySession *_proxy_ssn = nullptr;
   HttpSM *_sm              = nullptr;
@@ -274,3 +282,19 @@ ProxyTransaction::adjust_thread(Continuation *cont, int event, void *data)
   }
   return nullptr;
 }
+
+inline IOBufferReader *
+ProxyTransaction::get_remote_reader()
+{
+  return _reader;
+}
+
+inline sockaddr const *
+ProxyTransaction::get_remote_addr() const
+{
+  if (_proxy_ssn) {
+    return _proxy_ssn->get_remote_addr();
+  } else {
+    return nullptr;
+  }
+}

proxy/http/Http1ClientSession.cc

@@ -61,6 +61,8 @@ ClassAllocator<Http1ClientSession, true> http1ClientSessionAllocator("http1Clien
 
 Http1ClientSession::Http1ClientSession() : super(), trans(this) {}
 
+//
+// Will only close the connection if do_io_close has been called previously (to set read_state to HCS_CLOSED
 void
 Http1ClientSession::destroy()
 {
@@ -86,12 +88,16 @@ Http1ClientSession::release_transaction()
   if (transact_count == released_transactions) {
     // Make sure we previously called release() or do_io_close() on the session
     ink_release_assert(read_state != HCS_INIT);
-    if (read_state == HCS_ACTIVE_READER) {
+    if (is_active()) {
       // (in)active timeout
       do_io_close(HTTP_ERRNO);
+    } else if (read_state == HCS_ACTIVE_READER) {
+      release(&trans); // Put back to keep-alive state
     } else {
       destroy();
     }
+  } else {
+    ink_release_assert(transact_count == released_transactions);
   }
 }
 
@@ -209,6 +215,9 @@ void
 Http1ClientSession::do_io_close(int alerrno)
 {
   if (read_state == HCS_CLOSED) {
+    if (transact_count == released_transactions) {
+      this->destroy();
+    }
     return; // Don't double call session close
   }
   if (read_state == HCS_ACTIVE_READER) {
@@ -257,8 +266,6 @@ Http1ClientSession::do_io_close(int alerrno)
     HTTP_SUM_DYN_STAT(http_transactions_per_client_con, transact_count);
     read_state = HCS_CLOSED;
 
-    // Can go ahead and close the netvc now, but keeping around the session object
-    // until all the transactions are closed
     if (_vc) {
       _vc->do_io_close();
       _vc = nullptr;
@@ -347,11 +354,21 @@ Http1ClientSession::state_keep_alive(int event, void *data)
 {
   // Route the event.  It is either for vc or
   //  the origin server slave vc
-  if (data && data == slave_ka_vio) {
-    return state_slave_keep_alive(event, data);
-  } else {
-    ink_assert(data && data == ka_vio);
-    ink_assert(read_state == HCS_KEEP_ALIVE);
+  if (data) {
+    if (data == slave_ka_vio) {
+      return state_slave_keep_alive(event, data);
+    } else if (data == schedule_event) {
+      schedule_event = nullptr;
+    } else {
+      ink_assert(data && data == ka_vio);
+      ink_assert(read_state == HCS_KEEP_ALIVE);
+    }
+  }
+
+  // If we got here due to a network I/O event directly, go ahead and cancel any remaining schedule events
+  if (schedule_event) {
+    schedule_event->cancel();
+    schedule_event = nullptr;
   }
 
   STATE_ENTER(&Http1ClientSession::state_keep_alive, event, data);
@@ -387,8 +404,6 @@ Http1ClientSession::state_keep_alive(int event, void *data)
 void
 Http1ClientSession::release(ProxyTransaction *trans)
 {
-  ink_assert(read_state == HCS_ACTIVE_READER || read_state == HCS_INIT);
-
   // When release is called from start() to read the first transaction, get_sm()
   // will return null.
   HttpSM *sm                = trans->get_sm();
@@ -409,6 +424,7 @@ Http1ClientSession::release(ProxyTransaction *trans)
   //  buffer.  If there is, spin up a new state
   //  machine to process it.  Otherwise, issue an
   //  IO to wait for new data
+  /*  Start the new transaction once we finish completely the current transaction and unroll the stack */
   bool more_to_read = this->_reader->is_read_avail_more_than(0);
   if (more_to_read) {
     HttpSsnDebug("[%" PRId64 "] data already in buffer, starting new transaction", con_id);
@@ -429,19 +445,19 @@ Http1ClientSession::release(ProxyTransaction *trans)
   }
 }
 
-void
+ProxyTransaction *
 Http1ClientSession::new_transaction()
 {
   // If the client connection terminated during API callouts we're done.
   if (nullptr == _vc) {
     this->do_io_close(); // calls the SSN_CLOSE hooks to match the SSN_START hooks.
-    return;
+    return nullptr;
   }
 
   if (!_vc->add_to_active_queue()) {
     // no room in the active queue close the connection
     this->do_io_close();
-    return;
+    return nullptr;
   }
 
   // Defensive programming, make sure nothing persists across
@@ -453,6 +469,7 @@ Http1ClientSession::new_transaction()
   transact_count++;
 
   trans.new_transaction(read_from_early_data > 0 ? true : false);
+  return &trans;
 }
 
 bool
@@ -472,7 +489,7 @@ Http1ClientSession::attach_server_session(PoolableSession *ssession, bool transa
     //  have it call the client session back.  This IO also prevent
     //  the server net conneciton from calling back a dead sm
     SET_HANDLER(&Http1ClientSession::state_keep_alive);
-    slave_ka_vio = ssession->do_io_read(this, INT64_MAX, ssession->get_reader()->mbuf);
+    slave_ka_vio = ssession->do_io_read(this, INT64_MAX, ssession->get_remote_reader()->mbuf);
     ink_assert(slave_ka_vio != ka_vio);
 
     // Transfer control of the write side as well