Use Channel Instead of BufferBlock

[jwood803]

Updates for #4482.

@eerhardt I took a shot at a single replacement to make sure it looks good before continuing. Does this look good or am I missing anything?

[codecov]

Codecov Report

Merging #5123 into master will decrease coverage by 0.10%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##           master    #5123      +/-   ##
==========================================
- Coverage   75.66%   75.55%   -0.11%     
==========================================
  Files         993      995       +2     
  Lines      178168   179316    +1148     
  Branches    19122    19298     +176     
==========================================
+ Hits       134802   135487     +685     
- Misses      38143    38564     +421     
- Partials     5223     5265      +42     

Flag	Coverage Δ
#Debug	75.55% <100.00%> (-0.11%)	⬇️
#production	71.50% <100.00%> (-0.13%)	⬇️
#test	88.63% <ø> (-0.05%)	⬇️

Impacted Files	Coverage Δ
src/Microsoft.ML.Sweeper/AsyncSweeper.cs	72.60% <100.00%> (ø)
src/Microsoft.ML.Data/Transforms/Hashing.cs	78.03% <0.00%> (-14.41%)	⬇️
src/Microsoft.ML.OnnxConverter/SaveOnnxCommand.cs	71.42% <0.00%> (-2.11%)	⬇️
.../Microsoft.ML.Vision/ImageClassificationTrainer.cs	91.21% <0.00%> (-0.48%)	⬇️
...rosoft.ML.Data/DataView/RowToRowMapperTransform.cs	92.89% <0.00%> (-0.48%)	⬇️
...cenariosWithDirectInstantiation/TensorflowTests.cs	91.56% <0.00%> (-0.32%)	⬇️
test/Microsoft.ML.Benchmarks/HashBench.cs	0.00% <0.00%> (ø)
test/Microsoft.ML.Benchmarks/BenchmarkBase.cs	0.00% <0.00%> (ø)
test/Microsoft.ML.Functional.Tests/Debugging.cs	100.00% <0.00%> (ø)
test/Microsoft.ML.TestFrameworkCommon/Datasets.cs	100.00% <0.00%> (ø)
... and 18 more

[codecov]

Codecov Report

Merging #5123 into master will decrease coverage by 1.97%.
The diff coverage is 86.95%.

@@            Coverage Diff             @@
##           master    #5123      +/-   ##
==========================================
- Coverage   75.66%   73.68%   -1.98%     
==========================================
  Files         993     1022      +29     
  Lines      178168   190353   +12185     
  Branches    19122    20473    +1351     
==========================================
+ Hits       134802   140264    +5462     
- Misses      38143    44557    +6414     
- Partials     5223     5532     +309     

Flag	Coverage Δ
#Debug	73.68% <86.95%> (-1.98%)	⬇️
#production	69.42% <86.95%> (-2.21%)	⬇️
#test	87.62% <ø> (-1.06%)	⬇️

Impacted Files	Coverage Δ
test/Microsoft.ML.FSharp.Tests/SmokeTests.fs	88.23% <ø> (-7.85%)	⬇️
...soft.ML.Data/Transforms/RowShufflingTransformer.cs	71.25% <83.33%> (-0.53%)	⬇️
src/Microsoft.ML.Sweeper/AsyncSweeper.cs	71.42% <100.00%> (-1.18%)	⬇️
...ML.Tests/Scenarios/IrisPlantClassificationTests.cs	0.00% <0.00%> (-100.00%)	⬇️
test/Microsoft.ML.AutoML.Tests/Util.cs	50.00% <0.00%> (-50.00%)	⬇️
...st/Microsoft.ML.Functional.Tests/Datasets/Adult.cs	58.82% <0.00%> (-41.18%)	⬇️
...c/Microsoft.ML.FastTree/Utils/ThreadTaskManager.cs	79.48% <0.00%> (-20.52%)	⬇️
...rc/Microsoft.ML.AutoML/API/RunDetails/RunDetail.cs	77.27% <0.00%> (-18.19%)	⬇️
...ft.ML.Data/DataLoadSave/Database/DatabaseSource.cs	83.33% <0.00%> (-16.67%)	⬇️
...enerator/CodeGenerator/CSharp/PipelineExtension.cs	67.34% <0.00%> (-16.33%)	⬇️
... and 342 more

[eerhardt]

I think this looks good.

[jwood803]

Thanks @eerhardt! I'll make similar changes to the other items and will mark this PR ready.

[harishsk]

[eerhardt]

I think you can remove this line as well:

machinelearning/test/Microsoft.ML.FSharp.Tests/SmokeTests.fs

Line 25 in 665a366

#r @"../../bin/AnyCPU.Debug/Microsoft.ML.FSharp.Tests/net461/System.Threading.Tasks.Dataflow.dll"

[eerhardt]

        protected override void Dispose(bool disposing)

This whole Dispose method looks like it can be deleted now.

---

Refers to: src/Microsoft.ML.Data/Transforms/RowShufflingTransformer.cs:555 in 8a45c1d. [](commit_id = 8a45c1d, deletion_comment = False)

[eerhardt]

Previously _toConsume.Receive() would block the thread until something was available. But now TryRead will return false immediately if something wasn't available. In that case, we will keep spinning in the while loop and checking the reader constantly.

It might make sense to keep the previous behavior and block the thread until _toConsumeChannel has an item available, so we aren't wasting CPU cycles spinning here.

[mstfbl]

@eerhardt @harishsk Is there a reason why we haven't yet merged in this PR?

[eerhardt]

Is there a reason why we haven't yet merged in this PR?

I don't believe this comment has been addressed yet:

#5123 (comment)

[jwood803]

I don't believe this comment has been addressed yet

@eerhardt I may need some guidance on the best way to block the thread there. 😃 Would using the lock keyword work for that?

[eerhardt]

I believe you can call ChannelReader<T>.WaitToRead and block on it. This is similar to how Receive() works:

https://github.com/dotnet/runtime/blob/4cb924de165aa8e5373a109a68611d422aa72a09/src/libraries/System.Threading.Tasks.Dataflow/src/Base/DataflowBlock.cs#L952-L959

            // Get a TCS to represent the receive operation and wait for it to complete.
            // If it completes successfully, return the result. Otherwise, throw the
            // original inner exception representing the cause.  This could be an OCE.
            Task<TOutput> task = ReceiveCore(source, false, timeout, cancellationToken);
            try
            {
                return task.GetAwaiter().GetResult(); // block until the result is available
            }

[jwood803]

@eerhardt Updated to use WaitToReadAsync, but definitely let me know if more changes are needed.

[eerhardt]

Looks good! Thanks for the great work here, @jwood803.

[harishsk]

@jwood803 Thank you for your contribution. I am merging the PR now.