-- Copyright (C) 1981, 1984 by Xerox Corporation. All rights reserved.
-- File: SequinImplB.mesa
-- HGM: 10-Dec-84 21:24:46
-- last edited by Hankins: 6-Aug-84 13:38:30
-- Klamath update (pup changes, sequinClosure change)
-- Last edited by Levin: 6-Jul-81 16:15:00
-- Loosely derived (after extensive discussions with Wobber)
-- from Butterfield's Sequin.mesa of August 27, 1979 2:49 PM.
DIRECTORY
Buffer USING [Buffer, BufferObject, Enqueue, GetBuffer, ReturnBuffer],
PupDefs USING [
DequeuePup, EnqueuePup, GetPupContentsBytes, PupBuffer, SetPupContentsBytes],
PupTypes USING [BufferBody, PupType],
Sequin USING [Buffer, noBuffer],
SequinPrivate USING [
bufferPool, Handle, LongHandle, maxAllocate, maxBytes, maxPings, Seconds,
SequenceNumber, SequinID, SequinControl],
Time USING [Current];
SequinImplB: MONITOR LOCKS sequin.LOCK USING sequin: SequinPrivate.Handle
IMPORTS Buffer, PupDefs, SequinPrivate, Time EXPORTS Sequin, SequinPrivate =
BEGIN
-- Types --
Handle: PUBLIC TYPE = SequinPrivate.Handle;
-- Miscellaneous Declarations --
BogusBuffer: ERROR = CODE;
ServerIsSpeakingGreek: ERROR = CODE;
WhereDidIPutThatBuffer: ERROR = CODE;
idleLatency: SequinPrivate.Seconds = 10;
-- Procedures and Signals Exported to Sequin --
Broken: PUBLIC ERROR = CODE;
Get: PUBLIC ENTRY PROCEDURE [sequin: Handle] RETURNS [Sequin.Buffer] =
BEGIN
packet: PupDefs.PupBuffer;
DO
IF sequin.broken THEN RETURN WITH ERROR Broken;
IF (packet ← PupDefs.DequeuePup[@sequin.getQueue]) ~= NIL THEN EXIT;
WAIT sequin.goAhead;
ENDLOOP;
RETURN[PupToSequinBuffer[packet]]
END;
GetIfAvailable: PUBLIC ENTRY PROC [sequin: Handle] RETURNS [Sequin.Buffer] =
BEGIN
packet: PupDefs.PupBuffer;
IF sequin.broken THEN RETURN WITH ERROR Broken;
IF (packet ← PupDefs.DequeuePup[@sequin.getQueue]) = NIL THEN
RETURN[Sequin.noBuffer];
RETURN[PupToSequinBuffer[packet]]
END;
Put: PUBLIC PROCEDURE [sequin: Handle, buffer: Sequin.Buffer] =
BEGIN
packet: PupDefs.PupBuffer = SequinToPupBuffer[buffer];
PupDefs.SetPupContentsBytes[packet, buffer.nBytes];
Send[sequin, packet, data];
END;
GetEmptyBuffer: PUBLIC PROCEDURE RETURNS [Sequin.Buffer] =
BEGIN
packet: PupDefs.PupBuffer = Buffer.GetBuffer[
type: pup, aH: SequinPrivate.bufferPool, function: send];
--is send correct?
PupDefs.SetPupContentsBytes[packet, 0];
RETURN[PupToSequinBuffer[packet]]
END;
ReleaseBuffer: PUBLIC PROCEDURE [buffer: Sequin.Buffer] =
BEGIN
IF buffer = Sequin.noBuffer THEN RETURN;
Buffer.ReturnBuffer[SequinToPupBuffer[buffer]];
END;
-- Procedures exported to SequinPrivate --
Send: PUBLIC ENTRY PROCEDURE [
sequin: Handle, packet: PupDefs.PupBuffer,
control: SequinPrivate.SequinControl] =
BEGIN
UNTIL Compare[
sequin.retransmitSequence + MIN[sequin.allocate, SequinPrivate.maxAllocate],
sequin.id.sendSequence] = ahead OR sequin.broken DO
WAIT sequin.goAhead; ENDLOOP;
IF sequin.broken THEN {Buffer.ReturnBuffer[packet]; RETURN WITH ERROR Broken};
IF control = data AND sequin.state = init THEN {
control ← open; sequin.state ← open};
sequin.id.control ← control;
LOOPHOLE[@packet.pup.pupID, LONG POINTER TO SequinPrivate.SequinID]↑ ←
sequin.id;
IF PupDefs.GetPupContentsBytes[packet] ~= 0 THEN
sequin.id.sendSequence ← sequin.id.sendSequence + 1;
packet.pup.pupType ← sequin.pupType;
SendRequeueable[sequin, packet];
END;
SocketWarmer: PUBLIC PROCEDURE [sequin: Handle] =
-- This procedure is forked as a separate process which tries to keep the
-- socket as "clean" as possible. It absorbs incoming packets and initiates
-- retransmissions as necessary. Good packets are moved to the getQueue.
BEGIN
packet: PupDefs.PupBuffer;
packetID: LONG POINTER TO SequinPrivate.SequinID;
alive: BOOLEAN ← TRUE;
ProcessPacket: PROC [sequin: Handle] RETURNS [alive: BOOLEAN] = INLINE
BEGIN
alive ← TRUE;
IF packet = NIL THEN
BEGIN
SELECT CheckForPing[sequin] FROM
no => NULL;
retransmit => Retransmit[sequin];
check => RespondWithNewPacket[check];
dead => RETURN[FALSE];
ENDCASE;
RETURN
END;
IF PupDefs.GetPupContentsBytes[packet] > SequinPrivate.maxBytes THEN RETURN;
SELECT packet.pup.pupType FROM
error =>
BEGIN OPEN PupTypes;
SELECT packet.pup.errorCode FROM
noProcessPupErrorCode, cantGetTherePupErrorCode, hostDownPupErrorCode,
eightHopsPupErrorCode => RETURN[FALSE];
ENDCASE => RETURN;
END;
sequin.pupType => NULL;
ENDCASE => RETURN;
packetID ← LOOPHOLE[@packet.pup.pupID];
IF packetID.control = broken THEN RETURN[FALSE];
SELECT Compare[packetID.sendSequence, sequin.id.receiveSequence] FROM
duplicate => RETURN;
ahead =>
BEGIN
DiscardAckedPackets[sequin];
IF packetID.control = restart THEN Retransmit[sequin];
IF ~sequin.restartRequested THEN
BEGIN
sequin.restartRequested ← TRUE;
RespondWithCurrentPacket[sequin, restart];
END;
RETURN
END;
ENDCASE;
-- we've seen everything from the server
ResetPinging[];
DiscardAckedPackets[sequin];
SELECT packetID.control FROM
data => {EnqueueArrival[sequin]; RespondWithNewPacket[ack]};
dallying => {RespondWithCurrentPacket[sequin, quit]; RETURN[FALSE]};
restart => Retransmit[sequin];
check => RespondWithCurrentPacket[sequin, ack];
ack, nop => NULL;
ENDCASE => ERROR ServerIsSpeakingGreek;
END;
CheckForPing: ENTRY PROCEDURE [sequin: Handle]
RETURNS [{no, retransmit, check, dead}] = INLINE
BEGIN
SELECT sequin.pings FROM
0 =>
IF Time.Current[] - sequin.lastPacketTime <= idleLatency
AND sequin.retransmitQueue.length = 0 THEN RETURN[no];
SequinPrivate.maxPings => RETURN[dead];
ENDCASE;
IF sequin.state = init THEN RETURN[no]; -- don't ping until first real packet
sequin.pings ← sequin.pings + 1;
sequin.recentRestart ← sequin.restartRequested ← FALSE;
RETURN[
IF sequin.retransmitQueue.length = 1 AND sequin.pings = 1 THEN retransmit
ELSE check]
END;
ResetPinging: PROCEDURE = INLINE {
sequin.pings ← 0; sequin.lastPacketTime ← Time.Current[]};
DiscardAckedPackets: ENTRY PROCEDURE [sequin: Handle] =
BEGIN
sequin.allocate ← packetID.allocate;
IF Compare[packetID.receiveSequence, sequin.retransmitSequence] = ahead THEN
BEGIN
skipped: CARDINAL ← 0;
UNTIL packetID.receiveSequence = sequin.retransmitSequence
OR skipped = sequin.retransmitQueue.length DO
b: PupDefs.PupBuffer ← PupDefs.DequeuePup[@sequin.retransmitQueue];
IF b = NIL THEN EXIT; -- buffer hasn't made it to queue yet
IF sequin.retransmitSequence ~= LOOPHOLE[b.pup.pupID,
SequinPrivate.SequinID].sendSequence THEN {
PupDefs.EnqueuePup[@sequin.retransmitQueue, b];
skipped ← skipped + 1;
LOOP};
Buffer.ReturnBuffer[b];
sequin.retransmitSequence ← sequin.retransmitSequence + 1;
skipped ← 0;
ENDLOOP;
BROADCAST sequin.goAhead;
END;
END;
EnqueueArrival: ENTRY PROCEDURE [sequin: Handle] = INLINE
BEGIN
PupDefs.EnqueuePup[@sequin.getQueue, packet];
sequin.id.receiveSequence ← sequin.id.receiveSequence + 1;
sequin.recentRestart ← sequin.restartRequested ← FALSE;
NOTIFY sequin.goAhead;
END;
RespondWithNewPacket: PROCEDURE [control: SequinPrivate.SequinControl] =
BEGIN
packet ← Buffer.GetBuffer[
type: pup, aH: SequinPrivate.bufferPool, function: send];
--is send correct?
packetID ← LOOPHOLE[@packet.pup.pupID];
RespondWithCurrentPacket[sequin, control];
END;
RespondWithCurrentPacket: ENTRY PROCEDURE [
sequin: Handle, control: SequinPrivate.SequinControl] =
BEGIN
sequin.id.control ← control;
packetID↑ ← sequin.id;
packet.pup.pupType ← sequin.pupType;
PupDefs.SetPupContentsBytes[packet, 0];
sequin.socket.put[packet];
packet ← NIL;
END;
MarkDead: ENTRY PROCEDURE [sequin: Handle] =
BEGIN
sequin.broken ← TRUE;
sequin.state ← destroyed;
BROADCAST sequin.goAhead;
END;
Retransmit: ENTRY PROCEDURE [sequin: Handle] =
BEGIN
skipped: CARDINAL ← 0;
seq: SequinPrivate.SequenceNumber ← sequin.retransmitSequence;
IF sequin.recentRestart THEN RETURN;
UNTIL skipped = sequin.retransmitQueue.length DO
buffer: PupDefs.PupBuffer = PupDefs.DequeuePup[@sequin.retransmitQueue];
bufferID: LONG POINTER TO SequinPrivate.SequinID;
IF buffer = NIL THEN ERROR WhereDidIPutThatBuffer;
IF (bufferID ← LOOPHOLE[@buffer.pup.pupID]).sendSequence ~= seq THEN {
PupDefs.EnqueuePup[@sequin.retransmitQueue, buffer];
skipped ← skipped + 1;
LOOP};
bufferID.inPart ← sequin.id.inPart;
SendRequeueable[sequin, buffer];
sequin.recentRestart ← TRUE;
skipped ← 0;
seq ← IF seq = LAST[SequinPrivate.SequenceNumber] THEN 0 ELSE seq + 1;
ENDLOOP;
END;
ResetPinging[];
WHILE alive DO
packet ← sequin.socket.get[];
IF ~(alive ← ProcessPacket[sequin]) THEN MarkDead[sequin];
IF packet ~= NIL THEN Buffer.ReturnBuffer[packet];
ENDLOOP;
END;
Requeue: PROCEDURE [buffer: Buffer.Buffer] =
BEGIN
longUnspecified: SequinPrivate.LongHandle ← LOOPHOLE[buffer.requeueData];
DoRequeue: ENTRY PROCEDURE [sequin: Handle] = INLINE
BEGIN
sequin.buffersToRequeue ← sequin.buffersToRequeue - 1;
IF sequin.state = destroyed THEN
BEGIN
Buffer.ReturnBuffer[buffer];
IF sequin.buffersToRequeue = 0 THEN {NOTIFY sequin.goAhead; RETURN};
END
ELSE Buffer.Enqueue[@sequin.retransmitQueue, buffer];
END;
DoRequeue[longUnspecified.shortHandle];
END;
-- Internal Procedures --
SendRequeueable: INTERNAL PROC [sequin: Handle, buffer: PupDefs.PupBuffer] =
BEGIN
longUnspecified: SequinPrivate.LongHandle ← [sequin, 0];
buffer.requeueProcedure ← Requeue;
buffer.requeueData ← LOOPHOLE[longUnspecified];
sequin.buffersToRequeue ← sequin.buffersToRequeue + 1;
sequin.socket.put[buffer];
END;
Compare: PROCEDURE [a, b: SequinPrivate.SequenceNumber]
RETURNS [{equal, duplicate, ahead}] =
BEGIN
maxGetAhead: CARDINAL = 128;
RETURN[
SELECT TRUE FROM
a = b => equal,
a > b => IF a <= b + maxGetAhead THEN ahead ELSE duplicate,
ENDCASE => IF b <= a + maxGetAhead THEN duplicate ELSE ahead]
END;
PupToSequinBuffer: PROC [b: PupDefs.PupBuffer] RETURNS [Sequin.Buffer] = INLINE
{
RETURN[
[
data: LOOPHOLE[@b.pup.pupBody], nBytes: PupDefs.GetPupContentsBytes[b],
maxBytes: SequinPrivate.maxBytes]]};
positionInBuffer: CARDINAL = 42; -- data offset into PupDefs.PupBuffer
SequinToPupBuffer: PROC [b: Sequin.Buffer] RETURNS [PupDefs.PupBuffer] =
BEGIN
IF b.data = NIL THEN ERROR BogusBuffer;
RETURN[LOOPHOLE[b.data - positionInBuffer]];
-- old: RETURN[LOOPHOLE[b.data-SIZE[pupBytes Buffer.PupBufferObject]]]
-- **** it is possible for the defn of PupDefs.PupBuffer and
-- PupTypes.BufferBody to change such that this var positionInBuffer is
-- no longer correct even if check in mainline code does not fail.
END;
-- mainline code:
IF SIZE[pupBytes PupTypes.BufferBody] + SIZE[Buffer.BufferObject] # (67B + 12B)
THEN ERROR;
-- defn of these two types has changed and the positionInBuffer var must be altered.
END.