DIRECTORY IO USING [Error, StreamProcs, EndOfStream, CreateStream, CreateStreamProcs, STREAM], IOClasses; IOPipeImpl: CEDAR MONITOR IMPORTS IO EXPORTS IOClasses = BEGIN STREAM: TYPE = IO.STREAM; Data: TYPE = RECORD [ buffer: REF TEXT, pullPos: NAT _ 0, pushPos: NAT _ 0, pullerClosed: BOOL _ FALSE, pusherClosed: BOOL _ FALSE, pipeChange: CONDITION ]; DataHandle: TYPE = REF Data; CreatePipe: PUBLIC ENTRY PROC [bufferSize: NAT] RETURNS [push, pull: STREAM] = { ENABLE UNWIND => NULL; common: DataHandle _ NEW[Data _ [buffer: NEW[TEXT[bufferSize]]]]; common.buffer.length _ common.buffer.maxLength; push _ IO.CreateStream[ pushProcs, common ]; pull _ IO.CreateStream[ pullProcs, common ]; }; PullGetChar: ENTRY PROC [self: STREAM] RETURNS [c: CHAR] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; DO IF data.pullerClosed THEN ERROR IO.Error[StreamClosed, self]; IF data.pullPos # data.pushPos THEN EXIT; IF data.pusherClosed THEN 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; }; AddNAT: PROC [a, b: NAT] RETURNS [NAT] = INLINE { RETURN [MIN[CARDINAL[a]+CARDINAL[b], NAT.LAST]]; }; PullGetBlock: ENTRY PROC [self: STREAM, block: REF TEXT, startIndex: NAT, count: NAT] RETURNS [nBytesRead: NAT] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; stopIndexPlusOne: NAT _ AddNAT[startIndex, count]; stopIndexPlusOne _ MIN[ stopIndexPlusOne, block.length ]; DO IF data.pullerClosed THEN ERROR IO.Error[StreamClosed, self]; IF data.pullPos # data.pushPos THEN EXIT; IF data.pusherClosed THEN ERROR IO.EndOfStream[self]; WAIT data.pipeChange; ENDLOOP; IF data.pullPos < data.pushPos THEN { nBytesRead _ MIN[ data.pushPos - data.pullPos, stopIndexPlusOne - startIndex ]; FOR i: INT IN (data.pullPos..data.pullPos+nBytesRead] DO block[startIndex] _ data.buffer[i]; startIndex _ startIndex + 1; ENDLOOP; data.pullPos _ data.pullPos + nBytesRead; } ELSE { nBytesRead _ MIN[ data.buffer.maxLength - 1 - data.pullPos, stopIndexPlusOne - startIndex ]; FOR i: INT IN (data.pullPos..data.pullPos+nBytesRead] DO block[startIndex] _ data.buffer[i]; startIndex _ startIndex + 1; ENDLOOP; data.pullPos _ data.pullPos + nBytesRead; IF startIndex < stopIndexPlusOne THEN { data.pullPos _ MIN[ data.pushPos, stopIndexPlusOne - startIndex - 1]; nBytesRead _ nBytesRead + data.pullPos + 1; FOR i: INT IN [0..data.pullPos] DO block[startIndex] _ data.buffer[i]; startIndex _ startIndex + 1; ENDLOOP; }; }; BROADCAST data.pipeChange; }; PullCharsAvail: ENTRY PROC [self: STREAM, wait: BOOL] RETURNS [INT] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; diff: INT; DO IF data.pullerClosed THEN ERROR IO.Error[StreamClosed, self]; diff _ LONG[data.pushPos] - LONG[data.pullPos]; IF diff > 0 THEN RETURN [diff]; IF diff < 0 THEN RETURN [LONG[data.buffer.maxLength] + diff]; IF data.pusherClosed THEN RETURN [1]; IF NOT wait THEN RETURN [0]; WAIT data.pipeChange; ENDLOOP; }; PullEndOf: ENTRY PROC [self: STREAM] RETURNS [eof: BOOL] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; RETURN [data.pushPos = data.pullPos AND data.pusherClosed]; }; PullClose: ENTRY PROC [self: STREAM, abort: BOOL] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; data.pullerClosed _ TRUE; IF data.pushPos # data.pullPos AND ~abort THEN ERROR IO.Error[StreamClosed, self]; BROADCAST data.pipeChange; }; PushPutChar: ENTRY PROC [self: STREAM, char: CHAR] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; IF data.pusherClosed THEN ERROR IO.Error[StreamClosed, self]; IF data.pullerClosed THEN ERROR IO.Error[StreamClosed, self]; WHILE TRUE DO data.pushPos _ data.pushPos + 1; IF data.pushPos = data.buffer.length THEN data.pushPos _ 0; IF data.pushPos # data.pullPos THEN EXIT; IF data.pushPos = 0 THEN data.pushPos _ data.buffer.length; data.pushPos _ data.pushPos - 1; WAIT data.pipeChange; ENDLOOP; data.buffer[data.pushPos] _ char; BROADCAST data.pipeChange; }; PushPutBlock: ENTRY PROC[self: STREAM, block: REF READONLY TEXT, startIndex: NAT, count: NAT] = { ENABLE UNWIND => NULL; WaitForSpace: INTERNAL PROC [] = { WHILE data.pushPos + 1 = data.pullPos OR (data.pullPos = 0 AND data.pushPos = data.buffer.length - 1) DO WAIT data.pipeChange; ENDLOOP; }; data: DataHandle _ NARROW[self.streamData]; numChars: INT; stopIndexPlusOne: NAT _ AddNAT[startIndex, count]; stopIndexPlusOne _ MIN[ stopIndexPlusOne, block.length ]; WHILE startIndex < stopIndexPlusOne DO IF data.pusherClosed THEN ERROR IO.Error[StreamClosed, self]; IF data.pullerClosed THEN ERROR IO.Error[StreamClosed, self]; WaitForSpace[]; data.pushPos _ data.pushPos + 1; IF data.pushPos = data.buffer.length THEN data.pushPos _ 0; IF data.pushPos > data.pullPos THEN numChars _ MIN[ data.buffer.length-data.pushPos, stopIndexPlusOne-startIndex ] ELSE numChars _ MIN[ data.pullPos-data.pushPos, stopIndexPlusOne-startIndex ]; FOR i: INT IN [data.pushPos..data.pushPos+numChars-1] DO data.buffer[i] _ block[startIndex]; startIndex _ startIndex+1; ENDLOOP; data.pushPos _ data.pushPos+numChars-1; ENDLOOP; BROADCAST data.pipeChange; }; PushFlush: ENTRY PROC [self: STREAM] = { ENABLE UNWIND => NULL; PushFlusher[self]; }; PushFlusher: INTERNAL PROC [self: STREAM] = { data: DataHandle _ NARROW[self.streamData]; WHILE data.pushPos # data.pullPos DO IF data.pullerClosed THEN ERROR IO.Error[StreamClosed, self]; WAIT data.pipeChange; ENDLOOP; BROADCAST data.pipeChange; }; PushReset: ENTRY PROC [self: STREAM] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; data.pushPos _ data.pullPos _ 0; BROADCAST data.pipeChange; }; PushClose: ENTRY PROC [self: STREAM, abort: BOOL _ FALSE] = { ENABLE UNWIND => NULL; data: DataHandle _ NARROW[self.streamData]; IF data.pullerClosed AND data.pushPos # data.pullPos THEN ERROR IO.Error[StreamClosed, self]; IF ~abort THEN PushFlusher[self]; data.pusherClosed _ TRUE; BROADCAST data.pipeChange; }; pushProcs: REF IO.StreamProcs = IO.CreateStreamProcs[ variety: $output, class: $Pipe, putChar: PushPutChar, putBlock: PushPutBlock, 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. CHANGE LOG Created by Nix on September 9, 1983 3:04 pm Changed by MBrown on October 25, 1983 4:52 pm Converted to Cedar 5.0; exports IOClasses.CreatePipe instead of Pipe.Open. Did minimal conversions of block operations; these should be reworked to use count instead of stopIndexPlusOne. Did substantial conversion of PullerCharsAvail. Bug in PullEndOf: returned true with stuff in buffer. BIOPipeImpl.mesa -- Last Edited by -- Nix on September 13, 1983 4:58 pm -- MBrown on January 13, 1984 2:21 pm 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? I've decided to give him the 10. 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). 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. Waits until the reader has read all of the characters that are currently in the pipe. Raises IO.Error[StreamClosed..] if the puller closes the file while there's stuff in the pipe. 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. Э╩ ШJЪЬlЩlJШЪ╧k Ш JЪЭЬЭЬDЭЬШTJШ ЧJШЪ╧b ЬЭЬЭШJЪЭЬЭШ JЪЭЬШЧJЪЭШJЪЭЬЭЬЭЬЭЬШЪЬЭЬЭЬШЪЬЭЬЭЬШJЪЬDЩDЧЪЬ ЭЬШJЪЬxЩxЧЪЬ ЭЬШJЪЬкЩкЧЪЬЭЬЭЬШJЪЬ1Щ1ЧЪЬЭЬЭЬШJЪЬ1Щ1ЧЪЬЭ ШJЪЬAЩAЧJЪЬШЧJЪЬЭЬЭЬШIunitШЪ╧n ЬЭЬЭЬЭЬЭЬЭЬЭЬШPJЪЬеЩеJЪЭЬЭЬЭЬШJЪЬЭЬЭЬЭЬШAJШ/JЪЬЭЬ#Ш,JЪЬЭЬ#Ш,JШЧЪЯЬЭЬЭЬЭЬЭЬЭЬШ=JЪЬPЩPJЪЬГЩГJЪЭЬЭЬЭЬШJЪЬЭЬШ+ЪЭШJЪЭЬЭЬЭЬЭЬШ=JЪЭЬЭЬЭЬШ)JЪЭЬЭЬЭЬЭЬШ5JЪЭЬШJЪЭЬШЧJШ ЪЭЬ#ЭШ)JШЧJШJЪЭ ЬШJШЧJШЪЯЬЭЬЭЬЭЬЭЬЭЬШ1JЪЭЬЭЬЭЬЭЬЭЬЭЬШ0JШJШЧЪЯЬЭЬЭЬЭЬ ЭЬЭЬЭЬ ЭЬШUIcodeЪЭЬ ЭЬШJЪЬNЩNJЪЬ│Щ│LЪЭЬЭЬЭЬШLЪЬЭЬШ+LЪЬЭЬШ2LЪЬЭЬ#Ш9ЪЭШLЪЭЬЭЬЭЬЭЬШ=LЪЭЬЭЬЭЬШ)LЪЭЬЭЬЭЬЭЬШ5LЪЭЬШLЪЭЬШЧЪЭЬЭЬШ%LЪЬ ЭЬ?ШOЪЭЬЭЬЭЬ)ЭШ8LШ#LШLЪЭЬШЧLШ)LШЧЪЭЬШLЪЬ ЭЬLШ\ЪЭЬЭЬЭЬ)ЭШ8LШ#LШLЪЭЬШЧLШ)ЪЭЬЭЬШ'LЪЬЭЬ3ШELШ+ЪЭЬЭЬЭЬЭШ"LШ#LШLЪЭЬШЧLШЧLЪЬШЧLЪЭ ЬШLЪЬШЧЪЯЬЭЬЭЬЭЬЭЬЭЬЭЬШHJЪЭЬЭЬЭЬШJЪЬЭЬШ+JЪЬЭЬШ ЪЭШJЪЭЬЭЬЭЬЭЬШ=JЪЬЭЬЭЬШ/JЪЭЬ ЭЬЭЬШJЪЭЬ ЭЬЭЬ Ш=JЪЭЬЭЬЭЬШ%JЪЬЭЬЭЬЭЬШJЪЭЬШJЪЭЬШЧJШЧЪЯ ЬЭЬЭЬЭЬЭЬЭЬШ=JЪЬ=Щ=JЪЭЬЭЬЭЬШJЪЬЭЬШ+JЪЭЬЭЬШ;JШЧЪ Я ЬЭЬЭЬЭЬ ЭЬШ6JЪЬ(Щ(JЪЬ`Щ`JЪЭЬЭЬЭЬШJЪЬЭЬШ+JЪЬЭЬШJЪ ЭЬЭЬЭЬЭЬЭЬШRJЪЭ ЬШJШJШЧJШЪ ЯЬЭЬЭЬЭЬЭЬШ7JЪЬ\Щ\JЪЬвЩвJЪЭЬЭЬЭЬШJЪЬЭЬШ+JЪЭЬЭЬЭЬЭЬШ=JЪЭЬЭЬЭЬЭЬШ=ЪЭЬЭЬЭШ JШ JЪЭЬ#ЭЬШ;JЪЭЬЭЬЭЬШ)JЪЭЬЭЬ#Ш;JШ JЪЭЬШJЪЭЬШЧJШ!JЪЭ ЬШJШЧЪЯЬЭЬЭЬЭЬ ЭЬЭЬЭЬЭЬ ЭЬШaJЪЭЬЭЬЭЬШЪЯЬЭ ЬШ"ЪЭЬ!ЭЬЭЬ(ЭШhJЪЭЬШJЪЭЬШЧJШЧJЪЬЭЬШ+JЪЬ ЭЬШJЪЬЭЬШ2JЪЬЭЬ#Ш9ЪЭЬЭШ&JЪЭЬЭЬЭЬЭЬШ=JЪЭЬЭЬЭЬЭЬШ=JШJШ JЪЭЬ#ЭЬШ;ЪЭЬЭЬШ$JЪЬЭЬ@ШNЧЪЭШJЪЬЭЬ;ШIЧЪЭЬЭЬЭЬ)ЭШ8JШ#JШJЪЭЬШЧJЪЬ'Ш'JЪЭЬШЧJЪЭ ЬШJШЧЪЯ ЬЭЬЭЬЭЬШ(JЪЭЬЭЬЭЬШJЪЬШJШЧЪЯЬЭЬЭЬЭЬШ-JЪЬUЩUJЪЬ^Щ^JЪЬЭЬШ+ЪЭЬЭЬШ%JЪЭЬЭЬЭЬЭЬШ=JЪЭЬШJЪЭЬШЧJЪЭ ЬШJШЧЪЯ ЬЭЬЭЬЭЬШ(JЪЬЩJЪЭЬЭЬЭЬШJЪЬЭЬШ+JШ JЪЭ ЬШJШЧЪЯ ЬЭЬЭЬЭЬ ЭЬЭЬШ>JЪЬFЩFJЪЬgЩgJЪЭЬЭЬЭЬШJЪЬЭЬШ+JЪ ЭЬЭЬЭЬЭЬЭЬШ]JЪЭЬЭЬШ!JЪЬЭЬШJЪЭ ЬШJШЧKШЪЬЭЬЭЬЭЬШ5JШJЪЬШJШJЪЬШJЪЬШJЪЬШJШЧЪЬЭЬЭЬЭЬШ5JШJЪЬШJШJЪЬШJЪЬШJЪЬШJЪЬШЧJШJЪЭЬШJШJЪЭЬЭШ JШJЪЬ,Ш,ЪЬ-Ш-JЪЬгШгЧЧЕЧ┌-<