cylondata · mstaylor · Feb 4, 2026 · Mar 5, 2026 · Mar 12, 2026 · May 25, 2026
diff --git a/.github/workflows/conda-cpp.yml b/.github/workflows/conda-cpp.yml
@@ -68,11 +68,20 @@ jobs:
           ./configure --prefix=$HOME/ucc/install --with-ucx=$HOME/ucx/install
           make install
 
+      - name: Set library paths
+        run: |
+          mkdir -p $CONDA_PREFIX/etc/conda/activate.d
+          echo 'export LD_LIBRARY_PATH=$HOME/ucx/install/lib:$HOME/ucc/install/lib:$LD_LIBRARY_PATH' > $CONDA_PREFIX/etc/conda/activate.d/ucx_paths.sh
+
       - name: Build cylon, pycylon and run cpp test
         run: python build.py -cmake-flags="-DCYLON_UCX=1 -DCYLON_GLOO=1 -DGLOO_INSTALL_PREFIX=$HOME/gloo/install -DCYLON_UCC=1 -DUCC_INSTALL_PREFIX=$HOME/ucc/install -DUCX_INSTALL_PREFIX=$HOME/ucx/install" -ipath="$HOME/cylon/install" --cpp --python --test
 
       - name: Run pytest
-        run: python build.py -ipath="$HOME/cylon/install" --pytest
+        run: |
+          export LD_LIBRARY_PATH=$HOME/ucx/install/lib:$HOME/ucc/install/lib:$LD_LIBRARY_PATH
+          python build.py -ipath="$HOME/cylon/install" --pytest
 
       - name: Build Java
-        run: python build.py -ipath="$HOME/cylon/install" --java
+        run: |
+          export LD_LIBRARY_PATH=$HOME/ucx/install/lib:$HOME/ucc/install/lib:$LD_LIBRARY_PATH
+          python build.py -ipath="$HOME/cylon/install" --java
diff --git a/build.py b/build.py
@@ -173,6 +173,7 @@ def parse_cmake_flags(flag):
 CYLON_UCX = parse_cmake_flags('CYLON_UCX')
 CYLON_UCC = parse_cmake_flags('CYLON_UCC')
 CYLON_FMI = parse_cmake_flags('CYLON_FMI')
+CYLON_SIMD = parse_cmake_flags('CYLON_SIMD')
 UCX_INSTALL_PREFIX = parse_cmake_flags('UCX_INSTALL_PREFIX')
 UCC_PREFIX = parse_cmake_flags('UCC_INSTALL_PREFIX')
 REDIS_PREFIX = parse_cmake_flags('REDIS_INSTALL_PREFIX')
@@ -199,6 +200,7 @@ def print_line():
 logger.info(f" -CYLON_UCX    : {CYLON_UCX}")
 logger.info(f" -CYLON_UCC    : {CYLON_UCC}")
 logger.info(f" -CYLON_FMI    : {CYLON_FMI}")
+logger.info(f" -CYLON_SIMD   : {CYLON_SIMD}")
 logger.info(f" -UCC_PREFIX   : {UCC_PREFIX}")
 logger.info(f"Run C++ tests  : {RUN_CPP_TESTS}")
 logger.info(f"Build PyCylon  : {BUILD_PYTHON}")
@@ -310,6 +312,8 @@ def python_test():
                                          env['LD_LIBRARY_PATH']
             if CYLON_FMI:
                 env['CYLON_FMI'] = str(CYLON_FMI)
+            if CYLON_SIMD:
+                env['CYLON_SIMD'] = str(CYLON_SIMD)
 
         elif OS_NAME == 'Darwin':
             if 'DYLD_LIBRARY_PATH' in env:
@@ -364,6 +368,9 @@ def build_python():
         env['CYLON_REDIS'] = str(CYLON_REDIS)
         env['REDIS_PREFIX'] = REDIS_PREFIX
 
+    if CYLON_SIMD:
+        env['CYLON_SIMD'] = str(CYLON_SIMD)
+
     logger.info("Arrow prefix: " + str(Path(conda_prefix)))
 
     # Diagnostic logging for pycylon build
