Job management

.gitignore

@@ -4,6 +4,9 @@
 *.pyc
 *.swp
 
+# Ignore emacs backup files
+*~
+
 # Ignore these files
 .coverage
 

digits/dataset/tasks/analyze_db.py

@@ -107,16 +107,7 @@ def process_output(self, line):
         match = re.match(r'Progress: (\d+)\/(\d+)', message)
         if match:
             self.progress = float(match.group(1))/float(match.group(2))
-            socketio.emit('task update',
-                    {
-                        'task': self.html_id(),
-                        'update': 'progress',
-                        'percentage': int(round(100*self.progress)),
-                        'eta': utils.time_filters.print_time_diff(self.est_done()),
-                        },
-                    namespace='/jobs',
-                    room=self.job_id,
-                    )
+            self.emit_progress_update()
             return True
 
         # total count

digits/dataset/tasks/create_db.py

@@ -182,16 +182,7 @@ def process_output(self, line):
         match = re.match(r'Processed (\d+)\/(\d+)', message)
         if match:
             self.progress = float(match.group(1))/int(match.group(2))
-            socketio.emit('task update',
-                    {
-                        'task': self.html_id(),
-                        'update': 'progress',
-                        'percentage': int(round(100*self.progress)),
-                        'eta': utils.time_filters.print_time_diff(self.est_done()),
-                        },
-                    namespace='/jobs',
-                    room=self.job_id,
-                    )
+            self.emit_progress_update()
             return True
 
         # distribution

digits/dataset/tasks/parse_folder.py

@@ -148,16 +148,7 @@ def process_output(self, line):
         match = re.match(r'Progress: ([-+]?[0-9]*\.?[0-9]+(e[-+]?[0-9]+)?)', message)
         if match:
             self.progress = float(match.group(1))
-            socketio.emit('task update',
-                    {
-                        'task': self.html_id(),
-                        'update': 'progress',
-                        'percentage': int(round(100*self.progress)),
-                        'eta': utils.time_filters.print_time_diff(self.est_done()),
-                        },
-                    namespace='/jobs',
-                    room=self.job_id,
-                    )
+            self.emit_progress_update()
             return True
 
         # totals

digits/job.py

@@ -8,6 +8,7 @@
 
 import flask
 
+from digits import utils
 from digits.config import config_value
 from digits.utils import sizeof_fmt, filesystem as fs
 from status import Status, StatusCls
@@ -158,6 +159,7 @@ def on_status_update(self):
                 'status': self.status.name,
                 'css': self.status.css,
                 'running': self.status.is_running(),
+                'job_id': self.id(),
                 }
         with app.app_context():
             message['html'] = flask.render_template('status_updates.html', updates=self.status_history)
@@ -168,6 +170,13 @@ def on_status_update(self):
                 room=self.id(),
                 )
 
+        # send message to job_management room as well
+        socketio.emit('job update',
+                message,
+                namespace='/jobs',
+                room='job_management',
+                )
+
     def abort(self):
         """
         Abort a job and stop all running tasks
@@ -202,3 +211,34 @@ def disk_size_fmt(self):
         size = fs.get_tree_size(self._dir)
         return sizeof_fmt(size)
 
+    def get_progress(self):
+        """
+        Return job progress computed from task progress
+        """
+        if len(self.tasks) == 0:
+            return 0.0
+
+        progress = 0.0
+
+        for task in self.tasks:
+            progress += task.progress
+
+        progress /= len(self.tasks)
+        return progress
+
+    def emit_progress_update(self):
+        """
+        Call socketio.emit for task job update, by considering task progress.
+        """
+        progress = self.get_progress()
+
+        from digits.webapp import socketio
+        socketio.emit('job update',
+                      {
+                          'job_id': self.id(),
+                          'update': 'progress',
+                          'percentage': int(round(100*progress)),
+                      },
+                      namespace='/jobs',
+                      room='job_management'
+                  )

digits/model/tasks/train.py

@@ -207,17 +207,7 @@ def send_progress_update(self, epoch):
 
         self.current_epoch = epoch
         self.progress = epoch/self.train_epochs
-
-        socketio.emit('task update',
-                {
-                    'task': self.html_id(),
-                    'update': 'progress',
-                    'percentage': int(round(100*self.progress)),
-                    'eta': utils.time_filters.print_time_diff(self.est_done()),
-                    },
-                namespace='/jobs',
-                room=self.job_id,
-                )
+        self.emit_progress_update()
 
     def save_train_output(self, *args):
         """
@@ -247,6 +237,20 @@ def save_train_output(self, *args):
                     room=self.job_id,
                     )
 
