[BUG]: A channel with multiple producers splits batches produced by an individual producer

[steinybot]

Version

1.0-RC1+9-cc7446a2-SNAPSHOT

Scala Version

3.7.1

Expected Behavior

Given a channel:

Channel.init[Int](4)

If there are multiple producers to a channel that produce batches (where the channel capacity is not a multiple of the batch size):

Async.foreach((1 to 60).grouped(3).toSeq)(channel.putBatch)

Then with a single consumer:

Async.fill(20, 1)(channel.takeExactly(3))

I would expect that those batches could come in any order but the batches themselves will never be split.

takes.forall { chunk =>
    // Every group of 3 elements should be consecutive.
    val first    = chunk.head
    val expected = first to (first + 2)
    chunk == expected
}

Actual Behavior

This may be false.

Consider:

1. putBatch enqueues 1, 2, 3
2. putBatch enqueues 4, channel is now full, pending 5, 6
3. putBatch pending 7, 8, 9

Saving the pending 5, 6 and 7, 8, 9 can be in any order so you end up with a channel that effectively has: 1, 2, 3, 4, 7, 8, 9, 5, 6 which has split the batch of 4, 5, 6.

There are a few different variations of this that will store the remainder of a batch as a pending put including drainUpTo and poll (maybe that's all?) and for zero and non-zero capacity channels.

I think you always have to have more than one producer. In this example the producers race amongst themselves which is somewhat expected although I don't think splitting batches is correct. In other cases there needs to be one producer with a pending batch, some arbitrary time delay (no race), and then another producer races with the consumer that is pending the remainder from the first pending batch.

Steps to Reproduce

See tests in https://github.com/getkyo/kyo/pull/1284/files.

Current Workaround

The only practical workaround is to have a single producer or forget about any kind of order.

[steinybot]

Actually a far better workaround is to change the type of the channel to be Channel.init[Chunk[Int]] and have each element be a batch. Maybe that is in fact the solution and we document the caveats somewhere.

[johnhungerford]

Actually a far better workaround is to change the type of the channel to be Channel.init[Chunk[Int]] and have each element be a batch. Maybe that is in fact the solution and we document the caveats somewhere.

This is my preferred approach. I have already been using this for async streaming operations where I want to preserve chunking. I think putBatch should be regarded as an optimization for putting a chunk of elements, and not having any special guarantees other than those that exist for putting elements, e.g., the channel consumer will never receive elements from a single producer out of order, even though ordering is not guaranteed vis a vis other producers.