pub fn spawn_module(
    parsed_module: parity_wasm::elements::Module,
    logger: Logger,
    subgraph_id: SubgraphDeploymentId,
    host_metrics: Arc<HostMetrics>,
    runtime: tokio::runtime::Handle,
) -> Result<mpsc::Sender<MappingRequest>, Error> {
    let valid_module = Arc::new(ValidModule::new(parsed_module)?);

    // Create channel for event handling requests
    let (mapping_request_sender, mapping_request_receiver) = mpsc::channel(100);

    // wasmi modules are not `Send` therefore they cannot be scheduled by
    // the regular tokio executor, so we create a dedicated thread.
    //
    // In case of failure, this thread may panic or simply terminate,
    // dropping the `mapping_request_receiver` which ultimately causes the
    // subgraph to fail the next time it tries to handle an event.
    let conf =
        thread::Builder::new().name(format!("mapping-{}-{}", &subgraph_id, uuid::Uuid::new_v4()));
    conf.spawn(move || {
        runtime.enter(|| {
            // Pass incoming triggers to the WASM module and return entity changes;
            // Stop when canceled because all RuntimeHosts and their senders were dropped.
            match mapping_request_receiver
                .map_err(|()| unreachable!())
                .for_each(move |request| -> Result<(), Error> {
                    let MappingRequest {
                        ctx,
                        trigger,
                        result_sender,
                    } = request;

                    // Start the WASMI module runtime.
                    let section = host_metrics.stopwatch.start_section("module_init");
                    let module = WasmiModule::from_valid_module_with_ctx(
                        valid_module.clone(),
                        ctx,
                        host_metrics.clone(),
                    )?;
                    section.end();

                    let section = host_metrics.stopwatch.start_section("run_handler");
                    let result = match trigger {
                        MappingTrigger::Log {
                            transaction,
                            log,
                            params,
                            handler,
                        } => module.handle_ethereum_log(
                            handler.handler.as_str(),
                            transaction,
                            log,
                            params,
                        ),
                        MappingTrigger::Call {
                            transaction,
                            call,
                            inputs,
                            outputs,
                            handler,
                        } => module.handle_ethereum_call(
                            handler.handler.as_str(),
                            transaction,
                            call,
                            inputs,
                            outputs,
                        ),
                        MappingTrigger::Block { handler } => {
                            module.handle_ethereum_block(handler.handler.as_str())
                        }
                    };
                    section.end();

                    result_sender
                        .send((result, future::ok(Instant::now())))
                        .map_err(|_| err_msg("WASM module result receiver dropped."))
                })
                .wait()
            {
                Ok(()) => debug!(logger, "Subgraph stopped, WASM runtime thread terminated"),
                Err(e) => debug!(logger, "WASM runtime thread terminated abnormally";
                                    "error" => e.to_string()),
            }
        })
    })
    .map(|_| ())
    .map_err(|e| format_err!("Spawning WASM runtime thread failed: {}", e))?;

    Ok(mapping_request_sender)
}
/// Spawn a wasm module in its own thread.
Spawn a wasm module in its own thread.
Spawn a wasm module in its own thread .
pub fn spawn_module(
    parsed_module: parity_wasm::elements::Module,
    logger: Logger,
    subgraph_id: SubgraphDeploymentId,
    host_metrics: Arc<HostMetrics>,
    runtime: tokio::runtime::Handle,
) -> Result<mpsc::Sender<MappingRequest>, Error> {
    let valid_module = Arc::new(ValidModule::new(parsed_module)?);
    // Create channel for event handling requests
    let (mapping_request_sender, mapping_request_receiver) = mpsc::channel(100);
    // wasmi modules are not `Send` therefore they cannot be scheduled by
    // the regular tokio executor, so we create a dedicated thread.
    //
    // In case of failure, this thread may panic or simply terminate,
    // dropping the `mapping_request_receiver` which ultimately causes the
    // subgraph to fail the next time it tries to handle an event.
    let conf =
        thread::Builder::new().name(format!("mapping-{}-{}", &subgraph_id, uuid::Uuid::new_v4()));
    conf.spawn(move || {
        runtime.enter(|| {
            // Pass incoming triggers to the WASM module and return entity changes;
            // Stop when canceled because all RuntimeHosts and their senders were dropped.
            match mapping_request_receiver
                .map_err(|()| unreachable!())
                .for_each(move |request| -> Result<(), Error> {
                    let MappingRequest {
                        ctx,
                        trigger,
                        result_sender,
                    } = request;
                    // Start the WASMI module runtime.
                    let section = host_metrics.stopwatch.start_section("module_init");
                    let module = WasmiModule::from_valid_module_with_ctx(
                        valid_module.clone(),
                        ctx,
                        host_metrics.clone(),
                    )?;
                    section.end();
                    let section = host_metrics.stopwatch.start_section("run_handler");
                    let result = match trigger {
                        MappingTrigger::Log {
                            transaction,
                            log,
                            params,
                            handler,
                        } => module.handle_ethereum_log(
                            handler.handler.as_str(),
                            transaction,
                            log,
                            params,
                        ),
                        MappingTrigger::Call {
                            transaction,
                            call,
                            inputs,
                            outputs,
                            handler,
                        } => module.handle_ethereum_call(
                            handler.handler.as_str(),
                            transaction,
                            call,
                            inputs,
                            outputs,
                        ),
                        MappingTrigger::Block { handler } => {
                            module.handle_ethereum_block(handler.handler.as_str())
                        }
                    };
                    section.end();
                    result_sender
                        .send((result, future::ok(Instant::now())))
                        .map_err(|_| err_msg("WASM module result receiver dropped."))
                })
                .wait()
            {
                Ok(()) => debug!(logger, "Subgraph stopped, WASM runtime thread terminated"),
                Err(e) => debug!(logger, "WASM runtime thread terminated abnormally";
                                    "error" => e.to_string()),
            }
        })
    })
    .map(|_| ())
    .map_err(|e| format_err!("Spawning WASM runtime thread failed: {}", e))?;
    Ok(mapping_request_sender)
}
pub fn spawn_module ( parsed_module : parity_wasm :: elements :: Module , logger : Logger , subgraph_id : SubgraphDeploymentId , host_metrics : Arc < HostMetrics > , runtime : tokio :: runtime :: Handle , ) -> Result < mpsc :: Sender < MappingRequest > , Error > { let valid_module = Arc :: new ( ValidModule :: new ( parsed_module ) ? ) ; // Create channel for event handling requests let ( mapping_request_sender , mapping_request_receiver ) = mpsc :: channel ( 100 ) ; // wasmi modules are not `Send` therefore they cannot be scheduled by // the regular tokio executor, so we create a dedicated thread. // // In case of failure, this thread may panic or simply terminate, // dropping the `mapping_request_receiver` which ultimately causes the // subgraph to fail the next time it tries to handle an event. let conf = thread :: Builder :: new ( ) . name ( format ! ( " " subgraph_id uuid Uuid new_v4 ( ) ) ) ; conf . spawn ( move | | { runtime . enter ( | | { // Pass incoming triggers to the WASM module and return entity changes; // Stop when canceled because all RuntimeHosts and their senders were dropped. match mapping_request_receiver . map_err ( | ( ) | unreachable ! ( ) ) . for_each ( move | request | -> Result < ( ) , Error > { let MappingRequest { ctx , trigger , result_sender , } = request ; // Start the WASMI module runtime. let section = host_metrics . stopwatch . start_section ( " " ) ; let module = WasmiModule :: from_valid_module_with_ctx ( valid_module . clone ( ) , ctx , host_metrics . clone ( ) , ) ? ; section . end ( ) ; let section = host_metrics . stopwatch . start_section ( " " ) ; let result = match trigger { MappingTrigger :: Log { transaction , log , params , handler , } => module . handle_ethereum_log ( handler . handler . as_str ( ) , transaction , log , params , ) , MappingTrigger :: Call { transaction , call , inputs , outputs , handler , } => module . handle_ethereum_call ( handler . handler . as_str ( ) , transaction , call , inputs , outputs , ) , MappingTrigger :: Block { handler } => { module . handle_ethereum_block ( handler . handler . as_str ( ) ) } } ; section . end ( ) ; result_sender . send ( ( result , future :: ok ( Instant :: now ( ) ) ) ) . map_err ( | _ | err_msg ( " " ) ) } ) . wait ( ) { Ok ( ( ) ) => debug ! ( logger " " ) , Err ( e ) => debug ! ( logger " " " " e to_string ( ) ) , } } ) } ) . map ( | _ | ( ) ) . map_err ( | e | format_err ! ( " " e ) ) ? ; Ok ( mapping_request_sender ) }
Spawn a wasm module in its own thread.
Spawn a wasm module in its own thread .
parsed_module
parity_wasm::elements::Module
logger
Logger
subgraph_id
SubgraphDeploymentId
host_metrics
Arc<HostMetrics>
runtime
tokio::runtime::Handle
parsed_module
parity_wasm::elements::Module
logger
Logger
subgraph_id
SubgraphDeploymentId
host_metrics
Arc<HostMetrics>
runtime
tokio::runtime::Handle