Provided batched input delivery with a BufferedInputRouter

content/browser/renderer_host/input/input_queue.cc

+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "content/browser/renderer_host/input/input_queue.h"
+
+#include "base/bind.h"
+#include "content/browser/renderer_host/input/input_ack_observer.h"
+#include "content/browser/renderer_host/input/input_queue_client.h"
+#include "content/common/input/event_packet.h"
+#include "content/common/input/input_event.h"
+
+namespace content {
+namespace {
+
+// TODO(jdduke):
+//   INPUT_EVENT_NEEDS_ACK_AND_HAS_FOLLOWUP: Delay flush, hold future packets.
+//   INPUT_EVENT_NEEDS_ACK: Signal flush immediately, dispatch ack in-order.
+//   INPUT_EVENT_ONE_WAY: Signal flush immediately, ignore ack.
+InputEventType RestrictingEventType(const EventPacket& packet) {
+  InputEventType restricting_type = INPUT_EVENT_INVALID;
+  for (std::vector<InputEvent>::const_iterator event_it = packet.events.begin();
+       event_it != packet.events.end();
+       ++event_it) {
+    if (event_it->type > restricting_type)
+      restricting_type = event_it->type;
+  }
+  return restricting_type;
+}
+
+bool HasFollowupEvents(const EventPacket& packet) {
+  return RestrictingEventType(packet) == INPUT_EVENT_NEEDS_ACK_AND_HAS_FOLLOWUP;
+}
+
+bool HasAckEvents(const EventPacket& packet) {
+  return RestrictingEventType(packet) >= INPUT_EVENT_NEEDS_ACK;

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

Please avoid <=/>= on enums.  Please delete RestrictingEventType() and just use
std::find() for each enum type you're interested in.  There are only 4 anyway.

Or delete them since they seem unused now?

[jdduke (slow), 2013/08/15 23:22:26]

Yeah this was a speculative addition.  I'll remove now, and we can add something
similar in the future should the need arise. 

+}
+
+bool IsOneWay(const EventPacket& packet) {
+  return RestrictingEventType(packet) <= INPUT_EVENT_ONE_WAY;
+}
+
+}
+
+// Utilty class for storing input events and their ack observers, and

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

typo: "Utilty"

[jdduke (slow), 2013/08/15 23:22:26]

Done.

+// dispatching input event responses.
+class InputQueue::EventFrame {
+ public:
+  EventFrame() : weak_factory_(this) {}
+  virtual ~EventFrame() {}
+
+  void DeliverTo(InputQueueClient* client, int64 id) {
+    packet_.id = id;
+    client->Deliver(packet_);
+  }
+
+  void QueueEvent(const InputEvent& event, InputAckObserver* observer) {
+    packet_.events.push_back(event);
+    observers_.push_back(observer);
+  }
+
+  // Validate |acked_packet| and dispatch the acked events upon successful
+  // validation.  Returns |false| if the validation failed.
+  AckResult ValidateAndDispatchAcks(const EventPacket& acked_packet,
+                                    InputQueueClient* client) {
+    if (!Validate(acked_packet))
+      return ACK_INVALID;
+
+    std::vector<InputEvent> events;
+    std::swap(events, packet_.events);
+
+    std::vector<InputAckObserver*> observers;
+    std::swap(observers, observers_);
+
+    // The EventFrame could be deleted as a result of the event ack; use a local
+    // weak ref to ensure proper shutdown.
+    base::WeakPtr<EventFrame> weak_ref_this = weak_factory_.GetWeakPtr();
+
+    const std::vector<InputEvent>& acked_events = acked_packet.events;
+    for (size_t event_idx = 0; event_idx < acked_events.size(); ++event_idx) {
+      const InputEvent& acked_event = acked_events[event_idx];
+
+      // All events marked undeliverable should be re-enqueued.
+      switch (acked_event.state) {
+        case INPUT_EVENT_IMPL_THREAD_COULD_NOT_DELIVER:
+        case INPUT_EVENT_MAIN_THREAD_COULD_NOT_DELIVER:
+          QueueEvent(events[event_idx], observers[event_idx]);
+          continue;
+        default:
+          break;
+      }
+
+      // Update the original event's state, and use it for the ack.
+      InputEvent& event = events[event_idx];
+      event.state = acked_event.state;
+
+      std::vector<InputEvent> followup_events = client->OnInputEventAck(event);
+
+      // Exit immediately if the client ack triggered an early shutdown.
+      // TODO(jdduke): Look into making ack-triggered shutdown async.

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

Please file a bug for this.

[jdduke (slow), 2013/08/15 23:53:30]

Done.

+      if (!weak_ref_this.get())
+        return ACK_SHUTDOWN;
+
+      for (std::vector<InputEvent>::iterator iter = followup_events.begin();
+           iter != followup_events.end();
+           ++iter) {
+        QueueEvent(*iter, NULL);
+      }
+
+      if (observers[event_idx])
+        observers[event_idx]->OnInputEventAck(event);

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

Please DCHECK the size of "observers".

[jdduke (slow), 2013/08/15 23:22:26]

Done.

+    }
+    return ACK_OK;
+  };
+
+  bool IsEmpty() const { return packet_.events.empty(); }
+  size_t Size() const { return packet_.events.size(); }
+
+protected:
+
+  // Perform a sanity check of |acked_packet| against the queued events.
+  // The packet should contain the same number of events, and all events and
+  // the event messages should be of the same general type.
+  bool Validate(const EventPacket& acked_packet) {
+    if (acked_packet.id != packet_.id)
+      return false;
+
+    const std::vector<InputEvent>& source_events = packet_.events;
+    const std::vector<InputEvent>& acked_events = acked_packet.events;
+    if (source_events.size() != acked_events.size())
+      return false;
+
+    for (size_t event_idx = 0; event_idx < acked_events.size(); ++event_idx) {
+      const InputEvent& source_event = source_events[event_idx];
+      const InputEvent& acked_event = acked_events[event_idx];
+
+      if (acked_event.type != source_event.type)
+        return false;
+
+      if (acked_event.message.type() != source_event.message.type())
+        return false;
+    }
+
+    return true;
+  }
+
+private:
+  EventPacket packet_;
+  std::vector<InputAckObserver*> observers_;
+  base::WeakPtrFactory<EventFrame> weak_factory_;
+};
+
+InputQueue::InputQueue(InputQueueClient* client)
+    : client_(client),
+      last_packet_id_(0),
+      flush_requested_(false),
+      flush_signalled_(false),
+      current_frame_(new EventFrame()),
+      pending_frame_(new EventFrame()) {
+  DCHECK(client_);
+}
+
+InputQueue::~InputQueue() {}
+
+void InputQueue::QueueEvent(const InputEvent& event,
+                            InputAckObserver* ack_observer) {
+  pending_frame_->QueueEvent(event, ack_observer);
+  RequestFlushIfNecessary();
+}
+
+void InputQueue::FlushEventsInCurrentFrame() {
+  // Ignore repeated flush attempts.
+  if (!flush_requested_)
+    return;
+
+  if (IsFlushing()) {
+    NOTREACHED() << "InputQueue::Flush() - A flush is already in-progress.";

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

Another place that should just have a DCHECK, not a return.

[jdduke (slow), 2013/08/15 23:22:26]

Done.

+    return;
+  }
+
+  flush_requested_ = false;
+  flush_signalled_ = false;
+  current_frame_.swap(pending_frame_);
+
+  TryFinishFlush();

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

Do you ever want the request-flush behavior of TryFlush here?  Isn't this
supposed to be called in reaction to a request?

I don't think TryFinishFlush has clear semantics and I'd like it to be just
FinishFlush and just have the "deliver" line.  Move the signal and request
operations to OnEventPacketAck.

[jdduke (slow), 2013/08/15 23:22:26]

Yeah, I added this to allow a bit of flexibility if we ever sent early
"DidFlush()" signals, but it has since been decided that we will never do that. 
I'll remove.

+}
+
+InputQueue::AckResult InputQueue::OnEventPacketAck(
+    const EventPacket& acked_packet) {
+  if (!IsFlushing())
+    return ACK_UNEXPECTED;
+
+  AckResult ack_result =
+      current_frame_->ValidateAndDispatchAcks(acked_packet, client_);
+  if (ack_result == ACK_OK)
+    TryFinishFlush();
+
+  return ack_result;
+}
+
+size_t InputQueue::QueuedEventCount() const {
+  return current_frame_->Size() + pending_frame_->Size();
+}
+
+void InputQueue::TryFinishFlush() {
+  if (!IsFlushing()) {
+    SignalFlushedIfNecessary();
+    RequestFlushIfNecessary();
+    return;
+  }
+
+  // Provide a unique id for every delivered packet.
+  current_frame_->DeliverTo(client_, ++last_packet_id_);
+}
+
+void InputQueue::RequestFlushIfNecessary() {
+  if (flush_requested_)
+    return;
+
+  // No events to flush.
+  if (pending_frame_->IsEmpty())

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

Is it correct to check pending_frame_ here?  TryFinishFlush() is called after
the current_frame_.swap() above.

[jdduke (slow), 2013/08/15 23:22:26]

Well, I think I should clear up the expectations: A flush request should ONLY
occur for a non-empty pending_frame_.  So the TryFinishFlush() should never hit
this path.

TryFinishFlush probably makes it more confusing than it ought to be.  I'll
remove entirely.  Thanks.

+    return;
+
+  // We're still flushing; wait until it finishes before an additional request.
+  if (IsFlushing())
+    return;
+
+  flush_requested_ = true;
+  client_->SetNeedsFlush();
+}
+
+void InputQueue::SignalFlushedIfNecessary() {

[aelias_OOO_until_Jul13, 2013/08/15 00:52:15]

I don't think this method (nor the field flush_signalled_) is needed.  Just do
if (!IsFlushing()) client_->DidFlush() in OnEventPacketAck.

[jdduke (slow), 2013/08/15 23:22:26]

Right, deleting.

+  if (flush_signalled_)
+    return;
+
+  flush_signalled_ = true;
+  client_->DidFlush();
+}
+
+bool InputQueue::IsFlushing() const {
+  return !current_frame_->IsEmpty();
+}
+
+} // namespace content