[Indigo]<Voice>bluejay>VoiceStreamServer.mesa!17

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.

Ades, February 10, 1986 4:01:59 pm PST

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, RunComponent, singlePktLength, RunElementType, RunArrayRange],

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

SHARES 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.runData.runArray[buffer.runIndex].elementType ~= silence THEN

last thing placed in chirp was non-silent

{ buffer.runIndex ← buffer.runIndex + (IF buffer.runData.runArray[buffer.runIndex].elementType = soundEnergy THEN 2 ELSE 1);
buffer.runData.runArray[buffer.runIndex] ← [silence[0]]
};
buffer.runData.runArray[buffer.runIndex]← [silence[NARROW[buffer.runData.runArray[buffer.runIndex], Jukebox.RunComponent[silence]].length+(Jukebox.bytesPerChirp - buffer.bytesAccountedFor)]];

buffer.bytesAccountedFor ← Jukebox.bytesPerChirp -- just for purity

};

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.

This should be changed - nobody should ever let a tune be opened down here!!!!!

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];

}

ELSE IF NOT piece.playback AND NOT piece.tune.writable THEN ERROR;

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;

buffer.playedBytes ← 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] ← [silence[ourSize]];

}

ELSE { -- build a fake chirp with the right structure

buffer.block.startIndex ← 0;

buffer.block.stopIndexPlusOne ← ourSize;

buffer.runData.runArray[0] ← [silence[skipLeading]];

};

}

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;

The value of skipLeading is the amount to chase down the runData

buffer.block.startIndex ← 0;

WITH curr: buffer.runData.runArray[buffer.runIndex] SELECT FROM
silence => runSize ← curr.length;
singlePkt => runSize ← Jukebox.singlePktLength;
soundEnergy => runSize ← runSize ← NARROW[buffer.runData.runArray[buffer.runIndex+1], Jukebox.RunComponent[soundLength]].length;
ENDCASE => ERROR;

IF skipLeading >= runSize THEN {

skip the whole run

skipLeading ← skipLeading - runSize;

IF buffer.runData.runArray[buffer.runIndex].elementType # silence THEN buffer.block.startIndex ← buffer.block.startIndex + runSize;

buffer.runIndex ← buffer.runIndex + (IF buffer.runData.runArray[buffer.runIndex].elementType=soundEnergy THEN 2 ELSE 1);

}

ELSE { -- last run:

WITH curr: buffer.runData.runArray[buffer.runIndex] SELECT FROM

silence => curr.length ← runSize - skipLeading;
soundEnergy =>
buffer.runData.runArray[buffer.runIndex+1] ← [soundLength[IF piece.playback
THEN NARROW[buffer.runData.runArray[buffer.runIndex+1], Jukebox.RunComponent[soundLength]].length - skipLeading

i.e. on playback this is what is left to be played back

ELSE skipLeading]]; -- but on record this is what is already there
singlePkt => IF piece.playback THEN buffer.playedBytes ← skipLeading
ELSE

since we are recording it is okay to rewrite this and the next runArray entries

{buffer.runData.runArray[buffer.runIndex] ← [soundEnergy[NARROW[buffer.runData.runArray[buffer.runIndex], Jukebox.RunComponent[soundEnergy]].energy]];
buffer.runData.runArray[buffer.runIndex+1] ← [soundLength[skipLeading]];
};
ENDCASE;
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] ← [silence[0]]

};

buffer.toJukebox ← NOT piece.playback;

IF buffer.toJukebox THEN buffer.runData.ambientLevel ← piece.ambientLevel;

giveClientBuffer[handle];

ENDLOOP;

};

CheckRunData: PROC [buffer: REF Buffer] = {

total: CARDINAL ← 0;

run: CARDINAL;
currElement: Jukebox.RunElementType ← silence; -- must simply start as ~soundEnergy

FOR ri: Jukebox.RunArrayRange IN [0..95) DO
IF currElement = soundEnergy -- acutally it is prevElement is this context
THEN IF buffer.runData.runArray[ri].elementType = soundLength THEN
{currElement ←soundLength; LOOP} ELSE ERROR;

currElement ← buffer.runData.runArray[ri].elementType;
WITH curr: buffer.runData.runArray[ri] SELECT FROM
silence => run ← curr.length;
singlePkt => run ← Jukebox.singlePktLength;
soundEnergy => run ← NARROW[buffer.runData.runArray[ri+1],
Jukebox.RunComponent[soundLength]].length;
ENDCASE => ERROR;

total ← total + run;

IF total > Jukebox.bytesPerChirp THEN
{
WITH curr: buffer.runData.runArray[ri] SELECT FROM
silence => curr.length ← run - (total - Jukebox.bytesPerChirp);
soundEnergy => buffer.runData.runArray[ri+1] ← [soundLength[run - (total - Jukebox.bytesPerChirp)]];
singlePkt => -- since this is last valid entry in chirp, okay to trample 'nextElement'
{ buffer.runData.runArray[buffer.runIndex]← [soundEnergy[NARROW[buffer.runData.runArray[buffer.runIndex], Jukebox.RunComponent[singlePkt]].energy]];
buffer.runData.runArray[ri+1] ← [soundLength[Jukebox.singlePktLength - (total - Jukebox.bytesPerChirp)]]
};
ENDCASE;
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