Minimal working example (Re: set_buffer_mode)

Well, in trying to understand how to use it (in order to repair a *very*
obscure race condition in the pgsql driver), I find the following issues:

- There is a bug in the Stdio.Buffer implementation which prevents it
  from properly fetching new input (patch attached).
- After fixing that, I find that the smallest example which shows
  the functional equivalent I'm trying to insert into the pgsql driver is
  something like the following below.
- IMO the example is way too complicated to be considered straightforward,
  so maybe I'm missing something, but reducing it further seems impossible
  without sacrificing some functionality.
- The main() basically is intended to startup a new thread for every
  connection to the database, whereas the processpackets function is
  supposed to be able to perform blocking reads on the (network) connection
  using Stdio.Buffer directives which consume varying amounts of the input.
  I.e. in the example I use read(4), but it could easily be any sequence
  of read_hstring(), read_int32() etc.
- The artificial i%8 check is to simulate buffering output in order
  to send it as a single packet, instead of several small ones.
  The set_buffer_mode() setting for output packets is greedy and seems
  unsuitable for this purpose.

Any suggestions on how this could be simplified further are welcome.
-----------------------sample code-------------------
#!/usr/local/bin/pike

#define MAGICTERMINATE	42

class autofill
{ inherit Stdio.Buffer;
  private final Thread.Condition _cond;
  private final Thread.Mutex _mux;

  final void create(Thread.Mutex mux,Thread.Condition cond)
  { ::create(); _cond=cond; _mux=mux;
  }

  protected final bool range_error(int howmuch)
  { Thread.MutexKey lock=_mux->lock();
    if(_cond)
      _cond.wait(lock);
    else
      throw(MAGICTERMINATE);
    return true;
  }
};

void tim(object condition)
{ werror("Timeout\n");
  destruct(condition);
}

void processpackets(Stdio.Buffer inb,Stdio.Buffer outb,Stdio.File out)
{ werror("Processing ... \n");
  int i=0;
  mixed e=catch
  { for(;;i++)
    { string s=inb->read(4);
      werror("%O processed\n",s);
      outb->add(s);
      if(!(i%8))
        werror("Direct Output %O\n",outb->output_to(out));
    }
  };
  if(e!=MAGICTERMINATE)
    throw(e);
  werror("Terminating thread\n");
  out->close();
}

int read_cb(mixed id)
{ Thread.MutexKey lock=id[0]->lock();
  if(id[1])
    id[1].signal();
  return 0;
}

int write_cb(mixed id)
{ id[0]->output_to(id[1]);
  return 0;
}

int main(int argc, array(string) argv)
{ Stdio.Buffer o=Stdio.Buffer();
  Thread.Condition condition=Thread.Condition();
  Thread.Mutex mux=Thread.Mutex();

  autofill i = autofill(mux,condition);
  object in = Stdio.File(0,"r");
  object out = Stdio.File(1,"w");

  out->set_id(({o,out}));
  out->set_nonblocking(0,write_cb,0);

  in->set_id(({mux,condition}));
  in->set_buffer_mode(i,0);
  in->set_nonblocking(read_cb,0,0);
  Thread.Thread(processpackets,i,o,out);
  
  call_out(tim,10,condition);

  return -1;
}
-----------------------sample code-------------------

-------------------patch----------------------
commit e62d7678f282cbe63fe69a5a82484a471471e654
Author: Stephen R. van den Berg <[email protected]>
Date:   Thu Oct 9 11:35:40 2014 +0200

    Stdio.Buffer: interpret 0 differently for once==1.

diff --git a/src/modules/_Stdio/buffer.cmod b/src/modules/_Stdio/buffer.cmod
index feecf01..74067cd 100644
--- a/src/modules/_Stdio/buffer.cmod
+++ b/src/modules/_Stdio/buffer.cmod
@@ -915,14 +915,14 @@ PIKECLASS Buffer
     struct my_file *fd;
     int once = 0;
 
-    if( _nbytes && TYPEOF(*_nbytes) == PIKE_T_INT) {
-        nbytes = _nbytes->u.integer;
-       if (!nbytes) RETURN 0;
-    }
-
     if( _once )
         once = _once->u.integer;
 
+    if( _nbytes && (!once || _nbytes->u.integer)) {
+        nbytes = _nbytes->u.integer;
+        if (!nbytes) RETURN 0;
+    }
+
 
     if( (fd = get_storage( f, file_program )) )
     {
-------------------patch----------------------
-- 
Stephen.