[9fans] writefs.

[9fans] writefs.

[Dan Cross]

[-- Attachment #1: Type: text/plain, Size: 1106 bytes --]

I wrote this a long time ago; it's imperfect, and full of bugs,
I'm sure.  It crashes fairly often, too, but seems to work for
simple things.  Conceptually, it's like what Geoff posted,
only implemented as a filesystem instead of just a simple
file descriptor in /srv.  The idea is that it could thus be run
on a central CPU server and imported all over the place,
effectively giving network write service for free.  If someone
wants to clean it up and mail it back to me, feel free; I
never finished writing the ACL stuff (I'd probably do it
differently now, anyway).  Maybe in a few weeks I'll have
some time to clean it up some more and we can stick it in
the distribution or something.

To run it, compile writefs.c and put the binary somewhere.
Run it on a CPU server when it starts up.  Copy writelstn
and write somewhere, and modify them as needed (for
instance, to import /mnt/writefs from the CPU server if needed).
Run writelstn in a window somewhere to receive messages
(I usually run it in a `win' under acme) and run `write user' to
send messages to others.

	- Dan C.


[-- Attachment #2: write --]
[-- Type: application/octet-stream, Size: 77 bytes --]

#!/bin/rc
rfork n
mount /srv/writefs /mnt/write
exec cat > /mnt/write/$1

[-- Attachment #3: writefs.c --]
[-- Type: text/plain, Size: 4909 bytes --]

/*
 *  I miss the Unix write(1) command, but I think
 *  this is a better way to do it.
 */

#include <u.h>
#include <libc.h>
#include <fcall.h>
#include <thread.h>
#include <9p.h>

#define	min(A, B)	((A) < (B) ? (A) : (B))
#define	STREQ(A, B)	((A)[0] == (B)[0] && strcmp(A, B) == 0)

//  Messages are consumed by readers either when they're
//  posted, or when the reader sends a Tread message.

typedef	struct Message Message;
struct	Message {
	Ref;
	char	*msg;
};

typedef	struct Reader Reader;
struct	Reader {
	Reader	*next;
	Channel	*mq;
	Req	*req;
	char	*name;
};

typedef	struct Acl Acl;
struct	Acl {
	char	**acl;
	int	acllen;
};

typedef	struct Station Station;
struct	Station {
	Reader	*readers;
	Acl	*rallow;
	Acl	*rdisallow;
	Acl	*wallow;
	Acl	*wdisallow;
	Acl	*callow;
	Acl	*cdisallow;
};

static void	*ezmalloc(ulong);
static void	*estrdup(char *);

static void	consume(Reader *);
static void	produce(Message *, Reader *);

void	wfsopen(Req *);
void	wfsread(Req *);
void	wfswrite(Req *);

enum { POOLINCR = 20 };

Srv	serv = {
	.open =	wfsopen,
	.read =	wfsread,
	.write=	wfswrite
};
Tree	*tp;

static void *
ezmalloc(ulong size)
{
	void	*p;

	p = malloc(size);
	if (p == nil)
		sysfatal("malloc %lud bytes: %r\n", size);
	memset(p, 0, size);
	return(p);
}

static char *
estrdup(char *str)
{
	char	*p;

	p = strdup(str);
	if (p == nil)
		sysfatal("estrdup: malloc failed: %r\n");
	return(p);
}

static void
produce(Message *mp, Reader *readers)
{
	Reader	*rp;

	for (rp = readers; rp != nil; rp = rp->next)
		incref(mp);
	for (rp = readers; rp != nil; rp = rp->next)
		sendp(rp->mq, mp);
}

void
consume(void *reader)
{
	Reader	*rp;
	Message	*mp;
	Req	*req;
	long	len;

	rp = reader;
	for ( ; ; ) {
		req = rp->req;
		if (req == nil) {
			yield();
			continue;
		}
		mp = recvp(rp->mq);
		if (mp == nil)
			continue;
		len = min(strlen(mp->msg), req->ifcall.count);
		memcpy(req->ofcall.data, mp->msg, len);
		req->ofcall.count = len;
		if (decref(mp) == 0)
			free(mp);
		rp->req = nil;
		respond(req, nil);
	}
	threadexits(0);
}

Reader *
addreader(File *file, char *user)
{
	Station	*sp;
	Reader	*rp, *pp, *np;

	sp = file->aux;
	for (pp = nil, rp = sp->readers;
	    rp != nil && strcmp(rp->name, user) < 0;
	    pp = rp, rp = rp->next)
		;
	if (rp != nil && STREQ(rp->name, user))
		return(rp);
	np = ezmalloc(sizeof(*rp));
	np->name = estrdup(user);
	np->req = nil;
	np->mq = chancreate(sizeof(Message *), 128);
	np->next = rp;
	if (pp == nil) {
		//  At the beginning or empty.
		sp->readers = np;
	} else {
		pp->next = np;
	}
	proccreate(consume, np, 8192);
	return(np);
}

/*
 *  Add ACL testing code.
 */
void
wfsopen(Req *req)
{
	if (!STREQ(req->fid->file->name, "clone") &&
	    !(req->fid->file->mode & DMDIR) &&
	    (req->ifcall.mode == OREAD) || req->ifcall.mode == ORDWR)
		addreader(req->fid->file, req->fid->uid);
	respond(req, nil);
}

void
wfsread(Req *req)
{
	File	*file;
	Fid	*fid;
	char	*uid;
	Station	*sp;
	Reader	*rp;

	uid = req->fid->uid;
	file = req->fid->file;
	if (file == nil)
		sysfatal("fid->file == nil?");
	if (STREQ(file->name, "clone")) {
		if (req->ifcall.offset > 0) {
			req->ofcall.count = 0;
			respond(req, nil);
			return;
		}
		file = createfile(tp->root, uid, uid, 0622, ezmalloc(sizeof(*sp)));
		if (file == nil) {
			respond(req, "user already exists");
			return;
		}
		strcpy(req->ofcall.data, uid);
		req->ofcall.count = strlen(uid);
		respond(req, nil);
		return;
	}
	sp = file->aux;
	if (sp != nil)
		for (rp = sp->readers; rp != nil; rp = rp->next)
			if (STREQ(uid, rp->name)) {
				if (rp->req != nil)
					respond(rp->req, nil);
				rp->req = req;
			}
}

void
wfswrite(Req *req)
{
	Station	*sp;
	Reader	*rp;
	Message	*mp;
	Fid	*fid;
	int	len;
	char	date[32];

	fid = req->fid;
	sp = fid->file->aux;
	if (sp->readers == nil) {
		respond(req, "no listeners");
		return;
	}
	snprint(date, sizeof date, "%lld", (vlong)time(nil));
	len = strlen(date) + 1 + strlen(fid->uid) + 1 + strlen(fid->file->name) + 1 + req->ifcall.count + 1;
	mp = ezmalloc(sizeof(*mp) + len);
	mp->msg = (char *)mp + sizeof(*mp);
	snprint(mp->msg, len, "%s:%s:%s:%.*s", date,
	    fid->uid, fid->file->name,
	    req->ifcall.count, req->ifcall.data);
	req->ofcall.count = req->ifcall.count;
	produce(mp, sp->readers);
	respond(req, nil);
}

void
threadmain(int, char *[])
{
	File	*f;
	char	*user;

	rfork(RFNOTEG);
	user = getuser();
	if (user == nil)
		user = "unknown";
	tp = alloctree(user, user, DMDIR | 0664, nil);
	serv.tree = tp;
	f = createfile(tp->root, "clone", user, 0666, nil);
	if (f == nil)
		sysfatal("fcreate clone");
	threadpostmountsrv(&serv, "writefs", "/mnt/write", MREPL);

	exits(0);
}

[-- Attachment #4: writelstn --]
[-- Type: application/octet-stream, Size: 235 bytes --]

#!/bin/rc
rfork n
mount /srv/writefs /mnt/write
cat /mnt/write/clone > /dev/null >[2=1]
exec awk -F: '{
	time = $1
	from = $2
	to = $3
	msg = $0
	sub(/^[^:]*:[^:]*:[^:]*:/, "", msg)
	print from ": " msg
}' /mnt/write/$user