fix bug for scan ip

Author: xxnet

code/default/gae_proxy/lang/fa_IR/LC_MESSAGES/messages.po

@@ -3,7 +3,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PROJECT VERSION\n"
 "Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
-"POT-Creation-Date: 2016-04-17 22:25+0800\n"
+"POT-Creation-Date: 2016-04-30 14:57+0800\n"
 "PO-Revision-Date: 2015-12-06 09:14+0800\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language: fa_IR\n"
@@ -246,9 +246,6 @@ msgstr "خودکار تنظیم تعداد موضوع اسکن"
 msgid "Max scan thread count"
 msgstr "حداکثر تعداد موضوع اسکن"
 
-msgid "Update"
-msgstr "به روز رسانی"
-
 msgid "IP range list"
 msgstr "لیست محدوده IP"
 
@@ -270,6 +267,9 @@ msgstr "شبکه محلی ما شکست خورده است."
 msgid "IP Number"
 msgstr "IP شماره"
 
+msgid "Is Idle"
+msgstr "آماده به کار"
+
 msgid "IP Quality"
 msgstr "IP کیفیت"
 
@@ -336,12 +336,6 @@ msgstr "نمایش جزئیات"
 msgid "Post to Github issue needs to sign in your Github account"
 msgstr "بازخورد یک کلید نیاز به ثبت نام در حساب github خود را"
 
-msgid "To Google Groups"
-msgstr "به Google Groups"
-
-msgid "Post to Github Issue..."
-msgstr "بررسی مسائل _Github_..."
-
 msgid "GAE Proxy Status Info"
 msgstr "GAE پروکسی وضعیت"
 
@@ -392,6 +386,9 @@ msgstr ""
 msgid "No working appid. Please check. "
 msgstr "بدون APPID کارگر. لطفا بررسی کنید. "
 
+msgid "System is Idle. "
+msgstr ""
+
 msgid "Connection not established yet. "
 msgstr "اتصال برقرار نشده است"
 
@@ -429,7 +426,10 @@ msgstr "شما با استفاده از شناسه های عمومی است."
 msgid "new:"
 msgstr ""
 
-msgid " used:"
+msgid " h1:"
+msgstr ""
+
+msgid " h2:"
 msgstr ""
 
 msgid "Auto proxy enabled at "
@@ -441,3 +441,7 @@ msgstr "پروکسی را فعال کنید در "
 msgid "Proxy disabled"
 msgstr "پروکسی غیرفعال"
 
+
+
+
+

code/default/gae_proxy/lang/zh_CN/LC_MESSAGES/messages.po

@@ -3,7 +3,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: PROJECT VERSION\n"
 "Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
-"POT-Creation-Date: 2016-04-17 22:25+0800\n"
+"POT-Creation-Date: 2016-04-30 14:57+0800\n"
 "PO-Revision-Date: 2015-12-16 15:55+0800\n"
 "Last-Translator: XX-Net <[email protected]>\n"
 "Language: zh_Hans_CN\n"
@@ -242,9 +242,6 @@ msgstr "自动调整扫描线程数"
 msgid "Max scan thread count"
 msgstr "最大扫描线程数"
 
-msgid "Update"
-msgstr "更新"
-
 msgid "IP range list"
 msgstr "IP段列表"
 
@@ -266,6 +263,9 @@ msgstr "本地网络状态"
 msgid "IP Number"
 msgstr "IP数量"
 
+msgid "Is Idle"
+msgstr "空闲"
+
 msgid "IP Quality"
 msgstr "IP延迟"
 
@@ -332,12 +332,6 @@ msgstr "显示详细信息"
 msgid "Post to Github issue needs to sign in your Github account"
 msgstr "贴到Github问题区需要登录Github账号"
 
-msgid "To Google Groups"
-msgstr "前往Google论坛"
-
-msgid "Post to Github Issue..."
-msgstr "贴到Github问题区..."
-
 msgid "GAE Proxy Status Info"
 msgstr "GAEProxy状态信息"
 
@@ -390,6 +384,9 @@ msgstr ""
 msgid "No working appid. Please check. "
 msgstr "没有可用APPID，请检查。"
 
+msgid "System is Idle. "
+msgstr ""
+
 msgid "Connection not established yet. "
 msgstr "尚未建立连接"
 
@@ -427,8 +424,11 @@ msgstr "你正使用公共AppID"
 msgid "new:"
 msgstr "新:"
 
-msgid " used:"
-msgstr " 已用:"
+msgid " h1:"
+msgstr ""
+
+msgid " h2:"
+msgstr ""
 
 msgid "Auto proxy enabled at "
 msgstr "智能代理已启用："
@@ -439,3 +439,7 @@ msgstr "全局代理已启用："
 msgid "Proxy disabled"
 msgstr "全局代理已禁用"
 
+
+
+
+

code/default/gae_proxy/local/appids_manager.py

@@ -52,7 +52,7 @@ def report_not_exist(self, appid, ip):
         th.start()
 
     def process_appid_not_exist(self, appid, ip):
