// PupBSPProt.bcpl -- Byte Stream Protocol
// companion files are PupBSPStreams.bcpl and PupBSPa.asm
// Copyright Xerox Corporation 1979, 1981

// Last modified November 21, 1981  10:49 AM by Taft

get "Pup.decl"
get "PupRTPInternal.decl"

external
[
//outgoing procedures
BSPPupProc; BSPTimerProc; SendAck; TransmitStrategy; SetTimeout; SearchTQ

//incoming procedures
BSPHandleUncommonPup; RTPFilter; RTPFSM
ReleasePBI; GetPBI; CompletePup; SetPupID
Zero; MultEq; DoubleDifference; DoubleIncrement
SetTimer; TimerHasExpired
Enqueue; Dequeue; Unqueue; InsertAfter; FlushQueue
EnableInterrupts; DisableInterrupts; Min; Max

//statics
maxPupDataBytes; overlapDataWithAck
]

// If overlapDataWithAck is true, TransmitStrategy will generate
// an AData before allocation is completely exhausted, thereby permitting
// continued sending during the AData/Ack round-trip time.
// This improves performance when the round-trip time is large.
// Unfortunately, it has disasterous effects on performance in certain
// situations because of lost packets caused by Ethernet interface
// turnaround latency in Altos and Novas.  We therefore do not enable
// this feature at the present time.

static [ overlapDataWithAck = false ]

