5.6.1 refactor and fix bug of resource leaking.

Author: Michael-X-Net

code/default/launcher/web_control.py

@@ -855,7 +855,6 @@ def req_debug_handler(self):
             dat += "thread num:%d\r\n" % threading.active_count()
             for thread in threading.enumerate():
                 dat += "\nThread: %s \r\n" % (thread.name)
-                # dat += traceback.format_exc(sys._current_frames()[thread.ident])
                 stack = sys._current_frames()[thread.ident]
                 st = traceback.extract_stack(stack)
                 stl = traceback.format_list(st)

code/default/lib/noarch/front_base/boringssl_wrap.py

@@ -19,6 +19,7 @@ def __init__(self, context, sock, ip_str=None, sni=None, on_close=None):
         self._context = context
         self._sock = sock
         self._fileno = self._sock.fileno()
+        # self._context.logger.debug("sock %s init fd:%d", ip_str, self._fileno)
         self.ip_str = utils.to_bytes(ip_str)
         self.sni = sni
         self._makefile_refs = 0
@@ -240,18 +241,49 @@ def close(self):
             self.running = False
             if not self.socket_closed:
                 if self._connection:
-                    bssl.BSSL_SSL_shutdown(self._connection)
+                    res = bssl.BSSL_SSL_shutdown(self._connection)
+                    # res == 0: close_notify sent but not recv, means you need to call SSL_shutdown again if you want a full bidirectional shutdown.
+                    # res == 1: success, mean you previously received a close_notify alert from the other peer, and you're totally done
+                    # res == -1: failed
+                    # self._context.logger.debug("sock %s SSL_shutdown fd:%d res:%d", self.ip_str, self._fileno, res)
+
+                    if res < 0:
+                        error = bssl.BSSL_SSL_get_error(self._connection, res)
+                        # self._context.logger.debug("sock %s shutdown fd:%d error:%d", self.ip_str, self._fileno, error)
+                        if error == 1:
+                            p = ffi.new("char[]",
+                                        b"hello, worldhello, worldhello, worldhello, worldhello, world")  # p is a 'char *'
+                            q = ffi.new("char **", p)  # q is a 'char **'
+                            line_no = 0
+                            line_no_p = ffi.new("int *", line_no)
+                            error = bssl.BSSL_ERR_get_error_line(q, line_no_p)
+                            filename = ffi.string(q[0])
+                            # self._context.logger.error("error:%d file:%s, line:%s", error, filename, line_no_p[0])
+                            self._context.logger.debug("sock %s shutdown error: %s, file:%s, line:%d, sni:%s" %
+                                               (self.ip_str, error, filename, line_no_p[0], self.sni))
+                        else:
+                            self._context.logger.debug("sock %s shutdown error:%s" % (self.ip_str, error))
+
+                    bssl.BSSL_SSL_free(self._connection)
+                    self._connection = None
+
+                if self._sock:
+                    try:
+                        self._sock.close()
+                        # self._context.logger.debug("sock %s sock_close fd:%d", self.ip_str, self._fileno)
+                    except Exception as e:
+                        # self._context.logger.debug("sock %s sock_close fd:%d e:%r", self.ip_str, self._fileno, e)
+                        pass
+                    self._sock = None
 
                 self.socket_closed = True
+
                 if self._on_close:
                     self._on_close(self.ip_str, self.sni)
+                    self._on_close = None
 
     def __del__(self):
         self.close()
-        if self._connection:
-            bssl.BSSL_SSL_free(self._connection)
-            self._connection = None
-        self._sock = None
 
     def settimeout(self, t):
         if not self.running:

code/default/lib/noarch/front_base/connect_manager.py

@@ -312,8 +312,9 @@ def _create_ssl_connection(self, ip_str, sni, host):
             t = threading.Thread(target=self._connect_ssl, args=fn_args, name="connect_ssl_%s" % ip_str)
             t.start()
             try:
-                ssl_sock = q.get(timeout=3)
+                ssl_sock = q.get(timeout=30)
             except:
+                self.logger.warn("connect_ssl_timeout %s", ip_str)
                 raise socket.error("timeout")
 
             if not isinstance(ssl_sock, SSLConnection):

code/default/lib/noarch/front_base/http_common.py

@@ -1,3 +1,4 @@
+import threading
 import time
 import random
 
@@ -204,6 +205,8 @@ def __init__(self, logger, ip_manager, config, ssl_sock, close_cb, retry_task_cb
         self.retry_task_cb = retry_task_cb
         self.idle_cb = idle_cb
         self.log_debug_data = log_debug_data
+
+        self._lock = threading.Lock()
         self.accept_task = True
         self.keep_running = True
         self.processed_tasks = 0
@@ -215,6 +218,7 @@ def __init__(self, logger, ip_manager, config, ssl_sock, close_cb, retry_task_cb
         self.last_send_time = self.ssl_sock.create_time
         self.life_end_time = self.ssl_sock.create_time + \
                              random.randint(self.config.connection_max_life, int(self.config.connection_max_life * 1.5))
+        # self.logger.debug("worker.init %s", self.ip_str)
 
     def __str__(self):
         o = ""
@@ -260,24 +264,30 @@ def update_debug_data(self, rtt, sent, received, speed):
         # else:
         #     self.rtt = rtt
 
-        # self.log_debug_data(rtt, sent, received)
+        self.log_debug_data(rtt, sent, received)
         return
 
     def close(self, reason):
-        if not self.keep_running:
-            self.logger.warn("worker already closed %s", self.ip_str)
-            return
-
-        self.accept_task = False
-        self.keep_running = False
-        self.ssl_sock.close()
-        if reason not in ["idle timeout", "life end"]:
-            now = time.time()
-            inactive_time = now - self.last_recv_time
-            if inactive_time < self.config.http2_ping_min_interval:
-                self.logger.debug("%s worker close:%s inactive:%d", self.ip_str, reason, inactive_time)
-        self.ip_manager.report_connect_closed(self.ssl_sock.ip_str, self.ssl_sock.sni, reason)
-        self.close_cb(self)
+        with self._lock:
+            if not self.keep_running:
+                # self.logger.warn("worker %s already closed %s", self.ip_str, reason)
+                return
+
+            # self.logger.debug("worker.close %s reason:%s", self.ip_str, reason)
+            self.accept_task = False
+            self.keep_running = False
+            self.ssl_sock.close()
+            if reason not in ["idle timeout", "life end"]:
+                now = time.time()
+                inactive_time = now - self.last_recv_time
+                if inactive_time < self.config.http2_ping_min_interval:
+                    self.logger.debug("%s worker close:%s inactive:%d", self.ip_str, reason, inactive_time)
+            self.ip_manager.report_connect_closed(self.ssl_sock.ip_str, self.ssl_sock.sni, reason)
+            self.close_cb(self)
+
+    def __del__(self):
+        # self.logger.debug("__del__ %s", self.ip_str)
+        self.close("__del__")
 
     def get_score(self):
         # The smaller, the better

code/default/lib/noarch/simple_http_client.py

@@ -129,6 +129,17 @@ def __init__(self, sock):
         self.select2 = selectors.DefaultSelector()
         self.select2.register(sock, selectors.EVENT_READ)
 
+    def __del__(self):
+        try:
+            self.select2.unregister(self.sock)
+        except:
+            pass
+
+        try:
+            socket.socket.close(self.sock)
+        except:
+            pass
+
     def recv(self, to_read=8192, timeout=30.0):
         if timeout < 0:
             raise Exception("recv timeout")

code/default/lib/noarch/xlog.py

@@ -27,7 +27,7 @@ def __init__(self, name, buffer_size=0, file_name=None, roll_num=1,
                  log_path=None, save_start_log=0, save_warning_log=False):
         self.name = str(name)
         self.file_max_size = 1024 * 1024
-        self.buffer_lock = threading.Lock()
+        self.buffer_lock = threading.RLock()
         self.buffer = {}  # id => line
         self.buffer_size = buffer_size
         self.last_no = 0

code/default/smart_router/local/gfw_black_list.txt

@@ -1838,6 +1838,7 @@ google.ca
 google.calstate.edu
 google.cd
 google.ci
+google.cn
 google.co.id
 google.co.jp
 google.co.kr

code/default/smart_router/local/smart_route.py

@@ -580,13 +580,13 @@ def handle_domain_proxy(sock, host, port, client_address, left_buf=""):
         if not g.domain_cache.accept_gae(host):
             rule_list.remove("gae")
     elif g.config.country_code == "CN":
-        if g.gfwlist.in_white_list(host):
-            rule_list = ["direct", "gae", "socks", "redirect_https"]
-        elif g.gfwlist.in_block_list(host):
+        if g.gfwlist.in_block_list(host):
             if g.config.pac_policy == "black_X-Tunnel":
                 rule_list = ["socks", "redirect_https", "direct", "gae"]
             else:
                 rule_list = ["gae", "socks", "redirect_https", "direct"]
+        elif g.gfwlist.in_white_list(host):
+            rule_list = ["direct", "gae", "socks", "redirect_https"]
         else:
             ips = g.dns_query.query_recursively(host, 1)
             if g.ip_region.check_ips(ips):

code/default/version.txt

@@ -1 +1 @@
-5.6.0
+5.6.1