Feat: channel different comm managers

.github/workflows/master.yml

@@ -75,35 +75,37 @@ jobs:
       os: ubuntu-24.04-arm
       arch: arm64
 
-  build-acl-docker-amd64:
-    needs: [ push-buildenv-manifest ]
-    if: |
-      always() &&
-      (contains(needs.push-buildenv-manifest.result, 'success') || contains(needs.push-buildenv-manifest.result, 'skipped'))
-    uses: Algebraic-Programming/HiCR/.github/workflows/master-build-acl-workflow.yml@master
-    with:
-      os: ubuntu-24.04
-      arch: amd64
+  # disabled because of broken docker image creation
+
+  # build-acl-docker-amd64:
+  #   needs: [ push-buildenv-manifest ]
+  #   if: |
+  #     always() &&
+  #     (contains(needs.push-buildenv-manifest.result, 'success') || contains(needs.push-buildenv-manifest.result, 'skipped'))
+  #   uses: Algebraic-Programming/HiCR/.github/workflows/master-build-acl-workflow.yml@master
+  #   with:
+  #     os: ubuntu-24.04
+  #     arch: amd64
 
-  push-buildenv-acl-manifest:
-    runs-on: ubuntu-latest
-    needs: [ build-acl-docker-amd64, build-acl-docker-arm64 ]
-    if: |
-      always() &&
-      contains(needs.build-acl-docker-amd64.result, 'success') &&
-      contains(needs.build-acl-docker-arm64.result, 'success') 
-    steps:
-      - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@v3
+  # push-buildenv-acl-manifest:
+  #   runs-on: ubuntu-latest
+  #   needs: [ build-acl-docker-amd64, build-acl-docker-arm64 ]
+  #   if: |
+  #     always() &&
+  #     contains(needs.build-acl-docker-amd64.result, 'success') &&
+  #     contains(needs.build-acl-docker-arm64.result, 'success') 
+  #   steps:
+  #     - name: Set up Docker Buildx
+  #       uses: docker/setup-buildx-action@v3
 
-      - name: Log in to the Container registry
-        uses: docker/login-action@v3
-        with:
-          registry: ${{ env.REGISTRY }}
-          username: ${{ github.repository_owner }}
-          password: ${{ secrets.GITHUB_TOKEN }}
+  #     - name: Log in to the Container registry
+  #       uses: docker/login-action@v3
+  #       with:
+  #         registry: ${{ env.REGISTRY }}
+  #         username: ${{ github.repository_owner }}
+  #         password: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: Create and push manifest images
-        run:
-          docker buildx imagetools create --tag ${{ env.DOCKERIMAGE }}-acl:latest ${{ env.DOCKERIMAGE }}-acl:amd64-latest ${{ env.DOCKERIMAGE }}-acl:arm64-latest
+  #     - name: Create and push manifest images
+  #       run:
+  #         docker buildx imagetools create --tag ${{ env.DOCKERIMAGE }}-acl:latest ${{ env.DOCKERIMAGE }}-acl:amd64-latest ${{ env.DOCKERIMAGE }}-acl:arm64-latest
 

.github/workflows/style.yml

@@ -20,10 +20,52 @@ jobs:
       run: sudo apt update
     - name: Installing clang-tormat
       run: |
-            python3 -m pip install 'clang-format==18.1.0'
+        python3 -m pip install --upgrade pip
+        python3 -m pip install 'clang-format==18.1.0'
+        python3 -m pip install sphinx
+        python3 -m pip install sphinx_rtd_theme
+        python3 -m pip install sphinxcontrib-bibtex
+        python3 -m pip install doxysphinx
+        python3 -m pip install sphinxcontrib.needs
+        python3 -m pip install sphinxcontrib.plantuml
+        python3 -m pip install autoapi
+        python3 -m pip install sphinx-autoapi
+        python3 -m pip install myst_parser
+        python3 -m pip install sphinx_copybutton
+        python3 -m pip install sphinxcontrib.doxylink
+        python3 -m pip install sphinx_design
+        python3 -m pip install Sphinx-Substitution-Extensions
+        python3 -m pip install sphinx_toolbox
+        python3 -m pip install sphinx-theme
+        python3 -m pip install sphinx-book-theme
+        sudo apt update --fix-missing
+        sudo apt install -y build-essential \
+        doxygen \
+        graphviz \
+        fonts-freefont-ttf \
+        texlive \
+        texlive-latex-extra \
+        texlive-fonts-extra \
+        libffi-dev \
+        ghostscript \
+        texlive-extra-utils \
+        texlive-font-utils
+
+        python3 -m pip install 'sphinx==7.2.6'
+        sphinx-build --version
     - name: Checking style
       run: |
         echo "Checking HiCR source and test formatting..."
         .build-tools/style/check-style.sh check include
         .build-tools/style/check-style.sh check tests
         .build-tools/style/check-style.sh check examples
