From VisualWorks®, 7.2 of October 20, 2003 on October 20, 2003 at 12:40:30 pm (Dialog confirm: 'You are filing-in a Parcel source file!\\While this is possible it will not have\the same effect as loading the parcel.\None of the Parcel''s prerequisites will\be loaded and none of its load actions\will be performed.\\Are you sure you want to file-in?' withCRs) ifFalse: [self error: 'Parcel file-in abandoned. Choose terminate or close.'] <class> <name>NonBlockingPipeInterface</name> <environment>Examples</environment> <super>External.ExternalInterface</super> <private>false</private> none infd outfd generator writer readers results running <imports> Examples.NonBlockingPipeInterfaceDictionary.* </imports> <category>ExternalInterface-THAPI Example</category> <attributes> <includeFiles></includeFiles> <includeDirectories></includeDirectories> <libraryFiles><item>libc.so</item> <item>libc.sl</item> <item>msvcrt40.dll</item></libraryFiles> <libraryDirectories><item>/lib</item> <item>/usr/shlib</item> <item>/usr/lib</item> <item>$(windir)\system</item> <item>$(windir)\system32</item></libraryDirectories> <beVirtual>false</beVirtual> <optimizationLevel>full</optimizationLevel> <package>THAPIExample</package> </attributes> </class> <comment> Examples.NonBlockingPipeInterface
This class provides a simple test of the threaded API. Two + N processes cooperate to print the results of an nfib computation in the system transcript. One process opens a transcript, opens a pipe, forks off a writer process and N reader processes and then computes a series of nfib benchmark times which it writes to its transcript and puts into a shared queue. The writer process gets items from the shared queue and does a blocking write of each item as a string to the pipe. Each of the N reader processes loops doing blocking reads from the pipe and writing the result of the read to the system transcript. When its running one sees results of the nfib computation written to the interface's private transcript followed by the same information written to the system transcript from each reader. The writer process is an unnecessary frill. Some example doits: "10 readers. ctrl-y to terminate" NonBlockingPipeInterface new readers: 10 "10 readers in the background. Inspect the nbpi and send it terminate when you get bored." | nbpi | nbpi := NonBlockingPipeInterface new. [nbpi readers: 10] forkAt: Processor activePriority - 1. nbpi Please be aware of a bug on Solaris. If one of the threads in a multi-threaded process attempts to read from a closed pipe the entire process locks-up (the reading LWP is looping on the read in the kernel) and no signal delivery happens. Only kill -9 pid will terminate the program. Sun are aware of the bug and are working on it.... Instance Variables: infd <SmallInteger> read-side pipe file descriptor outfd <SmallInteger> write-side pipe file descriptor generator <Process> generator process (useful for debugging) writer <Process> writer process (useful for debugging) readers <Array of Process> reader process(es) (useful for debugging) results <SharedQueue> queue to buffer results between generator & writer running <Boolean> flag tested by writer and readers to see if they should continue Class Variables: TranscriptProtect <Semaphore> mutex for readers writing to Transcript
</comment> <class> <name>FSNBPI</name> <environment>Examples</environment> <super>Examples.NonBlockingPipeInterface</super> <private>false</private> none <imports> Examples.FSNBPIDictionary.* </imports> <category>ExternalInterface-THAPI Example</category> <attributes> <includeFiles></includeFiles> <includeDirectories></includeDirectories> <libraryFiles></libraryFiles> <libraryDirectories></libraryDirectories> <beVirtual>false</beVirtual> <optimizationLevel>full</optimizationLevel> <package>THAPIExample</package> </attributes> </class> <comment> Examples.FSNBPI
This class does exactly what its superclass NonBlockingPipeInterface does, except that instead of using buffers allocated on the C heap it uses normal Smalltalk objects in FixedSpace. Some example doits: "10 readers. ctrl-y to terminate" FSNBPI new readers: 10 "10 readers in the background. Inspect the nbpi and send it terminate when you get bored." | nbpi | nbpi := FSNBPI new. [nbpi readers: 10] forkAt: Processor activePriority - 2. nbpi
</comment> <name>TranscriptProtect</name> <environment>Examples.NonBlockingPipeInterface</environment> <private>false</private> <constant>false</constant> <category>local globals</category> <attributes> <package>THAPIExample</package> </attributes> <name>O_BINARY</name> <environment>Examples.NonBlockingPipeInterface</environment> <private>false</private> <constant>false</constant> <category>local globals</category> <attributes> <package>THAPIExample</package> </attributes> <methods> Examples.NonBlockingPipeInterface <category>initialize-release</category>
openPipes "Create the pipe. Note that pipe in the MSVC run-time library is different from the standard unix pipe." | fds | fds := CIntegerType int gcMalloc: 2. (OSHandle currentOS == #win32 ifTrue: [self pipe: fds ofSize: 1024 mode: O_BINARY] ifFalse: [self pipe: fds]) < 0 ifTrue: [self error: 'pipe open failed.']. infd := fds at: 0. outfd := fds at: 1.
terminate "Terminate all the relevant processes and close the pipe." running ifFalse: [^self]. running := false. generator == Processor activeProcess ifFalse: [generator terminate]. "Write sufficient data to the pipe so that all readers will get data and hence by checking running will stop." readers size * 2 timesRepeat: [results nextPut: 'so long!']. "Delay until the results have been written by the writer and then kill the writer. Yield doesn't work if terminate has a higher priority than the writer." [results isEmpty] whileFalse: [(Delay forMilliseconds: 20) wait]. writer terminate. self close: infd; close: outfd
</methods> <methods> Examples.NonBlockingPipeInterface <category>server</category>
reader: id "Read from the pipe as long as running is true. Print what ever's read from the pipe to the Transcript tagged with the reader id." | buffer count | "Allocate a buffer on the C heap to receiver characters from the read call." buffer := CIntegerType char gcMalloc: 1025. "Continue until running is false." [running] whileTrue: ["Make a blocking read call on its own thread." count := self read: infd with: buffer with: 1024. "Write results to the transcript. Use a Semaphore to ensure other reader processes don't collide when accessing the un-thread-safe Transcript." TranscriptProtect critical: [Transcript cr; print: id; tab. (count >= 1024 or: [count < 0]) ifTrue: [Transcript nextPutAll: 'READ RETURNED '; print: count] ifFalse: [buffer at: count put: 0. Transcript nextPutAll: buffer copyCStringFromHeap]. Transcript endEntry]]
startReader: id "Place an exception handler around the reader: method to catch ou-of-thread exceptions." self externalAccessFailedSignal handle: [:ex | ex localErrorString == #'out of threads' ifTrue: [TranscriptProtect critical: [Transcript cr; print: id; nextPutAll: ' CANNOT PROCEED. OUT OF THREADS'; endEntry]. Processor activeProcess terminate] ifFalse: [ex reject]] do: [self reader: id]
writer "Loop writing strings from the results queue to the pipe." | result buffer writeCount | "Allocate a buffer on the C heap to pass characters to the write call." buffer := CIntegerType char gcMalloc: 1024. [true] whileTrue: ["Get the next result from the results shared queue. This process will wait until one is available. Convert the result to a ByteArray since the type of the buffer is char (an integer)." result := results next asByteArray. "Copy the characters from the result into the buffer." buffer copyAt: 0 from: result size: result size startingAt: 1. "Write the buffer's contents to the pipe." writeCount := self write: outfd with: buffer with: result size. "Check the write operation succeeded." writeCount ~= result size ifTrue: [TranscriptProtect critical: [Transcript cr; nextPutAll: 'WRITE RETURNED '; print: writeCount; nextPutAll: ' EXPECTED '; print: result size; nextPut: $!; endEntry]]]
</methods> <methods> Examples.NonBlockingPipeInterface <category>public access</category>
readers: n "Run the example with n reader processes." results := SharedQueue new. self openPipes. generator := Processor activeProcess. "running flag is polled by other processes to make them terminate when we're done." running := true. "Fork a writer process to write data to the pipe. Its priority is higher than the generator to ensure writes happen promptly. It needs to reserve its own thread because there may be more readers than available threads, and if the writer can't get a thread because they have been used up by the readers the example will deadlock, since the readers will get no data unless some is written." writer := [self writer] forkAt: generator priority + 1. writer attachToThread. "Fork n readers at a higher priority so that results will get read and displayed." readers := (1 to: n) collect: [:i| [Processor yield. self startReader: i] forkAt: generator priority + 1]. "Generate some data. Use a benchmark the author is excessively fond of. But anything would do." [| i r t s nfibs | s := String new writeStream. i := 0. [t := Time millisecondsToRun: [r := i nfib]. s reset. nfibs := Number errorSignal handle: [:ex| '??'] do: [(r * 1000.0 / t) rounded]. s nextPutAll: 'nfib '; print: i; nextPutAll: ' = '; print: r; tab; tab; nextPutAll: 'nfibs '; print: nfibs; nextPutAll: ' ('; print: t / 1000.0; nextPutAll: ' seconds)'. "Put datum in results shared queue for the writer to consume." results nextPut: s contents. "Increase the value from which we compute nfib, limiting it to a value that takes 30 seconds or less to compute." i := t > 30000 ifTrue: [0] ifFalse: [i + 1]] repeat] ensure: [self terminate]
</methods> <methods> Examples.NonBlockingPipeInterface class <category>class initialization</category>
initialize "Initialize the mutex used to serialise access to the Transcript, and the flags used to open the pipe." TranscriptProtect := Semaphore forMutualExclusion. O_BINARY := 16r8000. "from msvc20\include\fcntl.h" "self initialize"
</methods> <methods> Examples.NonBlockingPipeInterface class <category>private</category>
libraryFilesSearchSignals "Answer a SignalCollection used to handle exceptions raised when scanning for library files. The signals answered by this method will result in those signals being ignored by the library search machinery. Clients should not answer signals they wish to receive." ^ExternalLibraryHolder libraryNotFoundSignal
</methods> <methods> Examples.FSNBPI <category>server</category>
reader: id "Read from the pipe as long as running is true. Print what ever's read from the pipe to the Transcript tagged with the reader id." | buffer count | "Use a normal Smalltalk string, but allocate it in FixedSpace." buffer := String defaultPlatformClass newInFixedSpace: 1024. [running] whileTrue: ["Make a blocking read call on its own thread." count := self read: infd with: buffer with: buffer size. "Write results to the transcript. Use a Semaphore to ensure other reader processes don't collide when accessing the un-thread-safe Transcript." TranscriptProtect critical: [Transcript cr; print: id; tab. (count > buffer size or: [count < 0]) ifTrue: [Transcript nextPutAll: 'READ RETURNED '; print: count] ifFalse: [1 to: count do: [:i| Transcript nextPut: (buffer at: i)]]. Transcript endEntry]]
writer "Loop writing strings from the results queue to the pipe." | result writeCount | [true] whileTrue: ["Get the next result from the results shared queue. This process will wait until one is available." result := results next. "Write the contents of the string to the pipe." writeCount := self write: outfd with: result with: result size. "Check the write operation succeeded." writeCount ~= result size ifTrue: [TranscriptProtect critical: [Transcript cr; nextPutAll: 'WRITE RETURNED '; print: writeCount; nextPutAll: ' EXPECTED '; print: result size; nextPut: $!; endEntry]]]
</methods> <methods> Examples.NonBlockingPipeInterface <category>procedures</category>
access: name with: mode "Unix access creation function." <C: unsigned long __syscall _threaded access(char *, int)> ^self externalAccessFailedWith: _errorCode "self basicNew access: 'e252.cha' with: 0" "self basicNew access: 'notOnYourNellie' with: 0"
close: fd "Close the file descriptor fd" <C: unsigned long __syscall close(int fd)> ^self externalAccessFailedWith: _errorCode
pipe: arg "Unix pipe creation function." <C: unsigned long __syscall pipe(int [])> ^self externalAccessFailedWith: _errorCode
pipe: arg ofSize: size mode: textMode "NT pipe creation function." <C: unsigned long __syscall pipe(int arg[], unsigned int size, int textMode)> ^self externalAccessFailedWith: _errorCode
read: fd with: buffer with: size "Invoke the read system call on its own thread and hence avoid blocking the ObjectEngine." <C: unsigned long __syscall _threaded read(int fd, _oopref *buffer, unsigned long size)> ^self externalAccessFailedWith: _errorCode
write: fd with: buffer with: size "Invoke the read system call on its own thread and hence avoid blocking the ObjectEngine." <C: unsigned long __syscall _threaded write(int fd, _oopref *buffer, unsigned long size)> ^self externalAccessFailedWith: _errorCode
</methods> <initialize> Examples.NonBlockingPipeInterface </initialize> "Imported Classes:" self error: 'Attempting to file-in parcel imports. Choose terminate or close' <class> <name>ExternalInterface</name> <environment>External</environment> <super>Core.Object</super> <private>false</private> none includeFiles includeDirectories libraryFiles libraryDirectories virtual optimizationLevel <imports> private Kernel.OpcodePool.* </imports> <category>External-Interface</category> <attributes> <package>External-Interface</package> </attributes> </class>