Fs2: FS2 Dynamically created subscribers: doesn't receive messages

[supperslonic]

I'm using pub-sub pattern in fs2. I dynamically create topics and subscribers while processing a stream of messages. For some reason, my subscribers receive only initial message, but further published messages never get to subscribers

def startPublisher2[In](inputStream: Stream[F, Event]): Stream[F, Unit] = {
    inputStream.through(processingPipe)
  }

  val processingPipe: Pipe[F, Event, Unit] = { inputStream =>
    inputStream.flatMap {
      case message: Message[_] => initSubscriber(message)
        .flatMap { topic => Stream.eval(topic.publish1(message)) }
    }
  }

  def initSubscriber[In](message: Message[In]): Stream[F,Topic[F, Event]] = {

    Option(sessions.get(message.sessionId)) match {
      case None =>
        println(s"=== Create new topic for sessionId=${message.sessionId}")

        val topic = Topic[F, Event](message)
        sessions.put(message.sessionId, topic)
         Stream.eval(topic) flatMap  {t =>
           //TODO: Is there a better solution?
           Stream.empty.interruptWhen(interrupter) concurrently startSubscribers2(t)
         }

      case Some(topic) =>
        println(s"=== Existing topic for sessionId=${message.sessionId}")
        Stream.eval(topic)
    }
  }

Subscriber code is simple:

def startSubscribers2(topic: Topic[F, Event]): Stream[F, Unit] = {
    def processEvent(): Pipe[F, Event, Unit] =
      _.flatMap {
        case e@Text(_) =>
          Stream.eval(F.delay(println(s"Subscriber processing event: $e")))
        case Message(content, sessionId) =>
          //Thread.sleep(2000)
          Stream.eval(F.delay(println(s"Subscriber #$sessionId got message: ${content}")))
        case Quit =>
          println("Quit")
          Stream.eval(interrupter.set(true))
      }

    topic.subscribe(10).through(processEvent())
  }

The output is the following:

=== Create new topic for sessionId=11111111-1111-1111-1111-111111111111
Subscriber #11111111-1111-1111-1111-111111111111 got message: 1
=== Existing topic for sessionId=11111111-1111-1111-1111-111111111111
=== Create new topic for sessionId=22222222-2222-2222-2222-222222222222
Subscriber #22222222-2222-2222-2222-222222222222 got message: 1
=== Create new topic for sessionId=33333333-3333-3333-3333-333333333333
Subscriber #33333333-3333-3333-3333-333333333333 got message: 1
=== Existing topic for sessionId=22222222-2222-2222-2222-222222222222
=== Existing topic for sessionId=22222222-2222-2222-2222-222222222222

I don't see messages published to existing topic. Also, I'm wondering if there is a better way to start an async stream of subscribers, instead of Stream.empty.interruptWhen(interrupter) concurrently startSubscribers2(t)

[SystemFw]

Your problem is Stream.empty concurrently (the interruptWhen there is irrelevant). Stream.empty terminates immediately, and concurrently interrupts the right hand side when the left terminates

[SystemFw]

I will try and give you a solution later when I have more time, ping me here or on our gitter

[supperslonic]

Your problem is Stream.empty concurrently (the interruptWhen there is irrelevant). Stream.empty terminates immediately, and concurrently interrupts the right hand side when the left terminates

Thank you for the response, I was suspecting that this is not a good solution and might be a cause of the issue, but I didn't find any other solution to make it work. I will appreciate your help.

[mpilquist]

Another issue is that you're caching the F[Topic[F, Event]] instead of a Topic[F, Event].

Given val topic = Topic[F, Event](message), think of topic as a factory that constructs a topic each time it's evaluated. To fix, you'd need to evaluate topic once (via Stream.eval) and cache the result.

One other note -- it appears the sessions cache is a mutable map. It can be really hard to reason about program behavior when mixing purely functional code with non purely functional code. I recommend replacing that mutable map with a Ref[F, scala.collection.immutable.Map[...]].

[SystemFw]

Can I also ask why you are using a Topic there? At a quick glance it appears to be only ever be one subscriber per topic, am I missing something?

[supperslonic]

Can I also ask why you are using a Topic there? At a quick glance it appears to be only ever be one subscriber per topic, am I missing something?

Yes, there is only one subscriber per topic. Basically, i'm trying to split an incoming stream of messages into sub streams for each dedicated consumer

[supperslonic]

Another issue is that you're caching the F[Topic[F, Event]] instead of a Topic[F, Event].

