runtime: Add a metric that counts currently active tasks

Adds a new metric for the number of active tasks  for the Runtime. We count
the elements on the OwnedTasks LinkedList, and access that through RuntimeMetrics.

Closes: #5400

Author: matildasmeds

tokio/src/runtime/metrics/runtime.rs

@@ -68,6 +68,25 @@ impl RuntimeMetrics {
         self.handle.inner.num_blocking_threads()
     }
 
+    /// Returns the number of active tasks in the runtime.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// use tokio::runtime::Handle;
+    ///
+    /// #[tokio::main]
+    /// async fn main() {
+    ///    let metrics = Handle::current().metrics();
+    ///
+    ///     let n = metrics.active_tasks_count();
+    ///     println!("Runtime has {} active tasks", n);
+    /// }
+    /// ```
+    pub fn active_tasks_count(&self) -> usize {
+        self.handle.inner.active_tasks_count()
+    }
+
     /// Returns the number of idle threads, which have spawned by the runtime
     /// for `spawn_blocking` calls.
     ///

tokio/src/runtime/scheduler/current_thread.rs

@@ -430,6 +430,10 @@ cfg_metrics! {
         pub(crate) fn blocking_queue_depth(&self) -> usize {
             self.blocking_spawner.queue_depth()
         }
+
+        pub(crate) fn active_tasks_count(&self) -> usize {
+            self.shared.owned.active_tasks_count()
+        }
     }
 }
 

tokio/src/runtime/scheduler/mod.rs

@@ -135,6 +135,14 @@ cfg_rt! {
                 }
             }
 
+            pub(crate) fn active_tasks_count(&self) -> usize {
+                match self {
+                    Handle::CurrentThread(handle) => handle.active_tasks_count(),
+                    #[cfg(all(feature = "rt-multi-thread", not(tokio_wasi)))]
+                    Handle::MultiThread(handle) => handle.active_tasks_count(),
+                }
+            }
+
             pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
                 match self {
                     Handle::CurrentThread(handle) => handle.scheduler_metrics(),

tokio/src/runtime/scheduler/multi_thread/handle.rs

@@ -69,6 +69,10 @@ cfg_metrics! {
             self.blocking_spawner.num_idle_threads()
         }
 
+        pub(crate) fn active_tasks_count(&self) -> usize {
+            self.shared.owned.active_tasks_count()
+        }
+
         pub(crate) fn scheduler_metrics(&self) -> &SchedulerMetrics {
             &self.shared.scheduler_metrics
         }

tokio/src/runtime/task/list.rs

@@ -153,6 +153,10 @@ impl<S: 'static> OwnedTasks<S> {
         }
     }
 
+    pub(crate) fn active_tasks_count(&self) -> usize {
+        self.inner.lock().list.count()
+    }
+
     pub(crate) fn remove(&self, task: &Task<S>) -> Option<Task<S>> {
         let task_id = task.header().get_owner_id();
         if task_id == 0 {

tokio/src/util/linked_list.rs

@@ -174,6 +174,21 @@ impl<L: Link> LinkedList<L, L::Target> {
         true
     }
 
+    // Counts the elements of the LinkedList
+    pub(crate) fn count(&self) -> usize {
+        let mut count = 0;
+        let mut current = self.head;
+
+        while let Some(node) = current {
+            unsafe {
+                count += 1;
+                current = L::pointers(node).as_ref().get_next()
+            }
+        }
+
+        count
+    }
+
     /// Removes the specified node from the list
     ///
     /// # Safety

tokio/tests/rt_metrics.rs

@@ -81,6 +81,23 @@ fn blocking_queue_depth() {
     assert_eq!(0, rt.metrics().blocking_queue_depth());
 }
 
+#[test]
+fn active_tasks_count() {
+    let rt = current_thread();
+    let metrics = rt.metrics();
+    assert_eq!(0, metrics.active_tasks_count());
+    rt.spawn(async move {
+        assert_eq!(1, metrics.active_tasks_count());
+    });
+
+    let rt = threaded();
+    let metrics = rt.metrics();
+    assert_eq!(0, metrics.active_tasks_count());
+    rt.spawn(async move {
+        assert_eq!(1, metrics.active_tasks_count());
+    });
+}
+
 #[test]
 fn remote_schedule_count() {
     use std::thread;