PYTHON-PROJECT-WIDE: replaced prints with a logger, fixed backtraces

.ci/scripts/check_prints.sh

@@ -0,0 +1,60 @@
+#!/bin/bash
+# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Check if a path is provided as an argument
+if [ -z "$1" ]; then
+  echo "Usage: $0 <path_to_python_directory>"
+  echo "Example: $0 ./path/to/python/directory"
+  exit 1
+fi
+
+DIR_PATH="$1"
+
+# Validate that the provided path is a directory
+if [ ! -d "$DIR_PATH" ]; then
+  echo "Error: The provided path '$DIR_PATH' is not a valid directory."
+  exit 1
+fi
+
+echo "Checking for BUILT-IN 'print()' calls in Python files within: $DIR_PATH"
+echo "---------------------------------------------------------------------"
+
+found_print=false
+
+# Find all Python files and process them
+while read -r py_file; do
+  # Use grep to find 'print()' calls with line numbers, then filter out method calls.
+  # First grep: finds all occurrences of 'print(' with word boundary.
+  # Second grep: filters out lines where 'print(' is preceded by a dot and optional whitespace.
+  MATCHES=$(grep -nE '\bprint\s*\(' "$py_file" | grep -vE '\.[[:space:]]*print\s*\(')
+
+  if [ -n "$MATCHES" ]; then
+    echo "Found built-in 'print()' in: $py_file"
+    echo "${MATCHES//$'\n'/$'\n' Line }" # Indent and prepend "Line "
+    echo # Add a blank line for readability
+    found_print=true
+  fi
+done < <(find "$DIR_PATH" -name "*.py")
+
+echo "---------------------------------------------------------------------"
+
+if [ "$found_print" = true ]; then
+  echo "One or more Python files in '$DIR_PATH' contain built-in 'print()' calls."
+  exit 1
+else
+  echo "No built-in 'print()' calls found in any Python files within '$DIR_PATH'."
+  exit 0
+fi

.github/workflows/python-checks.yml

@@ -0,0 +1,19 @@
+name: Python Checks
+
+on: [pull_request]
+
+jobs:
+  python-checks:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        path:
+          - ./src
+          - ./test
+          - ./benchmark
+          - ./examples/python
+    steps:
+    - uses: actions/checkout@v3
+    - name: Check for print() calls in ${{ matrix.path }}
+      run: |
+        ./.ci/scripts/check_prints.sh ${{ matrix.path }}

.gitlab/test_python.sh

@@ -44,6 +44,8 @@ export CPATH=${INSTALL_DIR}/include:$CPATH
 export PATH=${INSTALL_DIR}/bin:$PATH
 export PKG_CONFIG_PATH=${INSTALL_DIR}/lib/pkgconfig:$PKG_CONFIG_PATH
 export NIXL_PLUGIN_DIR=${INSTALL_DIR}/lib/$ARCH-linux-gnu/plugins
+# Raise exceptions for logging errors
+export NIXL_DEBUG_LOGGING=yes
 
 pip3 install --break-system-packages .
 pip3 install --break-system-packages pytest

benchmark/kvbench/commands/nixlbench.py

@@ -333,7 +333,6 @@ def should_include(name, value, include_defaults=False):
             for key, value in params.items():
                 if value is not None:
                     merged_params[key] = value
-            # print(json.dumps(merged_params))
             return merged_params
         else:  # for text format, exclude defaults to keep command concise
             for name, value in params.items():

benchmark/kvbench/models/model_config.py

@@ -20,6 +20,10 @@
 
 import yaml  # type: ignore
 
+from nixl.logging import get_logger
+
+logger = get_logger(__name__)
+
 
 @dataclass
 class StrategyConfig:
@@ -121,7 +125,7 @@ def from_yaml_files(cls, yaml_paths: List[str]) -> "ModelConfig":
                     config_dict = yaml.safe_load(f)
                 config = config.update(config_dict)
             else:
-                print(f"Warning: Config file not found: {path}")
+                logger.warning("Config file not found: %s", path)
 
         return config
 

benchmark/kvbench/runtime/etcd_rt.py

@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import os
 import pickle
 import re
@@ -23,9 +22,11 @@
 
 import etcd3
 
