ForceBatch method doesn't complete batch immediately about open.channelextensions HOT 12 CLOSED

Ok. I'll investigate.

from open.channelextensions.

Did you experience this bug with 3.3.2?

from open.channelextensions.

This test passes: https://github.com/electricessence/Open.ChannelExtensions/blob/master/Open.ChannelExtensions.Tests/BatchTests.cs#L112-L153

from open.channelextensions.

Now that the other bug is fixed, you should be able to 'flush' the reader periodically with a timer/interval.

var c = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });

var reader = c.Reader.Batch(2);
Task.Run(async ()=>{
  await Task.Delay(1000);
  while(!reader.Completion.IsCompleted)
  {
    reader.ForceBatch();
    await Task.Delay(1000);
  }
});
await reader.ReadAllAsync(()=>{/*...*/});

from open.channelextensions.

Here's my example:

var c = Channel.CreateUnbounded<int>(new UnboundedChannelOptions { SingleReader = false, SingleWriter = false });
var reader = c.Reader.Batch(3);
Task.Run(async () =>
{
    await Task.Delay(1000);
    c.Writer.TryWrite(1);
    c.Writer.TryWrite(2);
    c.Writer.TryWrite(3);
    c.Writer.TryWrite(4);
    c.Writer.TryWrite(5);

    await Task.Delay(3000);
    Console.WriteLine(reader.ForceBatch());
});

await reader.ReadAllAsync(async batch =>
{
    Console.WriteLine($"Batch: {string.Join(",", batch.Select(i => i.ToString()))}");
    await Task.Delay(500);
});

from open.channelextensions.

K. Adding to tests.

from open.channelextensions.

Thanks again for helping with this. This is a tricky one, that if the implementation was to buffer all incomming channel content to another channel then it probably wouldn't be an issue.
I want to retain the BatchingChannelReader as a 'reader' that only pulls from it's source and subsequently batches on demand. As you might imagine, there could be unintended side effects if that wasn't the case.

The difference between your test and mine is that mine delays the read operation and yours doesn't.
Yours reveals a situation that would only occur with .ForceBatch() where I'm 'awaiting' the source before proceeding but because the source doesn't have anything new, the await is stuck.
What I was trying to avoid before was having to double await which typically requires a Task.WhenAny instead of simply awaiting a ValueTask.

Again, appreciate the help. I'm investigating further to find a solution.

from open.channelextensions.

I can suggest that BatchingChannelReader could actually work with "wrapped" items and `.ForceBatch' could post some predefined "command" item that cause completion of the batch. Then you don't need to mess with tasks.

from open.channelextensions.

The problem is, when implementing the buffer differently, the code is quite simple and without any additional contention. But as soon as I add the .ForceBatch() method, everything gets way more complex and ultimately less performant because of contention.

Still working on it.

from open.channelextensions.

Released an update.
https://www.nuget.org/packages/Open.ChannelExtensions/3.4.0

I tried a few different methods. I'm not 100% happy with having to create Tasks under the hood, but it's actually not too terrible. It does work as expected. All tests are passing.

I avoided incurring the same overhead on the Join operation and implemented some minor improvements I should have already.

from open.channelextensions.

@alexeynikitin Thanks again. Lemme know if there's anything else.

from open.channelextensions.

@electricessence Ok, thank you!

from open.channelextensions.