@@ -389,7 +396,6 @@ def build_python():
         logger.error("setup.py egg_info failed - see error above")
         check_status(test_res.returncode, "PyCylon setup.py validation")
 
-    # Use legacy setup.py develop mode which avoids PEP 517 entirely
     cmd = f'{PYTHON_EXEC} setup.py build_ext --inplace && {PYTHON_EXEC} -m pip install -v --no-build-isolation {clean} .'
     res = subprocess.run(cmd, shell=True, env=env, cwd=PYTHON_SOURCE_DIR)
     check_status(res.returncode, "PyCylon build")

diff --git a/conda/environments/cylon.yml b/conda/environments/cylon.yml
@@ -12,6 +12,7 @@ dependencies:
   - glog
   - openmpi
   - ucx
+  - pip
   - cython>=0.29.31
   - numpy>=1.23,<2.0a0
   - pandas>=2.0,<2.2.3dev

diff --git a/conda/environments/cylon_NoUCX.yml b/conda/environments/cylon_NoUCX.yml
@@ -12,6 +12,7 @@ dependencies:
   - libarrow-dataset==16.1.0.*
   - libarrow==16.1.0.*
   - glog
+  - pip
   - openmpi
   - cython>=0.29.31
   - numpy>=1.23,<2.0a0

diff --git a/conda/environments/gcylon.yml b/conda/environments/gcylon.yml
@@ -20,6 +20,7 @@ dependencies:
   - glog
   - openmpi
   - ucx
+  - pip
   - numpy>=1.23,<2.0a0
   - pandas>=2.0,<2.2.3dev
   - fsspec>=0.6.0

diff --git a/conda/environments/gcylon_cuda13.yml b/conda/environments/gcylon_cuda13.yml
@@ -16,6 +16,7 @@ dependencies:
   - libcudf=26.02
   - rmm=26.02
   - librmm=26.02
+  - pip
   # CUDA 13.x toolkit
   - cuda-version=13
   - cuda-cudart

diff --git a/cpp/CMakeLists.txt b/cpp/CMakeLists.txt
@@ -403,6 +403,11 @@ if (CYLON_LIBFABRIC)
     message(STATUS "Libfabric lib: ${LIBFABRIC_LIB}")
 endif (CYLON_LIBFABRIC)
 
+# SIMD operations (cosine similarity, batch search)
+if (CYLON_SIMD)
+    message("Cylon SIMD Enabled")
+    add_definitions(-DBUILD_CYLON_SIMD)
+endif ()
 
 # Arrow
 if (NOT ARROW_BUILD_TYPE)

diff --git a/cpp/src/cylon/CMakeLists.txt b/cpp/src/cylon/CMakeLists.txt
@@ -124,9 +124,18 @@ else (CYLON_LIBFABRIC)
     set(CYLON_LIBFABRIC_FILES)
 endif (CYLON_LIBFABRIC)
 
