Merge pull request #1199 from tli2/tianyuli-brain

scaffolding for brain

Author: linmagit

src/common/notifiable_task.cpp

@@ -0,0 +1,59 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// notifiable_task.cpp
+//
+// Identification: src/common/notifiable_task.cpp
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#include "common/notifiable_task.h"
+#include "common/logger.h"
+#include "common/event_util.h"
+#include <cstring>
+
+namespace peloton {
+
+NotifiableTask::NotifiableTask(int task_id) : task_id_(task_id) {
+  base_ = EventUtil::EventBaseNew();
+  // For exiting a loop
+  terminate_ = RegisterManualEvent([](int, short, void *arg) {
+    EventUtil::EventBaseLoopExit((struct event_base *) arg, nullptr);
+  }, base_);
+};
+
+NotifiableTask::~NotifiableTask() {
+  for (struct event *event : events_) {
+    EventUtil::EventDel(event);
+    event_free(event);
+  }
+  event_base_free(base_);
+}
+
+struct event *NotifiableTask::RegisterEvent(int fd,
+                                            short flags,
+                                            event_callback_fn callback,
+                                            void *arg,
+                                            const struct timeval *timeout) {
+  struct event *event = event_new(base_, fd, flags, callback, arg);
+  events_.insert(event);
+  EventUtil::EventAdd(event, timeout);
+  return event;
+}
+
+void NotifiableTask::UnregisterEvent(struct event *event) {
+  auto it = events_.find(event);
+  if (it == events_.end()) return;
+  if (event_del(event) == -1) {
+    LOG_ERROR("Failed to delete event");
+    return;
+  }
+  event_free(event);
+  events_.erase(event);
+}
+
+}  // namespace peloton
+

src/include/brain/brain.h

@@ -0,0 +1,80 @@
+//===----------------------------------------------------------------------===//
+//
+//                         Peloton
+//
+// brain.h
+//
+// Identification: src/include/brain/brain.h
+//
+// Copyright (c) 2015-2018, Carnegie Mellon University Database Group
+//
+//===----------------------------------------------------------------------===//
+
+#pragma once
+#include <unordered_map>
+#include <string>
+#include <cstdio>
+#include <utility>
+#include "capnp/ez-rpc.h"
+#include "peloton/capnp/peloton_service.capnp.h"
+#include "common/notifiable_task.h"
+
+namespace peloton {
+namespace brain {
+
+class BrainEnvironment {
+  // TODO(tianyu): provide interface for accessing various resources of the brain,
+  // such as network connection to a peloton engine.
+};
+
+class BrainJob {
+ public:
+  // TODO(tianyu): Extend this interface for richer interaction
+  virtual void RunJob(BrainEnvironment *) = 0;
+};
+
+class ExampleBrainJob: public BrainJob {
+ public:
+   void RunJob(BrainEnvironment *) override {
+     // TODO(tianyu): Replace with real address
+     capnp::EzRpcClient client("localhost:15445");
+     PelotonService::Client peloton_service = client.getMain<PelotonService>();
+     auto request = peloton_service.createIndexRequest();
+     request.getRequest().setIndexKeys(42);
+     auto response = request.send().wait(client.getWaitScope());
+   }
+};
+
+class Brain {
+ public:
+  // TODO(tianyu): Add necessary parameters to initialize the brain's resources
+  Brain() : scheduler_(0) {}
+
+  inline void RegisterJob(const struct timeval *period,
+                          std::string name,
+                          BrainJob &job) {
+    auto callback = [](int, short, void *pair) {
+      auto *job_env_pair = reinterpret_cast<std::pair<BrainJob *, BrainEnvironment *> *>(pair);
+      job_env_pair->first->RunJob(job_env_pair->second);
+    };
+    // TODO(tianyu) Deal with this memory
+    auto *pair = new std::pair<BrainJob *, BrainEnvironment *>(&job, &env_);
+    job_handles_[name] =
+        scheduler_.RegisterPeriodicEvent(period, callback, pair);
+  }
+
+  inline void Run() {
+    scheduler_.EventLoop();
+  }
+
+  inline void Terminate() {
+    scheduler_.ExitLoop();
+  }
+
+ private:
+  NotifiableTask scheduler_;
+  std::unordered_map<std::string, struct event *> job_handles_;
+  BrainEnvironment env_;
+};
+}
+}

src/include/capnp/peloton_service.capnp

@@ -12,7 +12,7 @@ struct CreateIndexRequest {
 }
 
 struct CreateIndexResponse {
-
+  message @0 :Text;
 }
 
 interface PelotonService {

src/include/network/error_util.h renamed to src/include/common/event_util.h

@@ -2,20 +2,16 @@
 //
 //                         Peloton
 //
-// error_util.h
+// event_util.h
 //
-// Identification: src/include/network/error_util.h
+// Identification: src/include/common/event_util.h
 //
 // Copyright (c) 2015-2018, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
 
 #pragma once
 
-#include <string>
-#include <unordered_map>
-#include <utility>
-
 #include <event2/buffer.h>
 #include <event2/bufferevent.h>
 #include <event2/event.h>
@@ -25,7 +21,6 @@
 #include "common/logger.h"
 
 namespace peloton {
-namespace network {
 /**
  * Static utility class with wrappers for libevent functions.
  *
@@ -86,5 +81,4 @@ class EventUtil {
                 "Error in event base dispatch");
   }
 };
-}  // namespace network
 }  // namespace peloton

src/include/network/notifiable_task.h renamed to src/include/common/notifiable_task.h

@@ -4,31 +4,19 @@
 //
 // notifiable_task.h
 //
-// Identification: src/include/network/notifiable_task.h
+// Identification: src/include/common/notifiable_task.h
 //
 // Copyright (c) 2015-2018, Carnegie Mellon University Database Group
 //
 //===----------------------------------------------------------------------===//
 
 #pragma once
 
-#include <csignal>
-#include <cstdio>
-#include <cstdlib>
-#include <cstring>
-#include <vector>
 #include <unordered_set>
-
-#include <sys/file.h>
 #include <event2/thread.h>
-
-#include "common/exception.h"
-#include "common/logger.h"
-#include "network_state.h"
-#include "error_util.h"
+#include "common/event_util.h"
 
 namespace peloton {
-namespace network {
 
 /**
  * Convenient MACRO to use a method as a libevent callback function. Example
@@ -61,29 +49,13 @@ class NotifiableTask {
    * Constructs a new NotifiableTask instance.
    * @param task_id a unique id assigned to this task
    */
-  explicit NotifiableTask(int task_id) : task_id_(task_id) {
-    base_ = EventUtil::EventBaseNew();
-    // TODO(tianyu) Determine whether we actually need this line. Tianyi says we
-    // need it, libevent documentation says no
-    // evthread_make_base_notifiable(base_);
-
-    // For exiting a loop
-    terminate_ = RegisterManualEvent([](int, short, void *arg) {
-      EventUtil::EventBaseLoopExit((struct event_base *)arg, nullptr);
-    }, base_);
-  };
+  explicit NotifiableTask(int task_id);
 
   /**
- * Destructs this NotifiableTask. All events currently registered to its base
- * are also deleted and freed.
- */
-  virtual ~NotifiableTask() {
-    for (struct event *event : events_) {
-      EventUtil::EventDel(event);
-      event_free(event);
-    }
-    event_base_free(base_);
-  }
+   * Destructs this NotifiableTask. All events currently registered to its base
+   * are also deleted and freed.
+   */
+  virtual ~NotifiableTask();
 
   /**
    * @return unique id assigned to this task
@@ -114,13 +86,7 @@ class NotifiableTask {
    */
   struct event *RegisterEvent(int fd, short flags, event_callback_fn callback,
                               void *arg,
-                              const struct timeval *timeout = nullptr) {
-    struct event *event = event_new(base_, fd, flags, callback, arg);
-    events_.insert(event);
-    EventUtil::EventAdd(event, timeout);
-    return event;
-  }
-
+                              const struct timeval *timeout = nullptr);
   /**
    * @brief Register a signal event. This is a wrapper around RegisterEvent()
    *
@@ -133,8 +99,9 @@ class NotifiableTask {
    * null which will wait forever
    * @return pointer to the allocated event.
    */
-  struct event *RegisterSignalEvent(int signal, event_callback_fn callback,
-                                    void *arg) {
+  inline struct event *RegisterSignalEvent(int signal,
+                                           event_callback_fn callback,
+                                           void *arg) {
     return RegisterEvent(signal, EV_SIGNAL | EV_PERSIST, callback, arg);
   }
 
@@ -150,8 +117,9 @@ class NotifiableTask {
    * @param arg an argument to be passed to the callback function
    * @return pointer to the allocated event.
    */
-  struct event *RegisterPeriodicEvent(const struct timeval *timeout,
-                                      event_callback_fn callback, void *arg) {
+  inline struct event *RegisterPeriodicEvent(const struct timeval *timeout,
+                                             event_callback_fn callback,
+                                             void *arg) {
     return RegisterEvent(-1, EV_TIMEOUT | EV_PERSIST, callback, arg, timeout);
   }
 
@@ -166,14 +134,14 @@ class NotifiableTask {
    * @param arg an argument to be passed to the callback function
    * @return pointer to the allocated event.
    */
-  struct event *RegisterManualEvent(event_callback_fn callback, void *arg) {
+  inline struct event *RegisterManualEvent(event_callback_fn callback,
+                                           void *arg) {
     return RegisterEvent(-1, EV_PERSIST, callback, arg);
   }
 
   // TODO(tianyu): The original network code seems to do this as an
   //  optimization. Specifically it avoids new memory allocation by reusing
   //  an existing event. I am leaving this out until we get numbers.
-
   //  void UpdateEvent(struct event *event, int fd, short flags,
   //                   event_callback_fn callback, void *arg,
   //                   const struct timeval *timeout = nullptr) {
@@ -198,25 +166,15 @@ class NotifiableTask {
    *
    * @param event the event to be freed
    */
-  void UnregisterEvent(struct event *event) {
-    auto it = events_.find(event);
-    if (it == events_.end()) return;
-    if (event_del(event) == -1) {
-      LOG_ERROR("Failed to delete event");
-      return;
-    }
-    event_free(event);
-    events_.erase(event);
-  }
+  void UnregisterEvent(struct event *event);
 
   /**
    * In a loop, make this notifiable task wait and respond to incoming events
    */
-  void EventLoop() {
+  inline void EventLoop() {
     EventUtil::EventBaseDispatch(base_);
     LOG_TRACE("stop");
   }
-
   /**
    * Exits the event loop
    */
@@ -225,7 +183,7 @@ class NotifiableTask {
   /**
    * Wrapper around ExitLoop() to conform to libevent callback signature
    */
-  void ExitLoop(int, short) { ExitLoop(); }
+  inline void ExitLoop(int, short) { ExitLoop(); }
 
  private:
   const int task_id_;
@@ -236,5 +194,4 @@ class NotifiableTask {
   std::unordered_set<struct event *> events_;
 };
 
-}  // namespace network
 }  // namespace peloton

src/include/network/connection_dispatcher_task.h

@@ -12,7 +12,7 @@
 
 #pragma once
 
-#include "notifiable_task.h"
+#include "common/notifiable_task.h"
 #include "network_state.h"
 #include "concurrency/epoch_manager_factory.h"
 #include "connection_handler_task.h"

src/include/network/connection_handler_task.h

@@ -22,7 +22,7 @@
 #include "common/exception.h"
 #include "common/logger.h"
 #include "common/container/lock_free_queue.h"
-#include "network/notifiable_task.h"
+#include "common/notifiable_task.h"
 
 namespace peloton {
 namespace network {

src/include/network/peloton_server.h

@@ -34,7 +34,7 @@
 #include "common/dedicated_thread_owner.h"
 #include "connection_dispatcher_task.h"
 #include "network_state.h"
-#include "notifiable_task.h"
+#include "common/notifiable_task.h"
 #include "protocol_handler.h"
 
 #include <openssl/crypto.h>

src/include/settings/settings.h

@@ -156,6 +156,11 @@ SETTING_bool(brain,
             false,
             true, true)
 
+SETTING_string(peloton_address,
+               "ip and port of the peloton rpc service, address:port",
+               "127.0.0.1:15445",
+               false, false)
+
 // Size of the brain task queue
 SETTING_int(brain_task_queue_size,
             "Brain Task Queue Size (default: 32)",