Learn Hands-On Machine Learning with Scikit-Learn and TensorFlow-Chapter 12

Cuda and CuDNN. CuDNN is based on CUDA.

install tensorflow for GPU: pip install tensorflow-gpu

set the environment variable CUDA_VISIBLE_DEVICES=3,2 then run a tensorflow program to let the tensorflow program only see the GPU card 2 and 3;otherwise, a tensorflow program will grab all memory of all GPU cards installed on your system.

You can also use the tf.ConfigProto() class when constructing the Session object to let the program use only fraction of each GPU card on your system:

config=tf.ConfigProto()
config.gpu_options.per_process_gpu_memory_fraction=0.4
session=tf.Session(config=config)

Simple placement rule: tensorflow will put all nodes on GPU0, if no GPU, will place them on CPU. Of course you can specify the device a node is placed on and that takes higher privilege. To pin nodes on a device:

with tf.device("/cpu:0"):
    a=tf.Variable(3.0)
    b=tf.constant(4.0)

The placement is done when running the graph for the first time, for all nodes even they are not evaluated in the run. You can pass a function to the constructor of device to generate different device names for different ops(the function takes a op parameter).

An operation can be placed on a device only if it has implementation for that kind of device.

Tensorflow manages an inter-op thread pool for each device. The ops that can be parallelized will be sent to the inter-op pool of the device they were placed. Tensorflow also manages an intra-op thread pool for each device. If an op has a multithreaded kernel, it will be split and sent to the intra-op thread pool for parallel execution.

The arrangement of nodes in the compute graph determine the control dependency of these nodes,i.e., some nodes can only be executed after the nodes they depend on are executed. But you can create your own control dependency using tf.control_dependencies([tensor1, tensor2,…]). The nodes in the control dependency block will be evaluated after tensor1, tensor2,… are evaluated.

A task is a running tensorflow program(process, server). A job is a group is tasks. The tasks in a job can be run on different machines. A cluster describes a set of jobs. To start a task, create a tf.train.Server object passing the cluster as its parameter:

server=tf.train.Server(cluster_spec,job_name="worker",task_index=0)

The server uses the information in the cluster_spec to communicate with other servers(possibly on different machines).

After all the tasked described in the cluster are running, you can run a client in any process on any machine to open a session on any of the servers. The client process may even be the same process as one of the servers. The operations will be transferred from the client to the server and placed on the GPUs/CPUs of the server to executed. The evaluation result will be transported back to the client from the server.

a=tf.constant(1.0)
b=a+2
c=a*3
with tf.Session("grpc://machine-b.example.com:2222")assess:
    print(c.eval())

When the server’s master service receives the request from a client, it may distribute the operations to other tasks. The master service use the worker service to actually execute the operation. A client can pin its operations on different jobs/tasks/devices, but it does not need to open different session on different tasks to evaluate the operations. Just create one session on one task, and deliver the graph to it. The session task will handle the distribution of the operations among other tasks. If a node is not pinned on any devices, it will be executed on GPU#0 of machine of the session task.

Within a device block created with the tf.train.replica_device_setter(ps_tasks=2) function, the variables will be pinned to /job:ps/task:0 and /job:ps/task:1 in round-robin way. So you must name a job after “ps” in the cluster’s specification. Other operations apart from variables in the device block will be pinned to /job:worker automatically(defaults to task 0). So you must name a job after “worker” in the cluster. You can use embedded device block to override the settings in outer device block.

The value of nodes except variables does not persist across graph runs. The value of variables does not persist after local session(you do not provide a url when creating it) closes. The state of variables persists after distributed session(you provide a url when creating it) closes. Variables are stored in containers. If you do not specify a container, the variable is stored in the default container. Variables are maintained by servers in the containers of a cluster. If you run a client which creates a variable and opens a distributed session, after the client ends, you run the client again, then you will get the same variable as the one in the client’s first run. You can reset a container when opening a distributed session:

tf.Session.reset("grpc://machine-a.example.com:2222",["my_problem_1"])

If two clients define a variable with the same name and in the same container, they share the variable. If two clients define a queue with the same shared_name and in the same container, they share the same queue.

q=tf.FIFOQueue(capacity=10,dtypes=[tf.float32],shapes=[[2]],name="q",shared_name="shared_q")

This queue can contain 10 tensor, each of which contains 2 floats. The dtypes is a list because an item in the queue can be a tuple of tensors, and dtypes specifies the types of each tensor in the tuple. Similarly, shapes is a list specifying the shapes for all the tensors in the tuple. Every item in the queue must have the same shape and type. You need to create an operation by calling q.enqueue() to send data to the queue. To pull data from the queue, you need to create an operation by calling q.dequeue()(pull one tensor at each time) or q.dequeue_many(size)(pull multiple tensors each time). RandomShuffleQueue is a queue that pops up a random item in it than the first one. Its constructor has a min_after_dequeue parameter. If after current dequeue operation, there are less than min_after_dequeue items remaining in the queue, the current dequeue operation will be blocked. PaddingFIFOQueue allows different sizes(but same rank) with its elements. If you pop multiple elements using dequeue_many, the popped elements will be padded with zeroes to make them the same size.

