FileStreamMoreImpl.mesa
Copyright Ó 1985, 1986 by Xerox Corporation. All rights reserved.
Hagmann on December 6, 1983 4:50 pm
Russ Atkinson (RRA) February 4, 1985 3:27:09 pm PST
Bob Hagmann November 12, 1986 4:54:05 pm PST
Please maintain change log at end of file.
This compilation is logically part of FileStreamImpl, but must be separate since it uses a module monitor (FileStreamImpl uses object monitors) and we export to FS here (there is a type conflict on Close).
DIRECTORY
BasicTime USING [GMT],
FileStream USING [ErrorFromStream, OpenFileFromStream, StreamFromOpenFile, StreamFromOpenStream],
FileStreamPrivate USING [ BufferNodeHandle, BufferNode, FileDataHandle, FileData, NodeStatus, ProcessNode ],
FS USING [AccessOptions, ByteCount, Close, Create, Error, ErrorDesc, ExtendFileProc, FileType, InitialPosition, Lock, Open, OpenFile, OpenOrCreate, PagesForBytes, StreamBufferParms, StreamOptions],
IO USING [STREAM],
Process USING [Detach, Pause, Priority, priorityForeground, SecondsToTicks, SetPriority, Ticks],
Rope USING [ROPE];
FileStreamMoreImpl: CEDAR MONITOR
IMPORTS FileStream, FileStreamPrivate, FS, Process
EXPORTS FileStreamPrivate, FS = BEGIN
OPEN Rope, IO;
FileData: TYPE = FileStreamPrivate.FileData;
BufferNode: TYPE = FileStreamPrivate.BufferNode;
FileDataHandle: TYPE = FileStreamPrivate.FileDataHandle;
BufferNodeHandle: TYPE = FileStreamPrivate.BufferNodeHandle;
MaxNumberOfProcesses: NAT ← 8 ;
NumberOfProcesses: NAT = 3 ;
nProcesses: CARD ← 0;
ModuleCondition: CONDITION;
QueueRecord: TYPE = RECORD [
fileData: FileDataHandle,
node: BufferNodeHandle
,nodeStatus: FileStreamPrivate.NodeStatus
,nodeFirstFileByteInBuffer: INT
];
QUEUESIZE: INT = 20 ;
QueueRec: TYPE = RECORD [
queue: ARRAY [0..QUEUESIZE) OF QueueRecord,
putQueue: INT ← 0,
getQueue: INT ← 0,
noEntries: INT ← 0,
queueStuck: BOOLFALSE
];
Queue: REF QueueRec ← NIL;
StartRequest: PUBLIC ENTRY PROC [ fileData: FileDataHandle, node: BufferNodeHandle ] = {
Queue up a request. Normal case is to queue and BROADCAST without waiting. If the queue is full, then wait for an entry to free up and then queue request. Except when the queue is full (and thus we are really behind), this returns immediately.
tempCount: INT ← Queue.noEntries ;
ptr: INT ← Queue.getQueue ;
IF node.status # needsParallelRead AND node.status # needsParallelWrite THEN ERROR ;
WHILE tempCount > 0 DO
IF Queue.queue[ptr].node = node THEN ERROR ;
ptr ← IF ptr = QUEUESIZE-1 THEN 0 ELSE ptr+1 ;
tempCount ← tempCount - 1 ;
ENDLOOP;
WHILE Queue.noEntries >= QUEUESIZE - 1 DO
WAIT ModuleCondition ;
ENDLOOP;
Queue.queue[Queue.putQueue] ← [fileData, node --, node.status, node.firstFileByteInBuffer--];
Queue.putQueue ← IF Queue.putQueue = QUEUESIZE-1 THEN 0 ELSE Queue.putQueue+1 ;
Queue.noEntries ← Queue.noEntries + 1;
Queue.queueStuck ← FALSE;
BROADCAST ModuleCondition ;
};
FindSomethingToDo: ENTRY PROC [] RETURNS [fileData: FileDataHandle, node: BufferNodeHandle ] = {
This procedure is where the processes all wait for something to do.
WHILE Queue.noEntries = 0 DO
WAIT ModuleCondition ;
ENDLOOP;
[fileData, node ] ← Queue.queue[Queue.getQueue] ;
Queue.queue[Queue.getQueue] ← [NIL,NIL];
Queue.getQueue ← IF Queue.getQueue = QUEUESIZE-1 THEN 0 ELSE Queue.getQueue+1 ;
Queue.noEntries ← Queue.noEntries - 1;
Queue.queueStuck ← FALSE;
BROADCAST ModuleCondition ;
};
FileStreamForegroundProcess: PROC [] = {
This procedure is the top of the processes that do all the parallel reads and writes.
Process.SetPriority[Process.priorityForeground];
DO
fileData: FileDataHandle ;
node: BufferNodeHandle;
[fileData, node] ← FindSomethingToDo [];
FileStreamPrivate.ProcessNode[fileData, node] ;
ENDLOOP;
};
FileStreamForegroundWatcher: PROC [] = {
This procedure is the top of the processes that do all the parallel reads and writes.
threeSeconds: Process.Ticks ← Process.SecondsToTicks[3];
Process.SetPriority[Process.priorityForeground];
DO
IF Queue.noEntries # 0 AND nProcesses < MaxNumberOfProcesses THEN {
IF Queue.queueStuck THEN TRUSTED {
Process.Detach[FORK FileStreamForegroundProcess[]];
};
Queue.queueStuck ← TRUE;
};
Process.Pause[threeSeconds];
ENDLOOP;
};
Implementation of the "workstation FS" subclass of FileStream. This is mostly a veneer over the generic FileStream; only stream creation is special.
StreamOpen: PUBLIC PROC [fileName: ROPE, accessOptions: FS.AccessOptions,
streamOptions: FS.StreamOptions, keep: CARDINAL, createByteCount: FS.ByteCount,
streamBufferParms: FS.StreamBufferParms, extendFileProc: FS.ExtendFileProc,
wantedCreatedTime: BasicTime.GMT, remoteCheck: BOOL, wDir: ROPE, checkFileType: BOOL, fileType: FS.FileType]
RETURNS [STREAM] = {
fileHandle: FS.OpenFile = SELECT accessOptions FROM
$read => FS.Open[name: fileName, wantedCreatedTime: wantedCreatedTime, remoteCheck: remoteCheck, wDir: wDir, checkFileType: checkFileType, fileType: fileType],
$create => FS.Create[name: fileName, keep: keep, pages: FS.PagesForBytes[createByteCount], wDir: wDir, fileType: fileType],
$append => FS.OpenOrCreate[name: fileName, keep: keep, pages: FS.PagesForBytes[createByteCount], wDir: wDir, checkFileType: checkFileType, fileType: fileType],
$write => FS.Open[name: fileName, lock: $write, wantedCreatedTime: wantedCreatedTime, remoteCheck: remoteCheck, wDir: wDir, checkFileType: checkFileType, fileType: fileType],
ENDCASE => ERROR;
RETURN[FileStream.StreamFromOpenFile[openFile: fileHandle, accessRights: IF accessOptions = $read THEN $read ELSE $write, initialPosition: IF accessOptions = $append THEN $end ELSE $start, streamOptions: streamOptions, streamBufferParms: streamBufferParms, extendFileProc: extendFileProc
! FS.Error => FS.Close[file: fileHandle]]];
};
The following procedures simply re-export the ones from FileStream. However, directly exporting the procedure values imported from FileStream does not work for some reason.
StreamFromOpenFile: PUBLIC PROC [openFile: FS.OpenFile, accessRights: FS.Lock, initialPosition: FS.InitialPosition, streamOptions: FS.StreamOptions, streamBufferParms: FS.StreamBufferParms, extendFileProc: FS.ExtendFileProc] RETURNS [STREAM] = {
RETURN [FileStream.StreamFromOpenFile[openFile: openFile, accessRights: accessRights, initialPosition: initialPosition, streamOptions: streamOptions, streamBufferParms: streamBufferParms, extendFileProc: extendFileProc]];
};
StreamFromOpenStream: PUBLIC PROC [self: STREAM] RETURNS [STREAM] = {
RETURN [FileStream.StreamFromOpenStream[self]];
};
OpenFileFromStream: PUBLIC PROC [self: STREAM] RETURNS [FS.OpenFile] = {
RETURN [FileStream.OpenFileFromStream[self]];
};
ErrorFromStream: PUBLIC PROC [self: STREAM] RETURNS [FS.ErrorDesc] = {
RETURN [FileStream.ErrorFromStream[self]];
};
Initialization
Init: PROC [] = {
Queue ← NEW[QueueRec];
THROUGH [1..NumberOfProcesses] DO
nProcesses ← nProcesses + 1;
TRUSTED {Process.Detach[FORK FileStreamForegroundProcess[]]; };
ENDLOOP;
TRUSTED {Process.Detach[FORK FileStreamForegroundWatcher[]]; };
};
Init[];
END.
CHANGE LOG
Created by Hagmann on December 6, 1983 3:02 pm
From a suggestion by Mark Brown, implement a cache of processes on a package wide basis.
Modified by Hagmann on December 19, 1984 9:49:48 am PST
Reformatted, and changed the queue be a collectable object to reduce MDS and GFI use.
Bob Hagmann June 10, 1986 8:22:35 am PDT
merged FSStreamImpl and FileStreamProcessCacheImpl to form this program
Bob Hagmann November 12, 1986 4:48:12 pm PST
added FileStreamForegroundWatcher - grow the number of foreground processes if needed