Task step implementation
(
task: &PyObjectRef,
exc: Option<PyObjectRef>,
vm: &VirtualMachine,
)
| 2019 | |
| 2020 | /// Task step implementation |
| 2021 | fn task_step_impl( |
| 2022 | task: &PyObjectRef, |
| 2023 | exc: Option<PyObjectRef>, |
| 2024 | vm: &VirtualMachine, |
| 2025 | ) -> PyResult { |
| 2026 | let task_ref: PyRef<PyTask> = task |
| 2027 | .clone() |
| 2028 | .downcast() |
| 2029 | .map_err(|_| vm.new_type_error("task_step called with non-Task object"))?; |
| 2030 | |
| 2031 | if task_ref.base.fut_state.load() != FutureState::Pending { |
| 2032 | // Task is already done - report InvalidStateError via exception handler |
| 2033 | let loop_obj = task_ref.base.fut_loop.read().clone(); |
| 2034 | if let Some(loop_obj) = loop_obj { |
| 2035 | let exc = new_invalid_state_error(vm, "step(): already done"); |
| 2036 | let context = vm.ctx.new_dict(); |
| 2037 | context.set_item("message", vm.new_pyobj("step(): already done"), vm)?; |
| 2038 | context.set_item("exception", exc.clone().into(), vm)?; |
| 2039 | context.set_item("task", task.clone(), vm)?; |
| 2040 | let _ = vm.call_method(&loop_obj, "call_exception_handler", (context,)); |
| 2041 | } |
| 2042 | return Ok(vm.ctx.none()); |
| 2043 | } |
| 2044 | |
| 2045 | *task_ref.task_fut_waiter.write() = None; |
| 2046 | |
| 2047 | let coro = task_ref.task_coro.read().clone(); |
| 2048 | let coro = match coro { |
| 2049 | Some(c) => c, |
| 2050 | None => return Ok(vm.ctx.none()), |
| 2051 | }; |
| 2052 | |
| 2053 | // Get event loop for enter/leave task |
| 2054 | let loop_obj = task_ref.base.fut_loop.read().clone(); |
| 2055 | let loop_obj = match loop_obj { |
| 2056 | Some(l) => l, |
| 2057 | None => return Ok(vm.ctx.none()), |
| 2058 | }; |
| 2059 | |
| 2060 | // Get task context |
| 2061 | let context = task_ref.task_context.read().clone(); |
| 2062 | |
| 2063 | // Enter task - register as current task |
| 2064 | _enter_task(loop_obj.clone(), task.clone(), vm)?; |
| 2065 | |
| 2066 | // Determine the exception to throw (if any) |
| 2067 | // If task_must_cancel is set and exc is None or not CancelledError, create CancelledError |
| 2068 | let exc_to_throw = if task_ref.task_must_cancel.load(Ordering::Relaxed) { |
| 2069 | task_ref.task_must_cancel.store(false, Ordering::Relaxed); |
| 2070 | if let Some(ref e) = exc { |
| 2071 | if is_cancelled_error_obj(e, vm) { |
| 2072 | exc.clone() |
| 2073 | } else { |
| 2074 | Some(task_ref.make_cancelled_error_impl(vm).into()) |
| 2075 | } |
| 2076 | } else { |
| 2077 | Some(task_ref.make_cancelled_error_impl(vm).into()) |
| 2078 | } |
no test coverage detected