From 0041cc1f23fff4dd64c644c243ba8404122786c8 Mon Sep 17 00:00:00 2001
From: Abdulelah Bin Mahfoodh <abduh.bm@gmail.com>
Date: Tue, 26 May 2020 17:23:08 +0300
Subject: [PATCH] Include total number of tasks in the performance report
 (#3822)

* Include total number of tasks in the performance report

* Include tasks' timings in the performance report
---
 distributed/scheduler.py | 26 +++++++++++++++++++++-----
 1 file changed, 21 insertions(+), 5 deletions(-)

diff --git a/distributed/scheduler.py b/distributed/scheduler.py
index dd96550ed41..3eae685c355 100644
--- a/distributed/scheduler.py
+++ b/distributed/scheduler.py
@@ -1069,7 +1069,7 @@ def __init__(
         preload=None,
         preload_argv=(),
         plugins=(),
-        **kwargs
+        **kwargs,
     ):
         self._setup_logging(logger)
 
@@ -1352,7 +1352,7 @@ def __init__(
             connection_limit=connection_limit,
             deserialize=False,
             connection_args=self.connection_args,
-            **kwargs
+            **kwargs,
         )
 
         if self.worker_ttl:
@@ -2089,7 +2089,7 @@ def stimulus_task_erred(
                     exception=exception,
                     traceback=traceback,
                     worker=worker,
-                    **kwargs
+                    **kwargs,
                 )
         else:
             recommendations = {}
@@ -3389,7 +3389,7 @@ async def retire_workers(
         close_workers=False,
         names=None,
         lock=True,
-        **kwargs
+        **kwargs,
     ):
         """ Gracefully retire workers from cluster
 
@@ -4090,7 +4090,7 @@ def transition_processing_memory(
         typename=None,
         worker=None,
         startstops=None,
-        **kwargs
+        **kwargs,
     ):
         try:
             ts = self.tasks[key]
@@ -5055,6 +5055,15 @@ def profile_to_figure(state):
 
         # Task stream
         task_stream = self.get_task_stream(start=start)
+        total_tasks = len(task_stream)
+        timespent = defaultdict(int)
+        for d in task_stream:
+            for x in d["startstops"]:
+                timespent[x["action"]] += x["stop"] - x["start"]
+        tasks_timings = ""
+        for k in sorted(timespent.keys()):
+            tasks_timings += f"\n<li> {k} time: {format_time(timespent[k])} </li>"
+
         from .diagnostics.task_stream import rectangles
         from .dashboard.components.scheduler import task_stream_figure
 
@@ -5081,6 +5090,11 @@ def profile_to_figure(state):
         <i> Select different tabs on the top for additional information </i>
 
         <h2> Duration: {time} </h2>
+        <h2> Tasks Information </h2>
+        <ul>
+         <li> number of tasks: {ntasks} </li>
+         {tasks_timings}
+        </ul>
 
         <h2> Scheduler Information </h2>
         <ul>
@@ -5096,6 +5110,8 @@ def profile_to_figure(state):
         </pre>
         """.format(
             time=format_time(stop - start),
+            ntasks=total_tasks,
+            tasks_timings=tasks_timings,
             address=self.address,
             nworkers=len(self.workers),
             threads=sum(w.nthreads for w in self.workers.values()),