1

2

3

4

5

6

7

8

9

10

11

12

13

14

# coding=utf-8 

# 多臺機器，每臺機器有一個顯示卡、或者多個顯示卡，這種訓練叫做分散式訓練 

import  tensorflow as tf 

# 現在假設我們有A、B、C、D四臺機器，首先需要在各臺機器上寫一份程式碼，並跑起來，各機器上的程式碼內容大部分相同 

# 除了開始定義的時候，需要各自指定該臺機器的task之外。<br># 以機器A為例子，A機器上的程式碼如下： 

cluster

"worker": [ 

"A_IP:2222",           # 格式 IP地址：埠號，第一臺機器A的IP地址 ,在程式碼中需要用這臺機器計算的時候，就要定義：/job:worker/task:0 

"B_IP:1234"            # 第二臺機器的IP地址 /job:worker/task:1 

"C_IP:2222"            # 第三臺機器的IP地址 /job:worker/task:2 

], 

"ps": [ 

"D_IP:2222",           # 第四臺機器的IP地址 對應到程式碼塊：/job:ps/task:0 

]}) 

1

server=tf.train.Server(cluster,job_name='worker',task_index=0)  # 找到‘worker’名字下的，task0，也就是機器A 

1

2

3

4

5

6

7

8

9

10

with tf.device('/job:ps/task:0'):      # 引數定義在機器D上 

w=tf.get_variable('w',(2,2),tf.float32,initializer=tf.constant_initializer(2)) 

b=tf.get_variable('b',(2,2),tf.float32,initializer=tf.constant_initializer(5)) 

with tf.device('/job:worker/task:0/cpu:0'):     # 在機器A cpu上執行 

addwb=w+b 

with tf.device('/job:worker/task:1/cpu:0'):     # 在機器B cpu上執行 

mutwb=w*b 

with tf.device('/job:worker/task:2/cpu:0'):     # 在機器C cpu上執行 

divwb=w/b 

1

with tf.device(tf.train.replica_device_setter(worker_device='/job:worker/task:indexi',cluster=cluster)):

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

# 上面是因為worker計算內容各不相同，不過再深度學習中，一般每個worker的計算內容是一樣的， 

# 因為都是計算神經網路的每個batch的前向傳導，所以一般程式碼是重用的 

import  tensorflow as tf 

# 現在假設我們有A、B臺機器，首先需要在各臺機器上寫一份程式碼，並跑起來，各機器上的程式碼內容大部分相同 

# ，除了開始定義的時候，需要各自指定該臺機器的task之外。以機器A為例子，A機器上的程式碼如下： 

cluster=tf.train.ClusterSpec({ 

"worker": [ 

"192.168.11.105:1234",  # 格式 IP地址：埠號，在程式碼中需要用這臺機器計算的時候，就要定義：/job:worker/task:0 

], 

"ps": [ 

"192.168.11.130:2223"   # 第四臺機器的IP地址 對應到程式碼塊：/job:ps/task:0 

]}) 

# 不同的機器，下面這一行程式碼各不相同，server可以根據job_name、task_index兩個引數，查詢到叢集cluster中對應的機器 

isps=False 

if isps: 

server=tf.train.Server(cluster,job_name='ps',task_index=0)  # 找到‘ps’名字下的，task0，也就是機器A

server.join() 

else: 

server=tf.train.Server(cluster,job_name='worker',task_index=0)  # 找到‘worker’名字下的，task0，也就是機器B

with tf.device(tf.train.replica_device_setter(worker_device='/job:worker/task:0',cluster=cluster)): 

w=tf.get_variable('w',(2,2),tf.float32,initializer=tf.constant_initializer(2)) 

b=tf.get_variable('b',(2,2),tf.float32,initializer=tf.constant_initializer(5)) 

addwb=w+b 

mutwb=w*b 

divwb=w/b 

saver = tf.train.Saver() 

summary_op = tf.merge_all_summaries() 

init_op = tf.initialize_all_variables() 

sv = tf.train.Supervisor(init_op=init_op, summary_op=summary_op, saver=saver) 

with sv.managed_session(server.target) as sess: 

while 1: 

print sess.run([addwb,mutwb,divwb]) 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

# 上面是因為worker計算內容各不相同，不過再深度學習中，一般每個worker的計算內容是一樣的， 

# 因為都是計算神經網路的每個batch前向傳導，所以一般程式碼是重用的 

