Merge pull request #7 from Algebraic-Programming/feat/updateTaskR

feat: update TaskR

Author: lterrac

.github/workflows/master-test-workflow.yml

@@ -39,7 +39,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup
-        run: source /home/hicr/.bashrc && meson setup build -Dengines=mpi,cloudr -Db_coverage=true -DbuildTests=true -DbuildExamples=true -DcompileWarningsAsErrors=true
+        run: source /home/hicr/.bashrc && meson setup build -Dengines=mpi,cloudr -Db_coverage=true -DbuildTests=true -DbuildExamples=true -DcompileWarningsAsErrors=false
 
       - name: Compile
         run: source /home/hicr/.bashrc && meson compile -C build

.github/workflows/pr-development-workflow.yml

@@ -39,7 +39,7 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Setup
-        run: source /home/hicr/.bashrc && meson setup build -Db_coverage=true -Dengines=mpi,cloudr -DbuildTests=true -DbuildExamples=true -DcompileWarningsAsErrors=true
+        run: source /home/hicr/.bashrc && meson setup build -Db_coverage=true -Dengines=mpi,cloudr -DbuildTests=true -DbuildExamples=true -DcompileWarningsAsErrors=false
 
       - name: Compile
         run: source /home/hicr/.bashrc && meson compile -C build

extern/TaskR

@@ -188,14 +188,14 @@ __INLINE__ void createChannelBuffers(HiCR::MemoryManager
 }
 
 void createChannels(HiCR::MemoryManager                                                                         &memoryManager,
-                   HiCR::CommunicationManager                                                                  &communicationManager,
-                   std::shared_ptr<HiCR::MemorySpace>                                                          &memorySpace,
-                   std::map<HiCR::GlobalMemorySlot::tag_t, nlohmann::json>                                     &consumersData,
-                   std::map<HiCR::GlobalMemorySlot::tag_t, nlohmann::json>                                     &producersData,
-                   HiCR::GlobalMemorySlot::tag_t                                                                channelsIds,
-                   std::map<HiCR::GlobalMemorySlot::tag_t, std::string>                                        &channelTagNameMap,
-                   std::unordered_map<std::string, std::unique_ptr<hLLM::channel::channelConsumerInterface_t>> &consumers,
-                   std::unordered_map<std::string, std::unique_ptr<hLLM::channel::channelProducerInterface_t>> &producers)
+                    HiCR::CommunicationManager                                                                  &communicationManager,
+                    std::shared_ptr<HiCR::MemorySpace>                                                          &memorySpace,
+                    std::map<HiCR::GlobalMemorySlot::tag_t, nlohmann::json>                                     &consumersData,
+                    std::map<HiCR::GlobalMemorySlot::tag_t, nlohmann::json>                                     &producersData,
+                    HiCR::GlobalMemorySlot::tag_t                                                                channelsIds,
+                    std::map<HiCR::GlobalMemorySlot::tag_t, std::string>                                        &channelTagNameMap,
+                    std::unordered_map<std::string, std::unique_ptr<hLLM::channel::channelConsumerInterface_t>> &consumers,
+                    std::unordered_map<std::string, std::unique_ptr<hLLM::channel::channelProducerInterface_t>> &producers)
 {
   std::map<HiCR::GlobalMemorySlot::tag_t, std::vector<HiCR::CommunicationManager::globalKeyMemorySlotPair_t>> memorySlotsToExchange;
   std::map<HiCR::GlobalMemorySlot::tag_t, std::shared_ptr<HiCR::LocalMemorySlot>>                             coordinationBufferSizesSlots;
@@ -319,22 +319,22 @@ void Engine::createDependencies()
   }
 
   createChannels(*_unbufferedMemoryManager,
-                *_unbufferedCommunicationManager,
-                _unbufferedMemorySpace,
-                unbufferedConsumers,
-                unbufferedProducers,
-                unbufferedChannelId,
-                unbufferedChannelTagNameMap,
-                _consumers,
-                _producers);
+                 *_unbufferedCommunicationManager,
+                 _unbufferedMemorySpace,
+                 unbufferedConsumers,
+                 unbufferedProducers,
+                 unbufferedChannelId,
+                 unbufferedChannelTagNameMap,
+                 _consumers,
+                 _producers);
   createChannels(*_bufferedMemoryManager,
-                *_bufferedCommunicationManager,
-                _bufferedMemorySpace,
-                bufferedConsumers,
-                bufferedProducers,
-                bufferedChannelId,
-                bufferedChannelTagNameMap,
-                _consumers,
-                _producers);
+                 *_bufferedCommunicationManager,
+                 _bufferedMemorySpace,
+                 bufferedConsumers,
+                 bufferedProducers,
+                 bufferedChannelId,
+                 bufferedChannelTagNameMap,
+                 _consumers,
+                 _producers);
 }
 } // namespace hLLM

include/hllm/channelCreationImpl.hpp

@@ -498,10 +498,11 @@ class Engine final
     _taskr->setTaskCallbackHandler(HiCR::tasking::Task::callback_t::onTaskSuspend, [&](taskr::Task *task) { _taskr->resumeTask(task); });
 
     // Setting service to listen for incoming administrative messages
-    std::function<void()> RPCListeningService = [this]() {
+    std::function<void()> RPCListeningServiceFc = [this]() {
       if (_rpcEngine->hasPendingRPCs()) _rpcEngine->listen();
     };
-    _taskr->addService(&RPCListeningService);
+    _rpcListeningService = std::make_unique<taskr::Service>(RPCListeningServiceFc, 50);
+    _taskr->addService(_rpcListeningService.get());
 
     // Running TaskR
     _taskr->run();
@@ -626,6 +627,8 @@ class Engine final
   HiCR::Instance::instanceId_t _deployerInstanceId;
 
   std::map<partitionId_t, deployr::Runner::runnerId_t> _partitionRunnerIdMap;
+
+  std::unique_ptr<taskr::Service> _rpcListeningService;
 }; // class Engine
 
 } // namespace hLLM

include/hllm/engine.hpp

@@ -51,13 +51,13 @@ __INLINE__ std::unordered_map<std::string, nlohmann::json> Engine::parseDependen
       {
         if (dependenciesMap.contains(output) == false)
         {
-          HICR_THROW_RUNTIME("Producer %s defined in partition %s has no consumers defined", output.get<std::string>(), partition["Name"]);
+          HICR_THROW_RUNTIME("Producer %s defined in partition %s has no consumers defined", output.get<std::string>().c_str(), partition["Name"].get<std::string>().c_str());
         }
 
         if (dependenciesMap[output].contains("Producer"))
         {
           HICR_THROW_RUNTIME("Set two producers for buffered dependency %s is not allowed. Producer: %s, tried to add %s",
-                             output,
+                             output.get<std::string>().c_str(),
                              hicr::json::getNumber<partitionId_t>(dependenciesMap[output], "Producer"),
                              hicr::json::getString(partition, "Name").c_str());
         }