Repository: actix/actix-net (language: rust)

Issue:
actix-rt: leaking memory in every request
I just compiled example from github and run apache benchmarks `ab -n 1000000 -c 64 http://127.0.0.1:8001/test/john/index.html` and memory was constantly growing. After 1 milion requests it was 3.1 GB of heap.

I could reproduce the same on my catalina macbook pro and linux server. It's like every request is leaking few KB's of memory.


Discussion / hints:
Which example repo did you compile?
Recorded little demonstration of the issue on youtube https://www.youtube.com/watch?v=nQ0YxJ94B2s 

It's example from Readme
I've got circumstancial evidence (production service recently updated) of there being a leak introduced somewhere in v2.
@robjtede Do we have any leads as to where?
@cdbattags Not yet, personally noticed only recently, but we should keep an eye out and post any discoveries here.
@cdbattags I was able to debug it with MTuner and it showed me where are those allocations, but after turning it off It doesn't want to turn on again.

<details>
<summary>screencap</summary>

![image](https://user-images.githubusercontent.com/543264/78907538-1e8abc80-7a81-11ea-9a90-e32e6b2aa7e6.png)

</details>
Ok no problem @Lesiuk, sooner the better on getting any sort of debug information. This should be decently easy to squash we just need more data.
![image](https://user-images.githubusercontent.com/543264/78910339-146abd00-7a85-11ea-989b-64f2c44152f1.png)

First actix line is https://github.com/actix/actix-net/blob/master/actix-rt/src/arbiter.rs#L185
tokio::runtime:task::core::Cell is allocating memory which is never deallocated.
This change looks suspicious, I will build it with older tokio https://github.com/tokio-rs/tokio/commit/a78b1c65ccfb9692ca5d3ed8ddde934f40091d83
@Lesiuk why do you think that commit looks suspicious?
Looks like first usage of `unsafe` in this chain is https://github.com/tokio-rs/tokio/blob/master/tokio/src/task/local.rs#L200

I know that's not much to go off of but hmmm
@carllerche or @alexcrichton any ideas?
@Lesiuk deleted your screencap comment because?
Because I'm still debugging and I think I found out something.

It's working fine with actix-rt 1.0 and leaking with actix-rt 1.1.
There is a change in the line where it's leaking

![image](https://user-images.githubusercontent.com/543264/78919527-50584f00-7a92-11ea-8939-c861168a805a.png)

https://github.com/actix/actix-net/commit/f1685d82534e41a36ccff783e6af631d81db5247#diff-01ce174553e287daa3a0debf474e11c7

L177
Aren't elements never removed from this PENDING array?
@Jonathas-Conceicao any chance you can shed some light on this change?
The source of the increasing memory usage is the indeed the PENDING array. A call for `actix_rt::Arbiter::local_join` would replace the PENDING vector allowing memory to be freed, this small example can test it interactively.

```Rust
use actix_web::{get, post, App, HttpServer};

#[get("/")]
async fn hello() -> String {
    format!("Hello")
}

#[post("/")]
async fn free() -> String {
    let _ = actix_rt::Arbiter::local_join();
    format!("Clear")
}

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| App::new().service(hello).service(free))
        .bind("127.0.0.1:8080")?
        .run()
        .workers(1)
        .await
}
```

Nevertheless I do admit it's a poor design on my part for https://github.com/actix/actix-net/pull/94 and I'll try to think of a proper solution for it.
Sorry for my hastiness. I just want to make sure we're being proactive on these types of issues because we're getting a lot of bad "PR" for this 2+ release as it stands and a memory leak is the worst kind of news.

What's the ETA on a fix?
Since the leak only happens on `actix-rt:1.1.0`, users can just enforce a more stable version with `actix-rt = "=1.0.0"` on their `Cargo.toml`. As for the fix I'll have some time to work on it later today.
Maybe we can yank actix-rt:1.1.0 and release fixed version later on?
I agree yank it. Any point releasing the current version as `1.2.0-beta.1` for testing reasons?

ping @JohnTitor, thoughts on yanking?

Fix patch:
diff --git a/actix-rt/Cargo.toml b/actix-rt/Cargo.toml
index eb9965e078..2c0c1b5026 100644
--- a/actix-rt/Cargo.toml
+++ b/actix-rt/Cargo.toml
@@ -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"] }
diff --git a/actix-rt/src/arbiter.rs b/actix-rt/src/arbiter.rs
index 124a55802e..eff10ca3d0 100644
--- a/actix-rt/src/arbiter.rs
+++ b/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| {
-                    cell.borrow_mut().push(tokio::task::spawn_local(future));
-                })
+                let len = PENDING.with(move |cell| {
+                    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| {
-                            cell.borrow_mut().push(tokio::task::spawn_local(fut));
+                        let len = PENDING.with(move |cell| {
+                            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();


Test patch: