Dynamic import (#2516)

cli/state.rs

@@ -118,9 +118,9 @@ impl Loader for ThreadSafeState {
     &self,
     specifier: &str,
     referrer: &str,
-    is_root: bool,
+    is_main: bool,
   ) -> Result<ModuleSpecifier, ErrBox> {
-    if !is_root {
+    if !is_main {
       if let Some(import_map) = &self.import_map {
         let result = import_map.resolve(specifier, referrer)?;
         if result.is_some() {
@@ -138,12 +138,14 @@ impl Loader for ThreadSafeState {
     module_specifier: &ModuleSpecifier,
   ) -> Box<deno::SourceCodeInfoFuture> {
     self.metrics.resolve_count.fetch_add(1, Ordering::SeqCst);
+    let module_url_specified = module_specifier.to_string();
     Box::new(self.fetch_compiled_module(module_specifier).map(
       |compiled_module| deno::SourceCodeInfo {
         // Real module name, might be different from initial specifier
         // due to redirections.
         code: compiled_module.code,
-        module_name: compiled_module.name,
+        module_url_specified,
+        module_url_found: compiled_module.name,
       },
     ))
   }

cli/worker.rs

@@ -5,6 +5,7 @@ use crate::tokio_util;
 use deno;
 use deno::ErrBox;
 use deno::ModuleSpecifier;
+use deno::RecursiveLoad;
 use deno::StartupData;
 use futures::Async;
 use futures::Future;
@@ -28,10 +29,24 @@ impl Worker {
     let isolate = Arc::new(Mutex::new(deno::Isolate::new(startup_data, false)));
     {
       let mut i = isolate.lock().unwrap();
+
       let state_ = state.clone();
       i.set_dispatch(move |op_id, control_buf, zero_copy_buf| {
         state_.dispatch(op_id, control_buf, zero_copy_buf)
       });
+
+      let state_ = state.clone();
+      i.set_dyn_import(move |id, specifier, referrer| {
+        let load_stream = RecursiveLoad::dynamic_import(
+          id,
+          specifier,
+          referrer,
+          state_.clone(),
+          state_.modules.clone(),
+        );
+        Box::new(load_stream)
+      });
+
       let state_ = state.clone();
       i.set_js_error_create(move |v8_exception| {
         JSError::from_v8_exception(v8_exception, &state_.ts_compiler)
@@ -66,12 +81,9 @@ impl Worker {
     let loader = self.state.clone();
     let isolate = self.isolate.clone();
     let modules = self.state.modules.clone();
-    let recursive_load = deno::RecursiveLoad::new(
+    let recursive_load =
-      &module_specifier.to_string(),
+      RecursiveLoad::main(&module_specifier.to_string(), loader, modules)
-      loader,
+        .get_future(isolate);
-      isolate,
-      modules,
-    );
     recursive_load.and_then(move |id| -> Result<(), ErrBox> {
       worker.state.progress.done();
       if is_prefetch {

core/isolate.rs

@@ -1,8 +1,9 @@
 // Copyright 2018 the Deno authors. All rights reserved. MIT license.
 
-// Do not add dependenies to modules.rs. it should remain decoupled from the
+// Do not add any dependency to modules.rs!
-// isolate to keep the Isolate struct from becoming too bloating for users who
+// modules.rs is complex and should remain decoupled from isolate.rs to keep the
-// do not need asynchronous module loading.
+// Isolate struct from becoming too bloating for users who do not need
+// asynchronous module loading.
 
 use crate::any_error::ErrBox;
 use crate::js_errors::CoreJSError;
@@ -18,7 +19,9 @@ use crate::libdeno::Snapshot1;
 use crate::libdeno::Snapshot2;
 use crate::shared_queue::SharedQueue;
 use crate::shared_queue::RECOMMENDED_SIZE;
-use futures::stream::{FuturesUnordered, Stream};
+use futures::stream::FuturesUnordered;
+use futures::stream::Stream;
+use futures::stream::StreamFuture;
 use futures::task;
 use futures::Async::*;
 use futures::Future;
@@ -27,6 +30,7 @@ use libc::c_char;
 use libc::c_void;
 use std::ffi::CStr;
 use std::ffi::CString;
+use std::fmt;
 use std::ptr::null;
 use std::sync::{Arc, Mutex, Once};
 
@@ -57,6 +61,76 @@ pub struct Script<'a> {
   pub filename: &'a str,
 }
 
+/// Represent result of fetching the source code of a module. Found module URL
+/// might be different from specified URL used for loading due to redirections
+/// (like HTTP 303). E.G. Both https://example.com/a.ts and
+/// https://example.com/b.ts may point to https://example.com/c.ts
+/// By keeping track of specified and found URL we can alias modules and avoid
+/// recompiling the same code 3 times.
+#[derive(Debug, Eq, PartialEq)]
+pub struct SourceCodeInfo {
+  pub code: String,
+  pub module_url_specified: String,
+  pub module_url_found: String,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub enum RecursiveLoadEvent {
+  Fetch(SourceCodeInfo),
+  Instantiate(deno_mod),
+}
+
+pub trait ImportStream: Stream {
+  fn register(
+    &mut self,
+    source_code_info: SourceCodeInfo,
+    isolate: &mut Isolate,
+  ) -> Result<(), ErrBox>;
+}
+
+type DynImportStream =
+  Box<dyn ImportStream<Item = RecursiveLoadEvent, Error = ErrBox> + Send>;
+
+type DynImportFn = Fn(deno_dyn_import_id, &str, &str) -> DynImportStream;
+
+/// Wraps DynImportStream to include the deno_dyn_import_id, so that it doesn't
+/// need to be exposed.
+#[derive(Debug)]
+struct DynImport {
+  pub id: deno_dyn_import_id,
+  pub inner: DynImportStream,
+}
+
+impl fmt::Debug for DynImportStream {
+  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+    write!(f, "DynImportStream(..)")
+  }
+}
+
+impl Stream for DynImport {
+  type Item = (deno_dyn_import_id, RecursiveLoadEvent);
+  type Error = (deno_dyn_import_id, ErrBox);
+
+  fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+    match self.inner.poll() {
+      Ok(Ready(Some(event))) => Ok(Ready(Some((self.id, event)))),
+      Ok(Ready(None)) => unreachable!(),
+      Err(e) => Err((self.id, e)),
+      Ok(NotReady) => Ok(NotReady),
+    }
+  }
+}
+
+impl ImportStream for DynImport {
+  fn register(
+    &mut self,
+    source_code_info: SourceCodeInfo,
+    isolate: &mut Isolate,
+  ) -> Result<(), ErrBox> {
+    self.inner.register(source_code_info, isolate)
+  }
+}
+
 // TODO(ry) It's ugly that we have both Script and OwnedScript. Ideally we
 // wouldn't expose such twiddly complexity.
 struct OwnedScript {
@@ -83,31 +157,8 @@ pub enum StartupData<'a> {
   None,
 }
 
-pub type DynImportFuture =
-  Box<dyn Future<Item = deno_mod, Error = ErrBox> + Send>;
-type DynImportFn = dyn Fn(&str, &str) -> DynImportFuture;
-
 type JSErrorCreateFn = dyn Fn(V8Exception) -> ErrBox;
 
-/// Wraps DynImportFuture to include the deno_dyn_import_id, so that it doesn't
-/// need to be exposed.
-struct DynImport {
-  id: deno_dyn_import_id,
-  inner: DynImportFuture,
-}
-
-impl Future for DynImport {
-  type Item = (deno_dyn_import_id, deno_mod);
-  type Error = (deno_mod, ErrBox);
-  fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-    match self.inner.poll() {
-      Ok(Ready(mod_id)) => Ok(Ready((self.id, mod_id))),
-      Ok(NotReady) => Ok(NotReady),
-      Err(e) => Err((self.id, e)),
-    }
-  }
-}
-
 /// A single execution context of JavaScript. Corresponds roughly to the "Web
 /// Worker" concept in the DOM. An Isolate is a Future that can be used with
 /// Tokio.  The Isolate future complete when there is an error or when all
@@ -125,7 +176,7 @@ pub struct Isolate {
   needs_init: bool,
   shared: SharedQueue,
   pending_ops: FuturesUnordered<PendingOpFuture>,
-  pending_dyn_imports: FuturesUnordered<DynImport>,
+  pending_dyn_imports: FuturesUnordered<StreamFuture<DynImport>>,
   have_unpolled_ops: bool,
   startup_script: Option<OwnedScript>,
 }
@@ -208,7 +259,10 @@ impl Isolate {
 
   pub fn set_dyn_import<F>(&mut self, f: F)
   where
-    F: Fn(&str, &str) -> DynImportFuture + Send + Sync + 'static,
+    F: Fn(deno_dyn_import_id, &str, &str) -> DynImportStream
+      + Send
+      + Sync
+      + 'static,
   {
     self.dyn_import = Some(Arc::new(f));
   }
@@ -257,10 +311,10 @@ impl Isolate {
     debug!("dyn_import specifier {} referrer {} ", specifier, referrer);
 
     if let Some(ref f) = isolate.dyn_import {
-      let inner = f(specifier, referrer);
+      let inner = f(id, specifier, referrer);
-      let fut = DynImport { inner, id };
+      let stream = DynImport { inner, id };
       task::current().notify();
-      isolate.pending_dyn_imports.push(fut);
+      isolate.pending_dyn_imports.push(stream.into_future());
     } else {
       panic!("dyn_import callback not set")
     }
@@ -434,18 +488,19 @@ impl Isolate {
     let (mod_id, maybe_err_str) = match result {
       Ok(mod_id) => (mod_id, None),
       Err(None) => (0, None),
-      Err(Some(err_str)) => (0, Some(err_str)),
+      Err(Some(err_str)) => (0, Some(CString::new(err_str).unwrap())),
+    };
+    let err_str_ptr = match maybe_err_str {
+      Some(ref err_str) => err_str.as_ptr(),
+      None => std::ptr::null(),
     };
-    let err_ptr = maybe_err_str
-      .map(|e| e.as_ptr() as *const c_char)
-      .unwrap_or(std::ptr::null());
     unsafe {
       libdeno::deno_dyn_import_done(
         self.libdeno_isolate,
         self.as_raw_ptr(),
         id,
         mod_id,
-        err_ptr,
+        err_str_ptr,
       )
     };
     self.check_last_exception().map_err(|err| {
@@ -453,6 +508,46 @@ impl Isolate {
       err
     })
   }
+
+  fn poll_dyn_imports(&mut self) -> Poll<(), ErrBox> {
+    use RecursiveLoadEvent::*;
+    loop {
+      match self.pending_dyn_imports.poll() {
+        Ok(NotReady) | Ok(Ready(None)) => {
+          // There are no active dynamic import loaders, or none are ready.
+          return Ok(futures::Async::Ready(()));
+        }
+        Ok(Ready(Some((
+          Some((dyn_import_id, Fetch(source_code_info))),
+          mut stream,
+        )))) => {
+          // A module (not necessarily the one dynamically imported) has been
+          // fetched. Create and register it, and if successful, poll for the
+          // next recursive-load event related to this dynamic import.
+          match stream.register(source_code_info, self) {
+            Ok(()) => self.pending_dyn_imports.push(stream.into_future()),
+            Err(err) => {
+              self.dyn_import_done(dyn_import_id, Err(Some(err.to_string())))?
+            }
+          }
+        }
+        Ok(Ready(Some((Some((dyn_import_id, Instantiate(module_id))), _)))) => {
+          // The top-level module from a dynamic import has been instantiated.
+          match self.mod_evaluate(module_id) {
+            Ok(()) => self.dyn_import_done(dyn_import_id, Ok(module_id))?,
+            Err(..) => self.dyn_import_done(dyn_import_id, Err(None))?,
+          }
+        }
+        Err(((dyn_import_id, err), _)) => {
+          // A non-javascript error occurred; this could be due to a an invalid
+          // module specifier, or a problem with the source map, or a failure
+          // to fetch the module source code.
+          self.dyn_import_done(dyn_import_id, Err(Some(err.to_string())))?
+        }
+        Ok(Ready(Some((None, _)))) => unreachable!(),
+      }
+    }
+  }
 }
 
 /// Called during mod_instantiate() to resolve imports.
@@ -556,20 +651,7 @@ impl Future for Isolate {
 
     loop {
       // If there are any pending dyn_import futures, do those first.
-      loop {
+      self.poll_dyn_imports()?;
-        match self.pending_dyn_imports.poll() {
-          Ok(NotReady) | Ok(Ready(None)) => break,
-          Ok(Ready(Some((dyn_import_id, mod_id)))) => {
-            match self.mod_evaluate(mod_id) {
-              Ok(()) => self.dyn_import_done(dyn_import_id, Ok(mod_id))?,
-              Err(..) => self.dyn_import_done(dyn_import_id, Err(None))?,
-            }
-          }
-          Err((dyn_import_id, err)) => {
-            self.dyn_import_done(dyn_import_id, Err(Some(err.to_string())))?
-          }
-        }
-      }
 
       // Now handle actual ops.
       self.have_unpolled_ops = false;
@@ -612,7 +694,7 @@ impl Future for Isolate {
     self.check_last_exception()?;
 
     // We're idle if pending_ops is empty.
-    if self.pending_ops.is_empty() {
+    if self.pending_ops.is_empty() && self.pending_dyn_imports.is_empty() {
       Ok(futures::Async::Ready(()))
     } else {
       if self.have_unpolled_ops {
@@ -658,10 +740,11 @@ pub mod tests {
   use futures::future::lazy;
   use futures::future::ok;
   use futures::Async;
+  use std::io;
   use std::ops::FnOnce;
   use std::sync::atomic::{AtomicUsize, Ordering};
 
-  fn run_in_task<F, R>(f: F) -> R
+  pub fn run_in_task<F, R>(f: F) -> R
   where
     F: FnOnce() -> R,
   {
@@ -864,6 +947,40 @@ pub mod tests {
     });
   }
 
+  struct MockImportStream(Vec<Result<RecursiveLoadEvent, ErrBox>>);
+
+  impl Stream for MockImportStream {
+    type Item = RecursiveLoadEvent;
+    type Error = ErrBox;
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+      let event = if self.0.is_empty() {
+        None
+      } else {
+        Some(self.0.remove(0)?)
+      };
+      Ok(Ready(event))
+    }
+  }
+
+  impl ImportStream for MockImportStream {
+    fn register(
+      &mut self,
+      module_data: SourceCodeInfo,
+      isolate: &mut Isolate,
+    ) -> Result<(), ErrBox> {
+      let id = isolate.mod_new(
+        false,
+        &module_data.module_url_found,
+        &module_data.code,
+      )?;
+      println!(
+        "MockImportStream register {} {}",
+        id, module_data.module_url_found
+      );
+      Ok(())
+    }
+  }
+
   #[test]
   fn dyn_import_err() {
     // Test an erroneous dynamic import where the specified module isn't found.
@@ -871,13 +988,13 @@ pub mod tests {
       let count = Arc::new(AtomicUsize::new(0));
       let count_ = count.clone();
       let mut isolate = Isolate::new(StartupData::None, false);
-      isolate.set_dyn_import(move |specifier, referrer| {
+      isolate.set_dyn_import(move |_, specifier, referrer| {
         count_.fetch_add(1, Ordering::Relaxed);
         assert_eq!(specifier, "foo.js");
         assert_eq!(referrer, "dyn_import2.js");
-        Box::new(futures::future::err(
+        let err = io::Error::from(io::ErrorKind::NotFound);
-          std::io::Error::new(std::io::ErrorKind::Other, "oh no!").into(),
+        let stream = MockImportStream(vec![Err(err.into())]);
-        ))
+        Box::new(stream)
       });
       js_check(isolate.execute(
         "dyn_import2.js",
@@ -902,17 +1019,29 @@ pub mod tests {
       let count_ = count.clone();
 
       // Sometimes Rust is really annoying.
-      use std::sync::Mutex;
       let mod_b = Arc::new(Mutex::new(0));
       let mod_b2 = mod_b.clone();
 
       let mut isolate = Isolate::new(StartupData::None, false);
-      isolate.set_dyn_import(move |_specifier, referrer| {
+      isolate.set_dyn_import(move |_id, specifier, referrer| {
-        count_.fetch_add(1, Ordering::Relaxed);
+        let c = count_.fetch_add(1, Ordering::Relaxed);
-        // assert_eq!(specifier, "foo.js");
+        match c {
+          0 => assert_eq!(specifier, "foo1.js"),
+          1 => assert_eq!(specifier, "foo2.js"),
+          _ => unreachable!(),
+        }
         assert_eq!(referrer, "dyn_import3.js");
-        let mod_id = mod_b2.lock().unwrap();
+        let mod_id = *mod_b2.lock().unwrap();
-        Box::new(futures::future::ok(*mod_id))
+        let source_code_info = SourceCodeInfo {
+          module_url_specified: "foo.js".to_owned(),
+          module_url_found: "foo.js".to_owned(),
+          code: "".to_owned(),
+        };
+        let stream = MockImportStream(vec![
+          Ok(RecursiveLoadEvent::Fetch(source_code_info)),
+          Ok(RecursiveLoadEvent::Instantiate(mod_id)),
+        ]);
+        Box::new(stream)
       });
 
       // Instantiate mod_b

core/module_specifier.rs

@@ -41,7 +41,7 @@ impl fmt::Display for ModuleResolutionError {
   }
 }
 
-#[derive(Debug, Clone, PartialEq)]
+#[derive(Debug, Clone, Eq, Hash, PartialEq)]
 /// Resolved module specifier
 pub struct ModuleSpecifier(Url);
 
@@ -50,6 +50,10 @@ impl ModuleSpecifier {
     &self.0
   }
 
+  pub fn as_str(&self) -> &str {
+    self.0.as_str()
+  }
+
   /// Resolves module using this algorithm:
   /// https://html.spec.whatwg.org/multipage/webappapis.html#resolve-a-module-specifier
   pub fn resolve_import(

core/modules.rs

@@ -7,32 +7,26 @@
 // synchronously. The isolate.rs module should never depend on this module.
 
 use crate::any_error::ErrBox;
+use crate::isolate::ImportStream;
 use crate::isolate::Isolate;
+use crate::isolate::RecursiveLoadEvent as Event;
+use crate::isolate::SourceCodeInfo;
+use crate::libdeno::deno_dyn_import_id;
 use crate::libdeno::deno_mod;
 use crate::module_specifier::ModuleSpecifier;
-use futures::Async;
+use futures::future::loop_fn;
+use futures::future::Loop;
+use futures::stream::FuturesUnordered;
+use futures::stream::Stream;
+use futures::Async::*;
 use futures::Future;
 use futures::Poll;
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fmt;
-use std::marker::PhantomData;
 use std::sync::Arc;
 use std::sync::Mutex;
 
-/// Represent result of fetching the source code of a module.
-/// Contains both module name and code.
-/// Module name might be different from initial URL used for loading
-/// due to redirections.
-/// e.g. Both https://example.com/a.ts and https://example.com/b.ts
-/// may point to https://example.com/c.ts. By specifying module_name
-/// all be https://example.com/c.ts in module_name (for aliasing),
-/// we avoid recompiling the same code for 3 different times.
-pub struct SourceCodeInfo {
-  pub module_name: String,
-  pub code: String,
-}
-
 pub type SourceCodeInfoFuture =
   dyn Future<Item = SourceCodeInfo, Error = ErrBox> + Send;
 
@@ -45,7 +39,7 @@ pub trait Loader: Send + Sync {
     &self,
     specifier: &str,
     referrer: &str,
-    is_root: bool,
+    is_main: bool,
   ) -> Result<ModuleSpecifier, ErrBox>;
 
   /// Given ModuleSpecifier, load its source code.
@@ -55,123 +49,167 @@ pub trait Loader: Send + Sync {
   ) -> Box<SourceCodeInfoFuture>;
 }
 
-struct PendingLoad {
+#[derive(Debug, Eq, PartialEq)]
-  url: String,
+enum Kind {
-  is_root: bool,
+  Main,
-  source_code_info_future: Box<SourceCodeInfoFuture>,
+  DynamicImport(deno_dyn_import_id),
+}
+
+#[derive(Debug, Eq, PartialEq)]
+enum State {
+  ResolveMain(String),           // specifier
+  ResolveImport(String, String), // specifier, referrer
+  LoadingRoot,
+  LoadingImports(deno_mod),
+  Instantiated(deno_mod),
 }
 
 /// This future is used to implement parallel async module loading without
-/// complicating the Isolate API. Note that RecursiveLoad will take ownership of
+/// complicating the Isolate API.
-/// an Isolate during load.
+/// TODO: RecursiveLoad desperately needs to be merged with Modules.
 pub struct RecursiveLoad<L: Loader> {
+  kind: Kind,
+  state: State,
   loader: L,
-  isolate: Arc<Mutex<Isolate>>,
   modules: Arc<Mutex<Modules>>,
-  pending: Vec<PendingLoad>,
+  pending: FuturesUnordered<Box<SourceCodeInfoFuture>>,
-  is_pending: HashSet<String>,
+  is_pending: HashSet<ModuleSpecifier>,
-  phantom: PhantomData<L>,
-  // TODO(ry) The following can all be combined into a single enum State type.
-  root: Option<String>,           // Empty before polled.
-  root_specifier: Option<String>, // Empty after first poll
-  root_id: Option<deno_mod>,
 }
 
 impl<L: Loader> RecursiveLoad<L> {
-  /// Starts a new parallel load of the given URL.
+  /// Starts a new parallel load of the given URL of the main module.
-  pub fn new(
+  pub fn main(
-    url: &str,
+    specifier: &str,
+    loader: L,
+    modules: Arc<Mutex<Modules>>,
+  ) -> Self {
+    let kind = Kind::Main;
+    let state = State::ResolveMain(specifier.to_owned());
+    Self::new(kind, state, loader, modules)
+  }
+
+  pub fn dynamic_import(
+    id: deno_dyn_import_id,
+    specifier: &str,
+    referrer: &str,
+    loader: L,
+    modules: Arc<Mutex<Modules>>,
+  ) -> Self {
+    let kind = Kind::DynamicImport(id);
+    let state = State::ResolveImport(specifier.to_owned(), referrer.to_owned());
+    Self::new(kind, state, loader, modules)
+  }
+
+  pub fn dyn_import_id(&self) -> Option<deno_dyn_import_id> {
+    match self.kind {
+      Kind::Main => None,
+      Kind::DynamicImport(id) => Some(id),
+    }
+  }
+
+  fn new(
+    kind: Kind,
+    state: State,
     loader: L,
-    isolate: Arc<Mutex<Isolate>>,
     modules: Arc<Mutex<Modules>>,
   ) -> Self {
     Self {
+      kind,
+      state,
       loader,
-      isolate,
       modules,
-      root: None,
+      pending: FuturesUnordered::new(),
-      root_specifier: Some(url.to_string()),
-      root_id: None,
-      pending: Vec::new(),
       is_pending: HashSet::new(),
-      phantom: PhantomData,
     }
   }
 
-  fn add(
+  fn add_root(&mut self) -> Result<(), ErrBox> {
+    let module_specifier = match self.state {
+      State::ResolveMain(ref specifier) => {
+        self.loader.resolve(specifier, ".", true)?
+      }
+      State::ResolveImport(ref specifier, ref referrer) => {
+        self.loader.resolve(specifier, referrer, false)?
+      }
+      _ => unreachable!(),
+    };
+
+    // We deliberately do not check if this module is already present in the
+    // module map. That's because the module map doesn't track whether a
+    // a module's dependencies have been loaded and whether it's been
+    // instantiated, so if we did find this module in the module map and used
+    // its id, this could lead to a crash.
+    //
+    // For the time being code and metadata for a module specifier is fetched
+    // multiple times, register() uses only the first result, and assigns the
+    // same module id to all instances.
+    //
+    // TODO: this is very ugly. The module map and recursive loader should be
+    // integrated into one thing.
+    self
+      .pending
+      .push(Box::new(self.loader.load(&module_specifier)));
+    self.state = State::LoadingRoot;
+
+    Ok(())
+  }
+
+  fn add_import(
     &mut self,
     specifier: &str,
     referrer: &str,
-    parent_id: Option<deno_mod>,
+    parent_id: deno_mod,
-  ) -> Result<String, ErrBox> {
+  ) -> Result<(), ErrBox> {
-    let is_root = parent_id.is_none();
+    let module_specifier = self.loader.resolve(specifier, referrer, false)?;
-    let module_specifier = self.loader.resolve(specifier, referrer, is_root)?;
+    let module_name = module_specifier.as_str();
-    let module_name = module_specifier.to_string();
 
-    if !is_root {
+    let mut modules = self.modules.lock().unwrap();
+
+    modules.add_child(parent_id, module_name);
+
+    if !modules.is_registered(module_name)
+      && !self.is_pending.contains(&module_specifier)
     {
-        let mut m = self.modules.lock().unwrap();
+      self
-        m.add_child(parent_id.unwrap(), &module_name);
+        .pending
+        .push(Box::new(self.loader.load(&module_specifier)));
+      self.is_pending.insert(module_specifier);
+    }
+
+    Ok(())
+  }
+
+  /// Returns a future that resolves to the final module id of the root module.
+  /// This future needs to take ownership of the isolate.
+  pub fn get_future(
+    self,
+    isolate: Arc<Mutex<Isolate>>,
+  ) -> impl Future<Item = deno_mod, Error = ErrBox> {
+    loop_fn(self, move |load| {
+      let isolate = isolate.clone();
+      load.into_future().map_err(|(e, _)| e).and_then(
+        move |(event, mut load)| {
+          Ok(match event.unwrap() {
+            Event::Fetch(info) => {
+              let mut isolate = isolate.lock().unwrap();
+              load.register(info, &mut isolate)?;
+              Loop::Continue(load)
+            }
+            Event::Instantiate(id) => Loop::Break(id),
+          })
+        },
+      )
+    })
   }
 }
 
-    {
+impl<L: Loader> ImportStream for RecursiveLoad<L> {
-      // #B We only add modules that have not yet been resolved for RecursiveLoad.
+  // TODO: this should not be part of RecursiveLoad.
-      // Only short circuit after add_child().
+  fn register(
-      // This impacts possible conditions in #A.
+    &mut self,
-      let modules = self.modules.lock().unwrap();
+    source_code_info: SourceCodeInfo,
-      if modules.is_registered(&module_name) {
+    isolate: &mut Isolate,
-        return Ok(module_name);
+  ) -> Result<(), ErrBox> {
-      }
-    }
-
-    if !self.is_pending.contains(&module_name) {
-      self.is_pending.insert(module_name.to_string());
-      let source_code_info_future = { self.loader.load(&module_specifier) };
-      self.pending.push(PendingLoad {
-        url: module_name.to_string(),
-        source_code_info_future,
-        is_root,
-      });
-    }
-
-    Ok(module_name)
-  }
-}
-
-impl<L: Loader> Future for RecursiveLoad<L> {
-  type Item = deno_mod;
-  type Error = ErrBox;
-
-  fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-    if self.root.is_none() && self.root_specifier.is_some() {
-      let s = self.root_specifier.take().unwrap();
-      match self.add(&s, ".", None) {
-        Err(err) => {
-          return Err(err);
-        }
-        Ok(root) => {
-          self.root = Some(root);
-        }
-      }
-    }
-    assert!(self.root_specifier.is_none());
-    assert!(self.root.is_some());
-
-    let mut i = 0;
-    while i < self.pending.len() {
-      let pending = &mut self.pending[i];
-      match pending.source_code_info_future.poll() {
-        Err(err) => {
-          return Err(err);
-        }
-        Ok(Async::NotReady) => {
-          i += 1;
-        }
-        Ok(Async::Ready(source_code_info)) => {
-          // We have completed loaded one of the modules.
-          let completed = self.pending.remove(i);
-
     // #A There are 3 cases to handle at this moment:
     // 1. Source code resolved result have the same module name as requested
     //    and is not yet registered
@@ -181,83 +219,99 @@ impl<L: Loader> Future for RecursiveLoad<L> {
     //     -> alias
     //   2b. The module with resolved module name has not yet been registerd
     //     -> register & alias
-          let is_module_registered = {
+    let SourceCodeInfo {
-            let modules = self.modules.lock().unwrap();
+      code,
-            modules.is_registered(&source_code_info.module_name)
+      module_url_specified,
-          };
+      module_url_found,
+    } = source_code_info;
 
-          let need_alias = source_code_info.module_name != completed.url;
+    let is_main = self.kind == Kind::Main && self.state == State::LoadingRoot;
 
-          if !is_module_registered {
+    let module_id = {
-            let module_name = &source_code_info.module_name;
-
-            let mod_id = {
-              let isolate = self.isolate.lock().unwrap();
-              isolate.mod_new(
-                completed.is_root,
-                module_name,
-                &source_code_info.code,
-              )
-            }?;
-
-            if completed.is_root {
-              assert!(self.root_id.is_none());
-              self.root_id = Some(mod_id);
-            }
-
-            // Register new module.
-            {
       let mut modules = self.modules.lock().unwrap();
-              modules.register(mod_id, module_name);
+
-              // If necessary, register the alias.
+      // If necessary, register an alias.
-              if need_alias {
+      if module_url_specified != module_url_found {
-                let module_alias = &completed.url;
+        modules.alias(&module_url_specified, &module_url_found);
-                modules.alias(module_alias, module_name);
+      }
+
+      match modules.get_id(&module_url_found) {
+        // Module has already been registered.
+        Some(id) => {
+          debug!(
+            "Already-registered module fetched again: {}",
+            module_url_found
+          );
+          id
+        }
+        // Module not registered yet, do it now.
+        None => {
+          let id = isolate.mod_new(is_main, &module_url_found, &code)?;
+          modules.register(id, &module_url_found);
+          id
         }
       }
+    };
 
     // Now we must iterate over all imports of the module and load them.
-            let imports = {
+    let imports = isolate.mod_get_imports(module_id);
-              let isolate = self.isolate.lock().unwrap();
+    for import in imports {
-              isolate.mod_get_imports(mod_id)
+      self.add_import(&import, &module_url_found, module_id)?;
+    }
+
+    // If we just finished loading the root module, store the root module id.
+    match self.state {
+      State::LoadingRoot => self.state = State::LoadingImports(module_id),
+      State::LoadingImports(..) => {}
+      _ => unreachable!(),
     };
-            let referrer = module_name;
-            for specifier in imports {
-              self.add(&specifier, referrer, Some(mod_id))?;
-            }
-          } else if need_alias {
-            let mut modules = self.modules.lock().unwrap();
-            modules.alias(&completed.url, &source_code_info.module_name);
-          }
-        }
-      }
-    }
 
-    if !self.pending.is_empty() {
+    // If all imports have been loaded, instantiate the root module.
-      return Ok(Async::NotReady);
+    if self.pending.is_empty() {
-    }
+      let root_id = match self.state {
+        State::LoadingImports(mod_id) => mod_id,
+        _ => unreachable!(),
+      };
 
-    let root_id = self.root_id.unwrap();
+      let mut resolve_cb =
-
+        |specifier: &str, referrer_id: deno_mod| -> deno_mod {
-    let mut resolve_cb = |specifier: &str, referrer_id: deno_mod| -> deno_mod {
           let modules = self.modules.lock().unwrap();
           let referrer = modules.get_name(referrer_id).unwrap();
-      // this callback is only called for non-root modules
+          match self.loader.resolve(specifier, &referrer, is_main) {
-      match self.loader.resolve(specifier, &referrer, false) {
+            Ok(specifier) => modules.get_id(specifier.as_str()).unwrap_or(0),
-        Ok(specifier) => match modules.get_id(&specifier.to_string()) {
+            // We should have already resolved and Ready this module, so
-          Some(id) => id,
-          None => 0,
-        },
-        // We should have already resolved and loaded this module, so
             // resolve() will not fail this time.
-        Err(_err) => unreachable!(),
+            Err(..) => unreachable!(),
           }
         };
+      isolate.mod_instantiate(root_id, &mut resolve_cb)?;
 
-    let mut isolate = self.isolate.lock().unwrap();
+      self.state = State::Instantiated(root_id);
-    isolate
+    }
-      .mod_instantiate(root_id, &mut resolve_cb)
+
-      .map(|_| Async::Ready(root_id))
+    Ok(())
+  }
+}
+
+impl<L: Loader> Stream for RecursiveLoad<L> {
+  type Item = Event;
+  type Error = ErrBox;
+
+  fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+    Ok(match self.state {
+      State::ResolveMain(..) | State::ResolveImport(..) => {
+        self.add_root()?;
+        self.poll()?
+      }
+      State::LoadingRoot | State::LoadingImports(..) => {
+        match self.pending.poll()? {
+          Ready(None) => unreachable!(),
+          Ready(Some(info)) => Ready(Some(Event::Fetch(info))),
+          NotReady => NotReady,
+        }
+      }
+      State::Instantiated(id) => Ready(Some(Event::Instantiate(id))),
+    })
   }
 }
 
@@ -519,6 +573,7 @@ mod tests {
   use super::*;
   use crate::isolate::js_check;
   use crate::isolate::tests::*;
+  use futures::Async;
   use std::error::Error;
   use std::fmt;
 
@@ -557,7 +612,7 @@ mod tests {
       "/dir/redirect3.js" => Some((REDIRECT3_SRC, "file:///redirect3.js")),
       "/slow.js" => Some((SLOW_SRC, "file:///slow.js")),
       "/never_ready.js" => {
-        Some(("should never be loaded", "file:///never_ready.js"))
+        Some(("should never be Ready", "file:///never_ready.js"))
       }
       "/main.js" => Some((MAIN_SRC, "file:///main.js")),
       "/bad_import.js" => Some((BAD_IMPORT_SRC, "file:///bad_import.js")),
@@ -594,15 +649,20 @@ mod tests {
 
     fn poll(&mut self) -> Poll<Self::Item, ErrBox> {
       self.counter += 1;
-      if self.url == "file:///never_ready.js"
+      if self.url == "file:///never_ready.js" {
-        || (self.url == "file:///slow.js" && self.counter < 2)
+        return Ok(Async::NotReady);
-      {
+      }
+      if self.url == "file:///slow.js" && self.counter < 2 {
+        // TODO(ry) Hopefully in the future we can remove current task
+        // notification. See comment above run_in_task.
+        futures::task::current().notify();
         return Ok(Async::NotReady);
       }
       match mock_source_code(&self.url) {
         Some(src) => Ok(Async::Ready(SourceCodeInfo {
           code: src.0.to_owned(),
-          module_name: src.1.to_owned(),
+          module_url_specified: self.url.clone(),
+          module_url_found: src.1.to_owned(),
         })),
         None => Err(MockError::LoadErr.into()),
       }
@@ -679,20 +739,34 @@ mod tests {
     if (import.meta.url != 'file:///d.js') throw Error();
   "#;
 
+  // TODO(ry) Sadly FuturesUnordered requires the current task to be set. So
+  // even though we are only using poll() in these tests and not Tokio, we must
+  // nevertheless run it in the tokio executor. Ideally run_in_task can be
+  // removed in the future.
+  use crate::isolate::tests::run_in_task;
+
   #[test]
   fn test_recursive_load() {
+    run_in_task(|| {
       let loader = MockLoader::new();
       let modules = loader.modules.clone();
       let modules_ = modules.clone();
       let isolate = loader.isolate.clone();
       let isolate_ = isolate.clone();
       let loads = loader.loads.clone();
-    let mut recursive_load =
+      let mut recursive_load = RecursiveLoad::main("/a.js", loader, modules);
-      RecursiveLoad::new("/a.js", loader, isolate, modules);
+
+      let a_id = loop {
+        match recursive_load.poll() {
+          Ok(Ready(Some(Event::Fetch(info)))) => {
+            let mut isolate = isolate.lock().unwrap();
+            recursive_load.register(info, &mut isolate).unwrap();
+          }
+          Ok(Ready(Some(Event::Instantiate(id)))) => break id,
+          _ => panic!("unexpected result"),
+        };
+      };
 
-    let result = recursive_load.poll();
-    assert!(result.is_ok());
-    if let Async::Ready(a_id) = result.ok().unwrap() {
       let mut isolate = isolate_.lock().unwrap();
       js_check(isolate.mod_evaluate(a_id));
 
@@ -730,9 +804,7 @@ mod tests {
         Some(&vec!["file:///d.js".to_string()])
       );
       assert_eq!(modules.get_children(d_id), Some(&vec![]));
-    } else {
+    })
-      unreachable!();
-    }
   }
 
   const CIRCULAR1_SRC: &str = r#"
@@ -753,16 +825,16 @@ mod tests {
 
   #[test]
   fn test_circular_load() {
+    run_in_task(|| {
       let loader = MockLoader::new();
       let isolate = loader.isolate.clone();
       let isolate_ = isolate.clone();
       let modules = loader.modules.clone();
       let modules_ = modules.clone();
       let loads = loader.loads.clone();
-    let mut recursive_load =
+      let recursive_load =
-      RecursiveLoad::new("/circular1.js", loader, isolate, modules);
+        RecursiveLoad::main("/circular1.js", loader, modules);
-
+      let result = recursive_load.get_future(isolate.clone()).poll();
-    let result = recursive_load.poll();
       assert!(result.is_ok());
       if let Async::Ready(circular1_id) = result.ok().unwrap() {
         let mut isolate = isolate_.lock().unwrap();
@@ -805,6 +877,7 @@ mod tests {
       } else {
         unreachable!();
       }
+    })
   }
 
   const REDIRECT1_SRC: &str = r#"
@@ -823,16 +896,17 @@ mod tests {
 
   #[test]
   fn test_redirect_load() {
+    run_in_task(|| {
       let loader = MockLoader::new();
       let isolate = loader.isolate.clone();
       let isolate_ = isolate.clone();
       let modules = loader.modules.clone();
       let modules_ = modules.clone();
       let loads = loader.loads.clone();
-    let mut recursive_load =
+      let recursive_load =
-      RecursiveLoad::new("/redirect1.js", loader, isolate, modules);
+        RecursiveLoad::main("/redirect1.js", loader, modules);
-
+      let result = recursive_load.get_future(isolate.clone()).poll();
-    let result = recursive_load.poll();
+      println!(">> result {:?}", result);
       assert!(result.is_ok());
       if let Async::Ready(redirect1_id) = result.ok().unwrap() {
         let mut isolate = isolate_.lock().unwrap();
@@ -866,6 +940,7 @@ mod tests {
       } else {
         unreachable!();
       }
+    })
   }
 
   // main.js
@@ -886,28 +961,26 @@ mod tests {
 
   #[test]
   fn slow_never_ready_modules() {
+    run_in_task(|| {
       let loader = MockLoader::new();
       let isolate = loader.isolate.clone();
       let modules = loader.modules.clone();
       let loads = loader.loads.clone();
       let mut recursive_load =
-      RecursiveLoad::new("/main.js", loader, isolate, modules);
+        RecursiveLoad::main("/main.js", loader, modules).get_future(isolate);
 
       let result = recursive_load.poll();
       assert!(result.is_ok());
       assert!(result.ok().unwrap().is_not_ready());
 
-    {
+      // TODO(ry) Arguably the first time we poll only the following modules
-      let l = loads.lock().unwrap();
+      // should be loaded:
-      assert_eq!(
+      //      "file:///main.js",
-        l.to_vec(),
+      //      "file:///never_ready.js",
-        vec![
+      //      "file:///slow.js"
-          "file:///main.js",
+      // But due to current task notification in DelayedSourceCodeFuture they
-          "file:///never_ready.js",
+      // all get loaded in a single poll. Also see the comment above
-          "file:///slow.js"
+      // run_in_task.
-        ]
-      );
-    }
 
       for _ in 0..10 {
         let result = recursive_load.poll();
@@ -927,6 +1000,7 @@ mod tests {
           ]
         );
       }
+    })
   }
 
   // bad_import.js
@@ -936,18 +1010,20 @@ mod tests {
 
   #[test]
   fn loader_disappears_after_error() {
+    run_in_task(|| {
       let loader = MockLoader::new();
       let isolate = loader.isolate.clone();
       let modules = loader.modules.clone();
-    let mut recursive_load =
+      let recursive_load =
-      RecursiveLoad::new("/bad_import.js", loader, isolate, modules);
+        RecursiveLoad::main("/bad_import.js", loader, modules);
-    let result = recursive_load.poll();
+      let result = recursive_load.get_future(isolate).poll();
       assert!(result.is_err());
       let err = result.err().unwrap();
       assert_eq!(
         err.downcast_ref::<MockError>().unwrap(),
         &MockError::ResolveErr
       );
+    })
   }
 
   #[test]

tests/015_duplicate_parallel_import.js

@@ -0,0 +1,20 @@
+// Importing the same module in parallel, the module should only be
+// instantiated once.
+
+const promises = new Array(100)
+  .fill(null)
+  .map(() => import("./subdir/mod1.ts"));
+
+Promise.all(promises).then(imports => {
+  const mod = imports.reduce((first, cur) => {
+    if (typeof first !== "object") {
+      throw new Error("Expected an object.");
+    }
+    if (first !== cur) {
+      throw new Error("More than one instance of the same module.");
+    }
+    return first;
+  });
+
+  mod.printHello3();
+});

tests/015_duplicate_parallel_import.js.out

@@ -0,0 +1 @@
+Hello

tests/015_duplicate_parallel_import.test

@@ -0,0 +1,2 @@
+args: tests/015_duplicate_parallel_import.js --reload
+output: tests/015_duplicate_parallel_import.js.out

tests/error_014_catch_dynamic_import_error.js

@@ -0,0 +1,31 @@
+(async () => {
+  try {
+    await import("does not exist");
+  } catch (err) {
+    console.log("Caught direct dynamic import error.");
+    console.log(err);
+  }
+
+  try {
+    await import("./subdir/indirect_import_error.js");
+  } catch (err) {
+    console.log("Caught indirect direct dynamic import error.");
+    console.log(err);
+  }
+
+  try {
+    await import("./subdir/throws.js");
+  } catch (err) {
+    console.log("Caught error thrown by dynamically imported module.");
+    console.log(err);
+  }
+
+  try {
+    await import("./subdir/indirect_throws.js");
+  } catch (err) {
+    console.log(
+      "Caught error thrown indirectly by dynamically imported module."
+    );
+    console.log(err);
+  }
+})();

tests/error_014_catch_dynamic_import_error.js.out

@@ -0,0 +1,12 @@
+Caught direct dynamic import error.
+TypeError: relative import path "does not exist" not prefixed with / or ./ or ../
+
+Caught indirect direct dynamic import error.
+TypeError: relative import path "does not exist either" not prefixed with / or ./ or ../
+
+Caught error thrown by dynamically imported module.
+Error: An error
+    at file:///[WILDCARD]tests/subdir/throws.js:5:7
+Caught error thrown indirectly by dynamically imported module.
+Error: An error
+    at file:///[WILDCARD]tests/subdir/throws.js:5:7

tests/error_014_catch_dynamic_import_error.test

@@ -0,0 +1,2 @@
+args: tests/error_014_catch_dynamic_import_error.js --reload
+output: tests/error_014_catch_dynamic_import_error.js.out

tests/subdir/indirect_import_error.js

@@ -0,0 +1 @@
+export * from "does not exist either";

tests/subdir/indirect_throws.js

@@ -0,0 +1 @@
+export * from "./throws.js";

tests/subdir/throws.js

@@ -0,0 +1,5 @@
+export function boo() {
+  console.log("Boo!");
+}
+
+throw new Error("An error");