-        if check_ip.test_gae_ip(ip, "xxnet-1"):
+        if check_ip.test_gae_ip2(ip, "xxnet-1"):
             self.set_appid_not_exist(appid)
         else:
             xlog.warn("process_appid_not_exist, remove ip:%s", ip)

code/default/gae_proxy/local/check_ip.py

@@ -186,28 +186,6 @@ def check_goagent(ssl_sock, appid):
     return True
 
 
-# export api for google_ip, appid_manager
-def test_gae_ip(ip, appid=None):
-    try:
-        ssl_sock = connect_ssl(ip, timeout=max_timeout)
-        get_ssl_cert_domain(ssl_sock)
-
-        if not appid:
-            appid = "xxnet-1"
-        if not check_goagent(ssl_sock, appid):
-            return False
-
-        return ssl_sock
-    except socket.timeout:
-        if __name__ == "__main__":
-            xlog.warn("connect timeout")
-        return False
-    except Exception as e:
-        if __name__ == "__main__":
-            xlog.exception("test_gae_ip %s e:%r",ip, e)
-        return False
-
-
 def test_gae_ip2(ip, appid="xxnet-1", use_openssl=True):
     if use_openssl:
         try:
@@ -231,7 +209,9 @@ def test_gae_ip2(ip, appid="xxnet-1", use_openssl=True):
                     return False
                 else:
                     return ssl_sock
-            except:
+            except Exception as e:
+                if __name__ == "__main__":
+                    xlog.exception("check fail:%r", e)
                 return False
     else:
         ssl_sock = None

code/default/gae_proxy/local/check_local_network.py

@@ -70,6 +70,23 @@ def report_network_ok():
     last_check_time = time.time()
     continue_fail_count = 0
 
+
+def report_network_fail():
+    global network_stat, last_check_time, continue_fail_count
+    continue_fail_count += 1
+    last_check_time = time.time()
+
+    if continue_fail_count > 10:
+        network_stat = "unkown"
+        xlog.debug("report_connect_fail continue_fail_count:%d", continue_fail_count)
+        triger_check_network()
+
+
+def is_ok():
+    global network_stat, last_check_time, continue_fail_count
+    return network_stat == "OK"
+
+
 def _check_worker():
     global _checking_lock, _checking_num, network_stat, last_check_time
     time_now = time.time()

code/default/gae_proxy/local/config.py

@@ -86,7 +86,6 @@ def load(self):
         self.HOSTS_DIRECT = tuple(direct_hosts)
 
         self.AUTORANGE_MAXSIZE = self.CONFIG.getint('autorange', 'maxsize')
-        self.AUTORANGE_WAITSIZE = self.CONFIG.getint('autorange', 'waitsize')
         self.AUTORANGE_BUFSIZE = self.CONFIG.getint('autorange', 'bufsize')
         self.AUTORANGE_THREADS = self.CONFIG.getint('autorange', 'threads')
 

code/default/gae_proxy/local/gae_handler.py

@@ -67,6 +67,7 @@
 
 from config import config
 from google_ip import google_ip
+import check_local_network
 from http_dispatcher import http_dispatch
 from http_common import *
 
@@ -231,6 +232,7 @@ def fetch_by_gae(method, url, headers, body):
     if response.app_status != 200:
         return response
 
+    check_local_network.report_network_ok()
     try:
         data = response.body.get(2)
         if len(data) < 2:
@@ -390,7 +392,6 @@ class RangeFetch(object):
     threads = config.AUTORANGE_THREADS
     maxsize = config.AUTORANGE_MAXSIZE
     bufsize = config.AUTORANGE_BUFSIZE
-    waitsize = config.AUTORANGE_WAITSIZE
 
     def __init__(self, method, url, headers, body, response, wfile):
         self.method = method
@@ -441,9 +442,11 @@ def fetch(self):
         self.expect_begin = start
         for begin in range(end+1, length, self.maxsize):
             range_queue.put((begin, min(begin+self.maxsize-1, length-1), None))
-        for i in xrange(0, self.threads):
+
+        thread_num = min(self.threads, range_queue.qsize())
+        for i in xrange(0, thread_num):
             range_delay_size = i * self.maxsize
-            spawn_later(float(range_delay_size)/self.waitsize, self.__fetchlet, range_queue, data_queue, range_delay_size)
+            spawn_later(i*0.1, self.__fetchlet, range_queue, data_queue, range_delay_size)
 
         has_peek = hasattr(data_queue, 'peek')
         peek_timeout = 120
@@ -496,9 +499,9 @@ def __fetchlet(self, range_queue, data_queue, range_delay_size):
             except Queue.Empty:
                 continue
 
-            if self.expect_begin < start and data_queue.qsize() * self.bufsize + range_delay_size > 30*1024*1024:
+            if start > self.expect_begin and data_queue.qsize() * self.bufsize + range_delay_size > 30*1024*1024:
                 range_queue.put((start, end, response))
