=act #3623 unsubscribe when actors terminate, instead of `isTerminated` #2011

ktoso · 2014-02-14T00:49:01Z

Moved to removing actors proactively when they are terminated instead of checking isTerminated during publish.

Assumed having an init method on the EventBus is fine, as it is mutable already anyway, as @patriknw points out in the task.

Related question: should this change be documented in the docs too? I figured that it's more of an internals change so maybe not?

ktoso · 2014-02-14T00:53:38Z

akka-actor/src/main/scala/akka/event/EventStream.scala

@@ -59,4 +61,13 @@ class EventStream(private val debug: Boolean = false) extends LoggingBus with Su
    if (debug) publish(Logging.Debug(simpleName(this), this.getClass, "unsubscribing " + subscriber + " from all channels"))
  }

+  def initTerminatedUnsubscriber(unsubscriber: ActorRef) {
+    this.unsubscriber = Some(unsubscriber)
+    // todo what do we do about already subscribed ones?


ideas:

make subscribers set accessible here and add to unsubscriber? not sure if needed, since init is run so early - during _start

don't care?

How about changing unsubscriber to an AtomicReference[Either[Seq[ActorRef], ActorRef] which starts out at Left(Nil), and here we swap it out against Right(unsubscriber), registering all the early-comers.

sounds neat, will do!

ktoso · 2014-02-14T00:57:07Z

I had some unanswered questions during impl, brought them up in comments.
Thanks a lot in advance for any tips :-)

ktoso · 2014-02-14T09:06:21Z

Investigating - was green locally, will check why failed on jenkins.

ktoso · 2014-02-16T13:44:42Z

PLS REBUILD/pr-validator-per-commit@7da363c

akka-ci · 2014-02-16T13:50:36Z

(kitty-note-to-self: ignore 35196996)
🐱 Roger! Rebuilding pr-validator-per-commit for 7da363c. 🚨

ktoso · 2014-02-16T14:04:56Z

PLS REBUILD/pr-validator-per-commit@5adef0d
is what I meant to rebuild (didn't hear back about result).

// not sure why akka-ci didn't comment on passed build, was green: https://jenkins.akka.io:8498/view/PR%20Validators/job/pr-validator-per-commit/1903/

akka-ci · 2014-02-16T14:10:38Z

(kitty-note-to-self: ignore 35197449)
🐱 Roger! Rebuilding pr-validator-per-commit for 5adef0d. 🚨

ktoso · 2014-02-19T14:13:41Z

Feedback addressed with ktoso@4a12449 .
If cool I'll squash.

ktoso · 2014-02-19T23:26:27Z

Updated registerWithUnsubscriber to be CAS and squashed commits.
Should be good now :)

rkuhn · 2014-02-20T08:12:37Z

akka-actor/src/main/scala/akka/event/EventStream.scala

@@ -59,4 +64,34 @@ class EventStream(private val debug: Boolean = false) extends LoggingBus with Su
    if (debug) publish(Logging.Debug(simpleName(this), this.getClass, "unsubscribing " + subscriber + " from all channels"))
  }

+  @tailrec final def initTerminatedUnsubscriber(unsubscriber: ActorRef): Boolean = {


this method should be private[akka]

rkuhn · 2014-02-20T09:05:24Z

LGTM after fixing the last nitpicks. Then awaiting team review (but this won’t make 2.3.0 anymore, slated for 2.3.1).

patriknw · 2014-02-20T09:09:54Z

akka-actor-tests/src/test/scala/akka/event/EventStreamSpec.scala

+        val a1, a2 = TestProbe()
+        val tm = new A
+
+        val target = system.actorOf(Props(new Actor {


is there a reason to use system instead of sys here?

should have been sys, yes.

patriknw · 2014-02-20T09:16:23Z

akka-actor/src/main/scala/akka/actor/ActorSystem.scala

@@ -642,6 +643,10 @@ private[akka] class ActorSystemImpl(val name: String, applicationConfig: Config,
    guardian.stop()
  }

+  def startEventStreamTerminationWatcher(eventStream: EventStream) {
+    systemActorOf(Props(classOf[EventStreamTerminatedUnsubscriber], eventStream, DebugEventStream), "eventStreamTerminatedUnsubscriber")


the name EventStreamTerminatedUnsubscriber feels a bit clunky, would EventStreamUnsubscriber be enough?

true, might have even given the wrong idea about when it does it's job.
EventStreamUnsubscriber sounds good, changed!

patriknw · 2014-02-20T16:41:47Z

akka-actor/src/main/scala/akka/event/EventStream.scala

  }

  override def subscribe(subscriber: ActorRef, channel: Class[_]): Boolean = {
    if (subscriber eq null) throw new IllegalArgumentException("subscriber is null")
    if (debug) publish(Logging.Debug(simpleName(this), this.getClass, "subscribing " + subscriber + " to channel " + channel))
+    registerWithUnsubscriber(subscriber)


should we also unwatch when unsubscribe?

Yes, addressed in ktoso@8ead935#diff-a5669e25b3993c6bf5ae1e262bff557bR64

ktoso · 2014-02-24T00:12:34Z

Over the weekend I worked on killing isTerminated in ActorClassification, the result is here:
ktoso@5a2237f

Addressed also similar issues that we had here, about when to stop watching the subscriber etc. I hope I didn't miss anything this time (hints from implementing EventStream certainly helped). If seems cool, I can make it part of this pr or make a separate one "just about ActorClassification".

Q: Not sure if we can break ActorEventBus API (to require an actorSystem - for spinning up the unsubscriber system actor) in 2.3.1? If not... then init method on ActorClassification, similar to what we did on EventStream here?

rkuhn · 2014-02-24T07:51:07Z

Hmm, I have not fully reviewed the ActorClassification commit yet, but either way it would not be binary backwards compatible, so any such change cannot go into 2.3.1 but must wait for 2.4.0. It is probably best to create a separate pull request for that one.

ktoso · 2014-02-24T22:36:18Z

Roger that about separate PR. Squashed this one.

jozic · 2014-02-27T18:17:01Z

akka-actor/src/main/scala/akka/event/EventStreamUnsubscriber.scala

+
+  def receive = {
+    case Register(actor) ⇒
+      if (debug) eventStream.publish(Logging.Debug(simpleName(getClass), getClass, s"watching $actor in order to unsubscribe from EventStream when it terminates"))


there will be a debug message from DeathWatch saying that the actor is watched by unsubscriber.
Isn't it enough?

That message depends on akka.actor.debug.lifecycle, which should be orthogonal to the setting used here (akka.actor.debug.eventstream).

right, but I meant they are all debug messages at the end, and for that purpose one is probably enough

But not all debug messages are created equal: the Akka philosophy is to enable them separately because just switching them all on is rarely useful.

ok, i see the point, I guess more is better than less here

rkuhn · 2014-02-28T08:56:01Z

@ktoso I’ll get back to this next week after 2.3.0 is out.

ktoso · 2014-02-28T09:13:40Z

2.3.0, right - was expecting just that, thanks! :-)

patriknw · 2014-03-04T09:38:47Z

akka-actor/src/main/scala/akka/event/EventStream.scala

+   * INTERNAL API
+   */
+  private[akka] def unregisterIfNoMoreSubscribedChannels(unsubscribeHadDiff: Boolean, subscriber: ActorRef, channel: Class[_]) {
+    if (!unsubscribeHadDiff || !hasSubscriptions(subscriber)) {


I would like to have a second opinion about this hasSubscriptions. Is it to costly? Is it worth it? If we skip it the actor ref would remain until the subscriber terminates. //cc @rkuhn @bantonsson @drewhk

Thanks for keeping an eye out: you are right in that it has a potentially high cost. Not doing this means that we can leak state in the unsubscriber (if these actors do not terminate soon-ish); imagine a use-case where you keep starting actors which use the event bus only during initialization and which are long-running. Perhaps we need to do low-frequency pruning instead of doing it upon each unsubscription.

yes, it would be a leak, but only some references to objects that already exists in the jvm

Great that we re-reviewed it then. I'll work on an "pruning" approach.

patriknw · 2014-04-15T09:51:44Z

What is the status here? Should we put the two related changes into one pull request and re-start reviewing?

ktoso · 2014-04-15T09:52:44Z

Yeah, I'd love to revive these. Let me bring them up to date with master and let's have a re-look?

rkuhn · 2014-04-15T12:08:53Z

yes, sounds like a plan!

ktoso · 2014-04-15T15:25:20Z

Rebased, squashed, cleaned-up and went through the cases again. I hope I didn't miss anything - please review again @rkuhn, @patriknw. Thanks!

2m · 2014-04-15T15:56:22Z

akka-actor/src/main/scala/akka/event/EventStreamUnsubscriber.scala

+      if (debug) eventStream.publish(Logging.Debug(simpleName(getClass), getClass, s"watching $actor in order to unsubscribe from EventStream when it terminates"))
+      context watch actor
+
+    case UnregisterIfNoMoreSubscribedChannels(actor) if eventStream.hasSubscriptions(actor) ⇒


What kind of sorcery is this with an empty block? Should this guard be negated and moved to the case below?

I find those sometimes more readable than !event...., but yeah could be made an if in the case bellow... What do the others think?

Okay. Then maybe a comment like "// do nothing" after an arrow would be nice. Otherwise it may induce some head scratching after a month or two when someone gets back to this code.

yes, the convention is to put a // do nothing comment in the obvious place

okey, done :)

@patriknw

* Moved to removing actors proactively when they are terminated instead of checking `isTerminated` during publish. * Subscribers which have registered before initializing the unsubscriber will be aggregated in a Seq until one is registered and then it will take responsibility of unregistering them on termination. * Initialization of the unsubscriber can only be run once - attempting to initialize the event stream with another unsubscriber will fail, and init will return false. * Assumed having an init (mutable) method on the `EventBus` is fine, as it has such methods already and @patriknw's comment in the task for this. * since we must check if the subscriber has any subscribed channels left we had to expose this detail from SubchannelClassification via `hasSubscriptions`. Increases cost of ubsubscribe(actor, channel) a bit. * Evacuated the expensive `hasSubscription` call out of eventstream's `unsubscribe` call, and instead making the Unsubscriber check this before it stops watching an actor. If in the mean time the same actor got subscribed, there will be a new Subscribe message emited - so we're good on that side. Also, if it would terminate before the unsubscriber gets the Register message it will call `watch(actor)` on a dead actor, which results in getting Terminated for it, thus we'll stop watching it from the Unsubscriber as expected. Final squash and small cleanup. Please review again;

ktoso · 2014-04-16T09:53:29Z

Closing and will re-open as merged req to remove isTerminated from both evenbus + eventstream

ktoso reviewed Feb 14, 2014
View reviewed changes

ktoso added tested and removed needs-attention labels Feb 20, 2014

rkuhn reviewed Feb 20, 2014
View reviewed changes

patriknw reviewed Feb 20, 2014
View reviewed changes

ktoso added the reviewed label Feb 20, 2014

patriknw reviewed Feb 20, 2014
View reviewed changes

ktoso removed the tested label Feb 20, 2014

patriknw reviewed Feb 20, 2014
View reviewed changes

ktoso added the tested label Feb 23, 2014

ktoso removed the tested label Feb 24, 2014

ktoso mentioned this pull request Feb 24, 2014

!act,doc #3893 Removed isTerminated checks from ActorClassification #2037

Closed

ktoso added the tested label Feb 24, 2014

jozic reviewed Feb 27, 2014
View reviewed changes

patriknw reviewed Mar 4, 2014
View reviewed changes

ktoso added tested and removed tested labels Mar 21, 2014

viktorklang mentioned this pull request Feb 26, 2014

ActorClassification should unsubscribe terminated subscribers #13893

Closed

ktoso mentioned this pull request Apr 15, 2014

Remove deprecated isTerminated #13922

Closed

ktoso removed the tested label Apr 15, 2014

ktoso self-assigned this Apr 15, 2014

2m reviewed Apr 15, 2014
View reviewed changes

ktoso added the tested label Apr 16, 2014

ktoso closed this Apr 16, 2014

ktoso mentioned this pull request Apr 16, 2014

!act,doc Removing isTerminated from EventStream and ActorClassification #15024

Merged

ktoso deleted the wip-3623-eventstream-and-terminated-listeners-ktoso branch February 22, 2015 16:25

=act #3623 unsubscribe when actors terminate, instead of isTerminated #2011

=act #3623 unsubscribe when actors terminate, instead of isTerminated #2011

Conversation

ktoso commented Feb 14, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ktoso commented Feb 14, 2014

ktoso commented Feb 14, 2014

ktoso commented Feb 16, 2014

akka-ci commented Feb 16, 2014

ktoso commented Feb 16, 2014

akka-ci commented Feb 16, 2014

ktoso commented Feb 19, 2014

ktoso commented Feb 19, 2014

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rkuhn commented Feb 20, 2014

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

ktoso commented Feb 24, 2014

rkuhn commented Feb 24, 2014

ktoso commented Feb 24, 2014

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

rkuhn commented Feb 28, 2014

ktoso commented Feb 28, 2014

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 15, 2014

ktoso commented Apr 15, 2014

rkuhn commented Apr 15, 2014

ktoso commented Apr 15, 2014

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

ktoso commented Apr 16, 2014

=act #3623 unsubscribe when actors terminate, instead of `isTerminated` #2011

=act #3623 unsubscribe when actors terminate, instead of `isTerminated` #2011