loading data

If you use placeholders to feed data to the graph, you will read the data from filesystem to the client, transfer the data from client to server, the server will distribute the data to other tasks in the cluster. Every client must repeat the same process if they train the same model, which means the same data will be read/transferred multiple times in the filesystem/network.

You can load the data into a variable. Since the variable can be shared among clients, the clients are not required to read them multiple times. Only one client needs to load the data from file sytem and transfer it to the task to initialize the shared variable; other clients just use it directly.

If the data is too large to fit in memory, you can create operations to read it from file system directly. Since operations are executed on servers, clients do not need to read the data.

reader=tf.TextLineReader(skip_header_lines=1)
filename_queue=tf.FIFOQueue(capacity=10,dtypes=[tf.string],shapes=[()])
filename=tf.placeholder(tf.string)
enqueue_filename=filename_queue.enqueue([filename])
close_filename_queue=filename_queue.close()
key,value=reader.read(filename_queue)
x1,x2,target=tf.decode_csv(value,record_defaults=[[-1.],[-1.],[-1]])
features=tf.stack([x1,x2])
instance_queue=tf.RandomShuffleQueue(capacity=10,min_after_dequeue=2,dtypes=[tf.float32,tf.int32],shapes=[[2],[]],name="instance_q",shared_name="shared_instance_q")
enqueue_instance=instance_queue.enqueue([features,target])
close_instance_queue=instance_queue.close()

Every line creates an operation in the graph. The graph is ready to read the data from a file, but you need to evaluate the related operations to start the read process.

sess.run(enqueue_filename,feed_dict={filename:"my_test.csv"})
sess.run(close_filename_queue)
while True:
    sess.run(enqueue_instance)
sess.run(close_instance_queue)

First, you evaluate enqueue_filename to put a file name to the filename queue. Then you evaluate enqueue_instance multiple times in a loop. Note that enqueue_instance depends on the operation reader.read(filename_queue). Every time reader.read(filename_queue) is evaluated, it reads a line(record) from the file. If all records in current file are read, reader.read will try to dequeue another file name from the file name queue. Since the filename queue has been closed, an exception will occur and the loop ends. Now all records in the file have been enqueued to the instance_queue queue. The clients that need to pull data from this queue just create the RandomShuffleQueue node with the same shared name, and evaluate its dequeue_up_to operation.

Using the tasks in a cluster, you can parallelize neuron networks on multiple devices on multiple machines. Just create sessions using different urls, the neuron networks will be dispatched to different tasks on different servers. You can define compute graphs in different device blocks with different device #, so they will be dispatched to different devices on the same machine. You can specify job/task/device when define device blocks, then you just need a single session to dispatch the graph to any server in a cluster, and the task will be able to distribute the graph among servers/devices. If the multiple neuron networks are needed to be ensembled, i.e., their outputs are needed to be aggregated by other operations, you can define the multiple neuron networks and the aggregation operations in one graph, then dispatch the whole graph to one task, and let the task dispatch individual nns to proper devices, wait for their outputs, and do the aggregation operations. This is called in-graph replication. You can also define and run each nn in an independent client, then use another client to read the inputs for the nns, and use one more client to handle the outputs from each nn. You can use the queues to share and sync the input data and the output data between the clients.

The above parallelization techniques are good to parallelize different NNs to different devices. But can I parallelize a single NN among different devices? For full connected NNs, it is hard because the following layer is dependent on the output of previous layer so they cannot be parallelized. The operations in one layer cannot be parallelized too because if they are dispatched to different servers, their outputs need to be transferred to other devices to complete the operation of next layer. The data transfer cost makes the parallelization not economic. But for some partially connected NNs, the parallelization is feasible because fewer data need to be transferred across devices to complete the operations.

The parallelization can also be done to training data, i.e., replicating the NN and feeding the replicas with different mini-batches from the training set at the same time. After the gradients are computed in a replica, they are not use to adjust the parameters of the replica, but wait for the gradients of other replicas available, then aggravated with them (such as mean) and the aggravated gradients are used to adjust the parameters of every replica, then all replica start the next training step. This is called synchronous updates. While in asynchronous updates, the gradients computed by a replica immediately update the parameters of all replicas including itself. In other words, for a replica, some gradients may be computed based on different parameters than other gradients, or, the gradients do not point to the current descent direction because the current position has been changed(updated). The stale gradients may make the algorithm diverge.

Note that whether you use synchronous updates or asynchronous updates, the replicas share a same set of model parameters(variables) that are pinned on one device on parameter server. The parameters are transferred to every device of the replicas to compute gradients, and the gradients are transferred back to the parameter server to update the parameters. The data transfer may saturate the bandwidth between devices and networks.

Did you like this?
Tip admin with Cryptocurrency