    async def start_async(self, params:Union[dict, Parameters]=None, **kwargs):
        """Execute the task. Creates a new process
        (if execution='process'), a new thread
        (if execution='thread') or blocks and
        runs till the task is completed (if
        execution='main').

        Parameters
        ----------
        params : dict, Parameters, optional
            Extra parameters for the task. Also
            the session parameters, task parameters
            and extra parameters are acquired, by default None
        """

        # The parameters are handled in the following way:
        #   - First extra parameters are fetched. This includes:
        #       - session.parameters
        #       - _task_, _session_, _thread_terminate_
        #   - Then these extras are prefiltered (called params)
        #   - Then the task parameters are fetched (direct_params)
        #   - If process/thread, these parameters are pre_materialized
        #   - Then the params are post filtered
        #   - Then params and direct_params are fed to the execute method
        execution = self.get_execution()
        task_run = TaskRun(start=self.session.get_time(), task=None)
        try:
            self.force_run = False
            params = self.get_extra_params(params, execution=execution)
            direct_params = self._get_direct_params()

            task_run.run_id = self.get_run_id(task_run, params=params | direct_params)

            # Run the actual task
            if execution in ("main", "async"):
                async_task = asyncio.create_task(
                    self._run_as_async(
                        params=params,
                        direct_params=direct_params,
                        task_run=task_run,
                        execution=execution, **kwargs
                    )
                )
                if execution == "async":
                    task_run.task = async_task
                self._run_stack.append(task_run)
                self.log_running(task_run)
                if execution == "main":
                    await async_task
                if _IS_WINDOWS:
                    #! TODO: This probably is now solved
                    # There is an annoying bug (?) in Windows:
                    # https://bugs.python.org/issue44831
                    # If one checks whether the task has succeeded/failed
                    # already the log might show that the task finished
                    # 1 microsecond in the future if memory logging is used.
                    # Therefore we sleep that so condition checks especially
                    # in tests will succeed.
                    time.sleep(1e-6)
            elif execution == "process":
                self.run_as_process(params=params, direct_params=direct_params, task_run=task_run, **kwargs)
            elif execution == "thread":
                self.run_as_thread(params=params, direct_params=direct_params, task_run=task_run, **kwargs)
        except (SchedulerRestart, SchedulerExit):
            raise
        except TaskLoggingError:
            if self.status == "run" and execution not in ('thread', 'process'):
                # Task logging to run failed
                # so we log it to fail

                # NOTE: processes and threads log independently
                # and it is not aware the logging failed
                # (there is a log record still coming about the finish)
                self.log_failure(task_run)
            raise
        except Exception as exc:
            # Something went wrong in the initiation
            # and it did not reach to log_running
            if task_run.run_id is None:
                task_run.run_id = self.get_run_id(task_run)
            if self.status != "run":
                self.log_running(task_run)
            self.log_failure(task_run)
            raise TaskSetupError("Task failed before logging") from exc
        finally:
            # Clean up
            self._main_alive = False
            # Delete the "main" runs from run stack
            self._run_stack = [run for run in self._run_stack if run.task is not None]
Execute the task. Creates a new process
(if execution='process'), a new thread
(if execution='thread') or blocks and
runs till the task is completed (if
execution='main').

Parameters
----------
params : dict, Parameters, optional
Extra parameters for the task. Also
the session parameters, task parameters
and extra parameters are acquired, by default None
rocketry/core/task.py