Merge pull request #134 from JohnTitor/new-rt

Bump up `actix-rt` to 1.1.1

actix-macros/.gitignore

@@ -0,0 +1 @@
+/wip

actix-macros/Cargo.toml

@@ -8,7 +8,6 @@ documentation = "https://docs.rs/actix-macros/"
 categories = ["network-programming", "asynchronous"]
 license = "MIT/Apache-2.0"
 edition = "2018"
-workspace = ".."
 
 [lib]
 proc-macro = true
@@ -18,4 +17,7 @@ quote = "1.0.3"
 syn = { version = "^1", features = ["full"] }
 
 [dev-dependencies]
-actix-rt = { version = "1.0.0" }
+actix-rt = "1.0"
+
+futures = "0.3"
+trybuild = "1"

actix-macros/src/lib.rs

@@ -55,12 +55,11 @@ pub fn main(_: TokenStream, item: TokenStream) -> TokenStream {
 /// ```
 #[proc_macro_attribute]
 pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
-    let input = syn::parse_macro_input!(item as syn::ItemFn);
+    let mut input = syn::parse_macro_input!(item as syn::ItemFn);
-
-    let ret = &input.sig.output;
-    let name = &input.sig.ident;
-    let body = &input.block;
     let attrs = &input.attrs;
+    let vis = &input.vis;
+    let sig = &mut input.sig;
+    let body = &input.block;
     let mut has_test_attr = false;
 
     for attr in attrs {
@@ -69,7 +68,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
         }
     }
 
