Strict priority polling w/ backmerge #288

atyndall · 2016-12-13T23:56:18Z

@phstc This is the strict priority polling implementation from #263 with master backmerged, it didn't make it when you closed #263 in favour of #284.

phstc · 2016-12-14T20:59:36Z

@mariokostelac would you mind reviewing it?

mariokostelac · 2016-12-14T21:42:54Z

@atyndall could you rebase on master first please? We merged my change in the master.

atyndall · 2016-12-14T23:21:44Z

@mariokostelac Should be done now

mariokostelac · 2016-12-15T19:48:11Z

lib/shoryuken/polling.rb

@@ -19,11 +19,49 @@ def ==(other)
      end

      alias_method :eql?, :==
+
+      def to_s
+        options.empty? ? name : super


Could you explain this?

It was to tidy up log messages such as these:

Before change;

2016-12-15T23:38:08Z 88280 TID-oxeiaih3s DEBUG: Looking for new messages in '#<struct Shoryuken::Polling::QueueConfiguration name="development_atyndall_affinity_0", options={}>' 2016-12-15T23:38:08Z 88280 TID-oxeiaih3s DEBUG: Fetcher for '#<struct Shoryuken::Polling::QueueConfiguration name="development_atyndall_affinity_0", options={}>' completed in 268.11899999999997 ms

After change;

2016-12-15T23:39:56Z 88353 TID-out2291p8 DEBUG: Looking for new messages in 'development_atyndall_affinity_0' 2016-12-15T23:39:56Z 88353 TID-out2291p8 DEBUG: Fetcher for 'development_atyndall_affinity_0' completed in 217.197 ms

Seeing as QueueConfigurations are considered equal if have no options and their strings are equal, it made sense to me to make their string representation just the names of the queues to tidy up logging. This could also be handled by changing inspect, I'm open to either way.

Can we make a case where options does exist a little bit nicer? :)

Made some changes

mariokostelac · 2016-12-15T19:49:34Z

lib/shoryuken/polling.rb

+          .each_with_object({}) { |queue, h| h[queue] = [true, nil] }
+
+        # Most recently used queue
+        @current_queue = nil


From what I understand, that it actually last_used_queue. Can we name it like that? It will make more sense for the loop.

Rework has eliminated variable altogether

mariokostelac · 2016-12-15T19:51:51Z

lib/shoryuken/polling.rb

+          .map(&:first)
+
+        # Pause status of the queues
+        @queue_status = @queue_order


I think we can compress queue_status in just one field, something like "activation time". When Time.now < activation_time, queue is not active yet. When Time.now >= activation_time, queue is active.

mariokostelac · 2016-12-15T19:55:39Z

lib/shoryuken/polling.rb

+
+        # Priority ordering of the queues
+        @queue_order = @queue_priorities
+          .to_a


I think we need to simplify this code or add a comment of the datastructure coming out :).

mariokostelac · 2016-12-15T19:57:09Z

lib/shoryuken/polling.rb

+
+        # Loop through the queue order from the current queue until we find a
+        # queue that is next in line and is not paused
+        while true


Can we use a for loop? This external bookkeeping of variable i looks ugly.

It's not drop-in but we have similar code in our custom stuff that works like this:

queues = @queue_order.dup while true queue = queues.first # process queue queues.rotate! end

So rather than maintaining i or @current_queue you dup the ordered queues at some point and then Array#rotate! them as needed to move to the next queue.

@gshutler Excellent! I didn't know about Array#rotate!, I'll rework the code around it.

mariokostelac · 2016-12-15T19:57:49Z

lib/shoryuken/polling.rb

+
+          i += 1
+          return queue if active
+          return nil if i >= @queue_order.length # Prevents infinite looping


Also, using For loop gets rid of this line.

mariokostelac · 2016-12-15T20:01:44Z

lib/shoryuken/polling.rb

+        logger.debug "Paused '#{queue}'"
+      end
+
+      def unpause_queues


Once we compress the state, we do not have to unpause queues because we know their order, we just have to check if activation_time > Time.now or not :).

Ah yes, I'll implement this approach.

mariokostelac · 2016-12-15T22:08:06Z

spec/shoryuken/polling_spec.rb

+      expect(subject.next_queue).to eq(queue2)
+    end
+
+    it 'cycles when declared asc' do


why looks the same as previous one, while priorities are different?

The priorities aren't different, the priorities are the number. They are declaration order independent.

Therefore

[shoryuken, 2] [uppercut, 1]

and

[uppercut, 1] [shoryuken, 2]

Have the same meaning, because they have the same priority numbers.

Oh, did not see that queues are reversed!

mariokostelac · 2016-12-15T22:08:55Z

@atyndall could you please describe the algorithm you want to implement? That's the only way we can be sure we're on the same page :).

gshutler · 2016-12-15T22:25:13Z

lib/shoryuken/polling.rb

