Edit: In lack of an alternative I just multiply the task. I am using Flask as a webserver. As I call the Train_network endpoint, the Task is executed as many times as there are workers.
response = [fw.train_network.delay().get() for _ in range(Workers)]
For this to work, I also remove the -c 2 argument from the celery worker command and placed the amount into my config
celery.conf.worker_concurrency = cfg.celery_workers
Witht his I always know the amount of Subprocesses and how many times the Task should be repeated.
If there is an better option to solve this, I will update the post with an answer. Or maybe somebody else can provide insight.
Edit: Basically, I need to have access from all Subprocesses to a specific set of variables, which should be shared between these processes. Or, if every process got their own varibale, I need to be able to modify all of these variables by executing a task.
Edit: SO, Ive found out that the Task is indeed broadcasted at to all workers. But not the workers launched from the pool/ the concurrency but fro, the terminal.
So, if I start multiple terminals with celery worker ..... -c 2 These celery workers do receive the broadcast task. Which is good I guess. Now I want to broadcast these task to the PoolWorkers inside the celery workers too.
Basically I load a model and I want to relaod the model on all pool workers
Original:
I"ve been reading trough the user guide cat celery so that I can send a single task to all of my workers.
I am using RabbitMQ and everzthing elso works fine, but the broadcasted task are only processed by a single worker.
I define the exchange and the Queue
exchange = Exchange('broadcast_exchange', type='fanout')
celery.conf.task_queues = (Broadcast(name='broadcast_learning', exchange=exchange),)
And also the Task routes:
celery.conf.task_routes = {
'fworker.train_network':
{
'queue':'broadcast_learning',
'exchange':'broadcast_exchange'
},
....
}
But executing the task with .delay() or with .apply_async(queue='broadcast_learning') does not seem to send the task to ALL workers - instead only one is processing it.
After starting my worker it listens to the broadcast and the default queue, I see that they are registered in Celery (altough with a strange internal name)
[queues]
.> bcast.13bebf5c-f69c-40d9-a0e8-73f74efb9114 exchange=broadcast_exchange(fanout) key=celery
.> celery exchange=celery(direct) key=celery
I already changed from Redis backend to RabbitmQ, since some answers suggested that Redis is not working with broadcasting. But whatever I try, it does not seem to work.