-                time.sleep(10)
+                time.sleep(2)
                 continue
             headers['Range'] = 'bytes=%d-%d' % (start, end)
 

code/default/gae_proxy/local/google_ip.py

@@ -203,6 +203,11 @@ def save_ip_list(self, force=False):
         finally:
             self.ip_lock.release()
 
+    def _ip_rate(self, ip_info):
+        return ip_info['handshake_time'] + \
+                    (ip_info['fail_times'] * 1000 ) + \
+                    (ip_info['down_fail'] * 500 )
+
     def try_sort_gws_ip(self, force=False):
         if time.time() - self.last_sort_time_for_gws < 10 and not force:
             return
@@ -215,9 +220,7 @@ def try_sort_gws_ip(self, force=False):
             for ip in self.ip_dict:
                 if 'gws' not in self.ip_dict[ip]['server']:
                     continue
-                ip_rate[ip] = self.ip_dict[ip]['handshake_time'] + \
-                    (self.ip_dict[ip]['fail_times'] * 1000 ) + \
-                    (self.ip_dict[ip]['down_fail'] * 500 )
+                ip_rate[ip] = self._ip_rate(self.ip_dict[ip])
                 if self.ip_dict[ip]['fail_times'] == 0:
                     self.good_ip_num += 1
 
@@ -246,7 +249,7 @@ def adjust_scan_thread_num(self, max_scan_ip_thread_num=None):
         else:
             try:
                 the_100th_ip = self.gws_ip_list[99]
-                the_100th_handshake_time = self.ip_dict[the_100th_ip]['handshake_time']
+                the_100th_handshake_time = self._ip_rate(self.ip_dict[the_100th_ip])
                 scan_ip_thread_num = int( (the_100th_handshake_time - 200)/2 * self.max_scan_ip_thread_num/50 )
             except Exception as e:
                 xlog.warn("adjust_scan_thread_num fail:%r", e)
@@ -400,7 +403,7 @@ def update_ip(self, ip, handshake_time):
             return
 
         time_now = time.time()
-        check_local_network.network_stat = "OK"
+        check_local_network.report_network_ok()
         check_ip.last_check_time = time_now
         check_ip.continue_fail_count = 0
 
@@ -458,15 +461,12 @@ def report_connect_fail(self, ip, force_remove=False):
             self.ip_dict[ip]['links'] -= 1
 
             # ignore if system network is disconnected.
-            if check_local_network.network_stat == "Fail":
+            if not check_local_network.is_ok():
                 xlog.debug("report_connect_fail network fail")
                 return
 
-            check_local_network.continue_fail_count += 1
-            if check_local_network.continue_fail_count > 10:
-                check_local_network.network_stat = "unknown"
-                xlog.debug("report_connect_fail continue_fail_count:%d", check_local_network.continue_fail_count)
-                check_local_network.triger_check_network()
+            check_local_network.report_network_fail()
+            if not check_local_network.is_ok():
                 return
 
             fail_time = self.ip_dict[ip]["fail_time"]
@@ -480,7 +480,6 @@ def report_connect_fail(self, ip, force_remove=False):
             self.append_ip_history(ip, "fail")
             self.ip_dict[ip]["fail_time"] = time_now
 
-            check_local_network.triger_check_network()
             self.to_check_ip_queue.put((ip, time_now + 10))
             xlog.debug("report_connect_fail:%s", ip)
 
@@ -541,7 +540,7 @@ def check_ip_process(self):
             if time_wait > 0:
                 time.sleep(time_wait)
 
-            if check_local_network.network_stat == "Fail":
+            if not check_local_network.is_ok():
                 try:
                     if self.ip_dict[ip]['fail_times']:
                         self.ip_dict[ip]['fail_times'] = 0
@@ -550,7 +549,7 @@ def check_ip_process(self):
                     pass
                 continue
 
-            result = check_ip.test_gae_ip(ip)
+            result = check_ip.test_gae_ip2(ip)
             if result:
                 self.add_ip(ip, result.handshake_time, result.domain, "gws")
                 xlog.debug("restore ip:%s", ip)
@@ -588,6 +587,25 @@ def remove_slowest_ip(self):
         finally:
             self.ip_lock.release()
 
+    def recheck_ip(self, ip):
+        # recheck ip if not work.
+        # can block.
+        if not check_local_network.is_ok():
+            xlog.debug("recheck_ip:%s network is fail", ip)
+            return
+
+        self.report_connect_fail(ip)
+
+        connect_control.start_connect_register()
+        result = check_ip.test_gae_ip2(ip)
+        connect_control.end_connect_register()
+        if not result:
+            self.report_connect_fail(ip, force_remove=True)
+            xlog.debug("recheck_ip:%s real fail, removed.", ip)
+        else:
+            self.add_ip(ip, result.handshake_time, result.domain, "gws")
+            xlog.debug("recheck_ip:%s restore okl", ip)
+
     def scan_ip_worker(self):
         while self.scan_thread_count <= self.scan_ip_thread_num and connect_control.keep_running:
             if not connect_control.allow_scan():