etha.tensor_bus.command_queue
=============================

.. py:module:: etha.tensor_bus.command_queue

.. autoapi-nested-parse::

   LMDB-based Command Queue for Tensor Bus.



Attributes
----------

.. autoapisummary::

   etha.tensor_bus.command_queue.logger


Exceptions
----------

.. autoapisummary::

   etha.tensor_bus.command_queue.QueueFullError


Classes
-------

.. autoapisummary::

   etha.tensor_bus.command_queue.CommandQueue


Module Contents
---------------

.. py:exception:: QueueFullError

   Bases: :py:obj:`Exception`


   Raised when trying to enqueue to a full CommandQueue with block=False.

   Initialize self.  See help(type(self)) for accurate signature.


.. py:class:: CommandQueue(lmdb_path: str = '/tmp/tensor_bus.lmdb', capacity: int = 50000)

   LMDB based command queue.


   .. py:method:: clear(delete_commands: bool = True)

      Clear queue.



   .. py:method:: close(destroy: bool = True)

      Close queue connection.

      :param destroy: If True (default), completely remove all resources (LMDB files, semaphores).
                      If False, only close handles in this process (safe for multi-process).

      Default behavior (destroy=True):
          - Most common case: single process or final cleanup
          - Unlinks semaphores from the system
          - Deletes LMDB database files
          - Ensures no resource leaks

      Multi-process mode (destroy=False):
          - Use when other processes are still using the same queue
          - Only closes handles in this process
          - Resources remain available for other processes
          - Last process should call close() or close(destroy=True) to clean up

      .. admonition:: Examples

         # Single process (default)
         q = CommandQueue('/tmp/test.lmdb')
         q.enqueue(msg)
         q.close()  # Automatically cleans up everything
         
         # Multi-process
         # Worker process
         q = CommandQueue('/tmp/bus.lmdb')
         q.enqueue(msg)
         q.close(destroy=False)  # Don't delete, other processes need it
         
         # Main process (last one)
         q = CommandQueue('/tmp/bus.lmdb')
         q.close()  # Final cleanup, removes all resources



   .. py:method:: dequeue(*, block: bool = False, timeout: float | None = None) -> etha.tensor_bus.commands.Message | None

      Dequeue a message.

      :returns: Message object (auto-detected type), or None if queue is empty



   .. py:method:: dequeue_batch(max_count: int = 32) -> list[etha.tensor_bus.commands.Message]

      Batch dequeue (improve throughput).

      :param max_count: Maximum number of messages to dequeue

      :returns: List of messages



   .. py:method:: enqueue(msg: etha.tensor_bus.commands.Message, *, block: bool = True, timeout: float | None = None) -> int

      Enqueue a message.

      :param msg: Any Message type (PutCommand, GetCommand, etc.)
      :param block: If True, block when queue is full. If False, raise QueueFullError.
      :param timeout: Timeout in seconds for blocking wait (None = infinite)

      :returns: Message ID (queue position)

      :raises QueueFullError: If queue is full and block=False, or timeout expires



   .. py:method:: is_empty() -> bool

      Check if queue is empty.



   .. py:method:: peek() -> etha.tensor_bus.commands.Message | None

      View front message (without dequeuing).



   .. py:method:: size() -> int

      Return queue length.



   .. py:attribute:: capacity
      :value: 50000



   .. py:attribute:: lmdb_path


   .. py:attribute:: ready_name


   .. py:attribute:: sem_name


   .. py:attribute:: space_sem_name


.. py:data:: logger