Merge pull request #1613 from microsoft/ab-sched-refactor

scheduler: cleanup: simplify logic and cleanup

Author: anandbonde

src/catnap/linux/transport.rs

@@ -89,7 +89,7 @@ impl SharedCatnapTransport {
             options: TcpSocketOptions::new(config)?,
         }));
         let mut me2 = me.clone();
-        runtime.insert_io_polling_coroutine(
+        runtime.schedule_polling_coroutine(
             "bgc::catnap::transport::epoll",
             Box::pin(async move { me2.poll().await }.fuse()),
         )?;

src/catnap/win/overlapped.rs

@@ -474,9 +474,7 @@ mod tests {
         })
         .fuse();
 
-        let server_task = runtime
-            .insert_nonpolling_coroutine("ioc_server", Box::pin(server))
-            .unwrap();
+        let server_task = runtime.schedule_coroutine("ioc_server", Box::pin(server)).unwrap();
         post_completion(&iocp, overlapped.as_mut().marshal(), COMPLETION_KEY)?;
 
         iocp.process_events()?;
@@ -584,7 +582,7 @@ mod tests {
         );
 
         let mut runtime = SharedDemiRuntime::default();
-        let server_task = runtime.insert_nonpolling_coroutine("ioc_server", server).unwrap();
+        let server_task = runtime.schedule_coroutine("ioc_server", server).unwrap();
 
         let mut wait_for_state = |state| -> Result<(), Fail> {
             while server_state_view.load(Ordering::Relaxed) < state {
@@ -689,9 +687,7 @@ mod tests {
         .fuse();
 
         let mut runtime = SharedDemiRuntime::default();
-        let server_task = runtime
-            .insert_nonpolling_coroutine("ioc_server", Box::pin(server))
-            .unwrap();
+        let server_task = runtime.schedule_coroutine("ioc_server", Box::pin(server)).unwrap();
 
         ensure_eq!(
             server_state_view.load(Ordering::Relaxed),
@@ -702,7 +698,6 @@ mod tests {
         let iocp_ref = unsafe { &mut *iocp.get() };
         iocp_ref.process_events()?;
 
-        // Poll the runtime again, which
         let result = loop {
             // Move time forward, which should time out the operation.
             runtime.advance_clock(Instant::now());

src/catnap/win/transport.rs

@@ -78,7 +78,7 @@ impl SharedCatnapTransport {
             runtime: runtime.clone(),
         }));
 
-        runtime.insert_io_polling_coroutine(
+        runtime.schedule_polling_coroutine(
             "bgc::catnap::transport::epoll",
             Box::pin({
                 let mut me = me.clone();

src/demikernel/libos/network/libos.rs

@@ -162,7 +162,7 @@ impl<T: NetworkTransport> SharedNetworkLibOS<T> {
             let coroutine = Box::pin(self.clone().accept_coroutine(qd).fuse());
             self.runtime
                 .clone()
-                .insert_nonpolling_coroutine("ioc::network::libos::accept", coroutine)
+                .schedule_coroutine("ioc::network::libos::accept", coroutine)
         };
 
         queue.accept(coroutine_constructor)
@@ -213,7 +213,7 @@ impl<T: NetworkTransport> SharedNetworkLibOS<T> {
             let coroutine = Box::pin(self.clone().connect_coroutine(qd, remote).fuse());
             self.runtime
                 .clone()
-                .insert_nonpolling_coroutine("ioc::network::libos::connect", coroutine)
+                .schedule_coroutine("ioc::network::libos::connect", coroutine)
         };
 
         queue.connect(coroutine_constructor)
@@ -253,7 +253,7 @@ impl<T: NetworkTransport> SharedNetworkLibOS<T> {
             let coroutine = Box::pin(self.clone().close_coroutine(qd).fuse());
             self.runtime
                 .clone()
-                .insert_nonpolling_coroutine("ioc::network::libos::close", coroutine)
+                .schedule_coroutine("ioc::network::libos::close", coroutine)
         };
 
         queue.close(coroutine_constructor)
@@ -322,7 +322,7 @@ impl<T: NetworkTransport> SharedNetworkLibOS<T> {
             let coroutine = Box::pin(self.clone().push_coroutine(qd, bufs, None).fuse());
             self.runtime
                 .clone()
-                .insert_nonpolling_coroutine("ioc::network::libos::push", coroutine)
+                .schedule_coroutine("ioc::network::libos::push", coroutine)
         };
 
         queue.push(coroutine_constructor)
@@ -370,7 +370,7 @@ impl<T: NetworkTransport> SharedNetworkLibOS<T> {
             let coroutine = Box::pin(self.clone().push_coroutine(qd, bufs, Some(remote)).fuse());
             self.runtime
                 .clone()
-                .insert_nonpolling_coroutine("ioc::network::libos::pushto", coroutine)
+                .schedule_coroutine("ioc::network::libos::pushto", coroutine)
         };
 
         queue.push(coroutine_constructor)
@@ -390,7 +390,7 @@ impl<T: NetworkTransport> SharedNetworkLibOS<T> {
             let coroutine = Box::pin(self.clone().pop_coroutine(qd, size).fuse());
             self.runtime
                 .clone()
-                .insert_nonpolling_coroutine("ioc::network::libos::pop", coroutine)
+                .schedule_coroutine("ioc::network::libos::pop", coroutine)
         };
 
         queue.pop(coroutine_constructor)

src/inetstack/mod.rs

@@ -83,11 +83,11 @@ impl SharedInetStack {
             runtime: runtime.clone(),
             layer4_endpoint,
         }));
-        runtime.insert_io_polling_coroutine("bgc::inetstack::poll", Box::pin(me.clone().poll().fuse()))?;
+        runtime.schedule_polling_coroutine("bgc::inetstack::poll", Box::pin(me.clone().poll().fuse()))?;
         Ok(me)
     }
 
-    /// Scheduler will poll all futures that are ready to make progress.
+    /// Scheduler will run all futures that are ready to make progress.
     /// Then ask the runtime to receive new data which we will forward to the engine to parse and
     /// route to the correct protocol.
     pub async fn poll(mut self) {

src/inetstack/protocols/layer3/arp/peer.rs

@@ -82,7 +82,7 @@ impl SharedArpPeer {
             recv_queue: AsyncQueue::<DemiBuffer>::default(),
         }));
         // This is a future returned by the async function.
-        runtime.insert_nonpolling_coroutine("bgc::inetstack::arp::background", Box::pin(peer.clone().poll().fuse()))?;
+        runtime.schedule_coroutine("bgc::inetstack::arp::background", Box::pin(peer.clone().poll().fuse()))?;
         Ok(peer.clone())
     }
 

src/inetstack/protocols/layer3/arp/tests.rs

@@ -52,7 +52,7 @@ fn arp_immediate_reply() -> Result<()> {
     // Move clock forward and poll the engine.
     now += Duration::from_micros(1);
     engine.advance_clock(now);
-    engine.get_runtime().poll_background_tasks();
+    engine.get_runtime().run_background_tasks();
 
     // Check if the ARP cache outputs a reply message.
     let mut buffers = engine.pop_expected_frames(1);
@@ -120,7 +120,7 @@ fn arp_cache_update() -> Result<()> {
     // Move clock forward and poll the engine.
     now += Duration::from_micros(1);
     engine.advance_clock(now);
-    engine.get_runtime().poll_background_tasks();
+    engine.get_runtime().run_background_tasks();
 
     // Check if the ARP cache has been updated.
     let cache = engine.get_transport().export_arp_cache();
@@ -157,10 +157,10 @@ fn arp_cache_timeout() -> Result<()> {
     let _qt = engine
         .get_runtime()
         .clone()
-        .insert_nonpolling_coroutine("arp query", coroutine)?;
+        .schedule_coroutine("arp query", coroutine)?;
 
     for _ in 0..(ARP_RETRY_COUNT + 1) {
-        engine.get_runtime().poll_foreground_tasks();
+        engine.get_runtime().run_foreground_tasks();
         // Check if the ARP cache outputs a reply message.
         let buffers = engine.pop_expected_frames(1);
         crate::ensure_eq!(buffers.len(), 1);

src/inetstack/protocols/layer3/icmpv4/peer.rs

@@ -99,8 +99,7 @@ impl SharedIcmpv4Peer {
             rng,
             inflight: HashMap::<(u16, u16), InflightRequest>::new(),
         }));
-        runtime
-            .insert_nonpolling_coroutine("bgc::inetstack::icmp::background", Box::pin(peer.clone().poll().fuse()))?;
+        runtime.schedule_coroutine("bgc::inetstack::icmp::background", Box::pin(peer.clone().poll().fuse()))?;
         Ok(peer)
     }
 

src/inetstack/protocols/layer4/tcp/established/mod.rs

@@ -173,7 +173,7 @@ impl SharedEstablishedSocket {
             me.receive(header, data);
         }
         let me2 = me.clone();
-        runtime.insert_nonpolling_coroutine(
+        runtime.schedule_coroutine(
             "bgc::inetstack::tcp::established::background",
             Box::pin(async move { me2.background().await }.fuse()),
         )?;

src/inetstack/protocols/layer4/tcp/passive_open.rs

@@ -191,7 +191,7 @@ impl SharedPassiveSocket {
             .fuse();
         match self
             .runtime
-            .insert_nonpolling_coroutine("bgc::inetstack::tcp::passiveopen::background", Box::pin(future))
+            .schedule_coroutine("bgc::inetstack::tcp::passiveopen::background", Box::pin(future))
         {
             Ok(qt) => qt,
             Err(e) => {