Skip to content

Commit 359bad4

Browse files
xxnetxxnet
committed
fix bug: h2 fail and retry, too many worker
1 parent 49f2c05 commit 359bad4

11 files changed

Lines changed: 93 additions & 120 deletions

File tree

‎code/default/download.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11

22
## 下载(Download)：
33
测试版(Test)：
4-
https://codeload.github.com/XX-net/XX-Net/zip/3.1.9
4+
https://codeload.github.com/XX-net/XX-Net/zip/3.1.4
55

66
稳定版(Stable)：
77
https://codeload.github.com/XX-net/XX-Net/zip/3.1.4

‎code/default/gae_proxy/local/config.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,7 @@ def load(self):
127127

128128
self.https_max_connect_thread = config.CONFIG.getint("connect_manager", "https_max_connect_thread")
129129
self.connect_interval = config.CONFIG.getint("connect_manager", "connect_interval")
130+
self.max_worker_num = config.CONFIG.getint("connect_manager", "max_worker_num")
130131

131132
self.log_file = config.CONFIG.getint("system", "log_file")
132133

‎code/default/gae_proxy/local/connect_manager.py‎

Lines changed: 35 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -235,7 +235,7 @@ def __init__(self):
235235
p.start()
236236

237237
if self.connection_pool_min_num:
238-
p = threading.Thread(target=self.create_connection_daemon)
238+
p = threading.Thread(target=self.keep_connection_daemon)
239239
p.daemon = True
240240
p.start()
241241

@@ -306,62 +306,40 @@ def save_ssl_connection_for_reuse(self, ssl_sock, host=None, call_time=0):
306306
google_ip.report_connect_closed(ssl_sock.ip, "slowest %d" % ssl_sock.handshake_time)
307307
ssl_sock.close()
308308

309-
def create_connection_daemon(self):
310-
connect_start_num = 0
311-
309+
def keep_connection_daemon(self):
312310
while connect_control.keep_running:
313-
time.sleep(0.1)
314311
if not connect_control.allow_connect():
315312
time.sleep(5)
316313
continue
317314

318-
if self.thread_num > self.max_thread_num:
319-
continue
320-
321-
target_conn_num = self.connection_pool_min_num
322-
if self.new_conn_pool.qsize() > target_conn_num:
323-
time.sleep(1)
315+
if self.new_conn_pool.qsize() > self.connection_pool_min_num:
316+
time.sleep(5)
324317
continue
325318

326-
self.thread_num_lock.acquire()
327-
self.thread_num += 1
328-
self.thread_num_lock.release()
329-
p = threading.Thread(target=self.connect_process)
330-
p.start()
331-
connect_start_num += 1
332-
333-
if connect_start_num > 10:
334-
time.sleep(5)
335-
connect_start_num = 0
319+
self.connect_process()
336320

337321
def connect_process(self):
338322
try:
339323
ip_str = google_ip.get_gws_ip()
340324
if not ip_str:
341325
time.sleep(60)
342-
xlog.warning("no enough ip")
326+
# xlog.warning("no enough ip")
343327
return
344328

345-
port = 443
346329
#xlog.debug("create ssl conn %s", ip_str)
347-
ssl_sock = self._create_ssl_connection( (ip_str, port) )
348-
if ssl_sock:
349-
ssl_sock.last_use_time = time.time()
330+
ssl_sock = self._create_ssl_connection( (ip_str, 443) )
331+
if not ssl_sock:
332+
return
350333

351-
if self.new_conn_pool.qsize() >= self.connection_pool_max_num and self.ssl_timeout_cb:
352-
self.ssl_timeout_cb(ssl_sock)
353-
else:
354-
self.new_conn_pool.put((ssl_sock.handshake_time, ssl_sock))
355-
finally:
356-
self.thread_num_lock.acquire()
357-
self.thread_num -= 1
358-
self.thread_num_lock.release()
334+
self.new_conn_pool.put((ssl_sock.handshake_time, ssl_sock))
335+
except:
336+
pass
359337

