def testSerialize(self):
    # pylint: disable=g-import-not-at-top
    try:
      import portpicker
    except ImportError:
      return
    with context.graph_mode():
      worker_port = portpicker.pick_unused_port()
      ps_port = portpicker.pick_unused_port()
      cluster_dict = {
          "worker": ["localhost:%s" % worker_port],
          "ps": ["localhost:%s" % ps_port]
      }
      cs = server_lib.ClusterSpec(cluster_dict)

      worker = server_lib.Server(
          cs, job_name="worker", protocol="grpc", task_index=0, start=True)
      unused_ps = server_lib.Server(
          cs, job_name="ps", protocol="grpc", task_index=0, start=True)
      with ops.Graph().as_default(), session.Session(target=worker.target):
        with ops.device("/job:worker"):
          t = constant_op.constant([[1.0], [2.0]])
          l = list_ops.tensor_list_from_tensor(t, element_shape=[1])
        with ops.device("/job:ps"):
          l_ps = array_ops.identity(l)
          l_ps, e = list_ops.tensor_list_pop_back(
              l_ps, element_dtype=dtypes.float32)
        with ops.device("/job:worker"):
          worker_e = array_ops.identity(e)
        self.assertAllEqual(worker_e.eval(), [2.0])

def create_local_cluster(num_workers, num_ps, protocol="grpc"):
  """Create and start local servers and return the associated `Server` objects.

  Example:
  ```python
  workers, _ = tf.test.create_local_cluster(num_workers=2, num_ps=2)

  worker_sessions = [tf.Session(w.target) for w in workers]

  with tf.device("/job:ps/task:0"):
    ...
  with tf.device("/job:ps/task:1"):
    ...
  with tf.device("/job:worker/task:0"):
    ...
  with tf.device("/job:worker/task:1"):
    ...

  worker_sessions[0].run(...)
  ```

  Args:
    num_workers: Number of worker servers to start.
    num_ps: Number of PS servers to start.
    protocol: Communication protocol.  Allowed values are documented in
      the documentation of `tf.train.Server`.

  Returns:
    A tuple `(worker_servers, ps_servers)`.  `worker_servers` is a list
    of `num_workers` objects of type `tf.train.Server` (all running locally);
    and `ps_servers` is a list of `num_ps` objects of similar type.

  Raises:
    ImportError: if portpicker module was not found at load time
  """
  if _portpicker_import_error:
    raise _portpicker_import_error  # pylint: disable=raising-bad-type
  worker_ports = [portpicker.pick_unused_port() for _ in range(num_workers)]
  ps_ports = [portpicker.pick_unused_port() for _ in range(num_ps)]
  cluster_dict = {
      "worker": ["localhost:%s" % port for port in worker_ports],
      "ps": ["localhost:%s" % port for port in ps_ports]
  }
  cs = server_lib.ClusterSpec(cluster_dict)

  workers = [
      server_lib.Server(
          cs, job_name="worker", protocol=protocol, task_index=ix, start=True)
      for ix in range(num_workers)
  ]
  ps_servers = [
      server_lib.Server(
          cs, job_name="ps", protocol=protocol, task_index=ix, start=True)
      for ix in range(num_ps)
  ]

  return workers, ps_servers

def _create_local_cluster(num_workers,
                          num_ps,
                          has_eval=False,
                          protocol="grpc",
                          worker_config=None,
                          ps_config=None):
  if _portpicker_import_error:
    raise _portpicker_import_error  # pylint: disable=raising-bad-type
  worker_ports = [portpicker.pick_unused_port() for _ in range(num_workers)]
  ps_ports = [portpicker.pick_unused_port() for _ in range(num_ps)]

  cluster_dict = {
      "worker": ["localhost:%s" % port for port in worker_ports],
      "ps": ["localhost:%s" % port for port in ps_ports]
  }
  if has_eval:
    cluster_dict["evaluator"] = ["localhost:%s" % portpicker.pick_unused_port()]

  cs = server_lib.ClusterSpec(cluster_dict)

  workers = [
      server_lib.Server(
          cs,
          job_name="worker",
          protocol=protocol,
          task_index=ix,
          config=worker_config,
          start=True) for ix in range(num_workers)
  ]
  ps_servers = [
      server_lib.Server(
          cs,
          job_name="ps",
          protocol=protocol,
          task_index=ix,
          config=ps_config,
          start=True) for ix in range(num_ps)
  ]
  if has_eval:
    evals = [
        server_lib.Server(
            cs,
            job_name="evaluator",
            protocol=protocol,
            task_index=0,
            config=worker_config,
            start=True)
    ]
  else:
    evals = []

  return workers, ps_servers, evals