+            if data['columns']:
+                # isolate the Loss column data for the sparkline
+                graph_data = data['columns'][0][1:]
+                socketio.emit('task update',
+                              {
+                                  'task': self.html_id(),
+                                  'job_id': self.job_id,
+                                  'update': 'combined_graph',
+                                  'data': graph_data,
+                              },
+                              namespace='/jobs',
+                              room='job_management',
+                          )
+
         # lr graph data
         data = self.lr_graph_data()
         if data:

digits/scheduler.py

@@ -175,6 +175,35 @@ def add_job(self, job):
             return False
         else:
             self.jobs.append(job)
+
+            # Need to fix this properly
+            # if True or flask._app_ctx_stack.top is not None:
+            from digits.webapp import app
+            with app.app_context():
+                # send message to job_management room that the job is added
+                import flask
+                html = flask.render_template('job_row.html', job = job)
+
+                # Convert the html into a list for the jQuery
+                # DataTable.row.add() method.  This regex removes the <tr>
+                # and <td> tags, and splits the string into one element
+                # for each cell.
+                import re
+                html = re.sub('<tr[^<]*>[\s\n\r]*<td[^<]*>[\s\n\r]*', '', html)
+                html = re.sub('[\s\n\r]*</td>[\s\n\r]*</tr>', '', html)
+                html = re.split('</td>[\s\n\r]*<td[^<]*>', html)
+
+                from digits.webapp import socketio
+                socketio.emit('job update',
+                              {
+                                  'update': 'added',
+                                  'job_id': job.id(),
+                                  'html': html
+                              },
+                              namespace='/jobs',
+                              room='job_management',
+                          )
+
             if 'DIGITS_MODE_TEST' not in os.environ:
                 # Let the scheduler do a little work before returning
                 time.sleep(utils.wait_time())
@@ -238,6 +267,15 @@ def delete_job(self, job):
                 if os.path.exists(job.dir()):
                     shutil.rmtree(job.dir())
                 logger.info('Job deleted.', job_id=job_id)
+                from digits.webapp import socketio
+                socketio.emit('job update',
+                              {
+                                  'update': 'deleted',
+                                  'job_id': job.id()
+                              },
+                              namespace='/jobs',
+                              room='job_management',
+                )
                 return True
 
         # see if the folder exists on disk
@@ -417,6 +455,7 @@ def reserve_resources(self, task, resources):
                     for resource in self.resources[resource_type]:
                         if resource.identifier == identifier:
                             resource.allocate(task, value)
+                            self.emit_gpus_available()
                             found = True
                             break
                     if not found:
@@ -439,6 +478,7 @@ def release_resources(self, task, resources):
                 for resource in self.resources[resource_type]:
                     if resource.identifier == identifier:
                         resource.deallocate(task)
+                        self.emit_gpus_available()
         task.current_resources = None
 
     def run_task(self, task, resources):
@@ -457,3 +497,17 @@ def run_task(self, task, resources):
         finally:
             self.release_resources(task, resources)
 
+    def emit_gpus_available(self):
+        """
+        Call socketio.emit gpu availablity
+        """
+        from digits.webapp import scheduler, socketio
+        socketio.emit('server update',
+                      {
+                          'update': 'gpus_available',
+                          'total_gpu_count': len(self.resources['gpus']),
+                          'remaining_gpu_count': sum(r.remaining() for r in scheduler.resources['gpus']),
+                      },
+                      namespace='/jobs',
+                      room='job_management'
+                  )

digits/static/css/style.css

