Repository: jina-ai/serve
Pull request: fix: test random_port
resolves #741 

Problem statement:
add retry mechanism for port conflicts
**Describe the bug** 
The zmq ports conflict in some cases.

```
2020-07-29T16:56:40.0807186Z   File "/home/runner/work/jina/jina/jina/peapods/zmq.py", line 142, in init_sockets
2020-07-29T16:56:40.0807474Z     in_sock, in_addr = _init_socket(ctx, self.args.host_in, self.args.port_in, self.args.socket_in,
2020-07-29T16:56:40.0807935Z   File "/home/runner/work/jina/jina/jina/peapods/zmq.py", line 657, in _init_socket
2020-07-29T16:56:40.0808130Z     raise ex
2020-07-29T16:56:40.0808347Z   File "/home/runner/work/jina/jina/jina/peapods/zmq.py", line 654, in _init_socket
2020-07-29T16:56:40.0808832Z     sock.bind('tcp://%s:%d' % (host, port))
2020-07-29T16:56:40.0809098Z   File "zmq/backend/cython/socket.pyx", line 550, in zmq.backend.cython.socket.Socket.bind
2020-07-29T16:56:40.0809479Z   File "zmq/backend/cython/checkrc.pxd", line 26, in zmq.backend.cython.checkrc._check_rc
2020-07-29T16:56:40.0809669Z zmq.error.ZMQError: Address already in use
```

**Describe how you solve it**
<!-- copy past your code/pull request link -->

---

<!-- Optional, but really help us locate the problem faster -->

**Environment**
<!-- Run `jina --version-full` and copy paste the output here -->

**Screenshots**
<!-- If applicable, add screenshots to help explain your problem. -->

Hints from issue discussion:
warning, this is not as easy as "change to another port and retry". The Flow has no clue about the port-change at Pea/Pod level and has no way but to re-orchestrate the whole stack.

non-trivial 


> warning, this is not as easy as "change to another port and retry". The Flow has no clue about the port-change at Pea/Pod level and has no way but to re-orchestrate the whole stack.
> 
> non-trivial

Exactly, we have to rethink about the flow building process.
refactoring Flow just for solving port conflicting problem is over-engineering. i don't see any way to get it done in 2 hours.

need to think other ways
To fix this problem, we need to check the port conflicts. There are two options:

1. check when defining before running the flow
    *solution*: We scan all the available ports and check if the assigned ports is taken. If yes, a new port will be assigned **before** building the connections.
    - pros
        - No hassle to rebuild the flow
    - cons
        - the port might be **still** taken before running the flow
2. check when running the flow
    *solution*: We check whether the assigned port is available. If not, a new port will be assigned and **the flow will be re-built**.
    - pros
        - no worry about the port being taken before running the flow. New port will be assigned in the case of conflicts.
    - cons
        - the information of the new ports needs to be fed back from the pea level to the flow level
        - involves flow re-build for every conflict
It is not the right solution technically, but what if we try to add retry mechanism to the tests directly? It may do the jo After all is more a problem of too much concurrent tests right? It is not likely to happen in a prod env.
please notice that random_port already implement your 1


> To fix this problem, we need to check the port conflicts. There are two options:
> 
> 
> 
> 1. check when defining before running the flow
> 
>     *solution*: We scan all the available ports and check if the assigned ports is taken. If yes, a new port will be assigned **before** building the connections.
> 
>     - pros
> 
>         - No hassle to rebuild the flow
> 
>     - cons
> 
>         - the port might be **still** taken before running the flow
> 
> 2. check when running the flow
> 
>     *solution*: We check whether the assigned port is available. If not, a new port will be assigned and **the flow will be re-built**.
> 
>     - pros
> 
>         - no worry about the port being taken before running the flow. New port will be assigned in the case of conflicts.
> 
>     - cons
> 
>         - the information of the new ports needs to be fed back from the pea level to the flow level
> 
>         - involves flow re-build for every conflict

I agree with this. The ideal case is to have *each* test case inside a docker container, to avoid collided ports.

Note, we should not use parallelism when doing unit test. The original motivation of moving away from `pytest` is we want to use the process as a context manager so that when the process ends, it releases all resources. `pytest` runs all test cases in *one* process, and this is what we try to avoid.

> It is not the right solution technically, but what if we try to add retry mechanism to the tests directly? It may do the jo After all is more a problem of too much concurrent tests right? It is not likely to happen in a prod env.
We have `-n 1` option right? I am not familiar with the insights of `pytest` and `unittest`. But are we sure that `github` spins totally isolated environments for every run? Or could it be that the tests for python 3.7 and 3.8 collide?
I'm a pytest newbie, but quick Stackoverflowing shows that it can run each test in a separate process (need to install plugin), see https://stackoverflow.com/questions/48234032/run-py-test-test-in-different-process
> I'm a pytest newbie, but quick Stackoverflowing shows that it can run each test in a separate process (need to install plugin), see https://stackoverflow.com/questions/48234032/run-py-test-test-in-different-process

