Skip to content

Commit 39f35c4

Browse files
leoyvensleoyvens
committed
runtime: Refactor block_on to use local runtime
1 parent 66b3306 commit 39f35c4

File tree

9 files changed

+479
-532
lines changed

9 files changed

+479
-532
lines changed

‎Cargo.lock

Lines changed: 352 additions & 323 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎chain/ethereum/tests/network_indexer.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -83,7 +83,7 @@ fn run_network_indexer(
8383
.map_err(|_| ())
8484
.forward(event_sink.sink_map_err(|_| ()))
8585
.map(|_| ())
86-
.timeout(timeout)
86+
.timeout(timeout),
8787
);
8888

8989
future::ok((chains, event_stream.collect()))

‎core/src/subgraph/instance_manager.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -236,7 +236,7 @@ impl SubgraphInstanceManager {
236236
// Subgraph instance shutdown senders
237237
let instances: SharedInstanceKeepAliveMap = Default::default();
238238

239-
graph::spawn(receiver.compat().try_for_each(move |event| {
239+
graph::spawn_blocking(receiver.compat().try_for_each(move |event| {
240240
use self::SubgraphAssignmentProviderEvent::*;
241241

242242
match event {

‎graph/src/log/elastic.rs

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,8 +6,7 @@ use std::sync::{Arc, Mutex};
66
use std::time::Duration;
77

88
use chrono::prelude::{SecondsFormat, Utc};
9-
use futures::Future;
10-
use futures03::{FutureExt, TryFutureExt};
9+
use futures03::TryFutureExt;
1110
use reqwest;
1211
use reqwest::Client;
1312
use serde::ser::Serializer as SerdeSerializer;
@@ -183,7 +182,6 @@ impl ElasticDrain {
183182
}
184183

185184
fn periodically_flush_logs(&self) {
186-
use futures03::compat::Future01CompatExt;
187185
use futures03::stream::StreamExt;
188186

189187
let flush_logger = self.error_logger.clone();

‎runtime/wasm/src/host.rs

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,13 @@ where
7979
subgraph_id: SubgraphDeploymentId,
8080
metrics: Arc<HostMetrics>,
8181
) -> Result<Sender<Self::Req>, Error> {
82-
crate::mapping::spawn_module(parsed_module, logger, subgraph_id, metrics)
82+
crate::mapping::spawn_module(
83+
parsed_module,
84+
logger,
85+
subgraph_id,
86+
metrics,
87+
tokio::runtime::Handle::current(),
88+
)
8389
}
8490

8591
fn build(

‎runtime/wasm/src/host_exports.rs

Lines changed: 4 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,6 @@ impl HostExports {
203203
/// Returns `Ok(None)` if the call was reverted.
204204
pub(crate) fn ethereum_call(
205205
&self,
206-
task_sink: &mut impl Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>,
207206
logger: &Logger,
208207
block: &LightEthereumBlock,
209208
unresolved_call: UnresolvedContractCall,
@@ -349,7 +348,6 @@ impl HostExports {
349348
pub(crate) fn ipfs_cat(
350349
&self,
351350
logger: &Logger,
352-
task_sink: &mut impl Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>,
353351
link: String,
354352
) -> Result<Vec<u8>, HostExportError<impl ExportError>> {
355353
block_on(
@@ -366,28 +364,20 @@ impl HostExports {
366364
// which is identical to `module` when it was first started. The signature
367365
// of the callback must be `callback(JSONValue, Value)`, and the `userData`
368366
// parameter is passed to the callback without any changes
369-
pub(crate) fn ipfs_map<U>(
367+
pub(crate) fn ipfs_map(
370368
&self,
371-
module: &WasmiModule<U>,
369+
module: &WasmiModule,
372370
link: String,
373371
callback: &str,
374372
user_data: store::Value,
375373
flags: Vec<String>,
376-
) -> Result<Vec<BlockState>, HostExportError<impl ExportError>>
377-
where
378-
U: Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>
379-
+ Clone
380-
+ Send
381-
+ Sync
382-
+ 'static,
383-
{
374+
) -> Result<Vec<BlockState>, HostExportError<impl ExportError>> {
384375
const JSON_FLAG: &str = "json";
385376
if !flags.contains(&JSON_FLAG.to_string()) {
386377
return Err(HostExportError(format!("Flags must contain 'json'")));
387378
}
388379

389380
let host_metrics = module.host_metrics.clone();
390-
let task_sink = module.task_sink.clone();
391381
let valid_module = module.valid_module.clone();
392382
let ctx = module.ctx.clone();
393383
let callback = callback.to_owned();
@@ -409,7 +399,6 @@ impl HostExports {
409399
let module = WasmiModule::from_valid_module_with_ctx(
410400
valid_module.clone(),
411401
ctx.clone(),
412-
task_sink.clone(),
413402
host_metrics.clone(),
414403
)?;
415404
let result =
@@ -662,5 +651,5 @@ fn test_string_to_h160_with_0x() {
662651
fn block_on<I: Send + 'static, ER: Send + 'static>(
663652
future: impl Future<Item = I, Error = ER> + Send + 'static,
664653
) -> Result<I, ER> {
665-
futures03::executor::block_on(tokio::spawn(future.compat())).unwrap()
654+
tokio::spawn(future.compat()).compat().wait().unwrap()
666655
}

‎runtime/wasm/src/mapping.rs

Lines changed: 64 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ pub fn spawn_module(
1414
logger: Logger,
1515
subgraph_id: SubgraphDeploymentId,
1616
host_metrics: Arc<HostMetrics>,
17+
runtime: tokio::runtime::Handle,
1718
) -> Result<mpsc::Sender<MappingRequest>, Error> {
1819
let valid_module = Arc::new(ValidModule::new(parsed_module)?);
1920

@@ -23,86 +24,76 @@ pub fn spawn_module(
2324
// wasmi modules are not `Send` therefore they cannot be scheduled by
2425
// the regular tokio executor, so we create a dedicated thread.
2526
//
26-
// This thread can spawn tasks on the runtime by sending them to
27-
// `task_receiver`.
28-
let (task_sender, task_receiver) =
29-
mpsc::channel::<Box<dyn Future<Item = (), Error = ()> + Send>>(100);
30-
graph::spawn(task_receiver.compat().try_for_each(|f| {
31-
graph::spawn(f.compat());
32-
futures03::future::ok(())
33-
}));
34-
35-
// Spawn a dedicated thread for the runtime.
36-
//
3727
// In case of failure, this thread may panic or simply terminate,
3828
// dropping the `mapping_request_receiver` which ultimately causes the
3929
// subgraph to fail the next time it tries to handle an event.
4030
let conf =
4131
thread::Builder::new().name(format!("mapping-{}-{}", &subgraph_id, uuid::Uuid::new_v4()));
4232
conf.spawn(move || {
43-
// Pass incoming triggers to the WASM module and return entity changes;
44-
// Stop when canceled because all RuntimeHosts and their senders were dropped.
45-
match mapping_request_receiver
46-
.map_err(|()| unreachable!())
47-
.for_each(move |request| -> Result<(), Error> {
48-
let MappingRequest {
49-
ctx,
50-
trigger,
51-
result_sender,
52-
} = request;
53-
54-
// Start the WASMI module runtime.
55-
let section = host_metrics.stopwatch.start_section("module_init");
56-
let module = WasmiModule::from_valid_module_with_ctx(
57-
valid_module.clone(),
58-
ctx,
59-
task_sender.clone(),
60-
host_metrics.clone(),
61-
)?;
62-
section.end();
63-
64-
let section = host_metrics.stopwatch.start_section("run_handler");
65-
let result = match trigger {
66-
MappingTrigger::Log {
67-
transaction,
68-
log,
69-
params,
70-
handler,
71-
} => module.handle_ethereum_log(
72-
handler.handler.as_str(),
73-
transaction,
74-
log,
75-
params,
76-
),
77-
MappingTrigger::Call {
78-
transaction,
79-
call,
80-
inputs,
81-
outputs,
82-
handler,
83-
} => module.handle_ethereum_call(
84-
handler.handler.as_str(),
85-
transaction,
86-
call,
87-
inputs,
88-
outputs,
89-
),
90-
MappingTrigger::Block { handler } => {
91-
module.handle_ethereum_block(handler.handler.as_str())
92-
}
93-
};
94-
section.end();
95-
96-
result_sender
97-
.send((result, future::ok(Instant::now())))
98-
.map_err(|_| err_msg("WASM module result receiver dropped."))
99-
})
100-
.wait()
101-
{
102-
Ok(()) => debug!(logger, "Subgraph stopped, WASM runtime thread terminated"),
103-
Err(e) => debug!(logger, "WASM runtime thread terminated abnormally";
33+
runtime.enter(|| {
34+
// Pass incoming triggers to the WASM module and return entity changes;
35+
// Stop when canceled because all RuntimeHosts and their senders were dropped.
36+
match mapping_request_receiver
37+
.map_err(|()| unreachable!())
38+
.for_each(move |request| -> Result<(), Error> {
39+
let MappingRequest {
40+
ctx,
41+
trigger,
42+
result_sender,
43+
} = request;
44+
45+
// Start the WASMI module runtime.
46+
let section = host_metrics.stopwatch.start_section("module_init");
47+
let module = WasmiModule::from_valid_module_with_ctx(
48+
valid_module.clone(),
49+
ctx,
50+
host_metrics.clone(),
51+
)?;
52+
section.end();
53+
54+
let section = host_metrics.stopwatch.start_section("run_handler");
55+
let result = match trigger {
56+
MappingTrigger::Log {
57+
transaction,
58+
log,
59+
params,
60+
handler,
61+
} => module.handle_ethereum_log(
62+
handler.handler.as_str(),
63+
transaction,
64+
log,
65+
params,
66+
),
67+
MappingTrigger::Call {
68+
transaction,
69+
call,
70+
inputs,
71+
outputs,
72+
handler,
73+
} => module.handle_ethereum_call(
74+
handler.handler.as_str(),
75+
transaction,
76+
call,
77+
inputs,
78+
outputs,
79+
),
80+
MappingTrigger::Block { handler } => {
81+
module.handle_ethereum_block(handler.handler.as_str())
82+
}
83+
};
84+
section.end();
85+
86+
result_sender
87+
.send((result, future::ok(Instant::now())))
88+
.map_err(|_| err_msg("WASM module result receiver dropped."))
89+
})
90+
.wait()
91+
{
92+
Ok(()) => debug!(logger, "Subgraph stopped, WASM runtime thread terminated"),
93+
Err(e) => debug!(logger, "WASM runtime thread terminated abnormally";
10494
"error" => e.to_string()),
105-
}
95+
}
96+
})
10697
})
10798
.map(|_| ())
10899
.map_err(|e| format_err!("Spawning WASM runtime thread failed: {}", e))?;

‎runtime/wasm/src/module/mod.rs

Lines changed: 10 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -92,13 +92,12 @@ fn format_wasmi_error(e: Error) -> String {
9292
}
9393

9494
/// A WASM module based on wasmi that powers a subgraph runtime.
95-
pub(crate) struct WasmiModule<U> {
95+
pub(crate) struct WasmiModule {
9696
pub module: ModuleRef,
9797
memory: MemoryRef,
9898

9999
pub ctx: MappingContext,
100100
pub(crate) valid_module: Arc<ValidModule>,
101-
pub(crate) task_sink: U,
102101
pub(crate) host_metrics: Arc<HostMetrics>,
103102

104103
// Time when the current handler began processing.
@@ -118,19 +117,11 @@ pub(crate) struct WasmiModule<U> {
118117
timeout_checkpoint_count: u64,
119118
}
120119

121-
impl<U> WasmiModule<U>
122-
where
123-
U: Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>
124-
+ Clone
125-
+ Send
126-
+ Sync
127-
+ 'static,
128-
{
120+
impl WasmiModule {
129121
/// Creates a new wasmi module
130122
pub fn from_valid_module_with_ctx(
131123
valid_module: Arc<ValidModule>,
132124
ctx: MappingContext,
133-
task_sink: U,
134125
host_metrics: Arc<HostMetrics>,
135126
) -> Result<Self, FailureError> {
136127
// Build import resolver
@@ -158,7 +149,6 @@ where
158149
memory,
159150
ctx,
160151
valid_module: valid_module.clone(),
161-
task_sink,
162152
host_metrics,
163153
start_time: Instant::now(),
164154
running_start: true,
@@ -323,14 +313,7 @@ where
323313
}
324314
}
325315

326-
impl<U> AscHeap for WasmiModule<U>
327-
where
328-
U: Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>
329-
+ Clone
330-
+ Send
331-
+ Sync
332-
+ 'static,
333-
{
316+
impl AscHeap for WasmiModule {
334317
fn raw_new(&mut self, bytes: &[u8]) -> Result<u32, Error> {
335318
// We request large chunks from the AssemblyScript allocator to use as arenas that we
336319
// manage directly.
@@ -370,14 +353,7 @@ where
370353
impl<E> HostError for HostExportError<E> where E: fmt::Debug + fmt::Display + Send + Sync + 'static {}
371354

372355
// Implementation of externals.
373-
impl<U> WasmiModule<U>
374-
where
375-
U: Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>
376-
+ Clone
377-
+ Send
378-
+ Sync
379-
+ 'static,
380-
{
356+
impl WasmiModule {
381357
fn gas(&mut self) -> Result<Option<RuntimeValue>, Trap> {
382358
// This function is called so often that the overhead of calling `Instant::now()` every
383359
// time would be significant, so we spread out the checks.
@@ -490,12 +466,10 @@ where
490466
call_ptr: AscPtr<AscUnresolvedContractCall>,
491467
) -> Result<Option<RuntimeValue>, Trap> {
492468
let call = self.asc_get(call_ptr);
493-
let result = self.ctx.host_exports.ethereum_call(
494-
&mut self.task_sink,
495-
&mut self.ctx.logger,
496-
&self.ctx.block,
497-
call,
498-
)?;
469+
let result =
470+
self.ctx
471+
.host_exports
472+
.ethereum_call(&mut self.ctx.logger, &self.ctx.block, call)?;
499473
Ok(Some(match result {
500474
Some(tokens) => RuntimeValue::from(self.asc_new(tokens.as_slice())),
501475
None => RuntimeValue::from(0),
@@ -581,10 +555,7 @@ where
581555
/// function ipfs.cat(link: String): Bytes
582556
fn ipfs_cat(&mut self, link_ptr: AscPtr<AscString>) -> Result<Option<RuntimeValue>, Trap> {
583557
let link = self.asc_get(link_ptr);
584-
let ipfs_res = self
585-
.ctx
586-
.host_exports
587-
.ipfs_cat(&self.ctx.logger, &mut self.task_sink, link);
558+
let ipfs_res = self.ctx.host_exports.ipfs_cat(&self.ctx.logger, link);
588559
match ipfs_res {
589560
Ok(bytes) => {
590561
let bytes_obj: AscPtr<Uint8Array> = self.asc_new(&*bytes);
@@ -952,14 +923,7 @@ where
952923
}
953924
}
954925

955-
impl<U> Externals for WasmiModule<U>
956-
where
957-
U: Sink<SinkItem = Box<dyn Future<Item = (), Error = ()> + Send>>
958-
+ Clone
959-
+ Send
960-
+ Sync
961-
+ 'static,
962-
{
926+
impl Externals for WasmiModule {
963927
fn invoke_index(
964928
&mut self,
965929
index: usize,

0 commit comments

Comments
 (0)