PipelineOptimizer

  • class paddle.fluid.optimizer.PipelineOptimizer(optimizer, cut_list=None, place_list=None, concurrency_list=None, queue_size=30, sync_steps=1, start_cpu_core_id=0)[源代码]

使用流水线模式进行训练。 Program会根据切分列表cut_list进行分割。如果cut_list的长度是k,则整个program(包括反向部分)将被分割为2k-1个section。 所以place_list和concurrency_list的长度也必须是2k-1。

注解

虽然我们在流水线训练模式中采用异步更新的方式来加速,但最终的效果会依赖于每条流水线的训练进程。我们将在未来尝试同步模式。

  • 参数:
    • optimizer (Optimizer) - 基础优化器,如SGD
    • cut_list (list of Variable list) - main_program的cut变量列表
    • place_list (list of Place) - 对应section运行所在的place
    • concurrency_list (list of int) - 指定每个section的并发度列表
    • queue_size (int) - 每个section都会消费其输入队列(in-scope queue)中的scope,并向输出队列(out-scope queue)产出scope。 此参数的作用就是指定队列的大小。 可选,默认值:30
    • sync_steps (int) - 不同显卡之间的同步周期数。可选,默认值:1
    • start_cpu_core_id (int) - 指定所使用的第一个CPU核的id。可选,默认值:0

代码示例

  1. import paddle.fluid as fluid
  2. import paddle.fluid.layers as layers
  3. x = fluid.layers.data(name='x', shape=[1], dtype='int64', lod_level=0)
  4. y = fluid.layers.data(name='y', shape=[1], dtype='int64', lod_level=0)
  5. emb_x = layers.embedding(input=x, param_attr=fluid.ParamAttr(name="embx"), size=[10,2], is_sparse=False)
  6. emb_y = layers.embedding(input=y, param_attr=fluid.ParamAttr(name="emby",learning_rate=0.9), size=[10,2], is_sparse=False)
  7. concat = layers.concat([emb_x, emb_y], axis=1)
  8. fc = layers.fc(input=concat, name="fc", size=1, num_flatten_dims=1, bias_attr=False)
  9. loss = layers.reduce_mean(fc)
  10. optimizer = fluid.optimizer.SGD(learning_rate=0.5)
  11. optimizer = fluid.optimizer.PipelineOptimizer(optimizer,
  12. cut_list=[[emb_x, emb_y], [loss]],
  13. place_list=[fluid.CPUPlace(), fluid.CUDAPlace(0), fluid.CPUPlace()],
  14. concurrency_list=[1, 1, 4],
  15. queue_size=2,
  16. sync_steps=1,
  17. )
  18. optimizer.minimize(loss)
  19. place = fluid.CPUPlace()
  20. exe = fluid.Executor(place)
  21. exe.run(fluid.default_startup_program())
  22. filelist = [] # you should set your own filelist, e.g. filelist = ["dataA.txt"]
  23. dataset = fluid.DatasetFactory().create_dataset("FileInstantDataset")
  24. dataset.set_use_var([x,y])
  25. dataset.set_batch_size(batch_size)
  26. dataset.set_filelist(filelist)
  27. exe.train_from_dataset(
  28. fluid.default_main_program(),
  29. dataset,
  30. thread=2,
  31. debug=False,
  32. fetch_list=[],
  33. fetch_info=[],
  34. print_period=1)