We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
new_cassandra + Celery leaks memory
Continuing performing stress test I ran into a situation where next jobs done using Celery + new_cassandra backend leak memory.

Running tasks I get bigger and bigger ru_maxrss value

```
[2015-10-31 21:24:46,598: WARNING/Worker-1] resource.struct_rusage(ru_utime=0.010111, ru_stime=0.00629, ru_maxrss=25748, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=1289, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=2, ru_nivcsw=237)
[2015-10-31 21:24:46,601: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[f2695142-0680-4565-a51c-790c36f685f2]
[2015-10-31 21:25:07,208: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[c9cb66b1-d8ce-45c5-ab6b-05c516ec5464] succeeded in 20.61094686s: None
[2015-10-31 21:25:07,211: WARNING/Worker-1] resource.struct_rusage(ru_utime=1.141192, ru_stime=0.249278, ru_maxrss=31512, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=2940, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=19155, ru_nivcsw=1739)
[2015-10-31 21:25:07,221: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[cadb2d50-eb12-4e4c-aa74-57bfd4bf41d3]
[2015-10-31 21:25:23,323: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[5d2ee4a0-b25d-4107-98b7-d55200f18a72] succeeded in 16.112931506s: None
[2015-10-31 21:25:23,325: WARNING/Worker-1] resource.struct_rusage(ru_utime=2.084558, ru_stime=0.413746, ru_maxrss=32288, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=3154, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=34251, ru_nivcsw=2860)
[2015-10-31 21:25:23,338: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[a2b3b191-3b0a-4bb6-ab77-dbaf2113fe09]
[2015-10-31 21:25:43,668: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[d3e5539f-8f12-4bec-b1a8-19a26a6a665e] succeeded in 20.34311718s: None
[2015-10-31 21:25:43,671: WARNING/Worker-1] resource.struct_rusage(ru_utime=3.290886, ru_stime=0.659077, ru_maxrss=33072, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=3343, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=53488, ru_nivcsw=4137)
[2015-10-31 21:25:43,682: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[99a6ca5f-5734-4eab-b148-b2a077a754b6]
[2015-10-31 21:26:04,140: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[283055a5-b466-4435-812b-0d8dbaefb2bf] succeeded in 20.469805369s: None
[2015-10-31 21:26:04,143: WARNING/Worker-1] resource.struct_rusage(ru_utime=4.581904, ru_stime=0.922214, ru_maxrss=33860, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=3537, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=73128, ru_nivcsw=5329)
```