def main(argv):
  del argv  # Unused.

  if not flags.FLAGS.dest_server_config_path:
    raise ValueError("dest_server_config_path flag has to be provided.")

  if not flags.FLAGS.dest_client_config_path:
    raise ValueError("dest_client_config_path flag has to be provided.")

  admin_ui_port = portpicker.pick_unused_port()
  frontend_port = portpicker.pick_unused_port()

  source_server_config_path = package.ResourcePath(
      "grr-response-core", "install_data/etc/grr-server.yaml")
  config_lib.LoadConfig(config.CONFIG, source_server_config_path)
  config.CONFIG.SetWriteBack(flags.FLAGS.dest_server_config_path)

  # TODO(user): remove when AFF4 is gone.
  config.CONFIG.Set("Database.aff4_enabled", False)
  config.CONFIG.Set("Database.enabled", True)

  config.CONFIG.Set("Blobstore.implementation", "DbBlobStore")
  config.CONFIG.Set("Database.implementation", "MysqlDB")
  config.CONFIG.Set("Mysql.database", flags.FLAGS.config_mysql_database)
  if flags.FLAGS.config_mysql_username is not None:
    config.CONFIG.Set("Mysql.username", flags.FLAGS.config_mysql_username)
  if flags.FLAGS.config_mysql_password is not None:
    config.CONFIG.Set("Mysql.password", flags.FLAGS.config_mysql_password)
  config.CONFIG.Set("AdminUI.port", admin_ui_port)
  config.CONFIG.Set("AdminUI.headless", True)
  config.CONFIG.Set("Frontend.bind_address", "127.0.0.1")
  config.CONFIG.Set("Frontend.bind_port", frontend_port)
  config.CONFIG.Set("Server.initialized", True)
  config.CONFIG.Set("Client.poll_max", 1)
  config.CONFIG.Set("Client.server_urls",
                    ["http://localhost:%d/" % frontend_port])
  if flags.FLAGS.config_logging_path is not None:
    config.CONFIG.Set("Logging.path", flags.FLAGS.config_logging_path)

  config_updater_keys_util.GenerateKeys(config.CONFIG)
  config.CONFIG.Write()

  config_lib.SetPlatformArchContext()
  context = list(config.CONFIG.context)
  context.append("Client Context")
  deployer = build.ClientRepacker()
  config_data = deployer.GetClientConfig(
      context, validate=False, deploy_timestamp=False)
  with io.open(flags.FLAGS.dest_client_config_path, "w") as fd:
    fd.write(config_data)

  def setUpClass(cls):
    gpu_memory_fraction_opt = (
        "--gpu_memory_fraction=%f" % cls.PER_PROC_GPU_MEMORY_FRACTION)

    worker_port = portpicker.pick_unused_port()
    cluster_spec = "worker|localhost:%d" % worker_port
    tf_logging.info("cluster_spec: %s", cluster_spec)

    server_bin = test.test_src_dir_path("python/debug/grpc_tensorflow_server")

    cls.server_target = "grpc://localhost:%d" % worker_port

    cls.server_procs = {}
    cls.server_procs["worker"] = subprocess.Popen(
        [
            server_bin,
            "--cluster_spec=%s" % cluster_spec,
            "--job_name=worker",
            "--task_id=0",
            gpu_memory_fraction_opt,
        ],
        stdout=sys.stdout,
        stderr=sys.stderr)

    # Start debug server in-process, on separate thread.
    (cls.debug_server_port, cls.debug_server_url, _, cls.debug_server_thread,
     cls.debug_server
    ) = grpc_debug_test_server.start_server_on_separate_thread(
        dump_to_filesystem=False)
    tf_logging.info("debug server url: %s", cls.debug_server_url)

    cls.session_config = config_pb2.ConfigProto(
        gpu_options=config_pb2.GPUOptions(
            per_process_gpu_memory_fraction=cls.PER_PROC_GPU_MEMORY_FRACTION))

  def __init__(self, run_config, full_screen=False, **kwargs):
    self._proc = None
    self._sock = None
    self._controller = None
    self._tmp_dir = tempfile.mkdtemp(prefix="sc-", dir=run_config.tmp_dir)
    self._port = portpicker.pick_unused_port()
    self._check_exists(run_config.exec_path)

    args = [
        run_config.exec_path,
        "-listen", "127.0.0.1",
        "-port", str(self._port),
        "-dataDir", os.path.join(run_config.data_dir, ""),
        "-tempDir", os.path.join(self._tmp_dir, ""),
        "-displayMode", "1" if full_screen else "0",
    ]
    try:
      self._proc = self._launch(run_config, args, **kwargs)
      self._sock = self._connect(self._port)
      client = protocol.StarcraftProtocol(self._sock)
      self._controller = remote_controller.RemoteController(client)
      with sw("startup"):
        self._controller.ping()
    except:
      self.close()
      raise