We are using that one actually.

Commit message: fix: clean up

Patch to review:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index a5c44ce37fc23..f03dba8e06464 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -69,7 +69,7 @@ jobs:
           jina check
           export JINA_LOG_VERBOSITY="ERROR"
           pip install pytest pytest-xdist pytest-timeout pytest-cov
-          pytest --cov=jina --cov-report=xml -n 1 --timeout=60 -v tests/unit
+          pytest --cov=jina --cov-report=xml -n 1 --timeout=120 -v tests/unit
           mv coverage.xml unit-test-coverage.xml
         timeout-minutes: 15
       - name: Upload coverage result from unit-test
diff --git a/jina/flow/__init__.py b/jina/flow/__init__.py
index 090676a3893f2..e453ad08e1247 100644
--- a/jina/flow/__init__.py
+++ b/jina/flow/__init__.py
@@ -17,7 +17,7 @@
 from .. import JINA_GLOBAL
 from ..enums import FlowBuildLevel, FlowOptimizeLevel
 from ..excepts import FlowTopologyError, FlowMissingPodError, FlowBuildLevelError
-from ..helper import yaml, expand_env_var, get_non_defaults_args, deprecated_alias
+from ..helper import yaml, expand_env_var, get_non_defaults_args, deprecated_alias, random_port
 from ..logging import get_logger
 from ..logging.sse import start_sse_logger
 from ..peapods.pod import SocketType, FlowPod, GatewayFlowPod
@@ -171,6 +171,7 @@ def __init__(self, args: 'argparse.Namespace' = None, **kwargs):
         self._last_changed_pod = ['gateway']  #: default first pod is gateway, will add when build()
 
         self._update_args(args, **kwargs)
+        self._ports_in_use = []
 
     def _update_args(self, args, **kwargs):
         from ..main.parser import set_flow_parser
