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.

  • I got the same problem, I want to send a broadcast to every working subprocess to reload a config, but it cannot work, only one of subprocess do the work. – guichao Jun 22 at 4:25
  • Did you tried to duplicate the Task? response = [fw.train_network.delay().get() for _ in range(Workers)]. Its hacky but works for me – Rikku Porta Jul 18 at 11:47

Your Answer

 

By clicking "Post Your Answer", you acknowledge that you have read our updated terms of service, privacy policy and cookie policy, and that your continued use of the website is subject to these policies.

Browse other questions tagged or ask your own question.