MQTT streaming: SourceQueue backpressure #1577

longshorej · 2019-03-12T23:10:13Z

This adds a mechanism to mqtt-streaming to use Source.queue with OverflowStrategy.backpressure and wait for the backpressure signal before handling more messages.

I think we could probably increase from 1 to say 4 or 8, reducing latency at the expense of the cost of the memory required to buffer those events, but curious to get opinions on that.

While we're at it, it also removes BroadcastHub usage when there's only a single consumer, which should be a bit more performant.

Some funky things with the tests, in particular doc ones, so this is a draft for now.

/cc @huntc who vowed to pick up where I've left off 😃

longshorej · 2019-03-12T23:14:04Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/QueueOfferState.scala

+ case (_, other) =>
+ waitForQueueOfferCompleted(behavior, stash :+ other)
+ }
+ .orElse(behavior) // handle signals immediately


One thing I'm not clear on are the semantics here. Note that on L43, we catch all events that flow. Thus, the intention is that all signals go to behavior but nothing else (until we hit L34 in the future)

If behavior's signal handler (i.e. the behavior being run via orElse) returns Behavior.same, I'd like to clarify that it doesn't change the behavior of the actor to itself, i.e. waitForQueueOfferCompleted (L26-L46) remains the behavior. Hopefully my question makes sense..

I think the behaviour would indeed remain the same i.e. stay as per the behaviour of waitForQueueOfferCompleted. But I've not yet proved this.

However, given L46 with the orElse, I'm not seeing the signal handler of behavior being applied. When I run the tests I see a DeathPactException which means that there is no signal handler for Terminated, but there is such a signal handler for behavior...

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

huntc · 2019-03-13T04:09:46Z

I've been looking at this in some detail and ended up posing a question to the Lightbend forum. I've also tidied up a couple of things around the edge, but nothing specifically to do with this PR.

@patriknw 'hoping you may be able to enlighten me re my question. :-)

huntc · 2019-03-13T04:11:48Z

mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala

 .run()

 val connect = Connect("some-client-id", ConnectFlags.None)

 client.offer(Command(connect))

- result.failed.futureValue shouldBe a[WatchedActorTerminatedException]


This exhibits the core of my change here - this termination exception isn't something that the caller should be concerned with. It is internal to the session.

huntc · 2019-03-13T04:12:24Z

mqtt-streaming/src/test/scala/docs/scaladsl/MqttSessionSpec.scala

@@ -1271,7 +1271,7 @@ class MqttSessionSpec
 case Right(Event(cp: Unsubscribe, _)) if cp.topicFilters == unsubscribe.topicFilters =>
 unsubscribeReceived.success(Done)
 })
- .toMat(Sink.seq)(Keep.both)
+ .toMat(Sink.ignore)(Keep.left)
 .run()


The result wasn't being used, so I removed it.

ennru

Great to get the backpressure all the way.

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

ennru · 2019-03-13T09:15:31Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/scaladsl/MqttSession.scala

@@ -538,6 +544,9 @@ final class ActorMqttServerSession(settings: MqttSessionSettings)(implicit mat:
 case c: Command[A] => throw new IllegalStateException(c + " is not a server command")
 }
 )
+ .recover {
+ case _: WatchedActorTerminatedException => ByteString.empty


Great to clear those from the logs when they shouldn't concern anyone.

longshorej · 2019-03-19T19:26:00Z

FYI - This is waiting on an Akka release with akka/akka#26524 given its reliance on Behavior#orElse.

A number of commits here that improve the resilience of the mqtt-streaming connector. #1577 remains important as well, but will require an Akka release. ## Require a connection id to be specified for client flows This allows the MQTT session to better track which events are relevant, given that there can be races when old connections are torn down and new ones are established. ## Fix a bug where actor state was closed over This could cause duplicate publications to never be sent to new connections ## Republish messages on reconnect only by default Previously, messages for QoS1/2 were only republished on an interval after not receiving an ack. It is more conventional to instead republish everything only on connect, and indeed to be compliant for MQTT 5, that is the only time this is allowed. To accommodate this, the timeouts default to 0, but the previous behavior can still be restored by changing the default producer timeout settings.

longshorej · 2019-03-26T20:01:48Z

Rebased this to master, still waiting on an Akka release.

huntc · 2019-03-26T20:34:47Z

Apparently, there may be an Akka release next week.

ennru · 2019-04-03T17:02:44Z

I know you are keen to get this in now that Akka 2.5.22 is available.
Please hang in until we do Alpakka 1.0.1.

ennru · 2019-04-26T09:38:58Z

This is still a draft, anything more we need to solve?

longshorej · 2019-04-29T20:43:58Z

Ah, that was oversight! This is ready for review/merge.

ennru

Some minor comments. Needs a rebase.

ennru · 2019-05-06T10:14:51Z

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala

+
+ QueueOfferState.waitForQueueOfferCompleted(
+ serverConnected(data),
+ Seq.empty


Might be good to add (more of these further down)

Suggested change

Seq.empty

stash = Seq.empty

ennru · 2019-05-06T10:20:21Z

project/Dependencies.scala

 }

 val AwsSdkVersion = "1.11.476"
 val AwsSdk2Version = "2.5.20"
- val AkkaHttpVersion = "10.1.7"


This is not connected to this PR, is it?

thanks - removed

longshorej · 2019-05-16T02:10:38Z

Thanks @ennru - I'll get this updated tomorrow. FWIW, we've been running it in production since Akka 2.5.22 was released. It's been pretty stable, I think we're down to one remaining known issue with the connector, (ConsumeFaileds happen more regularly than expected) but we haven't tracked down the cause yet.

longshorej · 2019-05-16T17:34:53Z

Okay, this is all set.

longshorej · 2019-05-16T18:15:18Z

or not - MiMA failures. I think it's related to the new field MqttSessionSettings#clientSendBufferSize, but note that MqttSessionSettings has a private constructor. Can I just whitelist the change?

ennru · 2019-05-17T08:46:34Z

Yes, that is why we require the constructors to be private.

ennru · 2019-05-23T11:36:25Z

I took the liberty to add the MiMa rule.

2m

LGTM. Renamed the MiMa fitlers file, as we had one Alpakka release since then.

ennru · 2019-06-28T12:49:34Z

@longshorej @huntc Just to be sure, as I guess you've run this code for a while now -- are we good to make it part of Alpakka 1.1.0 next week?

huntc · 2019-06-28T21:40:11Z

Yeah, let’s get this in! Thanks

When offering commands to the internal queue, a backpressure strategy is now used to ensure that elements are not dropped. Instead, we wait for consumption downstream before continuing.

I noticed that a test wasn’t quite right - the client should only complete after we’ve received all elements. We now use “take” to ensure that the stream completes.

We want WatchedActorTerminatedException to complete stream processing as before, but not with a failure. A session shutting down isn’t in itself a failure.

2m · 2019-07-01T06:17:37Z

Rebased from the latest master, as the MiMa previous artifact version was derived to be 1.0.1 (because this PR was started before 1.0.2 release).

ennru · 2019-07-01T09:43:25Z

Thank you for this important improvement.

* mqtt streaming - introduce a mechanism to use SourceQueue backpressure When offering commands to the internal queue, a backpressure strategy is now used to ensure that elements are not dropped. Instead, we wait for consumption downstream before continuing. We want WatchedActorTerminatedException to complete stream processing as before, but not with a failure. A session shutting down isn’t in itself a failure.

longshorej commented Mar 12, 2019

View reviewed changes

huntc reviewed Mar 12, 2019

View reviewed changes

mqtt-streaming/src/main/scala/akka/stream/alpakka/mqtt/streaming/impl/ClientState.scala Outdated Show resolved Hide resolved

huntc reviewed Mar 13, 2019

View reviewed changes

ennru reviewed Mar 13, 2019

View reviewed changes

ennru added the p:mqtt-streaming label Mar 13, 2019

ennru mentioned this pull request Mar 14, 2019

akka.stream.alpakka.mqtt.streaming.ActorMqttClientSession backpressure #1581

Open

longshorej force-pushed the mqtt-source-queue-bp branch from 6511de7 to fc2d365 Compare March 19, 2019 17:21

longshorej mentioned this pull request Mar 20, 2019

mqtt-streaming: Various fixes #1595

Merged

longshorej force-pushed the mqtt-source-queue-bp branch 2 times, most recently from 1ac2674 to e683d5b Compare March 26, 2019 19:58

huntc mentioned this pull request Mar 26, 2019

Mqtt streaming: LocalPacketRouter$CannotRoute: null) #1609

