fs.c

这是一个同样来自贝尔实验室的和UNIX有着渊源的操作系统, 其简洁的设计和实现易于我们学习和理解

	if(f->file == &files[Qplayctl]){
		r->ifcall.data[cnt] = '\0';
		if (debug & DbgPlayer)
			fprint(2, "rwrite playctl: %s\n", r->ifcall.data);
		nf = tokenize(r->ifcall.data, fields, 4);
		if (nf == 0){
			r->ofcall.count = r->ifcall.count;
			return nil;
		}
		if (nf == 2)
			i = strtol(fields[1], nil, 0);
		else
			i = playstate.off;
		newstate = playstate;
		if ((cmd = lookup(fields[0], statetxt)) < 0)
			return  Ebadctl;
		switch(cmd){
		case Play:
			newstate.cmd = cmd;
			newstate.off = i;
			break;
		case Pause:
			if (playstate.cmd != Play)
				break;
			// fall through
		case Stop:
			newstate.cmd = cmd;
			newstate.off = playstate.off;
			break;
		case Resume:
			if(playstate.cmd == Stop)
				break;
			newstate.cmd = Resume;
			newstate.off = playstate.off;
			break;
		case Skip:
			if (nf == 2)
				i += playstate.off;
			else
				i = playstate.off +1;
			if(i < 0)
				i = 0;
			else if (i >= playlist.nlines)
				i = playlist.nlines - 1;
			newstate.cmd = Play;
			newstate.off = i;
		}
		if (newstate.off >= playlist.nlines){
			newstate.cmd = Stop;
			newstate.off = playlist.nlines;
		}
		if (debug & DbgPlayer)
			fprint(2, "new state %s-%ud\n",
				statetxt[newstate.cmd], newstate.off);
		if (newstate.m != playstate.m)
			sendul(playc, newstate.m);
		f->file->dir.qid.vers++;
	} else if(f->file == &files[Qplayvol]){
		char *subfields[nelem(volume)];
		int v[nelem(volume)];

		r->ifcall.data[cnt] = '\0';
		if (debug & DbgPlayer)
			fprint(2, "rwrite playvol: %s\n", r->ifcall.data);
		nf = tokenize(r->ifcall.data, fields, 4);
		if (nf == 0){
			r->ofcall.count = r->ifcall.count;
			return nil;
		}
		if (nf != 2 || strcmp(fields[0], "volume") != 0)
			return Ebadctl;
		if (debug & DbgPlayer)
			fprint(2, "new volume '");
		nf = tokenize(fields[1], subfields, nelem(subfields));
		if (nf <= 0 || nf > nelem(volume))
			return "volume";
		for (i = 0; i < nf; i++){
			v[i] = strtol(subfields[i], nil, 0);
			if (debug & DbgPlayer)
				fprint(2, " %d", v[i]);
		}
		if (debug & DbgPlayer)
			fprint(2, "'\n");
		while (i < nelem(volume))
			v[i++] = Undef;
		volumeset(v);
		r->ofcall.count = r->ifcall.count;
		return nil;
	} else if(f->file == &files[Qplaylist]){
		if (debug & DbgPlayer){
			fprint(2, "rwrite playlist: `");
			write(2, r->ifcall.data, cnt);
			fprint(2, "'\n");
		}
		playlist.data = realloc(playlist.data, playlist.ndata + cnt + 2);
		if (playlist.data == 0)
			sysfatal("realloc: %r");
		memmove(playlist.data + playlist.ndata, r->ifcall.data, cnt);
		if (playlist.data[playlist.ndata + cnt-1] != '\n')
			playlist.data[playlist.ndata + cnt++] = '\n';
		playlist.data[playlist.ndata + cnt] = '\0';
		p = playlist.data + playlist.ndata;
		while (*p){
			playlist.lines = realloc(playlist.lines, (playlist.nlines+1)*sizeof(playlist.lines[0]));
			if(playlist.lines == nil)
				sysfatal("realloc: %r");
			playlist.lines[playlist.nlines] = playlist.ndata;
			q = strchr(p, '\n');
			if (q == nil)
				break;
			if(debug & DbgPlayer)
				fprint(2, "[%lud]: ", playlist.nlines);
			playlist.nlines++;
			q++;
			if(debug & DbgPlayer)
				write(2, p, q-p);
			playlist.ndata += q - p;
			p = q;
		}
		f->file->dir.length = playlist.ndata;
		f->file->dir.qid.vers++;
	}else
		return Eperm;
	r->ofcall.count = r->ifcall.count;
	m.cmd = Check;
	m.off = 0;
	m.arg = nil;
	bcastmsg(f->file->workers, &m);
	return nil;
}

char *
rclunk(Worker *w)
{
	Fid *f;

	f = oldfid(w->r->ifcall.fid);
	if(f == nil)
		return Enotexist;
	f->flags &= ~(Open|Busy);
	return 0;
}

char *
rremove(Worker*)
{
	return Eperm;
}

char *
rstat(Worker *w)
{
	Req *r;

	r = w->r;
	r->fid = oldfid(r->ifcall.fid);
	if(r->fid == nil)
		return Enotexist;
	r->ofcall.nstat = convD2M(&r->fid->file->dir, r->indata, messagesize - IOHDRSZ);
	r->ofcall.stat = r->indata;
	return 0;
}

char *
rwstat(Worker*)
{
	return Eperm;
}

Fid *
oldfid(int fid)
{
	Fid *f;

	for(f = fids; f; f = f->next)
		if(f->fid == fid)
			return f;
	return nil;
}

Fid *
newfid(int fid)
{
	Fid *f, *ff;

	ff = nil;
	for(f = fids; f; f = f->next)
		if(f->fid == fid){
			return f;
		}else if(ff == nil && (f->flags & Busy) == 0)
			ff = f;
	if(ff == nil){
		ff = mallocz(sizeof *ff, 1);
		if (ff == nil)
			sysfatal("malloc: %r");
		ff->next = fids;
		ff->readers = 0;
		fids = ff;
	}
	ff->fid = fid;
	ff->file = nil;
	ff->vers = ~0;
	return ff;
}

void
work(Worker *w)
{
	Req *r;
	char *err;
	int n;

	r = w->r;
	r->ofcall.data = (char*)r->indata;
	if(!fcalls[r->ifcall.type])
		err = "bad fcall type";
	else
		err = (*fcalls[r->ifcall.type])(w);
	if(err != (char*)~0){	/* ~0 indicates Flush received */
		if(err){
			r->ofcall.type = Rerror;
			r->ofcall.ename = err;
		}else{
			r->ofcall.type = r->ifcall.type + 1;
			r->ofcall.fid = r->ifcall.fid;
		}
		r->ofcall.tag = r->ifcall.tag;
		if(debug & DbgFs)
			fprint(2, "io:->%F\n", &r->ofcall);/**/
		n = convS2M(&r->ofcall, r->outdata, messagesize);
		if(write(srvfd[0], r->outdata, n) != n)
			sysfatal("mount write");
	}
	reqfree(r);
	w->r = nil;
}

void
worker(void *arg)
{
	Worker *w;
	Wmsg m;

	w = arg;
	recv(w->eventc, &m);
	for(;;){
		assert(m.cmd == Work);
		w->r = m.arg;
		if(debug & DbgWorker)
			fprint(2, "worker 0x%p:<-%F\n", w, &w->r->ifcall);
		work(w);
		if(debug & DbgWorker)
			fprint(2, "worker 0x%p wait for next\n", w);
		m = waitmsg(w, workers);
	}
}

void
allocwork(Req *r)
{
	Worker *w;
	Wmsg m;

	m.cmd = Work;
	m.off = 0;
	m.arg = r;
	if(sendmsg(workers, &m))
		return;
	/* No worker ready to accept request, allocate one */
	w = malloc(sizeof(Worker));
	w->eventc = chancreate(sizeof(Wmsg), 1);
	if(debug & DbgWorker)
		fprint(2, "new worker 0x%p\n", w);/**/
	threadcreate(worker, w, 4096);
	send(w->eventc, &m);
}

