tcp: Passing ReadBuffer as rvalue and other optimizations

* Catching exceptions by reference
* Replaced asserts with Ensures/Expects

Author: AndreasAakesson

api/net/tcp/connection.hpp

@@ -283,7 +283,7 @@ class Connection : public std::enable_shared_from_this<Connection> {
     virtual size_t send(Connection&, WriteBuffer&);
 
     /** Read from a Connection [RECEIVE] */
-    virtual void receive(Connection&, ReadBuffer&);
+    virtual void receive(Connection&, ReadBuffer&&);
 
     /** Close a Connection [CLOSE] */
     virtual void close(Connection&);
@@ -542,8 +542,7 @@ class Connection : public std::enable_shared_from_this<Connection> {
     Buffer is cleared for data after every reset.
   */
   void read(size_t n, ReadCallback callback) {
-    ReadBuffer buffer = {new_shared_buffer(n), n};
-    read(buffer, callback);
+    read({new_shared_buffer(n), n}, callback);
   }
 
   /*
@@ -553,13 +552,13 @@ class Connection : public std::enable_shared_from_this<Connection> {
   void read(buffer_t buffer, size_t n, ReadCallback callback)
   { read({buffer, n}, callback); }
 
-  void read(ReadBuffer buffer, ReadCallback callback);
+  void read(ReadBuffer&& buffer, ReadCallback callback);
 
   /*
     Assign the read request (read buffer)
   */
-  void receive(ReadBuffer& buffer)
-  { read_request.buffer = {buffer}; }
+  void receive(ReadBuffer&& buffer)
+  { read_request = {buffer}; }
 
   /*
     Receive data into the current read requests buffer.

api/net/tcp/connection_states.hpp

@@ -169,7 +169,7 @@ class Connection::Established : public State {
 
   virtual size_t send(Connection&, WriteBuffer&) override;
 
-  virtual void receive(Connection&, ReadBuffer&) override;
+  virtual void receive(Connection&, ReadBuffer&&) override;
 
   virtual void close(Connection&) override;
 
@@ -207,7 +207,7 @@ class Connection::FinWait1 : public State {
     return instance;
   }
 
-  virtual void receive(Connection&, ReadBuffer&) override;
+  virtual void receive(Connection&, ReadBuffer&&) override;
 
   virtual void close(Connection&) override;
 
@@ -246,7 +246,7 @@ class Connection::FinWait2 : public State {
     return instance;
   }
 
-  virtual void receive(Connection&, ReadBuffer&) override;
+  virtual void receive(Connection&, ReadBuffer&&) override;
 
   virtual void close(Connection&) override;
 
@@ -284,7 +284,7 @@ class Connection::CloseWait : public State {
 
   virtual size_t send(Connection&, WriteBuffer&) override;
 
-  virtual void receive(Connection&, ReadBuffer&) override;
+  virtual void receive(Connection&, ReadBuffer&&) override;
 
   virtual void close(Connection&) override;
 

api/net/tcp/read_buffer.hpp

@@ -79,7 +79,7 @@ struct ReadBuffer {
   void renew() {
     remaining = capacity();
     offset = 0;
-    buffer = buffer_t(new uint8_t[remaining], std::default_delete<uint8_t[]>());
+    buffer = new_shared_buffer(remaining);
     push = false;
   }
 }; // < ReadBuffer

api/net/tcp/read_request.hpp

@@ -31,18 +31,22 @@ struct ReadRequest {
   ReadBuffer buffer;
   ReadCallback callback;
 
-  /*ReadRequest()
-    : buffer(nullptr, 0),
-      callback({this, &ReadRequest::default_read_callback})
-  {}*/
+  ReadRequest()
+    : buffer{nullptr, 0},
+      callback{nullptr}
+  {}
+
+  ReadRequest(ReadBuffer buf)
+    : ReadRequest(buf, nullptr)
+  {}
 
   ReadRequest(ReadBuffer buf, ReadCallback cb)
     : buffer(buf),
       callback(cb)
   {}
 
-  ReadRequest(size_t n = 0)
-    : buffer(buffer_t(new uint8_t[n], std::default_delete<uint8_t[]>()), n),
+  ReadRequest(size_t n)
+    : buffer({new_shared_buffer(n), n}),
       callback({this, &ReadRequest::default_read_callback})
   {}
 

src/net/tcp/connection.cpp

@@ -82,29 +82,29 @@ Socket Connection::local() const {
   return {host_.address(), local_port_};
 }
 
-void Connection::read(ReadBuffer buffer, ReadCallback callback) {
+void Connection::read(ReadBuffer&& buffer, ReadCallback callback) {
   try {
-    state_->receive(*this, buffer);
+    state_->receive(*this, std::forward<ReadBuffer>(buffer));
     read_request.callback = callback;
   }
-  catch (TCPException err) {
+  catch (const TCPException&) {
     callback(buffer.buffer, buffer.size());
   }
 }
 
 size_t Connection::receive(const uint8_t* data, size_t n, bool PUSH) {
   //printf("<Connection::receive> len=%u\n", n);
   // should not be called without an read request
-  assert(read_request.buffer.capacity());
-  assert(n);
+  Ensures(read_request.buffer.capacity());
+  Ensures(n);
   auto& buf = read_request.buffer;
   size_t received{0};
   while(n) {
     auto read = receive(buf, data+received, n);
     // nothing was read to buffer
     if(!buf.advance(read)) {
       // buffer should be full
-      assert(buf.full());
+      Expects(buf.full());
       // signal the user
       debug2("<Connection::receive> Buffer full - signal user\n");
       read_request.callback(buf.buffer, buf.size());
@@ -115,7 +115,7 @@ size_t Connection::receive(const uint8_t* data, size_t n, bool PUSH) {
     received += read;
   }
   // n shouldnt be negative
-  assert(n == 0);
+  Expects(n == 0);
 
   // end of data, signal the user
   if(PUSH) {
@@ -145,7 +145,7 @@ void Connection::write(WriteBuffer buffer, WriteCallback callback) {
       writeq.advance(written);
     }
   }
-  catch(TCPException err) {
+  catch(const TCPException&) {
     callback(0);
   }
 }
@@ -302,7 +302,7 @@ void Connection::open(bool active) {
     state_->open(*this, active);
   }
   // No remote host, or state isnt valid for opening.
-  catch (TCPException e) {
+  catch (const TCPException& e) {
     debug("<TCP::Connection::open> Cannot open Connection. \n");
     signal_error(e);
   }
@@ -314,7 +314,7 @@ void Connection::close() {
     state_->close(*this);
     if(is_state(Closed::instance()))
       signal_close();
-  } catch(TCPException err) {
+  } catch(const TCPException& err) {
     Ensures(on_error_);
     signal_error(err);
   }
@@ -342,7 +342,7 @@ void Connection::segment_arrived(Packet_ptr incoming) {
     try {
       parse_options(incoming);
     }
-    catch(TCPBadOptionException err) {
+    catch(const TCPBadOptionException& err) {
       debug("<TCP::Connection::receive> %s \n", err.what());
       drop(incoming, err.what());
       return;

src/net/tcp/connection_states.cpp

@@ -621,24 +621,24 @@ size_t Connection::CloseWait::send(Connection& tcp, WriteBuffer& buffer) {
 */
 /////////////////////////////////////////////////////////////////////
 
-void Connection::State::receive(Connection&, ReadBuffer&) {
+void Connection::State::receive(Connection&, ReadBuffer&&) {
   throw TCPException{"Connection closing."};
 }
 
-void Connection::Established::receive(Connection& tcp, ReadBuffer& buffer) {
-  tcp.receive(buffer);
+void Connection::Established::receive(Connection& tcp, ReadBuffer&& buffer) {
+  tcp.receive(std::forward<ReadBuffer>(buffer));
 }
 
-void Connection::FinWait1::receive(Connection& tcp, ReadBuffer& buffer) {
-  tcp.receive(buffer);
+void Connection::FinWait1::receive(Connection& tcp, ReadBuffer&& buffer) {
+  tcp.receive(std::forward<ReadBuffer>(buffer));
 }
 
-void Connection::FinWait2::receive(Connection& tcp, ReadBuffer& buffer) {
-  tcp.receive(buffer);
+void Connection::FinWait2::receive(Connection& tcp, ReadBuffer&& buffer) {
+  tcp.receive(std::forward<ReadBuffer>(buffer));
 }
 
-void Connection::CloseWait::receive(Connection& tcp, ReadBuffer& buffer) {
-  tcp.receive(buffer);
+void Connection::CloseWait::receive(Connection& tcp, ReadBuffer&& buffer) {
+  tcp.receive(std::forward<ReadBuffer>(buffer));
 }
 
 /////////////////////////////////////////////////////////////////////