def run_benchmark_distributed():
  ops = create_graph("/job:worker/task:0", "/job:worker/task:1")
  queues = [create_done_queue(0), create_done_queue(1)]

  # launch distributed service


  port0, port1 = [portpicker.pick_unused_port() for _ in range(2)]
  flags = " ".join(sys.argv)  # pass parent flags to children
  
  def run_worker(w):
    my_env = os.environ.copy()
    if not FLAGS.verbose:
      my_env["CUDA_VISIBLE_DEVICES"] = ""
      my_env["TF_CPP_MIN_LOG_LEVEL"] = "2"
    if FLAGS.profile:
      my_env["LD_PRELOAD"]="/usr/lib/libtcmalloc_and_profiler.so.4"
      my_env["CPUPROFILE"]="/tmp/profile.out.%s"%(w)
    cmd = "python %s --task=%d --port0=%s --port1=%s"%(flags, w, port0, port1)
    subprocess.Popen(cmd, shell=True, stderr=subprocess.STDOUT,
                     env=my_env)
    
  run_worker(0)
  run_worker(1)

  sess = tf.Session("grpc://%s:%s"%(host, port0), config=session_config())
  rate = run_benchmark(sess, *ops)

  # bring down workers
  if FLAGS.verbose:
    print("Killing workers.")
  sess.run(queues[1].enqueue(1))
  sess.run(queues[0].enqueue(1))  # bring down master last
  
  return rate

def _pick_unused_port():
  """For some reason portpicker returns the same port sometimes."""
  while True:
    p = portpicker.pick_unused_port()
    if p not in _PORTS:
      break
  _PORTS.add(p)
  return p

