fix redis balance server, test=develop (#139)

Signed-off-by: WangXi <wangxi16@baidu.com>

Author: wangxicoding

python/edl/distill/redis/balance_server.py

@@ -23,6 +23,19 @@
 import time
 
 
+def set_keepalive_linux(sock, after_idle_sec=30, interval_sec=10, max_fails=6):
+    """Set TCP keepalive on an open socket.
+
+    It activates after 30 second (after_idle_sec) of idleness,
+    then sends a keepalive ping once every 10 seconds (interval_sec),
+    and closes the connection after 6 failed ping (max_fails), or 60 seconds
+    """
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, after_idle_sec)
+    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, interval_sec)
+    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, max_fails)
+
+
 class Server(object):
     _READ = select.EPOLLIN | select.EPOLLHUP | select.EPOLLERR
     _WRITE = select.EPOLLOUT | select.EPOLLHUP | select.EPOLLERR
@@ -116,6 +129,7 @@ def _init_conn(self):
         # client.getpeername()
 
         client.setblocking(False)
+        set_keepalive_linux(client)
         fd = client.fileno()
         self._epoll.register(fd, self._READ)
         self._clients[fd] = client

python/edl/distill/redis/service_table.py

@@ -59,13 +59,13 @@ def is_servers_update(self, fd, version):
             return new_version, None
 
     def get_servers(self, fd, num):
-        if fd not in self._fd_to_service_name:
+        service_name = self._fd_to_service_name.get(fd)
+        if service_name is None:
             # not register
             return []
-        service_name = self._fd_to_service_name[fd]
 
         if service_name not in self._service_name_to_servers or \
-           self._service_name_to_update[service_name] is True:
+                self._service_name_to_update.get(service_name, False):
             self._refresh_service(service_name)
 
         return list(self._fd_to_servers[fd])
@@ -85,12 +85,11 @@ def add_service_name(self, fd, service_name, num):
         self._service_name_to_update[service_name] = True
 
     def rm_service_name(self, fd):
-        # client maybe exit before register
-        if fd not in self._fd_to_service_name:
+        service_name = self._fd_to_service_name.get(fd)
+        if service_name is None:
+            # client maybe exit before register
             return
 
-        service_name = self._fd_to_service_name[fd]
-
         with self._mutex:
             if service_name in self._service_name_to_fds:
                 try:
@@ -141,7 +140,7 @@ def _refresh_service(self, service_name):
 
         # no change
         if len(rm_servers) == 0 and len(add_servers) == 0 and \
-                self._service_name_to_update[service_name] is False:
+                not self._service_name_to_update.get(service_name, False):
             return
         self._service_name_to_update[service_name] = False
         update_fd = set()
@@ -182,7 +181,7 @@ def _refresh_service(self, service_name):
         # assign: {fd0:32, fd1:32, fd2:32}
         server_max_connect = int((fd_num + server_num - 1) / server_num)
         fd_max_connect = max(1, int(server_num / fd_num))
-        #fd_max_connect = int((server_num + fd_num - 1) / fd_num)
+        # fd_max_connect = int((server_num + fd_num - 1) / fd_num)
         print('fd_num={}, server_num={}, smax={}, mcon={}'.format(
             fd_num, server_num, server_max_connect, fd_max_connect))
 
@@ -253,9 +252,17 @@ def _refresh(self):
                 except KeyError:
                     pass
 
-            time.sleep(2)
+            time.sleep(3)
+
+    def refresh(self):
+        while True:
+            try:
+                self._refresh()
+            except Exception as e:
+                sys.stderr.write(str(e) + '\n')
+                time.sleep(6)
 
     def start(self):
-        self._thread = threading.Thread(target=self._refresh)
+        self._thread = threading.Thread(target=self.refresh)
         self._thread.daemon = True
         self._thread.start()