Fs2: RFC: ability to close Queues

[SystemFw]

A very common pattern in both user code and fs2 internals is sending a stream to a queue from which it is dequeued concurrently (with transformations in between).

Since dequeue is semantically blocking, there often is the need to stop waiting on something, and in most combinators this has to be done in a deterministic way that doesn't lose elements (so interruptWhen doesn't cut it).

What we do right now is noneTerminate on one hand plus unNoneTerminate on the other, and while this is elegant in some ways, and works really well, it forces us to wrap things in Option quite a lot, and that's not great for perf. You can see this pattern all over the place (the observe issue got me thinking about this again).

We could address this in a couple of ways:

• Create versions of unNoneTerminate/noneTerminate that work with an allocation-free Option replacement (I think the limitations of these data types as general replacements for Option won't really apply in this limited use case).
• Add logic to the Queue internals to allow closing the queue and unblocking a dequeued. Note that replicate the behaviour we have now, a close: F[Unit] will only trigger once all the elements that are in the queue before it gets called are dequeued, or we end up with the same concurrent characteristics of interruptWhen, which already exists.

What do you think? Is this a problem worth addressing? Any other solutions in mind?

EDIT:

• A third alternative is to create a single consumer channel since in most of the internals this is what queues are used for (I don't think we need to force a single producer limitation here though, since that's quite useful, e.g. join )

[pchlupacek]

To understand this correctly, once close is called

• What would be semantics of subscriber if
  
  
  
  • there is element in the queue
  
  
  • there are no elements in the queue
  
  
• What would be publisher semantics if
  
  
  
  • Publisher will wait on queue to be drained to finsih its publish while close is recevied
  
  
  • Publisher will try to publish to closed queue.
  
  

Would be excellent to design this, however I would like to have this as feature on top of the current queues and not include it as standard opt-in in default queue implementation.

[SystemFw]

To understand this correctly, once close is called, What would be semantics of subscriber if

there is element in the queue

subscriber would first get all the elements in the queue, then stop waiting. Additional logic needs to be devised if we're allowing multiple subscribers (but note that atm unNoneTerminate will also not work with multiple subscribers, you need extra machinery)

there are no elements in the queue

The dequeue stream terminates successfully

What would be publisher semantics if

Publisher will wait on queue to be drained to finish its publish while close is received

Not sure :) (also not sure I 100% understand the question)

Publisher will try to publish to closed queue.

Up for discussion, at the moment with the current solution this is allowed (you can publish a Some(a) after a None has been published). Maybe it should be an error, maybe not.

Would be excellent to design this, however I would like to have this as feature on top of the current queues and not include it as standard opt-in in default queue implementation.

I'm very open to doing it this way if possible

[pchlupacek]

Publisher will wait on queue to be drained to finish its publish while close is received

Not sure :) (also not sure I 100% understand the question)

Bounded queues may block, so publisher may wait for subscribers to consume before enqueueing more ... Not sure what would happen when you call close ....

I generally think the Publish semantics is what will count and we have to be very careful on corectness there.

[djspiewak]

I actually prefer things not being closable. Every bit of dynamic state which isn't reflected in any static form is a bug waiting in ambush. Queues were closable back in scalaz-stream and it was sometimes the source of really hard-to-trace issues. I definitely itch a little bit at the all-too-common pattern of enqueue/dequeue with noneTerminate/unNoneTerminate, but other than implicitly adding some utility functions when a queue contains Option values, I'm not sure what to do. We could alias the type as well (e.g. type CloseableQueue[F[_], A] = Queue[F, Option[A]]).

So I would be 👎

[SystemFw]

Right, that makes sense. My original idea is to have something that reflects the semantics we get with noneTerminate/unNoneTerminate, which works well in terms of correctness, but with less allocation.

In this case close (which should perhaps be called closeOnDrain or something) will only terminate the first dequeuer that encounters an empty queue after close has been called.

This is how things work now, but once you describe you realise it's pretty weird, and that's what had me thinking about a channel-specific implementation. In any case I'm open to ideas.

[SystemFw]

@djspiewak What about the allocation-free OptionLike combinators?

I'm also not super keen on adding complex semantics to queues (generally speaking), but this is really pervasive and I _suspect_ quite a perf drain

