File: VoiceStreamSocketImpl.mesa
Copyright Ó 1985, 1986, 1987 by Xerox Corporation. All rights reserved.
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.
Swinehart, July 17, 1987 6:34:18 pm PDT
Ades, January 28, 1986 7:05:42 pm PST
Ousterhout, March 15, 1983 3:22 pm compile
L. Stewart, December 27, 1983 2:06 pm
DIRECTORY
Basics USING [BITAND, Card16FromH, HFromCard16, HFromInt16, Int16FromH, LowHalf],
BasicTime,
IntervalTimer USING [Wait],
IO USING [card, int, PutF, PutRope, rope],
Process USING [Detach, MsecToTicks, Priority, priorityForeground, SetPriority, SetTimeout],
PupBuffer USING [ Buffer ],
PupSocket USING [ AllocBuffer, FreeBuffer, Get, GetUserBytes, Put, SetUserBytes, SetUserHWords, Socket ],
PupType USING [Type],
Rope USING [ROPE],
VoiceStream;
VoiceStreamSocketImpl: MONITOR LOCKS VoiceStream.Lock
IMPORTS Basics, BasicTime, IntervalTimer, Process, PupSocket, VoiceStream
EXPORTS VoiceStream =
BEGIN
socketPriority: Process.Priority ← Process.priorityForeground;
Truth for the constants immediately following is ProcessorFace.microsecondsPerHundredPulses.
MicrosecondsPerPulse: LONG CARDINAL = 32;
PulsesPerMillisecond: LONG CARDINAL = 31; -- really 31.25
packet types and formats
BluejayVoiceType: PupType.Type = VAL[250];
ProbeReplyType: PupType.Type = VAL[251];
LarkFirstVoiceType: PupType.Type = VAL[252];
RetransmitRequestType: PupType.Type = VAL[253];
LarkVoiceType: PupType.Type = VAL[254];
BluejayPacket: TYPE = LONG POINTER TO BluejayPacketObject;
BluejayVoiceType
id.a: timeToPlay
id.b: bluejay sequence number
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;
ProbeReplyType
id.a: Lark clock
id.b: Lark sequence number
ProbeReplyPacketObject: TYPE = MACHINE DEPENDENT RECORD [
replyID: CARDINAL,
maxPacketSize: CARDINAL,
maxBuffers: CARDINAL,
blank1: CARDINAL
];
LarkVoiceType
id.a: Lark time corresponding to first byte of silence.
id.b: Monotonic increasing packet sequence number (ignored).
LarkPacketObject: TYPE = MACHINE DEPENDENT RECORD [
blank1: [0..377B],
blank2: [0..17B],
encryptionKeyIndex: [0..17B],
energy: CARDINAL,
silenceMS: CARDINAL, -- preceding data
blank3: CARDINAL
];
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:
bytesPerPacket: INTEGER = 160;
bytesPerMs: NAT = 8;
debugSwitch: BOOLFALSE;
maxLostTime: LONG CARDINAL ← 100000 / MicrosecondsPerPulse;
now: LONG CARDINAL;
timesLate: LONG CARDINAL ← 0;
noPieceMS: NAT ← 200;
SetSocket: PUBLIC ENTRY PROC [handle: VoiceStream.Handle, socket: PupSocket.Socket] = {
ENABLE UNWIND => NULL;
IF handle.errorRope # NIL THEN VoiceStream.ReportError[handle];
Create a new process, if that is needed.
IF handle.connection = NIL THEN handle.connection ← NEW[VoiceStream.ConnectionObject];
handle.connection.socket ← NIL;
IF socket # NIL THEN {
handle.connection.socket ← socket;
Process.Detach[FORK ReceiveProc[handle, socket]];
Process.Detach[FORK SendProc[handle, socket]];
};
};
CheckHandle: ENTRY PROC [handle: VoiceStream.Handle, socket: PupSocket.Socket] RETURNS [Rope.ROPE] = {
ENABLE UNWIND => NULL;
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.
IF handle.connection.socket # socket THEN RETURN ["Socket changed."];
IF handle.errorRope # NIL THEN RETURN [handle.errorRope];
RETURN [NIL];
};
ForceToJukebox: ENTRY PROC [handle: VoiceStream.Handle] = {
This procedure outputs to the server any partially-full write buffer, then waits for the server to write it to disk.
ENABLE UNWIND => NULL;
VoiceStream.FlushBuffer[handle];
WHILE (handle.firstServerBuffer # NIL)
AND (handle.errorRope = NIL)
AND (handle.piece # NIL) DO WAIT VoiceStream.client ENDLOOP;
};
ReceiveProc: PROC [handle: VoiceStream.Handle, origSocket: PupSocket.Socket] = {
This process sets up a connection with an Etherphone, then receives incoming packets from the Etherphone.
ENABLE {
VoiceStream.Error => {
IF reason # StreamClosed THEN IO.PutF[ioStream, "Rx Network connection closed: %s.\n", IO.rope[rope]];
CONTINUE;
};
};
input: PupBuffer.Buffer;
expLarkTime: CARDINAL ← 0;   -- Expected time for next data.
larkTime: CARDINAL ← 0;     -- Time of current packet.
silenceMs: CARDINAL ← 0;
dataBytes: CARDINAL;
first: BOOLTRUE;  -- first packet of recording
reason: Rope.ROPENIL;
Process.SetPriority[socketPriority];
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.
WHILE (reason ← CheckHandle[handle, origSocket]) = NIL DO
input ← handle.connection.socket.Get[];
IF input = NIL THEN LOOP;
Decode the Lark packet.
larkTime ← Basics.Card16FromH[input.id.hi];
SELECT input.type FROM
On probe replies, just update information about the connection.
= ProbeReplyType => {
AdjustLocked[handle.connection, input];
handle.connection.packetSize ← Basics.Int16FromH[input.hWord[1]];
handle.connection.bufferSpace ← Basics.Int16FromH[input.hWord[2]] * (handle.connection.packetSize / bytesPerMs);
};
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.
= LarkVoiceType, = LarkFirstVoiceType => {
silenceMs ← larkTime - expLarkTime;
silenceMs ← MIN[silenceMs, 1000];
input.hWord[2] ← Basics.HFromInt16[MIN[260, Basics.Int16FromH[input.hWord[2]]]];
IF NOT first AND silenceMs NOT IN [0..250) THEN {
IO.PutF[ioStream, "Got packet time %d, expected %d.\n", IO.int[larkTime], IO.int[expLarkTime]];
silenceMs ← 0;
};
Write silence and data.
IF first THEN silenceMs ← 0
ELSE silenceMs ← silenceMs + Basics.Int16FromH[input.hWord[2]];
IF LOOPHOLE[silenceMs, INTEGER] < 0 THEN silenceMs ← 0;
The above statement prevents at least the initial manifestation of a problem with Lark-to-Bluejay traffic during Bluejay-to-Lark playback. A better solution awaits a fuller understanding. One-way Lark connections would be desirable. DCS July 17, 1987 6:34:16 pm PDT
dataBytes ← PupSocket.GetUserBytes[input] - 8;
dataBytes ← Basics.BITAND[dataBytes, 0177770B];
[] ← VoiceStream.Put[handle: handle, silentBytes: silenceMs * bytesPerMs, block: [@input.hWord[4], 0, dataBytes], energy: Basics.Int16FromH[input.hWord[1]]];
expLarkTime ← larkTime + Basics.Int16FromH[input.hWord[2]] + (dataBytes/bytesPerMs);
first ← FALSE;
};
ENDCASE;
PupSocket.FreeBuffer[input];
ENDLOOP;
IO.PutF[ioStream, "Rx process %b quitting: %g\n", IO.card[LOOPHOLE[Process.GetCurrent[], CARDINAL]], IO.rope[reason]];
};
AdjustLocked: ENTRY PROC [c: VoiceStream.Connection, b: PupBuffer.Buffer] = {
ENABLE UNWIND => NULL;
rp: ProbeReply;
deltaLark, deltaLocal: INT;
rp ← LOOPHOLE[@b.body];
IF rp^.replyID = c.probeSequenceNumber THEN {
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.id.high;
IF c.initialProbeFinished THEN {
c.roundTripMS ← (Basics.LowHalf[(BasicTime.GetClockPulses[] - c.probeLaunchTime) / PulsesPerMillisecond] + c.roundTripMS) / 2;
deltaLocal ← c.probeLaunchTime - c.localEpoch;
deltaLark ← LONG[Basics.Card16FromH[b.id.hi] - c.larkEpoch] * PulsesPerMillisecond;
positive drift means the local clock is too fast
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 => {
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 + 150;
c.drift ← c.drift - 150;
};
ENDCASE;
c.larkEpoch ← Basics.Card16FromH[b.id.hi];
c.localEpoch ← c.probeLaunchTime;
}
ELSE { -- initial probe case
c.roundTripMS ← Basics.LowHalf[(BasicTime.GetClockPulses[] - c.probeLaunchTime) / PulsesPerMillisecond];
c.localEpoch ← c.probeLaunchTime;
c.larkEpoch ← Basics.Card16FromH[b.id.hi];
c.drift ← 0;
c.initialProbeFinished ← TRUE;
NOTIFY c.initialProbeDone;
};
};
};
SendProc: PROC [handle: VoiceStream.Handle, origSocket: PupSocket.Socket] = {
This procedure runs the second process for each connection between Bluejay and a Lark. It is responsible for sending voice to Lark.
ENABLE {
VoiceStream.Error => {
IF reason # StreamClosed THEN IO.PutF[ioStream, "Tx Network connection closed: %s.\n", IO.rope[rope]];
CONTINUE;
};
};
output: PupBuffer.Buffer ← NIL;
pp: BluejayPacket;
c: VoiceStream.Connection ← handle.connection;
silenceLength: NAT;
voiceLength: NAT;
reason: Rope.ROPE;
silenceMS: NAT ← 0;
voiceMS: NAT ← 0;
lastPlayed: BasicTime.Pulses;
noPieceThisTime: BOOLFALSE;
wakeTime: BasicTime.Pulses;
temp1: INT;
{
Process.SetPriority[socketPriority];
c.initialProbeFinished ← FALSE;
Process.SetTimeout[condition: @c.initialProbeDone, ticks: Process.MsecToTicks[100]];
For starters, send a probe request to Lark, and keep sending it every quarter second until we get a probe packet back again.
EmptyProbe[handle: handle, origSocket: origSocket];
IF NOT c.initialProbeFinished THEN {
probe failed
handle.errorRope ← "Lark failed to respond";
handle.errorCode ← Timeout;
GOTO Cleanup;
};
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
c.timeToPlay ← c.larkEpoch + MIN[120, c.bufferSpace];
send the first batch right away . . . is this unsafe? (no, it's a simple assignment)
c.sendNext ← c.localEpoch;
lastPlayed ← BasicTime.GetClockPulses[] - (10000 / MicrosecondsPerPulse);
Now enter a loop to send voice data and/or silence to Lark.
WHILE (reason ← CheckHandle[handle, origSocket]) = NIL DO
IF output = NIL THEN {
output ← c.socket.AllocBuffer[];
pp ← LOOPHOLE[@output.body];
output.type ← BluejayVoiceType;
output.id.lo ← Basics.HFromCard16[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.id.hi ← Basics.HFromCard16[c.timeToPlay];
Poll the voice stream to see if there is any data there.
[silence: silenceLength, bytesTransferred: voiceLength, keyIndex: pp.encryptionKeyIndex] ← VoiceStream.Get[handle: handle, maxSilentBytes: 500 * bytesPerMs, block: [@output.hWord[4], 0, bytesPerPacket]];
voiceLength ← Basics.BITAND[voiceLength, 0177770B];
silenceMS ← silenceLength/bytesPerMs;
voiceMS ← voiceLength/bytesPerMs;
If either return value is non-zero, then there is data.
noPieceThisTime ← (silenceLength = 0) AND (voiceLength = 0);
IF noPieceThisTime THEN silenceMS ← noPieceMS;
c.packetMS ← silenceMS + voiceMS;
pp.silenceMS ← silenceMS;
PupSocket.SetUserBytes[output, SIZE[BluejayPacketObject] * 2 + voiceLength];
now ← BasicTime.GetClockPulses[];
IF (temp1←LOOPHOLE[(LOOPHOLE[lastPlayed, LONG CARDINAL] + (5000 / MicrosecondsPerPulse)) - now, INT]) > 0 THEN {
IntervalTimer.Wait[PulsesToSeconds[temp1]]; -- wait at least 5 mS
now ← BasicTime.GetClockPulses[];
IF (now - (lastPlayed + (5000 / MicrosecondsPerPulse))) > maxLostTime THEN {
timesLate ← timesLate + 1;
};
};
Else wait to send the packet
now ← BasicTime.GetClockPulses[];
wakeTime ← c.sendNext;
IF (temp1 ← LOOPHOLE[wakeTime - now, INT]) > 0 THEN {
IntervalTimer.Wait[PulsesToSeconds[temp1]];
now ← BasicTime.GetClockPulses[];
IF now - wakeTime > maxLostTime THEN {
timesLate ← timesLate + 1;
};
};
PlayPacketLocked[c, pp, output];
IF (reason ← CheckHandle[handle, origSocket]) # NIL THEN {
PupSocket.FreeBuffer[output];
output ← NIL;
EXIT;
};
c.socket.Put[output];
output ← NIL;
lastPlayed ← BasicTime.GetClockPulses[];
ENDLOOP;
GOTO Cleanup;
EXITS
Cleanup => {
IO.PutF[ioStream, "Tx process %b quitting: %g\n", IO.card[LOOPHOLE[Process.GetCurrent[], CARDINAL]], IO.rope[reason]];
};
};
};
PulsesToSeconds: PROC[intPulses: INT] RETURNS [seconds: REAL] = {
microseconds: INT ← MicrosecondsPerPulse * intPulses;
RETURN[(0.0+microseconds)*1E-6];
};
Call PlayPacketLocked only if the packet will be sent!
PlayPacketLocked: ENTRY PROC [c: VoiceStream.Connection, pp: BluejayPacket, b: PupBuffer.Buffer] = {
ENABLE UNWIND => NULL;
IF (c.sendNext - c.probeLaunchTime) NOT IN [0..450000 / MicrosecondsPerPulse) THEN {
pp^.probeRequest ← TRUE;
c.probeLaunchTime ← BasicTime.GetClockPulses[];
c.probeSequenceNumber ← Basics.Card16FromH[b.id.lo];
};
c.timeToPlay ← c.timeToPlay + c.packetMS;
c.sendNext ← c.sendNext + (LONG[c.packetMS] * PulsesPerMillisecond);
};
EmptyProbe: PROC [handle: VoiceStream.Handle, origSocket: PupSocket.Socket] = {
b: PupBuffer.Buffer;
pp: BluejayPacket;
trySN: CARDINAL;
c: VoiceStream.Connection ← handle.connection;
WHILE CheckHandle[handle, origSocket] = NIL DO
b ← c.socket.AllocBuffer[];
pp ← LOOPHOLE[@b.body];
trySN ← c.sequenceNumber;
c.sequenceNumber ← c.sequenceNumber + 1;
b.id.hi ← Basics.HFromCard16[0];
b.id.lo ← Basics.HFromCard16[trySN];
b.type ← BluejayVoiceType;
pp^ ← [
blank1: 0,
ignore: TRUE,
probeRequest: TRUE,
blank2: 0,
encryptionKeyIndex: 0,
energy: 0,
silenceMS: 0,
blank3: 0
];
PupSocket.SetUserHWords[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: PupBuffer.Buffer] = {
ENABLE UNWIND => NULL;
c.probeLaunchTime ← BasicTime.GetClockPulses[];
c.probeSequenceNumber ← Basics.Card16FromH[b.id.lo];
};
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, May 14, 1985 6:29:20 pm PDT, Cedar 5, L
Swinehart, May 14, 1985 6:29:12 pm PDT
Cedar 6.0
Ades, May 10, 1986 5:21:14 pm PDT
modified Put call to supply energy value, DIRECTORY, IMPORTS, AdjustLocked, PlayPacketLocked, SendProbeLocked
Swinehart, May 10, 1986 4:18:44 pm PDT
Convert to new communications package
changes to: VoiceStreamSocketImpl, EXPORTS, BEGIN, SetSocket, CheckHandle, ForceToJukebox, ReceiveProc, SendProc, EmptyProbe
Swinehart, April 8, 1987 8:35:08 am PDT
Cedar 7. IntervalTimer now requires a REAL argument.
changes to: AdjustLocked, SendProc, PulsesToSeconds, PlayPacketLocked, SendProbeLocked, DIRECTORY, ReceiveProc, EmptyProbe