message: Improve error reporting of multi worker strategy combinations
  It used to only show "some worker failed, see logs for details", now it can report the traceback. Though since we're unable to differentiate return values from different workers.
diff: @@ -22,6 +22,7 @@ from __future__ import absolute_import
  from __future__ import division
  from __future__ import print_function
  +import collections
  import re
  import sys
  import types
  @@ -332,6 +333,9 @@ def main():
  _running_in_worker = False
  +_TestResult = collections.namedtuple("_TestResult", ["status", "message"])
  +
  +
  def _test_runner(test_id):
  """Executes the test with the given test_id.
  @@ -353,15 +357,23 @@ def _test_runner(test_id):
  test = unittest.defaultTestLoader.loadTestsFromName(test_id)
  runner = unittest.TextTestRunner()
  result = runner.run(test)
  - # Print failures and errors to stdout and multi_process_runner will collect
  + # Treat expected failures as failures, so that the main process can get
  + # them and fail as expected. Also treat errors as failures to simplify the
  + # handling.
  + failures = result.failures + result.expectedFailures + result.errors
  + if failures:
  + ret = _TestResult(status="failure", message=failures[0][1])
  + elif result.skipped:
  + ret = _TestResult(status="skipped", message=result.skipped[0][1])
  + else:
  + # Treat unexpectedSuccesses as OK so that the test case in the main process
  + # succeed as well.
  + ret = _TestResult(status="ok", message=None)
  + # Print tracebacks to stdout and multi_process_runner will collect
  # them and stream back to the main process.
  - for _, msg in result.failures + result.errors:
  - print(msg)
  - # Return expected failures as failures, so that the main process can get
  - # them and fail as expected.
  - if result.expectedFailures:
  - return False
  - return result.wasSuccessful()
  + if ret.message:
  + print(ret.message)
  + return ret
  def _multi_worker_test(test_method):
  @@ -407,18 +419,29 @@ def _multi_worker_test(test_method):
  # [sub process]test_method()
  test_id = self.id()
  if runner:
  - result = runner.run(_test_runner, args=(test_id,))
  + results = runner.run(_test_runner, args=(test_id,))
  else:
  cluster_spec = multi_worker_test_base.create_cluster_spec(
  has_chief=has_chief,
  num_workers=num_workers,
  num_ps=0,
  has_eval=False)
  - result = multi_process_runner.run(
  + results = multi_process_runner.run(
  _test_runner, cluster_spec, args=(test_id,)).return_value
  - for was_successful in result:
  - if not was_successful:
  - raise AssertionError("some worker failed, see logs for details")
  +
  + skip_reason = None
  + for result in results:
  + if result.status == "failure":
  + # We can't tell which worker the return value come from, so we fail on
  + # the first error.
  + self.fail(result.message)
  + break
  + elif result.status == "skipped":
  + # Record the skip reason, but do not actually skip the test in case some
  + # processes fail instead.
  + skip_reason = result.message
  + if skip_reason is not None:
  + self.skipTest(skip_reason)
  argspec = tf_inspect.getfullargspec(test_method)
  decorator_args = (argspec.args or []) + ["has_chief", "num_workers", "runner"]
  