implement Logfile;

#
# Copyright © 1999 Vita Nuova Limited.  All rights reserved.
#

include "sys.m";
	sys: Sys;
	stderr: ref Sys->FD;
include "draw.m";

Logfile: module {
	init: fn(nil: ref Draw->Context, argv: list of string);
};

Fidrec: adt {
	fid: 	int;		# fid of read
	rq: 	list of (int, Sys->Rread);	# outstanding read requests
	pos:	int;		# current position in the logfile
};

Circbuf: adt {
	start: int;
	data: array of byte;
	new: fn(size: int): ref Circbuf;
	put: fn(b: self ref Circbuf, d: array of byte): int;
	get: fn(b: self ref Circbuf, s, n: int): (int, array of byte);
};

Fidhash: adt
{
	table: array of list of ref Fidrec;
	get: fn(ht: self ref Fidhash, fid: int): ref Fidrec;
	put: fn(ht: self ref Fidhash, fidrec: ref Fidrec);
	del: fn(ht: self ref Fidhash, fidrec: ref Fidrec);
	new: fn(): ref Fidhash;
};

usage()
{
	sys->fprint(stderr, "usage: logfile [-size] file\n");
	sys->raise("fail: usage");
}

init(nil: ref Draw->Context, argv: list of string)
{
	sys = load Sys Sys->PATH;
	stderr = sys->fildes(2);

	bufsize := Sys->ATOMICIO * 4;

	if (argv != nil)
		argv = tl argv;
	if (argv != nil && (hd argv)[0] == '-' && len hd argv > 1) {
		if ((bufsize = int ((hd argv)[1:])) <= 0) {
			sys->fprint(stderr, "logfile: can't have a zero buffer size\n");
			usage();
		}
		argv = tl argv;
	}
	if (argv == nil || tl argv != nil)
		usage();
	path := hd argv;

	(dir, f) := pathsplit(path);
	if (sys->bind("#s", dir, Sys->MBEFORE|Sys->MCREATE) == -1) {
		sys->fprint(stderr, "logfile: bind #s failed: %r\n");
		return;
	}
	fio := sys->file2chan(dir, f);
	if (fio == nil) {
		sys->fprint(stderr, "logfile: couldn't make %s: %r\n", path);
		return;
	}

	spawn logserver(fio, bufsize);
}

logserver(fio: ref Sys->FileIO, bufsize: int)
{
	waitlist: list of ref Fidrec;
	readers := Fidhash.new();
	availcount := 0;
	availchan := chan of int;
	workchan := chan of (Sys->Rread, array of byte);
	buf := Circbuf.new(bufsize);
	for (;;) alt {
	<-availchan =>
		availcount++;
	(off, count, fid, rc) := <-fio.read =>
		r := readers.get(fid);
		if (rc == nil) {
			if (r != nil)
				readers.del(r);
			continue;
		}
		if (r == nil) {
			r = ref Fidrec(fid, nil, buf.start);
			if (r.pos < len buf.data)
				r.pos = len buf.data;		# first buffer's worth is garbage
			readers.put(r);
		}

		(s, d) := buf.get(r.pos, count);
		r.pos = s + len d;

		if (d != nil) {
			rc <-= (d, nil);
		} else {
			if (r.rq == nil)
				waitlist = r :: waitlist;
			r.rq = (count, rc) :: r.rq;
		}

	(off, data, fid, wc) := <-fio.write =>
		if (wc == nil)
			continue;
		if ((n := buf.put(data)) < len data)
			wc <-= (n, "write too long for buffer");
		else
			wc <-= (n, nil);

		wl := waitlist;
		for (waitlist = nil; wl != nil; wl = tl wl) {
			r := hd wl;
			if (availcount == 0) {
				spawn worker(workchan, availchan);
				availcount++;
			}
			(count, rc) := hd r.rq;
			r.rq = tl r.rq;

			# optimisation: if the read request wants exactly the data provided
			# in the write request, then use the original data buffer.
			s: int;
			d: array of byte;
			if (count >= n && r.pos == buf.start + len buf.data - n)
				(s, d) = (r.pos, data);
			else
				(s, d) = buf.get(r.pos, count);
			r.pos = s + len d;
			workchan <-= (rc, d);
			availcount--;
			if (r.rq != nil)
				waitlist = r :: waitlist;
			d = nil;
		}
		data = nil;
		wl = nil;
	}
}

worker(work: chan of (Sys->Rread, array of byte), ready: chan of int)
{
	for (;;) {
		(rc, data) := <-work;	# blocks forever if the reading process is killed
		rc <-= (data, nil);
		(rc, data) = (nil, nil);
		ready <-= 1;
	}
}
		
Circbuf.new(size: int): ref Circbuf
{
	return ref Circbuf(0, array[size] of byte);
}

# return number of bytes actually written
Circbuf.put(b: self ref Circbuf, d: array of byte): int
{
	blen := len b.data;
	# if too big to fit in buffer, truncate the write.
	if (len d > blen)
		d = d[0:blen];
	dlen := len d;

	offset := b.start % blen;
	if (offset + dlen <= blen) {
		b.data[offset:] = d;
	} else {
		b.data[offset:] = d[0:blen - offset];
		b.data[0:] = d[blen - offset:];
	}
	b.start += dlen;
	return dlen;
}

# return (start, data)
Circbuf.get(b: self ref Circbuf, s, n: int): (int, array of byte)
{
	# if the beginning's been overrun, start from the earliest place we can.
	# we could put some indication of elided bytes in the buffer.
	if (s < b.start)
		s = b.start;
	blen := len b.data;
	if (s + n > b.start + blen)
		n = b.start + blen - s;
	if (n <= 0)
		return (s, nil);
	o := s % blen;
	d := array[n] of byte;
	if (o + n <= blen)
		d[0:] = b.data[o:o+n];
	else {
		d[0:] = b.data[o:];
		d[blen - o:] = b.data[0:o+n-blen];
	}
	return (s, d);
}

FIDHASHSIZE: con 32;

Fidhash.new(): ref Fidhash
{
	return ref Fidhash(array[FIDHASHSIZE] of list of ref Fidrec);
}

# put an entry in the hash table.
# assumes there is no current entry for the fid.
Fidhash.put(ht: self ref Fidhash, f: ref Fidrec)
{
	slot := f.fid & (FIDHASHSIZE-1);
	ht.table[slot] = f :: ht.table[slot];
}

Fidhash.get(ht: self ref Fidhash, fid: int): ref Fidrec
{
	for (l := ht.table[fid & (FIDHASHSIZE-1)]; l != nil; l = tl l)
		if ((hd l).fid == fid)
			return hd l;
	return nil;
}

Fidhash.del(ht: self ref Fidhash, f: ref Fidrec)
{
	slot := f.fid & (FIDHASHSIZE-1);
	nl: list of ref Fidrec;
	for (l := ht.table[slot]; l != nil; l = tl l)
		if ((hd l).fid != f.fid)
			nl = (hd l) :: nl;
	ht.table[slot] = nl;
}

pathsplit(p: string): (string, string)
{
	for (i := len p - 1; i >= 0; i--)
		if (p[i] != '/')
			break;
	if (i < 0)
		return (p, nil);
	p = p[0:i+1];
	for (i = len p - 1; i >=0; i--)
		if (p[i] == '/')
			break;
	if (i < 0)
		return (".", p);
	return (p[0:i+1], p[i+1:]);
}