void
srvio(void *arg)
{
	char e[32];
	int n;
	Req *r;
	Channel *dispatchc;

	threadsetname("file server IO");
	dispatchc = arg;

	r = reqalloc();
	for(;;){
		/*
		 * reading from a pipe or a network device
		 * will give an error after a few eof reads
		 * however, we cannot tell the difference
		 * between a zero-length read and an interrupt
		 * on the processes writing to us,
		 * so we wait for the error
		 */
		n = read9pmsg(srvfd[0], r->indata, messagesize);
		if(n == 0)
			continue;
		if(n < 0){
			rerrstr(e, sizeof e);
			if (strcmp(e, "interrupted") == 0){
				if (debug & DbgFs) fprint(2, "read9pmsg interrupted\n");
				continue;
			}
			sysfatal("srvio: %s", e);
		}
		if(convM2S(r->indata, n, &r->ifcall) == 0)
			continue;

		if(debug & DbgFs)
			fprint(2, "io:<-%F\n", &r->ifcall);
		sendp(dispatchc, r);
		r = reqalloc();
	}
}

char *
getplaylist(int n)
{
	Wmsg m;

	m.cmd = Preq;
	m.off = n;
	m.arg = nil;
	send(playlistreq, &m);
	recv(playlistreq, &m);
	if(m.cmd == Error)
		return nil;
	assert(m.cmd == Prep);
	assert(m.arg);
	return m.arg;
}

void
playlistsrv(void*)
{
	Wmsg m;
	char *p, *q, *r;
	char *fields[2];
	int n;
	/* Runs in the srv proc */

	threadsetname("playlistsrv");
	while(recv(playlistreq, &m)){
		assert(m.cmd == Preq);
		m.cmd = Error;
		if(m.off < playlist.nlines){
			p = playlist.data + playlist.lines[m.off];
			q = strchr(p, '\n');
			if (q == nil)
				sysfatal("playlistsrv: no newline character found");
			n = q-p;
			r = malloc(n+1);
			memmove(r, p, n);
			r[n] = 0;
			tokenize(r, fields, nelem(fields));
			assert(fields[0] == r);
			m.cmd = Prep;
			m.arg = r;
		}
		send(playlistreq, &m);
	}
}

void
srv(void*)
{
	Req *r;
	Channel *dispatchc;
	/*
	 * This is the proc with all the action.
	 * When a file request comes in, it is dispatched to this proc
	 * for processing.  Two extra threads field changes in play state
	 * and volume state.
	 * By keeping all the action in this proc, we won't need locks
	 */

	threadsetname("srv");
	close(srvfd[1]);

	dispatchc = chancreate(sizeof(Req*), 1);
	procrfork(srvio, dispatchc, 4096, RFFDG);

	threadcreate(volumeupdater, nil, 4096);
	threadcreate(playupdater, nil, 4096);
	threadcreate(playlistsrv, nil, 4096);

	while(r = recvp(dispatchc))
		allocwork(r);
		
}

void
playupdater(void*)
{
	Wmsg m;
	/* This is a thread in the srv proc */

	while(recv(playchan, &m)){
		if(debug & DbgPlayer)
			fprint(2, "playupdate: %s %d %s\n", statetxt[m.cmd], m.off, m.arg?m.arg:"");
		if(playstate.m == m.m)
			continue;
		if(m.cmd == Stop && m.off == 0xffff)
			m.off = playlist.nlines;
		if(m.cmd != Error){
			playstate.m = m.m;
			m.cmd = Check;
			assert(m.arg == nil);
		}
		files[Qplayctl].dir.qid.vers++;
		bcastmsg(files[Qplayctl].workers, &m);
	}
}

void
volumeupdater(void*)
{
	Wmsg m;
	int v[nelem(volume)];
	/* This is a thread in the srv proc */

	while(recv(volumechan, v)){
		if(debug & DbgPlayer)
			fprint(2, "volumeupdate: volume now %d %d %d %d\n", volume[0], volume[1], volume[2], volume[3]);
		memmove(volume, v, sizeof(volume));
		files[Qplayvol].dir.qid.vers++;
		m.cmd = Check;
		m.arg = nil;
		bcastmsg(files[Qplayvol].workers, &m);
	}
}

void
playupdate(Pmsg p, char *s)
{
	Wmsg m;

	m.m = p.m;
	m.arg = s ? strdup(s) : nil;
	send(playchan, &m);
}