compiled now, bug in inital packet

Author: Tianyi Chen

src/include/network/connection_handle.h

@@ -179,7 +179,7 @@ class ConnectionHandle {
    * Process SSL handshake to generate valid SSL connection context
    * for further communications
    */
-  Transition ConnectionHandle::SSLHandshake();
+  Transition SSLHandshake();
 
   /**
    * Set the socket to non-blocking mode
@@ -204,7 +204,7 @@ class ConnectionHandle {
    * Determine if there is still responses in the buffer
    */
   inline bool HasResponse() {
-    return (protocol_handler_->responses.size() != 0) || (wbuf_->buf_size != 0);
+    return (protocol_handler_->responses_.size() != 0) || (wbuf_->buf_size != 0);
   }
 
   int sock_fd_;                            // socket file descriptor

src/include/network/network_state.h

@@ -24,7 +24,7 @@ enum class ConnState {
   PROCESS,     // State that runs the network protocol on received data
   CLOSING,     // State for closing the client connection
   GET_RESULT,  // State when triggered by worker thread that completes the task.
-  PROCESS_WRITE_SSL_HANDSHAKE  // State to flush out responses and  doing (Real)
+  PROCESS_WRITE_SSL_HANDSHAKE,  // State to flush out responses and  doing (Real)
                                // SSL handshake
 };
 
@@ -48,7 +48,7 @@ enum class Transition {
   GET_RESULT,
   FINISH,
   RETRY,
-  NEED_SSL_HANDSHAKE
+  NEED_SSL_HANDSHAKE,
 };
 }  // namespace network
 }  // namespace peloton

src/include/network/postgres_protocol_handler.h