def start_server_on_separate_thread(dump_to_filesystem=True,
                                    server_start_delay_sec=0.0,
                                    poll_server=False,
                                    blocking=True,
                                    toggle_watch_on_core_metadata=None):
  """Create a test gRPC debug server and run on a separate thread.

  Args:
    dump_to_filesystem: (bool) whether the debug server will dump debug data
      to the filesystem.
    server_start_delay_sec: (float) amount of time (in sec) to delay the server
      start up for.
    poll_server: (bool) whether the server will be polled till success on
      startup.
    blocking: (bool) whether the server should be started in a blocking mode.
    toggle_watch_on_core_metadata: A list of
        (node_name, output_slot, debug_op) tuples to toggle the
        watchpoint status during the on_core_metadata calls (optional).

  Returns:
    server_port: (int) Port on which the server runs.
    debug_server_url: (str) grpc:// URL to the server.
    server_dump_dir: (str) The debug server's dump directory.
    server_thread: The server Thread object.
    server: The `EventListenerTestServicer` object.

  Raises:
    ValueError: If polling the server process for ready state is not successful
      within maximum polling count.
  """
  server_port = portpicker.pick_unused_port()
  debug_server_url = "grpc://localhost:%d" % server_port

  server_dump_dir = tempfile.mkdtemp() if dump_to_filesystem else None
  server = EventListenerTestServicer(
      server_port=server_port,
      dump_dir=server_dump_dir,
      toggle_watch_on_core_metadata=toggle_watch_on_core_metadata)

  def delay_then_run_server():
    time.sleep(server_start_delay_sec)
    server.run_server(blocking=blocking)

  server_thread = threading.Thread(target=delay_then_run_server)
  server_thread.start()

  if poll_server:
    if not _poll_server_till_success(
        50,
        0.2,
        debug_server_url,
        server_dump_dir,
        server,
        gpu_memory_fraction=0.1):
      raise ValueError(
          "Failed to start test gRPC debug server at port %d" % server_port)
    server.clear_data()
  return server_port, debug_server_url, server_dump_dir, server_thread, server

  def __init__(self,
               emulator_cmd=None,
               deadline=10,
               start_options=(),
               silent=False):
    """Constructs a DatastoreEmulator.

    Clients should use DatastoreEmulatorFactory to construct DatastoreEmulator
    instances.

    Args:
      emulator_cmd: A string representing the path to an executable script that
        invokes emulator binary.
      deadline: A integer representing number of seconds to wait for the
        datastore to start.
      start_options: A list of additional command-line options to pass to the
          emulator 'start' command.
      silent: A bool indicates if emulator runs in silent mode.

    Raises:
      IOError: if the emulator failed to start within the deadline
    """
    self._emulator_cmd = emulator_cmd
    self._http = httplib2.Http()
    self.__running = False

    self._silent = silent

    self._redirected_output = open(os.devnull, 'wb') if self._silent else None

    # Start the emulator and wait for it to start responding to requests.
    cmd = [self._emulator_cmd, 'start'] + _DEFAULT_EMULATOR_OPTIONS
    if start_options:
      cmd.extend(start_options)
    port = ParsePortFromOption(start_options or [])
    if not port:
      port = portpicker.pick_unused_port()
      cmd.append('--port=%d' % port)
    self._host = 'http://localhost:%d' % port
    cmd.append(tempfile.mkdtemp())

    # On windows, cloud_datastore_emulator.bat always prompts up
    # 'Terminate batch job (Y/N)'. Passing nul to this .bat avoids self.Stop()
    # hang at this prompt up.
    if sys.platform.startswith('win'):
      cmd.append('<nul')

    popen_kwargs = {}
    if self._silent:
      popen_kwargs.update(
          stdout=self._redirected_output, stderr=self._redirected_output)

    self.emulator_proc = subprocess.Popen(cmd, **popen_kwargs)
    if not self._WaitForStartup(deadline):
      raise IOError('emulator did not respond within %ds' % deadline)
    self.__datastore = None
    self.__running = True

  def test_basic(self):
    """Basic functionality test of START_PROCESS_REVERSE flavor."""
    portpicker.pick_unused_port().AndReturn(2345)
    # As the lock is mocked out, this provides a mox expectation.
    with self.proxy._process_lock:
      safe_subprocess.start_process_file(
          args=['/runtime'],
          input_string=self.runtime_config.SerializeToString(),
          env={'foo': 'bar',
               'PORT': '2345'},
          cwd=self.tmpdir,
          stderr=subprocess.PIPE).AndReturn(self.process)
    self.proxy._stderr_tee = FakeTee('')

    self.mox.ReplayAll()
    self.proxy.start()
    self.assertEquals(2345, self.proxy._proxy._port)
    self.mox.VerifyAll()

  def __init__(self, tmux_window, job, task_id):
    self.tmux_window = tmux_window
    self.job = job
    self.ip = '127.0.0.1' # hostname/ip address
    self.id = task_id
    self.port = portpicker.pick_unused_port()
    self.connect_instructions = 'tmux a -t '+self.tmux_window

    self.last_stdout = '<unavailable>'  # compatiblity with aws.py:Task
    self.last_stderr = '<unavailable>'

def create_local_cluster(num_workers, num_ps, protocol="grpc"):
  """Create local GRPC servers and return their servers."""
  worker_ports = [portpicker.pick_unused_port() for _ in range(num_workers)]
  ps_ports = [portpicker.pick_unused_port() for _ in range(num_ps)]
  cluster_dict = {
      "worker": ["localhost:%s" % port for port in worker_ports],
      "ps": ["localhost:%s" % port for port in ps_ports]}
  cs = tf.train.ClusterSpec(cluster_dict)

  workers = [
      tf.train.Server(
          cs, job_name="worker", protocol=protocol, task_index=ix, start=True)
      for ix in range(num_workers)]
  ps_servers = [
      tf.train.Server(
          cs, job_name="ps", protocol=protocol, task_index=ix, start=True)
      for ix in range(num_ps)]

  return workers, ps_servers

    def testPickUnusedCanSuccessfullyUsePortServer(self):

        with mock.patch.object(portpicker, '_pick_unused_port_without_server'):
            portpicker._pick_unused_port_without_server.side_effect = (
                Exception('eek!')
            )

            # Since _PickUnusedPortWithoutServer() raises an exception, if we
            # can successfully obtain a port, the portserver must be working.
            port = portpicker.pick_unused_port()
            self.assertTrue(self.IsUnusedTCPPort(port))
            self.assertTrue(self.IsUnusedUDPPort(port))

    def test_silent_request(self):
        """Call log_message once on the silent request handler for coverage's sake."""

        handler = AdderHandler()
        server = easyrpc.on(handler, adder_service, 'localhost', portpicker.pick_unused_port())

        class _DumbHandler(server.httpd.RequestHandlerClass):
            def __init__(self):
                pass

        http_request_handler = _DumbHandler()
        http_request_handler.log_message('unused')

