#
# Translation from styx to op for ofs.
# Either use the Cache or Op to operate on remote files.
#
# semantics:
# dir (+initial data) fetched/checked upon walk
# directories fully read.
# dir hierarchies leading to files/dirs are created (faked) on demand.
# files read on demand (after data read on first get)
# creates deferred until first write, but for "mkdirs".
# writethrough:
# data pushed for write at 0; and partially filled puts
# delayed write:
# data pushed with full puts.
#
# All files with open fids are alive in the cache.
# Op's fds correspond to cached files. This means that a clunk is
# likely to clunk an op fd. Read and Write modes would use different fds.
#
# A coherency lag of N ms can be set. In this case,
# files checked out not more than Nms ago are considered
# up to date. For directories, this is tricky, because we must
# still read directories for which we got just metadata, and not data.
# in particular, directories invented during a walk.
#
# Each close in a file being written signals an end-of-put (clunk)
# in the server. However, this does not make any distinction between different
# client processes updating the same file in the server (which is reasonable, because
# in that case there would be races in any case).
implement Stop;
include "sys.m";
sys: Sys;
fprint, sprint, pctl, millisec, DMDIR, write, nulldir, tokenize, fildes,
QTDIR, FD, Dir, Qid: import sys;
include "names.m";
names: Names;
cleanname: import names;
include "string.m";
str: String;
splitstrl: import str;
include "styx.m";
styx: Styx;
unpackdir: import styx;
include "styxservers.m";
Enotfound, Eexists: import Styxservers;
include "op.m";
OSTAT, NOFD, MAXDATA, OCREATE, ODATA, OMORE, Tmsg, Rmsg: import Op;
include "opmux.m";
opmux: Opmux;
include "error.m";
err: Error;
stderr, panic: import err;
include "stop.m";
include "ofstree.m";
ofstree: Ofstree;
Cfile: import ofstree;
# To put a limit on the number of processes that we might spawn
# mainly because of Tputs for huge files.
Nprocs: con 30;
Xop: adt {
fun: ref fn(x: Xop);
tag: int;
f: ref Cfile;
path: string;
d: ref Dir;
data: array of byte;
n: int;
o: big;
rc: chan of ref Crep;
};
cfsreqc: chan of (ref Creq, chan of ref Crep);
xcfsreqc: chan of Xop;
xcfsendc: chan of int;
donec: chan of ref Cfile;
coherencylag := 0;
maxwaiting := 0;
maxheld := 0;
init(s: Styx, m: Opmux, dir: string, lag: int): string
{
sys = load Sys Sys->PATH;
err = load Error Error->PATH;
err->init();
names = load Names Names->PATH;
str = load String String->PATH;
opmux = m;
ofstree = load Ofstree Ofstree->PATH;
if (sys == nil || names == nil || str == nil || opmux == nil || ofstree == nil)
return "load failed in cache";
styx = s;
coherencylag = lag;
cfsreqc = chan of (ref Creq, chan of ref Crep);
xcfsreqc = chan[5] of Xop;
xcfsendc= chan of int;
donec = chan[5] of ref Cfile;
e := ofstree->init(sys, str, styx, err, names, dir);
ofstree->debug = debug;
if (e != nil)
return e;
spawn cfsproc();
spawn xcfsproc();
return nil;
}
term()
{
cfsreqc <-= (nil, nil);
}
Creq.text(r: self ref Creq): string
{
s := sprint("t%d q%bx", r.tag, r.qid);
pick rr := r {
Dump =>
s += " dump";
Remove =>
s += " remove";
Stat =>
s += " stat";
Sync =>
s += " sync";
Validate =>
s += " validate [" + rr.path + "]";
Walk1 =>
s += " walk1 [" + rr.name + "]";
Readdir =>
s += sprint(" readdir c%d o%bd", rr.cnt, rr.off);
Pread =>
s += sprint(" pread c%d o%bd", rr.cnt, rr.off);
Pwrite =>
s += sprint(" pwrite [%3.3s...] c%d o%bd", string rr.data, len rr.data, rr.off);
Wstat =>
s += " wstat";
Create =>
s += " create [" + rr.d.name + "]";
Flush =>
s += sprint(" flush %d", rr.oldtag);
}
return s;
}
Crep.text(r: self ref Crep): string
{
s := "";
pick rr := r {
Remove =>
s += " remove";
Stat =>
s += " stat";
if (rr.d != nil)
s += sprint(" [%s] q%bx:%d", rr.d.name, rr.d.qid.path, rr.d.qid.vers);
Sync =>
s += " sync";
Validate =>
s += " validate";
if (rr.d != nil)
s += sprint(" [%s] q%bx:%d", rr.d.name, rr.d.qid.path, rr.d.qid.vers);
Walk1 =>
s += " walk1";
if (rr.d != nil)
s += sprint(" [%s] q%bx:%d", rr.d.name, rr.d.qid.path, rr.d.qid.vers);
Readdir =>
s += sprint(" readdir n%d", len rr.sons);
Pread =>
s += sprint(" pread [%3.3s...] c%d", string rr.data, len rr.data);
Pwrite =>
s += sprint(" pwrite c%d", rr.count);
Wstat =>
s += " wstat";
if (rr.d != nil)
s += sprint(" [%s] q%bx:%d", rr.d.name, rr.d.qid.path, rr.d.qid.vers);
Create =>
s += " create";
if (rr.d != nil)
s += sprint(" [%s] q%bx:%d", rr.d.name, rr.d.qid.path, rr.d.qid.vers);
Flush =>
s += sprint(" flush");
}
if (r.err != nil)
s += " errstr=" + r.err;
return s;
}
crepcs: list of chan of ref Crep;
getcrepc(): chan of ref Crep
{
if (crepcs == nil)
return chan of ref Crep;
else {
c := hd crepcs;
crepcs = tl crepcs;
return c;
}
}
putcrepc(c: chan of ref Crep)
{
crepcs = c::crepcs;
}
cacherpc(r: ref Creq) : ref Crep
{
c := getcrepc();
if (debug)
fprint(stderr, "<c %d\t%s\n", millisec(), r.text());
cfsreqc <-= (r, c);
rep := <-c;
putcrepc(c);
if (debug)
fprint(stderr, "c> %d\tt%d %s\n", millisec(), r.tag, rep.text());
return rep;
}
dump()
{
cacherpc(ref Creq.Dump(0, big 0));
}
validate(tag: int, qid: big, name: string): (ref Dir, string)
{
r := cacherpc(ref Creq.Validate(tag, qid, name));
pick rr := r {
Validate =>
return (rr.d, rr.err);
}
panic("validatebug");
return (nil, nil);
}
create(tag: int, qid: big, d: ref Sys->Dir): (ref Sys->Dir, string)
{
r := cacherpc(ref Creq.Create(tag, qid, ref *d));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, ""));
r = cacherpc(ref Creq.Create(tag, qid, ref *d));
}
pick rr := r {
Create =>
return (rr.d, rr.err);
}
panic("createbug");
return (nil, nil);
}
remove(tag: int, qid: big) : string
{
r := cacherpc(ref Creq.Remove(tag, qid));
pick rr := r {
Remove =>
return rr.err;
}
panic("removebug");
return nil;
}
walk1(tag: int, qid: big, elem: string): (ref Sys->Dir, string)
{
r := cacherpc(ref Creq.Walk1(tag, qid, elem));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, elem));
r = cacherpc(ref Creq.Walk1(tag, qid, elem));
}
pick rr := r {
Walk1 =>
return (rr.d, rr.err);
}
panic("walk1bug");
return (nil, nil);
}
readdir(tag: int, qid: big, cnt: int, off: int): (list of Sys->Dir, string)
{
r := cacherpc(ref Creq.Readdir(tag, qid, cnt, big off));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, ""));
r = cacherpc(ref Creq.Readdir(tag, qid, cnt, big off));
}
pick rr := r {
Readdir =>
return (rr.sons, rr.err);
}
panic("readdirbug");
return (nil, nil);
}
pread(tag: int, qid: big, cnt: int, off: big): (array of byte, string)
{
r := cacherpc(ref Creq.Pread(tag, qid, cnt, off));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, ""));
r = cacherpc(ref Creq.Pread(tag, qid, cnt, off));
}
pick rr := r {
Pread =>
return (rr.data, rr.err);
}
panic("preadbug");
return (nil, nil);
}
pwrite(tag: int, qid: big, data: array of byte, off: big) : (int, string)
{
r := cacherpc(ref Creq.Pwrite(tag, qid, data, off));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, ""));
r = cacherpc(ref Creq.Pwrite(tag, qid, data, off));
}
pick rr := r {
Pwrite =>
return (rr.count, rr.err);
}
panic("pwritebug");
return (0, nil);
}
stat(tag: int, qid: big): (ref Sys->Dir, string)
{
r := cacherpc(ref Creq.Stat(tag, qid));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, ""));
r = cacherpc(ref Creq.Stat(tag, qid));
}
pick rr := r {
Stat =>
return (rr.d, rr.err);
}
panic("statbug");
return (nil, nil);
}
wstat(tag: int, qid: big, d: ref Sys->Dir): (ref Sys->Dir, string)
{
r := cacherpc(ref Creq.Wstat(tag, qid, ref *d));
if (r.err == "invalid"){
cacherpc(ref Creq.Validate(tag, qid, ""));
r = cacherpc(ref Creq.Wstat(tag, qid, ref *d));
}
pick rr := r {
Wstat =>
return (rr.d, rr.err);
}
panic("wstatbug");
return (nil, nil);
}
flush(tag: int, old: int): string
{
r := cacherpc(ref Creq.Flush(tag, big 0, old));
pick rr := r {
Sync =>
return rr.err;
}
panic("flushbug");
return nil;
}
sync(tag: int, qid: big): string
{
r := cacherpc(ref Creq.Sync(tag, qid));
pick rr := r {
Sync =>
return rr.err;
}
panic("syncbug");
return nil;
}
opfname(d, n: string) : string
{
path: string;
if (n == "" || n == "/")
path = d;
else if (d == "/")
path = n;
else
path = d + "/" + n;
path = cleanname(path);
return path;
}
Wlist: type list of (ref Cfile, ref Creq, chan of ref Crep);
waitingop(l: Wlist, rf: ref Cfile): (Wlist, ref Creq, chan of ref Crep)
{
rr: ref Creq;
rrc: chan of ref Crep;
rl : Wlist;
for(; l != nil; l = tl l){
(f, r, rc) := hd l;
if (f == rf && rr == nil)
(rr, rrc) = (r, rc);
else
rl = (f, r, rc) :: rl;
}
return (rl, rr, rrc);
}
dumpwaiting(l: Wlist)
{
fprint(stderr, "\tcache: waiting:\n");
for(; l != nil; l = tl l){
(f, r, nil) := hd l;
fprint(stderr, "\t\t%s: %s\n", f.d.name, r.text());
}
}
# Main cache mux
# Requests are either being processed here, or spawned in auxiliary
# processes speaking op (and going through the opmux),
# or waiting for other requests to clear a file from being busy
cfsproc()
{
waiting: Wlist;
rootd := ref nulldir;
rootd.name = "/";
rootd.uid = rootd.gid = rootd.muid = "sys";
rootd.qid = Qid(big 0, 0, QTDIR);
root := Cfile.create(nil, rootd);
if (root == nil)
panic("cache: can't create root");
if (debug)
fprint(stderr, "cache: new: %s\n", root.text());
readyfile: ref Cfile;
taggen := 0;
Loop:
for (;;){
r: ref Creq;
rc: chan of ref Crep;
r = nil; rc = nil;
# Attend first pending ops on a file that was busy.
if (len waiting > maxwaiting)
maxwaiting = len waiting;
if (readyfile != nil){
(waiting, r, rc) = waitingop(waiting, readyfile);
if (r != nil && debug)
fprint(stderr, "\tcache: readyop: %s\n", r.text());
}
if (debug)
dumpwaiting(waiting);
if (r == nil){
readyfile = nil;
alt {
(r, rc) = <-cfsreqc =>
;
f := <-donec =>
f.busy = 0;
readyfile = f;
if (debug)
fprint(stderr, "\tcache: %s ready\n", f.d.name);
continue Loop;
}
}
if (r == nil){
if (debug) fprint(stderr, "mcache: stop\n");
break;
}
f := Cfile.find(r.qid);
if (f != nil && f.d == nil)
panic(sprint("cache: nil dir for file 16r%bx", r.qid));
pick rr := r {
Flush =>
xcfsreqc <-= Xop(xopflush, taggen++,nil,nil,nil,nil,rr.oldtag,big 0,rc);
Validate =>
if (f == nil){
rc <-= ref Crep.Validate("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
# 1. Create a fake dir hierarchy for all but the last component.
(nels, els) := tokenize(rr.path, "/");
wf := f;
for (i := 0 ; i < nels -1; i++){ # find or create a fake entry
(wf, nil) = wf.walkorcreate(hd els, nil);
if (wf != nil && wf.busy){
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (wf, r, rc) :: waiting;
continue Loop;
}
els = tl els;
}
f = wf;
# 2. try to validate the last component.
# els could be nil (e.g., for "/"), or for a pure clone
sf: ref Cfile;
if (els == nil)
sf = f;
else {
if (f == nil)
panic("mcache validate bug");
sf = f.walk(hd els);
}
# sf is either nil, or points to the file (and maybe sf == f)
if (sf != nil && sf.busy){
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (sf, r, rc) :: waiting;
continue;
}
now := millisec();
# This optimization does not seem to work in some cases
if (0)
if (sf == nil && f != sf && f.dirreaded && now - f.time < coherencylag){
rc <-= ref Crep.Validate("file does not exist", nil);
continue;
}
if (sf != nil && now - sf.time < coherencylag){
rc <-= ref Crep.Validate(nil, ref *sf.d);
continue;
}
# don't know. Ask the caller for help. Put the file on hold.
isnew := 0;
if (sf == nil){
# BUG: This is a problem. The cache should create the file
# only when necessary to keep cached data. Otherwise, we
# create the file here as a directory and it might later be a file,
# Things are fixed when the cache finds out the problem, but
# it would be better to make it right from the start.
(sf, isnew) = f.walkorcreate(hd els, nil);
}
if (sf.busy){
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (sf, r, rc) :: waiting;
continue;
}
sf.busy = 1;
path := sf.getpath();
xcfsreqc <-= Xop(xgetandvalidate, taggen++, sf, path, nil, nil, isnew, big 0, rc);
Create =>
if (f == nil){
rc <-= ref Crep.Create("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
(f,nil) = f.walkorcreate(rr.d.name, rr.d);
if (f.busy){
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
f.data = array[0] of byte;
f.created = f.dirtyd = 1;
if (f.d.qid.qtype&QTDIR){
f.busy = 1;
path := f.getpath();
xcfsreqc <-= Xop(xputandcreate, taggen++, f, path, rr.d, nil, 0, big 0, rc);
} else
rc <-= ref Crep.Create(nil, ref *f.d);
Remove =>
if (f == nil){
rc <-= ref Crep.Remove("invalid");
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
f.busy = 1;
path := f.getpath();
xcfsreqc <-= Xop(xdelandremove, taggen++, f, path, nil, nil, 0, big 0, rc);
Walk1 =>
if (f == nil){
rc <-= ref Crep.Walk1("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
f = f.walk(rr.name);
if (f != nil && f.busy){
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
if (f == nil){
# if this error ever happens, try to (re)validate this file
rc <-= ref Crep.Walk1("invalid", nil);
} else
rc <-= ref Crep.Walk1(nil, ref *f.d);
Readdir =>
if (f == nil){
rc <-= ref Crep.Readdir("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
rc <-= ref Crep.Readdir(nil, f.children(rr.cnt, int rr.off));
Pread =>
if (f == nil){
rc <-= ref Crep.Pread("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
# we cache prefixes only, and don't believe that files
# reporting zero length are indeed empty.
if (rr.off >= f.d.length && f.d.length > big 0)
rc <-= ref Crep.Pread(nil, array[0] of byte);
else if (rr.off < big len f.data){
n := int rr.off + rr.cnt;
if (n > len f.data)
n = len f.data;
rc <-= ref Crep.Pread(nil, f.data[int rr.off:n]);
} else if (!(f.d.qid.qtype&QTDIR) && (cdata := f.pread(rr.cnt, rr.off)) != nil)
rc <-= ref Crep.Pread(nil, cdata);
else {
path := f.getpath();
f.busy = 1;
xcfsreqc <-= Xop(xgetandread, taggen++, f, path, nil, nil, rr.cnt, rr.off, rc);
}
Pwrite =>
# This is write-through (perhaps asynchronously)
if (f == nil){
rc <-= ref Crep.Pwrite("invalid", -1);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
if (rr.off <= big len f.data){
if (int rr.off + len rr.data > MAXCACHED){
f.data = f.data[0:int rr.off];
} else {
off := int rr.off;
ndata := array[off + len rr.data] of byte;
ndata[0:] = f.data[0:off];
ndata[off:] = rr.data;
f.data = ndata;
if (f.d.length < big len f.data)
f.d.length = big len f.data;
}
}
f.pwrite(rr.data, rr.off);
path := f.getpath();
f.busy = 1;
# keep this condition in sync with async in putandwrite()
l := len f.d.name;
if ((rr.off != big 0 && (l > 4 && (f.d.name[l-4:l] == ".dis" || f.d.name[l-4:l] == ".sbl"))) ||
(len rr.data == 8192 && !f.created && rr.off != big 0))
rc <-= ref Crep.Pwrite(nil, len rr.data);
xcfsreqc <-= Xop(xputandwrite, taggen++, f, path, nil, rr.data, 0, rr.off, rc);
Stat =>
if (f == nil){
rc <-= ref Crep.Stat("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
rc <-= ref Crep.Stat(nil, ref *f.d);
Wstat =>
if (f == nil){
rc <-= ref Crep.Wstat("invalid", nil);
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
truncate := (f.d.length != big 0 && rr.d.length == big 0);
if (truncate){
f.data = array[0] of byte;
f.created = 1; # to truncate it
}
f.dirtyd = 1;
f.busy = 1;
path := f.getpath();
xcfsreqc <-= Xop(xputandwstat, taggen++, f, path, rr.d, nil, 0, big 0, rc);
Sync =>
if (f == nil){
rc <-= ref Crep.Sync("invalid");
continue;
} else if (f.busy) {
if (debug) fprint(stderr, "\tcache: busy\n");
waiting = (f, r, rc) :: waiting;
continue;
}
path := f.getpath();
f.busy = 1;
xcfsreqc <-= Xop(xputandsync, taggen++, f, path, nil, nil, 0, big 0, rc);
Dump =>
root.dump(0, "cfs tab:\n");
fprint(stderr, "cache: %d max busy files, %d max held ops\n", maxwaiting, maxheld);
rc <-= ref Crep.Dump(nil);
}
}
}
# BUG: cache xfuncs and put a reasonably high limit to the number of processes.
xcfsproc()
{
pending: list of Xop;
procs: list of chan of Xop;
nprocs := 0;
idlec := chan of chan of Xop;
for(;;){
alt {
x := <-xcfsreqc =>
if (x.fun== nil)
exit;
xc: chan of Xop;
if (procs != nil){
xc = hd procs;
procs = tl procs;
} else if (nprocs < Nprocs) {
nprocs++;
xc = chan of Xop;
spawn xproc(xc, idlec);
} else
xc = nil;
if (xc != nil)
xc <-= x;
else {
pending = x :: pending;
if (len pending > maxheld)
maxheld = len pending;
if (debug)
fprint(stderr, "o/ofs: %d processes, op held\n", nprocs);
}
xc := <-idlec =>
if (pending != nil){
x := hd pending;
pending = tl pending;
if (debug)
fprint(stderr, "o/ofs: held op released\n");
xc <-= x;
} else
procs = xc::procs;
}
}
}
xproc(xc: chan of Xop, idlec: chan of chan of Xop)
{
for(;;){
x := <-xc;
x.fun(x);
idlec <-= xc;
}
}
xopflush(nil: Xop)
{
# opflush(tag, n, rc); BUG: pretend it's flushed.
}
opflush(tag, oldtag: int, rc: chan of ref Crep)
{
repc := opmux->rpc(ref Tmsg.Flush(2*tag, 2*oldtag));
<-repc;
rc <-= ref Crep.Flush(nil);
}
# BUG: Doing a ls -ld on a file that has NOT read permission will make Tget fail
# because we ask for data as well, and oxport would report a permission denied while trying to
# open the file for reading. This must be changed. If OSTAT|ODATA is asked to oxport,
# and only stat could be retrieved, only stat should be reported back, with just OSTAT in the
# reply. Of course, this means that there are problems reading file data, and the client may issue
# a Tget(ODATA) should it want to learn the cause (or probably assume just a permission denied).
get(tag: int, path: string, fd: int, mode: int, cnt: int, off: big): (int, ref Dir, array of byte, string)
{
data := array[0] of byte;
d : ref Dir;
nm := 1;
if (mode&ODATA){
while(nm < MAXNMSGS && off + big (Op->MAXDATA*nm) < big MAXCACHED)
nm++;
if (nm > 1 || off < big MAXCACHED)
cnt = MAXDATA;
}
req := ref Tmsg.Get(tag, path, fd, mode, nm, off, cnt);
repc := opmux->rpc(req);
nmsgs := nm;
d = nil;
for (;;){
rep := <- repc;
pick r := rep {
Error =>
return (Op->NOFD, nil, nil, r.ename);
Get =>
if ((r.mode&OSTAT) != 0){
d = ref r.stat;
if ((d.qid.qtype&QTDIR) != 0)
nmsgs = 0;
}
if ((r.mode &ODATA) != 0){
ndata := array[len data + len r.data] of byte;
ndata[0:] = data;
ndata[len data:] = r.data;
data = ndata;
}
if (--nmsgs == 0 || (r.mode&OMORE) == 0) # last message
return(r.fd, d, data, nil);
* =>
panic("get0bug");
}
}
panic("get0bug");
return (Op->NOFD, nil, nil, "bug");
}
xgetandvalidate(x: Xop)
{
getandvalidate(x.tag, x.f, x.path, x.n, x.rc);
}
getandvalidate(tag: int, f: ref Cfile, path: string, isnew: int, rc: chan of ref Crep)
{
(fd, d, data, e) := get(tag, path, f.oprdfd, OSTAT|ODATA|OMORE, MAXDATA, big 0);
f.oprdfd = fd;
if (e != nil || d == nil){
f.remove();
rc <-= ref Crep.Validate(e, nil);
}
else {
if (isnew)
f.serverqid = d.qid;
else {
if (f.serverqid.path != d.qid.path || f.serverqid.vers != d.qid.vers)
if (!f.created)
f.data = array[0] of byte;
f.serverqid = d.qid;
}
f.wstat(d);
if (d.qid.qtype&QTDIR){
f.dirreaded = 1;
f.updatedirdata(data);
} else {
f.pwrite(data, big 0);
if (!f.dirtyd && !f.created)
f.data = data;
}
f.d.qid.vers = d.qid.vers;
f.time = millisec();
rc <-= ref Crep.Validate(nil, ref *f.d);
}
donec <-= f;
}
xgetandread(x: Xop)
{
getandread(x.tag, x.f, x.path, x.n, x.o, x.rc);
}
getandread(tag: int, f: ref Cfile, path: string, cnt: int, off: big, rc: chan of ref Crep)
{
(fd, nil, data, e) := get(tag, path, f.oprdfd, ODATA|OMORE, cnt, off);
f.oprdfd = fd;
if (e != nil){
# fsremove(fstab, f.d.qid.path); should?
rc <-= ref Crep.Pread(e, nil);
} else {
if (f.d.length != big 0){ # files with zero-length are probably streams.
f.pwrite(data, off); # do not cache.
odata := f.data;
o := int off;
if (o <= len odata){
if (o + len data > MAXCACHED)
f.data = odata[0:o];
else {
ndata := array[o + len data] of byte;
ndata[0:] = odata[0:o];
ndata[o:] = data;
f.data = ndata;
if (f.d.length < big len f.data)
f.d.length = big len f.data;
}
}
}
f.time = millisec();
rc <-= ref Crep.Pread(nil, data);
}
donec <-= f;
}
# BUG: Too many args; put does it all.
put(tag: int, path: string, opfd: int, d: ref Dir, data: array of byte, off: big, mkit, more: int) : (int, int, int, string)
{
if (d == nil && data == nil && opfd == Op->NOFD)
return (NOFD, 0, 0, nil);
flag := 0;
if (d != nil)
flag |= OSTAT;
if (data != nil && (!mkit || (d.qid.qtype&QTDIR) == 0))
flag |= ODATA;
if (mkit)
flag |= OCREATE;
if (more)
flag |= OMORE;
if (data == nil)
data = array[0] of byte;
tot := 0;
rfd := NOFD;
mtime := 0;
for(;;){
nw := len data[tot:];
if (nw > MAXDATA)
nw = MAXDATA;
if (d == nil)
d = ref nulldir;
rc := opmux->rpc(ref Tmsg.Put(tag, path, opfd, flag, *d, off, data[tot:tot+nw]));
vers := 0;
r := <- rc;
pick rr := r {
Put =>
if (rr.count != nw)
return (NOFD, 0, 0, "short write");
vers = rr.qid.vers;
rfd = rr.fd;
mtime = rr.mtime;
Error =>
return (NOFD, 0, 0, rr.ename);
* =>
panic("putbug");
}
tot += nw;
if (tot == len data)
return (rfd, vers, mtime, nil);
flag &= ~(OSTAT|OCREATE);
}
panic("putbug");
return (NOFD, 0, 0, "putbug");
}
xputandcreate(x: Xop)
{
putandcreate(x.tag, x.f, x.path, x.d, x.rc);
}
putandcreate(tag: int, f: ref Cfile, path: string, d: ref Dir, rc: chan of ref Crep)
{
(fd, vers, mtime, e) := put(tag, path, f.opwrfd, f.d, nil, big 0, 1, 1);
f.opwrfd = fd;
if (e != nil){
f.remove();
rc <-= ref Crep.Create(e, nil);
} else {
f.created = f.dirtyd = 0;
f.wstat(d);
if (vers != 0 || mtime != 0){
f.serverqid.vers = vers;
f.d.qid.vers = vers;
f.d.mtime = mtime;
}
f.time = millisec();
rc <-= ref Crep.Create(nil, ref *f.d);
}
donec <-= f;
}
xputandwrite(x: Xop)
{
putandwrite(x.tag, x.f, x.path, x.data, x.o, x.rc);
}
putandwrite(tag: int, f: ref Cfile, path: string, data: array of byte, off: big, rc: chan of ref Crep)
{
async := (len data == 8192 && !f.created && off != big 0); # (off % big (4*8192)) != big 0);
l := len f.d.name;
if (off != big 0 && (l > 4 && (f.d.name[l-4:l] == ".dis" || f.d.name[l-4:l] == ".sbl")))
async = 1;
d: ref Dir;
if (f.dirtyd)
d = f.d;
f.dirtyd = 0;
mkit := f.created;
f.created = 0;
if (async)
donec <-= f;
(fd, vers, mtime, e) := put(tag, path, f.opwrfd, d, data, off, mkit, 1);
f.opwrfd = fd;
if (e != nil){
f.remove();
if (!async)
rc <-= ref Crep.Pwrite(e, len data);
else
fprint(stderr, "\n*** ofs: delayed write error ***\n\n");
} else {
if (!async){
f.serverqid.vers = vers;
f.d.qid.vers = vers;
f.d.mtime= mtime;
f.time = millisec();
rc <-= ref Crep.Pwrite(nil, len data);
} else {
# Race, but safer this way.
f.d.qid.vers = vers;
if (mtime)
f.d.mtime = mtime;
}
}
if (!async)
donec <-= f;
}
xputandwstat(x: Xop)
{
putandwstat(x.tag, x.f, x.path, x.d, x.rc);
}
putandwstat(tag: int, f: ref Cfile, path: string, d: ref Dir, rc: chan of ref Crep)
{
mkit := f.created;
f.created = 0;
f.dirtyd = 0;
(fd, vers, mtime, e) := put(tag, path, f.opwrfd, d, nil, big 0, mkit, 1);
f.opwrfd = fd;
if (e != nil){
f.remove();
rc <-= ref Crep.Wstat(e, nil);
} else {
f.serverqid.vers = vers;
f.wstat(d);
f.d.mtime = mtime;
f.time = millisec();
rc <-= ref Crep.Wstat(nil, ref *f.d);
}
donec <-= f;
}
xputandsync(x: Xop)
{
putandsync(x.tag, x.f, x.path, x.rc);
}
putandsync(tag: int, f: ref Cfile, path: string, rc: chan of ref Crep)
{
mkit := f.created;
f.created = 0;
d: ref Dir;
if (f.dirtyd)
d = ref *f.d;
f.dirtyd = 0;
f.oldname = nil;
f.fsfd = nil;
rc <-= ref Crep.Sync(nil);
(fd, vers, mtime, e) := put(tag, path, f.opwrfd, d, nil, big 0, mkit, 0);
f.opwrfd = fd;
if (f.oprdfd != NOFD){
(fd, nil, nil, nil) = get(tag, path, f.oprdfd, OSTAT, 0, big 0);
f.oprdfd = fd;
}
if (e != nil)
f.remove();
else {
f.serverqid.vers = vers;
if (vers != 0)
f.d.qid.vers = vers;
if (mtime != 0)
f.d.mtime = mtime;
f.time = millisec();
}
donec <-= f;
}
xdelandremove(x: Xop)
{
delandremove(x.tag, x.f, x.path, x.rc);
}
delandremove(tag: int, f: ref Cfile, path: string, rc: chan of ref Crep)
{
oprc := opmux->rpc(ref Tmsg.Remove(2*tag, path));
opr := <-oprc;
e: string;
pick oprr := opr {
Remove =>
f.remove();
Error =>
e = oprr.ename;
}
rc <-= ref Crep.Remove(e);
if (f.oprdfd != NOFD){
(fd, nil, nil, nil) := get(tag, path, f.oprdfd, OSTAT, 0, big 0);
f.oprdfd = fd;
}
if (f.opwrfd != NOFD){
(fd, nil, nil, nil) := put(tag, path, f.opwrfd, nil, nil, big 0, 0, 0);
f.opwrfd = fd;
}
donec <-= f;
}
|
