Improve workflow task deadlock and eviction (#806)

Author: cretz

temporalio/worker/_replayer.py

@@ -51,15 +51,22 @@ def __init__(
         """Create a replayer to replay workflows from history.
 
         See :py:meth:`temporalio.worker.Worker.__init__` for a description of
-        most of the arguments. The same arguments need to be passed to the
-        replayer that were passed to the worker when the workflow originally
+        most of the arguments. Most of the same arguments need to be passed to
+        the replayer that were passed to the worker when the workflow originally
         ran.
+
+        Note, unlike the worker, for the replayer the workflow_task_executor
+        will default to a new thread pool executor with no max_workers set that
+        will be shared across all replay calls and never explicitly shut down.
+        Users are encouraged to provide their own if needing more control.
         """
         if not workflows:
             raise ValueError("At least one workflow must be specified")
         self._config = ReplayerConfig(
             workflows=list(workflows),
-            workflow_task_executor=workflow_task_executor,
+            workflow_task_executor=(
+                workflow_task_executor or concurrent.futures.ThreadPoolExecutor()
+            ),
             workflow_runner=workflow_runner,
             unsandboxed_workflow_runner=unsandboxed_workflow_runner,
             namespace=namespace,
@@ -195,6 +202,7 @@ def on_eviction_hook(
                 task_queue=task_queue,
                 workflows=self._config["workflows"],
                 workflow_task_executor=self._config["workflow_task_executor"],
+                max_concurrent_workflow_tasks=5,
                 workflow_runner=self._config["workflow_runner"],
                 unsandboxed_workflow_runner=self._config["unsandboxed_workflow_runner"],
                 data_converter=self._config["data_converter"],

temporalio/worker/_worker.py

@@ -109,9 +109,10 @@ def __init__(
             workflow_task_executor: Thread pool executor for workflow tasks. If
                 this is not present, a new
                 :py:class:`concurrent.futures.ThreadPoolExecutor` will be
-                created with ``max_workers`` set to ``max(os.cpu_count(), 4)``.
-                The default one will be properly shutdown, but if one is
-                provided, the caller is responsible for shutting it down after
+                created with ``max_workers`` set to
+                ``max_concurrent_workflow_tasks`` if it is present, or 500
+                otherwise. The default one will be properly shutdown, but if one
+                is provided, the caller is responsible for shutting it down after
                 the worker is shut down.
             workflow_runner: Runner for workflows.
             unsandboxed_workflow_runner: Runner for workflows that opt-out of
@@ -312,6 +313,7 @@ def __init__(
                 task_queue=task_queue,
                 workflows=workflows,
                 workflow_task_executor=workflow_task_executor,
+                max_concurrent_workflow_tasks=max_concurrent_workflow_tasks,
                 workflow_runner=workflow_runner,
                 unsandboxed_workflow_runner=unsandboxed_workflow_runner,
                 data_converter=client_config["data_converter"],