def _create_cluster_spec(has_chief=False,
                         num_workers=1,
                         num_ps=0,
                         has_eval=False):
  if _portpicker_import_error:
    raise _portpicker_import_error  # pylint: disable=raising-bad-type

  cluster_spec = {}
  if has_chief:
    cluster_spec[CHIEF] = ["localhost:%s" % portpicker.pick_unused_port()]
  if num_workers:
    cluster_spec[WORKER] = [
        "localhost:%s" % portpicker.pick_unused_port()
        for _ in range(num_workers)
    ]
  if num_ps:
    cluster_spec[PS] = [
        "localhost:%s" % portpicker.pick_unused_port() for _ in range(num_ps)
    ]
  if has_eval:
    cluster_spec[EVALUATOR] = ["localhost:%s" % portpicker.pick_unused_port()]
  return cluster_spec

  def setUp(self):
    super(RpcOpTest, self).setUp()

    service_port = portpicker.pick_unused_port()

    server = grpc.server(logging_pool.pool(max_workers=25))
    servicer = rpc_op_test_servicer.RpcOpTestServicer()
    test_example_pb2_grpc.add_TestCaseServiceServicer_to_server(
        servicer, server)
    self._address = 'localhost:%d' % service_port
    server.add_insecure_port(self._address)
    server.start()
    self._server = server

  def setUp(self):
    self._debugger_data_server_grpc_port = portpicker.pick_unused_port()
    self._debug_url = (
        "grpc://localhost:%d" % self._debugger_data_server_grpc_port)
    self._logdir = tempfile.mkdtemp(prefix="tensorboard_dds_")

    self._debug_data_server = debugger_server_lib.DebuggerDataServer(
        self._debugger_data_server_grpc_port, self._logdir, always_flush=True)
    self._server_thread = threading.Thread(
        target=self._debug_data_server.start_the_debugger_data_receiving_server)
    self._server_thread.start()

    self.assertTrue(self._poll_server_till_success(50, 0.2))

def main(argv):
  logging.basicConfig(
    level=logging.INFO,
    format='%(levelname)s: %(threadName)s %(message)s')

  directory = os.getcwd()

  if argv[1:] == ['server']:
    port = gflags.FLAGS.port or portpicker.pick_unused_port()
    compilation_server_lib.RunServerAtPort(port)

  elif argv[1:] == ['local']:
    port = gflags.FLAGS.port or portpicker.pick_unused_port()
    server_proc = subprocess.Popen(
      ["python", argv[0], "--port", str(port), "server"])
    server_address = '127.0.0.1:{}'.format(port)
    compilation_client_lib.WaitTillHealthy(server_address)
    try:
      RunEditingEnvironment(
        directory,
        server_address=server_address)
    finally:
      try:
        logging.info("Terminating latex compilation server.")
        server_proc.kill()
      except OSError:
        pass

  elif argv[1:] == []:
    if not gflags.FLAGS.using_server:
      logging.fatal(
        "Please specify the server address when running in the client mode "
        "via --using_server")
    RunEditingEnvironment(directory, server_address=gflags.FLAGS.using_server)

  else:
    print "Unable to recognize command %r" % argv
    print __doc__
    sys.exit(1)

def pick_unused_port():
  """Returns an unused and unassigned local port."""
  if _portpicker_import_error:
    raise _portpicker_import_error  # pylint: disable=raising-bad-type

  global ASSIGNED_PORTS
  with lock:
    while True:
      port = portpicker.pick_unused_port()
      if port > 10000 and port not in ASSIGNED_PORTS:
        ASSIGNED_PORTS.add(port)
        logging.info('Using local port %r', port)
        return port