+    - name: Build documentation
+      run: |
+          echo "Building code documentation..."
+          make -j1 -C docs
+          mkdir public
+          cp -r docs/build/html/* public
+    - uses: actions/upload-artifact@v4
+      with:
+        name: docs
+        path: docs/build/html

examples/channels/fixedSize/mpsc/locking/include/consumer.hpp

@@ -22,41 +22,45 @@
 #include <hicr/frontends/channel/fixedSize/mpsc/locking/consumer.hpp>
 #include "common.hpp"
 
-void consumerFc(HiCR::MemoryManager               &memoryManager,
-                HiCR::CommunicationManager        &communicationManager,
-                std::shared_ptr<HiCR::MemorySpace> bufferMemorySpace,
+void consumerFc(HiCR::MemoryManager               &coordinationMemoryManager,
+                HiCR::MemoryManager               &payloadMemoryManager,
+                HiCR::CommunicationManager        &coordinationCommunicationManager,
+                HiCR::CommunicationManager        &payloadCommunicationManager,
+                std::shared_ptr<HiCR::MemorySpace> coordinationMemorySpace,
+                std::shared_ptr<HiCR::MemorySpace> payloadMemorySpace,
                 const size_t                       channelCapacity,
                 const size_t                       producerCount)
 {
   // Getting required buffer sizes
   auto tokenBufferSize = HiCR::channel::fixedSize::Base::getTokenBufferSize(sizeof(ELEMENT_TYPE), channelCapacity);
 
   // Registering token buffer as a local memory slot
-  auto tokenBufferSlot = memoryManager.allocateLocalMemorySlot(bufferMemorySpace, tokenBufferSize);
+  auto tokenBufferSlot = payloadMemoryManager.allocateLocalMemorySlot(payloadMemorySpace, tokenBufferSize);
 
   // Getting required buffer size
   auto coordinationBufferSize = HiCR::channel::fixedSize::Base::getCoordinationBufferSize();
 
   // Registering token buffer as a local memory slot
-  auto coordinationBuffer = memoryManager.allocateLocalMemorySlot(bufferMemorySpace, coordinationBufferSize);
+  auto coordinationBuffer = coordinationMemoryManager.allocateLocalMemorySlot(coordinationMemorySpace, coordinationBufferSize);
 
   // Initializing coordination buffer (sets to zero the counters)
   HiCR::channel::fixedSize::Base::initializeCoordinationBuffer(coordinationBuffer);
 
   // Exchanging local memory slots to become global for them to be used by the remote end
-  communicationManager.exchangeGlobalMemorySlots(CHANNEL_TAG,                          /* global tag */
-                                                 {{TOKEN_BUFFER_KEY, tokenBufferSlot}, /* key-slot pairs */
-                                                  {CONSUMER_COORDINATION_BUFFER_KEY, coordinationBuffer}});
+  coordinationCommunicationManager.exchangeGlobalMemorySlots(CHANNEL_TAG, {{CONSUMER_COORDINATION_BUFFER_KEY, coordinationBuffer}});
+  payloadCommunicationManager.exchangeGlobalMemorySlots(CHANNEL_TAG, {{TOKEN_BUFFER_KEY, tokenBufferSlot}});
 
   // Synchronizing so that all actors have finished registering their global memory slots
-  communicationManager.fence(CHANNEL_TAG);
+  coordinationCommunicationManager.fence(CHANNEL_TAG);
+  payloadCommunicationManager.fence(CHANNEL_TAG);
 
   // Obtaining the globally exchanged memory slots
