<commit_before># Copyright 2016 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================

"""Tests and benchmarks for creating RPC clusters on localhost."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import time

import numpy as np
import tensorflow as tf

from tensorflow.python.util import net_lib


def create_local_cluster(num_workers, num_ps, protocol="grpc"):
  """Create local GRPC servers and return their servers."""
  worker_ports = [net_lib.pick_unused_port_or_die() for _ in range(num_workers)]
  ps_ports = [net_lib.pick_unused_port_or_die() 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


class CreateLocalClusterTest(tf.test.TestCase):

  def testCreateLocalCluster(self):
    workers, _ = 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"):
      var0 = tf.Variable(0.0)
    with tf.device("/job:ps/task:1"):
      var1 = tf.Variable(1.0)
    worker_sessions[0].run([var0.initializer, var1.initializer])
    with tf.device("/job:ps/task:0"):
      var2 = tf.Variable(2.0)
    with tf.device("/job:ps/task:1"):
      var3 = tf.Variable(3.0)
    worker_sessions[1].run([var2.initializer, var3.initializer])

    # Read values back in the opposite session
    self.assertAllEqual(0.0, var0.eval(session=worker_sessions[1]))
    self.assertAllEqual(1.0, var1.eval(session=worker_sessions[1]))
    self.assertAllEqual(2.0, var2.eval(session=worker_sessions[0]))
    self.assertAllEqual(3.0, var3.eval(session=worker_sessions[0]))


class CreateLocalClusterBenchmark(tf.test.Benchmark):

  def benchmarkCreateLocalCluster(self):
    deltas = []
    iters = 50
    for _ in range(iters):
      start_time = time.time()<commit_msg>Reduce iterations in localhost cluster performance test. Change: 129480234<commit_after># Copyright 2016 The TensorFlow Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ==============================================================================

"""Tests and benchmarks for creating RPC clusters on localhost."""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import time

import numpy as np
import tensorflow as tf

from tensorflow.python.util import net_lib


def create_local_cluster(num_workers, num_ps, protocol="grpc"):
  """Create local GRPC servers and return their servers."""
  worker_ports = [net_lib.pick_unused_port_or_die() for _ in range(num_workers)]
  ps_ports = [net_lib.pick_unused_port_or_die() 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


class CreateLocalClusterTest(tf.test.TestCase):

  def testCreateLocalCluster(self):
    workers, _ = 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"):
      var0 = tf.Variable(0.0)
    with tf.device("/job:ps/task:1"):
      var1 = tf.Variable(1.0)
    worker_sessions[0].run([var0.initializer, var1.initializer])
    with tf.device("/job:ps/task:0"):
      var2 = tf.Variable(2.0)
    with tf.device("/job:ps/task:1"):
      var3 = tf.Variable(3.0)
    worker_sessions[1].run([var2.initializer, var3.initializer])

    # Read values back in the opposite session
    self.assertAllEqual(0.0, var0.eval(session=worker_sessions[1]))
    self.assertAllEqual(1.0, var1.eval(session=worker_sessions[1]))
    self.assertAllEqual(2.0, var2.eval(session=worker_sessions[0]))
    self.assertAllEqual(3.0, var3.eval(session=worker_sessions[0]))


class CreateLocalClusterBenchmark(tf.test.Benchmark):

  def benchmarkCreateLocalCluster(self):
    deltas = []
    iters = 5
    for _ in range(iters):
      start_time = time.time()
{'fraction_non_alphanumeric': 0.1348314606741573, 'fraction_numerical': 0.018386108273748723, 'mean_word_length': 4.340909090909091, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4669638', 'n_tokens_mistral': 1842, 'n_tokens_neox': 1804, 'n_words': 499}