Merge pull request #1 from AndreasAakesson/tcp_rcvwnd

TCP, pmr and microlb fixes

Author: KristianJerpetjon

api/util/detail/alloc_pmr.hpp

@@ -33,6 +33,7 @@ namespace os::mem::detail {
 
     void* do_allocate(size_t size, size_t align) override {
       if (UNLIKELY(size + allocated_ > cap_total_)) {
+        //printf("pmr about to throw bad alloc: sz=%zu alloc=%zu cap=%zu\n", size, allocated_, cap_total_);
         throw std::bad_alloc();
       }
 
@@ -46,6 +47,7 @@ namespace os::mem::detail {
       void* buf = memalign(align, size);
 
       if (buf == nullptr) {
+        //printf("pmr memalign return nullptr, throw bad alloc\n");
         throw std::bad_alloc();
       }
 
@@ -152,7 +154,9 @@ namespace os::mem::detail {
 
     std::size_t resource_capacity() {
       if (cap_suballoc_ == 0)
+      {
         return cap_total_ / (used_resources_ + os::mem::Pmr_pool::resource_division_offset);
+      }
       return cap_suballoc_;
     }
 
@@ -244,7 +248,9 @@ namespace os::mem {
   // Pmr_resource implementation
   //
   Pmr_resource::Pmr_resource(Pool_ptr p) : pool_{p} {}
-  std::size_t Pmr_resource::capacity() { return pool_->resource_capacity(); }
+  std::size_t Pmr_resource::capacity() {
+    return std::min(pool_->resource_capacity(), pool_->allocatable());
+  }
   std::size_t Pmr_resource::allocatable() {
     auto cap = capacity();
     if (used > cap)
@@ -267,10 +273,8 @@ namespace os::mem {
     }
 
     void* buf = pool_->allocate(size, align);
-
     used += size;
     allocs++;
-
     return buf;
   }
 

lib/microLB/micro_lb/balancer.cpp

@@ -259,8 +259,12 @@ namespace microLB
       auto &session=get_session(idx);
 
       // free session destroying potential unique ptr objects
-      session.incoming =nullptr;
-      session.outgoing=nullptr;
+      session.incoming = nullptr;
+      auto out_tcp = dynamic_cast<net::tcp::Stream*>(session.outgoing->bottom_transport())->tcp();
+      session.outgoing = nullptr;
+      // if we don't have anything to write to the backend, abort it.
+      if(not out_tcp->sendq_size())
+        out_tcp->abort();
       free_sessions.push_back(session.self);
       LBOUT("Session %d destroyed  (total = %d)\n", session.self, session_cnt);
     }
@@ -359,7 +363,17 @@ namespace microLB
   }
   void Node::connect()
   {
-    auto outgoing = this->stack.tcp().connect(this->addr);
+    net::tcp::Connection_ptr outgoing;
+    try
+    {
+      outgoing = this->stack.tcp().connect(this->addr);
+    }
+    catch([[maybe_unused]]const net::TCP_error& err)
+    {
+      LBOUT("Got exception: %s\n", err.what());
+      this->restart_active_check();
+      return;
+    }
     // connecting to node atm.
     this->connecting++;
     // retry timer when connect takes too long

src/net/tcp/connection.cpp

@@ -53,8 +53,9 @@ Connection::Connection(TCP& host, Socket local, Socket remote, ConnectCallback c
 
 Connection::~Connection()
 {
-  //printf("<Connection> Deleted %p %s  ACTIVE: %u\n", this,
+  //printf("<Connection> Deleted %p %s  ACTIVE: %zu\n", this,
   //        to_string().c_str(), host_.active_connections());
+
   rtx_clear();
 }
 
@@ -434,6 +435,8 @@ bool Connection::handle_ack(const Packet_view& in)
 
   if(is_win_update(in, true_win))
   {
+    //if(cb.SND.WND < SMSS()*2)
+    //  printf("Win update: %u => %u\n", cb.SND.WND, true_win);
     cb.SND.WND = true_win;
     cb.SND.WL1 = in.seq();
     cb.SND.WL2 = in.ack();
@@ -613,7 +616,7 @@ void Connection::on_dup_ack(const Packet_view& in)
   // 3 dup acks
   else if(dup_acks_ == 3)
   {
-    debug("<TCP::Connection::on_dup_ack> Dup ACK == 3 - UNA=%u recover=%u\n", cb.SND.UNA, cb.recover);
+    //printf("<TCP::Connection::on_dup_ack> Dup ACK == 3 - UNA=%u recover=%u\n", cb.SND.UNA, cb.recover);
 
     if(cb.SND.UNA - 1 > cb.recover)
       goto fast_rtx;
@@ -997,8 +1000,8 @@ void Connection::retransmit() {
 
     // TODO: Finish to send window zero probe, but only on rtx timeout
 
-    debug2("<Connection::retransmit> With data (wq.sz=%u) buf.unacked=%u\n",
-           writeq.size(), buf->size(), buf->size() - writeq.acked());
+    //printf("<Connection::retransmit> With data (wq.sz=%zu) buf.size=%zu buf.unacked=%zu SND.WND=%u CWND=%u\n",
+    //       writeq.size(), buf->size(), buf->size() - writeq.acked(), cb.SND.WND, cb.cwnd);
     fill_packet(*packet, buf->data() + writeq.acked(), buf->size() - writeq.acked());
       packet->set_flag(PSH);
   }
@@ -1070,8 +1073,8 @@ void Connection::rtx_clear() {
        begins (i.e., after the three-way handshake completes).
 */
 void Connection::rtx_timeout() {
-  debug("<Connection::RTX@timeout> Timed out (RTO %lld ms). FS: %u\n",
-    rttm.rto_ms().count(), flight_size());
+  //printf("<Connection::RTX@timeout> Timed out (RTO %lld ms). FS: %u usable=%u\n",
+  //  rttm.rto_ms().count(), flight_size(), usable_window());
 
   signal_rtx_timeout();
   // experimental
@@ -1421,12 +1424,12 @@ void Connection::reduce_ssthresh() {
     fs = (fs >= two_seg) ? fs - two_seg : 0;
 
   cb.ssthresh = std::max( (fs / 2), two_seg );
-  debug2("<TCP::Connection::reduce_ssthresh> Slow start threshold reduced: %u\n",
-    cb.ssthresh);
+  //printf("<TCP::Connection::reduce_ssthresh> Slow start threshold reduced: %u\n",
+  //  cb.ssthresh);
 }
 
 void Connection::fast_retransmit() {
-  debug("<TCP::Connection::fast_retransmit> Fast retransmit initiated.\n");
+  //printf("<TCP::Connection::fast_retransmit> Fast retransmit initiated.\n");
   // reduce sshtresh
   reduce_ssthresh();
   // retransmit segment starting SND.UNA
@@ -1441,5 +1444,5 @@ void Connection::finish_fast_recovery() {
   fast_recovery_ = false;
   //cb.cwnd = std::min(cb.ssthresh, std::max(flight_size(), (uint32_t)SMSS()) + SMSS());
   cb.cwnd = cb.ssthresh;
-  debug("<TCP::Connection::finish_fast_recovery> Finished Fast Recovery - Cwnd: %u\n", cb.cwnd);
+  //printf("<TCP::Connection::finish_fast_recovery> Finished Fast Recovery - Cwnd: %u\n", cb.cwnd);
 }

src/net/tcp/tcp.cpp

@@ -492,28 +492,43 @@ bool TCP::unbind(const Socket& socket)
   return false;
 }
 
-bool TCP::add_connection(tcp::Connection_ptr conn) {
+bool TCP::add_connection(tcp::Connection_ptr conn)
+{
+  const size_t alloc_thres = max_bufsize() * Read_request::buffer_limit;
   // Stat increment number of incoming connections
   (*incoming_connections_)++;
 
   debug("<TCP::add_connection> Connection added %s \n", conn->to_string().c_str());
-  conn->bufalloc = mempool_.get_resource();
+  auto resource = mempool_.get_resource();
 
   // Reject connection if we can't allocate memory
-  if (conn->bufalloc == nullptr
-      or conn->bufalloc->allocatable() < max_bufsize() * Read_request::buffer_limit){
+  if(UNLIKELY(resource == nullptr or resource->allocatable() < alloc_thres))
+  {
     conn->_on_cleanup_ = nullptr;
     conn->abort();
     return false;
   }
 
+  conn->bufalloc = std::move(resource);
+
+  //printf("New inc conn %s allocatable=%zu\n", conn->to_string().c_str(), conn->bufalloc->allocatable());
+
   Expects(conn->bufalloc != nullptr);
   conn->_on_cleanup({this, &TCP::close_connection});
   return connections_.emplace(conn->tuple(), conn).second;
 }
 
 Connection_ptr TCP::create_connection(Socket local, Socket remote, ConnectCallback cb)
 {
+  const size_t alloc_thres = max_bufsize() * Read_request::buffer_limit;
+
+  auto resource = mempool_.get_resource();
+  // Don't create connection if we can't allocate memory
+  if(UNLIKELY(resource == nullptr or resource->allocatable() < alloc_thres))
+  {
+    throw TCP_error{"Unable to create new connection: Not enough allocatable memory"};
+  }
+
   // Stat increment number of outgoing connections
   (*outgoing_connections_)++;
 
@@ -523,7 +538,10 @@ Connection_ptr TCP::create_connection(Socket local, Socket remote, ConnectCallba
       )
     ).first->second;
   conn->_on_cleanup({this, &TCP::close_connection});
-  conn->bufalloc = mempool_.get_resource();
+  conn->bufalloc = std::move(resource);
+
+  //printf("New out conn %s allocatable=%zu\n", conn->to_string().c_str(), conn->bufalloc->allocatable());
+
   Expects(conn->bufalloc != nullptr);
   return conn;
 }