We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Continuous memory leak
There is a memory leak in the parent process of Celery's worker.
It is not a child process executing a task.
It happens suddenly every few days.
Unless you stop Celery, it consumes server memory in tens of hours.

This problem happens at least in Celery 4.1, and it also occurs in Celery 4.2.
Celery is running on Ubuntu 16 and brokers use RabbitMQ.

![memory](https://user-images.githubusercontent.com/280983/41805959-fcdd7700-76ed-11e8-9e02-db322818e924.jpeg)



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `celery/events/receiver.py`
Content:
```
1 """Event receiver implementation."""
2 from __future__ import absolute_import, unicode_literals
3 
4 import time
5 from operator import itemgetter
6 
7 from kombu import Queue
8 from kombu.connection import maybe_channel
9 from kombu.mixins import ConsumerMixin
10 
11 from celery import uuid
12 from celery.app import app_or_default
13 from celery.utils.time import adjust_timestamp
14 
15 from .event import get_exchange
16 
17 __all__ = ('EventReceiver',)
18 
19 CLIENT_CLOCK_SKEW = -1
20 
21 _TZGETTER = itemgetter('utcoffset', 'timestamp')
22 
23 
24 class EventReceiver(ConsumerMixin):
25     """Capture events.
26 
27     Arguments:
28         connection (kombu.Connection): Connection to the broker.
29         handlers (Mapping[Callable]): Event handlers.
30             This is  a map of event type names and their handlers.
31             The special handler `"*"` captures all events that don't have a
32             handler.
33     """
34 
35     app = None
36 
37     def __init__(self, channel, handlers=None, routing_key='#',
38                  node_id=None, app=None, queue_prefix=None,
39                  accept=None, queue_ttl=None, queue_expires=None):
40         self.app = app_or_default(app or self.app)
41         self.channel = maybe_channel(channel)
42         self.handlers = {} if handlers is None else handlers
43         self.routing_key = routing_key
44         self.node_id = node_id or uuid()
45         self.queue_prefix = queue_prefix or self.app.conf.event_queue_prefix
46         self.exchange = get_exchange(
47             self.connection or self.app.connection_for_write(),
48             name=self.app.conf.event_exchange)
49         if queue_ttl is None:
50             queue_ttl = self.app.conf.event_queue_ttl
51         if queue_expires is None:
52             queue_expires = self.app.conf.event_queue_expires
53         self.queue = Queue(
54             '.'.join([self.queue_prefix, self.node_id]),
55             exchange=self.exchange,
56             routing_key=self.routing_key,
57             auto_delete=True, durable=False,
58             message_ttl=queue_ttl,
59             expires=queue_expires,
60         )
61         self.clock = self.app.clock
62         self.adjust_clock = self.clock.adjust
63         self.forward_clock = self.clock.forward
64         if accept is None:
65             accept = {self.app.conf.event_serializer, 'json'}
66         self.accept = accept
67 
68     def process(self, type, event):
69         """Process event by dispatching to configured handler."""
70         handler = self.handlers.get(type) or self.handlers.get('*')
71         handler and handler(event)
72 
73     def get_consumers(self, Consumer, channel):
74         return [Consumer(queues=[self.queue],
75                          callbacks=[self._receive], no_ack=True,
76                          accept=self.accept)]
77 
78     def on_consume_ready(self, connection, channel, consumers,
79                          wakeup=True, **kwargs):
80         if wakeup:
81             self.wakeup_workers(channel=channel)
82 
83     def itercapture(self, limit=None, timeout=None, wakeup=True):
84         return self.consume(limit=limit, timeout=timeout, wakeup=wakeup)
85 
86     def capture(self, limit=None, timeout=None, wakeup=True):
87         """Open up a consumer capturing events.
88 
89         This has to run in the main process, and it will never stop
90         unless :attr:`EventDispatcher.should_stop` is set to True, or
91         forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`.
92         """
93         return list(self.consume(limit=limit, timeout=timeout, wakeup=wakeup))
94 
95     def wakeup_workers(self, channel=None):
96         self.app.control.broadcast('heartbeat',
97                                    connection=self.connection,
98                                    channel=channel)
99 
100     def event_from_message(self, body, localize=True,
101                            now=time.time, tzfields=_TZGETTER,
102                            adjust_timestamp=adjust_timestamp,
103                            CLIENT_CLOCK_SKEW=CLIENT_CLOCK_SKEW):
104         type = body['type']
105         if type == 'task-sent':
106             # clients never sync so cannot use their clock value
107             _c = body['clock'] = (self.clock.value or 1) + CLIENT_CLOCK_SKEW
108             self.adjust_clock(_c)
109         else:
110             try:
111                 clock = body['clock']
112             except KeyError:
113                 body['clock'] = self.forward_clock()
114             else:
115                 self.adjust_clock(clock)
116 
117         if localize:
118             try:
119                 offset, timestamp = tzfields(body)
120             except KeyError:
121                 pass
122             else:
123                 body['timestamp'] = adjust_timestamp(timestamp, offset)
124         body['local_received'] = now()
125         return type, body
126 
127     def _receive(self, body, message, list=list, isinstance=isinstance):
128         if isinstance(body, list):  # celery 4.0: List of events
129             process, from_message = self.process, self.event_from_message
130             [process(*from_message(event)) for event in body]
131         else:
132             self.process(*self.event_from_message(body))
133 
134     @property
135     def connection(self):
136         return self.channel.connection.client if self.channel else None
137
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/celery/events/receiver.py b/celery/events/receiver.py
--- a/celery/events/receiver.py
+++ b/celery/events/receiver.py
@@ -90,7 +90,8 @@
         unless :attr:`EventDispatcher.should_stop` is set to True, or
         forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`.
         """
-        return list(self.consume(limit=limit, timeout=timeout, wakeup=wakeup))
+        for _ in self.consume(limit=limit, timeout=timeout, wakeup=wakeup):
+            pass
 
     def wakeup_workers(self, channel=None):
         self.app.control.broadcast('heartbeat',

{"golden_diff": "diff --git a/celery/events/receiver.py b/celery/events/receiver.py\n--- a/celery/events/receiver.py\n+++ b/celery/events/receiver.py\n@@ -90,7 +90,8 @@\n         unless :attr:`EventDispatcher.should_stop` is set to True, or\n         forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`.\n         \"\"\"\n-        return list(self.consume(limit=limit, timeout=timeout, wakeup=wakeup))\n+        for _ in self.consume(limit=limit, timeout=timeout, wakeup=wakeup):\n+            pass\n \n     def wakeup_workers(self, channel=None):\n         self.app.control.broadcast('heartbeat',\n", "issue": "Continuous memory leak\nThere is a memory leak in the parent process of Celery's worker.\nIt is not a child process executing a task.\nIt happens suddenly every few days.\nUnless you stop Celery, it consumes server memory in tens of hours.\n\nThis problem happens at least in Celery 4.1, and it also occurs in Celery 4.2.\nCelery is running on Ubuntu 16 and brokers use RabbitMQ.\n\n![memory](https://user-images.githubusercontent.com/280983/41805959-fcdd7700-76ed-11e8-9e02-db322818e924.jpeg)\n\n\n", "before_files": [{"content": "\"\"\"Event receiver implementation.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport time\nfrom operator import itemgetter\n\nfrom kombu import Queue\nfrom kombu.connection import maybe_channel\nfrom kombu.mixins import ConsumerMixin\n\nfrom celery import uuid\nfrom celery.app import app_or_default\nfrom celery.utils.time import adjust_timestamp\n\nfrom .event import get_exchange\n\n__all__ = ('EventReceiver',)\n\nCLIENT_CLOCK_SKEW = -1\n\n_TZGETTER = itemgetter('utcoffset', 'timestamp')\n\n\nclass EventReceiver(ConsumerMixin):\n    \"\"\"Capture events.\n\n    Arguments:\n        connection (kombu.Connection): Connection to the broker.\n        handlers (Mapping[Callable]): Event handlers.\n            This is  a map of event type names and their handlers.\n            The special handler `\"*\"` captures all events that don't have a\n            handler.\n    \"\"\"\n\n    app = None\n\n    def __init__(self, channel, handlers=None, routing_key='#',\n                 node_id=None, app=None, queue_prefix=None,\n                 accept=None, queue_ttl=None, queue_expires=None):\n        self.app = app_or_default(app or self.app)\n        self.channel = maybe_channel(channel)\n        self.handlers = {} if handlers is None else handlers\n        self.routing_key = routing_key\n        self.node_id = node_id or uuid()\n        self.queue_prefix = queue_prefix or self.app.conf.event_queue_prefix\n        self.exchange = get_exchange(\n            self.connection or self.app.connection_for_write(),\n            name=self.app.conf.event_exchange)\n        if queue_ttl is None:\n            queue_ttl = self.app.conf.event_queue_ttl\n        if queue_expires is None:\n            queue_expires = self.app.conf.event_queue_expires\n        self.queue = Queue(\n            '.'.join([self.queue_prefix, self.node_id]),\n            exchange=self.exchange,\n            routing_key=self.routing_key,\n            auto_delete=True, durable=False,\n            message_ttl=queue_ttl,\n            expires=queue_expires,\n        )\n        self.clock = self.app.clock\n        self.adjust_clock = self.clock.adjust\n        self.forward_clock = self.clock.forward\n        if accept is None:\n            accept = {self.app.conf.event_serializer, 'json'}\n        self.accept = accept\n\n    def process(self, type, event):\n        \"\"\"Process event by dispatching to configured handler.\"\"\"\n        handler = self.handlers.get(type) or self.handlers.get('*')\n        handler and handler(event)\n\n    def get_consumers(self, Consumer, channel):\n        return [Consumer(queues=[self.queue],\n                         callbacks=[self._receive], no_ack=True,\n                         accept=self.accept)]\n\n    def on_consume_ready(self, connection, channel, consumers,\n                         wakeup=True, **kwargs):\n        if wakeup:\n            self.wakeup_workers(channel=channel)\n\n    def itercapture(self, limit=None, timeout=None, wakeup=True):\n        return self.consume(limit=limit, timeout=timeout, wakeup=wakeup)\n\n    def capture(self, limit=None, timeout=None, wakeup=True):\n        \"\"\"Open up a consumer capturing events.\n\n        This has to run in the main process, and it will never stop\n        unless :attr:`EventDispatcher.should_stop` is set to True, or\n        forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`.\n        \"\"\"\n        return list(self.consume(limit=limit, timeout=timeout, wakeup=wakeup))\n\n    def wakeup_workers(self, channel=None):\n        self.app.control.broadcast('heartbeat',\n                                   connection=self.connection,\n                                   channel=channel)\n\n    def event_from_message(self, body, localize=True,\n                           now=time.time, tzfields=_TZGETTER,\n                           adjust_timestamp=adjust_timestamp,\n                           CLIENT_CLOCK_SKEW=CLIENT_CLOCK_SKEW):\n        type = body['type']\n        if type == 'task-sent':\n            # clients never sync so cannot use their clock value\n            _c = body['clock'] = (self.clock.value or 1) + CLIENT_CLOCK_SKEW\n            self.adjust_clock(_c)\n        else:\n            try:\n                clock = body['clock']\n            except KeyError:\n                body['clock'] = self.forward_clock()\n            else:\n                self.adjust_clock(clock)\n\n        if localize:\n            try:\n                offset, timestamp = tzfields(body)\n            except KeyError:\n                pass\n            else:\n                body['timestamp'] = adjust_timestamp(timestamp, offset)\n        body['local_received'] = now()\n        return type, body\n\n    def _receive(self, body, message, list=list, isinstance=isinstance):\n        if isinstance(body, list):  # celery 4.0: List of events\n            process, from_message = self.process, self.event_from_message\n            [process(*from_message(event)) for event in body]\n        else:\n            self.process(*self.event_from_message(body))\n\n    @property\n    def connection(self):\n        return self.channel.connection.client if self.channel else None\n", "path": "celery/events/receiver.py"}], "after_files": [{"content": "\"\"\"Event receiver implementation.\"\"\"\nfrom __future__ import absolute_import, unicode_literals\n\nimport time\nfrom operator import itemgetter\n\nfrom kombu import Queue\nfrom kombu.connection import maybe_channel\nfrom kombu.mixins import ConsumerMixin\n\nfrom celery import uuid\nfrom celery.app import app_or_default\nfrom celery.utils.time import adjust_timestamp\n\nfrom .event import get_exchange\n\n__all__ = ('EventReceiver',)\n\nCLIENT_CLOCK_SKEW = -1\n\n_TZGETTER = itemgetter('utcoffset', 'timestamp')\n\n\nclass EventReceiver(ConsumerMixin):\n    \"\"\"Capture events.\n\n    Arguments:\n        connection (kombu.Connection): Connection to the broker.\n        handlers (Mapping[Callable]): Event handlers.\n            This is  a map of event type names and their handlers.\n            The special handler `\"*\"` captures all events that don't have a\n            handler.\n    \"\"\"\n\n    app = None\n\n    def __init__(self, channel, handlers=None, routing_key='#',\n                 node_id=None, app=None, queue_prefix=None,\n                 accept=None, queue_ttl=None, queue_expires=None):\n        self.app = app_or_default(app or self.app)\n        self.channel = maybe_channel(channel)\n        self.handlers = {} if handlers is None else handlers\n        self.routing_key = routing_key\n        self.node_id = node_id or uuid()\n        self.queue_prefix = queue_prefix or self.app.conf.event_queue_prefix\n        self.exchange = get_exchange(\n            self.connection or self.app.connection_for_write(),\n            name=self.app.conf.event_exchange)\n        if queue_ttl is None:\n            queue_ttl = self.app.conf.event_queue_ttl\n        if queue_expires is None:\n            queue_expires = self.app.conf.event_queue_expires\n        self.queue = Queue(\n            '.'.join([self.queue_prefix, self.node_id]),\n            exchange=self.exchange,\n            routing_key=self.routing_key,\n            auto_delete=True, durable=False,\n            message_ttl=queue_ttl,\n            expires=queue_expires,\n        )\n        self.clock = self.app.clock\n        self.adjust_clock = self.clock.adjust\n        self.forward_clock = self.clock.forward\n        if accept is None:\n            accept = {self.app.conf.event_serializer, 'json'}\n        self.accept = accept\n\n    def process(self, type, event):\n        \"\"\"Process event by dispatching to configured handler.\"\"\"\n        handler = self.handlers.get(type) or self.handlers.get('*')\n        handler and handler(event)\n\n    def get_consumers(self, Consumer, channel):\n        return [Consumer(queues=[self.queue],\n                         callbacks=[self._receive], no_ack=True,\n                         accept=self.accept)]\n\n    def on_consume_ready(self, connection, channel, consumers,\n                         wakeup=True, **kwargs):\n        if wakeup:\n            self.wakeup_workers(channel=channel)\n\n    def itercapture(self, limit=None, timeout=None, wakeup=True):\n        return self.consume(limit=limit, timeout=timeout, wakeup=wakeup)\n\n    def capture(self, limit=None, timeout=None, wakeup=True):\n        \"\"\"Open up a consumer capturing events.\n\n        This has to run in the main process, and it will never stop\n        unless :attr:`EventDispatcher.should_stop` is set to True, or\n        forced via :exc:`KeyboardInterrupt` or :exc:`SystemExit`.\n        \"\"\"\n        for _ in self.consume(limit=limit, timeout=timeout, wakeup=wakeup):\n            pass\n\n    def wakeup_workers(self, channel=None):\n        self.app.control.broadcast('heartbeat',\n                                   connection=self.connection,\n                                   channel=channel)\n\n    def event_from_message(self, body, localize=True,\n                           now=time.time, tzfields=_TZGETTER,\n                           adjust_timestamp=adjust_timestamp,\n                           CLIENT_CLOCK_SKEW=CLIENT_CLOCK_SKEW):\n        type = body['type']\n        if type == 'task-sent':\n            # clients never sync so cannot use their clock value\n            _c = body['clock'] = (self.clock.value or 1) + CLIENT_CLOCK_SKEW\n            self.adjust_clock(_c)\n        else:\n            try:\n                clock = body['clock']\n            except KeyError:\n                body['clock'] = self.forward_clock()\n            else:\n                self.adjust_clock(clock)\n\n        if localize:\n            try:\n                offset, timestamp = tzfields(body)\n            except KeyError:\n                pass\n            else:\n                body['timestamp'] = adjust_timestamp(timestamp, offset)\n        body['local_received'] = now()\n        return type, body\n\n    def _receive(self, body, message, list=list, isinstance=isinstance):\n        if isinstance(body, list):  # celery 4.0: List of events\n            process, from_message = self.process, self.event_from_message\n            [process(*from_message(event)) for event in body]\n        else:\n            self.process(*self.event_from_message(body))\n\n    @property\n    def connection(self):\n        return self.channel.connection.client if self.channel else None\n", "path": "celery/events/receiver.py"}]}