Fix by comments in PR. Make refresh before request

Author: denis-ignatenko

.travis.yml

@@ -10,28 +10,15 @@ env:
       - TARGET=test
       - OS=el DIST=6
       - OS=el DIST=7
-      - OS=fedora DIST=24
-      - OS=fedora DIST=25
-      - OS=ubuntu DIST=precise
+      - OS=fedora DIST=28
+      - OS=fedora DIST=29
       - OS=ubuntu DIST=trusty
       - OS=ubuntu DIST=xenial
-#      - OS=debian DIST=wheezy
+      - OS=ubuntu DIST=bionic
+      - OS=ubuntu DIST=disco
       - OS=debian DIST=jessie
       - OS=debian DIST=stretch
-
-matrix:
-    allow_failures:
-#      - env: OS=el DIST=6
-#      - env: OS=el DIST=7
-#      - env: OS=fedora DIST=24
-#      - env: OS=fedora DIST=25
-#      - env: OS=ubuntu DIST=precise
-#      - env: OS=ubuntu DIST=trusty
-#      - env: OS=ubuntu DIST=xenial
-#      - env: OS=ubuntu DIST=yakkety
-#      - env: OS=debian DIST=wheezy
-#      - env: OS=debian DIST=jessie
-#      - env: OS=debian DIST=stretch
+      - OS=debian DIST=buster
 
 script:
   - git describe --long
@@ -100,6 +87,16 @@ deploy:
     on:
       branch: master
       condition: -n "${OS}" && -n "${DIST}" && -n "${PACKAGECLOUD_TOKEN}"
