add test for non static block_on.

actix-rt/CHANGES.md

@@ -9,8 +9,7 @@
 ### Changed
 
 * Remove `'static` lifetime requirement for `Runtime::block_on` and `SystemRunner::block_on`. 
-  Remove `'static` lifetime requirement for `System::run`. 
-  Deprecate `Arbiter::is_running` and `Arbiter::local_join`. 
+  Remove `'static` lifetime requirement for `System::run` and `Builder::run`.
   `Arbiter::spawn` would panic when `System` is not in scope. [#207]
 
 ### Fixed

actix-rt/src/arbiter.rs

@@ -291,7 +291,7 @@ impl Future for CleanupPending {
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         HANDLE.with(move |handle| {
-            let _ = Pin::new(&mut *handle.borrow_mut()).poll_next(cx);
+            recycle_join_handle(&mut *handle.borrow_mut(), cx);
         });
 
         Poll::Ready(())
@@ -328,7 +328,7 @@ impl Future for ArbiterController {
                         HANDLE.with(|handle| {
                             let mut handle = handle.borrow_mut();
                             handle.push(tokio::task::spawn_local(fut));
-                            let _ = Pin::new(&mut *handle).poll_next(cx);
+                            recycle_join_handle(&mut *handle, cx);
                         });
                     }
                     ArbiterCommand::ExecuteFn(f) => {
@@ -341,6 +341,20 @@ impl Future for ArbiterController {
     }
 }
 
+fn recycle_join_handle(handle: &mut FuturesUnordered<JoinHandle<()>>, cx: &mut Context<'_>) {
+    let _ = Pin::new(&mut *handle).poll_next(cx);
+
+    // Try to recycle more join handles and free up memory.
+    //
+    // this is a guess. The yield limit for FuturesUnordered is 32.
+    // So poll an extra 3 times would make the total poll below 128.
+    if handle.len() > 64 {
+        (0..3).for_each(|_| {
+            let _ = Pin::new(&mut *handle).poll_next(cx);
+        })
+    }
+}
+
 #[derive(Debug)]
 pub(crate) enum SystemCommand {
     Exit(i32),

actix-rt/src/builder.rs

@@ -65,7 +65,7 @@ impl Builder {
     /// Function `f` get called within tokio runtime context.
     pub fn run<F>(self, f: F) -> io::Result<()>
     where
-        F: FnOnce() + 'static,
+        F: FnOnce(),
     {
         self.create_runtime(f).run()
     }
@@ -88,7 +88,7 @@ impl Builder {
 
     fn create_runtime<F>(self, f: F) -> SystemRunner
     where
-        F: FnOnce() + 'static,
+        F: FnOnce(),
     {
         let (stop_tx, stop) = channel();
         let (sys_sender, sys_receiver) = unbounded();

actix-rt/src/system.rs

@@ -256,7 +256,7 @@ impl System {
     /// Function `f` get called within tokio runtime context.
     pub fn run<F>(f: F) -> io::Result<()>
     where
-        F: FnOnce() + 'static,
+        F: FnOnce(),
     {
         Self::builder().run(f)
     }

actix-rt/tests/integration_tests.rs

@@ -98,3 +98,29 @@ fn join_current_arbiter() {
         "local_join should await only for the already spawned futures"
     );
 }
+
+#[test]
+fn non_static_block_on() {
+    let string = String::from("test_str");
+    let str = string.as_str();
+
+    let mut sys = actix_rt::System::new("borrow some");
+
+    sys.block_on(async {
+        actix_rt::time::delay_for(Duration::from_millis(1)).await;
+        assert_eq!("test_str", str);
+    });
+
+    let mut rt = actix_rt::Runtime::new().unwrap();
+
+    rt.block_on(async {
+        actix_rt::time::delay_for(Duration::from_millis(1)).await;
+        assert_eq!("test_str", str);
+    });
+
+    actix_rt::System::run(|| {
+        assert_eq!("test_str", str);
+        actix_rt::System::current().stop();
+    })
+    .unwrap();
+}