-- File: PupPktHot.mesa,  Last Edit:
  -- HGM  October 13, 1980  8:08 PM
  -- MAS  August 20, 1980  12:57 PM

-- Copyright  Xerox Corporation 1979, 1980

DIRECTORY
  InlineDefs: FROM "InlineDefs" USING [LowHalf],
  StatsDefs: FROM "StatsDefs" USING [StatBump, StatIncr],
  CommUtilDefs: FROM "CommUtilDefs" USING [
    GetTicks, SetTimeout, MsecToTicks],
  DriverDefs: FROM "DriverDefs" USING [
    MaybeGetFreePupBuffer, doDebug, doStats, Glitch],
  PupStream: FROM "PupStream" USING [StreamClosing],
  PupPktPrivateDefs: FROM "PupPktPrivateDefs",
  PupPktDefs: FROM "PupPktDefs",
  PupDefs: FROM "PupDefs" USING [
    Byte, PupBuffer, DequeuePup, EnqueuePup,
    GetFreePupBuffer, ReturnFreePupBuffer, PupRouterSendThis],
  BufferDefs: FROM "BufferDefs" USING [ReturnFreeBuffer],
  PupTypes: FROM "PupTypes" USING [PupAddress, maxDataBytesPerGatewayPup];

PupPktHot: MONITOR
    RETURNS [POINTER TO PupPktPrivateDefs.InstanceData]
 LOCKS mine.lock
  IMPORTS
    InlineDefs,
    StatsDefs, CommUtilDefs, PupStream, DriverDefs,
    PupPktPrivateDefs, PupDefs, BufferDefs
  EXPORTS PupPktPrivateDefs =
BEGIN OPEN StatsDefs, DriverDefs, PupPktPrivateDefs, PupDefs;

mine: PupPktPrivateDefs.InstanceData;

NeitherDataNorMark: PUBLIC ERROR = CODE;
BufferAlreadyRequeued: PUBLIC ERROR = CODE;
StreamNotOpen: PUBLIC ERROR = CODE;

-- Only called by InputPacket
Diff: PROCEDURE [a, b: LONG INTEGER] RETURNS [INTEGER] =
  BEGIN
  maxInteger: INTEGER = 77777B;
  temp: LONG INTEGER _ a-b;
  SELECT TRUE FROM
    temp>maxInteger => RETURN [maxInteger];
    temp<-maxInteger => RETURN [-maxInteger];
    ENDCASE => RETURN[InlineDefs.LowHalf[temp]];
  END;

Retransmitter: ENTRY PROCEDURE =
  -- Retransmit things which have not been acknowledged in a reasonable time.
  -- Such things include RFCs and ENDs as well as data.
  BEGIN OPEN mine;
  now:  CARDINAL;
  UNTIL pleaseDie DO
    now _ CommUtilDefs.GetTicks[];
    SELECT state FROM
      open =>
        BEGIN
        IF now-timer>pingTicks AND ping THEN
          BEGIN
          probeCounter _ pingRetransmissions;
          allocatedPups _ 0;  -- will start probing
          END;
        IF now-timer>ctlRetransmitTicks AND (outEnd=0 OR allocatedPups=0) THEN
          ProbeForAck[];
        END;
      talking, finishing =>
        BEGIN
        DO  -- recycle things that have timed out
          IF sentBuffer=NIL THEN sentBuffer _ DequeuePup[@sentQueue];
          IF sentBuffer=NIL THEN EXIT;
          IF ackedID-Flip[sentBuffer.pupID]>0 THEN
            BEGIN -- this packet has been ACKed already
            IF (unackedPups_unackedPups-1)=0 THEN
              BEGIN
              SELECT state FROM  -- last packet has been ACKed
                talking => state _ open;
                finishing => state _ end;
                ENDCASE;
              BROADCAST stateChange;
              END;
            ReturnFreePupBuffer[sentBuffer]; sentBuffer_NIL;
            END
            ELSE EXIT;
          ENDLOOP;
        IF sentBuffer#NIL AND now-timer>retransmitTicks THEN
          BEGIN
          ProbeForAck[];
          IF now-timer>retransmitTicks THEN
            BEGIN  -- couldn't get buffer, use one of ours
            IF sentBuffer.pupType=data THEN sentBuffer.pupType_aData;
            timer _ CommUtilDefs.GetTicks[];
            PupRouterSendThis[sentBuffer];
            IF doStats THEN StatIncr[statDataPacketsRetransmitted];
            sentBuffer _ DequeuePup[@sentQueue];
            END;
          END;
        END;
      halfOpen => IF now-timer>ctlRetransmitTicks THEN SendRfc[@mine];
      end => IF now-timer>ctlRetransmitTicks THEN SendEnd[@mine];
      closed =>
        BEGIN
        DO  -- flush anything left on the sentQueue
          IF sentBuffer=NIL THEN sentBuffer _ DequeuePup[@sentQueue];
          IF sentBuffer=NIL THEN EXIT;
          unackedPups _ unackedPups-1;
          ReturnFreePupBuffer[sentBuffer];
          sentBuffer _ NIL;
          ENDLOOP;
        END;
      ENDCASE;
    IF outIntPending AND now-outIntTime>ctlRetransmitTicks THEN SendInt[@mine];
    WAIT retransmitterReady;
    ENDLOOP;
  UNTIL unackedPups=0 DO
    DO  -- flush anything left on the sentQueue
      IF sentBuffer=NIL THEN sentBuffer _ DequeuePup[@sentQueue];
      IF sentBuffer=NIL THEN EXIT;
      unackedPups _ unackedPups-1;
      ReturnFreePupBuffer[sentBuffer];
      sentBuffer _ NIL;
      ENDLOOP;
    -- If a buffer is on a device output queue, wait until it comes back.
    IF unackedPups#0 THEN WAIT retransmitterReady;
    ENDLOOP;
  END;