Given val topic = Topic[F, Event](message), think of topic as a factory that constructs a topic each time it's evaluated. To fix, you'd need to evaluate topic once (via Stream.eval) and cache the result.

One other note -- it appears the sessions cache is a mutable map. It can be really hard to reason about program behavior when mixing purely functional code with non purely functional code. I recommend replacing that mutable map with a Ref[F, scala.collection.immutable.Map[...]].

Totally agree with F[Topic[F, Event]], it is just a version of the code when i was experimenting with different solutions and i forgot to fix this.

[SystemFw]

Yes, there is only one subscriber per topic. Basically, i'm trying to split an incoming stream of messages into sub streams for each dedicated consumer

Right, you don't need a topic then, just a queue.

The starting point is Ref[F, immutable.Map[Id, Queue[F]]]. I will sketch out a prototype in a bit

[SystemFw]

Oh, do you have any way to know when it's safe to consider a Session over, or alternatively if the number of distinct SessionId is bounded? In the current form, the map keeps growing over time.

[supperslonic]

Thanks for your help!
We plan to make it bounded (grows from 0 up to N), I was thinking to use Semaphore for that.

[SystemFw]

That's not (the entirety) of what I mean. Every sessionId creates a new subscriber. If there is never a notion of "unsubscribe" or "terminate session", the map will keep growing.
Or in other words, if you use a semaphore that you decrement on each subscription, what's the condition for incrementing it?

(grows from 0 up to N)

Btw Semaphores work the other way, start at N, block at 0, unblock when the count grows

[supperslonic]

if there is never a notion of "unsubscribe" or "terminate session", the map will keep growing.
Based on a specific message, for example Quite, the session is terminated and a new one can be initiated for another client

[Daenyth]

@supperslonic I was recently working on a groupBy structure that I wanted to be more generic, I'll see if I can gist it for you to use as a starting point.

Edit: https://gist.github.com/Daenyth/bcb0b4b8978dc9fa00735f374f5f2d92

[supperslonic]

@supperslonic I was recently working on a groupBy structure that I wanted to be more generic, I'll see if I can gist it for you to use as a starting point.

Edit: https://gist.github.com/Daenyth/bcb0b4b8978dc9fa00735f374f5f2d92

This does look like what I need, thanks! I will try it out

[SystemFw]

I'm going to close this issue for now, feel free to keep commenting if you need anything else

[Daenyth]

@supperslonic I had planned to add a key-bounded version of that grouping structure but realized I didn't need it for my use case. (Envelope math: 10K keys assuming 10 object allocations per key * 16 bytes/obj < 1.6MiB. Assuming 100 object allocations per key at 10K keys <16 MiB)

A structure that terminates the streams just needs a method to decide how to do it, and possibly a lock on updates for ease of coding; I sketched out this state model for limiting the number of distinct keys:

state object: Ref of Map (K -> (NoneTerminatedQueue, Deferred of Unit) )
on enqueue:

• has key?
  
  
  
  • yes -> add to queue
  
  
  • no
  
  
  • with lock:
    
    
    
    
    
    • check if map has space
    
    
    
    • yes -> create queue
      
      
      
      
      
      
      
      • add element to queue
      
      
      
      
      • return queue.dequeue ++ Stream.eval_(deferred.complete())
      
      
      
      
      • release lock
      
      
      
      
    
    
    
    • no -> select queue for terminatio)
      
      
      
      
      
      
      
      • enqueue None, wait for queue to drain (deferred.get)
      
      
      
      
      • remove old (q, d) from state, create new as with above
      
      
      
      
    
    
    
  
  

[supperslonic]

@Daenyth Thanks for sharing the code with me, I have implemented a KeyBoundedKeyedEnqueue, it seems to be working fine except for one thing. For some reason, if i send standard messages and terminate messages (which should remove a queue from map), they behaviour is not determenistic.

override def enqueue1(key: K, item: A): F[Option[(K, fs2.Stream[F, A])]] = {
      keyedEnqueue.hasKey(key).flatMap {
        case true =>
          println(s"Using existing queue with key=$key, message=$item")
          keyedEnqueue.enqueue1(key, item)
        case false =>
          println(s"Creating new queue for key=$key, message=$item")
          limit.acquire >> keyedEnqueue.enqueue1(key, item).map(_.map {
            case (key, stream) =>
              key -> Stream.bracket(F.delay(key))(cleanSession).flatMap(_ => stream)
          })
      }
  }

This is a clean function:

private def cleanSession(key: K): F[Unit] = {
    for {
     _ <- removeKey(key)
      _ <- limit.releaseN(1)
      _ <- F.delay(println(s"Released $key"))
    } yield ()
  }

