Struct holochain::conductor::manager::ManagedTaskAdd

pub struct ManagedTaskAdd { /* private fields */ }

A message sent to the TaskManager, registering an ManagedTask of a given kind.

Implementations

impl ManagedTaskAdd

pub fn ignore(handle: JoinHandle<ManagedTaskResult>, name: &str) -> Self

You just want the task in the task manager but don’t want to react to an error

Examples found in repository

src/conductor/conductor.rs (lines 421-424)

        pub(crate) async fn add_admin_interfaces(
            self: Arc<Self>,
            configs: Vec<AdminInterfaceConfig>,
        ) -> ConductorResult<()> {
            let admin_api = RealAdminInterfaceApi::new(self.clone());
            let stop_tx = self.task_manager.share_ref(|tm| {
                tm.as_ref()
                    .expect("Task manager not started yet")
                    .task_stop_broadcaster()
                    .clone()
            });

            // Closure to process each admin config item
            let spawn_from_config = |AdminInterfaceConfig { driver, .. }| {
                let admin_api = admin_api.clone();
                let stop_tx = stop_tx.clone();
                async move {
                    match driver {
                        InterfaceDriver::Websocket { port } => {
                            let (listener_handle, listener) =
                                spawn_websocket_listener(port).await?;
                            let port = listener_handle.local_addr().port().unwrap_or(port);
                            let handle: ManagedTaskHandle = spawn_admin_interface_task(
                                listener_handle,
                                listener,
                                admin_api.clone(),
                                stop_tx.subscribe(),
                            )?;
                            InterfaceResult::Ok((port, handle))
                        }
                    }
                }
            };

            // spawn interface tasks, collect their JoinHandles,
            // panic on errors.
            let handles: Result<Vec<_>, _> =
                future::join_all(configs.into_iter().map(spawn_from_config))
                    .await
                    .into_iter()
                    .collect();
            // Exit if the admin interfaces fail to be created
            let handles = handles.map_err(Box::new)?;

            {
                let mut ports = Vec::new();

                // First, register the keepalive task, to ensure the conductor doesn't shut down
                // in the absence of other "real" tasks
                self.manage_task(ManagedTaskAdd::ignore(
                    tokio::spawn(keep_alive_task(stop_tx.subscribe())),
                    "keepalive task",
                ))
                .await?;

                // Now that tasks are spawned, register them with the TaskManager
                for (port, handle) in handles {
                    ports.push(port);
                    self.manage_task(ManagedTaskAdd::ignore(
                        handle,
                        &format!("admin interface, port {}", port),
                    ))
                    .await?
                }
                for p in ports {
                    self.add_admin_port(p);
                }
            }
            Ok(())
        }

        /// Spawn a new app interface task, register it with the TaskManager,
        /// and modify the conductor accordingly, based on the config passed in
        /// which is just a networking port number (or 0 to auto-select one).
        /// Returns the given or auto-chosen port number if giving an Ok Result
        pub async fn add_app_interface(
            self: Arc<Self>,
            port: either::Either<u16, AppInterfaceId>,
        ) -> ConductorResult<u16> {
            let interface_id = match port {
                either::Either::Left(port) => AppInterfaceId::new(port),
                either::Either::Right(id) => id,
            };
            let port = interface_id.port();
            tracing::debug!("Attaching interface {}", port);
            let app_api = RealAppInterfaceApi::new(self.clone());
            // This receiver is thrown away because we can produce infinite new
            // receivers from the Sender
            let (signal_tx, _r) = tokio::sync::broadcast::channel(SIGNAL_BUFFER_SIZE);
            let stop_rx = self.task_manager.share_ref(|tm| {
                tm.as_ref()
                    .expect("Task manager not initialized")
                    .task_stop_broadcaster()
                    .subscribe()
            });
            let (port, task) = spawn_app_interface_task(port, app_api, signal_tx.clone(), stop_rx)
                .await
                .map_err(Box::new)?;
            // TODO: RELIABILITY: Handle this task by restarting it if it fails and log the error
            self.manage_task(ManagedTaskAdd::ignore(
                task,
                &format!("app interface, port {}", port),
            ))
            .await?;
            let interface = AppInterfaceRuntime::Websocket { signal_tx };

            self.app_interfaces.share_mut(|app_interfaces| {
                if app_interfaces.contains_key(&interface_id) {
                    return Err(ConductorError::AppInterfaceIdCollision(
                        interface_id.clone(),
                    ));
                }

                app_interfaces.insert(interface_id.clone(), interface);
                Ok(())
            })?;
            let config = AppInterfaceConfig::websocket(port);
            self.update_state(|mut state| {
                state.app_interfaces.insert(interface_id, config);
                Ok(state)
            })
            .await?;
            tracing::debug!("App interface added at port: {}", port);
            Ok(port)
        }

pub fn unrecoverable(handle: JoinHandle<ManagedTaskResult>, name: &str) -> Self

If this task fails, the entire conductor must be shut down

pub fn cell_critical(
    handle: JoinHandle<ManagedTaskResult>,
    cell_id: CellId,
    name: &str
) -> Self

