ClearML/clearml-session
1
0
Fork 0
mirror of https://github.com/clearml/clearml-session synced 2025-04-30 02:40:20 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Improve Windows stable connection support layer

Browse Source
This commit is contained in:
allegroai 2020-12-31 22:19:32 +02:00
parent efc23cc96a
commit fcfa20bade
1 changed files with 42 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
clearml_session/single_thread_proxy.py
View File

@ -13,10 +13,11 @@ class SingleThreadProxy(object):
self.forward = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.forward = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def start(self, host, port): def start(self, host, port):
# noinspection PyBroadException
try: try:
self.forward.connect((host, port)) self.forward.connect((host, port))
return self.forward return self.forward
except Exception as e: except Exception:
return False return False
def __init__(self, port, tgtport, host="127.0.0.1", tgthost="127.0.0.1", def __init__(self, port, tgtport, host="127.0.0.1", tgthost="127.0.0.1",
@ -51,39 +52,48 @@ class SingleThreadProxy(object):
def main_loop(self): def main_loop(self):
self.input_list.append(self.server) self.input_list.append(self.server)
ss = select.select
while 1: while 1:
time.sleep(self.delay) time.sleep(self.delay)
# noinspection PyBroadException
try: try:
inputready, outputready, exceptready = ss(self.input_list, [], []) inputready, outputready, exceptready = select.select(self.input_list, [], [])
except: except Exception:
continue continue
for self.s in inputready: for s in inputready:
if self.s == self.server: if s == self.server:
# noinspection PyBroadException
try: try:
self.on_accept() self.on_accept()
except: except Exception:
pass pass
break break
# noinspection PyBroadException
try: try:
self.data = self.s.recv(self.buffer_size) data = s.recv(self.buffer_size)
except: except ConnectionResetError:
# this will trigger on_close
data = []
except Exception:
continue continue
if len(self.data) == 0:
if len(data) == 0:
# noinspection PyBroadException
try: try:
self.on_close() self.on_close(s)
except: except Exception:
pass pass
break break
else: else:
# noinspection PyBroadException
try: try:
self.on_recv() self.on_recv(s, data)
except: except Exception:
pass pass
def on_accept(self): def on_accept(self):
clientsock, clientaddr = self.server.accept() clientsock, clientaddr = self.server.accept()
forward = None
for i in range(self.max_timeout_for_remote_connection): for i in range(self.max_timeout_for_remote_connection):
forward = self.Forward().start(self.tgthost, self.tgtport) forward = self.Forward().start(self.tgthost, self.tgtport)
if forward: if forward:
@ -92,35 +102,35 @@ class SingleThreadProxy(object):
time.sleep(1) time.sleep(1)
if forward: if forward:
# logger.info("{0} has connected".format(clientaddr)) # print("{0} has connected".format(clientaddr))
self.input_list.append(clientsock) self.input_list.append(clientsock)
self.input_list.append(forward) self.input_list.append(forward)
self.channel[clientsock] = forward self.channel[clientsock] = forward
self.channel[forward] = clientsock self.channel[forward] = clientsock
_sidbase = "{0}_{1}_{2}_{3}".format(self.tgthost, self.tgtport, clientaddr[0], clientaddr[1]) sidbase = "{0}_{1}_{2}_{3}".format(self.tgthost, self.tgtport, clientaddr[0], clientaddr[1])
self.sidmap[clientsock] = (_sidbase, 1) self.sidmap[clientsock] = (sidbase, 1)
self.sidmap[forward] = (_sidbase, -1) self.sidmap[forward] = (sidbase, -1)
else: else:
# logger.warn("Can't establish connection with remote server.\n" # print("Can't establish connection with remote server.\n"
# "Closing connection with client side{0}".format(clientaddr)) # "Closing connection with client side{0}".format(clientaddr))
clientsock.close() clientsock.close()
def on_close(self): def on_close(self, s):
# logger.info("{0} has disconnected".format(self.s.getpeername())) # logger.info("{0} has disconnected".format(self.s.getpeername()))
# print("has disconnected")
self.input_list.remove(self.s) self.input_list.remove(s)
self.input_list.remove(self.channel[self.s]) self.input_list.remove(self.channel[s])
out = self.channel[self.s] out = self.channel[s]
self.channel[out].close() self.channel[out].close()
self.channel[self.s].close() self.channel[s].close()
del self.channel[out] del self.channel[out]
del self.channel[self.s] del self.channel[s]
del self.sidmap[out] del self.sidmap[out]
del self.sidmap[self.s] del self.sidmap[s]
def on_recv(self): def on_recv(self, s, data):
_sidbase = self.sidmap[self.s][0] _sidbase = self.sidmap[s][0]
_c_or_s = self.sidmap[self.s][1] _c_or_s = self.sidmap[s][1]
data = self.data
# logger.debug(ctrl_less(data.strip())) # logger.debug(ctrl_less(data.strip()))
self.channel[self.s].send(data) self.channel[s].send(data)