Bug in dataflow?

[schneems]

I tried using dataflow. Kept getting an error. Re-wrote to only use a future, and it worked fine. Was getting this error:

argumentError: Not all dependencies are IVars
/Users/richardschneeman/.gem/ruby/2.3.0/gems/concurrent-ruby-1.0.0/lib/concurrent/dataflow.rb:61:in `call_dataflow'
/Users/richardschneeman/.gem/ruby/2.3.0/gems/concurrent-ruby-1.0.0/lib/concurrent/dataflow.rb:40:in `dataflow_with'
/Users/richardschneeman/.gem/ruby/2.3.0/gems/concurrent-ruby-1.0.0/lib/concurrent/dataflow.rb:35:in `dataflow'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/manifest.rb:203:in `block in compile'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/manifest.rb:131:in `block (2 levels) in find'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/base.rb:76:in `find_all_linked_assets'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/manifest.rb:130:in `block in find'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/manifest.rb:129:in `each'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/manifest.rb:129:in `find'
/Users/richardschneeman/Documents/projects/sprockets/lib/sprockets/manifest.rb:164:in `compile'
/Users/richardschneeman/.gem/ruby/2.3.0/bundler/gems/sprockets-rails-ad4a43bd1bb1/lib/sprockets/rails/task.rb:68:in `block (3 levels) in define'
/Users/richardschneeman/Documents/projects/sprockets/lib/rake/sprocketstask.rb:147:in `with_logger'
/Users/richardschneeman/.gem/ruby/2.3.0/bundler/gems/sprockets-rails-ad4a43bd1bb1/lib/sprockets/rails/task.rb:67:in `block (2 levels) in define'
Tasks: TOP => assets:precompile

Here is the diff to get the program working

         if File.exist?(target)
           logger.debug "Skipping #{target}, already exists"
-          file_is_written = Concurrent.dataflow { }
         else
           logger.info "Writing #{target}"
-          file_is_written = Concurrent.dataflow { asset.write_to target }
+          file_is_written = Concurrent::Future.execute { asset.write_to target }
           concurrent_writers << file_is_written
         end
         filenames << asset.filename
@@ -198,7 +197,10 @@ module Sprockets
           logger.debug "Skipping #{target}.gz, already exists"
         else
           logger.info "Writing #{target}.gz"
-          concurrent_compressors << Concurrent.dataflow(file_is_written) { gzip.compress(target) }
+          concurrent_compressors << Concurrent::Future.execute do
+            file_is_written.wait if file_is_written
+            gzip.compress(target)
+          end
         end

Then below I make sure all the futures have executed:

concurrent_writers.each(&:wait)
concurrent_compressors.each(&:wait)

Does anything jump out at you as being off with the dataflow implementation?

[schneems]

if you're curious the value of the dependency was false when i ran that code with #477

[schneems]

I shipped this implementation in Sprockets 3.5.0 rails/sprockets#193.

[jdantonio]

@schneems Thanks for updating your PR and for your patience in waiting for me to review it. I'm out of town for work this week and I haven't had much free time. I plan to catch up on my open source commitments this weekend.

I'm glad that you were able to get sprockets working!

[schneems]

No worries. Take your time. I wish github had an "out of office feature". When I was out for paternity leave I keep getting all these emails, and the only good way to deal with them was to reply and say I was out. Had to do that manually.

Thanks for all your work on this gem, so far it's been great to work with. Also thanks for your talk at RubyConf. I'm not blocked on this issue, but I did want to share to see if you think maybe this is a problem inside of ruby-concurrency. Take a look and let me know if you've got any questions when you get back in the flow.

[RoryO]

@schneems I had this issue when working with Dataflow. Where I think your issue lies is require 'concurrent/dataflow' instead of require 'concurrent'. I have seen a number of interesting, difficult to trace issues with the Concurrent inner classes when not requiring the base concurrent file. (e.x. doing require 'concurrent/promise'; x = Concurrent::Promise.new { true }.execute never actually advances past the pending state).

I've seen this exact one by only requiring data flow. Perhaps try doing full require 'concurrent' and see if that helps?

[chrisseaton]

I'll aim to take a look at this tomorrow.

[chrisseaton]

Sorry, I can't seem to reproduce this:

require 'concurrent'
file_is_written = Concurrent.dataflow { }
concurrent_writers = [file_is_written]
concurrent_compressors = [Concurrent.dataflow(file_is_written) { |x| p x }]
concurrent_compressors.each(&:wait)

require 'concurrent/dataflow doesn't work - but doesn't give the error message you saw.

Can you see any significant difference between my reduced test case and what you were doing? It looks like you (@schneems) already made a change to report a better error message. Did that tell you anything useful?

[schneems]

I think we can close this issue. It was a bug in my code. If you were following along with #477 I actually had bad code in there that would always raise an error. I was using that to debug, and when I added the extra Concurrent.dataflow {} to the other side of the if statement it was hiding the working code (since it was always raising an error). 🙊 🙊 🙊 Sorry for the noise, thanks for the sanity check ❤️ ❤️ ❤️

[chrisseaton]

Great - are you trying dataflow again? You can see in your code example above that it simplifies some things compared to futures.

[schneems]

I might try it in the future. For now i'm going to need to move on to other things.

In my code sample the only complex case that we have to handle when the file is already written to disk. In the future solution we check for presence of the future before calling wait on it. In the dataflow case we create an empty dataflow. Creating an empty dataflow is probably more expensive than the nil check with a future that already shipped.

It might be interesting to have a null dataflow so I could do something like

if File.exist?(target)
 logger.debug "Skipping #{target}, already exists"
 file_is_written = Concurrent::Dataflow::Null
else

So that way we don't have to spin cycles checking inputs, and executing an empty block. I'll be looking for other places around sprockets to parallelize. I think we can maybe re-implement the way we read in files, but not sure. I'll likely use dataflows in the future, but for now i've got a working solution. Thanks again for the help.

[pitr-ch]

@schneems you may be interested in http://ruby-concurrency.github.io/concurrent-ruby/Concurrent/Edge/FutureShortcuts.html a new future implementation going into c-r 2.0. It allows chaining as dataflow and there is also Concurrent.succeeded_future to create futures without blocks. For now it's in edge though.