Metadata-Version: 2.4
Name: iap_messenger
Version: 2.1.1
Classifier: Programming Language :: Python :: 3
Classifier: Operating System :: OS Independent
Requires-Dist: nats-py~=2.12.0
Requires-Dist: pillow>=10.4,<13.0
Requires-Dist: pyyaml~=6.0.2
Requires-Dist: yamlcore~=0.0.4
Requires-Dist: json-tricks~=3.17.3
Requires-Dist: msgpack~=1.1.0
Requires-Dist: msgpack-numpy~=0.4.8
Requires-Dist: urllib3~=2.5.0
Requires-Dist: lz4~=4.4.4
Requires-Dist: pyzmq>=26.4,<28.0
License-File: LICENSE
Summary: Messaging service package for IA PARC
Author: Bertrand Louargant
License-Expression: Apache-2.0
Requires-Python: >=3.10
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Homepage, https://gitlab.com/ia-parc/public/pypi/iap_messenger
Project-URL: Issues, https://gitlab.com/ia-parc/public/pypi/iap_messenger/issues

# iap_messenger


[![PyPI version](https://badge.fury.io/py/iaparc-inference.svg)](https://badge.fury.io/py/iaparc-inference)
![PyPI - License](https://img.shields.io/pypi/l/iaparc-inference)


The IA Parc inference plugin allows developers to easily integrate their inference pipeline into IA Parc's production module.   


## Installation

```bash
pip install iaparc-inference
```

## Usage

* If your inference pipeline support batching:
  
    ```python
    from iap_messenger import MsgListener, Message

    # Define a callback to query your inference pipeline
    # To load your model only once it is recommended to use a class:
    class MyModel:
        def __init__(self, model_path: str):
            ## Load your model in pytorch, tensorflow or any other backend
        
        def batch_query(msgs: list[Message]) -> list[Message]:
            ''' execute your pipeline on a batch input
                Note:   "parameters" is an optional argument.
                        It can be used to handle URL's query parameters
                        It's a list of key(string)/value(string) dictionaries
            '''

    if __name__ == '__main__':
        # Initiate your model class
        my_model = MyModel("path/to/my/model")

        # Initiate IAParc listener
        listener = MsgListener(my_model.batch_query)
        # Start the listener
        listener.run()

    ```
* If your inference pipeline **do not** support batching:
  
    ```python
    from iap_messenger import MsgListener, Message

    # Define a callback to query your inference pipeline
    # To load your model only once it is recommended to use a class:
    class MyModel:
        def __init__(self, model_path: str):
            ## Load your model in pytorch, tensorflow or any other backend
        
        def single_query(msg: Message) -> Message:
            ''' execute your pipeline on a single input
                Note:   "parameters" is an optional argument.
                        It can be used to handle URL's query parameters
                        It's a key(string)/value(string) dictionary
            '''

    if __name__ == '__main__':
        # Initiate your model class
        my_model = MyModel("path/to/my/model")

        # Initiate IAParc listener
        listener = MsgListener(my_model.single_query, batch=1)  # Note that batch size is forced to 1 here
        # Start the listener
        listener.run()

## Message consumption and replicas

Each input link is served by one durable JetStream consumer named
`{queue}-{link}`, shared by every replica of the worker. Since the
`IAP_CONSUMER_MODE` release the consumer is a **pull** consumer: a replica only
fetches a message (or a batch) when it is free, so nothing ever waits in the
buffer of a busy or dead replica while another one is idle, and a crash only
leaves the batch in flight to be redelivered after `ack_wait`. Older versions
used a push consumer with a queue group, where the server spreads messages
across replicas at random and up to `max_ack_pending` messages could sit in a
dead pod until `ack_wait` expired.

| Variable | Default | Role |
|---|---|---|
| `IAP_CONSUMER_MODE` | `auto` | `auto` uses pull unless the existing consumer is a push consumer that still has subscribers bound (workers of a push-only version), in which case it joins it in push mode. `pull` forces the migration: the push consumer is deleted and recreated as pull, and bound push subscribers stop receiving messages until restarted. `push` keeps the historical behaviour. The mode chosen for each link is logged at startup. |
| `IAP_ACK_WAIT_SECONDS` | `300` | JetStream `ack_wait`. Long callbacks are kept alive with `in_progress` heartbeats, so this only bounds two things: how long a NATS outage may last before in-flight work is redelivered, and how long a message held by a crashed replica waits before another one gets it. |
| `IAP_PULL_IDLE_TIMEOUT_SECONDS` | `30` | Longest an idle pull fetch blocks. After each idle timeout the worker checks that its consumer still exists and rebuilds the subscription if the API recreated the stream. |
| `IAP_SHUTDOWN_DDL_GRACE_SECONDS` | `30` | On shutdown, how long to keep serving large replies from the local DDL store so the next component can still fetch them. |

### Upgrading a pipeline from push to pull

The stream is a `WorkQueue`, which refuses two consumers on the same subjects,
so a pipeline is either all push or all pull for a given link. `auto` makes a
rolling update safe: new pods join the push consumer while old pods are bound
to it, nothing is delivered twice and no old pod breaks. Once every pod of the
queue runs a pull-capable version, the migration happens on its own the next
time no pod is bound (scale to zero, NATS restart), or immediately with a
`rollout restart` after setting `IAP_CONSUMER_MODE=pull`. Note that after a
NATS restart, a pod of the old version reconnecting to a consumer that has
just been migrated fails to subscribe and exits; the rollout replaces it.

### Graceful shutdown

On `SIGTERM` (or `SIGINT`) the listener stops fetching, lets the task or batch
in flight complete and acks it, hands back with a NAK any message already
pushed to the process but not started (push mode), waits up to
`IAP_SHUTDOWN_DDL_GRACE_SECONDS` for large replies to be fetched downstream,
then exits with code 0. Set the pod's `terminationGracePeriodSeconds` above the
longest inference plus that grace, otherwise Kubernetes kills the pod mid-task
and the message is redelivered after `ack_wait`. A second signal exits at once.

## Large payloads (DDL)

Payloads above 8 MB never travel through NATS: the sender keeps them in its
local DDL store and the message only carries a `uuid@host:port` pointer that the
recipient resolves over ZMQ. Environment variables:

| Variable | Role |
|---|---|
| `POD_IP` | Enables DDL (the address advertised to recipients). Without it, payloads fall back to the NATS object store. |
| `ZMQ_PORT` | DDL server port (default `5555`). |
| `DDL_STORAGE` | Directory for the blob store. Strongly recommended in production: blobs then never transit through the `multiprocessing.Manager` pipe. Mount it on a disk-backed `emptyDir` — a `medium: Memory` volume is counted against the container's memory limit. |

The reply format is negotiated per request: a client that can read a two-frame
reply (`[status]`, `[payload]`) asks with the `MP:` prefix and the server then
hands the stored bytes to ZMQ without copying them; any older peer keeps
getting the historical single frame. No configuration and no deployment order
are required — mixed versions of workers and `inferences-api` interoperate.

## Features
* Dynamic batching
* Autoscalling 
* Support both synchronous and asynchronous queries
* Data agnostic


## License
This project is licensed under the Apache License Version 2.0  - see the Apache [LICENSE](https://www.apache.org/licenses/LICENSE-2.0) file for details.
