Made some fixes and small enhancements.

Author: s3rius

docs/guide/cli.md

@@ -18,6 +18,24 @@ Like this:
 taskiq worker mybroker:broker_var my_project.module1 my_project.module2
 ```
 
+### Sync function
+
+Taskiq can run sync function. But since it's asynchronous system, it runs it in separate thread or process.
+By default process pool is used. But if you are going to use taskiq for heavy computations, such as neural network
+model training or other calculations, you might want to use processpool instead.
+
+More precisely about the difference you can read in [python docs about executors](https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.Executor).
+
+As a rule of thumb:
+* If you're using sync functions for IO then use `threadpool`;
+* If you're using sync functions for CPU bound workloads then use processpool.
+
+By default taskiq uses threadpool. If you want to change this behavior, use these parameters:
+
+* `--use-process-pool` to switch to processpools;
+* `--max-process-pool-processes` to manually specify worker processes;
+* `--max-threadpool-threads` to configure maximum threads for threadpool if it's being used;
+
 ### Auto importing
 
 Enumerating all modules with tasks is not an option sometimes.

taskiq/cli/worker/args.py

@@ -31,7 +31,7 @@ class WorkerArgs:
     configure_logging: bool = True
     log_level: LogLevel = LogLevel.INFO
     workers: int = 2
-    max_threadpool_threads: int = 10
+    max_threadpool_threads: Optional[int] = None
     max_process_pool_processes: Optional[int] = None
     no_parse: bool = False
     shutdown_timeout: float = 5
@@ -147,6 +147,7 @@ def from_cli(
         parser.add_argument(
             "--max-threadpool-threads",
             type=int,
+            default=None,
             help="Maximum number of threads for executing sync functions.",
         )
         parser.add_argument(

taskiq/schedule_sources/label_based.py

@@ -31,8 +31,8 @@ async def get_schedules(self) -> List["ScheduledTask"]:
             if task.broker != self.broker:
                 # if task broker doesn't match self, something is probably wrong
                 logger.warning(
-                    f"Broker for {task_name} ({task.broker}) doesn't "
-                    f"match scheduler's broker ({self.broker})"
+                    f"Broker for {task_name} `{task.broker}` doesn't "
+                    f"match scheduler's broker `{self.broker}`",
                 )
                 continue
             for schedule in task.labels.get("schedule", []):
@@ -69,8 +69,8 @@ def post_send(self, scheduled_task: ScheduledTask) -> None:
             if task.broker != self.broker:
                 # if task broker doesn't match self, something is probably wrong
                 logger.warning(
-                    f"Broker for {task_name} ({task.broker}) doesn't "
-                    f"match scheduler's broker ({self.broker})"
+                    f"Broker for {task_name} `{task.broker}` doesn't "
+                    f"match scheduler's broker `{self.broker}`",
                 )
                 continue
             if scheduled_task.task_name != task_name: