We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Add ability to force sync slicing
# Description

The `LayerSlicer` should perform asynchronous slicing when all layers being sliced are asynchronously sliceable. In some cases, we might want to temporarily force slicing to be synchronous. For example, we might consider usage like

```python
with viewer.force_sync():
    viewer.dims.current_step = (5, 0, 0)
image = viewer.screenshot()
```

where we need to wait for slicing to finish before taking a screenshot.

This task should define public ways to allow synchronous slicing to be forced. This should include a temporary force with the context manager above, and may also include forcing sync slicing for a particular run/session of napari (so that plugins can easily force it once). We may also consider adding a setting that causes sync slicing to always be forced.

## Update

Kim: I started to add this to the initial base slicer layer PR, but ended up taking it out. Its not much code, but might be useful for reference. https://github.com/napari/napari/pull/5170/commits/56d4244112d0bc001e5f82ae31a7e5058cca1e51

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `napari/components/_layer_slicer.py`
Content:
```
1 from __future__ import annotations
2 
3 import logging
4 from concurrent.futures import Executor, Future, ThreadPoolExecutor
5 from threading import RLock
6 from typing import (
7     Callable,
8     Dict,
9     Iterable,
10     Optional,
11     Protocol,
12     Tuple,
13     TypeVar,
14     runtime_checkable,
15 )
16 
17 from napari.components import Dims
18 from napari.layers import Layer
19 from napari.utils.events.event import EmitterGroup, Event
20 
21 logger = logging.getLogger("napari.components._layer_slicer")
22 
23 
24 # Layers that can be asynchronously sliced must be able to make
25 # a slice request that can be called and will produce a slice
26 # response. The request and response types will vary per layer
27 # type, which means that the values of the dictionary result of
28 # ``_slice_layers`` cannot be fixed to a single type.
29 
30 _SliceResponse = TypeVar('_SliceResponse')
31 _SliceRequest = Callable[[], _SliceResponse]
32 
33 
34 @runtime_checkable
35 class _AsyncSliceable(Protocol[_SliceResponse]):
36     def _make_slice_request(self, dims: Dims) -> _SliceRequest[_SliceResponse]:
37         ...
38 
39 
40 class _LayerSlicer:
41     """
42     High level class to control the creation of a slice (via a slice request),
43     submit it (synchronously or asynchronously) to a thread pool, and emit the
44     results when complete.
45 
46     Events
47     ------
48     ready
49         emitted after slicing is done with a dict value that maps from layer
50         to slice response. Note that this may be emitted on the main or
51         a non-main thread. If usage of this event relies on something happening
52         on the main thread, actions should be taken to ensure that the callback
53         is also executed on the main thread (e.g. by decorating the callback
54         with `@ensure_main_thread`).
55     """
56 
57     def __init__(self):
58         """
59         Attributes
60         ----------
61         _executor : concurrent.futures.ThreadPoolExecutor
62             manager for the slicing threading
63         _layers_to_task : dict
64             task storage for cancellation logic
65         _lock_layers_to_task : threading.RLock
66             lock to guard against changes to `_layers_to_task` when finding,
67             adding, or removing tasks.
68         """
69         self.events = EmitterGroup(source=self, ready=Event)
70         self._executor: Executor = ThreadPoolExecutor(max_workers=1)
71         self._layers_to_task: Dict[Tuple[Layer], Future] = {}
72         self._lock_layers_to_task = RLock()
73 
74     def slice_layers_async(
75         self, layers: Iterable[Layer], dims: Dims
76     ) -> Future[dict]:
77         """This should only be called from the main thread.
78 
79         Creates a new task and adds it to the _layers_to_task dict. Cancels
80         all tasks currently pending for that layer tuple.
81 
82         Submitting multiple layers at one generates multiple requests (stored in a dict),
83         but only ONE task.
84 
85         If multiple layers are sliced, any task that contains only one of those
86         layers can safely be cancelled. If a single layer is sliced, it will
87         wait for any existing tasks that include that layer AND another layer,
88         In other words, it will only cancel if the new task will replace the
89         slices of all the layers in the pending task.
90         """
91         # Cancel any tasks that are slicing a subset of the layers
92         # being sliced now. This allows us to slice arbitrary sets of
93         # layers with some sensible and not too complex cancellation
94         # policy.
95         if existing_task := self._find_existing_task(layers):
96             logger.debug('Cancelling task for %s', layers)
97             existing_task.cancel()
98 
99         # Not all layer types will initially be asynchronously sliceable.
100         # The following logic gives us a way to handle those in the short
101         # term as we develop, and also in the long term if there are cases
102         # when we want to perform sync slicing anyway.
103         requests = {}
104         for layer in layers:
105             if isinstance(layer, _AsyncSliceable):
106                 requests[layer] = layer._make_slice_request(dims)
107             else:
108                 layer._slice_dims(dims.point, dims.ndisplay, dims.order)
109         # create task for slicing of each request/layer
110         task = self._executor.submit(self._slice_layers, requests)
111 
112         # store task for cancellation logic
113         # this is purposefully done before adding the done callback to ensure
114         # that the task is added before the done callback can be executed
115         with self._lock_layers_to_task:
116             self._layers_to_task[tuple(requests)] = task
117 
118         task.add_done_callback(self._on_slice_done)
119 
120         return task
121 
122     def shutdown(self) -> None:
123         """Should be called from the main thread when this is no longer needed."""
124         self._executor.shutdown()
125 
126     def _slice_layers(self, requests: Dict) -> Dict:
127         """
128         Iterates through a dictionary of request objects and call the slice
129         on each individual layer. Can be called from the main or slicing thread.
130 
131         Attributes
132         ----------
133         requests: dict[Layer, SliceRequest]
134             Dictionary of request objects to be used for constructing the slice
135 
136         Returns
137         -------
138         dict[Layer, SliceResponse]: which contains the results of the slice
139         """
140         return {layer: request() for layer, request in requests.items()}
141 
142     def _on_slice_done(self, task: Future[Dict]) -> None:
143         """
144         This is the "done_callback" which is added to each task.
145         Can be called from the main or slicing thread.
146         """
147         if not self._try_to_remove_task(task):
148             logger.debug('Task not found')
149             return
150 
151         if task.cancelled():
152             logger.debug('Cancelled task')
153             return
154         result = task.result()
155         self.events.ready(Event('ready', value=result))
156 
157     def _try_to_remove_task(self, task: Future[Dict]) -> bool:
158         """
159         Attempt to remove task, return false if task not found, return true
160         if task is found and removed from layers_to_task dict.
161 
162         This function provides a lock to ensure that the layers_to_task dict
163         is unmodified during this process.
164         """
165         with self._lock_layers_to_task:
166             for k_layers, v_task in self._layers_to_task.items():
167                 if v_task == task:
168                     del self._layers_to_task[k_layers]
169                     return True
170         return False
171 
172     def _find_existing_task(
173         self, layers: Iterable[Layer]
174     ) -> Optional[Future[Dict]]:
175         """Find the task associated with a list of layers. Returns the first
176         task found for which the layers of the task are a subset of the input
177         layers.
178 
179         This function provides a lock to ensure that the layers_to_task dict
180         is unmodified during this process.
181         """
182         with self._lock_layers_to_task:
183             layer_set = set(layers)
184             for task_layers, task in self._layers_to_task.items():
185                 if set(task_layers).issubset(layer_set):
186                     logger.debug(f'Found existing task for {task_layers}')
187                     return task
188         return None
189
```
--- 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/napari/components/_layer_slicer.py b/napari/components/_layer_slicer.py
--- a/napari/components/_layer_slicer.py
+++ b/napari/components/_layer_slicer.py
@@ -1,7 +1,8 @@
 from __future__ import annotations
 
 import logging