+  - provider: packagecloud
+    username: tarantool
+    repository: "2_2"
+    token: ${PACKAGECLOUD_TOKEN}
+    dist: ${OS}/${DIST}
+    package_glob: build/*.{rpm,deb}
+    skip_cleanup: true
+    on:
+      branch: master
+      condition: -n "${OS}" && -n "${DIST}" && -n "${PACKAGECLOUD_TOKEN}"
 
 notifications:
   email:

tarantool/connection.py

@@ -126,14 +126,14 @@ def __init__(self, host, port,
         self.encoding = encoding
         self.call_16 = call_16
         self.connection_timeout = connection_timeout
-        self.authenticated = False
         if connect_now:
             self.connect()
 
     def close(self):
         '''
         Close connection to the server
         '''
+        self.connected = False
         self._socket.close()
         self._socket = None
 
@@ -201,11 +201,9 @@ def connect(self):
         :raise: `NetworkError`
         '''
         try:
-            self.authenticated = False
             self.connect_basic()
             self.handshake()
             self.load_schema()
-            self.authenticated = True
         except Exception as e:
             self.connected = False
             raise NetworkError(e)
@@ -362,47 +360,34 @@ def flush_schema(self):
         self.schema.flush()
         self.load_schema()
 
-    def call(self, func_name, *args):
+    def call(self, func_name, *args, reconnect=True):
         '''
         Execute CALL request. Call stored Lua function.
 
         :param func_name: stored Lua function name
         :type func_name: str
-        :param args: list of function arguments
-        :type args: list or tuple
+        :param args: function arguments
+        :type args: tuple
+        :param reconnect: reconnect before call
+        :type reconnect: boolean
 
         :rtype: `Response` instance
         '''
 
+        assert isinstance(func_name, str)
+
         # This allows to use a tuple or list as an argument
         if len(args) == 1 and isinstance(args[0], (list, tuple)):
             args = args[0]
 
-        return self.call_ex(func_name, args, True)
-
-    def call_ex(self, func_name, args=[], reconnect=True):
-        '''
-        Execute CALL request. Call stored Lua function.
-
-        :param func_name: stored Lua function name
-        :type func_name: str
-        :param args: list of function arguments
-        :type args: list or tuple
-        :param reconnect: reconnect before call
-        :type reconnect: boolean
-
-        :rtype: `Response` instance
-        '''
-        assert isinstance(func_name, str)
-
         request = RequestCall(self, func_name, args, self.call_16)
         if reconnect:
             response = self._send_request(request)
         else:
             response = self._send_request_wo_reconnect(request)
         return response
 
-    def eval(self, expr, *args):
+    def eval(self, expr, *args, reconnect=True):
         '''
         Execute EVAL request. Eval Lua expression.
 
@@ -413,28 +398,12 @@ def eval(self, expr, *args):
 
         :rtype: `Response` instance
         '''
+        assert isinstance(expr, str)
 
         # This allows to use a tuple or list as an argument
         if len(args) == 1 and isinstance(args[0], (list, tuple)):
             args = args[0]
-
-        return self.eval_ex(expr, args, True)
-
-    def eval_ex(self, expr, args=[], reconnect=True):
-        '''
-        Execute EVAL request. Eval Lua expression.
-
-        :param expr: Lua expression
-        :type expr: str
-        :param args: list of function arguments
-        :type args: list or tuple
-        :param reconnect: reconnect before call
-        :type reconnect: boolean
-
-        :rtype: `Response` instance
-        '''
-        assert isinstance(expr, str)
-
+        
         request = RequestEval(self, expr, args)
         if reconnect:
             response = self._send_request(request)

tarantool/const.py

@@ -87,4 +87,4 @@
 # Default delay between attempts to reconnect (seconds)
 RECONNECT_DELAY = 0.1
 # Default cluster nodes list refresh interval (seconds)
-NODES_REFRESH_INTERVAL = 300
+DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS = 60000

tarantool/mesh_connection.py

@@ -9,12 +9,16 @@
 from tarantool.error import NetworkError
 from tarantool.utils import ENCODING_DEFAULT
 from tarantool.const import (
+    CONNECTION_TIMEOUT,
     SOCKET_TIMEOUT,
     RECONNECT_MAX_ATTEMPTS,
     RECONNECT_DELAY,
-    NODES_REFRESH_INTERVAL
+    DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS,
 )
 
+from tarantool.utils import (
+    ENCODING_DEFAULT
+)
 
 class RoundRobinStrategy(object):
     def __init__(self, addrs):
@@ -88,26 +92,36 @@ class MeshConnection(Connection):
         end
     '''
 
-    def __init__(self, addrs,
+    def __init__(self, host, port,
                  user=None,
                  password=None,
                  socket_timeout=SOCKET_TIMEOUT,
                  reconnect_max_attempts=RECONNECT_MAX_ATTEMPTS,
                  reconnect_delay=RECONNECT_DELAY,
                  connect_now=True,
                  encoding=ENCODING_DEFAULT,
+                 call_16=False,
+                 connection_timeout=CONNECTION_TIMEOUT,
+                 cluster_list = None,
                  strategy_class=RoundRobinStrategy,
                  get_nodes_function_name=None,
-                 nodes_refresh_interval=NODES_REFRESH_INTERVAL):
-        self.nattempts = 2 * len(addrs) + 1
+                 nodes_refresh_interval=DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS):
+
+        addrs = [{"host":host, "port":port}]
+        if cluster_list:
+            for i in cluster_list:
+                if i["host"] == host or i["port"] == port:
+                    continue
+                addrs.append(i)
+        
         self.strategy = strategy_class(addrs)
         self.strategy_class = strategy_class
         addr = self.strategy.getnext()
         host = addr['host']
         port = addr['port']
         self.get_nodes_function_name = get_nodes_function_name
-        self.nodes_refresh_interval = nodes_refresh_interval >= 30 and nodes_refresh_interval or 30
-        self.last_nodes_refresh = 0
+        self.nodes_refresh_interval = nodes_refresh_interval
+        self.last_nodes_refresh = time.time()
         super(MeshConnection, self).__init__(host=host,
                                              port=port,
                                              user=user,
@@ -116,54 +130,69 @@ def __init__(self, addrs,
                                              reconnect_max_attempts=reconnect_max_attempts,
                                              reconnect_delay=reconnect_delay,
                                              connect_now=connect_now,
-                                             encoding=encoding)
-
-    def _opt_reconnect(self):
-        nattempts = self.nattempts
-        while nattempts > 0:
-            try:
-                super(MeshConnection, self)._opt_reconnect()
-                break
-            except NetworkError:
-                nattempts -= 1
-                addr = self.strategy.getnext()
-                self.host = addr['host']
-                self.port = addr['port']
-        else:
-            raise NetworkError
+                                             encoding=encoding,
+                                             call_16=call_16,
+                                             connection_timeout=connection_timeout)
 
-        if self.authenticated and self.get_nodes_function_name:
-            now = time.time()
-            if now - self.last_nodes_refresh > self.nodes_refresh_interval:
-                self.refresh_nodes(now)
+    def _opt_refresh_instances(self):
+        """
+        Refresh list of cluster instances. If current connection not in server list will change connection.
+        """
+        now = time.time()
+
+        if self.connected and now - self.last_nodes_refresh > self.nodes_refresh_interval/1000:
+            resp = self.call(self.get_nodes_function_name, reconnect=False)
+
+            # got data to refresh        
+            if resp.data and resp.data[0]:
+                addrs = list(parse_uri(i) for i in resp.data[0])
+                self.strategy = self.strategy_class(addrs)
+                self.last_nodes_refresh = now
 
-    def refresh_nodes(self, cur_time):
-        '''
-        Refreshes nodes list by calling Lua function with name
-        self.get_nodes_function_name on the current node. If this field is None
-        no refresh occurs. Usually you don't need to call this function manually
-        since it's called automatically during reconnect every
-        self.nodes_refresh_interval seconds.
-        '''
-        resp = super(MeshConnection, self).call_ex(self.get_nodes_function_name,
-                                                   [], reconnect=False)
-
-        if not (resp.data and resp.data[0]):
-            return
-
-        addrs_raw = resp.data[0]
-        if type(addrs_raw) is list:
-            addrs = []
-            for uri_str in addrs_raw:
-                addr = parse_uri(uri_str)
-                if addr:
-                    addrs.append(addr)
-
-            self.strategy = self.strategy_class(addrs)
-            self.last_nodes_refresh = cur_time
             if not {'host': self.host, 'port': self.port} in addrs:
                 addr = self.strategy.getnext()
                 self.host = addr['host']
                 self.port = addr['port']
                 self.close()
-                self._opt_reconnect()
+        
+        if not self.connected:
+            
+            nattempts = (len(self.strategy.addrs) * 2) + 1
+
+            while nattempts >= 0:
+                try:
+                    addr = self.strategy.getnext()
+                    if addr['host'] != self.host or addr['port'] != self.port:
+                        self.host = addr['host']
+                        self.port = addr['port']
+                        self._opt_reconnect()
+                        break
+                    else:
+                        nattempts -= 1
+                except NetworkError:
+                    continue
+            else:
+                raise NetworkError
+    
+    def _send_request(self, request):
+        '''
+        Send the request to the server through the socket.
+        Return an instance of `Response` class.
+
+        Update instances list from server `get_nodes_function_name` function. 
+
+        :param request: object representing a request
+        :type request: `Request` instance
+
+        :rtype: `Response` instance
+        '''
+        if self.get_nodes_function_name:
+            self._opt_refresh_instances()
+        
+        try:
+            return super(MeshConnection, self)._send_request(request)
+        except NetworkError:
+            self.connected = False
+            self._opt_refresh_instances()
+        finally:
+            return super(MeshConnection, self)._send_request(request)

test/cluster-py/multi.test.py

@@ -42,7 +42,7 @@ def check_connection(con):
 # Make a list of servers
 sources = []
 for server in cluster[1:]:
-    sources.append(yaml.load(server.admin('box.cfg.listen', silent=True))[0])
+    sources.append(yaml.safe_load(server.admin('box.cfg.listen', silent=True))[0])
 
 addrs = []
 for addr in sources:

unit/suites/__init__.py

@@ -8,9 +8,10 @@
 from .test_dml import TestSuite_Request
 from .test_protocol import TestSuite_Protocol
 from .test_reconnect import TestSuite_Reconnect
+from .test_mesh import TestSuite_Mesh
 
 test_cases = (TestSuite_Schema, TestSuite_Request, TestSuite_Protocol,
-              TestSuite_Reconnect)
+              TestSuite_Reconnect, TestSuite_Mesh)
 
 def load_tests(loader, tests, pattern):
     suite = unittest.TestSuite()

unit/suites/lib/tarantool_admin.py

@@ -61,4 +61,4 @@ def execute(self, command):
             if (res.rfind("\n...\n") >= 0 or res.rfind("\r\n...\r\n") >= 0):
                 break
 
-        return yaml.load(res)
+        return yaml.safe_load(res)