Stdio.Buffer.set_buffer_mode revisited

Ok, below is the more or less minimal class that can be used
like a blocking connection that reads using Stdio.Buffer input
commands, and outputs using Stdio.Buffer commands followed by a
transmit() to actually send off the prepared output buffer.
The underlying filedescriptor is put in non-blocking mode to make
this work.  It works like a charm in the 8.0/8.1 pgsql driver, it just
looks a bit convoluted in the backtrace because of the recurring
blocks on the range_error() method.

Would it be easy and/or desirable to be able to get rid of possibly
the fillread/shortmux at the input side by incorporating this functionality
natively into Stdio.Buffer?

How about the shortmux & write callback at the output side?  I guess
one could use the Shuffler at that side; then again the Shuffler
currently does not support Stdio.Buffer input, so it would result in
an unnecessary copy of the data.

#define MAGICTERMINATE	42

class conxion {
  inherit Stdio.Buffer:i;
  inherit Stdio.Buffer:o;

  private Thread.Condition fillread;
  private Thread.Mutex shortmux;
  private Stdio.File socket;
  private int towrite;

  private function(:void) release_cb;

  protected bool range_error(int howmuch) {
    if(!howmuch)
      return false;
    if(fillread) {
      Thread.MutexKey lock=shortmux->lock();
      fillread.wait(lock);
      lock=0;
    } else
      throw(MAGICTERMINATE);
    return true;
  }

  private int read_cb(mixed id,mixed b) {
    Thread.MutexKey lock=shortmux->lock();
    if(fillread)
      fillread.signal();
    lock=0;
    return 0;
  }

  private int write_cb() {
    Thread.MutexKey lock=shortmux->lock();
    towrite-=output_to(socket,towrite);
    lock=0;
    if(!fillread && !sizeof(this))
      release_cb();
    return 0;
  }

  final inline    int consume(int w)     { return i::consume(w);     }
  final inline    int unread(int w)      { return i::unread(w);      }
  final inline string read(int w)        { return i::read(w);        }
  final inline object read_buffer(int w) { return i::read_buffer(w); }
  final inline    int read_sint(int w)   { return i::read_sint(w);   }
  final inline    int read_int8()        { return i::read_int8();    }
  final inline    int read_int16()       { return i::read_int16();   }
  final inline    int read_int32()       { return i::read_int32();   }
  final inline string read_cstring()     { return i::read_cstring(); }

  final void transmit() {
    Thread.MutexKey lock=shortmux->lock();
    if(towrite=sizeof(this))
      towrite-=output_to(socket,towrite);
    lock=0;
  }

  final void close() {
    destruct(fillread);  // Delayed close() after flushing the output buffer
  }

  protected void destroy() {
    catch(close());		// Exceptions don't work inside destructors
    Sql.pgsql_util.unregister_backend();
  }

  private string _sprintf(int type, void|mapping flags) {
    string res=UNDEFINED;
    switch(type) {
      case 'O':
        res=predef::sprintf("conxion  fd: %d input queue: %d/%d "
                    "queued portals: %d  output queue: %d/%d\n",
                    socket&&socket->query_fd(),
                    sizeof(i::this),i::_size_object(),
                    qportals->size(),sizeof(this),_size_object());
        break;
    }
    return res;
  }

  protected void create(Stdio.File _socket, function (:void) _release_cb) {
    i::create(); o::create();
    fillread=Thread.Condition();
    shortmux=Thread.Mutex();
    socket=_socket;
    release_cb=_release_cb;
    Sql.pgsql_util.register_backend();
    _socket->set_backend(Sql.pgsql.local_backend);
    _socket->set_buffer_mode(i::this,0);
    _socket->set_nonblocking(read_cb,write_cb,0);
  }
}
-- 
Stephen.