-  auto globalTokenBufferSlot      = communicationManager.getGlobalMemorySlot(CHANNEL_TAG, TOKEN_BUFFER_KEY);
-  auto consumerCoordinationBuffer = communicationManager.getGlobalMemorySlot(CHANNEL_TAG, CONSUMER_COORDINATION_BUFFER_KEY);
+  auto consumerCoordinationBuffer = coordinationCommunicationManager.getGlobalMemorySlot(CHANNEL_TAG, CONSUMER_COORDINATION_BUFFER_KEY);
+  auto globalTokenBufferSlot      = payloadCommunicationManager.getGlobalMemorySlot(CHANNEL_TAG, TOKEN_BUFFER_KEY);
 
   // Creating producer and consumer channels
-  auto consumer = HiCR::channel::fixedSize::MPSC::locking::Consumer(communicationManager,
+  auto consumer = HiCR::channel::fixedSize::MPSC::locking::Consumer(coordinationCommunicationManager,
+                                                                    payloadCommunicationManager,
                                                                     globalTokenBufferSlot, /* tokenBuffer */
                                                                     coordinationBuffer,    /* internalCoordinationBuffer */
                                                                     consumerCoordinationBuffer,
@@ -90,18 +94,20 @@ void consumerFc(HiCR::MemoryManager               &memoryManager,
   }
 
   // Synchronizing so that all actors have finished registering their global memory slots
-  communicationManager.fence(CHANNEL_TAG);
+  coordinationCommunicationManager.fence(CHANNEL_TAG);
+  payloadCommunicationManager.fence(CHANNEL_TAG);
 
   // De-registering global slots
-  communicationManager.deregisterGlobalMemorySlot(globalTokenBufferSlot);
-  communicationManager.deregisterGlobalMemorySlot(consumerCoordinationBuffer);
+  payloadCommunicationManager.deregisterGlobalMemorySlot(globalTokenBufferSlot);
+  coordinationCommunicationManager.deregisterGlobalMemorySlot(consumerCoordinationBuffer);
 
-  communicationManager.destroyGlobalMemorySlot(globalTokenBufferSlot);
-  communicationManager.destroyGlobalMemorySlot(consumerCoordinationBuffer);
+  payloadCommunicationManager.destroyGlobalMemorySlot(globalTokenBufferSlot);
+  coordinationCommunicationManager.destroyGlobalMemorySlot(consumerCoordinationBuffer);
 
-  communicationManager.fence(CHANNEL_TAG);
+  coordinationCommunicationManager.fence(CHANNEL_TAG);
+  payloadCommunicationManager.fence(CHANNEL_TAG);
 
   // Freeing up local memory
-  memoryManager.freeLocalMemorySlot(tokenBufferSlot);
-  memoryManager.freeLocalMemorySlot(coordinationBuffer);
+  payloadMemoryManager.freeLocalMemorySlot(tokenBufferSlot);
+  coordinationMemoryManager.freeLocalMemorySlot(coordinationBuffer);
 }

examples/channels/fixedSize/mpsc/locking/include/producer.hpp

@@ -22,33 +22,39 @@
 #include <hicr/frontends/channel/fixedSize/mpsc/locking/producer.hpp>
 #include "common.hpp"
 
-void producerFc(HiCR::MemoryManager               &memoryManager,
-                HiCR::CommunicationManager        &communicationManager,
-                std::shared_ptr<HiCR::MemorySpace> bufferMemorySpace,
+void producerFc(HiCR::MemoryManager               &coordinationMemoryManager,
+                HiCR::MemoryManager               &payloadMemoryManager,
+                HiCR::CommunicationManager        &coordinationCommunicationManager,
+                HiCR::CommunicationManager        &payloadCommunicationManager,
+                std::shared_ptr<HiCR::MemorySpace> coordinationMemorySpace,
+                std::shared_ptr<HiCR::MemorySpace> payloadMemorySpace,
                 const size_t                       channelCapacity,
                 const size_t                       producerId)
 {
   // Getting required buffer size
   auto coordinationBufferSize = HiCR::channel::fixedSize::Base::getCoordinationBufferSize();
 
   // Registering token buffer as a local memory slot
-  auto coordinationBuffer = memoryManager.allocateLocalMemorySlot(bufferMemorySpace, coordinationBufferSize);
+  auto coordinationBuffer = coordinationMemoryManager.allocateLocalMemorySlot(coordinationMemorySpace, coordinationBufferSize);
 
   // Initializing coordination buffer (sets to zero the counters)
   HiCR::channel::fixedSize::Base::initializeCoordinationBuffer(coordinationBuffer);
 
   // Exchanging local memory slots to become global for them to be used by the remote end
-  communicationManager.exchangeGlobalMemorySlots(CHANNEL_TAG, {});
+  coordinationCommunicationManager.exchangeGlobalMemorySlots(CHANNEL_TAG, {});
+  payloadCommunicationManager.exchangeGlobalMemorySlots(CHANNEL_TAG, {});
 
   // Synchronizing so that all actors have finished registering their global memory slots
-  communicationManager.fence(CHANNEL_TAG);
+  coordinationCommunicationManager.fence(CHANNEL_TAG);
+  payloadCommunicationManager.fence(CHANNEL_TAG);
 
   // Obtaining the globally exchanged memory slots
-  auto globalTokenBufferSlot      = communicationManager.getGlobalMemorySlot(CHANNEL_TAG, TOKEN_BUFFER_KEY);
-  auto consumerCoordinationBuffer = communicationManager.getGlobalMemorySlot(CHANNEL_TAG, CONSUMER_COORDINATION_BUFFER_KEY);
+  auto globalTokenBufferSlot      = payloadCommunicationManager.getGlobalMemorySlot(CHANNEL_TAG, TOKEN_BUFFER_KEY);
+  auto consumerCoordinationBuffer = coordinationCommunicationManager.getGlobalMemorySlot(CHANNEL_TAG, CONSUMER_COORDINATION_BUFFER_KEY);
 
   // Creating producer and consumer channels
-  auto producer = HiCR::channel::fixedSize::MPSC::locking::Producer(communicationManager,
+  auto producer = HiCR::channel::fixedSize::MPSC::locking::Producer(coordinationCommunicationManager,
+                                                                    payloadCommunicationManager,
                                                                     globalTokenBufferSlot, /* tokenBuffer */
                                                                     coordinationBuffer,    /* internalCoordinationBuffer */
                                                                     consumerCoordinationBuffer,
@@ -58,7 +64,7 @@ void producerFc(HiCR::MemoryManager               &memoryManager,
   // Allocating a send slot to put the values we want to communicate
   ELEMENT_TYPE sendBuffer    = 0;
   auto         sendBufferPtr = &sendBuffer;
-  auto         sendSlot      = memoryManager.registerLocalMemorySlot(bufferMemorySpace, sendBufferPtr, sizeof(ELEMENT_TYPE));
+  auto         sendSlot      = payloadMemoryManager.registerLocalMemorySlot(payloadMemorySpace, sendBufferPtr, sizeof(ELEMENT_TYPE));
 
   // Pushing values to the channel, one by one, suspending when/if the channel is full
   for (size_t i = 0; i < MESSAGES_PER_PRODUCER; i++)
@@ -74,15 +80,17 @@ void producerFc(HiCR::MemoryManager               &memoryManager,
   }
 
   // Synchronizing so that all actors have finished registering their global memory slots
-  communicationManager.fence(CHANNEL_TAG);
+  coordinationCommunicationManager.fence(CHANNEL_TAG);
+  payloadCommunicationManager.fence(CHANNEL_TAG);
 
   // De-registering global slots
-  communicationManager.deregisterGlobalMemorySlot(globalTokenBufferSlot);
-  communicationManager.deregisterGlobalMemorySlot(consumerCoordinationBuffer);
+  payloadCommunicationManager.deregisterGlobalMemorySlot(globalTokenBufferSlot);
+  coordinationCommunicationManager.deregisterGlobalMemorySlot(consumerCoordinationBuffer);
 
   // Destroying global slots
-  communicationManager.fence(CHANNEL_TAG);
+  coordinationCommunicationManager.fence(CHANNEL_TAG);
+  payloadCommunicationManager.fence(CHANNEL_TAG);
 
   // Freeing up local memory
-  memoryManager.freeLocalMemorySlot(coordinationBuffer);
+  coordinationMemoryManager.freeLocalMemorySlot(coordinationBuffer);
 }

examples/channels/fixedSize/mpsc/locking/meson.build

@@ -1,21 +1,6 @@
-testSuite = [ 'examples', 'channels', 'fixedSize', 'mpsc', 'locking']
-test_timeout = 60
-
 exampleBuildIncludes = include_directories([
    'include'
   ])
-
-if 'mpi' in enabledBackends and 'hwloc' in enabledBackends 
-	mpi = executable('mpi', [ 'source/mpi.cpp' ], dependencies: hicrBuildDep, include_directories: [exampleBuildIncludes] )
-
-	if get_option('buildTests')
-	test('mpi', mpirunExecutable, args : [ '-n', '4', '--oversubscribe', '--host', 'localhost:16', mpi.full_path(), '3' ], timeout: test_timeout, suite: testSuite )
-	endif
-endif
 
-if 'lpf' in enabledBackends and 'hwloc' in enabledBackends 
-    lpf = executable('lpf', [ 'source/lpf.cpp' ], dependencies: hicrBuildDep, include_directories: [exampleBuildIncludes] )
-    if get_option('buildTests')
-        test('lpf', lpfrunExecutable, args : [ '-np', '4', '-engine', 'zero', lpf.full_path(), '3' ], timeout: test_timeout, suite: testSuite )
-    endif
-endif
+subdir('source/distributed') 
+subdir('source/local') 

examples/channels/fixedSize/mpsc/locking/source/lpf.cpp → examples/channels/fixedSize/mpsc/locking/source/distributed/lpf.cpp

@@ -20,8 +20,8 @@
 #include <hicr/backends/lpf/memoryManager.hpp>
 #include <hicr/backends/lpf/communicationManager.hpp>
 #include <hicr/backends/hwloc/topologyManager.hpp>
-#include "include/consumer.hpp"
-#include "include/producer.hpp"
+#include "../include/consumer.hpp"
+#include "../include/producer.hpp"
 
 // flag needed when using MPI to launch
 const int LPF_MPI_AUTO_INITIALIZE = 0;
@@ -81,8 +81,8 @@ void spmd(lpf_t lpf, lpf_pid_t pid, lpf_pid_t nprocs, lpf_args_t args)
   size_t rankId        = pid;
 
   // Rank 0 is consumer, the rest are producers
-  if (rankId == 0) consumerFc(m, c, firstMemorySpace, channelCapacity, producerCount);
-  if (rankId >= 1) producerFc(m, c, firstMemorySpace, channelCapacity, rankId);
+  if (rankId == 0) consumerFc(m, m, c, c, firstMemorySpace, firstMemorySpace, channelCapacity, producerCount);
+  if (rankId >= 1) producerFc(m, m, c, c, firstMemorySpace, firstMemorySpace, channelCapacity, rankId);
 }
 
 int main(int argc, char **argv)

examples/channels/fixedSize/mpsc/locking/source/distributed/meson.build

@@ -0,0 +1,45 @@
+testSuite = ['examples', 'channels', 'fixedSize', 'mpsc', 'locking', 'distributed']
+test_timeout = 60
+
+if 'mpi' in enabledBackends and 'hwloc' in enabledBackends
+    mpi = executable(
+        'mpi',
+        ['mpi.cpp'],
+        dependencies: hicrBuildDep,
+        include_directories: [exampleBuildIncludes],
+    )
+
+    if get_option('buildTests')
+        test(
+            'mpi',
+            mpirunExecutable,
+            args: [
+                '-n', '4',
+                '--oversubscribe',
+                '--host', 'localhost:16',
+                mpi.full_path(),
+                '3',
+            ],
+            timeout: test_timeout,
+            suite: testSuite,
+        )
+    endif
+endif
+
+if 'lpf' in enabledBackends and 'hwloc' in enabledBackends
+    lpf = executable(
+        'lpf',
+        ['lpf.cpp'],
+        dependencies: hicrBuildDep,
+        include_directories: [exampleBuildIncludes],
+    )
+    if get_option('buildTests')
+        test(
+            'lpf',
+            lpfrunExecutable,
+            args: ['-np', '4', '-engine', 'zero', lpf.full_path(), '3'],
+            timeout: test_timeout,
+            suite: testSuite,
+        )
+    endif
+endif