@@ -47,16 +47,7 @@ class PostgresProtocolHandler : public ProtocolHandler {
    */
   ProcessResult Process(Buffer &rbuf, size_t thread_id);
 
-  void Reset();
-
-  void GetResult();
-
- private:
-  //===--------------------------------------------------------------------===//
-  // STATIC HELPERS
-  //===--------------------------------------------------------------------===//
-
-  // Deserialize the parameter types from packet
+  // Deserialize the parame types from packet
   static size_t ReadParamType(InputPacket *pkt, int num_params,
                               std::vector<int32_t> &param_types);
 
@@ -70,6 +61,15 @@ class PostgresProtocolHandler : public ProtocolHandler {
       std::vector<std::pair<type::TypeId, std::string>> &bind_parameters,
       std::vector<type::Value> &param_values, std::vector<int16_t> &formats);
 
+  void Reset();
+
+  void GetResult();
+
+ private:
+  //===--------------------------------------------------------------------===//
+  // STATIC HELPERS
+  //===--------------------------------------------------------------------===//
+
   // Parse the input packet based on if it is the startup packet
   static bool ParseInputPacket(Buffer &rbuf, InputPacket &rpkt,
                                bool startup_format);
@@ -90,12 +90,12 @@ class PostgresProtocolHandler : public ProtocolHandler {
   /**
    * @brief Routine to deal with the first packet from the client
    */
-  ProcessResult ProcessInitialPackets(InputPacket *pkt);
+  ProcessResult ProcessInitialPacket(InputPacket *pkt);
 
   /**
    * @brief Main Switch function to process general packets
    */
-  ProcessResult ProcessNormalPackets(InputPacket *pkt, const size_t thread_id);
+  ProcessResult ProcessNormalPacket(InputPacket *pkt, const size_t thread_id);
 
   /**
    * @brief Helper function to process startup packet

src/include/network/protocol_handler.h

@@ -29,11 +29,9 @@ class ProtocolHandler {
 
   virtual ~ProtocolHandler();
 
-  /* Main switch case wrapper to process every packet apart from the startup
-   * packet. Avoid flushing the response for extended protocols. */
-
   /**
-   * 
+   * Main switch case wrapper to process every packet apart from the startup
+   * packet. Avoid flushing the response for extended protocols. 
    */
   virtual ProcessResult Process(Buffer &rbuf, const size_t thread_id);
 

src/network/connection_handle.cpp

@@ -105,7 +105,7 @@ DEF_TRANSITION_GRAPH
     ON(NEED_DATA) SET_STATE_TO(READ) AND_INVOKE(FillReadBuffer)
     ON(GET_RESULT) SET_STATE_TO(GET_RESULT) AND_WAIT
     ON(FINISH) SET_STATE_TO(CLOSING) AND_INVOKE(CloseSocket)
-    ON(NEED_SSL_HANDSHAKE) SET_STATE_TO(PROCESS_WRITE_SSL_HANDSHKE) 
+    ON(NEED_SSL_HANDSHAKE) SET_STATE_TO(PROCESS_WRITE_SSL_HANDSHAKE) 
       AND_INVOKE(ProcessWrite_SSLHandshake)
   END_DEF
 
@@ -120,11 +120,6 @@ DEF_TRANSITION_GRAPH
     ON(PROCEED) SET_STATE_TO(WRITE) AND_INVOKE(ProcessWrite)
   END_DEF
 
-  DEFINE_STATE(CLOSING)
-    ON(PROCEED) SET_STATE_TO(CLOSED) AND_WAIT
-    ON(NEED_DATA) SET_STATE_TO(CLOSING) AND_WAIT
-  END_DEF
-
 END_DEF
 // clang-format on
 
@@ -188,9 +183,9 @@ void ConnectionHandle::UpdateEventFlags(short flags) {
 
 WriteState ConnectionHandle::WritePackets() {
   // iterate through all the packets
-  for (; next_response_ < protocol_handler_->responses.size();
+  for (; next_response_ < protocol_handler_->responses_.size();
        next_response_++) {
-    auto pkt = protocol_handler_->responses[next_response_].get();
+    auto pkt = protocol_handler_->responses_[next_response_].get();
     LOG_TRACE("To send packet with type: %c, len %lu",
               static_cast<char>(pkt->msg_type), pkt->len);
     // write is not ready during write. transit to WRITE
@@ -201,7 +196,7 @@ WriteState ConnectionHandle::WritePackets() {
   }
 
   // Done writing all packets. clear packets
-  protocol_handler_->responses.clear();
+  protocol_handler_->responses_.clear();
   next_response_ = 0;
 
   if (protocol_handler_->GetFlushFlag()) {
@@ -442,6 +437,7 @@ std::string ConnectionHandle::WriteBufferToString() {
   return std::string(wbuf_->buf.begin(), wbuf_->buf.end());
 }
 
+// TODO (Tianyi) Make this to be protocol specific
 // Writes a packet's header (type, size) into the write buffer.
 // Return false when the socket is not ready for write
 WriteState ConnectionHandle::BufferWriteBytesHeader(OutputPacket *pkt) {
@@ -473,14 +469,13 @@ WriteState ConnectionHandle::BufferWriteBytesHeader(OutputPacket *pkt) {
   // make len include its field size as well
   len_nb = htonl(len + sizeof(int32_t));
 
-  if (finish_startup_packet_) {
-    // append the bytes of this integer in network-byte order
-    std::copy(reinterpret_cast<uchar *>(&len_nb),
-              reinterpret_cast<uchar *>(&len_nb) + 4,
-              std::begin(wbuf_->buf) + wbuf_->buf_ptr);
-    // move the write buffer pointer and update size of the socket buffer
-    wbuf_->buf_ptr += sizeof(int32_t);
-  }
+  // TODO (Tianyi) Check why we need finish startup packet before
+  // append the bytes of this integer in network-byte order
+  std::copy(reinterpret_cast<uchar *>(&len_nb),
+            reinterpret_cast<uchar *>(&len_nb) + 4,
+            std::begin(wbuf_->buf) + wbuf_->buf_ptr);
+  // move the write buffer pointer and update size of the socket buffer
+  wbuf_->buf_ptr += sizeof(int32_t);
 
   wbuf_->buf_size = wbuf_->buf_ptr;
 
@@ -680,8 +675,11 @@ Transition ConnectionHandle::Process() {
       return Transition::GET_RESULT;
     case ProcessResult::TERMINATE:
       return Transition::FINISH;
-    case ProcessResult::NEED_SSL_HANDSHAKE：
+    case ProcessResult::NEED_SSL_HANDSHAKE:
       return Transition::NEED_SSL_HANDSHAKE;
+    default:
+      LOG_ERROR("Unknown process result");
+      throw NetworkProcessException("Unknown process result");
   }
 }
 
@@ -690,7 +688,6 @@ Transition ConnectionHandle::ProcessWrite() {
   switch (WritePackets()) {
     case WriteState::COMPLETE:
       UpdateEventFlags(EV_READ | EV_PERSIST);
-      if (!finish_startup_packet_) return Transition::NEED_DATA;
       return Transition::PROCEED;
     case WriteState::NOT_READY:
       return Transition::NONE;

src/network/postgres_protocol_handler.cpp

@@ -21,6 +21,7 @@
 #include "expression/expression_util.h"
 #include "network/marshal.h"
 #include "network/postgres_protocol_handler.h"
+#include "network/peloton_server.h"
 #include "parser/postgresparser.h"
 #include "planner/abstract_plan.h"
 #include "planner/delete_plan.h"
@@ -57,7 +58,6 @@ const std::unordered_map<std::string, std::string>
 
 PostgresProtocolHandler::PostgresProtocolHandler(tcop::TrafficCop *traffic_cop)
     : ProtocolHandler(traffic_cop),
-      stage_(CommStage::SSL_SETUP),
       txn_state_(NetworkTransactionStateType::IDLE) {}
 
 PostgresProtocolHandler::~PostgresProtocolHandler() {}
@@ -429,7 +429,7 @@ void PostgresProtocolHandler::ExecBindMessage(InputPacket *pkt) {
     std::unique_ptr<OutputPacket> response(new OutputPacket());
     // Send Bind complete response
     response->msg_type = NetworkMessageType::BIND_COMPLETE;
-    responses.push_back(std::move(response));
+    responses_.push_back(std::move(response));
     // TODO(Tianyi) This is a hack to respond correct describe message
     // as well as execute message
     skipped_stmt_ = true;
@@ -456,7 +456,7 @@ void PostgresProtocolHandler::ExecBindMessage(InputPacket *pkt) {
     std::unique_ptr<OutputPacket> response(new OutputPacket());
     // Send Bind complete response
     response->msg_type = NetworkMessageType::BIND_COMPLETE;
-    responses.push_back(std::move(response));
+    responses_.push_back(std::move(response));
     return;
   }
 
@@ -536,7 +536,7 @@ void PostgresProtocolHandler::ExecBindMessage(InputPacket *pkt) {
   // send bind complete
   std::unique_ptr<OutputPacket> response(new OutputPacket());
   response->msg_type = NetworkMessageType::BIND_COMPLETE;
-  responses.push_back(std::move(response));
+  responses_.push_back(std::move(response));
 }
 
 size_t PostgresProtocolHandler::ReadParamType(
@@ -694,7 +694,7 @@ ProcessResult PostgresProtocolHandler::ExecDescribeMessage(InputPacket *pkt) {
     // send 'no-data' message
     std::unique_ptr<OutputPacket> response(new OutputPacket());
     response->msg_type = NetworkMessageType::NO_DATA_RESPONSE;
-    responses.push_back(std::move(response));
+    responses_.push_back(std::move(response));
     return ProcessResult::COMPLETE;
   }
 
@@ -868,28 +868,29 @@ void PostgresProtocolHandler::ExecCloseMessage(InputPacket *pkt) {
   // Send close complete response
   std::unique_ptr<OutputPacket> response(new OutputPacket());
   response->msg_type = NetworkMessageType::CLOSE_COMPLETE;
-  responses.push_back(std::move(response));
+  responses_.push_back(std::move(response));
 }
 
-bool PostgresProtocolHandler::ParseInputPacket(Buffer &rbuf, InputPacket,
-                                               bool starup_format) {
-  if (request.header_parsed == false) {
+bool PostgresProtocolHandler::ParseInputPacket(Buffer &rbuf, InputPacket &rpkt,
+                                               bool startup_format) {
+  if (rpkt.header_parsed == false) {
     // parse out the header first
-    if (ReadPacketHeader(rbuf, request, startup_format) == false) {
+    if (ReadPacketHeader(rbuf, rpkt, startup_format) == false) {
       // need more data
       return false;
     }
   }
 
-  PL_ASSERT(request.header_parsed == true);
+  PL_ASSERT(rpkt.header_parsed == true);
 
-  if (request.is_initialized == false) {
+  if (rpkt.is_initialized == false) {
     // packet needs to be initialized with rest of the contents
-    if (PostgresProtocolHandler::ReadPacket(rbuf, request) == false) {
+    if (PostgresProtocolHandler::ReadPacket(rbuf, rpkt) == false) {
       // need more data
       return false;
     }
   }
+  return true;
 }
 
 // The function tries to do a preliminary read to fetch the size value and
@@ -967,27 +968,14 @@ bool PostgresProtocolHandler::ReadPacket(Buffer &rbuf, InputPacket &rpkt) {
   return true;
 }
 
-ProcessResult PostgresProtocolHandler::Process(Buffer &rbuf,
-                                               const size_t thread_id) {
-  if (!ParseInputPacket(rbuf, request_, init_stage_))
-    return ProcessResult::MORE_DATA_REQUIRED;
-
-  ProcessResult process_status;
-  if (init_stage_) {
-    process_status = ProcessInitialPacket(request_);
-  } else {
-    process_status = ProcessNormalPacket(request_);
-  }
-  request_.Reset();
-
-  return process_status;
-}
-
+/*
+ * process_startup_packet - Processes the startup packet
+ *  (after the size field of the header).
+ */
 ProcessResult PostgresProtocolHandler::ProcessInitialPacket(InputPacket *pkt) {
   int32_t proto_version = PacketGetInt(pkt, sizeof(int32_t));
   LOG_INFO("protocol version: %d", proto_version);
 
-  force_flush_ = true;
   // TODO(Yuchen): consider more about return value
   if (proto_version == SSL_MESSAGE_VERNO) {
     LOG_TRACE("process SSL MESSAGE");
@@ -996,6 +984,7 @@ ProcessResult PostgresProtocolHandler::ProcessInitialPacket(InputPacket *pkt) {
     response->msg_type =
         ssl_able ? NetworkMessageType::SSL_YES : NetworkMessageType::SSL_NO;
     responses_.push_back(std::move(response));
+    force_flush_ = true;
     return ssl_able ? ProcessResult::NEED_SSL_HANDSHAKE
                     : ProcessResult::COMPLETE;
   } else {
@@ -1023,7 +1012,7 @@ ProcessResult PostgresProtocolHandler::ProcessStartupPacket(
     if (pkt->ptr >= pkt->len) break;
     GetStringToken(pkt, value);
     LOG_TRACE("Option value is %s", token.c_str());
-    client.cmdline_options[token] = value;
+    cmdline_options_[token] = value;
     if (token.compare("database") == 0) {
       traffic_cop_->SetDefaultDatabaseName(value);
     }
@@ -1035,10 +1024,32 @@ ProcessResult PostgresProtocolHandler::ProcessStartupPacket(
   SendStartupResponse();
 
   init_stage_ = false;
+  force_flush_ = true;
   return ProcessResult::COMPLETE;
 }
 
-ProcessResult PostgresProtocolHandler::ProcessNormalPackets(
+ProcessResult PostgresProtocolHandler::Process(Buffer &rbuf,
+                                               const size_t thread_id) {
+  InputPacket rpkt;
+  rpkt.Reset();
+  if (!ParseInputPacket(rbuf, rpkt, init_stage_))
+    return ProcessResult::MORE_DATA_REQUIRED;
+
+  ProcessResult process_status;
+  if (init_stage_) {
+    process_status = ProcessInitialPacket(&rpkt);
+  } else {
+    process_status = ProcessNormalPacket(&rpkt, thread_id);
+  }
+
+  return process_status;
+}
+
+/*
+ * process_packet - Main switch block; process incoming packets,
+ *  Returns false if the session needs to be closed.
+ */
+ProcessResult PostgresProtocolHandler::ProcessNormalPacket(
     InputPacket *pkt, const size_t thread_id) {
   LOG_TRACE("Message type: %c", static_cast<unsigned char>(pkt->msg_type));
   // We don't set force_flush to true for `PBDE` messages because they're
@@ -1093,7 +1104,6 @@ ProcessResult PostgresProtocolHandler::ProcessNormalPackets(
   }
   return ProcessResult::COMPLETE;
 }
-
 void PostgresProtocolHandler::MakeHardcodedParameterStatus(
     const std::pair<std::string, std::string> &kv) {
   std::unique_ptr<OutputPacket> response(new OutputPacket());

src/network/protocol_handler.cpp

@@ -23,19 +23,15 @@ ProtocolHandler::ProtocolHandler(tcop::TrafficCop *traffic_cop) {
 
 ProtocolHandler::~ProtocolHandler() {}
 
-/* Manage the startup packet */
-//  bool ManageStartupPacket();
-void ProtocolHandler::SendInitialResponse() {}
-
 ProcessResult ProtocolHandler::Process(
     UNUSED_ATTRIBUTE Buffer &rbuf, UNUSED_ATTRIBUTE const size_t thread_id) {
   return ProcessResult::TERMINATE;
 }
 
 void ProtocolHandler::Reset() {
   SetFlushFlag(false);
-  responses.clear();
-  request.Reset();
+  responses_.clear();
+  request_.Reset();
 }
 
 void ProtocolHandler::GetResult() {}