Repository: abhinavsingh/proxy.py
Pull request #763 (2021-11-19 21:11:36)

Enhancements

codecov[bot]: # [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) Report
> Merging [#763](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (88a53bd) into [develop](https://codecov.io/gh/abhinavsingh/proxy.py/commit/5eb4fa4acd9c5a0a3389437fffe1f38ca1736a10?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) (5eb4fa4) will **decrease** coverage by `0.17%`.
> The diff coverage is `95.65%`.

> :exclamation: Current head 88a53bd differs from pull request most recent head 7a9a340. Consider uploading reports for the commit 7a9a340 to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/graphs/tree.svg?width=650&height=150&src=pr&token=Zh9J7b4la2&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)

```diff
@@             Coverage Diff             @@
##           develop     #763      +/-   ##
===========================================
- Coverage    88.12%   87.94%   -0.18%     
===========================================
  Files          125      125              
  Lines         5422     5426       +4     
  Branches       432      434       +2     
===========================================
- Hits          4778     4772       -6     
- Misses         558      564       +6     
- Partials        86       90       +4     
```

| Flag | Coverage Δ | |
|---|---|---|
| pytest | `87.89% <95.65%> (-0.11%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) | Coverage Δ | |
|---|---|---|
| [proxy/proxy.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvcHJveHkucHk=) | `100.00% <ø> (ø)` | |
| [tests/core/test\_acceptor.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvY29yZS90ZXN0X2FjY2VwdG9yLnB5) | `100.00% <ø> (ø)` | |
| [tests/core/test\_acceptor\_pool.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvY29yZS90ZXN0X2FjY2VwdG9yX3Bvb2wucHk=) | `100.00% <ø> (ø)` | |
| [tests/test\_main.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvdGVzdF9tYWluLnB5) | `100.00% <ø> (ø)` | |
| [proxy/core/acceptor/executors.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9hY2NlcHRvci9leGVjdXRvcnMucHk=) | `89.61% <94.44%> (+0.56%)` | :arrow_up: |
| [proxy/common/utils.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29tbW9uL3V0aWxzLnB5) | `90.98% <100.00%> (-2.57%)` | :arrow_down: |
| [proxy/core/acceptor/acceptor.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9hY2NlcHRvci9hY2NlcHRvci5weQ==) | `98.38% <100.00%> (+0.02%)` | :arrow_up: |
| [proxy/core/acceptor/pool.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-cHJveHkvY29yZS9hY2NlcHRvci9wb29sLnB5) | `94.23% <100.00%> (+0.11%)` | :arrow_up: |
| [tests/http/test\_proxy\_protocol.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvaHR0cC90ZXN0X3Byb3h5X3Byb3RvY29sLnB5) | `96.00% <0.00%> (-4.00%)` | :arrow_down: |
| [tests/test\_set\_open\_file\_limit.py](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh#diff-dGVzdHMvdGVzdF9zZXRfb3Blbl9maWxlX2xpbWl0LnB5) | `96.42% <0.00%> (-3.58%)` | :arrow_down: |
| ... and [4 more](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Last update [5eb4fa4...7a9a340](https://codecov.io/gh/abhinavsingh/proxy.py/pull/763?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Abhinav+Singh).


Patch:
diff --git a/Makefile b/Makefile
index d34b72c..f522d01 100644
--- a/Makefile
+++ b/Makefile
@@ -128,7 +128,16 @@ lib-coverage:
 	$(OPEN) htmlcov/index.html
 
 lib-profile:
-	sudo py-spy record -o profile.svg -t -F -s -- python -m proxy
+	sudo py-spy record \
+		-o profile.svg \
+		-t -F -s -- \
+		python -m proxy \
+			--num-acceptors 1 \
+			--num-workers 1 \
+			--disable-http-proxy \
+			--enable-web-server \
+			--plugin proxy.plugin.WebServerPlugin \
+			--log-file /tmp/proxy.log
 
 devtools:
 	pushd dashboard && npm run devtools && popd
diff --git a/docs/conf.py b/docs/conf.py
index 7e56b35..ec96d7c 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -273,4 +273,5 @@ nitpick_ignore = [
     (_py_class_role, 'unittest.result.TestResult'),
     (_py_class_role, 'UUID'),
     (_py_class_role, 'WebsocketFrame'),
+    (_py_class_role, 'Url'),
 ]
diff --git a/proxy/common/utils.py b/proxy/common/utils.py
index 14a1623..b25b680 100644
--- a/proxy/common/utils.py
+++ b/proxy/common/utils.py
@@ -180,12 +180,10 @@ def find_http_line(raw: bytes) -> Tuple[Optional[bytes], bytes]:
     """Find and returns first line ending in CRLF along with following buffer.
 
     If no ending CRLF is found, line is None."""
-    pos = raw.find(CRLF)
-    if pos == -1:
+    parts = raw.split(CRLF)
+    if len(parts) == 1:
         return None, raw
-    line = raw[:pos]
-    rest = raw[pos + len(CRLF):]
-    return line, rest
+    return parts[0], CRLF.join(parts[1:])
 
 
 def wrap_socket(
diff --git a/proxy/core/acceptor/acceptor.py b/proxy/core/acceptor/acceptor.py
index b49f854..64b4d06 100644
--- a/proxy/core/acceptor/acceptor.py
+++ b/proxy/core/acceptor/acceptor.py
@@ -78,6 +78,7 @@ class Acceptor(multiprocessing.Process):
             lock: multiprocessing.synchronize.Lock,
             executor_queues: List[connection.Connection],
             executor_pids: List[int],
+            executor_locks: List[multiprocessing.synchronize.Lock],
             event_queue: Optional[EventQueue] = None,
     ) -> None:
         super().__init__()
@@ -94,6 +95,7 @@ class Acceptor(multiprocessing.Process):
         # Available executors
         self.executor_queues = executor_queues
         self.executor_pids = executor_pids
+        self.executor_locks = executor_locks
         # Selector
         self.running = multiprocessing.Event()
         self.selector: Optional[selectors.DefaultSelector] = None
@@ -152,6 +154,7 @@ class Acceptor(multiprocessing.Process):
             ThreadlessPool.delegate(
                 self.executor_pids[index],
                 self.executor_queues[index],
+                self.executor_locks[index],
                 conn,
                 addr,
                 self.flags.unix_socket_path,
diff --git a/proxy/core/acceptor/executors.py b/proxy/core/acceptor/executors.py
index 3bafe48..3d40841 100644
--- a/proxy/core/acceptor/executors.py
+++ b/proxy/core/acceptor/executors.py
@@ -94,6 +94,7 @@ class ThreadlessPool:
         # Threadless worker communication states
         self.work_queues: List[connection.Connection] = []
         self.work_pids: List[int] = []
+        self.work_locks: List[multiprocessing.synchronize.Lock] = []
         # List of threadless workers
         self._workers: List[Threadless] = []
 
@@ -108,22 +109,24 @@ class ThreadlessPool:
     def delegate(
             worker_pid: int,
             work_queue: connection.Connection,
+            work_lock: multiprocessing.synchronize.Lock,
             conn: socket.socket,
             addr: Optional[Tuple[str, int]],
             unix_socket_path: Optional[str] = None,
     ) -> None:
         """Utility method to delegate a work to threadless executor pool."""
-        # Accepted client address is empty string for
-        # unix socket domain, avoid sending empty string
-        # for optimization.
-        if not unix_socket_path:
-            work_queue.send(addr)
-        send_handle(
-            work_queue,
-            conn.fileno(),
-            worker_pid,
-        )
-        conn.close()
+        with work_lock:
+            # Accepted client address is empty string for
+            # unix socket domain, avoid sending empty string
+            # for optimization.
+            if not unix_socket_path:
+                work_queue.send(addr)
+            send_handle(
+                work_queue,
+                conn.fileno(),
+                worker_pid,
+            )
+            conn.close()
 
     @staticmethod
     def start_threaded_work(
@@ -168,8 +171,7 @@ class ThreadlessPool:
     def shutdown(self) -> None:
         """Shutdown threadless processes."""
         if is_threadless(self.flags.threadless, self.flags.threaded):
-            for _ in range(self.flags.num_workers):
-                self._shutdown_worker()
+            self._shutdown_workers()
             logger.info(
                 'Stopped {0} threadless workers'.format(
                     self.flags.num_workers,
@@ -178,6 +180,7 @@ class ThreadlessPool:
 
     def _start_worker(self, index: int) -> None:
         """Starts a threadless worker."""
+        self.work_locks.append(multiprocessing.Lock())
         pipe = multiprocessing.Pipe()
         self.work_queues.append(pipe[0])
         w = Threadless(
@@ -191,12 +194,14 @@ class ThreadlessPool:
         self.work_pids.append(w.pid)
         logger.debug('Started threadless#%d process#%d', index, w.pid)
 
-    def _shutdown_worker(self) -> None:
+    def _shutdown_workers(self) -> None:
         """Pop a running threadless worker and clean it up."""
-        w = self._workers.pop()
-        pid = w.pid
-        w.running.set()
-        w.join()
-        self.work_pids.pop()
-        self.work_queues.pop().close()
-        logger.debug('Stopped threadless process#%d', pid)
+        for index in range(self.flags.num_workers):
+            self._workers[index].running.set()
+        for index in range(self.flags.num_workers):
+            pid = self._workers[index].pid
+            self._workers[index].join()
+            self.work_pids.pop()
+            self.work_queues.pop().close()
+            logger.debug('Stopped threadless process#%d', pid)
+        self.work_locks = []
diff --git a/proxy/core/acceptor/pool.py b/proxy/core/acceptor/pool.py
index 66d0421..9114bcf 100644
--- a/proxy/core/acceptor/pool.py
+++ b/proxy/core/acceptor/pool.py
@@ -72,6 +72,7 @@ class AcceptorPool:
             listener: Listener,
             executor_queues: List[connection.Connection],
             executor_pids: List[int],
+            executor_locks: List[multiprocessing.synchronize.Lock],
             event_queue: Optional[EventQueue] = None,
     ) -> None:
         self.flags = flags
@@ -80,6 +81,7 @@ class AcceptorPool:
         # Available executors
         self.executor_queues: List[connection.Connection] = executor_queues
         self.executor_pids: List[int] = executor_pids
+        self.executor_locks: List[multiprocessing.synchronize.Lock] = executor_locks
         # Eventing core queue
         self.event_queue: Optional[EventQueue] = event_queue
         # Acceptor process instances
@@ -128,6 +130,7 @@ class AcceptorPool:
                 event_queue=self.event_queue,
                 executor_queues=self.executor_queues,
                 executor_pids=self.executor_pids,
+                executor_locks=self.executor_locks,
             )
             acceptor.start()
             logger.debug(
diff --git a/proxy/plugin/reverse_proxy.py b/proxy/plugin/reverse_proxy.py
index 8eb3b91..0715d18 100644
--- a/proxy/plugin/reverse_proxy.py
+++ b/proxy/plugin/reverse_proxy.py
@@ -52,7 +52,6 @@ class ReverseProxyPlugin(TcpUpstreamConnectionHandler, HttpWebServerBasePlugin):
           "url": "http://localhost/get"
         }
         ```