@@ -1,3 +1,5 @@
+/* Copyright (c) 2014-2015, NVIDIA CORPORATION.  All rights reserved. */
+
 body {
     padding-top: 50px;
     padding-bottom: 50px;
@@ -27,6 +29,13 @@ path.c3-line {
 .autocomplete-group { padding: 2px 5px; }
 .autocomplete-group strong { display: block; border-bottom: 1px solid #000; }
 
+/* To fix the tooltip in the sparkline */
+.jqstooltip {
+  -webkit-box-sizing: content-box;
+  -moz-box-sizing: content-box;
+  box-sizing: content-box;
+}
+
 .btn-file {
     position: relative;
     overflow: hidden;
@@ -66,5 +75,3 @@ a.active {
 ul.inline li {
 	display: inline-block;
 }
-
-

digits/static/js/jquery.time_filters.js

@@ -0,0 +1,32 @@
+// Copyright (c) 2014-2015, NVIDIA CORPORATION.  All rights reserved.
+
+function print_time_diff(diff) {
+    if (diff < 0) {
+        return 'Negative Time';
+    }
+
+    var total_seconds = Math.round(diff);
+    var days = Math.round(total_seconds/(24*3600));
+    var hours = Math.round((total_seconds % (24*3600))/3600);
+    var minutes = Math.round((total_seconds % 3600)/60);
+    var seconds = Math.round(total_seconds % 60);
+
+    function plural(number, name) {
+        return number + ' ' + name + (number == 1 ? '' : 's');
+    }
+
+    function pair(number1, name1, number2, name2) {
+        if (number2 > 0)
+            return plural(number1, name1) + ', ' + plural(number2, name2);
+        else
+            return plural(number1, name1);
+    }
+
+    if (days >= 1)
+        return pair(days, 'day', hours, 'hour');
+    else if (hours >= 1)
+        return pair(hours, 'hour', minutes, 'minute');
+    else if (minutes >= 1)
+        return pair(minutes, 'minute', seconds, 'second');
+    return plural(seconds, 'second');
+}

digits/static/js/jquery.timer.min.js

@@ -0,0 +1,29 @@
+/**
+ * jquery.timer.js
+ *
+ * Copyright (c) 2011 Jason Chavannes <jason.chavannes@gmail.com>
+ *
+ * http://jchavannes.com/jquery-timer
+ *
+ * Permission is hereby granted, free of charge, to any person
+ * obtaining a copy of this software and associated documentation
+ * files (the "Software"), to deal in the Software without
+ * restriction, including without limitation the rights to use, copy,
+ * modify, merge, publish, distribute, sublicense, and/or sell copies
+ * of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be
+ * included in all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+ * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+ * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+ * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
+ * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
+ * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+ * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+!function($){$.timer=function(func,time,autostart){return this.set=function(func,time,autostart){if(this.init=!0,"object"==typeof func){var paramList=["autostart","time"];for(var arg in paramList)void 0!=func[paramList[arg]]&&eval(paramList[arg]+" = func[paramList[arg]]");func=func.action}return"function"==typeof func&&(this.action=func),isNaN(time)||(this.intervalTime=time),autostart&&!this.isActive&&(this.isActive=!0,this.setTimer()),this},this.once=function(t){var i=this;return isNaN(t)&&(t=0),window.setTimeout(function(){i.action()},t),this},this.play=function(t){return this.isActive||(t?this.setTimer():this.setTimer(this.remaining),this.isActive=!0),this},this.pause=function(){return this.isActive&&(this.isActive=!1,this.remaining-=new Date-this.last,this.clearTimer()),this},this.stop=function(){return this.isActive=!1,this.remaining=this.intervalTime,this.clearTimer(),this},this.toggle=function(t){return this.isActive?this.pause():t?this.play(!0):this.play(),this},this.reset=function(){return this.isActive=!1,this.play(!0),this},this.clearTimer=function(){window.clearTimeout(this.timeoutObject)},this.setTimer=function(t){var i=this;"function"==typeof this.action&&(isNaN(t)&&(t=this.intervalTime),this.remaining=t,this.last=new Date,this.clearTimer(),this.timeoutObject=window.setTimeout(function(){i.go()},t))},this.go=function(){if(this.isActive)try{this.action()}finally{this.setTimer()}},this.init?new $.timer(func,time,autostart):(this.set(func,time,autostart),this)}}(jQuery);

digits/status.py

@@ -117,6 +117,8 @@ def status(self, value):
         # If the status is Done, then force the progress to 100%
         if value == Status.DONE:
             self.progress = 1.0
+            if hasattr(self, 'emit_progress_update'):
+                self.emit_progress_update()
 
         # Don't invoke callback for INIT
         if value != Status.INIT:

digits/task.py

@@ -266,6 +266,7 @@ def abort(self):
         if self.status.is_running():
             self.aborted.set()
 
+
     def preprocess_output_digits(self, line):
         """
         Takes line of output and parses it according to DIGITS's log format
@@ -325,3 +326,23 @@ def after_runtime_error(self):
         """
         pass
 
+    def emit_progress_update(self):
+        """
+        Call socketio.emit for task progess update, and trigger job progress update.
+        """
+        from digits.webapp import socketio
+        socketio.emit('task update',
+                {
+                    'task': self.html_id(),
+                    'update': 'progress',
+                    'percentage': int(round(100*self.progress)),
+                    'eta': utils.time_filters.print_time_diff(self.est_done()),
+                    },
+                namespace='/jobs',
+                room=self.job_id,
+                )
+
+        from digits.webapp import scheduler
+        job = scheduler.get_job(self.job_id)
+        if job:
+            job.emit_progress_update()