<<>> <> <> <> <> DIRECTORY Basics USING [BITAND, BITLSHIFT, BITOR, BITRSHIFT, ByteBlt, FillBytes, RawBytes, RawChars], CStrings USING [CString], IO USING [CreateStream, CreateStreamProcs, EndOfStream, PutFR, PutFR1, STREAM, StreamProcs, UnsafeBlock], IOUtils USING [AmbushStream, closedStreamProcs], MentatInterface USING [maxXNSPacketSize, xnsSPPDevice], Process USING [Detach, EnableAborts, MsecToTicks, SetTimeout, Ticks], Rope USING [Concat, ROPE], RuntimeError USING [BoundsFault], TLI USING [CONNECT, ConnectionData, ConnectionDataRep, DATA, DISCONNECT, EXDATA, EXPEDITED, LISTEN, MORE, ORDREL, States, TBind, TBindCall, TCall, TClose, TConnect, TERROR, TGetState, TLook, TOpen, TRcv, TRcvRel, TSnd, TSndDis, TSndRel, UDERR ], UXStrings USING [Create], UnixErrno USING [Errno, GetErrno], UnixTypes USING [CHARPtr], XNS USING [Address, IsMulticastHost], XNSSPPTypes USING [defaultSST, SubSequenceType], XNSStream USING [AttentionType, CloseReason, dontWait, Milliseconds, State, waitForever]; XNSStreamImpl: CEDAR MONITOR LOCKS handle USING handle: Handle IMPORTS Basics, IO, IOUtils, Process, Rope, RuntimeError, TLI, UXStrings, UnixErrno, XNS EXPORTS XNSStream ~ BEGIN <> ROPE: TYPE ~ Rope.ROPE; Milliseconds: TYPE ~ XNSStream.Milliseconds; -- TYPE ~ CARD32; waitForever: Milliseconds ~ XNSStream.waitForever; -- 1800000 1/2 hour of milliseconds dontWait: Milliseconds ~ XNSStream.dontWait; -- 0; CloseReason: TYPE ~ XNSStream.CloseReason; State: TYPE ~ XNSStream.State; SubSequenceType: TYPE ~ XNSSPPTypes.SubSequenceType; defaultSST: SubSequenceType ~ XNSSPPTypes.defaultSST; AttentionType: TYPE ~ XNSStream.AttentionType; -- BYTE; STREAM: TYPE ~ IO.STREAM; <> <> <<>> BufferIndex: TYPE ~ [0 .. MentatInterface.maxXNSPacketSize]; BufferState: TYPE ~ { empty, emptying, filling, full }; Buffer: TYPE ~ REF BufferRep; BufferRep: TYPE ~ RECORD [ next: Buffer Ź NIL, -- the next one on the chain. This is so that we can get more buffers filled ahead. state: BufferState Ź empty, attention: BOOL Ź FALSE, -- does the buffer contain expidited data sequence: INT Ź 0, -- the inband attention sequence number. This is used to remove OOB attentions from the queue eom: BOOL Ź FALSE, -- was eom set in this packet sst: SubSequenceType Ź 0, -- the data stream type of this packet index: BufferIndex Ź 0, bytes: BufferIndex Ź 0, buffer: POINTER, -- conviently setup for us ahead of time charPTR: UnixTypes.CHARPtr, -- also set up for us bufferRef: REF TEXT, -- so we can hold onto the bytes flags: REF INT -- this is here so we dont allocate one on each call to ReadBuffer ]; OBAttn: TYPE ~ REF OBAttnObject; OBAttnObject: TYPE ~ RECORD [ next: OBAttn, sequence: INT, -- which attention packet is this. The idea is that this number is assigned by us rather than by the packet sequence number. It's used to correlate both the OOB and in band copy. attentionType: AttentionType ]; Calls: TYPE ~ REF CallsRep; CallsRep: TYPE ~ RECORD [ reqTBind: REF TLI.TBindCall, -- what socket should try to use during TBind reqTBindAddress: REF XNS.Address, -- the storage behind the reqTBind retTBind: REF TLI.TBindCall, -- what socket did we actually bind to retTBindAddress: REF XNS.Address, -- the storage behind the retTBind sendTCall: REF TLI.TCall, -- used as the requested address of the TConnect sendTCallAddress: REF XNS.Address, -- the storage behind sendTCall rcvTCall: REF TLI.TCall, -- used as the answer from TConnect rcvTCallAddress: REF XNS.Address -- the storage behind rcvTCall ]; Handle: TYPE ~ REF HandleRep; HandleRep: TYPE ~ MONITORED RECORD [ next: Handle Ź NIL, cd: TLI.ConnectionData, calls: Calls, -- the various call data structures we need local: XNS.Address Ź NULL, -- what socket are we using remote: XNS.Address Ź NULL, -- who are we connected to closed: BOOL Ź FALSE, -- for later calls to do things on this stream closeReason: CloseReason Ź unknown, closeText: ROPE Ź NIL, <> auxProcCnt: CARDINAL Ź 0, auxProcsDone: CONDITION, <> getTimeout: Milliseconds Ź 0, putTimeout: Milliseconds Ź 0, <> <> currentInputBuffer: Buffer Ź NIL, -- the buffer we are working on currentInputSST: SubSequenceType Ź defaultSST, -- what the sst is now inBuffers: Buffer, -- the input buffer chain. This contains all buffers that contain data to be passed to the client inFreeBuffers: Buffer, -- the free buffers available for input inputWaiting: BOOL Ź FALSE, -- are we currently in ReceiveBuffer inputReady: CONDITION, inputBufferFree: CONDITION, -- this indicates that there is a newly freed buffer outOfBandAttnNumber: INT Ź 0, -- this number is assigned to and then incremented whenever an out of band attention is received. This number should never be found in the recvdOBAttnList since that means we wrapped around inBandAttnNumber: INT Ź 0, -- this number is assigned to the inband copy of the attention packet. It is used to locate the oob attention to kill when this one is delivered. recvdOBAttnList: OBAttn Ź NIL, -- Out-of-Band Attentions received and not yet read by WaitAttention recvdOBAttn: CONDITION, -- wake up for WaitAttention inputSSType: SubSequenceType Ź 0, -- SubSequenceType last seen by the stream client <> currentOutputBuffer: Buffer Ź NIL, -- the buffer we are working on currentOutputSST: SubSequenceType Ź defaultSST, -- what the sst is now outputBuffers: Buffer, -- the output buffer list. These are ready for sending outputFreeBuffers: Buffer, -- the list of unused output buffers outputReady: CONDITION, outputBufferFree: CONDITION, -- indicates a buffer is available for output flusher: CONDITION, -- there is a process watching us fill up buffers. When this fires, the buffer gets sent <> spare: REF Ź NIL -- for no good reason ]; ByteArrayPtr: TYPE = LONG POINTER TO Basics.RawBytes; <> packetSize: INT ~ MentatInterface.maxXNSPacketSize; -- for now, only deal with TRcv and TSnd calls using this size buffer. This size also determines the amount of storage in the buffer.buffer part. flushTime: Process.Ticks ~ Process.MsecToTicks[500]; -- flush every 1/2 second <> ConnectionClosed: PUBLIC ERROR [why: CloseReason, text: Rope.ROPE] ~ CODE; Timeout: PUBLIC SIGNAL ~ CODE; BadSysCall: ERROR [why: ROPE] ~ CODE; LookEvent: ERROR [lookEvent: INT] ~ CODE; SystemError: ERROR ~ CODE; <> streamProcs: REF IO.StreamProcs; CreateStreamProcs: PROC ~ { streamProcs Ź IO.CreateStreamProcs [ variety: $inputOutput, class: $XNS, getChar: GetChar, getBlock: GetBlock, unsafeGetBlock: UnsafeGetBlock, endOf: EndOf, charsAvail: CharsAvail, putChar: PutChar, unsafePutBlock: UnsafePutBlock, flush: Flush, close: Close ]; }; GetChar: PROC [self: STREAM] RETURNS [CHAR] = TRUSTED { buff: PACKED ARRAY [0..3] OF CHAR; bp: ByteArrayPtr = LOOPHOLE[LONG[@buff]]; IF UnsafeGetBlock[self, [base: bp, startIndex: 0, count: 1]] = 0 THEN ERROR IO.EndOfStream[self]; RETURN[buff[0]] }; GetBlock: PROC [self: STREAM, block: REF TEXT, startIndex: NAT, count: NAT] RETURNS [nBytesRead: NAT] = TRUSTED { nBytesRead Ź UnsafeGetBlock[self, [ base: LOOPHOLE[block, ByteArrayPtr]+SIZE[TEXT[0]], startIndex: startIndex, count: MAX[MIN[INT[count], INT[block.maxLength]-startIndex], 0] ]]; block.length Ź startIndex + nBytesRead; RETURN[nBytesRead]; }; UnsafeGetBlock: UNSAFE PROC [self: STREAM, block: IO.UnsafeBlock] RETURNS [nBytesRead: INT Ź 0] ~ { handle: Handle ~ NARROW[self.streamData]; IF block.startIndex < 0 OR block.count < 0 THEN ERROR RuntimeError.BoundsFault; TRUSTED {nBytesRead Ź EntryUnsafeGetBlock[handle: handle, block: block]}; RETURN[nBytesRead]; }; EntryUnsafeGetBlock: ENTRY PROC [handle: Handle, block: IO.UnsafeBlock] RETURNS [nBytesRead: INT Ź 0] ~ { -- with all these things that change the state kept by the handle, we best be in a monitor ENABLE UNWIND => NULL; startIndex: INT Ź block.startIndex; stop: INT ~ block.startIndex+block.count; nBytes: NAT; buffer: Buffer Ź handle.currentInputBuffer; WHILE startIndex < stop DO IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; IF buffer = NIL THEN { -- we need to get a buffer buffer Ź GetFirstInputBuffer[handle: handle]; -- this will block until a packet arrives handle.currentInputBuffer Ź buffer; -- so that we know where we are. PrintF[IO.PutFR1["buffer = %g\n", [text[buffer.bufferRef]]]]; }; IF buffer.attention THEN RETURN[nBytesRead]; -- pass along any of the bytes that might have already been read into the block. If this is really the buffer from the previous call, it's ok since nBytesRead will still be 0. Let the caller worry about EndOfStream. IF buffer.sst # handle.currentInputSST THEN RETURN[nBytesRead]; -- the currentSST isn't updated until the client calls GetStatus[..., reset: TRUE] to get passed it. This means EndOfStream IF buffer.index = buffer.bytes THEN { -- this buffer is used up. Check for EOM IF buffer.eom THEN RETURN [nBytesRead] -- no more bytes in the packet and it's EOM ELSE { -- we need to get more bytes since this buffer is empty and it's not EOM FreeInputBuffer[handle: handle, buffer: buffer]; -- give back the empty one buffer Ź NIL; -- This is really cheating since the buffer = NIL check is going to fetch a fresh one for us when we go around LOOP; }; }; <> TRUSTED { nBytes Ź Basics.ByteBlt[ to: [ blockPointer: block.base, startIndex: startIndex, stopIndexPlusOne: stop], from: [ blockPointer: buffer.buffer, startIndex: buffer.index, stopIndexPlusOne: buffer.bytes] ]; }; <<>> <> buffer.index Ź buffer.index + nBytes; nBytesRead Ź nBytesRead + nBytes; startIndex Ź startIndex + nBytes; ENDLOOP; -- WHILE startIndex < stop }; EndOf: PROC [self: STREAM] RETURNS [BOOL] ~ { handle: Handle ~ NARROW[self.streamData]; RETURN [EntryEndOf[handle]]; }; EntryEndOf: ENTRY PROC [handle: Handle] RETURNS [BOOL] ~ { ENABLE UNWIND => NULL; buffer: Buffer ~ handle.currentInputBuffer; IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; IF buffer = NIL THEN RETURN[FALSE]; -- I'm not really sure what this means RETURN [ buffer.attention OR buffer.sst # handle.currentInputSST OR ((buffer.index = buffer.bytes) AND buffer.eom) ]; }; CharsAvail: PROC [self: STREAM, wait: BOOL] RETURNS [nChars: INT] ~ { handle: Handle ~ NARROW[self.streamData]; RETURN[EntryCharsAvail[handle: handle, wait: wait]]; }; EntryCharsAvail: ENTRY PROC [handle: Handle, wait: BOOL] RETURNS [INT] = { ENABLE UNWIND => NULL; DO buffer: Buffer ~ handle.currentInputBuffer; IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; IF buffer = NIL THEN { -- there isn't a buffer now IF NOT wait THEN RETURN[0]; -- there aren't any bytes WAIT handle.inputReady; LOOP; -- there should be one now }; IF buffer.attention OR buffer.sst # handle.currentInputSST THEN RETURN[LAST[INT]]; -- GetChar will raise EndOfStream IF buffer.index < buffer.bytes THEN RETURN [(buffer.bytes - buffer.index)]; IF buffer.eom THEN RETURN [LAST[INT]]; ENDLOOP; }; PutChar: PROC [self: STREAM, char: CHAR] = TRUSTED { buff: PACKED ARRAY [0..3] OF CHAR; bp: ByteArrayPtr Ź LOOPHOLE[LONG[@buff]]; buff[0] Ź char; UnsafePutBlock[self, [base: bp, startIndex: 0, count: 1]]; }; UnsafePutBlock: PROC [self: STREAM, block: IO.UnsafeBlock] ~ { handle: Handle ~ NARROW[self.streamData]; IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; IF block.startIndex < 0 OR block.count < 0 THEN ERROR RuntimeError.BoundsFault; TRUSTED {EntryUnsafePutBlock[handle: handle, block: block]}; }; EntryUnsafePutBlock: ENTRY PROC [handle: Handle, block: IO.UnsafeBlock] ~ { startIndex: INT Ź block.startIndex; stop: INT ~ block.startIndex+block.count; nBytes: NAT; buffer: Buffer Ź handle.currentOutputBuffer; WHILE startIndex < stop DO IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; IF buffer = NIL THEN { -- we need to get a buffer buffer Ź AllocateOutputBuffer[handle: handle]; -- this will block one is free handle.currentOutputBuffer Ź buffer; -- so that we know where we are. buffer.index Ź 0; -- insure that the buffer is empty before we start filling it up buffer.bytes Ź LAST[BufferIndex]; -- set the last byte position }; TRUSTED { nBytes Ź Basics.ByteBlt[ to: [ blockPointer: buffer.buffer, startIndex: buffer.bytes, stopIndexPlusOne: LAST[BufferIndex]], from: [ blockPointer: block.base, startIndex: startIndex, stopIndexPlusOne: stop] ]; }; buffer.index Ź buffer.index + nBytes; startIndex Ź startIndex + nBytes; IF buffer.index = buffer.bytes THEN { -- the buffer is full buffer.state Ź full; buffer.attention Ź FALSE; -- this isn't an attention byte buffer.eom Ź FALSE; -- not eom either buffer.sst Ź handle.currentOutputSST; -- set the SST. The currentOutputSST is changed by SetSSType. AppendOutputBuffer[handle: handle, buffer: buffer]; buffer Ź NIL; -- so that we get a fresh one LOOP; -- go around again to see if we got the whole block }; ENDLOOP; -- WHILE startIndex < stop }; Flush: PROC [self: STREAM] ~ { handle: Handle ~ NARROW[self.streamData]; IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; }; Close: PROC [self: STREAM, abort: BOOL] ~ { handle: Handle ~ NARROW[self.streamData]; OrderlyClose[handle: handle, initiate: TRUE]; EntryNotifyClosed[handle: handle, closeReason: localClose, closeText: "close called"]; CloseTheStream[self: self]; }; BlastClose: PROC [handle: Handle] = { cd: TLI.ConnectionData ~ handle.cd; [] Ź TLI.TClose[cd: cd]; }; OrderlyClose: PROC [handle: Handle, initiate: BOOL] = { cd: TLI.ConnectionData ~ handle.cd; state: TLI.States; IF initiate THEN [] Ź TLI.TSndRel[cd: cd]; state Ź TLI.TGetState[cd: cd]; SELECT state FROM inrel => [] Ź TLI.TRcvRel[cd: cd]; unbnd, idle, outrel => NULL; ENDCASE => [] Ź TLI.TSndDis[cd: cd, call: NIL]; [] Ź TLI.TClose[cd: cd]; RETURN; }; CloseTheStream: PROC [self: STREAM] = { IF self.streamProcs = IOUtils.closedStreamProcs THEN RETURN; IOUtils.AmbushStream[self: self, streamProcs: IOUtils.closedStreamProcs, streamData: self.streamData]; }; NotifyClosed: INTERNAL PROC [handle: Handle, closeReason: CloseReason, closeText: ROPE] ~ { IF NOT handle.closed THEN { handle.closeReason Ź closeReason; handle.closeText Ź closeText; handle.closed Ź TRUE; }; BROADCAST handle.inputReady; BROADCAST handle.inputBufferFree; BROADCAST handle.recvdOBAttn; BROADCAST handle.outputReady; BROADCAST handle.outputBufferFree; BROADCAST handle.flusher; }; EntryNotifyClosed: ENTRY PROC [handle: Handle, closeReason: CloseReason, closeText: ROPE] ~ { ENABLE UNWIND => NULL; NotifyClosed[handle, closeReason, closeText] }; <> createBuffers: CARDINAL Ź 6; -- ???? Create: PUBLIC PROC [remote: XNS.Address, getTimeout: Milliseconds Ź waitForever, putTimeout: Milliseconds Ź waitForever] RETURNS [IO.STREAM] = { handle: Handle Ź NIL; cd: TLI.ConnectionData ~ AllocateCD[]; calls: Calls ~ AllocateCalls[]; path: CStrings.CString ~ UXStrings.Create[MentatInterface.xnsSPPDevice]; return: INT; retTBind: REF TLI.TBindCall ~ calls.retTBind; sendCall: REF TLI.TCall ~ calls.sendTCall; rcvCall: REF TLI.TCall ~ calls.rcvTCall; calls.sendTCallAddress Ź remote; -- load up the address we want to talk to IF XNS.IsMulticastHost[remote.host] -- check to see if it's multicast THEN ERROR ConnectionClosed[noRoute, "multicast remote"]; <> <> IF TLI.TOpen[cd: cd, path: path, flags: [access: RDWR]] < 0 THEN { -- can't open the device error: ROPE ~ IO.PutFR["TOpen failed: cd.errno = %g, errno = %g", [integer[LOOPHOLE[cd.errno]]], [integer[LOOPHOLE[UnixErrno.GetErrno[]]]]]; ERROR ConnectionClosed[localClose, error]; }; IF TLI.TBind[cd: cd, request: NIL, return: retTBind] < 0 THEN { -- can't bind to any socket error: ROPE ~ IO.PutFR["TBind failed: cd.errno = %g, errno = %g", [integer[LOOPHOLE[cd.errno]]], [integer[LOOPHOLE[UnixErrno.GetErrno[]]]]]; [] Ź TLI.TClose[cd: cd]; ERROR ConnectionClosed[localClose, error]; }; return Ź TLI.TConnect[cd: cd, sndCall: sendCall, rcvCall: rcvCall]; IF return < 0 THEN { error: ROPE Ź IO.PutFR["TConnect failed: cd.errno = %g, errno = %g ", [integer[LOOPHOLE[cd.errno]]], [integer[LOOPHOLE[UnixErrno.GetErrno[]]]]]; IF cd.errno = TLOOK THEN { IsEvent: PROC [event: INT] RETURNS [BOOL] = INLINE { RETURN[ Basics.BITAND[events, event] # 0]; }; events: INT Ź TLI.TLook[cd: cd]; error Ź IO.PutFR["%g\nTLook event = %g\n", [rope[error]], [integer[events]]]; IF IsEvent[TLI.LISTEN] THEN error Ź Rope.Concat[error, "LISTEN "]; IF IsEvent[TLI.CONNECT] THEN error Ź Rope.Concat[error, "CONNECT "]; IF IsEvent[TLI.DATA] THEN error Ź Rope.Concat[error, "DATA "]; IF IsEvent[TLI.EXDATA] THEN error Ź Rope.Concat[error, "EXDATA "]; IF IsEvent[TLI.DISCONNECT] THEN { error Ź Rope.Concat[error, "DISCONNECT "]; ERROR ConnectionClosed[remoteClose, error]; }; IF IsEvent[TLI.TERROR] THEN error Ź Rope.Concat[error, "TERROR "]; IF IsEvent[TLI.UDERR] THEN error Ź Rope.Concat[error, "UDERR "]; IF IsEvent[TLI.ORDREL] THEN error Ź Rope.Concat[error, "ORDREL "]; }; [] Ź TLI.TClose[cd: cd]; ERROR ConnectionClosed[localClose, error]; }; <> handle Ź MakeHandle[sendBuffers: createBuffers, recvBuffers: createBuffers, getTimeout: getTimeout, putTimeout: putTimeout]; handle.cd Ź cd; handle.calls Ź calls; handle.remote Ź calls.rcvTCallAddress; handle.local Ź calls.retTBindAddress; RETURN [IO.CreateStream[streamProcs: streamProcs, streamData: handle]]; }; GetLocal: PUBLIC PROC [self: STREAM] RETURNS [local: XNS.Address] = { handle: Handle = NARROW[self.streamData]; RETURN [EntryGetLocal[handle: handle]]; }; EntryGetLocal: ENTRY PROC [handle: Handle] RETURNS [local: XNS.Address] = { ENABLE UNWIND => NULL; RETURN[handle.local]; }; GetRemote: PUBLIC PROC [self: STREAM] RETURNS [remote: XNS.Address] = { handle: Handle = NARROW[self.streamData]; RETURN [EntryGetRemote[handle: handle]]; }; EntryGetRemote: ENTRY PROC [handle: Handle] RETURNS [local: XNS.Address] = { ENABLE UNWIND => NULL; RETURN[handle.remote]; }; GetTimeouts: PUBLIC PROC [self: STREAM] RETURNS [getTimeout, putTimeout: Milliseconds Ź 0] = { <> }; SetTimeouts: PUBLIC PROC [self: STREAM, getTimeout, putTimeout: Milliseconds Ź waitForever] = { <> }; SetSSType: PUBLIC PROC [self: STREAM, ssType: SubSequenceType] = { handle: Handle ~ NARROW[self.streamData]; EntrySetSSType[handle: handle, ssType: ssType]; }; EntrySetSSType: ENTRY PROC [handle: Handle, ssType: SubSequenceType] = { ENABLE UNWIND => NULL; buffer: Buffer Ź handle.currentOutputBuffer; IF buffer # NIL THEN { -- we aren't at a boundry. We need to force one buffer.state Ź full; buffer.attention Ź FALSE; -- this isn't an attention byte buffer.eom Ź FALSE; -- not eom either buffer.sst Ź handle.currentOutputSST; -- set the SST. The currentOutputSST is changed by SetSSType. AppendOutputBuffer[handle: handle, buffer: buffer]; handle.currentOutputSST Ź ssType; -- set the SST for all new packets RETURN; -- we did what we needed to }; <> handle.currentOutputSST Ź ssType; -- set the SST for all new packets }; SendEndOfMessage: PUBLIC PROC [self: STREAM] = { handle: Handle ~ NARROW[self.streamData]; EntrySendEndOfMessage[handle: handle]; }; EntrySendEndOfMessage: ENTRY PROC [handle: Handle] = { ENABLE UNWIND => NULL; buffer: Buffer Ź handle.currentOutputBuffer; IF buffer # NIL THEN { -- we aren't at a boundry. We need to force one buffer.state Ź full; buffer.attention Ź FALSE; -- this isn't an attention byte buffer.eom Ź TRUE; -- not eom either buffer.sst Ź handle.currentOutputSST; -- set the SST. The currentOutputSST is changed by SetSSType. AppendOutputBuffer[handle: handle, buffer: buffer]; handle.currentOutputBuffer Ź NIL; -- we have set eom and forced the buffer onto the queue. RETURN; -- we did what we needed to }; <> buffer Ź AllocateOutputBuffer[handle: handle]; -- this will block one is free buffer.index Ź 0; -- insure that the buffer is empty before we start filling it up buffer.bytes Ź 0; buffer.state Ź full; buffer.attention Ź FALSE; -- this isn't an attention byte buffer.eom Ź TRUE; buffer.sst Ź handle.currentOutputSST; AppendOutputBuffer[handle: handle, buffer: buffer]; handle.currentOutputBuffer Ź NIL; }; SendAttention: PUBLIC PROC [self: STREAM, attentionType: AttentionType] = { <> }; SendClose: PUBLIC PROC [self: STREAM] RETURNS [ok: BOOL Ź FALSE] = { <> }; SendCloseReply: PUBLIC PROC [self: STREAM] RETURNS [ok: BOOL Ź FALSE] = { <> }; SendNow: PUBLIC PROC [self: STREAM] = { <> }; FlushInput: PUBLIC PROC [self: STREAM, wait: BOOL Ź FALSE] RETURNS [bytesSkipped: CARD Ź 0] = { <> }; WaitAttention: PUBLIC PROC [self: STREAM, waitTimeout: Milliseconds Ź waitForever] RETURNS [AttentionType] = { RETURN[0]; }; GetStatus: PUBLIC PROC [self: STREAM, reset: BOOL Ź TRUE] RETURNS [state: State, ssType: SubSequenceType, attentionType: AttentionType] = { handle: Handle = NARROW[self.streamData]; [state, ssType, attentionType] Ź EntryGetStatus[handle, reset]; }; <> EntryGetStatus: ENTRY PROC [handle: Handle, reset: BOOL] RETURNS [state: State, ssType: SubSequenceType, attentionType: AttentionType] ~ { ENABLE UNWIND => NULL; buffer: Buffer Ź handle.currentInputBuffer; IF handle.closed THEN RETURN WITH ERROR ConnectionClosed[handle.closeReason, handle.closeText]; IF buffer = NIL THEN -- I think that the only state we can have now is -- RETURN[state: open, ssType: handle.currentInputSST, attentionType: 0]; IF buffer.attention THEN { state Ź attention; ssType Ź buffer.sst; -- the sst of the attention packet attentionType Ź LOOPHOLE[buffer.bufferRef[0]]; -- still ugly IF reset THEN { -- dispose of this packet p: OBAttn ~ handle.recvdOBAttnList; IF (p # NIL) AND (p.sequence = handle.inBandAttnNumber) THEN -- delete it from the expidited queue handle.recvdOBAttnList Ź p.next; FreeInputBuffer[handle: handle, buffer: buffer]; handle.currentInputBuffer Ź NIL; -- so that UnsafeGetBlock can make progress }; RETURN; }; -- IF buffer.attention IF buffer.sst # handle.currentInputSST THEN { IF reset THEN handle.currentInputSST Ź buffer.sst; RETURN[state: ssTypeChange, ssType: buffer.sst, attentionType: 0]; }; IF buffer.eom THEN { IF reset THEN { -- give the packet back FreeInputBuffer[handle: handle, buffer: buffer]; handle.currentInputBuffer Ź NIL; -- so that UnsafeGetBlock can make progress }; RETURN [state: endOfMessage, ssType: handle.currentInputSST, attentionType: 0]; }; RETURN [state: open, ssType: handle.currentInputSST, attentionType: 0]; }; <> AllocateCD: PROC RETURNS [cd: TLI.ConnectionData] = { bytes: CARD32 ~ BYTES[TLI.ConnectionDataRep]; cd Ź NEW[TLI.ConnectionDataRep]; -- the rest gets filled in in the call to TOpen TRUSTED {Basics.FillBytes[dstBase: LOOPHOLE[cd], dstStart: 0, count: bytes, value: 0]}; }; AllocateCalls: PROC RETURNS [calls: Calls] = { calls Ź NEW[CallsRep]; calls.reqTBindAddress Ź NEW[XNS.Address]; calls.reqTBind Ź NEW[TLI.TBindCall]; calls.reqTBind.addr.maxlen Ź calls.reqTBind.addr.len Ź BYTES[XNS.Address]; TRUSTED {calls.reqTBind.addr.char Ź LOOPHOLE[calls.reqTBindAddress, UnixTypes.CHARPtr]}; calls.retTBindAddress Ź NEW[XNS.Address]; calls.retTBind Ź NEW[TLI.TBindCall]; calls.retTBind.addr.maxlen Ź calls.retTBind.addr.len Ź BYTES[XNS.Address]; TRUSTED {calls.retTBind.addr.char Ź LOOPHOLE[calls.retTBindAddress, UnixTypes.CHARPtr]}; calls.sendTCallAddress Ź NEW[XNS.Address]; calls.sendTCall Ź NEW[TLI.TCall]; calls.sendTCall.addr.maxlen Ź calls.sendTCall.addr.len Ź BYTES[XNS.Address]; TRUSTED {calls.sendTCall.addr.char Ź LOOPHOLE[calls.sendTCallAddress, UnixTypes.CHARPtr]}; calls.rcvTCallAddress Ź NEW[XNS.Address]; calls.rcvTCall Ź NEW[TLI.TCall]; calls.rcvTCall.addr.maxlen Ź calls.rcvTCall.addr.len Ź BYTES[XNS.Address]; TRUSTED {calls.rcvTCall.addr.char Ź LOOPHOLE[calls.rcvTCallAddress, UnixTypes.CHARPtr]}; }; MakeHandle: PROC [sendBuffers, recvBuffers: CARD, getTimeout, putTimeout: Milliseconds] RETURNS [handle: Handle] ~ { handle Ź NEW[HandleRep]; handle.cd Ź NIL; -- filled in by our caller since it was needed before we got called handle.calls Ź NIL; -- filled in by our caller since it was needed before we got called InitConditions[handle: handle]; InitTimeouts[handle, getTimeout, putTimeout]; handle.inBuffers Ź NIL; handle.inFreeBuffers Ź AllocateBuffers[number: recvBuffers]; handle.outputBuffers Ź NIL; handle.outputFreeBuffers Ź AllocateBuffers[number: sendBuffers]; Process.Detach[FORK Pull[handle]]; handle.auxProcCnt Ź handle.auxProcCnt.SUCC; Process.Detach[FORK Push[handle]]; handle.auxProcCnt Ź handle.auxProcCnt.SUCC; }; InitConditions: PROC [handle: Handle] ~ { TRUSTED { Process.EnableAborts[@handle.auxProcsDone]; -- ???? Process.EnableAborts[@handle.inputReady]; Process.EnableAborts[@handle.inputBufferFree]; Process.EnableAborts[@handle.recvdOBAttn]; Process.EnableAborts[@handle.outputReady]; Process.EnableAborts[@handle.outputBufferFree]; Process.EnableAborts[@handle.flusher]; }; }; InitTimeouts: PROC [handle: Handle, getTimeout, putTimeout: Milliseconds] ~ { TRUSTED { Process.SetTimeout[@handle.flusher, flushTime]; }; }; AllocateBuffers: PROC [number: INT] RETURNS [buffer: Buffer Ź NIL] = { FOR i: INT IN [0 .. number) DO new: Buffer ~ NEW[BufferRep]; new.bufferRef Ź NEW[TEXT[LAST[BufferIndex]]]; new.buffer Ź LOOPHOLE[new.bufferRef, POINTER] + BYTES[TEXT[0]]; -- to the bits new.charPTR Ź LOOPHOLE[new.buffer, UnixTypes.CHARPtr]; new.flags Ź NEW[INT Ź 0]; IF buffer = NIL THEN buffer Ź new ELSE {buffer.next Ź new; buffer Ź buffer.next}; ENDLOOP; }; <> AllocateInputBuffer: ENTRY PROC [handle: Handle] RETURNS [buffer: Buffer]~ { ENABLE UNWIND => NULL; WHILE handle.inFreeBuffers = NIL DO -- we need to wait until one becomes available WAIT handle.inputBufferFree; ENDLOOP; buffer Ź handle.inFreeBuffers; handle.inFreeBuffers Ź handle.inFreeBuffers.next; -- who cares it it's nil }; FreeInputBuffer: INTERNAL PROC [handle: Handle, buffer: Buffer] = { ENABLE UNWIND => NULL; tmp: Buffer Ź handle.inFreeBuffers; -- might be nil buffer.state Ź empty; handle.inFreeBuffers Ź buffer; handle.inFreeBuffers.next Ź tmp; -- hang the free ones on the list BROADCAST handle.inputBufferFree; -- tell someone who is waiting for a buffer }; AppendInputBuffer: ENTRY PROC [handle: Handle, buffer: Buffer] = { ENABLE UNWIND => NULL; tmp: Buffer Ź handle.inBuffers; -- the current head IF tmp = NIL THEN { -- first time handle.inBuffers Ź buffer; BROADCAST handle.inputReady; RETURN; }; UNTIL tmp.next = NIL DO -- find the end tmp Ź tmp.next; ENDLOOP; tmp.next Ź buffer; -- tack it on BROADCAST handle.inputReady; }; GetFirstInputBuffer: INTERNAL PROC [handle: Handle] RETURNS [buffer: Buffer] = { ENABLE UNWIND => NULL; WHILE handle.inBuffers = NIL DO -- we need to wait until one becomes available WAIT handle.inputReady; ENDLOOP; buffer Ź handle.inBuffers; -- take off the first one handle.inBuffers Ź handle.inBuffers.next; -- might be nil, but who cares }; AllocateOutputBuffer: INTERNAL PROC [handle: Handle] RETURNS [buffer: Buffer]~ { ENABLE UNWIND => NULL; WHILE handle.outputFreeBuffers = NIL DO -- we need to wait until one becomes available WAIT handle.outputBufferFree; ENDLOOP; buffer Ź handle.outputFreeBuffers; handle.outputFreeBuffers Ź handle.outputFreeBuffers.next; -- who cares if it's nil }; FreeOutputBuffer: ENTRY PROC [handle: Handle, buffer: Buffer] = { ENABLE UNWIND => NULL; tmp: Buffer Ź handle.outputFreeBuffers; -- might be nil buffer.state Ź empty; handle.outputFreeBuffers Ź buffer; handle.outputFreeBuffers.next Ź tmp; -- hang the free ones on the list BROADCAST handle.outputBufferFree; -- tell someone who is waiting for a buffer }; AppendOutputBuffer: INTERNAL PROC [handle: Handle, buffer: Buffer] = { ENABLE UNWIND => NULL; tmp: Buffer Ź handle.outputBuffers; -- the current head IF tmp = NIL THEN { -- first time handle.outputBuffers Ź buffer; BROADCAST handle.outputReady; RETURN; }; UNTIL tmp.next = NIL DO -- find the end tmp Ź tmp.next; ENDLOOP; tmp.next Ź buffer; -- tack it on BROADCAST handle.outputReady; }; GetFirstOutputBuffer: ENTRY PROC [handle: Handle] RETURNS [buffer: Buffer] = { ENABLE UNWIND => NULL; WHILE handle.outputBuffers = NIL DO -- we need to wait until one becomes available WAIT handle.outputReady; ENDLOOP; buffer Ź handle.outputBuffers; -- take off the first one handle.outputBuffers Ź handle.outputBuffers.next; -- might be nil, but who cares }; <> NotifyAuxProcDone: ENTRY PROC [handle: Handle] ~ { ENABLE UNWIND => NULL; IF handle.auxProcCnt = 0 THEN ERROR; handle.auxProcCnt Ź handle.auxProcCnt.PRED; IF handle.auxProcCnt = 0 THEN NOTIFY handle.auxProcsDone; }; WaitAuxProcsDone: ENTRY PROC [handle: Handle] ~ { ENABLE UNWIND => NULL; WHILE handle.auxProcCnt > 0 DO WAIT handle.auxProcsDone; ENDLOOP; }; <> Pull: PROC [handle: Handle] ~ { SetGetting: ENTRY PROC [handle: Handle] = {handle.inputWaiting Ź TRUE}; UnSetGetting: ENTRY PROC [handle: Handle] = {handle.inputWaiting Ź FALSE}; WHILE NOT handle.closed DO nRcv: INT; cd: TLI.ConnectionData ~ handle.cd; event: INT Ź 0; look: BOOL Ź FALSE; buffer: Buffer Ź AllocateInputBuffer[handle: handle]; -- get (wait) for a free buffer. This really doesn't allocate anything, but that's the function it performs SetGetting[handle: handle]; nRcv Ź ReceiveBuffer[cd: cd, buffer: buffer ! LookEvent => {event Ź lookEvent; look Ź TRUE; CONTINUE} ]; -- get the packet UnSetGetting[handle: handle]; IF NOT look THEN { ProcessBuffer[handle: handle, buffer: buffer, nRcv: nRcv]; AppendInputBuffer[handle: handle, buffer: buffer]; } ELSE { SELECT event FROM TLI.ORDREL => { -- the other end wants us to close in an orderly way PrintF["Orderly close requested"]; OrderlyClose[handle: handle, initiate: FALSE]; -- set up for the close. This proc returns only when we are closed with the other end EXIT; }; TLI.DISCONNECT => { -- the other end just wants us to go away PrintF["Disconnect requested"]; BlastClose[handle: handle]; -- don't really care how we close, just close EXIT; }; TLI.LISTEN => {ERROR BadSysCall["LISTEN"]}; -- an connect indication received TLI.CONNECT => {ERROR BadSysCall["CONNECT"]}; -- a connect confirmation received TLI.DATA => {ERROR BadSysCall["DATA"]}; -- normal data TLI.EXDATA => {ERROR BadSysCall["EXDATA"]}; -- expidited data TLI.TERROR => {ERROR BadSysCall["TERROR"]}; -- a real problem TLI.UDERR => {ERROR BadSysCall["UDERR"]}; -- should probably not happen either ENDCASE => {ERROR BadSysCall["ENDCASE"]}; -- unknown event type }; ENDLOOP; NotifyAuxProcDone[handle]; }; ReceiveBuffer: PROC [cd: TLI.ConnectionData, buffer: Buffer] RETURNS [INT] = { flags: REF INT ~ buffer.flags; -- will also contain the sst field nRcv: INT Ź 0; IF buffer.state # empty THEN SystemError; -- this is probably not right, but I don't know what to do for the moment buffer.state Ź filling; -- set the flag for anyone that might be waiting on it DO -- the loop is in here because of the errno and # TLOOK stuff in the error handler nRcv Ź TLI.TRcv[cd: cd, buf: buffer.charPTR, nbytes: packetSize, flags: flags]; IF nRcv < 0 THEN { -- something is wrong error: UnixErrno.Errno ~ UnixErrno.GetErrno[]; IF (error = EINTR) AND (cd.errno # TLOOK) THEN LOOP; -- I'm not sure what this does SELECT cd.errno FROM TNODATA => { -- this is set when the underlying GetMsg returns < 0 and EAGAIN is the errno. I think that this can't happen according to the SunOS documentation IF error = EWOULDBLOCK THEN RETURN[nRcv]; -- I'm not sure that this can happen under pcr since the XRŹGetMsg is done by pcr as blocked IO. ERROR BadSysCall["TNODATA"]; }; TSYSERR => { -- this is the general catch-all for things that have gone wrong. I don't think that this can be recovered from ERROR BadSysCall["TSYSERR"]; }; TLOOK => { -- an asynchonous event happened event: INT ~ TLI.TLook[cd: cd]; ERROR LookEvent[event]; }; TBADF => -- a bad fd on TAccept -- ERROR BadSysCall["TBADF"]; TOUTSTATE => -- reports out of state -- ERROR BadSysCall["TOUTSTATE"]; TBUFOVFLW => -- control buffer to small -- ERROR BadSysCall["TBUFOVFLW"]; TFLOW => { -- indicates that flow control is invoked on this call. I think that this needs to be retried ERROR BadSysCall["TFLOW"]; }; TNODIS => -- we didn't get the right answer on a dissconnect -- ERROR BadSysCall["TNODIS"]; TNOREL => -- we didn't get the orderly release -- ERROR BadSysCall["TNOREL"]; TNOTSUPPORT => -- something is really wrong -- ERROR BadSysCall["TNOTSUPPORT"]; TSTATECHNG => -- state is changing -- ERROR BadSysCall["TSTATECHNG"]; ENDCASE => ERROR BadSysCall["something else"]; }; -- something is wrong buffer.state Ź full; -- set the flag for anyone that might be waiting on it RETURN[nRcv]; ENDLOOP; }; ProcessBuffer: ENTRY PROC [handle: Handle, buffer: Buffer, nRcv: INT] = { ENABLE UNWIND => NULL; cd: TLI.ConnectionData ~ handle.cd; flags: REF INT ~ buffer.flags; -- also contains the sst field PrintF["ProcessBuffer"]; PrintF[IO.PutFR1[" nRcv = %g ", [integer[nRcv]]]]; buffer.index Ź 0; -- reset the index buffer.bytes Ź nRcv; -- set the number of bytes in the packet <> buffer.sst Ź GetSST[flags: flags]; <<>> <> IF GetAttention[flags: flags] THEN { -- attention byte IF GetMore[flags: flags] THEN { -- this is the expedited (OOB) copy of the attention sequence: INT ~ handle.outOfBandAttnNumber; IF buffer.bytes # 1 THEN ERROR SystemError; -- not good since there's supposed to be only one byte GotOBAttn[handle: handle, sequence: sequence, type: LOOPHOLE[buffer.bufferRef[0]]]; -- ugly ain't it handle.outOfBandAttnNumber Ź handle.outOfBandAttnNumber + 1; -- update the number to get ready for the next one buffer.state Ź full; -- mark our condition PrintF[" attention, OOB\n"]; BROADCAST handle.inputReady; RETURN; } ELSE { -- this is the second (InBand) copy sequence: INT ~ handle.inBandAttnNumber; IF buffer.bytes # 1 THEN ERROR SystemError; -- not good since there's supposed to be only one byte buffer.attention Ź TRUE; -- this is an attention packet buffer.sequence Ź sequence; -- to correlate with the OOB one handle.inBandAttnNumber Ź handle.inBandAttnNumber + 1; buffer.state Ź full; -- mark our condition PrintF[" attention, inline\n"]; BROADCAST handle.inputReady; RETURN; }; }; <<>> <> buffer.eom Ź NOT GetMore[flags: flags]; buffer.state Ź full; -- mark our condition BROADCAST handle.inputReady; PrintF[IO.PutFR1["eom = %g\n", [boolean[buffer.eom]]]]; buffer.bufferRef.length Ź nRcv; }; GotOBAttn: INTERNAL PROC [handle: Handle, sequence: INT, type: AttentionType] ~ { p, prev, new: OBAttn; prev Ź NIL; p Ź handle.recvdOBAttnList; DO IF p = NIL THEN EXIT; prev Ź p; p Ź p.next; ENDLOOP; new Ź NEW[OBAttnObject Ź [next: p, sequence: sequence, attentionType: type]]; IF prev = NIL THEN handle.recvdOBAttnList Ź new ELSE prev.next Ź new; BROADCAST handle.recvdOBAttn; }; <> Push: PROC [handle: Handle] ~ { -- there had best be only one of these per stream or things might get real confused cd: TLI.ConnectionData ~ handle.cd; sent: INT Ź 0; WHILE NOT handle.closed DO buffer: Buffer Ź GetFirstOutputBuffer[handle: handle]; -- get a buffer to send. This will block until one is ready flags: INT Ź SetSST[sst: buffer.sst, flags: 0]; flags Ź SetMore[more: NOT buffer.eom, flags: flags]; flags Ź SetAttention[attention: buffer.attention, flags: flags]; sent Ź TLI.TSnd[cd: cd, buf: buffer.charPTR, nbytes: buffer.bytes, flags: flags]; IF sent < 0 OR sent # buffer.bytes THEN { -- we need to disconnect BlastClose[handle: handle]; handle.closed Ź TRUE; EXIT; }; FreeOutputBuffer[handle: handle, buffer: buffer]; ENDLOOP; NotifyAuxProcDone[handle]; }; <> GetAttention: PROC [flags: INT] RETURNS [attention: BOOL] = { RETURN[ Basics.BITAND[flags, TLI.EXPEDITED] # 0 ]; }; GetMore: PROC [flags: INT] RETURNS [more: BOOL] = { RETURN[ Basics.BITAND[flags, TLI.MORE] # 0 ]; }; GetSST: PROC [flags: INT] RETURNS [SubSequenceType] = { tmp: INT ~ Basics.BITAND[Basics.BITRSHIFT[value: flags, count: 8], 0FFh]; -- the sst IF tmp <= 255 THEN RETURN[LOOPHOLE[tmp]] ELSE ERROR SystemError; }; SetAttention: PROC [attention: BOOL, flags: INT Ź 0] RETURNS [INT] = { RETURN[(IF attention THEN Basics.BITOR[flags, TLI.EXPEDITED] ELSE flags)]; }; SetMore: PROC [more: BOOL, flags: INT Ź 0] RETURNS [INT] = { RETURN[(IF more THEN Basics.BITOR[flags, TLI.MORE] ELSE flags)]; }; SetSST: PROC [sst: SubSequenceType, flags: INT Ź 0] RETURNS [INT] = { tmp: INT Ź sst; tmp Ź Basics.BITLSHIFT[value: flags, count: 8]; -- move it up 1 byte tmp Ź Basics.BITAND[tmp, 0FF00h]; -- make sure that it's only the sst tmp Ź Basics.BITOR[tmp, flags]; RETURN[tmp]; }; <> UnixPrintF: PROC [format: CStrings.CString] RETURNS [INT] = TRUSTED MACHINE CODE {"printf"}; PrintF: PROC [rope: ROPE] = { s: CStrings.CString ~ UXStrings.Create[rope]; [] Ź UnixPrintF[s]; }; <> CreateStreamProcs[]; END.