(fix): refactor batch_processor to start on first process and use mutex to wait till deadline if nothing in the queue (#232)

* initial commit of refactor

* fix project spec

* 2.4.4 is too quick

* trying to get 2.4.4 queue to fill

* cleanup one test

* update to look for debug statement

* reduce size of max queue

Author: thomaszurkan-optimizely

.rubocop.yml

@@ -46,7 +46,7 @@ Style/SignalException:
   Enabled: false
 
 Lint/RescueException:
-  Enabled: false
+  Enabled: true
 
 Layout/EndOfLine:
   EnforcedStyle: lf

.travis.yml

@@ -9,7 +9,7 @@ rvm:
   - 2.5.1
   - 2.6.0
 before_install:
-  - gem update --system
+#  - gem update --system
   - gem install bundler
 install:
   - bundle install

lib/optimizely/event/batch_event_processor.rb

@@ -31,7 +31,6 @@ class BatchEventProcessor < EventProcessor
     DEFAULT_BATCH_INTERVAL = 30_000 # interval in milliseconds
     DEFAULT_QUEUE_CAPACITY = 1000
     DEFAULT_TIMEOUT_INTERVAL = 5 # interval in seconds
-    MAX_NIL_COUNT = 3
 
     FLUSH_SIGNAL = 'FLUSH_SIGNAL'
     SHUTDOWN_SIGNAL = 'SHUTDOWN_SIGNAL'
@@ -62,7 +61,7 @@ def initialize(
       @notification_center = notification_center
       @current_batch = []
       @started = false
-      start!
+      @stopped = false
     end
 
     def start!
@@ -72,26 +71,37 @@ def start!
       end
       @flushing_interval_deadline = Helpers::DateTimeUtils.create_timestamp + @flush_interval
       @logger.log(Logger::INFO, 'Starting scheduler.')
-      @thread = Thread.new { run }
+      if @wait_mutex.nil?
+        @wait_mutex = Mutex.new
+        @resource = ConditionVariable.new
+      end
+      @thread = Thread.new { run_queue }
       @started = true
+      @stopped = false
     end
 
     def flush
       @event_queue << FLUSH_SIGNAL
+      @wait_mutex.synchronize { @resource.signal }
     end
 
     def process(user_event)
       @logger.log(Logger::DEBUG, "Received userEvent: #{user_event}")
 
-      if !@started || !@thread.alive?
+      # if the processor has been explicitly stopped. Don't accept tasks
+      if @stopped
         @logger.log(Logger::WARN, 'Executor shutdown, not accepting tasks.')
         return
       end
 
+      # start if the processor hasn't been started
+      start! unless @started
+
       begin
         @event_queue.push(user_event, true)
-      rescue Exception
-        @logger.log(Logger::WARN, 'Payload not accepted by the queue.')
+        @wait_mutex.synchronize { @resource.signal }
+      rescue => e
+        @logger.log(Logger::WARN, 'Payload not accepted by the queue: ' + e.message)
         return
       end
     end
@@ -101,42 +111,20 @@ def stop!
 
       @logger.log(Logger::INFO, 'Stopping scheduler.')
       @event_queue << SHUTDOWN_SIGNAL
+      @wait_mutex.synchronize { @resource.signal }
       @thread.join(DEFAULT_TIMEOUT_INTERVAL)
       @started = false
+      @stopped = true
     end
 
     private
 
-    def run
-      # if we receive a number of item nils that reach MAX_NIL_COUNT,
-      # then we hang on the pop via setting use_pop to false
-      @nil_count = 0
-      # hang on pop if true
-      @use_pop = false
-      loop do
-        if Helpers::DateTimeUtils.create_timestamp >= @flushing_interval_deadline
-          @logger.log(Logger::DEBUG, 'Deadline exceeded flushing current batch.')
-          flush_queue!
-          @flushing_interval_deadline = Helpers::DateTimeUtils.create_timestamp + @flush_interval
-          @use_pop = true if @nil_count > MAX_NIL_COUNT
-        end
-
-        item = @event_queue.pop if @event_queue.length.positive? || @use_pop
-
-        if item.nil?
-          # when nil count is greater than MAX_NIL_COUNT, we hang on the pop until there is an item available.
-          # this avoids to much spinning of the loop.
-          @nil_count += 1
-          next
-        end
-
-        # reset nil_count and use_pop if we have received an item.
-        @nil_count = 0
-        @use_pop = false
-
+    def process_queue
+      while @event_queue.length.positive?
+        item = @event_queue.pop
         if item == SHUTDOWN_SIGNAL
           @logger.log(Logger::DEBUG, 'Received shutdown signal.')
-          break
+          return false
         end
 
         if item == FLUSH_SIGNAL
@@ -147,15 +135,35 @@ def run
 
         add_to_batch(item) if item.is_a? Optimizely::UserEvent
       end
+      true
+    end
+
+    def run_queue
+      loop do
+        if Helpers::DateTimeUtils.create_timestamp >= @flushing_interval_deadline
+          @logger.log(Logger::DEBUG, 'Deadline exceeded flushing current batch.')
+
+          break unless process_queue
+
+          flush_queue!
+          @flushing_interval_deadline = Helpers::DateTimeUtils.create_timestamp + @flush_interval
+        end
+
+        break unless process_queue
+
+        # what is the current interval to flush in seconds
+        interval = (@flushing_interval_deadline - Helpers::DateTimeUtils.create_timestamp) * 0.001
+
+        next unless interval.positive?
+
+        @wait_mutex.synchronize { @resource.wait(@wait_mutex, interval) }
+      end
     rescue SignalException
       @logger.log(Logger::ERROR, 'Interrupted while processing buffer.')
-    rescue Exception => e
+    rescue => e
       @logger.log(Logger::ERROR, "Uncaught exception processing buffer. #{e.message}")
     ensure
-      @logger.log(
-        Logger::INFO,
-        'Exiting processing loop. Attempting to flush pending events.'
-      )
+      @logger.log(Logger::INFO, 'Exiting processing loop. Attempting to flush pending events.')
       flush_queue!
     end
 

spec/event/batch_event_processor_spec.rb

@@ -32,6 +32,7 @@
   let(:event) { project_config.get_event_from_key('test_event') }
 
   before(:example) do
+    spy_logger = spy('logger')
     @event_queue = SizedQueue.new(100)
     @event_dispatcher = Optimizely::EventDispatcher.new
     allow(@event_dispatcher).to receive(:dispatch_event).with(instance_of(Optimizely::Event))
@@ -46,6 +47,8 @@
   it 'should log waring when service is already started' do
     @event_processor = Optimizely::BatchEventProcessor.new(logger: spy_logger)
     @event_processor.start!
+    @event_processor.start!
+
     expect(spy_logger).to have_received(:log).with(Logger::WARN, 'Service already started.').once
   end
 
@@ -61,22 +64,23 @@
     )
 
     @event_processor.process(conversion_event)
-    # flush interval is set to 100ms. Wait for 300ms and assert that event is dispatched.
-    sleep 1
+    # flush interval is set to 100ms. Wait for 200ms and assert that event is dispatched.
+    sleep 0.2
 
     expect(@event_dispatcher).to have_received(:dispatch_event).with(log_event).once
     expect(@notification_center).to have_received(:send_notifications).with(
       Optimizely::NotificationCenter::NOTIFICATION_TYPES[:LOG_EVENT],
       log_event
     ).once
     expect(spy_logger).to have_received(:log).with(Logger::INFO, 'Flushing Queue.').once
+    expect(spy_logger).to have_received(:log).with(Logger::DEBUG, 'Deadline exceeded flushing current batch.').at_most(2).times
   end
 
   it 'should flush the current batch when max batch size met' do
     @event_processor = Optimizely::BatchEventProcessor.new(
       event_dispatcher: @event_dispatcher,
       batch_size: 11,
-      flush_interval: 100_000,
+      flush_interval: 10_000,
       logger: spy_logger
     )
 
@@ -92,8 +96,9 @@
     end
 
     # Wait until other thread has processed the event.
-    until @event_processor.event_queue.empty?; end
-    until @event_processor.current_batch.empty?; end
+    sleep 0.1 until @event_processor.event_queue.empty?
+
+    sleep 0.1 until @event_processor.current_batch.empty?
 
     expect(Optimizely::EventFactory).to have_received(:create_log_event).with(expected_batch, spy_logger).once
     expect(@event_dispatcher).to have_received(:dispatch_event).with(
@@ -109,7 +114,7 @@
     @event_processor = Optimizely::BatchEventProcessor.new(
       event_queue: @event_queue,
       event_dispatcher: @event_dispatcher,
-      flush_interval: 100_000,
+      flush_interval: 10_000,
       logger: spy_logger
     )
 
@@ -120,8 +125,9 @@
     @event_processor.flush
 
     # Wait until other thread has processed the event.
-    until @event_processor.event_queue.empty?; end
-    until @event_processor.current_batch.empty?; end
+    sleep 0.1 until @event_processor.event_queue.empty?
+
+    sleep 0.1 until @event_processor.current_batch.empty?
 
     expect(@event_dispatcher).to have_received(:dispatch_event).with(log_event).twice
     expect(@event_processor.event_queue.length).to eq(0)
@@ -143,13 +149,13 @@
     expect(user_event1.event_context[:revision]).to eq('1')
     @event_processor.process(user_event1)
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 1; end
+    sleep 0.1 while @event_processor.current_batch.length != 1
 
     expect(user_event2.event_context[:revision]).to eq('2')
     @event_processor.process(user_event2)
     @event_processor.process(user_event2)
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 2; end
+    sleep 0.1 while @event_processor.current_batch.length != 2
 
     expect(@event_dispatcher).to have_received(:dispatch_event).with(log_event).once
     expect(spy_logger).to have_received(:log).with(Logger::DEBUG, 'Revisions mismatched: Flushing current batch.').once
@@ -170,13 +176,13 @@
     expect(user_event1.event_context[:project_id]).to eq('X')
     @event_processor.process(user_event1)
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 1; end
+    sleep 0.1 while @event_processor.current_batch.length != 1
 
     expect(user_event2.event_context[:project_id]).to eq('Y')
     @event_processor.process(user_event2)
     @event_processor.process(user_event2)
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 2; end
+    sleep 0.1 while @event_processor.current_batch.length != 2
 
     expect(@event_dispatcher).to have_received(:dispatch_event).with(log_event).once
     expect(spy_logger).to have_received(:log).with(Logger::DEBUG, 'Project Ids mismatched: Flushing current batch.').once
@@ -252,10 +258,11 @@
     @event_processor.process(conversion_event)
 
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 1; end
+    sleep 0.1 while @event_processor.current_batch.length != 1
+
     @event_processor.flush
     # Wait until other thread has processed the event.
-    until @event_processor.current_batch.empty?; end
+    sleep 0.1 until @event_processor.current_batch.empty?
 
     expect(@notification_center).to have_received(:send_notifications).with(
       Optimizely::NotificationCenter::NOTIFICATION_TYPES[:LOG_EVENT],
@@ -281,10 +288,11 @@
 
     @event_processor.process(conversion_event)
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 1; end
+    sleep 0.1 while @event_processor.current_batch.length != 1
+
     @event_processor.flush
     # Wait until other thread has processed the event.
-    until @event_processor.current_batch.empty?; end
+    sleep 0.1 until @event_processor.current_batch.empty?
 
     expect(@notification_center).not_to have_received(:send_notifications)
     expect(spy_logger).to have_received(:log).once.with(
@@ -315,7 +323,7 @@
     end
 
     # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 4; end
+    sleep 0.1 while @event_processor.current_batch.length != 4
     expect(@event_dispatcher).not_to have_received(:dispatch_event)
 
     @event_processor.stop!
@@ -329,32 +337,31 @@
 
   it 'should log a warning when Queue gets full' do
     @event_processor = Optimizely::BatchEventProcessor.new(
-      event_queue: SizedQueue.new(10),
+      event_queue: SizedQueue.new(5),
       event_dispatcher: @event_dispatcher,
-      batch_size: 100,
-      flush_interval: 100_000,
+      batch_size: 1000,
+      flush_interval: 10_000,
       logger: spy_logger
     )
 
     user_event = Optimizely::UserEventFactory.create_conversion_event(project_config, event, 'test_user', nil, nil)
-    11.times do
+    900.times do
       @event_processor.process(user_event)
     end
 
-    # Wait until other thread has processed the event.
-    while @event_processor.current_batch.length != 10; end
     expect(@event_dispatcher).not_to have_received(:dispatch_event)
-    expect(spy_logger).to have_received(:log).with(Logger::WARN, 'Payload not accepted by the queue.').once
+    expect(spy_logger).to have_received(:log).with(Logger::WARN, 'Payload not accepted by the queue: queue full').at_least(:once)
   end
 
   it 'should not process and log when Executor is not running' do
     @event_processor = Optimizely::BatchEventProcessor.new(
       event_dispatcher: @event_dispatcher,
       batch_size: 100,
-      flush_interval: 100_000,
+      flush_interval: 10_000,
       logger: spy_logger
     )
 
+    @event_processor.start!
     @event_processor.stop!
 
     user_event = Optimizely::UserEventFactory.create_conversion_event(project_config, event, 'test_user', nil, nil)

spec/project_spec.rb

@@ -2814,6 +2814,8 @@ def callback(_args); end
       project_instance = Optimizely::Project.new(nil, nil, nil, nil, true, nil, nil, config_manager, nil, event_processor)
 
       expect(config_manager.stopped).to be false
+      expect(event_processor.started).to be false
+      event_processor.start!
       expect(event_processor.started).to be true
 
       project_instance.close