Passing buffer for reading data from userland

Author: hamidr

includes/connection.hpp

@@ -40,12 +40,12 @@ namespace async_redis {
             req_queue_.emplace(reply_cb, nullptr);
 
             if (req_queue_.size() == 1)
-              socket_->async_read(std::bind(&connection::reply_received, this, std::placeholders::_1, std::placeholders::_2));
+              socket_->async_read(data_, max_data_size, std::bind(&connection::reply_received, this, std::placeholders::_1));
           });
       }
 
     private:
-      void reply_received(const char* data, int len) {
+      void reply_received(int len) {
         ssize_t acc = 0;
 
         while (acc < len && req_queue_.size()) {
@@ -57,7 +57,7 @@ namespace async_redis {
           if (0 != len && -1 != len) {
 
             bool is_finished = false;
-            acc += ParserPolicy(parser).append_chunk(data + acc, len - acc, is_finished);
+            acc += ParserPolicy(parser).append_chunk(data_ + acc, len - acc, is_finished);
 
             if (!is_finished)
               break;
@@ -68,13 +68,15 @@ namespace async_redis {
         }
 
         if (req_queue_.size() != 0)
-          socket_->async_read(std::bind(&connection::reply_received, this, std::placeholders::_1, std::placeholders::_2));
+          socket_->async_read(data_, max_data_size, std::bind(&connection::reply_received, this, std::placeholders::_1));
       }
 
     private:
       std::unique_ptr<SocketType> socket_;
       InputOutputHandler& event_loop_;
       std::queue<std::tuple<reply_cb_t, parser_t>> req_queue_;
+      enum { max_data_size = 1024 };
+      char data_[max_data_size];
     };
 
   }

includes/network/async_socket.hpp

@@ -27,7 +27,7 @@ namespace async_redis {
     {
     public:
       using socket_identifier_t  = typename InputOutputHanler::socket_identifier_t;
-      using recv_cb_t         = std::function<void (const char*, int)>;
+      using recv_cb_t         = std::function<void (int)>;
       using ready_cb_t        = std::function<void ()>;
       using connect_handler_t = std::function<void (bool)>;
 
@@ -71,8 +71,11 @@ namespace async_redis {
         return ::accept(fd_, nullptr, nullptr);
       }
 
-      inline bool close() {
-        return ::close(fd_) == 0;
+      bool close() {
+        auto res = ::close(fd_) == 0;
+        is_connected_ = false;
+        fd_ = -1;
+        return res;
       }
 
       void async_write(const string& data, const ready_cb_t& cb)
@@ -86,21 +89,18 @@ namespace async_redis {
           });
       }
 
-      void async_read(const recv_cb_t& cb)
+      void async_read(char *buffer, int max_len, const recv_cb_t& cb)
       {
         if (!is_connected())
           return;
 
-        return io_.async_read(id_, [&, cb]() {
-            auto l = receive(buffer_, max_length);
+        return io_.async_read(id_, [&, buffer, max_len,  cb]() {
+            auto l = receive(buffer, max_len);
             if (l == 0) {
-              is_connected_ = false;
               io_.unwatch(id_);
-              id_ = nullptr;
               close();
-              fd_ = -1;
             }
-            cb(buffer_, l);
+            cb(l);
           });
       }
 
@@ -162,8 +162,6 @@ namespace async_redis {
       InputOutputHanler& io_;
       socket_identifier_t id_ = nullptr;
       int fd_ = -1;
-      enum { max_length = 1024 };
-      char buffer_[max_length]; //TODO: move to user land!
     };
   }
 }

includes/network/tcp_server.hpp

@@ -30,14 +30,14 @@ namespace tcp_server {
     }
 
     void accept(std::shared_ptr<tcp_socket> socket) {
-      auto receiver = std::bind(&tcp_server::chunk_received, this, std::placeholders::_1, std::placeholders::_2, socket);
-      socket->async_read(receiver);
+      auto receiver = std::bind(&tcp_server::chunk_received, this, std::placeholders::_1, socket);
+      socket->async_read(buffer_, max_buffer_length, receiver);
 
       conns_.emplace(socket, nullptr);
     }
 
   private:
-    void chunk_received(const char* data, int len, std::shared_ptr<tcp_socket>& socket)
+    void chunk_received(int len, std::shared_ptr<tcp_socket>& socket)
     {
       std::string command;
 
@@ -48,7 +48,7 @@ namespace tcp_server {
 
       for(int n = 0; n < len; ++n) {
 
-        char c = data[n];
+        char c = buffer_[n];
         switch(c)
         {
         case '\r':
@@ -73,8 +73,8 @@ namespace tcp_server {
         return; // dont read
       }
 
-      auto receiver = std::bind(&tcp_server::chunk_received, this, std::placeholders::_1, std::placeholders::_2, socket);
-      socket->async_read(receiver);
+      auto receiver = std::bind(&tcp_server::chunk_received, this, std::placeholders::_1, socket);
+      socket->async_read(buffer_, max_buffer_length, receiver);
     }
 
   private:
@@ -83,6 +83,8 @@ namespace tcp_server {
     socket_t listener_;
     InputOutputHandler& loop_;
     std::unordered_map<socket_t, void*> conns_;
+    enum { max_buffer_length = 1024 };
+    char buffer_[max_buffer_length];
   };
 
 }

src/event_loop/event_loop_ev.cpp

@@ -48,7 +48,7 @@ void event_loop_ev::async_timeout(double time, const action& cb )
 
 void event_loop_ev::read_handler(EV_P_ ev_io* w, int revents)
 {
-  if (!(revents & EV_READ)) {
+  if (revents & EV_ERROR) {
     // LOG_ERR("WRONG EVENT ON read_handler");
     return;
   }
@@ -78,7 +78,7 @@ void event_loop_ev::read_handler(EV_P_ ev_io* w, int revents)
 
 void event_loop_ev::write_handler(EV_P_ ev_io* w, int revents)
 {
-  if (!(revents & EV_WRITE)) {
+  if (revents & EV_ERROR) {
     // LOG_ERR("WRONG EVENT ON read_handler");
     return;
   }
@@ -133,6 +133,7 @@ event_loop_ev::socket_identifier_t event_loop_ev::watch(int fd)
 void event_loop_ev::unwatch(socket_identifier_t& id)
 {
   id->free_me = true;
+  id = nullptr;
 }
 
 }}