diff --git a/.gitignore b/.gitignore index 2280041c..76b1fcdb 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,7 @@ .DS_Store .ipynb_checkpoints *.pyc +*.gz __pycache__ *.egg-info build/* @@ -11,3 +12,4 @@ dist/* checkpoint htmlcov mnist +.vscode \ No newline at end of file diff --git a/examples/2_BasicModels/word2vec.py b/examples/2_BasicModels/word2vec.py index 45e87775..71fe9662 100644 --- a/examples/2_BasicModels/word2vec.py +++ b/examples/2_BasicModels/word2vec.py @@ -48,7 +48,7 @@ data_path = 'text8.zip' if not os.path.exists(data_path): print("Downloading the dataset... (It may take some time)") - filename, _ = urllib.urlretrieve(url, data_path) + filename, _ = urllib.request.urlretrieve(url, data_path) print("Done!") # Unzip the dataset file. Text has already been processed with zipfile.ZipFile(data_path) as f: @@ -168,7 +168,7 @@ def next_batch(batch_size, num_skips, skip_window): x_test = np.array([word2id[w] for w in eval_words]) average_loss = 0 - for step in xrange(1, num_steps + 1): + for step in range(1, num_steps + 1): # Get a new batch of data batch_x, batch_y = next_batch(batch_size, num_skips, skip_window) # Run training op @@ -186,10 +186,10 @@ def next_batch(batch_size, num_skips, skip_window): if step % eval_step == 0 or step == 1: print("Evaluation...") sim = sess.run(cosine_sim_op, feed_dict={X: x_test}) - for i in xrange(len(eval_words)): + for i in range(len(eval_words)): top_k = 8 # number of nearest neighbors nearest = (-sim[i, :]).argsort()[1:top_k + 1] log_str = '"%s" nearest neighbors:' % eval_words[i] - for k in xrange(top_k): + for k in range(top_k): log_str = '%s %s,' % (log_str, id2word[nearest[k]]) print(log_str) diff --git a/examples/3_NeuralNetworks/convolutional_network_raw.py b/examples/3_NeuralNetworks/convolutional_network_raw.py index d063f21f..4808738e 100644 --- a/examples/3_NeuralNetworks/convolutional_network_raw.py +++ b/examples/3_NeuralNetworks/convolutional_network_raw.py @@ -6,6 +6,8 @@ Author: Aymeric Damien Project: https://github.com/aymericdamien/TensorFlow-Examples/ + +广泛用于图像识别 """ from __future__ import division, print_function, absolute_import @@ -23,14 +25,14 @@ display_step = 10 # Network Parameters -num_input = 784 # MNIST data input (img shape: 28*28) -num_classes = 10 # MNIST total classes (0-9 digits) -dropout = 0.75 # Dropout, probability to keep units +num_input = 784 # MNIST data input (img shape: 28*28) +num_classes = 10 # MNIST total classes (0-9 digits) +dropout = 0.75 # Dropout, probability to keep units # tf Graph input X = tf.placeholder(tf.float32, [None, num_input]) Y = tf.placeholder(tf.float32, [None, num_classes]) -keep_prob = tf.placeholder(tf.float32) # dropout (keep probability) +keep_prob = tf.placeholder(tf.float32) # dropout (keep probability) # Create some wrappers for simplicity @@ -41,6 +43,8 @@ def conv2d(x, W, b, strides=1): return tf.nn.relu(x) +# 池化层夹在连续的卷积层中间， 用于压缩数据和参数的量，减小过拟合 +# use max_pool, get the max value of k*k matrix as the value def maxpool2d(x, k=2): # MaxPool2D wrapper return tf.nn.max_pool(x, ksize=[1, k, k, 1], strides=[1, k, k, 1], @@ -52,6 +56,11 @@ def conv_net(x, weights, biases, dropout): # MNIST data input is a 1-D vector of 784 features (28*28 pixels) # Reshape to match picture format [Height x Width x Channel] # Tensor input become 4-D: [Batch Size, Height, Width, Channel] + # a channel in this case is extended to be the grayscale image based on + # any such conventional primary color. + # By extension, a channel is any grayscale image with the same size + # as the "proper" image, and associated with it. + # channel可以用来表示灰度/透明度等等 x = tf.reshape(x, shape=[-1, 28, 28, 1]) # Convolution Layer @@ -76,11 +85,12 @@ def conv_net(x, weights, biases, dropout): out = tf.add(tf.matmul(fc1, weights['out']), biases['out']) return out + # Store layers weight & bias weights = { - # 5x5 conv, 1 input, 32 outputs + # 5x5 conv, 1 input, 32 outputs 卷积层第一个神经元 'wc1': tf.Variable(tf.random_normal([5, 5, 1, 32])), - # 5x5 conv, 32 inputs, 64 outputs + # 5x5 conv, 32 inputs, 64 outputs 卷积层第二个神经元 'wc2': tf.Variable(tf.random_normal([5, 5, 32, 64])), # fully connected, 7*7*64 inputs, 1024 outputs 'wd1': tf.Variable(tf.random_normal([7*7*64, 1024])), @@ -128,14 +138,14 @@ def conv_net(x, weights, biases, dropout): loss, acc = sess.run([loss_op, accuracy], feed_dict={X: batch_x, Y: batch_y, keep_prob: 1.0}) - print("Step " + str(step) + ", Minibatch Loss= " + \ - "{:.4f}".format(loss) + ", Training Accuracy= " + \ + print("Step " + str(step) + ", Minibatch Loss= " + + "{:.4f}".format(loss) + ", Training Accuracy= " + "{:.3f}".format(acc)) print("Optimization Finished!") # Calculate accuracy for 256 MNIST test images - print("Testing Accuracy:", \ - sess.run(accuracy, feed_dict={X: mnist.test.images[:256], - Y: mnist.test.labels[:256], - keep_prob: 1.0})) + print("Testing Accuracy:", + sess.run(accuracy, feed_dict={X: mnist.test.images[:256], + Y: mnist.test.labels[:256], + keep_prob: 1.0})) diff --git a/examples/3_NeuralNetworks/distributed/convolutional_network/Dockerfile b/examples/3_NeuralNetworks/distributed/convolutional_network/Dockerfile new file mode 100644 index 00000000..da642c36 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/convolutional_network/Dockerfile @@ -0,0 +1,3 @@ +FROM tensorflow/tensorflow:1.10.0 +ADD distributed_convolutional_network.py /opt +#ENTRYPOINT ["python", "/opt/distributed_convolutional_network.py"] \ No newline at end of file diff --git a/examples/3_NeuralNetworks/distributed/convolutional_network/distributed_convolutional_network.py b/examples/3_NeuralNetworks/distributed/convolutional_network/distributed_convolutional_network.py new file mode 100644 index 00000000..731c0b42 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/convolutional_network/distributed_convolutional_network.py @@ -0,0 +1,343 @@ +""" Convolutional Neural Network. + +Build and train a distributed convolutional neural network with TensorFlow. +This example is using the MNIST database of handwritten digits +(http://yann.lecun.com/exdb/mnist/) + +Author: Violet Cheng +""" + +from __future__ import division, print_function, absolute_import + +import tensorflow as tf +import json +import os +import sys +import tempfile +import time +# Import MNIST data +from tensorflow.examples.tutorials.mnist import input_data + + +flags = tf.app.flags +flags.DEFINE_string("data_dir", "/tmp/mnist-data", + "Directory for storing mnist data") +flags.DEFINE_boolean("download_only", False, + "Only perform downloading of data; Do not proceed to " + "session preparation, model definition or training") +flags.DEFINE_integer("task_index", None, + "Worker task index, should be >= 0. task_index=0 is " + "the master worker task the performs the variable " + "initialization ") +flags.DEFINE_integer("num_gpus", 1, "Total number of gpus for each machine." + "If you don't use GPU, please set it to '0'") +flags.DEFINE_integer("replicas_to_aggregate", None, + "Number of replicas to aggregate before parameter update" + "is applied (For sync_replicas mode only; default: " + "num_workers)") +flags.DEFINE_integer("hidden_units", 100, + "Number of units in the hidden layer of the NN") +flags.DEFINE_integer("train_steps", 1000, + "Number of (global) training steps to perform") +flags.DEFINE_integer("batch_size", 128, "Training batch size") +flags.DEFINE_float("learning_rate", 0.01, "Learning rate") +flags.DEFINE_boolean( + "sync_replicas", False, + "Use the sync_replicas (synchronized replicas) mode, " + "wherein the parameter updates from workers are aggregated " + "before applied to avoid stale gradients") +flags.DEFINE_boolean( + "existing_servers", False, "Whether servers already exists. If True, " + "will use the worker hosts via their GRPC URLs (one client process " + "per worker host). Otherwise, will create an in-process TensorFlow " + "server.") +flags.DEFINE_string("ps_hosts", "localhost:2222", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("worker_hosts", "localhost:2223,localhost:2224", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("job_name", None, "job name: worker or ps") + +FLAGS = flags.FLAGS + +# globals +IMAGE_PIXELS = 28 + +learning_rate = 0.001 +training_steps = 200 +batch_size = 128 +display_step = 10 + + +num_input = IMAGE_PIXELS * IMAGE_PIXELS +num_classes = 10 +dropout = 0.75 + + +# Create some wrappers for simplicity +def conv2d(x, W, b, strides=1): + # Conv2D wrapper, with bias and relu activation + x = tf.nn.conv2d(x, W, strides=[1, strides, strides, 1], padding='SAME') + x = tf.nn.bias_add(x, b) + return tf.nn.relu(x) + + +def maxpool2d(x, k=2): + # MaxPool2D wrapper + return tf.nn.max_pool(x, ksize=[1, k, k, 1], strides=[1, k, k, 1], + padding='SAME') + + +# Create model +def conv_net(x, weights, biases, dropout): + # MNIST data input is a 1-D vector of 784 features (28*28 pixels) + # Reshape to match picture format [Height x Width x Channel] + # Tensor input become 4-D: [Batch Size, Height, Width, Channel] + x = tf.reshape(x, shape=[-1, 28, 28, 1]) + + # Convolution Layer + conv1 = conv2d(x, weights['wc1'], biases['bc1']) + # Max Pooling (down-sampling) + conv1 = maxpool2d(conv1, k=2) + + # Convolution Layer + conv2 = conv2d(conv1, weights['wc2'], biases['bc2']) + # Max Pooling (down-sampling) + conv2 = maxpool2d(conv2, k=2) + + # Fully connected layer + # Reshape conv2 output to fit fully connected layer input + fc1 = tf.reshape(conv2, [-1, weights['wd1'].get_shape().as_list()[0]]) + fc1 = tf.add(tf.matmul(fc1, weights['wd1']), biases['bd1']) + fc1 = tf.nn.relu(fc1) + # Apply Dropout + fc1 = tf.nn.dropout(fc1, dropout) + + # Output, class prediction + out = tf.add(tf.matmul(fc1, weights['out']), biases['out']) + return out + + +def main(unused_argv): + # Parse environment variable TF_CONFIG to get job_name and task_index + + # If not explicitly specified in the constructor and the TF_CONFIG + # environment variable is present, load cluster_spec from TF_CONFIG. + tf_config = json.loads(os.environ.get('TF_CONFIG') or '{}') + task_config = tf_config.get('task', {}) + task_type = task_config.get('type') + task_index = task_config.get('index') + + FLAGS.job_name = task_type + FLAGS.task_index = task_index + + mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True) + if FLAGS.download_only: + sys.exit(0) + + if FLAGS.job_name is None or FLAGS.job_name == "": + raise ValueError("Must specify an explicit `job_name`") + if FLAGS.task_index is None or FLAGS.task_index == "": + raise ValueError("Must specify an explicit `task_index`") + + print("job name = %s" % FLAGS.job_name) + print("task index = %d" % FLAGS.task_index) + + cluster_config = tf_config.get('cluster', {}) + ps_hosts = cluster_config.get('ps') + worker_hosts = cluster_config.get('worker') + + ps_hosts_str = ','.join(ps_hosts) + worker_hosts_str = ','.join(worker_hosts) + + FLAGS.ps_hosts = ps_hosts_str + FLAGS.worker_hosts = worker_hosts_str + + # Construct the cluster and start the server + ps_spec = FLAGS.ps_hosts.split(",") + worker_spec = FLAGS.worker_hosts.split(",") + + # Get the number of workers. + num_workers = len(worker_spec) + + cluster = tf.train.ClusterSpec({"ps": ps_spec, "worker": worker_spec}) + + if not FLAGS.existing_servers: + # Not using existing servers. Create an in-process server. + server = tf.train.Server( + cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index) + if FLAGS.job_name == "ps": + server.join() + + is_chief = (FLAGS.task_index == 0) + if FLAGS.num_gpus > 0: + # Avoid gpu allocation conflict: now allocate task_num -> #gpu + # for each worker in the corresponding machine + gpu = (FLAGS.task_index % FLAGS.num_gpus) + worker_device = "/job:worker/task:%d/gpu:%d" % (FLAGS.task_index, gpu) + elif FLAGS.num_gpus == 0: + # Just allocate the CPU to worker server + cpu = 0 + worker_device = "/job:worker/task:%d/cpu:%d" % (FLAGS.task_index, cpu) + # The device setter will automatically place Variables ops on separate + # parameter servers (ps). The non-Variable ops will be placed on the workers. + # The ps use CPU and workers use corresponding GPU + with tf.device( + tf.train.replica_device_setter( + worker_device=worker_device, + ps_device="/job:ps/cpu:0", + cluster=cluster)): + global_step = tf.Variable(0, name="global_step", trainable=False) + + # Store layers weight & bias + weights = { + # 5x5 conv, 1 input, 32 outputs + 'wc1': tf.Variable(tf.random_normal([5, 5, 1, 32])), + # 5x5 conv, 32 inputs, 64 outputs + 'wc2': tf.Variable(tf.random_normal([5, 5, 32, 64])), + # fully connected, 7*7*64 inputs, 1024 outputs + 'wd1': tf.Variable(tf.random_normal([7*7*64, 1024])), + # 1024 inputs, 10 outputs (class prediction) + 'out': tf.Variable(tf.random_normal([1024, num_classes])) + } + + biases = { + 'bc1': tf.Variable(tf.random_normal([32])), + 'bc2': tf.Variable(tf.random_normal([64])), + 'bd1': tf.Variable(tf.random_normal([1024])), + 'out': tf.Variable(tf.random_normal([num_classes])) + } + # tf Graph input need in "with device" + X = tf.placeholder(tf.float32, [None, num_input]) + Y = tf.placeholder(tf.float32, [None, num_classes]) + keep_prob = tf.placeholder(tf.float32) # dropout (keep probability) + + # Construct model + logits = conv_net(X, weights, biases, keep_prob) + prediction = tf.nn.softmax(logits) + + # Define loss and optimizer + loss_op = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits( + logits=logits, labels=Y)) + opt = tf.train.AdamOptimizer(learning_rate=learning_rate) + + if FLAGS.sync_replicas: + if FLAGS.replicas_to_aggregate is None: + replicas_to_aggregate = num_workers + else: + replicas_to_aggregate = FLAGS.replicas_to_aggregate + + opt = tf.train.SyncReplicasOptimizer( + opt, + replicas_to_aggregate=replicas_to_aggregate, + total_num_replicas=num_workers, + name="mnist_sync_replicas") + + train_op = opt.minimize(loss_op, global_step=global_step) + + # supervisor's ops + if FLAGS.sync_replicas: + local_init_op = opt.local_step_init_op + if is_chief: + local_init_op = opt.chief_init_op + + ready_for_local_init_op = opt.ready_for_local_init_op + + # Initial token and chief queue runners required by the sync_replicas mode + chief_queue_runner = opt.get_chief_queue_runner() + sync_init_op = opt.get_init_tokens_op() + + # Evaluate model + correct_pred = tf.equal(tf.argmax(prediction, 1), tf.argmax(Y, 1)) + accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32)) + + # Initialize the variables (i.e. assign their default value) + init_op = tf.global_variables_initializer() + train_dir = tempfile.mkdtemp() + # Supervisor for distributed training + if FLAGS.sync_replicas: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + local_init_op=local_init_op, + ready_for_local_init_op=ready_for_local_init_op, + recovery_wait_secs=1, + global_step=global_step) + else: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + recovery_wait_secs=1, + global_step=global_step) + + sess_config = tf.ConfigProto( + allow_soft_placement=True, + log_device_placement=False, + device_filters=["/job:ps", + "/job:worker/task:%d" % FLAGS.task_index]) + + # The chief worker (task_index==0) session will prepare the session, + # while the remaining workers will wait for the preparation to complete. + if is_chief: + print("Worker %d: Initializing session..." % FLAGS.task_index) + else: + print("Worker %d: Waiting for session to be initialized..." % + FLAGS.task_index) + + if FLAGS.existing_servers: + server_grpc_url = "grpc://" + worker_spec[FLAGS.task_index] + print("Using existing server at: %s" % server_grpc_url) + + sess = sv.prepare_or_wait_for_session( + server_grpc_url, config=sess_config) + else: + sess = sv.prepare_or_wait_for_session( + server.target, config=sess_config) + + print("Worker %d: Session initialization complete." % FLAGS.task_index) + + if FLAGS.sync_replicas and is_chief: + # Chief worker will start the chief queue runner and call the init op. + sess.run(sync_init_op) + sv.start_queue_runners(sess, [chief_queue_runner]) + + # Perform training + time_begin = time.time() + print("Training begins @ %f" % time_begin) + + local_step = 0 + while True: + # Training feed + # put your data and train op here + batch_xs, batch_ys = mnist.train.next_batch(batch_size) + _, step = sess.run([train_op, global_step], + feed_dict={X: batch_xs, Y: batch_ys, keep_prob: 0.8}) + local_step += 1 + + now = time.time() + if local_step % display_step == 0: + loss, acc = sess.run([loss_op, accuracy], feed_dict={X: batch_xs, + Y: batch_ys, keep_prob: 1.0}) + + print("%f: Worker %d: training step %d done (global step: %d) Minibatch Loss=%.4f Accuracy=%.3f" % + (now, FLAGS.task_index, local_step, step, loss, acc)) + + if step >= FLAGS.train_steps: + break + + time_end = time.time() + print("Training ends @ %f" % time_end) + training_time = time_end - time_begin + print("Training elapsed time: %f s" % training_time) + + # Validation feed + test_len = 128 + test_data = mnist.test.images[:test_len] + test_label = mnist.test.labels[:test_len] + print("Testing Accuracy:", + sess.run(accuracy, feed_dict={X: test_data, Y: test_label, keep_prob: 1.0})) + + +if __name__ == "__main__": + tf.app.run() diff --git a/examples/3_NeuralNetworks/distributed/convolutional_network/ps/env.py b/examples/3_NeuralNetworks/distributed/convolutional_network/ps/env.py new file mode 100644 index 00000000..9acc8808 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/convolutional_network/ps/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'ps', 'index': 0}}) diff --git a/examples/3_NeuralNetworks/distributed/convolutional_network/tf_job_convolutional_network.yaml b/examples/3_NeuralNetworks/distributed/convolutional_network/tf_job_convolutional_network.yaml new file mode 100644 index 00000000..b0c51f41 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/convolutional_network/tf_job_convolutional_network.yaml @@ -0,0 +1,32 @@ +apiVersion: "kubeflow.org/v1alpha2" +kind: "TFJob" +metadata: + name: "dist-convolutional-network" +spec: + tfReplicaSpecs: + PS: + replicas: 2 + restartPolicy: Never + template: + spec: + containers: + - name: tensorflow + args: + - python + - /opt/distributed_convolutional_network.py + - --train_steps=1000 + image: hanacolor/tf-dist-convolutional-network:1.0 + imagePullPolicy: Always + Worker: + replicas: 2 + restartPolicy: Never + template: + spec: + containers: + - name: tensorflow + args: + - python + - /opt/distributed_convolutional_network.py + - --train_steps=1000 + image: hanacolor/tf-dist-convolutional-network:1.0 + imagePullPolicy: Always \ No newline at end of file diff --git a/examples/3_NeuralNetworks/distributed/convolutional_network/worker0/env.py b/examples/3_NeuralNetworks/distributed/convolutional_network/worker0/env.py new file mode 100644 index 00000000..6ab7b3ec --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/convolutional_network/worker0/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'worker', 'index': 0}}) diff --git a/examples/3_NeuralNetworks/distributed/convolutional_network/worker1/env.py b/examples/3_NeuralNetworks/distributed/convolutional_network/worker1/env.py new file mode 100644 index 00000000..eb816f9b --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/convolutional_network/worker1/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'worker', 'index': 1}}) diff --git a/examples/3_NeuralNetworks/distributed/example_framework/Readme.md b/examples/3_NeuralNetworks/distributed/example_framework/Readme.md new file mode 100644 index 00000000..0bf6677d --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/Readme.md @@ -0,0 +1,11 @@ +## How to create a distributed Tensorflow workload from single node version + +This folder is a demo example and framework of distributed version tf workload. + +We have 1 parameter server folder and 2 worker folders in this example. The python code file "example.py" are exact the same in the 3 folders, but "env.py" indicated the different roles they will play. + +To run the example, simply cd into 3 folders and run + +``` +python example.py +``` diff --git a/examples/3_NeuralNetworks/distributed/example_framework/ps/env.py b/examples/3_NeuralNetworks/distributed/example_framework/ps/env.py new file mode 100644 index 00000000..9acc8808 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/ps/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'ps', 'index': 0}}) diff --git a/examples/3_NeuralNetworks/distributed/example_framework/ps/example.py b/examples/3_NeuralNetworks/distributed/example_framework/ps/example.py new file mode 100644 index 00000000..e3fbc80f --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/ps/example.py @@ -0,0 +1,309 @@ +# 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. +# ============================================================================== +"""Distributed MNIST training and validation, with model replicas. + +A simple softmax model with one hidden layer is defined. The parameters +(weights and biases) are located on one parameter server (ps), while the ops +are executed on two worker nodes by default. The TF sessions also run on the +worker node. +Multiple invocations of this script can be done in parallel, with different +values for --task_index. There should be exactly one invocation with +--task_index, which will create a master session that carries out variable +initialization. The other, non-master, sessions will wait for the master +session to finish the initialization before proceeding to the training stage. + +The coordination between the multiple worker invocations occurs due to +the definition of the parameters on the same ps devices. The parameter updates +from one worker is visible to all other workers. As such, the workers can +perform forward computation and gradient calculation in parallel, which +should lead to increased training speed for the simple model. +""" + +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function + +import json +import math +import os +import sys +import tempfile +import time +import env +import tensorflow as tf +from tensorflow.examples.tutorials.mnist import input_data +# 1. All flags +# We could move the flags to a separate file +flags = tf.app.flags +flags.DEFINE_string("data_dir", "/tmp/mnist-data", + "Directory for storing mnist data") +flags.DEFINE_boolean("download_only", False, + "Only perform downloading of data; Do not proceed to " + "session preparation, model definition or training") +flags.DEFINE_integer("task_index", None, + "Worker task index, should be >= 0. task_index=0 is " + "the master worker task the performs the variable " + "initialization ") +flags.DEFINE_integer("num_gpus", 1, "Total number of gpus for each machine." + "If you don't use GPU, please set it to '0'") +flags.DEFINE_integer("replicas_to_aggregate", None, + "Number of replicas to aggregate before parameter update" + "is applied (For sync_replicas mode only; default: " + "num_workers)") +flags.DEFINE_integer("hidden_units", 100, + "Number of units in the hidden layer of the NN") +flags.DEFINE_integer("train_steps", 20000, + "Number of (global) training steps to perform") +flags.DEFINE_integer("batch_size", 100, "Training batch size") +flags.DEFINE_float("learning_rate", 0.01, "Learning rate") +flags.DEFINE_boolean( + "sync_replicas", False, + "Use the sync_replicas (synchronized replicas) mode, " + "wherein the parameter updates from workers are aggregated " + "before applied to avoid stale gradients") +flags.DEFINE_boolean( + "existing_servers", False, "Whether servers already exists. If True, " + "will use the worker hosts via their GRPC URLs (one client process " + "per worker host). Otherwise, will create an in-process TensorFlow " + "server.") +flags.DEFINE_string("ps_hosts", "localhost:2222", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("worker_hosts", "localhost:2223,localhost:2224", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("job_name", None, "job name: worker or ps") + +FLAGS = flags.FLAGS + +IMAGE_PIXELS = 28 + + +def main(unused_argv): + # 2. tf config envrionment variables + + # Parse environment variable TF_CONFIG to get job_name and task_index + + # If not explicitly specified in the constructor and the TF_CONFIG + # environment variable is present, load cluster_spec from TF_CONFIG. + tf_config = json.loads(os.environ.get('TF_CONFIG') or '{}') + task_config = tf_config.get('task', {}) + task_type = task_config.get('type') + task_index = task_config.get('index') + + FLAGS.job_name = task_type + FLAGS.task_index = task_index + # 3. input dataset + mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True) + if FLAGS.download_only: + sys.exit(0) + + if FLAGS.job_name is None or FLAGS.job_name == "": + raise ValueError("Must specify an explicit `job_name`") + if FLAGS.task_index is None or FLAGS.task_index == "": + raise ValueError("Must specify an explicit `task_index`") + + print("job name = %s" % FLAGS.job_name) + print("task index = %d" % FLAGS.task_index) + # 4. generate ClusterSpec + cluster_config = tf_config.get('cluster', {}) + ps_hosts = cluster_config.get('ps') + worker_hosts = cluster_config.get('worker') + + ps_hosts_str = ','.join(ps_hosts) + worker_hosts_str = ','.join(worker_hosts) + + FLAGS.ps_hosts = ps_hosts_str + FLAGS.worker_hosts = worker_hosts_str + + # Construct the cluster and start the server + ps_spec = FLAGS.ps_hosts.split(",") + worker_spec = FLAGS.worker_hosts.split(",") + + # Get the number of workers. + num_workers = len(worker_spec) + + cluster = tf.train.ClusterSpec({"ps": ps_spec, "worker": worker_spec}) + + # 5. create server for job + if not FLAGS.existing_servers: + # Not using existing servers. Create an in-process server. + server = tf.train.Server( + cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index) + if FLAGS.job_name == "ps": + server.join() + + is_chief = (FLAGS.task_index == 0) + if FLAGS.num_gpus > 0: + # Avoid gpu allocation conflict: now allocate task_num -> #gpu + # for each worker in the corresponding machine + gpu = (FLAGS.task_index % FLAGS.num_gpus) + worker_device = "/job:worker/task:%d/gpu:%d" % (FLAGS.task_index, gpu) + elif FLAGS.num_gpus == 0: + # Just allocate the CPU to worker server + cpu = 0 + worker_device = "/job:worker/task:%d/cpu:%d" % (FLAGS.task_index, cpu) + # The device setter will automatically place Variables ops on separate + # parameter servers (ps). The non-Variable ops will be placed on the workers. + # The ps use CPU and workers use corresponding GPU + with tf.device( + tf.train.replica_device_setter( + worker_device=worker_device, + ps_device="/job:ps/cpu:0", + cluster=cluster)): + global_step = tf.Variable(0, name="global_step", trainable=False) + # 6. Define variables for training + + # Variables of the hidden layer + hid_w = tf.Variable( + tf.truncated_normal( + [IMAGE_PIXELS * IMAGE_PIXELS, FLAGS.hidden_units], + stddev=1.0 / IMAGE_PIXELS), + name="hid_w") + hid_b = tf.Variable(tf.zeros([FLAGS.hidden_units]), name="hid_b") + + # Variables of the softmax layer + sm_w = tf.Variable( + tf.truncated_normal( + [FLAGS.hidden_units, 10], + stddev=1.0 / math.sqrt(FLAGS.hidden_units)), + name="sm_w") + sm_b = tf.Variable(tf.zeros([10]), name="sm_b") + + # Training Ops: located on the worker specified with FLAGS.task_index + x = tf.placeholder(tf.float32, [None, IMAGE_PIXELS * IMAGE_PIXELS]) + y_ = tf.placeholder(tf.float32, [None, 10]) + + hid_lin = tf.nn.xw_plus_b(x, hid_w, hid_b) + hid = tf.nn.relu(hid_lin) + + y = tf.nn.softmax(tf.nn.xw_plus_b(hid, sm_w, sm_b)) + cross_entropy = - \ + tf.reduce_sum(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0))) + + # 7. define optimizer + opt = tf.train.AdamOptimizer(FLAGS.learning_rate) + + if FLAGS.sync_replicas: + if FLAGS.replicas_to_aggregate is None: + replicas_to_aggregate = num_workers + else: + replicas_to_aggregate = FLAGS.replicas_to_aggregate + + opt = tf.train.SyncReplicasOptimizer( + opt, + replicas_to_aggregate=replicas_to_aggregate, + total_num_replicas=num_workers, + name="mnist_sync_replicas") + + train_step = opt.minimize(cross_entropy, global_step=global_step) + + # supervisor's ops + if FLAGS.sync_replicas: + local_init_op = opt.local_step_init_op + if is_chief: + local_init_op = opt.chief_init_op + + ready_for_local_init_op = opt.ready_for_local_init_op + + # Initial token and chief queue runners required by the sync_replicas mode + chief_queue_runner = opt.get_chief_queue_runner() + sync_init_op = opt.get_init_tokens_op() + + init_op = tf.global_variables_initializer() + train_dir = tempfile.mkdtemp() + # Supervisor for distributed training + if FLAGS.sync_replicas: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + local_init_op=local_init_op, + ready_for_local_init_op=ready_for_local_init_op, + recovery_wait_secs=1, + global_step=global_step) + else: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + recovery_wait_secs=1, + global_step=global_step) + + sess_config = tf.ConfigProto( + allow_soft_placement=True, + log_device_placement=False, + device_filters=["/job:ps", + "/job:worker/task:%d" % FLAGS.task_index]) + + # The chief worker (task_index==0) session will prepare the session, + # while the remaining workers will wait for the preparation to complete. + if is_chief: + print("Worker %d: Initializing session..." % FLAGS.task_index) + else: + print("Worker %d: Waiting for session to be initialized..." % + FLAGS.task_index) + + if FLAGS.existing_servers: + server_grpc_url = "grpc://" + worker_spec[FLAGS.task_index] + print("Using existing server at: %s" % server_grpc_url) + + sess = sv.prepare_or_wait_for_session( + server_grpc_url, config=sess_config) + else: + sess = sv.prepare_or_wait_for_session( + server.target, config=sess_config) + + print("Worker %d: Session initialization complete." % FLAGS.task_index) + + if FLAGS.sync_replicas and is_chief: + # Chief worker will start the chief queue runner and call the init op. + sess.run(sync_init_op) + sv.start_queue_runners(sess, [chief_queue_runner]) + + # Perform training + time_begin = time.time() + print("Training begins @ %f" % time_begin) + + local_step = 0 + while True: + # Training feed + # put your data and train op here + batch_xs, batch_ys = mnist.train.next_batch(FLAGS.batch_size) + train_feed = {x: batch_xs, y_: batch_ys} + + _, step = sess.run([train_step, global_step], feed_dict=train_feed) + local_step += 1 + + now = time.time() + print("%f: Worker %d: training step %d done (global step: %d)" % + (now, FLAGS.task_index, local_step, step)) + + if step >= FLAGS.train_steps: + break + + time_end = time.time() + print("Training ends @ %f" % time_end) + training_time = time_end - time_begin + print("Training elapsed time: %f s" % training_time) + + # Validation feed + val_feed = {x: mnist.validation.images, y_: mnist.validation.labels} + val_xent = sess.run(cross_entropy, feed_dict=val_feed) + print("After %d training step(s), validation cross entropy = %g" % + (FLAGS.train_steps, val_xent)) + + +# call main +if __name__ == "__main__": + tf.app.run() diff --git a/examples/3_NeuralNetworks/distributed/example_framework/worker0/env.py b/examples/3_NeuralNetworks/distributed/example_framework/worker0/env.py new file mode 100644 index 00000000..6ab7b3ec --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/worker0/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'worker', 'index': 0}}) diff --git a/examples/3_NeuralNetworks/distributed/example_framework/worker0/example.py b/examples/3_NeuralNetworks/distributed/example_framework/worker0/example.py new file mode 100644 index 00000000..b39daeab --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/worker0/example.py @@ -0,0 +1,303 @@ +# 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. +# ============================================================================== +"""Distributed MNIST training and validation, with model replicas. + +A simple softmax model with one hidden layer is defined. The parameters +(weights and biases) are located on one parameter server (ps), while the ops +are executed on two worker nodes by default. The TF sessions also run on the +worker node. +Multiple invocations of this script can be done in parallel, with different +values for --task_index. There should be exactly one invocation with +--task_index, which will create a master session that carries out variable +initialization. The other, non-master, sessions will wait for the master +session to finish the initialization before proceeding to the training stage. + +The coordination between the multiple worker invocations occurs due to +the definition of the parameters on the same ps devices. The parameter updates +from one worker is visible to all other workers. As such, the workers can +perform forward computation and gradient calculation in parallel, which +should lead to increased training speed for the simple model. +""" + +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function + +import json +import math +import os +import sys +import tempfile +import time +import env +import tensorflow as tf +from tensorflow.examples.tutorials.mnist import input_data + +flags = tf.app.flags +flags.DEFINE_string("data_dir", "/tmp/mnist-data", + "Directory for storing mnist data") +flags.DEFINE_boolean("download_only", False, + "Only perform downloading of data; Do not proceed to " + "session preparation, model definition or training") +flags.DEFINE_integer("task_index", None, + "Worker task index, should be >= 0. task_index=0 is " + "the master worker task the performs the variable " + "initialization ") +flags.DEFINE_integer("num_gpus", 1, "Total number of gpus for each machine." + "If you don't use GPU, please set it to '0'") +flags.DEFINE_integer("replicas_to_aggregate", None, + "Number of replicas to aggregate before parameter update" + "is applied (For sync_replicas mode only; default: " + "num_workers)") +flags.DEFINE_integer("hidden_units", 100, + "Number of units in the hidden layer of the NN") +flags.DEFINE_integer("train_steps", 20000, + "Number of (global) training steps to perform") +flags.DEFINE_integer("batch_size", 100, "Training batch size") +flags.DEFINE_float("learning_rate", 0.01, "Learning rate") +flags.DEFINE_boolean( + "sync_replicas", False, + "Use the sync_replicas (synchronized replicas) mode, " + "wherein the parameter updates from workers are aggregated " + "before applied to avoid stale gradients") +flags.DEFINE_boolean( + "existing_servers", False, "Whether servers already exists. If True, " + "will use the worker hosts via their GRPC URLs (one client process " + "per worker host). Otherwise, will create an in-process TensorFlow " + "server.") +flags.DEFINE_string("ps_hosts", "localhost:2222", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("worker_hosts", "localhost:2223,localhost:2224", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("job_name", None, "job name: worker or ps") + +FLAGS = flags.FLAGS + +IMAGE_PIXELS = 28 + + +def main(unused_argv): + # Parse environment variable TF_CONFIG to get job_name and task_index + + # If not explicitly specified in the constructor and the TF_CONFIG + # environment variable is present, load cluster_spec from TF_CONFIG. + tf_config = json.loads(os.environ.get('TF_CONFIG') or '{}') + task_config = tf_config.get('task', {}) + task_type = task_config.get('type') + task_index = task_config.get('index') + + FLAGS.job_name = task_type + FLAGS.task_index = task_index + + mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True) + if FLAGS.download_only: + sys.exit(0) + + if FLAGS.job_name is None or FLAGS.job_name == "": + raise ValueError("Must specify an explicit `job_name`") + if FLAGS.task_index is None or FLAGS.task_index == "": + raise ValueError("Must specify an explicit `task_index`") + + print("job name = %s" % FLAGS.job_name) + print("task index = %d" % FLAGS.task_index) + + cluster_config = tf_config.get('cluster', {}) + ps_hosts = cluster_config.get('ps') + worker_hosts = cluster_config.get('worker') + + ps_hosts_str = ','.join(ps_hosts) + worker_hosts_str = ','.join(worker_hosts) + + FLAGS.ps_hosts = ps_hosts_str + FLAGS.worker_hosts = worker_hosts_str + + # Construct the cluster and start the server + ps_spec = FLAGS.ps_hosts.split(",") + worker_spec = FLAGS.worker_hosts.split(",") + + # Get the number of workers. + num_workers = len(worker_spec) + + cluster = tf.train.ClusterSpec({"ps": ps_spec, "worker": worker_spec}) + + if not FLAGS.existing_servers: + # Not using existing servers. Create an in-process server. + server = tf.train.Server( + cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index) + if FLAGS.job_name == "ps": + server.join() + + is_chief = (FLAGS.task_index == 0) + if FLAGS.num_gpus > 0: + # Avoid gpu allocation conflict: now allocate task_num -> #gpu + # for each worker in the corresponding machine + gpu = (FLAGS.task_index % FLAGS.num_gpus) + worker_device = "/job:worker/task:%d/gpu:%d" % (FLAGS.task_index, gpu) + elif FLAGS.num_gpus == 0: + # Just allocate the CPU to worker server + cpu = 0 + worker_device = "/job:worker/task:%d/cpu:%d" % (FLAGS.task_index, cpu) + # The device setter will automatically place Variables ops on separate + # parameter servers (ps). The non-Variable ops will be placed on the workers. + # The ps use CPU and workers use corresponding GPU + with tf.device( + tf.train.replica_device_setter( + worker_device=worker_device, + ps_device="/job:ps/cpu:0", + cluster=cluster)): + global_step = tf.Variable(0, name="global_step", trainable=False) + + # Variables of the hidden layer + hid_w = tf.Variable( + tf.truncated_normal( + [IMAGE_PIXELS * IMAGE_PIXELS, FLAGS.hidden_units], + stddev=1.0 / IMAGE_PIXELS), + name="hid_w") + hid_b = tf.Variable(tf.zeros([FLAGS.hidden_units]), name="hid_b") + + # Variables of the softmax layer + sm_w = tf.Variable( + tf.truncated_normal( + [FLAGS.hidden_units, 10], + stddev=1.0 / math.sqrt(FLAGS.hidden_units)), + name="sm_w") + sm_b = tf.Variable(tf.zeros([10]), name="sm_b") + + # Training Ops: located on the worker specified with FLAGS.task_index + x = tf.placeholder(tf.float32, [None, IMAGE_PIXELS * IMAGE_PIXELS]) + y_ = tf.placeholder(tf.float32, [None, 10]) + + hid_lin = tf.nn.xw_plus_b(x, hid_w, hid_b) + hid = tf.nn.relu(hid_lin) + + y = tf.nn.softmax(tf.nn.xw_plus_b(hid, sm_w, sm_b)) + cross_entropy = - \ + tf.reduce_sum(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0))) + + # optimizer + opt = tf.train.AdamOptimizer(FLAGS.learning_rate) + + if FLAGS.sync_replicas: + if FLAGS.replicas_to_aggregate is None: + replicas_to_aggregate = num_workers + else: + replicas_to_aggregate = FLAGS.replicas_to_aggregate + + opt = tf.train.SyncReplicasOptimizer( + opt, + replicas_to_aggregate=replicas_to_aggregate, + total_num_replicas=num_workers, + name="mnist_sync_replicas") + + train_step = opt.minimize(cross_entropy, global_step=global_step) + + # supervisor's ops + if FLAGS.sync_replicas: + local_init_op = opt.local_step_init_op + if is_chief: + local_init_op = opt.chief_init_op + + ready_for_local_init_op = opt.ready_for_local_init_op + + # Initial token and chief queue runners required by the sync_replicas mode + chief_queue_runner = opt.get_chief_queue_runner() + sync_init_op = opt.get_init_tokens_op() + + init_op = tf.global_variables_initializer() + train_dir = tempfile.mkdtemp() + # Supervisor for distributed training + if FLAGS.sync_replicas: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + local_init_op=local_init_op, + ready_for_local_init_op=ready_for_local_init_op, + recovery_wait_secs=1, + global_step=global_step) + else: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + recovery_wait_secs=1, + global_step=global_step) + + sess_config = tf.ConfigProto( + allow_soft_placement=True, + log_device_placement=False, + device_filters=["/job:ps", + "/job:worker/task:%d" % FLAGS.task_index]) + + # The chief worker (task_index==0) session will prepare the session, + # while the remaining workers will wait for the preparation to complete. + if is_chief: + print("Worker %d: Initializing session..." % FLAGS.task_index) + else: + print("Worker %d: Waiting for session to be initialized..." % + FLAGS.task_index) + + if FLAGS.existing_servers: + server_grpc_url = "grpc://" + worker_spec[FLAGS.task_index] + print("Using existing server at: %s" % server_grpc_url) + + sess = sv.prepare_or_wait_for_session( + server_grpc_url, config=sess_config) + else: + sess = sv.prepare_or_wait_for_session( + server.target, config=sess_config) + + print("Worker %d: Session initialization complete." % FLAGS.task_index) + + if FLAGS.sync_replicas and is_chief: + # Chief worker will start the chief queue runner and call the init op. + sess.run(sync_init_op) + sv.start_queue_runners(sess, [chief_queue_runner]) + + # Perform training + time_begin = time.time() + print("Training begins @ %f" % time_begin) + + local_step = 0 + while True: + # Training feed + # put your data and train op here + batch_xs, batch_ys = mnist.train.next_batch(FLAGS.batch_size) + train_feed = {x: batch_xs, y_: batch_ys} + + _, step = sess.run([train_step, global_step], feed_dict=train_feed) + local_step += 1 + + now = time.time() + print("%f: Worker %d: training step %d done (global step: %d)" % + (now, FLAGS.task_index, local_step, step)) + + if step >= FLAGS.train_steps: + break + + time_end = time.time() + print("Training ends @ %f" % time_end) + training_time = time_end - time_begin + print("Training elapsed time: %f s" % training_time) + + # Validation feed + val_feed = {x: mnist.validation.images, y_: mnist.validation.labels} + val_xent = sess.run(cross_entropy, feed_dict=val_feed) + print("After %d training step(s), validation cross entropy = %g" % + (FLAGS.train_steps, val_xent)) + + +if __name__ == "__main__": + tf.app.run() diff --git a/examples/3_NeuralNetworks/distributed/example_framework/worker1/env.py b/examples/3_NeuralNetworks/distributed/example_framework/worker1/env.py new file mode 100644 index 00000000..eb816f9b --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/worker1/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'worker', 'index': 1}}) diff --git a/examples/3_NeuralNetworks/distributed/example_framework/worker1/example.py b/examples/3_NeuralNetworks/distributed/example_framework/worker1/example.py new file mode 100644 index 00000000..b39daeab --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/example_framework/worker1/example.py @@ -0,0 +1,303 @@ +# 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. +# ============================================================================== +"""Distributed MNIST training and validation, with model replicas. + +A simple softmax model with one hidden layer is defined. The parameters +(weights and biases) are located on one parameter server (ps), while the ops +are executed on two worker nodes by default. The TF sessions also run on the +worker node. +Multiple invocations of this script can be done in parallel, with different +values for --task_index. There should be exactly one invocation with +--task_index, which will create a master session that carries out variable +initialization. The other, non-master, sessions will wait for the master +session to finish the initialization before proceeding to the training stage. + +The coordination between the multiple worker invocations occurs due to +the definition of the parameters on the same ps devices. The parameter updates +from one worker is visible to all other workers. As such, the workers can +perform forward computation and gradient calculation in parallel, which +should lead to increased training speed for the simple model. +""" + +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function + +import json +import math +import os +import sys +import tempfile +import time +import env +import tensorflow as tf +from tensorflow.examples.tutorials.mnist import input_data + +flags = tf.app.flags +flags.DEFINE_string("data_dir", "/tmp/mnist-data", + "Directory for storing mnist data") +flags.DEFINE_boolean("download_only", False, + "Only perform downloading of data; Do not proceed to " + "session preparation, model definition or training") +flags.DEFINE_integer("task_index", None, + "Worker task index, should be >= 0. task_index=0 is " + "the master worker task the performs the variable " + "initialization ") +flags.DEFINE_integer("num_gpus", 1, "Total number of gpus for each machine." + "If you don't use GPU, please set it to '0'") +flags.DEFINE_integer("replicas_to_aggregate", None, + "Number of replicas to aggregate before parameter update" + "is applied (For sync_replicas mode only; default: " + "num_workers)") +flags.DEFINE_integer("hidden_units", 100, + "Number of units in the hidden layer of the NN") +flags.DEFINE_integer("train_steps", 20000, + "Number of (global) training steps to perform") +flags.DEFINE_integer("batch_size", 100, "Training batch size") +flags.DEFINE_float("learning_rate", 0.01, "Learning rate") +flags.DEFINE_boolean( + "sync_replicas", False, + "Use the sync_replicas (synchronized replicas) mode, " + "wherein the parameter updates from workers are aggregated " + "before applied to avoid stale gradients") +flags.DEFINE_boolean( + "existing_servers", False, "Whether servers already exists. If True, " + "will use the worker hosts via their GRPC URLs (one client process " + "per worker host). Otherwise, will create an in-process TensorFlow " + "server.") +flags.DEFINE_string("ps_hosts", "localhost:2222", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("worker_hosts", "localhost:2223,localhost:2224", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("job_name", None, "job name: worker or ps") + +FLAGS = flags.FLAGS + +IMAGE_PIXELS = 28 + + +def main(unused_argv): + # Parse environment variable TF_CONFIG to get job_name and task_index + + # If not explicitly specified in the constructor and the TF_CONFIG + # environment variable is present, load cluster_spec from TF_CONFIG. + tf_config = json.loads(os.environ.get('TF_CONFIG') or '{}') + task_config = tf_config.get('task', {}) + task_type = task_config.get('type') + task_index = task_config.get('index') + + FLAGS.job_name = task_type + FLAGS.task_index = task_index + + mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True) + if FLAGS.download_only: + sys.exit(0) + + if FLAGS.job_name is None or FLAGS.job_name == "": + raise ValueError("Must specify an explicit `job_name`") + if FLAGS.task_index is None or FLAGS.task_index == "": + raise ValueError("Must specify an explicit `task_index`") + + print("job name = %s" % FLAGS.job_name) + print("task index = %d" % FLAGS.task_index) + + cluster_config = tf_config.get('cluster', {}) + ps_hosts = cluster_config.get('ps') + worker_hosts = cluster_config.get('worker') + + ps_hosts_str = ','.join(ps_hosts) + worker_hosts_str = ','.join(worker_hosts) + + FLAGS.ps_hosts = ps_hosts_str + FLAGS.worker_hosts = worker_hosts_str + + # Construct the cluster and start the server + ps_spec = FLAGS.ps_hosts.split(",") + worker_spec = FLAGS.worker_hosts.split(",") + + # Get the number of workers. + num_workers = len(worker_spec) + + cluster = tf.train.ClusterSpec({"ps": ps_spec, "worker": worker_spec}) + + if not FLAGS.existing_servers: + # Not using existing servers. Create an in-process server. + server = tf.train.Server( + cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index) + if FLAGS.job_name == "ps": + server.join() + + is_chief = (FLAGS.task_index == 0) + if FLAGS.num_gpus > 0: + # Avoid gpu allocation conflict: now allocate task_num -> #gpu + # for each worker in the corresponding machine + gpu = (FLAGS.task_index % FLAGS.num_gpus) + worker_device = "/job:worker/task:%d/gpu:%d" % (FLAGS.task_index, gpu) + elif FLAGS.num_gpus == 0: + # Just allocate the CPU to worker server + cpu = 0 + worker_device = "/job:worker/task:%d/cpu:%d" % (FLAGS.task_index, cpu) + # The device setter will automatically place Variables ops on separate + # parameter servers (ps). The non-Variable ops will be placed on the workers. + # The ps use CPU and workers use corresponding GPU + with tf.device( + tf.train.replica_device_setter( + worker_device=worker_device, + ps_device="/job:ps/cpu:0", + cluster=cluster)): + global_step = tf.Variable(0, name="global_step", trainable=False) + + # Variables of the hidden layer + hid_w = tf.Variable( + tf.truncated_normal( + [IMAGE_PIXELS * IMAGE_PIXELS, FLAGS.hidden_units], + stddev=1.0 / IMAGE_PIXELS), + name="hid_w") + hid_b = tf.Variable(tf.zeros([FLAGS.hidden_units]), name="hid_b") + + # Variables of the softmax layer + sm_w = tf.Variable( + tf.truncated_normal( + [FLAGS.hidden_units, 10], + stddev=1.0 / math.sqrt(FLAGS.hidden_units)), + name="sm_w") + sm_b = tf.Variable(tf.zeros([10]), name="sm_b") + + # Training Ops: located on the worker specified with FLAGS.task_index + x = tf.placeholder(tf.float32, [None, IMAGE_PIXELS * IMAGE_PIXELS]) + y_ = tf.placeholder(tf.float32, [None, 10]) + + hid_lin = tf.nn.xw_plus_b(x, hid_w, hid_b) + hid = tf.nn.relu(hid_lin) + + y = tf.nn.softmax(tf.nn.xw_plus_b(hid, sm_w, sm_b)) + cross_entropy = - \ + tf.reduce_sum(y_ * tf.log(tf.clip_by_value(y, 1e-10, 1.0))) + + # optimizer + opt = tf.train.AdamOptimizer(FLAGS.learning_rate) + + if FLAGS.sync_replicas: + if FLAGS.replicas_to_aggregate is None: + replicas_to_aggregate = num_workers + else: + replicas_to_aggregate = FLAGS.replicas_to_aggregate + + opt = tf.train.SyncReplicasOptimizer( + opt, + replicas_to_aggregate=replicas_to_aggregate, + total_num_replicas=num_workers, + name="mnist_sync_replicas") + + train_step = opt.minimize(cross_entropy, global_step=global_step) + + # supervisor's ops + if FLAGS.sync_replicas: + local_init_op = opt.local_step_init_op + if is_chief: + local_init_op = opt.chief_init_op + + ready_for_local_init_op = opt.ready_for_local_init_op + + # Initial token and chief queue runners required by the sync_replicas mode + chief_queue_runner = opt.get_chief_queue_runner() + sync_init_op = opt.get_init_tokens_op() + + init_op = tf.global_variables_initializer() + train_dir = tempfile.mkdtemp() + # Supervisor for distributed training + if FLAGS.sync_replicas: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + local_init_op=local_init_op, + ready_for_local_init_op=ready_for_local_init_op, + recovery_wait_secs=1, + global_step=global_step) + else: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + recovery_wait_secs=1, + global_step=global_step) + + sess_config = tf.ConfigProto( + allow_soft_placement=True, + log_device_placement=False, + device_filters=["/job:ps", + "/job:worker/task:%d" % FLAGS.task_index]) + + # The chief worker (task_index==0) session will prepare the session, + # while the remaining workers will wait for the preparation to complete. + if is_chief: + print("Worker %d: Initializing session..." % FLAGS.task_index) + else: + print("Worker %d: Waiting for session to be initialized..." % + FLAGS.task_index) + + if FLAGS.existing_servers: + server_grpc_url = "grpc://" + worker_spec[FLAGS.task_index] + print("Using existing server at: %s" % server_grpc_url) + + sess = sv.prepare_or_wait_for_session( + server_grpc_url, config=sess_config) + else: + sess = sv.prepare_or_wait_for_session( + server.target, config=sess_config) + + print("Worker %d: Session initialization complete." % FLAGS.task_index) + + if FLAGS.sync_replicas and is_chief: + # Chief worker will start the chief queue runner and call the init op. + sess.run(sync_init_op) + sv.start_queue_runners(sess, [chief_queue_runner]) + + # Perform training + time_begin = time.time() + print("Training begins @ %f" % time_begin) + + local_step = 0 + while True: + # Training feed + # put your data and train op here + batch_xs, batch_ys = mnist.train.next_batch(FLAGS.batch_size) + train_feed = {x: batch_xs, y_: batch_ys} + + _, step = sess.run([train_step, global_step], feed_dict=train_feed) + local_step += 1 + + now = time.time() + print("%f: Worker %d: training step %d done (global step: %d)" % + (now, FLAGS.task_index, local_step, step)) + + if step >= FLAGS.train_steps: + break + + time_end = time.time() + print("Training ends @ %f" % time_end) + training_time = time_end - time_begin + print("Training elapsed time: %f s" % training_time) + + # Validation feed + val_feed = {x: mnist.validation.images, y_: mnist.validation.labels} + val_xent = sess.run(cross_entropy, feed_dict=val_feed) + print("After %d training step(s), validation cross entropy = %g" % + (FLAGS.train_steps, val_xent)) + + +if __name__ == "__main__": + tf.app.run() diff --git a/examples/3_NeuralNetworks/distributed/recurrent_network/Dockerfile b/examples/3_NeuralNetworks/distributed/recurrent_network/Dockerfile new file mode 100644 index 00000000..bb07571e --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/recurrent_network/Dockerfile @@ -0,0 +1,3 @@ +FROM tensorflow/tensorflow:1.10.0 +ADD distributed_recurrent_network.py /opt +#ENTRYPOINT ["python", "/opt/distributed_recurrent_network.py"] diff --git a/examples/3_NeuralNetworks/distributed/recurrent_network/distributed_recurrent_network.py b/examples/3_NeuralNetworks/distributed/recurrent_network/distributed_recurrent_network.py new file mode 100644 index 00000000..2855e668 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/recurrent_network/distributed_recurrent_network.py @@ -0,0 +1,326 @@ +# 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. +# ============================================================================== +"""Distributed MNIST training and validation, with model replicas. + +A simple softmax model with one hidden layer is defined. The parameters +(weights and biases) are located on one parameter server (ps), while the ops +are executed on two worker nodes by default. The TF sessions also run on the +worker node. +Multiple invocations of this script can be done in parallel, with different +values for --task_index. There should be exactly one invocation with +--task_index, which will create a master session that carries out variable +initialization. The other, non-master, sessions will wait for the master +session to finish the initialization before proceeding to the training stage. + +The coordination between the multiple worker invocations occurs due to +the definition of the parameters on the same ps devices. The parameter updates +from one worker is visible to all other workers. As such, the workers can +perform forward computation and gradient calculation in parallel, which +should lead to increased training speed for the simple model. +""" + +from __future__ import absolute_import +from __future__ import division +from __future__ import print_function + +import json +import math +import os +import sys +import tempfile +import time +import tensorflow as tf +from tensorflow.examples.tutorials.mnist import input_data +from tensorflow.contrib import rnn + +flags = tf.app.flags +flags.DEFINE_string("data_dir", "/tmp/mnist-data", + "Directory for storing mnist data") +flags.DEFINE_boolean("download_only", False, + "Only perform downloading of data; Do not proceed to " + "session preparation, model definition or training") +flags.DEFINE_integer("task_index", None, + "Worker task index, should be >= 0. task_index=0 is " + "the master worker task the performs the variable " + "initialization ") +flags.DEFINE_integer("num_gpus", 1, "Total number of gpus for each machine." + "If you don't use GPU, please set it to '0'") +flags.DEFINE_integer("replicas_to_aggregate", None, + "Number of replicas to aggregate before parameter update" + "is applied (For sync_replicas mode only; default: " + "num_workers)") +flags.DEFINE_integer("hidden_units", 100, + "Number of units in the hidden layer of the NN") +flags.DEFINE_integer("train_steps", 1000, + "Number of (global) training steps to perform") +flags.DEFINE_integer("batch_size", 128, "Training batch size") +flags.DEFINE_float("learning_rate", 0.01, "Learning rate") +flags.DEFINE_boolean( + "sync_replicas", False, + "Use the sync_replicas (synchronized replicas) mode, " + "wherein the parameter updates from workers are aggregated " + "before applied to avoid stale gradients") +flags.DEFINE_boolean( + "existing_servers", False, "Whether servers already exists. If True, " + "will use the worker hosts via their GRPC URLs (one client process " + "per worker host). Otherwise, will create an in-process TensorFlow " + "server.") +flags.DEFINE_string("ps_hosts", "localhost:2222", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("worker_hosts", "localhost:2223,localhost:2224", + "Comma-separated list of hostname:port pairs") +flags.DEFINE_string("job_name", None, "job name: worker or ps") + +FLAGS = flags.FLAGS + +# globals +IMAGE_PIXELS = 28 + +learning_rate = 0.001 +batch_size = 128 +display_step = 50 + + +num_input = IMAGE_PIXELS +# rnn_size = 100 +timesteps = IMAGE_PIXELS +num_hidden = 128 +num_classes = 10 + + +def main(unused_argv): + # Parse environment variable TF_CONFIG to get job_name and task_index + + # If not explicitly specified in the constructor and the TF_CONFIG + # environment variable is present, load cluster_spec from TF_CONFIG. + tf_config = json.loads(os.environ.get('TF_CONFIG') or '{}') + task_config = tf_config.get('task', {}) + task_type = task_config.get('type') + task_index = task_config.get('index') + + FLAGS.job_name = task_type + FLAGS.task_index = task_index + + mnist = input_data.read_data_sets(FLAGS.data_dir, one_hot=True) + if FLAGS.download_only: + sys.exit(0) + + if FLAGS.job_name is None or FLAGS.job_name == "": + raise ValueError("Must specify an explicit `job_name`") + if FLAGS.task_index is None or FLAGS.task_index == "": + raise ValueError("Must specify an explicit `task_index`") + + print("job name = %s" % FLAGS.job_name) + print("task index = %d" % FLAGS.task_index) + print("total training steps are %d" % FLAGS.train_steps) + cluster_config = tf_config.get('cluster', {}) + ps_hosts = cluster_config.get('ps') + worker_hosts = cluster_config.get('worker') + + ps_hosts_str = ','.join(ps_hosts) + worker_hosts_str = ','.join(worker_hosts) + + FLAGS.ps_hosts = ps_hosts_str + FLAGS.worker_hosts = worker_hosts_str + + # Construct the cluster and start the server + ps_spec = FLAGS.ps_hosts.split(",") + worker_spec = FLAGS.worker_hosts.split(",") + + # Get the number of workers. + num_workers = len(worker_spec) + + cluster = tf.train.ClusterSpec({"ps": ps_spec, "worker": worker_spec}) + + if not FLAGS.existing_servers: + # Not using existing servers. Create an in-process server. + server = tf.train.Server( + cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index) + if FLAGS.job_name == "ps": + server.join() + + is_chief = (FLAGS.task_index == 0) + if FLAGS.num_gpus > 0: + # Avoid gpu allocation conflict: now allocate task_num -> #gpu + # for each worker in the corresponding machine + gpu = (FLAGS.task_index % FLAGS.num_gpus) + worker_device = "/job:worker/task:%d/gpu:%d" % (FLAGS.task_index, gpu) + elif FLAGS.num_gpus == 0: + # Just allocate the CPU to worker server + cpu = 0 + worker_device = "/job:worker/task:%d/cpu:%d" % (FLAGS.task_index, cpu) + # The device setter will automatically place Variables ops on separate + # parameter servers (ps). The non-Variable ops will be placed on the workers. + # The ps use CPU and workers use corresponding GPU + with tf.device( + tf.train.replica_device_setter( + worker_device=worker_device, + ps_device="/job:ps/cpu:0", + cluster=cluster)): + global_step = tf.Variable(0, name="global_step", trainable=False) + + # Define weights + weights = { + 'out': tf.Variable(tf.random_normal([num_hidden, num_classes])) + } + + biases = { + 'out': tf.Variable(tf.random_normal([num_classes])) + } + + # Training Ops: located on the worker specified with FLAGS.task_index + X = tf.placeholder("float", [None, timesteps, num_input]) + Y = tf.placeholder("float", [None, num_classes]) + + img = tf.unstack(X, timesteps, 1) + + # Define a lstm cell with tensorflow + lstm_cell = rnn.BasicLSTMCell(num_hidden, forget_bias=1.0) + + # Get lstm cell output + outputs, _ = rnn.static_rnn(lstm_cell, img, dtype=tf.float32) + + # Linear activation, using rnn inner loop last output + logits = tf.matmul(outputs[-1], weights['out']) + biases['out'] + prediction = tf.nn.softmax(logits) + # Define loss and optimizer + loss_op = tf.reduce_mean(tf.nn.softmax_cross_entropy_with_logits( + logits=logits, labels=Y)) + # Evaluate model (with test logits, for dropout to be disabled) + correct_pred = tf.equal( + tf.argmax(prediction, 1), tf.argmax(Y, 1)) + accuracy = tf.reduce_mean(tf.cast(correct_pred, tf.float32)) + + # optimizer + opt = tf.train.AdamOptimizer(FLAGS.learning_rate) + + if FLAGS.sync_replicas: + if FLAGS.replicas_to_aggregate is None: + replicas_to_aggregate = num_workers + else: + replicas_to_aggregate = FLAGS.replicas_to_aggregate + + opt = tf.train.SyncReplicasOptimizer( + opt, + replicas_to_aggregate=replicas_to_aggregate, + total_num_replicas=num_workers, + name="mnist_sync_replicas") + + train_step = opt.minimize(loss_op, global_step=global_step) + + # supervisor's ops + if FLAGS.sync_replicas: + local_init_op = opt.local_step_init_op + if is_chief: + local_init_op = opt.chief_init_op + + ready_for_local_init_op = opt.ready_for_local_init_op + + # Initial token and chief queue runners required by the sync_replicas mode + chief_queue_runner = opt.get_chief_queue_runner() + sync_init_op = opt.get_init_tokens_op() + + init_op = tf.global_variables_initializer() + train_dir = tempfile.mkdtemp() + # Supervisor for distributed training + if FLAGS.sync_replicas: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + local_init_op=local_init_op, + ready_for_local_init_op=ready_for_local_init_op, + recovery_wait_secs=1, + global_step=global_step) + else: + sv = tf.train.Supervisor( + is_chief=is_chief, + logdir=train_dir, + init_op=init_op, + recovery_wait_secs=1, + global_step=global_step) + + sess_config = tf.ConfigProto( + allow_soft_placement=True, + log_device_placement=False, + device_filters=["/job:ps", + "/job:worker/task:%d" % FLAGS.task_index]) + + # The chief worker (task_index==0) session will prepare the session, + # while the remaining workers will wait for the preparation to complete. + if is_chief: + print("Worker %d: Initializing session..." % FLAGS.task_index) + else: + print("Worker %d: Waiting for session to be initialized..." % + FLAGS.task_index) + + if FLAGS.existing_servers: + server_grpc_url = "grpc://" + worker_spec[FLAGS.task_index] + print("Using existing server at: %s" % server_grpc_url) + + sess = sv.prepare_or_wait_for_session( + server_grpc_url, config=sess_config) + else: + sess = sv.prepare_or_wait_for_session( + server.target, config=sess_config) + + print("Worker %d: Session initialization complete." % FLAGS.task_index) + + if FLAGS.sync_replicas and is_chief: + # Chief worker will start the chief queue runner and call the init op. + sess.run(sync_init_op) + sv.start_queue_runners(sess, [chief_queue_runner]) + + # Perform training + time_begin = time.time() + print("Training begins @ %f" % time_begin) + + local_step = 0 + while True: + # Training feed + # put your data and train op here + batch_xs, batch_ys = mnist.train.next_batch(batch_size) + batch_xs = batch_xs.reshape((batch_size, timesteps, num_input)) + _, step = sess.run([train_step, global_step], + feed_dict={X: batch_xs, Y: batch_ys}) + local_step += 1 + + now = time.time() + if local_step % display_step == 0: + loss, acc = sess.run([loss_op, accuracy], feed_dict={X: batch_xs, + Y: batch_ys}) + + print("%f: Worker %d: training step %d done (global step: %d) Minibatch Loss=%.4f Accuracy=%.3f" % + (now, FLAGS.task_index, local_step, step, loss, acc)) + + if step >= FLAGS.train_steps: + break + + time_end = time.time() + print("Training ends @ %f" % time_end) + training_time = time_end - time_begin + print("Training elapsed time: %f s" % training_time) + + # Validation feed + test_len = 128 + test_data = mnist.test.images[:test_len].reshape( + (-1, timesteps, num_input)) + test_label = mnist.test.labels[:test_len] + print("Testing Accuracy:", + sess.run(accuracy, feed_dict={X: test_data, Y: test_label})) + + +if __name__ == "__main__": + tf.app.run() diff --git a/examples/3_NeuralNetworks/distributed/recurrent_network/ps/env.py b/examples/3_NeuralNetworks/distributed/recurrent_network/ps/env.py new file mode 100644 index 00000000..9acc8808 --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/recurrent_network/ps/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'ps', 'index': 0}}) diff --git a/examples/3_NeuralNetworks/distributed/recurrent_network/tf_job_recurrent_network.yaml b/examples/3_NeuralNetworks/distributed/recurrent_network/tf_job_recurrent_network.yaml new file mode 100644 index 00000000..c202e7cd --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/recurrent_network/tf_job_recurrent_network.yaml @@ -0,0 +1,32 @@ +apiVersion: "kubeflow.org/v1alpha2" +kind: "TFJob" +metadata: + name: "dist-recurrent-network" +spec: + tfReplicaSpecs: + PS: + replicas: 2 + restartPolicy: Never + template: + spec: + containers: + - name: tensorflow + args: + - python + - /opt/distributed_recurrent_network.py + - --train_steps=2000 + image: hanacolor/tf-dist-recurrent-network:1.0 + imagePullPolicy: Always + Worker: + replicas: 2 + restartPolicy: Never + template: + spec: + containers: + - name: tensorflow + args: + - python + - /opt/distributed_recurrent_network.py + - --train_steps=2000 + image: hanacolor/tf-dist-recurrent-network:1.0 + imagePullPolicy: Always \ No newline at end of file diff --git a/examples/3_NeuralNetworks/distributed/recurrent_network/worker0/env.py b/examples/3_NeuralNetworks/distributed/recurrent_network/worker0/env.py new file mode 100644 index 00000000..6ab7b3ec --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/recurrent_network/worker0/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'worker', 'index': 0}}) diff --git a/examples/3_NeuralNetworks/distributed/recurrent_network/worker1/env.py b/examples/3_NeuralNetworks/distributed/recurrent_network/worker1/env.py new file mode 100644 index 00000000..eb816f9b --- /dev/null +++ b/examples/3_NeuralNetworks/distributed/recurrent_network/worker1/env.py @@ -0,0 +1,7 @@ +import json +import os +cluster = {'ps': ['localhost:2224'], + 'worker': ['localhost:2226', 'localhost:2227']} +os.environ['TF_CONFIG'] = json.dumps( + {'cluster': cluster, + 'task': {'type': 'worker', 'index': 1}}) diff --git a/examples/3_NeuralNetworks/recurrent_network.py b/examples/3_NeuralNetworks/recurrent_network.py index fbc3d271..b013d1aa 100644 --- a/examples/3_NeuralNetworks/recurrent_network.py +++ b/examples/3_NeuralNetworks/recurrent_network.py @@ -33,10 +33,10 @@ display_step = 200 # Network Parameters -num_input = 28 # MNIST data input (img shape: 28*28) -timesteps = 28 # timesteps -num_hidden = 128 # hidden layer num of features -num_classes = 10 # MNIST total classes (0-9 digits) +num_input = 28 # MNIST data input (img shape: 28*28) +timesteps = 28 # timesteps +num_hidden = 128 # hidden layer num of features +num_classes = 10 # MNIST total classes (0-9 digits) # tf Graph input X = tf.placeholder("float", [None, timesteps, num_input]) @@ -44,6 +44,7 @@ # Define weights weights = { + # 128 rows, 10 cols 'out': tf.Variable(tf.random_normal([num_hidden, num_classes])) } biases = { @@ -64,11 +65,12 @@ def RNN(x, weights, biases): lstm_cell = rnn.BasicLSTMCell(num_hidden, forget_bias=1.0) # Get lstm cell output - outputs, states = rnn.static_rnn(lstm_cell, x, dtype=tf.float32) + outputs, _ = rnn.static_rnn(lstm_cell, x, dtype=tf.float32) # Linear activation, using rnn inner loop last output return tf.matmul(outputs[-1], weights['out']) + biases['out'] + logits = RNN(X, weights, biases) prediction = tf.nn.softmax(logits) @@ -101,15 +103,16 @@ def RNN(x, weights, biases): # Calculate batch loss and accuracy loss, acc = sess.run([loss_op, accuracy], feed_dict={X: batch_x, Y: batch_y}) - print("Step " + str(step) + ", Minibatch Loss= " + \ - "{:.4f}".format(loss) + ", Training Accuracy= " + \ + print("Step " + str(step) + ", Minibatch Loss= " + + "{:.4f}".format(loss) + ", Training Accuracy= " + "{:.3f}".format(acc)) print("Optimization Finished!") # Calculate accuracy for 128 mnist test images test_len = 128 - test_data = mnist.test.images[:test_len].reshape((-1, timesteps, num_input)) + test_data = mnist.test.images[:test_len].reshape( + (-1, timesteps, num_input)) test_label = mnist.test.labels[:test_len] - print("Testing Accuracy:", \ - sess.run(accuracy, feed_dict={X: test_data, Y: test_label})) + print("Testing Accuracy:", + sess.run(accuracy, feed_dict={X: test_data, Y: test_label}))