Can run fast stream multi thread?

[alm0ra]

hey guys
is fast stream thread safe?
how can I run it multithreaded not with --wrokers=x

[alm0ra]

i tryed with this code :

import asyncio
import concurrent.futures

from faststream import FastStream
from faststream.kafka import KafkaBroker

from app.routers.base_consumer import router

broker = KafkaBroker("kafka:9092")
broker.include_router(router)

app = FastStream(broker)


def runner(a):
    print(a)
    print("start in new thread")
    asyncio.run(app.run())


def main():
    # loop = asyncio.get_running_loop()
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        tasks = [pool.submit(runner, i) for i in range(2)]

        for task in concurrent.futures.as_completed(tasks):
            print("complete, result:", task.result())
    print("finish")


if __name__ == "__main__":
    main()

and i am getting

0
start in new thread
1
start in new thread
2024-02-10 17:56:21,115 INFO     - FastStream app starting...
2024-02-10 17:56:21,115 INFO     - FastStream app starting...
Unclosed AIOKafkaProducer
producer: <aiokafka.producer.producer.AIOKafkaProducer object at 0x7fc4e041fbb0>
Unclosed AIOKafkaProducer
producer: <aiokafka.producer.producer.AIOKafkaProducer object at 0x7fc4e046d840>
Traceback (most recent call last):
  File "/home/ali/websocket-sever/app/main.py", line 32, in <module>
    main()
  File "/home/ali/websocket-sever/app/main.py", line 27, in main
    print("complete, result:", task.result())
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 451, in result
    return self.__get_result()
  File "/usr/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/usr/lib/python3.10/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
  File "/home/ali/websocket-sever/app/main.py", line 18, in runner
    asyncio.run(app.run())
  File "/usr/lib/python3.10/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/home/ali/websocket-sever/venv/lib/python3.10/site-packages/faststream/app.py", line 344, in run
    raise ex from None
  File "/home/ali/websocket-sever/venv/lib/python3.10/site-packages/faststream/app.py", line 340, in run
    await self._stop(log_level)
  File "/home/ali/websocket-sever/venv/lib/python3.10/site-packages/faststream/app.py", line 376, in _stop
    with anyio.open_signal_receiver(*HANDLED_SIGNALS) as signals:
  File "/home/ali/websocket-sever/venv/lib/python3.10/site-packages/anyio/_backends/_asyncio.py", line 1784, in __enter__
    self._loop.add_signal_handler(sig, self._deliver, sig)
  File "/usr/lib/python3.10/asyncio/unix_events.py", line 107, in add_signal_handler
    raise RuntimeError(str(exc))
RuntimeError: set_wakeup_fd only works in main thread of the main interpreter

Process finished with exit code 1

[alm0ra]

@Lancetnik

[Lancetnik]

Is it important for u to use threads? I am not sure about it possibility due thread safety of basic classes. But, multiprocessing works fine (our —workers implementation at least)

[alm0ra]

@Lancetnik i was looking for optimizing ram usage with multiThreading . i think running in threads is so much optimizer that running in subprocess

[Lancetnik]

Python is not thread safety language, also the problem is in underlying libraries. Unfortunately, we can't guarantee thread safety running and I can't help you with such decision, sorry

I strongly recommend you to use processes. It shouldn't be much expensive than processes and this decision already works. Btw, one instance scaling with multiprocessing/threading is a bad decision at all. The preferred way in current infrastructure - container-based horizontal scaling.

[alm0ra]

thanks for your answer.

what do you mean also the problem is in underlying libraries.?? aiokafka or asyncio ??

[Lancetnik]

Aiokafka doesn't allow you to setup multiple connections in different threads