+
+        # Priority ordering of the queues
+        @queue_order = @queue_priorities
+          .to_a


@queue_order = queues.group_by(&:itself).sort_by { |q, qs| -qs.count }.map(&:first)

May have to do the longhand of Object#itself if you need support for Ruby < 2.2.0

Nice, will use instead

atyndall · 2016-12-16T01:38:09Z

I have made significant changes to the implementation taking into account the feedback, thanks!

@mariokostelac Sure, at a high level each queue is assigned a unique priority, describing how critical its work is.

When shoryuken goes to select a new job, it chooses the highest priority unpaused queue to check first.

If that queue contains no work, it will proceed to check the next highest priority unpaused queue, etc.

When new work is received or when a queue unpauses, the checking order is reset to check the highest priority queue once more.

mariokostelac · 2016-12-16T13:48:11Z

@atyndall so that's priority queue, right? As long as there are jobs in a queue with the highest priority, poll that queue. Once it has no jobs, pause it for delay period and continue with the rest of queues?

If yes, we can just use https://ruby-doc.org/stdlib-2.3.1/libdoc/set/rdoc/SortedSet.html.

atyndall · 2016-12-16T13:53:59Z

@mariokostelac Yeah that's correct.
I don't really understand what changes you're envisioning with the use of a SortedSet

mariokostelac

I think we have to figure out how to release that. This change is actually moving from "one polling strategy" model to "choose your polling strategy" model.

I suggest we do something similar to:

we do not share options between polling strategies (no sharing delay option, it's confusing)
we cut a new version where we deprecate delay parameter (it's confusing anyway). The new version also has a way to choose polling strategy and parameters are scoped per polling strategy (while weighted round robin still supporting delay for some time).

WDYT @phstc?

mariokostelac · 2016-12-18T12:41:34Z

lib/shoryuken/polling.rb

+        @queue_order = @initial_order.dup
+
+        # Pause status of the queues, default to past time (unpaused)
+        @paused_until = queues


mariokostelac · 2016-12-18T12:43:20Z

lib/shoryuken/polling.rb

+          pause(queue)
+        else
+          # Reset the queue order to the initial ordering
+          @queue_order = @initial_order.dup


Since we're using this reset_queue_order several times, can we isolate that into a method? If we do so, we do not even need a comment. Let the method name talk! :)

Will add such a method

mariokostelac · 2016-12-18T12:45:32Z

lib/shoryuken/polling.rb

+
+        # `rotate!` through the queue list until we find an unpaused queue
+        begin
+          next_queue = @queue_order.first


What is stored in here? It seems that .map(&:first) in initializer stores just the q in here, no?

It's getting refactored out now

mariokostelac · 2016-12-18T12:46:03Z

spec/shoryuken/polling_spec.rb

+
+      allow(subject).to receive(:delay).and_return(10)
+
+      now = Time.now


Good practice 👍

mariokostelac · 2016-12-18T12:53:05Z

lib/shoryuken/polling.rb

+        now = Time.now
+
+        # Return nil if all queues are paused to prevent infinite loop
+        return nil if @paused_until.values.all? { |t| t > now }


Hmmm, I know I am picky here but this code seems a little bit complicated to me :). Since our number of queues has to be low (100 at most), we do not have to think about speed too much and we can implement a straightforward solution - iterate over queues from index last_index (like you did before) and just check if a queue is paused. The first one that is not - has to be returned. If it ever gets slow (I really doubt that :)), we can optimize this class, but I'd rather have something that is very simple and everybody can understand, fast!

I think it's far simpler than maintaining that initial_order, having a check from line 174, making sure that we call dup every time we want to get a copy etc.

The whole next_active_queue would be very simple, something like:

def next_active_queue size = @queues.length for 0..limit do queue = @queues[@next_queue_index] @next_queue_index = (@next_queue_index + 1) % size return queue unless queue_paused?(queue) end return nil end

We would be able to have very simple, reusable queue_paused? method.
Also, reset_next_queue is simple as @next_queue_index = 0. We have to call it in initializer, too.

WDYT?

Sounds good, I'll implement that approach.

mariokostelac · 2016-12-18T13:00:14Z

lib/shoryuken/polling.rb

+      end
+
+      def active_queues
+        @paused_until


Also, once we have queue_paused?, we can use something simple as

@queues.reject{ |q| queue_paused?(q) }

I think it's far easier to understand

I'll add the method

atyndall

Made additional changes to simplify methods further, and move them into smaller submethods.

atyndall · 2016-12-19T00:16:26Z

lib/shoryuken/polling.rb

+        now = Time.now
+
+        # Return nil if all queues are paused to prevent infinite loop
+        return nil if @paused_until.values.all? { |t| t > now }


Sounds good, I'll implement that approach.

atyndall · 2016-12-19T00:16:38Z

lib/shoryuken/polling.rb