+from nixl.logging import get_logger
+
 from .rt_base import ReduceOp, _RTUtils
 
-log = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
 
 def int_to_bytes(val: int) -> bytes:
@@ -75,15 +76,15 @@ def __init__(
                 f"Invalid etcd endpoint format: {etcd_endpoints}, expected format is [http://]host[:port]"
             )
 
-        log.info(f"ETCD client initialized with host {host} & port {port}")
+        logger.info("ETCD client initialized with host %s & port %d", host, port)
 
         try:
             self.client = etcd3.client(host=host, port=port)
         except Exception as e:
             raise ValueError(f"Failed to initialize ETCD client: {e}")
 
         if self.rank == 0:
-            log.info(f"Wiping ETCD prefix {self.prefix}")
+            logger.info("Wiping ETCD prefix %s", self.prefix)
             self.client.delete_prefix(self.prefix)
 
     def destroy_dist(self):
@@ -124,7 +125,13 @@ def barrier(self, ranks: Optional[List[int]] = None, timeout_sec=600):
             ):
                 if timeout_sec and time.time() - start_time > timeout_sec:
                     raise TimeoutError(
-                        f"[Rank {self.rank}] ROOT - Barrier {key} timed out after {timeout_sec} seconds, current value: {self.client.get(key)}, waiting for val={len(ranks)} (i.e all the ranks have entered the barrier), (ranks: {ranks})"
+                        "[Rank %d] ROOT - Barrier %s timed out after %.3f seconds, current value: %s, waiting for val=%d (i.e all the ranks have entered the barrier), (ranks: %s)",
+                        self.rank,
+                        key,
+                        timeout_sec,
+                        self.client.get(key),
+                        len(ranks),
+                        ranks,
                     )
         else:
             my_index = ranks.index(self.rank)
