You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "luoyuliuyin (via GitHub)" <gi...@apache.org> on 2024/05/08 11:32:55 UTC
[PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
luoyuliuyin opened a new pull request, #39484:
URL: https://github.com/apache/airflow/pull/39484
PR of https://github.com/apache/airflow/issues/39482
When the scheduler send task to celery, if there is only 1 task in the current cycle, the task will be sent to the main thread; if there are multiple tasks, a thread pool will be created based on the number of CPU cores, and then all tasks will be consumed by the thread pool.
There are some problems with the current implementation. The scheduler creates a thread pool every time it schedules, which will bring a very large performance overhead. In fact, the thread pool can be reused.
When I tested, sometimes it would take almost 4 seconds to consume 32 tasks.
If the thread pool is reused, it only takes 10 milliseconds.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2103450422
> We conducted a throughput benchmarking test, executing 15k DAG's comprised of simple BashOperator tasks. We measured the success rate of tasks completed within a 2-minute interval and observed a 3.7% increase in throughput with this PR. @luoyuliuyin, would like to know how we can replicate the scenario under which your throughput increased by 50%.
When I did performance testing, I also used a dag composed of a simple BashOperator. You may not see the difference when the load is relatively low. When I tested, the load of the airflow scheduler was very high. There were 40,000 scheduled dags (every 2 hours) and 10,000 manually triggered dags. The number of schedulers was 30, and the number of workers was 100
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on code in PR #39484:
URL: https://github.com/apache/airflow/pull/39484#discussion_r1596007476
##########
airflow/providers/celery/executors/celery_executor.py:
##########
@@ -267,6 +267,7 @@ def __init__(self):
self.tasks = {}
self.task_publish_retries: Counter[TaskInstanceKey] = Counter()
self.task_publish_max_retries = conf.getint("celery", "task_publish_max_retries")
+ self.send_pool = ProcessPoolExecutor(max_workers=self._sync_parallelism)
Review Comment:
When I did performance testing, I also used a dag composed of a simple BashOperator. You may not see the difference when the load is relatively low. When I tested, the load of the airflow scheduler was very high. There were 40,000 scheduled dags (every 2 hours) and 10,000 manually triggered dags. The number of schedulers was 30, and the number of workers was 100.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101323443
> broker
I use rabbitmq,
AIRFLOW__CELERY__BROKER_URL=amqp://admin:admin@xxx.xxx:5672
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2104545760
> I am a little skeptical that the overhead for process pool creation is meaningful. Because i think the dominant performance consideration is the blocking IO sending the task to redis and waiting for response.
>
> In order for the process pool creation to be dominant, you'd have to be sending very few tasks i'd think, in which case it would be fast anyway. No?
>
> E.g. in your example above, you claim that it sometimes took 4 seconds to send 32 tasks. But, you did not establish that the bottleneck was process pool creation. How do you know it wasn't mostly just waiting for the broker to respond? Are you claiming that it takes 4 seconds to create the process pool? Maybe you would be able to get at this by adding a log message immediately after entering the context?
I added some logs for comparison, the longest time it takes is the processing of the task, and then the closing of the ProcessPoolExecutor.
**once ProcessPoolExecutor:**
**reuse ProcessPoolExecutor:**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2133206232
> > @luoyuliuyin have we verified if it's due to a resource leak caused by Rabbitmq?
>
> I'm not sure, I've upgraded rabbitmq to the latest version (3.13.2), but rabbitmq consumption is still getting slower and slower
What happens when you restart rabbitmq rather than scheduler?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100957025
> Here are some benefits:
> The maximum scheduling delay between two nodes is reduced from 10 seconds to less than 1 second.
> System throughput increased by more than 50%
> The machine load has dropped a lot
I assume it's only on MacOS?
Do you see the same gains on Linux?
Maybe this shoud be MacOS only change ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2112697556
@luoyuliuyin have we verified if it's due to a resource leak caused by Rabbitmq?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2103758103
> What happens if you `kill -9` one (or all) of the ProcessPool processes and then try to schedule/enqueue new tasks?
After using kill -9 to kill all child processes created by ProcessPoolExecutor, the scheduler process will exit abnormally. Therefore, how to restore or roll back to the previous implementation needs to be considered.
By the way, the reason why I can run it stably in production is because my startup script has the ability to automatically restart, and the scheduler will restart after exiting abnormally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on code in PR #39484:
URL: https://github.com/apache/airflow/pull/39484#discussion_r1596007476
##########
airflow/providers/celery/executors/celery_executor.py:
##########
@@ -267,6 +267,7 @@ def __init__(self):
self.tasks = {}
self.task_publish_retries: Counter[TaskInstanceKey] = Counter()
self.task_publish_max_retries = conf.getint("celery", "task_publish_max_retries")
+ self.send_pool = ProcessPoolExecutor(max_workers=self._sync_parallelism)
Review Comment:
When I did performance testing, I also used a dag composed of a simple BashOperator. You may not see the difference when the load is relatively low. When I tested, the load of the airflow scheduler was very high. There were 40,000 scheduled dags (every 2 hours) and 10,000 manually triggered dags. The number of schedulers was 30, and the number of workers was 100.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101176516
What Celery broker are you using?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2102503655
@vatsrahul1001 is going to run some performance benchmarks with the changes in this PR too in upcoming days.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2134339192
> There is no change. It seems that the resource leak is caused by the client rather than the server.
Is this possible that you have a network proxy (might be transparent, provided by Kubernetes or other networking configuration) that gets slower and slower as more and more connections are opened? Maybe simply your proxy does not properly clean opened connections to rabbitmq? Do you see a growht in other resources used by any of the components of the system? Maybe - for some reason - there is a problem with killing the processes that are forked and they remain as zombies ?
Also - do you happen to have some CPU limits set for the Scheduler POD from the deployment point of view (like K8S CPU limits?)
I really would love to understand where the slow-down is coming from
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2134270804
> Is this trying to save the initialisation overhead, or the shutdown overhead after each send? Reducing the shutdown overhead is most more dangerous as discussed above, but it is not obvious to me from the above discussion that’s what we’re looking for. The former can be easily avoided without consequences if lines 341-344/345 are changed to
>
> ```python
> with self.send_pool as pool:
> key_and_async_results = list(pool.map(...))
> ```
In this way, self.send_pool will be closed after processing the first batch of data, and then an error `RuntimeError: cannot schedule new futures after shutdown` will be reported when processing the second batch of data.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101002106
> > Here are some benefits:
>
> > The maximum scheduling delay between two nodes is reduced from 10 seconds to less than 1 second.
> > System throughput increased by more than 50%
> > The machine load has dropped a lot
>
> I assume it's only on MacOS? Do you see the same gains on Linux?
>
> Maybe this shoud be MacOS only change ?
The above data is my running results in the Linux environment. If ProcessPoolExecutor is created every time, no problem will be seen at the beginning, but over time the scheduling will become slower and slower, especially under high load conditions. If ProcessPoolExecutor can be reused, these problems will not occur.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101370206
> What happens if you `kill -9` one (or all) of the ProcessPool processes and then try to schedule/enqueue new tasks?
I'll test it later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100988477
Also cc: @ashb @kaxil => maybe there is another reason why we are doing it this way for Celery Executor. I think this change has the potential to change resource usage of Airflow using Celery, so it needs to be carefully considered and I think it needs few more pairs of eyes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100903473
Regarding risk considerations, in our company I have already released this change to production. Our production environment is centos8 and it has been running well until now. Our airflow scale has 30,000 DAGs, and the average daily task scheduling number is 1 million.
Here are some benefits:
1. The maximum scheduling delay between two nodes is reduced from 10 seconds to less than 1 second.
2. System throughput increased by more than 50%
3. The machine load has dropped a lot
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "hussein-awala (via GitHub)" <gi...@apache.org>.
hussein-awala commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100812157
> On the other hand, the change you propose will impact memory used in a long term as the long running processes might accumulate memory when run for a long time and run different callables. So it might cause memory leak, not mentioning that in the original model the forked processes are only needed for a very short time - to send tasks to celery and for example they are not consuming any resources when there are no tasks to run (because they are stopped right after they complete,
This pool executor is only used to execute short-running tasks, I don't think it will impact the memory usage, and I believe that this kind of objects (process/thread pool executor) should be created once and reused.
If it's really a risky change, we can add a new configuration to switch between the behaviors and change the default value later when the new behavior is tested in production.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2103879926
> At the beginning, there may be no obvious difference in the task throughput of the two implementation solutions. After a long time (such as 1 day), the throughput of the existing implementation solution will gradually decrease. What I compare is the stable result.
I cannot think of a single reason (on the cliend side) why it would happen and why changing to a long running pool would change it. The way pool context manager works is that it will wait until all processes started in the pool complete their tasks and close the pool including all the processes freeing all the resources. So there is no rason why "airflow" would slow down.
However maybe that is a problem of your firewall/networking/rabbitmq wrong behaviour. The main difference between long running pool of processes and processes started temporarily to send the tasks, is that the long running processes **might**. (depending on implementation of the way client API works) reuse an open connection to the broker to send the tasks rather than open a new one. But if rabbitmq server is implemented properly, then it should have no effect of getting longer and longer over time, because after closing the processes sending the tasks, rabbitmq should free all the resources on the server side.
So maybe the problem is a that your rabbitmq leaks resources when processes sending tasks to it are closing down? Do you have some monitoring / can you please provide some data to back-up the statement that rabbitmq is actually leaking resources in this case? I think if airflow gets slower and slower over time, you should be able to see some resources leaking on either side - memory, CPU being the most likely candidates and my hypothesis is that it's rabbitmq misbehaving (which might be for example a known bug in some old version of rabbitmq. Quick search reveals similar behaviour observed https://groups.google.com/g/rabbitmq-users/c/v630G6OCxuU in some old versions of rabbitmq (but I have not looked in details of that conversation, it's just likely that it might be something similar). Can you please take a close look at the rabbitmq side of yours and see if you can observe some resource leaks when you go back to the current solution and maybe upgrade rabbitmq to latest versions
to exclude the possibility it is some old bug?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2104551800
> > At the beginning, there may be no obvious difference in the task throughput of the two implementation solutions. After a long time (such as 1 day), the throughput of the existing implementation solution will gradually decrease. What I compare is the stable result.
>
> I cannot think of a single reason (on the cliend side) why it would happen and why changing to a long running pool would change it. The way pool context manager works is that it will wait until all processes started in the pool complete their tasks and close the pool including all the processes freeing all the resources. So there is no rason why "airflow" would slow down.
>
> However maybe that is a problem of your firewall/networking/rabbitmq wrong behaviour. The main difference between long running pool of processes and processes started temporarily to send the tasks, is that the long running processes **might**. (depending on implementation of the way client API works) reuse an open connection to the broker to send the tasks rather than open a new one. But if rabbitmq server is implemented properly, then it should have no effect of getting longer and longer over time, because after closing the processes sending the tasks, rabbitmq should free all the resources on the server side.
>
> So maybe the problem is a that your rabbitmq leaks resources when processes sending tasks to it are closing down? Do you have some monitoring / can you please provide some data to back-up the statement that rabbitmq is actually leaking resources in this case? I think if airflow gets slower and slower over time, you should be able to see some resources leaking on either side - memory, CPU being the most likely candidates and my hypothesis is that it's rabbitmq misbehaving (which might be for example a known bug in some old version of rabbitmq. Quick search reveals similar behaviour observed https://groups.google.com/g/rabbitmq-users/c/v630G6OCxuU in some old versions of rabbitmq (but I have not looked in details of that conversation, it's just likely that it might be something similar). Can you please take a close look at the rabbitmq side of yours and see if you can observe some resource leaks when you go back to the current solution and maybe upgrade rabbitmq to latest versions
to exclude the possibility it is some old bug?
Maybe as you said, the reason is the resource leak caused by rabbitmq. I will investigate this aspect first and synchronize it if there are other new clues.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100975510
@hussein-awala
> This pool executor is only used to execute short-running tasks, I don't think it will impact the memory usage, and I believe that this kind of objects (process/thread pool executor) should be created once and reused.
> If it's really a risky change, we can add a new configuration to switch between the behaviors and change the default value later when the new behavior is tested in production.
yes I think it's not **significant** overhead, I just wonder if there are no cases here which will accumulate over time. Actually in Python 3.11 there is a new feature in ProcessPoolExecutor that would be even better `max_tasks_per_child` that would be BEST approach - because then we could be able to run the process pool only once and let it recycle the processes after say 1000 tasks scheduled - and that would be rather safe change if we only do it in Python 3.11 and have a reasonable default for `max_tasks_per_child`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2103466344
> > We conducted a throughput benchmarking test, executing 15k DAG's comprised of simple BashOperator tasks. We measured the success rate of tasks completed within a 2-minute interval and observed a 3.7% increase in throughput with this PR. @luoyuliuyin, would like to know how we can replicate the scenario under which your throughput increased by 50%.
>
> When I did performance testing, I also used a dag composed of a simple BashOperator. You may not see the difference when the load is relatively low. When I tested, the load of the airflow scheduler was very high. There were 40,000 scheduled dags (every 2 hours) and 10,000 manually triggered dags. The number of schedulers was 30, and the number of workers was 100
At the beginning, there may be no obvious difference in the task throughput of the two implementation solutions. After a long time (such as 1 day), the throughput of the existing implementation solution will gradually decrease. What I compare is the stable result.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "dirrao (via GitHub)" <gi...@apache.org>.
dirrao commented on code in PR #39484:
URL: https://github.com/apache/airflow/pull/39484#discussion_r1594061166
##########
airflow/providers/celery/executors/celery_executor.py:
##########
@@ -267,6 +267,7 @@ def __init__(self):
self.tasks = {}
self.task_publish_retries: Counter[TaskInstanceKey] = Counter()
self.task_publish_max_retries = conf.getint("celery", "task_publish_max_retries")
+ self.send_pool = ProcessPoolExecutor(max_workers=self._sync_parallelism)
Review Comment:
Don't we need to gracefully shutdown it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101071101
> The above data is my running results in the Linux environment. If ProcessPoolExecutor is created every time, no problem will be seen at the beginning, but over time the scheduling will become slower and slower, especially under high load conditions. If ProcessPoolExecutor can be reused, these problems will not occur.
That is very interesting why. I would not expect it - do you have some more data showing trace of it and what gets slower?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2133240615
Is this trying to save the initialisation overhead, or the shutdown overhead after each send? Reducing the shutdown overhead is most more dangerous as discussed above, but it is not obvious to me from the above discussion that’s what we’re looking for. The former can be easily avoided without consequences if lines 341-344/345 are changed to
```python
with self.send_pool as pool:
key_and_async_results = list(pool.map(...))
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100687710
This is a very risky change if you have not thoroughly tested various configuration and I think it's for the worse on production systems. When you are using it on MacOS, yes it will be a bit slow because default multiprocessing context on MacOS is `spawn` https://docs.python.org/3/library/multiprocessing.html#multiprocessing-start-methods but on linux it is `fork` and this implementation is optimized for production case where Airflow is run on Linux.
Have you considered impact on production for this change?
The overhead for production (Linux and system where fork is supported) is minimal, because `fork` model is used for the processes. Forking a process that does very little is very inexpensive in this model, because the heap memory is in "copy on write" mode, which means that forking is nearly instanteneous and until some memory is modiified, no extra memory is used.
On the other hand, the change you propose will impact memory used in a long term as the long running processes might accumulate memory when run for a long time and run different callables. So it might cause memory leak, not mentioning that in the original model the forked processes are only needed for a very short time - to send tasks to celery and for example they are not consuming any resources when there are no tasks to run (because they are stopped right after they complete,
Have you tested it beyond MacOS to see what kind of performance gains you have and what impact on memory usage and resource usage at scale?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101300305
I am skeptical that the overhead for process pool creation is meaningful. Because i think the dominant performance consideration is the blocking IO sending the task to redis and waiting for response.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101380558
> I am a little skeptical that the overhead for process pool creation is meaningful. Because i think the dominant performance consideration is the blocking IO sending the task to redis and waiting for response.
>
> In order for the process pool creation to be dominant, you'd have to be sending very few tasks i'd think, in which case it would be fast anyway. No?
>
> E.g. in your example above, you claim that it sometimes took 4 seconds to send 32 tasks. But, you did not establish that the bottleneck was process pool creation. How do you know it wasn't mostly just waiting for the broker to respond? Are you claiming that it takes 4 seconds to create the process pool? Maybe you would be able to get at this by adding a log message immediately after entering the context?
I'm testing it through the logs. It takes a while. It usually takes about 2 days to see that the scheduling process slows down significantly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101122299
And just to add a little bit of context - why I am surprised to see it .... It's not that I am stubborn. I just looked deeper:
The current context manager behaviour has been introduced in 2020: https://github.com/apache/airflow/pull/7542
And previously the implementation looked like that:
```
# Recreate the process pool each sync in case processes in the pool die
self._sync_pool = Pool(processes=num_processes)
```
So at least we know **why** we are recreating the pool in the first place with every sync - to handle the case where the processes die. And indeed the Process Pool does not have "self-healing" option, so if the processes will die, they will not be recreated and eventually Celery executor will stop sending tasks.
This is why I consider those changes as "risky" - because there might be some good reasons why we are doing it in the first place. And the current proposal does not handle the case to be honest.
So if we change the behaviour here, we need to handle this case as well and be sure that what we are doing is not causing more problems that might hit us back.
And we need to know if we **really** need to do it. I would be really surprises that no-one noticed the slow-down you mentioned :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2100964084
And I am not totally against it - I am just surprised to see such numbers on Linux. It's been running like that for years, so it would be interesting to see if indeed this is is only the "spawn" vs "fork" issue or maybe there are other factors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "dstandish (via GitHub)" <gi...@apache.org>.
dstandish commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101387857
> I'm testing it through the logs. It takes a while. It usually takes about 2 days to see that the scheduling process slows down significantly.
Understood. Did you see my suggestion about adding a message inside the context? If it's true that the pool is created at that point (i'm not sure) then that would show you how long it takes to actually create the pool and that would separate it from the sending. Maybe it's something else that is slow after 2 days?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101185520
What happens if you `kill -9` one (or all) of the ProcessPool processes and then try to schedule/enqueue new tasks?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2103162890
We conducted a throughput benchmarking test, executing 15k DAG's comprised of simple BashOperator tasks. We measured the success rate of tasks completed within a 2-minute interval and observed a 3.7% increase in throughput with this PR. @luoyuliuyin, would like to know how we can replicate the scenario under which your throughput increased by 50%.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101183817
I suspect if we want to roll out this change we should do it behind a config flag so that it is a) not on by default, and b) easy to fall back to the old behavior if it causes problems for some people
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2101367892
> > The above data is my running results in the Linux environment. If ProcessPoolExecutor is created every time, no problem will be seen at the beginning, but over time the scheduling will become slower and slower, especially under high load conditions. If ProcessPoolExecutor can be reused, these problems will not occur.
>
> That is very interesting why. I would not expect it - do you have some more data showing trace of it and what gets slower?
Not yet, I can only locate here through the logs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2112887809
> @luoyuliuyin have we verified if it's due to a resource leak caused by Rabbitmq?
I'm not sure, I've upgraded rabbitmq to the latest version (3.13.2), but rabbitmq consumption is still getting slower and slower
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix: Reuse ProcessPoolExecutor in CeleryExecutor [airflow]
Posted by "luoyuliuyin (via GitHub)" <gi...@apache.org>.
luoyuliuyin commented on PR #39484:
URL: https://github.com/apache/airflow/pull/39484#issuecomment-2134269202
> > > @luoyuliuyin have we verified if it's due to a resource leak caused by Rabbitmq?
> >
> >
> > I'm not sure, I've upgraded rabbitmq to the latest version (3.13.2), but rabbitmq consumption is still getting slower and slower
>
> What happens when you restart rabbitmq rather than scheduler?
There is no change. It seems that the resource leak is caused by the client rather than the server.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org