FileWriterImpl.Mesa
Written by Paxton. March 1981
Last Edited by: Lamming, June 10, 1983 2:47 pm
DIRECTORY
File    USING [Capability],
FileIO    USING [StreamFromCapability],
FileWriter  USING [BlockList, BlockListBody, blockSize, FileWriterBody, Offset, OfWriter, Ref, WriteChar],
IO     USING [Close, Handle, PutBlock, SetIndex, SetLength],
MonitoredQueue USING [Add, Close, Create, EndOfQueue, MQ, Remove, Reset],
Rope    USING [Concat, FromRefText, ROPE],
RopeReader  USING [],
T2FileOps  USING [controlTrailerId, DataPaddedFlag, DataUnpaddedFlag];
FileWriterImpl: CEDAR MONITOR
IMPORTS IO, FileIO, FileWriter, MonitoredQueue, Rope
EXPORTS FileWriter
SHARES FileWriter = BEGIN
ROPE: TYPE = Rope.ROPE;
versionNumber:INT ← 13; -- Tioga2 version stamp
---------------------------------------------------------------
PRINCIPAL PUBLIC INTERFACES
Close: PUBLIC PROC [control, data: FileWriter.Ref, textOnly: BOOLEAN] RETURNS [dataLen, count: FileWriter.Offset, output: ROPE] = {
Close the file - i.e. compute the trailer information and write out all the buffers.
deltaBytes: PACKED ARRAY [0..3] OF CHARACTER;
controlStart, controlLen, filePropsOffset: FileWriter.Offset ← 0;
stream: IO.Handle ← data.stream;
len, controlBlocks, dataBlocks: NAT ← 0;
toRope: BOOLEAN ← data.toRope;
controlList, dataList: FileWriter.BlockList;
controlBlock, dataBlock: REF TEXT;
pad:BOOLEAN ←FALSE; -- says if a data pad byte has been output
FinishWriter: PROC [writer: FileWriter.Ref] RETURNS [list: FileWriter.BlockList, block: REF TEXT] = {
Theblocks are chained in reverse order and need re-ordering before output
list ← Reverse[writer.blockList]; block ← writer.block; FreeFileWriter[writer]
};
GetChars: PROC [num: NAT] RETURNS [chars: REF TEXT] = {
GetIt: PROC [blocks: NAT, list: FileWriter.BlockList, block: REF TEXT] RETURNS [ok: BOOLEAN] = {
IF num NOT IN [0..blocks) THEN RETURN [FALSE];
FOR lst: FileWriter.BlockList ← list, lst.next UNTIL lst=NIL DO
IF num = 0 THEN { chars ← lst.block; RETURN [TRUE] };
num ← num-1; ENDLOOP;
IF num = 0 THEN { chars ← block; RETURN [TRUE] };
ERROR
};
IF GetIt[dataBlocks, dataList, dataBlock] THEN RETURN [chars];
num ← num-dataBlocks;
IF GetIt[controlBlocks, controlList, controlBlock] THEN RETURN [chars];
ERROR
};
LastBlock: PROC [lst: FileWriter.BlockList] RETURNS [block: REF TEXT] = {
Find the last block in a a writers block list
FOR l: FileWriter.BlockList ← lst, l.next DO
IF l.next=NIL THEN RETURN [l.block];
ENDLOOP
};
MakeRope: PROC [first, num: NAT] RETURNS [ROPE] = {
half: NAT;
SELECT num FROM
0 => RETURN [NIL];
1 => RETURN [Rope.FromRefText[GetChars[first]]];
ENDCASE;
half ← num/2;
RETURN [Rope.Concat[MakeRope[first, half], MakeRope[first+half, num-half]]]
};
Reverse: PROC [lst: FileWriter.BlockList] RETURNS [prev: FileWriter.BlockList] = {
Reverse a writers block list
next: FileWriter.BlockList;
UNTIL lst=NIL DO
next ← lst.next; lst.next ← prev; prev ← lst; lst ← next; ENDLOOP
};
dataBlocks ← data.blockCount;
controlStart ← dataBlocks*LONG[FileWriter.blockSize];
IF (len�ta.block.length) MOD 2 = 1 THEN {
round up the data to an even number of bytes (word boundary) because this makes the copying to file buffers go much faster
FileWriter.WriteChar[0C, data]; -- this constitutes an extra EOF
pad ← TRUE; -- remind ourselves to mark trailer to say data is padded
len ← len+1
};
controlStart ← controlStart+len; -- calculate the size in bytes
dataBlocks ← dataBlocks+1;  -- counts number of full and partial blocks needed
IF control # NIL THEN {
IF ~textOnly THEN {
chars: REF TEXT;
WriteControlInt: PROC [int: INT] = {
32 bits written as four bytes (unencoded)
deltaBytes ← LOOPHOLE[int];
FOR i:NAT IN [0..3] DO
FileWriter.WriteChar[deltaBytes[i], control];
ENDLOOP
};
Check: PROC [writer: FileWriter.Ref, kind: FileWriter.OfWriter] = {
Check consistency of writer structure
IF writer.kind # kind THEN ERROR;
IF writer.toRope # toRope THEN ERROR;
IF writer.closeStream # data.closeStream THEN ERROR;
IF writer.stream # data.stream THEN ERROR
};
Check[control, control];
filePropsOffset ← LONG[FileWriter.blockSize]*control.blockCount+control.block.length;
WriteControlInt[versionNumber]; -- Tioga2 version stamp
-- this is the place to write out the file props, if any
-- this is the place to write out the symbol table
Write out the trailer block which contains a one byte magic number, a pad flag, the offset of the first byte of the file properties, the offset of the control information, and the total file length = 14 bytes
FileWriter.WriteChar[T2FileOps.controlTrailerId, control];
FileWriter.WriteChar[IF pad THEN T2FileOps.DataPaddedFlag ELSE T2FileOps.DataUnpaddedFlag, control];
WriteControlInt[filePropsOffset]; -- Where the file properties begin
WriteControlInt[controlStart];  -- index of first control byte
controlLen ← LONG[FileWriter.blockSize]*control.blockCount+control.block.length;
controlBlocks ← control.blockCount+1; -- counts number of full and partial blocks needed
WriteControlInt[controlStart+controlLen+4]; -- file length (+4 accounts for self)
chars ← IF control.blockList=NIL THEN control.block ELSE LastBlock[control.blockList];
};
[controlList, controlBlock] ← FinishWriter[control]; -- reverse blocklist, free writer structure
};
IF data.kind # data THEN ERROR; -- checking for impossible error
dataBlock ← data.block;
IF toRope THEN {
dataList ← Reverse[data.blockList];
output ← MakeRope[0, dataBlocks+controlBlocks]
} -- return a balanced rope
ELSE {
MonitoredQueue.Close[data.blockQueue];
TRUSTED {JOIN data.consumer};
FreeMQ[data.blockQueue];
IO.PutBlock[stream, dataBlock];
FreeBLOCK[dataBlock];
IF control # NIL THEN {
FOR lst:FileWriter.BlockList ← controlList, lst.next UNTIL lst=NIL DO
IF ~textOnly THEN IO.PutBlock[stream, lst.block];
FreeBLOCK[lst.block];
ENDLOOP;
IF ~textOnly THEN IO.PutBlock[stream, controlBlock];
FreeBLOCK[controlBlock]
}
};
count ← controlStart+controlLen;
dataLen ← controlStart;
IF data.closeStream THEN IO.Close[stream];
FreeFileWriter[data]
};
OpenC: PUBLIC PROC [capability: File.Capability, start: FileWriter.Offset ← 0, makeControl: BOOLEAN ← TRUE] RETURNS [control, data: FileWriter.Ref] = {
Get ready to output document to a file (file is specified as a capability)
stream: IO.Handle = FileIO.StreamFromCapability[capability, overwrite];
IO.SetLength[stream, start];
IO.SetIndex[stream, start];
[control, data] ← CreateWriters[stream, FALSE, makeControl, TRUE]
};
ToRope: PUBLIC PROC [makeControl: BOOLEAN ← TRUE] RETURNS [control,data: FileWriter.Ref] = {
Get ready to write document to a rope
[control, data] ← CreateWriters[NIL, TRUE, makeControl, FALSE]
};
ToStream: PUBLIC PROC [stream: IO.Handle, makeControl: BOOLEAN ← TRUE] RETURNS [control, data: FileWriter.Ref] = {
Get ready to write document to a stream
[control, data] ← CreateWriters[stream, FALSE, makeControl, FALSE]
};
---------------------------------------------------------------
MONITORED QUEUE MANAGEMENT
A two element cache (q1,q2) of Monitored queues is maintained from which new queues are initially allocated. Note that only one queue is needed per file write in progress.
q1, q2: MonitoredQueue.MQ;
FreeMQ: ENTRY PROC [q: MonitoredQueue.MQ] = {
Looks for a spare slot in the MQ cache and if found plugs in the returned queue. If there are no spare slots the queue is ignored - presumably to be recovered by the garbage collector
ENABLE UNWIND => NULL;
IF q=NIL THEN RETURN;
MonitoredQueue.Reset[q];
IF q1 = NIL THEN q1 ← q
ELSE
IF q2 = NIL THEN q2 ← q
};
GetMQ: ENTRY PROC RETURNS [q: MonitoredQueue.MQ] = {
Searches the cache for a spare queue. If none available allocates a new one
ENABLE UNWIND => NULL;
IF q1 # NIL THEN { q ← q1; q1 ← NIL }
ELSE
IF q2 # NIL THEN { q ← q2; q2 ← NIL }
ELSE q ← MonitoredQueue.Create[]
};
---------------------------------------------------------------
FILE WRITER MANAGEMENT
A two element cache (fw1, fw2) of FileWriterBodys is maintained from which new FileWriterBodys are initially allocated.
fw1, fw2: FileWriter.Ref;
CreateWriters: PROC [stream: IO.Handle, toRope: BOOLEAN ← FALSE, makeControl, closeStream: BOOLEAN ← TRUE] RETURNS [control, data: FileWriter.Ref] = {
Make a writer for the data stream and, if required, a writer for the control stream. Note that the 'stream' and 'clostream' information is identical in both writers
MakeOne: PROC [kind: FileWriter.OfWriter] RETURNS [writer: FileWriter.Ref] = {
writer ← GetFileWriter[];
writer.kind ← kind;
writer.block ← NewBLOCK[];
writer.stream ← stream;
writer.toRope ← toRope;
writer.closeStream ← closeStream
};
IF makeControl THEN { -- make control writer
control ← MakeOne[control];
THROUGH [0..1] DO -- these zeros terminate the data portion
FileWriter.WriteChar[0C, control];
ENDLOOP;
};
data ← MakeOne[data];
IF ~toRope THEN { -- fork process to write data as it is produced
data.blockQueue ← GetMQ[];
data.consumer ← FORK Consumer[data]
}
};
FreeFileWriter: ENTRY PROC [fw: FileWriter.Ref] = {
Reinitialise the file writer and if there is room, put it back in the cache. If there is no room drop it on the floor where the GC will presumably sweep it up.
ENABLE UNWIND => NULL;
fw.block ← NIL;
fw.blockList ← NIL;
fw.blockQueue ← NIL;
fw.blockCount ← 0;
fw.stream ← NIL;
fw.kind ← unused;
IF fw1 = NIL THEN fw1 ← fw
ELSE
IF fw2 = NIL THEN fw2 ← fw
};
GetFileWriter: ENTRY PROC RETURNS [fw: FileWriter.Ref] = {
Do a linear search of the file cache for a spare file writer. If none available then allocate a new one
ENABLE UNWIND => NULL;
IF fw1 # NIL THEN { fw ← fw1; fw1 ← NIL }
ELSE
IF fw2 # NIL THEN { fw ← fw2; fw2 ← NIL }
ELSE fw ← NEW[FileWriter.FileWriterBody];
IF fw.kind # unused THEN ERROR
};
---------------------------------------------------------------
BLOCK CACHE MANAGEMENT
"blocks" contains a cache of blocks produced by the writers as they write blocks out.
blks: NAT = 8;
BlockArray: TYPE = ARRAY [0..blks) OF REF TEXT;
blocks: REF BlockArray ← NEW[BlockArray];
freeBlock: CONDITION;
BumpWriter: PUBLIC PROC [writer: FileWriter.Ref] RETURNS [block: REF TEXT] = {
Queues a block for output. If the output stream is "data" then the block is simply placed on a monitored queue to be consumed and written out by Consumer. Otherwise a queue node is created and chained onto the head (sic) of the block queue (later to be reversed). The queue node points to the block.
A new block is allocated and returned.
writer.blockCount ← writer.blockCount+1; -- number added to queue/list
IF writer.kind=data AND ~writer.toRope THEN {
MonitoredQueue.Add[writer.block, writer.blockQueue];
writer.block ← block ← GetBLOCK[]} -- get a block from the cache
ELSE { -- create a new block since it won't be returned for a while
writer.blockList ← NEW[FileWriter.BlockListBody ← [writer.block, writer.blockList]];
writer.block ← block ← NewBLOCK[]};
};
Consumer: PROC [data: FileWriter.Ref] = {
The "data" portion of the file goes at the front of the file so it can be written out as it is produced. This procedure is the kernel of a process, forked inside MakeOne. It writes out blocks placed on the data stream.
queue: MonitoredQueue.MQ ← data.blockQueue;
stream: IO.Handle ← data.stream;
DO
x: REF ANY ← MonitoredQueue.Remove[queue ! MonitoredQueue.EndOfQueue => EXIT];
WITH x SELECT FROM
xx: REF TEXT => {
IO.PutBlock[stream, xx];
FreeBLOCK[xx]
};
ENDCASE => ERROR;
ENDLOOP
};
FreeBLOCK: ENTRY PROC [b: REF TEXT] = {
If there is a slot in the cache the block is inserted, else it is ignored - presumably to be recovered by the garbage collector.
ENABLE UNWIND => NULL;
FOR i:NAT IN [0..blks) DO
IF blocks[i] = NIL THEN {
blocks[i] ← b;
BROADCAST freeBlock;
RETURN
};
ENDLOOP
};
GetBLOCK: ENTRY PROC RETURNS [b: REF TEXT] = {
Does a linear search of the block cache to find a free block. If no block can be found then it allocates a new block
ENABLE UNWIND => NULL;
WHILE TRUE DO
FOR i:NAT IN [0..blks) DO
IF blocks[i] # NIL THEN {
b ← blocks[i];
blocks[i] ← NIL;
b.length ← 0;
RETURN[b];
};
ENDLOOP;
WAIT freeBlock;
ENDLOOP;
};
NewBLOCK: PROC RETURNS [b: REF TEXT] = INLINE {
RETURN[NEW[TEXT[FileWriter.blockSize]]]};
FOR i:CARDINAL IN [0..blks) DO
blocks[i] ← NewBLOCK[];
ENDLOOP;
END.