Re: Stop thread-ing when condition is met

Panagiotis Atmatzidis <[email protected]> Wed, 30 Mar 2016 20:48:09 +0300
Newsgroups	gmane.comp.lang.ruby.general
Message-ID	<[email protected]>
Hello,

> On 30 Mar 2016, at 19:29, Robert Klemme <[email protected]> wrote:
> 
> On Tue, Mar 29, 2016 at 10:27 PM, Panagiotis Atmatzidis
> <[email protected] <mailto:[email protected]>> wrote:
>> Hello,
>> 
>> I am using ruby-thread[1] for the pool. Here is the code:
>> 
>> 
>> def collect
>>  items = []
>>  pages = 59
>>  pool = Thread.pool(10)
>>  mutex = Mutex.new
>>  1.upto(pages) do |n|
>>    pool.process {
>>      uri = URI(@uri)
>>      params = {
>>        :page => n,
>>        :page_size => 1000
>>      }
>>      uri.query = URI.encode_www_form(params)
>>      puts "Fetching from #{uri}"
>>      http = Net::HTTP.new(uri.host, uri.port)
>>      http.use_ssl = true
>>      request = Net::HTTP::Get.new(uri.request_uri)
>>      request.basic_auth(@user, @password)
>>      begin
>>        response = http.request(request)
>>        response.is_a?(Net::HTTPSuccess) ? data = JSON.parse(response.body)
>> : data = {}
>>      rescue SocketError
>>        @log.info(“Something happened on the way to heaven: #{e}"
>>      end
>>      begin
>>        data.each do |d|
>>          # ...mangle...
>>          mutex.synchronize {
>>            items.push(p) # avoid race condition
> 
> Where does p come from?

“p" comes from block of code inside ‘data’, transforms the data into a suitable format.

> And why do you hold the mutex only for each chunk of data that you
> fetch from a URL? This will lead to content of different pages to be
> mixed in items. Is this what you want to do?

Yes, I want to create an array of hashes.

> 
>>          }
>>        end
>>      rescue => e
>>        puts "Trouble parsing event: #{e}"
>>      end
>>    }
>>  end
>>  pool.shutdown
>>  File.open('sample-mutex.txt', 'w') {|f| f.write(items)}
>> end
>> 
>> The above version works but it’s not dynamic, depends on the number of
>> pages. I’ve tried implementing “break" as suggested by M. Kerwin in various
>> ways without success.
>> 
>> The condition I want to meet is: data.key?(“next”) # => false - When this
>> condition is met, I’d like to have to stop creating new threads while
>> allowing current threads to finish their work.
> 
> First of all, most likely you are not creating threads. They will be
> created by the pool upfront or on demand - you cannot influence that
> other than by shutting down the pool.
> 
> The only thing you could stop doing is place work in the pools inbound
> queue. But this is unlikely to work the way you imagine because the
> work is put in the queue _initially_ and most likely all the 59 tasks
> are placed into the queue before even data comes in from fetching
> those URLs.

Yes I came to realise that this is what is happening, the loop (as another user pointed in a previous email) is just too fast and I have no control over that.

> So by the time you detect that it's enough work all the
> work is scheduled and will be processed - unless you somehow
> immediately terminate the processing (e.g. with #shutdown!) which you
> don't.
> 
> But if the number of URLs to process depends on the data downloaded
> from those URLs then maybe doing this in parallel is not such a good
> option.
>> Is it true the Fibers are better suited for this work as Daniele suggested?
> 
> I don't know. First you should step back and tell us what you are
> trying to achieve and what the purpose of all this is. Until then it's
> really a lot guesswork and anything we can tell you based on guesswork
> might as well be wrong.

I am working on a module which fetches a series of JSON objects. Each object is the processed and the result is saved in an array and passed to another module. Currently the code fetches page1, then process the data and dumps the result on the ‘items’ Array. Finally checks if there’s a value in “data.next”. If the value of “data.next” is a URI pointing to the next page. This way iterates until there are no pages (no URI value in ‘next’ key). This process takes ~ 62 seconds to complete for 59 pages.

I made a version using threads which does the same in ~ 6 seconds. But I since these pages might become 65 or 45 at any given time by a third party, I am trying to make this more dynamic. I just want to speed things up!

> There are a few things apart from those mentioned above:
> - Your method does not have inputs, instead it uses instance variables.
> - The method does not return anything meaningful. In terms of
> modularity I'd rather have it return items and do the writing
> elsewhere. This also helps with testing.
> - For items you could use an instance of Queue instead of an Array plus a Mutex.

Hm, thanks for the input, I need to test that.

> - If you really always need to write what you download you could push
> results into another queue with a single thread which reads and
> immediately writs to "sample-mutex.txt”.

I’m creating the file to test the data for consistency agains the working version, so there’s no need to actually write anything.

thanks!

> Cheers
> 
> robert
> 
> --
> [guy, jim, charlie].each {|him| remember.him do |as, often| as.you_can
> - without end}
> http://blog.rubybestpractices.com/ <http://blog.rubybestpractices.com/>
> 
> Unsubscribe: <mailto:[email protected]?subject=unsubscribe <mailto:[email protected]?subject=unsubscribe>>
> <http://lists.ruby-lang.org/cgi-bin/mailman/options/ruby-talk <http://lists.ruby-lang.org/cgi-bin/mailman/options/ruby-talk>>


Unsubscribe: <mailto:[email protected]?subject=unsubscribe>
<http://lists.ruby-lang.org/cgi-bin/mailman/options/ruby-talk>
signature.asc (application/pgp-signature, 832 B)