2nd release of messaging refactor

.github/workflows/cd.yml

@@ -19,7 +19,7 @@ jobs:
 
     - name: Build artifacts 
       run: |
-        pipx run poetry build
+        poetry build
 
     - name: Archive artifacts
       uses: actions/upload-artifact@v3
@@ -57,6 +57,7 @@ jobs:
           pip install --no-input $artifact
           python -c 'import lava.magma.compiler.subcompilers'
           python -c 'import lava.magma.core.model'
+          python -c 'from lava.magma.runtime.message_infrastructure.multiprocessing import MultiProcessing'
           pip uninstall -y lava-nc
           deactivate
           rm -rf artifact-test
@@ -69,11 +70,11 @@ jobs:
           pip install --no-input $artifact
           python -c 'import lava.magma.compiler.subcompilers'
           python -c 'import lava.magma.core.model'
+          python -c 'from lava.magma.runtime.message_infrastructure.multiprocessing import MultiProcessing'
           pip uninstall -y lava-nc
           deactivate
           rm -rf artifact-test
 
-
   test-artifact-use:
     name: Test Artifact With Unit Tests
     runs-on: ubuntu-latest

.github/workflows/ci.yml

@@ -21,10 +21,24 @@ jobs:
     - name: setup CI
       uses: lava-nc/[email protected]
       with:
-        repository: 'Lava'    
+        repository: 'Lava'
 
     - name: Run flakeheaven (flake8)
-      run: poetry run flakeheaven lint src/lava tests/
+      run: poetry run flakeheaven lint src/lava tests/ prebuild.py
+
+    - name: Run cpplint (cpplint)
+      run: |
+        poetry run cpplint --recursive --quiet \
+        --repository=src/lava/magma/runtime/_c_message_infrastructure \
+        --root=csrc \
+        --exclude=src/lava/magma/runtime/_c_message_infrastructure/csrc/channel/dds/protos \
+        src/lava/magma/runtime/_c_message_infrastructure/ \
+        src/lava/magma/runtime/_c_message_infrastructure/test/
+
+        poetry run cpplint --recursive --quiet \
+        --repository=src/lava/magma/runtime/_c_message_infrastructure/examples/ros2/ros_talk_with_dds_cpp \
+        --root=src \
+        src/lava/magma/runtime/_c_message_infrastructure/examples/ros2/ros_talk_with_dds_cpp/src
 
   security-lint:
     name: Security Lint Code
@@ -44,14 +58,15 @@ jobs:
       with:
         targets: |
           src/lava/.      
-        options: "-r  --format custom --msg-template '{abspath}:{line}: {test_id}[bandit]: {severity}: {msg}'"
+        options: "-r --exclude build --format custom --msg-template '{abspath}:{line}: {test_id}[bandit]: {severity}: {msg}'"
 
   unit-tests:
     name: Unit Test Code + Coverage
     runs-on: ${{ matrix.operating-system }}
     strategy:
       matrix:
-        operating-system: [ubuntu-latest, windows-latest, macos-latest]
+        # TODO: restore it before merge into ML
+        operating-system: [ubuntu-latest,] # windows-latest, macos-latest]
     steps:
     - uses: actions/checkout@v3
       with:
@@ -67,3 +82,26 @@ jobs:
         poetry run git lfs fetch
         poetry run git lfs pull
         poetry run pytest
+
+  msg-infr-unit-tests:
+    name: Message Infrastructure Unit Test
+    runs-on: ${{ matrix.operating-system }}
+    env:
+      DEBUG: 1
+
+    strategy:
+      matrix:
+        # TODO: restore it before merge into ML
+        operating-system: [ubuntu-latest,] #windows-latest, macos-latest]
+    steps:
+    - uses: actions/checkout@v3
+      with:
+        lfs: true
+
+    - name: setup CI
+      uses: lava-nc/[email protected]
+      with:
+        repository: 'Lava'
+
+    - name: Run message infrastructure cpp unit tests
+      run: ./build/test/test_messaging_infrastructure

.gitignore

@@ -16,6 +16,7 @@ src/lava/utils/dataloader/mnist.npy
 # Distribution / packaging
 .Python
 build/
+log/
 develop-eggs/
 dist/
 downloads/

prebuild.py

