Asyncore fixes

- TCP fixes
This commit is contained in:
Peter Šurda 2017-05-27 21:52:56 +02:00
parent 99e714c432
commit 21f6d38ec2
Signed by untrusted user: PeterSurda
GPG Key ID: 0C5F50C0B5F37D87
5 changed files with 15 additions and 60 deletions

View File

@ -34,7 +34,7 @@ class AdvancedDispatcher(asyncore.dispatcher):
return True return True
def process(self): def process(self):
if self.state not in ["init", "tls_handshake"] and len(self.read_buf) == 0: if self.state != "tls_handshake" and len(self.read_buf) == 0:
return return
if not self.connected: if not self.connected:
return return
@ -54,7 +54,7 @@ class AdvancedDispatcher(asyncore.dispatcher):
self.state = state self.state = state
def writable(self): def writable(self):
return self.connecting or len(self.write_buf) > 0 or not self.writeQueue.empty() return self.connected and (len(self.write_buf) > 0 or not self.writeQueue.empty())
def readable(self): def readable(self):
return self.connecting or len(self.read_buf) < AdvancedDispatcher._buf_len return self.connecting or len(self.read_buf) < AdvancedDispatcher._buf_len

View File

@ -57,7 +57,7 @@ import warnings
import os import os
from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \ from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \
ENOTCONN, ESHUTDOWN, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \ ENOTCONN, ESHUTDOWN, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \
ECONNREFUSED, \ ECONNREFUSED, EHOSTUNREACH, \
errorcode errorcode
try: try:
from errno import WSAEWOULDBLOCK from errno import WSAEWOULDBLOCK
@ -66,7 +66,7 @@ except:
from ssl import SSLError, SSL_ERROR_WANT_READ, SSL_ERROR_WANT_WRITE from ssl import SSLError, SSL_ERROR_WANT_READ, SSL_ERROR_WANT_WRITE
_DISCONNECTED = frozenset((ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE, _DISCONNECTED = frozenset((ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE,
EBADF, ECONNREFUSED)) EBADF, ECONNREFUSED, EHOSTUNREACH))
OP_READ = 1 OP_READ = 1
OP_WRITE = 2 OP_WRITE = 2

View File

@ -469,28 +469,6 @@ class BMProto(AdvancedDispatcher, ObjectTracker):
payload += struct.pack('>H', peer.port) # remote port payload += struct.pack('>H', peer.port) # remote port
return protocol.CreatePacket('addr', payload) return protocol.CreatePacket('addr', payload)
def handle_connect_event(self):
try:
asyncore.dispatcher.handle_connect_event(self)
self.connectedAt = time.time()
except socket.error as e:
#print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e))
self.close()
def handle_read_event(self):
try:
asyncore.dispatcher.handle_read_event(self)
except socket.error as e:
#print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e))
self.close()
def handle_write_event(self):
try:
asyncore.dispatcher.handle_write_event(self)
except socket.error as e:
#print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e))
self.close()
def close(self, reason=None): def close(self, reason=None):
self.set_state("close") self.set_state("close")
# if reason is None: # if reason is None:

View File

@ -63,14 +63,8 @@ class TCPConnection(BMProto, TLSDispatcher):
shared.connectedHostsList[self.destination] = 0 shared.connectedHostsList[self.destination] = 0
ObjectTracker.__init__(self) ObjectTracker.__init__(self)
UISignalQueue.put(('updateNetworkStatusTab', 'no data')) UISignalQueue.put(('updateNetworkStatusTab', 'no data'))
def state_init(self):
self.bm_proto_reset() self.bm_proto_reset()
if self.isOutbound:
self.writeQueue.put(protocol.assembleVersionMessage(self.destination.host, self.destination.port, network.connectionpool.BMConnectionPool().streams, False))
print "%s:%i: Sending version" % (self.destination.host, self.destination.port)
self.set_state("bm_header") self.set_state("bm_header")
return True
def antiIntersectionDelay(self, initial = False): def antiIntersectionDelay(self, initial = False):
# estimated time for a small object to propagate across the whole network # estimated time for a small object to propagate across the whole network
@ -148,35 +142,28 @@ class TCPConnection(BMProto, TLSDispatcher):
def handle_connect_event(self): def handle_connect_event(self):
try: try:
asyncore.dispatcher.handle_connect_event(self) asyncore.dispatcher.handle_connect_event(self)
self.connectedAt = time.time()
except socket.error as e: except socket.error as e:
#print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e)) if e.errno in asyncore._DISCONNECTED:
self.close() self.close("Connection failed")
return
self.writeQueue.put(protocol.assembleVersionMessage(self.destination.host, self.destination.port, network.connectionpool.BMConnectionPool().streams, False))
#print "%s:%i: Sending version" % (self.destination.host, self.destination.port)
self.connectedAt = time.time()
def handle_read_event(self): def handle_read(self):
try: try:
asyncore.dispatcher.handle_read_event(self) AdvancedDispatcher.handle_read(self)
except socket.error as e: except socket.error as e:
#print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e)) #print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e))
self.close() self.close()
def handle_write_event(self): def handle_write(self):
try: try:
asyncore.dispatcher.handle_write_event(self) AdvancedDispatcher.handle_write(self)
except socket.error as e: except socket.error as e:
#print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e)) #print "%s:%i: socket error: %s" % (self.destination.host, self.destination.port, str(e))
self.close() self.close()
def close(self, reason=None):
self.set_state("close")
# if reason is None:
# print "%s:%i: closing" % (self.destination.host, self.destination.port)
# #traceback.print_stack()
# else:
# print "%s:%i: closing, %s" % (self.destination.host, self.destination.port, reason)
network.connectionpool.BMConnectionPool().removeConnection(self)
asyncore.dispatcher.close(self)
class Socks5BMConnection(Socks5Connection, TCPConnection): class Socks5BMConnection(Socks5Connection, TCPConnection):
def __init__(self, address): def __init__(self, address):

View File

@ -127,7 +127,7 @@ class UDPSocket(BMProto):
def bm_command_version(self): def bm_command_version(self):
return True return True
def handle_connect_event(self): def handle_connect(self):
return return
def writable(self): def writable(self):
@ -168,16 +168,6 @@ class UDPSocket(BMProto):
print "socket error on sendato: %s" % (e) print "socket error on sendato: %s" % (e)
self.writeQueue.task_done() self.writeQueue.task_done()
def close(self, reason=None):
self.set_state("close")
# if reason is None:
# print "%s:%i: closing" % (self.destination.host, self.destination.port)
# #traceback.print_stack()
# else:
# print "%s:%i: closing, %s" % (self.destination.host, self.destination.port, reason)
network.connectionpool.BMConnectionPool().removeConnection(self)
asyncore.dispatcher.close(self)
if __name__ == "__main__": if __name__ == "__main__":
# initial fill # initial fill