If this task fails, only the Cell which it runs under must be stopped

Examples found in repository

src/core/queue_consumer.rs (lines 105-109)

pub async fn spawn_queue_consumer_tasks(
    cell_id: CellId,
    network: HolochainP2pDna,
    space: &Space,
    conductor_handle: ConductorHandle,
    task_sender: sync::mpsc::Sender<ManagedTaskAdd>,
    stop: sync::broadcast::Sender<()>,
) -> (QueueTriggers, InitialQueueTriggers) {
    let Space {
        authored_db,
        dht_db,
        cache_db: cache,
        dht_query_cache,
        ..
    } = space;

    let keystore = conductor_handle.keystore().clone();
    let dna_hash = Arc::new(cell_id.dna_hash().clone());
    let queue_consumer_map = conductor_handle.get_queue_consumer_workflows();

    // Publish
    let (tx_publish, handle) = spawn_publish_dht_ops_consumer(
        cell_id.agent_pubkey().clone(),
        authored_db.clone(),
        conductor_handle.clone(),
        stop.subscribe(),
        Box::new(network.clone()),
    );
    task_sender
        .send(ManagedTaskAdd::cell_critical(
            handle,
            cell_id.clone(),
            "publish_dht_ops_consumer",
        ))
        .await
        .expect("Failed to manage workflow handle");

    // Validation Receipt
    // One per space.
    let (tx_receipt, handle) =
        queue_consumer_map.spawn_once_validation_receipt(dna_hash.clone(), || {
            spawn_validation_receipt_consumer(
                dna_hash.clone(),
                dht_db.clone(),
                conductor_handle.clone(),
                stop.subscribe(),
                network.clone(),
            )
        });

    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "validation_receipt_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    // Integration
    // One per space.
    let (tx_integration, handle) =
        queue_consumer_map.spawn_once_integration(dna_hash.clone(), || {
            spawn_integrate_dht_ops_consumer(
                dna_hash.clone(),
                dht_db.clone(),
                dht_query_cache.clone(),
                stop.subscribe(),
                tx_receipt.clone(),
                network.clone(),
            )
        });

    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "integrate_dht_ops_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    let dna_def = conductor_handle
        .get_dna_def(&*dna_hash)
        .expect("Dna must be in store");

    // App validation
    // One per space.
    let (tx_app, handle) = queue_consumer_map.spawn_once_app_validation(dna_hash.clone(), || {
        spawn_app_validation_consumer(
            dna_hash.clone(),
            AppValidationWorkspace::new(
                authored_db.clone().into(),
                dht_db.clone(),
                space.dht_query_cache.clone(),
                cache.clone(),
                keystore.clone(),
                Arc::new(dna_def),
            ),
            conductor_handle.clone(),
            stop.subscribe(),
            tx_integration.clone(),
            network.clone(),
            dht_query_cache.clone(),
        )
    });
    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "app_validation_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    let dna_def = conductor_handle
        .get_dna_def(&*dna_hash)
        .expect("Dna must be in store");

    // Sys validation
    // One per space.
    let (tx_sys, handle) = queue_consumer_map.spawn_once_sys_validation(dna_hash.clone(), || {
        spawn_sys_validation_consumer(
            SysValidationWorkspace::new(
                authored_db.clone().into(),
                dht_db.clone().into(),
                dht_query_cache.clone(),
                cache.clone(),
                Arc::new(dna_def),
            ),
            space.clone(),
            conductor_handle.clone(),
            stop.subscribe(),
            tx_app.clone(),
            network.clone(),
        )
    });

    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "sys_validation_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    let (tx_cs, handle) = queue_consumer_map.spawn_once_countersigning(dna_hash.clone(), || {
        spawn_countersigning_consumer(
            space.clone(),
            stop.subscribe(),
            network.clone(),
            tx_sys.clone(),
        )
    });
    if let Some(handle) = handle {
        task_sender
            .send(ManagedTaskAdd::cell_critical(
                handle,
                cell_id.clone(),
                "countersigning_consumer",
            ))
            .await
            .expect("Failed to manage workflow handle");
    }

    (
        QueueTriggers {
            sys_validation: tx_sys.clone(),
            publish_dht_ops: tx_publish.clone(),
            countersigning: tx_cs,
            integrate_dht_ops: tx_integration.clone(),
        },
        InitialQueueTriggers::new(tx_sys, tx_publish, tx_app, tx_integration, tx_receipt),
    )
}

pub fn dna_critical(
    handle: JoinHandle<ManagedTaskResult>,
    dna_hash: Arc<DnaHash>,
    name: &str
) -> Self

If this task fails, only the Cells with this DnaHash must be stopped

pub fn generic(handle: JoinHandle<ManagedTaskResult>, f: OnDeath) -> Self

Handle a task’s completion with a generic callback

Trait Implementations

impl Debug for ManagedTaskAdd

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Future for ManagedTaskAdd

type Output = TaskOutcome

The type of value produced on completion.

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>

Attempt to resolve the future to a final value, registering the current task for wakeup if the value is not yet available.