fix macos tests

Author: kzhdev

slick_queue_py.py

@@ -311,7 +311,7 @@ def __getitem__(self, index: int) -> memoryview:
         off = self._data_offset + (index & self.mask) * self.element_size
         return self._buf[off: off + self.element_size]
 
-    def read(self, read_index: Union[int, AtomicCursor]) -> Union[Tuple[Optional[bytes], int, int], Tuple[Optional[bytes], int]]:
+    def read(self, read_index: Union[int, AtomicCursor]) -> Union[Tuple[Optional[bytes], int, int], Tuple[Optional[bytes], int, int]]:
         """
         Read data from the queue.
 
@@ -345,7 +345,7 @@ def read(self, read_index: Union[int, AtomicCursor]) -> Union[Tuple[Optional[byt
 
             # Multi-consumer work-stealing
             cursor = AtomicCursor(cursor_shm.buf, 0)
-            data, size = q.read(cursor)  # Atomically claim next item
+            data, size, index = q.read(cursor)  # Atomically claim next item
         """
         if isinstance(read_index, AtomicCursor):
             return self._read_atomic_cursor(read_index)
@@ -396,7 +396,7 @@ def _read_single_consumer(self, read_index: int) -> Tuple[Optional[bytes], int,
             new_read_index = data_index + slot_size
             return data, slot_size, new_read_index
 
-    def _read_atomic_cursor(self, read_index: AtomicCursor) -> Tuple[Optional[bytes], int]:
+    def _read_atomic_cursor(self, read_index: AtomicCursor) -> Tuple[Optional[bytes], int, int]:
         """
         Multi-consumer read using a shared atomic cursor (work-stealing pattern).
 
@@ -407,8 +407,8 @@ def _read_atomic_cursor(self, read_index: AtomicCursor) -> Tuple[Optional[bytes]
             read_index: Shared AtomicCursor for coordinating multiple consumers
 
         Returns:
-            Tuple of (data_bytes or None, item_size).
-            If no data available returns (None, 0).
+            Tuple of (data_bytes or None, item_size, data_index).
+            If no data available returns (None, 0, -1).
         """
         if self._buf is None:
             raise RuntimeError("Queue buffer is not initialized")
@@ -433,7 +433,7 @@ def _read_atomic_cursor(self, read_index: AtomicCursor) -> Tuple[Optional[bytes]
 
             # Check if data is ready (C++ lines 296-299)
             if data_index == (2**64 - 1) or data_index < current_index:
-                return None, 0
+                return None, 0, -1
 
             # Check for wrap (C++ lines 300-304)
             if data_index > current_index and ((data_index & self.mask) != idx):
@@ -449,7 +449,7 @@ def _read_atomic_cursor(self, read_index: AtomicCursor) -> Tuple[Optional[bytes]
                 # Successfully claimed the item, read and return it
                 data_off = self._data_offset + (current_index & self.mask) * self.element_size
                 data = bytes(self._buf[data_off: data_off + slot_size * self.element_size])
-                return data, slot_size
+                return data, slot_size, current_index
 
             # CAS failed, another consumer claimed it, retry
 

tests/test_atomic_cursor.py

@@ -59,7 +59,7 @@ def test_atomic_cursor_local_mode_basic():
     # Consume with atomic cursor
     consumed = []
     for _ in range(10):
-        data, size = q.read(cursor)
+        data, size,  _ = q.read(cursor)
         if data is not None:
             value = struct.unpack("<I", data[:4])[0]
             consumed.append(value)
@@ -68,7 +68,7 @@ def test_atomic_cursor_local_mode_basic():
     assert consumed == list(range(10)), f"Expected {list(range(10))}, got {consumed}"
 
     # Verify no more data
-    data, size = q.read(cursor)
+    data, size, _ = q.read(cursor)
     assert data is None, "Expected no more data"
 
     q.close()
@@ -104,7 +104,7 @@ def consumer_worker(worker_id, results):
         max_consecutive_none = 20  # Give more chances
         time.sleep(random.uniform(0.03, 0.05))
         while len(consumed) < num_items and consecutive_none < max_consecutive_none:
-            data, _ = q.read(cursor)
+            data, size, data_index = q.read(cursor)
             if data is None:
                 consecutive_none += 1
             else:
@@ -173,15 +173,15 @@ def consumer_worker(worker_id, results):
         consumed = []
         no_data_count = 0
         while len(consumed) < num_items // num_threads + 50:  # Over-subscribe
-            data, size = q.read(cursor)
+            data, size, _ = q.read(cursor)
             if data is not None:
                 no_data_count = 0
                 value = struct.unpack("<I", data[:4])[0]
                 consumed.append(value)
                 time.sleep(random.uniform(0.0003, 0.0005))
             else:
                 no_data_count += 1
-                if no_data_count > 1000:
+                if no_data_count > 10000:
                     break
                 time.sleep(0.00001)  # Wait for producer
         results[worker_id] = consumed
@@ -234,19 +234,21 @@ def consumer_process_worker(queue_name, cursor_name, worker_id, num_items, resul
     cursor = AtomicCursor(cursor_shm.buf, 0)
 
     consumed = []
-    num_no_data = 0
 
-    with open('ready', 'a'):
-        # Keep consuming until we've collectively consumed all items
-        while cursor.load() < num_items:
-            data, size = q.read(cursor)
-            if data is not None:
-                num_no_data = 0
-                value = struct.unpack("<I", data[:4])[0]
-                consumed.append(value)
-                time.sleep(random.uniform(0.003, 0.005))
-            else:
-                time.sleep(0.000001)
+    # Signal that consumer is ready (create/touch the file, then close it immediately)
+    with open('ready', 'a') as f:
+        pass  # Just create/touch the file
+
+    # Keep consuming until we've collectively consumed all items
+    while cursor.load() < num_items:
+        data, size, data_index = q.read(cursor)
+        if data is not None:
+            value = struct.unpack("<I", data[:4])[0]
+            print(f"Process {worker_id} consumed an item {value} @ index {data_index}")
+            consumed.append(value)
+            time.sleep(random.uniform(0.0003, 0.0005))
+        else:
+            time.sleep(0.000001)
     results.put((worker_id, consumed))
 
     cursor_shm.close()
@@ -267,6 +269,7 @@ def producer_worker(num_items: int, q: SlickQueue):
         data = struct.pack("<I", i)
         q[idx][:len(data)] = data
         q.publish(idx)
+        print(f"Produced item {idx} {i}")
         time.sleep(random.uniform(0.001, 0.003))  # Slow producer
 
 def test_atomic_cursor_shared_memory_mode():
@@ -394,7 +397,7 @@ def test_atomic_cursor_compare_with_int_cursor():
 
     consumed_atomic = []
     for _ in range(num_items):
-        data, size = q2.read(cursor)
+        data, size, _ = q2.read(cursor)
         if data is not None:
             value = struct.unpack("<I", data[:4])[0]
             consumed_atomic.append(value)
@@ -436,7 +439,7 @@ def test_atomic_cursor_wraparound():
         q.publish(idx, sz)
 
         # Consume immediately
-        data, size = q.read(cursor)
+        data, size, _ = q.read(cursor)
         if data is not None:
             offset = 0
             while size > 0:
@@ -538,93 +541,6 @@ def test_atomic_cursor_python_cpp_work_stealing_cursor_created_by_py():
     print(f"[PASS] C++ work-stealing consumer cursor created by Python test passed: {num_items} items consumed")
 
 
-def test_atomic_cursor_python_cpp_work_stealing_cursor_created_by_py():
-    """Test AtomicCursor with a C++ work-stealing consumer, cursor created by Python."""
-    print("\n=== Test: AtomicCursor with C++ Work-Stealing Consumer, cursor created by Python ===")
-
-    queue_name = 'test_atomic_cursor_cpp_queue'
-    cursor_name = 'test_cur_cpp_py_created'
-    num_items = 100
-
-    try:
-        os.remove('ready')
-    except Exception:
-        pass
-
-    # Create shared queue and cursor
-    q = SlickQueue(name=queue_name, size=128, element_size=32)
-    cursor_shm = SharedMemory(name=cursor_name, create=True, size=8)
-    cursor = AtomicCursor(cursor_shm.buf, 0)
-    cursor.store(0)
-
-    actual_name = q.get_shm_name()
-    actual_cursor_name = cursor_shm._name
-
-    # Start C++ consumer process
-    cpp_consumer = find_cpp_executable("cpp_work_stealing_consumer")
-    output_file = Path(__file__).parent / "cpp_work_stealing_consumer_output.txt"
-    cpp_consumer_proc = subprocess.Popen([
-        cpp_consumer,
-        actual_name,
-        str(num_items),
-        str(32),  # element size
-        actual_cursor_name,
-        str(output_file)
-    ], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-
-    # Start consumer processes
-    results = MPQueue()
-    py_consumer_proc = Process(
-        target=consumer_process_worker,
-        args=(queue_name, cursor_name, 0, num_items, results)
-    )
-    py_consumer_proc.start()
-
-    # Start producer
-    producer = Thread(target=producer_worker, args=(num_items, q))
-    producer.start()
-
-    producer.join()
-    py_consumer_proc.join()
-
-    stdout, stderr = cpp_consumer_proc.communicate()
-
-    print(stdout.decode())
-    print(stderr.decode())
-    if cpp_consumer_proc.returncode != 0:
-        print(stderr, file=sys.stderr)
-        raise RuntimeError(f"C++ consumer failed with code {cpp_consumer_proc.returncode}")
-    
-    # Verify consumed data
-    consumed = []
-    with open(output_file, 'r') as f:
-        for line in f:
-            item, _ = map(int, line.strip().split())
-            consumed.append(item)
-
-    py_result = results.get(timeout=5)
-    py_consumed = py_result[1]
-    print("Python consumer consumed:", len(py_consumed))
-    consumed.extend(item for item in py_consumed)
-
-    # Check all items consumed
-    assert len(consumed) == num_items, f"Expected {num_items} items, got {len(consumed)}"
-
-    # Check data integrity
-    expected = set(range(num_items))
-    actual = set(consumed)
-    assert expected == actual, f"Data mismatch between produced and consumed"
-
-    # Cleanup
-    cursor_shm.close()
-    cursor_shm.unlink()
-    q.close()
-    q.unlink()
-    if output_file.exists():
-        output_file.unlink()
-
-    print(f"[PASS] C++ work-stealing consumer cursor created by Python test passed: {num_items} items consumed")
-
 def test_atomic_cursor_python_cpp_work_stealing_cursor_created_by_cpp():
     """Test AtomicCursor with a C++ work-stealing consumer, cursor created by C++."""
     print("\n=== Test: AtomicCursor with C++ Work-Stealing Consumer, cursor created by C++ ===")
@@ -641,15 +557,20 @@ def test_atomic_cursor_python_cpp_work_stealing_cursor_created_by_cpp():
     # Create shared queue and cursor
     q = SlickQueue(name=queue_name, size=128, element_size=32)
 
+    actual_queue_name = q.get_shm_name()
+
+    from slick_queue_py.atomic_ops import validate_shm_name
+    actual_cursor_name = validate_shm_name(cursor_name)  # Created by C++ process
+
     # Start C++ consumer process
     cpp_consumer = find_cpp_executable("cpp_work_stealing_consumer")
     output_file = Path(__file__).parent / "cpp_work_stealing_consumer_output.txt"
     cpp_consumer_proc = subprocess.Popen([
         cpp_consumer,
-        queue_name,
+        actual_queue_name,
         str(num_items),
         str(32),  # element size
-        cursor_name,
+        actual_cursor_name,
         str(output_file)
     ], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 

tests/test_interop.py

@@ -70,34 +70,37 @@ def python_producer(queue_name, num_items, worker_id, results_queue, size=64):
 def python_consumer(queue_name, num_items, results_queue, starting_index):
     """Python consumer process."""
     try:
-        with open('ready', 'a'):
-            # On Windows, we need to specify size when opening existing shared memory
-            # q = SlickQueue(name=queue_name, size=size, element_size=32)
-            q = SlickQueue(name=queue_name, element_size=32)
-            consumed = []
-            # read_index = 0
-            read_index = starting_index
-            attempts = 0
-            max_attempts = 10000
-
-            while len(consumed) < num_items and attempts < max_attempts:
-                attempts += 1
-                prev_read_index = read_index
-                data, size, read_index = q.read(read_index)
-
-                if data is not None:
-                    worker_id, item_num = struct.unpack("<I I", data[:8])
-                    consumed.append((worker_id, item_num))
-                    # print(f"consume {worker_id} {item_num} (from index {prev_read_index}, consumed: {len(consumed)})", flush=True)
-                else:
-                    time.sleep(0.001)
-
-            # print(f"CONSUMER: About to put {len(consumed)} items into results queue", flush=True)
-            results_queue.put(('success', consumed))
-            # print(f"CONSUMER: Successfully put results into queue", flush=True)
-            q.close()
-            time.sleep(1)
-            # print(f"CONSUMER: Exiting normally", flush=True)
+        # Signal that consumer is ready (create/touch the file, then close it immediately)
+        with open('ready', 'a') as f:
+            pass  # Just create/touch the file
+
+        # On Windows, we need to specify size when opening existing shared memory
+        # q = SlickQueue(name=queue_name, size=size, element_size=32)
+        q = SlickQueue(name=queue_name, element_size=32)
+        consumed = []
+        # read_index = 0
+        read_index = starting_index
+        attempts = 0
+        max_attempts = 10000
+
+        while len(consumed) < num_items and attempts < max_attempts:
+            attempts += 1
+            prev_read_index = read_index
+            data, size, read_index = q.read(read_index)
+
+            if data is not None:
+                worker_id, item_num = struct.unpack("<I I", data[:8])
+                consumed.append((worker_id, item_num))
+                # print(f"consume {worker_id} {item_num} (from index {prev_read_index}, consumed: {len(consumed)})", flush=True)
+            else:
+                time.sleep(0.001)
+
+        # print(f"CONSUMER: About to put {len(consumed)} items into results queue", flush=True)
+        results_queue.put(('success', consumed))
+        # print(f"CONSUMER: Successfully put results into queue", flush=True)
+        q.close()
+        time.sleep(1)
+        # print(f"CONSUMER: Exiting normally", flush=True)
     except Exception as e:
         print(f"CONSUMER: Exception - {str(e)}", flush=True)
         results_queue.put(('error', str(e)))