Skip to main content

Cuda Pipeline 同步机制

pipeline

它实现上是一个proxy pattern, cuda::pipeline是每个thread访问pipeline_shared_state的proxy

  • pipeline_shared_state的模板参数也仅仅是描述pipeline会被共享的范围,和barrier类似.
  • pipeline_shared_state需要在共享的内存区域创建
  • thread_scope的pipeline是性能最优秀的, 它不使用任何共享资源, 用cuda::pipeline<cuda::thread_scope_thread> pipeline = cuda::make_pipeline()直接创建
  • make_pipeline是一个同步操作,它用于初始化pipeline_shared_state,为当前线程确定role, 并通过线程间通信确定group内producer/consumer的数量
  • pipeline逻辑上是一个fifo, head in, tail out, 这个pipeline的元素称为stage
  • pipeline这个proxy有三种可能的角色, consumer, producer, both
  • fifo的最大容量是编译期创建pipeline_shared_state时指定的,当fifo中的stage满时,后续的producer将在acquire时被阻塞.
  • 对于role为producer的pipeline:
    • pipeline.producer_acquire(); 表明当前thread开始尝试向fifo中push数据,并需要lock相关的资源
    • 当producer_acquire成功后,当前thread的后续async指令都可以发射到acquire获得的stage中
    • pipeline.producer_commit();提交当前thread的async任务
    • 当group内所有的producer都commit后, 对应的stage将正式被push到fifo中,这个stage将会在内部所有async任务完成后被标记为ready
  •  对于role为consumer的pipeline
    • pipeline.consumer_wait(); 表明当前thread需要从fifo中取数据,并需要锁定相关资源
    • 当consumer_wait()成功后,说明group中的producer提交的一个stage已经ready,可以开始处理数据
    • pipeline.consumer_release(); 表明当前thread的任务已经完成.
    • 当group内所有的consumer都release后,这个stage将被pop出fifo
  • 注意: stage只有执行顺序的逻辑意义, fifo tail对应的是哪些数据搬运或者计算需要由用户自行track
// 申请一个pipeline，同步API，会自动记录producer和consumer的数量
cuda::pipeline pipeline = cuda::make_pipeline(block, &shared_state, thread_role);
if (thread_role == cuda::pipeline_role::producer) {
  // Only the producer threads schedule asynchronous memcpys:
  pipeline.producer_acquire();
  size_t shared_idx = fetch_batch % stages_count;
  size_t batch_idx = fetch_batch;
  size_t global_batch_idx = block_batch(batch_idx) + thread_idx;
  size_t shared_batch_idx = shared_offset[shared_idx] + thread_idx;
  cuda::memcpy_async(shared + shared_batch_idx, global_in + global_batch_idx, sizeof(int), pipeline);
  // 同步接口？
  pipeline.producer_commit();
}
if (thread_role == cuda::pipeline_role::consumer) {
  // Only the consumer threads compute:
  // 同步接口？
  pipeline.consumer_wait();
  size_t shared_idx = compute_batch % stages_count;
  size_t global_batch_idx = block_batch(compute_batch) + thread_idx;
  size_t shared_batch_idx = shared_offset[shared_idx] + thread_idx;
  compute(global_out + global_batch_idx, *(shared + shared_batch_idx));
  pipeline.consumer_release();
}
Back to top