+if (CYLON_SIMD)
+    set(CYLON_SIMD_FILES
+            simd/simd_ops.hpp
+            simd/simd_ops.cpp)
+else ()
+    set(CYLON_SIMD_FILES)
+endif ()
+
 add_library(cylon SHARED
         ${UCX_CYLON_FILES}
         ${UCC_CYLON_FILES}
+        ${CYLON_SIMD_FILES}
         ${CYLON_GLOO_FILES}
         ${CYLON_LIBFABRIC_FILES}
         ${UCX_REDIS_CYLON_FILES}
@@ -383,6 +392,9 @@ add_subdirectory(thridparty/fmi/comm)
 add_subdirectory(thridparty/fmi/utils)
 add_subdirectory(compute)
 add_subdirectory(checkpoint)
+if (CYLON_SIMD)
+    add_subdirectory(simd)
+endif ()
 
 set_target_properties(cylon PROPERTIES VERSION ${CYLON_VERSION})
 

diff --git a/cpp/src/cylon/net/fmi/fmi_channel.cpp b/cpp/src/cylon/net/fmi/fmi_channel.cpp
@@ -254,7 +254,15 @@ namespace cylon::fmi {
                 if (rank == recvRank) {
                     continue;//FMI does not support local receives, so process during sends
                 }
-            }
+                // Init a new pending receive for the request
+                auto *buf = new PendingReceive();
+                buf->receiveId = recvRank;
+                // Add to pendingReceive object to pendingReceives map
+                pendingReceives.insert(std::pair<int, PendingReceive *>(recvRank, buf));
+                // Receive for the initial header buffer
+                // Init context
+                buf->context = new FMI::Utils::fmiContext;
+                buf->context->completed = 0;
 
                 auto send_data_byte_size = CYLON_CHANNEL_HEADER_SIZE * sizeof(int);
                 auto send_void_ptr = const_cast<void *>(static_cast<const void *>(buf->headerBuf));
@@ -404,34 +412,12 @@ namespace cylon::fmi {
                 peerBusy = true;
                 break;
             }
-        } else if (pend_send->status == SEND_FINISH) {
-            // we are going to send complete
-            send_comp_fn->sendFinishComplete(finishRequests[rank]);
-            pend_send->status = SEND_DONE;
-        } else if (pend_send->status != SEND_DONE) {
-            // throw an exception and log
-            LOG(FATAL) << "At an un-expected state " << pend_send->status;
-        }
-    }
-
-    void FMIChannel::progressSendTo(int peer_id) {
-        // Role-based ordering: Only send first if rank < peer_id
-        //if (worldSize > 2) {
-        //   if (rank >= peer_id) return;
-        //}
-
-        PendingSend *ps = sends[peer_id];
-
-        if (peer_id == rank) {
-            progressSendsLocal(sends[peer_id]);
-            return;
         }
 
         if (peerBusy) {
             release_lock(lock_key, lock_val);
             return;
         }
-    }
 
         // Check if peer is RECEIVING
         auto peerRecvStatusStr = redis->get(peer_receive_status_key);
