robin-coac
commented
1 year ago Further insights into the problem, hacky solution
1. The Problem Scenario
I intend to use nats as celery+rabbitmq replacement. I want to deploy things in kubernetes, where each container will deal with processing a single file that can take upto 20 mins. Thus I kept max_ack_pending to 1. I would rather scale pods horizontally based on message queue size.
2. Setting max_ack_pending > 1 (say 20) in above consumer code, I get a partial parallelism.
Say If I were running consumer.py in 4 separate terminals, two of the process would pick up task. And after they were done, rest of two would pick up task and original two processes just stay idle. And so forth in an alternate fashion. Which I don't think should be the case. By the way, There were 1000 messages in the queue.
3. This is my hacky solution using pull-subsriber approach.
Even here, Noticed a a major issue.
Suppose I have 1000 messages in a freshly created stream/queue.
Then I start multiple consumer processes in 3 terminals. Now If I abruptly close my consumer processes (Ctrl+C), before the process can do acknowledgement to the nats-server, something strange happens. When I restart start my consumer processes now, they cannot fetch any message from the server. I get Timeout Error while fetching.
The only solution that worked was doing await msg.ack_sync() before I start processing my message as shown below. Even using await msg.ack() doesn't work. This is not a good solution because I can still loose my message due to some exception while processing.
import asyncio
import signal
import sys
import logging.config
logging.config.fileConfig("logging.conf", disable_existing_loggers=False)
import nats
from nats.js.api import RetentionPolicy, ConsumerConfig, AckPolicy
consumer_config = ConsumerConfig(
ack_wait=900,
max_deliver=1,
max_ack_pending=1,
ack_policy=AckPolicy.EXPLICIT,
)
# Nats
NATS_HOST = '0.0.0.0'
NATS_PORT = '4222'
DOCUMENT_EXT_SUBJECT = "file_process"
MAX_RECONNECT_ATTEMPTS = 10
## A cpu bound task
def time_consuming_task():
import time; time.sleep(50)
return
async def task_cb(msg):
time_consuming_task()
from nats.errors import TimeoutError
async def run():
logging.info("Started Document processing Consumer ...")
nc = await nats.connect(
servers=f"nats://{NATS_HOST}:{NATS_PORT}",
max_reconnect_attempts=MAX_RECONNECT_ATTEMPTS,
)
# Create JetStream context.
js = nc.jetstream()
## PERSIST ON THIS SUBJECT
await js.add_stream(
subjects=[DOCUMENT_EXT_SUBJECT],
name="sample-stream",
## Extra
retention=RetentionPolicy.WORK_QUEUE,
)
psub = await js.pull_subscribe(
subject=DOCUMENT_EXT_SUBJECT,
stream="sample-stream",
durable="worker",
config=consumer_config,
)
## Keep pulling from Server Every 2 secs.
async def constant_fetch():
while True:
try:
await asyncio.sleep(2)
msgs = await psub.fetch(1, timeout=5)
for msg in msgs:
# This is a life saver. Idk what it does.
await msg.ack_sync()
await document_extraction_cb(msg)
except TimeoutError:
print("fetch timed out . Retrying")
pass
await asyncio.get_event_loop().create_task( constant_fetch() )
def signal_handler():
sys.exit()
for sig in ('SIGINT', 'SIGTERM'):
asyncio.get_running_loop().add_signal_handler(getattr(signal, sig), signal_handler)
logging.info("Done Extracting .. ?")
if __name__ == '__main__':
loop = asyncio.get_event_loop()
try:
loop.create_task(run())
loop.run_forever()
except Exception as e:
print("Got error : ", e)
finally:
loop.close()Extra Notes :
I suspected one reason for this was my file-processing being synchronous. That why I tried doing this. But didn't work as well in both push and pull examples.
async def task_cb(msg):
with concurrent.futures.ProcessPoolExecutor(max_workers=2) as pool:
future = await asyncio.get_running_loop().run_in_executor(
pool, time_consuming_task
)
result = await asyncio.gather(*future)
ripienaar
bruth
Problem
I am trying to use nats for job queuing (producer-consumer problem) where a time consuming task is distributed randomly across multiple consumer.
I have two scripts.
producer.pyNATS_HOST = '0.0.0.0' NATS_PORT = '4222' DOCUMENT_EXT_SUBJECT = "file_process"
nc = await nats.connect(servers=f"nats://{NATS_HOST}:{NATS_PORT}") js = nc.jetstream() await js.add_stream( name="sample-stream", subjects=[DOCUMENT_EXT_SUBJECT], retention=RetentionPolicy.WORK_QUEUE, )
for i in range(20): ack = await js.publish( subject=DOCUMENT_EXT_SUBJECT, payload=json.dumps( { "some_load" : "lsome_load", }).encode() )
import asyncio import signal import sys import logging
import nats from nats.js.api import RetentionPolicy, ConsumerConfig, AckPolicy
consumer_config = ConsumerConfig( ack_wait=900, max_deliver=1, max_ack_pending=1, ack_policy=AckPolicy.EXPLICIT )
Nats
NATS_HOST = '0.0.0.0' NATS_PORT = '4222' DOCUMENT_EXT_SUBJECT = "file_process"
MAX_RECONNECT_ATTEMPTS = 10
A cpu bound task
def time_consuming_task(): import time; time.sleep(50) return
async def task_cb(msg): time_consuming_task() await msg.ack() print("acknowledged document-extraction !")
async def run():
if name == 'main': loop = asyncio.get_event_loop() try: loop.run_until_complete(run()) loop.run_forever() except Exception as e: print("Got error : ", e) finally: loop.close()