Just read an old blogpost of Nicolas Cellier's today that suggested something like this, and thought it sounded interesting. It's use would be for parallell operations that involve IO delays (the example in the post was MCRepositoryGroup >> #includesVersionNamed:) An important assumption would be that the blocks are free of side-effects, so evaluation completion/order is not relied upon for further correct operation. The goal would be to be able to write the method in question like this: includesVersionNamed: aString " check for existing version name in parallel over all repositories " ^self repositories threadedAnySatisfy: [:repository | repository includesVersionNamed: aString ] while preserving the important property of anySatisfy: that is returns as soon as an element evaluates true. Here's a sample implementation: Collection >> #threadedAnySatisfy: aBlock "Evaluate aBlock with the elements of the receiver. If aBlock returns true for any element return true. Otherwise return false." | threads results processed | threads := OrderedCollection new: self size. results := AtomicSharedQueue new. processed := 0. self do: [ :each | threads add: ([ results nextPut: (aBlock value: each) ] newProcess priority: Processor activeProcess priority - 1) ]. threads do: #resume. [ processed = self size ] whileFalse: [ results next ifTrue: [ threads do: #terminate. ^ true ]. processed := processed + 1 ]. ^ false Test: [test := #(2000 1000 ). test threadedAnySatisfy: [:each | (Delay forMilliseconds: each) wait. each = 1000]] timeToRun 1000 test anySatisfy: [:each | (Delay forMilliseconds: each) wait. each = 1000]] timeToRun 3000 If one considers referencing AtomicSharedQueue in a Collection method bad form, one could also use Arrays + explicit Semaphore, along the lines of: threadedAnySatisfy: aBlock "Evaluate aBlock with the elements of the receiver. If aBlock returns true for any element return true. Otherwise return false." | threads results processed semaphore| threads := Array new: self size. results := Array new: self size withAll: false. processed := 0. self inject: 1 into: [:index :each | threads at: index put: ([ results at: index put: (aBlock value: each) ] newProcess priority: Processor activeProcess priority - 1). index + 1 ]. threads do: #resume. [ processed = self size ] whileFalse: [ semaphore wait. (results anySatisfy: #yourself) ifTrue: [ threads do: #terminate. ^ true ]. processed := processed + 1 ]. ^ false The cost of that is needing an anySatisfy: per signal received. (on an array of booleans, so the cost isn't large, but still). IMHO, the better option would be the version using a shared queue, but putting it in a non-core package, say, Collections - Parallell. Anyone else think it'd be a useful addition? Cheers, Henry
Hi Henry, On Thu, Sep 25, 2014 at 8:25 AM, Henrik Johansen < henrik.s.johansen@veloxit.no> wrote:
Just read an old blogpost of Nicolas Cellier's today that suggested something like this, and thought it sounded interesting. It's use would be for parallell operations that involve IO delays (the example in the post was MCRepositoryGroup >> #includesVersionNamed:) An important assumption would be that the blocks are free of side-effects, so evaluation completion/order is not relied upon for further correct operation.
The goal would be to be able to write the method in question like this:
includesVersionNamed: aString " check for existing version name in parallel over all repositories "
^self repositories threadedAnySatisfy: [:repository | repository includesVersionNamed: aString ]
while preserving the important property of anySatisfy: that is returns as soon as an element evaluates true.
Here's a sample implementation:
Collection >> #threadedAnySatisfy: aBlock "Evaluate aBlock with the elements of the receiver. If aBlock returns true for any element return true. Otherwise return false."
| threads results processed | threads := OrderedCollection new: self size. results := AtomicSharedQueue new. processed := 0. self do: [ :each | threads add: ([ results nextPut: (aBlock value: each) ] newProcess priority: Processor activeProcess priority - 1) ]. threads do: #resume. [ processed = self size ] whileFalse: [ results next ifTrue: [ threads do: #terminate. ^ true ]. processed := processed + 1 ]. ^ false
Test:
[test := #(2000 1000 ). test threadedAnySatisfy: [:each | (Delay forMilliseconds: each) wait. each = 1000]] timeToRun 1000 test anySatisfy: [:each | (Delay forMilliseconds: each) wait. each = 1000]] timeToRun 3000
Cool. But please call it parallelAnySatisfy: or anySatisfyInParallel:. Are you sure there are no parallel hazards in the includesVersionNamed code? If one considers referencing AtomicSharedQueue in a Collection method bad
form, one could also use Arrays + explicit Semaphore, along the lines of:
threadedAnySatisfy: aBlock "Evaluate aBlock with the elements of the receiver. If aBlock returns true for any element return true. Otherwise return false."
| threads results processed semaphore| threads := Array new: self size. results := Array new: self size withAll: false. processed := 0. self inject: 1 into: [:index :each | threads at: index put: ([ results at: index put: (aBlock value: each) ] newProcess priority: Processor activeProcess priority - 1). index + 1 ]. threads do: #resume. [ processed = self size ] whileFalse: [ semaphore wait. (results anySatisfy: #yourself) ifTrue: [ threads do: #terminate. ^ true ]. processed := processed + 1 ]. ^ false
The cost of that is needing an anySatisfy: per signal received. (on an array of booleans, so the cost isn't large, but still). IMHO, the better option would be the version using a shared queue, but putting it in a non-core package, say, Collections - Parallell.
Anyone else think it'd be a useful addition?
Cheers, Henry
-- best, Eliot
On 25 Sep 2014, at 6:23 , Eliot Miranda <eliot.miranda@gmail.com> wrote:
Hi Henry,
On Thu, Sep 25, 2014 at 8:25 AM, Henrik Johansen <henrik.s.johansen@veloxit.no> wrote: Just read an old blogpost of Nicolas Cellier's today that suggested something like this, and thought it sounded interesting. It's use would be for parallell operations that involve IO delays (the example in the post was MCRepositoryGroup >> #includesVersionNamed:) An important assumption would be that the blocks are free of side-effects, so evaluation completion/order is not relied upon for further correct operation.
The goal would be to be able to write the method in question like this:
includesVersionNamed: aString " check for existing version name in parallel over all repositories "
^self repositories threadedAnySatisfy: [:repository | repository includesVersionNamed: aString ]
while preserving the important property of anySatisfy: that is returns as soon as an element evaluates true.
Here's a sample implementation:
Collection >> #threadedAnySatisfy: aBlock Cool. But please call it parallelAnySatisfy: or anySatisfyInParallel:.
I agree, parallelAnySatisfy: is a better name. (In Norwegian it's written parallell , I guess I was subconsciously trying to prevent a high rate of spelling errors, at the cost of an inferior name ;) )
Are you sure there are no parallel hazards in the includesVersionNamed code?
Good question, I would imagine it might do some caching that could lead to inconsistent state if arbitrarily terminated as in the example implementation, but I haven't checked closely*... Cheers, Henry * I'm don't think it's used, but this might be a candidate: cacheAllFileNamesDuring: aBlock cacheFileNames == true ifTrue: [ ^ aBlock value ]. allFileNames := nil. cacheFileNames := true. <-- If this is a suspension point, if terminated, the file names would be static from the next call to allFileNames until the next time cacheAllFiles... is used --> ^ aBlock ensure: [ allFileNames := nil. cacheFileNames := false] If the need is guarding against arbitrary termination, not just errors in the block itself, I think it would have to be: ^[allFileNames := nil. cacheFileNames := true. aBlock value ] ensure: [allFileNames := nil. cacheFileNames := false] Of the actually called methods from includesVersionNamed:, I stopped checking after stumbling into, and spending too much time wondering why the code did what it did: MCFileBasedRepository >> #readableFileNames | all cached new emptyFilenamelength | "<hyphenated-package-name>.<dotted.branch.tag>-<initials>.<count>.mcz" emptyFilenamelength := 'P-i.c.mcz' size. all := self allFileNames. "from repository" all := all reject: [ :each | each size < emptyFilenamelength]. "first stupid way to filter first level broken files. Ideally we should remove any files not following the naming pattern: PackageName-author.number[(branch)].mcz" cached := self cachedFileNames. "in memory" new := all difference: cached. ^ (cached asArray, new) select: [:ea | self canReadFileNamed: ea] Removing cached entries from all entries, then iterating on the remaining + cached? That's an aweful lot of nonsense for the clearly essential capability to say you are able to read files that have recently been deleted from the file system...
We got some problems with the cache logic recently. To the point we would like to add a UUID to the file name. It needs a real look. Stef
* I'm don't think it's used, but this might be a candidate: cacheAllFileNamesDuring: aBlock cacheFileNames == true ifTrue: [ ^ aBlock value ]. allFileNames := nil. cacheFileNames := true. <-- If this is a suspension point, if terminated, the file names would be static from the next call to allFileNames until the next time cacheAllFiles... is used --> ^ aBlock ensure: [ allFileNames := nil. cacheFileNames := false]
If the need is guarding against arbitrary termination, not just errors in the block itself, I think it would have to be:
^[allFileNames := nil. cacheFileNames := true. aBlock value ] ensure: [allFileNames := nil. cacheFileNames := false]
Of the actually called methods from includesVersionNamed:, I stopped checking after stumbling into, and spending too much time wondering why the code did what it did:
MCFileBasedRepository >> #readableFileNames | all cached new emptyFilenamelength | "<hyphenated-package-name>.<dotted.branch.tag>-<initials>.<count>.mcz" emptyFilenamelength := 'P-i.c.mcz' size. all := self allFileNames."from repository" all := all reject: [ :each | each size < emptyFilenamelength]. "first stupid way to filter first level broken files. Ideally we should remove any files not following the naming pattern: PackageName-author.number[(branch)].mcz" cached := self cachedFileNames."in memory" new := all difference: cached. ^ (cached asArray, new) select: [:ea | self canReadFileNamed: ea]
Removing cached entries from all entries, then iterating on the remaining + cached? That's an aweful lot of nonsense for the clearly essential capability to say you are able to read files that have recently been deleted from the file system...
IMHO, the better option would be the version using a shared queue, but putting it in a non-core package, say, Collections - Parallell. Anyone else think it'd be a useful addition? Cheers, Henry Yes! Now it would be nice to be able to get more information about methods (with Spur object format probably) to know if a method is performing side effect or not. Stef
participants (3)
-
Eliot Miranda -
Henrik Johansen -
stepharo