vbatts/shadowsocks
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

optimize performance for multiple ports

Browse source 
UDPRelay is broken now
This commit is contained in:
clowwindy 2015-08-01 18:22:35 +08:00
parent d95e5ce680
commit 4a8d0774b4
7 changed files with 2171 additions and 168 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
setup.py
View file

@ -7,7 +7,7 @@ with codecs.open('README.rst', encoding='utf-8') as f:
setup(
name="shadowsocks",
version="2.6.12",
version="2.7",
license='http://www.apache.org/licenses/LICENSE-2.0',
description="A fast tunnel proxy that help you get through firewalls",
author='clowwindy',

28
shadowsocks/asyncdns.py
View file

@ -256,7 +256,6 @@ class DNSResolver(object):
self._hostname_to_cb = {}
self._cb_to_hostname = {}
self._cache = lru_cache.LRUCache(timeout=300)
self._last_time = time.time()
self._sock = None
self._servers = None
self._parse_resolv()
@ -304,7 +303,7 @@ class DNSResolver(object):
except IOError:
self._hosts['localhost'] = '127.0.0.1'
def add_to_loop(self, loop, ref=False):
def add_to_loop(self, loop):
if self._loop:
raise Exception('already add to loop')
self._loop = loop
@ -312,8 +311,8 @@ class DNSResolver(object):
self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM,
socket.SOL_UDP)
self._sock.setblocking(False)
loop.add(self._sock, eventloop.POLL_IN)
loop.add_handler(self.handle_events, ref=ref)
loop.add(self._sock, eventloop.POLL_IN, self)
loop.add_periodic(self.handle_periodic)
def _call_callback(self, hostname, ip, error=None):
callbacks = self._hostname_to_cb.get(hostname, [])
@ -354,30 +353,27 @@ class DNSResolver(object):
self._call_callback(hostname, None)
break
def handle_events(self, events):
for sock, fd, event in events:
def handle_event(self, sock, fd, event):
if sock != self._sock:
continue
return
if event & eventloop.POLL_ERR:
logging.error('dns socket err')
self._loop.remove(self._sock)
self._loop.remove(self._sock, self)
self._sock.close()
# TODO when dns server is IPv6
self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM,
socket.SOL_UDP)
self._sock.setblocking(False)
self._loop.add(self._sock, eventloop.POLL_IN)
self._loop.add(self._sock, eventloop.POLL_IN, self)
else:
data, addr = sock.recvfrom(1024)
if addr[0] not in self._servers:
logging.warn('received a packet other than our dns')
break
return
self._handle_data(data)
break
now = time.time()
if now - self._last_time > CACHE_SWEEP_INTERVAL:
def handle_periodic(self):
self._cache.sweep()
self._last_time = now
def remove_callback(self, callback):
hostname = self._cb_to_hostname.get(callback)
@ -385,7 +381,7 @@ class DNSResolver(object):
del self._cb_to_hostname[callback]
arr = self._hostname_to_cb.get(hostname, None)
if arr:
arr.remove(callback)
arr.remove(callback, self)
if not arr:
del self._hostname_to_cb[hostname]
if hostname in self._hostname_status:
@ -430,6 +426,7 @@ class DNSResolver(object):
def close(self):
if self._sock:
self._loop.remove(self._sock, self)
self._sock.close()
self._sock = None
@ -451,7 +448,6 @@ def test():
print(result, error)
counter += 1
if counter == 9:
loop.remove_handler(dns_resolver.handle_events)
dns_resolver.close()
a_callback = callback
return a_callback

111
shadowsocks/eventloop.py
View file

@ -22,6 +22,7 @@ from __future__ import absolute_import, division, print_function, \
with_statement
import os
import time
import socket
import select
import errno
@ -51,23 +52,8 @@ EVENT_NAMES = {
POLL_NVAL: 'POLL_NVAL',
}
class EpollLoop(object):
def __init__(self):
self._epoll = select.epoll()
def poll(self, timeout):
return self._epoll.poll(timeout)
def add_fd(self, fd, mode):
self._epoll.register(fd, mode)
def remove_fd(self, fd):
self._epoll.unregister(fd)
def modify_fd(self, fd, mode):
self._epoll.modify(fd, mode)
# we check timeouts every TIMEOUT_PRECISION seconds
TIMEOUT_PRECISION = 10
class KqueueLoop(object):
@ -100,17 +86,17 @@ class KqueueLoop(object):
results[fd] |= POLL_OUT
return results.items()
def add_fd(self, fd, mode):
def register(self, fd, mode):
self._fds[fd] = mode
self._control(fd, mode, select.KQ_EV_ADD)
def remove_fd(self, fd):
def unregister(self, fd):
self._control(fd, self._fds[fd], select.KQ_EV_DELETE)
del self._fds[fd]
def modify_fd(self, fd, mode):
self.remove_fd(fd)
self.add_fd(fd, mode)
def modify(self, fd, mode):
self.unregister(fd)
self.register(fd, mode)
class SelectLoop(object):
@ -129,7 +115,7 @@ class SelectLoop(object):
results[fd] |= p[1]
return results.items()
def add_fd(self, fd, mode):
def register(self, fd, mode):
if mode & POLL_IN:
self._r_list.add(fd)
if mode & POLL_OUT:
@ -137,7 +123,7 @@ class SelectLoop(object):
if mode & POLL_ERR:
self._x_list.add(fd)
def remove_fd(self, fd):
def unregister(self, fd):
if fd in self._r_list:
self._r_list.remove(fd)
if fd in self._w_list:
@ -145,16 +131,15 @@ class SelectLoop(object):
if fd in self._x_list:
self._x_list.remove(fd)
def modify_fd(self, fd, mode):
self.remove_fd(fd)
self.add_fd(fd, mode)
def modify(self, fd, mode):
self.unregister(fd)
self.register(fd, mode)
class EventLoop(object):
def __init__(self):
self._iterating = False
if hasattr(select, 'epoll'):
self._impl = EpollLoop()
self._impl = select.epoll()
model = 'epoll'
elif hasattr(select, 'kqueue'):
self._impl = KqueueLoop()
@ -166,48 +151,50 @@ class EventLoop(object):
raise Exception('can not find any available functions in select '
'package')
self._fd_to_f = {}
self._handlers = []
self._ref_handlers = []
self._handlers_to_remove = []
self._fd_to_handler = {}
self._last_time = time.time()
self._periodic_callbacks = []
self._stopping = False
logging.debug('using event model: %s', model)
def poll(self, timeout=None):
events = self._impl.poll(timeout)
return [(self._fd_to_f[fd], fd, event) for fd, event in events]
def add(self, f, mode):
def add(self, f, mode, handler):
fd = f.fileno()
self._fd_to_f[fd] = f
self._impl.add_fd(fd, mode)
self._impl.register(fd, mode)
self._fd_to_handler[fd] = handler
def remove(self, f):
def remove(self, f, handler):
fd = f.fileno()
del self._fd_to_f[fd]
self._impl.remove_fd(fd)
self._impl.unregister(fd)
if handler is not None:
del self._fd_to_handler[fd]
def modify(self, f, mode):
def add_periodic(self, callback):
self._periodic_callbacks.append(callback)
def remove_periodic(self, callback):
self._periodic_callbacks.remove(callback)
def modify(self, f, mode, handler):
fd = f.fileno()
self._impl.modify_fd(fd, mode)
self._impl.modify(fd, mode)
if handler is not None:
self._fd_to_handler[fd] = handler
def add_handler(self, handler, ref=True):
self._handlers.append(handler)
if ref:
# when all ref handlers are removed, loop stops
self._ref_handlers.append(handler)
def remove_handler(self, handler):
if handler in self._ref_handlers:
self._ref_handlers.remove(handler)
if self._iterating:
self._handlers_to_remove.append(handler)
else:
self._handlers.remove(handler)
def stop(self):
self._stopping = True
def run(self):
events = []
while self._ref_handlers:
while not self._stopping:
now = time.time()
try:
events = self.poll(1)
events = self.poll(TIMEOUT_PRECISION)
except (OSError, IOError) as e:
if errno_from_exception(e) in (errno.EPIPE, errno.EINTR):
# EPIPE: Happens when the client closes the connection
@ -219,18 +206,18 @@ class EventLoop(object):
import traceback
traceback.print_exc()
continue
self._iterating = True
for handler in self._handlers:
# TODO when there are a lot of handlers
for sock, fd, event in events:
handler = self._fd_to_handler.get(fd, None)
if handler is not None:
try:
handler(events)
handler.handle_event(sock, fd, event)
except (OSError, IOError) as e:
shell.print_exception(e)
if self._handlers_to_remove:
for handler in self._handlers_to_remove:
self._handlers.remove(handler)
self._handlers_to_remove = []
self._iterating = False
if now - self._last_time >= TIMEOUT_PRECISION:
for callback in self._periodic_callbacks:
callback()
self._last_time = now
# from tornado

