X-Git-Url: http://www.chiark.greenend.org.uk/ucgi/~ian/git?a=blobdiff_plain;f=server;h=a34cc3cf3082ab6047cf0a6463b8fde2a6528f2d;hb=2d2a54a45c59769081a906418963e5fad2f7a31e;hp=80bdd807202cdd8a20f09eb1a5a1cc431dd4202d;hpb=040ff5118442b3a69e2da76c0f4970b8743c3647;p=hippotat.git diff --git a/server b/server index 80bdd80..a34cc3c 100755 --- a/server +++ b/server @@ -1,31 +1,19 @@ #!/usr/bin/python3 -import signal -signal.signal(signal.SIGINT, signal.SIG_DFL) +from hippotat import * 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 twisted.web.server import Site #from twisted.web.resource import Resource -from optparse import OptionParser -from configparser import ConfigParser -from configparser import NoOptionError - -import collections - import syslog -from hippotat import * - clients = { } defcfg = ''' @@ -33,6 +21,8 @@ defcfg = ''' max_batch_down = 65536 max_queue_time = 10 max_request_time = 54 +target_requests_outstanding = 3 +ipif = userv root ipif %(local)s,%(peer)s,%(mtu)s,slip %(rnets)s [virtual] mtu = 1500 @@ -41,7 +31,6 @@ mtu = 1500 # [relay] [server] -ipif = userv root ipif %(host)s,%(relay)s,%(mtu)s,slip %(network)s addrs = 127.0.0.1 ::1 port = 8099 @@ -49,22 +38,9 @@ port = 8099 max_batch_down = 262144 max_queue_time = 121 max_request_time = 121 +target_requests_outstanding = 10 ''' -#---------- error handling ---------- - -def crash(err): - print('CRASH ', err, file=sys.stderr) - try: reactor.stop() - except twisted.internet.error.ReactorNotRunning: pass - -def crash_on_defer(defer): - defer.addErrback(lambda err: crash(err)) - -def crash_on_critical(event): - if event.get('log_level') >= LogLevel.critical: - crash(twisted.logger.formatEvent(event)) - #---------- "router" ---------- def route(packet, saddr, daddr): @@ -75,7 +51,7 @@ def route(packet, saddr, daddr): dclient.queue_outbound(packet) elif saddr.is_link_local or daddr.is_link_local: log_discard(packet, saddr, daddr, 'link-local') - elif daddr == host or daddr not in network: + elif daddr == c.host or daddr not in c.network: print('TRACE INBOUND ', saddr, daddr, packet) queue_inbound(packet) elif daddr == relay: @@ -97,15 +73,18 @@ 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): @@ -121,7 +100,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') @@ -138,15 +117,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: @@ -155,8 +127,8 @@ 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) @@ -174,6 +146,10 @@ 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): @@ -186,7 +162,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) @@ -217,66 +194,47 @@ def start_http(): #---------- config and setup ---------- def process_cfg(): - global network - global host - global relay - global ipif_command + process_cfg_common_always() - network = ipnetwork(cfg.get('virtual','network')) - if network.num_addresses < 3 + 2: + c.network = ipnetwork(cfg.get('virtual','network')) + if c.network.num_addresses < 3 + 2: raise ValueError('network needs at least 2^3 addresses') try: - host = cfg.get('virtual','host') + c.host = cfg.get('virtual','host') except NoOptionError: - host = next(network.hosts()) + c.host = next(c.network.hosts()) try: - relay = cfg.get('virtual','relay') + c.relay = cfg.get('virtual','relay') except NoOptionError: - for search in network.hosts(): - if search == host: continue - relay = search + for search in c.network.hosts(): + if search == c.host: continue + c.relay = search break for cs in cfg.sections(): if not (':' in cs or '.' in cs): continue ci = ipaddr(cs) - if ci not in network: + if ci not in c.network: raise ValueError('client %s not in network' % ci) if ci in clients: raise ValueError('multiple client cfg sections for %s' % ci) clients[ci] = Client(ci, cs) - global mtu - mtu = cfg.get('virtual','mtu') - - iic_vars = { } - for k in ('host','relay','mtu','network'): - iic_vars[k] = globals()[k] - - ipif_command = cfg.get('server','ipif', vars=iic_vars) - -def startup(): - global cfg - - op = OptionParser() - op.add_option('-c', '--config', dest='configfile', - default='/etc/hippottd/server.conf') - global opts - (opts, args) = op.parse_args() - if len(args): op.error('no non-option arguments please') - - twisted.logger.globalLogPublisher.addObserver(crash_on_critical) + for d, s in (('local', 'host'), + ('peer','relay'), + ('rnets','network')): + try: v = getattr(c, s) + except KeyError: pass + setattr(c, d, v) - cfg = ConfigParser() - cfg.read_string(defcfg) - cfg.read(opts.configfile) - process_cfg() + print(repr(c)) - start_ipif(ipif_command, route) - start_http() + c.ipif_command = cfg.get('server','ipif', vars=c.__dict__) -startup() -reactor.run() -print('CRASHED (end)', file=sys.stderr) +common_startup(defcfg) +process_cfg() +start_ipif(c.ipif_command, route) +start_http() +common_run()