[djspiewak]

which works well in terms of correctness, but with less allocation.

Oh, I forgot to address this point…

Frankly, Queue has oodles of allocation already. People simply should not be threading performance sensitive values through it. The same goes for Stream itself. When you need performance (i.e. you're sensitive to these kinds of allocations), you should be enqueueing large chunks all at once (where the type A reflects the chunkiness, rather than being a single semantic value).

If this is a performance drain for people, then they are using Queue incorrectly, and I'd be really surprised if the Option boxing is measurable relative to the other boxing that is in and around Queue.

[SystemFw]

Also note that even the current solution is fragile: imagine enqueing None, Some(a) and then None with multiple consumers. What's the behaviour there?

and I'd be really surprised if the Option boxing is measurable relative to the other boxing that is in and around Queue.

Right, that was why I asked if this is worth addressing, it _strikes_ me as wasteful intuitively, but I'm could quite possibly be entirely wrong

[djspiewak]

Also note that even the current solution is fragile: imagine enqueing None, Some(a) and then None with multiple consumers.

Do people actually run multiple consumers off a single queue on a regular basis? I agree that None termination is fragile in that case, but that case is already kind of horrifying. Maybe we should do something to address that a bit more directly?

[pchlupacek]

I think multiple consumers from the queue are very rare. But, it may be used for performance and paralelization of the work.

[SystemFw]

Do people actually run multiple consumers off a single queue on a regular basis?

Yeah that's part of the point. There are definitely several cases in which there are multiple consumers, but the one I'm talking about here is the MPSC case we see in most of our combinators. I'm not sure how to exploit this extra info to improve things (this is the angle I was coming from with the extra Channel type or something along those lines).

[pchlupacek]

Re performance, we could have queue that buffers chunks and then delivers to subscribers chunks of certain _size_. This would be on top of current queue and still would have signature of Queue[F, A]

[pchlupacek]

@SystemFw also perhaps how about if we introduce something, lets call it for example Joint. That thing will allow one publisher to publish messages to it and one subscriber to consume messages with possible configurable size of queue. When either of them terminates, then all stream will terminate. I.e.

trait Joint[F, A] {
   def sink: Sink[F, A]
   def stream: Stream[F, A]
}

//then on async

def joint[F[_], A](queueSize: Long): F[Joint[F, A]]

obviously sink won't pull unless stream starts to pull.

[SystemFw]

We need Joint to allow multiple consumers I think, if we want it to be general enough to use it in join or merge

[pchlupacek]

I think this won't work, definitely will be problematic with Close semantics of sink/stream. To have multiple consumers, look on distribute proposal in observe discussion

[SystemFw]

ah, sorry, I misspoke. I meant multiple producers, but single consumer

[pchlupacek]

Thats interesting idea. I still think that in that case termination of the stream will be problematic (not to implement but reason about) and you can have that by Q.dequeue.to(joint.sink) in fact ?

[pchlupacek]

The whole point of Joint is to be very minimalistic and thus simple to implement and reason about. I believe then you can use it to build higher level primitives. I also see Joint to be a little bit "enchanted" synchronous queue (with termination and cancellation semantics) and that it.

[pchlupacek]

Also the queueSize won't be necessary, joint may operate only on chunks, thus queueing depth will be configurable by size of chunks.

[djspiewak]

Multiple producer, single consumer is a really really common model in my experience. I know it makes the termination semantics complicated, but I think it's probably worth supporting.

[pchlupacek]

But thats where we have queue? Or what different semantics to queue we
need?

On Fri, 3 Aug 2018 at 18:58, Daniel Spiewak [e-mail]
wrote:

Multiple producer, single consumer is a really really common model in my
experience. I know it makes the termination semantics complicated, but I
think it's probably worth supporting.

—
You are receiving this because you commented.

Reply to this email directly, view it on GitHub
https://github.com/functional-streams-for-scala/fs2/issues/1207#issuecomment-410314596,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AApaoMoEFaAJ-5046vf3f7mRGE0hsaQ7ks5uNIFFgaJpZM4Vsm4W
.

[pchlupacek]

@SystemFw do you think this can be closed now?