+      end
+
+      def active_queues
+        @paused_until


I'll add the method

atyndall · 2016-12-19T00:18:00Z

lib/shoryuken/polling.rb

+          pause(queue)
+        else
+          # Reset the queue order to the initial ordering
+          @queue_order = @initial_order.dup


Will add such a method

atyndall · 2016-12-19T00:18:10Z

lib/shoryuken/polling.rb

@@ -19,11 +19,49 @@ def ==(other)
      end

      alias_method :eql?, :==
+
+      def to_s
+        options.empty? ? name : super


Made some changes

atyndall · 2016-12-19T00:18:25Z

lib/shoryuken/polling.rb

+
+        # `rotate!` through the queue list until we find an unpaused queue
+        begin
+          next_queue = @queue_order.first


It's getting refactored out now

phstc · 2016-12-19T01:07:10Z

@mariokostelac I think most of people they don't care about which polling strategy they need to use, the only want to make it work. The shoryuken.yml by default (in the samples/README) should be something like this

concurrency: 25  # The number of allocated threads to process messages. Default 25
queues:
  - [high_priority, 6]
  - [default, 2]
  - [low_priority, 1]

I would remove the delay from the sample, I agree it's confusing. Most people, only want to install, configure the queues and run it. Maybe we could also omit concurrency from the samples/README, not sure if it's needed. Most people only need the defaults, these extra options can be documented in the wiki. Including the polling strategy, I'm still seeing it as "edge case".

With that, I wouldn't do any major release, I would just let "choose your polling strategy" something special for now, documented in the wiki. Keeping shoryuken as simple and straightforward as possible, I don't mind about keeping the delay, default strategy will be probably the most used one.

WDYT?

Is this change introducing any breaking changes?

atyndall · 2016-12-19T01:18:25Z

@phstc AFAIK there are no breaking changes, merging this in should not have any impact on anyone who doesn't use it.

For your information, this is what our queueing config looks like atm;

polling_strategy: !ruby/class 'Shoryuken::Polling::StrictPriority'
timeout: 1140   # Seconds grace to shutdown
concurrency: 1  # The number of allocated threads to process messages. Default 25
delay: 60       # The delay in seconds to pause a queue when it's empty. Default 0
queues:
  - [high_priority,       3]
  - [medium_priority,     2]
  - [low_priority,        1]

phstc · 2016-12-19T01:22:42Z

@atyndall nice! polling_strategy, timeout and delay are still optional, right?

atyndall · 2016-12-19T01:24:09Z

@phstc Yep

mariokostelac · 2016-12-21T23:24:09Z

@atyndall @phstc I am fine with having a default polling strategy being set and all associated parameters, but I am not really happy with sharing these parameters between polling strategies. They could have slightly or completely different meanings. That's why I'd like to namespace these parameters in new version :). Also, delay is not descriptive at all.

phstc · 2016-12-21T23:28:52Z

That's why I'd like to namespace these parameters in new version :). Also, delay is not descriptive at all.

@mariokostelac I agree, I just want to make the basic shoryuken.yml as trivial as possible. Most of people don't care about these extra settings, as long as we can have a good set of defaults, all is fine 🍻

Add StrictPriority strategy

05b467b

atyndall force-pushed the strict_priority_polling branch from 1eab6b6 to 05b467b Compare December 14, 2016 23:21

mariokostelac suggested changes Dec 15, 2016

View reviewed changes

mariokostelac reviewed Dec 15, 2016

View reviewed changes

gshutler reviewed Dec 15, 2016

View reviewed changes

Restructure strategy based on feedback to be more clear and concise

160bba7

mariokostelac reviewed Dec 18, 2016

View reviewed changes

atyndall added 3 commits December 19, 2016 08:15

Better QueueConfiguration options to_s

5cb87b0

Unused variable change to _

37256ae

Restructure to use an indexing approach with more submethods

298f560

atyndall commented Dec 19, 2016

View reviewed changes

phstc merged commit b48342e into ruby-shoryuken:master Dec 21, 2016


		allow(subject).to receive(:delay).and_return(10)

		now = Time.now

Strict priority polling w/ backmerge #288

Strict priority polling w/ backmerge #288

Conversation

atyndall commented Dec 13, 2016

phstc commented Dec 14, 2016

mariokostelac commented Dec 14, 2016

atyndall commented Dec 14, 2016

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

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

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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mariokostelac commented Dec 15, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

atyndall commented Dec 16, 2016

mariokostelac commented Dec 16, 2016

atyndall commented Dec 16, 2016

mariokostelac left a comment

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

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

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

Choose a reason for hiding this comment

atyndall left a comment

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

phstc commented Dec 19, 2016

atyndall commented Dec 19, 2016

phstc commented Dec 19, 2016

atyndall commented Dec 19, 2016

mariokostelac commented Dec 21, 2016

phstc commented Dec 21, 2016