ref: da7d6df6faf18e289fe0f3f61524dcc7fddeef18
dir: /appl/lib/iobuf.b/
implement IOBuf;
include "sys.m";
sys: Sys;
sprint: import sys;
include "iobuf.m";
LF: array of byte;
init()
{
sys = load Sys Sys->PATH;
LF = array[] of { byte '\n' };
}
ReadBuf.new(fd: ref Sys->FD, bufsize: int): ref ReadBuf
{
r := ref ReadBuf;
r.buf = array[bufsize] of byte;
r.s = 0;
r.e = 0;
r.setsep("\n", 1);
r.fd = fd;
r.reader = sysread;
r.is_eof = 0;
return r;
}
ReadBuf.newc(queuesize, bufsize: int): ref ReadBuf
{
r := ReadBuf.new(nil, bufsize);
r.queue = chan[queuesize] of array of byte;
r.pending = chan[1] of (array of byte, Sys->Rwrite);
r.is_pending = chan[1] of int;
r.reader = chanread;
return r;
}
ReadBuf.setsep(r: self ref ReadBuf, sep: string, strip: int)
{
if(sep == nil)
raise "iobuf:empty separator";
r.sep = array of byte sep;
r.strip = strip;
}
ReadBuf.reads(r: self ref ReadBuf): array of byte
{
if(len r.sep != 1)
raise "iobuf:multibyte separator not implemented yet";
c := r.sep[0];
for(;;){
if(r.is_eof)
if(r.s == r.e)
return nil;
else{
s := r.s;
r.s = r.e;
return r.buf[s:r.e];
}
for(i := r.s; i < r.e; i++)
if(r.buf[i] == c){
s := r.s;
r.s = i+1;
return r.buf[s:i + 1 * !r.strip];
}
if(r.s != 0){
r.buf[0:] = r.buf[r.s:r.e];
r.e -= r.s;
r.s = 0;
}
if(r.e == len r.buf)
raise "iobuf:no separator found in full buffer";
if(r.reader(r) == 0)
r.is_eof = 1;
}
}
sysread(r: ref ReadBuf): int
{
n := sys->read(r.fd, r.buf[r.e:], len r.buf - r.e);
if(n < 0)
raise sprint("iobuf:%r");
r.e += n;
return n;
}
bufread(r: ref ReadBuf, buf: array of byte): int
{
n := len buf;
if(len r.buf - r.e < n)
n = len r.buf - r.e;
r.buf[r.e:] = buf[0:n];
r.e += n;
if(len buf > n)
r.leftover = buf[n:];
else
r.leftover = nil;
return n;
}
chanread(r: ref ReadBuf): int
{
if(r.leftover != nil)
return bufread(r, r.leftover);
alt{
buf := <-r.queue =>
if(buf == nil)
return 0;
else
return bufread(r, buf);
(buf, wc) := <-r.pending =>
n := len buf;
alt{
buf2 := <-r.queue =>
r.queue <-= buf;
buf = buf2;
* =>
;
}
<-r.is_pending;
if(wc != nil)
wc <-= (n, nil);
if(buf == nil)
return 0;
else
return bufread(r, buf);
}
}
ReadBuf.readn(r: self ref ReadBuf, n: int): array of byte
{
if(r.is_eof)
return nil;
if(r.e - r.s >= n){
s := r.s;
r.s += n;
return r.buf[s:r.s];
}
oldbuf : array of byte;
if(len r.buf >= n){
if(len r.buf - r.s < n){
r.buf[0:] = r.buf[r.s:r.e];
r.e -= r.s;
r.s = 0;
}
}
else{
oldbuf = r.buf;
r.buf = array[n] of byte;
r.buf[0:] = oldbuf[r.s:r.e];
r.e -= r.s;
r.s = 0;
}
while(r.e - r.s < n)
if(r.reader(r) == 0){
r.is_eof = 1;
n = r.e - r.s;
}
if(oldbuf == nil){
s := r.s;
r.s += n;
return r.buf[s:r.s];
}
else{
tmp := r.buf;
r.buf = oldbuf;
r.s = r.e = 0;
return tmp[:n];
}
}
ReadBuf.fill(r: self ref ReadBuf, data: array of byte, wc: Sys->Rwrite)
{
alt{
r.is_pending <-= 1 =>
<-r.is_pending;
alt{
r.queue <-= data =>
if(wc != nil)
wc <-= (len data, nil);
* =>
r.is_pending <-= 1;
r.pending <-= (data, wc);
}
* =>
if(wc != nil)
wc <-= (0, "concurrent writes not supported");
}
}
#
WriteBuf.new(fd: ref Sys->FD, bufsize: int): ref WriteBuf
{
w := ref WriteBuf;
w.buf = array[bufsize] of byte;
w.s = 0;
w.e = 0;
w.fd = fd;
w.writer = syswrite;
return w;
}
WriteBuf.newc(bufsize: int): ref WriteBuf
{
w := WriteBuf.new(nil, bufsize);
w.pending = chan[1] of (int, Sys->Rread);
w.writer = chanwrite;
return w;
}
WriteBuf.write(w: self ref WriteBuf, buf: array of byte)
{
n := 0;
if(w.e != 0){
n = len w.buf - w.e;
if(n > len buf)
n = len buf;
w.buf[w.e:] = buf[:n];
w.e += n;
if(len w.buf == w.e)
w.flush();
}
if(len buf > n){
n2 := int((len buf - n) / len w.buf) * len w.buf;
if(n2 > 0){
tmp := w.buf;
w.buf = buf[n:n + n2];
w.s = 0;
w.e = n2;
w.flush();
w.buf = tmp;
n += n2;
}
w.buf[0:] = buf[n:];
w.e = len buf - n;
}
if(w.fd == nil && w.s != w.e)
optchanwrite(w);
}
WriteBuf.writeln(w: self ref WriteBuf, buf: array of byte)
{
w.write(buf);
w.write(LF);
}
syswrite(w: ref WriteBuf)
{
n := sys->write(w.fd, w.buf[w.s:w.e], w.e - w.s);
if(n != w.e - w.s)
raise sprint("iobuf:%r");
w.s = 0;
w.e = 0;
}
chanwrite(w: ref WriteBuf)
{
(n, rc) := <-w.pending;
if(rc == nil)
raise "iobuf:broken pipe";
if(n > w.e - w.s)
n = w.e - w.s;
buf := array[n] of byte;
buf[0:] = w.buf[w.s:w.s + n];
rc <-= (buf, nil);
w.s += n;
}
optchanwrite(w: ref WriteBuf)
{
alt{
(n, rc) := <-w.pending =>
if(rc == nil)
raise "iobuf:broken pipe";
if(n > w.e - w.s)
n = w.e - w.s;
buf := array[n] of byte;
buf[0:] = w.buf[w.s:w.s + n];
rc <-= (buf, nil);
w.s += n;
* =>
;
}
}
WriteBuf.flush(w: self ref WriteBuf)
{
while(w.s != w.e)
w.writer(w);
w.s = w.e = 0;
}
WriteBuf.eof(w: self ref WriteBuf)
{
w.flush();
if(w.fd != nil)
return;
for(;;){
(nil, rc) := <-w.pending;
if(rc == nil)
break;
rc <-= (nil, nil);
}
}
WriteBuf.request(w: self ref WriteBuf, n: int, rc: Sys->Rread)
{
if(rc == nil)
alt{
<-w.pending => ;
* => ;
}
alt{
w.pending <-= (n, rc) =>;
* => rc <-= (nil, "concurrent reads not supported");
}
}