我正在尝试在两台机器上分配训练和评估:
- PC1:在 GPU 和参数服务器上训练
- PC2:仅评估，只要有新的检查点可用
为此，我正在尝试调整 tf.contrib.learn.Experiment 框架，但我似乎无法获得正确的集群规范。
这是我的代码的精简版:
def get_schedule(run_config):
if run_config.task_type == 'ps':
return 'run_std_server'
if run_config.task_type == 'worker':
return 'train'
if run_config.task_type == 'evaluator':
return 'continuous_eval'
if run_config.task_type == 'master':
return 'train'
raise ValueError('Unknown task type "{}"'.format(run_config.task_type))
def deeplpr_model_fn(features, labels, mode, cluster_spec={}):
with tf.device(tf.train.replica_device_setter(cluster=cluster_spec)):
logits = build_model(features['images'], mode)
#[...] Standard Estimator setup for training & evaluation
return tf.estimator.EstimatorSpec(mode=mode,
predictions=predictions,
export_outputs=export_outputs,
loss=loss,
train_op=train_op,
eval_metric_ops=metrics)
def get_experiment(run_config=None, hparams=None):
# Create the Estimator
estimator = tf.estimator.Estimator(
model_fn=lambda features, labels, mode : my_model_fn(features, labels, mode, run_config.cluster_spec),
model_dir=FLAGS.model_dir,
config=run_config)
# Set up input functions for training and evaluation
train_input_fn = lambda : input_fn(tf.estimator.ModeKeys.TRAIN, FLAGS.batch_size)
eval_input_fn = lambda : input_fn(tf.estimator.ModeKeys.EVAL, FLAGS.batch_size)
# Set up the experiment
experiment = tf.contrib.learn.Experiment(
estimator=estimator,
train_input_fn=train_input_fn,
eval_input_fn=eval_input_fn,
train_steps=FLAGS.steps,
eval_steps=None,
eval_delay_secs=20, # time to wait before running the first evaluation
train_steps_per_iteration=2000)
return experiment
主要功能如下:
def distributed_main(unused_argv):
import json
# Set up environment variables according to the parameters passed to the process
TF_CONFIG = {
'cluster': {
"worker": [
"pc1:2222",
],
"ps": [
"pc1:2223",
],
"evaluator": [
"pc2:2224",
]
},
'environment': 'cluster',
'task': {
'type': unused_argv[1].strip(),
'index': unused_argv[2].strip() if len(unused_argv) > 2 else 0
}
}
os.environ['TF_CONFIG'] = json.dumps(TF_CONFIG)
session_config = tf.ConfigProto(device_filters=device_filters,
allow_soft_placement=True)
config = tf.contrib.learn.RunConfig(model_dir=FLAGS.model_dir,
session_config=session_config)
schedule = get_schedule(config)
tf.logging.info('Beginning task {}:{}'.format(config.task_type, config.task_id))
# Run the function
tf.contrib.learn.learn_runner.run(get_experiment, schedule=schedule, run_config=config)
其中 unused_argv 包含作业名称和可选的索引(默认为 0)。
使用适当的作业名称和任务 ID 运行三个进程，我无法让工作人员通过 session 初始化步骤，因为它期望 evaluator 与首席工作人员通信(它没有t，如果 continuous_eval 被调用，显然)。
研究这个问题，我发现 this answer他们建议在其中添加 device_filter，所以我尝试添加:
device_filters=["/job:ps", "/job:worker"]
if unused_argv[1] != 'worker':
device_filters += ['/job:evaluator']
session_config = tf.ConfigProto(device_filters=device_filters)
config = tf.contrib.learn.RunConfig(model_dir=FLAGS.model_dir, session_config=session_config)
这有效地解锁了 worker 和 ps，但是当试图恢复最新的检查点时评估器会崩溃:
Traceback (most recent call last):
File "deeplpr.py", line 431, in <module>
tf.app.run(main=distributed_main, argv=[sys.argv[0]] + unparsed)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\platform\app.py", line 126, in run
_sys.exit(main(argv))
File "deeplpr.py", line 420, in distributed_main
tf.contrib.learn.learn_runner.run(get_experiment, schedule=schedule, run_config=config)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\contrib\learn\python\learn\learn_runner.py", line 218, in run
return _execute_schedule(experiment, schedule)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\contrib\learn\python\learn\learn_runner.py", line 46, in _execute_schedule
return task()
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\contrib\learn\python\learn\experiment.py", line 573, in continuous_eval
continuous_eval_predicate_fn=continuous_eval_predicate_fn)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\contrib\learn\python\learn\experiment.py", line 533, in _continuous_eval
hooks=self._eval_hooks)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\contrib\learn\python\learn\experiment.py", line 894, in _call_evaluate
hooks=hooks)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\estimator\estimator.py", line 414, in evaluate
name=name)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\estimator\estimator.py", line 949, in _evaluate_model
config=self._session_config)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\evaluation.py", line 209, in _evaluate_once
session_creator=session_creator, hooks=hooks) as session:
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\monitored_session.py", line 795, in __init__
stop_grace_period_secs=stop_grace_period_secs)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\monitored_session.py", line 518, in __init__
self._sess = _RecoverableSession(self._coordinated_creator)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\monitored_session.py", line 981, in __init__
_WrappedSession.__init__(self, self._create_session())
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\monitored_session.py", line 986, in _create_session
return self._sess_creator.create_session()
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\monitored_session.py", line 675, in create_session
self.tf_sess = self._session_creator.create_session()
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\monitored_session.py", line 446, in create_session
init_fn=self._scaffold.init_fn)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\session_manager.py", line 275, in prepare_session
config=config)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\session_manager.py", line 191, in _restore_checkpoint
saver.restore(sess, checkpoint_filename_with_path)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\training\saver.py", line 1760, in restore
{self.saver_def.filename_tensor_name: save_path})
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\client\session.py", line 905, in run
run_metadata_ptr)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\client\session.py", line 1137, in _run
feed_dict_tensor, options, run_metadata)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\client\session.py", line 1355, in _do_run
options, run_metadata)
File "C:\Users\1\Anaconda3\envs\tensorflow\lib\site-packages\tensorflow\python\client\session.py", line 1374, in _do_call
raise type(e)(node_def, op, message)
tensorflow.python.framework.errors_impl.InvalidArgumentError: Cannot assign a device for operation 'save/RestoreV2_1': Operation was explicitly assigned to /job:ps/task:0/device:CPU:0 but available devices are [ /job:localhost/replica:0/task:0/device:CPU:0 ]. Make sure the device specification refers to a valid device.
[[Node: save/RestoreV2_1 = RestoreV2[dtypes=[DT_FLOAT, DT_FLOAT, DT_FLOAT, DT_FLOAT, DT_FLOAT, DT_FLOAT, DT_FLOAT, DT_FLOAT], _device="/job:ps/task:0/device:CPU:0"](save/Const, save/RestoreV2_1/tensor_names, save/RestoreV2_1/shape_and_slices)]]
指定 worker 仅进行评估的正确方法是什么？ 在 Tensorflow 的日志中，我看到使用的 RunConfig 有一个参数 '_evaluation_master': ''，但我找不到任何关于它的文档。这有什么关系吗？是否有任何工作示例说明如何分配训练和评估分开的实验？
更新:
按照建议，我在定义 session_config 时添加了 log_device_placement=True。 然而，日志输出似乎在记录设备放置之前崩溃了:
INFO:tensorflow:Waiting 20.000000 secs before starting eval.
INFO:tensorflow:Calling model_fn.
INFO:tensorflow:Done calling model_fn.
INFO:tensorflow:Starting evaluation at 2018-02-19-15:35:04
INFO:tensorflow:Graph was finalized.
2018-02-19 16:35:04.888165: I C:\tf_jenkins\workspace\rel-win\M\windows-gpu\PY\35\tensorflow\core\platform\cpu_feature_guard.cc:140] Your CPU supports instructions that this TensorFlow binary was not compiled to use: AVX2
INFO:tensorflow:Restoring parameters from models\test_distributed\model.ckpt-2373
# Here starts the traceback of the error, same as above
这有点令人困惑，我不应该在 session 创建时看到放置日志吗？ restore op 不需要 session 来运行吗？
设置 allow_device_placement=True 也没有改变日志和错误中的任何内容。
更新 2:
为 所有 机器设置 log_device_placement=True 仅将其记录在 worker:0(又名主机)中，我认为这是预期的行为
编辑:
更新了上面的代码以反射(reflect)我如何设置 allow_device_placement=True(仅更改了主要功能)。
请您参考如下方法:
在 PC2 上，将 allow_soft_placement=True 添加到 session_config。
session_config = tf.ConfigProto(allow_soft_placement=True)
阅读this section有关此参数的更多信息。
If you would like TensorFlow to automatically choose an existing and supported device to run the operations in case the specified one doesn't exist, you can set allow_soft_placement to True in the configuration option when creating the session
更新
仔细查看错误日志后，我发现错误是在这一行触发的。
tf.contrib.learn.learn_runner.run(
get_experiment,
schedule=schedule,
run_config=config
)
因此，allow_soft_placement=True 应该在其 run_config 中设置参数，如下所示。
config = tf.contrib.learn.RunConfig(
model_dir=FLAGS.model_dir,
session_config=tf.ConfigProto(allow_soft_placement=True)
)
