> UNTIL pleaseStop DO <> now _ ProcessorFace.GetClockPulses[]; THROUGH (0..RetransmissionCount[]] DO <> <> <> <> IF (retransBuffer _ GetFromRetransmissionQueue[]) = NIL THEN EXIT; IF LOOPHOLE[retransBuffer.ois.sequenceNumber - unackedOutputSeq, INTEGER] < 0 THEN OISCP.ReturnFreeOisBuffer[retransBuffer] <> ELSE BEGIN <> retransBuffer.tries _ retransBuffer.tries + 1; PutPacketOnSocketChannel[retransBuffer]; -- requeues buffer on sentQueue retransBuffer _ NIL; IF CommFlags.doStats THEN StatIncr[statDataPacketsRetransmitted]; END; ENDLOOP; IF TimeToSendAProbe[] THEN SendSystemPacket[TRUE]; WaitForAWhile[]; UpdateRetransmitTime[]; ENDLOOP; END; -- Retransmitter PulsesToTicks: PROCEDURE [pulses: Pulses] RETURNS [Process.Ticks] = INLINE BEGIN lastCard: LONG CARDINAL = LAST[CARDINAL]; msecs: LONG CARDINAL _ (((pulses+50)/100)*LONG[ProcessorFace.microsecondsPerHundredPulses]+500)/1000 + 1; RETURN[Process.MsecToTicks[IF msecs < lastCard THEN msecs ELSE LAST[CARDINAL]]]; END; <> <> <> Receiver: PROCEDURE = BEGIN b: BufferDefs.OisBuffer; UNTIL pleaseStop DO <> <> b _ Socket.GetPacket[ socketCH ! Socket.TimeOut => RETRY; Socket.ChannelAborted => EXIT]; <> SELECT LOOPHOLE[b.status, XmitStatus] FROM goodCompletion => SELECT b.ois.transCntlAndPktTp.packetType FROM sequencedPacket => GotSequencedPacket[LOOPHOLE[b, BufferDefs.OisBuffer]]; error => GotErrorPacket[b]; ENDCASE => BEGIN <> OISCP.ReturnFreeOisBuffer[b]; IF CommFlags.doStats THEN StatIncr[statPacketsRejectedBadType]; END; ENDCASE => OISCP.ReturnFreeOisBuffer[b]; ENDLOOP; END; -- Receiver <> GotSequencedPacket: PROCEDURE [b: BufferDefs.OisBuffer] = BEGIN returnAnAck: BOOLEAN _ FALSE; <> giveBackToSocket: BOOLEAN _ TRUE; -- unless we put on some other queue nowPulse: LONG CARDINAL; -- set if we get a packet for our connection GotSequencedPacketLocked: ENTRY PROCEDURE = INLINE BEGIN SELECT state FROM unestablished, suspended, terminating => NULL; -- not interested. activeEstablish, waitEstablish => [returnAnAck, giveBackToSocket] _ EstablishThisConnection[b]; established, open => BEGIN <> IF b.ois.source.host = remoteAddr.host AND b.ois.source.socket = remoteAddr.socket AND b.ois.sourceConnectionID = remoteConnectionID AND b.ois.destinationConnectionID = localConnectionID THEN BEGIN -- the packet is for this connection probeCounter _ 0; -- other end is alive lastPacketReceivedPulse _ nowPulse _ ProcessorFace.GetClockPulses[]; <> SELECT LOOPHOLE[b.ois.sequenceNumber - nextInputSeq, INTEGER] FROM <> IN [0..duplicateWindow] => RightOrEarlyPacket[]; <> IN [-duplicateWindow..0) => DuplicatePacket[]; <> ENDCASE => IF CommFlags.doStats THEN StatIncr[statDataPacketsReceivedVeryLate] END ELSE <> BEGIN IF CommFlags.doStats AND (b.ois.source.host # remoteAddr.host OR b.ois.source.socket # remoteAddr.socket) THEN StatIncr[statPacketsRejectedBadSource]; IF b.ois.sourceConnectionID # remoteConnectionID OR b.ois.destinationConnectionID # localConnectionID THEN BEGIN IF CommFlags.doStats THEN StatIncr[statPacketsRejectedBadID]; returnAnAck _ TRUE; -- probably our connection response got lost END; END; END; ENDCASE; <> <> <> <> <> <> <> <> <> <> IF LOOPHOLE[maxInputSeq - nextInputSeq, INTEGER] < minRcvAllocation THEN WAIT letClientRun; END; -- GotSequencedPacketLocked <> <> <> RightOrEarlyPacket: INTERNAL PROCEDURE = INLINE BEGIN IF b.ois.sendAck THEN BEGIN returnAnAck _ TRUE; IF CommFlags.doStats THEN StatIncr[statAckRequestsReceived]; END; <> IF LOOPHOLE[b.ois.allocationNumber - maxOutputSeq, INTEGER] > 0 THEN BEGIN maxOutputSeq _ b.ois.allocationNumber; NOTIFY newAllocation; END; IF LOOPHOLE[b.ois.acknowledgeNumber - unackedOutputSeq, INTEGER] > 0 THEN BEGIN unackedOutputSeq _ b.ois.acknowledgeNumber; ackRequestPulse _ nowPulse; -- if we get an ack we move the ack req. time END; <> <> IF sentBuffer = NIL THEN sentBuffer _ DequeueSpp[sentQueue]; UNTIL sentBuffer = NIL OR LOOPHOLE[sentBuffer.ois.sequenceNumber - unackedOutputSeq, INTEGER] >= 0 DO IF sentBuffer.ois.sendAck AND sentBuffer.tries = 0 THEN BEGIN -- constitutes a measure of lossless delay (no retransmissions) delaySum _ [delaySum + (nowPulse - sentBuffer.time)]; <> delayCount _ delayCount + 1; END; OISCP.ReturnFreeOisBuffer[sentBuffer]; sentBuffer _ DequeueSpp[sentQueue]; ENDLOOP; <> IF b.ois.systemPacket THEN BEGIN IF CommFlags.doStats THEN BEGIN StatIncr[statSystemPacketsReceived]; StatIncr[statAcksReceived]; END; END ELSE <> BEGIN <> <> IF ~b.ois.attention OR (b.ois.attention AND AttentionPacketProcessed[b]) THEN BEGIN <> IF LOOPHOLE[b.ois.sequenceNumber - nextInputSeq, INTEGER] = 0 THEN <> BEGIN state _ open; DO nextInputSeq _ nextInputSeq + 1; IF attnCount # 0 THEN UpdateAttnSeqTable[]; EnqueueSpp[inOrderQueue, b]; giveBackToSocket _ FALSE; NOTIFY inOrderQueueNotEmpty; <> <> DO IF BufferDefs.QueueLength[outOfOrderQueue] = 0 THEN BEGIN b _ NIL; EXIT; END; b _ DequeueSpp[outOfOrderQueue]; SELECT LOOPHOLE[b.ois.sequenceNumber - nextInputSeq, INTEGER] FROM 0 => EXIT; <> IN (0..duplicateWindow] => EnqueueSpp[tempQueue, b]; <> ENDCASE => OISCP.ReturnFreeOisBuffer[b]; <> ENDLOOP; <> UNTIL (tempQueue.first) = NIL DO EnqueueSpp[outOfOrderQueue, DequeueSpp[tempQueue]]; ENDLOOP; IF b = NIL THEN EXIT; <> ENDLOOP; END ELSE IF CommFlags.doStats THEN StatIncr[statDataPacketsReceivedEarly]; <> END; END; END; -- RightOrEarlyPacket DuplicatePacket: INTERNAL PROCEDURE = INLINE BEGIN IF CommFlags.doStats THEN StatIncr[statDataPacketsReceivedAgain]; <> IF b.ois.sendAck THEN returnAnAck _ TRUE; END; -- DuplicatePacket <

> IF b.ois.pktLength >= bytesPerSequencedPktHeader THEN BEGIN GotSequencedPacketLocked[]; IF returnAnAck THEN SendSystemPacket[FALSE]; -- we may have to send the ack END ELSE IF CommFlags.doStats THEN StatIncr[statEmptyFunnys]; IF giveBackToSocket THEN OISCP.ReturnFreeOisBuffer[b]; END; -- GotSequencedPacket <> <> <> <> <> <> <> <> AttentionPacketProcessed: INTERNAL PROCEDURE [b: BufferDefs.OisBuffer] RETURNS [BOOLEAN] = BEGIN i: CARDINAL _ 1; alreadyThere: BOOLEAN _ FALSE; advanceB: BufferDefs.OisBuffer; nBytes: CARDINAL; IF attnCount = maxPendingAttns THEN RETURN[FALSE]; -- no space <> IF attnCount # 0 -- see if an entry is already there THEN FOR i IN (0..attnCount] DO IF attnSeqTable[i] = b.ois.sequenceNumber THEN BEGIN alreadyThere _ TRUE; EXIT; END; ENDLOOP; IF NOT alreadyThere THEN -- because attnCount=0 or because really not there! BEGIN IF (advanceB _ MaybeGetFreeReceiveOisBufferFromPool[pool]) = NIL THEN RETURN[FALSE]; -- cop out attnSeqTable[i] _ b.ois.sequenceNumber; attnCount _ attnCount + 1; <> nBytes _ PrincOpsUtils.ByteBlt[to: [@advanceB.ois, 0, b.ois.pktLength], from: [@b.ois, 0, b.ois.pktLength]]; EnqueueSpp[inAttnQueue, advanceB]; NOTIFY newInputAttn; END; IF CommFlags.doStats THEN StatIncr[statAttentionsReceived]; RETURN[TRUE]; END; -- AttentionPacketProcessed <> <> <> UpdateAttnSeqTable: INTERNAL PROCEDURE = BEGIN i: CARDINAL _ 1; DO IF i > attnCount THEN EXIT; IF LOOPHOLE[(attnSeqTable[i] - nextInputSeq), INTEGER] < 0 THEN BEGIN attnSeqTable[i] _ attnSeqTable[attnCount]; attnSeqTable[attnCount] _ 0; attnCount _ attnCount - 1; END ELSE i _ i + 1; ENDLOOP; END; -- UpdateAttnSeqTable <> GotErrorPacket: ENTRY PROCEDURE [b: BufferDefs.OisBuffer] = BEGIN ENABLE UNWIND => NULL; IF CommFlags.doStats THEN StatIncr[statErrorPacketsReceived]; SELECT b.ois.errorType FROM noSocketOisErrorCode => SELECT state FROM unestablished, activeEstablish, waitEstablish => {whyFailed _ noServiceAtDestination; SuspendStream[noRouteToDestination]}; established, open => SuspendStream[remoteServiceDisappeared]; ENDCASE => NULL; ENDCASE => NULL; OISCP.ReturnFreeOisBuffer[b]; END; -- GotErrorPacket <> <> SetWaitTime: ENTRY PROCEDURE [time: OISCPTypes.WaitTime] = BEGIN waitPulses _ MilliSecondsToPulses[time]; END; -- SetWaitTime <> FindAddresses: ENTRY PROCEDURE RETURNS [local, remote: NetworkAddress] = BEGIN local _ localAddr; remote _ remoteAddr; END; -- FindAddresses <> GetSenderSizeLimit: ENTRY PROCEDURE RETURNS [CARDINAL] = BEGIN RETURN[maxOisPktLength - bytesPerSequencedPktHeader]; END; <> <> <> EstablishThisConnection: INTERNAL PROCEDURE [b: BufferDefs.OisBuffer] RETURNS [returnAnAck, giveBackToSocket: BOOLEAN] = BEGIN returnAnAck _ FALSE; giveBackToSocket _ TRUE; <> <> IF b.ois.source.host = remoteAddr.host AND b.ois.source.socket = remoteAddr.socket AND b.ois.destinationConnectionID = localConnectionID AND b.ois.sourceConnectionID # unknownConnID AND b.ois.sequenceNumber = 0 THEN BEGIN remoteConnectionID _ b.ois.sourceConnectionID; maxOutputSeq _ b.ois.allocationNumber; state _ established; IF b.ois.sendAck THEN returnAnAck _ TRUE; IF NOT b.ois.systemPacket THEN BEGIN <> <> IF ~b.ois.attention OR (b.ois.attention AND AttentionPacketProcessed[b]) THEN BEGIN nextInputSeq _ nextInputSeq + 1; EnqueueSpp[inOrderQueue, b]; giveBackToSocket _ FALSE; IF attnCount # 0 THEN UpdateAttnSeqTable[]; <> state _ open; NOTIFY inOrderQueueNotEmpty; END; END ELSE IF CommFlags.doStats THEN StatIncr[statSystemPacketsReceived]; NOTIFY connectionIsEstablished; END <> <> <> ELSE IF b.ois.source.host = remoteAddr.host AND b.ois.source.socket = remoteAddr.socket AND b.ois.destinationConnectionID = unknownConnID AND b.ois.sourceConnectionID # unknownConnID AND b.ois.sequenceNumber = 0 THEN BEGIN remoteConnectionID _ b.ois.sourceConnectionID; maxOutputSeq _ b.ois.allocationNumber; IF state = waitEstablish THEN returnAnAck _ TRUE; state _ established; IF b.ois.sendAck THEN returnAnAck _ TRUE; IF CommFlags.doStats AND b.ois.systemPacket THEN StatIncr[statSystemPacketsReceived]; NOTIFY connectionIsEstablished; END <> <> ELSE IF state = activeEstablish AND b.ois.source.host = remoteAddr.host AND b.ois.source.socket # remoteAddr.socket AND b.ois.destinationConnectionID = localConnectionID AND b.ois.sourceConnectionID # unknownConnID AND b.ois.sequenceNumber = 0 THEN BEGIN remoteAddr.socket _ b.ois.source.socket; remoteConnectionID _ b.ois.sourceConnectionID; maxOutputSeq _ b.ois.allocationNumber; state _ established; IF b.ois.sendAck THEN returnAnAck _ TRUE; IF NOT b.ois.systemPacket THEN BEGIN <> <> IF ~b.ois.attention OR (b.ois.attention AND AttentionPacketProcessed[b]) THEN BEGIN nextInputSeq _ nextInputSeq + 1; EnqueueSpp[inOrderQueue, b]; giveBackToSocket _ FALSE; IF attnCount # 0 THEN UpdateAttnSeqTable[]; <> state _ open; NOTIFY inOrderQueueNotEmpty; END; END ELSE IF CommFlags.doStats THEN StatIncr[statSystemPacketsReceived]; NOTIFY connectionIsEstablished; END <> ELSE IF CommFlags.doStats THEN StatIncr[statPacketsRejectedBadID]; IF returnAnAck AND CommFlags.doStats THEN StatIncr[statAckRequestsReceived]; END; -- EstablishThisConnection <> <> <> <> ActivelyEstablish: PROCEDURE = BEGIN currentPulse, startPulse: LONG CARDINAL; WaitUntilActivelyEstablished: ENTRY PROCEDURE RETURNS [BOOLEAN] = INLINE BEGIN ENABLE UNWIND => NULL; WAIT connectionIsEstablished; IF state = established OR state = open THEN RETURN[TRUE] ELSE BEGIN IF state = suspended THEN RETURN WITH ERROR ConnectionFailed[whyFailed]; currentPulse _ ProcessorFace.GetClockPulses[]; IF (currentPulse - startPulse) >= waitPulses THEN BEGIN SIGNAL ConnectionFailed[timeout]; startPulse _ ProcessorFace.GetClockPulses[]; END; RETURN[FALSE]; END; END; -- WaitUntilActivelyEstablished startPulse _ ProcessorFace.GetClockPulses[]; DO SendSystemPacket[TRUE]; <> IF WaitUntilActivelyEstablished[] -- this returns or raises a SIGNAL -- THEN RETURN; ENDLOOP; END; -- ActivelyEstablish <> WaitUntilEstablished: ENTRY PROCEDURE = BEGIN ENABLE UNWIND => NULL; currentPulse, startPulse: LONG CARDINAL; startPulse _ ProcessorFace.GetClockPulses[]; DO <> WAIT connectionIsEstablished; IF state = established OR state = open THEN RETURN ELSE BEGIN currentPulse _ ProcessorFace.GetClockPulses[]; IF (currentPulse - startPulse) >= waitPulses THEN BEGIN SIGNAL ConnectionFailed[timeout]; startPulse _ ProcessorFace.GetClockPulses[]; END; END; ENDLOOP; END; -- WaitUntilEstablished <> <> <> <> Delete: PUBLIC PROCEDURE = BEGIN DeleteActivate[]; Socket.Abort[socketCH]; JOIN retransmitterFork; JOIN receiverFork; DeleteCleanup[]; Socket.Delete[socketCH]; LoadState.SelfDestruct[]; END; -- Delete DeleteActivate: ENTRY PROCEDURE = INLINE BEGIN state _ terminating; pleaseStop _ TRUE; NOTIFY retransmitterWakeupTimer; END; -- DeleteActivate DeleteCleanup: ENTRY PROCEDURE = INLINE BEGIN OPEN SocketInternal; ENABLE UNWIND => NULL; IF sentBuffer # NIL THEN OISCP.ReturnFreeOisBuffer[sentBuffer]; BufferDefs.QueueCleanup[inOrderQueue]; BufferDefs.QueueCleanup[inAttnQueue]; BufferDefs.QueueCleanup[outOfOrderQueue]; BufferDefs.QueueCleanup[tempQueue]; WHILE sentQueue.length> WAIT letClientRun; -- wait on any convenient variable - ABORTs must be DISABLED! ENDLOOP; BufferDefs.QueueCleanup[sentQueue]; END; -- DeleteCleanup MilliSecondsToPulses: PROCEDURE [ms: LONG CARDINAL] RETURNS [pulses: Pulses] = BEGIN <> <> micro: LONG CARDINAL = MIN[LAST[LONG CARDINAL]/1000,ms] * 1000; hundreds: LONG CARDINAL = micro / ProcessorFace.microsecondsPerHundredPulses; pulses _ [MIN[LAST[LONG CARDINAL]/100,hundreds]*100]; END; -- end MilliSecondsToPulses <> <> <> <> <> <> <> <> state _ unestablished; whySuspended _ notSuspended; stateBeforeSuspension _ unestablished; Process.SetTimeout[@connectionIsEstablished, Process.MsecToTicks[2000]]; nextInputSeq _ unackedOutputSeq _ nextOutputSeq _ 0; maxOutputSeq _ nextOutputSeq + defaultSendAllocation - 1; SELECT classOfService FROM bulk => {socketCH _ Socket.Create[localAddr, bulkSendBufs, bulkReceiveBufs, 0, TRUE]; maxInputSeq _ bulkInitialAllocation}; transactional => {socketCH _ Socket.Create[localAddr, transactSendBufs, transactReceiveBufs, 0, TRUE]; maxInputSeq _ transactInitialAllocation}; ENDCASE => ERROR; pool _ ps.pool _ SocketInternal.GetBufferPool[ SocketInternal.ChannelHandleToSocketHandle[socketCH]]; localAddr _ Socket.GetStatus[socketCH].localAddr; IF remoteConnectionID # unknownConnID THEN state _ established; Process.SetTimeout[@newAllocation, Process.MsecToTicks[1000]]; CommUtilDefs.EnableAborts[@newAllocation]; sendAnAck _ FALSE; maxOisPktLength _ OISCPTypes.maxBytesPerPkt; Process.SetTimeout[@newInputAttn, Process.MsecToTicks[1000]]; CommUtilDefs.EnableAborts[@newInputAttn]; FOR attnCount IN (0..maxPendingAttns] DO attnSeqTable[attnCount] _ 0; ENDLOOP; attnCount _ 0; sentBuffer _ NIL; BufferDefs.QueueInitialize[inOrderQueue]; BufferDefs.QueueInitialize[inAttnQueue]; BufferDefs.QueueInitialize[outOfOrderQueue]; BufferDefs.QueueInitialize[tempQueue]; BufferDefs.QueueInitialize[sentQueue]; Process.SetTimeout[@inOrderQueueNotEmpty, Process.MsecToTicks[1000]]; CommUtilDefs.EnableAborts[@inOrderQueueNotEmpty]; ackRequestPulse _ lastProbePulse _ lastDelayCalculationPulse _ lastPacketReceivedPulse _ ProcessorFace.GetClockPulses[]; probeCounter _ 0; dataPacketRetransmitPulses _ MilliSecondsToPulses[ initialDataPacketRetransmitTime]; -- msecs converted to Pulses probeRetransmitPulses _ MilliSecondsToPulses[ probeMultiplier*initialDataPacketRetransmitTime + 40]; <> delaySum _ [0]; delayCount _ 0; normalRetransUpdates _ doubleRetransUpdates _ 0; -- **temp waitPulses _ MilliSecondsToPulses[timeout]; -- msecs converted to Pulses pleaseStop _ FALSE; Process.SetTimeout[@letClientRun, Process.MsecToTicks[250]]; -- this is also used in DeleteCleanup - do not ENABLE ABORTs Process.SetTimeout[ @retransmitterWakeupTimer, Process.MsecToTicks[initialRetransmitterTime]]; <> <> <> IF state = unestablished THEN BEGIN IF establish THEN BEGIN state _ activeEstablish; retransmitterFork _ FORK Retransmitter[]; receiverFork _ FORK Receiver[]; Process.Yield[]; ActivelyEstablish[]; END ELSE BEGIN state _ waitEstablish; retransmitterFork _ FORK Retransmitter[]; receiverFork _ FORK Receiver[]; Process.Yield[]; WaitUntilEstablished[]; END; END ELSE BEGIN SendSystemPacket[FALSE]; -- gratuitous ack in case we are a server agent now. <> retransmitterFork _ FORK Retransmitter[]; receiverFork _ FORK Receiver[]; Process.Yield[]; END; waitPulses _ MilliSecondsToPulses[PacketStream.infiniteWaitTime]; -- Gets, WaitAttentions default to very long RETURN[@ps, remoteAddr, remoteConnectionID]; END. -- of PacketStreamInstance module LOG Time: January 26, 1981 10:03 AM By: Garlick Action: 1.) added classOfService parameter to reduce buffers in interactive case 2.) added interpretation of Error Protocol packets 3.) increased inactivity timeouts for internet cases 4.) initial data operation timeout set to infinite 5.) acks no longer stimulated by Gets, rather in receiver 6.) send an alocation packet if just opening the window 7.) reinstated slow media adaptive algorithm 8.) enabled Process.Aborting for Gets, Puts, WaitAttn.(635)\23386v14V