We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
"Connection refused" when using keep-alive with gthread
Gunicorn version: 19.7.1 (also tried master branch)

This bug could be reproduced with attached files:
- test_http_gunicorn_raw_keep-alive-requests.py (test runner)
- myapp.py (gunicorn app that should be in the same directory)

Test case starts command:
```
gunicorn --worker-class gthread --workers 1 --threads 20 --keep-alive 9999 \
           --log-level DEBUG --bind localhost:7777 myapp:app
```
with logs redirected to `/tmp/_test_gunicorn.out`
and then create separate threads to open http session and send 1000 requests within it.

In my case gunicorn server received request `/request/1096` and then reset connection (see `tcp.stream eq 10` from attached `gunicorn_reset_keep_alived_connection.pcapng`

NOTE: this is race condition so it may happen that all requests finish with success (see myapp.py -> without `time.sleep` it almost never fails) but with that `time.sleep` it usually fails and then script should output line similar to:
`[1] Failed request id=1096 with ('Connection aborted.', error(104, 'Connection reset by peer'))`
it mean that thread sent request `GET /request/1096` and received `Connection reset by peer` (so the last one request from that session that succeed is `GET /request/1095`).

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `gunicorn/workers/gthread.py`
Content:
```
1 # -*- coding: utf-8 -
2 #
3 # This file is part of gunicorn released under the MIT license.
4 # See the NOTICE for more information.
5 
6 # design:
7 # a threaded worker accepts connections in the main loop, accepted
8 # connections are are added to the thread pool as a connection job. On
9 # keepalive connections are put back in the loop waiting for an event.
10 # If no event happen after the keep alive timeout, the connectoin is
11 # closed.
12 
13 from collections import deque
14 from datetime import datetime
15 import errno
16 from functools import partial
17 import os
18 import socket
19 import ssl
20 import sys
21 from threading import RLock
22 import time
23 
24 from .. import http
25 from ..http import wsgi
26 from .. import util
27 from . import base
28 from .. import six
29 
30 
31 try:
32     import concurrent.futures as futures
33 except ImportError:
34     raise RuntimeError("""
35     You need to install the 'futures' package to use this worker with this
36     Python version.
37     """)
38 
39 try:
40     from asyncio import selectors
41 except ImportError:
42     from gunicorn import selectors
43 
44 
45 class TConn(object):
46 
47     def __init__(self, cfg, sock, client, server):
48         self.cfg = cfg
49         self.sock = sock
50         self.client = client
51         self.server = server
52 
53         self.timeout = None
54         self.parser = None
55 
56         # set the socket to non blocking
57         self.sock.setblocking(False)
58 
59     def init(self):
60         self.sock.setblocking(True)
61         if self.parser is None:
62             # wrap the socket if needed
63             if self.cfg.is_ssl:
64                 self.sock = ssl.wrap_socket(self.sock, server_side=True,
65                         **self.cfg.ssl_options)
66 
67             # initialize the parser
68             self.parser = http.RequestParser(self.cfg, self.sock)
69 
70     def set_timeout(self):
71         # set the timeout
72         self.timeout = time.time() + self.cfg.keepalive
73 
74     def close(self):
75         util.close(self.sock)
76 
77     def __lt__(self, other):
78         return self.timeout < other.timeout
79 
80     __cmp__ = __lt__
81 
82 
83 class ThreadWorker(base.Worker):
84 
85     def __init__(self, *args, **kwargs):
86         super(ThreadWorker, self).__init__(*args, **kwargs)
87         self.worker_connections = self.cfg.worker_connections
88         self.max_keepalived = self.cfg.worker_connections - self.cfg.threads
89         # initialise the pool
90         self.tpool = None
91         self.poller = None
92         self._lock = None
93         self.futures = deque()
94         self._keep = deque()
95         self.nr_conns = 0
96 
97     @classmethod
98     def check_config(cls, cfg, log):
99         max_keepalived = cfg.worker_connections - cfg.threads
100 
101         if max_keepalived <= 0 and cfg.keepalive:
102             log.warning("No keepalived connections can be handled. " +
103                     "Check the number of worker connections and threads.")
104 
105     def init_process(self):
106         self.tpool = futures.ThreadPoolExecutor(max_workers=self.cfg.threads)
107         self.poller = selectors.DefaultSelector()
108         self._lock = RLock()
109         super(ThreadWorker, self).init_process()
110 
111     def handle_quit(self, sig, frame):
112         self.alive = False
113         # worker_int callback
114         self.cfg.worker_int(self)
115         self.tpool.shutdown(False)
116         time.sleep(0.1)
117         sys.exit(0)
118 
119     def _wrap_future(self, fs, conn):
120         fs.conn = conn
121         self.futures.append(fs)
122         fs.add_done_callback(self.finish_request)
123 
124     def enqueue_req(self, conn):
125         conn.init()
126         # submit the connection to a worker
127         fs = self.tpool.submit(self.handle, conn)
128         self._wrap_future(fs, conn)
129 
130     def accept(self, server, listener):
131         try:
132             sock, client = listener.accept()
133             # initialize the connection object
134             conn = TConn(self.cfg, sock, client, server)
135             self.nr_conns += 1
136             # enqueue the job
137             self.enqueue_req(conn)
138         except EnvironmentError as e:
139             if e.errno not in (errno.EAGAIN,
140                     errno.ECONNABORTED, errno.EWOULDBLOCK):
141                 raise
142 
143     def reuse_connection(self, conn, client):
144         with self._lock:
145             # unregister the client from the poller
146             self.poller.unregister(client)
147             # remove the connection from keepalive
148             try:
149                 self._keep.remove(conn)
150             except ValueError:
151                 # race condition
152                 return
153 
154         # submit the connection to a worker
155         self.enqueue_req(conn)
156 
157     def murder_keepalived(self):
158         now = time.time()
159         while True:
160             with self._lock:
161                 try:
162                     # remove the connection from the queue
163                     conn = self._keep.popleft()
164                 except IndexError:
165                     break
166 
167             delta = conn.timeout - now
168             if delta > 0:
169                 # add the connection back to the queue
170                 with self._lock:
171                     self._keep.appendleft(conn)
172                 break
173             else:
174                 self.nr_conns -= 1
175                 # remove the socket from the poller
176                 with self._lock:
177                     try:
178                         self.poller.unregister(conn.sock)
179                     except EnvironmentError as e:
180                         if e.errno != errno.EBADF:
181                             raise
182                     except KeyError:
183                         # already removed by the system, continue
184                         pass
185 
186                 # close the socket
187                 conn.close()
188 
189     def is_parent_alive(self):
190         # If our parent changed then we shut down.
191         if self.ppid != os.getppid():
192             self.log.info("Parent changed, shutting down: %s", self)
193             return False
194         return True
195 
196     def run(self):
197         # init listeners, add them to the event loop
198         for sock in self.sockets:
199             sock.setblocking(False)
200             # a race condition during graceful shutdown may make the listener
201             # name unavailable in the request handler so capture it once here
202             server = sock.getsockname()
203             acceptor = partial(self.accept, server)
204             self.poller.register(sock, selectors.EVENT_READ, acceptor)
205 
206         while self.alive:
207             # notify the arbiter we are alive
208             self.notify()
209 
210             # can we accept more connections?
211             if self.nr_conns < self.worker_connections:
212                 # wait for an event
213                 events = self.poller.select(1.0)
214                 for key, _ in events:
215                     callback = key.data
216                     callback(key.fileobj)
217 
218                 # check (but do not wait) for finished requests
219                 result = futures.wait(self.futures, timeout=0,
220                         return_when=futures.FIRST_COMPLETED)
221             else:
222                 # wait for a request to finish
223                 result = futures.wait(self.futures, timeout=1.0,
224                         return_when=futures.FIRST_COMPLETED)
225 
226             # clean up finished requests
227             for fut in result.done:
228                 self.futures.remove(fut)
229 
230             if not self.is_parent_alive():
231                 break
232 
233             # hanle keepalive timeouts
234             self.murder_keepalived()
235 
236         self.tpool.shutdown(False)
237         self.poller.close()
238 
239         for s in self.sockets:
240             s.close()
241 
242         futures.wait(self.futures, timeout=self.cfg.graceful_timeout)
243 
244     def finish_request(self, fs):
245         if fs.cancelled():
246             self.nr_conns -= 1
247             fs.conn.close()
248             return
249 
250         try:
251             (keepalive, conn) = fs.result()
252             # if the connection should be kept alived add it
253             # to the eventloop and record it
254             if keepalive:
255                 # flag the socket as non blocked
256                 conn.sock.setblocking(False)
257 
258                 # register the connection
259                 conn.set_timeout()
260                 with self._lock:
261                     self._keep.append(conn)
262 
263                     # add the socket to the event loop
264                     self.poller.register(conn.sock, selectors.EVENT_READ,
265                             partial(self.reuse_connection, conn))
266             else:
267                 self.nr_conns -= 1
268                 conn.close()
269         except:
270             # an exception happened, make sure to close the
271             # socket.
272             self.nr_conns -= 1
273             fs.conn.close()
274 
275     def handle(self, conn):
276         keepalive = False
277         req = None
278         try:
279             req = six.next(conn.parser)
280             if not req:
281                 return (False, conn)
282 
283             # handle the request
284             keepalive = self.handle_request(req, conn)
285             if keepalive:
286                 return (keepalive, conn)
287         except http.errors.NoMoreData as e:
288             self.log.debug("Ignored premature client disconnection. %s", e)
289 
290         except StopIteration as e:
291             self.log.debug("Closing connection. %s", e)
292         except ssl.SSLError as e:
293             if e.args[0] == ssl.SSL_ERROR_EOF:
294                 self.log.debug("ssl connection closed")
295                 conn.sock.close()
296             else:
297                 self.log.debug("Error processing SSL request.")
298                 self.handle_error(req, conn.sock, conn.client, e)
299 
300         except EnvironmentError as e:
301             if e.errno not in (errno.EPIPE, errno.ECONNRESET):
302                 self.log.exception("Socket error processing request.")
303             else:
304                 if e.errno == errno.ECONNRESET:
305                     self.log.debug("Ignoring connection reset")
306                 else:
307                     self.log.debug("Ignoring connection epipe")
308         except Exception as e:
309             self.handle_error(req, conn.sock, conn.client, e)
310 
311         return (False, conn)
312 
313     def handle_request(self, req, conn):
314         environ = {}
315         resp = None
316         try:
317             self.cfg.pre_request(self, req)
318             request_start = datetime.now()
319             resp, environ = wsgi.create(req, conn.sock, conn.client,
320                     conn.server, self.cfg)
321             environ["wsgi.multithread"] = True
322             self.nr += 1
323             if self.alive and self.nr >= self.max_requests:
324                 self.log.info("Autorestarting worker after current request.")
325                 resp.force_close()
326                 self.alive = False
327 
328             if not self.cfg.keepalive:
329                 resp.force_close()
330             elif len(self._keep) >= self.max_keepalived:
331                 resp.force_close()
332 
333             respiter = self.wsgi(environ, resp.start_response)
334             try:
335                 if isinstance(respiter, environ['wsgi.file_wrapper']):
336                     resp.write_file(respiter)
337                 else:
338                     for item in respiter:
339                         resp.write(item)
340 
341                 resp.close()
342                 request_time = datetime.now() - request_start
343                 self.log.access(resp, req, environ, request_time)
344             finally:
345                 if hasattr(respiter, "close"):
346                     respiter.close()
347 
348             if resp.should_close():
349                 self.log.debug("Closing connection.")
350                 return False
351         except EnvironmentError:
352             # pass to next try-except level
353             six.reraise(*sys.exc_info())
354         except Exception:
355             if resp and resp.headers_sent:
356                 # If the requests have already been sent, we should close the
357                 # connection to indicate the error.
358                 self.log.exception("Error handling request")
359                 try:
360                     conn.sock.shutdown(socket.SHUT_RDWR)
361                     conn.sock.close()
362                 except EnvironmentError:
363                     pass
364                 raise StopIteration()
365             raise
366         finally:
367             try:
368                 self.cfg.post_request(self, req, environ, resp)
369             except Exception:
370                 self.log.exception("Exception in post_request hook")
371 
372         return True
373
```
--- 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/gunicorn/workers/gthread.py b/gunicorn/workers/gthread.py
--- a/gunicorn/workers/gthread.py
+++ b/gunicorn/workers/gthread.py
@@ -74,11 +74,6 @@
     def close(self):
         util.close(self.sock)
 
-    def __lt__(self, other):
-        return self.timeout < other.timeout
-
-    __cmp__ = __lt__
-
 
 class ThreadWorker(base.Worker):

{"golden_diff": "diff --git a/gunicorn/workers/gthread.py b/gunicorn/workers/gthread.py\n--- a/gunicorn/workers/gthread.py\n+++ b/gunicorn/workers/gthread.py\n@@ -74,11 +74,6 @@\n     def close(self):\n         util.close(self.sock)\n \n-    def __lt__(self, other):\n-        return self.timeout < other.timeout\n-\n-    __cmp__ = __lt__\n-\n \n class ThreadWorker(base.Worker):\n", "issue": "\"Connection refused\" when using keep-alive with gthread\nGunicorn version: 19.7.1 (also tried master branch)\r\n\r\nThis bug could be reproduced with attached files:\r\n- test_http_gunicorn_raw_keep-alive-requests.py (test runner)\r\n- myapp.py (gunicorn app that should be in the same directory)\r\n\r\nTest case starts command:\r\n```\r\ngunicorn --worker-class gthread --workers 1 --threads 20 --keep-alive 9999 \\\r\n           --log-level DEBUG --bind localhost:7777 myapp:app\r\n```\r\nwith logs redirected to `/tmp/_test_gunicorn.out`\r\nand then create separate threads to open http session and send 1000 requests within it.\r\n\r\nIn my case gunicorn server received request `/request/1096` and then reset connection (see `tcp.stream eq 10` from attached `gunicorn_reset_keep_alived_connection.pcapng`\r\n\r\nNOTE: this is race condition so it may happen that all requests finish with success (see myapp.py -> without `time.sleep` it almost never fails) but with that `time.sleep` it usually fails and then script should output line similar to:\r\n`[1] Failed request id=1096 with ('Connection aborted.', error(104, 'Connection reset by peer'))`\r\nit mean that thread sent request `GET /request/1096` and received `Connection reset by peer` (so the last one request from that session that succeed is `GET /request/1095`).\n", "before_files": [{"content": "# -*- coding: utf-8 -\n#\n# This file is part of gunicorn released under the MIT license.\n# See the NOTICE for more information.\n\n# design:\n# a threaded worker accepts connections in the main loop, accepted\n# connections are are added to the thread pool as a connection job. On\n# keepalive connections are put back in the loop waiting for an event.\n# If no event happen after the keep alive timeout, the connectoin is\n# closed.\n\nfrom collections import deque\nfrom datetime import datetime\nimport errno\nfrom functools import partial\nimport os\nimport socket\nimport ssl\nimport sys\nfrom threading import RLock\nimport time\n\nfrom .. import http\nfrom ..http import wsgi\nfrom .. import util\nfrom . import base\nfrom .. import six\n\n\ntry:\n    import concurrent.futures as futures\nexcept ImportError:\n    raise RuntimeError(\"\"\"\n    You need to install the 'futures' package to use this worker with this\n    Python version.\n    \"\"\")\n\ntry:\n    from asyncio import selectors\nexcept ImportError:\n    from gunicorn import selectors\n\n\nclass TConn(object):\n\n    def __init__(self, cfg, sock, client, server):\n        self.cfg = cfg\n        self.sock = sock\n        self.client = client\n        self.server = server\n\n        self.timeout = None\n        self.parser = None\n\n        # set the socket to non blocking\n        self.sock.setblocking(False)\n\n    def init(self):\n        self.sock.setblocking(True)\n        if self.parser is None:\n            # wrap the socket if needed\n            if self.cfg.is_ssl:\n                self.sock = ssl.wrap_socket(self.sock, server_side=True,\n                        **self.cfg.ssl_options)\n\n            # initialize the parser\n            self.parser = http.RequestParser(self.cfg, self.sock)\n\n    def set_timeout(self):\n        # set the timeout\n        self.timeout = time.time() + self.cfg.keepalive\n\n    def close(self):\n        util.close(self.sock)\n\n    def __lt__(self, other):\n        return self.timeout < other.timeout\n\n    __cmp__ = __lt__\n\n\nclass ThreadWorker(base.Worker):\n\n    def __init__(self, *args, **kwargs):\n        super(ThreadWorker, self).__init__(*args, **kwargs)\n        self.worker_connections = self.cfg.worker_connections\n        self.max_keepalived = self.cfg.worker_connections - self.cfg.threads\n        # initialise the pool\n        self.tpool = None\n        self.poller = None\n        self._lock = None\n        self.futures = deque()\n        self._keep = deque()\n        self.nr_conns = 0\n\n    @classmethod\n    def check_config(cls, cfg, log):\n        max_keepalived = cfg.worker_connections - cfg.threads\n\n        if max_keepalived <= 0 and cfg.keepalive:\n            log.warning(\"No keepalived connections can be handled. \" +\n                    \"Check the number of worker connections and threads.\")\n\n    def init_process(self):\n        self.tpool = futures.ThreadPoolExecutor(max_workers=self.cfg.threads)\n        self.poller = selectors.DefaultSelector()\n        self._lock = RLock()\n        super(ThreadWorker, self).init_process()\n\n    def handle_quit(self, sig, frame):\n        self.alive = False\n        # worker_int callback\n        self.cfg.worker_int(self)\n        self.tpool.shutdown(False)\n        time.sleep(0.1)\n        sys.exit(0)\n\n    def _wrap_future(self, fs, conn):\n        fs.conn = conn\n        self.futures.append(fs)\n        fs.add_done_callback(self.finish_request)\n\n    def enqueue_req(self, conn):\n        conn.init()\n        # submit the connection to a worker\n        fs = self.tpool.submit(self.handle, conn)\n        self._wrap_future(fs, conn)\n\n    def accept(self, server, listener):\n        try:\n            sock, client = listener.accept()\n            # initialize the connection object\n            conn = TConn(self.cfg, sock, client, server)\n            self.nr_conns += 1\n            # enqueue the job\n            self.enqueue_req(conn)\n        except EnvironmentError as e:\n            if e.errno not in (errno.EAGAIN,\n                    errno.ECONNABORTED, errno.EWOULDBLOCK):\n                raise\n\n    def reuse_connection(self, conn, client):\n        with self._lock:\n            # unregister the client from the poller\n            self.poller.unregister(client)\n            # remove the connection from keepalive\n            try:\n                self._keep.remove(conn)\n            except ValueError:\n                # race condition\n                return\n\n        # submit the connection to a worker\n        self.enqueue_req(conn)\n\n    def murder_keepalived(self):\n        now = time.time()\n        while True:\n            with self._lock:\n                try:\n                    # remove the connection from the queue\n                    conn = self._keep.popleft()\n                except IndexError:\n                    break\n\n            delta = conn.timeout - now\n            if delta > 0:\n                # add the connection back to the queue\n                with self._lock:\n                    self._keep.appendleft(conn)\n                break\n            else:\n                self.nr_conns -= 1\n                # remove the socket from the poller\n                with self._lock:\n                    try:\n                        self.poller.unregister(conn.sock)\n                    except EnvironmentError as e:\n                        if e.errno != errno.EBADF:\n                            raise\n                    except KeyError:\n                        # already removed by the system, continue\n                        pass\n\n                # close the socket\n                conn.close()\n\n    def is_parent_alive(self):\n        # If our parent changed then we shut down.\n        if self.ppid != os.getppid():\n            self.log.info(\"Parent changed, shutting down: %s\", self)\n            return False\n        return True\n\n    def run(self):\n        # init listeners, add them to the event loop\n        for sock in self.sockets:\n            sock.setblocking(False)\n            # a race condition during graceful shutdown may make the listener\n            # name unavailable in the request handler so capture it once here\n            server = sock.getsockname()\n            acceptor = partial(self.accept, server)\n            self.poller.register(sock, selectors.EVENT_READ, acceptor)\n\n        while self.alive:\n            # notify the arbiter we are alive\n            self.notify()\n\n            # can we accept more connections?\n            if self.nr_conns < self.worker_connections:\n                # wait for an event\n                events = self.poller.select(1.0)\n                for key, _ in events:\n                    callback = key.data\n                    callback(key.fileobj)\n\n                # check (but do not wait) for finished requests\n                result = futures.wait(self.futures, timeout=0,\n                        return_when=futures.FIRST_COMPLETED)\n            else:\n                # wait for a request to finish\n                result = futures.wait(self.futures, timeout=1.0,\n                        return_when=futures.FIRST_COMPLETED)\n\n            # clean up finished requests\n            for fut in result.done:\n                self.futures.remove(fut)\n\n            if not self.is_parent_alive():\n                break\n\n            # hanle keepalive timeouts\n            self.murder_keepalived()\n\n        self.tpool.shutdown(False)\n        self.poller.close()\n\n        for s in self.sockets:\n            s.close()\n\n        futures.wait(self.futures, timeout=self.cfg.graceful_timeout)\n\n    def finish_request(self, fs):\n        if fs.cancelled():\n            self.nr_conns -= 1\n            fs.conn.close()\n            return\n\n        try:\n            (keepalive, conn) = fs.result()\n            # if the connection should be kept alived add it\n            # to the eventloop and record it\n            if keepalive:\n                # flag the socket as non blocked\n                conn.sock.setblocking(False)\n\n                # register the connection\n                conn.set_timeout()\n                with self._lock:\n                    self._keep.append(conn)\n\n                    # add the socket to the event loop\n                    self.poller.register(conn.sock, selectors.EVENT_READ,\n                            partial(self.reuse_connection, conn))\n            else:\n                self.nr_conns -= 1\n                conn.close()\n        except:\n            # an exception happened, make sure to close the\n            # socket.\n            self.nr_conns -= 1\n            fs.conn.close()\n\n    def handle(self, conn):\n        keepalive = False\n        req = None\n        try:\n            req = six.next(conn.parser)\n            if not req:\n                return (False, conn)\n\n            # handle the request\n            keepalive = self.handle_request(req, conn)\n            if keepalive:\n                return (keepalive, conn)\n        except http.errors.NoMoreData as e:\n            self.log.debug(\"Ignored premature client disconnection. %s\", e)\n\n        except StopIteration as e:\n            self.log.debug(\"Closing connection. %s\", e)\n        except ssl.SSLError as e:\n            if e.args[0] == ssl.SSL_ERROR_EOF:\n                self.log.debug(\"ssl connection closed\")\n                conn.sock.close()\n            else:\n                self.log.debug(\"Error processing SSL request.\")\n                self.handle_error(req, conn.sock, conn.client, e)\n\n        except EnvironmentError as e:\n            if e.errno not in (errno.EPIPE, errno.ECONNRESET):\n                self.log.exception(\"Socket error processing request.\")\n            else:\n                if e.errno == errno.ECONNRESET:\n                    self.log.debug(\"Ignoring connection reset\")\n                else:\n                    self.log.debug(\"Ignoring connection epipe\")\n        except Exception as e:\n            self.handle_error(req, conn.sock, conn.client, e)\n\n        return (False, conn)\n\n    def handle_request(self, req, conn):\n        environ = {}\n        resp = None\n        try:\n            self.cfg.pre_request(self, req)\n            request_start = datetime.now()\n            resp, environ = wsgi.create(req, conn.sock, conn.client,\n                    conn.server, self.cfg)\n            environ[\"wsgi.multithread\"] = True\n            self.nr += 1\n            if self.alive and self.nr >= self.max_requests:\n                self.log.info(\"Autorestarting worker after current request.\")\n                resp.force_close()\n                self.alive = False\n\n            if not self.cfg.keepalive:\n                resp.force_close()\n            elif len(self._keep) >= self.max_keepalived:\n                resp.force_close()\n\n            respiter = self.wsgi(environ, resp.start_response)\n            try:\n                if isinstance(respiter, environ['wsgi.file_wrapper']):\n                    resp.write_file(respiter)\n                else:\n                    for item in respiter:\n                        resp.write(item)\n\n                resp.close()\n                request_time = datetime.now() - request_start\n                self.log.access(resp, req, environ, request_time)\n            finally:\n                if hasattr(respiter, \"close\"):\n                    respiter.close()\n\n            if resp.should_close():\n                self.log.debug(\"Closing connection.\")\n                return False\n        except EnvironmentError:\n            # pass to next try-except level\n            six.reraise(*sys.exc_info())\n        except Exception:\n            if resp and resp.headers_sent:\n                # If the requests have already been sent, we should close the\n                # connection to indicate the error.\n                self.log.exception(\"Error handling request\")\n                try:\n                    conn.sock.shutdown(socket.SHUT_RDWR)\n                    conn.sock.close()\n                except EnvironmentError:\n                    pass\n                raise StopIteration()\n            raise\n        finally:\n            try:\n                self.cfg.post_request(self, req, environ, resp)\n            except Exception:\n                self.log.exception(\"Exception in post_request hook\")\n\n        return True\n", "path": "gunicorn/workers/gthread.py"}], "after_files": [{"content": "# -*- coding: utf-8 -\n#\n# This file is part of gunicorn released under the MIT license.\n# See the NOTICE for more information.\n\n# design:\n# a threaded worker accepts connections in the main loop, accepted\n# connections are are added to the thread pool as a connection job. On\n# keepalive connections are put back in the loop waiting for an event.\n# If no event happen after the keep alive timeout, the connectoin is\n# closed.\n\nfrom collections import deque\nfrom datetime import datetime\nimport errno\nfrom functools import partial\nimport os\nimport socket\nimport ssl\nimport sys\nfrom threading import RLock\nimport time\n\nfrom .. import http\nfrom ..http import wsgi\nfrom .. import util\nfrom . import base\nfrom .. import six\n\n\ntry:\n    import concurrent.futures as futures\nexcept ImportError:\n    raise RuntimeError(\"\"\"\n    You need to install the 'futures' package to use this worker with this\n    Python version.\n    \"\"\")\n\ntry:\n    from asyncio import selectors\nexcept ImportError:\n    from gunicorn import selectors\n\n\nclass TConn(object):\n\n    def __init__(self, cfg, sock, client, server):\n        self.cfg = cfg\n        self.sock = sock\n        self.client = client\n        self.server = server\n\n        self.timeout = None\n        self.parser = None\n\n        # set the socket to non blocking\n        self.sock.setblocking(False)\n\n    def init(self):\n        self.sock.setblocking(True)\n        if self.parser is None:\n            # wrap the socket if needed\n            if self.cfg.is_ssl:\n                self.sock = ssl.wrap_socket(self.sock, server_side=True,\n                        **self.cfg.ssl_options)\n\n            # initialize the parser\n            self.parser = http.RequestParser(self.cfg, self.sock)\n\n    def set_timeout(self):\n        # set the timeout\n        self.timeout = time.time() + self.cfg.keepalive\n\n    def close(self):\n        util.close(self.sock)\n\n\nclass ThreadWorker(base.Worker):\n\n    def __init__(self, *args, **kwargs):\n        super(ThreadWorker, self).__init__(*args, **kwargs)\n        self.worker_connections = self.cfg.worker_connections\n        self.max_keepalived = self.cfg.worker_connections - self.cfg.threads\n        # initialise the pool\n        self.tpool = None\n        self.poller = None\n        self._lock = None\n        self.futures = deque()\n        self._keep = deque()\n        self.nr_conns = 0\n\n    @classmethod\n    def check_config(cls, cfg, log):\n        max_keepalived = cfg.worker_connections - cfg.threads\n\n        if max_keepalived <= 0 and cfg.keepalive:\n            log.warning(\"No keepalived connections can be handled. \" +\n                    \"Check the number of worker connections and threads.\")\n\n    def init_process(self):\n        self.tpool = futures.ThreadPoolExecutor(max_workers=self.cfg.threads)\n        self.poller = selectors.DefaultSelector()\n        self._lock = RLock()\n        super(ThreadWorker, self).init_process()\n\n    def handle_quit(self, sig, frame):\n        self.alive = False\n        # worker_int callback\n        self.cfg.worker_int(self)\n        self.tpool.shutdown(False)\n        time.sleep(0.1)\n        sys.exit(0)\n\n    def _wrap_future(self, fs, conn):\n        fs.conn = conn\n        self.futures.append(fs)\n        fs.add_done_callback(self.finish_request)\n\n    def enqueue_req(self, conn):\n        conn.init()\n        # submit the connection to a worker\n        fs = self.tpool.submit(self.handle, conn)\n        self._wrap_future(fs, conn)\n\n    def accept(self, server, listener):\n        try:\n            sock, client = listener.accept()\n            # initialize the connection object\n            conn = TConn(self.cfg, sock, client, server)\n            self.nr_conns += 1\n            # enqueue the job\n            self.enqueue_req(conn)\n        except EnvironmentError as e:\n            if e.errno not in (errno.EAGAIN,\n                    errno.ECONNABORTED, errno.EWOULDBLOCK):\n                raise\n\n    def reuse_connection(self, conn, client):\n        with self._lock:\n            # unregister the client from the poller\n            self.poller.unregister(client)\n            # remove the connection from keepalive\n            try:\n                self._keep.remove(conn)\n            except ValueError:\n                # race condition\n                return\n\n        # submit the connection to a worker\n        self.enqueue_req(conn)\n\n    def murder_keepalived(self):\n        now = time.time()\n        while True:\n            with self._lock:\n                try:\n                    # remove the connection from the queue\n                    conn = self._keep.popleft()\n                except IndexError:\n                    break\n\n            delta = conn.timeout - now\n            if delta > 0:\n                # add the connection back to the queue\n                with self._lock:\n                    self._keep.appendleft(conn)\n                break\n            else:\n                self.nr_conns -= 1\n                # remove the socket from the poller\n                with self._lock:\n                    try:\n                        self.poller.unregister(conn.sock)\n                    except EnvironmentError as e:\n                        if e.errno != errno.EBADF:\n                            raise\n                    except KeyError:\n                        # already removed by the system, continue\n                        pass\n\n                # close the socket\n                conn.close()\n\n    def is_parent_alive(self):\n        # If our parent changed then we shut down.\n        if self.ppid != os.getppid():\n            self.log.info(\"Parent changed, shutting down: %s\", self)\n            return False\n        return True\n\n    def run(self):\n        # init listeners, add them to the event loop\n        for sock in self.sockets:\n            sock.setblocking(False)\n            # a race condition during graceful shutdown may make the listener\n            # name unavailable in the request handler so capture it once here\n            server = sock.getsockname()\n            acceptor = partial(self.accept, server)\n            self.poller.register(sock, selectors.EVENT_READ, acceptor)\n\n        while self.alive:\n            # notify the arbiter we are alive\n            self.notify()\n\n            # can we accept more connections?\n            if self.nr_conns < self.worker_connections:\n                # wait for an event\n                events = self.poller.select(1.0)\n                for key, _ in events:\n                    callback = key.data\n                    callback(key.fileobj)\n\n                # check (but do not wait) for finished requests\n                result = futures.wait(self.futures, timeout=0,\n                        return_when=futures.FIRST_COMPLETED)\n            else:\n                # wait for a request to finish\n                result = futures.wait(self.futures, timeout=1.0,\n                        return_when=futures.FIRST_COMPLETED)\n\n            # clean up finished requests\n            for fut in result.done:\n                self.futures.remove(fut)\n\n            if not self.is_parent_alive():\n                break\n\n            # hanle keepalive timeouts\n            self.murder_keepalived()\n\n        self.tpool.shutdown(False)\n        self.poller.close()\n\n        for s in self.sockets:\n            s.close()\n\n        futures.wait(self.futures, timeout=self.cfg.graceful_timeout)\n\n    def finish_request(self, fs):\n        if fs.cancelled():\n            self.nr_conns -= 1\n            fs.conn.close()\n            return\n\n        try:\n            (keepalive, conn) = fs.result()\n            # if the connection should be kept alived add it\n            # to the eventloop and record it\n            if keepalive:\n                # flag the socket as non blocked\n                conn.sock.setblocking(False)\n\n                # register the connection\n                conn.set_timeout()\n                with self._lock:\n                    self._keep.append(conn)\n\n                    # add the socket to the event loop\n                    self.poller.register(conn.sock, selectors.EVENT_READ,\n                            partial(self.reuse_connection, conn))\n            else:\n                self.nr_conns -= 1\n                conn.close()\n        except:\n            # an exception happened, make sure to close the\n            # socket.\n            self.nr_conns -= 1\n            fs.conn.close()\n\n    def handle(self, conn):\n        keepalive = False\n        req = None\n        try:\n            req = six.next(conn.parser)\n            if not req:\n                return (False, conn)\n\n            # handle the request\n            keepalive = self.handle_request(req, conn)\n            if keepalive:\n                return (keepalive, conn)\n        except http.errors.NoMoreData as e:\n            self.log.debug(\"Ignored premature client disconnection. %s\", e)\n\n        except StopIteration as e:\n            self.log.debug(\"Closing connection. %s\", e)\n        except ssl.SSLError as e:\n            if e.args[0] == ssl.SSL_ERROR_EOF:\n                self.log.debug(\"ssl connection closed\")\n                conn.sock.close()\n            else:\n                self.log.debug(\"Error processing SSL request.\")\n                self.handle_error(req, conn.sock, conn.client, e)\n\n        except EnvironmentError as e:\n            if e.errno not in (errno.EPIPE, errno.ECONNRESET):\n                self.log.exception(\"Socket error processing request.\")\n            else:\n                if e.errno == errno.ECONNRESET:\n                    self.log.debug(\"Ignoring connection reset\")\n                else:\n                    self.log.debug(\"Ignoring connection epipe\")\n        except Exception as e:\n            self.handle_error(req, conn.sock, conn.client, e)\n\n        return (False, conn)\n\n    def handle_request(self, req, conn):\n        environ = {}\n        resp = None\n        try:\n            self.cfg.pre_request(self, req)\n            request_start = datetime.now()\n            resp, environ = wsgi.create(req, conn.sock, conn.client,\n                    conn.server, self.cfg)\n            environ[\"wsgi.multithread\"] = True\n            self.nr += 1\n            if self.alive and self.nr >= self.max_requests:\n                self.log.info(\"Autorestarting worker after current request.\")\n                resp.force_close()\n                self.alive = False\n\n            if not self.cfg.keepalive:\n                resp.force_close()\n            elif len(self._keep) >= self.max_keepalived:\n                resp.force_close()\n\n            respiter = self.wsgi(environ, resp.start_response)\n            try:\n                if isinstance(respiter, environ['wsgi.file_wrapper']):\n                    resp.write_file(respiter)\n                else:\n                    for item in respiter:\n                        resp.write(item)\n\n                resp.close()\n                request_time = datetime.now() - request_start\n                self.log.access(resp, req, environ, request_time)\n            finally:\n                if hasattr(respiter, \"close\"):\n                    respiter.close()\n\n            if resp.should_close():\n                self.log.debug(\"Closing connection.\")\n                return False\n        except EnvironmentError:\n            # pass to next try-except level\n            six.reraise(*sys.exc_info())\n        except Exception:\n            if resp and resp.headers_sent:\n                # If the requests have already been sent, we should close the\n                # connection to indicate the error.\n                self.log.exception(\"Error handling request\")\n                try:\n                    conn.sock.shutdown(socket.SHUT_RDWR)\n                    conn.sock.close()\n                except EnvironmentError:\n                    pass\n                raise StopIteration()\n            raise\n        finally:\n            try:\n                self.cfg.post_request(self, req, environ, resp)\n            except Exception:\n                self.log.exception(\"Exception in post_request hook\")\n\n        return True\n", "path": "gunicorn/workers/gthread.py"}]}