//----------------------------------------------------------------------------
let BSPPupProc(pbi) be
//----------------------------------------------------------------------------
// Incoming Pups that are not RTP come here.  pbi is always disposed of,
// either here or by higher-level handler that we call.
[
let soc = pbi>>PBI.socket

// ignore pbi if socket's state is not Open, EndIn, or EndOut
if soc>>BSPSoc.state ls stateOpen % soc>>BSPSoc.state gr stateEndOut then
   [ ReleasePBI(pbi); return ]

switchon pbi>>PBI.pup.type into
   [
   case typeAMark:
   case typeAData:
      soc>>BSPSoc.ackPending = true
   case typeData:
   case typeMark:
      [
      // discard pbi unless all the following conditions are met:
      // from correct source port?
      unless RTPFilter(pbi, true, false) &

      // socket allocation sufficient?
       soc>>BSPSoc.numTPBI ne 0 & soc>>BSPSoc.numIPBI ne 0 &

      // pbi non-empty?
       pbi>>PBI.pup.length gr pupOvBytes &

      // all of pbi's bytes in allocation window?
       DoubleDifference(lv pbi>>PBI.pup.id, lv soc>>BSPSoc.rcvByteID) +
       pbi>>PBI.pup.length - (pupOvBytes+1) uls
       (soc>>BSPSoc.maxIPBI-1)*maxPupDataBytes &

      // non-duplicate?
       InsertInSequence(lv soc>>BSPSoc.bspIQ, pbi) do
         [ ReleasePBI(pbi); endcase ]  // discard

      while pbi ne 0 &
       MultEq(lv soc>>BSPSoc.rcvByteID, lv pbi>>PBI.pup.id) do
         [  // Filled in first hole, advance rcvByteID
         DoubleIncrement(lv soc>>BSPSoc.rcvByteID,
          pbi>>PBI.pup.length-pupOvBytes)
         soc>>BSPSoc.unReadPups = soc>>BSPSoc.unReadPups+1
         pbi = pbi!0
         ]

      SetTimer(lv soc>>BSPSoc.inactivityTimer, inactivityTimeout)
      endcase
      ]

   case typeAck:
      [
      unless RTPFilter(pbi, true, false) &
       DoubleDifference(lv pbi>>PBI.pup.id, lv soc>>BSPSoc.lastAckID) ge 0 do
         [ ReleasePBI(pbi); endcase ]  // wrong source or out of order

      if soc>>BSPSoc.ackRequested then
         [  // Update the adaptive timeout = 2 times the average round-trip
            // delay, exponentially aged over the last 8 samples.
         let t = nil; SetTimer(lv t, 0)
         compiletest alto
            ifso [ t = (t-soc>>BSPSoc.lastADataTime) lshift 3 +4 ]
            ifnot [ t = (t-soc>>BSPSoc.lastADataTime) lshift 1 +1 ]
         soc>>BSPSoc.aDataTimeout =
          (7*soc>>BSPSoc.aDataTimeout+Max(5, Min(1000, t))) rshift 3
         ]

      soc>>BSPSoc.lastAckID^1 = pbi>>PBI.pup.id^1
      soc>>BSPSoc.lastAckID^2 = pbi>>PBI.pup.id^2

// BSPPupProc (cont'd)

      // move all PBIs from bspTQ to a private queue
      let tQ = vec 1
      DisableInterrupts()
      tQ!0 = soc>>BSPSoc.bspTQ.head; tQ!1 = soc>>BSPSoc.bspTQ.tail
      soc>>BSPSoc.bspTQ.head = 0
      EnableInterrupts()

      // decide whether to release, keep, or retransmit each pbi
      let rTQ = vec 1; rTQ!0 = 0  // q of pbis to retransmit
         [ // repeat
         let npbi = Dequeue(tQ); if npbi eq 0 break
         if npbi>>PBI.pup.type eq typeInterrupt then
            [ CompletePup(npbi); loop ]  // Retransmit interrupts
         test DoubleDifference(lv soc>>BSPSoc.lastAckID,
          lv npbi>>PBI.pup.id) ge npbi>>PBI.pup.length-pupOvBytes
            ifso
               [  // This pbi has been acked, release it
               soc>>BSPSoc.unAckedPups = soc>>BSPSoc.unAckedPups-1
               soc>>BSPSoc.unAckedBytes = soc>>BSPSoc.unAckedBytes-
                  (npbi>>PBI.pup.length-pupOvBytes)
               ReleasePBI(npbi)
               soc>>BSPSoc.pupAllocCount = soc>>BSPSoc.pupAllocCount+1
               ]
            ifnot
               [  // pbi not yet acked, decide whether to retransmit
               test soc>>BSPSoc.aDataCount-npbi>>PBI.timer gr 0 &
                soc>>BSPSoc.ackRequested
                  ifso  // Yes, retain for retransmission
                     [
                     TransmitStrategy(npbi, false)
                     InsertInSequence(rTQ, npbi) // can't be duplicate
                     ]
                  ifnot  // No, just put back on bspTQ
                     Enqueue(lv soc>>BSPSoc.bspTQ, npbi)
               ]
         ] repeat

      // Update allocations
      soc>>BSPSoc.ackRequested = false
      if soc>>BSPSoc.pupAllocCount ge pupAllocHysteresis then
         [  // Time to attempt to increase maxPupAlloc
         soc>>BSPSoc.pupAllocCount = 0
         soc>>BSPSoc.maxPupAlloc =
          Min(soc>>BSPSoc.maxPupAlloc+1, soc>>BSPSoc.maxOPBI)
         ]
      soc>>BSPSoc.bytesPerPup = Min(pbi>>PBI.pup.words^1, maxPupDataBytes)
      soc>>BSPSoc.pupAlloc = Min(soc>>BSPSoc.maxPupAlloc,
       pbi>>PBI.pup.words^2 - soc>>BSPSoc.unAckedPups)
      soc>>BSPSoc.byteAlloc = pbi>>PBI.pup.words^3 - soc>>BSPSoc.unAckedBytes
      ReleasePBI(pbi)

      // if we have any PBIs to retransmit, do so now
      if rTQ!0 ne 0 then
         [
         TransmitStrategy(rTQ!1)  // Maybe turn tail into AData
         while rTQ!0 ne 0 do CompletePup(Dequeue(rTQ))
         ]

      SetTimeout(soc)
      SetTimer(lv soc>>BSPSoc.inactivityTimer, inactivityTimeout)
      endcase
      ]

// BSPPupProc (cont'd)

   case typeInterrupt:
   case typeInterruptReply:
   case typeEnd:
      BSPHandleUncommonPup(soc, pbi)  // it always disposes of pbi
      endcase

   case typeError:
      if pbi>>PBI.pup.words^11 eq 3 %  // port IQ overflow
       pbi>>PBI.pup.words^11 eq #1007 then  // gateway OQ overflow
         [
         // Got a congestion error.  Attempt to throttle back output
         // by decreasing the maximum outgoing Pup allocation.
         soc>>BSPSoc.maxPupAlloc = Max(soc>>BSPSoc.maxPupAlloc-1, 1)
         soc>>BSPSoc.pupAllocCount = 0
         ]
      // let higher-level handler look at it also

   default:
      // non-BSP Pups get passed to higher-level handler, which always
      // disposes of pbi
      (soc>>BSPSoc.bspOtherPupProc)(pbi)
   ]

if soc>>BSPSoc.ackPending then SendAck(soc)
]

//----------------------------------------------------------------------------
and InsertInSequence(queue, pbi) = valof
//----------------------------------------------------------------------------
// Inserts pbi into the proper place in queue to maintain ascending order of
// Pup IDs.  Queue must already be in order.
// Returns true normally; false if the pbi's ID duplicates one already in
// the queue.
[
// Test first for the most common case of the new pbi going on the end of
// the queue.  If it doesn't belong there, then initiate a search from the
// front of the queue for the right place to insert it.
test queue!0 eq 0 %
 DoubleDifference(lv pbi>>PBI.pup.id, lv (queue!1)>>PBI.pup.id) gr 0
   ifso Enqueue(queue, pbi)
   ifnot
      [ // Search queue for right place to insert new pbi
      let pred = queue
      let d = nil

      // assert: the queue is non-empty and pbi's ID <= the tail item's ID.
      // Therefore this loop is guaranteed to terminate.
         [ // repeat
         d = DoubleDifference(lv pbi>>PBI.pup.id, lv (pred!0)>>PBI.pup.id)
         if d le 0 then break
         pred = pred!0
         ] repeat

      if d eq 0 then resultis false  // duplicate
      InsertAfter(queue, pred, pbi)
      ]

resultis true
]

//----------------------------------------------------------------------------
and BSPTimerProc(soc) be
//----------------------------------------------------------------------------
// Called whenever the socket bspTimer expires
[
// do nothing if socket's state is not Open, EndIn, or EndOut
unless soc>>BSPSoc.state ls stateOpen % soc>>BSPSoc.state gr stateEndOut do
   [
   unless soc>>BSPSoc.noInactivityTimeout do
      if TimerHasExpired(lv soc>>BSPSoc.inactivityTimer) then
         [ RTPFSM(soc, CLST, 0); return ]  // Connection has died, abort it

   if soc>>BSPSoc.ackPending then SendAck(soc)
   if soc>>BSPSoc.interruptOut then
      [
      let pbi = SearchTQ(soc, true)  // Search for outstanding interrupt
      if pbi ne 0 then CompletePup(pbi)  // Retransmit if found one
      ]

   // Generate AData probe unconditionally.  If we have no data to move,
   // probe once every 15 seconds to ensure that the connection is alive.
   let pbi = GetPBI(soc, true)
   test pbi ne 0
      ifso
         [  // Probe with zero-length AData if can get pbi
         unless soc>>BSPSoc.ackRequested do
            SetTimer(lv soc>>BSPSoc.lastADataTime, 0)
         soc>>BSPSoc.ackRequested = true
         soc>>BSPSoc.aDataCount = soc>>BSPSoc.aDataCount+1
         SetPupID(pbi, lv soc>>BSPSoc.xmitByteID)
         CompletePup(pbi, typeAData, pupOvBytes)
         ]
      ifnot
         [  // Probe with a pbi on the TQ otherwise
         pbi = SearchTQ(soc, false)  // Look for a non-interrupt
         if pbi ne 0 then
            [ TransmitStrategy(pbi, true); CompletePup(pbi) ]
         ]
   ]

SetTimeout(soc)
]

//----------------------------------------------------------------------------
and SendAck(soc) be
//----------------------------------------------------------------------------
[
let pbi = GetPBI(soc, true); if pbi ne 0 then
   [
   soc>>BSPSoc.ackPending = false
   SetPupID(pbi, lv soc>>BSPSoc.rcvByteID)
   pbi>>PBI.pup.words^1 = maxPupDataBytes  // Bytes per pup
   pbi>>PBI.pup.words^2 =  // Pup allocation
      soc>>BSPSoc.maxIPBI-1-soc>>BSPSoc.unReadPups
   pbi>>PBI.pup.words^3 =  // Byte allocation
      maxPupDataBytes*pbi>>PBI.pup.words^2
   soc>>BSPSoc.sentZeroAlloc = pbi>>PBI.pup.words^2 eq 0
   CompletePup(pbi, typeAck, pupOvBytes+6)

   // Flush received data packets we aren't acknowledging, since
   // we know they will be retransmitted and we don't want the
   // retransmissions to overflow our socket IQ if we are slow
   // in processing them.
   while soc>>BSPSoc.bspIQ.head ne 0 do
      [
      pbi = soc>>BSPSoc.bspIQ.tail
      if DoubleDifference(lv pbi>>PBI.pup.id,
       lv soc>>BSPSoc.rcvByteID) ls 0 then break
      Unqueue(lv soc>>BSPSoc.bspIQ, pbi)
      ReleasePBI(pbi)
      ]
   ]
]

