futurewei-cloud · lfu-ps · Jan 27, 2022 · Oct 11, 2021 · Oct 18, 2021 · Oct 19, 2021
diff --git a/alcor b/alcor
diff --git a/build/Dockerfile b/build/Dockerfile
@@ -194,3 +194,14 @@ RUN echo "6--- installing pulsar dependacies ---" && \
     apt install -y ./apache-pulsar-client*.deb && \
     rm -rf /var/local/git/pulsar
 
+RUN echo "7--- installing marl ---" && \
+    mkdir -p /var/local/git/marl && \
+    cd /var/local/git/marl && \
+    git clone https://github.com/google/marl.git && \
+    cd /var/local/git/marl/marl && \
+    git submodule update --init && \
+    mkdir /var/local/git/marl/marl/build && \
+    cd /var/local/git/marl/marl/build && \
+    cmake .. -DMARL_BUILD_EXAMPLES=1 -DMARL_BUILD_TESTS=1 && \
+    make && \
+    cd ~
diff --git a/build/aca-machine-init.sh b/build/aca-machine-init.sh
@@ -199,8 +199,20 @@ echo "6--- installing openvswitch dependancies ---" && \
     test -f /usr/bin/ovs-vsctl && rm -rf /usr/local/sbin/ov* /usr/local/bin/ov* /usr/local/bin/vtep* && \
     cd ~
 
+echo "7--- installing marl ---" && \
+    mkdir -p /var/local/git/marl && \
+    cd /var/local/git/marl && \
+    git clone https://github.com/google/marl.git && \
+    cd /var/local/git/marl/marl && \
+	  git submodule update --init && \
+	  mkdir /var/local/git/marl/marl/build && \
+	  cd /var/local/git/marl/marl/build && \
+	  cmake .. -DMARL_BUILD_EXAMPLES=1 -DMARL_BUILD_TESTS=1 && \
+    make && \
+    cd ~
+
 PULSAR_RELEASE_TAG='pulsar-2.8.1'
-echo "7--- installing pulsar dependacies ---" && \
+echo "8--- installing pulsar dependacies ---" && \
     mkdir -p /var/local/git/pulsar && \
     wget https://archive.apache.org/dist/pulsar/${PULSAR_RELEASE_TAG}/DEB/apache-pulsar-client.deb -O /var/local/git/pulsar/apache-pulsar-client.deb && \
     wget https://archive.apache.org/dist/pulsar/${PULSAR_RELEASE_TAG}/DEB/apache-pulsar-client-dev.deb -O /var/local/git/pulsar/apache-pulsar-client-dev.deb && \
@@ -209,20 +221,23 @@ echo "7--- installing pulsar dependacies ---" && \
     rm -rf /var/local/git/pulsar 
     cd ~
 
-echo "8--- building alcor-control-agent"
+echo "9--- building alcor-control-agent"
 cd $BUILD/.. && cmake . && \
 # after cmake ., modify the generated link.txt s so that the "-lssl" and "-lcrypto" appears after the openvswitch, so that it can compile
 sed -i 's/\(-ldl -lrt -lm -lpthread\)/-lssl -lcrypto \1/' src/CMakeFiles/AlcorControlAgent.dir/link.txt && \
 sed -i 's/\(-ldl -lrt -lm -lpthread\)/-lssl -lcrypto \1/' test/CMakeFiles/aca_tests.dir/link.txt && \
 sed -i 's/\(-ldl -lrt -lm -lpthread\)/-lssl -lcrypto \1/' test/CMakeFiles/gs_tests.dir/link.txt && \
 make
 if [ -n "$1" -a "$1" = "delete-bridges" ]; then
-  echo "9--- deleting br-tun and br-int if requested"
+  echo "10--- deleting br-tun and br-int if requested"
   PATH=$PATH:/usr/local/share/openvswitch/scripts \
       LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/usr/local/lib
   ovs-ctl --system-id=random --delete-bridges restart
 fi
 
-echo "10--- running alcor-control-agent"
-# sends output to null device, but stderr to console 
-nohup $BUILD/bin/AlcorControlAgent -d > /dev/null 2>&1 &
+echo "11--- running alcor-control-agent"
+# kill current AlcorControlAgent process if any
+pkill -f AlcorControlAgent
+# launch newly built AlcorControlAgent with default debug mode as 'false'
+# sends output to null device, but stderr to console
+nohup $BUILD/bin/AlcorControlAgent > /dev/null 2>&1 &
diff --git a/include/aca_on_demand_engine.h b/include/aca_on_demand_engine.h
@@ -20,9 +20,8 @@
 #define STDOUT_FILENO 1 /* Standard output.  */
 
 #include "common.pb.h"
-#include <openvswitch/ofp-errors.h>
-//#include <openvswitch/ofp-packet.h>
-#include <openvswitch/ofp-util.h>
+#include "libfluid-msg/of10msg.hh"
+#include "libfluid-msg/of13msg.hh"
 #include <string>
 #include <thread>
 #include <netinet/ether.h>
@@ -31,11 +30,14 @@
 #include <unordered_map>
 #include "aca_log.h"
 #include "goalstateprovisioner.grpc.pb.h"
-#include "ctpl/ctpl_stl.h"
+
+#include "marl/defer.h"
+#include "marl/event.h"
+#include "marl/scheduler.h"
+#include "marl/waitgroup.h"
 
 using namespace alcor::schema;
 using namespace std;
-using namespace ctpl;
 
 extern int thread_pools_size;
 
@@ -74,8 +76,6 @@ class ACA_On_Demand_Engine {
   its initial value should be the time  when clean_remaining_payload() was first called*/
   std::chrono::_V2::steady_clock::time_point last_time_cleaned_remaining_payload;
 
-  ctpl::thread_pool thread_pool_;
-
   static ACA_On_Demand_Engine &get_instance();
 
   /*
@@ -175,22 +175,19 @@ class ACA_On_Demand_Engine {
     int cores = std::thread::hardware_concurrency();
     ACA_LOG_DEBUG("This host has %ld cores, setting the size of the thread pools to be %ld\n",
                   cores, thread_pools_size);
-    on_demand_reply_processing_thread = new std::thread(
-            std::bind(&ACA_On_Demand_Engine::process_async_grpc_replies, this));
-
-    on_demand_reply_processing_thread->detach();
-    on_demand_payload_cleaning_thread = new std::thread(
-            std::bind(&ACA_On_Demand_Engine::clean_remaining_payload, this));
-    on_demand_payload_cleaning_thread->detach();
-    thread_pool_.resize(thread_pools_size);
+    marl::schedule([=]{
+      process_async_grpc_replies();
+    });
+    marl::schedule([=]{
+      clean_remaining_payload();
+    });
   };
   ~ACA_On_Demand_Engine()
   {
     _cq.Shutdown();
     request_uuid_on_demand_payload_map.clear();
     delete on_demand_reply_processing_thread;
     delete on_demand_payload_cleaning_thread;
-    thread_pool_.stop();
   };
 };
 } // namespace aca_on_demand_engine

diff --git a/include/libfluid-base/base/EventLoop.hh b/include/libfluid-base/base/EventLoop.hh
@@ -17,6 +17,9 @@
 #ifndef __EVENTLOOP_HH__
 #define __EVENTLOOP_HH__
 
+#include "marl/defer.h"
+#include "marl/scheduler.h"
+
 namespace fluid_base {
 
 class BaseOFServer;
@@ -41,7 +44,7 @@ public:
 
     @param id event loop id
     */
-    EventLoop(int id);
+    EventLoop(int id, marl::Scheduler* scheduler);
     ~EventLoop();
 
     /**
@@ -80,6 +83,7 @@ private:
     class LibEventEventLoop;
     friend class LibEventEventLoop;
     LibEventEventLoop* m_implementation;
+    marl::Scheduler* m_scheduler;
 };
 
 }

diff --git a/include/of_controller.h b/include/of_controller.h
@@ -25,6 +25,11 @@
 #include "libfluid-msg/of10msg.hh"
 #include "libfluid-msg/of13msg.hh"
 
+#include "marl/defer.h"
+#include "marl/event.h"
+#include "marl/scheduler.h"
+#include "marl/waitgroup.h"
+
 #include <arpa/inet.h>
 #include <atomic>
 #include <vector>
@@ -40,6 +45,8 @@
 #include <set>
 #include <stdlib.h>
 #include <unordered_map>
+#include <chrono>
+
 
 using namespace fluid_base;
 using namespace fluid_msg;
@@ -56,8 +63,10 @@ class OFController : public OFServer {
             switch_dpid_map(switch_dpid_map),
             port_id_map(port_id_map),
             OFServer(address, port, nthreads, secure,
-                     OFServerSettings().supported_version(4) // OF version 0x04 is OF 1.3
-                         .echo_interval(30)) { }
+                     OFServerSettings()
+                         .supported_version(4) // OF version 1 is OF 1.0 and version 4 is 1.3
+                         .echo_interval(30)) {
+                          }
 
     ~OFController() = default;
 
@@ -86,6 +95,7 @@ class OFController : public OFServer {
     void packet_out(const char* br, const char* opt);
 
 private:
+
     // tracking xid (ovs transaction id)
     std::atomic<uint32_t> xid;
 

diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
@@ -66,12 +66,14 @@ link_libraries(/usr/lib/x86_64-linux-gnu/libuuid.so)
 link_libraries(/usr/lib/x86_64-linux-gnu/libevent_pthreads.so)
 link_libraries(/usr/lib/x86_64-linux-gnu/libpthread.so)
 link_libraries(/usr/local/lib/libopenvswitch.a) #this was installed by aca-machine-init.sh
+link_libraries(/var/local/git/marl/marl/build/libmarl.a)   #this was built by aca-machine-init.sh
 include_directories(${RDKAFKA_INCLUDE_DIR} ${CPPKAFKA_INCLUDE_DIR} ${PULSAR_INCLUDE_DIR} ${LIBEVENT_INCLUDE_DIR})
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/../include)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/proto3)
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/grpc)
 include_directories(/usr/local/include/openvswitch)
 include_directories(/usr/local/include/openflow)
+include_directories(/var/local/git/marl/marl/include)
 
 # Find Protobuf installation
 # Looks for protobuf-config.cmake file installed by Protobuf's cmake installation.

diff --git a/src/aca_main.cpp b/src/aca_main.cpp
@@ -35,6 +35,11 @@
 #include <grpcpp/grpcpp.h>
 #include <cmath>
 
+#include "marl/defer.h"
+#include "marl/event.h"
+#include "marl/scheduler.h"
+#include "marl/waitgroup.h"
+
 using aca_message_pulsar::ACA_Message_Pulsar_Consumer;
 using aca_ovs_control::ACA_OVS_Control;
 using std::string;
@@ -240,6 +245,7 @@ int main(int argc, char *argv[])
     }
   }
 
+
   // fill in the information if not provided in command line args
   if (g_broker_list == EMPTY_STRING) {
     g_broker_list = BROKER_LIST;
@@ -261,15 +267,25 @@ int main(int argc, char *argv[])
   }
 
   g_grpc_server = new GoalStateProvisionerAsyncServer();
-  g_grpc_server_thread = new std::thread(std::bind(
-          &GoalStateProvisionerAsyncServer::RunServer, g_grpc_server, thread_pools_size));
-  g_grpc_server_thread->detach();
-
+
   // Create a separate thread to run the grpc client.
   g_grpc_client = new GoalStateProvisionerClientImpl();
-  g_grpc_client_thread = new std::thread(
-          std::bind(&GoalStateProvisionerClientImpl::RunClient, g_grpc_client));
-  g_grpc_client_thread->detach();
+
+  // Create a marl scheduler using all the logical processors available to the process.
+  // Bind this scheduler to the main thread so we can call marl::schedule()
+  marl::Scheduler::Config cfg_bind_hw_cores;
+  cfg_bind_hw_cores.setWorkerThreadCount(thread_pools_size * 2);
+  marl::Scheduler task_scheduler(cfg_bind_hw_cores);
+  task_scheduler.bind();
+  defer(task_scheduler.unbind());
+
+  marl::schedule([=]{
+    g_grpc_server->RunServer(thread_pools_size);
+  });
+
+  marl::schedule([=]{
+    g_grpc_client->RunClient();
+  });
 
   aca_ovs_l2_programmer::ACA_OVS_L2_Programmer::get_instance().get_local_host_ips();
 

diff --git a/src/comm/aca_grpc.cpp b/src/comm/aca_grpc.cpp
@@ -31,6 +31,11 @@
 #include "aca_log.h"
 #include "aca_grpc.h"
 
+#include "marl/defer.h"
+#include "marl/event.h"
+#include "marl/scheduler.h"
+#include "marl/waitgroup.h"
+
 extern string g_grpc_server_port;
 extern string g_ncm_address;
 extern string g_ncm_port;
@@ -43,7 +48,6 @@ Status GoalStateProvisionerAsyncServer::ShutDownServer()
   ACA_LOG_INFO("%s", "Shutdown server");
   server_->Shutdown();
   cq_->Shutdown();
-  thread_pool_.stop();
   keepReadingFromCq_ = false;
   return Status::OK;
 }
@@ -275,15 +279,7 @@ void GoalStateProvisionerAsyncServer::AsyncWorkder()
 void GoalStateProvisionerAsyncServer::RunServer(int thread_pool_size)
 {
   ACA_LOG_INFO("Start of RunServer, pool size %ld\n", thread_pool_size);
-
-  thread_pool_.resize(thread_pool_size);
-  ACA_LOG_DEBUG("Async GRPC SERVER: Resized thread pool to %ld threads, start waiting for the pool to have enough threads\n",
-                thread_pool_size);
-  /* wait for thread pool to initialize*/
-  while (thread_pool_.n_idle() != thread_pool_.size()) {
-    ACA_LOG_DEBUG("%s\n", "Still waiting...sleep 1 ms");
-    std::this_thread::sleep_for(std::chrono::milliseconds(1));
-  };
+
   ACA_LOG_DEBUG("Async GRPC SERVER: finised resizing thread pool to %ld threads\n",
                 thread_pool_size);
   //  Create the server
@@ -331,6 +327,8 @@ void GoalStateProvisionerAsyncServer::RunServer(int thread_pool_size)
 
   for (int i = 0; i < thread_pool_size; i++) {
     ACA_LOG_DEBUG("Pushing the %ldth async worker into the pool", i);
-    thread_pool_.push(std::bind(&GoalStateProvisionerAsyncServer::AsyncWorkder, this));
+    marl::schedule([=]{
+        AsyncWorkder();
+    });
   }
 }
