prompt: 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()
  
  ```
  
golden_diff: 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',
  
verification_info: {"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"}]}