Method go

src/python/py_runtime.rs:144–233

(
        &self,
        py: Python<'_>,
        spec: &Bound<'_, PyAny>,
        args: &Bound<'_, PyTuple>,
        out: Option<PyChannel>,
        kwargs: Option<&Bound<'_, PyDict>>,
    )

    /// If `out` is provided, the result is delivered to that channel asynchronously.
    #[pyo3(signature = (spec, *args, out=None, **kwargs))]
    fn go(
        &self,
        py: Python<'_>,
        spec: &Bound<'_, PyAny>,
        args: &Bound<'_, PyTuple>,
        out: Option<PyChannel>,
        kwargs: Option<&Bound<'_, PyDict>>,
    ) -> PyResult<PyTaskHandle> {
        let out_channel = out.map(|c| c.inner);

        if let Ok(map_spec) = spec.extract::<PyMapSpec>() {
            reject_callable_args_for_spec(args, kwargs)?;
            let handle = TaskHandle::new();
            let h2 = handle.clone();

            py.allow_threads(|| {
                h2.set_running_compute();
                match execute_map(&map_spec) {
                    Ok(buf) => {
                        if let Some(ch) = out_channel {
                            h2.set_delivery_queued(TaskResult::Buffer(buf.clone()));
                            submit_delivery(DeliveryJob {
                                task: h2,
                                channel: ch,
                                buffer: buf,
                            });
                        } else {
                            h2.complete(TaskResult::Buffer(buf));
                        }
                    }
                    Err(e) => h2.fail(e),
                }
            });

            Ok(PyTaskHandle::new(handle))
        } else if let Ok(reduce_spec) = spec.extract::<PyReduceSpec>() {
            reject_callable_args_for_spec(args, kwargs)?;
            let handle = TaskHandle::new();
            let h2 = handle.clone();

            py.allow_threads(|| {
                h2.set_running_compute();
                match execute_reduce(&reduce_spec) {
                    Ok(v) => {
                        if let Some(ch) = out_channel {
                            h2.set_delivery_queued(TaskResult::Scalar(v));
                            submit_delivery(DeliveryJob {
                                task: h2,
                                channel: ch,
                                buffer: Buffer::from_f64_vec(vec![v]),
                            });
                        } else {
                            h2.complete(TaskResult::Scalar(v));
                        }
                    }
                    Err(e) => h2.fail(e),
                }
            });