@@ -0,0 +1,76 @@
+import os
+import multiprocessing
+import numpy
+import subprocess  # nosec
+import sys
+
+
+class CMake:
+    def __init__(self, sourcedir, targetdir):
+        self.sourcedir = os.path.abspath(sourcedir)
+        self.targetdir = os.path.abspath(targetdir)
+        self.env = os.environ.copy()
+        self.from_poetry = self._check_poetry()
+        self.from_cd_action = self._check_cd_action()
+        self.cmake_command = ["poetry", "run", "cmake"] \
+            if self.from_cd_action and self.from_poetry else ["cmake"]
+        self.cmake_args = []
+        self.build_args = []
+
+    def _check_poetry(self):
+        exec_code = self.env.get('_', "").rsplit('/')[-1]
+        if exec_code == 'poetry':
+            return True
+        return False
+
+    def _check_cd_action(self):
+        event_name = self.env.get('GITHUB_EVENT_NAME', '')
+        return event_name == 'workflow_dispatch'
+
+    def _set_cmake_path(self):
+        self.temp_path = os.path.join(os.path.abspath(""), "build")
+        if not os.path.exists(self.temp_path):
+            os.makedirs(self.temp_path)
+
+    def _set_cmake_args(self):
+        debug = int(os.environ.get("DEBUG", 0))
+        cfg = "Debug" if debug else "Release"
+        if self.from_cd_action and self.from_poetry:
+            python_env = subprocess.check_output(["poetry", "env", "info", "-p"]) \
+                .decode().strip() + "/bin/python3" # nosec # noqa
+            numpy_include_dir = subprocess.check_output(["poetry", "run", # nosec # noqa
+                "python3", "-c", "import numpy; print(numpy.get_include())"]).decode().strip() # nosec # noqa
+        else:
+            python_env = sys.executable
+            numpy_include_dir = numpy.get_include()
+        self.cmake_args += [
+            f"-DCMAKE_LIBRARY_OUTPUT_DIRECTORY={self.targetdir}",
+            f"-DPYTHON_EXECUTABLE={python_env}",
+            f"-DCMAKE_BUILD_TYPE={cfg}",  # not used on MSVC, but no harm
+        ]
+        if "CMAKE_ARGS" in os.environ:
+            self.cmake_args += [item for item in
+                                os.environ["CMAKE_ARGS"].split(" ") if item]
+        # Set numpy include header to cpplib
+        self.cmake_args += [
+            f"-DNUMPY_INCLUDE_DIRS={numpy_include_dir}"]
+
+        # Set CMAKE_BUILD_PARALLEL_LEVEL to control the parallel build level
+        # across all generators.
+        if "CMAKE_BUILD_PARALLEL_LEVEL" not in os.environ:
+            self.parallel = multiprocessing.cpu_count()
+            self.build_args += [f"-j{self.parallel}"]
+
+    def run(self):
+        self._set_cmake_path()
+        self._set_cmake_args()
+        subprocess.check_call([*self.cmake_command, self.sourcedir] + self.cmake_args, cwd=self.temp_path, env=self.env)  # nosec # noqa
+        subprocess.check_call([*self.cmake_command, "--build", "."] + self.build_args,  cwd=self.temp_path, env=self.env)  # nosec # noqa
+
+
+if __name__ == '__main__':
+    base_runtime_path = "src/lava/magma/runtime/"
+    sourcedir = f"{base_runtime_path}_c_message_infrastructure"
+    targetdir = f"{base_runtime_path}message_infrastructure"
+    cmake = CMake(sourcedir, targetdir)
+    cmake.run()

pyproject.toml

@@ -1,6 +1,6 @@
 
 [build-system]
-requires = ["poetry-core>=1.2.0"]
+requires = ["poetry-core>=1.2.0", "wheel", "cmake>=3.12", "numpy>=1.22.2", "pybind11[global]>=2.10.1"]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
@@ -9,8 +9,12 @@ packages = [
     {include = "lava", from = "src"},
     {include = "tests"}
 ]
-include = ["tutorials"]
+
+include = ["tutorials",
+           "src/lava/magma/runtime/message_infrastructure/*.so",
+           "src/lava/magma/runtime/message_infrastructure/install/lib/lib*"]
 version = "0.6.0"
+
 description = "A Software Framework for Neuromorphic Computing"
 homepage = "https://lava-nc.org/"
 repository = "https://github.com/lava-nc/lava"
@@ -51,6 +55,7 @@ numpy = "^1.22.2"
 scipy = "^1.8.0"
 networkx = "<=2.8.7"
 asteval = "^0.9.27"
+pybind11 = {extras = ["global"], version = "^2.10.1"}
 
 [tool.poetry.dev-dependencies]
 bandit = "1.7.4"
@@ -82,6 +87,12 @@ autopep8 = "^1.6.0"
 ipykernel = "^6.15.0"
 nbformat = "^5.3.0"
 nbconvert = "^7.2.0"
+cpplint = "^1.6.0"
+psutil = "^5.9.4"
+
+[tool.poetry.build]
+generate-setup-file = false
+script = "prebuild.py"
 
 [tool.black]
 line-length = 80
