Merge pull request #8 from fixstars/feature/runtime-layer

Introduced runtime layer

.gitignore

@@ -2,4 +2,5 @@ GPATH
 GTAGS
 GRTAGS
 build*
+.vscode
 *~

CMakeLists.txt

@@ -34,7 +34,9 @@ endif()
 if(LIBDOCAGPU_FOUND)
     set(LNG_WITH_DOCA 1)
     add_compile_definitions(LNG_WITH_DOCA=1)
-elseif(LIBDPDK_FOUND)
+endif()
+
+if(LIBDPDK_FOUND)
     set(LNG_WITH_DPDK 1)
     add_compile_definitions(LNG_WITH_DPDK=1)
 endif()
@@ -59,10 +61,11 @@ set(LNG_CORE_SRC
     src/receiver-actor.cc
     src/event.cc
     src/log.cc
+    src/runtime.cc
     src/stream.cc
     src/system.cc
     )
-if(LNG_WITH_CUDA)
+if(LNG_WITH_CUDA AND LNG_WITH_DOCA)
     list(APPEND LNG_CORE_SRC
         src/protocol_tcp.cu
         src/protocol_udp.cu)
@@ -120,13 +123,15 @@ target_link_libraries(test_basic lng-core)
 if(LNG_WITH_DOCA OR LNG_WITH_DPDK)
     add_executable(test_dpdk_echo_udp test/dpdk_echo_udp.cc)
     target_link_libraries(test_dpdk_echo_udp lng-core ${LIBDPDK_STATIC_LDFLAGS})
-    add_executable(dpdk_frame_builder_tcp test/dpdk_frame_builder_tcp.cc)
-    target_link_libraries(dpdk_frame_builder_tcp lng-core ${LIBDPDK_STATIC_LDFLAGS})
+
+    add_executable(test_dpdk_build_frame_tcp test/dpdk_build_frame_tcp.cc)
+    target_link_libraries(test_dpdk_build_frame_tcp lng-core ${LIBDPDK_STATIC_LDFLAGS})
 endif()
 
 if(LNG_WITH_DOCA)
     add_executable(test_doca_echo_udp test/doca_echo_udp.cc)
     target_link_libraries(test_doca_echo_udp lng-core ${LIBDPDK_STATIC_LDFLAGS})
+
     add_executable(test_doca_build_frame_tcp test/doca_build_frame_tcp.cc)
     target_link_libraries(test_doca_build_frame_tcp lng-core ${LIBDPDK_STATIC_LDFLAGS})
 endif()

include/lng/actor.h

@@ -43,7 +43,7 @@ class Actor {
         int cpu_id;
 
         Impl(Actor* obj, const std::string& id, int cpu)
-            : th(entry_point, obj)
+            : th()
             , mutex()
             , cvar()
             , id(id)
@@ -69,6 +69,8 @@ class Actor {
     void wait_until(State to);
 
 protected:
+    virtual void setup() {}
+    virtual void teardown() {}
     virtual void main() = 0;
 
 private:

include/lng/receiver-actor.h

@@ -9,54 +9,70 @@ class Receiver : public Actor {
 public:
     Receiver(const std::string& id,
         int cpu_id,
-        DPDKStream* dpdk_st,
-        Stream<Payloads*>* valid,
-        Stream<Payloads*>* ready)
+        const std::shared_ptr<DPDKStream>& dpdk_st,
+        const std::shared_ptr<Queueable<Payload*>>& valid,
+        const std::shared_ptr<Queueable<Payload*>>& ready)
         : Actor(id, cpu_id)
         , nic_stream_(dpdk_st)
-        , vaild_payload_stream_(valid)
+        , valid_payload_stream_(valid)
         , ready_payload_stream_(ready)
+        , payload_(nullptr)
     {
     }
 
 protected:
+    virtual void setup() override;
     virtual void main() override;
 
 private:
-    DPDKStream* nic_stream_;
-    Stream<Payloads*>* vaild_payload_stream_;
-    Stream<Payloads*>* ready_payload_stream_;
+    std::shared_ptr<DPDKStream> nic_stream_;
+    std::shared_ptr<Queueable<Payload*>> valid_payload_stream_;
+    std::shared_ptr<Queueable<Payload*>> ready_payload_stream_;
+
+    Payload *payload_;
 };
 
 class FrameBuilder : public Actor {
 public:
     FrameBuilder(const std::string& id,
         int cpu_id,
-        Stream<Payloads*>* valid_payload,
-        Stream<Payloads*>* ready_payload,
-        Stream<Frame*>* valid_frame,
-        Stream<Frame*>* ready_frame)
+        const std::shared_ptr<Queueable<Payload*>>& valid_payload,
+        const std::shared_ptr<Queueable<Payload*>>& ready_payload,
+        const std::shared_ptr<Queueable<Frame*>>& valid_frame,
+        const std::shared_ptr<Queueable<Frame*>>& ready_frame)
         : Actor(id, cpu_id)
-        , vaild_payload_stream_(valid_payload)
+        , valid_payload_stream_(valid_payload)
         , ready_payload_stream_(ready_payload)
-        , vaild_frame_stream_(valid_frame)
+        , valid_frame_stream_(valid_frame)
         , ready_frame_stream_(ready_frame)
+        , payload_(nullptr)
+        , payload_segment_id_(0)
+        , payload_segment_read_offset_(0)
+        , frame_(nullptr)
         , frame_id_(0)
-        , write_head_(0)
-        , next_frame_(nullptr)
+        , frame_write_offset_(0)
     {
     }
 
 protected:
     virtual void main() override;
 
 private:
-    Stream<Payloads*>* vaild_payload_stream_;
-    Stream<Payloads*>* ready_payload_stream_;
-    Stream<Frame*>* vaild_frame_stream_;
-    Stream<Frame*>* ready_frame_stream_;
+    std::shared_ptr<Queueable<Payload*>> valid_payload_stream_;
+    std::shared_ptr<Queueable<Payload*>> ready_payload_stream_;
+    std::shared_ptr<Queueable<Frame*>> valid_frame_stream_;
+    std::shared_ptr<Queueable<Frame*>> ready_frame_stream_;
+    #if 0
+    Frame* next_frame_;
     size_t frame_id_;
     size_t write_head_;
-    Frame* next_frame_;
+#else
+    Payload* payload_;
+    size_t payload_segment_id_;
+    size_t payload_segment_read_offset_;
+    Frame* frame_;
+    size_t frame_id_;
+    size_t frame_write_offset_;
+    #endif
 };
 }

include/lng/runtime.h

@@ -0,0 +1,51 @@
+#ifndef LNG_RUNTIME_H
+#define LNG_RUNTIME_H
+
+#if defined(LNG_WITH_DPDK)
+struct rte_mempool;
+#endif
+
+namespace lng {
+
+class Runtime {
+public:
+    enum Type {
+        DPDK,
+        DOCA
+    };
+
+    virtual void start() = 0;
+    virtual void stop() = 0;
+};
+
+#if defined(LNG_WITH_DOCA)
+class DOCARuntime : public Runtime {
+public:
+    DOCARuntime();
+
+    virtual void start() { }
+
+    virtual void stop() { }
+};
+#endif
+
+#if defined(LNG_WITH_DPDK)
+
+class DPDKRuntime : public Runtime {
+public:
+    virtual void start();
+
+    virtual void stop();
+
+    rte_mempool* get_mempool() {
+        return mbuf_pool_;
+    }
+
+private:
+    rte_mempool* mbuf_pool_;
+};
+#endif
+
+} // lng
+
+#endif