--- 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/backends/cassandra.py`
Content:
```
1 # -* coding: utf-8 -*-
2 """
3     celery.backends.cassandra
4     ~~~~~~~~~~~~~~~~~~~~~~~~~
5 
6     Apache Cassandra result store backend using DataStax driver
7 
8 """
9 from __future__ import absolute_import
10 
11 import sys
12 try:  # pragma: no cover
13     import cassandra
14     import cassandra.cluster
15 except ImportError:  # pragma: no cover
16     cassandra = None   # noqa
17 
18 from celery import states
19 from celery.exceptions import ImproperlyConfigured
20 from celery.utils.log import get_logger
21 from .base import BaseBackend
22 
23 __all__ = ['CassandraBackend']
24 
25 logger = get_logger(__name__)
26 
27 E_NO_CASSANDRA = """
28 You need to install the cassandra-driver library to
29 use the Cassandra backend. See https://github.com/datastax/python-driver
30 """
31 
32 Q_INSERT_RESULT = """
33 INSERT INTO {table} (
34     task_id, status, result, date_done, traceback, children) VALUES (
35         %s, %s, %s, %s, %s, %s) {expires};
36 """
37 
38 Q_SELECT_RESULT = """
39 SELECT status, result, date_done, traceback, children
40 FROM {table}
41 WHERE task_id=%s
42 LIMIT 1
43 """
44 
45 Q_CREATE_RESULT_TABLE = """
46 CREATE TABLE {table} (
47     task_id text,
48     status text,
49     result blob,
50     date_done timestamp,
51     traceback blob,
52     children blob,
53     PRIMARY KEY ((task_id), date_done)
54 ) WITH CLUSTERING ORDER BY (date_done DESC);
55 """
56 
57 Q_EXPIRES = """
58     USING TTL {0}
59 """
60 
61 if sys.version_info[0] == 3:
62     def buf_t(x):
63         return bytes(x, 'utf8')
64 else:
65     buf_t = buffer  # noqa
66 
67 
68 class CassandraBackend(BaseBackend):
69     """Cassandra backend utilizing DataStax driver
70 
71     :raises celery.exceptions.ImproperlyConfigured: if
72         module :mod:`cassandra` is not available.
73 
74     """
75 
76     #: List of Cassandra servers with format: ``hostname``.
77     servers = None
78 
79     supports_autoexpire = True      # autoexpire supported via entry_ttl
80 
81     def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,
82                  port=9042, **kwargs):
83         """Initialize Cassandra backend.
84 
85         Raises :class:`celery.exceptions.ImproperlyConfigured` if
86         the :setting:`cassandra_servers` setting is not set.
87 
88         """
89         super(CassandraBackend, self).__init__(**kwargs)
90 
91         if not cassandra:
92             raise ImproperlyConfigured(E_NO_CASSANDRA)
93 
94         conf = self.app.conf
95         self.servers = (servers or
96                         conf.get('cassandra_servers', None))
97         self.port = (port or
98                      conf.get('cassandra_port', None))
99         self.keyspace = (keyspace or
100                          conf.get('cassandra_keyspace', None))
101         self.table = (table or
102                       conf.get('cassandra_table', None))
103 
104         if not self.servers or not self.keyspace or not self.table:
105             raise ImproperlyConfigured('Cassandra backend not configured.')
106 
107         expires = (entry_ttl or conf.get('cassandra_entry_ttl', None))
108 
109         self.cqlexpires = (Q_EXPIRES.format(expires)
110                            if expires is not None else '')
111 
112         read_cons = conf.get('cassandra_read_consistency') or 'LOCAL_QUORUM'
113         write_cons = conf.get('cassandra_write_consistency') or 'LOCAL_QUORUM'
114 
115         self.read_consistency = getattr(
116             cassandra.ConsistencyLevel, read_cons,
117             cassandra.ConsistencyLevel.LOCAL_QUORUM,
118         )
119         self.write_consistency = getattr(
120             cassandra.ConsistencyLevel, write_cons,
121             cassandra.ConsistencyLevel.LOCAL_QUORUM,
122         )
123 
124         self._connection = None
125         self._session = None
126         self._write_stmt = None
127         self._read_stmt = None
128 
129     def process_cleanup(self):
130         if self._connection is not None:
131             self._connection = None
132         if self._session is not None:
133             self._session.shutdown()
134             self._session = None
135 
136     def _get_connection(self, write=False):
137         """Prepare the connection for action
138 
139         :param write: bool - are we a writer?
140 
141         """
142         if self._connection is None:
143             try:
144                 self._connection = cassandra.cluster.Cluster(self.servers,
145                                                              port=self.port)
146                 self._session = self._connection.connect(self.keyspace)
147 
148                 # We are forced to do concatenation below, as formatting would
149                 # blow up on superficial %s that will be processed by Cassandra
150                 self._write_stmt = cassandra.query.SimpleStatement(
151                     Q_INSERT_RESULT.format(
152                         table=self.table, expires=self.cqlexpires),
153                 )
154                 self._write_stmt.consistency_level = self.write_consistency
155 
156                 self._read_stmt = cassandra.query.SimpleStatement(
157                     Q_SELECT_RESULT.format(table=self.table),
158                 )
159                 self._read_stmt.consistency_level = self.read_consistency
160 
161                 if write:
162                     # Only possible writers "workers" are allowed to issue
163                     # CREATE TABLE. This is to prevent conflicting situations
164                     # where both task-creator and task-executor would issue it
165                     # at the same time.
166 
167                     # Anyway; if you're doing anything critical, you should
168                     # have created this table in advance, in which case
169                     # this query will be a no-op (AlreadyExists)
170                     self._make_stmt = cassandra.query.SimpleStatement(
171                         Q_CREATE_RESULT_TABLE.format(table=self.table),
172                     )
173                     self._make_stmt.consistency_level = self.write_consistency
174                     try:
175                         self._session.execute(self._make_stmt)
176                     except cassandra.AlreadyExists:
177                         pass
178 
179             except cassandra.OperationTimedOut:
180                 # a heavily loaded or gone Cassandra cluster failed to respond.
181                 # leave this class in a consistent state
182                 self._connection = None
183                 if self._session is not None:
184                     self._session.shutdown()
185 
186                 raise   # we did fail after all - reraise
187 
188     def _store_result(self, task_id, result, status,
189                       traceback=None, request=None, **kwargs):
190         """Store return value and status of an executed task."""
191         self._get_connection(write=True)
192 
193         self._session.execute(self._write_stmt, (
194             task_id,
195             status,
196             buf_t(self.encode(result)),
197             self.app.now(),
198             buf_t(self.encode(traceback)),
199             buf_t(self.encode(self.current_task_children(request)))
200         ))
201 
202     def _get_task_meta_for(self, task_id):
203         """Get task metadata for a task by id."""
204         self._get_connection()
205 
206         res = self._session.execute(self._read_stmt, (task_id, ))
207         if not res:
208             return {'status': states.PENDING, 'result': None}
209 
210         status, result, date_done, traceback, children = res[0]
211 
212         return self.meta_from_decoded({
213             'task_id': task_id,
214             'status': status,
215             'result': self.decode(result),
216             'date_done': date_done.strftime('%Y-%m-%dT%H:%M:%SZ'),
217             'traceback': self.decode(traceback),
218             'children': self.decode(children),
219         })
220 
221     def __reduce__(self, args=(), kwargs={}):
222         kwargs.update(
223             dict(servers=self.servers,
224                  keyspace=self.keyspace,
225                  table=self.table))
226         return super(CassandraBackend, self).__reduce__(args, kwargs)
227
```
--- 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/backends/cassandra.py b/celery/backends/cassandra.py
--- a/celery/backends/cassandra.py
+++ b/celery/backends/cassandra.py
@@ -125,13 +125,14 @@
         self._session = None
         self._write_stmt = None
         self._read_stmt = None
+        self._make_stmt = None
 
     def process_cleanup(self):
         if self._connection is not None:
-            self._connection = None
-        if self._session is not None:
-            self._session.shutdown()
-            self._session = None
+            self._connection.shutdown() # also shuts down _session
+
+        self._connection = None
+        self._session = None
 
     def _get_connection(self, write=False):
         """Prepare the connection for action
@@ -171,6 +172,7 @@
                         Q_CREATE_RESULT_TABLE.format(table=self.table),
                     )
                     self._make_stmt.consistency_level = self.write_consistency
+
                     try:
                         self._session.execute(self._make_stmt)
                     except cassandra.AlreadyExists:
@@ -179,10 +181,11 @@
             except cassandra.OperationTimedOut:
                 # a heavily loaded or gone Cassandra cluster failed to respond.
                 # leave this class in a consistent state
-                self._connection = None
-                if self._session is not None:
-                    self._session.shutdown()
+                if self._connection is not None:
+                    self._connection.shutdown()     # also shuts down _session
 
+                self._connection = None
+                self._session = None
                 raise   # we did fail after all - reraise
 
     def _store_result(self, task_id, result, status,

{"golden_diff": "diff --git a/celery/backends/cassandra.py b/celery/backends/cassandra.py\n--- a/celery/backends/cassandra.py\n+++ b/celery/backends/cassandra.py\n@@ -125,13 +125,14 @@\n         self._session = None\n         self._write_stmt = None\n         self._read_stmt = None\n+        self._make_stmt = None\n \n     def process_cleanup(self):\n         if self._connection is not None:\n-            self._connection = None\n-        if self._session is not None:\n-            self._session.shutdown()\n-            self._session = None\n+            self._connection.shutdown() # also shuts down _session\n+\n+        self._connection = None\n+        self._session = None\n \n     def _get_connection(self, write=False):\n         \"\"\"Prepare the connection for action\n@@ -171,6 +172,7 @@\n                         Q_CREATE_RESULT_TABLE.format(table=self.table),\n                     )\n                     self._make_stmt.consistency_level = self.write_consistency\n+\n                     try:\n                         self._session.execute(self._make_stmt)\n                     except cassandra.AlreadyExists:\n@@ -179,10 +181,11 @@\n             except cassandra.OperationTimedOut:\n                 # a heavily loaded or gone Cassandra cluster failed to respond.\n                 # leave this class in a consistent state\n-                self._connection = None\n-                if self._session is not None:\n-                    self._session.shutdown()\n+                if self._connection is not None:\n+                    self._connection.shutdown()     # also shuts down _session\n \n+                self._connection = None\n+                self._session = None\n                 raise   # we did fail after all - reraise\n \n     def _store_result(self, task_id, result, status,\n", "issue": "new_cassandra + Celery leaks memory\nContinuing performing stress test I ran into a situation where next jobs done using Celery + new_cassandra backend leak memory.\n\nRunning tasks I get bigger and bigger ru_maxrss value\n\n```\n[2015-10-31 21:24:46,598: WARNING/Worker-1] resource.struct_rusage(ru_utime=0.010111, ru_stime=0.00629, ru_maxrss=25748, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=1289, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=2, ru_nivcsw=237)\n[2015-10-31 21:24:46,601: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[f2695142-0680-4565-a51c-790c36f685f2]\n[2015-10-31 21:25:07,208: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[c9cb66b1-d8ce-45c5-ab6b-05c516ec5464] succeeded in 20.61094686s: None\n[2015-10-31 21:25:07,211: WARNING/Worker-1] resource.struct_rusage(ru_utime=1.141192, ru_stime=0.249278, ru_maxrss=31512, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=2940, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=19155, ru_nivcsw=1739)\n[2015-10-31 21:25:07,221: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[cadb2d50-eb12-4e4c-aa74-57bfd4bf41d3]\n[2015-10-31 21:25:23,323: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[5d2ee4a0-b25d-4107-98b7-d55200f18a72] succeeded in 16.112931506s: None\n[2015-10-31 21:25:23,325: WARNING/Worker-1] resource.struct_rusage(ru_utime=2.084558, ru_stime=0.413746, ru_maxrss=32288, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=3154, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=34251, ru_nivcsw=2860)\n[2015-10-31 21:25:23,338: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[a2b3b191-3b0a-4bb6-ab77-dbaf2113fe09]\n[2015-10-31 21:25:43,668: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[d3e5539f-8f12-4bec-b1a8-19a26a6a665e] succeeded in 20.34311718s: None\n[2015-10-31 21:25:43,671: WARNING/Worker-1] resource.struct_rusage(ru_utime=3.290886, ru_stime=0.659077, ru_maxrss=33072, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=3343, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=53488, ru_nivcsw=4137)\n[2015-10-31 21:25:43,682: INFO/MainProcess] Received task: s4jobs.tasks.decommission_device.decommission_device[99a6ca5f-5734-4eab-b148-b2a077a754b6]\n[2015-10-31 21:26:04,140: INFO/MainProcess] Task s4jobs.tasks.decommission_device.decommission_device[283055a5-b466-4435-812b-0d8dbaefb2bf] succeeded in 20.469805369s: None\n[2015-10-31 21:26:04,143: WARNING/Worker-1] resource.struct_rusage(ru_utime=4.581904, ru_stime=0.922214, ru_maxrss=33860, ru_ixrss=0, ru_idrss=0, ru_isrss=0, ru_minflt=3537, ru_majflt=0, ru_nswap=0, ru_inblock=0, ru_oublock=0, ru_msgsnd=0, ru_msgrcv=0, ru_nsignals=0, ru_nvcsw=73128, ru_nivcsw=5329)\n```\n\n", "before_files": [{"content": "# -* coding: utf-8 -*-\n\"\"\"\n    celery.backends.cassandra\n    ~~~~~~~~~~~~~~~~~~~~~~~~~\n\n    Apache Cassandra result store backend using DataStax driver\n\n\"\"\"\nfrom __future__ import absolute_import\n\nimport sys\ntry:  # pragma: no cover\n    import cassandra\n    import cassandra.cluster\nexcept ImportError:  # pragma: no cover\n    cassandra = None   # noqa\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.utils.log import get_logger\nfrom .base import BaseBackend\n\n__all__ = ['CassandraBackend']\n\nlogger = get_logger(__name__)\n\nE_NO_CASSANDRA = \"\"\"\nYou need to install the cassandra-driver library to\nuse the Cassandra backend. See https://github.com/datastax/python-driver\n\"\"\"\n\nQ_INSERT_RESULT = \"\"\"\nINSERT INTO {table} (\n    task_id, status, result, date_done, traceback, children) VALUES (\n        %s, %s, %s, %s, %s, %s) {expires};\n\"\"\"\n\nQ_SELECT_RESULT = \"\"\"\nSELECT status, result, date_done, traceback, children\nFROM {table}\nWHERE task_id=%s\nLIMIT 1\n\"\"\"\n\nQ_CREATE_RESULT_TABLE = \"\"\"\nCREATE TABLE {table} (\n    task_id text,\n    status text,\n    result blob,\n    date_done timestamp,\n    traceback blob,\n    children blob,\n    PRIMARY KEY ((task_id), date_done)\n) WITH CLUSTERING ORDER BY (date_done DESC);\n\"\"\"\n\nQ_EXPIRES = \"\"\"\n    USING TTL {0}\n\"\"\"\n\nif sys.version_info[0] == 3:\n    def buf_t(x):\n        return bytes(x, 'utf8')\nelse:\n    buf_t = buffer  # noqa\n\n\nclass CassandraBackend(BaseBackend):\n    \"\"\"Cassandra backend utilizing DataStax driver\n\n    :raises celery.exceptions.ImproperlyConfigured: if\n        module :mod:`cassandra` is not available.\n\n    \"\"\"\n\n    #: List of Cassandra servers with format: ``hostname``.\n    servers = None\n\n    supports_autoexpire = True      # autoexpire supported via entry_ttl\n\n    def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,\n                 port=9042, **kwargs):\n        \"\"\"Initialize Cassandra backend.\n\n        Raises :class:`celery.exceptions.ImproperlyConfigured` if\n        the :setting:`cassandra_servers` setting is not set.\n\n        \"\"\"\n        super(CassandraBackend, self).__init__(**kwargs)\n\n        if not cassandra:\n            raise ImproperlyConfigured(E_NO_CASSANDRA)\n\n        conf = self.app.conf\n        self.servers = (servers or\n                        conf.get('cassandra_servers', None))\n        self.port = (port or\n                     conf.get('cassandra_port', None))\n        self.keyspace = (keyspace or\n                         conf.get('cassandra_keyspace', None))\n        self.table = (table or\n                      conf.get('cassandra_table', None))\n\n        if not self.servers or not self.keyspace or not self.table:\n            raise ImproperlyConfigured('Cassandra backend not configured.')\n\n        expires = (entry_ttl or conf.get('cassandra_entry_ttl', None))\n\n        self.cqlexpires = (Q_EXPIRES.format(expires)\n                           if expires is not None else '')\n\n        read_cons = conf.get('cassandra_read_consistency') or 'LOCAL_QUORUM'\n        write_cons = conf.get('cassandra_write_consistency') or 'LOCAL_QUORUM'\n\n        self.read_consistency = getattr(\n            cassandra.ConsistencyLevel, read_cons,\n            cassandra.ConsistencyLevel.LOCAL_QUORUM,\n        )\n        self.write_consistency = getattr(\n            cassandra.ConsistencyLevel, write_cons,\n            cassandra.ConsistencyLevel.LOCAL_QUORUM,\n        )\n\n        self._connection = None\n        self._session = None\n        self._write_stmt = None\n        self._read_stmt = None\n\n    def process_cleanup(self):\n        if self._connection is not None:\n            self._connection = None\n        if self._session is not None:\n            self._session.shutdown()\n            self._session = None\n\n    def _get_connection(self, write=False):\n        \"\"\"Prepare the connection for action\n\n        :param write: bool - are we a writer?\n\n        \"\"\"\n        if self._connection is None:\n            try:\n                self._connection = cassandra.cluster.Cluster(self.servers,\n                                                             port=self.port)\n                self._session = self._connection.connect(self.keyspace)\n\n                # We are forced to do concatenation below, as formatting would\n                # blow up on superficial %s that will be processed by Cassandra\n                self._write_stmt = cassandra.query.SimpleStatement(\n                    Q_INSERT_RESULT.format(\n                        table=self.table, expires=self.cqlexpires),\n                )\n                self._write_stmt.consistency_level = self.write_consistency\n\n                self._read_stmt = cassandra.query.SimpleStatement(\n                    Q_SELECT_RESULT.format(table=self.table),\n                )\n                self._read_stmt.consistency_level = self.read_consistency\n\n                if write:\n                    # Only possible writers \"workers\" are allowed to issue\n                    # CREATE TABLE. This is to prevent conflicting situations\n                    # where both task-creator and task-executor would issue it\n                    # at the same time.\n\n                    # Anyway; if you're doing anything critical, you should\n                    # have created this table in advance, in which case\n                    # this query will be a no-op (AlreadyExists)\n                    self._make_stmt = cassandra.query.SimpleStatement(\n                        Q_CREATE_RESULT_TABLE.format(table=self.table),\n                    )\n                    self._make_stmt.consistency_level = self.write_consistency\n                    try:\n                        self._session.execute(self._make_stmt)\n                    except cassandra.AlreadyExists:\n                        pass\n\n            except cassandra.OperationTimedOut:\n                # a heavily loaded or gone Cassandra cluster failed to respond.\n                # leave this class in a consistent state\n                self._connection = None\n                if self._session is not None:\n                    self._session.shutdown()\n\n                raise   # we did fail after all - reraise\n\n    def _store_result(self, task_id, result, status,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and status of an executed task.\"\"\"\n        self._get_connection(write=True)\n\n        self._session.execute(self._write_stmt, (\n            task_id,\n            status,\n            buf_t(self.encode(result)),\n            self.app.now(),\n            buf_t(self.encode(traceback)),\n            buf_t(self.encode(self.current_task_children(request)))\n        ))\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task metadata for a task by id.\"\"\"\n        self._get_connection()\n\n        res = self._session.execute(self._read_stmt, (task_id, ))\n        if not res:\n            return {'status': states.PENDING, 'result': None}\n\n        status, result, date_done, traceback, children = res[0]\n\n        return self.meta_from_decoded({\n            'task_id': task_id,\n            'status': status,\n            'result': self.decode(result),\n            'date_done': date_done.strftime('%Y-%m-%dT%H:%M:%SZ'),\n            'traceback': self.decode(traceback),\n            'children': self.decode(children),\n        })\n\n    def __reduce__(self, args=(), kwargs={}):\n        kwargs.update(\n            dict(servers=self.servers,\n                 keyspace=self.keyspace,\n                 table=self.table))\n        return super(CassandraBackend, self).__reduce__(args, kwargs)\n", "path": "celery/backends/cassandra.py"}], "after_files": [{"content": "# -* coding: utf-8 -*-\n\"\"\"\n    celery.backends.cassandra\n    ~~~~~~~~~~~~~~~~~~~~~~~~~\n\n    Apache Cassandra result store backend using DataStax driver\n\n\"\"\"\nfrom __future__ import absolute_import\n\nimport sys\ntry:  # pragma: no cover\n    import cassandra\n    import cassandra.cluster\nexcept ImportError:  # pragma: no cover\n    cassandra = None   # noqa\n\nfrom celery import states\nfrom celery.exceptions import ImproperlyConfigured\nfrom celery.utils.log import get_logger\nfrom .base import BaseBackend\n\n__all__ = ['CassandraBackend']\n\nlogger = get_logger(__name__)\n\nE_NO_CASSANDRA = \"\"\"\nYou need to install the cassandra-driver library to\nuse the Cassandra backend. See https://github.com/datastax/python-driver\n\"\"\"\n\nQ_INSERT_RESULT = \"\"\"\nINSERT INTO {table} (\n    task_id, status, result, date_done, traceback, children) VALUES (\n        %s, %s, %s, %s, %s, %s) {expires};\n\"\"\"\n\nQ_SELECT_RESULT = \"\"\"\nSELECT status, result, date_done, traceback, children\nFROM {table}\nWHERE task_id=%s\nLIMIT 1\n\"\"\"\n\nQ_CREATE_RESULT_TABLE = \"\"\"\nCREATE TABLE {table} (\n    task_id text,\n    status text,\n    result blob,\n    date_done timestamp,\n    traceback blob,\n    children blob,\n    PRIMARY KEY ((task_id), date_done)\n) WITH CLUSTERING ORDER BY (date_done DESC);\n\"\"\"\n\nQ_EXPIRES = \"\"\"\n    USING TTL {0}\n\"\"\"\n\nif sys.version_info[0] == 3:\n    def buf_t(x):\n        return bytes(x, 'utf8')\nelse:\n    buf_t = buffer  # noqa\n\n\nclass CassandraBackend(BaseBackend):\n    \"\"\"Cassandra backend utilizing DataStax driver\n\n    :raises celery.exceptions.ImproperlyConfigured: if\n        module :mod:`cassandra` is not available.\n\n    \"\"\"\n\n    #: List of Cassandra servers with format: ``hostname``.\n    servers = None\n\n    supports_autoexpire = True      # autoexpire supported via entry_ttl\n\n    def __init__(self, servers=None, keyspace=None, table=None, entry_ttl=None,\n                 port=9042, **kwargs):\n        \"\"\"Initialize Cassandra backend.\n\n        Raises :class:`celery.exceptions.ImproperlyConfigured` if\n        the :setting:`cassandra_servers` setting is not set.\n\n        \"\"\"\n        super(CassandraBackend, self).__init__(**kwargs)\n\n        if not cassandra:\n            raise ImproperlyConfigured(E_NO_CASSANDRA)\n\n        conf = self.app.conf\n        self.servers = (servers or\n                        conf.get('cassandra_servers', None))\n        self.port = (port or\n                     conf.get('cassandra_port', None))\n        self.keyspace = (keyspace or\n                         conf.get('cassandra_keyspace', None))\n        self.table = (table or\n                      conf.get('cassandra_table', None))\n\n        if not self.servers or not self.keyspace or not self.table:\n            raise ImproperlyConfigured('Cassandra backend not configured.')\n\n        expires = (entry_ttl or conf.get('cassandra_entry_ttl', None))\n\n        self.cqlexpires = (Q_EXPIRES.format(expires)\n                           if expires is not None else '')\n\n        read_cons = conf.get('cassandra_read_consistency') or 'LOCAL_QUORUM'\n        write_cons = conf.get('cassandra_write_consistency') or 'LOCAL_QUORUM'\n\n        self.read_consistency = getattr(\n            cassandra.ConsistencyLevel, read_cons,\n            cassandra.ConsistencyLevel.LOCAL_QUORUM,\n        )\n        self.write_consistency = getattr(\n            cassandra.ConsistencyLevel, write_cons,\n            cassandra.ConsistencyLevel.LOCAL_QUORUM,\n        )\n\n        self._connection = None\n        self._session = None\n        self._write_stmt = None\n        self._read_stmt = None\n        self._make_stmt = None\n\n    def process_cleanup(self):\n        if self._connection is not None:\n            self._connection.shutdown() # also shuts down _session\n\n        self._connection = None\n        self._session = None\n\n    def _get_connection(self, write=False):\n        \"\"\"Prepare the connection for action\n\n        :param write: bool - are we a writer?\n\n        \"\"\"\n        if self._connection is None:\n            try:\n                self._connection = cassandra.cluster.Cluster(self.servers,\n                                                             port=self.port)\n                self._session = self._connection.connect(self.keyspace)\n\n                # We are forced to do concatenation below, as formatting would\n                # blow up on superficial %s that will be processed by Cassandra\n                self._write_stmt = cassandra.query.SimpleStatement(\n                    Q_INSERT_RESULT.format(\n                        table=self.table, expires=self.cqlexpires),\n                )\n                self._write_stmt.consistency_level = self.write_consistency\n\n                self._read_stmt = cassandra.query.SimpleStatement(\n                    Q_SELECT_RESULT.format(table=self.table),\n                )\n                self._read_stmt.consistency_level = self.read_consistency\n\n                if write:\n                    # Only possible writers \"workers\" are allowed to issue\n                    # CREATE TABLE. This is to prevent conflicting situations\n                    # where both task-creator and task-executor would issue it\n                    # at the same time.\n\n                    # Anyway; if you're doing anything critical, you should\n                    # have created this table in advance, in which case\n                    # this query will be a no-op (AlreadyExists)\n                    self._make_stmt = cassandra.query.SimpleStatement(\n                        Q_CREATE_RESULT_TABLE.format(table=self.table),\n                    )\n                    self._make_stmt.consistency_level = self.write_consistency\n\n                    try:\n                        self._session.execute(self._make_stmt)\n                    except cassandra.AlreadyExists:\n                        pass\n\n            except cassandra.OperationTimedOut:\n                # a heavily loaded or gone Cassandra cluster failed to respond.\n                # leave this class in a consistent state\n                if self._connection is not None:\n                    self._connection.shutdown()     # also shuts down _session\n\n                self._connection = None\n                self._session = None\n                raise   # we did fail after all - reraise\n\n    def _store_result(self, task_id, result, status,\n                      traceback=None, request=None, **kwargs):\n        \"\"\"Store return value and status of an executed task.\"\"\"\n        self._get_connection(write=True)\n\n        self._session.execute(self._write_stmt, (\n            task_id,\n            status,\n            buf_t(self.encode(result)),\n            self.app.now(),\n            buf_t(self.encode(traceback)),\n            buf_t(self.encode(self.current_task_children(request)))\n        ))\n\n    def _get_task_meta_for(self, task_id):\n        \"\"\"Get task metadata for a task by id.\"\"\"\n        self._get_connection()\n\n        res = self._session.execute(self._read_stmt, (task_id, ))\n        if not res:\n            return {'status': states.PENDING, 'result': None}\n\n        status, result, date_done, traceback, children = res[0]\n\n        return self.meta_from_decoded({\n            'task_id': task_id,\n            'status': status,\n            'result': self.decode(result),\n            'date_done': date_done.strftime('%Y-%m-%dT%H:%M:%SZ'),\n            'traceback': self.decode(traceback),\n            'children': self.decode(children),\n        })\n\n    def __reduce__(self, args=(), kwargs={}):\n        kwargs.update(\n            dict(servers=self.servers,\n                 keyspace=self.keyspace,\n                 table=self.table))\n        return super(CassandraBackend, self).__reduce__(args, kwargs)\n", "path": "celery/backends/cassandra.py"}]}