-    if input.sig.asyncness.is_none() {
+    if sig.asyncness.is_none() {
         return syn::Error::new_spanned(
             input.sig.fn_token,
             format!("only async fn is supported, {}", input.sig.ident),
@@ -78,10 +77,12 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
         .into();
     }
 
+    sig.asyncness = None;
+
     let result = if has_test_attr {
         quote! {
             #(#attrs)*
-            fn #name() #ret {
+            #vis #sig {
                 actix_rt::System::new("test")
                     .block_on(async { #body })
             }
@@ -90,7 +91,7 @@ pub fn test(_: TokenStream, item: TokenStream) -> TokenStream {
         quote! {
             #[test]
             #(#attrs)*
-            fn #name() #ret {
+            #vis #sig {
                 actix_rt::System::new("test")
                     .block_on(async { #body })
             }

actix-macros/tests/trybuild.rs

@@ -0,0 +1,9 @@
+#[test]
+fn compile_macros() {
+    let t = trybuild::TestCases::new();
+    t.pass("tests/trybuild/main-01-basic.rs");
+    t.compile_fail("tests/trybuild/main-02-only-async.rs");
+
+    t.pass("tests/trybuild/test-01-basic.rs");
+    t.pass("tests/trybuild/test-02-keep-attrs.rs");
+}

actix-macros/tests/trybuild/main-01-basic.rs

@@ -0,0 +1,4 @@
+#[actix_rt::main]
+async fn main() {
+    println!("Hello world");
+}

actix-macros/tests/trybuild/main-02-only-async.rs

@@ -0,0 +1,4 @@
+#[actix_rt::main]
+fn main() {
+    futures::future::ready(()).await
+}

actix-macros/tests/trybuild/main-02-only-async.stderr

@@ -0,0 +1,14 @@
+error: only async fn is supported
+ --> $DIR/main-02-only-async.rs:2:1
+  |
+2 | fn main() {
+  | ^^
+
+error[E0601]: `main` function not found in crate `$CRATE`
+ --> $DIR/main-02-only-async.rs:1:1
+  |
+1 | / #[actix_rt::main]
+2 | | fn main() {
+3 | |     futures::future::ready(()).await
+4 | | }
+  | |_^ consider adding a `main` function to `$DIR/tests/trybuild/main-02-only-async.rs`

actix-macros/tests/trybuild/test-01-basic.rs

@@ -0,0 +1,6 @@
+#[actix_rt::test]
+async fn my_test() {
+    assert!(true);
+}
+
+fn main() {}

actix-macros/tests/trybuild/test-02-keep-attrs.rs

@@ -0,0 +1,7 @@
+#[actix_rt::test]
+#[should_panic]
+async fn my_test() {
+    todo!()
+}
+
+fn main() {}

actix-rt/CHANGES.md

@@ -1,7 +1,17 @@
 # Changes
 
+## [1.1.1] - 2020-04-30
+
+### Fixed
+
+* Fix memory leak due to [#94] (see [#129] for more detail)
+
+[#129]: https://github.com/actix/actix-net/issues/129
+
 ## [1.1.0] - 2020-04-08
 
+**This version has been yanked.**
+
 ### Added
 
 * Expose `System::is_set` to check if current system has ben started [#99]

actix-rt/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "actix-rt"
-version = "1.1.0"
+version = "1.1.1"
 authors = ["Nikolay Kim <fafhrd91@gmail.com>"]
 description = "Actix runtime"
 keywords = ["network", "framework", "async", "futures"]
@@ -21,4 +21,5 @@ actix-threadpool = "0.3"
 futures-channel = { version = "0.3.4", default-features = false }
 futures-util = { version = "0.3.4", default-features = false, features = ["alloc"] }
 copyless = "0.1.4"
+smallvec = "1"
 tokio = { version = "0.2.6", default-features = false, features = ["rt-core", "rt-util", "io-driver", "tcp", "uds", "udp", "time", "signal", "stream"] }

actix-rt/src/arbiter.rs

@@ -18,13 +18,14 @@ use crate::system::System;
 
 use copyless::BoxHelper;
 
+use smallvec::SmallVec;
 pub use tokio::task::JoinHandle;
 
 thread_local!(
     static ADDR: RefCell<Option<Arbiter>> = RefCell::new(None);
     static RUNNING: Cell<bool> = Cell::new(false);
     static Q: RefCell<Vec<Pin<Box<dyn Future<Output = ()>>>>> = RefCell::new(Vec::new());
-    static PENDING: RefCell<Vec<JoinHandle<()>>> = RefCell::new(Vec::new());
+    static PENDING: RefCell<SmallVec<[JoinHandle<()>; 8]>> = RefCell::new(SmallVec::new());
     static STORAGE: RefCell<HashMap<TypeId, Box<dyn Any>>> = RefCell::new(HashMap::new());
 );
 
@@ -181,9 +182,15 @@ impl Arbiter {
         RUNNING.with(move |cell| {
             if cell.get() {
                 // Spawn the future on running executor
-                PENDING.with(move |cell| {
+                let len = PENDING.with(move |cell| {
-                    cell.borrow_mut().push(tokio::task::spawn_local(future));
+                    let mut p = cell.borrow_mut();
-                })
+                    p.push(tokio::task::spawn_local(future));
+                    p.len()
+                });
+                if len > 7 {
+                    // Before reaching the inline size
+                    tokio::task::spawn_local(CleanupPending);
+                }
             } else {
                 // Box the future and push it to the queue, this results in double boxing
                 // because the executor boxes the future again, but works for now
@@ -311,12 +318,36 @@ impl Arbiter {
     /// have completed.
     pub fn local_join() -> impl Future<Output = ()> {
         PENDING.with(move |cell| {
-            let current = cell.replace(Vec::new());
+            let current = cell.replace(SmallVec::new());
             future::join_all(current).map(|_| ())
         })
     }
 }
 
+/// Future used for cleaning-up already finished `JoinHandle`s
+/// from the `PENDING` list so the vector doesn't grow indefinitely
+struct CleanupPending;
+
+impl Future for CleanupPending {
+    type Output = ();
+
+    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        PENDING.with(move |cell| {
+            let mut pending = cell.borrow_mut();
+            let mut i = 0;
+            while i != pending.len() {
+                if let Poll::Ready(_) = Pin::new(&mut pending[i]).poll(cx) {
+                    pending.remove(i);
+                } else {
+                    i += 1;
+                }
+            }
+        });
+
+        Poll::Ready(())
+    }
+}
+
 struct ArbiterController {
     stop: Option<Sender<i32>>,
     rx: UnboundedReceiver<ArbiterCommand>,
@@ -350,9 +381,15 @@ impl Future for ArbiterController {
                         return Poll::Ready(());
                     }
                     ArbiterCommand::Execute(fut) => {
-                        PENDING.with(move |cell| {
+                        let len = PENDING.with(move |cell| {
-                            cell.borrow_mut().push(tokio::task::spawn_local(fut));
+                            let mut p = cell.borrow_mut();
+                            p.push(tokio::task::spawn_local(fut));
+                            p.len()
                         });
+                        if len > 7 {
+                            // Before reaching the inline size
+                            tokio::task::spawn_local(CleanupPending);
+                        }
                     }
                     ArbiterCommand::ExecuteFn(f) => {
                         f.call_box();

actix-server/src/config.rs

@@ -218,7 +218,7 @@ impl ServiceRuntime {
         // let name = name.to_owned();
         if let Some(token) = self.names.get(name) {
             self.services.insert(
-                token.clone(),
+                *token,
                 Box::new(ServiceFactory {
                     inner: service.into_factory(),
                 }),

string/src/lib.rs

@@ -1,10 +1,11 @@
-//! A utl-8 encoded read-only string with Bytes as a storage.
+//! A UTF-8 encoded read-only string using Bytes as storage.
+
 use std::convert::TryFrom;
 use std::{borrow, fmt, hash, ops, str};
 
 use bytes::Bytes;
 
-/// A utf-8 encoded string with [`Bytes`] as a storage.
+/// A UTF-8 encoded string with [`Bytes`] as a storage.
 ///
 /// [`Bytes`]: https://docs.rs/bytes/0.5.3/bytes/struct.Bytes.html
 #[derive(Clone, Eq, Ord, PartialOrd, Default)]