Concurrent mirror services #17

SGospodinov · 2017-07-07T17:48:34Z

The tests pass when mirror service is defined, in the config file, but I'm not quite sure how to test the threading itself.

georgeclaghorn · 2017-07-07T18:11:12Z

lib/active_storage/service/mirror_service.rb

      end
+      threads.collect { |thread| thread.join }


(Deleted my previous comment because I realized it was wrong after I posted it.)

Eliminate the local variable:

services.collect do |service| Thread.new { service.public_send method, *args } end.each(&:join)

georgeclaghorn · 2017-07-07T18:17:14Z

Thanks! I don’t see a need to test that the service is threaded. It’s enough for its existing tests to pass.

SGospodinov · 2017-07-07T18:17:35Z

I'll do it later because my flight is about to take off. 😀

Sent from my LGE Nexus 5X using FastHub

SGospodinov · 2017-07-08T00:55:10Z

Just pushed the edited version.

jeremy

Note that we can do concurrent uploads, too. Check out https://github.com/lloeki/ruby-tee/blob/master/lib/tee.rb for a nice approach.

Generally speaking, passing around IOs forces us into a "pull" execution model: read from this stream indiscriminately. Makes it awkward to push—or "pump"—a chunked input stream out through multiple channels concurrently, like one to calculate a streaming digest and two more to upload to storage services.

jeremy · 2017-07-08T02:56:36Z

lib/active_storage/service/mirror_service.rb

-        service.public_send method, *args
-      end
+        Thread.new { service.public_send method, *args }
+      end.map(&:join)


Can use concurrent-ruby's promises (or a lower level thread pool executor). Check out its usage in Action Cable.

I've changed the code to use thread pool executor, and now it looks like this:

def perform_across_services(method, *args) pool = Concurrent::ThreadPoolExecutor.new(max_threads: @services.count) # pool is local only for the example purpose services.each do |service| pool.post { service.public_send method, *args } end end

But this causes troubles with the test because threads do not join.
I can make them pass by sleeping for a while, but this does not seem like a good solution to the problem. Can you advise me how to handle this?

As @jeremy suggested, consider using promises:

def perform_across_services(method, *args) promises = services.collect do |service| Concurrent::Promise.execute { service.public_send method, *args } end Concurrent::Promise.zip(promises).value end

SGospodinov · 2017-07-08T19:29:50Z

Thank you for the advice. Just pushed version with promises but I'm using map to get all of the values because there is seems to be a problem with the zip method. It produces promise which rejected, even if all of the given promises are fulfilled.

georgeclaghorn · 2017-07-08T20:10:59Z

Sorry, my example was incorrect. You’ll need to splat the promises you pass to Concurrent::Promise.zip. This works:

def perform_across_services(method, *args)
  promises = services.collect do |service|
    Concurrent::Promise.execute { service.public_send method, *args }
  end
  Concurrent::Promise.zip(*promises).value
end

georgeclaghorn · 2017-07-08T22:38:33Z

Looking good! Are you up for implementing concurrent uploads, too? That’d add the most value, since deletion will usually be fast.

SGospodinov · 2017-07-08T22:41:31Z

Yes, I can start working on it.

jeremy · 2017-07-09T00:56:00Z

lib/active_storage/service/mirror_service.rb

@@ -1,4 +1,5 @@
 require "active_support/core_ext/module/delegation"
+require "concurrent"


Can tighten to require "concurrent/promise"?

Let's add concurrent-ruby to the gemspec, too, rather than rely on transitive dep via Active Support.

Ok, I'll do it.

jeremy · 2017-07-09T01:33:24Z

lib/active_storage/service/mirror_service.rb

      end
+      Concurrent::Promise.zip(*promises).value


Makes one wish for a simple services.pmap do |service| /cc @jdantonio 😊

Ref. ruby-concurrency/concurrent-ruby#493 (comment)

SGospodinov · 2017-07-09T17:27:09Z

@jeremy I think to add queues for service operations because concurrency causes some problems. E.g. if we have the following code:

service.upload(key, ...)
service.delete(key)
service.exists?(key)

delete will start desperately that upload is not finished yet. Then when exists?' is called and upload' is probably finished it will return true when the expected behaviour is to be false.
Can you help me to choose a gem for the implementing it because I'm not that familiar with ruby multithreading?

georgeclaghorn · 2017-07-09T17:31:31Z

Active Storage won’t, by itself, check the existence of or delete a key while uploading data for it, so we can ignore those issues.

SGospodinov · 2017-07-09T17:56:05Z

Ok, but how to test this behaviour?

georgeclaghorn · 2017-07-09T18:20:06Z

No need to, IMO. It’s still enough for the service’s existing tests to pass.

SGospodinov · 2017-07-10T15:02:45Z

This code works, but a few tests do not pass with the current mirror configuration. I don't know why but when disc services are used tests fail. When I configure test with GCS, everything passes.

georgeclaghorn · 2017-07-10T15:08:43Z

lib/active_storage/service/mirror_service.rb

-    each_service.collect do |service|
-      service.upload key, io.tap(&:rewind), checksum: checksum
-    end
+    perform_across_services :upload, key, io.tap(&:rewind), checksum: checksum


This isn’t going to work. We can’t share a single IO between concurrent uploads. Jeremy pointed you in the right direction above.

SGospodinov · 2017-07-10T22:56:20Z

lib/active_storage/service/mirror_service.rb

-    each_service.collect do |service|
-      service.upload key, io.tap(&:rewind), checksum: checksum
+    perform_async_across_services do |service|
+      service.upload key, StringIO.new(io.tap(&:rewind).read), checksum: checksum


