When used for recovery, pin query to journal dispatcher #880

ignasi35 · 2021-04-16T08:56:33Z

Recovering an entity should not be clogged by the load on the read-side.

When using a query for recovery, it should run on the journal dispatcher no the read dispatcher.

patriknw

agreed, good to run those on the journal dispatcher

ignasi35 · 2021-04-16T09:39:18Z

Note this targets release-0.x. Needs forward-port to master.

ignasi35 · 2021-04-16T10:20:16Z

Travis can't download the Jabba installer script. 🤦🏼‍♂️
I've raised #883 to install the JDKs in travis using apt.

patriknw · 2021-04-16T10:27:28Z

core/src/main/scala/akka/persistence/cassandra/journal/CassandraRecovery.scala

@@ -88,6 +92,8 @@ trait CassandraRecovery extends CassandraTagRecovery with TaggedPreparedStatemen
          someReadConsistency,
          someReadRetryPolicy,
          extractor = Extractors.persistentRepr(eventDeserializer, serialization))
+        // run the query on the journal dispatcher (not the queries dispatcher)
+        .withAttributes(ActorAttributes.dispatcher(sessionSettings.pluginDispatcher))


We have to be sure that this isn't just adding an outer asynchronous boundary. Might be more clear and safe to pass in the dispatcher as parameter to queries. eventsByPersistenceId?

ignasi35 · 2021-04-16T12:03:56Z

I've had to rebase this on top of #881 and target the branch of #881 to get travis to work.

ignasi35 · 2021-04-16T16:39:49Z

The job that runs tests on JDK11 seems to be slow enough that the 50min timeout in travis fails the job.

core/src/main/scala/akka/persistence/cassandra/CassandraPluginConfig.scala

johanandren · 2021-04-19T08:40:43Z

core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala

@@ -569,7 +569,8 @@ class CassandraReadJournal(system: ExtendedActorSystem, cfg: Config)
      queryPluginConfig.fetchSize,
      None,
      s"currentEventsByPersistenceId-$persistenceId",
-      extractor = Extractors.persistentRepr(eventsByPersistenceIdDeserializer, serialization))
+      extractor = Extractors.persistentRepr(eventsByPersistenceIdDeserializer, serialization),
+      dispatcher = queryPluginConfig.pluginDispatcher)


Since the user part of the stream run through the public API queries will run on the default dispatcher this means an async boundary is introduced in every app that uses the query side, is that really something we want to do?

I'm not sure I understand the previous comment.

@johanandren this is not changing that, the dispatchers parameter is delegated down to https://github.com/akka/akka-persistence-cassandra/pull/880/files#diff-279ff4d092baf64b28404c59938d5a8f4053dffe99b8679826391980f66a6f24R647

We must have a separate dispatcher at the inner stage because some operations are blocking.

Yes, so if we run that inner EventsByPersistenceIdStage and mapAsync stage on an internal dispatcher, the user flow running on the default dispatcher will always introduce an async boundary since different dispatchers.

If that's unavoidable I guess that is fine, just that it is understood that is the decision.

That's intended, because otherwise we would propagate the blocking out to user responsibility (and it would end up on the default-dispatcher). That doesn't change by this PR, or did I miss something?

See #870

That same problem goes for this entire PR though. So it wouldn't be more "blindly" than the current changes?

I tried to keep things as small as possible.

I only extracted a hardcoded dispatcher name and set the appropriate value in 5 places (CassandraJournal, EventsByTagMigration and CassandreRecovery (x3)).

I got the impression you were suggesting a full review of the whole codebase. It's different orders of magnitude.

we should trust that Akka Streams does the right thing so additional automated tests shouldn't be needed here, but since we know there were problems with setting the attributes a manual verification with println is what I was suggesting. I can do that before approving this PR.

Actually after reviewing this a bit more in detail I see how tricky it is because of the futureSource bug, setting the dispatcher needs to be repeated in lots of places.

Perhaps it's even preferrable to set the dispatcher where you'd expect to set it (outermost) so that this is fixed once that is fixed in Akka?

johanandren · 2021-04-20T11:53:42Z

core/src/main/scala/akka/persistence/cassandra/journal/CassandraJournal.scala

-        extractor = Extractors.sequenceNumber(eventDeserializer, serialization))
+        extractor = Extractors.sequenceNumber(eventDeserializer, serialization),
+        // run the query on the journal dispatcher (not the queries dispatcher)
+        dispatcher = sessionSettings.pluginDispatcher)
      .map(_.sequenceNr)
      .runWith(Sink.headOption)


Here the .map and sink is running on default dispatcher, so one async boundary added that we probably do not want, since it is an entirely internal stream

I just did the tests @patriknw mentioned above and came back to comment this exact thing:

the (for example) mapAsync in

akka-persistence-cassandra/core/src/main/scala/akka/persistence/cassandra/journal/CassandraRecovery.scala

Line 129 in dc79aed

.mapAsync(1) { t =>

runs in the default dispatcher

I did check that all instances of EventsByPersistenceIdStage run on the assigned dispatcher, though.

johanandren · 2021-04-20T11:55:29Z

core/src/main/scala/akka/persistence/cassandra/journal/CassandraRecovery.scala

-              extractor = Extractors.taggedPersistentRepr(eventDeserializer, serialization))
+              extractor = Extractors.taggedPersistentRepr(eventDeserializer, serialization),
+              // run the query on the journal dispatcher (not the queries dispatcher)
+              dispatcher = sessionSettings.pluginDispatcher)
            .mapAsync(1)(sendMissingTagWrite(tp, tagWrites.get))


