Merge branch 'master' into master

Author: apavlo

src/include/common/exception.h

@@ -56,7 +56,8 @@ enum class ExceptionType {
   CONNECTION = 21,        // connection related
   SYNTAX = 22,            // syntax related
   SETTINGS = 23,          // settings related
-  BINDER = 24             // settings related
+  BINDER = 24,             // settings related
+  NETWORK = 25
 };
 
 class Exception : public std::runtime_error {
@@ -427,6 +428,13 @@ class ConnectionException : public Exception {
       : Exception(ExceptionType::CONNECTION, msg) {}
 };
 
+class NetworkProcessException : public Exception {
+  NetworkProcessException() = delete;
+
+  public:
+  NetworkProcessException(std::string msg) : Exception(ExceptionType::NETWORK, msg) {}
+};
+
 class SettingsException : public Exception {
   SettingsException() = delete;
 

src/include/common/thread_pool.h

@@ -10,30 +10,29 @@
 //
 //===----------------------------------------------------------------------===//
 
-
 #pragma once
 
-#include <vector>
 #include <thread>
+#include <vector>
 
 #include <boost/asio/io_service.hpp>
-#include <boost/thread/thread.hpp>
 #include <boost/bind.hpp>
 #include <boost/function.hpp>
+#include <boost/thread/thread.hpp>
 
 #include "common/macros.h"
 
 namespace peloton {
 // a wrapper for boost worker thread pool.
 class ThreadPool {
  public:
-  ThreadPool() : pool_size_(0), 
-                 dedicated_thread_count_(0),
-                 work_(io_service_) { }
+  ThreadPool()
+      : pool_size_(0), dedicated_thread_count_(0), work_(io_service_) {}
 
-  ~ThreadPool() { }
+  ~ThreadPool() {}
 
-  void Initialize(const size_t &pool_size, const size_t &dedicated_thread_count) {
+  void Initialize(const size_t &pool_size,
+                  const size_t &dedicated_thread_count) {
     current_thread_count_ = ATOMIC_VAR_INIT(0);
     pool_size_ = pool_size;
     // PL_ASSERT(pool_size_ != 0);
@@ -73,7 +72,8 @@ class ThreadPool {
     size_t thread_id =
         current_thread_count_.fetch_add(1, std::memory_order_relaxed);
     // assign task to dedicated thread.
-    dedicated_threads_[thread_id].reset(new std::thread(std::thread(func, params...)));
+    dedicated_threads_[thread_id].reset(
+        new std::thread(std::thread(func, params...)));
   }
 
  private:

src/include/network/connection_dispatcher_task.h

@@ -0,0 +1,69 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// connection_dispatcher_task.h
+//
+// Identification: src/include/network/connection_dispatcher_task.h
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+
+#include "notifiable_task.h"
+#include "network_state.h"
+#include "concurrency/epoch_manager_factory.h"
+#include "connection_handler_task.h"
+
+namespace peloton {
+namespace network {
+
+/**
+ * @brief A ConnectionDispatcherTask on the main server thread and dispatches
+ * incoming connections to handler threads.
+ *
+ * On construction, the dispatcher also spawns a number of handlers running on
+ * their own threads. The dispatcher is
+ * then responsible for maintain, and when shutting down, shutting down the
+ * spawned handlers also.
+ */
+class ConnectionDispatcherTask : public NotifiableTask {
+ public:
+  /**
+   * Creates a new ConnectionDispatcherTask, spawning the specified number of
+   * handlers, each running on their own threads.
+   *
+   * @param num_handlers The number of handler tasks to spawn.
+   * @param listen_fd The server socket fd to listen on.
+   */
+  ConnectionDispatcherTask(int num_handlers, int listen_fd);
+
+  /**
+   * @brief Dispatches the client connection at fd to a handler.
+   * Currently, the dispatch uses round-robin, and thread communication is
+   * achieved
+   * through channels. The dispatch writes a symbol to the fd that the handler
+   * is configured
+   * to receive updates on.
+   *
+   * @param fd the socket fd of the client connection being dispatched
+   * @param flags Unused. This is here to conform to libevent callback function
+   * signature.
+   */
+  void DispatchConnection(int fd, short flags);
+
+  /**
+   * Breaks the dispatcher and managed handlers from their event loops.
+   */
+  void ExitLoop() override;
+
+ private:
+  std::vector<std::shared_ptr<ConnectionHandlerTask>> handlers_;
+  // TODO: have a smarter dispatch scheduler, we currently use round-robin
+  std::atomic<int> next_handler_;
+};
+
+}  // namespace network
+}  // namespace peloton

src/include/network/connection_handle.h

@@ -0,0 +1,221 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// connection_handle.h
+//
+// Identification: src/include/network/connection_handle.h
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+#include <event2/buffer.h>
+#include <event2/bufferevent.h>
+#include <event2/event.h>
+#include <event2/listener.h>
+#include <unordered_map>
+
+#include <csignal>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <vector>
+
+#include <arpa/inet.h>
+#include <netinet/tcp.h>
+#include <sys/file.h>
+
+#include "common/exception.h"
+#include "common/logger.h"
+
+#include "marshal.h"
+#include "network/connection_handler_task.h"
+#include "network_state.h"
+#include "protocol_handler.h"
+
+#include <openssl/err.h>
+#include <openssl/ssl.h>
+
+namespace peloton {
+namespace network {
+
+// TODO(tianyu) This class is not refactored in full as rewriting the logic is
+// not cost-effective. However, readability
+// improvement and other changes may become desirable in the future. Other than
+// code clutter, responsibility assignment
+// is not well thought-out in this class. Abstracting out some type of socket
+// wrapper would be nice.
+/**
+ * @brief A ConnectionHandle encapsulates all information about a client
+ * connection for its entire duration.
+ * One should not use the constructor to construct a new ConnectionHandle
+ * instance every time as it is expensive
+ * to allocate buffers. Instead, use the ConnectionHandleFactory.
+ *
+ * @see ConnectionHandleFactory
+ */
+class ConnectionHandle {
+ public:
+  /**
+   * Update the existing event to listen to the passed flags
+   */
+  void UpdateEventFlags(short flags);
+
+  WriteState WritePackets();
+
+  std::string WriteBufferToString();
+
+  inline void HandleEvent(int, short) {
+    state_machine_.Accept(Transition::WAKEUP, *this);
+  }
+
+  // Exposed for testing
+  const std::unique_ptr<ProtocolHandler> &GetProtocolHandler() const {
+    return protocol_handler_;
+  }
+
+  // State Machine actions
+  /**
+   * refill_read_buffer - Used to repopulate read buffer with a fresh
+   * batch of data from the socket
+   */
+  Transition FillReadBuffer();
+  Transition Wait();
+  Transition Process();
+  Transition ProcessWrite();
+  Transition GetResult();
+  Transition CloseSocket();
+
+ private:
+  /**
+   * A state machine is defined to be a set of states, a set of symbols it
+   * supports, and a function mapping each
+   * state and symbol pair to the state it should transition to. i.e.
+   * transition_graph = state * symbol -> state
+   *
+   * In addition to the transition system, our network state machine also needs
+   * to perform actions. Actions are
+   * defined as functions (lambdas, or closures, in various other languages) and
+   * is promised to be invoked by the
+   * state machine after each transition if registered in the transition graph.
+   *
+   * So the transition graph overall has type transition_graph = state * symbol
+   * -> state * action
+   */
+  class StateMachine {
+   public:
+    using action = Transition (*)(ConnectionHandle &);
+    using transition_result = std::pair<ConnState, action>;
+    /**
+     * Runs the internal state machine, starting from the symbol given, until no
+     * more
+     * symbols are available.
+     *
+     * Each state of the state machine defines a map from a transition symbol to
+     * an action
+     * and the next state it should go to. The actions can either generate the
+     * next symbol,
+     * which means the state machine will continue to run on the generated
+     * symbol, or signal
+     * that there is no more symbols that can be generated, at which point the
+     * state machine
+     * will stop running and return, waiting for an external event (user
+     * interaction, or system event)
+     * to generate the next symbol.
+     *
+     * @param action starting symbol
+     * @param connection the network connection object to apply actions to
+     */
+    void Accept(Transition action, ConnectionHandle &connection);
+
+   private:
+    /**
+     * delta is the transition function that defines, for each state, its
+     * behavior and the
+     * next state it should go to.
+     */
+    static transition_result Delta_(ConnState state, Transition transition);
+    ConnState current_state_ = ConnState::READ;
+  };
+
+  friend class StateMachine;
+  friend class ConnectionHandleFactory;
+
+  ConnectionHandle(int sock_fd, ConnectionHandlerTask *handler,
+                   std::shared_ptr<Buffer> rbuf, std::shared_ptr<Buffer> wbuf,
+                   bool ssl_able);
+
+  ProcessResult ProcessInitial();
+
+  /**
+   * Extracts the header of a Postgres start up packet from the read socket
+   * buffer
+   */
+  static bool ReadStartupPacketHeader(Buffer &rbuf, InputPacket &rpkt);
+
+  /**
+   * Writes a packet's header (type, size) into the write buffer
+   */
+  WriteState BufferWriteBytesHeader(OutputPacket *pkt);
+
+  /**
+   * Writes a packet's content into the write buffer
+   */
+  WriteState BufferWriteBytesContent(OutputPacket *pkt);
+
+  /**
+   * Used to invoke a write into the Socket, returns false if the socket is not
+   * ready for write
+   */
+  WriteState FlushWriteBuffer();
+
+  /**
+   * Set the socket to non-blocking mode
+   */
+  inline void SetNonBlocking(evutil_socket_t fd) {
+    auto flags = fcntl(fd, F_GETFL);
+    flags |= O_NONBLOCK;
+    if (fcntl(fd, F_SETFL, flags) < 0) {
+      LOG_ERROR("Failed to set non-blocking socket");
+    }
+  }
+
+  /**
+   * Set TCP No Delay for lower latency
+   */
+  inline void SetTCPNoDelay(evutil_socket_t fd) {
+    int one = 1;
+    setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);
+  }
+
+  int sock_fd_;                            // socket file descriptor
+  struct event *network_event = nullptr;   // something to read from network
+  struct event *workpool_event = nullptr;  // worker thread done the job
+
+  SSL *conn_SSL_context = nullptr;  // SSL context for the connection
+
+  ConnectionHandlerTask *handler_;  // reference to the network thread
+  std::unique_ptr<ProtocolHandler>
+      protocol_handler_;  // Stores state for this socket
+  tcop::TrafficCop traffic_cop_;
+
+  std::shared_ptr<Buffer> rbuf_;    // Socket's read buffer
+  std::shared_ptr<Buffer> wbuf_;    // Socket's write buffer
+  unsigned int next_response_ = 0;  // The next response in the response buffer
+  Client client_;
+  StateMachine state_machine_;
+
+  // TODO(Tianyi) Can we encapsulate these flags?
+  bool ssl_handshake_ = false;
+  bool finish_startup_packet_ = false;
+  bool ssl_able_;
+
+  // TODO(Tianyi) hide this in protocol handler
+  InputPacket initial_packet_;
+
+  short curr_event_flag_;  // current libevent event flag
+};
+}  // namespace network
+}  // namespace peloton