Open

longshorej force-pushed the mqtt-source-queue-bp branch from e683d5b to 1994542 Compare April 4, 2019 15:51

longshorej marked this pull request as ready for review April 29, 2019 20:44

ennru reviewed May 6, 2019

View reviewed changes

longshorej force-pushed the mqtt-source-queue-bp branch 2 times, most recently from 23546ee to 767e4d9 Compare May 16, 2019 17:34

2m force-pushed the mqtt-source-queue-bp branch from 998f067 to 067917f Compare June 19, 2019 12:45

2m added this to the 1.0.3 milestone Jun 19, 2019

2m approved these changes Jun 19, 2019

View reviewed changes

longshorej and others added 6 commits July 1, 2019 09:15

mqtt streaming - introduce a mechanism to use SourceQueue backpressure

e445205

When offering commands to the internal queue, a backpressure strategy is now used to ensure that elements are not dropped. Instead, we wait for consumption downstream before continuing.

Harden test

c7a0ed2

I noticed that a test wasn’t quite right - the client should only complete after we’ve received all elements. We now use “take” to ensure that the stream completes.

Don’t let WatchedActorTerminatedException fail the session streams

a925fe2

We want WatchedActorTerminatedException to complete stream processing as before, but not with a failure. A session shutting down isn’t in itself a failure.

Mima exclude; improve toString

930046b

Scalafmt

3d9c2aa

Rename MiMa excludes file to the latest released version

e096ddb

2m force-pushed the mqtt-source-queue-bp branch from 067917f to e096ddb Compare July 1, 2019 06:16

ennru changed the title ~~mqtt streaming - introduce a mechanism to use SourceQueue backpressure~~ MQTT streaming: SourceQueue backpressure Jul 1, 2019

ennru merged commit ed5770c into akka:master Jul 1, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

MQTT streaming: SourceQueue backpressure #1577

MQTT streaming: SourceQueue backpressure #1577

longshorej commented Mar 12, 2019

longshorej Mar 12, 2019

huntc Mar 13, 2019

huntc commented Mar 13, 2019 •

edited

Loading

huntc Mar 13, 2019

huntc Mar 13, 2019

ennru left a comment

ennru Mar 13, 2019

longshorej commented Mar 19, 2019

longshorej commented Mar 26, 2019

huntc commented Mar 26, 2019

ennru commented Apr 3, 2019

ennru commented Apr 26, 2019

longshorej commented Apr 29, 2019

ennru left a comment

ennru May 6, 2019

longshorej May 16, 2019

ennru May 6, 2019

longshorej May 16, 2019

longshorej commented May 16, 2019 •

edited

Loading

longshorej commented May 16, 2019

longshorej commented May 16, 2019

ennru commented May 17, 2019

ennru commented May 23, 2019

2m left a comment

ennru commented Jun 28, 2019

huntc commented Jun 28, 2019

2m commented Jul 1, 2019

ennru commented Jul 1, 2019

MQTT streaming: SourceQueue backpressure #1577

MQTT streaming: SourceQueue backpressure #1577

Conversation

longshorej commented Mar 12, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

huntc commented Mar 13, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

longshorej commented Mar 19, 2019

longshorej commented Mar 26, 2019

huntc commented Mar 26, 2019

ennru commented Apr 3, 2019

ennru commented Apr 26, 2019

longshorej commented Apr 29, 2019

ennru left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

longshorej commented May 16, 2019 • edited Loading

longshorej commented May 16, 2019

longshorej commented May 16, 2019

ennru commented May 17, 2019

ennru commented May 23, 2019

2m left a comment

Choose a reason for hiding this comment

ennru commented Jun 28, 2019

huntc commented Jun 28, 2019

2m commented Jul 1, 2019

ennru commented Jul 1, 2019

huntc commented Mar 13, 2019 •

edited

Loading

longshorej commented May 16, 2019 •

edited

Loading