X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?a=blobdiff_plain;f=server;h=db13f25954d01ff4ad24fe6fbd6bb4305e0d2995;hb=650a3251620a44d22f30a4242cea6fff88ccfba5;hp=3abebe96bfb69bdf5921dfa64cf1a8da2f70bbf0;hpb=8e279651cf76bb0e7d532307c0afbfbd7dc4986b;p=hippotat.git diff --git a/server b/server index 3abebe9..db13f25 100755 --- a/server +++ b/server @@ -6,16 +6,11 @@ signal.signal(signal.SIGINT, signal.SIG_DFL) import sys import os -import twisted import twisted.internet import twisted.internet.endpoints -from twisted.internet import reactor from twisted.web.server import NOT_DONE_YET from twisted.logger import LogLevel -import ipaddress -from ipaddress import AddressValueError - #import twisted.web.server import Site #from twisted.web.resource import Resource @@ -27,27 +22,16 @@ import collections import syslog -clients = { } - -def ipaddr(input): - try: - r = ipaddress.IPv4Address(input) - except AddressValueError: - r = ipaddress.IPv6Address(input) - return r +from hippotat import * -def ipnetwork(input): - try: - r = ipaddress.IPv4Network(input) - except NetworkValueError: - r = ipaddress.IPv6Network(input) - return r +clients = { } defcfg = ''' [DEFAULT] max_batch_down = 65536 max_queue_time = 10 max_request_time = 54 +target_requests_outstanding = 3 [virtual] mtu = 1500 @@ -64,6 +48,7 @@ port = 8099 max_batch_down = 262144 max_queue_time = 121 max_request_time = 121 +target_requests_outstanding = 10 ''' #---------- error handling ---------- @@ -103,89 +88,6 @@ def log_discard(packet, saddr, daddr, why): # syslog.syslog(syslog.LOG_DEBUG, # 'discarded packet %s -> %s (%s)' % (saddr, daddr, why)) -#---------- ipif (slip subprocess) ---------- - -class IpifProcessProtocol(twisted.internet.protocol.ProcessProtocol): - def __init__(self): - self._buffer = b'' - def connectionMade(self): pass - def outReceived(self, data): - #print('RECV ', repr(data)) - self._buffer += data - packets = slip_decode(self._buffer) - self._buffer = packets.pop() - for packet in packets: - if not len(packet): continue - (saddr, daddr) = packet_addrs(packet) - route(packet, saddr, daddr) - def processEnded(self, status): - status.raiseException() - -def start_ipif(): - global ipif - ipif = IpifProcessProtocol() - reactor.spawnProcess(ipif, - '/bin/sh',['sh','-xc', ipif_command], - childFDs={0:'w', 1:'r', 2:2}) - -def queue_inbound(packet): - ipif.transport.write(slip_delimiter) - ipif.transport.write(slip_encode(packet)) - ipif.transport.write(slip_delimiter) - -#---------- SLIP handling ---------- - -slip_end = b'\300' -slip_esc = b'\333' -slip_esc_end = b'\334' -slip_esc_esc = b'\335' -slip_delimiter = slip_end - -def slip_encode(packet): - return (packet - .replace(slip_esc, slip_esc + slip_esc_esc) - .replace(slip_end, slip_esc + slip_esc_end)) - -def slip_decode(data): - print('DECODE ', repr(data)) - out = [] - for packet in data.split(slip_end): - pdata = b'' - while True: - eix = packet.find(slip_esc) - if eix == -1: - pdata += packet - break - #print('ESC ', repr((pdata, packet, eix))) - pdata += packet[0 : eix] - ck = packet[eix+1] - #print('ESC... %o' % ck) - if ck == slip_esc_esc[0]: pdata += slip_esc - elif ck == slip_esc_end[0]: pdata += slip_end - else: raise ValueError('invalid SLIP escape') - packet = packet[eix+2 : ] - out.append(pdata) - print('DECODED ', repr(out)) - return out - -#---------- packet parsing ---------- - -def packet_addrs(packet): - version = packet[0] >> 4 - if version == 4: - addrlen = 4 - saddroff = 3*4 - factory = ipaddress.IPv4Address - elif version == 6: - addrlen = 16 - saddroff = 2*4 - factory = ipaddress.IPv6Address - else: - raise ValueError('unsupported IP version %d' % version) - saddr = factory(packet[ saddroff : saddroff + addrlen ]) - daddr = factory(packet[ saddroff + addrlen : saddroff + addrlen*2 ]) - return (saddr, daddr) - #---------- client ---------- class Client(): @@ -195,18 +97,21 @@ class Client(): self._cs = cs self.pw = cfg.get(cs, 'password') self._rq = collections.deque() # requests - self._pq = collections.deque() # packets + # self._pq = PacketQueue(...) # plus from config: # .max_batch_down # .max_queue_time # .max_request_time - for k in ('max_batch_down','max_queue_time','max_request_time'): + # .target_requests_outstanding + for k in ('max_batch_down','max_queue_time','max_request_time', + 'target_requests_outstanding'): req = cfg.getint(cs, k) limit = cfg.getint('limits',k) self.__dict__[k] = min(req, limit) + self._pq = PacketQueue(self.max_queue_time) def process_arriving_data(self, d): - for packet in slip_decode(d): + for packet in slip.decode(d): (saddr, daddr) = packet_addrs(packet) if saddr != self._ip: raise ValueError('wrong source address %s' % saddr) @@ -219,7 +124,7 @@ class Client(): self._req_cancel(request) def queue_outbound(self, packet): - self._pq.append((time.monotonic(), packet)) + self._pq.append(packet) def http_request(self, request): request.setHeader('Content-Type','application/octet-stream') @@ -236,15 +141,8 @@ class Client(): self._rq.popleft() continue - # now request is an unfinished request, or None - try: (queuetime, packet) = self._pq[0] - except IndexError: + if not self._pq.nonempty(): # no packets, oh well - break - - age = time.monotonic() - queuetime - if age > self.max_queue_time: - self._pq.popleft() continue if request is None: @@ -253,16 +151,16 @@ class Client(): # request, and also some non-expired packets while True: - try: (dummy, packet) = self._pq[0] - except IndexError: break + packet = self.pq.popleft() + if packet is None: break - encoded = slip_encode(packet) + encoded = slip.encode(packet) if request.sentLength > 0: - if (request.sentLength + len(slip_delimiter) + if (request.sentLength + len(slip.delimiter) + len(encoded) > self.max_batch_down): break - request.write(slip_delimiter) + request.write(slip.delimiter) request.write(encoded) self._pq.popLeft() @@ -272,7 +170,12 @@ class Client(): request.finish() # round again, looking for more to do + while len(self._rq) > self.target_requests_outstanding: + request = self._rq.popleft() + request.finish() + class IphttpResource(twisted.web.resource.Resource): + isLeaf = True def render_POST(self, request): # find client, update config, etc. ci = ipaddr(request.args['i']) @@ -283,7 +186,8 @@ class IphttpResource(twisted.web.resource.Resource): # update config for r, w in (('mbd', 'max_batch_down'), ('mqt', 'max_queue_time'), - ('mrt', 'max_request_time')): + ('mrt', 'max_request_time'), + ('tro', 'target_requests_outstanding')): try: v = request.args[r] except KeyError: continue v = int(v) @@ -296,11 +200,11 @@ class IphttpResource(twisted.web.resource.Resource): c.new_request(request) def render_GET(self, request): - return 'hippotit' + return b'hippotat' def start_http(): resource = IphttpResource() - sitefactory = twisted.web.server.Site(resource) + site = twisted.web.server.Site(resource) for addrspec in cfg.get('server','addrs').split(): try: addr = ipaddress.IPv4Address(addrspec) @@ -309,7 +213,7 @@ def start_http(): addr = ipaddress.IPv6Address(addrspec) endpointfactory = twisted.internet.endpoints.TCP6ServerEndpoint ep = endpointfactory(reactor, cfg.getint('server','port'), addr) - crash_on_defer(ep.listen(sitefactory)) + crash_on_defer(ep.listen(site)) #---------- config and setup ---------- @@ -371,7 +275,7 @@ def startup(): cfg.read(opts.configfile) process_cfg() - start_ipif() + start_ipif(ipif_command, route) start_http() startup()