#coding=utf-8 

#多臺機器，每臺機器有一個顯示卡、或者多個顯示卡，這種訓練叫做分散式訓練 

import  tensorflow as tf 

# 現在假設我們有A、B、C、D四臺機器，首先需要在各臺機器上寫一份程式碼，並跑起來，各機器上的程式碼內容大部分相同 

# ，除了開始定義的時候，需要各自指定該臺機器的task之外。以機器A為例子，A機器上的程式碼如下： 

cluster=tf.train.ClusterSpec({ 

"worker": [ 

"192.168.11.105:1234",  # 格式 IP地址：埠號，在程式碼中需要用這臺機器計算的時候，就要定義：/job:worker/task:0 

], 

"ps": [ 

"192.168.11.130:2223"   # 第四臺機器的IP地址 對應到程式碼塊：/job:ps/task:0 

]}) 

# 不同的機器，下面這一行程式碼各不相同，server可以根據job_name、task_index兩個引數，查詢到叢集cluster中對應的機器 

isps=True 

if isps: 

server=tf.train.Server(cluster,job_name='ps',task_index=0)  # 找到‘ps’名字下的，task0，也就是機器A

server.join() 

else: 

server=tf.train.Server(cluster,job_name='worker',task_index=0)  # 找到‘worker’名字下的，task0，也就是機器B

with tf.device(tf.train.replica_device_setter(worker_device='/job:worker/task:0',cluster=cluster)): 

w=tf.get_variable('w',(2,2),tf.float32,initializer=tf.constant_initializer(2)) 

b=tf.get_variable('b',(2,2),tf.float32,initializer=tf.constant_initializer(5)) 

addwb=w+b 

mutwb=w*b 

divwb=w/b 

saver = tf.train.Saver() 

summary_op = tf.merge_all_summaries() 

init_op = tf.initialize_all_variables() 

sv = tf.train.Supervisor(init_op=init_op, summary_op=summary_op, saver=saver) 

with sv.managed_session(server.target) as sess: 

while 1: 

print sess.run([addwb,mutwb,divwb]) 

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

# 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 math

import sys

import tempfile

import time

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", 200,

"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):

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)

# 解析叢集引數

# ps作業的ip埠，可以有多個ps作業，之間用逗號分割

ps_spec = FLAGS.ps_hosts.split(",")

# worker作業的ip埠，可以有多個worker作業，之間用逗號分割

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.

# 任務內部伺服器（ps或者worker的上層抽象）的抽象物件，接收叢集物件，接收當前任務資訊

server = tf.train.Server(

cluster, job_name=FLAGS.job_name, task_index=FLAGS.task_index)

# 如果本次程式運行於ps任務，則啟動監聽（join持續等待，不會返回）

if FLAGS.job_name == "ps":

server.join()

# 選擇本任務所處GPU

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

# 裝置設定器將自動將變數OPS放置在單獨的引數伺服器（PS）上。不可變的OPS將放在worker身上。

with tf.device(

# 裝置放置器，可以返回被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")

# 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)))

opt = tf.train.AdamOptimizer(FLAGS.learning_rate)

# 如果使用同步訓練機制

if FLAGS.sync_replicas:

# 如果並行副本數(期望)沒有指定

if FLAGS.replicas_to_aggregate is None:

# 勒令並行數等於worker數

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")

# 優化器或者同步優化器單步優化節點（注意此句在if外）

train_step = opt.minimize(cross_entropy, global_step=global_step)

# 如果是worker，則編號0的worker設定為chief worker

is_chief = (FLAGS.task_index == 0)

# 同步訓練機制下的初始化操作

if FLAGS.sync_replicas:

# local_step初始化（chief_worker會改寫此句，所以實際上本句針對非chief_worker）

local_init_op = opt.local_step_init_op

if is_chief:

# chief_worker使用的時global_step，也需要初始化

local_init_op = opt.chief_init_op

# 為未初始化的Variable初始化

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()

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)

# 配置分散式會話

#    沒有可用GPU時使用CPU

#    不列印裝置放置資訊

#    過濾未繫結在ps或者worker的操作

sess_config = tf.ConfigProto(

allow_soft_placement=True,

log_device_placement=False,

device_filters=["/job:ps",

"/job:worker/task:%d" % FLAGS.task_index])

# chief會初始化所有worker的會話，否則等待chief返回會話

# 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)

# 同步更新模式的chief worker

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

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()