Protocol Layer Refactor

src/common/notifiable_task.cpp

@@ -17,7 +17,7 @@
 
 namespace peloton {
 
-NotifiableTask::NotifiableTask(int task_id) : task_id_(task_id) {
+NotifiableTask::NotifiableTask(size_t task_id) : task_id_(task_id) {
   base_ = EventUtil::EventBaseNew();
   // For exiting a loop
   terminate_ = RegisterManualEvent([](int, short, void *arg) {

src/common/portal.cpp

@@ -18,12 +18,10 @@ namespace peloton {
 
 Portal::Portal(const std::string& portal_name,
                std::shared_ptr<Statement> statement,
-               std::vector<type::Value> bind_parameters,
-               std::shared_ptr<stats::QueryMetric::QueryParams> param_stat)
+               std::vector<type::Value> bind_parameters)
     : portal_name_(portal_name),
       statement_(statement),
-      bind_parameters_(std::move(bind_parameters)),
-      param_stat_(param_stat) {}
+      bind_parameters_(std::move(bind_parameters)) {}
 
 Portal::~Portal() { statement_.reset(); }
 

src/common/statement.cpp

@@ -70,11 +70,11 @@ std::string Statement::GetQueryTypeString() const { return query_type_string_; }
 
 QueryType Statement::GetQueryType() const { return query_type_; }
 
-void Statement::SetParamTypes(const std::vector<int32_t>& param_types) {
+void Statement::SetParamTypes(const std::vector<PostgresValueType>& param_types) {
   param_types_ = param_types;
 }
 
-std::vector<int32_t> Statement::GetParamTypes() const { return param_types_; }
+std::vector<PostgresValueType> Statement::GetParamTypes() const { return param_types_; }
 
 void Statement::SetTupleDescriptor(
     const std::vector<FieldInfo>& tuple_descriptor) {

src/executor/copy_executor.cpp

@@ -22,9 +22,9 @@
 #include "executor/logical_tile_factory.h"
 #include "planner/export_external_file_plan.h"
 #include "storage/table_factory.h"
-#include "network/postgres_protocol_handler.h"
 #include "common/exception.h"
 #include "common/macros.h"
+#include "network/marshal.h"
 
 namespace peloton {
 namespace executor {
@@ -202,7 +202,7 @@ bool CopyExecutor::DExecute() {
           // Read param types
           types.resize(num_params);
           //TODO: Instead of passing packet to executor, some data structure more generic is need
-          network::PostgresProtocolHandler::ReadParamType(&packet, num_params, types);
+          network::OldReadParamType(&packet, num_params, types);
 
           // Write all the types to output file
           for (int i = 0; i < num_params; i++) {
@@ -219,7 +219,7 @@ bool CopyExecutor::DExecute() {
           // Read param formats
           formats.resize(num_params);
           //TODO: Instead of passing packet to executor, some data structure more generic is need
-          network::PostgresProtocolHandler::ReadParamFormat(&packet, num_params, formats);
+          network::OldReadParamFormat(&packet, num_params, formats);
 
         } else if (origin_col_id == param_val_col_id) {
           // param_values column
@@ -230,9 +230,9 @@ bool CopyExecutor::DExecute() {
           bind_parameters.resize(num_params);
           param_values.resize(num_params);
           //TODO: Instead of passing packet to executor, some data structure more generic is need
-          network::PostgresProtocolHandler::ReadParamValue(&packet, num_params, types,
-                                              bind_parameters, param_values,
-                                              formats);
+          network::OldReadParamValue(&packet, num_params, types,
+                                     bind_parameters, param_values,
+                                     formats);
 
           // Write all the values to output file
           for (int i = 0; i < num_params; i++) {

src/include/common/internal_types.h

@@ -862,10 +862,10 @@ enum class ResultType {
   SUCCESS = 1,
   FAILURE = 2,
   ABORTED = 3,  // aborted
-  NOOP = 4,     // no op
-  UNKNOWN = 5,
-  QUEUING = 6,
-  TO_ABORT = 7,
+  NOOP = 4,     // no op  // TODO Remove this type
+  UNKNOWN = 5,            // TODO Remove this type
+  QUEUING = 6,            // TODO Remove this type
+  TO_ABORT = 7,           // TODO Remove this type
 };
 std::string ResultTypeToString(ResultType type);
 ResultType StringToResultType(const std::string &str);
@@ -1419,25 +1419,14 @@ typedef std::map<oid_t, std::pair<oid_t, oid_t>> column_map_type;
 //===--------------------------------------------------------------------===//
 // Wire protocol typedefs
 //===--------------------------------------------------------------------===//
-#define SOCKET_BUFFER_SIZE 8192
+#define SOCKET_BUFFER_CAPACITY 8192
 
 /* byte type */
 typedef unsigned char uchar;
 
 /* type for buffer of bytes */
 typedef std::vector<uchar> ByteBuf;
 
-//===--------------------------------------------------------------------===//
-// Packet Manager: ProcessResult
-//===--------------------------------------------------------------------===//
-enum class ProcessResult {
-  COMPLETE,
-  TERMINATE,
-  PROCESSING,
-  MORE_DATA_REQUIRED,
-  NEED_SSL_HANDSHAKE,
-};
-
 enum class NetworkProtocolType {
   POSTGRES_JDBC,
   POSTGRES_PSQL,
@@ -1449,6 +1438,19 @@ enum class SSLLevel {
   SSL_VERIIFY = 2,
 };
 
+using CallbackFunc = std::function<void(void)>;
+using BindParameter = std::pair<type::TypeId, std::string>;
+
+enum class PostgresDataFormat : int16_t {
+  TEXT = 0,
+  BINARY = 1
+};
+
+enum class PostgresNetworkObjectType : uchar {
+  PORTAL = 'P',
+  STATEMENT = 'S'
+};
+
 // Eigen/Matrix types used in brain
 // TODO(saatvik): Generalize Eigen utilities across all types
 typedef std::vector<std::vector<float>> matrix_t;

src/include/common/notifiable_task.h

@@ -49,7 +49,7 @@ class NotifiableTask {
    * Constructs a new NotifiableTask instance.
    * @param task_id a unique id assigned to this task
    */
-  explicit NotifiableTask(int task_id);
+  explicit NotifiableTask(size_t task_id);
 
   /**
    * Destructs this NotifiableTask. All events currently registered to its base
@@ -60,7 +60,7 @@ class NotifiableTask {
   /**
    * @return unique id assigned to this task
    */
-  inline int Id() const { return task_id_; }
+  inline size_t Id() const { return task_id_; }
 
   /**
    * @brief Register an event with the event base associated with this
@@ -183,7 +183,7 @@ class NotifiableTask {
   inline void ExitLoop(int, short) { ExitLoop(); }
 
  private:
-  const int task_id_;
+  const size_t task_id_;
   struct event_base *base_;
 
   // struct event and lifecycle management

src/include/common/portal.h

@@ -31,19 +31,14 @@ class Portal {
   Portal &operator=(Portal &&) = delete;
 
   Portal(const std::string &portal_name, std::shared_ptr<Statement> statement,
-         std::vector<type::Value> bind_parameters,
-         std::shared_ptr<stats::QueryMetric::QueryParams> param_stat);
+         std::vector<type::Value> bind_parameters);
 
   ~Portal();
 
   std::shared_ptr<Statement> GetStatement() const;
 
   const std::vector<type::Value> &GetParameters() const;
 
-  inline std::shared_ptr<stats::QueryMetric::QueryParams> GetParamStat() const {
-    return param_stat_;
-  }
-
   // Portal name
   std::string portal_name_;
 
@@ -52,9 +47,6 @@ class Portal {
 
   // Values bound to the statement of this portal
   std::vector<type::Value> bind_parameters_;
-
-  // The serialized params for stats collection
-  std::shared_ptr<stats::QueryMetric::QueryParams> param_stat_;
 };
 
 }  // namespace peloton

src/include/common/statement.h

@@ -65,21 +65,21 @@ class Statement : public Printable {
 
   QueryType GetQueryType() const;
 
-  void SetParamTypes(const std::vector<int32_t> &param_types);
+  void SetParamTypes(const std::vector<PostgresValueType> &param_types);
 
-  std::vector<int32_t> GetParamTypes() const;
+  std::vector<PostgresValueType> GetParamTypes() const;
 
   void SetTupleDescriptor(const std::vector<FieldInfo> &tuple_descriptor);
 
-  void SetReferencedTables(const std::set<oid_t> table_ids);
+  void SetReferencedTables(std::set<oid_t> table_ids);
 
   const std::set<oid_t> GetReferencedTables() const;
 
   void SetPlanTree(std::shared_ptr<planner::AbstractPlan> plan_tree);
 
   const std::shared_ptr<planner::AbstractPlan> &GetPlanTree() const;
 
-  std::unique_ptr<parser::SQLStatementList> const &GetStmtParseTreeList() {
+  const std::unique_ptr<parser::SQLStatementList> &GetStmtParseTreeList() {
     return sql_stmt_list_;
   }
 
@@ -113,7 +113,7 @@ class Statement : public Printable {
   std::string query_type_string_;
 
   // format codes of the parameters
-  std::vector<int32_t> param_types_;
+  std::vector<PostgresValueType> param_types_;
 
   // schema of result tuple
   std::vector<FieldInfo> tuple_descriptor_;

src/include/network/connection_dispatcher_task.h

@@ -15,7 +15,7 @@
 #include "common/notifiable_task.h"
 #include "concurrency/epoch_manager_factory.h"
 #include "connection_handler_task.h"
-#include "network_state.h"
+#include "network_types.h"
 
 namespace peloton {
 namespace network {

src/include/network/connection_handle.h

@@ -33,8 +33,9 @@
 #include "marshal.h"
 #include "network/connection_handler_task.h"
 #include "network/network_io_wrappers.h"
-#include "network_state.h"
-#include "protocol_handler.h"
+#include "network/network_types.h"
+#include "network/protocol_interpreter.h"
+#include "network/postgres_protocol_interpreter.h"
 
 #include <openssl/err.h>
 #include <openssl/ssl.h>
@@ -43,19 +44,23 @@ namespace peloton {
 namespace network {
 
 /**
- * @brief A ConnectionHandle encapsulates all information about a client
- * connection for its entire duration. This includes a state machine and the
- * necessary libevent infrastructure for a handler to work on this connection.
+ * A ConnectionHandle encapsulates all information we need to do IO about
+ * a client connection for its entire duration. This includes a state machine
+ * and the necessary libevent infrastructure for a handler to work on this
+ * connection.
  */
 class ConnectionHandle {
  public:
+
   /**
    * Constructs a new ConnectionHandle
    * @param sock_fd Client's connection fd
    * @param handler The handler responsible for this handle
    */
   ConnectionHandle(int sock_fd, ConnectionHandlerTask *handler);
 
+  DISALLOW_COPY_AND_MOVE(ConnectionHandle);
+
   /**
    * @brief Signal to libevent that this ConnectionHandle is ready to handle
    * events
@@ -70,14 +75,6 @@ class ConnectionHandle {
     workpool_event_ = conn_handler_->RegisterManualEvent(
         METHOD_AS_CALLBACK(ConnectionHandle, HandleEvent), this);
 
-    // TODO(Tianyi): should put the initialization else where.. check
-    // correctness first.
-    tcop_.SetTaskCallback(
-        [](void *arg) {
-          struct event *event = static_cast<struct event *>(arg);
-          event_active(event, EV_WRITE, 0);
-        },
-        workpool_event_);
 
     network_event_ = conn_handler_->RegisterEvent(
         io_wrapper_->GetSocketFd(), EV_READ | EV_PERSIST,
@@ -94,8 +91,20 @@ class ConnectionHandle {
   /* State Machine Actions */
   // TODO(Tianyu): Write some documentation when feeling like it
   inline Transition TryRead() { return io_wrapper_->FillReadBuffer(); }
-  Transition TryWrite();
-  Transition Process();
+
+  inline Transition TryWrite() {
+    if (io_wrapper_->ShouldFlush())
+      return io_wrapper_->FlushAllWrites();
+    return Transition::PROCEED;
+  }
+
+  inline Transition Process() {
+    return protocol_interpreter_->
+        Process(io_wrapper_->GetReadBuffer(),
+                io_wrapper_->GetWriteQueue(),
+                [=] { event_active(workpool_event_, EV_WRITE, 0); });
+  }
+
   Transition GetResult();
   Transition TrySslHandshake();
   Transition TryCloseConnection();
@@ -173,26 +182,15 @@ class ConnectionHandle {
   };
 
   friend class StateMachine;
-  friend class NetworkIoWrapperFactory;
-
-  /**
-   * @brief: Determine if there is still responses in the buffer
-   * @return true if there is still responses to flush out in either wbuf or
-   * responses
-   */
-  inline bool HasResponse() {
-    return (protocol_handler_->responses_.size() != 0) ||
-           (io_wrapper_->wbuf_->size_ != 0);
-  }
+  friend class ConnectionHandleFactory;
 
+  // A raw pointer is used here because references cannot be rebound.
   ConnectionHandlerTask *conn_handler_;
-  std::shared_ptr<NetworkIoWrapper> io_wrapper_;
-  StateMachine state_machine_;
+  std::unique_ptr<NetworkIoWrapper> io_wrapper_;
+  // TODO(Tianyu): Probably use a factory for this
+  std::unique_ptr<ProtocolInterpreter> protocol_interpreter_;
+  StateMachine state_machine_{};
   struct event *network_event_ = nullptr, *workpool_event_ = nullptr;
-  std::unique_ptr<ProtocolHandler> protocol_handler_ = nullptr;
-  tcop::TrafficCop tcop_;
-  // TODO(Tianyu): Put this into protocol handler in a later refactor
-  unsigned int next_response_ = 0;
 };
 }  // namespace network
 }  // namespace peloton