Merge branch 'master' of https://github.com/poojanilangekar/peloton

Author: poojanilangekar

src/brain/query_logger.cpp

@@ -11,22 +11,40 @@
 //===----------------------------------------------------------------------===//
 
 #include "brain/query_logger.h"
+
 #include "catalog/query_history_catalog.h"
 #include "concurrency/transaction_context.h"
 #include "concurrency/transaction_manager_factory.h"
-#include "parser/pg_query.h"
 
 namespace peloton {
 namespace brain {
 
+QueryLogger::Fingerprint::Fingerprint(const std::string &query)
+    : query_(query),
+      fingerprint_(""),
+      fingerprint_result_(pg_query_fingerprint(query.c_str())) {
+  if (fingerprint_result_.hexdigest != nullptr) {
+    fingerprint_ = fingerprint_result_.hexdigest;
+  }
+}
+
+QueryLogger::Fingerprint::~Fingerprint() {
+  pg_query_free_fingerprint_result(fingerprint_result_);
+}
+
 void QueryLogger::LogQuery(std::string query_string, uint64_t timestamp) {
   auto &txn_manager = concurrency::TransactionManagerFactory::GetInstance();
-  auto txn = txn_manager.BeginTransaction();
-  std::string fingerprint = pg_query_fingerprint(query_string.c_str()).hexdigest;
+  auto *txn = txn_manager.BeginTransaction();
+
+  // Perform fingerprint
+  Fingerprint fingerprint{query_string};
 
-  catalog::QueryHistoryCatalog::GetInstance()->InsertQueryHistory(
-      query_string, fingerprint, timestamp, nullptr, txn);
+  // Log query + fingerprint
+  auto &query_history_catalog = catalog::QueryHistoryCatalog::GetInstance();
+  query_history_catalog.InsertQueryHistory(
+      query_string, fingerprint.GetFingerprint(), timestamp, nullptr, txn);
 
+  // We're done
   txn_manager.CommitTransaction(txn);
 }
 

src/catalog/query_history_catalog.cpp

@@ -6,25 +6,23 @@
 //
 // Identification: src/catalog/query_history_catalog.cpp
 //
-// Copyright (c) 2015-18, Carnegie Mellon University Database Group
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
 
 #include "catalog/query_history_catalog.h"
 
 #include "catalog/catalog.h"
-#include "executor/logical_tile.h"
-#include "parser/pg_query.h"
 #include "storage/data_table.h"
 #include "type/value_factory.h"
 
 namespace peloton {
 namespace catalog {
 
-QueryHistoryCatalog *QueryHistoryCatalog::GetInstance(
+QueryHistoryCatalog &QueryHistoryCatalog::GetInstance(
     concurrency::TransactionContext *txn) {
   static QueryHistoryCatalog query_history_catalog{txn};
-  return &query_history_catalog;
+  return query_history_catalog;
 }
 
 QueryHistoryCatalog::QueryHistoryCatalog(concurrency::TransactionContext *txn)
@@ -36,22 +34,23 @@ QueryHistoryCatalog::QueryHistoryCatalog(concurrency::TransactionContext *txn)
                       "timestamp      TIMESTAMP NOT NULL);",
                       txn) {}
 
-QueryHistoryCatalog::~QueryHistoryCatalog() {}
+QueryHistoryCatalog::~QueryHistoryCatalog() = default;
 
-bool QueryHistoryCatalog::InsertQueryHistory(const std::string &query_string, 
-                                  std::string &fingerprint, uint64_t timestamp,
-                                  type::AbstractPool *pool,
-                                  concurrency::TransactionContext *txn) {
+bool QueryHistoryCatalog::InsertQueryHistory(
+    const std::string &query_string, const std::string &fingerprint,
+    uint64_t timestamp, type::AbstractPool *pool,
+    concurrency::TransactionContext *txn) {
   std::unique_ptr<storage::Tuple> tuple(
       new storage::Tuple(catalog_table_->GetSchema(), true));
 
-  auto val0 = type::ValueFactory::GetVarcharValue(query_string, pool);
-  auto val1 = type::ValueFactory::GetVarcharValue(fingerprint, pool);
+  auto val0 = type::ValueFactory::GetVarcharValue(query_string);
+  auto val1 = type::ValueFactory::GetVarcharValue(fingerprint);
   auto val2 = type::ValueFactory::GetTimestampValue(timestamp);
 
-  tuple->SetValue(ColumnId::QUERY_STRING, val0, pool);
-  tuple->SetValue(ColumnId::FINGERPRINT, val1, pool);
-  tuple->SetValue(ColumnId::TIMESTAMP, val2, pool);
+  tuple->SetValue(ColumnId::QUERY_STRING, val0,
+                  pool != nullptr ? pool : &pool_);
+  tuple->SetValue(ColumnId::FINGERPRINT, val1, pool != nullptr ? pool : &pool_);
+  tuple->SetValue(ColumnId::TIMESTAMP, val2, pool != nullptr ? pool : &pool_);
 
   // Insert the tuple
   return InsertTuple(std::move(tuple), txn);

src/include/brain/query_logger.h

@@ -12,8 +12,11 @@
 
 #pragma once
 
+#include <cstdint>
 #include <string>
 
+#include "parser/pg_query.h"
+
 namespace peloton {
 namespace brain {
 
@@ -25,8 +28,28 @@ class QueryLogger {
  public:
   QueryLogger() = default;
 
-  /*
+  class Fingerprint {
+   public:
+    /// Constructor
+    explicit Fingerprint(const std::string &query);
+
+    /// Destructor
+    ~Fingerprint();
+
+    /// Get original string
+    const std::string &GetQueryString() { return query_; }
+    const std::string &GetFingerprint() { return fingerprint_; }
+
+   private:
+    // Accessors
+    std::string query_;
+    std::string fingerprint_;
+    PgQueryFingerprintResult fingerprint_result_;
+  };
+
+  /**
    * @brief This function logs the query into query_history_catalog
+   *
    * @param the sql string corresponding to the query
    * @param timestamp of the transaction that executed the query
    */

src/include/catalog/query_history_catalog.h

@@ -6,7 +6,7 @@
 //
 // Identification: src/include/catalog/query_history_catalog.h
 //
-// Copyright (c) 2015-18, Carnegie Mellon University Database Group
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
 
@@ -23,6 +23,7 @@
 #pragma once
 
 #include "catalog/abstract_catalog.h"
+#include "type/ephemeral_pool.h"
 
 #define QUERY_HISTORY_CATALOG_NAME "pg_query_history"
 
@@ -34,14 +35,14 @@ class QueryHistoryCatalog : public AbstractCatalog {
   ~QueryHistoryCatalog();
 
   // Global Singleton
-  static QueryHistoryCatalog *GetInstance(
+  static QueryHistoryCatalog &GetInstance(
       concurrency::TransactionContext *txn = nullptr);
 
   //===--------------------------------------------------------------------===//
   // write Related API
   //===--------------------------------------------------------------------===//
   bool InsertQueryHistory(const std::string &query_string,
-                          std::string &fingerprint, uint64_t timestamp,
+                          const std::string &fingerprint, uint64_t timestamp,
                           type::AbstractPool *pool,
                           concurrency::TransactionContext *txn);
 
@@ -54,6 +55,8 @@ class QueryHistoryCatalog : public AbstractCatalog {
  private:
   QueryHistoryCatalog(concurrency::TransactionContext *txn);
 
+  // Pool to use for variable length strings
+  type::EphemeralPool pool_;
 };
 
 }  // namespace catalog

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