ProbeForAck: INTERNAL PROCEDURE =
  BEGIN OPEN mine;
  b: PupBuffer;
  IF (b_MaybeGetFreePupBuffer[])=NIL THEN RETURN;
  b.pupLength _ bytesPerPupHeader;
  b.pupType _ aData;
  b.pupID _ Flop[nextOutputID];
  b.source.socket _ local.socket;
  b.dest _ remote;
  timer _ CommUtilDefs.GetTicks[];
  aDataOut _ FALSE;
  clumpsSinceBump _ 0;
  PupRouterSendThis[b];
  IF doStats THEN StatIncr[statProbesSent];
  IF (probeCounter_probeCounter+1)>retransmitionsBeforeAbort THEN
    SmashClosed[@mine,transmissionTimeout];
  IF probeCounter>probesBeforePanic THEN
    retransmitTicks _ MIN[2*retransmitTicks,maxRetransmitTicks];
  END;

ThrottleForward: PROCEDURE =
  BEGIN OPEN mine;
  old: CARDINAL _ pathMaxAllocate;
  clumpsSinceBump _ 0;
  -- We can actually get one packet ahead at this point.
  IF retransmitTicks=maxRetransmitTicks THEN RETURN;
  pathMaxAllocate _ MIN[pathMaxAllocate+1,myMaxAllocate];
  retransmitTicks _ (retransmitTicks*pathMaxAllocate)/old;
  END;

ThrottleBack: INTERNAL PROCEDURE =
  BEGIN OPEN mine;
  old: CARDINAL;
  IF pathMaxAllocate=1 THEN
    BEGIN  -- This is a desperate attempt to avoid an instability
    -- It is/(was?) also a nasty bug under some strange case that Andrew found
    pause: CONDITION;
CommUtilDefs.SetTimeout[@pause,CommUtilDefs.MsecToTicks[maxRetransmitTime]];
    WAIT pause;
    END;
  UNTIL throttle=0 OR retransmitTicks=minRetransmitTicks OR pathMaxAllocate=1 DO
    old _ pathMaxAllocate;
    pathMaxAllocate _ pathMaxAllocate-1;
    -- Beware of rounding down, assume return ack takes as long as a send packet
-- This goes unstable if much of the time is due to somebody else's packets
--    retransmitTicks _ ((retransmitTicks+old)*old)/(old+1);
    throttle _ throttle-1;
    ENDLOOP;
  clumpsSinceBump _ throttle _ 0;
  END;

SendAck: INTERNAL PROCEDURE =
  BEGIN OPEN mine;
  b: PupBuffer;
  IF c#NIL THEN BEGIN b_c; c_NIL; END
  ELSE IF (b_MaybeGetFreePupBuffer[])=NIL THEN RETURN;
  b.pupBody _ ack [
    dataBytesPerPup,
    MAX[0,INTEGER[myMaxAllocate-inputQueue.length]],
    byteAllocate];
  allocatedID _ nextInputID+byteAllocate;
  b.pupType _ ack;
  b.pupID _ Flop[nextInputID];
  b.pupLength _ bytesPerAck;
  b.source.socket _ local.socket;
  b.dest _ remote;
  PupRouterSendThis[b];
  IF doStats THEN StatIncr[statAcksSent];
  IF state=open AND outEnd#0 AND allocatedPups#0 THEN
    timer _ CommUtilDefs.GetTicks[];  -- avoid pinging
  sendAck _ FALSE;
  END;

Get: ENTRY PROCEDURE RETURNS [b: PupBuffer] =
  BEGIN OPEN mine; ENABLE UNWIND => NULL;
  IF inputQueue.length=0 THEN
    SELECT TRUE FROM
      (state=closed) => ERROR PupStream.StreamClosing[whyClosed,text];
      dontWait => RETURN[NIL];
      ENDCASE => WAIT inputReady;
  IF inputQueue.length#0 THEN b _ DequeuePup[@inputQueue]
  ELSE b _ NIL;
  IF b=NIL THEN
    BEGIN
    IF state=closed THEN ERROR PupStream.StreamClosing[whyClosed,text];
    RETURN;
    END;
  IF doStats THEN SELECT b.pupType FROM
    data, aData =>
      BEGIN
      StatIncr[statDataPacketsReceived];
      StatBump[statDataBytesReceived,b.pupLength-bytesPerPupHeader];
      END;
    mark, aMark => StatIncr[statMarksReceived];
    ENDCASE => Glitch[NeitherDataNorMark];
  IF inputQueue.length=0 AND sendAck THEN SendAck[];
  END;

Put: PROCEDURE [b: PupBuffer]=
  BEGIN OPEN mine;
  IF doDebug AND b.requeueProcedure#BufferDefs.ReturnFreeBuffer THEN
    Glitch[BufferAlreadyRequeued];
  SELECT state FROM
    open, talking  =>
      BEGIN
        IF doStats THEN
          BEGIN
          StatIncr[statDataPacketsSent];
          StatBump[statDataBytesSent,b.pupLength-bytesPerPupHeader];
          END;
        SendPacket[b];
        END;
    idle, halfOpen => Glitch[StreamNotOpen];
    ENDCASE --end, closed, finishing-- => StreamDied[@mine,b];
  END;

PutMark: PROCEDURE [byte: Byte] =
  BEGIN OPEN mine;
  b: PupBuffer;
  b _ GetFreePupBuffer[];
  b.pupBytes[0] _ byte;
  b.pupLength _ bytesPerPupHeader+1;
  b.pupType _ aMark;
  SendPacket[b];
  IF doStats THEN StatIncr[statMarksSent];
  END;

SendPacket: ENTRY PROCEDURE [b: PupBuffer] =
  BEGIN OPEN mine;
  SELECT state FROM
    open, talking =>
      BEGIN
      b.pupID _ Flop[nextOutputID];
      IF b.pupLength=bytesPerPupHeader THEN probeCounter _ 1
      ELSE
        BEGIN
        b.requeueProcedure _ LOOPHOLE[PutOnSentQueue];
        state _ talking;
        unackedPups _ unackedPups+1;
        END;
      b.source.socket _ local.socket;
      b.dest _ remote;
      nextOutputID _ nextOutputID+(b.pupLength-bytesPerPupHeader);
      IF b.pupType=data  -- we don't use mark, only aMark
      AND ~((maxOutputID-nextOutputID)>0 AND allocatedPups>unackedPups) THEN
        b.pupType _ aData;
      IF b.pupType#data THEN  -- aMark or aData
        BEGIN
        timer _ CommUtilDefs.GetTicks[];
        aDataOut _ TRUE;
        END;
      PupRouterSendThis[b];
      IF b.pupType#data THEN WaitToSend[ ! UNWIND => NULL];
      END;
    idle, halfOpen => Glitch[StreamNotOpen];
    ENDCASE --end, closed, finishing-- => StreamDied[@mine, b ! UNWIND => NULL];
  END;

WaitToSend: INTERNAL PROCEDURE =
  BEGIN OPEN mine;
  SELECT state FROM
    open, talking =>
      BEGIN
      -- Wait until all our packets have been acked so we don't shoot them down.
      DO
        SELECT state FROM
          open => EXIT;
          talking => WAIT stateChange;
          ENDCASE => StreamDied[@mine,NIL];
        ENDLOOP;
      -- now wait for allocate
      UNTIL (maxOutputID-nextOutputID)>0 AND allocatedPups>unackedPups DO
        SELECT state FROM
          -- not really, but .....
          open, talking => WAIT stateChange;
          ENDCASE => StreamDied[@mine,NIL];
        ENDLOOP;
      IF throttle>0 THEN ThrottleBack[];
      END;
    idle, halfOpen => Glitch[StreamNotOpen];
    ENDCASE --end, closed, finishing-- => StreamDied[@mine,NIL];
  END;

PutOnSentQueue: ENTRY PROCEDURE [b: PupBuffer] =
  BEGIN OPEN mine;  -- better not SIGNAL from here
  IF ackedID-Flip[b.pupID]>0 THEN
    BEGIN
    IF (unackedPups_unackedPups-1)=0 THEN
      BEGIN
      SELECT state FROM  -- last packet has been ACKed
        talking => state _ open;
        finishing => state _ end;
        ENDCASE => SendAbort[@mine];
      BROADCAST stateChange;
      END;
    ReturnFreePupBuffer[b];
    END
  ELSE IF sentBuffer=NIL THEN sentBuffer _ b ELSE EnqueuePup[@sentQueue,b];
  END;

Slurp: PROCEDURE =
  BEGIN OPEN mine;
  b: PupBuffer;
  UNTIL pleaseDie DO
    b _ socket.get[];
    IF b#NIL THEN InputPacket[b];
    ENDLOOP;
  END;

InputPacket: ENTRY PROCEDURE [b: PupBuffer] =
  BEGIN OPEN mine;
  thisID: LONG INTEGER _ Flip[b.pupID];
  c _ b;

  IF ~(b.pupType=rfc OR b.pupType=error)
  AND b.source.socket#remote.socket THEN
    BEGIN
    IF doStats THEN StatIncr[statPacketsRejectedBadSource];
    ReturnFreePupBuffer[b];
    c _ NIL;
    END;

  SELECT b.pupType FROM
    data, aData, mark, aMark =>
    SELECT state FROM
    open, talking, end =>
      BEGIN
      SELECT b.pupType FROM
        aData, aMark => sendAck _ TRUE;
        ENDCASE;
      IF b.pupLength>bytesPerPupHeader THEN
        BEGIN
        offset: INTEGER _ Diff[thisID,nextInputID];
        SELECT offset FROM
          0 =>
            BEGIN  -- nice - just what we wanted
            nextInputID _ nextInputID+(c.pupLength-bytesPerPupHeader);
            EnqueuePup[@inputQueue,c];
            NOTIFY inputReady;
            c _ NIL;
            END;
          ENDCASE =>
            BEGIN
            IF doStats THEN SELECT offset FROM
              IN (0..duplicateWindow] => StatIncr[statDataPacketsReceivedEarly];
              IN (-duplicateWindow..0) => StatIncr[statDataPacketsReceivedAgain];
              ENDCASE => StatIncr[statDataPacketsReceivedVeryLate];
            ReturnFreePupBuffer[c];
            c _ NIL;
            END;
        END
      ELSE  -- funny length from way back there
        BEGIN
        IF doStats THEN
          IF b.pupLength=bytesPerPupHeader AND b.pupType=aData
          THEN StatIncr[statProbesReceived]
          ELSE StatIncr[statEmptyFunnys];
        END;
      -- answer probes immediately
      IF sendAck AND inputQueue.length=0 THEN SendAck[];
      END;
    halfOpen => SendRfc[@mine];
    ENDCASE --idle, closed, finishing-- => SendAbort[@mine];
   
  ack =>
    BEGIN
    IF b.pupLength<bytesPerAck THEN
      BEGIN IF doStats THEN StatIncr[statMouseTrap]; GOTO SkipThisAck; END;
    IF doStats THEN StatIncr[statAcksReceived];
    -- Try to avoid the funny stable SLOW case
    IF (thisID-ackedID)<0 OR (thisID=ackedID AND probeCounter=0) THEN
      BEGIN IF doStats THEN StatIncr[statDuplicateAcks]; GOTO SkipThisAck; END;
    probeCounter _ 0;
    IF aDataOut THEN
      BEGIN
      responseTime, myRetrTime: CARDINAL;
      responseTime _ CommUtilDefs.GetTicks[]-timer;
      responseTime _ MIN[responseTime,maxRetransmitTicks];
      -- retransmitTicks _ 2*Smooth[responseTime]
      -- retransmitTicks _ 2*((7/8)*(retransmitTicks/2)+(1/8)*responseTime)
      -- retransmitTicks _ (7*retransmitTicks+2*responseTime)/8
      myRetrTime _ retransmitTicks;
      myRetrTime _ (6*myRetrTime+myRetrTime+2*responseTime)/8;
      myRetrTime _ MAX[minRetransmitTicks,MIN[myRetrTime,maxRetransmitTicks]];
      retransmitTicks _ myRetrTime;
      aDataOut _ FALSE;
      clumpsSinceBump _ clumpsSinceBump+1;
      IF clumpsSinceBump>clumpsBeforeBump THEN ThrottleForward[];
      END;
    IF allocatedPups=0 THEN BROADCAST stateChange;
    hisMaxAllocate _ b.numberOfPupsAhead;
    IF hisMaxAllocate=0 THEN
      BEGIN
      probeCounter _ 1;
      IF doStats THEN StatIncr[statEmptyAlloc];
      END;
    allocatedPups _ MIN[hisMaxAllocate,pathMaxAllocate];
    IF outEnd=0 THEN BROADCAST stateChange;  -- in case first time
    outEnd _ MIN[b.maximumBytesPerPup,dataBytesPerPup];
    IF ~sameNet THEN outEnd_MIN[outEnd,PupTypes.maxDataBytesPerGatewayPup];
    IF (thisID-maxOutputID)+b.numberOfBytesAhead>0
    THEN maxOutputID _ b.numberOfBytesAhead+thisID;
    ackedID _ thisID;
    IF sentBuffer=NIL THEN sentBuffer _ DequeuePup[@sentQueue];
    WHILE sentBuffer#NIL AND thisID>Flip[sentBuffer.pupID] DO
      IF (unackedPups_unackedPups-1)=0 THEN
        BEGIN
        SELECT state FROM  -- last packet has been ACKed
          talking => state _ open;
          finishing => state _ end;
          ENDCASE => SendAbort[@mine];
        BROADCAST stateChange;
        END;
      ReturnFreePupBuffer[sentBuffer];
      sentBuffer _ DequeuePup[@sentQueue];
      ENDLOOP;
    UNTIL sentBuffer=NIL DO
      PupRouterSendThis[sentBuffer];
      IF doStats THEN StatIncr[statDataPacketsRetransmitted];
      sentBuffer _ DequeuePup[@sentQueue];
      ENDLOOP;
    EXITS SkipThisAck => NULL;
    END;
   
  ENDCASE => GotOther[@mine,c];

IF c#NIL THEN BEGIN ReturnFreePupBuffer[c]; c_NIL; END;
END;

MyGetLocalAddress: PROCEDURE RETURNS [PupTypes.PupAddress] =
  BEGIN
  RETURN[GetLocalAddress[@mine]];
  END;

MyGetSenderSizeLimit: PROCEDURE RETURNS [CARDINAL] =
  BEGIN
  RETURN[GetSenderSizeLimit[@mine]];
  END;

MySendAttention: ENTRY PROCEDURE =
  BEGIN OPEN mine; ENABLE UNWIND => NULL;
  WHILE outIntPending DO WAIT stateChange; ENDLOOP;
  outIntPending _ TRUE;
  SendInt[@mine];
  END;

MyWaitForAttention: ENTRY PROCEDURE =
  BEGIN OPEN mine; ENABLE UNWIND => NULL;
  WHILE seenIntSeq=inIntSeq DO WAIT waitingForInterrupt; ENDLOOP;
  seenIntSeq _ seenIntSeq+1;
  END;

-- initialization
mine.me.get _ Get;  -- do cold stuff here to avoid recompilation hassels
mine.me.put _ Put;
mine.me.putMark _ PutMark;
mine.me.getSenderSizeLimit _ MyGetSenderSizeLimit;
mine.me.sendAttention _ MySendAttention;
mine.me.waitForAttention _ MyWaitForAttention;
mine.me.getLocalAddress _ MyGetLocalAddress;
mine.slurp _ Slurp;
mine.retransmitter _ Retransmitter;
RETURN[@mine];
END.  -- PupPktHot