We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Monitoring visualization show negative CPU usage with retries
**Describe the bug**
Currently, the logged CPU time in monitoring db is cumulative. This works for most of the cases, except when a task is failed and retried. As a result, multiple figures in the visualization will show a sawtooth.

**To Reproduce**
Steps to reproduce the behavior, for e.g:
1. Setup Parsl master with Python 3.6 on cluster
2. Run all the tests and open the visualization
3. See error

**Expected behavior**
Each retried task should be on a new row.

**Possible solution**
Add a new column (e.g., `retries`) to the table as a new primary key.
Fix visualization based on that


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `parsl/monitoring/visualization/plots/default/workflow_resource_plots.py`
Content:
```
1 import math
2 import numpy as np
3 import pandas as pd
4 import plotly.graph_objs as go
5 from plotly.offline import plot
6 
7 
8 def resource_distribution_plot(df_resources, df_task, type='psutil_process_time_user', label='CPU Time Distribution', option='avg', columns=20,):
9     # E.g., psutil_process_time_user or psutil_process_memory_percent
10 
11     min_range = min(df_resources[type].astype('float'))
12     max_range = max(df_resources[type].astype('float'))
13     time_step = (max_range - min_range) / columns
14 
15     if min_range == max_range:
16         x_axis = [min_range]
17     else:
18         x_axis = []
19         for i in np.arange(min_range, max_range + time_step, time_step):
20             x_axis.append(i)
21 
22     apps_dict = dict()
23     for i in range(len(df_task)):
24         row = df_task.iloc[i]
25         apps_dict[row['task_id']] = []
26 
27     def y_axis_setup():
28         items = [0] * len(x_axis)
29 
30         for app, tasks in apps_dict.items():
31             if option == 'avg':
32                 task = df_resources[df_resources['task_id'] ==
33                                     app][type].astype('float').mean()
34             elif option == 'max':
35                 task = df_resources[df_resources['task_id'] == app][type].astype('float').max()
36 
37             for i in range(len(x_axis) - 1):
38                 a = task >= x_axis[i]
39                 b = task < x_axis[i + 1]
40                 if a and b:
41                     items[i] += 1
42                     break
43             if task >= x_axis[-1]:
44                 items[-1] += 1
45         return items
46 
47     if "memory" not in type:
48         xaxis = dict(autorange=True,
49                      title='CPU user time (seconds)')
50     else:
51         xaxis = dict(autorange=True,
52                      title='Memory usage (bytes)')
53     fig = go.Figure(
54         data=[go.Bar(x=x_axis,
55                      y=y_axis_setup(),
56                      name='tasks')],
57         layout=go.Layout(xaxis=xaxis,
58                          yaxis=dict(title='Tasks'),
59                          title=label + '(' + option + ')'))
60 
61     return plot(fig, show_link=False, output_type="div", include_plotlyjs=False)
62 
63 
64 def resource_time_series(tasks, type='psutil_process_time_user', label='CPU user time'):
65     tasks['epoch_time'] = (pd.to_datetime(
66         tasks['timestamp']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
67     step = int(tasks['resource_monitoring_interval'][0])
68     start = tasks['epoch_time'].min()
69     end = tasks['epoch_time'].max()
70     tasks['relative_time'] = tasks['epoch_time'] - start
71     if end != start:
72         bins = pd.cut(tasks['relative_time'],
73                       range(0, end - start + 1, step),
74                       include_lowest=True)
75         df = tasks.groupby(bins, as_index=False)[type].mean()
76         df['time'] = step * df.index
77         fig = go.Figure(
78             data=[go.Scatter(x=df['time'],
79                              y=df[type],
80                              )],
81             layout=go.Layout(xaxis=dict(autorange=True,
82                                         title='Time (seconds)'),
83                              yaxis=dict(title=label),
84                              title=label))
85     else:
86         fig = go.Figure(
87             data=[go.Scatter(x=[0],
88                              y=[tasks[type].mean()],
89                              )],
90             layout=go.Layout(xaxis=dict(autorange=True,
91                                         title='Time (seconds)'),
92                              yaxis=dict(title=label),
93                              title=label))
94     return plot(fig, show_link=False, output_type="div", include_plotlyjs=False)
95 
96 
97 def worker_efficiency(task, node):
98     try:
99         node['epoch_time'] = (pd.to_datetime(
100             node['timestamp']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
101         task['epoch_time_start'] = (pd.to_datetime(
102             task['task_try_time_launched']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
103         task['epoch_time_running'] = (pd.to_datetime(
104             task['task_try_time_running']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
105         task['epoch_time_returned'] = (pd.to_datetime(
106             task['task_try_time_returned']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
107         start = int(min(task['epoch_time_start'].min(), node['epoch_time'].min()))
108         end = int(task['epoch_time_returned'].max())
109 
110         worker_plot = [0] * (end - start + 1)
111         total_workers = node['worker_count'].sum()
112 
113         for i, row in task.iterrows():
114             if math.isnan(row['epoch_time_running']):
115                 # skip tasks with no running start time
116                 continue
117             if math.isnan(row['epoch_time_returned']):
118                 # there is no end time for this, so we should assume the "end" time
119                 etr = end
120             else:
121                 etr = int(row['epoch_time_returned'])
122             for j in range(int(row['epoch_time_running']), etr + 1):
123                 worker_plot[j - start] += 1
124         fig = go.Figure(
125             data=[go.Scatter(x=list(range(0, end - start + 1)),
126                              y=worker_plot,
127                              name='Total busy workers',
128                              ),
129                   go.Scatter(x=list(range(0, end - start + 1)),
130                              y=[total_workers] * (end - start + 1),
131                              name='Total online workers',
132                              )
133                  ],
134             layout=go.Layout(xaxis=dict(autorange=True,
135                                         title='Time (seconds)'),
136                              yaxis=dict(title='Number of workers'),
137                              title="Worker efficiency"))
138         return plot(fig, show_link=False, output_type="div", include_plotlyjs=False)
139     except Exception as e:
140         print(e)
141         return "The worker efficiency plot cannot be generated due to missing data."
142 
143 
144 def resource_efficiency(resource, node, label='CPU'):
145     try:
146         resource['epoch_time'] = (pd.to_datetime(
147             resource['timestamp']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
148         node['epoch_time'] = (pd.to_datetime(
149             node['timestamp']) - pd.Timestamp("1970-01-01")) // pd.Timedelta('1s')
150         resource = resource.sort_values(by='epoch_time')
151         start = min(resource['epoch_time'].min(), node['epoch_time'].min())
152         end = resource['epoch_time'].max()
153         resource['relative_time'] = resource['epoch_time'] - start
154         node['relative_time'] = node['epoch_time'] - start
155 
156         task_plot = [0] * (end - start + 1)
157         if label == 'CPU':
158             total = node['cpu_count'].sum()
159         elif label == 'mem':
160             total = node['total_memory'].sum() / 1024 / 1024 / 1024
161 
162         resource['total_cpu_time'] = resource['psutil_process_time_user'] + resource['psutil_process_time_system']
163         for task_id in resource['task_id'].unique():
164             tmp = resource[resource['task_id'] == task_id]
165             tmp['last_timestamp'] = tmp['relative_time'].shift(1)
166             if label == 'CPU':
167                 tmp['last_cputime'] = tmp['total_cpu_time'].shift(1)
168             for index, row in tmp.iterrows():
169                 if np.isnan(row['last_timestamp']):
170                     continue
171                 for i in range(int(row['last_timestamp']), int(row['relative_time'])):
172                     if label == 'CPU':
173                         diff = (row['total_cpu_time'] - row['last_cputime']) / (row['relative_time'] - row['last_timestamp'])
174                     elif label == 'mem':
175                         diff = row['psutil_process_memory_resident'] / 1024 / 1024 / 1024
176                     task_plot[i] += diff
177 
178         if label == 'CPU':
179             name1 = 'Used CPU cores'
180             name2 = 'Total CPU cores'
181             yaxis = 'Number of CPU cores'
182             title = 'CPU usage'
183         elif label == 'mem':
184             name1 = 'Used memory'
185             name2 = 'Total memory'
186             yaxis = 'Memory (GB)'
187             title = 'Memory usage'
188 
189         fig = go.Figure(
190             data=[go.Scatter(x=list(range(0, end - start + 1)),
191                              y=task_plot,
192                              name=name1,
193                              ),
194                   go.Scatter(x=list(range(0, end - start + 1)),
195                              y=[total] * (end - start + 1),
196                              name=name2,
197                              )
198                  ],
199             layout=go.Layout(xaxis=dict(autorange=True,
200                                         title='Time (seconds)'),
201                              yaxis=dict(title=yaxis),
202                              title=title))
203         return plot(fig, show_link=False, output_type="div", include_plotlyjs=False)
204     except Exception as e:
205         print(e)
206         return "The resource efficiency plot cannot be generated because of exception {}.".format(e)
207
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py b/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py
--- a/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py
+++ b/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py
@@ -152,6 +152,7 @@
         end = resource['epoch_time'].max()
         resource['relative_time'] = resource['epoch_time'] - start
         node['relative_time'] = node['epoch_time'] - start
+        resource['key'] = resource['task_id'].astype(str) + "-" + resource['try_id'].astype(str)
 
         task_plot = [0] * (end - start + 1)
         if label == 'CPU':
@@ -160,8 +161,8 @@
             total = node['total_memory'].sum() / 1024 / 1024 / 1024
 
         resource['total_cpu_time'] = resource['psutil_process_time_user'] + resource['psutil_process_time_system']
-        for task_id in resource['task_id'].unique():
-            tmp = resource[resource['task_id'] == task_id]
+        for key in resource['key'].unique():
+            tmp = resource[resource['key'] == key]
             tmp['last_timestamp'] = tmp['relative_time'].shift(1)
             if label == 'CPU':
                 tmp['last_cputime'] = tmp['total_cpu_time'].shift(1)

{"golden_diff": "diff --git a/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py b/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py\n--- a/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py\n+++ b/parsl/monitoring/visualization/plots/default/workflow_resource_plots.py\n@@ -152,6 +152,7 @@\n         end = resource['epoch_time'].max()\n         resource['relative_time'] = resource['epoch_time'] - start\n         node['relative_time'] = node['epoch_time'] - start\n+        resource['key'] = resource['task_id'].astype(str) + \"-\" + resource['try_id'].astype(str)\n \n         task_plot = [0] * (end - start + 1)\n         if label == 'CPU':\n@@ -160,8 +161,8 @@\n             total = node['total_memory'].sum() / 1024 / 1024 / 1024\n \n         resource['total_cpu_time'] = resource['psutil_process_time_user'] + resource['psutil_process_time_system']\n-        for task_id in resource['task_id'].unique():\n-            tmp = resource[resource['task_id'] == task_id]\n+        for key in resource['key'].unique():\n+            tmp = resource[resource['key'] == key]\n             tmp['last_timestamp'] = tmp['relative_time'].shift(1)\n             if label == 'CPU':\n                 tmp['last_cputime'] = tmp['total_cpu_time'].shift(1)\n", "issue": "Monitoring visualization show negative CPU usage with retries\n**Describe the bug**\r\nCurrently, the logged CPU time in monitoring db is cumulative. This works for most of the cases, except when a task is failed and retried. As a result, multiple figures in the visualization will show a sawtooth.\r\n\r\n**To Reproduce**\r\nSteps to reproduce the behavior, for e.g:\r\n1. Setup Parsl master with Python 3.6 on cluster\r\n2. Run all the tests and open the visualization\r\n3. See error\r\n\r\n**Expected behavior**\r\nEach retried task should be on a new row.\r\n\r\n**Possible solution**\r\nAdd a new column (e.g., `retries`) to the table as a new primary key.\r\nFix visualization based on that\r\n\n", "before_files": [{"content": "import math\nimport numpy as np\nimport pandas as pd\nimport plotly.graph_objs as go\nfrom plotly.offline import plot\n\n\ndef resource_distribution_plot(df_resources, df_task, type='psutil_process_time_user', label='CPU Time Distribution', option='avg', columns=20,):\n    # E.g., psutil_process_time_user or psutil_process_memory_percent\n\n    min_range = min(df_resources[type].astype('float'))\n    max_range = max(df_resources[type].astype('float'))\n    time_step = (max_range - min_range) / columns\n\n    if min_range == max_range:\n        x_axis = [min_range]\n    else:\n        x_axis = []\n        for i in np.arange(min_range, max_range + time_step, time_step):\n            x_axis.append(i)\n\n    apps_dict = dict()\n    for i in range(len(df_task)):\n        row = df_task.iloc[i]\n        apps_dict[row['task_id']] = []\n\n    def y_axis_setup():\n        items = [0] * len(x_axis)\n\n        for app, tasks in apps_dict.items():\n            if option == 'avg':\n                task = df_resources[df_resources['task_id'] ==\n                                    app][type].astype('float').mean()\n            elif option == 'max':\n                task = df_resources[df_resources['task_id'] == app][type].astype('float').max()\n\n            for i in range(len(x_axis) - 1):\n                a = task >= x_axis[i]\n                b = task < x_axis[i + 1]\n                if a and b:\n                    items[i] += 1\n                    break\n            if task >= x_axis[-1]:\n                items[-1] += 1\n        return items\n\n    if \"memory\" not in type:\n        xaxis = dict(autorange=True,\n                     title='CPU user time (seconds)')\n    else:\n        xaxis = dict(autorange=True,\n                     title='Memory usage (bytes)')\n    fig = go.Figure(\n        data=[go.Bar(x=x_axis,\n                     y=y_axis_setup(),\n                     name='tasks')],\n        layout=go.Layout(xaxis=xaxis,\n                         yaxis=dict(title='Tasks'),\n                         title=label + '(' + option + ')'))\n\n    return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n\n\ndef resource_time_series(tasks, type='psutil_process_time_user', label='CPU user time'):\n    tasks['epoch_time'] = (pd.to_datetime(\n        tasks['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n    step = int(tasks['resource_monitoring_interval'][0])\n    start = tasks['epoch_time'].min()\n    end = tasks['epoch_time'].max()\n    tasks['relative_time'] = tasks['epoch_time'] - start\n    if end != start:\n        bins = pd.cut(tasks['relative_time'],\n                      range(0, end - start + 1, step),\n                      include_lowest=True)\n        df = tasks.groupby(bins, as_index=False)[type].mean()\n        df['time'] = step * df.index\n        fig = go.Figure(\n            data=[go.Scatter(x=df['time'],\n                             y=df[type],\n                             )],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title=label),\n                             title=label))\n    else:\n        fig = go.Figure(\n            data=[go.Scatter(x=[0],\n                             y=[tasks[type].mean()],\n                             )],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title=label),\n                             title=label))\n    return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n\n\ndef worker_efficiency(task, node):\n    try:\n        node['epoch_time'] = (pd.to_datetime(\n            node['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        task['epoch_time_start'] = (pd.to_datetime(\n            task['task_try_time_launched']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        task['epoch_time_running'] = (pd.to_datetime(\n            task['task_try_time_running']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        task['epoch_time_returned'] = (pd.to_datetime(\n            task['task_try_time_returned']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        start = int(min(task['epoch_time_start'].min(), node['epoch_time'].min()))\n        end = int(task['epoch_time_returned'].max())\n\n        worker_plot = [0] * (end - start + 1)\n        total_workers = node['worker_count'].sum()\n\n        for i, row in task.iterrows():\n            if math.isnan(row['epoch_time_running']):\n                # skip tasks with no running start time\n                continue\n            if math.isnan(row['epoch_time_returned']):\n                # there is no end time for this, so we should assume the \"end\" time\n                etr = end\n            else:\n                etr = int(row['epoch_time_returned'])\n            for j in range(int(row['epoch_time_running']), etr + 1):\n                worker_plot[j - start] += 1\n        fig = go.Figure(\n            data=[go.Scatter(x=list(range(0, end - start + 1)),\n                             y=worker_plot,\n                             name='Total busy workers',\n                             ),\n                  go.Scatter(x=list(range(0, end - start + 1)),\n                             y=[total_workers] * (end - start + 1),\n                             name='Total online workers',\n                             )\n                 ],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title='Number of workers'),\n                             title=\"Worker efficiency\"))\n        return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n    except Exception as e:\n        print(e)\n        return \"The worker efficiency plot cannot be generated due to missing data.\"\n\n\ndef resource_efficiency(resource, node, label='CPU'):\n    try:\n        resource['epoch_time'] = (pd.to_datetime(\n            resource['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        node['epoch_time'] = (pd.to_datetime(\n            node['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        resource = resource.sort_values(by='epoch_time')\n        start = min(resource['epoch_time'].min(), node['epoch_time'].min())\n        end = resource['epoch_time'].max()\n        resource['relative_time'] = resource['epoch_time'] - start\n        node['relative_time'] = node['epoch_time'] - start\n\n        task_plot = [0] * (end - start + 1)\n        if label == 'CPU':\n            total = node['cpu_count'].sum()\n        elif label == 'mem':\n            total = node['total_memory'].sum() / 1024 / 1024 / 1024\n\n        resource['total_cpu_time'] = resource['psutil_process_time_user'] + resource['psutil_process_time_system']\n        for task_id in resource['task_id'].unique():\n            tmp = resource[resource['task_id'] == task_id]\n            tmp['last_timestamp'] = tmp['relative_time'].shift(1)\n            if label == 'CPU':\n                tmp['last_cputime'] = tmp['total_cpu_time'].shift(1)\n            for index, row in tmp.iterrows():\n                if np.isnan(row['last_timestamp']):\n                    continue\n                for i in range(int(row['last_timestamp']), int(row['relative_time'])):\n                    if label == 'CPU':\n                        diff = (row['total_cpu_time'] - row['last_cputime']) / (row['relative_time'] - row['last_timestamp'])\n                    elif label == 'mem':\n                        diff = row['psutil_process_memory_resident'] / 1024 / 1024 / 1024\n                    task_plot[i] += diff\n\n        if label == 'CPU':\n            name1 = 'Used CPU cores'\n            name2 = 'Total CPU cores'\n            yaxis = 'Number of CPU cores'\n            title = 'CPU usage'\n        elif label == 'mem':\n            name1 = 'Used memory'\n            name2 = 'Total memory'\n            yaxis = 'Memory (GB)'\n            title = 'Memory usage'\n\n        fig = go.Figure(\n            data=[go.Scatter(x=list(range(0, end - start + 1)),\n                             y=task_plot,\n                             name=name1,\n                             ),\n                  go.Scatter(x=list(range(0, end - start + 1)),\n                             y=[total] * (end - start + 1),\n                             name=name2,\n                             )\n                 ],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title=yaxis),\n                             title=title))\n        return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n    except Exception as e:\n        print(e)\n        return \"The resource efficiency plot cannot be generated because of exception {}.\".format(e)\n", "path": "parsl/monitoring/visualization/plots/default/workflow_resource_plots.py"}], "after_files": [{"content": "import math\nimport numpy as np\nimport pandas as pd\nimport plotly.graph_objs as go\nfrom plotly.offline import plot\n\n\ndef resource_distribution_plot(df_resources, df_task, type='psutil_process_time_user', label='CPU Time Distribution', option='avg', columns=20,):\n    # E.g., psutil_process_time_user or psutil_process_memory_percent\n\n    min_range = min(df_resources[type].astype('float'))\n    max_range = max(df_resources[type].astype('float'))\n    time_step = (max_range - min_range) / columns\n\n    if min_range == max_range:\n        x_axis = [min_range]\n    else:\n        x_axis = []\n        for i in np.arange(min_range, max_range + time_step, time_step):\n            x_axis.append(i)\n\n    apps_dict = dict()\n    for i in range(len(df_task)):\n        row = df_task.iloc[i]\n        apps_dict[row['task_id']] = []\n\n    def y_axis_setup():\n        items = [0] * len(x_axis)\n\n        for app, tasks in apps_dict.items():\n            if option == 'avg':\n                task = df_resources[df_resources['task_id'] ==\n                                    app][type].astype('float').mean()\n            elif option == 'max':\n                task = df_resources[df_resources['task_id'] == app][type].astype('float').max()\n\n            for i in range(len(x_axis) - 1):\n                a = task >= x_axis[i]\n                b = task < x_axis[i + 1]\n                if a and b:\n                    items[i] += 1\n                    break\n            if task >= x_axis[-1]:\n                items[-1] += 1\n        return items\n\n    if \"memory\" not in type:\n        xaxis = dict(autorange=True,\n                     title='CPU user time (seconds)')\n    else:\n        xaxis = dict(autorange=True,\n                     title='Memory usage (bytes)')\n    fig = go.Figure(\n        data=[go.Bar(x=x_axis,\n                     y=y_axis_setup(),\n                     name='tasks')],\n        layout=go.Layout(xaxis=xaxis,\n                         yaxis=dict(title='Tasks'),\n                         title=label + '(' + option + ')'))\n\n    return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n\n\ndef resource_time_series(tasks, type='psutil_process_time_user', label='CPU user time'):\n    tasks['epoch_time'] = (pd.to_datetime(\n        tasks['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n    step = int(tasks['resource_monitoring_interval'][0])\n    start = tasks['epoch_time'].min()\n    end = tasks['epoch_time'].max()\n    tasks['relative_time'] = tasks['epoch_time'] - start\n    if end != start:\n        bins = pd.cut(tasks['relative_time'],\n                      range(0, end - start + 1, step),\n                      include_lowest=True)\n        df = tasks.groupby(bins, as_index=False)[type].mean()\n        df['time'] = step * df.index\n        fig = go.Figure(\n            data=[go.Scatter(x=df['time'],\n                             y=df[type],\n                             )],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title=label),\n                             title=label))\n    else:\n        fig = go.Figure(\n            data=[go.Scatter(x=[0],\n                             y=[tasks[type].mean()],\n                             )],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title=label),\n                             title=label))\n    return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n\n\ndef worker_efficiency(task, node):\n    try:\n        node['epoch_time'] = (pd.to_datetime(\n            node['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        task['epoch_time_start'] = (pd.to_datetime(\n            task['task_try_time_launched']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        task['epoch_time_running'] = (pd.to_datetime(\n            task['task_try_time_running']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        task['epoch_time_returned'] = (pd.to_datetime(\n            task['task_try_time_returned']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        start = int(min(task['epoch_time_start'].min(), node['epoch_time'].min()))\n        end = int(task['epoch_time_returned'].max())\n\n        worker_plot = [0] * (end - start + 1)\n        total_workers = node['worker_count'].sum()\n\n        for i, row in task.iterrows():\n            if math.isnan(row['epoch_time_running']):\n                # skip tasks with no running start time\n                continue\n            if math.isnan(row['epoch_time_returned']):\n                # there is no end time for this, so we should assume the \"end\" time\n                etr = end\n            else:\n                etr = int(row['epoch_time_returned'])\n            for j in range(int(row['epoch_time_running']), etr + 1):\n                worker_plot[j - start] += 1\n        fig = go.Figure(\n            data=[go.Scatter(x=list(range(0, end - start + 1)),\n                             y=worker_plot,\n                             name='Total busy workers',\n                             ),\n                  go.Scatter(x=list(range(0, end - start + 1)),\n                             y=[total_workers] * (end - start + 1),\n                             name='Total online workers',\n                             )\n                 ],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title='Number of workers'),\n                             title=\"Worker efficiency\"))\n        return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n    except Exception as e:\n        print(e)\n        return \"The worker efficiency plot cannot be generated due to missing data.\"\n\n\ndef resource_efficiency(resource, node, label='CPU'):\n    try:\n        resource['epoch_time'] = (pd.to_datetime(\n            resource['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        node['epoch_time'] = (pd.to_datetime(\n            node['timestamp']) - pd.Timestamp(\"1970-01-01\")) // pd.Timedelta('1s')\n        resource = resource.sort_values(by='epoch_time')\n        start = min(resource['epoch_time'].min(), node['epoch_time'].min())\n        end = resource['epoch_time'].max()\n        resource['relative_time'] = resource['epoch_time'] - start\n        node['relative_time'] = node['epoch_time'] - start\n        resource['key'] = resource['task_id'].astype(str) + \"-\" + resource['try_id'].astype(str)\n\n        task_plot = [0] * (end - start + 1)\n        if label == 'CPU':\n            total = node['cpu_count'].sum()\n        elif label == 'mem':\n            total = node['total_memory'].sum() / 1024 / 1024 / 1024\n\n        resource['total_cpu_time'] = resource['psutil_process_time_user'] + resource['psutil_process_time_system']\n        for key in resource['key'].unique():\n            tmp = resource[resource['key'] == key]\n            tmp['last_timestamp'] = tmp['relative_time'].shift(1)\n            if label == 'CPU':\n                tmp['last_cputime'] = tmp['total_cpu_time'].shift(1)\n            for index, row in tmp.iterrows():\n                if np.isnan(row['last_timestamp']):\n                    continue\n                for i in range(int(row['last_timestamp']), int(row['relative_time'])):\n                    if label == 'CPU':\n                        diff = (row['total_cpu_time'] - row['last_cputime']) / (row['relative_time'] - row['last_timestamp'])\n                    elif label == 'mem':\n                        diff = row['psutil_process_memory_resident'] / 1024 / 1024 / 1024\n                    task_plot[i] += diff\n\n        if label == 'CPU':\n            name1 = 'Used CPU cores'\n            name2 = 'Total CPU cores'\n            yaxis = 'Number of CPU cores'\n            title = 'CPU usage'\n        elif label == 'mem':\n            name1 = 'Used memory'\n            name2 = 'Total memory'\n            yaxis = 'Memory (GB)'\n            title = 'Memory usage'\n\n        fig = go.Figure(\n            data=[go.Scatter(x=list(range(0, end - start + 1)),\n                             y=task_plot,\n                             name=name1,\n                             ),\n                  go.Scatter(x=list(range(0, end - start + 1)),\n                             y=[total] * (end - start + 1),\n                             name=name2,\n                             )\n                 ],\n            layout=go.Layout(xaxis=dict(autorange=True,\n                                        title='Time (seconds)'),\n                             yaxis=dict(title=yaxis),\n                             title=title))\n        return plot(fig, show_link=False, output_type=\"div\", include_plotlyjs=False)\n    except Exception as e:\n        print(e)\n        return \"The resource efficiency plot cannot be generated because of exception {}.\".format(e)\n", "path": "parsl/monitoring/visualization/plots/default/workflow_resource_plots.py"}]}