From: Peter Amstutz <peter.amstutz@curoverse.com>
Date: Thu, 23 Oct 2014 19:22:21 +0000 (-0400)
Subject: 4295: First pass at more efficient database utilization.  A few tests fail.
X-Git-Tag: 1.1.0~2069^2~5
X-Git-Url: https://git.arvados.org/arvados.git/commitdiff_plain/74e7f636d3ec2f3a20640b721a3fd76fd23cd788

4295: First pass at more efficient database utilization.  A few tests fail.
---

diff --git a/services/api/lib/eventbus.rb b/services/api/lib/eventbus.rb
index bccbeea4bb..080138ba4c 100644
--- a/services/api/lib/eventbus.rb
+++ b/services/api/lib/eventbus.rb
@@ -49,65 +49,95 @@ class EventBus
   end
 
   # Push out any pending events to the connection +ws+
-  # +id+  the id of the most recent row in the log table, may be nil
-  def push_events ws, id = nil
-      begin
-        # Must have at least one filter set up to receive events
-        if ws.filters.length > 0
-          # Start with log rows readable by user, sorted in ascending order
-          logs = Log.readable_by(ws.user).order("id asc")
-
-          cond_id = nil
-          cond_out = []
-          param_out = []
-
-          if ws.last_log_id
-            # Client is only interested in log rows that are newer than the
-            # last log row seen by the client.
-            cond_id = "logs.id > ?"
-            param_out << ws.last_log_id
-          elsif id
-            # No last log id, so only look at the most recently changed row
-            cond_id = "logs.id = ?"
-            param_out << id.to_i
-          else
-            return
-          end
+  # +notify_id+  the id of the most recent row in the log table, may be nil
+  #
+  # This accepts a websocket and a notify_id (this is the row id from Postgres LISTEN/NOTIFY, it may nil)
+  # It queries the database for log rows that are either
+  #  a) greater than ws.last_log_id, which is the last log id which was a candidate to be sent out
+  #  b) if ws.last_log_id is nil, then it queries rows starting with notify_id
+  #
+  # Regular Arvados permissions are applied using readable_by() and filters using record_filters()
+  # To avoid clogging up the database, queries are limited to batches of 100.  It will schedule a new
+  # push_events call if there are more log rows to send.
+  def push_events ws, notify_id
+    begin
+      if !notify_id.nil? and !ws.last_log_id.nil? and notify_id <= ws.last_log_id
+        # This notify is for a row we've handled already.
+        return
+      end
 
-          # Now process filters provided by client
-          ws.filters.each do |filter|
-            ft = record_filters filter.filters, Log
-            if ft[:cond_out].any?
-              # Join the clauses within a single subscription filter with AND
-              # so it is consistent with regular queries
-              cond_out << "(#{ft[:cond_out].join ') AND ('})"
-              param_out += ft[:param_out]
-            end
-          end
+      # Must have at least one filter set up to receive events
+      if ws.filters.length > 0
+        # Start with log rows readable by user, sorted in ascending order
+        logs = Log.readable_by(ws.user).order("id asc")
 
-          # Add filters to query
-          if cond_out.any?
-            # Join subscriptions with OR
-            logs = logs.where(cond_id + " AND ((#{cond_out.join ') OR ('}))", *param_out)
-          else
-            logs = logs.where(cond_id, *param_out)
-          end
+        cond_id = nil
+        cond_out = []
+        param_out = []
 
-          # Finally execute query and actually send the matching log rows
-          logs.each do |l|
-            ws.send(l.as_api_response.to_json)
-            ws.last_log_id = l.id
+        if !ws.last_log_id.nil?
+          # Client is only interested in log rows that are newer than the
+          # last log row seen by the client.
+          cond_id = "logs.id > ?"
+          param_out << ws.last_log_id
+        elsif !notify_id.nil?
+          # No last log id, so look at rows starting with notify id
+          cond_id = "logs.id >= ?"
+          param_out << notify_id
+        else
+          # No log id to start from, nothing to do, return
+          return
+        end
+
+        # Now build filters provided by client
+        ws.filters.each do |filter|
+          ft = record_filters filter.filters, Log
+          if ft[:cond_out].any?
+            # Join the clauses within a single subscription filter with AND
+            # so it is consistent with regular queries
+            cond_out << "(#{ft[:cond_out].join ') AND ('})"
+            param_out += ft[:param_out]
           end
-        elsif id
-          # No filters set up, so just record the sequence number
-          ws.last_log_id = id.to_i
         end
-      rescue Exception => e
-        Rails.logger.warn "Error publishing event: #{$!}"
-        Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
-        ws.send ({status: 500, message: 'error'}.to_json)
-        ws.close
+
+        # Add filters to query
+        if cond_out.any?
+          # Join subscriptions with OR
+          logs = logs.where(cond_id + " AND ((#{cond_out.join ') OR ('}))", *param_out)
+        else
+          logs = logs.where(cond_id, *param_out)
+        end
+
+        # Execute query and actually send the matching log rows
+        count = 0
+        limit = 100
+
+        logs.limit(limit).each do |l|
+          ws.send(l.as_api_response.to_json)
+          ws.last_log_id = l.id
+          count += 1
+        end
+
+        if count == limit
+          # Number of rows returned was capped by limit(), we need to schedule
+          # another query to get more logs (will start from last_log_id
+          # reported by current query)
+          @channel.push nil
+        elsif !notify_id.nil? and notify_id > ws.last_log_id
+          # Number of rows returned was less than cap, but the notify id is
+          # higher than the last id visible to the client, so update last_log_id
+          ws.last_log_id = notify_id
+        end
+      elsif !notify_id.nil?
+        # No filters set up, so just record the sequence number
+        ws.last_log_id = notify_id
       end
+    rescue => e
+      Rails.logger.warn "Error publishing event: #{$!}"
+      Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
+      ws.send ({status: 500, message: 'error'}.to_json)
+      ws.close
+    end
   end
 
   # Handle inbound subscribe or unsubscribe message.
@@ -132,7 +162,7 @@ class EventBus
           ws.send ({status: 200, message: 'subscribe ok', filter: p}.to_json)
 
           # Send any pending events
-          push_events ws
+          push_events ws, nil
         else
           ws.send ({status: 403, message: "maximum of #{MAX_FILTERS} filters allowed per connection"}.to_json)
         end
@@ -153,7 +183,7 @@ class EventBus
       end
     rescue Oj::Error => e
       ws.send ({status: 400, message: "malformed request"}.to_json)
-    rescue Exception => e
+    rescue => e
       Rails.logger.warn "Error handling message: #{$!}"
       Rails.logger.warn "Backtrace:\n\t#{e.backtrace.join("\n\t")}"
       ws.send ({status: 500, message: 'error'}.to_json)
@@ -215,7 +245,7 @@ class EventBus
                 # channel and calls #push_events to actually dispatch the
                 # events to the client.
                 conn.wait_for_notify do |channel, pid, payload|
-                  @channel.push payload
+                  @channel.push payload.to_i
                 end
               end
             ensure