Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

Please format your code with pixi run format

@coderabbitai review

@coderabbitai review

Also, I have recently spotted that btop reports large numbers of threads for the main snakemake process on linux. Although they all seem to sleep, it seems to me that this might be related. Maybe the async spawns are not getting closed by Python?

Just wanted to chime in quickly that i debugged the same issue last week and came to a different fix by persisting an async_runner on the workflow object so that the job scheduler can just reuse the same async event loop: #3911.

The solution here is also quite elegant i find.

IMHO, ideally one would instead rewrite quite some of the snakemake inner logic to be async, and then just keep within the same async loop. I started doing that, too, but did not find a good approach for the eval statements in eval_resource_expression.

@johanneskoester

Looks good to me. Wouldn't the same make sense on Unix-like OSes as well?

I am not sure. The stalling/hanging is specific to Windows where python implements sockets with real TCP connections, which have a "cooling off" period before they are freed. I think linux uses some sort pipe to communicate which are freed immediately.

Also, using asyncio.run(), guarantees a completely clean slate for every call. There is no risk of hanging task or unclosed resources attached to the loop.

So I think it safest to leave this fix a Windows only.

@coroa Was your problem also on Windows?

@johanneskoester

Also, I have recently spotted that btop reports large numbers of threads for the main snakemake process on linux. Although they all seem to sleep, it seems to me that this might be related. Maybe the async spawns are not getting closed by Python?

Yes. They are related to the same underlying pattern: calling asyncio.run() inside the parallel job scheduler.

When we run concurrently (e.g., -j 20), the scheduler invokes asyncio.run for each worker thread, which creates a new Event Loop.

I believe every new Event Loop creates its own default ThreadPoolExecutor (usually cpu_count workers) if it needs to run blocking code.

On Linux: This results in a "pool of pools." 20 jobs × ~20 default threads = ~400 idle threads. They sleep (as you saw) so they don't consume CPU, but they do bloat the thread count.

On Windows: We hit a harder limit. Each new loop also creates a TCP socket pair. 20 jobs × rapid scheduling cycles = thousands of TCP connections which eventually exhausts the port range and crashes the process.

This PR fixes the crash on Windows by reusing the loop. On Linux, the high thread count is "safe" (if messy), but applying this same fix there would likely reduce that thread bloat as wel.

I keept the ON_WINDOWSguard, to minimize the effect of my change, but I am happy to enable it globally.

@coroa Was your problem also on Windows?

Yes, the actual issue only surfaced on windows in the proactor. Same stack trace as your py-spy one, but i have observed hangs on linux too (which might be unrelated).

    async_run(postprocess())
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\site-packages\snakemake\common\__init__.py",
line 100, in async_run
    return asyncio.run(coroutine)
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\runners.py", line 194, in run
    with Runner(debug=debug, loop_factory=loop_factory) as runner:
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\runners.py", line 58, in __enter__
    self._lazy_init()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\runners.py", line 137, in _lazy_init
    self._loop = events.new_event_loop()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\events.py", line 823, in new_event_loop
    return get_event_loop_policy().new_event_loop()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\events.py", line 720, in new_event_loop
    return self._loop_factory()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\windows_events.py", line 316, in
__init__
    super().__init__(proactor)
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\proactor_events.py", line 640, in
__init__
    self._make_self_pipe()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\asyncio\proactor_events.py", line 787, in
_make_self_pipe
    self._ssock, self._csock = socket.socketpair()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\socket.py", line 627, in _fallback_socketpair
    ssock, _ = lsock.accept()
  File "C:\Users\jonas\open-tyndp\.pixi\envs\open-tyndp\Lib\socket.py", line 295, in accept
    fd, addr = self._accept()

@johanneskoester, I could make a fix like this for all platforms:

_thread_local = threading.local()

def async_run(coroutine):
    """Attaches to running event loop or creates a new one to execute a
    coroutine.
    .. seealso::
         https://github.com/snakemake/snakemake/issues/1105
         https://stackoverflow.com/a/65696398
    """
    # We reuse a thread-local loop to avoid:
    # 1. On Windows: Port exhaustion (ProactorEventLoop creates TCP sockets)
    # 2. On Linux: Thread bloat (New loops spawn new ThreadPoolExecutors)
    if not hasattr(_thread_local, "loop") or _thread_local.loop.is_closed():
        _thread_local.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(_thread_local.loop)
        # Limit inner threads to 1, effectively flattening the "Pool of Pools"
        _thread_local.loop.set_default_executor(
            concurrent.futures.ThreadPoolExecutor(max_workers=1)
        )
    try:
        return _thread_local.loop.run_until_complete(coroutine)
    except RuntimeError as e:
        coroutine.close()
        raise WorkflowError("Error running coroutine in event loop.", e)

Crucially, to fix the bloat, I can also limit the executor size on these internal loops. Since the code is already running inside a parallel Snakemake worker, it shouldn't need a massive thread pool of its own.

@johanneskoester /@coroa , do you think that there are any sideeffects of doing this?

@melund, this looks good to me as well. Now I am unsure which PR is prefable to be honest. @melund and @coroa, what are your opinions?

