Verify Stream correctness in Barrier operators in cudf

[karthikeyann]

Description

Barrier operators like CudfHashBuild, CudfHashAggregation, CudfTopN in cudf could accept inputs from multiple streams. If the input streams are not synchronized or not connected to current stream via cuda event by creating a dependency, the results could be wrong and it's hard to debug these issues, similar to multi-threading issues.
So, All Barrier cudf operators should be validated for stream dependency.

Here are Barrier operators

• CudfTopN (fix(cudf): Fix stream synchronization issue with CudfTopN #16431 (comment), https://github.com/facebookincubator/velox/pull/16431/changes#r2855932924, https://github.com/facebookincubator/velox/pull/16542/changes#diff-a509b844fb8ad906739976759b3b13d69007bcff0709092b42eca75272ab560cR150 )
• CudfOrderBy
• CudfHashAggregation
• CudfHashBuild
• New Batching operator (recent stream bug fix in 956e000)

Suggested stream dependency injection:

void streamsWaitForStream(
    CudaEvent& event,
    cudf::host_span<rmm::cuda_stream_view const> streams,
    rmm::cuda_stream_view stream) {
  event.recordFrom(stream);
  std::for_each(
      streams.begin(), streams.end(), [&](auto& strm) { event.waitOn(strm); });
}