41
shadowsocks/tcprelay.py
View file

@ -32,9 +32,6 @@ from shadowsocks.common import parse_header
# we clear at most TIMEOUTS_CLEAN_SIZE timeouts each time
TIMEOUTS_CLEAN_SIZE = 512
# we check timeouts every TIMEOUT_PRECISION seconds
TIMEOUT_PRECISION = 4
MSG_FASTOPEN = 0x20000000
# SOCKS command definition
@ -126,7 +123,8 @@ class TCPRelayHandler(object):
fd_to_handlers[local_sock.fileno()] = self
local_sock.setblocking(False)
local_sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
loop.add(local_sock, eventloop.POLL_IN | eventloop.POLL_ERR)
loop.add(local_sock, eventloop.POLL_IN | eventloop.POLL_ERR,
self._server)
self.last_activity = 0
self._update_activity()
@ -175,14 +173,14 @@ class TCPRelayHandler(object):
event |= eventloop.POLL_OUT
if self._upstream_status & WAIT_STATUS_READING:
event |= eventloop.POLL_IN
self._loop.modify(self._local_sock, event)
self._loop.modify(self._local_sock, event, self._server)
if self._remote_sock:
event = eventloop.POLL_ERR
if self._downstream_status & WAIT_STATUS_READING:
event |= eventloop.POLL_IN
if self._upstream_status & WAIT_STATUS_WRITING:
event |= eventloop.POLL_OUT
self._loop.modify(self._remote_sock, event)
self._loop.modify(self._remote_sock, event, self._server)
def _write_to_sock(self, data, sock):
# write data to sock
@ -238,7 +236,7 @@ class TCPRelayHandler(object):
remote_sock = \
self._create_remote_socket(self._chosen_server[0],
self._chosen_server[1])
self._loop.add(remote_sock, eventloop.POLL_ERR)
self._loop.add(remote_sock, eventloop.POLL_ERR, self._server)
data = b''.join(self._data_to_write_to_remote)
l = len(data)
s = remote_sock.sendto(data, MSG_FASTOPEN, self._chosen_server)
@ -375,7 +373,8 @@ class TCPRelayHandler(object):
errno.EINPROGRESS:
pass
self._loop.add(remote_sock,
eventloop.POLL_ERR | eventloop.POLL_OUT)
eventloop.POLL_ERR | eventloop.POLL_OUT,
self._server)
self._stage = STAGE_CONNECTING
self._update_stream(STREAM_UP, WAIT_STATUS_READWRITING)
self._update_stream(STREAM_DOWN, WAIT_STATUS_READING)
@ -535,13 +534,13 @@ class TCPRelayHandler(object):
logging.debug('destroy')
if self._remote_sock:
logging.debug('destroying remote')
self._loop.remove(self._remote_sock)
self._loop.remove(self._remote_sock, self._server)
del self._fd_to_handlers[self._remote_sock.fileno()]
self._remote_sock.close()
self._remote_sock = None
if self._local_sock:
logging.debug('destroying local')
self._loop.remove(self._local_sock)
self._loop.remove(self._local_sock, self._server)
del self._fd_to_handlers[self._local_sock.fileno()]
self._local_sock.close()
self._local_sock = None
@ -557,7 +556,6 @@ class TCPRelay(object):
self._closed = False
self._eventloop = None
self._fd_to_handlers = {}
self._last_time = time.time()
self._timeout = config['timeout']
self._timeouts = [] # a list for all the handlers
@ -598,10 +596,9 @@ class TCPRelay(object):
if self._closed:
raise Exception('already closed')
self._eventloop = loop
loop.add_handler(self._handle_events)
self._eventloop.add(self._server_socket,
eventloop.POLL_IN | eventloop.POLL_ERR)
eventloop.POLL_IN | eventloop.POLL_ERR, self)
self._eventloop.add_periodic(self.handle_periodic)
def remove_handler(self, handler):
index = self._handler_to_timeouts.get(hash(handler), -1)
@ -613,7 +610,7 @@ class TCPRelay(object):
def update_activity(self, handler):
# set handler to active
now = int(time.time())
if now - handler.last_activity < TIMEOUT_PRECISION:
if now - handler.last_activity < eventloop.TIMEOUT_PRECISION:
# thus we can lower timeout modification frequency
return
handler.last_activity = now
@ -659,9 +656,8 @@ class TCPRelay(object):
pos = 0
self._timeout_offset = pos
def _handle_events(self, events):
def handle_event(self, sock, fd, event):
# handle events and dispatch to handlers
for sock, fd, event in events:
if sock:
logging.log(shell.VERBOSE_LEVEL, 'fd %d %s', fd,
eventloop.EVENT_NAMES.get(event, event))
@ -679,7 +675,7 @@ class TCPRelay(object):
error_no = eventloop.errno_from_exception(e)
if error_no in (errno.EAGAIN, errno.EINPROGRESS,
errno.EWOULDBLOCK):
continue
return
else:
shell.print_exception(e)
if self._config['verbose']:
@ -692,18 +688,17 @@ class TCPRelay(object):
else:
logging.warn('poll removed fd')
now = time.time()
if now - self._last_time > TIMEOUT_PRECISION:
def handle_periodic(self):
self._sweep_timeout()
self._last_time = now
if self._closed:
if self._server_socket:
self._eventloop.remove(self._server_socket)
self._eventloop.remove(self._server_socket, self)
self._eventloop.remove_periodic(self.handle_periodic)
self._server_socket.close()
self._server_socket = None
logging.info('closed listen port %d', self._listen_port)
if not self._fd_to_handlers:
self._eventloop.remove_handler(self._handle_events)
self._eventloop.stop()
def close(self, next_tick=False):
self._closed = True

20
shadowsocks/udprelay.py
View file

@ -106,7 +106,6 @@ class UDPRelay(object):
self._dns_cache = lru_cache.LRUCache(timeout=300)
self._eventloop = None
self._closed = False
self._last_time = time.time()
self._sockets = set()
if 'forbidden_ip' in config:
self._forbidden_iplist = config['forbidden_ip']
@ -137,7 +136,7 @@ class UDPRelay(object):
def _close_client(self, client):
if hasattr(client, 'close'):
self._sockets.remove(client.fileno())
self._eventloop.remove(client)
self._eventloop.remove(client, self)
client.close()
else:
# just an address
@ -199,7 +198,7 @@ class UDPRelay(object):
self._client_fd_to_server_addr[client.fileno()] = r_addr
self._sockets.add(client.fileno())
self._eventloop.add(client, eventloop.POLL_IN)
self._eventloop.add(client, eventloop.POLL_IN, self)
if self._is_local:
data = encrypt.encrypt_all(self._password, self._method, 1, data)
@ -257,14 +256,13 @@ class UDPRelay(object):
if self._closed:
raise Exception('already closed')
self._eventloop = loop
loop.add_handler(self._handle_events)
server_socket = self._server_socket
self._eventloop.add(server_socket,
eventloop.POLL_IN | eventloop.POLL_ERR)
eventloop.POLL_IN | eventloop.POLL_ERR, self)
loop.add_periodic(self.handle_periodic)
def _handle_events(self, events):
for sock, fd, event in events:
def handle_event(self, sock, fd, event):
if sock == self._server_socket:
if event & eventloop.POLL_ERR:
logging.error('UDP server_socket err')
@ -273,18 +271,18 @@ class UDPRelay(object):
if event & eventloop.POLL_ERR:
logging.error('UDP client_socket err')
self._handle_client(sock)
now = time.time()
if now - self._last_time > 3:
def handle_periodic(self):
self._cache.sweep()
self._client_fd_to_server_addr.sweep()
self._last_time = now
if self._closed:
self._server_socket.close()
for sock in self._sockets:
sock.close()
self._eventloop.remove_handler(self._handle_events)
self._eventloop.remove_periodic(self.handle_periodic)
def close(self, next_tick=False):
self._closed = True
if not next_tick:
self._eventloop.remove(self._server_socket, self)
self._server_socket.close()

19
tests/gen_multiple_passwd.py Normal file
View file

@ -0,0 +1,19 @@
#!/usr/bin/python
import os
import json
with open('server-multi-passwd-performance.json', 'wb') as f:
r = {
'server': '127.0.0.1',
'local_port': 1081,
'timeout': 60,
'method': 'aes-256-cfb'
}
ports = {}
for i in range(7000, 9000):
ports[str(i)] = 'aes_password'
r['port_password'] = ports
print(r)
f.write(json.dumps(r, indent=4).encode('utf-8'))

2008
tests/server-multi-passwd-performance.json Normal file
View file

File diff suppressed because it is too large Load diff