@johanneskoester, Just merge this windows PR right away.

I also made a PR (#3912) with the extended version for all OS'es. But I have no way to really test this on Linux, so I would think that the extended PR should only be merged once someone has used it for some real world applications.

@melund and @coroa, what are your opinions?

asyncio.run is not intended to be run repeatedly. for example the asyncio.run docs still have this comment: This function should be used as a main entry point for asyncio programs, and should ideally only be called once (that it became part of the loop_factory argument is a docs issue, refer to the original 3.7 docs).

I actually was also able to produce infrequent lock-ups on linux when using asyncio.to_thread in a storage plugin (PyPSA/snakemake-storage-plugin-cached-http@838c28d).

So i would not make it a windows only fix, but would suggest either #3912 or #3911 .

Which one? Puh. I originally thought, the unsolved issues in #3911 (not thread safe and resource expressions need to spawn a new async runner), made the global variable approach here a clear winner, but then i found solutions for both.

The main differences i see are:

#3911 uses the high-level asyncio.Runner interface (which was introduced in 3.11, which might be an issue) and cleans up the event loops when the runner is closed on tear down of the Workflow object, but it needs to jump through several hoops to make it available everywhere consistently.

#3912 uses the low-level event loop management interfaces to ensure that each thread reuses the same event loop, so the final effect is the same, except that the event loops are not closed at the end, which might hide exceptions in asyncio tasks (which snakemake is typically not using) and reuse the same event loops between several workflow instances (which snakemake typically does not have). But, it is a very local code change and will practically achieve the same outcome.

So, 🤷 .

I would not use set_default_executor, my understanding is that the executor is only instantiated when you want to make something synchronuous behave well within an async loop by using the high-level asyncio.to_thread or low level loop.run_in_executor.. as mentioned in run_in_executor the default executor is lazy-instantiated, so there is a high chance instantiating one and setting it as default does actually increase the number of threads rather than decrease it, with the negative effect that the utility of asyncio.to_thread is limited, i think.

@coroa You mentioned: "there is a high chance instantiating one ... does actually increase the number of threads rather than decrease it"

Actually, the motivation for set_default_executor(max_workers=1) is specifically because @johanneskoester reported seeing massive thread counts on Linux. I thought it indicated that that something is indeed lazily instantiating these executors.

The default ThreadPoolExecutor spawns lots of workers, and this happens for all the loops.

By forcing max_workers=1, we ensure that when the executor is instantiated, it caps at 1 thread. I was thinking it would solve the "Pool of Pools" explosion while keeping the change local and simple.

I think someone needs to actual try to run snakemake with the change in #3912, and see what it does with the thread bloat on Linux. Until then I wote for merging this PR to fix the critical issue on Windows with a minimal code.

@coroa You mentioned: "there is a high chance instantiating one ... does actually increase the number of threads rather than decrease it"

Actually, the motivation for set_default_executor(max_workers=1) is specifically because @johanneskoester reported seeing massive thread counts on Linux. I thought it indicated that that something is indeed lazily instantiating these executors.

I understand that, and your fix w/o the set_default_executor(ThreadPoolExecutor(max_workers=1)) will also already fix the thread explosion, since you limit the number of async event loops to the number of workers plus the one on the main thread.

What i am unclear/worried about is that if you don't set the default executor, the event loop will instantiate one, WHEN it is needed, and i think it is not needed in most cases. If you instantiate an executor to set it as the default then the additional single thread is there regardless if it is used.

Yes. You are right. In this case it would always instantiate a single worker-thread for every concurrent job. But I suspect that there could be something that is indeed lazily instantiating these executors (likely aiofiles, implicit run_in_executor calls, or plugins).

The default ThreadPoolExecutor spawns min(32, os.cpu_count() + 4) worker-threads. With 20 concurrent jobs --> 20 loops -> 20 Default Executors × ~20 worker-threads = 400+ threads.

My fist PR, and yours does nothing to fix that issue (I think 😅 ). It just prevents that they are not constantly closed and created again.

Of course if you could remove the places that instantiate the ThreadPoolExecutor's, you could maybe solve the problem in an other way.

@johanneskoester I fine with @coroa's solution to the issues on Windows. I haven't run my large pipelines with his PR yet, but I assume they would solve the problem as well.

So if you like that solution better, just merge that. I am looking forward to new version of snakemake which doesn't fail randomly on Windows.

But I suspect that there could be something that is indeed lazily instantiating these executors (likely aiofiles, implicit run_in_executor calls, or plugins).

Ok, i did a very loose review job of how often run_in_executor appears in base libraries and you are right it's more often than i thought. it's used by aiofiles, as by aiohttp. So this might launch it more often.

What do you think about sharing a default ThreadPoolExecutor across all asyncio loops?

What do you think about sharing a default ThreadPoolExecutor across all asyncio loops?

I don't see any problems with that, but that doesn't count for much since I am not well versed in asyncio, and how the different parts play together. I think that part could also come in a follow up PR - I am really hoping for a new snakemake release soon 😅

Closing in favor of #3911.