@@ -137,7 +148,7 @@ extended_default_ignore=[]  # Fix for bug while using newer flake8 ver.
 format = "grouped"
 max_line_length = 80
 show_source = true
-exclude = ["./docs/"]
+exclude = ["./docs/", "./src/lava/magma/runtime/message_infrastructure/install"]
 
 [tool.flakeheaven.plugins]
 flake8-bandit = ["+*", "-S322", "-B101", "-S404", "-S602"]       # Enable a plugin, disable specific checks

src/lava/magma/compiler/builders/channel_builder.py

@@ -10,9 +10,11 @@
     AbstractProcessModel
 from lava.magma.compiler.builders. \
     runtimeservice_builder import RuntimeServiceBuilder
-from lava.magma.compiler.channels.interfaces import (
+from lava.magma.runtime.message_infrastructure import (
     Channel,
-    ChannelType,
+    ChannelBackend,
+    ChannelQueueSize,
+    SyncChannelBytes
 )
 from lava.magma.compiler.utils import PortInitializer
 from lava.magma.runtime.message_infrastructure \
@@ -29,7 +31,7 @@ class ChannelBuilderMp(AbstractChannelBuilder):
     and multi processing backbone.
     """
 
-    channel_type: ChannelType
+    channel_type: ChannelBackend
     src_process: "AbstractProcess"
     dst_process: "AbstractProcess"
     src_port_initializer: PortInitializer
@@ -54,17 +56,12 @@ def build(
         Exception
             Can't build channel of type specified
         """
-        channel_class = messaging_infrastructure.channel_class(
-            channel_type=self.channel_type
-        )
-        return channel_class(
-            messaging_infrastructure,
-            self.src_port_initializer.name,
-            self.dst_port_initializer.name,
-            self.src_port_initializer.shape,
-            self.src_port_initializer.d_type,
-            self.src_port_initializer.size,
-        )
+        return Channel(self.channel_type,
+                       ChannelQueueSize,
+                       self.src_port_initializer.bytes,
+                       self.src_port_initializer.name,
+                       self.dst_port_initializer.name
+                       )
 
 
 @dataclass
@@ -73,7 +70,7 @@ class ServiceChannelBuilderMp(AbstractChannelBuilder):
     as messaging and multi processing backbone.
     """
 
-    channel_type: ChannelType
+    channel_type: ChannelBackend
     src_process: ty.Union[RuntimeServiceBuilder,
                           ty.Type["AbstractProcessModel"]]
     dst_process: ty.Union[RuntimeServiceBuilder,
@@ -99,19 +96,11 @@ def build(
         Exception
             Can't build channel of type specified
         """
-        channel_class = messaging_infrastructure.channel_class(
-            channel_type=self.channel_type
-        )
-
-        channel_name: str = self.port_initializer.name
-        return channel_class(
-            messaging_infrastructure,
-            channel_name + "_src",
-            channel_name + "_dst",
-            self.port_initializer.shape,
-            self.port_initializer.d_type,
-            self.port_initializer.size,
-        )
+        return Channel(ChannelBackend.SHMEMCHANNEL,
+                       ChannelQueueSize,
+                       SyncChannelBytes,
+                       self.port_initializer.name,
+                       self.port_initializer.name)
 
 
 @dataclass
@@ -120,7 +109,7 @@ class RuntimeChannelBuilderMp(AbstractChannelBuilder):
     used as messaging and multi processing backbone.
     """
 
-    channel_type: ChannelType
+    channel_type: ChannelBackend
     src_process: ty.Union[RuntimeServiceBuilder, ty.Type["Runtime"]]
     dst_process: ty.Union[RuntimeServiceBuilder, ty.Type["Runtime"]]
     port_initializer: PortInitializer
@@ -144,19 +133,11 @@ def build(
         Exception
             Can't build channel of type specified
         """
-        channel_class = messaging_infrastructure.channel_class(
-            channel_type=self.channel_type
-        )
-
-        channel_name: str = self.port_initializer.name
-        return channel_class(
-            messaging_infrastructure,
-            channel_name + "_src",
-            channel_name + "_dst",
-            self.port_initializer.shape,
-            self.port_initializer.d_type,
-            self.port_initializer.size,
-        )
+        return Channel(ChannelBackend.SHMEMCHANNEL,
+                       ChannelQueueSize,
+                       SyncChannelBytes,
+                       self.port_initializer.name,
+                       self.port_initializer.name)
 
 
 @dataclass
@@ -165,7 +146,7 @@ class ChannelBuilderNx(AbstractChannelBuilder):
     infrastructure.
     """
 
-    channel_type: ChannelType
+    channel_type: ChannelBackend
     src_process: "AbstractProcess"
     dst_process: "AbstractProcess"
     src_port_initializer: PortInitializer