@@ -398,8 +399,8 @@ def add(self,
 
         kwargs.update(op_flow._common_kwargs)
         kwargs['name'] = pod_name
+        op_flow._ports_in_use += self._check_port_collision(kwargs)
         op_flow._pod_nodes[pod_name] = FlowPod(kwargs=kwargs, needs=needs)
-
         op_flow.set_last_pod(pod_name, False)
 
         return op_flow
@@ -834,3 +835,17 @@ def use_grpc_gateway(self):
     def use_rest_gateway(self):
         """Change to use REST gateway for IO """
         self._common_kwargs['rest_api'] = True
+
+    def _check_port_collision(self, kwargs):
+        """Check if the Pods' ports collide"""
+        for _port_name in ('port_in', 'port_out', 'port_ctrl', 'port_expose'):
+            _port = kwargs.get(_port_name)
+            while _port in self._ports_in_use or _port is None:
+                _new = random_port()
+                if _port is not None:
+                    self.logger.info(f'{_port_name} collision detected. set from {_port} to {_new}')
+                _port = _new
+            kwargs[_port_name] = _port
+            self.logger.info(f'{kwargs.get("name")} {_port_name}: {_port}')
+            self._ports_in_use.append(_port)
+        return self._ports_in_use
diff --git a/jina/helper.py b/jina/helper.py
index 6a9ba22d9f5fc..dcd63c988d729 100644
--- a/jina/helper.py
+++ b/jina/helper.py
@@ -260,20 +260,33 @@ def random_name() -> str:
 
 
 def random_port() -> int:
+    from contextlib import closing
+    import socket
     if 'JINA_RANDOM_PORTS' not in os.environ:
         # feel like this gives higher chance of collision in unit test
-        from contextlib import closing
-        import socket
         import threading
         with threading.Lock():
             with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
                 s.bind(('', 0))
                 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-                return s.getsockname()[1]
+                _port = s.getsockname()[1]
+                return _port
     else:
         import random
+
+        def is_port_in_use(p):
+            if p is None:
+                return True
+            with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+                result = s.connect_ex(('', p))
+                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                return result == 0
+
         min_port, max_port = 49152, 65535
-        return random.randrange(min_port, max_port)
+        _port = None
+        while is_port_in_use(_port):
+            _port = random.randrange(min_port, max_port)
+        return _port
 
 
 def get_registered_ports(stack_id: int = JINA_GLOBAL.stack.id):
diff --git a/jina/peapods/zmq.py b/jina/peapods/zmq.py
index c75e75e66364a..eb499f549be80 100644
--- a/jina/peapods/zmq.py
+++ b/jina/peapods/zmq.py
@@ -316,15 +316,18 @@ def close(self):
             for s in self.opened_socks:
                 s.flush()
             super().close()
-            self.io_loop.stop()
-            # Replace handle events function, to skip
-            # None event after sockets are closed.
-            if hasattr(self.in_sock, '_handle_events'):
-                self.in_sock._handle_events = lambda *args, **kwargs: None
-            if hasattr(self.out_sock, '_handle_events'):
-                self.out_sock._handle_events = lambda *args, **kwargs: None
-            if hasattr(self.ctrl_sock, '_handle_events'):
-                self.ctrl_sock._handle_events = lambda *args, **kwargs: None
+            try:
+                self.io_loop.stop()
+                # Replace handle events function, to skip
+                # None event after sockets are closed.
+                if hasattr(self.in_sock, '_handle_events'):
+                    self.in_sock._handle_events = lambda *args, **kwargs: None
+                if hasattr(self.out_sock, '_handle_events'):
+                    self.out_sock._handle_events = lambda *args, **kwargs: None
+                if hasattr(self.ctrl_sock, '_handle_events'):
+                    self.ctrl_sock._handle_events = lambda *args, **kwargs: None
+            except AttributeError as e:
+                self.logger.info(f'failed to stop. {e}')
 
     def pause_pollin(self):
         """Remove :attr:`in_sock` from the poller """
diff --git a/tests/unit/flow/test_flow.py b/tests/unit/flow/test_flow.py
index 39bea5e205e20..60b2b38126d42 100644
--- a/tests/unit/flow/test_flow.py
+++ b/tests/unit/flow/test_flow.py
@@ -278,7 +278,6 @@ def test_shards(self):
         with f:
             pass
         self.add_tmpfile('test-docshard-tmp')
-        time.sleep(2)
 
     def test_py_client(self):
         f = (Flow().add(name='r1', uses='_pass')
@@ -393,9 +392,7 @@ def test_dry_run_with_two_pathways_diverging_at_non_gateway(self):
                 self.assertEqual(node.peas_args['peas'][0], node.tail_args)
             f.dry_run()
 
-    @pytest.mark.skip('this leads to zmq address conflicts on github')
     def test_refactor_num_part(self):
-        sleep(3)
         f = (Flow().add(name='r1', uses='_logforward', needs='gateway')
              .add(name='r2', uses='_logforward', needs='gateway')
              .join(['r1', 'r2']))
@@ -403,11 +400,11 @@ def test_refactor_num_part(self):
         with f:
             node = f._pod_nodes['gateway']
             self.assertEqual(node.head_args.socket_in, SocketType.PULL_CONNECT)
-            self.assertEqual(node.tail_args.socket_out, SocketType.PUSH_CONNECT)
+            self.assertEqual(node.tail_args.socket_out, SocketType.PUB_BIND)
 
             node = f._pod_nodes['r1']
-            self.assertEqual(node.head_args.socket_in, SocketType.PULL_BIND)
-            self.assertEqual(node.tail_args.socket_out, SocketType.PUB_BIND)
+            self.assertEqual(node.head_args.socket_in, SocketType.SUB_CONNECT)
+            self.assertEqual(node.tail_args.socket_out, SocketType.PUSH_CONNECT)
 
             node = f._pod_nodes['r2']
             self.assertEqual(node.head_args.socket_in, SocketType.SUB_CONNECT)
@@ -517,6 +514,17 @@ def input_fn():
         with flow:
             flow.index(input_fn=input_fn, output_fn=validate)
 
+    def test_flow_with_collision(self):
+        collision_port = 55555
+
+        flow = (Flow()
+                .add(name='p1', uses='_pass', port_out=collision_port)
+                .add(name='p2', uses='_pass', port_out=collision_port))
+
+        self.assertNotEqual(
+            flow._pod_nodes['p1']._args.port_out,
+            flow._pod_nodes['p2']._args.port_out)
+
 
 if __name__ == '__main__':
     unittest.main()
diff --git a/tests/unit/flow/test_flow_merge.py b/tests/unit/flow/test_flow_merge.py
index 3e266c6f5aa85..ba90b7174c279 100644
--- a/tests/unit/flow/test_flow_merge.py
+++ b/tests/unit/flow/test_flow_merge.py
@@ -43,6 +43,7 @@ def test_this_will_fail(self):
         with f:
             f.index(input_fn=random_docs(10), output_fn=self.validate)
 
+    @pytest.mark.timeout(180)
     def test_this_should_work(self):
         f = (Flow()
              .add(name='a1', uses='_pass')
diff --git a/tests/unit/test_container.py b/tests/unit/test_container.py
index c65847408e34d..f6950ba1d7b8b 100644
--- a/tests/unit/test_container.py
+++ b/tests/unit/test_container.py
@@ -120,7 +120,7 @@ def test_flow_topo_parallel(self):
             f.dry_run()
             f.index(input_fn=random_docs(1000))
 
-    @pytest.mark.skip('this leads to zmq address conflicts on github')
+    @pytest.mark.skip('extract_docs throw an error')
     def test_container_volume(self):
         time.sleep(5)
         f = (Flow()
@@ -131,7 +131,7 @@ def test_container_volume(self):
             f.index(input_fn=random_docs(10))
 
         out_file = 'abc/ext-mwu-encoder.bin'
-        self.assertTrue(os.path.exists(out_file))
+        self.assertTrue(os.path.exists(os.path.join(cur_dir, out_file)))
         self.add_tmpfile(out_file, './abc')
 
     def test_container_ping(self):
diff --git a/tests/unit/test_quant.py b/tests/unit/test_quant.py
index 1f3a0036e8d7a..663b0daab9a31 100644
--- a/tests/unit/test_quant.py
+++ b/tests/unit/test_quant.py
@@ -52,8 +52,6 @@ def f1(self, quant):
         with f as fl:
             fl.index(random_docs, output_fn=get_output)
 
-        time.sleep(4)
-
     def f2(self, quant):
         os.environ['JINA_ARRAY_QUANT'] = quant
 
@@ -61,8 +59,6 @@ def f2(self, quant):
         with f as fl:
             fl.index(random_docs, output_fn=get_output)
 
-        time.sleep(4)
-
     def test_quant(self):
         for j in ('fp32', 'fp16', 'uint8'):
             self.f1(j)


Code review comments:
diff_hunk: @@ -260,20 +260,33 @@ def random_name() -> str:
 
 
 def random_port() -> int:
+    from contextlib import closing
+    import socket
     if 'JINA_RANDOM_PORTS' not in os.environ:
         # feel like this gives higher chance of collision in unit test
-        from contextlib import closing
-        import socket
         import threading
         with threading.Lock():
             with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
                 s.bind(('', 0))
                 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-                return s.getsockname()[1]
+                _port = s.getsockname()[1]
+                return _port
     else:
         import random
+
+        def is_port_in_use(p):
+            if p is None:
+                return True
+            with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+                result = s.connect_ex(('', p))
+                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+                return result == 0
+
         min_port, max_port = 49152, 65535
-        return random.randrange(min_port, max_port)
+        _port = None
+        while is_port_in_use(_port):
+            _port = random.randrange(min_port, max_port)
path: jina/helper.py
text: @user1:
the coding style needs to be improved.

can't you set `_port = random.randrange(min_port, max_port)` first and then do the `while`? this remove `if p is None return True`

diff_hunk: @@ -316,15 +316,18 @@ def close(self):
             for s in self.opened_socks:
                 s.flush()
             super().close()
-            self.io_loop.stop()
-            # Replace handle events function, to skip
-            # None event after sockets are closed.
-            if hasattr(self.in_sock, '_handle_events'):
-                self.in_sock._handle_events = lambda *args, **kwargs: None
-            if hasattr(self.out_sock, '_handle_events'):
-                self.out_sock._handle_events = lambda *args, **kwargs: None
-            if hasattr(self.ctrl_sock, '_handle_events'):
-                self.ctrl_sock._handle_events = lambda *args, **kwargs: None
+            try:
+                self.io_loop.stop()
+                # Replace handle events function, to skip
+                # None event after sockets are closed.
+                if hasattr(self.in_sock, '_handle_events'):
+                    self.in_sock._handle_events = lambda *args, **kwargs: None
+                if hasattr(self.out_sock, '_handle_events'):
+                    self.out_sock._handle_events = lambda *args, **kwargs: None
+                if hasattr(self.ctrl_sock, '_handle_events'):
+                    self.ctrl_sock._handle_events = lambda *args, **kwargs: None
+            except AttributeError as e:
+                self.logger.info(f'failed to stop. {e}')
path: jina/peapods/zmq.py
text: @user1:
what could possibly happen here?

@author:

> what could possibly happen here?

When the address conflict happens, these attributes will not be set and this leads to an exception  in closing the process and messed up the error messages.

@user1:
then it should be error not info

diff_hunk: @@ -260,20 +260,33 @@ def random_name() -> str:
 
 
 def random_port() -> int:
+    from contextlib import closing
+    import socket
     if 'JINA_RANDOM_PORTS' not in os.environ:
         # feel like this gives higher chance of collision in unit test
-        from contextlib import closing
-        import socket
         import threading
         with threading.Lock():
             with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
                 s.bind(('', 0))
                 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-                return s.getsockname()[1]
+                _port = s.getsockname()[1]
+                return _port
path: jina/helper.py
text: @user1:

```suggestion
                return s.getsockname()[1]
```