We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
to_thread.run_sync() foils meta-class destructors 
Invoking `to_thread.run_sync()` seems to prevent meta-class destructors of completely unrelated classes from being called on shutdown.

```python
import trio

class deltype(type):
    def __del__(self):
        print(f"{self.__name__} class deleted")

Test = deltype("Test", (object,), {})

async def async_main():
    await trio.to_thread.run_sync(lambda: print('hi'))
    await trio.sleep(0)

trio.run(async_main)
```

expected output:
```
hi
Test class deleted
```

actual output:
```
hi
```

if the `run_sync()` is commented out, the meta-class destructor is called as expected:
```
Test class deleted
```


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `trio/_core/_thread_cache.py`
Content:
```
1 from threading import Thread, Lock
2 import outcome
3 from itertools import count
4 
5 # The "thread cache" is a simple unbounded thread pool, i.e., it automatically
6 # spawns as many threads as needed to handle all the requests its given. Its
7 # only purpose is to cache worker threads so that they don't have to be
8 # started from scratch every time we want to delegate some work to a thread.
9 # It's expected that some higher-level code will track how many threads are in
10 # use to avoid overwhelming the system (e.g. the limiter= argument to
11 # trio.to_thread.run_sync).
12 #
13 # To maximize sharing, there's only one thread cache per process, even if you
14 # have multiple calls to trio.run.
15 #
16 # Guarantees:
17 #
18 # It's safe to call start_thread_soon simultaneously from
19 # multiple threads.
20 #
21 # Idle threads are chosen in LIFO order, i.e. we *don't* spread work evenly
22 # over all threads. Instead we try to let some threads do most of the work
23 # while others sit idle as much as possible. Compared to FIFO, this has better
24 # memory cache behavior, and it makes it easier to detect when we have too
25 # many threads, so idle ones can exit.
26 #
27 # This code assumes that 'dict' has the following properties:
28 #
29 # - __setitem__, __delitem__, and popitem are all thread-safe and atomic with
30 #   respect to each other. This is guaranteed by the GIL.
31 #
32 # - popitem returns the most-recently-added item (i.e., __setitem__ + popitem
33 #   give you a LIFO queue). This relies on dicts being insertion-ordered, like
34 #   they are in py36+.
35 
36 # How long a thread will idle waiting for new work before gives up and exits.
37 # This value is pretty arbitrary; I don't think it matters too much.
38 IDLE_TIMEOUT = 10  # seconds
39 
40 name_counter = count()
41 
42 
43 class WorkerThread:
44     def __init__(self, thread_cache):
45         self._job = None
46         self._thread_cache = thread_cache
47         # This Lock is used in an unconventional way.
48         #
49         # "Unlocked" means we have a pending job that's been assigned to us;
50         # "locked" means that we don't.
51         #
52         # Initially we have no job, so it starts out in locked state.
53         self._worker_lock = Lock()
54         self._worker_lock.acquire()
55         thread = Thread(target=self._work, daemon=True)
56         thread.name = f"Trio worker thread {next(name_counter)}"
57         thread.start()
58 
59     def _work(self):
60         while True:
61             if self._worker_lock.acquire(timeout=IDLE_TIMEOUT):
62                 # We got a job
63                 fn, deliver = self._job
64                 self._job = None
65                 result = outcome.capture(fn)
66                 # Tell the cache that we're available to be assigned a new
67                 # job. We do this *before* calling 'deliver', so that if
68                 # 'deliver' triggers a new job, it can be assigned to us
69                 # instead of spawning a new thread.
70                 self._thread_cache._idle_workers[self] = None
71                 deliver(result)
72             else:
73                 # Timeout acquiring lock, so we can probably exit. But,
74                 # there's a race condition: we might be assigned a job *just*
75                 # as we're about to exit. So we have to check.
76                 try:
77                     del self._thread_cache._idle_workers[self]
78                 except KeyError:
79                     # Someone else removed us from the idle worker queue, so
80                     # they must be in the process of assigning us a job - loop
81                     # around and wait for it.
82                     continue
83                 else:
84                     # We successfully removed ourselves from the idle
85                     # worker queue, so no more jobs are incoming; it's safe to
86                     # exit.
87                     return
88 
89 
90 class ThreadCache:
91     def __init__(self):
92         self._idle_workers = {}
93 
94     def start_thread_soon(self, fn, deliver):
95         try:
96             worker, _ = self._idle_workers.popitem()
97         except KeyError:
98             worker = WorkerThread(self)
99         worker._job = (fn, deliver)
100         worker._worker_lock.release()
101 
102 
103 THREAD_CACHE = ThreadCache()
104 
105 
106 def start_thread_soon(fn, deliver):
107     """Runs ``deliver(outcome.capture(fn))`` in a worker thread.
108 
109     Generally ``fn`` does some blocking work, and ``deliver`` delivers the
110     result back to whoever is interested.
111 
112     This is a low-level, no-frills interface, very similar to using
113     `threading.Thread` to spawn a thread directly. The main difference is
114     that this function tries to re-use threads when possible, so it can be
115     a bit faster than `threading.Thread`.
116 
117     Worker threads have the `~threading.Thread.daemon` flag set, which means
118     that if your main thread exits, worker threads will automatically be
119     killed. If you want to make sure that your ``fn`` runs to completion, then
120     you should make sure that the main thread remains alive until ``deliver``
121     is called.
122 
123     It is safe to call this function simultaneously from multiple threads.
124 
125     Args:
126 
127         fn (sync function): Performs arbitrary blocking work.
128 
129         deliver (sync function): Takes the `outcome.Outcome` of ``fn``, and
130           delivers it. *Must not block.*
131 
132     Because worker threads are cached and reused for multiple calls, neither
133     function should mutate thread-level state, like `threading.local` objects
134     – or if they do, they should be careful to revert their changes before
135     returning.
136 
137     Note:
138 
139         The split between ``fn`` and ``deliver`` serves two purposes. First,
140         it's convenient, since most callers need something like this anyway.
141 
142         Second, it avoids a small race condition that could cause too many
143         threads to be spawned. Consider a program that wants to run several
144         jobs sequentially on a thread, so the main thread submits a job, waits
145         for it to finish, submits another job, etc. In theory, this program
146         should only need one worker thread. But what could happen is:
147 
148         1. Worker thread: First job finishes, and calls ``deliver``.
149 
150         2. Main thread: receives notification that the job finished, and calls
151            ``start_thread_soon``.
152 
153         3. Main thread: sees that no worker threads are marked idle, so spawns
154            a second worker thread.
155 
156         4. Original worker thread: marks itself as idle.
157 
158         To avoid this, threads mark themselves as idle *before* calling
159         ``deliver``.
160 
161         Is this potential extra thread a major problem? Maybe not, but it's
162         easy enough to avoid, and we figure that if the user is trying to
163         limit how many threads they're using then it's polite to respect that.
164 
165     """
166     THREAD_CACHE.start_thread_soon(fn, deliver)
167
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/trio/_core/_thread_cache.py b/trio/_core/_thread_cache.py
--- a/trio/_core/_thread_cache.py
+++ b/trio/_core/_thread_cache.py
@@ -69,6 +69,8 @@
                 # instead of spawning a new thread.
                 self._thread_cache._idle_workers[self] = None
                 deliver(result)
+                del fn
+                del deliver
             else:
                 # Timeout acquiring lock, so we can probably exit. But,
                 # there's a race condition: we might be assigned a job *just*

{"golden_diff": "diff --git a/trio/_core/_thread_cache.py b/trio/_core/_thread_cache.py\n--- a/trio/_core/_thread_cache.py\n+++ b/trio/_core/_thread_cache.py\n@@ -69,6 +69,8 @@\n                 # instead of spawning a new thread.\n                 self._thread_cache._idle_workers[self] = None\n                 deliver(result)\n+                del fn\n+                del deliver\n             else:\n                 # Timeout acquiring lock, so we can probably exit. But,\n                 # there's a race condition: we might be assigned a job *just*\n", "issue": "to_thread.run_sync() foils meta-class destructors \nInvoking `to_thread.run_sync()` seems to prevent meta-class destructors of completely unrelated classes from being called on shutdown.\r\n\r\n```python\r\nimport trio\r\n\r\nclass deltype(type):\r\n    def __del__(self):\r\n        print(f\"{self.__name__} class deleted\")\r\n\r\nTest = deltype(\"Test\", (object,), {})\r\n\r\nasync def async_main():\r\n    await trio.to_thread.run_sync(lambda: print('hi'))\r\n    await trio.sleep(0)\r\n\r\ntrio.run(async_main)\r\n```\r\n\r\nexpected output:\r\n```\r\nhi\r\nTest class deleted\r\n```\r\n\r\nactual output:\r\n```\r\nhi\r\n```\r\n\r\nif the `run_sync()` is commented out, the meta-class destructor is called as expected:\r\n```\r\nTest class deleted\r\n```\r\n\n", "before_files": [{"content": "from threading import Thread, Lock\nimport outcome\nfrom itertools import count\n\n# The \"thread cache\" is a simple unbounded thread pool, i.e., it automatically\n# spawns as many threads as needed to handle all the requests its given. Its\n# only purpose is to cache worker threads so that they don't have to be\n# started from scratch every time we want to delegate some work to a thread.\n# It's expected that some higher-level code will track how many threads are in\n# use to avoid overwhelming the system (e.g. the limiter= argument to\n# trio.to_thread.run_sync).\n#\n# To maximize sharing, there's only one thread cache per process, even if you\n# have multiple calls to trio.run.\n#\n# Guarantees:\n#\n# It's safe to call start_thread_soon simultaneously from\n# multiple threads.\n#\n# Idle threads are chosen in LIFO order, i.e. we *don't* spread work evenly\n# over all threads. Instead we try to let some threads do most of the work\n# while others sit idle as much as possible. Compared to FIFO, this has better\n# memory cache behavior, and it makes it easier to detect when we have too\n# many threads, so idle ones can exit.\n#\n# This code assumes that 'dict' has the following properties:\n#\n# - __setitem__, __delitem__, and popitem are all thread-safe and atomic with\n#   respect to each other. This is guaranteed by the GIL.\n#\n# - popitem returns the most-recently-added item (i.e., __setitem__ + popitem\n#   give you a LIFO queue). This relies on dicts being insertion-ordered, like\n#   they are in py36+.\n\n# How long a thread will idle waiting for new work before gives up and exits.\n# This value is pretty arbitrary; I don't think it matters too much.\nIDLE_TIMEOUT = 10  # seconds\n\nname_counter = count()\n\n\nclass WorkerThread:\n    def __init__(self, thread_cache):\n        self._job = None\n        self._thread_cache = thread_cache\n        # This Lock is used in an unconventional way.\n        #\n        # \"Unlocked\" means we have a pending job that's been assigned to us;\n        # \"locked\" means that we don't.\n        #\n        # Initially we have no job, so it starts out in locked state.\n        self._worker_lock = Lock()\n        self._worker_lock.acquire()\n        thread = Thread(target=self._work, daemon=True)\n        thread.name = f\"Trio worker thread {next(name_counter)}\"\n        thread.start()\n\n    def _work(self):\n        while True:\n            if self._worker_lock.acquire(timeout=IDLE_TIMEOUT):\n                # We got a job\n                fn, deliver = self._job\n                self._job = None\n                result = outcome.capture(fn)\n                # Tell the cache that we're available to be assigned a new\n                # job. We do this *before* calling 'deliver', so that if\n                # 'deliver' triggers a new job, it can be assigned to us\n                # instead of spawning a new thread.\n                self._thread_cache._idle_workers[self] = None\n                deliver(result)\n            else:\n                # Timeout acquiring lock, so we can probably exit. But,\n                # there's a race condition: we might be assigned a job *just*\n                # as we're about to exit. So we have to check.\n                try:\n                    del self._thread_cache._idle_workers[self]\n                except KeyError:\n                    # Someone else removed us from the idle worker queue, so\n                    # they must be in the process of assigning us a job - loop\n                    # around and wait for it.\n                    continue\n                else:\n                    # We successfully removed ourselves from the idle\n                    # worker queue, so no more jobs are incoming; it's safe to\n                    # exit.\n                    return\n\n\nclass ThreadCache:\n    def __init__(self):\n        self._idle_workers = {}\n\n    def start_thread_soon(self, fn, deliver):\n        try:\n            worker, _ = self._idle_workers.popitem()\n        except KeyError:\n            worker = WorkerThread(self)\n        worker._job = (fn, deliver)\n        worker._worker_lock.release()\n\n\nTHREAD_CACHE = ThreadCache()\n\n\ndef start_thread_soon(fn, deliver):\n    \"\"\"Runs ``deliver(outcome.capture(fn))`` in a worker thread.\n\n    Generally ``fn`` does some blocking work, and ``deliver`` delivers the\n    result back to whoever is interested.\n\n    This is a low-level, no-frills interface, very similar to using\n    `threading.Thread` to spawn a thread directly. The main difference is\n    that this function tries to re-use threads when possible, so it can be\n    a bit faster than `threading.Thread`.\n\n    Worker threads have the `~threading.Thread.daemon` flag set, which means\n    that if your main thread exits, worker threads will automatically be\n    killed. If you want to make sure that your ``fn`` runs to completion, then\n    you should make sure that the main thread remains alive until ``deliver``\n    is called.\n\n    It is safe to call this function simultaneously from multiple threads.\n\n    Args:\n\n        fn (sync function): Performs arbitrary blocking work.\n\n        deliver (sync function): Takes the `outcome.Outcome` of ``fn``, and\n          delivers it. *Must not block.*\n\n    Because worker threads are cached and reused for multiple calls, neither\n    function should mutate thread-level state, like `threading.local` objects\n    \u2013 or if they do, they should be careful to revert their changes before\n    returning.\n\n    Note:\n\n        The split between ``fn`` and ``deliver`` serves two purposes. First,\n        it's convenient, since most callers need something like this anyway.\n\n        Second, it avoids a small race condition that could cause too many\n        threads to be spawned. Consider a program that wants to run several\n        jobs sequentially on a thread, so the main thread submits a job, waits\n        for it to finish, submits another job, etc. In theory, this program\n        should only need one worker thread. But what could happen is:\n\n        1. Worker thread: First job finishes, and calls ``deliver``.\n\n        2. Main thread: receives notification that the job finished, and calls\n           ``start_thread_soon``.\n\n        3. Main thread: sees that no worker threads are marked idle, so spawns\n           a second worker thread.\n\n        4. Original worker thread: marks itself as idle.\n\n        To avoid this, threads mark themselves as idle *before* calling\n        ``deliver``.\n\n        Is this potential extra thread a major problem? Maybe not, but it's\n        easy enough to avoid, and we figure that if the user is trying to\n        limit how many threads they're using then it's polite to respect that.\n\n    \"\"\"\n    THREAD_CACHE.start_thread_soon(fn, deliver)\n", "path": "trio/_core/_thread_cache.py"}], "after_files": [{"content": "from threading import Thread, Lock\nimport outcome\nfrom itertools import count\n\n# The \"thread cache\" is a simple unbounded thread pool, i.e., it automatically\n# spawns as many threads as needed to handle all the requests its given. Its\n# only purpose is to cache worker threads so that they don't have to be\n# started from scratch every time we want to delegate some work to a thread.\n# It's expected that some higher-level code will track how many threads are in\n# use to avoid overwhelming the system (e.g. the limiter= argument to\n# trio.to_thread.run_sync).\n#\n# To maximize sharing, there's only one thread cache per process, even if you\n# have multiple calls to trio.run.\n#\n# Guarantees:\n#\n# It's safe to call start_thread_soon simultaneously from\n# multiple threads.\n#\n# Idle threads are chosen in LIFO order, i.e. we *don't* spread work evenly\n# over all threads. Instead we try to let some threads do most of the work\n# while others sit idle as much as possible. Compared to FIFO, this has better\n# memory cache behavior, and it makes it easier to detect when we have too\n# many threads, so idle ones can exit.\n#\n# This code assumes that 'dict' has the following properties:\n#\n# - __setitem__, __delitem__, and popitem are all thread-safe and atomic with\n#   respect to each other. This is guaranteed by the GIL.\n#\n# - popitem returns the most-recently-added item (i.e., __setitem__ + popitem\n#   give you a LIFO queue). This relies on dicts being insertion-ordered, like\n#   they are in py36+.\n\n# How long a thread will idle waiting for new work before gives up and exits.\n# This value is pretty arbitrary; I don't think it matters too much.\nIDLE_TIMEOUT = 10  # seconds\n\nname_counter = count()\n\n\nclass WorkerThread:\n    def __init__(self, thread_cache):\n        self._job = None\n        self._thread_cache = thread_cache\n        # This Lock is used in an unconventional way.\n        #\n        # \"Unlocked\" means we have a pending job that's been assigned to us;\n        # \"locked\" means that we don't.\n        #\n        # Initially we have no job, so it starts out in locked state.\n        self._worker_lock = Lock()\n        self._worker_lock.acquire()\n        thread = Thread(target=self._work, daemon=True)\n        thread.name = f\"Trio worker thread {next(name_counter)}\"\n        thread.start()\n\n    def _work(self):\n        while True:\n            if self._worker_lock.acquire(timeout=IDLE_TIMEOUT):\n                # We got a job\n                fn, deliver = self._job\n                self._job = None\n                result = outcome.capture(fn)\n                # Tell the cache that we're available to be assigned a new\n                # job. We do this *before* calling 'deliver', so that if\n                # 'deliver' triggers a new job, it can be assigned to us\n                # instead of spawning a new thread.\n                self._thread_cache._idle_workers[self] = None\n                deliver(result)\n                del fn\n                del deliver\n            else:\n                # Timeout acquiring lock, so we can probably exit. But,\n                # there's a race condition: we might be assigned a job *just*\n                # as we're about to exit. So we have to check.\n                try:\n                    del self._thread_cache._idle_workers[self]\n                except KeyError:\n                    # Someone else removed us from the idle worker queue, so\n                    # they must be in the process of assigning us a job - loop\n                    # around and wait for it.\n                    continue\n                else:\n                    # We successfully removed ourselves from the idle\n                    # worker queue, so no more jobs are incoming; it's safe to\n                    # exit.\n                    return\n\n\nclass ThreadCache:\n    def __init__(self):\n        self._idle_workers = {}\n\n    def start_thread_soon(self, fn, deliver):\n        try:\n            worker, _ = self._idle_workers.popitem()\n        except KeyError:\n            worker = WorkerThread(self)\n        worker._job = (fn, deliver)\n        worker._worker_lock.release()\n\n\nTHREAD_CACHE = ThreadCache()\n\n\ndef start_thread_soon(fn, deliver):\n    \"\"\"Runs ``deliver(outcome.capture(fn))`` in a worker thread.\n\n    Generally ``fn`` does some blocking work, and ``deliver`` delivers the\n    result back to whoever is interested.\n\n    This is a low-level, no-frills interface, very similar to using\n    `threading.Thread` to spawn a thread directly. The main difference is\n    that this function tries to re-use threads when possible, so it can be\n    a bit faster than `threading.Thread`.\n\n    Worker threads have the `~threading.Thread.daemon` flag set, which means\n    that if your main thread exits, worker threads will automatically be\n    killed. If you want to make sure that your ``fn`` runs to completion, then\n    you should make sure that the main thread remains alive until ``deliver``\n    is called.\n\n    It is safe to call this function simultaneously from multiple threads.\n\n    Args:\n\n        fn (sync function): Performs arbitrary blocking work.\n\n        deliver (sync function): Takes the `outcome.Outcome` of ``fn``, and\n          delivers it. *Must not block.*\n\n    Because worker threads are cached and reused for multiple calls, neither\n    function should mutate thread-level state, like `threading.local` objects\n    \u2013 or if they do, they should be careful to revert their changes before\n    returning.\n\n    Note:\n\n        The split between ``fn`` and ``deliver`` serves two purposes. First,\n        it's convenient, since most callers need something like this anyway.\n\n        Second, it avoids a small race condition that could cause too many\n        threads to be spawned. Consider a program that wants to run several\n        jobs sequentially on a thread, so the main thread submits a job, waits\n        for it to finish, submits another job, etc. In theory, this program\n        should only need one worker thread. But what could happen is:\n\n        1. Worker thread: First job finishes, and calls ``deliver``.\n\n        2. Main thread: receives notification that the job finished, and calls\n           ``start_thread_soon``.\n\n        3. Main thread: sees that no worker threads are marked idle, so spawns\n           a second worker thread.\n\n        4. Original worker thread: marks itself as idle.\n\n        To avoid this, threads mark themselves as idle *before* calling\n        ``deliver``.\n\n        Is this potential extra thread a major problem? Maybe not, but it's\n        easy enough to avoid, and we figure that if the user is trying to\n        limit how many threads they're using then it's polite to respect that.\n\n    \"\"\"\n    THREAD_CACHE.start_thread_soon(fn, deliver)\n", "path": "trio/_core/_thread_cache.py"}]}