diff --git a/.ci/jenkins/lib/build-matrix.yaml b/.ci/jenkins/lib/build-matrix.yaml index 55b4f09b8f..3627b2aa27 100644 --- a/.ci/jenkins/lib/build-matrix.yaml +++ b/.ci/jenkins/lib/build-matrix.yaml @@ -43,7 +43,7 @@ env: TEST_TIMEOUT: 30 UCX_TLS: "^shm" STORAGE_DRIVER: 'overlay' - CI_IMAGE_TAG: "20260310-1" + CI_IMAGE_TAG: "20260313-1" runs_on_dockers: diff --git a/.ci/jenkins/lib/test-matrix.yaml b/.ci/jenkins/lib/test-matrix.yaml index cc0768e5fe..19810a06a5 100644 --- a/.ci/jenkins/lib/test-matrix.yaml +++ b/.ci/jenkins/lib/test-matrix.yaml @@ -49,7 +49,7 @@ env: SLURM_JOB_TIMEOUT: '02:20:00' SLURM_IMMEDIATE_TIMEOUT: "3600" STORAGE_DRIVER: overlay - CI_IMAGE_TAG: "20260310-1" + CI_IMAGE_TAG: "20260313-1" empty_volumes: - {mountPath: /var/lib/containers/storage, memory: false} diff --git a/.ci/scripts/common.sh b/.ci/scripts/common.sh index f714f765e8..f0e225795c 100755 --- a/.ci/scripts/common.sh +++ b/.ci/scripts/common.sh @@ -132,3 +132,25 @@ wait_for_etcd() { done echo "Etcd is ready" } + +start_etcd_server() { + local namespace_prefix=$1 + if [ -z "${namespace_prefix}" ]; then + echo "Usage: start_etcd_server " + exit 1 + fi + + echo "==== Running ETCD server ====" + etcd_port=$(get_next_tcp_port) + etcd_peer_port=$(get_next_tcp_port) + export NIXL_ETCD_ENDPOINTS="http://127.0.0.1:${etcd_port}" + export NIXL_ETCD_PEER_URLS="http://127.0.0.1:${etcd_peer_port}" + export NIXL_ETCD_NAMESPACE="${namespace_prefix}/${etcd_port}" + + etcd --listen-client-urls "${NIXL_ETCD_ENDPOINTS}" --advertise-client-urls "${NIXL_ETCD_ENDPOINTS}" \ + --listen-peer-urls "${NIXL_ETCD_PEER_URLS}" --initial-advertise-peer-urls "${NIXL_ETCD_PEER_URLS}" \ + --initial-cluster "default=${NIXL_ETCD_PEER_URLS}" & + ETCD_PID=$! + + wait_for_etcd +} diff --git a/.gitlab/test_cpp.sh b/.gitlab/test_cpp.sh index 7674ba819c..33e8a3e4d5 100755 --- a/.gitlab/test_cpp.sh +++ b/.gitlab/test_cpp.sh @@ -72,18 +72,7 @@ else echo "/sys/module/nv_peer_mem/version not found" fi -echo "==== Running ETCD server ====" -etcd_port=$(get_next_tcp_port) -etcd_peer_port=$(get_next_tcp_port) -export NIXL_ETCD_ENDPOINTS="http://127.0.0.1:${etcd_port}" -export NIXL_ETCD_PEER_URLS="http://127.0.0.1:${etcd_peer_port}" -export NIXL_ETCD_NAMESPACE="/nixl/cpp_ci/${etcd_port}" -etcd --listen-client-urls ${NIXL_ETCD_ENDPOINTS} --advertise-client-urls ${NIXL_ETCD_ENDPOINTS} \ - --listen-peer-urls ${NIXL_ETCD_PEER_URLS} --initial-advertise-peer-urls ${NIXL_ETCD_PEER_URLS} \ - --initial-cluster default=${NIXL_ETCD_PEER_URLS} & -ETCD_PID=$! - -wait_for_etcd +start_etcd_server "/nixl/cpp_ci" echo "==== Running C++ tests ====" cd ${INSTALL_DIR} diff --git a/.gitlab/test_nixlbench.sh b/.gitlab/test_nixlbench.sh index 37f94d07c8..84cb471207 100755 --- a/.gitlab/test_nixlbench.sh +++ b/.gitlab/test_nixlbench.sh @@ -45,18 +45,7 @@ nvidia-smi topo -m || true ibv_devinfo || true uname -a || true -echo "==== Running ETCD server ====" -etcd_port=$(get_next_tcp_port) -etcd_peer_port=$(get_next_tcp_port) -export NIXL_ETCD_ENDPOINTS="http://127.0.0.1:${etcd_port}" -export NIXL_ETCD_PEER_URLS="http://127.0.0.1:${etcd_peer_port}" -export NIXL_ETCD_NAMESPACE="/nixl/nixlbench_ci/${etcd_port}" -etcd --listen-client-urls ${NIXL_ETCD_ENDPOINTS} --advertise-client-urls ${NIXL_ETCD_ENDPOINTS} \ - --listen-peer-urls ${NIXL_ETCD_PEER_URLS} --initial-advertise-peer-urls ${NIXL_ETCD_PEER_URLS} \ - --initial-cluster default=${NIXL_ETCD_PEER_URLS} & -ETCD_PID=$! - -wait_for_etcd +start_etcd_server "/nixl/nixlbench_ci" echo "==== Running Nixlbench tests ====" cd ${INSTALL_DIR} diff --git a/.gitlab/test_python.sh b/.gitlab/test_python.sh index 6d9ab8d4e5..78e732b20b 100755 --- a/.gitlab/test_python.sh +++ b/.gitlab/test_python.sh @@ -68,18 +68,7 @@ $pip3 install --break-system-packages pytest $pip3 install --break-system-packages pytest-timeout $pip3 install --break-system-packages zmq -echo "==== Running ETCD server ====" -etcd_port=$(get_next_tcp_port) -etcd_peer_port=$(get_next_tcp_port) -export NIXL_ETCD_ENDPOINTS="http://127.0.0.1:${etcd_port}" -export NIXL_ETCD_PEER_URLS="http://127.0.0.1:${etcd_peer_port}" -export NIXL_ETCD_NAMESPACE="/nixl/python_ci/${etcd_port}" -etcd --listen-client-urls ${NIXL_ETCD_ENDPOINTS} --advertise-client-urls ${NIXL_ETCD_ENDPOINTS} \ - --listen-peer-urls ${NIXL_ETCD_PEER_URLS} --initial-advertise-peer-urls ${NIXL_ETCD_PEER_URLS} \ - --initial-cluster default=${NIXL_ETCD_PEER_URLS} & -ETCD_PID=$! - -wait_for_etcd +start_etcd_server "/nixl/python_ci" echo "==== Running python tests ====" pytest -s test/python diff --git a/.gitlab/test_rust.sh b/.gitlab/test_rust.sh index 317dc35a5e..6b59490682 100755 --- a/.gitlab/test_rust.sh +++ b/.gitlab/test_rust.sh @@ -1,5 +1,5 @@ #!/bin/sh -# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -14,6 +14,10 @@ # See the License for the specific language governing permissions and # limitations under the License. + +# shellcheck disable=SC1091 +. "$(dirname "$0")/../.ci/scripts/common.sh" + set -e set -x @@ -45,6 +49,9 @@ export NIXL_PLUGIN_DIR=${INSTALL_DIR}/lib/$ARCH-linux-gnu/plugins export NIXL_PREFIX=${INSTALL_DIR} export NIXL_NO_STUBS_FALLBACK=1 +start_etcd_server "/nixl/rust_ci" +trap 'kill -9 $ETCD_PID 2>/dev/null || true' EXIT + cargo test -- --test-threads=1 # test that stubs and real wrapper defined APIs / symbols match diff --git a/src/bindings/rust/src/agent.rs b/src/bindings/rust/src/agent.rs index 68dad55cc0..51b8be8480 100644 --- a/src/bindings/rust/src/agent.rs +++ b/src/bindings/rust/src/agent.rs @@ -1,4 +1,4 @@ -// SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -697,7 +697,7 @@ impl Agent { tracing::trace!(remote_agent = %remote_name, "Fetching remote metadata from etcd"); let c_remote_name = CString::new(remote_name)?; - let inner_guard = self.inner.write().unwrap(); + let mut inner_guard = self.inner.write().unwrap(); let status = unsafe { bindings::nixl_capi_fetch_remote_md( @@ -709,9 +709,7 @@ impl Agent { match status { NIXL_CAPI_SUCCESS => { - self.inner - .write() - .unwrap() + inner_guard .remotes .insert(remote_name.to_string()); tracing::trace!(remote_agent = %remote_name, "Successfully fetched remote metadata from etcd"); @@ -1111,7 +1109,10 @@ impl AgentInner { fn invalidate_remote_md(&mut self, remote_agent: &str) -> Result<(), NixlError> { unsafe { if self.remotes.remove(remote_agent) { - nixl_capi_invalidate_remote_md(self.handle.as_ptr(), remote_agent.as_ptr().cast()); + nixl_capi_invalidate_remote_md( + self.handle.as_ptr(), + CString::new(remote_agent)?.as_ptr().cast(), + ); } else { return Err(NixlError::InvalidParam); } @@ -1122,7 +1123,10 @@ impl AgentInner { fn invalidate_all_remotes(&mut self) -> Result<(), NixlError> { unsafe { for remote in self.remotes.drain() { - nixl_capi_invalidate_remote_md(self.handle.as_ptr(), remote.as_ptr().cast()); + nixl_capi_invalidate_remote_md( + self.handle.as_ptr(), + CString::new(remote.as_str())?.as_ptr().cast(), + ); } } Ok(()) @@ -1136,7 +1140,17 @@ impl Drop for AgentInner { // invalidate all remotes for remote in self.remotes.iter() { tracing::trace!(remote.agent = %remote, "Invalidating remote agent"); - nixl_capi_invalidate_remote_md(self.handle.as_ptr(), remote.as_ptr().cast()); + + let c_remote = match CString::new(remote.as_str()) { + Ok(s) => s, + Err(e) => { + tracing::warn!(remote.agent = %remote, error = ?e, + "Skipping remote invalidation: remote name contains interior NULL"); + continue; + } + }; + + nixl_capi_invalidate_remote_md(self.handle.as_ptr(), c_remote.as_ptr().cast()); } // destroy all backends diff --git a/src/bindings/rust/tests/tests.rs b/src/bindings/rust/tests/tests.rs index c0cac1ba11..e530233b40 100644 --- a/src/bindings/rust/tests/tests.rs +++ b/src/bindings/rust/tests/tests.rs @@ -1,4 +1,4 @@ -// SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 // // Licensed under the Apache License, Version 2.0 (the "License"); @@ -20,7 +20,8 @@ //! `nixl` crate. use nixl_sys::*; -use std::env; +use std::time::{Duration, Instant}; +use std::{env, thread}; use std::collections::HashMap; // Helper function to create an agent with error handling fn create_test_agent(name: &str) -> Result { @@ -701,24 +702,58 @@ fn test_etcd_metadata_exchange() -> Result<(), NixlError> { let plugins = agent1.get_available_plugins()?; let plugin_name = find_plugin(&plugins, "UCX")?; let (_mems, params) = agent1.get_plugin_params(&plugin_name)?; - let backend = agent1.create_backend(&plugin_name, ¶ms)?; + let backend1 = agent1.create_backend(&plugin_name, ¶ms)?; + + // Backend on agent2 too + let (_mems2, params2) = agent2.get_plugin_params(&plugin_name)?; + let backend2 = agent2.create_backend(&plugin_name, ¶ms2)?; + // Create OptArgs with backend let mut opt_args = OptArgs::new()?; - opt_args.add_backend(&backend)?; + opt_args.add_backend(&backend1)?; + let mut opt_args2 = OptArgs::new()?; + opt_args2.add_backend(&backend2)?; // Send agent1's metadata to etcd agent1.send_local_md(Some(&opt_args))?; println!("Successfully sent agent1 metadata to etcd"); // Fetch agent1's metadata from etcd with agent2 - agent2.fetch_remote_md("EtcdAgent1", Some(&opt_args))?; + agent2.fetch_remote_md("EtcdAgent1", Some(&opt_args2))?; println!("Successfully fetched agent1 metadata from etcd"); + let poll_timeout = Duration::from_secs(10); + let poll_interval = Duration::from_millis(100); + let metadata_visible_start = Instant::now(); + loop { + if agent2.check_remote_metadata("EtcdAgent1", None) { + break; + } + if metadata_visible_start.elapsed() >= poll_timeout { + panic!("Timed out waiting for EtcdAgent1 metadata to become visible"); + } + thread::sleep(poll_interval); + } + // Invalidate agent1's metadata in etcd agent1.invalidate_local_md(Some(&opt_args))?; println!("Successfully invalidated agent1 metadata in etcd"); + let local_invalidate_start = Instant::now(); + loop { + if !agent2.check_remote_metadata("EtcdAgent1", None) { + break; + } + if local_invalidate_start.elapsed() >= poll_timeout { + panic!("Timed out waiting for EtcdAgent1 metadata invalidation in etcd"); + } + thread::sleep(poll_interval); + } + + agent2.invalidate_remote_md("EtcdAgent1")?; + println!("Successfully invalidated agent2 cached remote metadata"); + Ok(()) }