@@ -207,7 +214,7 @@ def all_reduce(
                 val = self.client.get(f"{self.prefix}/all_reduce/{dest_rank}")[0]
                 vals.append(pickle.loads(val))
 
-            print(vals)
+            logger.debug("All reduce values: %s", vals)
             if op == ReduceOp.SUM:
                 final_val = [sum(col) for col in zip(*vals)]
             elif op == ReduceOp.AVG:
@@ -235,7 +242,7 @@ def _get_group_id(self, ranks: List[int]) -> int:
 
 
 if not os.environ.get("NIXL_ETCD_NAMESPACE"):
-    log.warning(
+    logger.warning(
         "Environment variable NIXL_ETCD_NAMESPACE is not set, using default prefix /nixl/kvbench. "
         "Note that it can lead to conflicts if multiple instances of KVBench are running. "
         "To avoid this, set NIXL_ETCD_NAMESPACE to a unique value for each instance of KVBench. "

benchmark/kvbench/test/custom_traffic_perftest.py

@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import time
 from test.traffic_pattern import TrafficPattern
 from typing import Literal, Optional, Tuple
@@ -24,8 +23,9 @@
 from tabulate import tabulate
 
 from nixl._api import nixl_agent
+from nixl.logging import get_logger
 
-log = logging.getLogger(__name__)
+logger = get_logger(__name__)
 
 
 class NixlHandle:
@@ -62,13 +62,20 @@ def __init__(
         if shards > 1:
             raise ValueError("Sharding is not supported yet")
 
-        log.debug(
-            f"[Rank {dist_rt.get_rank()}] Initializing NixlBuffer with size {size}, device {device}, shards {shards}, fill_value {fill_value}"
+        logger.debug(
+            "[Rank %d] Initializing NixlBuffer with size %d, device %s, shards %d, fill_value %d",
+            dist_rt.get_rank(),
+            size,
+            device,
+            shards,
+            fill_value,
         )
         self.buf = torch.full((size,), fill_value, dtype=dtype, device=device)
 
-        log.debug(
-            f"[Rank {dist_rt.get_rank()}] Registering memory for buffer {self.buf}"
+        logger.debug(
+            "[Rank %d] Registering memory for buffer %s",
+            dist_rt.get_rank(),
+            self.buf,
         )
         self.reg_descs = nixl_agent.get_reg_descs(self.buf)
         assert (
@@ -137,7 +144,7 @@ def _barrier_tp(self, tp: TrafficPattern, senders_only=True):
 
     def _share_md(self) -> None:
         """Share agent metadata between all ranks. (Need to be run after registering buffers)"""
-        log.debug(f"[Rank {self.my_rank}] Sharing MD")
+        logger.debug("[Rank %d] Sharing MD", self.my_rank)
         md = self.nixl_agent.get_agent_metadata()
         mds = dist_rt.allgather_obj(md)
         for other_rank, metadata in enumerate(mds):
@@ -202,7 +209,7 @@ def _prepare_tp(
 
         send_bufs, recv_bufs = self._get_bufs(tp)
 
-        log.debug(f"[Rank {self.my_rank}] Sharing recv buf descs")
+        logger.debug("[Rank %d] Sharing recv buf descs", self.my_rank)
         dst_bufs_descs = self._share_recv_buf_descs(recv_bufs)
 
         handles: list[NixlHandle] = []
@@ -212,8 +219,12 @@ def _prepare_tp(
 
             xfer_desc = self.nixl_agent.get_xfer_descs(buf)
 
-            log.debug(
-                f"[Rank {self.my_rank}] Initializing xfer for {other} - xfer desc: {xfer_desc}, dst buf desc: {dst_bufs_descs[other]}"
+            logger.debug(
+                "[Rank %d] Initializing xfer for %d - xfer desc: %s, dst buf desc: %s",
+                self.my_rank,
+                other,
+                xfer_desc,
+                dst_bufs_descs[other],
             )
             handle = self.nixl_agent.initialize_xfer(
                 "WRITE",
@@ -268,11 +279,11 @@ def _wait(self, handles: list[NixlHandle]):
             handles = pending
 
     def _destroy(self, handles: list[NixlHandle]):
-        log.debug(f"[Rank {self.my_rank}] Releasing XFER handles")
+        logger.debug("[Rank %d] Releasing XFER handles", self.my_rank)
         for handle in handles:
             self.nixl_agent.release_xfer_handle(handle.handle)
 
-        log.debug(f"[Rank {self.my_rank}] Removing remote agents")
+        logger.debug("[Rank %d] Removing remote agents", self.my_rank)
         for other_rank in range(self.world_size):
             if other_rank == self.my_rank:
                 continue
@@ -281,7 +292,7 @@ def _destroy(self, handles: list[NixlHandle]):
         self._destroy_buffers()
 
     def _destroy_buffers(self):
-        log.debug(f"[Rank {self.my_rank}] Destroying buffers")
+        logger.debug("[Rank %d] Destroying buffers", self.my_rank)
         self.send_buf.destroy()
         self.recv_buf.destroy()
 
@@ -301,14 +312,14 @@ def _verify_tp(
         for r, recv_buf in enumerate(recv_bufs):
             if recv_buf is None:
                 if tp.matrix[r][self.my_rank] > 0:
-                    log.error(
+                    logger.error(
                         f"Rank {self.my_rank} expected {tp.matrix[r][self.my_rank]} bytes from rank {r}, but got 0"
                     )
                     raise RuntimeError("Buffer verification failed")
                 continue
 
             if print_recv_buffers:
-                log.info(f"Recv buffer {r}:\n{recv_buf.buf}")
+                logger.info("Recv buffer %d:\n%s", r, recv_buf.buf)
 
             # recv_buf has to be filled with the rank of the sender
             # and its size has to be the same as matrix[r][my_rank]
@@ -337,7 +348,7 @@ def run(
         Returns:
             Total execution time in seconds
         """
-        log.debug(f"[Rank {self.my_rank}] Running CT perftest")
+        logger.debug("[Rank %d] Running CT perftest", self.my_rank)
         self._share_md()
 
         handles, send_bufs, recv_bufs = self._prepare_tp(self.traffic_pattern)
@@ -377,7 +388,10 @@ def run(
                     total_size_gb,
                 ]
             ]
-            print(tabulate(data, headers=headers, floatfmt=".6f"))
+            logger.info(
+                "Performance metrics:\n%s",
+                tabulate(data, headers=headers, floatfmt=".6f"),
+            )
 
         if verify_buffers:
             self._verify_tp(self.traffic_pattern, recv_bufs, print_recv_buffers)