Add RCCL primitive testing (#70)

[ROCm/rccl commit: 1bb6d2104c]

projects/rccl/tools/rccl-prim-test/Makefile

@@ -0,0 +1,16 @@
+HIP_PATH?= $(wildcard /opt/rocm/hip)
+ifeq (,$(HIP_PATH))
+	HIP_PATH=../../..
+endif
+HIPCC=$(HIP_PATH)/bin/hipcc
+
+EXE=rccl_prim_test
+CXXFLAGS = -O3 -g -I/opt/rocm/rocrand/include
+
+all: $(EXE)
+
+$(EXE): rccl_prim_test.cpp
+	$(HIPCC) $(CXXFLAGS) $^ -o $@
+
+clean:
+	rm -f *.o $(EXE)

projects/rccl/tools/rccl-prim-test/copy_kernel.h

@@ -0,0 +1,255 @@
+/*************************************************************************
+ * Copyright (c) 2015, NVIDIA CORPORATION. All rights reserved.
+ *
+ * See LICENSE.txt for license information
+ ************************************************************************/
+
+
+#ifndef COPY_KERNEL_H_
+#define COPY_KERNEL_H_
+#include <cstdio>
+#include <cstdint>
+
+// Define min for ssize_t
+static __device__ int min(int a, ssize_t b) { return (a < b) ? a : b; }
+
+typedef uint64_t PackType;
+
+template<class FUNC, typename T>
+struct MULTI {
+    __device__ PackType operator()(const PackType x, const PackType y) const
+    {
+        return FUNC()(x, y);
+    }
+};
+
+#define ALIGNUP(x, a)   ((((x)-1) & ~((a)-1)) + (a))
+
+template<typename T>
+__device__ inline volatile T* AlignUp(volatile T * ptr, size_t align) {
+  size_t ptrval = reinterpret_cast<size_t>(ptr);
+  return reinterpret_cast<volatile T*>(ALIGNUP(ptrval, align));
+}
+
+template<typename T> inline __device__
+T vFetch(const volatile T* ptr) {
+  return *ptr;
+}
+
+template<typename T> inline __device__
+void vStore(volatile T* ptr, const T val) {
+  *ptr = val;
+}
+
+template<class FUNC, typename T, bool TWO_INPUTS, bool TWO_OUTPUTS>
+__attribute__((noinline))
+__device__ inline void ReduceCopy(
+    const int tid, const int nthreads,
+    const volatile T * __restrict__ const src0,
+    const volatile T * __restrict__ const src1,
+    volatile T * __restrict__ const dest0,
+    volatile T * __restrict__ const dest1, const int N) {
+  for (int idx = tid; idx < N; idx += nthreads) {
+    T val = vFetch(src0+idx);
+    if (TWO_INPUTS) {
+      val = FUNC()(val, vFetch(src1+idx));
+    }
+    vStore(dest0+idx, val);
+    if (TWO_OUTPUTS) {
+      vStore(dest1+idx, val);
+    }
+  }
+}
+
+typedef ulong2 Pack128;
+
+template<class FUNC, typename T>
+struct MULTI128 {
+  __device__ void operator()(Pack128& x, Pack128& y) {
+    x.x = MULTI<FUNC, T>()(x.x, y.x);
+    x.y = MULTI<FUNC, T>()(x.y, y.y);
+  }
+};
+
+inline __device__ void Fetch128(Pack128& v, Pack128* p) {
+  v.x = p->x;
+  v.y = p->y;
+}
+
+inline __device__ void Store128(Pack128* p, Pack128& v) {
+  p->x = v.x;
+  p->y = v.y;
+}
+
+#define WARP_SIZE 32
+template<class FUNC, typename T, bool TWO_INPUTS, bool TWO_OUTPUTS, int UNROLL>
+__attribute__((noinline))
+__device__ inline void ReduceCopy128b( const int w, const int nw, const int t,
+    Pack128 * src0, Pack128 * src1, Pack128 * dest0, Pack128 * dest1,
+    const int N) {
+  Pack128 t0[UNROLL];
+  Pack128 t1[UNROLL];
+  const Pack128* src0_end = src0 + N;
+  const int inc = nw * UNROLL * WARP_SIZE;
+  const int offset = w * UNROLL * WARP_SIZE + t;
+  src0 += offset;  if (TWO_INPUTS)  src1 += offset;
+  dest0 += offset; if (TWO_OUTPUTS) dest1 += offset;
+
+  while (src0 < src0_end) {
+#pragma unroll
+    for (int u = 0; u < UNROLL; ++u) {
+      Fetch128(t0[u], src0+u*WARP_SIZE);
+      if (TWO_INPUTS) Fetch128(t1[u], src1+u*WARP_SIZE);
+    }
+#pragma unroll
+    for (int u = 0; u < UNROLL; ++u) {
+      if (TWO_INPUTS) MULTI128<FUNC, T>()(t0[u], t1[u]);
+      Store128(dest0+u*WARP_SIZE, t0[u]);
+      if (TWO_OUTPUTS) Store128(dest1+u*WARP_SIZE, t0[u]);
+    }
+    src0 += inc;  if (TWO_INPUTS)  src1 += inc;
+    dest0 += inc; if (TWO_OUTPUTS) dest1 += inc;
+  }
+}
+
+template<int UNROLL, class FUNC, typename T, bool HAS_DEST1, bool HAS_SRC1>
+__attribute__((noinline))
+__device__ inline void ReduceOrCopy(const int tid, const int nthreads,
+    volatile T * __restrict__ dest0, volatile T * __restrict__ dest1,
+    const volatile T * __restrict__ src0, const volatile T * __restrict__ src1,
+    int N) {
+  int Nrem = N;
+  if (Nrem <= 0) return;
+
+  int Npreamble = (Nrem<alignof(Pack128)) ? Nrem : AlignUp(dest0, alignof(Pack128)) - dest0;
+
+  // stage 0: check if we'll be able to use the fast, 128-bit aligned path.
+  // If not, we'll just use the slow preamble path for the whole operation
+  bool alignable = (((AlignUp(src0,  alignof(Pack128)) == src0  + Npreamble)) &&
+          (!HAS_DEST1 || (AlignUp(dest1, alignof(Pack128)) == dest1 + Npreamble)) &&
+          (!HAS_SRC1  || (AlignUp(src1,  alignof(Pack128)) == src1  + Npreamble)));
+
+  if (!alignable) {
+    Npreamble = Nrem;
+  }
+
+  // stage 1: preamble: handle any elements up to the point of everything coming
+  // into alignment
+  ReduceCopy<FUNC, T, HAS_SRC1, HAS_DEST1>(tid, nthreads, src0, src1, dest0, dest1, Npreamble);
+
+  Nrem -= Npreamble;
+  if (Nrem == 0) return;
+
+  dest0 += Npreamble; if (HAS_DEST1) { dest1 += Npreamble; }
+  src0  += Npreamble; if (HAS_SRC1)  { src1  += Npreamble; }
+
+  // stage 2: fast path: use 128b loads/stores to do the bulk of the work,
+  // assuming the pointers we have are all 128-bit alignable.
+  int w = tid / WARP_SIZE;       // Warp number
+  int nw = nthreads / WARP_SIZE; // Number of warps
+  int t = tid % WARP_SIZE;       // Thread (inside the warp)
+
+  const int PackFactor = sizeof(Pack128) / sizeof(T);
+
+  // stage 2a: main loop
+  int Nalign2a = (Nrem / (PackFactor * UNROLL * nthreads))
+      * (UNROLL * nthreads); // round down
+
+  ReduceCopy128b<FUNC, T, HAS_SRC1, HAS_DEST1, UNROLL>(w, nw, t, (Pack128*)src0, (Pack128*)src1, (Pack128*)dest0, (Pack128*)dest1, Nalign2a);
+
+  int Ndone2a = Nalign2a * PackFactor;
+  Nrem -= Ndone2a;
+  if (Nrem == 0) return;
+  dest0 += Ndone2a; if (HAS_DEST1) { dest1 += Ndone2a; }
+  src0  += Ndone2a; if (HAS_SRC1)  { src1  += Ndone2a; }
+
+  // stage 2b: slightly less optimized for section when we don't have full
+  // UNROLLs
+
+  int Nalign2b = Nrem / PackFactor;
+
+  ReduceCopy128b<FUNC, T, HAS_SRC1, HAS_DEST1, 1>(w, nw, t, (Pack128*)src0, (Pack128*)src1, (Pack128*)dest0, (Pack128*)dest1, Nalign2b);
+
+  int Ndone2b = Nalign2b * PackFactor;
+  Nrem -= Ndone2b;
+  if (Nrem == 0) return;
+  dest0 += Ndone2b; if (HAS_DEST1) { dest1 += Ndone2b; }
+  src0  += Ndone2b; if (HAS_SRC1)  { src1  += Ndone2b; }
+
+  // stage 2c: tail
+  ReduceCopy<FUNC, T, HAS_SRC1, HAS_DEST1>(tid, nthreads, src0, src1, dest0, dest1, Nrem);
+}
+
+template<typename T>
+struct FuncPassA {
+  __device__ T operator()(const T x, const T y) const {
+    return x;
+  }
+};
+
+template<typename T>
+struct FuncSum {
+  __device__ T operator()(const T x, const T y) const {
+    return x + y;
+  }
+};
+
+template<class FUNC>
+struct MULTI<FUNC, float> {
+  static_assert(sizeof(PackType) == 2 * sizeof(float),
+      "PackType must be twice the size of float.");
+  union converter {
+    PackType storage;
+    struct {
+      float a, b;
+    };
+  };
+
+  __device__ PackType operator()(const PackType x, const PackType y) const {
+    converter cx, cy, cr;
+    cx.storage = x;
+    cy.storage = y;
+
+    cr.a = FUNC()(cx.a, cy.a);
+    cr.b = FUNC()(cx.b, cy.b);
+
+    return cr.storage;
+  }
+};
+
+// Assumptions:
+// - there is exactly 1 block
+// - THREADS is the number of producer threads
+// - this function is called by all producer threads
+template<int UNROLL, int THREADS, typename T>
+__device__ void Copy(volatile T * __restrict__ const dest,
+    const volatile T * __restrict__ const src, const int N) {
+  ReduceOrCopy<UNROLL, FuncPassA<T>, T, false, false>(threadIdx.x, THREADS,
+      dest, nullptr, src, nullptr, N);
+}
+
+template<int UNROLL, int THREADS, typename T>
+__device__ void DoubleCopy(volatile T * __restrict__ const dest0,
+    volatile T * __restrict__ const dest1,
+    const volatile T * __restrict__ const src, const int N) {
+  ReduceOrCopy<UNROLL, FuncPassA<T>, T, true, false>(threadIdx.x, THREADS,
+      dest0, dest1, src, nullptr, N);
+}
+
+template<int UNROLL, int THREADS, typename T>
+__device__ void Reduce(volatile T * __restrict__ const dest,
+    const volatile T * __restrict__ const src0,
+    const volatile T * __restrict__ const src1, const int N) {
+  ReduceOrCopy<UNROLL, FuncSum<T>, T, false, true>(threadIdx.x, THREADS,
+      dest, nullptr, src0, src1, N);
+}
+
+template<int UNROLL, int THREADS, typename T>
+__device__ void ReduceCopy(volatile T * __restrict__ const dest0,
+    volatile T * __restrict__ const dest1,
+    const volatile T * __restrict__ const src0,
+    const volatile T * __restrict__ const src1, const int N) {
+  ReduceOrCopy<UNROLL, FuncSum<T>, T, true, true>(threadIdx.x, THREADS,
+      dest0, dest1, src0, src1, N);
+}
+#endif // COPY_KERNEL_H_

projects/rccl/tools/rccl-prim-test/rccl_prim_test.cpp

@@ -0,0 +1,362 @@
+/*
+Copyright (c) 2019 - present Advanced Micro Devices, Inc. All rights reserved.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+*/
+
+/**
+ * @file rccl_prim_test.cpp
+ *
+ * test performance if individual rccl primitives
+ */
+#include <cstdio>  //fprintf
+#include <iostream> //cerr
+#include <unistd.h> //usleep
+#include <cstring>
+#include <hip/hip_runtime_api.h>
+#include <hip/hip_runtime.h>
+#include "copy_kernel.h"
+
+#define MAX_WORKGROUPS 8
+#define THREADS 256
+#define UNROLL 8
+
+#define NUM_ITERS 10
+
+struct transfer_data_t {
+  float *dest0; //remote fine grain
+  float *src0;  //local fine grain
+  float *dest1; //local coarse grain
+  float *src1;  //local coarse grain
+  int N;
+  int gpu;
+};
+
+struct profiling_data_t {
+  uint64_t write_cycles;
+  uint64_t bytes_transferred;
+};
+
+
+#define LOAD(VAR) __atomic_load_n((VAR), __ATOMIC_SEQ_CST)
+#define STORE(DST, SRC) __atomic_store_n((DST), (SRC), __ATOMIC_SEQ_CST)
+
+enum Ops {
+  OP_COPY,
+  OP_LOCALCOPY,
+  OP_DOUBLECOPY,
+  OP_REDUCE,
+  OP_REDUCECOPY,
+  NUM_OPS,
+};
+
+template<int op>
+__global__ void flag_sync_kernel(struct transfer_data_t* transfer_data, struct profiling_data_t* profiling_data) {
+  size_t idx = threadIdx.x;
+  uint64_t curr_time, next_time;
+
+  if (idx == 0) {
+    curr_time = clock();
+  }
+  __syncthreads();
+
+  int offset = transfer_data->N * blockIdx.x / gridDim.x;
+  int n = transfer_data->N / gridDim.x;
+  if (op == OP_COPY) Copy<UNROLL, THREADS, float>(transfer_data->dest0 + offset, transfer_data->src0 + offset, n);
+  if (op == OP_LOCALCOPY) Copy<UNROLL, THREADS, float>(transfer_data->dest1 + offset, transfer_data->src0 + offset, n);
+  if (op == OP_DOUBLECOPY) DoubleCopy<UNROLL, THREADS, float>(transfer_data->dest0 + offset, transfer_data->dest1 + offset, transfer_data->src0 + offset, n);
+  if (op == OP_REDUCE) Reduce<UNROLL, THREADS, float>(transfer_data->dest0 + offset, transfer_data->src0 + offset, transfer_data->src1 + offset, n);
+  if (op == OP_REDUCECOPY) ReduceCopy<UNROLL, THREADS, float>(transfer_data->dest0 + offset, transfer_data->dest1 + offset, transfer_data->src0 + offset, transfer_data->src1 + offset, n);
+
+  if (idx == 0) {
+    next_time = clock();
+    __atomic_fetch_add(&(profiling_data->write_cycles), next_time - curr_time, __ATOMIC_SEQ_CST);
+    curr_time = next_time;
+    __atomic_fetch_add(&(profiling_data->bytes_transferred), n * sizeof(float), __ATOMIC_SEQ_CST);
+  }
+}
+
+typedef void(*flag_sync_kernel_t)(struct transfer_data_t* transfer_data, struct profiling_data_t* profiling_data);
+
+static flag_sync_kernel_t const flagSyncKerns[NUM_OPS] = {
+  flag_sync_kernel<OP_COPY>,
+  flag_sync_kernel<OP_LOCALCOPY>,
+  flag_sync_kernel<OP_DOUBLECOPY>,
+  flag_sync_kernel<OP_REDUCE>,
+  flag_sync_kernel<OP_REDUCECOPY>,
+};
+
+__global__ void initTestDataKernel(float* data, const size_t N, const int gpu) {
+  int tid = threadIdx.x + blockIdx.x * blockDim.x;
+  while (tid < N) {
+    data[tid] = 1.0/(float)(gpu*17 + tid%77);
+    tid += blockDim.x * gridDim.x;
+  }
+}
+
+#define HIPCHECK(cmd)                                                          \
+do {                                                                           \
+  hipError_t error = (cmd);                                                    \
+  if (error != hipSuccess)                                                     \
+  {                                                                            \
+    std::cerr << "Encountered HIP error (" << error << ") at line "            \
+              << __LINE__ << " in file " << __FILE__ << "\n";                  \
+    exit(-1);                                                                  \
+  }                                                                            \
+} while (0)
+
+static void setupPeers() {
+    int deviceCnt, dev;
+
+     HIPCHECK(hipGetDeviceCount(&deviceCnt));
+     HIPCHECK(hipGetDevice(&dev));
+    //! If gpus are not peer enabled, enable them
+    for (int i = 0; i < deviceCnt; i++) {
+         HIPCHECK(hipSetDevice(i));
+        for (int j = 0; j < deviceCnt; j++) {
+            if (i != j) {
+                HIPCHECK(hipDeviceEnablePeerAccess(j, 0));
+            }
+        }
+    }
+     HIPCHECK(hipSetDevice(dev));
+}
+
+char* getCmdOption(char ** begin, char ** end, const std::string & option) {
+    char ** itr = std::find(begin, end, option);
+    if (itr != end && ++itr != end)
+    {
+        return *itr;
+    }
+    return 0;
+}
+
+bool cmdOptionExists(char** begin, char** end, const std::string& option) {
+    return std::find(begin, end, option) != end;
+}
+
+int main(int argc,char* argv[])
+{
+  if (cmdOptionExists(argv, argv + argc, "-h")) {
+    printf("./rccl_prim_test -w num_workgroups -p copy|localcopy|doublecopy|reduce|reducecopy|all\n");
+    exit(0);
+  }
+
+  int workgroups = 1;
+  char *wg = getCmdOption(argv, argv + argc, "-w");
+  if (wg)
+    workgroups = atol(wg);
+  printf("Benchmarking using %d workgroups\n", workgroups);
+
+  const char *ops[] = {"copy", "localcopy", "doublecopy", "reduce", "reducecopy", "all"};
+  char *prim = getCmdOption(argv, argv + argc, "-p");
+  int op = 5, begin_op, end_op;
+  if (prim) {
+    for (op = 0; op < sizeof(ops); op++)
+      if (!strcmp((const char *)prim, ops[op]))
+        break;
+  }
+  if (op < NUM_OPS ) {
+    begin_op = op;
+    end_op = op + 1;
+  } else {
+    begin_op = 0;
+    end_op = NUM_OPS;
+    printf("Benchmarking all ops\n");
+  }
+
+  // Enable peer access
+  setupPeers();
+
+  // data buffers
+  float *buff_0, *buff_1, *buff_coarse_0, *buff_coarse_1;
+  struct transfer_data_t h_transfer_data_0, h_transfer_data_1, *transfer_data_0, *transfer_data_1;
+  struct profiling_data_t *profiling_data_0, *profiling_data_1, *d_profiling_data_0, *d_profiling_data_1;
+  uint64_t N = 2097152*4*MAX_WORKGROUPS;
+
+  HIPCHECK(hipSetDevice(0));
+  HIPCHECK(hipExtMallocWithFlags((void**) &transfer_data_0, sizeof(struct transfer_data_t), hipDeviceMallocFinegrained));
+  //printf("GPU 0: allocated fine grain VRAM at %llx\n", (unsigned long long)transfer_data_0);
+  HIPCHECK(hipExtMallocWithFlags((void**) &buff_0, 2*N*sizeof(float), hipDeviceMallocFinegrained));
+  //printf("GPU 0: allocated fine grain VRAM at %llx\n", (unsigned long long)buff_0);
+  HIPCHECK(hipMalloc((void**) &buff_coarse_0, 2*N*sizeof(float)));
+  //printf("GPU 0: allocated coarse grain VRAM at %llx\n", (unsigned long long)buff_coarse_0);
+  profiling_data_0 = (struct profiling_data_t *)malloc(sizeof(struct profiling_data_t));
+  HIPCHECK(hipMalloc((void**) &d_profiling_data_0, sizeof(struct profiling_data_t)));
+  //create stream
+  hipStream_t stream_0;
+  HIPCHECK(hipStreamCreate(&stream_0));
+  //randomize test data
+  hipLaunchKernelGGL(initTestDataKernel,
+      /*grid dim x,y,z*/        dim3(32, 1, 1),
+      /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+      /*dynamic shared mem*/    0,
+      /*stream*/                stream_0,
+      /*kernel args*/           buff_0, 2*N, 0);
+  hipLaunchKernelGGL(initTestDataKernel,
+      /*grid dim x,y,z*/        dim3(32, 1, 1),
+      /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+      /*dynamic shared mem*/    0,
+      /*stream*/                stream_0,
+      /*kernel args*/           buff_coarse_0, 2*N, 0);
+
+  HIPCHECK(hipSetDevice(1));
+  HIPCHECK(hipExtMallocWithFlags((void**) &transfer_data_1, sizeof(struct transfer_data_t), hipDeviceMallocFinegrained));
+  //printf("GPU 1: allocated fine grain VRAM at %llx\n", (unsigned long long)transfer_data_1);
+  HIPCHECK(hipExtMallocWithFlags((void**) &buff_1, 2*N*sizeof(float), hipDeviceMallocFinegrained));
+  //printf("GPU 1: allocated fine grain VRAM at %llx\n", (unsigned long long)buff_1);
+  HIPCHECK(hipMalloc((void**) &buff_coarse_1, 2*N*sizeof(float)));
+  //printf("GPU 1: allocated coarse grain VRAM at %llx\n", (unsigned long long)buff_coarse_1);
+  profiling_data_1 = (struct profiling_data_t *)malloc(sizeof(struct profiling_data_t));
+  HIPCHECK(hipMalloc((void**) &d_profiling_data_1, sizeof(struct profiling_data_t)));
+  //create stream
+  hipStream_t stream_1;
+  HIPCHECK(hipStreamCreate(&stream_1));
+  //randomize test data
+  hipLaunchKernelGGL(initTestDataKernel,
+      /*grid dim x,y,z*/        dim3(32, 1, 1),
+      /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+      /*dynamic shared mem*/    0,
+      /*stream*/                stream_1,
+      /*kernel args*/           buff_1, 2*N, 1);
+  hipLaunchKernelGGL(initTestDataKernel,
+      /*grid dim x,y,z*/        dim3(32, 1, 1),
+      /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+      /*dynamic shared mem*/    0,
+      /*stream*/                stream_1,
+      /*kernel args*/           buff_coarse_1, 2*N, 1);
+
+  h_transfer_data_0.dest0 = buff_1;
+  h_transfer_data_0.dest1 = buff_coarse_0 + N;
+  h_transfer_data_0.src0 = buff_0;
+  h_transfer_data_0.src1 = buff_coarse_0;
+  h_transfer_data_0.N = N;
+  h_transfer_data_0.gpu = 0;
+
+  h_transfer_data_1.dest0 = buff_0 + N;
+  h_transfer_data_1.dest1 = buff_coarse_1;
+  h_transfer_data_1.src0 = buff_1 + N;
+  h_transfer_data_1.src1 = buff_coarse_1 + N;
+  h_transfer_data_1.N = N;
+  h_transfer_data_1.gpu = 1;
+
+  HIPCHECK(hipSetDevice(0));
+  HIPCHECK(hipMemcpyAsync(transfer_data_0, &h_transfer_data_0,
+                          sizeof(struct transfer_data_t), hipMemcpyHostToDevice,
+                          stream_0));
+  HIPCHECK(hipStreamSynchronize(stream_0));
+
+  HIPCHECK(hipSetDevice(1));
+  HIPCHECK(hipMemcpyAsync(transfer_data_1, &h_transfer_data_1,
+                          sizeof(struct transfer_data_t), hipMemcpyHostToDevice,
+                          stream_1));
+  HIPCHECK(hipStreamSynchronize(stream_1));
+
+  for (int op = begin_op; op < end_op; op ++) {
+    const char *OpsName[] = {"Copy", "Local Copy", "Double Copy", "Reduce", "ReduceCopy"};
+    printf("Testing %s: \n", OpsName[op]);
+    // 2 warm up cycles
+    for (int i = 0; i < 2; i ++) {
+      HIPCHECK(hipSetDevice(0));
+      //launch the kernel
+      hipLaunchKernelGGL(flagSyncKerns[op],
+          /*grid dim x,y,z*/        dim3(workgroups, 1, 1),
+          /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+          /*dynamic shared mem*/    0,
+          /*stream*/                stream_0,
+          /*kernel args*/           transfer_data_0, d_profiling_data_0);
+
+      HIPCHECK(hipSetDevice(1));
+      //launch the kernel
+      hipLaunchKernelGGL(flagSyncKerns[op],
+          /*grid dim x,y,z*/        dim3(workgroups, 1, 1),
+          /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+          /*dynamic shared mem*/    0,
+          /*stream*/                stream_1,
+          /*kernel args*/           transfer_data_1, d_profiling_data_1);
+    }
+
+    HIPCHECK(hipSetDevice(0));
+    HIPCHECK(hipStreamSynchronize(stream_0));
+    HIPCHECK(hipMemset(d_profiling_data_0, 0, sizeof(struct profiling_data_t)));
+    HIPCHECK(hipSetDevice(1));
+    HIPCHECK(hipStreamSynchronize(stream_1));
+    HIPCHECK(hipMemset(d_profiling_data_1, 0, sizeof(struct profiling_data_t)));
+
+    auto start = std::chrono::high_resolution_clock::now();
+    for (int i = 0; i < NUM_ITERS; i ++) {
+      HIPCHECK(hipSetDevice(0));
+      //launch the kernel
+      hipLaunchKernelGGL(flagSyncKerns[op],
+          /*grid dim x,y,z*/        dim3(workgroups, 1, 1),
+          /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+          /*dynamic shared mem*/    0,
+          /*stream*/                stream_0,
+          /*kernel args*/           transfer_data_0, d_profiling_data_0);
+
+      HIPCHECK(hipSetDevice(1));
+      //launch the kernel
+      hipLaunchKernelGGL(flagSyncKerns[op],
+          /*grid dim x,y,z*/        dim3(workgroups, 1, 1),
+          /*block dim x,y,z*/       dim3(THREADS, 1, 1),
+          /*dynamic shared mem*/    0,
+          /*stream*/                stream_1,
+          /*kernel args*/           transfer_data_1, d_profiling_data_1);
+    }
+
+    HIPCHECK(hipSetDevice(0));
+    HIPCHECK(hipStreamSynchronize(stream_0));
+    HIPCHECK(hipSetDevice(1));
+    HIPCHECK(hipStreamSynchronize(stream_1));
+    auto delta = std::chrono::high_resolution_clock::now() - start;
+    double deltaSec = std::chrono::duration_cast<std::chrono::duration<double>>(delta).count();
+
+    HIPCHECK(hipMemcpyAsync(profiling_data_0, d_profiling_data_0,
+                            sizeof(struct profiling_data_t), hipMemcpyDeviceToHost,
+                            stream_0));
+    HIPCHECK(hipStreamSynchronize(stream_0));
+
+    HIPCHECK(hipMemcpyAsync(profiling_data_1, d_profiling_data_1,
+                            sizeof(struct profiling_data_t), hipMemcpyDeviceToHost,
+                            stream_1));
+    HIPCHECK(hipStreamSynchronize(stream_1));
+
+    double speed = (double)(profiling_data_0->bytes_transferred) / (deltaSec*1.0E9);
+    printf("Transfered %lu bytes in %f s. Throughput %f GB/s\n", profiling_data_0->bytes_transferred, deltaSec, speed);
+
+    fprintf(stderr, "GPU 0: write_cycles %ld bytes_transferred %ld\n",
+      profiling_data_0->write_cycles, profiling_data_0->bytes_transferred);
+
+    fprintf(stderr, "GPU 1: write_cycles %ld bytes_transferred %ld\n",
+      profiling_data_1->write_cycles, profiling_data_1->bytes_transferred);
+  }
+
+  HIPCHECK(hipStreamDestroy(stream_0));
+  HIPCHECK(hipStreamDestroy(stream_1));
+  HIPCHECK(hipFree((void*) transfer_data_0));
+  HIPCHECK(hipFree((void*) buff_0));
+  HIPCHECK(hipFree((void*) buff_coarse_0));
+  HIPCHECK(hipFree((void*) d_profiling_data_0));
+  free(profiling_data_0);
+  HIPCHECK(hipFree((void*) transfer_data_1));
+  HIPCHECK(hipFree((void*) buff_1));
+  HIPCHECK(hipFree((void*) buff_coarse_1));
+  HIPCHECK(hipFree((void*) d_profiling_data_1));
+  free(profiling_data_1);
+}