[mercury-users] Native garbage collector for Mercury

[Thomas Charles CONWAY]

Peter Schachte, you write:
> But in this case you don't want sequentiality.  Or, looked at
> differently, you just can't express what you want logically.  Let's
> say you have a character count predicate
> 
> 	count(Buffer, Len0, Len, IO0, IO) :-
> 		(   is_empty(Buffer, IO0) ->
> 			IO = IO0,
> 			Len = Len0
> 		;   next_char(Buffer, _Char, IO0, IO1),
> 		    count(Buffer, Len0+1, Len, IO1, IO)
> 		).
> 
> Remember, the buffer is being written concurrently (maybe it's a
> pipe).  Suppose that is_empty/2 and next_char/4 are implemented to
> look in the io__state and do the appropriate thing if the buffer has
> characters in it or if it's empty and has been closed.  If someone is
> still writing to it but we've emptied the buffer, then we suspend.
> 
> So how do we write a call to count/5 that says that the buffer is to
> be filled by a different process?  And how to we write the interrupt
> handler (in Mercury, no destructive C hacks please) to modify the
> io__state that count/5 is using, possibly waking count/5 if it's
> suspended waiting for input?
> 

First your count code is illegal (since the final io__state from main
must be `uo' not `muo'): the call to is_empty must be deterministic.
Try code like:

	:- pred count(stream(char), int, int, io__state, io__state).
	:- mode count(in, in, out, di, uo) is det.

 	count(Buffer, Len0, Len, IO0, IO) :-
			% for the sake of simplicity, we'll assume
			% an eof marker.
 		get_item(Buffer, Char, IO0, IO1),
 		(   Char = eof ->
 			IO = IO1,
 			Len = Len0
 		    count(Buffer, Len0+1, Len, IO1, IO)
 		).

So, what is this magic thing called `stream(T)' ?

Roughly, I have a module stream.m:

	:- module stream.

	:- interface.

	:- import_module io.

	:- type stream(T).

	:- pred init_stream(stream(T), io__state, io__state).
	:- mode init_stream(out, di, uo) is det.

	:- pred put_item(stream(T), T, io__state, io__state).
	:- mode put_item(in, in, di, uo) is det.

	:- pred get_item(stream(T), T, io__state, io__state).
	:- mode get_item(in, out, di, uo) is det.

	:- implementation.

	:- import_module mutex, references, semaphore.
	:- import_module queue.

	:- type stream(T)
		--->	stream(
				semaphore,
				mutex,
				ref(queue(T))
			).


	init_stream(stream(Sem, Lock, QRef)) -->
		init_semaphore(Sem),
		init_mutex(Lock),
		{ queue__init(Q) },
		new_ref(Q, QRef).

	put_item(stream(Sem, Lock, QRef), Item) -->
		lock(Lock),
		deref(QRef, Q0),
		{ queue__put(Q0, Item, Q1) },
		setref(QRef, Q1),
		unlock(Lock),
		signal(Sem).
		
	get_item(stream(Sem, Lock, QRef), Item) -->
		wait(Sem),
		lock(Lock),
		deref(QRef, Q0),
		( { queue__get(Q0, Item0, Q1) } ->
			{ Item = Item0 },
			setref(QRef, Q1),
			unlock(Lock)
		;
			{ error("queue underflow!") }
		).

The interface of references.m looks something like:

	:- module references.

	:- interface.

	:- import_module io.

	:- type ref(T).

	:- pred new_ref(T, ref(T), io__state, io__state).
	:- mode new_ref(in, out, di, uo) is det.

	:- pred deref(ref(T), T, io__state, io__state).
	:- mode deref(in, out, di, uo) is det.

	:- pred setref(ref(T), T, io__state, io__state).
	:- mode setref(in, in, di, uo) is det.

We can implement this in a variety of ways, but the most efficient is
to use C to implement ref(T) as a pointer to a cell containing a pointer
to the data (this data structure is sometimes known as a handle).
However, if you insist on hobbling me so I can't use C, you can use the
user data field of the io__state to hold a map from ref(T) (which we'll
just make an int) to univ. The semantics are the same, since the io__state
itself is stored in a global variable.

As for lock/unlock and signal/wait, I have code for those too (the former
is implemented in terms of the latter). Semaphores are implemented in C
in cooperation with the runtime which handles the suspension and resumption
themselves.
(For locals, have a look in /home/hydra/conway/mercury/extras/concurrency)

Thomas
-- 
Thomas Conway <conway at cs.mu.oz.au>
Nail here [] for new monitor.  )O+