Wait until buffer flush

helper/monitor_open_files.sh

@@ -20,15 +20,20 @@ if [[ -z "$PROXY_PY_PID" ]]; then
   exit 1
 fi
 
-OPEN_FILES_BY_MAIN=$(lsof -p "$PROXY_PY_PID" | wc -l)
-echo "[$PROXY_PY_PID] Main process: $OPEN_FILES_BY_MAIN"
+while true;
+do
+  OPEN_FILES_BY_MAIN=$(lsof -p "$PROXY_PY_PID" | wc -l)
+  echo "[$PROXY_PY_PID] Main process: $OPEN_FILES_BY_MAIN"
 
-pgrep -P "$PROXY_PY_PID" | while read -r acceptorPid; do
-  OPEN_FILES_BY_ACCEPTOR=$(lsof -p "$acceptorPid" | wc -l)
-  echo "[$acceptorPid] Acceptor process: $OPEN_FILES_BY_ACCEPTOR"
+  pgrep -P "$PROXY_PY_PID" | while read -r acceptorPid; do
+    OPEN_FILES_BY_ACCEPTOR=$(lsof -p "$acceptorPid" | wc -l)
+    echo "[$acceptorPid] Acceptor process: $OPEN_FILES_BY_ACCEPTOR"
 
-  pgrep -P "$acceptorPid" | while read -r childPid; do
-    OPEN_FILES_BY_CHILD_PROC=$(lsof -p "$childPid" | wc -l)
-    echo "  [$childPid] child process: $OPEN_FILES_BY_CHILD_PROC"
+    pgrep -P "$acceptorPid" | while read -r childPid; do
+      OPEN_FILES_BY_CHILD_PROC=$(lsof -p "$childPid" | wc -l)
+      echo "  [$childPid] child process: $OPEN_FILES_BY_CHILD_PROC"
+    done
   done
-done
+
+  sleep 1
+done

proxy/http/handler.py

@@ -37,8 +37,6 @@ class HttpProtocolHandler(BaseTcpServerHandler[HttpClientConnection]):
     Accepts `Client` connection and delegates to HttpProtocolHandlerPlugin.
     """
 
-    server_teared = False
-
     def __init__(self, *args: Any, **kwargs: Any):
         super().__init__(*args, **kwargs)
         self.start_time: float = time.time()
@@ -131,6 +129,9 @@ async def get_events(self) -> SelectableEvents:
                     events[wfileno] |= selectors.EVENT_WRITE
         return events
 
+    writes_teardown: bool = False
+    reads_teardown: bool = False
+
     # We override super().handle_events and never call it
     async def handle_events(
             self,
@@ -139,26 +140,26 @@ async def handle_events(
     ) -> bool:
         """Returns True if proxy must tear down."""
         # Flush buffer for ready to write sockets
-        teardown = await self.handle_writables(writables)
-        if teardown:
-            return True
-        # Invoke plugin.write_to_descriptors
-        if self.plugin:
-            teardown = await self.plugin.write_to_descriptors(writables)
-            if teardown:
-                return True
-        if not self.server_teared:
-            # Read from ready to read sockets
-            teardown = await self.handle_readables(readables)
-            if teardown:
-                self.server_teared = True
-            # Invoke plugin.read_from_descriptors
+        self.writes_teardown = await self.handle_writables(writables)
+        if not self.writes_teardown:
+            # Invoke plugin.write_to_descriptors
             if self.plugin:
-                teardown = await self.plugin.read_from_descriptors(readables)
-                if teardown:
-                    self.server_teared = True
-        if self.server_teared and not self.work.has_buffer():
-            # Wait until client buffer flushed when server teared down.
+                self.writes_teardown = await self.plugin.write_to_descriptors(writables)
+            if not self.writes_teardown:
+                # Read from ready to read sockets
+                self.reads_teardown = await self.handle_readables(readables)
+                if not self.reads_teardown:
+                    # Invoke plugin.read_from_descriptors
+                    if self.plugin:
+                        self.reads_teardown = await self.plugin.read_from_descriptors(
+                            readables
+                        )
+        # Wait until client buffer has flushed when reads has teared down, but we can still write
+        if (
+            self.reads_teardown
+            and not self.writes_teardown
+            and not self.work.has_buffer()
+        ):
             return True
         return False
 

proxy/http/proxy/server.py

@@ -175,7 +175,11 @@ async def get_descriptors(self) -> Descriptors:
         return r, w
 
     async def write_to_descriptors(self, w: Writables) -> bool:
-        if (self.upstream and self.upstream.connection.fileno() not in w) or not self.upstream:
+        if (
+            self.upstream
+            and not self.upstream.closed
+            and self.upstream.connection.fileno() not in w
+        ) or not self.upstream:
             # Currently, we just call write/read block of each plugins.  It is
             # plugins responsibility to ignore this callback, if passed descriptors
             # doesn't contain the descriptor they registered.
@@ -208,9 +212,9 @@ async def write_to_descriptors(self, w: Writables) -> bool:
 
     async def read_from_descriptors(self, r: Readables) -> bool:
         if (
-            self.upstream and not
-            self.upstream.closed and
-            self.upstream.connection.fileno() not in r
+            self.upstream
+            and not self.upstream.closed
+            and self.upstream.connection.fileno() not in r
         ) or not self.upstream:
             # Currently, we just call write/read block of each plugins.  It is
             # plugins responsibility to ignore this callback, if passed descriptors