Commit ca14231a authored by Avery Pennarun's avatar Avery Pennarun

Whoops, we were spinning in the server when the mux outbuf was full.

Once again, the buffering gets mixed up with the selecting.  Seems to be the
story of my life.
parent 6c2dc54b
...@@ -35,12 +35,14 @@ def main(): ...@@ -35,12 +35,14 @@ def main():
handlers = filter(lambda s: s.ok, handlers) handlers = filter(lambda s: s.ok, handlers)
for s in handlers: for s in handlers:
s.pre_select(r,w,x) s.pre_select(r,w,x)
debug2('Waiting: %d[%d,%d,%d]...\n' debug2('Waiting: %d[%d,%d,%d] (fullness=%d/%d)...\n'
% (len(handlers), len(r), len(w), len(x))) % (len(handlers), len(r), len(w), len(x),
mux.fullness, mux.too_full))
(r,w,x) = select.select(r,w,x) (r,w,x) = select.select(r,w,x)
#log('r=%r w=%r x=%r\n' % (r,w,x)) #log('r=%r w=%r x=%r\n' % (r,w,x))
ready = set(r) | set(w) | set(x) ready = set(r) | set(w) | set(x)
for s in handlers: for s in handlers:
#debug2('check: %r: %r\n' % (s, s.socks & ready))
if s.socks & ready: if s.socks & ready:
s.callback() s.callback()
mux.check_fullness() mux.check_fullness()
......
...@@ -101,6 +101,9 @@ class SockWrapper: ...@@ -101,6 +101,9 @@ class SockWrapper:
except socket.error, e: except socket.error, e:
self.seterr(e) self.seterr(e)
def too_full(self):
return False # fullness is determined by the socket's select() state
def uwrite(self, buf): def uwrite(self, buf):
if self.connect_to: if self.connect_to:
return 0 # still connecting return 0 # still connecting
...@@ -181,14 +184,16 @@ class Proxy(Handler): ...@@ -181,14 +184,16 @@ class Proxy(Handler):
if self.wrap1.connect_to: if self.wrap1.connect_to:
w.add(self.wrap1.rsock) w.add(self.wrap1.rsock)
elif self.wrap1.buf: elif self.wrap1.buf:
w.add(self.wrap2.wsock) if not self.wrap2.too_full():
w.add(self.wrap2.wsock)
elif not self.wrap1.shut_read: elif not self.wrap1.shut_read:
r.add(self.wrap1.rsock) r.add(self.wrap1.rsock)
if self.wrap2.connect_to: if self.wrap2.connect_to:
w.add(self.wrap2.rsock) w.add(self.wrap2.rsock)
elif self.wrap2.buf: elif self.wrap2.buf:
w.add(self.wrap1.wsock) if not self.wrap1.too_full():
w.add(self.wrap1.wsock)
elif not self.wrap2.shut_read: elif not self.wrap2.shut_read:
r.add(self.wrap2.rsock) r.add(self.wrap2.rsock)
...@@ -232,7 +237,9 @@ class Mux(Handler): ...@@ -232,7 +237,9 @@ class Mux(Handler):
return sum(len(b) for b in self.outbuf) return sum(len(b) for b in self.outbuf)
def check_fullness(self): def check_fullness(self):
if self.fullness > 65536: if self.fullness > 32768:
if not self.too_full:
self.send(0, CMD_PING, 'rttest')
self.too_full = True self.too_full = True
#ob = [] #ob = []
#for b in self.outbuf: #for b in self.outbuf:
...@@ -247,8 +254,6 @@ class Mux(Handler): ...@@ -247,8 +254,6 @@ class Mux(Handler):
self.outbuf.append(p) self.outbuf.append(p)
debug2(' > channel=%d cmd=%s len=%d (fullness=%d)\n' debug2(' > channel=%d cmd=%s len=%d (fullness=%d)\n'
% (channel, cmd_to_name[cmd], len(data), self.fullness)) % (channel, cmd_to_name[cmd], len(data), self.fullness))
if self.fullness < 32768 and self.fullness+len(data) > 32768:
self.send(0, CMD_PING, 'rttest')
self.fullness += len(data) self.fullness += len(data)
def got_packet(self, channel, cmd, data): def got_packet(self, channel, cmd, data):
...@@ -274,7 +279,7 @@ class Mux(Handler): ...@@ -274,7 +279,7 @@ class Mux(Handler):
self.wsock.setblocking(False) self.wsock.setblocking(False)
if self.outbuf and self.outbuf[0]: if self.outbuf and self.outbuf[0]:
wrote = _nb_clean(os.write, self.wsock.fileno(), self.outbuf[0]) wrote = _nb_clean(os.write, self.wsock.fileno(), self.outbuf[0])
debug2('wrote: %d/%d\n' % (wrote, len(self.outbuf[0]))) debug2('mux wrote: %d/%d\n' % (wrote, len(self.outbuf[0])))
if wrote: if wrote:
self.outbuf[0] = self.outbuf[0][wrote:] self.outbuf[0] = self.outbuf[0][wrote:]
while self.outbuf and not self.outbuf[0]: while self.outbuf and not self.outbuf[0]:
...@@ -346,6 +351,9 @@ class MuxWrapper(SockWrapper): ...@@ -346,6 +351,9 @@ class MuxWrapper(SockWrapper):
self.shut_write = True self.shut_write = True
self.mux.send(self.channel, CMD_EOF, '') self.mux.send(self.channel, CMD_EOF, '')
def too_full(self):
return self.mux.too_full
def uwrite(self, buf): def uwrite(self, buf):
if self.mux.too_full: if self.mux.too_full:
return 0 # too much already enqueued return 0 # too much already enqueued
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment