3227: shutdown socket when worker is not alive

gunicorn/http/message.py

@@ -17,6 +17,7 @@
 from gunicorn.http.errors import InvalidProxyLine, ForbiddenProxyRequest
 from gunicorn.http.errors import InvalidSchemeHeaders
 from gunicorn.util import bytes_to_str, split_request_uri
+from gunicorn.util import Status
 
 MAX_REQUEST_LINE = 8190
 MAX_HEADERS = 32768
@@ -31,7 +32,7 @@
 
 
 class Message(object):
-    def __init__(self, cfg, unreader, peer_addr):
+    def __init__(self, cfg, unreader, peer_addr, status=None):
         self.cfg = cfg
         self.unreader = unreader
         self.peer_addr = peer_addr
@@ -42,6 +43,7 @@ def __init__(self, cfg, unreader, peer_addr):
         self.body = None
         self.scheme = "https" if cfg.is_ssl else "http"
         self.must_close = False
+        self.status = status if status else Status()
 
         # set headers limits
         self.limit_request_fields = cfg.limit_request_fields
@@ -227,6 +229,8 @@ def set_body_reader(self):
     def should_close(self):
         if self.must_close:
             return True
+        if not status.is_alive():
+            return True
         for (h, v) in self.headers:
             if h == "CONNECTION":
                 v = v.lower().strip(" \t")
@@ -239,7 +243,7 @@ def should_close(self):
 
 
 class Request(Message):
-    def __init__(self, cfg, unreader, peer_addr, req_number=1):
+    def __init__(self, cfg, unreader, peer_addr, req_number=1, status=None):
         self.method = None
         self.uri = None
         self.path = None
@@ -254,7 +258,7 @@ def __init__(self, cfg, unreader, peer_addr, req_number=1):
 
         self.req_number = req_number
         self.proxy_protocol_info = None
-        super().__init__(cfg, unreader, peer_addr)
+        super().__init__(cfg, unreader, peer_addr, status=status)
 
     def get_data(self, unreader, buf, stop=False):
         data = unreader.read()

gunicorn/http/parser.py

@@ -5,14 +5,16 @@
 
 from gunicorn.http.message import Request
 from gunicorn.http.unreader import SocketUnreader, IterUnreader
+from gunicorn.util import Status
 
 
 class Parser(object):
 
     mesg_class = None
 
-    def __init__(self, cfg, source, source_addr):
+    def __init__(self, cfg, source, source_addr, status=None):
         self.cfg = cfg
+        self.status = status if status else Status()
         if hasattr(source, "recv"):
             self.unreader = SocketUnreader(source)
         else:
@@ -39,7 +41,7 @@ def __next__(self):
 
         # Parse the next request
         self.req_count += 1
-        self.mesg = self.mesg_class(self.cfg, self.unreader, self.source_addr, self.req_count)
+        self.mesg = self.mesg_class(self.cfg, self.unreader, self.source_addr, self.req_count, status=self.status)
         if not self.mesg:
             raise StopIteration()
         return self.mesg

gunicorn/util.py

@@ -652,3 +652,20 @@ def bytes_to_str(b):
 
 def unquote_to_wsgi_str(string):
     return urllib.parse.unquote_to_bytes(string).decode('latin-1')
+
+
+class Status:
+    """
+    This class's object holds information about the worker's status.
+    The status is represented by an object instead of a simple type to enable
+    sharing between different parts of the application.
+    """
+
+    def __init__(self, is_alive=True):
+        self._is_alive = is_alive
+
+    def is_alive(self):
+        return self._is_alive
+
+    def set_alive(self, is_alive):
+        self._is_alive = is_alive

gunicorn/workers/base.py

@@ -24,6 +24,7 @@
 from gunicorn.http.wsgi import Response, default_environ
 from gunicorn.reloader import reloader_engines
 from gunicorn.workers.workertmp import WorkerTmp
+from gunicorn.util import Status
 
 
 class Worker(object):
@@ -59,13 +60,22 @@ def __init__(self, age, ppid, sockets, app, timeout, cfg, log):
         else:
             self.max_requests = sys.maxsize
 
+        self.status = Status()
         self.alive = True
         self.log = log
         self.tmp = WorkerTmp(cfg)
 
     def __str__(self):
         return "<Worker %s>" % self.pid
 
+    @property
+    def alive(self):
+        return self.status.is_alive()
+
+    @alive.setter
+    def alive(self, value):
+        self.status.set_alive(value)
+
     def notify(self):
         """\
         Your worker subclass must arrange to have this method called

gunicorn/workers/base_async.py

@@ -33,7 +33,7 @@ def is_already_handled(self, respiter):
     def handle(self, listener, client, addr):
         req = None
         try:
-            parser = http.RequestParser(self.cfg, client, addr)
+            parser = http.RequestParser(self.cfg, client, addr, status=self.status)
             try:
                 listener_name = listener.getsockname()
                 if not self.cfg.keepalive: