-- RPC: Reliable transmission and reception of packets

-- RPCPktIO.mesa

-- Andrew Birrell  March 15, 1983 3:27 pm

DIRECTORY
BufferDefs			USING[ PupBuffer ],
CedarSnapshot		USING[ After, Register ],
CommUtilDefs		USING[ AllocateIocbs ],
DriverDefs			USING[ ChangeNumberOfInputBuffers ],
DriverTypes			USING[ Encapsulation, ethernetEncapsulationBytes,
								 ethernetEncapsulationOffset ],
EthernetOneFace	USING[ ControlBlock, ControlBlockRecord, controlBlockSize,
								 DeviceHandle, GetNextDevice, GetStatus, hearSelf,
								 nullDeviceHandle, QueueOutput ],
Frame				USING[ Free ],
Heap					USING[ systemZone ],
Inline				USING[ LongCOPY ],
Process				USING[ Detach, MsecToTicks, Pause, Yield ],
ProcessOperations	USING[ HandleToIndex, ReadPSB ],
PSB					USING[ PsbIndex, PsbNull ],
PupDefs				USING[ GetFreePupBuffer, GetHopsToNetwork, GetLocalPupAddress,
								 PupAddress, PupPackageMake, PupRouterSendThis,
								 PupSocket, PupSocketDestroy, PupSocketMake,
								 ReturnFreePupBuffer, veryLongWait],
PupTypes			USING[ PupAddress, PupHostID, PupNetID, PupSocketID ],
MesaRPC			USING[ CallFailed, unencrypted ],
RPCInternal			USING[ DecryptPkt, DoSignal, EncryptPkt, ReplyToRFA,
								 RPCBinding, RPCPktStreams, RPCSecurity, ServerMain ],
RPCLupine			USING[ DataLength, Dispatcher, Header, RPCPkt ],
RPCPkt				USING[ CallCount, Header, Machine,
								 PktExchangeState, PktID, pktLengthOverhead ],
RPCPrivate			USING[ GetRPCPackets, rpcSocket, ReturnBuffer ],
RPCWatch			USING[],
Runtime				USING[ IsBound ],
System				USING[ GetClockPulses, MicrosecondsToPulses, Pulses ];

RPCPktIO: MONITOR
IMPORTS CedarSnapshot, CommUtilDefs, DriverDefs, EthernetOneFace, Frame, Heap, Inline,
        Process, ProcessOperations, PupDefs, MesaRPC, RPCInternal,
        RPCPrivate, Runtime, System
EXPORTS RPCLupine--Encapsulation,Header--, RPCPkt--PktExchange,IdleReceive--,
        RPCWatch--SetSpyProc--
SHARES  BufferDefs, RPCLupine =

BEGIN

Encapsulation: PUBLIC TYPE = DriverTypes.Encapsulation;

Header: PUBLIC TYPE = RPCPkt.Header;
      
ConcreteHeader: PROC[abstract: LONG POINTER TO RPCLupine.Header]
             RETURNS[LONG POINTER TO Header] = INLINE
  { RETURN[ abstract ] };

callSequence: RPCPkt.CallCount _ 0; -- monotonic from this host --



-- ******** IOCB management ******** --

-- IOCB's are kept on free chain.  Allocation and freeing are imbedded in
-- "PktExchange" procedure.
-- IOCB's are allocated in first 64K of memory, in resident memory.
-- Use "ControlBlock" proc to get pointer to give to EthernetFace.

IOCBrecord: TYPE = RECORD[next: IOCB];
IOCB: TYPE = LONG POINTER TO IOCBrecord;
iocbSize: CARDINAL = MAX[EthernetOneFace.controlBlockSize, SIZE[IOCBrecord]];

freeIOCBs: IOCB _ NIL;



-- ******** sending and receiving packets ******** --

firstDevice: EthernetOneFace.DeviceHandle =
  EthernetOneFace.GetNextDevice[EthernetOneFace.nullDeviceHandle];

myAddr: PupTypes.PupAddress;
myNet: PupTypes.PupNetID;
myDeviceHost: PupTypes.PupHostID;
myHost: RPCPkt.Machine;  

sent: CARDINAL _ 0;
recvd: CARDINAL _ 0;
retransmitted: CARDINAL _ 0;

-- "PktExchange" implements a reliable packet exchange.  There are five cases:
--    sending: transmit data, wait for ack
--    receiving: transmit ack, wait for data
--    call: transmit data, wait for data
--    endCall: transmit data, wait for ack or data (data => start of new call)
--    authReq: transmit RFA, wait for data (like "call", but no retransmissions)
-- Data and RFA packets are retransmitted until an ack is received.
-- Ack packets are retransmitted only in response to "pleaseAck" requests
-- No acknowledgement after 14 transmissions is fatal (CallFailed[timeout])
-- When the transmitted packet has been acknowledged (if needed), a ping is
-- sent periodically (ack packet saying pleaseAck).  If necessary the ping is
-- retransmitted until it has been acked.  Failure to receive an ack for the ping
-- is fatal (CallFailed[timeout]).  Provided pings continue to be acknowledged,
-- there is no limit on how long we will wait for the next data packet.  If the
-- client gets impatient, he can abort this process.  (This corresponds to the
-- local-machine semantics of a procedure call.)

minRetransmitMsecs:	CARDINAL = 100;	-- retransmit delay for local net --
msecsPerHop:			CARDINAL = 500;	-- approximate typical gateway hop delay? --
minPingMsecs:			CARDINAL = 5000;	-- initial interval between pings --
maxPingSecs:			CARDINAL = 300;	-- long-term ping interval --
maxTransmissions:		CARDINAL _ 20;		-- give up after too many transmissions --
signalTimeout:			BOOLEAN _ TRUE;	-- debugging switch --

-- The retransmission delay is incremented by minPingMsecs each time we timeout,
-- but is reset when we receive the desired packet.  If n=maxTransmissions
-- and i=minRetransmitMsecs and j=hops*msecsPerHop, we give up after
--   i*(n*n+n)/2+n*j msecs.
-- For the local network, that comes to 21 seconds.
-- For a two-hop route, that comes to 41 seconds.
-- The ping delay is doubled at each ping, up to maxPingSecs, which is 5 minutes;
-- The maxPingSecs value is reached after about 5 minutes.

minRetransmitPulses: System.Pulses =
                  System.MicrosecondsToPulses[LONG[1000] * minRetransmitMsecs];
pulsesPerHop: System.Pulses =
                  System.MicrosecondsToPulses[LONG[1000] * msecsPerHop];
minPingPulses: System.Pulses =
                  System.MicrosecondsToPulses[LONG[1000] * minPingMsecs];
maxPingPulses: System.Pulses =
                    System.MicrosecondsToPulses[LONG[1000]*LONG[1000] * maxPingSecs];
transmitLocalPkts: BOOLEAN _ TRUE;


-- Constants for WaitUntilSent (inside PktExchange) --

shortWait: System.Pulses = -- 10 msecs --
                  System.MicrosecondsToPulses[LONG[1000] * 10];
longerWait: System.Pulses = -- 5 secs --
                    System.MicrosecondsToPulses[LONG[1000]*LONG[1000] * 5];


PktExchange: PUBLIC PROC[inPkt: RPCLupine.RPCPkt, length: RPCLupine.DataLength,
                           maxlength: RPCLupine.DataLength,
                         state: RPCPkt.PktExchangeState,
                           signalHandler: RPCLupine.Dispatcher _ NIL]
                  RETURNS[newPkt: BOOLEAN, newLength: RPCLupine.DataLength] =
  BEGIN
  -- On exit if newPkt=TRUE, the packet has been decrypted if needed. --

  -- Normally, transmits inPkt and copies result into inPkt.
  -- If a signal occurs, calls DoSignal which handles the signal protocol
  -- up to the last resumption packet.  DoSignal returns the last resume
  -- packet for us to transmit, which we do by assigning it to outPkt;
  -- that packet was allocated in DoSignal's local frame, which we must
  -- later deallocate.
  outPkt:      RPCLupine.RPCPkt _ inPkt; -- altered after a signal --
  outPktFrame: POINTER _ NIL; -- DoSignal's local frame --

  DO -- loop for signal handlers --

     sentTime: System.Pulses; -- initialized after sending any packet --
  
     iocb:      IOCB _ NIL;
     -- IOCB management procs are imbedded for efficiency! --

     GetIOCB: ENTRY PROC = INLINE
       BEGIN
       IF freeIOCBs # NIL
       THEN { iocb _ freeIOCBs; freeIOCBs _ freeIOCBs.next }
       ELSE { iocb _ CommUtilDefs.AllocateIocbs[iocbSize] }
       END;

     ControlBlock: PROC RETURNS[ EthernetOneFace.ControlBlock ] = INLINE
       { RETURN[LOOPHOLE[iocb]] };

     ReturnIOCB: PROC = INLINE
       BEGIN
       InnerReturn: ENTRY PROC = INLINE
         { iocb.next _ freeIOCBs; freeIOCBs _ iocb };
       IF iocb # NIL THEN { WaitUntilSent[]; InnerReturn[] };
       END;

     WaitUntilSent: PROC = INLINE
       BEGIN
       -- Horrible!  We must ensure that the ethernet microcode isn't still
       -- using our IOCB when we free or re-use it. There are two problems:
       --  (a) a suitable response coming in before we've finished
       --      retransmitting our packet;
       --  (b) the standard EthernetOneDriver resetting the microcode
       --      (so it forgets our request).
       -- To handle (a) we busy-wait for up to 10msec, then wait for single
       -- ticks thereafter;  to handle (b) we assume our iocb is no longer
       -- being used after it's pending for 5 secs.  To do better seems to
       -- require changes to the EthernetOneFace.
       WHILE EthernetOneFace.GetStatus[ControlBlock[]] = pending
       DO IF System.GetClockPulses[] - sentTime < shortWait
          THEN Process.Yield[]
          ELSE IF System.GetClockPulses[] - sentTime < longerWait
               THEN Process.Pause[1]
               ELSE EXIT -- assume someone reset the microcode --;
       ENDLOOP;
       END;

     NewCallNumber: ENTRY PROC RETURNS[RPCPkt.CallCount] = INLINE
       { RETURN[ callSequence _ callSequence+1 ] };

     reply: BufferDefs.PupBuffer;
     recvdHeader: LONG POINTER TO Header;
     myPSB:     PSB.PsbIndex =
                   ProcessOperations.HandleToIndex[ProcessOperations.ReadPSB[]];
     acked:     BOOLEAN;
     thisPktID: RPCPkt.PktID;
     header:    LONG POINTER TO Header = @outPkt.header;
     localHost: BOOLEAN = header.destHost = myHost;
     localNet:  BOOLEAN = header.destHost.net = myNet;
     pingPulses: System.Pulses _ minPingPulses;
     header.srceHost _ myHost;
     header.srceSoc _ RPCPrivate.rpcSocket;
     header.srcePSB _ myPSB;
     IF header.pktID.pktSeq = 0 -- first packet of a call;  yucky interface! --
     THEN BEGIN
          header.type _ SELECT state FROM
              sending =>   [0, rpc, notEnd, pleaseAck, call],
              call =>      [0, rpc, end,    dontAck,   call],
              authReq =>   [0, rpc, end,    dontAck,   rfa],
            ENDCASE => --receiving, endCall-- ERROR;
          header.pktID.callSeq _ NewCallNumber[];
          header.pktID.pktSeq _ 1;
          acked _ FALSE;
          END
     ELSE BEGIN
          header.type _ SELECT state FROM
              sending =>   [0, rpc, notEnd, pleaseAck, data],
              receiving => [0, rpc, end,    dontAck,   ack],
              call =>      [0, rpc, end,    dontAck,   data],
              endCall =>    [0, rpc, end,    dontAck,   data],
            ENDCASE => --authReq-- ERROR;
          IF state # receiving --header.type.class = data --
          THEN { header.pktID.pktSeq _ header.pktID.pktSeq+1;
                 acked_FALSE }
          ELSE acked _ TRUE;
          END;
     thisPktID _ header.pktID;
     SetWanting[myPSB];

     DO -- loop for pings --
        ENABLE UNWIND => { ClearWanting[myPSB]; ReturnIOCB[];
                           IF outPktFrame # NIL THEN Frame.Free[outPktFrame] };
        BEGIN
        transmissions: CARDINAL _ 0;
        retransmitPulses: System.Pulses _
          IF localNet
          THEN minRetransmitPulses
          ELSE [minRetransmitPulses +
                       pulsesPerHop * PupDefs.GetHopsToNetwork[header.destHost.net]];

        IF outPkt.convHandle # MesaRPC.unencrypted
        THEN header.length _ RPCInternal.EncryptPkt[outPkt, length]
        ELSE header.length _ length + RPCPkt.pktLengthOverhead;
        header.oddByte _ no;

        DO -- loop for retransmissions --
           BEGIN
           IF NOT localNet
           THEN GeneralSend[outPkt]
           ELSE BEGIN
                IF localHost
                AND NOT( EthernetOneFace.hearSelf AND transmitLocalPkts )
                THEN LocalSend[outPkt];
                IF NOT localHost OR transmitLocalPkts
                THEN BEGIN
                     IF iocb = NIL
                     THEN GetIOCB[]
                     ELSE WaitUntilSent[];
		     -- Pup checksum --
                     outPkt.data[header.length-RPCPkt.pktLengthOverhead] _ 177777B;
                     outPkt.encapsulation _ Encapsulation[ethernetOne[,,,,,,
                                          header.destHost.host, myDeviceHost, pup]];
                     EthernetOneFace.QueueOutput[firstDevice,
                         @outPkt.encapsulation + DriverTypes.ethernetEncapsulationOffset,
                         header.length + (1+DriverTypes.ethernetEncapsulationBytes)/2,
                         ControlBlock[]]
                     END;
                  END;
           sentTime _ System.GetClockPulses[];
           sent _ sent+1;
           -- wait for response: an ack or the reply or a timeout --

           DO -- loop for each incoming packet (including garbage) --
              reply _ MyReceive[myPSB, sentTime,
                                (IF acked THEN pingPulses ELSE retransmitPulses)];
              IF reply = NIL
              THEN IF acked
                   THEN GOTO ping
                   ELSE { header.type.ack _ pleaseAck; GOTO retransmit };
              recvdHeader _ LOOPHOLE[@reply.pupLength];
              IF recvdHeader.type.class = rfa
              THEN BEGIN
                   IF RPCInternal.ReplyToRFA[reply,
                                 header,--encrypted--
                                 thisPktID--clear--,
                                 inPkt.convHandle]
                   THEN NULL-- can't set "acked", because we must retransmit our data
                        -- packet until we obtain the correct destPSB from some ack pkt --;
                   LOOP
                   END;
               IF outPkt.convHandle # MesaRPC.unencrypted
               AND recvdHeader.conv = header.conv
               AND recvdHeader.srceHost = header.destHost
               THEN [,newLength] _ RPCInternal.DecryptPkt[recvdHeader, outPkt.convHandle]
               ELSE newLength _ recvdHeader.length - RPCPkt.pktLengthOverhead;
               IF recvdHeader.conv = header.conv
               AND recvdHeader.srceHost = header.destHost
               AND recvdHeader.pktID.activity = thisPktID.activity
               THEN -- pkt is related to our call --
                   SELECT TRUE FROM
                     recvdHeader.pktID.callSeq = thisPktID.callSeq =>
                       -- pkt is in our call --
                       SELECT TRUE FROM
                         -- a) pkt has next sequence number to ours --
                         recvdHeader.pktID.pktSeq = 1+thisPktID.pktSeq =>
                           BEGIN
                           IF state = sending OR state = endCall
                           THEN -- he's not allowed to generate that pktSeq! --
                                ERROR MesaRPC.CallFailed[runtimeProtocol !
                                                 UNWIND => GiveBackBuffer[reply] ];
                           SELECT recvdHeader.type.class FROM
                             data => GOTO done;
                             ack =>
                               BEGIN
                               -- This can happen if state=call and callee sent next data
                               -- pkt, but it was lost and now he is responding to our
                               -- retransmission or ping.  Soon, he will retransmit his
                               -- data pkt. --
                               acked _ TRUE;
                               GiveBackBuffer[reply];
                               END;
                           ENDCASE => --call,rfa--
                             ERROR MesaRPC.CallFailed[runtimeProtocol !
                                                 UNWIND => GiveBackBuffer[reply] ];
                           END;
                         -- b) pkt has same sequence number as ours --
                         recvdHeader.pktID.pktSeq = thisPktID.pktSeq =>
                           BEGIN
                           SELECT recvdHeader.type.class FROM
                             ack => -- acknowledgement of our packet --
                               BEGIN
                               IF header.type.class = call
                               THEN header.destPSB _ recvdHeader.srcePSB;
                               acked _ TRUE;
                               IF state = sending OR state = endCall
                               THEN {GiveBackBuffer[reply]; reply_NIL; GOTO done}
                               ELSE -- state = call, authReq, or receiving --
                                    -- Even if "pleaseAck", we don't need to ack it,
                                    -- because other end should send data pkt soon --
                                    GiveBackBuffer[reply];
                               END;
                             data, call => -- retransmisson of his packet --
                               IF state = receiving
                               THEN IF recvdHeader.type.ack = pleaseAck
                                    THEN BEGIN
                                         GiveBackBuffer[reply]; reply _ NIL;
                                         GOTO retransmit -- we're already sending an ack --
                                         END
                                    ELSE GiveBackBuffer[reply]
                               ELSE ERROR MesaRPC.CallFailed[runtimeProtocol !
                                                 UNWIND => GiveBackBuffer[reply] ];
                           ENDCASE => --rfa --
                             ERROR MesaRPC.CallFailed[runtimeProtocol !
                                                 UNWIND => GiveBackBuffer[reply] ];
                           END;
                         -- c) pkt has earlier sequence number than ours --
                         recvdHeader.pktID.pktSeq < thisPktID.pktSeq =>
                           GiveBackBuffer[reply]; -- no need to ack it --
                         -- d) pkt has some future sequence number --
                       ENDCASE =>
                           ERROR MesaRPC.CallFailed[runtimeProtocol !
                                            UNWIND => GiveBackBuffer[reply] ];
                     recvdHeader.pktID.callSeq > thisPktID.callSeq AND state=endCall =>
                       BEGIN
                       IF recvdHeader.type.class # call
                       THEN ERROR MesaRPC.CallFailed[runtimeProtocol !
                                             UNWIND => GiveBackBuffer[reply] ];
                       -- acks our last packet, so we can handle the call --
                       GOTO done
                       END
                   ENDCASE => -- wrong call: let someone else do it --
                     { recvdHeader.destPSB _ PSB.PsbNull; EnqueueAgain[reply] }
              ELSE -- wrong conversation or activity: let someone else do it --
                { recvdHeader.destPSB _ PSB.PsbNull; EnqueueAgain[reply] };
           -- Incoming RFA packets don't reach here. --
           ENDLOOP--for each incoming packet--;
           EXITS
             retransmit =>
               BEGIN
               transmissions _ transmissions+1;
               IF (transmissions = maxTransmissions AND signalTimeout)
               OR state = authReq
                  -- Don't retransmit RFA: caller will retransmit call pkt.
                  -- Otherwise, if a spurious worker process occurred because of call pkt
                  -- retransmission before response to RFA, then the spurious worker
                  -- process sits needlessly retransmitting RFA's until it times out.
               THEN { SIGNAL MesaRPC.CallFailed[timeout]; transmissions _ 0 };
               retransmitted _ retransmitted+1;
               retransmitPulses _ [retransmitPulses + minRetransmitPulses];
               END;
           END;
        ENDLOOP-- for each transmission --;
        EXITS
          ping =>
            BEGIN
            header.type _ [0, rpc, end, pleaseAck, ack];
            length _ 0; header.pktID _ thisPktID;
            acked _ FALSE;
            pingPulses _ [MIN[maxPingPulses, pingPulses * 2]];
            END;
        END;
     -- only exit from loop is "GOTO done" or UNWIND --
     REPEAT done =>
       BEGIN
       -- This isn't covered by any UNWIND --
       ClearWanting[myPSB]; ReturnIOCB[];
       IF outPktFrame # NIL THEN Frame.Free[outPktFrame];
       IF reply = NIL
       THEN { --restore clear pktID-- header.pktID _ thisPktID; RETURN[FALSE,] }
       ELSE BEGIN
            IF recvdHeader.outcome = signal AND state = call
            AND signalHandler # NIL
            THEN [outPkt, length, outPktFrame] _
                     RPCInternal.DoSignal[reply, newLength, signalHandler, inPkt.convHandle]
            ELSE BEGIN
                 IF newLength > maxlength
                 THEN ERROR MesaRPC.CallFailed[runtimeProtocol !
                                      UNWIND => GiveBackBuffer[reply]];
                 Inline.LongCOPY[from: recvdHeader, to: @inPkt.header,
                                 nwords: recvdHeader.length];
                 GiveBackBuffer[reply];
                 RETURN[TRUE,newLength]
                 END;
            END
       END;
     ENDLOOP-- for each ping--;
     -- We get here only after coming back from a call of DoSignal --
  ENDLOOP-- for signal handlers --;
  -- we can't get here! --
  END;
  
GeneralSend: PROC[pkt: RPCLupine.RPCPkt] =
  BEGIN
  b: BufferDefs.PupBuffer = PupDefs.GetFreePupBuffer[];
  Inline.LongCOPY[from: @(pkt.header), to: @(b.pupLength),
                  nwords: ConcreteHeader[@pkt.header].length];
  PupDefs.PupRouterSendThis[b];
  END;

LocalSend: PROC[pkt: RPCLupine.RPCPkt] =
  BEGIN
  -- handle talking to ourselves if the ethernet head can't --
  b: BufferDefs.PupBuffer = PupDefs.GetFreePupBuffer[];
  Inline.LongCOPY[from: @(pkt.header), to: @(b.pupLength),
                  nwords: ConcreteHeader[@pkt.header].length];
  IF NOT AcceptPkt[b] THEN PupDefs.ReturnFreePupBuffer[b];
  END;

 
idlerPkt:      BufferDefs.PupBuffer _ NIL;
idlerCond:     CONDITION _ [timeout:0];
waiterCond:    CONDITION _ [timeout:Process.MsecToTicks[minRetransmitMsecs]];
WaiterArray:   TYPE = ARRAY PSB.PsbIndex OF BufferDefs.PupBuffer;
waiterPkts:    LONG POINTER TO WaiterArray _
                                 Heap.systemZone.NEW[WaiterArray _ ALL[NIL]];
wanting:       PACKED ARRAY PSB.PsbIndex OF BOOLEAN _ ALL[FALSE]; -- PSB expects a pkt --

SetWanting: ENTRY PROC[myPSB: PSB.PsbIndex] = INLINE
  { wanting[myPSB] _ TRUE };

ClearWanting: PROC[myPSB: PSB.PsbIndex] = INLINE
  BEGIN
  spare: BufferDefs.PupBuffer;
  InnerClear: ENTRY PROC = INLINE
    BEGIN
    wanting[myPSB] _ FALSE;
    IF (spare _ waiterPkts[myPSB]) # NIL THEN waiterPkts[myPSB] _ NIL;
    END;
  InnerClear[];
  IF spare # NIL THEN GiveBackBuffer[spare] --ignore it, outside monitor--;
  END;
  

MyReceive: ENTRY PROC[myPSB: PSB.PsbIndex, sentTime, waitTime: System.Pulses]
              RETURNS[recvd: BufferDefs.PupBuffer] = INLINE
  BEGIN
  ENABLE UNWIND => NULL;
  -- Returns NIL if no packet arrives within waitTime pulses after sentTime --
  DO IF (recvd _ waiterPkts[myPSB]) = NIL
     THEN IF System.GetClockPulses[] - sentTime < waitTime
          THEN WAIT waiterCond
          ELSE EXIT
     ELSE { waiterPkts[myPSB] _ NIL; EXIT };
  ENDLOOP;
  END;

RecvdPktTooLong: ERROR = CODE;
KillServer: ERROR = CODE;
servers:    CARDINAL _ 0;
serversMax: CARDINAL _ 20;
idlers:     CARDINAL _ 0;
idlersMax:  CARDINAL = 6;
idlersMin:  CARDINAL = 2;
-- Number of server processes will never exceed "serversMax".
-- If number of idle servers exceeds "idlersMax", one will abort.
-- If number of idle servers drops below "idlersMin", one is forked.

IdleReceive: PUBLIC PROC[pkt: RPCLupine.RPCPkt, maxlength: CARDINAL] =
  BEGIN
  b: BufferDefs.PupBuffer;
  InnerIdler: ENTRY PROC = INLINE
    BEGIN
    IF idlers >= idlersMax
    THEN { servers_servers-1; RETURN WITH ERROR KillServer[] };
    idlers _ idlers + 1;
    DO WAIT idlerCond; IF idlerPkt # NIL THEN EXIT ENDLOOP;
    IF idlers < idlersMin AND servers < serversMax
    THEN { servers_servers+1; Process.Detach[FORK Server[]] };
    b _ idlerPkt;
    BEGIN
      recvdHeader: LONG POINTER TO Header = LOOPHOLE[@b.pupLength];
      idlerPkt _ LOOPHOLE[idlerPkt.next,BufferDefs.PupBuffer]; b.next _ NIL;
      IF recvdHeader.length > maxlength
      THEN ERROR RecvdPktTooLong[] --NULL??--
      ELSE Inline.LongCOPY[from: @b.pupLength, to: @pkt.header,
                               nwords: recvdHeader.length];
    END;
    END;
  InnerIdler[];
  GiveBackBuffer[b];--outside monitor--
  END;

QueuesScrambled: ERROR = CODE;

spyProc: PROC[BufferDefs.PupBuffer] _ NIL;

SetSpyProc: PUBLIC ENTRY PROC[p: PROC[BufferDefs.PupBuffer]] =
  { spyProc _ p };

EnqueueRecvd: ENTRY PROC[b: BufferDefs.PupBuffer, report: BOOLEAN]
                                  RETURNS[BOOLEAN] =
  BEGIN
  -- despatch packet to appropriate RPC process, if any --
  -- Packet is known to be a Pup, and is addressed to our socket. --
  header: LONG POINTER TO Header = LOOPHOLE[@b.pupLength];
  IF report AND spyProc # NIL THEN spyProc[b];
  IF header.destHost = myHost AND header.type.subType = rpc
  THEN BEGIN
       destPSB: PSB.PsbIndex = header.destPSB;
       recvd _ recvd+1;
       IF destPSB NOT IN (PSB.PsbNull..LAST[PSB.PsbIndex]]
       OR NOT wanting[destPSB]
       THEN -- give to idle process to deal with --
            -- Pkts are dealt with LIFO by idlers, but it doesn't matter (much)
            BEGIN
            IF idlers = 0
            THEN -- server too busy: throw it away! --
                 RETURN[FALSE]
            ELSE { b.next _ idlerPkt; idlerPkt _ b;
                   idlers _ idlers-1; NOTIFY idlerCond };
            END
       ELSE -- someone wants this packet: give them it --
            BEGIN
            IF waiterPkts[destPSB] # NIL THEN RETURN[FALSE];
            waiterPkts[destPSB] _ b; BROADCAST waiterCond;
            END;
       RETURN[TRUE]
       END
  ELSE RETURN[FALSE];
  END;

GiveBackBuffer: PROC[b: BufferDefs.PupBuffer] =
  -- NOTE: calls of this must be made outside our monitor, because
  -- RPCPrivate.ReturnBuffer acquires the EthernetDriver monitor,
  -- and the EthernetDriver may call EnqueueRecvd which acquires
  -- our monitor!
  IF Runtime.IsBound[RPCPrivate.ReturnBuffer]
  THEN RPCPrivate.ReturnBuffer
  ELSE PupDefs.ReturnFreePupBuffer;


AcceptPkt: PROC[b: BufferDefs.PupBuffer] RETURNS[BOOLEAN] =
  { RETURN[ EnqueueRecvd[b: b, report: TRUE] ] };

EnqueueAgain: PUBLIC PROC[b: BufferDefs.PupBuffer] =
  { IF NOT EnqueueRecvd[b: b, report: FALSE] THEN GiveBackBuffer[b] };


Listener: PROC =
  BEGIN
  -- Catch any packets that get as far as the Pup socket level --
  soc: PupDefs.PupSocket = PupDefs.PupSocketMake[local: RPCPrivate.rpcSocket,
                                                 remote:,
                                                 ticks: PupDefs.veryLongWait];
  DO ENABLE ABORTED => EXIT;
     b: BufferDefs.PupBuffer = soc.get[];
     IF NOT AcceptPkt[b] THEN PupDefs.ReturnFreePupBuffer[b];
  ENDLOOP;
  PupDefs.PupSocketDestroy[soc]; 
  END;

listenerProcess: PROCESS;

Server: PROC =
  BEGIN
  RPCInternal.ServerMain[ ! KillServer => CONTINUE ];
  END;




-- ******** Initialization ******** --

Initialize: ENTRY PROC =
  BEGIN
  PupDefs.PupPackageMake[];
  DriverDefs.ChangeNumberOfInputBuffers[TRUE];
  myAddr _ PupDefs.GetLocalPupAddress[RPCPrivate.rpcSocket,NIL];
  myNet _ myAddr.net;
  myDeviceHost _ myAddr.host;
  myHost _ [myNet, myAddr.host];
  START RPCInternal.RPCBinding; -- exports "RPCInternal.exportTable" --
  START RPCInternal.RPCSecurity; -- exports "RPCInternal.firstConversation" --
  START RPCInternal.RPCPktStreams; -- initialize connection states --
  CedarSnapshot.Register[r: Rollback];
  servers _ servers+1; Process.Detach[FORK Server[]];
  listenerProcess _ FORK Listener[];
  RPCPrivate.GetRPCPackets[AcceptPkt];
  END;
  
Restart: INTERNAL PROC =
  BEGIN
  RESTART RPCInternal.RPCPktStreams; -- forget connection counts --
  RESTART RPCInternal.RPCSecurity; -- invalidate local conversations --
  RESTART RPCInternal.RPCBinding; -- nothing? --
  END;

Rollback: ENTRY PROC[after: CedarSnapshot.After] =
  { IF after = rollback THEN Restart[] };

Initialize[];

END.
Κ†–
"Mesa" style˜Iprocš›Οc7œ
œ
*œ
Οk	œžœžœ#žœ žœ.žœežœ€
žœžœžœžœ9žœžœžœ"žœί
žœ<žœ+žœ{žœ6žœbžœ7žœžœžœ3Οbœžœ
žœ·
žœ
œœœ
žœžœžœ
žœ'žœ
žœΟnœžœžœ
žœ
žœ žœ
žœ
žœ
žœžœžœ5œ'œν
œ
žœžœžœžœžœžœ
žœ
žœžœžœ#žœžœžœ5œμ
žœ
žœžœΦœžœ$œžœ,œžœ	$œžœœžœ*œžœžœœπœTžœjžœdžœgžœžœ*žœžœ6œœ/žœ*œ1žœžœ œžœ
žœξ
žœžœ	žœ'žœFœ;œHœGœEœFœœ)œžœžœœžœ
œ *œžœžœ8œ œžœ
žœžœžœžœ
žœžœ9žœ8žœ œžœ
žœ#žœ
žœ
žœ 
œžœžœžœ œžœ
žœžœ>žœžœ
žœ,žœ 
œžœžœžœEœEœ<œ#œ?œ%œEœDœDœ+œžœ<žœ
žœ:žœžœ
žœ@žœ!žœ
žœ
(œ	žœ	žœ 
œžœ
žœ
žœžœ
žœZžœ
žœ
žœžœmžœ0žœ
žœ
žœ*žœ-žœ½
žœ/œžœ
žœžœžœΖ
žœœ
žœižœžœžœ
žœžœžœ…žœœ
žœžœœžœGžœ
žœ	žœžœ?žœ
œ	žœ
žœDžœžœ
žœ$žœžœ:žœžœžœwžœ1žœ@žœZžœ
œžœžœ
žœžœ žœ
žœžœžœ
žœžœ%žœ$žœ
žœžœ#žœ
žœžœžœžœ žœœ€žœžœV9œ
žœ
7œSžœžœžœ#žœ	žœžœ
žœžœ
žœžœ žœ+žœ"žœ,žœ
žœžœG
œ+	œHžœ
žMœBœžœžœžœ8žœ/žœ7žœVžœKžœ/žœ7žœ@žœ
 œžœ
žœ
žœ^œžœ
žœ
žœ-œfžœžœžœ,žœ
/œ!žœWžœ8žœžœ&žœJžœ ˜œ(žœWžœžœœžœWžœ8žœ-œdžœžœžœ%#œ žœ žœ9žœNžœ!žœžœ0žœžœžœ&žœ
)œ%†
œ
$œ6žœ-!œ žœ2žœ
žœFžœ
žœIžœ+žœ"œ*žœ%žœ6žœ
žœWžœ8žœœžœWžœ8žœ2œ{œ,œžœžœRžœ`žœ)žœžœ6žœ
žœSžœ45œžœžœžœ(œ.žœ.žœ
<œ)žœ,,œžœ
žœ+žœ@žœ#žœžœ#:œσ
œžœžœΎ
žœ
žœ
žœ
žœžœ€
žœžœ.žœ
žœ2œžœžœ&œ2žœžœ
žœ!žœ	žœžœœžœ
žœžœ
žœ
žœžœžœžœ
žœŠ
žœ
žœžœ(žœ
žœLžœΪ
žœ
žœžœžœžœžœ@œžœœžœ œžœžœΥ
žœ 	œžœžœ<œ΅
žœ
žœžœ#žœ+žœž	œž	œEžœžœ
žœ
žœ&žœ
žœ
žœ@žœžœ
žœžœ
žœ
žœ
žœ
žœžœ
žœœ 
œžœ
žœžœ
žœžœ œžœžœ
žœžœ" 
œžœ
žœžœžœžœžœžœ
žœžœžœžœ	žœ
žœœžœ 	œžœ
žœžœ<žœ žœžœžœ
žœžœLœžœ
žœžœžœ
žœ9žœ
žœžœ
žœžœžœžœžœžœžœžœžœžœžœžœžœžœžœΕ
œ
 œžœ
žœ#žœžœ 
œžœ
žœžœžœžœžœžœ
žœ
žœ.žœ
žœžœžœ
žœ
žœ
žœžœžœžœ%žœ$žœžœ
žœ
žœ
žœ!žœ/žœžœ&žœ
žœ
œžœvžœžœ&œžœžœžœžœžœ 
œžœ
žœ
žœžœ. œžœ
žœ"žœ$žœ
žœžœ8œBœžœ
žœ
žœ
žœžœžœžœ
žœžœžœžœ
žœžœ;žœ	žœ
žœžœ
žœ
žœžœ
žœžœ
'œ
Dœžœ
žœžœ
%œžœ
žœžœJžœžœžœ
-œ
žœ
žœžœ
žœ
žœ
žœ(ž	œžœ	žœ
žœ	žœžœ
žœ
žœžœ œžœAœ@œ?œœžœ,žœžœ  	œžœžœ
žœžœžœ œžœ
žœ žœ
žœžœžœ œžœžœ?œΪ
žœ
žœ
žœžœ1žœ
žœžœ#žœ&žœžœ œžœžœ+žœžœ&œ 
œžœ
žœžœEžœ>žœZžœ'œžœ-œžœ"œNžœ žœ6žœ œžœ
žœžœžœœžœ$œžœœžœ œžœ
žœ#žœžœžœ
˜ΫΩ
—…—lήxj