File: VoiceStreamServer.mesa
This file contains part of the implementation of voicestreams. The
routines in this file implement the voicestream server, a background
process that transfers information between disk and buffers in
voicestreams.
Last Edited by: Ousterhout, March 8, 1983 11:39 am compile
Last Edited by: Stewart, January 3, 1984 11:47 am
DIRECTORY
FS USING [Read, Write],
IO USING [int, PutFR, STREAM],
Jukebox USING [bytesPerChirp, CloseTune, CreateTune, EOF, Error, FindChirp, MissingChirp, OpenTune, TuneSize],
Process USING [Detach, Priority, priorityNormal, SetPriority],
Rope USING [ROPE],
VM USING [AddressForPageNumber],
VoiceStream;
VoiceStreamServer: MONITOR LOCKS Lock
IMPORTS FS, IO, Jukebox, Process, VM, VoiceStream
EXPORTS VoiceStream
BEGIN OPEN VoiceStream;
Foo: SIGNAL = CODE;
debugRuns: BOOL ← FALSE;
serverPriority: Process.Priority ← Process.priorityNormal;
getServerBuffer:
ENTRY
PROC
RETURNS [handle: Handle] = {
This is a monitor procedure used by the server to wait
for work to do. For the server to have work to do, there
must not be any errors associated with the stream. Furthermore,
either there must be a buffer waiting for the server, or there
must be an idle buffer and there must be a valid piece waiting
for the server. If there is a server buffer waiting, this routine
transfers the frontmost server buffer to the front of the
idle list, from which position the server will process it.
Note: it doesn't make any difference how the buffers on
the idle list are ordered.
ENABLE UNWIND => NULL;
buffer: REF Buffer;
WHILE
TRUE
DO
handle ← VSList;
WHILE handle #
NIL
DO
IF (handle.errorRope =
NIL)
THEN {
IF (handle.firstServerBuffer #
NIL)
THEN {
buffer ← handle.firstServerBuffer;
handle.firstServerBuffer ← buffer.next;
buffer.next ← handle.firstIdleBuffer;
handle.firstIdleBuffer ← buffer;
RETURN [handle];
}
ELSE
IF (handle.piece #
NIL)
THEN {
IF handle.firstClientBuffer = NIL THEN RETURN [handle];
IF (
NOT handle.piece.flush
AND (handle.piece.nBytes > 0)
AND (handle.firstIdleBuffer # NIL))
THEN RETURN [handle];
};
};
handle ← handle.next;
ENDLOOP;
Notify anybody trying to close a stream that we've done all
the work we can, and are waiting.
BROADCAST closeCondition;
WAIT serverCondition;
ENDLOOP;
};
giveClientBuffer:
ENTRY
PROC [handle: Handle] = {
ENABLE UNWIND => NULL;
buffer, b2: REF Buffer;
buffer ← handle.firstIdleBuffer;
handle.firstIdleBuffer ← buffer.next;
buffer.next ← NIL;
IF handle.firstClientBuffer = NIL
THEN {
handle.firstClientBuffer ← buffer;
NOTIFY handle.newPiece;
}
ELSE {
b2 ← handle.firstClientBuffer;
WHILE b2.next # NIL DO b2 ← b2.next ENDLOOP;
b2.next ← buffer;
};
BROADCAST client;
};
setError:
PUBLIC
ENTRY
PROC [handle: Handle, rope: Rope.
ROPE, code: ErrorCode] = {
This procedure just records error information in a voice
stream and wakes up waiting clients. This is made a separate
procedure because exclusive access is needed on the voice
stream, which the server doesn't have.
ENABLE UNWIND => NULL;
handle.errorRope ← rope;
handle.errorCode ← code;
BROADCAST client;
BROADCAST waitCondition;
};
getPiece: ENTRY PROC [handle: Handle] RETURNS [REF Piece] = {
This procedure is called only by the server.
It is made a separate procedure in order to guarantee exclusive
access to the voice stream information. Because of the high overheads
in opening and closing tunes, we check the next piece and if it's
the same tune, then don't bother closing the current one. NIL is
returned if either there are no pieces left, or if all the space of the
frontmost piece is allocated to buffers but the buffers haven't been
processed completely. This means we can't buffer ahead across
pieces, but the alternative results in messy synchronization
problems.
ENABLE UNWIND => NULL;
p: REF Piece;
WHILE
TRUE
DO
p ← handle.piece;
IF p = NIL THEN RETURN [NIL];
IF p.flush THEN p.nBytes ← 0;
IF p.nBytes > 0 THEN RETURN [p];
IF (handle.firstClientBuffer #
NIL)
OR (handle.firstServerBuffer # NIL) THEN RETURN [NIL];
IF p.tune #
NIL
THEN {
p ← p.next;
IF p #
NIL
THEN
IF p.tuneId = handle.piece.tuneId
THEN {
p.tune ← handle.piece.tune;
handle.piece.tune ← NIL;
};
};
IF handle.piece.tune # NIL THEN Jukebox.CloseTune[handle.jukebox, handle.piece.tune];
handle.piece ← handle.piece.next;
IF handle.piece = NIL THEN BROADCAST client;
ENDLOOP;
RETURN [NIL];
};
server: PROC [] = {
This procedure is responsible for buffering ahead and behind the
client processes. Be VERY careful with this code. The procedure
runs unmonitored, so that other processes can be getting and putting
voice data while we do disk operations. However, you must be
extremely careful to make sure that the processes don't get in each
others' way.
handle: Handle;
buffer: REF Buffer;
ourSize: INT;
skipLeading: INT;
piece: REF Piece;
noChirp: BOOLEAN;
Process.SetPriority[serverPriority];
WHILE
TRUE
DO
ENABLE {
Jukebox.Error => {
setError[handle: handle, code: BadPiece,
rope:
IO.PutFR["Piece for tune id %d starting at byte %d can't be accessed.",
IO.int[piece.tuneId], IO.int[piece.firstByte]]];
CONTINUE;
};
On EOF, just discard the rest of the current piece.
Jukebox.
EOF => {
piece.nBytes ← 0;
CONTINUE;
};
ANY => {
setError[handle: handle, code: Bug,
rope: "Unknown error in voice stream demon."];
IO.PutRope[ioStream, "Unknown error in voice stream demon.\n"];
REJECT;
};
};
handle ← getServerBuffer[];
buffer ← handle.firstIdleBuffer;
If this buffer is to be written, and the buffer is valid, then write
it out.
IF buffer.toJukebox
AND buffer.valid
THEN {
buffer.valid ← FALSE;
buffer.window ← Jukebox.FindChirp[self: handle.jukebox, tune: handle.piece.tune,
chirp: buffer.chirp, signalMissingChirp: FALSE, signalEOF: FALSE];
make sure that no partial chirps are recorded!
IF buffer.bytesAccountedFor < Jukebox.bytesPerChirp
THEN {
fill the rest of the chirp with silence
IF (buffer.runIndex
MOD 2) # 0
THEN {
IF buffer.runData.runArray[buffer.runIndex] = 0 THEN buffer.runIndex ← buffer.runIndex - 1
ELSE {
buffer.runIndex ← buffer.runIndex + 1;
buffer.runData.runArray[buffer.runIndex] ← 0;
};
};
buffer.runData.runArray[buffer.runIndex] ← buffer.runData.runArray[buffer.runIndex] + (Jukebox.bytesPerChirp - buffer.bytesAccountedFor);
};
CheckRunData[buffer];
FS.Write[file: buffer.window.file, to: buffer.window.base, nPages: buffer.chirpSpace.count, from: VM.AddressForPageNumber[buffer.chirpSpace.page]];
IO.PutF[ioStream, "Wrote chirp %d to disk.\n", IO.int[buffer.chirp]];
};
Collect information for the next buffer's worth of voice. First of all,
find a piece that isn't empty.
piece ← getPiece[handle];
IF piece = NIL THEN LOOP;
ioStream.PutF["Number of bytes left: %d\n", IO.int[piece.nBytes]];
Set up the encryption key index (copy over from piece).
buffer.keyIndex ← piece.keyIndex;
Make sure that the tune is open.
IF (piece.tune =
NIL)
THEN {
IF (piece.create) THEN piece.tune ← Jukebox.CreateTune[self: handle.jukebox, tuneId: piece.tuneId]
ELSE piece.tune ← Jukebox.OpenTune[self: handle.jukebox, tuneId: piece.tuneId, write: NOT piece.playback];
};
Figure out which chirp this is.
IF piece.firstByte < 0 THEN piece.firstByte ← Jukebox.TuneSize[piece.tune] * Jukebox.bytesPerChirp;
buffer.chirp ← piece.firstByte / Jukebox.bytesPerChirp;
Figure out which portion of the chirp we'll actually use.
skipLeading ← piece.firstByte MOD Jukebox.bytesPerChirp;
buffer.bytesAccountedFor ← skipLeading;
ourSize ← Jukebox.bytesPerChirp - skipLeading;
Trim the tail if needed
IF piece.nBytes < ourSize
THEN {
disallow a recording that ends in the middle of a chirp
IF piece.playback THEN ourSize ← piece.nBytes
ELSE piece.nBytes ← ourSize;
IF piece.playback THEN buffer.bytesAccountedFor ← Jukebox.bytesPerChirp - ourSize;
};
Account for the part of the piece in this chirp.
piece.nBytes ← piece.nBytes - ourSize;
piece.firstByte ← piece.firstByte + ourSize;
Read in the chirp, if necessary. Note: we don't look up the disk location until its time to actually transfer information. This way, the jukebox routines won't allocate the space until it's actually needed. Also, for playback we substitute a homemade all zero chirp if there isn't a chirp on the disk.
buffer.runIndex ← 0;
IF (ourSize # Jukebox.bytesPerChirp)
OR piece.playback
THEN {
noChirp ← FALSE;
buffer.window ← Jukebox.FindChirp[self: handle.jukebox, tune: piece.tune,
chirp: buffer.chirp, signalEOF: piece.playback,
signalMissingChirp: TRUE
! Jukebox.MissingChirp => {noChirp ← TRUE; CONTINUE}];
IF noChirp
THEN {
IF piece.playback
THEN {
Construct a chirp with a single run of silence
buffer.block.startIndex ← 0;
buffer.block.stopIndexPlusOne ← 0;
buffer.runData.runArray[0] ← ourSize;
buffer.runData.packetSize ← 0;
}
ELSE {
-- build a fake chirp with the right structure
buffer.block.startIndex ← 0;
buffer.block.stopIndexPlusOne ← ourSize;
buffer.runData.runArray[0] ← skipLeading;
buffer.runData.packetSize ← 0;
};
}
ELSE {
Read in the whole chirp, including its runData
FS.Read[file: buffer.window.file, from: buffer.window.base, nPages: buffer.chirpSpace.count, to: VM.AddressForPageNumber[buffer.chirpSpace.page]];
IF skipLeading > 0
THEN {
runSize: CARDINAL;
silenceRunType: BOOL;
The value of skipLeading is the amount to chase down the runData
buffer.block.startIndex ← 0;
DO
runSize ← buffer.runData.runArray[buffer.runIndex];
silenceRunType ← (buffer.runIndex MOD 2) = 0;
IF skipLeading >= runSize
THEN {
skip the whole run
skipLeading ← skipLeading - runSize;
IF NOT silenceRunType THEN buffer.block.startIndex ← buffer.block.startIndex + runSize;
buffer.runIndex ← buffer.runIndex + 1;
}
ELSE {
-- last run
IF NOT silenceRunType THEN buffer.block.startIndex ← buffer.block.startIndex + skipLeading;
buffer.runData.runArray[buffer.runIndex] ← runSize - skipLeading;
EXIT;
};
ENDLOOP;
buffer.block.stopIndexPlusOne ← buffer.block.startIndex + ourSize;
}
ELSE {
buffer.block.startIndex ← 0;
Not all of this is needed, since some might be silence, but we don't know how much and don't want to spend the time analyzing the runData.
buffer.block.stopIndexPlusOne ← ourSize;
};
};
IO.PutF[ioStream, "Read chirp %d from disk.\n", IO.int[buffer.chirp]];
}
ELSE {
-- whole chirp is involved and it is record
buffer.block.startIndex ← 0;
buffer.block.stopIndexPlusOne ← ourSize;
buffer.runData.runArray[0] ← 0;
buffer.runData.packetSize ← 0;
};
buffer.toJukebox ← NOT piece.playback;
giveClientBuffer[handle];
ENDLOOP;
};
PrivRunDataObject: PRIVATE TYPE = RECORD [
This record represents the extra space in each chirp, after the end of recorded voice.
runArray: ARRAY [0..95) OF CARDINAL,
packetSize: CARDINAL
];
CheckRunData:
PROC [buffer:
REF VoiceStream.Buffer] = {
mrd: LONG POINTER TO PrivRunDataObject ← LOOPHOLE[buffer.runData];
total: CARDINAL ← 0;
run: CARDINAL;
FOR ri:
NAT
IN [0..95)
DO
run ← mrd.runArray[ri];
IF run > Jukebox.bytesPerChirp
THEN {
mrd.runArray[ri] ← Jukebox.bytesPerChirp - total;
IF debugRuns THEN SIGNAL Foo;
EXIT;
};
total ← total + run;
IF total > Jukebox.bytesPerChirp
THEN {
mrd.runArray[ri] ← run - (total - Jukebox.bytesPerChirp);
IF debugRuns THEN SIGNAL Foo;
EXIT;
};
IF total = Jukebox.bytesPerChirp THEN EXIT;
ENDLOOP;
};
StartServer: PUBLIC PROC [stream: IO.STREAM] = { ioStream ← stream; };
Init: PROC = {
This procedure simply forks the stream buffer server and
initializes shared state.
p: PROCESS;
p ← FORK server[];
Process.Detach[p];
};
Init[];
END.
1983, Stewart, run-encoded chirps
December 27, 1983 1:56 pm, Stewart, Cedar 5