360338
def create_more_connection_worker(self):
361339
while connect_control.allow_connect() and \
362340
self.thread_num < self.max_thread_num and \
363-
(self.new_conn_pool.qsize() < self.https_new_connect_num or \
364-
self.new_conn_pool.qsize(only_h1=True) < 1):
341+
(self.new_conn_pool.qsize() < self.https_new_connect_num or
342+
self.new_conn_pool.qsize(only_h1=True) < 1):
365343

366344
self.thread_num_lock.acquire()
367345
self.thread_num += 1
@@ -381,25 +359,33 @@ def connect_thread(self, sleep_time=0):
381359
time.sleep(sleep_time)
382360
try:
383361
while self.new_conn_pool.qsize() < self.https_new_connect_num or \
384-
self.new_conn_pool.qsize(only_h1=True) < 1:
362+
self.new_conn_pool.qsize(only_h1=True) < 1:
363+
364+
if self.new_conn_pool.qsize() > self.connection_pool_max_num:
365+
break
366+
367+
if not connect_control.allow_connect():
368+
break
385369

386370
ip_str = google_ip.get_gws_ip()
387371
if not ip_str:
388-
time.sleep(60)
389372
xlog.warning("no enough ip")
373+
time.sleep(60)
390374
break
391375

392-
port = 443
393376
#xlog.debug("create ssl conn %s", ip_str)
394-
ssl_sock = self._create_ssl_connection( (ip_str, port) )
395-
if ssl_sock:
396-
ssl_sock.last_use_time = time.time()
397-
if self.new_conn_pool.qsize() >= self.connection_pool_max_num and self.ssl_timeout_cb:
398-
self.ssl_timeout_cb(ssl_sock)
399-
else:
400-
self.new_conn_pool.put((ssl_sock.handshake_time, ssl_sock))
401-
elif not connect_control.allow_connect():
402-
break
377+
ssl_sock = self._create_ssl_connection( (ip_str, 443) )
378+
if not ssl_sock:
379+
continue
380+
381+
if self.ssl_timeout_cb and \
382+
self.new_conn_pool.qsize() > self.connection_pool_max_num + 1 and \
383+
self.new_conn_pool.qsize() > self.https_new_connect_num + 1 and \
384+
self.new_conn_pool.qsize(only_h1=True) > 2:
385+
386+
self.ssl_timeout_cb(ssl_sock)
387+
else:
388+
self.new_conn_pool.put((ssl_sock.handshake_time, ssl_sock))
403389
time.sleep(1)
404390
finally:
405391
self.thread_num_lock.acquire()
@@ -482,6 +468,7 @@ def verify_SSL_certificate_issuer(ssl_sock):
482468
xlog.debug("create_ssl update ip:%s time:%d h2:%d", ip, handshake_time, ssl_sock.h2)
483469
ssl_sock.fd = sock.fileno()
484470
ssl_sock.create_time = time_begin
471+
ssl_sock.last_use_time = time.time()
485472
ssl_sock.received_size = 0
486473
ssl_sock.load = 0
487474
ssl_sock.handshake_time = handshake_time

‎code/default/gae_proxy/local/gae_handler.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -653,14 +653,14 @@ def fetch(self, begin, end, first_response):
653653
if data_readed >= content_length:
654654
percent = begin * 100 / self.req_end
655655