This is a handling of Terminate message:

override def shutdown(key: K): F[Option[(K, Stream[F, A])]] = withKey(key)(_.enqueue1(None))

Testing stream:

 for {
      _ <- new Processor[IO]().start(2 /*max number of active sessions*/, Stream(
        Message(1, sessionId1),
        Terminate(sessionId1),
        Message(1, sessionId2),
        Message(1, sessionId3),
        Message(2, sessionId1)))
    } yield ()

If the number of active queues is 2 it works correctly:

Creating new queue for key=11111111-1111-1111-1111-111111111111, message=Message(1,11111111-1111-1111-1111-111111111111)
Creating new queue for key=22222222-2222-2222-2222-222222222222, message=Message(1,22222222-2222-2222-2222-222222222222)
Creating new queue for key=33333333-3333-3333-3333-333333333333, message=Message(1,33333333-3333-3333-3333-333333333333)
Processing Message(1,11111111-1111-1111-1111-111111111111) for sessionId=11111111-1111-1111-1111-111111111111
Processing Message(1,22222222-2222-2222-2222-222222222222) for sessionId=22222222-2222-2222-2222-222222222222
Released 11111111-1111-1111-1111-111111111111
Creating new queue for key=11111111-1111-1111-1111-111111111111, message=Message(2,11111111-1111-1111-1111-111111111111)
Processing Message(1,33333333-3333-3333-3333-333333333333) for sessionId=33333333-3333-3333-3333-333333333333

But if i set a number of active queues to 3 for example, for some reason it doesn't try to create a new queue for sessionId1, but tries to reuse an existing one:

Creating new queue for key=11111111-1111-1111-1111-111111111111, message=Message(1,11111111-1111-1111-1111-111111111111)
Creating new queue for key=22222222-2222-2222-2222-222222222222, message=Message(1,22222222-2222-2222-2222-222222222222)
Creating new queue for key=33333333-3333-3333-3333-333333333333, message=Message(1,33333333-3333-3333-3333-333333333333)
Processing Message(1,11111111-1111-1111-1111-111111111111) for sessionId=11111111-1111-1111-1111-111111111111
Processing Message(1,22222222-2222-2222-2222-222222222222) for sessionId=22222222-2222-2222-2222-222222222222
Using existing queue with key=11111111-1111-1111-1111-111111111111, message=Message(666,11111111-1111-1111-1111-111111111111)
Processing Message(1,33333333-3333-3333-3333-333333333333) for sessionId=33333333-3333-3333-3333-333333333333

It prints Using existing queue with key=11111111-1111-1111-1111-111111111111... instead of Released 11111111-1111-1111-1111-111111111111 and Creating new queue for key=11111111-1111-1111-1111-111111111111...

[supperslonic]

I actually think the problem is in the way i'm consuming the stream in this code, but I can't understand what is wrong:

def start[In](maxSessions: Int, inputStream: Stream[F, AlmondEvent]): Stream[F, Unit] = {
    val multiQueue = new MultiQueue[F]()
    inputStream
      .through(multiQueue.groupBy[AlmondEvent, UUID](maxSessions)(_.sessionId)(_.isInstanceOf[Terminate]))
      .map(message => {
        message._2.through(processSubStream(message._1))
      }).parJoin(maxSessions)
  }

  def processSubStream(sessionId: UUID): Pipe[F, AlmondEvent, Unit] =
    _.flatMap(event => {
      Stream.eval(F.delay(println(s"Processing $event for sessionId=$sessionId")))
    })

And i just do compile.drain to the output of start method. Looks like some messages get stuck

[Daenyth]

Does it still stick if you use maxSessions + 1 in parJoin? I believe what's happening here is that parJoin stops pulling from upstream because it has maxSession live substreams, so the groupBy structure doesn't get pulled, so the eviction doesn't happen - therefore deadlock

[supperslonic]

I actually managed to make it work, I don't think it was due to deadlock, it was more of a race condition. I fixed it by removing the key from the map when enqueing the None element.

/** Gracefully terminate a sub-streams for the specified key */
  override def shutdown(key: K): F[Option[KeyedStream[F, K, A]]] = for {
    qm <- withKey(key)(_.enqueue1(None))
    _ <- removeKey(key)
  } yield qm

[Daenyth]

I think you need to use a lock around the state map management because concurrent enqueue aimed at the key you want to remove might enter the Queue after the None, losing the input element. I haven't confirmed this is possible but I also haven't confirmed that is impossible. I'm not sure how I'd structure a test to expose that condition yet.