21.3 Streams and Ports

A stream is an infinite list. Streams are important for concurrent programming. Concurrent processes may produce an infinite stream of items when running forever. Of course, only a finite part of a stream is known at every time point. Infiniteness accounts the unboundedness of its length only.

A port is a stateful data structure which gives access to a logic variable at the ``end'' of a partially known stream. A process can instantiate this variable by sending an item to the port.

declare S P
{Browse S}
P = {Port.new S}
{Browse S} 

If you enter the following statements incrementally you will observe that S gets incrementally more defined.

{Port.send P 1}
{Port.send P 2}

A good example for illustrating ports and streams is to write a procedure which merges two streams in a fair manner. This means that each of the input streams is merged into the result at some time point even if one of them is infinite.

Fair merging of streams can be easily implemented in Oz which supports fair scheduling of threads. This means that at some time point, each runnable thread will be runned for a while. This also means that running threads have to be preempted in order to give a time slot to some other runnable thread.

declare 
proc{Merge L1 L2 L}
   P={Port.new L}
   proc{Forward X}
      {Port.send P X}
   end 
in 
   thread {ForAll L1 proc{$ X} {Forward X} {Time.delay 100} end} end 
   thread {ForAll L2 proc{$ X} {Forward X} {Time.delay 200} end} end 
end

You can verify the fairness of this Merge procedure by entering the following lines of code:

declare 
L1={MakeList 1000} {ForAll L1 proc{$ X} X=a end}
L2={MakeList 1000} {ForAll L2 proc{$ X} X=b end}
 
{Browse {Merge L1 L2}}

Finally note that ports allows to programm client-server applications. A server simply posts items to the port of a stream whereas a clients read all items of a stream and accesses those that he is interested in. More about streams in concurrency can be found in chapter 8 of the Application Tutorial written by Denys Duchier, Leif Kornsteadt, and Christian Schulte, and in chapter 9 of Seif Haridi's tutorial.