@@ -609,15 +595,6 @@ namespace cylon::fmi {
                             // Check if request is in finish
                             if (finishRequests.count(dest)) {
                                 sendFinishHeader(x);
-                            } else {
-                                // If pending data is empty
-                                // Notify about send completion
-                                send_comp_fn->sendComplete(x.second->currentSend);
-                                x.second->currentSend = {};
-
-                            // Check if request is in finish
-                            if (finishRequests.find(x.first) != finishRequests.end()) {
-                                sendFinishHeader(x);
                             } else {
                                 // If req is not in finish then re-init
                                 x.second->status = SEND_INIT;
@@ -645,15 +622,7 @@ namespace cylon::fmi {
 
     void FMIChannel::progressReceiveFrom(int peer_id) {
 
-        //check if ok to receive (can't rely on sender to block sending
-        //so, we need to check for socket activity in blocking mode
-        /*if (!communicator->checkIfOkToReceive(peer_id)) {
-            LOG(INFO) << "unable to receive -- releasing lock key: " << lock_key << "peerId: " << peer_id;
-            release_lock(lock_key, lock_val);
-            return;
-        }*/
-
-        publishStatus(rank, peer_id, RECEIVING, RECEIVE);
+        if (peer_id == rank) return;
 
         PendingReceive *recv = pendingReceives[peer_id];
 
@@ -775,101 +744,12 @@ namespace cylon::fmi {
             publishStatus(rank, peer_id, IDLE, RECEIVE);
         }
 
-        if (recv->status == RECEIVE_LENGTH_POSTED && recv->context->completed == 1) {
-
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-                LOG(INFO) << "finished RECEIVE_INIT -- releasing lock key: " << lock_key << " peer_id: " << peer_id;
-                //release_lock(lock_key, lock_val);
-                return;
-            } else {
-                release_lock(lock_key, lock_val);
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-                return;
-            }
-        } else if (recv->status == RECEIVE_LENGTH_POSTED && recv->context->completed == 1) {
-
-            int length     = recv->headerBuf[0];
-            int finFlag    = recv->headerBuf[1];
-
-            if (finFlag == CYLON_MSG_FIN) {
-                release_lock(lock_key, lock_val);
-                recv->status = RECEIVED_FIN;
-                rcv_fn->receivedHeader(peer_id, finFlag, nullptr, 0);
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-                LOG(INFO) << "[rank " << rank << "] ✅ Received FIN from " << peer_id;
-                LOG(INFO) << "finished CYLON_MSG_FIN -- releasing lock key: " << lock_key << " peer_id: " << peer_id;
-
-                return;
-            }
-
-            if (communicator->checkIfOkToReceive(peer_id, FMI::Utils::BLOCKING)) {
-                release_lock(lock_key, lock_val);
-                delete recv->context;
-                recv->context = new FMI::Utils::fmiContext();
-                recv->context->completed = 0;
-
-                allocator->Allocate(length, &recv->data);
-                recv->length = length;
-
-                FMI::Comm::Data<void *> payload(recv->data->GetByteBuffer(), length,
-                                                FMI::Comm::noop_deleter);
-                FMI_Irecv(payload, peer_id, recv->context);
-                recv->status = RECEIVE_POSTED;
-
-                int *header = new int[6];
-                std::memcpy(header, &recv->headerBuf[2], 6 * sizeof(int));
-                rcv_fn->receivedHeader(peer_id, finFlag, header, 6);
-
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-                //LOG(INFO) << "finished RECEIVE_LENGTH_POSTED -- releasing lock key: " << lock_key << " peer_id: "
-                //          << peer_id;
-
-                return;
-            } else {
-                release_lock(lock_key, lock_val);
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-                return;
-            }
-        } else if  (recv->status == RECEIVE_POSTED && recv->context->completed == 1) {
-            if (communicator->checkIfOkToReceive(peer_id, FMI::Utils::BLOCKING)) {
-                release_lock(lock_key, lock_val);
-                rcv_fn->receivedData(peer_id, recv->data, recv->length);
-
-                std::fill_n(recv->headerBuf, CYLON_CHANNEL_HEADER_SIZE, 0);
-                delete recv->context;
-                recv->context = new FMI::Utils::fmiContext();
-                recv->context->completed = 0;
-
-                FMI::Comm::Data<void *> next_header(recv->headerBuf,
-                                                    CYLON_CHANNEL_HEADER_SIZE * sizeof(int),
-                                                    FMI::Comm::noop_deleter);
-                FMI_Irecv(next_header, peer_id, recv->context);
-                recv->status = RECEIVE_LENGTH_POSTED;
-
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-                LOG(INFO) << "finished RECEIVE_POSTED -- releasing lock key: " << lock_key << " peer_id: " << peer_id;
-            } else {
-                release_lock(lock_key, lock_val);
-                publishStatus(rank, peer_id, IDLE, RECEIVE);
-            }
-        } else {
-            publishStatus(rank, peer_id, IDLE, RECEIVE);
-        }
-
-    }
-
     }
 
     void FMIChannel::progressReceives() {
 
         if (mode == FMI::Utils::NONBLOCKING) {
 
-        if (mode == FMI::Utils::BLOCKING) {
-            /*for (auto x: pendingReceives) {
-                progressReceiveFrom(x.first);
-            }*/
-        } else {
-
             communicator->communicator_event_progress(FMI::Utils::Operation::RECEIVE);
 
             // Iterate through the pending receives
@@ -1162,25 +1042,6 @@ namespace cylon::fmi {
             });
         }
 
-        bool FMIChannel::isSendComplete(int peer_id) {
-            auto it = sends.find(peer_id);
-            if (it == sends.end() || it->second == nullptr) {
-                return false;
-            }
-            PendingSend* ps = sends[peer_id];
-            return ps->status == SEND_DONE;
-        }
-
-        bool FMIChannel::isReceiveComplete(int peer_id) {
-            auto it = pendingReceives.find(peer_id);
-            if (it == pendingReceives.end() || it->second == nullptr) {
-                return false;
-            }
-            PendingReceive* recv = pendingReceives[peer_id];
-            return recv->status == RECEIVED_FIN;
-        }
-
-
     }
 
     std::shared_ptr<std::mutex> FMIChannel::getSendMutex(int peer_id) {
-Original file line number
+Diff line change
@@ Expand Up / @@ -12,6 +12,7 @@ dependencies: @@
       - glog
       - openmpi
       - ucx
+      - pip
       - cython>=0.29.31
       - numpy>=1.23,<2.0a0
       - pandas>=2.0,<2.2.3dev
@@ Expand Down @@