 
   std::vector<std::thread> ths;
   for (size_t i = 0; i < gpu_list.size(); ++i) {
-    std::thread th(DeviceProgram<float>, gpu_list[i], *op2,
-                   &g_scope.NewScope());
+    std::thread th(&NCCLTester::PerThreadProgram<float>, this, gpu_list[i],
+                   *op2.get(), &g_scope.NewScope());
     ths.emplace_back(std::move(th));
   }
 
   for (size_t i = 0; i < gpu_list.size(); ++i) {
     ths[i].join();
   }
-  g_scope->reset(nullptr);
 }
 
 // ncclReduceOp with desc
 TEST(NCCL, ncclReduceOp) {
-  std::unique_ptr<p::DeviceContext> ctx(new p::CPUDeviceContext(p::CPUPlace()));
-  std::unique_ptr<f::Scope> g_scope(new Scope);
-
-  std::unique_ptr<f::OpDescBind> op1(new f::OpDescBind);
-  op1->SetType("ncclInit");
-  op1->SetOutput("Communicator", {"comm"});
-  op1->SetAttr("gpus", {gpu_list});
-
-  auto *var = g_scope.Var("comm");
-  var->GetMutable<p::Communicator>();
-
-  auto op = f::OpRegistry::CreateOp(*op1);
-  VLOG(1) << "invoke NCCLInitOp.";
-  op->Run(g_scope, *ctx);
-  VLOG(1) << "NCCLInitOp finished.";
-  delete ctx;
-
   std::unique_ptr<f::OpDescBind> op2(new f::OpDescBind);
   op2->SetType("ncclReduce");
   op2->SetInput("X", {"st"});


  std::vector<std::thread> ths;
  for (size_t i = 0; i < gpu_list.size(); ++i) {
    std::thread th(DeviceProgram<float>, gpu_list[i], *op2,
                   &g_scope.NewScope());
    ths.emplace_back(std::move(th));
  }

  for (size_t i = 0; i < gpu_list.size(); ++i) {
    ths[i].join();
  }
  g_scope->reset(nullptr);
}

// ncclBcastOp with desc
TEST(NCCL, ncclBcastOp) {
  f::ProgramDescBind program;
  f::BlockDescBind *block = program.Block(0);
  f::OpDescBind *op1 = block->AppendOp();

  p::DeviceContext *ctx = new p::CPUDeviceContext(p::CPUPlace());

  op1->SetType("ncclInit");
  op1->SetOutput("Communicator", {"comm"});
  op1->SetAttr("gpus", {gpu_list});

  auto *var = g_scope.Var("comm");
  var->GetMutable<p::Communicator>();

  auto op = f::OpRegistry::CreateOp(*op1);
  VLOG(1) << "invoke NCCLInitOp.";
  op->Run(g_scope, *ctx);
  VLOG(1) << "NCCLInitOp finished.";

  f::OpDescBind *op2 = new f::OpDescBind;
  op2->SetType("ncclBcastSend");
  op2->SetInput("X", {"st"});
  op2->SetInput("Communicator", {"comm"});
  op2->SetOutput("Out", {"rt"});

  std::vector<std::thread> ths;
  for (size_t i = 0; i < gpu_list.size(); ++i) {
    std::thread th(DeviceProgram<float>, gpu_list[i], *op2);
    ths.emplace_back(std::move(th));
  }

  for (size_t i = 0; i < gpu_list.size(); ++i) {
    ths[i].join();
  }
}

int main(int argc, char **argv) {
  const int dev_count = p::GetCUDADeviceCount();


  // device context should be release before scope.
  // otherwise driver will down.
  return RUN_ALL_TESTS();
}
