jingrow/gunicorn
1
0
Fork 0
mirror of https://github.com/frappe/gunicorn.git synced 2026-01-14 11:09:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

make sure we can alwas accept connection.

Browse Source 
This is done by limiting the number of kept alived requests to MAX WORKER
CONNECTIONS - N Threads so at any time we can accept N connection.
This commit is contained in:
benoitc 2014-12-20 16:52:22 +01:00
parent fcd9d04515
commit 1f92511430
1 changed files with 39 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

67
gunicorn/workers/gthread.py
View File

@ -18,6 +18,7 @@ import os
import socket import socket
import ssl import ssl
import sys import sys
from threading import RLock
import time import time
from .. import http from .. import http
@ -84,16 +85,19 @@ class ThreadWorker(base.Worker):
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super(ThreadWorker, self).__init__(*args, **kwargs) super(ThreadWorker, self).__init__(*args, **kwargs)
self.worker_connections = self.cfg.worker_connections self.worker_connections = self.cfg.worker_connections
self.max_keepalived = self.cfg.worker_connections - self.cfg.threads
# initialise the pool # initialise the pool
self.tpool = None self.tpool = None
self.poller = None self.poller = None
self._lock = None
self.futures = deque() self.futures = deque()
self._keep = deque() self._keep = deque()
def init_process(self): def init_process(self):
self.tpool = futures.ThreadPoolExecutor(max_workers=self.cfg.threads) self.tpool = futures.ThreadPoolExecutor(max_workers=self.cfg.threads)
self.poller = selectors.DefaultSelector() self.poller = selectors.DefaultSelector()
self._lock = RLock()
super(ThreadWorker, self).init_process() super(ThreadWorker, self).init_process()
def _wrap_future(self, fs, conn): def _wrap_future(self, fs, conn):
@ -121,15 +125,15 @@ class ThreadWorker(base.Worker):
raise raise
def reuse_connection(self, conn, client): def reuse_connection(self, conn, client):
# unregister the client from the poller with self._lock:
self.poller.unregister(client) # unregister the client from the poller
self.poller.unregister(client)
# remove the connection from keepalive # remove the connection from keepalive
try: try:
self._keep.remove(conn) self._keep.remove(conn)
except ValueError: except ValueError:
# race condition # race condition
return return
# submit the connection to a worker # submit the connection to a worker
self.enqueue_req(conn) self.enqueue_req(conn)
@ -137,26 +141,28 @@ class ThreadWorker(base.Worker):
def murder_keepalived(self): def murder_keepalived(self):
now = time.time() now = time.time()
while True: while True:
try: with self._lock:
# remove the connection from the queue try:
conn = self._keep.popleft() # remove the connection from the queue
except IndexError: conn = self._keep.popleft()
break except IndexError:
break
delta = conn.timeout - now delta = conn.timeout - now
if delta > 0: if delta > 0:
# add the connection back to the queue # add the connection back to the queue
self._keep.appendleft(conn) with self._lock:
self._keep.appendleft(conn)
break break
else: else:
self.nr -= 1 self.nr -= 1
# remove the socket from the poller # remove the socket from the poller
try: with self._lock:
self.poller.unregister(conn.sock) try:
except socket.error as e: self.poller.unregister(conn.sock)
if e.args[0] == errno.EBADF: except socket.error as e:
pass if e.args[0] != errno.EBADF:
raise raise
# close the socket # close the socket
conn.close() conn.close()
@ -200,12 +206,14 @@ class ThreadWorker(base.Worker):
return_when=futures.FIRST_COMPLETED) return_when=futures.FIRST_COMPLETED)
if not result.done: if not result.done:
self.tpool.shutdown(False) break
self.poller.close()
return
else: else:
[self.futures.remove(f) for f in result.done] [self.futures.remove(f) for f in result.done]
self.tpool.shutdown(False)
self.poller.close()
def finish_request(self, fs): def finish_request(self, fs):
if fs.cancelled(): if fs.cancelled():
@ -222,11 +230,12 @@ class ThreadWorker(base.Worker):
# register the connection # register the connection
conn.set_timeout() conn.set_timeout()
self._keep.append(conn) with self._lock:
self._keep.append(conn)
# add the socket to the event loop # add the socket to the event loop
self.poller.register(conn.sock, selectors.EVENT_READ, self.poller.register(conn.sock, selectors.EVENT_READ,
partial(self.reuse_connection, conn)) partial(self.reuse_connection, conn))
else: else:
self.nr -= 1 self.nr -= 1
conn.close() conn.close()
@ -291,6 +300,8 @@ class ThreadWorker(base.Worker):
if not self.cfg.keepalive: if not self.cfg.keepalive:
resp.force_close() resp.force_close()
elif len(self._keep) >= self.max_keepalived:
resp.force_close()
respiter = self.wsgi(environ, resp.start_response) respiter = self.wsgi(environ, resp.start_response)
try: try: