

DIRECTORY
Process USING [EnableAborts], 
IO USING [Error, StreamProcs, EndOfStream, CreateStream, CreateStreamProcs, STREAM, UnsafeBlock],
IOClasses,
Basics USING [ByteBlt];

IOPipeImpl: CEDAR MONITOR
LOCKS data USING data: DataHandle
IMPORTS IO, Basics, Process
EXPORTS IOClasses = BEGIN
STREAM: TYPE = IO.STREAM;
Data: TYPE = MONITORED RECORD [
buffer: REF TEXT,
pullPos: NAT м 0,
pushPos: NAT м 0,
pullerClosed: BOOL м FALSE,
pusherClosed: BOOL м FALSE,
pipeChange: CONDITION
];
DataHandle: TYPE = REF Data;

CreatePipe: PUBLIC PROC [bufferSize: NAT] RETURNS [push, pull: STREAM] = {
ENABLE UNWIND => NULL;
nChars: NAT м SELECT bufferSize FROM
< 2 => 2, > LAST[NAT]/2 => LAST[NAT]/2, ENDCASE => bufferSize;
common: DataHandle м NEW[Data м [buffer: NEW[TEXT[nChars]]]];
common.buffer.length м nChars;
push м IO.CreateStream[pushProcs, common];
pull м IO.CreateStream[pullProcs, common];
TRUSTED {Process.EnableAborts[@common.pipeChange]};
};

PullGetChar: PROC [self: STREAM] RETURNS [c: CHAR м 0C] = {
innerGetChar: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
DO
IF data.pullerClosed THEN RETURN WITH ERROR IO.Error[StreamClosed, self];
IF data.pullPos # data.pushPos THEN EXIT;
IF data.pusherClosed THEN RETURN WITH ERROR IO.EndOfStream[self];
WAIT data.pipeChange;
ENDLOOP;
data.pullPos м data.pullPos + 1;
IF data.pullPos = data.buffer.length THEN
data.pullPos м 0;
c м data.buffer[data.pullPos];
BROADCAST data.pipeChange;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerGetChar[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

AddNAT: PROC [a, b: NAT] RETURNS [NAT] = INLINE {
RETURN [MIN[CARDINAL[a]+CARDINAL[b], NAT.LAST]];
};

PullGetBlock: PROC [self: STREAM, block: REF TEXT, startIndex: NAT, count: NAT] RETURNS [nBytesRead: NAT м 0] = {
innerGetBlock: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
pullPos: NAT м data.pullPos;
WHILE delta > 0 DO
DO
IF data.pullerClosed THEN RETURN WITH ERROR IO.Error[StreamClosed, self];
IF pullPos # data.pushPos THEN EXIT;
IF data.pusherClosed THEN RETURN;
WAIT data.pipeChange;
pullPos м data.pullPos;
ENDLOOP;
IF pullPos > data.pushPos THEN {
nBytes: NAT ~ MIN[data.buffer.maxLength - 1 - pullPos, delta];
FOR i: NAT IN (pullPos..pullPos+nBytes] DO
block[startIndex] м data.buffer[i];
startIndex м startIndex + 1;
ENDLOOP;
pullPos м pullPos + nBytes;
nBytesRead м nBytesRead + nBytes;
delta м delta - nBytes;
};
IF pullPos = data.buffer.maxLength - 1 AND delta > 0 THEN {
block[startIndex] м data.buffer[pullPos м 0];
startIndex м startIndex + 1;
nBytesRead м nBytesRead + 1;
delta м delta - 1;
};
IF pullPos < data.pushPos AND delta > 0 THEN {
nBytes: NAT ~ MIN[data.pushPos - pullPos, delta];
FOR i: NAT IN (pullPos..pullPos+nBytes] DO
block[startIndex] м data.buffer[i];
startIndex м startIndex + 1;
ENDLOOP;
pullPos м pullPos + nBytes;
nBytesRead м nBytesRead + nBytes;
delta м delta - nBytes;
};
data.pullPos м pullPos;
block.length м startIndex;
BROADCAST data.pipeChange;
ENDLOOP;
};
stopIndexPlusOne: NAT м MIN[AddNAT[startIndex, count], block.maxLength];
delta: NAT м IF stopIndexPlusOne > startIndex THEN stopIndexPlusOne - startIndex ELSE 0;
WITH self.streamData SELECT FROM
data: DataHandle => innerGetBlock[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PullCharsAvail: PROC [self: STREAM, wait: BOOL] RETURNS [diff: INT м 0] = {
innerCharsAvail: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
DO
IF data.pullerClosed THEN RETURN WITH ERROR IO.Error[StreamClosed, self];
diff м INT[data.pushPos] - INT[data.pullPos];
IF diff > 0 THEN EXIT;
IF diff < 0 THEN {diff м diff + data.buffer.maxLength; EXIT};
IF data.pusherClosed THEN {diff м 1; EXIT};
IF NOT wait THEN {diff м 0; EXIT};
WAIT data.pipeChange;
ENDLOOP;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerCharsAvail[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PullEndOf: PROC [self: STREAM] RETURNS [eof: BOOL м FALSE] = {
innerEndOf: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
eof м data.pushPos = data.pullPos AND data.pusherClosed;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerEndOf[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PullClose: PROC [self: STREAM, abort: BOOL] = {
innerClose: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
data.pullerClosed м TRUE;
IF ~abort AND data.pushPos # data.pullPos THEN
RETURN WITH ERROR IO.Error[StreamClosed, self];
data.pullPos м data.pushPos;
BROADCAST data.pipeChange;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerClose[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PushPutChar: PROC [self: STREAM, char: CHAR] = {
innerPutChar: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
IF data.pusherClosed OR data.pullerClosed THEN
RETURN WITH ERROR IO.Error[StreamClosed, self];
DO
new: NAT м data.pushPos + 1;
IF new = data.buffer.length THEN new м 0;
IF new # data.pullPos THEN {
data.pushPos м new;
data.buffer[data.pushPos] м char;
BROADCAST data.pipeChange;  
RETURN;
};
WAIT data.pipeChange;
ENDLOOP;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerPutChar[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PushPutBlock: PROC [self: STREAM, block: REF READONLY TEXT, startIndex, count: NAT] = {
innerPutBlock: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
numChars: INTEGER;
WHILE startIndex < stopIndexPlusOne DO
new: NAT м data.pushPos + 1;
IF data.pusherClosed OR data.pullerClosed THEN
RETURN WITH ERROR IO.Error[StreamClosed, self];
IF new = data.buffer.length THEN new м 0;
IF new = data.pullPos THEN {WAIT data.pipeChange; LOOP};
numChars м MIN[stopIndexPlusOne-startIndex,
IF new > data.pullPos THEN data.buffer.length-new ELSE data.pullPos-new];
FOR i: NAT IN [new..new+numChars-1] DO
data.buffer[i] м block[startIndex];
startIndex м startIndex+1;
ENDLOOP;
data.pushPos м new+numChars-1;
BROADCAST data.pipeChange;
ENDLOOP;
};
stopIndexPlusOne: NAT м AddNAT[startIndex, count];
stopIndexPlusOne м MIN[stopIndexPlusOne, block.length];
WITH self.streamData SELECT FROM
data: DataHandle => innerPutBlock[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PushUnsafePutBlock: PROC [self: STREAM, block: IO.UnsafeBlock] = {
startIndex: NAT м block.startIndex;
innerPutBlock: ENTRY PROC [data: DataHandle] = TRUSTED {
ENABLE UNWIND => NULL;
numChars: INTEGER;
bufferBase: LONG POINTER м LOOPHOLE[data.buffer, LONG POINTER] + SIZE[TEXT[0]];
WHILE startIndex < stopIndexPlusOne DO
new: NAT м data.pushPos + 1;
IF data.pusherClosed OR data.pullerClosed THEN
RETURN WITH ERROR IO.Error[StreamClosed, self];
IF new = data.buffer.length THEN new м 0;
IF new = data.pullPos THEN {WAIT data.pipeChange; LOOP};
numChars м Basics.ByteBlt[
to: [bufferBase, new, IF new > data.pullPos THEN data.buffer.length ELSE data.pullPos],
from: [block.base, startIndex, stopIndexPlusOne]
];
data.pushPos м new+numChars-1;
startIndex м startIndex+numChars;
BROADCAST data.pipeChange;
ENDLOOP;
};
stopIndexPlusOne: NAT м AddNAT[startIndex, block.count];
WITH self.streamData SELECT FROM
data: DataHandle => innerPutBlock[data];
ENDCASE => ERROR IO.Error[Failure, self];
};


PushFlush: PROC [self: STREAM] = {
innerFlush: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
WHILE data.pushPos # data.pullPos DO
IF data.pullerClosed THEN RETURN WITH ERROR IO.Error[StreamClosed, self];
WAIT data.pipeChange;
ENDLOOP;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerFlush[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PushReset: PROC [self: STREAM] = {
innerReset: ENTRY PROC [data: DataHandle] = INLINE {
data.pushPos м data.pullPos м 0;
BROADCAST data.pipeChange;
};
WITH self.streamData SELECT FROM
data: DataHandle => innerReset[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

PushClose: PROC [self: STREAM, abort: BOOL м FALSE] = {
innerClose: ENTRY PROC [data: DataHandle] = {
ENABLE UNWIND => NULL;
IF NOT data.pusherClosed THEN {
IF ~abort THEN
WHILE data.pushPos # data.pullPos AND NOT data.pullerClosed DO
WAIT data.pipeChange;
ENDLOOP;
IF data.pullerClosed AND data.pushPos # data.pullPos THEN
RETURN WITH ERROR IO.Error[StreamClosed, self];
data.pusherClosed м TRUE;
BROADCAST data.pipeChange;
};
};
WITH self.streamData SELECT FROM
data: DataHandle => innerClose[data];
ENDCASE => ERROR IO.Error[Failure, self];
};

pushProcs: REF IO.StreamProcs = IO.CreateStreamProcs[
variety: $output, class: $Pipe,
putChar: PushPutChar,
putBlock: PushPutBlock,
unsafePutBlock: PushUnsafePutBlock,
flush: PushFlush,
reset: PushReset,
close: PushClose
];

pullProcs: REF IO.StreamProcs = IO.CreateStreamProcs[
variety: $input, class: $Pipe,
getChar: PullGetChar,
getBlock: PullGetBlock,
endOf: PullEndOf,
charsAvail: PullCharsAvail,
close: PullClose
];

END.

  	└  IOPipeImpl.mesa
Copyright ╙ 1985, 1986, 1990, 1991 by Xerox Corporation.  All rights reserved.
Russ Atkinson (RRA) January 22, 1986 8:50:55 pm PST
Mike Spreitzer February 27, 1987 1:32:36 pm PST
Carl Hauser, December 7, 1988 3:19:19 pm PST
JKF, September 12, 1990 3:06:01 pm PDT
Circular buffer that holds the characters in the middle of the pipe.
Position of the puller's (reader's) cursor into the buffer. The puller increments pullPos and then grabs that character.
Position of the pusher's (writer's) cursor into the buffer.  The pusher increments pushPos and then deposits the character there.  The pipe is empty if pullPos = pushPos.
TRUE iff the puller (reader) has closed the pipe.
TRUE iff the pusher (writer) has closed the pipe.
Condition raised if the state of the pipe has changed in any way.
Creates a pipe and returns two streams that each access the ends of the pipe.  The push stream is used to add data to the pipe; the pull stream is used to remove it.
Returns the next character in the pipe, blocking until a character is available.
Raises IO.EndOfStream ERROR if the pusher has closed the pipe.  Raises IO.StreamClosed ERROR if the stream has already been closed.
Reads a block from the pipe, returning the number of characters actually read.
What should the approach be?  For example, if there are 10 characters in the pipe, and the guy wants 50, do we give him the 10 or wait for the 50? IO.mesa and IODoc.Tioga clearly say to wait for the 50, so that's what we do.
The first run of bytes to get lie between pullPos and data.buffer.maxLength-1
The first byte in the buffer should be read.
There are bytes to read between pullPos and data.pushPos-1
Returns TRUE iff the pusher has closed his end of the stream.
Closes the puller's end of the stream.  
Raises ERROR IO.Error[StreamClosed ..] if there is something in the pipe (unless abort if true).
To avoid problems if/when the pusher closes
Pushes a character onto the pipe, waiting until space is available in the buffer to hold it.
Raises ERROR IO.Error[StreamClosed..] if the pusher has already closed the stream.  Raises the same error if the puller has closed the stream and broken the pipe.
There is room to place the character
Clears out the pipe.
Closes the pusher's end of the pipe.  Flushes out the pipe iff ~abort.
Raises IO.Error[StreamClosed..] if the puller has closed the stream and there is still something in it.
Carl Hauser, November 19, 1987 6:16:54 pm PST
Added PushUnsafePutBlock., IOPipeImpl


Э╩:  ХNewlineDelimiter
Ц(cedarcode) styleШ codeЪЬЩKЪЬ
╧eЬCЩNKЩ3KЩ/KЩ,KЩ&ЧKШ Ъ╧k	Ш	KЪЬЮЬШKЪЮЬЮЬDЮЬШaKШ
KШЧKШ Ъ╧b
ЬЮЬЮШKЪЮЬЮЬШ!KЪЮЬЮЬШKЪЮЬЮШKЪЮЬЮЬЮЬЮЬШЪЬЮЬЮ	ЬЮЬШЪЬЮЬЮЬШKЪЬDЩDЧЪЬ	ЮЬШKЪЬxЩxЧЪЬ	ЮЬШKЪЬкЩкЧЪЬЮЬЮЬШKЪЬ1Щ1ЧЪЬЮЬЮЬШKЪЬ1Щ1ЧЪЬЮ	ШKЪЬAЩAЧKЪЬШЧKЪЬЮЬЮЬШЧKШ Ъ╧n
ЬЮЬЮЬЮЬЮЬЮЬШJKЪЬеЩеKЪЮЬЮЬЮЬШЪЬЮЬЮЬЮШ$KЪЬЮЬЮЬЮЬЮЬЮЬШ>ЧKЪЬЮЬЮЬЮЬШ=KЪЬШKЪЬЮЬ!Ш*KЪЬЮЬ!Ш*KЪЮЬ,Ш3KШKШ ЧЪ
аЬЮЬЮЬЮЬЮЬ
Ш;KЪЬPЩPKЪЬГЩГЪЬЮЬЮЬШ/KЪЮЬЮЬЮЬШЪЮШKЪЮЬЮЬЮЬЮЬЮЬЮЬШIKЪЮЬЮЬЮЬШ)KЪЮЬЮЬЮЬЮЬЮЬЮЬШAKЪЮЬШKЪЮЬШЧKШ ЪЮЬ#ЮШ)KШЧKШKЪЮ	ЬШKШЧЪЮЬЮЬЮШ KЪЬ'Ш'KЪЮЬЮЬЮЬШ)ЧKШЧKШ ЪаЬЮЬЮЬЮЬЮЬЮЬШ1KЪЮЬЮЬЮЬЮЬЮЬЮЬШ0KШKШ ЧЪаЬЮЬЮЬ	ЮЬЮЬЮЬ	ЮЬЮЬЮЬ	ШqKЪЬNЩNKЪЬрЩрЪЬЮЬЮЬШ0KЪЮЬЮЬЮЬШKЪЬ	ЮЬШЪЮЬЮШЪЮШKЪЮЬЮЬЮЬЮЬЮЬЮЬШIKЪЮЬЮЬЮЬШ$KЪЮЬЮЬЮЬШ!KЪЮЬШKЪЬШKЪЮЬШЧЪЮЬЮЬШ KЪЬMЩMKЪЬЮЬЮЬ-Ш>ЪЮЬЮЬЮЬЮШ*KШ#KШKЪЮЬШЧKЪЬШKЪЬ!Ш!KЪЬШKШЧЪЮЬ%ЮЬЮЬШ;KЩ,KЪЬ-Ш-KШKЪЬШKЪЬШKШЧЪЮЬЮЬЮЬШ.KЪЬ:Щ:KЪЬЮЬЮЬ Ш1ЪЮЬЮЬЮЬЮШ*KШ#KШKЪЮЬШЧKЪЬШKЪЬ!Ш!KЪЬШKЪЬШЧKШKШKЪЮ	ЬШKЪЮЬШЧKШЧKЪЬЮЬЮЬ"╧r	ЬШHKЪ	ЬЮЬЮЬЮЬЮЬШXЪЮЬЮЬЮШ KЪЬ(Ш(KЪЮЬЮЬЮЬШ)ЧKЪЬШKШ ЧЪаЬЮЬЮЬЮЬЮЬЮЬ	ШKЪЬЮЬЮЬШ2KЪЮЬЮЬЮЬШЪЮШKЪЮЬЮЬЮЬЮЬЮЬЮЬШIKЪЬЮЬЮЬШ-KЪЮЬ
ЮЬЮЬШKЪЮЬ
ЮЬ'ЮЬШ=KЪЮЬЮЬЮЬШ+KЪЮЬЮЬЮЬЮЬШ"KЪЮЬШKЪЮЬШЧKШЧЪЮЬЮЬЮШ KЪЬ*Ш*KЪЮЬЮЬЮЬШ)ЧKШKШ ЧЪа	ЬЮЬЮЬЮЬЮЬЮЬШ>KЪЬ=Щ=ЪЬЮЬЮЬШ-KЪЮЬЮЬЮЬШKЪЬ"ЮЬШ8KШЧЪЮЬЮЬЮШ KЪЬ%Ш%KЪЮЬЮЬЮЬШ)ЧKШKШ ЧЪа	ЬЮЬЮЬ	ЮЬШ/KЪЬ(Щ(KЪЬ`Щ`ЪЬЮЬЮЬШ-KЪЮЬЮЬЮЬШKЪЬЮЬШЪЮЬЮЬЮШ.KЪЮЬЮЬЮЬЮЬШ/ЧЪЬШKЪЬ+Щ+ЧKЪЮ	ЬШKШЧЪЮЬЮЬЮШ KЪЬ%Ш%KЪЮЬЮЬЮЬШ)ЧKШЧKШ ЪаЬЮЬЮЬЮЬШ0KЪЬ\Щ\KЪЬвЩвЪЬЮЬЮЬШ/KЪЮЬЮЬЮЬШЪЮЬЮЬЮШ.KЪЮЬЮЬЮЬЮЬШ/ЧЪЮШKЪЬЮЬШKЪЮЬЮЬ	Ш)ЪЮЬЮЬШKЪЬ$Щ$KЪЬШKШ!KЪЮ	ЬШKЪЮЬШKЪЬШЧKЪЮЬШKЪЮЬШЧKШЧЪЮЬЮЬЮШ KЪЬ'Ш'KЪЮЬЮЬЮЬШ)ЧKШKШ ЧЪаЬЮЬЮЬ	ЮЬЮЬЮЬЮЬШWЪЬЮЬЮЬШ0KЪЮЬЮЬЮЬШKЪЬ
ЮЬШЪЮЬЮШ&KЪЬЮЬШЪЮЬЮЬЮШ.KЪЮЬЮЬЮЬЮЬШ/ЧKЪЮЬЮЬ	Ш)KЪЮЬЮЬЮЬЮЬШ8ЪЬЮЬШ+KЪЮЬЮЬЮЬШIЧЪЮЬЮЬЮЬЮШ&KШ#KШKЪЮЬШЧKЪЬШKЪЮ	ЬШKЪЮЬШЧKШЧKЪЬЮЬШ2KЪЬЮЬ!Ш7ЪЮЬЮЬЮШ KЪЬ(Ш(KЪЮЬЮЬЮЬШ)ЧKШKШ ЧЪаЬЮЬЮЬШBKЪЬЮЬШ#ЪЬЮЬЮЬЮЬШ8KЪЮЬЮЬЮЬШKЪЬ
ЮЬШKЪЬЮЬЮЬЮЬЮЬЮЬЮЬЮЬШOЪЮЬЮШ&KЪЬЮЬШЪЮЬЮЬЮШ.KЪЮЬЮЬЮЬЮЬШ/ЧKЪЮЬЮЬ	Ш)KЪЮЬЮЬЮЬЮЬШ8ЪЬШKЪЬЮЬЮЬЮЬШWKЪЬ0Ш0KЪЬШЧKЪЬШKЪЬ!Ш!KЪЮ	ЬШKЪЮЬШЧKШЧKЪЬЮЬ#Ш8ЪЮЬЮЬЮШ KЪЬ(Ш(KЪЮЬЮЬЮЬШ)ЧKШKШ KШ ЧЪа	ЬЮЬЮЬШ"ЪЬЮЬЮЬШ-KЪЮЬЮЬЮЬШЪЮЬЮШ$KЪЮЬЮЬЮЬЮЬЮЬЮЬШIKЪЮЬШKЪЮЬШЧKШЧЪЮЬЮЬЮШ KЪЬ%Ш%KЪЮЬЮЬЮЬШ)ЧKШKШ ЧЪа	ЬЮЬЮЬШ"KЪЬЩЪЬЮЬЮЬЮЬШ4KШ KЪЮ	ЬШKШЧЪЮЬЮЬЮШ KЪЬ%Ш%KЪЮЬЮЬЮЬШ)ЧKШKШ ЧЪ
а	ЬЮЬЮЬ	ЮЬЮЬШ7KЪЬFЩFKЪЬgЩgЪЬЮЬЮЬШ-KЪЮЬЮЬЮЬШЪЮЬЮЬЮЬШЪЮЬЮШЪЮЬЮЬЮЬЮШ>KЪЮЬШKЪЮЬШЧЧЪЮЬЮЬЮШ9KЪЮЬЮЬЮЬЮЬШ/ЧKЪЬЮЬШKЪЮ	ЬШKШЧKШЧЪЮЬЮЬЮШ KЪЬ%Ш%KЪЮЬЮЬЮЬШ)ЧKШЧKШ ЪЬЮЬЮЬЮЬШ5KШKЪЬШKШKШ#KЪЬШKЪЬШKЪЬШKШKШ ЧЪЬЮЬЮЬЮЬШ5KШKЪЬШKШKЪЬШKЪЬШKЪЬШKЪЬШЧKШ KЪЮЬШKШ Щ-KЪЬбЩ%ЧKЩ KЩ Ч ЕЧ    "  :  