-- File: PtPkt.mesa,  Last Edit:
  -- MAS  April 21, 1980  6:52 PM
  -- HGM  November 19, 1979  12:02 PM

DIRECTORY
  IODefs: FROM "IODefs" USING [WriteChar, WriteLine, WriteOctal, WriteString],
  ProcessDefs: FROM "ProcessDefs" USING [Detach],
  StatsDefs: FROM "StatsDefs" USING [StatBump, StatIncr],
  PupStream: FROM "PupStream" USING [StreamClosing],
  PupPktDefs: FROM "PupPktDefs" USING [
    CreatePupPktStreamListener, DestroyPupListener, PupListener,
    PupPktStream, PupPktStreamCreate, PupPktStreamDestroy, PupPktStreamMake],
  PupDefs: FROM "PupDefs" USING [
    GetFreePupBuffer, ReturnFreePupBuffer, PupAddress, PupBuffer,
    GetPupContentsBytes, SetPupContentsWords, SecondsToTocks],
  PtDefs: FROM "PtDefs" USING [
    CursorBits, recvIndex, sendIndex, slow,
    DataMismatch, LengthMismatch, packetCycle, PtInterface,
    PtLookerReady, Random, SeeStorms, UnListen, Done, WaitUntilDone];

PtPkt: PROGRAM [pt: PtDefs.PtInterface]
  IMPORTS
    IODefs, ProcessDefs,
    StatsDefs, PupStream, PupPktDefs, PupDefs, PtDefs
  EXPORTS PtDefs =
BEGIN OPEN pt, IODefs, StatsDefs, PupPktDefs, PupDefs, PtDefs;

listener: PupListener _ NIL;
globalPushPS: PupPktStream _ NIL;
secondPushPS: PupPktStream _ NIL;
globalPullPS: PupPktStream _ NIL;

Header: PROCEDURE [s, t: STRING, where: PupAddress] =
  BEGIN
  WriteString[s];
  SELECT length FROM
    short => WriteString[" short"];
    long => WriteString[" long"];
    random => WriteString[" random length"];
    cyclic => WriteString[" cyclic length"];
    ignore => WriteString[" unknown length"];
    ENDCASE => ERROR;
  WriteString[" packets"];
  IF where.host=myHost AND where.net=myNet
   THEN WriteString[" locally"]
   ELSE BEGIN WriteString[" "]; WriteString[t];
     IF where.net#myNet
      THEN BEGIN WriteString[" net "]; WriteOctal[where.net] END;
     WriteString[" host "]; WriteOctal[where.host]; END;
  WriteLine["."];
  WriteString["The data pattern is"];
  SELECT data FROM
    alternating => WriteString[" alternating ones and zeros"];
    ones => WriteString[" all ones"];
    zeros => WriteString[" all zeros"];
    random => WriteString[" random"];
    cyclic => WriteString[" cyclic"];
    ignore => WriteString[" not being generated/checked"];
    ENDCASE => ERROR;
  WriteLine["."];
  SeeStorms[];
  END;

