Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fixed to notification on DeathWatch watching. #899

Merged
merged 1 commit into from
Apr 25, 2015
Merged

Fixed to notification on DeathWatch watching. #899

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 45 additions & 43 deletions src/core/Akka/Actor/ActorCell.DeathWatch.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ public IActorRef Watch(IActorRef subject)
MaintainAddressTerminatedSubscription(() =>
{
a.Tell(new Watch(a, Self)); // ➡➡➡ NEVER SEND THE SAME SYSTEM MESSAGE OBJECT TO TWO ACTORS
_state = _state.AddWatching(a);
_state = _state.AddWatching(a);
}, a);
}
return a;
Expand All @@ -34,12 +34,12 @@ public IActorRef Watch(IActorRef subject)
public IActorRef Unwatch(IActorRef subject)
{
var a = (IInternalActorRef)subject;
if (! a.Equals(Self) && WatchingContains(a))
if (!a.Equals(Self) && WatchingContains(a))
{
a.Tell(new Unwatch(a, Self));
MaintainAddressTerminatedSubscription(() =>
{
_state = _state.RemoveWatching(a);
_state = _state.RemoveWatching(a);
}, a);
}
_state = _state.RemoveTerminated(a);
Expand All @@ -48,11 +48,11 @@ public IActorRef Unwatch(IActorRef subject)

protected void ReceivedTerminated(Terminated t)
{
if (_state.ContainsTerminated(t.ActorRef))
{
_state = _state.RemoveTerminated(t.ActorRef); // here we know that it is the SAME ref which was put in
ReceiveMessage(t);
}
if (!_state.ContainsTerminated(t.ActorRef))
return;

_state = _state.RemoveTerminated(t.ActorRef); // here we know that it is the SAME ref which was put in
ReceiveMessage(t);
}

/// <summary>
Expand Down Expand Up @@ -87,12 +87,15 @@ public void TerminatedQueuedFor(IActorRef subject)
private bool WatchingContains(IActorRef subject)
{
return _state.ContainsWatching(subject) ||
(subject.Path.Uid != ActorCell.UndefinedUid && _state.ContainsWatching(new UndefinedUidActorRef(subject)));
(subject.Path.Uid != UndefinedUid && _state.ContainsWatching(new UndefinedUidActorRef(subject)));
}

protected void TellWatchersWeDied()
{
var watchedBy = _state.GetWatchedBy();
var watchedBy = _state
.GetWatchedBy()
.ToList();

if (!watchedBy.Any()) return;
try
{
Expand Down Expand Up @@ -124,26 +127,28 @@ private void SendTerminated(bool ifLocal, IActorRef watcher)
{
if (((IActorRefScope)watcher).IsLocal == ifLocal && !watcher.Equals(Parent))
{
((IInternalActorRef)watcher).Tell(new DeathWatchNotification(Self, true, false));
watcher.Tell(new DeathWatchNotification(Self, true, false));
}
}

protected void UnwatchWatchedActors(ActorBase actor)
{
var watching = _state.GetWatching();
if(!watching.Any()) return;
var watching = _state
.GetWatching()
.ToList();

if (!watching.Any()) return;

MaintainAddressTerminatedSubscription(() =>
{
try
{
foreach ( // ➡➡➡ NEVER SEND THE SAME SYSTEM MESSAGE OBJECT TO TWO ACTORS
var watchee in watching.OfType<IInternalActorRef>())
// ➡➡➡ NEVER SEND THE SAME SYSTEM MESSAGE OBJECT TO TWO ACTORS
foreach (var watchee in watching.OfType<IInternalActorRef>())
watchee.Tell(new Unwatch(watchee, Self));
}
finally
{
//_watching = new HashSet<IActorRef>();
//_terminatedQueue = new HashSet<IActorRef>();
_state = _state.ClearWatching();
_state = _state.ClearTerminated();
}
Expand All @@ -157,13 +162,13 @@ protected void AddWatcher(IActorRef watchee, IActorRef watcher)

if (watcheeSelf && !watcherSelf)
{
if(!_state.ContainsWatchedBy(watcher)) MaintainAddressTerminatedSubscription(() =>
{
//_watchedBy.Add(watcher);
_state = _state.AddWatchedBy(watcher);
if(System.Settings.DebugLifecycle) Publish(new Debug(Self.Path.ToString(), Actor.GetType(), string.Format("now watched by {0}", watcher)));
}, watcher);
if (!_state.ContainsWatchedBy(watcher)) MaintainAddressTerminatedSubscription(() =>
{
//_watchedBy.Add(watcher);
_state = _state.AddWatchedBy(watcher);

if (System.Settings.DebugLifecycle) Publish(new Debug(Self.Path.ToString(), Actor.GetType(), string.Format("now watched by {0}", watcher)));
}, watcher);
}
else if (!watcheeSelf && watcherSelf)
{
Expand All @@ -182,13 +187,13 @@ protected void RemWatcher(IActorRef watchee, IActorRef watcher)

if (watcheeSelf && !watcherSelf)
{
if( _state.ContainsWatchedBy(watcher)) MaintainAddressTerminatedSubscription(() =>
if (_state.ContainsWatchedBy(watcher)) MaintainAddressTerminatedSubscription(() =>
{
//_watchedBy.Remove(watcher);
_state = _state.RemoveWatchedBy(watcher);

if (System.Settings.DebugLifecycle) Publish(new Debug(Self.Path.ToString(), Actor.GetType(), string.Format("no longer watched by {0}", watcher)));
} , watcher);
}, watcher);
}
else if (!watcheeSelf && watcherSelf)
{
Expand All @@ -202,27 +207,23 @@ protected void RemWatcher(IActorRef watchee, IActorRef watcher)

protected void AddressTerminated(Address address)
{
var watchedBy = _state.GetWatchedBy();
// cleanup watchedBy since we know they are dead
MaintainAddressTerminatedSubscription(() =>
{
foreach (var a in watchedBy.Where(a => a.Path.Address == address))
foreach (var a in _state.GetWatchedBy().Where(a => a.Path.Address == address))
{
//_watchedBy.Remove(a);
_state = _state.RemoveWatchedBy(a);
}
});

//
watchedBy = _state.GetWatchedBy();

// send DeathWatchNotification to self for all matching subjects
// that are not child with existenceConfirmed = false because we could have been watching a
// non-local ActorRef that had never resolved before the other node went down
// When a parent is watching a child and it terminates due to AddressTerminated
// it is removed by sending DeathWatchNotification with existenceConfirmed = true to support
// immediate creation of child with same name.
foreach (var a in watchedBy.Where(a => a.Path.Address == address))
foreach (var a in _state.GetWatching().Where(a => a.Path.Address == address))
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is the fix

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So was this a bug that got added when we extracted the ActorCell state, or has this been a persistent issue for some time? Only asking because it may explain some of the cluster deathwatch behavior I've observed during multi-node test runs.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like this was introduced about 29 days ago. It affects Cluster DeathWatch directly because a node that has been removed from the cluster can't tell the DeathWatchers from another node it's dead:)

{
Self.Tell(new DeathWatchNotification(a, true /*TODO: childrenRefs.getByRef(a).isDefined*/, true));
}
Expand All @@ -234,15 +235,18 @@ protected void AddressTerminated(Address address)
/// Ends subscription to AddressTerminated if subscribing and the
/// block removes the last non-local ref from watching and watchedBy.
/// </summary>
private void MaintainAddressTerminatedSubscription(Action block, IActorRef change= null)
private void MaintainAddressTerminatedSubscription(Action block, IActorRef change = null)
{
if (IsNonLocal(change))
{
var had = HasNonLocalAddress();
block();
var has = HasNonLocalAddress();
if (had && !has) UnsubscribeAddressTerminated();
else if (!had && has) SubscribeAddressTerminated();

if (had && !has)
UnsubscribeAddressTerminated();
else if (!had && has)
SubscribeAddressTerminated();
}
else
{
Expand All @@ -252,10 +256,11 @@ private void MaintainAddressTerminatedSubscription(Action block, IActorRef chang

private static bool IsNonLocal(IActorRef @ref)
{
if (@ref == null) return true;
if (@ref == null)
return true;

var a = @ref as IInternalActorRef;
if (a != null && !a.IsLocal) return true;
return false;
return a != null && !a.IsLocal;
}

private bool HasNonLocalAddress()
Expand Down Expand Up @@ -298,7 +303,4 @@ public override IActorRefProvider Provider
}
}
}


}

}