-from concurrent.futures import Executor, Future, ThreadPoolExecutor
+from concurrent.futures import Executor, Future, ThreadPoolExecutor, wait
+from contextlib import contextmanager
 from threading import RLock
 from typing import (
     Callable,
@@ -60,6 +61,8 @@
         ----------
         _executor : concurrent.futures.ThreadPoolExecutor
             manager for the slicing threading
+        _force_sync: bool
+            if true, forces slicing to execute synchronously
         _layers_to_task : dict
             task storage for cancellation logic
         _lock_layers_to_task : threading.RLock
@@ -68,9 +71,49 @@
         """
         self.events = EmitterGroup(source=self, ready=Event)
         self._executor: Executor = ThreadPoolExecutor(max_workers=1)
+        self._force_sync = False
         self._layers_to_task: Dict[Tuple[Layer], Future] = {}
         self._lock_layers_to_task = RLock()
 
+    @contextmanager
+    def force_sync(self):
+        """Context manager to temporarily force slicing to be synchronous.
+        This should only be used from the main thread.
+
+        >>> layer_slicer = _LayerSlicer()
+        >>> layer = Image(...)  # an async-ready layer
+        >>> with layer_slice.force_sync():
+        >>>     layer_slicer.slice_layers_async(layers=[layer], dims=Dims())
+        """
+        prev = self._force_sync
+        self._force_sync = True
+        try:
+            yield None
+        finally:
+            self._force_sync = prev
+
+    def wait_until_idle(self, timeout: Optional[float] = None) -> None:
+        """Wait for all slicing tasks to complete before returning.
+
+        Attributes
+        ----------
+        timeout: float or None
+            (Optional) time in seconds to wait before raising TimeoutError. If set as None,
+            there is no limit to the wait time. Defaults to None
+
+        Raises
+        ------
+        TimeoutError: when the timeout limit has been exceeded and the task is
+            not yet complete
+        """
+        futures = self._layers_to_task.values()
+        _, not_done_futures = wait(futures, timeout=timeout)
+
+        if len(not_done_futures) > 0:
+            raise TimeoutError(
+                f'Slicing {len(not_done_futures)} tasks did not complete within timeout ({timeout}s).'
+            )
+
     def slice_layers_async(
         self, layers: Iterable[Layer], dims: Dims
     ) -> Future[dict]:
@@ -101,14 +144,20 @@
         # term as we develop, and also in the long term if there are cases
         # when we want to perform sync slicing anyway.
         requests = {}
+        sync_layers = []
         for layer in layers:
-            if isinstance(layer, _AsyncSliceable):
+            if isinstance(layer, _AsyncSliceable) and not self._force_sync:
                 requests[layer] = layer._make_slice_request(dims)
             else:
-                layer._slice_dims(dims.point, dims.ndisplay, dims.order)
+                sync_layers.append(layer)
+
         # create task for slicing of each request/layer
         task = self._executor.submit(self._slice_layers, requests)
 
+        # submit the sync layers (purposefully placed after async submission)
+        for layer in sync_layers:
+            layer._slice_dims(dims.point, dims.ndisplay, dims.order)
+
         # store task for cancellation logic
         # this is purposefully done before adding the done callback to ensure
         # that the task is added before the done callback can be executed

{"golden_diff": "diff --git a/napari/components/_layer_slicer.py b/napari/components/_layer_slicer.py\n--- a/napari/components/_layer_slicer.py\n+++ b/napari/components/_layer_slicer.py\n@@ -1,7 +1,8 @@\n from __future__ import annotations\n \n import logging\n-from concurrent.futures import Executor, Future, ThreadPoolExecutor\n+from concurrent.futures import Executor, Future, ThreadPoolExecutor, wait\n+from contextlib import contextmanager\n from threading import RLock\n from typing import (\n     Callable,\n@@ -60,6 +61,8 @@\n         ----------\n         _executor : concurrent.futures.ThreadPoolExecutor\n             manager for the slicing threading\n+        _force_sync: bool\n+            if true, forces slicing to execute synchronously\n         _layers_to_task : dict\n             task storage for cancellation logic\n         _lock_layers_to_task : threading.RLock\n@@ -68,9 +71,49 @@\n         \"\"\"\n         self.events = EmitterGroup(source=self, ready=Event)\n         self._executor: Executor = ThreadPoolExecutor(max_workers=1)\n+        self._force_sync = False\n         self._layers_to_task: Dict[Tuple[Layer], Future] = {}\n         self._lock_layers_to_task = RLock()\n \n+    @contextmanager\n+    def force_sync(self):\n+        \"\"\"Context manager to temporarily force slicing to be synchronous.\n+        This should only be used from the main thread.\n+\n+        >>> layer_slicer = _LayerSlicer()\n+        >>> layer = Image(...)  # an async-ready layer\n+        >>> with layer_slice.force_sync():\n+        >>>     layer_slicer.slice_layers_async(layers=[layer], dims=Dims())\n+        \"\"\"\n+        prev = self._force_sync\n+        self._force_sync = True\n+        try:\n+            yield None\n+        finally:\n+            self._force_sync = prev\n+\n+    def wait_until_idle(self, timeout: Optional[float] = None) -> None:\n+        \"\"\"Wait for all slicing tasks to complete before returning.\n+\n+        Attributes\n+        ----------\n+        timeout: float or None\n+            (Optional) time in seconds to wait before raising TimeoutError. If set as None,\n+            there is no limit to the wait time. Defaults to None\n+\n+        Raises\n+        ------\n+        TimeoutError: when the timeout limit has been exceeded and the task is\n+            not yet complete\n+        \"\"\"\n+        futures = self._layers_to_task.values()\n+        _, not_done_futures = wait(futures, timeout=timeout)\n+\n+        if len(not_done_futures) > 0:\n+            raise TimeoutError(\n+                f'Slicing {len(not_done_futures)} tasks did not complete within timeout ({timeout}s).'\n+            )\n+\n     def slice_layers_async(\n         self, layers: Iterable[Layer], dims: Dims\n     ) -> Future[dict]:\n@@ -101,14 +144,20 @@\n         # term as we develop, and also in the long term if there are cases\n         # when we want to perform sync slicing anyway.\n         requests = {}\n+        sync_layers = []\n         for layer in layers:\n-            if isinstance(layer, _AsyncSliceable):\n+            if isinstance(layer, _AsyncSliceable) and not self._force_sync:\n                 requests[layer] = layer._make_slice_request(dims)\n             else:\n-                layer._slice_dims(dims.point, dims.ndisplay, dims.order)\n+                sync_layers.append(layer)\n+\n         # create task for slicing of each request/layer\n         task = self._executor.submit(self._slice_layers, requests)\n \n+        # submit the sync layers (purposefully placed after async submission)\n+        for layer in sync_layers:\n+            layer._slice_dims(dims.point, dims.ndisplay, dims.order)\n+\n         # store task for cancellation logic\n         # this is purposefully done before adding the done callback to ensure\n         # that the task is added before the done callback can be executed\n", "issue": "Add ability to force sync slicing\n# Description\n\nThe `LayerSlicer` should perform asynchronous slicing when all layers being sliced are asynchronously sliceable. In some cases, we might want to temporarily force slicing to be synchronous. For example, we might consider usage like\n\n```python\nwith viewer.force_sync():\n    viewer.dims.current_step = (5, 0, 0)\nimage = viewer.screenshot()\n```\n\nwhere we need to wait for slicing to finish before taking a screenshot.\n\nThis task should define public ways to allow synchronous slicing to be forced. This should include a temporary force with the context manager above, and may also include forcing sync slicing for a particular run/session of napari (so that plugins can easily force it once). We may also consider adding a setting that causes sync slicing to always be forced.\n\n## Update\n\nKim: I started to add this to the initial base slicer layer PR, but ended up taking it out. Its not much code, but might be useful for reference. https://github.com/napari/napari/pull/5170/commits/56d4244112d0bc001e5f82ae31a7e5058cca1e51\n", "before_files": [{"content": "from __future__ import annotations\n\nimport logging\nfrom concurrent.futures import Executor, Future, ThreadPoolExecutor\nfrom threading import RLock\nfrom typing import (\n    Callable,\n    Dict,\n    Iterable,\n    Optional,\n    Protocol,\n    Tuple,\n    TypeVar,\n    runtime_checkable,\n)\n\nfrom napari.components import Dims\nfrom napari.layers import Layer\nfrom napari.utils.events.event import EmitterGroup, Event\n\nlogger = logging.getLogger(\"napari.components._layer_slicer\")\n\n\n# Layers that can be asynchronously sliced must be able to make\n# a slice request that can be called and will produce a slice\n# response. The request and response types will vary per layer\n# type, which means that the values of the dictionary result of\n# ``_slice_layers`` cannot be fixed to a single type.\n\n_SliceResponse = TypeVar('_SliceResponse')\n_SliceRequest = Callable[[], _SliceResponse]\n\n\n@runtime_checkable\nclass _AsyncSliceable(Protocol[_SliceResponse]):\n    def _make_slice_request(self, dims: Dims) -> _SliceRequest[_SliceResponse]:\n        ...\n\n\nclass _LayerSlicer:\n    \"\"\"\n    High level class to control the creation of a slice (via a slice request),\n    submit it (synchronously or asynchronously) to a thread pool, and emit the\n    results when complete.\n\n    Events\n    ------\n    ready\n        emitted after slicing is done with a dict value that maps from layer\n        to slice response. Note that this may be emitted on the main or\n        a non-main thread. If usage of this event relies on something happening\n        on the main thread, actions should be taken to ensure that the callback\n        is also executed on the main thread (e.g. by decorating the callback\n        with `@ensure_main_thread`).\n    \"\"\"\n\n    def __init__(self):\n        \"\"\"\n        Attributes\n        ----------\n        _executor : concurrent.futures.ThreadPoolExecutor\n            manager for the slicing threading\n        _layers_to_task : dict\n            task storage for cancellation logic\n        _lock_layers_to_task : threading.RLock\n            lock to guard against changes to `_layers_to_task` when finding,\n            adding, or removing tasks.\n        \"\"\"\n        self.events = EmitterGroup(source=self, ready=Event)\n        self._executor: Executor = ThreadPoolExecutor(max_workers=1)\n        self._layers_to_task: Dict[Tuple[Layer], Future] = {}\n        self._lock_layers_to_task = RLock()\n\n    def slice_layers_async(\n        self, layers: Iterable[Layer], dims: Dims\n    ) -> Future[dict]:\n        \"\"\"This should only be called from the main thread.\n\n        Creates a new task and adds it to the _layers_to_task dict. Cancels\n        all tasks currently pending for that layer tuple.\n\n        Submitting multiple layers at one generates multiple requests (stored in a dict),\n        but only ONE task.\n\n        If multiple layers are sliced, any task that contains only one of those\n        layers can safely be cancelled. If a single layer is sliced, it will\n        wait for any existing tasks that include that layer AND another layer,\n        In other words, it will only cancel if the new task will replace the\n        slices of all the layers in the pending task.\n        \"\"\"\n        # Cancel any tasks that are slicing a subset of the layers\n        # being sliced now. This allows us to slice arbitrary sets of\n        # layers with some sensible and not too complex cancellation\n        # policy.\n        if existing_task := self._find_existing_task(layers):\n            logger.debug('Cancelling task for %s', layers)\n            existing_task.cancel()\n\n        # Not all layer types will initially be asynchronously sliceable.\n        # The following logic gives us a way to handle those in the short\n        # term as we develop, and also in the long term if there are cases\n        # when we want to perform sync slicing anyway.\n        requests = {}\n        for layer in layers:\n            if isinstance(layer, _AsyncSliceable):\n                requests[layer] = layer._make_slice_request(dims)\n            else:\n                layer._slice_dims(dims.point, dims.ndisplay, dims.order)\n        # create task for slicing of each request/layer\n        task = self._executor.submit(self._slice_layers, requests)\n\n        # store task for cancellation logic\n        # this is purposefully done before adding the done callback to ensure\n        # that the task is added before the done callback can be executed\n        with self._lock_layers_to_task:\n            self._layers_to_task[tuple(requests)] = task\n\n        task.add_done_callback(self._on_slice_done)\n\n        return task\n\n    def shutdown(self) -> None:\n        \"\"\"Should be called from the main thread when this is no longer needed.\"\"\"\n        self._executor.shutdown()\n\n    def _slice_layers(self, requests: Dict) -> Dict:\n        \"\"\"\n        Iterates through a dictionary of request objects and call the slice\n        on each individual layer. Can be called from the main or slicing thread.\n\n        Attributes\n        ----------\n        requests: dict[Layer, SliceRequest]\n            Dictionary of request objects to be used for constructing the slice\n\n        Returns\n        -------\n        dict[Layer, SliceResponse]: which contains the results of the slice\n        \"\"\"\n        return {layer: request() for layer, request in requests.items()}\n\n    def _on_slice_done(self, task: Future[Dict]) -> None:\n        \"\"\"\n        This is the \"done_callback\" which is added to each task.\n        Can be called from the main or slicing thread.\n        \"\"\"\n        if not self._try_to_remove_task(task):\n            logger.debug('Task not found')\n            return\n\n        if task.cancelled():\n            logger.debug('Cancelled task')\n            return\n        result = task.result()\n        self.events.ready(Event('ready', value=result))\n\n    def _try_to_remove_task(self, task: Future[Dict]) -> bool:\n        \"\"\"\n        Attempt to remove task, return false if task not found, return true\n        if task is found and removed from layers_to_task dict.\n\n        This function provides a lock to ensure that the layers_to_task dict\n        is unmodified during this process.\n        \"\"\"\n        with self._lock_layers_to_task:\n            for k_layers, v_task in self._layers_to_task.items():\n                if v_task == task:\n                    del self._layers_to_task[k_layers]\n                    return True\n        return False\n\n    def _find_existing_task(\n        self, layers: Iterable[Layer]\n    ) -> Optional[Future[Dict]]:\n        \"\"\"Find the task associated with a list of layers. Returns the first\n        task found for which the layers of the task are a subset of the input\n        layers.\n\n        This function provides a lock to ensure that the layers_to_task dict\n        is unmodified during this process.\n        \"\"\"\n        with self._lock_layers_to_task:\n            layer_set = set(layers)\n            for task_layers, task in self._layers_to_task.items():\n                if set(task_layers).issubset(layer_set):\n                    logger.debug(f'Found existing task for {task_layers}')\n                    return task\n        return None\n", "path": "napari/components/_layer_slicer.py"}], "after_files": [{"content": "from __future__ import annotations\n\nimport logging\nfrom concurrent.futures import Executor, Future, ThreadPoolExecutor, wait\nfrom contextlib import contextmanager\nfrom threading import RLock\nfrom typing import (\n    Callable,\n    Dict,\n    Iterable,\n    Optional,\n    Protocol,\n    Tuple,\n    TypeVar,\n    runtime_checkable,\n)\n\nfrom napari.components import Dims\nfrom napari.layers import Layer\nfrom napari.utils.events.event import EmitterGroup, Event\n\nlogger = logging.getLogger(\"napari.components._layer_slicer\")\n\n\n# Layers that can be asynchronously sliced must be able to make\n# a slice request that can be called and will produce a slice\n# response. The request and response types will vary per layer\n# type, which means that the values of the dictionary result of\n# ``_slice_layers`` cannot be fixed to a single type.\n\n_SliceResponse = TypeVar('_SliceResponse')\n_SliceRequest = Callable[[], _SliceResponse]\n\n\n@runtime_checkable\nclass _AsyncSliceable(Protocol[_SliceResponse]):\n    def _make_slice_request(self, dims: Dims) -> _SliceRequest[_SliceResponse]:\n        ...\n\n\nclass _LayerSlicer:\n    \"\"\"\n    High level class to control the creation of a slice (via a slice request),\n    submit it (synchronously or asynchronously) to a thread pool, and emit the\n    results when complete.\n\n    Events\n    ------\n    ready\n        emitted after slicing is done with a dict value that maps from layer\n        to slice response. Note that this may be emitted on the main or\n        a non-main thread. If usage of this event relies on something happening\n        on the main thread, actions should be taken to ensure that the callback\n        is also executed on the main thread (e.g. by decorating the callback\n        with `@ensure_main_thread`).\n    \"\"\"\n\n    def __init__(self):\n        \"\"\"\n        Attributes\n        ----------\n        _executor : concurrent.futures.ThreadPoolExecutor\n            manager for the slicing threading\n        _force_sync: bool\n            if true, forces slicing to execute synchronously\n        _layers_to_task : dict\n            task storage for cancellation logic\n        _lock_layers_to_task : threading.RLock\n            lock to guard against changes to `_layers_to_task` when finding,\n            adding, or removing tasks.\n        \"\"\"\n        self.events = EmitterGroup(source=self, ready=Event)\n        self._executor: Executor = ThreadPoolExecutor(max_workers=1)\n        self._force_sync = False\n        self._layers_to_task: Dict[Tuple[Layer], Future] = {}\n        self._lock_layers_to_task = RLock()\n\n    @contextmanager\n    def force_sync(self):\n        \"\"\"Context manager to temporarily force slicing to be synchronous.\n        This should only be used from the main thread.\n\n        >>> layer_slicer = _LayerSlicer()\n        >>> layer = Image(...)  # an async-ready layer\n        >>> with layer_slice.force_sync():\n        >>>     layer_slicer.slice_layers_async(layers=[layer], dims=Dims())\n        \"\"\"\n        prev = self._force_sync\n        self._force_sync = True\n        try:\n            yield None\n        finally:\n            self._force_sync = prev\n\n    def wait_until_idle(self, timeout: Optional[float] = None) -> None:\n        \"\"\"Wait for all slicing tasks to complete before returning.\n\n        Attributes\n        ----------\n        timeout: float or None\n            (Optional) time in seconds to wait before raising TimeoutError. If set as None,\n            there is no limit to the wait time. Defaults to None\n\n        Raises\n        ------\n        TimeoutError: when the timeout limit has been exceeded and the task is\n            not yet complete\n        \"\"\"\n        futures = self._layers_to_task.values()\n        _, not_done_futures = wait(futures, timeout=timeout)\n\n        if len(not_done_futures) > 0:\n            raise TimeoutError(\n                f'Slicing {len(not_done_futures)} tasks did not complete within timeout ({timeout}s).'\n            )\n\n    def slice_layers_async(\n        self, layers: Iterable[Layer], dims: Dims\n    ) -> Future[dict]:\n        \"\"\"This should only be called from the main thread.\n\n        Creates a new task and adds it to the _layers_to_task dict. Cancels\n        all tasks currently pending for that layer tuple.\n\n        Submitting multiple layers at one generates multiple requests (stored in a dict),\n        but only ONE task.\n\n        If multiple layers are sliced, any task that contains only one of those\n        layers can safely be cancelled. If a single layer is sliced, it will\n        wait for any existing tasks that include that layer AND another layer,\n        In other words, it will only cancel if the new task will replace the\n        slices of all the layers in the pending task.\n        \"\"\"\n        # Cancel any tasks that are slicing a subset of the layers\n        # being sliced now. This allows us to slice arbitrary sets of\n        # layers with some sensible and not too complex cancellation\n        # policy.\n        if existing_task := self._find_existing_task(layers):\n            logger.debug('Cancelling task for %s', layers)\n            existing_task.cancel()\n\n        # Not all layer types will initially be asynchronously sliceable.\n        # The following logic gives us a way to handle those in the short\n        # term as we develop, and also in the long term if there are cases\n        # when we want to perform sync slicing anyway.\n        requests = {}\n        sync_layers = []\n        for layer in layers:\n            if isinstance(layer, _AsyncSliceable) and not self._force_sync:\n                requests[layer] = layer._make_slice_request(dims)\n            else:\n                sync_layers.append(layer)\n\n        # create task for slicing of each request/layer\n        task = self._executor.submit(self._slice_layers, requests)\n\n        # submit the sync layers (purposefully placed after async submission)\n        for layer in sync_layers:\n            layer._slice_dims(dims.point, dims.ndisplay, dims.order)\n\n        # store task for cancellation logic\n        # this is purposefully done before adding the done callback to ensure\n        # that the task is added before the done callback can be executed\n        with self._lock_layers_to_task:\n            self._layers_to_task[tuple(requests)] = task\n\n        task.add_done_callback(self._on_slice_done)\n\n        return task\n\n    def shutdown(self) -> None:\n        \"\"\"Should be called from the main thread when this is no longer needed.\"\"\"\n        self._executor.shutdown()\n\n    def _slice_layers(self, requests: Dict) -> Dict:\n        \"\"\"\n        Iterates through a dictionary of request objects and call the slice\n        on each individual layer. Can be called from the main or slicing thread.\n\n        Attributes\n        ----------\n        requests: dict[Layer, SliceRequest]\n            Dictionary of request objects to be used for constructing the slice\n\n        Returns\n        -------\n        dict[Layer, SliceResponse]: which contains the results of the slice\n        \"\"\"\n        return {layer: request() for layer, request in requests.items()}\n\n    def _on_slice_done(self, task: Future[Dict]) -> None:\n        \"\"\"\n        This is the \"done_callback\" which is added to each task.\n        Can be called from the main or slicing thread.\n        \"\"\"\n        if not self._try_to_remove_task(task):\n            logger.debug('Task not found')\n            return\n\n        if task.cancelled():\n            logger.debug('Cancelled task')\n            return\n        result = task.result()\n        self.events.ready(Event('ready', value=result))\n\n    def _try_to_remove_task(self, task: Future[Dict]) -> bool:\n        \"\"\"\n        Attempt to remove task, return false if task not found, return true\n        if task is found and removed from layers_to_task dict.\n\n        This function provides a lock to ensure that the layers_to_task dict\n        is unmodified during this process.\n        \"\"\"\n        with self._lock_layers_to_task:\n            for k_layers, v_task in self._layers_to_task.items():\n                if v_task == task:\n                    del self._layers_to_task[k_layers]\n                    return True\n        return False\n\n    def _find_existing_task(\n        self, layers: Iterable[Layer]\n    ) -> Optional[Future[Dict]]:\n        \"\"\"Find the task associated with a list of layers. Returns the first\n        task found for which the layers of the task are a subset of the input\n        layers.\n\n        This function provides a lock to ensure that the layers_to_task dict\n        is unmodified during this process.\n        \"\"\"\n        with self._lock_layers_to_task:\n            layer_set = set(layers)\n            for task_layers, task in self._layers_to_task.items():\n                if set(task_layers).issubset(layer_set):\n                    logger.debug(f'Found existing task for {task_layers}')\n                    return task\n        return None\n", "path": "napari/components/_layer_slicer.py"}]}