Update comparison logic for worker state

[rockwellw]

Addresses #3256 , based on @TomAugspurger 's comment.

[TomAugspurger]

Thanks. Do you think this will actually fix #3256? This being the cause was a shot in the dark.

And do you know if / how WorkerState.__eq__ is defined? I don't see one at a glance. Does implementing __slots__ give us one?

[rockwellw]

This may just address a symptom and not the root problem, but it seems like a good starting point considering the log messages @bnaul was seeing.

[TomAugspurger]

Ok. Do you know how __eq__ is implemented ok WorkerState?

[TomAugspurger]

I think WorkerState.__eq__ may actually be using identity. I would have expected these to be equal, if we were comparing attributes

In [15]: from distributed.scheduler import WorkerState

In [16]: a = WorkerState()

In [17]: b = WorkerState()

In [18]: a == b
Out[18]: False

This may not be a problem if WorkerState instances are expected to be singletons per worker, which I suspect they are. But it does mean that this likely isn't fixing the underlying issue.

[StephanErb]

There is another occurrence of the same identity comparison here

distributed/distributed/scheduler.py

Line 2560 in d747f63

if ts.processing_on is not ws:

To reduce surprises, I guess both should be fixed.

While I have no proof, I could totally imagine that race conditions between add_worker and remove_worker break the single instance assumption.

[bnaul]

I think WorkerState.__eq__ may actually be using identity. I would have expected these to be equal, if we were comparing attributes

@TomAugspurger it looks like identity() has the info that we would want to compare against, do you think something like this might do the trick?

class WorkerState():
    def __eq__(self, value):
        if hasattr(value, 'identity'):
            return self.identity() == value.identity()
        else:
            return False

[mrocklin]

Rather than conditionals I usually do something like ...

return type(self) == type(other) and self.identity() == other.identity()

Also, if we're going to implement __eq__ then we'll probably also have to implement __hash__ (should be more or less the same)

[bnaul]

Rather than conditionals I usually do something like ...

return type(self) == type(other) and self.identity() == other.identity()

Probably not at all important for this case but doesn't this pattern kind of break subclassing? How about isinstance(other, type(self))?

Also, if we're going to implement __eq__ then we'll probably also have to implement __hash__ (should be more or less the same)

Do we currently rely on WorkerStates being hashable? It seems to me like they're mutable so wouldn't we want hash to be None?

[mrocklin]

If they're of different classes then can they be equal? (but yeah, in truth it doesn't matter)

Fair enough on hashability (assuming that it doesn't come up, which it may)

[bnaul]

@mrocklin you were right that hashability is needed so we added that as well and changed __eq__ to rely on __hash__ instead. it seems to me like just name+host is enough to uniquely specify the WorkerState but if you think we should add more attributes we can do that.

@TomAugspurger anecdotally we haven't seen #3256 since we started running with this change, but I wouldn't say we've conclusively proven it's resolved since unfortunately that's a random unpredictable failure. regardless this seems like the "right" check to be performing and should help w/ possible race conditions as pointed out by @StephanErb, so my vote would be to merge this once everyone's happy with it and close #3256 until someone observes that same behavior again.

[TomAugspurger]

Has anyone seen this CI failure before?

=================================== FAILURES ===================================

____________________________ test_pickle_functions _____________________________

    def test_pickle_functions():

        def make_closure():

            value = 1

    

            def f(x):  # closure

                return x + value

    

            return f

    

        def funcs():

            yield make_closure()

            yield (lambda x: x + 1)

            yield partial(add, 1)

    

        for func in funcs():

            wr = weakref.ref(func)

            func2 = loads(dumps(func))

            wr2 = weakref.ref(func2)

            assert func2(1) == func(1)

            del func, func2

            gc.collect()

>           assert wr() is None

E           AssertionError: assert <function test_pickle_functions.<locals>.funcs.<locals>.<lambda> at 0x7f0ab6f535f0> is None

E            +  where <function test_pickle_functions.<locals>.funcs.<locals>.<lambda> at 0x7f0ab6f535f0> = <weakref at 0x7f0a8cc05350; to 'function' at 0x7f0ab6f535f0 (<lambda>)>()

distributed/protocol/tests/test_pickle.py:47: AssertionError`

I've restarted that build.

[mrocklin]

Not recently, but yes, it used to be an intermittent failure.

…

On Tue, Feb 11, 2020 at 5:23 AM Tom Augspurger ***@***.***> wrote: Has anyone seen this CI failure before? =================================== FAILURES =================================== ____________________________ test_pickle_functions _____________________________ def test_pickle_functions(): def make_closure(): value = 1 def f(x): # closure return x + value return f def funcs(): yield make_closure() yield (lambda x: x + 1) yield partial(add, 1) for func in funcs(): wr = weakref.ref(func) func2 = loads(dumps(func)) wr2 = weakref.ref(func2) assert func2(1) == func(1) del func, func2 gc.collect() > assert wr() is None E AssertionError: assert <function test_pickle_functions.<locals>.funcs.<locals>.<lambda> at 0x7f0ab6f535f0> is None E + where <function test_pickle_functions.<locals>.funcs.<locals>.<lambda> at 0x7f0ab6f535f0> = <weakref at 0x7f0a8cc05350; to 'function' at 0x7f0ab6f535f0 (<lambda>)>() distributed/protocol/tests/test_pickle.py:47: AssertionError` I've restarted that build. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3321?email_source=notifications&email_token=AACKZTER3ZA3WJTT5US4FMTRCKRGRA5CNFSM4J2TLRWKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOELMMN5I#issuecomment-584632053>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AACKZTDCXKDCFHYSALANVELRCKRGRANCNFSM4J2TLRWA> .

[TomAugspurger]

OK, thanks. We'll merge this and keep an eye out for that failure again.

Thanks @rockwellw!

[StephanErb]

Sorry for being late to the party. There is a bug in here: It is not guaranteed that two WorkerState objects are actually the same even though there hash value is the same. Hashing is not a bijective function.

I think we need to go the extra route here and check for (self.name, self.host) == (other.name, other.host)

[StephanErb]

@TomAugspurger @bnaul please see above.

While I totally agree that hash collisions are very very unlikely, Dask has a massive install base with sometimes hundreds to thousands of workers per cluster. So as far as we know the once in a million event could happen next Tuesday. 🤷‍♂