-
     """
 
     # TODO: We must use nginx python parser and
@@ -69,7 +68,6 @@ class ReverseProxyPlugin(TcpUpstreamConnectionHandler, HttpWebServerBasePlugin):
 
     def __init__(self, *args: Any, **kwargs: Any):
         super().__init__(*args, **kwargs)
-        # Chosen upstream proxy_pass url
         self.choice: Optional[Url] = None
 
     def handle_upstream_data(self, raw: memoryview) -> None:
diff --git a/proxy/proxy.py b/proxy/proxy.py
index 899b723..c05c6cf 100644
--- a/proxy/proxy.py
+++ b/proxy/proxy.py
@@ -179,6 +179,7 @@ class Proxy:
             listener=self.listener,
             executor_queues=self.executors.work_queues,
             executor_pids=self.executors.work_pids,
+            executor_locks=self.executors.work_locks,
             event_queue=event_queue,
         )
         self.acceptors.setup()


Test patch:
diff --git a/tests/core/test_acceptor.py b/tests/core/test_acceptor.py
index 25c092a..48ee7a4 100644
--- a/tests/core/test_acceptor.py
+++ b/tests/core/test_acceptor.py
@@ -34,6 +34,7 @@ class TestAcceptor(unittest.TestCase):
             lock=multiprocessing.Lock(),
             executor_queues=[],
             executor_pids=[],
+            executor_locks=[],
         )
 
     @mock.patch('selectors.DefaultSelector')
diff --git a/tests/core/test_acceptor_pool.py b/tests/core/test_acceptor_pool.py
index e2a17c1..5241bd4 100644
--- a/tests/core/test_acceptor_pool.py
+++ b/tests/core/test_acceptor_pool.py
@@ -42,7 +42,7 @@ class TestAcceptorPool(unittest.TestCase):
 
         pool = AcceptorPool(
             flags=flags, listener=mock_listener.return_value,
-            executor_queues=[], executor_pids=[],
+            executor_queues=[], executor_pids=[], executor_locks=[],
         )
         pool.setup()
 
diff --git a/tests/test_main.py b/tests/test_main.py
index d2bc641..1f83153 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -111,6 +111,7 @@ class TestMain(unittest.TestCase):
             listener=mock_listener.return_value,
             executor_queues=mock_executor_pool.return_value.work_queues,
             executor_pids=mock_executor_pool.return_value.work_pids,
+            executor_locks=mock_executor_pool.return_value.work_locks,
             event_queue=None,
         )
         mock_acceptor_pool.return_value.setup.assert_called_once()
@@ -156,6 +157,7 @@ class TestMain(unittest.TestCase):
             listener=mock_listener.return_value,
             executor_queues=mock_executor_pool.return_value.work_queues,
             executor_pids=mock_executor_pool.return_value.work_pids,
+            executor_locks=mock_executor_pool.return_value.work_locks,
             event_queue=None,
         )
         mock_acceptor_pool.return_value.setup.assert_called_once()
@@ -197,6 +199,7 @@ class TestMain(unittest.TestCase):
             event_queue=mock_event_manager.return_value.queue,
             executor_queues=mock_executor_pool.return_value.work_queues,
             executor_pids=mock_executor_pool.return_value.work_pids,
+            executor_locks=mock_executor_pool.return_value.work_locks,
         )
         mock_acceptor_pool.return_value.setup.assert_called_once()
         mock_acceptor_pool.return_value.shutdown.assert_called_once()