Repository: agronholm/anyio

Issue:
to_process.run_sync() causing high memory usage and exceeding defined process limit
### Things to check first

- [X] I have searched the existing issues and didn't find my bug already reported there

- [X] I have checked that my bug is still present in the latest release


### AnyIO version

3.7.0

### Python version

3.10.11

### What happened?

I encountered an issue while utilizing `to_process.run_sync()` in conjunction with a process pool. The problem arises when the process pool loses track of instances, leading to excessive memory consumption.

In my scenario, I needed to execute a CPU-intensive task asynchronously multiple times at unspecified intervals. Recognizing the overhead involved in process creation, I turned to the solution of maintaining a process pool. I implemented your provided code with the following parameters:

1. The capacity limiter was set to the number of available CPUs (I used 16 CPUs).
2. The CPU-intensive task was run multiple times (over 100).
3. Uncertainty of when the CPU-intensive task would be invoked.
4. The CPU-intensive task utilizes 1GB of memory by default.

While I built my code upon your process pool maintenance solutions, I encountered a noticeable surge in memory consumption. Additionally, I observed that the count of active processes surpassed the limit I had set (exceeding 30 processes).

Assumption Regarding the Root Cause:
Upon investigating your code, I've observed that the issue may be related to AnyIO losing the process pointer while identifying idle processes that are meant to be terminated.

Following my investigation, I engaged with your code locally and successfully resolved the problem. The code now functions as intended.

### How can we reproduce the bug?

```
from time import sleep

import anyio
import psutil
from anyio import to_process
from anyio._backends._asyncio import CapacityLimiter
from anyio.to_process import WORKER_MAX_IDLE_TIME


def cpu_intensive_callable() -> None:
    sleep(WORKER_MAX_IDLE_TIME + 5)  # 5 minutes + 5 seconds


class CpuIntensiveClass:
    async def cpu_intensive_method(self, capacity_limiter: CapacityLimiter) -> None:
        print("Running cpu intensive method")
        await to_process.run_sync(cpu_intensive_callable, cancellable=True, limiter=capacity_limiter)


async def main() -> None:
    capacity_limiter = CapacityLimiter(total_tokens=2)  # limit the number of concurrent processes to 2
    cpu_intensive_class = CpuIntensiveClass()
    for _ in range(10):
        await cpu_intensive_class.cpu_intensive_method(capacity_limiter)
        print(f"Running python processes number is {sum(1 for p in psutil.process_iter(['pid', 'name']) if 'python' in str(p.name).lower())}")


if __name__ == "__main__":
    anyio.run(main)
```

Notice that I used psutil package in order to track the current number of python processes.


Discussion:
agronholm: I don't understand what this is supposed to fix. The `process` variable isn't used anymore by the time it's overridden. Please explain. Also, a basic requirement for any fix is at least one accompanying test that proves that it fixes an existing problem.
zxul767: This is supposed to fix a leak of alive but unused processes *in a very specific condition.*

It happens when a process is selected from `idle_workers` *and* there is some pruning to do (i.e., there are processes that have been idle for too long). 

In such a case, by the time execution reaches these lines: 
```
    finally:
        if process in workers:
            idle_workers.append((process, current_time()))
```

`process` is pointing to one of the processes that were killed and removed from `idle_workers` and from `workers`, and not to the process that actually ran the request. As a result, the `if` condition fails and the process is never put back on the `idle_workers` pool, so it goes effectively untracked (i.e., never re-used again) from then on.
agronholm: Ah, I finally understand now. Thank you for explaining. But like I said before, a bug fix requires an accompanying test (and 
a changelog entry) to be merged.
agronholm: I was planning to release v4.0.0 final today, but I want this fix in the release. I could help you write a test for this, if you need help that is.
zxul767: That would be appreciated @agronholm. I ran into a similar issue as @AnaelGorfinkel but I hadn't had time to write a minimal reproduction of the issue to serve as a regression test.
agronholm: Alright, I'm currently writing said test.
agronholm: I guess it's been a while since I touched this code because I'm having trouble understanding some aspects of it, like why is an idle but functional process being removed from the set of worker processes.
agronholm: This is truly bizarre. On my machine, the test suite passed on all Python versions.
zxul767: I'm not sufficiently familiar with the codebase to know for sure, but is it possible that it is not entering the pruning loop because `elapsed_time = now - idle_workers[0][1]` is `< WORKERS_MAX_IDLE_TIME`?

```
if now - idle_workers[0][1] < WORKER_MAX_IDLE_TIME:
    break
```

Perhaps the test needs to set the fake idle worker's time to `WORKER_MAX_IDLE_TIME + current_time()`?

There's another error in the logs but it's related to the `Mock(Process)` not mocking `_stdin`.
agronholm: > I'm not sufficiently familiar with the codebase to know for sure, but is it possible that it is not entering the pruning loop because elapsed_time = now - idle_workers[0][1] is < WORKERS_MAX_IDLE_TIME?

Given that `now` should be in the hundreds of thousands, and `idle_workers[0][1] ` is set to 0, I don't see how it would not get pruned. But I will continue to investigate.
agronholm: My second fix made 100% sure that the mock process gets removed from the workers set. Yet, it 100% always fails due to asyncio shutdown code tripping on it. None of this makes a lick of sense! It would be one thing if this failed on my local machine but everything always passes for all Python versions!
agronholm: > I'm not sufficiently familiar with the codebase to know for sure, but is it possible that it is not entering the pruning loop because `elapsed_time = now - idle_workers[0][1]` is `< WORKERS_MAX_IDLE_TIME`?

Ah, my assumption was wrong: `now` is actually a relatively low number on the runners! Thus, the idle_since time for the mock process must be negative to always satisfy the condition.
zxul767: Great! Thanks for helping get this PR to the finish line!

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 5ed601d..a7c5073 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -18,6 +18,10 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Re-added the ``item_type`` argument to ``create_memory_object_stream()`` (but using it
   raises a deprecation warning and does nothing with regards to the static types of the
   returned streams)
+- Fixed processes spawned by ``anyio.to_process()`` being "lost" as unusable to the
+  process pool when processes that have idled over 5 minutes are pruned at part of the
+  ``to_process.run_sync()`` call, leading to increased memory consumption
+  (PR by Anael Gorfinkel)
 
 **4.0.0rc1**
 
diff --git a/src/anyio/to_process.py b/src/anyio/to_process.py
index dcfda5a..f3ccef5 100644
--- a/src/anyio/to_process.py
+++ b/src/anyio/to_process.py
@@ -118,14 +118,14 @@ async def run_sync(
                     if now - idle_workers[0][1] < WORKER_MAX_IDLE_TIME:
                         break
 
-                    process, idle_since = idle_workers.popleft()
-                    process.kill()
-                    workers.remove(process)
-                    killed_processes.append(process)
+                    process_to_kill, idle_since = idle_workers.popleft()
+                    process_to_kill.kill()
+                    workers.remove(process_to_kill)
+                    killed_processes.append(process_to_kill)
 
                 with CancelScope(shield=True):
-                    for process in killed_processes:
-                        await process.aclose()
+                    for killed_process in killed_processes:
+                        await killed_process.aclose()
 
                 break
 


Test patch:
diff --git a/tests/test_to_process.py b/tests/test_to_process.py
index 4c6507d..ac0b4b5 100644
--- a/tests/test_to_process.py
+++ b/tests/test_to_process.py
@@ -5,6 +5,7 @@ import platform
 import sys
 import time
 from functools import partial
+from unittest.mock import Mock
 
 import pytest
 
@@ -15,6 +16,7 @@ from anyio import (
     to_process,
     wait_all_tasks_blocked,
 )
+from anyio.abc import Process
 
 pytestmark = pytest.mark.anyio
 
@@ -95,3 +97,30 @@ async def test_cancel_during() -> None:
 
     # The previous worker was killed so we should get a new one now
     assert await to_process.run_sync(os.getpid) != worker_pid
+
+
+async def test_exec_while_pruning() -> None:
+    """
+    Test that in the case when one or more idle workers are pruned, the originally
+    selected idle worker is re-added to the queue of idle workers.
+    """
+
+    worker_pid1 = await to_process.run_sync(os.getpid)
+    workers = to_process._process_pool_workers.get()
+    idle_workers = to_process._process_pool_idle_workers.get()
+    real_worker = next(iter(workers))
+
+    fake_idle_process = Mock(Process)
+    workers.add(fake_idle_process)
+    try:
+        # Add a mock worker process that's guaranteed to be eligible for pruning
+        idle_workers.appendleft(
+            (fake_idle_process, -to_process.WORKER_MAX_IDLE_TIME - 1)
+        )
+
+        worker_pid2 = await to_process.run_sync(os.getpid)
+        assert worker_pid1 == worker_pid2
+        fake_idle_process.kill.assert_called_once_with()
+        assert idle_workers[0][0] is real_worker
+    finally:
+        workers.discard(fake_idle_process)