diff --git a/src/dp_abstraction/aca_dataplane_ovs.cpp b/src/dp_abstraction/aca_dataplane_ovs.cpp
@@ -329,9 +329,9 @@ int ACA_Dataplane_OVS::update_port_state_workitem(const PortState current_PortSt
           culminative_network_configuration_time, operation_total_time);
 
   if (overall_rc == EXIT_SUCCESS) {
-    ACA_LOG_INFO("%s", "Successfully configured the port state.\n");
+    ACA_LOG_DEBUG("%s", "Successfully configured the port state.\n");
   } else if (overall_rc == EINPROGRESS) {
-    ACA_LOG_INFO("Port state returned pending: rc=%d\n", overall_rc);
+    ACA_LOG_DEBUG("Port state returned pending: rc=%d\n", overall_rc);
   } else {
     ACA_LOG_ERROR("Unable to configure the port state: rc=%d\n", overall_rc);
   }
@@ -541,9 +541,9 @@ int ACA_Dataplane_OVS::update_port_state_workitem(const PortState current_PortSt
           culminative_network_configuration_time, operation_total_time);
 
   if (overall_rc == EXIT_SUCCESS) {
-    ACA_LOG_INFO("%s", "Successfully configured the port state.\n");
+    ACA_LOG_DEBUG("%s", "Successfully configured the port state.\n");
   } else if (overall_rc == EINPROGRESS) {
-    ACA_LOG_INFO("Port state returned pending: rc=%d\n", overall_rc);
+    ACA_LOG_DEBUG("Port state returned pending: rc=%d\n", overall_rc);
   } else {
     ACA_LOG_ERROR("Unable to configure the port state: rc=%d\n", overall_rc);
   }
