Revision: 3451 http://trac.macosforge.org/projects/ruby/changeset/3451 Author: ernest.prabhakar@gmail.com Date: 2010-02-08 15:04:57 -0800 (Mon, 08 Feb 2010) Log Message: ----------- Borked implementation of mapreduce Modified Paths: -------------- MacRuby/trunk/lib/dispatch/dispatch.rb MacRuby/trunk/lib/dispatch/enumerable.rb MacRuby/trunk/spec/macruby/library/dispatch/enumerable_spec.rb Modified: MacRuby/trunk/lib/dispatch/dispatch.rb =================================================================== --- MacRuby/trunk/lib/dispatch/dispatch.rb 2010-02-08 23:04:45 UTC (rev 3450) +++ MacRuby/trunk/lib/dispatch/dispatch.rb 2010-02-08 23:04:57 UTC (rev 3451) @@ -5,9 +5,10 @@ # Returns a unique label based on the ancestor chain and ID of +obj+ # plus the current time def label_for(obj) - label = obj.class.ancestors.uniq.reverse.join("_").downcase + ancestors = obj.class.ancestors.map {|a| a.to_s} + label = ancestors.uniq.reverse.join("_").downcase now = Time.now.to_f.to_s.gsub(".","_") - "#{label}_%x_%s" % [obj.object_id, now] + "#{label}__%x__%s" % [obj.object_id, now] end # Returns a new serial queue with a unique label based on +obj+ Modified: MacRuby/trunk/lib/dispatch/enumerable.rb =================================================================== --- MacRuby/trunk/lib/dispatch/enumerable.rb 2010-02-08 23:04:45 UTC (rev 3450) +++ MacRuby/trunk/lib/dispatch/enumerable.rb 2010-02-08 23:04:57 UTC (rev 3451) @@ -30,17 +30,23 @@ end # Parallel +collect+ plus +inject+ - # Accumulates in +initial+ via +op+ (default = '<<') - def p_mapreduce(initial, op=:<<, &block) - raise ArgumentError if not initial.respond_to? :op + # Accumulates from +initial+ via +op+ (default = '+') + def p_mapreduce(initial, op=:+, &block) + raise ArgumentError if not initial.respond_to? op # Since exceptions from a Dispatch block act funky - p_result = Dispatch.wrap(initial) - self.p_each_with_index do |obj, i| - val = block.call(obj) - p_result.send(op, val) + q = Dispatch.queue_for(initial) + ivar = "@#{Dispatch.label_for(initial)}" + p ivar + self.instance_variable_set(ivar, initial) #is creating an ivar thread-safe? + self.p_each do |obj| + val = block.call(obj) + q.async do + sum = self.instance_variable_get(ivar).send(op, val) + self.instance_variable_set(ivar, sum) + end end - p_result - ._done_ + q.sync {} + self.send(:remove_instance_variable, ivar) end Modified: MacRuby/trunk/spec/macruby/library/dispatch/enumerable_spec.rb =================================================================== --- MacRuby/trunk/spec/macruby/library/dispatch/enumerable_spec.rb 2010-02-08 23:04:45 UTC (rev 3450) +++ MacRuby/trunk/spec/macruby/library/dispatch/enumerable_spec.rb 2010-02-08 23:04:57 UTC (rev 3451) @@ -18,10 +18,10 @@ @sum2 = 0 @q = Dispatch.queue_for(@sum2) @ary.p_each {|v| temp = v*v; @q.sync {@sum2 += temp} } - @sum1.should == @sum2 + @sum2.should == @sum1 end - it "executes concurrently" do + it "should execute concurrently" do true.should == true end end @@ -37,10 +37,10 @@ @sum2 = 0 @q = Dispatch.queue_for(@sum2) @ary.p_each_with_index {|v, i| temp = v**i; @q.sync {@sum2 += temp} } - @sum1.should == @sum2 + @sum2.should == @sum1 end - it "executes concurrently" do + it "should execute concurrently" do true.should == true end end @@ -52,10 +52,10 @@ it "should behave like map" do map1 = @ary.map {|v| v*v} map2 = @ary.p_map {|v| v*v} - map1.should == map2 + map2.should == map1 end - it "executes concurrently" do + it "should execute concurrently" do true.should == true end end @@ -71,20 +71,22 @@ map1.should == map2.sort end - it "should use any result that takes :<< " do - map1 = @ary.map {|v| "%x" % 10+v} - map2 = @ary.p_mapreduce("") {|v| "%x" % 10+v} + it "should accumulate any object that takes :<< " do + map1 = @ary.map {|v| "%x" % (10+v)} + map2 = @ary.p_mapreduce("") {|v| "%x" % (10+v)} map1.each do |s| map2.index(s).should_not == nil end end it "should allow custom accumulator methods" do - map2 = @ary.p_mapreduce(0, :|) {|v| v**2} - map2.should == (2 | 4 | 8) + map1 = @ary.map {|v| v**2} + sum1 = map1.inject(0) {|s,v| s | v} + sum2 = @ary.p_mapreduce(0, :|) {|v| v**2} + sum2.should == sum1 end - it "executes concurrently" do + it "should execute concurrently" do true.should == true end end