I'm almost sure that this is not the best way to create new IO instance but clone or dup don't work. When any method is called inside upload it mutates all IO instances.

Right. I said we can’t just share a single IO instance between multiple threads, so cloning the given IO instance for each thread was a sensible thing to try. But we can’t do that, either: each clone will share the same underlying file descriptor.

So what can we do? Jeremy pointed you to ruby-tee. We probably can’t use it as-is, but its implementation is a good example of a way to do exactly what we want to do: safely share a single IO between multiple threads.

SGospodinov · 2017-07-12T02:23:26Z

lib/active_storage/async_uploader.rb

+  attr_reader :promise
+
+  def initialize service, key, checksum: nil
+    @data = ""


I'm not quite sure if using a string as a buffer, instead of an enumerator, is a problem. I'm also open for a new class name ideas.

SGospodinov · 2017-07-12T02:27:29Z

lib/active_storage/service/mirror_service.rb

    end
+    io.rewind
+    while chunk = io.read(1024)


This is the chunk size but I don't know where to places the constant. In MirrorService or in the AsyncUploader?

SGospodinov · 2017-07-25T21:04:07Z

@georgeclaghorn can you review the PR, please?

georgeclaghorn reviewed Jul 7, 2017

View reviewed changes

SGospodinov force-pushed the master branch from b5d86dc to 245f5f9 Compare July 8, 2017 00:53

jeremy reviewed Jul 8, 2017

View reviewed changes

SGospodinov force-pushed the master branch 2 times, most recently from 89b8e94 to a5e4715 Compare July 8, 2017 19:22

SGospodinov force-pushed the master branch from a5e4715 to 6d2cebb Compare July 8, 2017 20:15

SGospodinov changed the title ~~Mirror services execution is now threaded~~ Concurrent mirror services Jul 8, 2017

jeremy reviewed Jul 9, 2017

View reviewed changes

SGospodinov force-pushed the master branch from 6d2cebb to a9f9c46 Compare July 10, 2017 14:55

georgeclaghorn reviewed Jul 10, 2017

View reviewed changes

SGospodinov force-pushed the master branch from a9f9c46 to b5b88bd Compare July 10, 2017 22:51

SGospodinov commented Jul 10, 2017

View reviewed changes

SGospodinov force-pushed the master branch from b5b88bd to cd3063d Compare July 12, 2017 02:19

SGospodinov commented Jul 12, 2017

View reviewed changes

SGospodinov added 3 commits July 14, 2017 16:18

Concurrent deleting in mirror services

894cbcb

Adding concurrent-ruby as a dependency

daa40ac

Concurrent uploading in mirror services

5929b0e

SGospodinov force-pushed the master branch from cd3063d to 5929b0e Compare July 14, 2017 14:39

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Concurrent mirror services #17

Concurrent mirror services #17

SGospodinov commented Jul 7, 2017

georgeclaghorn Jul 7, 2017

georgeclaghorn Jul 7, 2017

georgeclaghorn commented Jul 7, 2017 •

edited

Loading

SGospodinov commented Jul 7, 2017 •

edited

Loading

SGospodinov commented Jul 8, 2017 •

edited

Loading

jeremy left a comment •

edited

Loading

jeremy Jul 8, 2017

SGospodinov Jul 8, 2017

georgeclaghorn Jul 8, 2017

SGospodinov commented Jul 8, 2017

georgeclaghorn commented Jul 8, 2017

georgeclaghorn commented Jul 8, 2017

SGospodinov commented Jul 8, 2017

jeremy Jul 9, 2017

jeremy Jul 9, 2017

SGospodinov Jul 9, 2017

jeremy Jul 9, 2017

SGospodinov commented Jul 9, 2017

georgeclaghorn commented Jul 9, 2017

SGospodinov commented Jul 9, 2017

georgeclaghorn commented Jul 9, 2017

SGospodinov commented Jul 10, 2017

georgeclaghorn Jul 10, 2017 •

edited

Loading

SGospodinov Jul 10, 2017 •

edited

Loading

georgeclaghorn Jul 11, 2017

SGospodinov Jul 12, 2017 •

edited

Loading

SGospodinov Jul 12, 2017

SGospodinov commented Jul 25, 2017

		@@ -1,4 +1,5 @@
		require "active_support/core_ext/module/delegation"
		require "concurrent"

Concurrent mirror services #17

Are you sure you want to change the base?

Concurrent mirror services #17

Conversation

SGospodinov commented Jul 7, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

georgeclaghorn commented Jul 7, 2017 • edited Loading

SGospodinov commented Jul 7, 2017 • edited Loading

SGospodinov commented Jul 8, 2017 • edited Loading

jeremy left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SGospodinov commented Jul 8, 2017

georgeclaghorn commented Jul 8, 2017

georgeclaghorn commented Jul 8, 2017

SGospodinov commented Jul 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SGospodinov commented Jul 9, 2017

georgeclaghorn commented Jul 9, 2017

SGospodinov commented Jul 9, 2017

georgeclaghorn commented Jul 9, 2017

SGospodinov commented Jul 10, 2017

georgeclaghorn Jul 10, 2017 • edited Loading

Choose a reason for hiding this comment

SGospodinov Jul 10, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SGospodinov Jul 12, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SGospodinov commented Jul 25, 2017

georgeclaghorn commented Jul 7, 2017 •

edited

Loading

SGospodinov commented Jul 7, 2017 •

edited

Loading

SGospodinov commented Jul 8, 2017 •

edited

Loading

jeremy left a comment •

edited

Loading

georgeclaghorn Jul 10, 2017 •

edited

Loading

SGospodinov Jul 10, 2017 •

edited

Loading

SGospodinov Jul 12, 2017 •

edited

Loading