Maybe fix celery#4699 (celery#4719)

* Maybe fix celery#4699

* Add unit test for celery#4699

* Fix test on PyPy

* Factor WeakMethod logic and add comment

celery/utils/dispatch/signal.py

@@ -37,6 +37,28 @@ def _make_id(target):  # pragma: no cover
     return id(target)
 
 
+def _boundmethod_safe_weakref(obj):
+    """Get weakref constructor appropriate for `obj`.  `obj` may be a bound method.
+
+    Bound method objects must be special-cased because they're usually garbage
+    collected immediately, even if the instance they're bound to persists.
+
+    Returns:
+        a (weakref constructor, main object) tuple. `weakref constructor` is
+        either :class:`weakref.ref` or :class:`weakref.WeakMethod`.  `main
+        object` is the instance that `obj` is bound to if it is a bound method;
+        otherwise `main object` is simply `obj.
+    """
+    try:
+        obj.__func__
+        obj.__self__
+        # Bound method
+        return WeakMethod, obj.__self__
+    except AttributeError:
+        # Not a bound method
+        return weakref.ref, obj
+
+
 def _make_lookup_key(receiver, sender, dispatch_uid):
     if dispatch_uid:
         return (dispatch_uid, _make_id(sender))
@@ -183,8 +205,7 @@ def _connect_signal(self, receiver, sender, weak, dispatch_uid):
         lookup_key = _make_lookup_key(receiver, sender, dispatch_uid)
 
         if weak:
-            ref = weakref.ref
-            receiver_object = receiver
+            ref, receiver_object = _boundmethod_safe_weakref(receiver)
             if PY3:
                 receiver = ref(receiver)
                 weakref.finalize(receiver_object, self._remove_receiver)

t/unit/utils/test_dispatcher.py

@@ -173,3 +173,14 @@ def test_retry_with_dispatch_uid(self):
         assert a_signal.receivers[0][0][0] == uid
         a_signal.disconnect(receiver_1_arg, sender=self, dispatch_uid=uid)
         self._testIsClean(a_signal)
+
+    def test_boundmethod(self):
+        a = Callable()
+        a_signal.connect(a.a, sender=self)
+        expected = [(a.a, 'test')]
+        garbage_collect()
+        result = a_signal.send(sender=self, val='test')
+        assert result == expected
+        del a, result, expected
+        garbage_collect()
+        self._testIsClean(a_signal)