@@ -730,9 +730,7 @@ int ACA_Dataplane_OVS::update_neighbor_state_workitem(NeighborState current_Neig
           overall_rc, culminative_dataplane_programming_time,
           culminative_network_configuration_time, operation_total_time);
 
-  if (overall_rc == EXIT_SUCCESS) {
-    ACA_LOG_INFO("%s", "Successfully configured the neighbor state.\n");
-  } else {
+  if (overall_rc != EXIT_SUCCESS) {
     ACA_LOG_ERROR("Unable to configure the neighbor state: rc=%d\n", overall_rc);
   }
 
@@ -907,9 +905,7 @@ int ACA_Dataplane_OVS::update_neighbor_state_workitem(NeighborState current_Neig
     overall_rc = -EFAULT;
   }
 
-  if (overall_rc == EXIT_SUCCESS) {
-    ACA_LOG_INFO("%s", "Successfully configured the neighbor state.\n");
-  } else {
+  if (overall_rc != EXIT_SUCCESS) {
     ACA_LOG_ERROR("Unable to configure the neighbor state: rc=%d\n", overall_rc);
   }
 
@@ -965,7 +961,7 @@ int ACA_Dataplane_OVS::update_router_state_workitem(RouterState current_RouterSt
           culminative_network_configuration_time, operation_total_time);
 
   if (overall_rc == EXIT_SUCCESS) {
-    ACA_LOG_INFO("%s", "Successfully configured the router state.\n");
+    ACA_LOG_DEBUG("%s", "Successfully configured the router state.\n");
   } else {
     ACA_LOG_ERROR("Unable to configure the router state: rc=%d\n", overall_rc);
   }
@@ -1022,7 +1018,7 @@ int ACA_Dataplane_OVS::update_router_state_workitem(RouterState current_RouterSt
           culminative_network_configuration_time, operation_total_time);
 
   if (overall_rc == EXIT_SUCCESS) {
-    ACA_LOG_INFO("%s", "Successfully configured the router state.\n");
+    ACA_LOG_DEBUG("%s", "Successfully configured the router state.\n");
   } else {
     ACA_LOG_ERROR("Unable to configure the router state: rc=%d\n", overall_rc);
   }