pass timestamps as status (#68)

mars/scheduler/graph.py

@@ -18,7 +18,6 @@
 import time
 import os
 from collections import deque, defaultdict
-from datetime import datetime
 
 from .assigner import AssignerActor
 from .resource import ResourceActor
@@ -171,7 +170,7 @@ def execute_graph(self):
         """
         Start graph execution
         """
-        self._start_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+        self._start_time = time.time()
         self.state = GraphState.PREPARING
 
         try:
@@ -630,7 +629,7 @@ def mark_terminal_finished(self, op_key, final_state=None):
                 self._terminated_tensors.add(tensor_key)
                 if len(self._terminated_tensors) == len(self._target_tensor_chunk_ops):
                     self.state = self.final_state if self.final_state is not None else GraphState.SUCCEEDED
-                    self._end_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+                    self._end_time = time.time()
 
     def get_state(self):
         return self.state

mars/scheduler/resource.py

@@ -13,11 +13,10 @@
 # limitations under the License.
 
 import copy
-import time
 import logging
 import os
+import time
 from collections import defaultdict
-from datetime import datetime, timedelta
 
 from .utils import SchedulerActor
 from ..config import options
@@ -49,15 +48,12 @@ def clean_worker(self):
         """
         Remove worker when it does not update its status for a long time
         """
-        timeout = options.scheduler.status_timeout
         for worker in list(self._meta_cache.keys()):
             worker_meta = self._meta_cache[worker]
             if 'update_time' not in worker_meta:
                 continue
 
-            last_time = datetime.strptime(worker_meta['update_time'], '%Y-%m-%d %H:%M:%S')
-            time_delta = timedelta(seconds=timeout)
-            if last_time + time_delta < datetime.now():
+            if time.time() - worker_meta['update_time'] > options.scheduler.status_timeout:
                 del self._meta_cache[worker]
 
         self.ref().clean_worker(_tell=True, _delay=1)

mars/web/server.py

@@ -97,8 +97,8 @@ def get_tasks_info(self):
                 task_desc['state'] = graph_ref.get_state().value
                 start_time, end_time, graph_size = graph_ref.get_graph_info()
                 task_desc['start_time'] = start_time
-                task_desc['end_time'] = end_time or 'N/A'
-                task_desc['graph_size'] = graph_size or 'N/A'
+                task_desc['end_time'] = end_time
+                task_desc['graph_size'] = graph_size
 
                 session_desc['tasks'][graph_key] = task_desc
         return sessions

mars/web/templates/task_list.html

@@ -25,8 +25,16 @@ <h3>{{ sessions[session_id]['name'] }}</h3>
                         {% else %}
                             <td><a href="/task?session_id={{ session_id }}&task_id={{ task_id }}">{{ task_id }}</a></td>
                         {% endif %}
-                        <td>{{ task_desc['start_time']  if 'start_time' in task_desc else 'N/A' }}</td>
-                        <td>{{ task_desc['end_time'] if 'end_time' in task_desc else 'N/A' }}</td>
+                        {% if task_desc['start_time'] is defined %}
+                            <td>{{ task_desc['start_time'] | format_ts }}</td>
+                        {% else %}
+                            <td>N/A</td>
+                        {% endif %}
+                        {% if task_desc['end_time'] is defined %}
+                            <td>{{ task_desc['end_time'] | format_ts }}</td>
+                        {% else %}
+                            <td>N/A</td>
+                        {% endif %}
                         <td>{{ task_desc['graph_size'] if 'graph_size' in task_desc else 'N/A' }}</td>
                         <td>{{ task_desc['state'] if 'state' in task_desc else 'N/A'}}</td>
                     </tr>

mars/web/templates/worker_detail.html

@@ -162,7 +162,7 @@
             </tr>
             <tr>
                 <td>Update Time</td>
-                <td>{{ worker_metrics['update_time']}}</td>
+                <td>{{ worker_metrics['update_time'] | format_ts }}</td>
             </tr>
             </tbody>
         </table>

mars/web/templates/worker_list.html

@@ -35,7 +35,7 @@ <h2>Worker List</h2>
                         <td>{{ '%0.2f' % metrics['cpu_used'] }} / {{ metrics['cpu_total'] }}</td>
                         <td>{{ metrics['memory_used'] | readable_size }} / {{ metrics['memory_total'] | readable_size }}</td>
                         <td>{{ metrics['cached_hold'] | readable_size }} / {{ metrics['cached_total'] | readable_size }}</td>
-                        <td>{{ worker_metrics[endpoint]['update_time'] }}</td>
+                        <td>{{ worker_metrics[endpoint]['update_time'] | format_ts }}</td>
                         <td></td>
                     </tr>
                 {% endfor -%}

mars/worker/status.py

@@ -17,7 +17,6 @@
 import json
 import time
 import logging
-from datetime import datetime
 from collections import defaultdict
 
 from .utils import WorkerActor, ExpMeanHolder
@@ -116,7 +115,7 @@ def collect_status(self):
 
             meta_dict = dict()
             meta_dict['hardware'] = metrics
-            meta_dict['update_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+            meta_dict['update_time'] = time.time()
             meta_dict['stats'] = dict()
             meta_dict['slots'] = dict()