//----------------------------------------------------------------------------
and SearchTQ(soc, lookForInterrupt) = valof
//----------------------------------------------------------------------------
// Unqueues and returns the last pbi on the BSP TQ, looking for
// an Interrupt if lookForInterrupt is true, anything else if false.
// We return the last pbi (rather than the first) so that retransmitting it
// leaves the order of pbis on the TQ unchanged.  Returns zero if none found.
[
let rpbi = 0
let pbi = soc>>BSPSoc.bspTQ.head
while pbi ne 0 do
   [
   if (pbi>>PBI.pup.type eq typeInterrupt) eqv lookForInterrupt then
      [ rpbi = pbi; break ]
   pbi = pbi!0
   ]
if rpbi ne 0 then Unqueue(lv soc>>BSPSoc.bspTQ, rpbi)
resultis rpbi
]

//----------------------------------------------------------------------------
and TransmitStrategy(pbi, makeA; numargs na) be
//----------------------------------------------------------------------------
// Diddles type depending on makeA or allocation remaining.
// Current algorithm is that we send AData when allocation
// falls below 33% of that given in the last received Ack.
// This may be expressed as:
//   pupAlloc le k * (pupAlloc + unAckedPups)
// where k = 0.33 and it is assumed that pupAlloc+unAckedPups
// gives the original Pup allocation.  This is equivalent to
//   (1-k)/k * pupAlloc le unAckedPups.
// For k = 0.33, (1-k)/k = 2.  This is the derivation of the
// comparison "pupAlloc lshift 1 le soc>>BSPSoc.unAckedPups" below.
[
let soc = pbi>>PBI.socket
unless na gr 1 & makeA do
   test soc>>BSPSoc.ackRequested
      ifso makeA = false
      ifnot
         [
         let pupAlloc = Min(Min(Min(soc>>BSPSoc.pupAlloc,
            soc>>BSPSoc.byteAlloc/soc>>BSPSoc.bytesPerPup),
            soc>>BSPSoc.numOPBI-1), soc>>BSPSoc.numTPBI-1)
         makeA = (overlapDataWithAck?
            pupAlloc lshift 1 le soc>>BSPSoc.unAckedPups,
            pupAlloc le 0)
         ]
if makeA then
   [
   // Unless an AData is already outstanding, start measuring
   // the round-trip delay for computing the adaptive timeout.
   unless soc>>BSPSoc.ackRequested do
      SetTimer(lv soc>>BSPSoc.lastADataTime, 0)
   soc>>BSPSoc.ackRequested = true
   soc>>BSPSoc.aDataCount = soc>>BSPSoc.aDataCount+1
   ]
pbi>>PBI.pup.type = selecton pbi>>PBI.pup.type into
   [
   case typeData:
   case typeAData: makeA? typeAData, typeData
   case typeMark:
   case typeAMark: makeA? typeAMark, typeMark
   ]
]

//----------------------------------------------------------------------------
and SetTimeout(soc) be
//----------------------------------------------------------------------------
SetTimer(lv soc>>BSPSoc.bspTimer, (soc>>BSPSoc.interruptOut %
 soc>>BSPSoc.unAckedPups gr 0 % soc>>BSPSoc.pupAlloc le 0 %
 soc>>BSPSoc.byteAlloc le 0 % soc>>BSPSoc.bytesPerPup le 0) ?
  (soc>>BSPSoc.ackRequested? soc>>BSPSoc.aDataTimeout,
   Max(soc>>BSPSoc.aDataTimeout, outstandingDataTimeout)),
  idleTimeout)