656-
xlog.debug('RangeFetch [thread %s] %d%% length:%s range:%s %s %s',
657-
threading.currentThread().ident, percent,
656+
xlog.debug('RangeFetch [%s] %d%% length:%s range:%s %s %s',
657+
response.ssl_sock.ip, percent,
658658
content_length, content_range, self.url, response.task.get_trace())
659659
break
660660

661661
data = response.task.read()
662662
if not data:
663-
xlog.warn("RangeFetch [%s] get body fail %s", threading.currentThread().ident, self.url)
663+
xlog.warn("RangeFetch [%s] get body fail %s", response.ssl_sock.ip, self.url)
664664
break
665665

666666
data_len = len(data)

‎code/default/gae_proxy/local/http2_connection.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -180,7 +180,7 @@ def recv_loop(self):
180180
try:
181181
self._consume_single_frame()
182182
except Exception as e:
183-
#xlog.warn("recv fail:%r", e)
183+
xlog.exception("recv fail:%r", e)
184184
self.close("recv fail:%r" % e)
185185

186186
def get_rtt_rate(self):

‎code/default/gae_proxy/local/http2_stream.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -259,7 +259,7 @@ def receive_frame(self, frame):
259259
# We've handled the headers, zero them out.
260260
self.response_header_datas = None
261261

262-
self.task.content_length = int(self.response_headers.get("Content-Length", "0"))
262+
self.task.content_length = int(self.response_headers["Content-Length"][0])
263263
time_now = self.task.set_state("h2_get_head")
264264
self.get_head_time = time_now
265265
self.send_response()

‎code/default/gae_proxy/local/http_common.py‎

Lines changed: 0 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -6,47 +6,6 @@
66
xlog = getLogger("gae_proxy")
77

88

9-
class ReadBuffer(object):
10-
def __init__(self, buf, begin=0, size=None):
11-
buf_len = len(buf)
12-
if size is None:
13-
if begin > buf_len:
14-
raise Exception("ReadBuffer buf_len:%d, start:%d" % (buf_len, begin))
15-
size = buf_len - begin
16-
elif begin + size > buf_len:
17-
raise Exception("ReadBuffer buf_len:%d, start:%d len:%d" % (buf_len, begin, size))
18-
19-
self.size = size
20-
self.buf = buf
21-
self.begin = begin
22-
23-
def __len__(self):
24-
return self.size
25-
26-
def get(self, size=None):
27-
if size is None:
28-
size = self.size
29-
elif size > self.size:
30-
raise Exception("ReadBuffer get %d but left %d" % (size, self.size))
31-
32-
data = self.buf[self.begin:self.begin + size]
33-
self.begin += size
34-
self.size -= size
35-
return data
36-
37-
def get_buf(self, size=None):
38-
if size is None:
39-
size = self.size
40-
elif size > self.size:
41-
raise Exception("ReadBuffer get %d but left %d" % (size, self.size))
42-
43-
buf = ReadBuffer(self.buf, self.begin, size)
44-
45-
self.begin += size
46-
self.size -= size
47-
return buf
48-
49-
509
class GAE_Exception(Exception):
5110
def __init__(self, type, message):
5211
xlog.debug("GAE_Exception %r %r", type, message)
@@ -59,7 +18,6 @@ def __init__(self, status=601, reason="", headers={}, body=""):
5918
self.status = status
6019
self.reason = reason
6120
self.headers = headers
62-
self.body = ReadBuffer(body)
6321

6422

6523
class Task(object):

‎code/default/gae_proxy/local/http_dispatcher.py‎

Lines changed: 33 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -24,15 +24,14 @@
2424
import operator
2525
import Queue
2626

27-
import socks
2827

2928
from config import config
3029
from appids_manager import appid_manager
3130
import connect_control
3231
from connect_manager import https_manager
3332
from http1 import HTTP1_worker
3433
from http2_connection import HTTP2_worker
35-
from http_common import *
34+
import http_common
3635
from xlog import getLogger
3736
xlog = getLogger("gae_proxy")
3837

@@ -41,23 +40,6 @@
4140
g_cacertfile = os.path.join(current_path, "cacert.pem")
4241

4342

44-
def load_proxy_config():
45-
if config.PROXY_ENABLE:
46-
if config.PROXY_TYPE == "HTTP":
47-
proxy_type = socks.HTTP
48-
elif config.PROXY_TYPE == "SOCKS4":
49-
proxy_type = socks.SOCKS4
50-
elif config.PROXY_TYPE == "SOCKS5":
51-
proxy_type = socks.SOCKS5
52-
else:
53-
xlog.error("proxy type %s unknown, disable proxy", config.PROXY_TYPE)
54-
config.PROXY_ENABLE = 0
55-
return
56-
57-
socks.set_default_proxy(proxy_type, config.PROXY_HOST, config.PROXY_PORT, config.PROXY_USER, config.PROXY_PASSWD)
58-
load_proxy_config()
59-
60-
6143
class HttpsDispatcher(object):
6244
min_worker_num = 20
6345
idle_time = 20 * 60
@@ -83,7 +65,7 @@ def on_ssl_created_cb(self, ssl_sock):
8365
if not appid:
8466
time.sleep(60)
8567
ssl_sock.close()
86-
raise GAE_Exception(1, "no appid can use")
68+
raise http_common.GAE_Exception(1, "no appid can use")
8769

8870
ssl_sock.appid = appid
8971
ssl_sock.host = ssl_sock.appid + ".appspot.com"
@@ -96,6 +78,8 @@ def on_ssl_created_cb(self, ssl_sock):
9678
self.h1_num += 1
9779

9880
self.workers.append(worker)
81+
self.check_worker_num()
82+
9983
return worker
10084

10185
def create_worker_thread(self):
@@ -152,17 +136,44 @@ def get_worker(self):
152136

153137
ssl_sock = https_manager.get_ssl_connection()
154138
if not ssl_sock:
155-
raise GAE_Exception(1, "no ssl_sock")
139+
raise http_common.GAE_Exception(1, "no ssl_sock")
156140

157141
worker = self.on_ssl_created_cb(ssl_sock)
158142

159143
return worker
160144

145+
def check_worker_num(self):
146+
if len(self.workers) <= config.max_worker_num:
147+
return
148+
149+
slowest_rtt = 9999
150+
slowest_worker = None
151+
idle_num = 0
152+
for worker in self.workers:
153+
if not worker.accept_task:
154+
continue
155+
156+
if worker.version == "2" and len(worker.streams) > 0:
157+
continue
158+
159+
idle_num += 1
160+
161+
rtt = worker.get_rtt_rate()
162+
163+
if rtt > slowest_rtt:
164+
slowest_rtt = rtt
165+
slowest_worker = worker
166+
167+
if idle_num < 3 or slowest_worker is None:
168+
return
169+
170+
self.close_cb(slowest_worker)
171+
161172
def request(self, headers, body):
162173
# xlog.debug("task start request")
163174
self.last_request_time = time.time()
164175
q = Queue.Queue()
165-
task = Task(headers, body, q)
176+
task = http_common.Task(headers, body, q)
166177
task.set_state("start_request")
167178
self.request_queue.put(task)
168179
response = q.get(True)

‎code/default/gae_proxy/local/openssl_wrap.py‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,21 @@ def __iowait(self, io_func, *args, **kwargs):
7575
time_now = time.time()
7676
if time_now - time_start > timeout:
7777
break
78+
except OpenSSL.SSL.SysCallError as e:
79+
if e[0] == 10035 and 'WSAEWOULDBLOCK' in e[1]:
80+
sys.exc_clear()
81+
if io_func == self._connection.send:
82+
_, _, errors = select.select([], [fd], [fd], timeout)
83+
else:
84+
_, _, errors = select.select([fd], [], [fd], timeout)
85+
86+
if errors:
87+
raise
88+
time_now = time.time()
89+
if time_now - time_start > timeout:
90+
break
91+
else:
92+
raise e
7893
except Exception as e:
7994
#xlog.exception("e:%r", e)
8095
raise e

‎code/default/gae_proxy/local/proxy.ini‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ https_new_connect_num = 3
103103
https_connection_pool_min = 1
104104

105105
; if exceed max, put ssl to worker.
106-
https_connection_pool_max = 5
106+
https_connection_pool_max = 30
107107

108108
; keep connection pool until no active request timeout
109109
keep_active_timeout = 600
@@ -113,6 +113,7 @@ https_keep_alive = 50
113113

114114
connect_interval = 40
115115

116+
max_worker_num = 50
116117

117118
[system]
118119
log_file = 0

0 commit comments

Comments
 (0)