here mapAsync and the "outer" map and runForeach runs on default dispatcher, we probably do not want that since it is an entirely internal stream.

This is one case where the bug in the futureSource operator does not propagate attributes like it should means the dispatcher needs to be set both on the complete "inner" stream and the "outer" stream.

This one can be very important. Cause a performance regression. We should fix that here or in follow up before releasing.

johanandren · 2021-04-20T12:00:14Z

core/src/main/scala/akka/persistence/cassandra/query/scaladsl/CassandraReadJournal.scala

@@ -569,7 +569,8 @@ class CassandraReadJournal(system: ExtendedActorSystem, cfg: Config)
      queryPluginConfig.fetchSize,
      None,
      s"currentEventsByPersistenceId-$persistenceId",
-      extractor = Extractors.persistentRepr(eventsByPersistenceIdDeserializer, serialization))
+      extractor = Extractors.persistentRepr(eventsByPersistenceIdDeserializer, serialization),
+      dispatcher = queryPluginConfig.pluginDispatcher)


Actually after reviewing this a bit more in detail I see how tricky it is because of the futureSource bug, setting the dispatcher needs to be repeated in lots of places.

Perhaps it's even preferrable to set the dispatcher where you'd expect to set it (outermost) so that this is fixed once that is fixed in Akka?

patriknw · 2021-04-20T12:03:49Z

Actually after reviewing this a bit more in detail I see how tricky it is because of the futureSource bug, setting the dispatcher needs to be repeated in lots of places.

Ok, we have an urgent need to fix this now and release. So let's follow up on the outer dispatcher things.

Created issue #886 for follow up.

johanandren

LGTM, given that we follow up the async boundary introduction, especially the completely internal streams for recovery etc that could live entirely on the same dispatcher/stream island and now has gotten an async boundary/two actors.

patriknw · 2021-04-20T12:57:32Z

I agree, for the internal usage, recovery, we have to fix it before releasing. Could cause performance regression otherwise.

ignasi35 · 2021-04-20T13:37:42Z

I'll forward port this first PR to master. #886 can then start on master or release-0.x.

patriknw reviewed Apr 16, 2021

View reviewed changes

This comment has been minimized.

Sign in to view

ignasi35 marked this pull request as ready for review April 16, 2021 09:38

patriknw reviewed Apr 16, 2021

View reviewed changes

ignasi35 force-pushed the pin-query-to-journal-dispatcher branch from cc4abdb to 12bbf34 Compare April 16, 2021 12:02

ignasi35 changed the base branch from release-0.x to inherit-scala-travis April 16, 2021 12:02

This comment has been minimized.

Sign in to view

johanandren reviewed Apr 19, 2021

View reviewed changes

ignasi35 added 5 commits April 20, 2021 11:34

When used for recovery, pin query to journal dispatcher

d58c39c

scalafmt

0ea1249

More queries used on write-side need journal dispatcher

1eff0b0

Pass the dispatcher name as argument

dc79aed

Remove leftover

d76e066

ignasi35 force-pushed the pin-query-to-journal-dispatcher branch from 2343699 to d76e066 Compare April 20, 2021 09:36

ignasi35 changed the base branch from inherit-scala-travis to migrate-installer-to-apt April 20, 2021 09:37

johanandren reviewed Apr 20, 2021

View reviewed changes

patriknw mentioned this pull request Apr 20, 2021

Set dispatcher attribute at the outer stages #886

Open

johanandren approved these changes Apr 20, 2021

View reviewed changes

ignasi35 merged commit 30e08e0 into migrate-installer-to-apt Apr 20, 2021

ignasi35 deleted the pin-query-to-journal-dispatcher branch April 20, 2021 13:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

When used for recovery, pin query to journal dispatcher #880

When used for recovery, pin query to journal dispatcher #880

ignasi35 commented Apr 16, 2021

patriknw left a comment

This comment has been minimized.

ignasi35 commented Apr 16, 2021

ignasi35 commented Apr 16, 2021 •

edited

Loading

patriknw Apr 16, 2021

ignasi35 commented Apr 16, 2021

ignasi35 commented Apr 16, 2021

This comment has been minimized.

johanandren Apr 19, 2021

ignasi35 Apr 19, 2021

patriknw Apr 19, 2021

johanandren Apr 19, 2021

patriknw Apr 19, 2021

johanandren Apr 20, 2021

ignasi35 Apr 20, 2021

ignasi35 Apr 20, 2021

patriknw Apr 20, 2021

johanandren Apr 20, 2021

johanandren Apr 20, 2021

ignasi35 Apr 20, 2021

johanandren Apr 20, 2021

patriknw Apr 20, 2021

johanandren Apr 20, 2021

patriknw commented Apr 20, 2021 •

edited

Loading

johanandren left a comment

patriknw commented Apr 20, 2021

ignasi35 commented Apr 20, 2021

When used for recovery, pin query to journal dispatcher #880

When used for recovery, pin query to journal dispatcher #880

Conversation

ignasi35 commented Apr 16, 2021

patriknw left a comment

Choose a reason for hiding this comment

This comment has been minimized.

ignasi35 commented Apr 16, 2021

ignasi35 commented Apr 16, 2021 • edited Loading

Choose a reason for hiding this comment

ignasi35 commented Apr 16, 2021

ignasi35 commented Apr 16, 2021

This comment has been minimized.

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

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

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

patriknw commented Apr 20, 2021 • edited Loading

johanandren left a comment

Choose a reason for hiding this comment

patriknw commented Apr 20, 2021

ignasi35 commented Apr 20, 2021

ignasi35 commented Apr 16, 2021 •

edited

Loading

patriknw commented Apr 20, 2021 •

edited

Loading