Adapt transitive crates based on port of task_executor and fs to toki…

…o 0.2 and stdlib futures.

src/rust/engine/ASYNC.md

@@ -0,0 +1,9 @@
+
+# async-await port notes
+
+Many functions at the boundary between ported async-await, stdlib futures code and legacy
+future 0.1 code temporarily return futures 0.3 BoxFuture and use explicit lifetimes, because that
+is easier for a futures 0.1 consumer. stdlib futures consumers can easily call async functions with
+references (because they can remain "on the stack"), but an 0.1 future cannot. These methods be
+swapped back to async once all callers are using async-await.
+

src/rust/engine/Cargo.toml

@@ -81,12 +81,14 @@ default-members = [
 ]
 
 [dependencies]
+async-trait = "0.1"
 boxfuture = { path = "boxfuture" }
 bytes = "0.4.5"
 concrete_time = { path = "concrete_time" }
 fnv = "1.0.5"
 fs = { path = "fs" }
 futures01 = { package = "futures", version = "0.1" }
+futures = { version = "0.3", features = ["compat"] }
 graph = { path = "graph" }
 hashing = { path = "hashing" }
 indexmap = "1.0.2"
@@ -109,6 +111,7 @@ ui = { path = "ui" }
 url = "1.7.1"
 uuid = { version = "0.7", features = ["v4"] }
 task_executor = { path = "task_executor" }
+tokio = { version = "0.2", features = ["rt-threaded"] }
 workunit_store = { path = "workunit_store" }
 
 [patch.crates-io]

src/rust/engine/engine_cffi/Cargo.toml

@@ -11,6 +11,7 @@ crate-type = ["cdylib"]
 [dependencies]
 engine = { path = ".." }
 futures01 = { package = "futures", version = "0.1" }
+futures = { version = "0.3", features = ["compat"] }
 hashing = { path = "../hashing" }
 log = "0.4"
 logging = { path = "../logging" }

src/rust/engine/engine_cffi/src/lib.rs

@@ -44,6 +44,7 @@ use engine::{
   externs, nodes, Core, ExecutionRequest, Function, Handle, Key, Params, RootResult, Rule,
   Scheduler, Session, Tasks, TypeId, Types, Value,
 };
+use futures::compat::Future01CompatExt;
 use futures01::{future, Future};
 use hashing::{Digest, EMPTY_DIGEST};
 use log::{error, warn, Log};
@@ -888,7 +889,8 @@ pub extern "C" fn capture_snapshots(
           })
           .collect::<Vec<_>>(),
       )
-      .map(|values| externs::store_tuple(&values)),
+      .map(|values| externs::store_tuple(&values))
+      .compat(),
     )
   })
   .into()
@@ -918,11 +920,10 @@ pub extern "C" fn merge_directories(
     scheduler
       .core
       .executor
-      .block_on(store::Snapshot::merge_directories(
-        scheduler.core.store(),
-        digests,
-        workunit_store,
-      ))
+      .block_on(
+        store::Snapshot::merge_directories(scheduler.core.store(), digests, workunit_store)
+          .compat(),
+      )
       .map(|dir| nodes::Snapshot::store_directory(&scheduler.core, &dir))
       .into()
   })
@@ -967,13 +968,11 @@ pub extern "C" fn run_local_interactive_process(
               None => unreachable!()
             };
 
-            let write_operation = scheduler.core.store().materialize_directory(
+            scheduler.core.store().materialize_directory(
               destination,
               digest,
               session.workunit_store(),
-            );
-
-            scheduler.core.executor.spawn_on_io_pool(write_operation).wait()?;
+            ).wait()?;
           }
         }
 
@@ -1050,7 +1049,7 @@ pub extern "C" fn materialize_directories(
     let types = &scheduler.core.types;
     let construct_materialize_directories_results = types.construct_materialize_directories_results;
     let construct_materialize_directory_result = types.construct_materialize_directory_result;
-    let work_future = future::join_all(
+    future::join_all(
       digests_and_path_prefixes
         .into_iter()
         .map(|(digest, path_prefix)| {
@@ -1097,9 +1096,8 @@ pub extern "C" fn materialize_directories(
         &[externs::store_tuple(&entries)],
       );
       output
-    });
-
-    scheduler.core.executor.spawn_on_io_pool(work_future).wait()
+    })
+    .wait()
   })
   .into()
 }
@@ -1194,7 +1192,7 @@ where
   F: FnOnce(&Scheduler) -> T,
 {
   let scheduler = unsafe { Box::from_raw(scheduler_ptr) };
-  let t = f(&scheduler);
+  let t = scheduler.core.runtime.enter(|| f(&scheduler));
   mem::forget(scheduler);
   t
 }

src/rust/engine/fs/Cargo.toml

@@ -6,9 +6,9 @@ authors = [ "Pants Build <pantsbuild@gmail.com>" ]
 publish = false
 
 [dependencies]
-boxfuture = { path = "../boxfuture" }
+async-trait = "0.1"
 bytes = "0.4.5"
-futures01 = { package = "futures", version = "0.1" }
+futures = "0.3"
 glob = "0.2.11"
 ignore = "0.4.4"
 lazy_static = "1"
@@ -19,5 +19,4 @@ tempfile = "3"
 
 [dev-dependencies]
 testutil = { path = "../testutil" }
-tokio = "0.1"
-tokio-compat = "0.1"
+tokio = { version = "0.2", features = ["rt-core", "macros"] }