task_runner: always use run_isolated

appengine/swarming/swarming_bot/bot_code/task_runner.py

 # Copyright 2013 The LUCI Authors. All rights reserved.
 # Use of this source code is governed under the Apache License, Version 2.0
 # that can be found in the LICENSE file.
 
 """Runs a Swarming task.
 
 Downloads all the necessary files to run the task, executes the command and
 streams results back to the Swarming server.
 
 The process exit code is 0 when the task was executed, even if the task itself
@@ skipping 96 matching lines @@
         '--cache', os.path.join(bot_dir, 'cache'),
         '--root-dir', os.path.join(work_dir, 'isolated'),
       ])
   if min_free_space:
     cmd.extend(('--min-free-space', str(min_free_space)))
 
   if task_details.hard_timeout:
     cmd.extend(('--hard-timeout', str(task_details.hard_timeout)))
   if task_details.grace_period:
     cmd.extend(('--grace-period', str(task_details.grace_period)))
-  if task_details.extra_args:
-    cmd.append('--')
+
+  cmd.append('--')
+  if task_details.command:
+    cmd.extend(task_details.command)
+  elif task_details.extra_args:
     cmd.extend(task_details.extra_args)
   return cmd
 
 
 class TaskDetails(object):
   def __init__(self, data):
     """Loads the raw data from a manifest file specified by --in-file."""
     logging.info('TaskDetails(%s)', data)
     if not isinstance(data, dict):
       raise ValueError('Expected dict, got %r' % data)
@@ skipping 163 matching lines @@
   # TODO(maruel): This function is incomprehensible, split and refactor.
   # Signal the command is about to be started.
   last_packet = start = now = monotonic_time()
   params = {
     'cost_usd': cost_usd_hour * (now - task_start) / 60. / 60.,
     'id': task_details.bot_id,
     'task_id': task_details.task_id,
   }
   post_update(swarming_server, params, None, '', 0)
 
-  if task_details.command:
-    # Raw command.
-    cmd = task_details.command
-    isolated_result = None
-  else:
-    # Isolated task.
-    isolated_result = os.path.join(work_dir, 'isolated_result.json')
-    cmd = get_isolated_cmd(
-        work_dir, task_details, isolated_result, min_free_space)
-    # Hard timeout enforcement is deferred to run_isolated. Grace is doubled to
-    # give one 'grace_period' slot to the child process and one slot to upload
-    # the results back.
-    task_details.hard_timeout = 0
-    if task_details.grace_period:
-      task_details.grace_period *= 2
+  isolated_result = os.path.join(work_dir, 'isolated_result.json')
+  cmd = get_isolated_cmd(
+      work_dir, task_details, isolated_result, min_free_space)
+  # Hard timeout enforcement is deferred to run_isolated. Grace is doubled to
+  # give one 'grace_period' slot to the child process and one slot to upload
+  # the results back.
+  task_details.hard_timeout = 0
+  if task_details.grace_period:
+    task_details.grace_period *= 2
 
   try:
     # TODO(maruel): Support both channels independently and display stderr in
     # red.
     env = None
     if task_details.env:
       env = os.environ.copy()
       for key, value in task_details.env.iteritems():
         if not value:
           env.pop(key, None)
@@ skipping 23 matching lines @@
         u'exit_code': 1,
         u'hard_timeout': False,
         u'io_timeout': False,
         u'must_signal_internal_failure': None,
         u'version': OUT_VERSION,
       }
 
     output_chunk_start = 0
     stdout = ''
     exit_code = None
-    had_hard_timeout = False
     had_io_timeout = False
     must_signal_internal_failure = None
     kill_sent = False
     timed_out = None
     try:
       calc = lambda: calc_yield_wait(
           task_details, start, last_io, timed_out, stdout)
       maxsize = lambda: MAX_CHUNK_SIZE - len(stdout)
       last_io = monotonic_time()
       for _, new_data in proc.yield_any(maxsize=maxsize, timeout=calc):
@@ skipping 14 matching lines @@
         # Send signal on timeout if necessary. Both are failures, not
         # internal_failures.
         # Eventually kill but return 0 so bot_main.py doesn't cancel the task.
         if not timed_out:
           if (task_details.io_timeout and
               now - last_io > task_details.io_timeout):
             had_io_timeout = True
             logging.warning('I/O timeout; sending SIGTERM')
             proc.terminate()
             timed_out = monotonic_time()
-          elif (task_details.hard_timeout and
-              now - start > task_details.hard_timeout):
-            had_hard_timeout = True
-            logging.warning('Hard timeout; sending SIGTERM')
-            proc.terminate()
-            timed_out = monotonic_time()
         else:
           # During grace period.
           if not kill_sent and now >= timed_out + task_details.grace_period:
             # Now kill for real. The user can distinguish between the following
             # states:
             # - signal but process exited within grace period,
             #   (hard_|io_)_timed_out will be set but the process exit code will
             #   be script provided.
             # - processed exited late, exit code will be -9 on posix.
             logging.warning('Grace exhausted; sending SIGKILL')
             proc.kill()
             kill_sent = True
       logging.info('Waiting for proces exit')
       exit_code = proc.wait()
     except MustExit as e:
       # TODO(maruel): Do the send SIGTERM to child process and give it
       # task_details.grace_period to terminate.
       must_signal_internal_failure = (
           u'task_runner received signal %s' % e.signal)
       exit_code = kill_and_wait(
           proc, task_details.grace_period, 'signal %d' % e.signal)
     except (IOError, OSError):
       # Something wrong happened, try to kill the child process.
-      had_hard_timeout = True
       exit_code = kill_and_wait(
           proc, task_details.grace_period, 'exception %s' % e)
 
     # This is the very last packet for this command. It if was an isolated task,
     # include the output reference to the archived .isolated file.
     now = monotonic_time()
     params['cost_usd'] = cost_usd_hour * (now - task_start) / 60. / 60.
     params['duration'] = now - start
     params['io_timeout'] = had_io_timeout
-    params['hard_timeout'] = had_hard_timeout
-    if isolated_result:
-      try:
-        if ((had_io_timeout or had_hard_timeout) and
-            not os.path.isfile(isolated_result)):
-          # It's possible that run_isolated failed to quit quickly enough; it
-          # could be because there was too much data to upload back or something
-          # else. Do not create an internal error, just send back the (partial)
-          # view as task_runner saw it, for example the real exit_code is
-          # unknown.
-          logging.warning('TIMED_OUT and there\'s no result file')
+    had_hard_timeout = False
+    try:
+      if not os.path.isfile(isolated_result):
+        # It's possible if
+        # - run_isolated.py did not start
+        # - run_isolated.py started, but arguments were invalid
+        # - host in a situation unable to fork
+        # - grand child process outliving the child process deleting everything
+        #   it can
+        # Do not create an internal error, just send back the (partial)
+        # view as task_runner saw it, for example the real exit_code is
+        # unknown.
+        logging.warning('there\'s no result file')
+        if exit_code is None:
           exit_code = -1
-        else:
-          # See run_isolated.py for the format.
-          with open(isolated_result, 'rb') as f:
-            run_isolated_result = json.load(f)
-          logging.debug('run_isolated:\n%s', run_isolated_result)
-          # TODO(maruel): Grab statistics (cache hit rate, data downloaded,
-          # mapping time, etc) from run_isolated and push them to the server.
-          if run_isolated_result['outputs_ref']:
-            params['outputs_ref'] = run_isolated_result['outputs_ref']
-          had_hard_timeout = (
-              had_hard_timeout or run_isolated_result['had_hard_timeout'])
-          params['hard_timeout'] = had_hard_timeout
-          if not had_io_timeout and not had_hard_timeout:
-            if run_isolated_result['internal_failure']:
-              must_signal_internal_failure = (
-                  run_isolated_result['internal_failure'])
-              logging.error('%s', must_signal_internal_failure)
-            elif exit_code:
-              # TODO(maruel): Grab stdout from run_isolated.
-              must_signal_internal_failure = (
-                  'run_isolated internal failure %d' % exit_code)
-              logging.error('%s', must_signal_internal_failure)
-          exit_code = run_isolated_result['exit_code']
-          if run_isolated_result.get('duration') is not None:
-            # Calculate the real task duration as measured by run_isolated and
-            # calculate the remaining overhead.
-            params['bot_overhead'] = params['duration']
-            params['duration'] = run_isolated_result['duration']
-            params['bot_overhead'] -= params['duration']
-            params['bot_overhead'] -= run_isolated_result.get(
-                'download', {}).get('duration', 0)
-            params['bot_overhead'] -= run_isolated_result.get(
-                'upload', {}).get('duration', 0)
-            if params['bot_overhead'] < 0:
-              params['bot_overhead'] = 0
-          stats = run_isolated_result.get('stats')
-          if stats:
-            params['isolated_stats'] = stats
-      except (IOError, OSError, ValueError) as e:
-        logging.error('Swallowing error: %s', e)
-        if not must_signal_internal_failure:
-          must_signal_internal_failure = str(e)
+      else:
+        # See run_isolated.py for the format.
+        with open(isolated_result, 'rb') as f:
+          run_isolated_result = json.load(f)
+        logging.debug('run_isolated:\n%s', run_isolated_result)
+        # TODO(maruel): Grab statistics (cache hit rate, data downloaded,
+        # mapping time, etc) from run_isolated and push them to the server.
+        if run_isolated_result['outputs_ref']:
+          params['outputs_ref'] = run_isolated_result['outputs_ref']
+        had_hard_timeout = run_isolated_result['had_hard_timeout']
+        if not had_io_timeout and not had_hard_timeout:
+          if run_isolated_result['internal_failure']:
+            must_signal_internal_failure = (
+                run_isolated_result['internal_failure'])
+            logging.error('%s', must_signal_internal_failure)
+          elif exit_code:
+            # TODO(maruel): Grab stdout from run_isolated.
+            must_signal_internal_failure = (
+                'run_isolated internal failure %d' % exit_code)
+            logging.error('%s', must_signal_internal_failure)
+        exit_code = run_isolated_result['exit_code']
+        if run_isolated_result.get('duration') is not None:
+          # Calculate the real task duration as measured by run_isolated and
+          # calculate the remaining overhead.
+          params['bot_overhead'] = params['duration']
+          params['duration'] = run_isolated_result['duration']
+          params['bot_overhead'] -= params['duration']
+          params['bot_overhead'] -= run_isolated_result.get(
+              'download', {}).get('duration', 0)
+          params['bot_overhead'] -= run_isolated_result.get(
+              'upload', {}).get('duration', 0)
+          if params['bot_overhead'] < 0:
+            params['bot_overhead'] = 0
+        stats = run_isolated_result.get('stats')
+        if stats:
+          params['isolated_stats'] = stats
+    except (IOError, OSError, ValueError) as e:
+      logging.error('Swallowing error: %s', e)
+      if not must_signal_internal_failure:
+        must_signal_internal_failure = str(e)
     # TODO(maruel): Send the internal failure here instead of sending it through
     # bot_main, this causes a race condition.
     if exit_code is None:
       exit_code = -1
+    params['hard_timeout'] = had_hard_timeout
     post_update(swarming_server, params, exit_code, stdout, output_chunk_start)
     return {
       u'exit_code': exit_code,
       u'hard_timeout': had_hard_timeout,
       u'io_timeout': had_io_timeout,
       u'must_signal_internal_failure': must_signal_internal_failure,
       u'version': OUT_VERSION,
     }
   finally:
-    if isolated_result:
-      try:
-        os.remove(isolated_result)
-      except OSError:
-        pass
+    try:
+      os.remove(isolated_result)
+    except OSError:
+      pass
 
 
 def main(args):
   parser = optparse.OptionParser(description=sys.modules[__name__].__doc__)
   parser.add_option('--in-file', help='Name of the request file')
   parser.add_option(
       '--out-file', help='Name of the JSON file to write a task summary to')
   parser.add_option(
       '--swarming-server', help='Swarming server to send data back')
   parser.add_option(
@@ skipping 14 matching lines @@
   if options.start > now:
     options.start = now
 
   try:
     load_and_run(
         options.in_file, options.swarming_server, options.cost_usd_hour,
         options.start, options.out_file, options.min_free_space)
     return 0
   finally:
     logging.info('quitting')