DIRECTORY Basics USING [BITAND, LowHalf], IntervalTimer USING [NowInPulses, Pulses, WaitForExpirationTimeInPulses], Process USING [Detach, GetCurrent, MsecToTicks, Priority, priorityForeground, SetPriority, SetTimeout], PupDefs USING [GetFreePupBuffer, GetPupContentsBytes, PupBuffer, PupSocket, ReturnFreePupBuffer, SetPupContentsBytes, SetPupContentsWords], PupTypes USING [PupType], Rope USING [ROPE], SpyOps USING [Record, StackType], VoiceStream; VoiceStreamSocket: MONITOR LOCKS Lock IMPORTS Basics, IntervalTimer, Process, PupDefs, SpyOps, VoiceStream EXPORTS VoiceStream SHARES VoiceStream = BEGIN OPEN VoiceStream; socketPriority: Process.Priority _ Process.priorityForeground; MicrosecondsPerPulse: LONG CARDINAL = 32; PulsesPerMillisecond: LONG CARDINAL = 31; -- really 31.25 BluejayVoiceType: PupTypes.PupType = LOOPHOLE[250]; ProbeReplyType: PupTypes.PupType = LOOPHOLE[251]; LarkFirstVoiceType: PupTypes.PupType = LOOPHOLE[252]; RetransmitRequestType: PupTypes.PupType = LOOPHOLE[253]; LarkVoiceType: PupTypes.PupType = LOOPHOLE[254]; BluejayPacket: TYPE = LONG POINTER TO BluejayPacketObject; BluejayPacketObject: TYPE = MACHINE DEPENDENT RECORD [ blank1: [0..77B], ignore: BOOL, probeRequest: BOOL, blank2: [0..17B], encryptionKeyIndex: [0..17B], energy: CARDINAL, silenceMS: CARDINAL, blank3: CARDINAL ]; ProbeReply: TYPE = LONG POINTER TO ProbeReplyPacketObject; ProbeReplyPacketObject: TYPE = MACHINE DEPENDENT RECORD [ replyID: CARDINAL, maxPacketSize: CARDINAL, maxBuffers: CARDINAL, blank1: CARDINAL ]; LarkPacketObject: TYPE = MACHINE DEPENDENT RECORD [ blank1: [0..377B], blank2: [0..17B], encryptionKeyIndex: [0..17B], energy: CARDINAL, silenceMS: CARDINAL, -- preceding data blank3: CARDINAL ]; bytesPerPacket: INTEGER = 160; bytesPerMs: NAT = 8; debugSwitch: BOOL _ FALSE; maxLostTime: LONG CARDINAL _ 100000 / MicrosecondsPerPulse; now: LONG CARDINAL; timesLate: LONG CARDINAL _ 0; noPieceMS: NAT _ 200; SetSocket: PUBLIC ENTRY PROC [handle: Handle, socket: PupDefs.PupSocket] = { ENABLE UNWIND => NULL; IF handle.errorRope # NIL THEN ERROR Error[handle.errorCode, handle.errorRope]; IF handle.connection = NIL THEN handle.connection _ NEW[VoiceStream.ConnectionObject]; handle.connection.socket _ NIL; IF socket # NIL THEN { handle.connection.socket _ socket; handle.notified _ TRUE; Process.Detach[FORK receiveProc[handle, socket]]; Process.Detach[FORK sendProc[handle, socket]]; }; }; checkHandle: ENTRY PROC [handle: Handle, socket: PupDefs.PupSocket] RETURNS [Rope.ROPE] = { ENABLE UNWIND => NULL; IF handle.connection.socket # socket THEN RETURN ["Socket changed."]; IF handle.errorRope # NIL THEN RETURN [handle.errorRope]; RETURN [NIL]; }; forceToJukebox: ENTRY PROC [handle: Handle] = { ENABLE UNWIND => NULL; flushBuffer[handle]; WHILE (handle.firstServerBuffer # NIL) AND (handle.errorRope = NIL) AND (handle.piece # NIL) DO WAIT client ENDLOOP; }; receiveProc: PROC [handle: Handle, origSocket: PupDefs.PupSocket] = { ENABLE { Error => { CONTINUE; }; }; input: PupDefs.PupBuffer; expLarkTime: CARDINAL _ 0; -- Expected time for next data. larkTime: CARDINAL _ 0; -- Time of current packet. silenceMs: CARDINAL _ 0; dataBytes: CARDINAL; first: BOOL _ TRUE; -- first packet of recording reason: Rope.ROPE _ NIL; Process.SetPriority[socketPriority]; WHILE (reason _ checkHandle[handle, origSocket]) = NIL DO input _ handle.connection.socket.get[]; IF input = NIL THEN LOOP; larkTime _ input.pupID.a; SELECT input.pupType FROM = ProbeReplyType => { AdjustLocked[handle.connection, input]; handle.connection.packetSize _ input.pupWords[1]; handle.connection.bufferSpace _ input.pupWords[2] * (handle.connection.packetSize / bytesPerMs); }; = LarkVoiceType, = LarkFirstVoiceType => { silenceMs _ larkTime - expLarkTime; silenceMs _ MIN[silenceMs, 1000]; input.pupWords[2] _ MIN[260, input.pupWords[2]]; IF NOT first AND silenceMs NOT IN [0..250) THEN { silenceMs _ 0; }; IF first THEN silenceMs _ 0 ELSE silenceMs _ silenceMs + input.pupWords[2]; dataBytes _ PupDefs.GetPupContentsBytes[input] - 8; dataBytes _ Basics.BITAND[dataBytes, 0177770B]; [] _ Put[handle: handle, silentBytes: silenceMs * bytesPerMs, block: [@input.pupWords[4], 0, dataBytes]]; expLarkTime _ larkTime + input.pupWords[2] + (dataBytes/bytesPerMs); first _ FALSE; }; ENDCASE; PupDefs.ReturnFreePupBuffer[input]; ENDLOOP; }; AdjustLocked: ENTRY PROC [c: VoiceStream.Connection, b: PupDefs.PupBuffer] = { ENABLE UNWIND => NULL; rp: ProbeReply; deltaLark, deltaLocal: INT; rp _ LOOPHOLE[@b.pupBody]; IF rp^.replyID = c.probeSequenceNumber THEN { IF c.initialProbeFinished THEN { c.roundTripMS _ (Basics.LowHalf[(IntervalTimer.NowInPulses[] - c.probeLaunchTime) / PulsesPerMillisecond] + c.roundTripMS) / 2; deltaLocal _ c.probeLaunchTime - c.localEpoch; deltaLark _ LONG[b.pupID.a - c.larkEpoch] * PulsesPerMillisecond; c.drift _ c.drift + deltaLocal - deltaLark; SELECT TRUE FROM c.drift < -312 => { c.sendNext _ c.sendNext - 150; c.drift _ c.drift + 150; }; c.drift > 312 => { c.sendNext _ c.sendNext + 150; c.drift _ c.drift - 150; }; ENDCASE; c.larkEpoch _ b.pupID.a; c.localEpoch _ c.probeLaunchTime; } ELSE { -- initial probe case c.roundTripMS _ Basics.LowHalf[(IntervalTimer.NowInPulses[] - c.probeLaunchTime) / PulsesPerMillisecond]; c.localEpoch _ c.probeLaunchTime; c.larkEpoch _ b.pupID.a; c.drift _ 0; c.initialProbeFinished _ TRUE; NOTIFY c.initialProbeDone; }; }; }; sendProc: PROC [handle: Handle, origSocket: PupDefs.PupSocket] = { ENABLE { Error => { CONTINUE; }; }; output: PupDefs.PupBuffer _ NIL; pp: BluejayPacket; c: VoiceStream.Connection _ handle.connection; silenceLength: NAT; voiceLength: NAT; reason: Rope.ROPE; silenceMS: NAT _ 0; voiceMS: NAT _ 0; lastPlayed: IntervalTimer.Pulses; noPieceThisTime: BOOL _ FALSE; wakeTime: IntervalTimer.Pulses; { Process.SetPriority[socketPriority]; c.initialProbeFinished _ FALSE; Process.SetTimeout[condition: @c.initialProbeDone, ticks: Process.MsecToTicks[100]]; EmptyProbe[handle: handle, origSocket: origSocket]; IF NOT c.initialProbeFinished THEN { handle.errorRope _ "Lark failed to respond"; handle.errorCode _ Timeout; GOTO Cleanup; }; c.timeToPlay _ c.larkEpoch + MIN[120, c.bufferSpace]; c.sendNext _ c.localEpoch; lastPlayed _ IntervalTimer.NowInPulses[] - (10000 / MicrosecondsPerPulse); WHILE (reason _ checkHandle[handle, origSocket]) = NIL DO IF output = NIL THEN { output _ PupDefs.GetFreePupBuffer[]; pp _ LOOPHOLE[@output.pupBody]; output.pupType _ BluejayVoiceType; output.pupID.b _ c.sequenceNumber; c.sequenceNumber _ c.sequenceNumber + 1; pp^ _ [ blank1: 0, ignore: FALSE, probeRequest: FALSE, blank2: 0, encryptionKeyIndex: 0, energy: 0, silenceMS: 0, blank3: 0 ]; }; output.pupID.a _ c.timeToPlay; [silence: silenceLength, bytesTransferred: voiceLength, keyIndex: pp.encryptionKeyIndex] _ VoiceStream.Get[handle: handle, maxSilentBytes: 500 * bytesPerMs, block: [@output.pupWords[4], 0, bytesPerPacket]]; voiceLength _ Basics.BITAND[voiceLength, 0177770B]; silenceMS _ silenceLength/bytesPerMs; voiceMS _ voiceLength/bytesPerMs; noPieceThisTime _ (silenceLength = 0) AND (voiceLength = 0); IF noPieceThisTime THEN silenceMS _ noPieceMS; c.packetMS _ silenceMS + voiceMS; pp.silenceMS _ silenceMS; PupDefs.SetPupContentsBytes[output, SIZE[BluejayPacketObject] * 2 + voiceLength]; now _ IntervalTimer.NowInPulses[]; IF LOOPHOLE[(LOOPHOLE[lastPlayed, LONG CARDINAL] + (5000 / MicrosecondsPerPulse)) - now, INT] > 0 THEN { IntervalTimer.WaitForExpirationTimeInPulses[lastPlayed + (5000 / MicrosecondsPerPulse)]; -- wait at least 5 mS now _ IntervalTimer.NowInPulses[]; IF (now - (lastPlayed + (5000 / MicrosecondsPerPulse))) > maxLostTime THEN { timesLate _ timesLate + 1; IF debugSwitch THEN SpyOps.Record[type: LAST[SpyOps.StackType]]; }; }; now _ IntervalTimer.NowInPulses[]; wakeTime _ c.sendNext; IF LOOPHOLE[wakeTime - now, INT] > 0 THEN { IntervalTimer.WaitForExpirationTimeInPulses[wakeTime]; now _ IntervalTimer.NowInPulses[]; IF now - wakeTime > maxLostTime THEN { timesLate _ timesLate + 1; IF debugSwitch THEN SpyOps.Record[type: LAST[SpyOps.StackType]]; }; }; PlayPacketLocked[c, pp, output]; IF (reason _ checkHandle[handle, origSocket]) # NIL THEN { PupDefs.ReturnFreePupBuffer[output]; output _ NIL; EXIT; }; c.socket.put[output]; output _ NIL; lastPlayed _ IntervalTimer.NowInPulses[]; ENDLOOP; GOTO Cleanup; EXITS Cleanup => { }; }; }; PlayPacketLocked: ENTRY PROC [c: VoiceStream.Connection, pp: BluejayPacket, b: PupDefs.PupBuffer] = { ENABLE UNWIND => NULL; IF (c.sendNext - c.probeLaunchTime) NOT IN [0..450000 / MicrosecondsPerPulse) THEN { pp^.probeRequest _ TRUE; c.probeLaunchTime _ IntervalTimer.NowInPulses[]; c.probeSequenceNumber _ b.pupID.b; }; c.timeToPlay _ c.timeToPlay + c.packetMS; c.sendNext _ c.sendNext + (LONG[c.packetMS] * PulsesPerMillisecond); }; EmptyProbe: PROC [handle: VoiceStream.Handle, origSocket: PupDefs.PupSocket] = { b: PupDefs.PupBuffer; pp: BluejayPacket; trySN: CARDINAL; c: VoiceStream.Connection _ handle.connection; WHILE checkHandle[handle, origSocket] = NIL DO b _ PupDefs.GetFreePupBuffer[]; pp _ LOOPHOLE[@b.pupBody]; trySN _ c.sequenceNumber; c.sequenceNumber _ c.sequenceNumber + 1; b.pupID.a _ 0; b.pupID.b _ trySN; b.pupType _ BluejayVoiceType; pp^ _ [ blank1: 0, ignore: TRUE, probeRequest: TRUE, blank2: 0, encryptionKeyIndex: 0, energy: 0, silenceMS: 0, blank3: 0 ]; PupDefs.SetPupContentsWords[b, SIZE[BluejayPacketObject]]; SendProbeLocked[c, b]; c.socket.put[b]; WaitLocked[c]; IF c.initialProbeFinished THEN EXIT; ENDLOOP; }; SendProbeLocked: ENTRY PROC [c: VoiceStream.Connection, b: PupDefs.PupBuffer] = { ENABLE UNWIND => NULL; c.probeLaunchTime _ IntervalTimer.NowInPulses[]; c.probeSequenceNumber _ b.pupID.b; }; WaitLocked: ENTRY PROC [c: VoiceStream.Connection] = { WAIT c.initialProbeDone; }; END. L. Stewart, June 5, 1983 4:30 pm, no notify on SetSocket L. Stewart, June 6, 1983 1:15 pm, LarkFirstVoiceType packet type L. Stewart, June 20, 1983 4:52 pm, Try using Pulses instead of Microseconds L. Stewart, July 9, 1983 4:48 pm, added noPieceMS in place of constant L. Stewart, December 27, 1983 2:06 pm, Cedar 5 æFile: VoiceStreamSocket.mesa This file provides part of the implementation of voicestreams. Its procedures provide for transfer of voice data to and from remote sockets on the network. Last Edited by: Ousterhout, March 15, 1983 3:22 pm compile Last Edited by: L. Stewart, December 27, 1983 2:06 pm Last Edited by: Swinehart, March 29, 1983 3:25 pm IntervalTimer USING [Microseconds, Now, WaitForExpirationTime], IO USING [card, int, PutF, PutRope, rope], Truth for the constants immediately following is ProcessorFace.microsecondsPerHundredPulses. packet types and formats BluejayVoiceType id.a: timeToPlay id.b: bluejay sequence number ProbeReplyType id.a: Lark clock id.b: Lark sequence number LarkVoiceType id.a: Lark time corresponding to first byte of silence. id.b: Monotonic increasing packet sequence number (ignored). RetransmitRequestType id.a: Lark time (in milliseconds) of first data not received. id.b: Monotonic increasing packet sequence number. Some constants for packet shipment on the net: Create a new process, if that is needed. This procedure just provides a safe way to interrogate handle information to make sure the current socket process should continue to run. A NIL return value means the socket process should continue to run. Otherwise, a rope is returned to desribe why the process should quit. This procedure outputs to the server any partially-full write buffer, then waits for the server to write it to disk. This process sets up a connection with an Etherphone, then receives incoming packets from the Etherphone. IF reason # StreamClosed THEN IO.PutF[ioStream, "Rx Network connection closed: %s.\n", IO.rope[rope]]; Read packets continuously until either the voice stream is closed or we cease to be the official socket for the stream. Note that we never time the stream out. Decode the Lark packet. On probe replies, just update information about the connection. On Lark data, throw away packets that don't correspond to the immediate future, then add silence to the tune (if necessary), then add the data from the packet. IO.PutF[ioStream, "Got packet time %d, expected %d.\n", IO.int[larkTime], IO.int[expLarkTime]]; Write silence and data. IO.PutF[ioStream, "Rx process %b quitting: %g\n", IO.card[LOOPHOLE[Process.GetCurrent[], CARDINAL]], IO.rope[reason]]; average roundtrip time and check for clock drift The initial association between Bluejay time and Lark time was localEpoch and larkEpoch. The new data point is probeLaunchTime and b.pupID.a; c.roundTripMS _ (Basics.LowHalf[(IntervalTimer.Now[] - c.probeLaunchTime) / 1000] + c.roundTripMS) / 2; deltaLark _ LONG[b.pupID.a - c.larkEpoch] * 1000; positive drift means the local clock is too fast since the local clock is too fast, we need to wait some more ticks of the local clock to delay the same real time SELECT TRUE FROM c.drift < -10000 => { c.sendNext _ c.sendNext - 5000; c.drift _ c.drift + 5000; }; c.drift > 10000 => { since the local clock is too fast, we need to wait some more ticks of the local clock to delay the same real time c.sendNext _ c.sendNext + 5000; c.drift _ c.drift - 5000; }; ENDCASE; c.roundTripMS _ Basics.LowHalf[(IntervalTimer.Now[] - c.probeLaunchTime) / 1000]; This procedure runs the second process for each connection between Bluejay and a Lark. It is responsible for sending voice to Lark. IF reason # StreamClosed THEN IO.PutF[ioStream, "Tx Network connection closed: %s.\n", IO.rope[rope]]; lastPlayed: IntervalTimer.Microseconds; For starters, send a probe request to Lark, and keep sending it every quarter second until we get a probe packet back again. probe failed now we have a clock correspondence, a packet transmitted at localEpoch arrived at the lark at larkEpoch we want to buffer up as much as possible, but not too much send the first batch right away . . . is this unsafe? (no, it's a simple assignment) lastPlayed _ IntervalTimer.Now[] - 10000; Now enter a loop to send voice data and/or silence to Lark. Poll the voice stream to see if there is any data there. If either return value is non-zero, then there is data. now _ IntervalTimer.Now[]; IF LOOPHOLE[(lastPlayed + 5000) - now, INT] > 0 THEN { IntervalTimer.WaitForExpirationTime[lastPlayed + 5000]; -- wait at least 5 mS now _ IntervalTimer.Now[]; IF (now - (lastPlayed + 5000)) > maxLostTime THEN { timesLate _ timesLate + 1; IF debugSwitch THEN SpyOps.Record[type: LAST[SpyOps.StackType]]; }; }; now _ IntervalTimer.Now[]; IF LOOPHOLE[c.sendNext - now, INT] > 0 THEN { IntervalTimer.WaitForExpirationTime[c.sendNext]; now _ IntervalTimer.Now[]; IF now - c.sendNext > maxLostTime THEN { timesLate _ timesLate + 1; IF debugSwitch THEN SpyOps.Record[type: LAST[SpyOps.StackType]]; }; }; Else wait to send the packet lastPlayed _ IntervalTimer.Now[]; IO.PutF[ioStream, "Tx process %b quitting: %g\n", IO.card[LOOPHOLE[Process.GetCurrent[], CARDINAL]], IO.rope[reason]]; Call PlayPacketLocked only if the packet will be sent! c.probeLaunchTime _ IntervalTimer.Now[]; c.sendNext _ c.sendNext + (LONG[c.packetMS] * LONG[1000]); c.probeLaunchTime _ IntervalTimer.Now[]; Ê(˜Jšœ™Jšœ™J˜Jšœ:™:Jšœ5™5J™1J˜šÏk ˜ Jšœœœ ˜Jšœœ,™?Jšœœ6˜IJšœœ"™*JšœœZ˜gJšœœ~˜‹Jšœ œ ˜Jšœœœ˜Jšœœ˜!J˜ J˜—Jšœœœ˜%Jšœ=˜DJšœ ˜šœ˜J˜—Jšœœ ˜˜J˜?J˜J™\Jšœœœ˜)JšœœœÏc˜:Jšœ™J˜Jšœ%œ˜3Jšœ#œ˜1Jšœ'œ˜5Jšœ*œ˜8Jšœ"œ˜0J˜Jš œœœœœ˜:J˜Jšœ™Jšœ™Jšœ™š œœœ œœ˜6J˜Jšœœ˜ Jšœœ˜J˜J˜Jšœœ˜Jšœ œ˜Jšœ˜J˜J˜—Jš œ œœœœ˜:J˜Jšœ™Jšœ™Jšœ™š œœœ œœ˜9Jšœ œ˜Jšœœ˜Jšœ œ˜Jšœ˜J˜J˜—Jšœ ™ Jšœ7™7Jšœ<™<š œœœ œœ˜3J˜J˜J˜Jšœœ˜Jšœ œž˜'Jšœ˜J˜J˜—Jšœ™Jšœ=™=Jšœ2™2J˜Jšœ.™.J˜Jšœœ˜Jšœ œ˜—˜Jšœ œœ˜Jšœ œœ!˜;Jšœœœ˜Jšœ œœ˜J˜Jšœ œ˜J˜šÏn œœœœ0˜LJšœœœ˜J˜Jšœœœœ+˜OJ˜Jšœ(™(J˜Jšœœœœ˜VJšœœ˜šœ œœ˜J˜"Jšœœ˜Jšœœ˜1Jšœœ˜.J˜—J˜J˜—š œ œœ-œœ˜[Jšœœœ˜J™Jšœ•™•J˜Jšœ#œœ˜EJšœœœœ˜9Jšœœ˜ Jšœ˜J˜—šœœœ˜/J™Jšœt™tJ˜Jšœœœ˜J˜J˜šœœ˜&Jšœœ˜Jš œœœœœ˜0—J˜J˜—šœ œ4˜EJ™Jšœi™iJ˜šœ˜˜ Jšœœ7œ ™fJšœ˜ J˜J˜—J˜—J˜J˜Jšœ œž˜