Push: PROCEDURE =
  BEGIN
  pushPS: PupPktStream _ NIL;
  pushI: CARDINAL;
  k, l: CARDINAL;
  r: CARDINAL _ 0;
  b: PupBuffer;
  BEGIN  -- extra BEGIN for EXITS
  pushPS _ PupPktStreamCreate[pushHim,SecondsToTocks[1] !
    PupStream.StreamClosing =>
      BEGIN
      WriteString["Connection failed: "];
      IF text#NIL THEN
        BEGIN
        WriteChar['(];
        WriteString[text];
        WriteString[") "];
        END;
      WriteLine[whyText[why]];
      GOTO Failed;
      END];
  SELECT NIL FROM
    globalPushPS => globalPushPS _ pushPS;
    secondPushPS => secondPushPS _ pushPS;
    ENDCASE => ERROR;
    IF info THEN WriteChar['o];
  PtLookerReady[];
  SELECT length FROM  -- can't use zero
    short => l_1;
    cyclic => l_1;  -- start at short
    long, ignore => l_dataWordsPerPup;
    random => NULL;
    ENDCASE => ERROR;
  UNTIL stopFlag DO
    FOR pushI IN [0..packetCycle) UNTIL stopFlag DO
      ENABLE PupStream.StreamClosing =>
        BEGIN
        WriteString["Push PktStream closed: "];
        IF text#NIL THEN
          BEGIN
          WriteChar['(];
          WriteString[text];
          WriteString[") "];
          END;
        WriteLine[whyText[why]];
        GOTO Closed;
        END;
      b_GetFreePupBuffer[];
      SELECT length FROM
        short, long, ignore => NULL;
        cyclic => l_pushI;
        random => l_((r_Random[r]) MOD dataWordsPerPup)+1;
        ENDCASE => ERROR;
      SELECT data FROM
        ignore => NULL;
        cyclic => FOR k IN [0..l) DO b.pupWords[k]_k+pushI; ENDLOOP;
        ones => FOR k IN [0..l) DO b.pupWords[k]_177777B; ENDLOOP;
        zeros => FOR k IN [0..l) DO b.pupWords[k]_0; ENDLOOP;
        alternating => FOR k IN [0..l) DO b.pupWords[k]_125252B; ENDLOOP;
        random => FOR k IN [0..l) DO b.pupWords[k]_(r_Random[r]); ENDLOOP;
        ENDCASE => ERROR;
      SetPupContentsWords[b,l];
      IF ~doStats THEN
        BEGIN
        StatIncr[statDataPacketsSent];
        StatBump[statDataBytesSent,l*2];
        END;
      pushPS.put[b]; 
      IF slow THEN CursorBits[sendIndex] _ CursorBits[sendIndex]+1;
      ENDLOOP;
    IF ~slow THEN CursorBits[sendIndex] _ CursorBits[sendIndex]+1;
    IF info THEN WriteChar['s];
    ENDLOOP;
  EXITS Closed, Failed => NULL;
  END;
  IF pushPS#NIL THEN PupPktStreamDestroy[pushPS];
  IF globalPushPS=pushPS THEN globalPushPS_NIL;
  IF secondPushPS=pushPS THEN secondPushPS_NIL;
  Done[];
  END;

Pull: PROCEDURE [pullPS: PupPktStream] =
  BEGIN
  listener: BOOLEAN _ pullPS#NIL;
  pullI: CARDINAL;
  bytes: CARDINAL;
  k, l: CARDINAL;
  r: CARDINAL _ 0;
  b: PupBuffer;
  IF pullPS=NIL THEN
    BEGIN
    pullPS _ PupPktStreamMake[
        pullMe.socket,
        pullHim,
        SecondsToTocks[1],
        wait,
        [0,0]];
    globalPullPS _ pullPS;
    END;
  SELECT length FROM	-- can't use zero
    short => l_1;
    cyclic => l_1;	-- start at short
    long => l_dataWordsPerPup;
    random, ignore => NULL;
    ENDCASE => ERROR;
  UNTIL stopFlag DO
    FOR pullI IN [0..packetCycle) UNTIL stopFlag DO
      ENABLE PupStream.StreamClosing => 
        BEGIN
        WriteString["Pull PktStream closed: "];
        IF text#NIL THEN
          BEGIN
          WriteChar['(];
          WriteString[text];
          WriteString[") "];
          END;
        WriteLine[whyText[why]];
        GOTO Closed;
        END;
      UNTIL (b_pullPS.get[])#NIL DO
        IF stopFlag THEN GOTO AllDone;
        WriteChar['?];
        ENDLOOP;
      IF slow THEN CursorBits[recvIndex] _ CursorBits[recvIndex]+1;
      bytes _ GetPupContentsBytes[b];
      IF ~doStats THEN
        BEGIN
        StatIncr[statDataPacketsReceived];
        StatBump[statDataBytesReceived,bytes];
        END;
      SELECT length FROM
        short, long, ignore => NULL;
        cyclic => IF (bytes)#(l_pullI)*2 THEN SIGNAL LengthMismatch;
        random =>
          BEGIN
          l _ ((r_Random[r]) MOD dataWordsPerPup)+1;
          IF (bytes)#l*2 THEN SIGNAL LengthMismatch;
          END;
        ENDCASE => ERROR;
      SELECT data FROM
        ignore => NULL;
        cyclic => FOR k IN [0..l) DO IF b.pupWords[k]#(k+pullI) THEN SIGNAL DataMismatch; ENDLOOP;
        ones => FOR k IN [0..l) DO IF b.pupWords[k]#177777B THEN SIGNAL DataMismatch; ENDLOOP;
        zeros => FOR k IN [0..l) DO IF b.pupWords[k]#0 THEN SIGNAL DataMismatch; ENDLOOP;
        alternating => FOR k IN [0..l) DO IF b.pupWords[k]#125252B THEN SIGNAL DataMismatch; ENDLOOP;
        random => FOR k IN [0..l) DO
          IF b.pupWords[k]#(r_Random[r]) THEN SIGNAL DataMismatch; ENDLOOP;
        ENDCASE => ERROR;
      ReturnFreePupBuffer[b];
      ENDLOOP;
    IF ~slow THEN CursorBits[recvIndex] _ CursorBits[recvIndex]+1;
    IF info THEN WriteChar['r];
    REPEAT AllDone, Closed=>NULL;
    ENDLOOP;
  PupPktStreamDestroy[pullPS];
  IF pullPS=globalPullPS THEN globalPullPS _ NIL;
  IF ~listener THEN Done[];
  END;

Hello: PROCEDURE [ps: PupPktStream, remote: PupAddress] =
  BEGIN
  IF info THEN Header["receiving","from",remote];
  ProcessDefs.Detach[FORK Pull[ps]];
  END;

PtPktListen: PUBLIC PROCEDURE =
  BEGIN
  listener _ CreatePupPktStreamListener[socNum,Hello,SecondsToTocks[1]];
  END;

PtPktPushPull: PUBLIC PROCEDURE =
  BEGIN
  UnListen[];
  IF pullHim.socket=[0,0] THEN BEGIN
    WriteLine["Need a socket number."]; RETURN; END;
  Header["Exchanging","with",pullHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Push[]];
  ProcessDefs.Detach[FORK Pull[NIL]];
  WaitUntilDone[2];
  END;

PtPktRecv: PUBLIC PROCEDURE =
  BEGIN
  UnListen[];
  IF pullHim.socket=[0,0] THEN BEGIN
    WriteLine["Need a socket number."]; RETURN; END;
  Header["receiving","from",pullHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Pull[NIL]];
  WaitUntilDone[1];
  END;

PtPktSend: PUBLIC PROCEDURE =
  BEGIN
  Header["Sending","to",pushHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Push[]];
  WaitUntilDone[1];
  END;

PtPktDoubleSend: PUBLIC PROCEDURE =
  BEGIN
  Header["Sending Double","to",pushHim];
  PtLookerReady[];
  ProcessDefs.Detach[FORK Push[]];
  ProcessDefs.Detach[FORK Push[]];
  WaitUntilDone[2];
  END;

PtPktUnListen: PUBLIC PROCEDURE =
  BEGIN
  IF listener=NIL THEN RETURN[];
  DestroyPupListener[listener]; listener_NIL;
  WriteLine["PktStream Unlisten."];
  END;

-- here is what gets STARTed
END.(2048)\124t2 8t0 6t2 1t0 786t7 4t0 7t7 11t0 121t7 2t2 2t0 181b6B1035b4B169t10 18t0 283t3 6t0 11t2 2t0 714t3 6t0 827t2 3t0 4t2 54t0 20t2 3t0 5t2 54t0 54t3 14t0 19t10 3t0 6t10 29t0 125b4B214t10 16t0 96t2 3t0 713t2 3t0 4t2 17t0 9t2 15t0 9t2 3t0 1113t2 3t0 5t2 17t0 9t2 15t0 9t2 3t0 83t10 19t0 97b5B156b11B110b13B292b9B256b9B145b15B187b13B