[PATCH] Finalize the stream-to-stream protocol
Locked
[PATCH] Finalize the stream-to-stream protocol
 
|
This patch adds #nextAvailable:putAllOn: and #next:putAllOn:. These are
important because, by using #next:putAll:startingAt: on the destination stream, they can pass their internal buffers directly without making useless copies of the data. It also renames the #nextHunkPutAllOn: to #nextAvailablePutAllOn:. It also eliminates #nextHunk, which is a much worse approach considering the alternatives that are available now! Applications that need it had better use the stream-to-stream protocol or at least the block input protocol, because #nextHunk caused too many allocations -- sometimes up to 50% of the time was spent on GC. However, for now I switched the uses of #nextHunk to use #nextAvailable: instead, as it's the simplest thing to do. The beauty of all the new methods is that only three need to be overridden in Stream subclasses if they want to support block streaming. These are #nextAvailable:putAllOn:, #nextAvailable:into:startingAt: and the pre-existing #next:putAll:startingAt:. Everything else is provided by Stream and is already optimized to use these three methods. Paolo 2008-08-06 Paolo Bonzini <[hidden email]> * kernel/PosStream.st: Add #nextAvailable:putAllOn:. * kernel/FileStream.st: Add #nextAvailable:putAllOn:, and remove #nextHunkPutAllOn: and #nextHunk. * kernel/Stream.st: Add #nextAvailable:putAllOn: and #next:putAllOn:, remove #nextHunk, rename #nextHunkPutAllOn: to #nextAvailablePutAllOn: and rewrite it. * kernel/VFSZip.st: Remove #nextHunk. libgst: 2008-08-06 Paolo Bonzini <[hidden email]> * input.c: Do not use #nextHunk. packages/httpd: 2008-08-06 Paolo Bonzini <[hidden email]> * FileServer.st: Do not use #nextHunk. packages/iconv: 2008-08-06 Paolo Bonzini <[hidden email]> * Sets.st: Add #nextAvailable:putAllOn:, remove #nextHunkPutAllOn: and #nextHunk. packages/net: 2008-04-07 Paolo Bonzini <[hidden email]> * FTP.st: Do not use #nextHunk. packages/sockets: 2008-08-06 Paolo Bonzini <[hidden email]> * Buffers.st: Add #nextAvailable:putAllOn:, remove #nextHunkPutAllOn: and #nextHunk. * Sockets.st: Likewise. * tests.st: Change #nextHunkPutAllOn: to #nextAvailablePutAllOn:. packages/zlib: 2008-08-06 Paolo Bonzini <[hidden email]> * ZLibReadStream.st: Add #nextAvailable:putAllOn:, remove #nextHunkPutAllOn: and #nextHunk. diff --git a/NEWS b/NEWS index 3bbcce2..aa07f61 100644 --- a/NEWS +++ b/NEWS @@ -26,9 +26,7 @@ o CObjects can be backed with garbage-collected (as opposed to o Error backtraces include line numbers and filenames. o FileDescriptor and FileStream raise an exception if #next: cannot - return the given number of bytes. They also implement #nextAvailable: - which is similar to #nextHunk but returns at most the number of bytes - given by the argument. + return the given number of bytes. o FileDescriptor is now a subclass of Stream. @@ -68,9 +66,6 @@ o It is possible to create C call-outs that are not attached override the #link method (the existing CFunctionDescriptor class is now implemented on top of this). -o A new method #nextHunkPutAllOn: allows to copy from stream to stream - while minimizing the number of allocated objects. - o ObjectDumper now accepts normal String streams. The class ByteStream has been removed. @@ -80,6 +75,23 @@ o ObjectMemory>>#snapshot and ObjectMemory>>#snapshot: return false in snapshot. Note that this does not apply to CallinProcesses, since those are stopped in saved images (will this be true in 3.1?). +o Streams have a set of new methods that allow to eliminate useless copies + when moving data from stream to stream, as well as to eliminate useless + creation of collections to store the data read from a stream. Besides + the standard methods #next, #nextPut: and #atEnd, two more methods can be + implemented by Stream subclasses in order to achieve better performance + on block transfers, namely #nextAvailable:into:startingAt:, + #nextAvailable:putAllOn: and #next:putAll:startingAt:. + + Another set of methods is placed in Stream; they all use the above + three methods and you should not need to override them. These are + #next:into:startingAt:, #next:putAllOn:, #nextAvailable:, + #nextAvailable:putAllOn:, #nextAvailablePutAllOn:. + + In addition, #nextHunk was removed. Applications can use #nextAvailable: + or, better, should be rewritten to use new stream-to-stream protocol + such as #nextAvailablePutAllOn:. + o The VFS subsystem was rewritten. Virtual filesystems are now accessible via special methods on File (such as File>>#zip, for example "(File name: 'abc.zip') zip") and not anymore with diff --git a/kernel/FileStream.st b/kernel/FileStream.st index c8d3eb4..1aa5635 100644 --- a/kernel/FileStream.st +++ b/kernel/FileStream.st @@ -477,6 +477,23 @@ file object, such as /dev/rmt0 on UNIX or MTA0: on VMS).'> ^resultStream contents ] + nextAvailable: anInteger putAllOn: aStream [ + "Copy up to anInteger bytes from the stream into + aStream. Return the number of bytes read." + + <category: 'buffering'> + | last n | + writePtr notNil ifTrue: [self flush]. + ptr > endPtr ifTrue: [self fill]. + + "Fetch data from the buffer, without doing more than one I/O operation." + last := endPtr min: ptr + anInteger - 1. + n := last - ptr + 1. + aStream next: n putAll: collection startingAt: ptr. + ptr := ptr + n. + ^n + ] + nextAvailable: anInteger into: aCollection startingAt: pos [ "Read up to anInteger bytes from the stream and store them into aCollection. Return the number of bytes read." @@ -611,34 +628,6 @@ file object, such as /dev/rmt0 on UNIX or MTA0: on VMS).'> self flush ] - nextHunkPutAllOn: aStream [ - "Copy the next buffers worth of stuff from the receiver to - aStream. For n consecutive calls to this method, we do - n - 1 or n actual input operation." - - <category: 'buffering'> - writePtr notNil ifTrue: [self flush]. - (ptr > endPtr or: [endPtr < collection size]) ifTrue: [self fill]. - ptr > endPtr ifTrue: [^self pastEnd]. - aStream next: endPtr - ptr + 1 putAll: collection startingAt: ptr. - ptr := endPtr + 1. - ] - - nextHunk [ - "Answer the next buffers worth of stuff in the Stream represented - by the receiver. For n consecutive calls to this method, we do - n - 1 or n actual input operation." - - <category: 'buffering'> - | result | - writePtr notNil ifTrue: [self flush]. - (ptr > endPtr or: [endPtr < collection size]) ifTrue: [self fill]. - ptr > endPtr ifTrue: [^self pastEnd]. - result := collection copyFrom: ptr to: endPtr. - ptr := endPtr + 1. - ^result - ] - next: n bufferAll: aCollection startingAt: pos [ "Private - Assuming that the buffer has space for n characters, store n characters of aCollection in the buffer, starting from the pos-th." diff --git a/kernel/PosStream.st b/kernel/PosStream.st index 1788d69..2eb3283 100644 --- a/kernel/PosStream.st +++ b/kernel/PosStream.st @@ -88,10 +88,25 @@ or ReadWriteStream instead of me to create and use streams.'> aStream next: endPtr putAll: collection startingAt: 1. ] + nextAvailable: anInteger putAllOn: aStream [ + "Copy up to anInteger objects from the receiver into + aStream, stopping if no more data is available." + + <category: 'accessing-reading'> + | n | + n := anInteger min: endPtr - ptr + 1. + aStream + next: n + putAll: collection + startingAt: ptr. + ptr := ptr + n. + ^n + ] + nextAvailable: anInteger into: aCollection startingAt: pos [ "Place up to anInteger objects from the receiver into - aCollection, starting from position pos and stopping if - no more data is available." + aCollection, starting from position pos in the collection + and stopping if no more data is available." <category: 'accessing-reading'> | n | diff --git a/kernel/Stream.st b/kernel/Stream.st index 542c9cd..7fa7749 100644 --- a/kernel/Stream.st +++ b/kernel/Stream.st @@ -67,6 +67,24 @@ provide for writing collections sequentially.'> ^answer ] + next: anInteger putAllOn: aStream [ + "Read up to anInteger bytes from the stream and store them + into aStream. Return the number of bytes that were read, raising + an exception if we could not read the full amount of data." + + <category: 'buffering'> + | read | + read := 0. + [ read = anInteger ] whileFalse: [ + self atEnd ifTrue: [ + ^SystemExceptions.NotEnoughElements signalOn: anInteger - read]. + read := read + (self + nextAvailable: anInteger - read + putAllOn: aStream) + ]. + ^read + ] + next: anInteger into: answer startingAt: pos [ "Read up to anInteger bytes from the stream and store them into answer. Return the number of bytes that were read, raising @@ -86,6 +104,24 @@ provide for writing collections sequentially.'> ^answer ] + nextAvailable: anInteger putAllOn: aStream [ + "Copy up to anInteger objects in the receiver to aStream. Besides + stopping if the end of the stream is reached, this may return + less than this number of bytes for various reasons. For example, + on files and sockets this operation could be non-blocking, + or could do at most one I/O operation." + + <category: 'accessing-reading'> + | n coll | + n := anInteger min: 1024. + n := self + nextAvailable: n + into: (coll := self species new: n) + startingAt: 1. + aStream next: n putAll: coll startingAt: 1. + ^n + ] + nextAvailable: anInteger [ "Return up to anInteger objects in the receiver. Besides stopping if the end of the stream is reached, this may return less than this @@ -266,7 +302,7 @@ provide for writing collections sequentially.'> "Write all the objects in the receiver to aStream" [self atEnd] whileFalse: - [self nextHunkPutAllOn: aStream]. + [self nextAvailablePutAllOn: aStream]. ] next: anInteger put: anObject [ @@ -535,32 +571,14 @@ provide for writing collections sequentially.'> repeat ] - nextHunkPutAllOn: aStream [ + nextAvailablePutAllOn: aStream [ "Copy to aStream a more-or-less arbitrary amount of data. When used on files, this does at most one I/O operation. For other kinds of - stream, the definition may vary. This method is used by the VM - when loading data from a Smalltalk stream, and by various kind - of Stream decorators supplied with GNU Smalltalk (including - zlib streams). Subclasses that implement nextHunk can implement - this method to avoid useless work." - - | coll | - coll := self nextHunk. - aStream - next: coll size - putAll: coll - startingAt: 1 - ] - - nextHunk [ - "Answer a more-or-less arbitrary amount of data. When used on files, this - does at most one I/O operation. For other kinds of stream, the definition - may vary. This method is used by the VM when loading data from a Smalltalk - stream, and by various kind of Stream decorators supplied with GNU - Smalltalk (including zlib streams)." + stream, the definition may vary. This method is used to do + stream-to-stream copies." - <category: 'positioning'> - ^self nextAvailable: 1024 + <category: 'streaming protocol'> + self nextAvailable: 16rFFFFFFF putAllOn: aStream. ] prefixTableFor: aCollection [ diff --git a/kernel/VFSZip.st b/kernel/VFSZip.st index 71e24e8..3ccac61 100644 --- a/kernel/VFSZip.st +++ b/kernel/VFSZip.st @@ -236,15 +236,18 @@ Stream subclass: LimitedStream [ ^true ] - next [ + nextAvailable: n into: aCollection startingAt: pos [ <category: 'stream operations'> - self atEnd ifTrue: [^self pastEnd]. - ^stream next + ^stream + nextAvailable: (n min: limit - stream position) + into: aCollection + startingAt: pos ] - nextHunk [ + next [ <category: 'stream operations'> - ^stream next: (1024 min: limit - stream position) + self atEnd ifTrue: [^self pastEnd]. + ^stream next ] peek [ diff --git a/libgst/input.c b/libgst/input.c index cf400f9..734a116 100644 --- a/libgst/input.c +++ b/libgst/input.c @@ -406,7 +406,7 @@ my_getc (input_stream stream) if (stream->st_oop.ptr == stream->st_oop.end) { char *buf; - _gst_msg_sendf(&buf, "%s %o nextHunk", stream->st_oop.oop); + _gst_msg_sendf(&buf, "%s %o nextAvailable: %i", stream->st_oop.oop, 1024); if (!buf || !*buf) return EOF; diff --git a/packages/httpd/FileServer.st b/packages/httpd/FileServer.st index e1b8977..062802f 100644 --- a/packages/httpd/FileServer.st +++ b/packages/httpd/FileServer.st @@ -280,7 +280,7 @@ to upload a file.'> <category: 'multipart'> request stream isPeerAlive ifFalse: [^'']. - ^request stream nextHunk + ^request stream nextAvailable: 1024 ] localFileFor: remoteName [ diff --git a/packages/iconv/Sets.st b/packages/iconv/Sets.st index 6045b27..30b28a7 100644 --- a/packages/iconv/Sets.st +++ b/packages/iconv/Sets.st @@ -966,42 +966,38 @@ Iconv is skipped altogether and only Smalltalk converters are used.'> ^answer ] - nextAvailable: anInteger into: aCollection startingAt: pos [ - "Answer the next buffer's worth of data from the receiver." + nextAvailable: anInteger putAllOn: aStream [ + "Copy up to anInteger bytes from the next buffer's worth of data + from the receiver to aStream." <category: 'stream operation'> | n | (self atEndOfBuffer and: [self convertMore]) ifTrue: [^self pastEnd]. n := anInteger min: recodedEnd - recodedPos + 1. - aCollection - replaceFrom: pos to: pos + n - 1 - with: recodedBuffer + aStream + next: n + putAll: recodedBuffer startingAt: recodedPos. recodedPos := recodedPos + n. ^n ] - nextHunkPutAllOn: aStream [ - "Copy the next buffer's worth of data from the receiver onto - aStream." + nextAvailable: anInteger into: aCollection startingAt: pos [ + "Store up to anInteger bytes from the next buffer's worth of data + from the receiver onto aCollection." <category: 'stream operation'> + | n | (self atEndOfBuffer and: [self convertMore]) ifTrue: [^self pastEnd]. - aStream - next: recodedEnd - recodedPos + 1 - putAll: recodedBuffer + n := anInteger min: recodedEnd - recodedPos + 1. + aCollection + replaceFrom: pos to: pos + n - 1 + with: recodedBuffer startingAt: recodedPos. - - recodedPos := recodedEnd + 1. - ] - - nextHunk [ - "Answer the next buffer's worth of data from the receiver." - - <category: 'stream operation'> - ^self nextAvailable: recodedEnd - recodedPos + 1 + recodedPos := recodedPos + n. + ^n ] release [ diff --git a/packages/net/FTP.st b/packages/net/FTP.st index 8504fd0..7e51aa2 100644 --- a/packages/net/FTP.st +++ b/packages/net/FTP.st @@ -354,7 +354,7 @@ NetProtocolInterpreter subclass: FTPProtocolInterpreter [ getDataWithType: type into: aStream do: controlBlock [ <category: 'ftp protocol'> - | dataStream totalByte | + | dataStream totalByte coll | (#(#ascii #binary) includes: type) ifFalse: [^self error: 'type must be #ascii or #binary']. type == #ascii ifTrue: [self ftpTypeAscii] ifFalse: [self ftpTypeBinary]. @@ -367,7 +367,7 @@ NetProtocolInterpreter subclass: FTPProtocolInterpreter [ [[dataStream atEnd] whileFalse: [| byte | - byte := dataStream nextHunk. + byte := dataStream nextAvailable: 1024. self reporter readByte: byte size. type == #ascii ifTrue: [aStream nextPutAll: (self decode: byte)] diff --git a/packages/sockets/Buffers.st b/packages/sockets/Buffers.st index fc2e498..f83f996 100644 --- a/packages/sockets/Buffers.st +++ b/packages/sockets/Buffers.st @@ -120,24 +120,6 @@ evaluates an user defined block to try to get some more data.'> ^contents ] - nextHunkPutAllOn: aStream [ - "Copy a buffer's worth of data from the receiver to aStream, doing - at most one call to the fill block." - - <category: 'accessing-reading'> - self atEnd ifTrue: [^super pastEnd]. - aStream next: endPtr - ptr + 1 putAll: self collection startingAt: ptr. - endPtr := ptr - 1. "Empty the buffer" - ] - - nextHunk [ - "Answer a buffer's worth of data, doing at most one call - to the fill block." - - <category: 'accessing-reading'> - ^self nextAvailable: endPtr - ptr + 1 - ] - availableBytes [ "Answer how many bytes are available in the buffer." @@ -146,6 +128,15 @@ evaluates an user defined block to try to get some more data.'> ^endPtr + 1 - ptr ] + nextAvailable: anInteger putAllOn: aStream [ + "Copy the next anInteger objects from the receiver to aStream. + Return the number of items stored." + + <category: 'accessing-reading'> + self isEmpty ifTrue: [ self fill ]. + ^super nextAvailable: anInteger putAllOn: aStream + ] + nextAvailable: anInteger into: aCollection startingAt: pos [ "Place the next anInteger objects from the receiver into aCollection, starting at position pos. Return the number of items stored." diff --git a/packages/sockets/Sockets.st b/packages/sockets/Sockets.st index 11de715..ba77857 100644 --- a/packages/sockets/Sockets.st +++ b/packages/sockets/Sockets.st @@ -1207,10 +1207,9 @@ This class adds a read buffer to the basic model of AbstractSocket.'> ^self readBuffer next ] - nextAvailable: anInteger into: aCollection startingAt: pos [ - "Place up to anInteger objects from the receiver into - aCollection, starting from position pos and stopping if - no more data is available." + nextAvailable: anInteger putAllOn: aStream [ + "Copy up to anInteger objects from the receiver to + aStream, stopping if no more data is available." <category: 'accessing-reading'> | available read | @@ -1221,32 +1220,29 @@ This class adds a read buffer to the basic model of AbstractSocket.'> [ (available := self availableBytes) > 0 ] whileTrue: [ read := read + (self readBuffer nextAvailable: available - into: aCollection - startingAt: pos + read) ]. + putAllOn: aStream) ]. ^read ] - nextHunkPutAllOn: aStream [ - "Copy the next buffers worth of stuff from the receiver to aStream. - Do at most one actual input operation." - - <category: 'stream protocol'> - "Ensure that the buffer is full" + nextAvailable: anInteger into: aCollection startingAt: pos [ + "Place up to anInteger objects from the receiver into + aCollection, starting from position pos and stopping if + no more data is available." + <category: 'accessing-reading'> + | available read | readBuffer isNil ifTrue: [ ^self pastEnd ]. - self readBuffer nextHunkPutAllOn: aStream - ] - - nextHunk [ - "Answer the next buffers worth of stuff in the Stream represented - by the receiver. Do at most one actual input operation." + self ensureReadable. - "Ensure that the buffer is full" + read := 0. + [ (available := self availableBytes) > 0 ] whileTrue: [ + read := read + (self readBuffer + nextAvailable: available + into: aCollection + startingAt: pos + read) ]. - <category: 'stream protocol'> - readBuffer isNil ifTrue: [ ^self pastEnd ]. - ^self readBuffer nextHunk + ^read ] peek [ diff --git a/packages/sockets/Tests.st b/packages/sockets/Tests.st index d11f9c5..86450d9 100644 --- a/packages/sockets/Tests.st +++ b/packages/sockets/Tests.st @@ -113,7 +113,7 @@ Socket class extend [ [server nextPutAll: sendBuf. bytesSent := bytesSent + sendBuf size. [client canRead] whileTrue: - [client nextHunkPutAllOn: recvBuf. + [client nextAvailablePutAllOn: recvBuf. bytesReceived := recvBuf size]. bytesSent >= bytesToSend and: [bytesReceived = bytesSent]] whileFalse]. @@ -193,7 +193,7 @@ Socket class extend [ client := Socket remote: queue localAddress port: (self testPortFor: addressClass). [[client canRead] whileTrue: - [client nextHunkPutAllOn: recvBuf. + [client nextAvailablePutAllOn: recvBuf. bytesReceived := recvBuf size]. bytesSent >= bytesToSend and: [bytesReceived = bytesSent]] whileFalse: [Processor yield]. diff --git a/packages/zlib/ZLibReadStream.st b/packages/zlib/ZLibReadStream.st index 8684ab5..4ad72a8 100644 --- a/packages/zlib/ZLibReadStream.st +++ b/packages/zlib/ZLibReadStream.st @@ -70,26 +70,19 @@ used for communication with zlib.'> ^result ] - nextHunkPutAllOn: aStream [ - "Copy the next buffers worth of stuff from the receiver to - aStream. Do at most one actual compression/decompression - operation." + nextAvailable: anInteger putAllOn: aStream [ + "Copy up to anInteger objects from the receiver to + aStream, stopping if no more data is available." - <category: 'streaming'> - | result | - self atEnd ifTrue: [^self pastEnd]. - aStream next: endPtr - ptr putAll: outBytes startingAt: ptr + 1. - ptr := endPtr. - ^result - ] - - nextHunk [ - "Answer the next buffers worth of stuff in the Stream represented - by the receiver. Do at most one actual compression/decompression - operation." - - <category: 'streaming'> - ^self nextAvailable: endPtr - ptr + <category: 'accessing-reading'> + | n | + n := anInteger min: endPtr - ptr. + aStream + next: n + putAll: outBytes + startingAt: ptr + 1. + ptr := ptr + n. + ^n ] nextAvailable: anInteger into: aCollection startingAt: pos [ diff --git a/packages/zlib/zlibtests.st b/packages/zlib/zlibtests.st index 5309368..947805d 100644 --- a/packages/zlib/zlibtests.st +++ b/packages/zlib/zlibtests.st @@ -155,25 +155,25 @@ TestCase subclass: ZlibStreamTest [ assertFooVector: (InflateStream on: self doDeflate readStream) contents ] - testNextHunk [ - "Test accessing data with nextHunk (needed to file-in compressed data)." + testNextAvailable [ + "Test accessing data with nextAvailable (needed to file-in compressed data)." <category: 'testing'> | stream data | stream := InflateStream on: self doDeflate readStream. data := String new. - [stream atEnd] whileFalse: [data := data , stream nextHunk]. + [stream atEnd] whileFalse: [data := data , (stream nextAvailable: 1024) ]. self assertFooVector: data ] - testNextHunkPutAllOn [ - "Test accessing data with nextHunkPutAllOn." + testNextAvailablePutAllOn [ + "Test accessing data with nextAvailablePutAllOn." <category: 'testing'> | stream data | stream := InflateStream on: self doDeflate readStream. data := String new writeStream. - [stream atEnd] whileFalse: [stream nextHunkPutAllOn: data]. + [stream atEnd] whileFalse: [stream nextAvailablePutAllOn: data]. self assertFooVector: data contents ] _______________________________________________ help-smalltalk mailing list [hidden email] http://lists.gnu.org/mailman/listinfo/help-smalltalk |
«
Return to Gnu
|
1 view|%1 views
| Free forum by Nabble | Edit this page |