Allow MessageFilters to restrict listening to specific message classes

ChannelProxy currently offers messages to all member MessageFilters.  It turns
out that a good portion of the most common message types will never be filtered,
making the O(N) filter walk an unnecessary affair.  To prevent this, allow
MessageFilters to indicate which (if any) subset of message classes they may
filter, allowing the ChannelProxy to refine the list of filters that are offered
a particular message.  This saves ~35us per message received on the browser IO
thread for a typical Android device.

BUG=340881
TBR=asargent@chromium.org

Review URL: https://codereview.chromium.org/142923005

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@251622 0039d316-1c4b-4281-b951-d872f2087c98

4 files changed, 383 insertions, 6 deletions

diff --git a/ipc/ipc.gyp b/ipc/ipc.gyp
index 95c6fd12..e546a99 100644
--- a/ipc/ipc.gyp
+++ b/ipc/ipc.gyp
@@ -46,6 +46,7 @@
       'sources': [
         'file_descriptor_set_posix_unittest.cc',
         'ipc_channel_posix_unittest.cc',
+        'ipc_channel_proxy_unittest.cc',
         'ipc_channel_unittest.cc',
         'ipc_fuzzing_tests.cc',
         'ipc_message_unittest.cc',
diff --git a/ipc/ipc_channel_proxy.cc b/ipc/ipc_channel_proxy.cc
index 3a0cf30..776b423 100644
--- a/ipc/ipc_channel_proxy.cc
+++ b/ipc/ipc_channel_proxy.cc
@@ -14,12 +14,96 @@
 #include "ipc/ipc_listener.h"
 #include "ipc/ipc_logging.h"
 #include "ipc/ipc_message_macros.h"
+#include "ipc/ipc_message_start.h"
 #include "ipc/ipc_message_utils.h"
 
 namespace IPC {
 
 //------------------------------------------------------------------------------
 
+class ChannelProxy::Context::MessageFilterRouter {
+ public:
+  typedef std::vector<MessageFilter*> MessageFilters;
+
+  MessageFilterRouter() {}
+  ~MessageFilterRouter() {}
+
+  void AddFilter(MessageFilter* filter) {
+    // Determine if the filter should be applied to all messages, or only
+    // messages of a certain class.
+    std::vector<uint32> supported_message_classes;
+    if (filter->GetSupportedMessageClasses(&supported_message_classes)) {
+      DCHECK(!supported_message_classes.empty());
+      for (size_t i = 0; i < supported_message_classes.size(); ++i) {
+        DCHECK(ValidMessageClass(supported_message_classes[i]));
+        message_class_filters_[supported_message_classes[i]].push_back(filter);
+      }
+    } else {
+      global_filters_.push_back(filter);
+    }
+  }
+
+  void RemoveFilter(MessageFilter* filter) {
+    if (RemoveFilter(global_filters_, filter))
+      return;
+
+    for (size_t i = 0; i < arraysize(message_class_filters_); ++i)
+      RemoveFilter(message_class_filters_[i], filter);
+  }
+
+  bool TryFilters(const Message& message) {
+    if (TryFilters(global_filters_, message))
+      return true;
+
+    const int message_class = IPC_MESSAGE_CLASS(message);
+    if (!ValidMessageClass(message_class))
+      return false;
+
+    return TryFilters(message_class_filters_[message_class], message);
+  }
+
+  void Clear() {
+    global_filters_.clear();
+    for (size_t i = 0; i < arraysize(message_class_filters_); ++i)
+      message_class_filters_[i].clear();
+  }
+
+ private:
+  static bool TryFilters(MessageFilters& filters, const IPC::Message& message) {
+    for (size_t i = 0; i < filters.size(); ++i) {
+      if (filters[i]->OnMessageReceived(message)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  static bool RemoveFilter(MessageFilters& filters, MessageFilter* filter) {
+    MessageFilters::iterator it =
+        std::find(filters.begin(), filters.end(), filter);
+    if (it == filters.end())
+      return false;
+
+    filters.erase(it);
+    return true;
+  }
+
+  static bool ValidMessageClass(int message_class) {
+    return message_class >= 0 && message_class < LastIPCMsgStart;
+  }
+
+  // List of global and selective filters; a given filter will exist in either
+  // |message_global_filters_| OR |message_class_filters_|, but not both.
+  // Note that |message_global_filters_| will be given first offering of any
+  // given message.  It's the filter implementer and installer's
+  // responsibility to ensure that a filter is either global or selective to
+  // ensure proper message filtering order.
+  MessageFilters global_filters_;
+  MessageFilters message_class_filters_[LastIPCMsgStart];
+};
+
+//------------------------------------------------------------------------------
+
 ChannelProxy::MessageFilter::MessageFilter() {}
 
 void ChannelProxy::MessageFilter::OnFilterAdded(Channel* channel) {}
@@ -36,6 +120,11 @@ bool ChannelProxy::MessageFilter::OnMessageReceived(const Message& message) {
   return false;
 }
 
+bool ChannelProxy::MessageFilter::GetSupportedMessageClasses(
+    std::vector<uint32>* /*supported_message_classes*/) const {
+  return false;
+}
+
 ChannelProxy::MessageFilter::~MessageFilter() {}
 
 //------------------------------------------------------------------------------
@@ -46,6 +135,7 @@ ChannelProxy::Context::Context(Listener* listener,
       listener_(listener),
       ipc_task_runner_(ipc_task_runner),
       channel_connected_called_(false),
+      message_filter_router_(new MessageFilterRouter()),
       peer_pid_(base::kNullProcessId) {
   DCHECK(ipc_task_runner_.get());
 }
@@ -65,20 +155,19 @@ void ChannelProxy::Context::CreateChannel(const IPC::ChannelHandle& handle,
 }
 
 bool ChannelProxy::Context::TryFilters(const Message& message) {
+  DCHECK(message_filter_router_);
 #ifdef IPC_MESSAGE_LOG_ENABLED
   Logging* logger = Logging::GetInstance();
   if (logger->Enabled())
     logger->OnPreDispatchMessage(message);
 #endif
 
-  for (size_t i = 0; i < filters_.size(); ++i) {
-    if (filters_[i]->OnMessageReceived(message)) {
+  if (message_filter_router_->TryFilters(message)) {
 #ifdef IPC_MESSAGE_LOG_ENABLED
-      if (logger->Enabled())
-        logger->OnPostDispatchMessage(message, channel_id_);
+    if (logger->Enabled())
+      logger->OnPostDispatchMessage(message, channel_id_);
 #endif
-      return true;
-    }
+    return true;
   }
   return false;
 }
@@ -157,6 +246,7 @@ void ChannelProxy::Context::OnChannelClosed() {
 
   // We don't need the filters anymore.
   filters_.clear();
+  message_filter_router_->Clear();
 
   channel_.reset();
 
@@ -190,6 +280,8 @@ void ChannelProxy::Context::OnAddFilter() {
   for (size_t i = 0; i < new_filters.size(); ++i) {
     filters_.push_back(new_filters[i]);
 
+    message_filter_router_->AddFilter(new_filters[i].get());
+
     // If the channel has already been created, then we need to send this
     // message so that the filter gets access to the Channel.
     if (channel_.get())
@@ -205,6 +297,8 @@ void ChannelProxy::Context::OnRemoveFilter(MessageFilter* filter) {
   if (!channel_.get())
     return;  // The filters have already been deleted.
 
+  message_filter_router_->RemoveFilter(filter);
+
   for (size_t i = 0; i < filters_.size(); ++i) {
     if (filters_[i].get() == filter) {
       filter->OnFilterRemoved();
diff --git a/ipc/ipc_channel_proxy.h b/ipc/ipc_channel_proxy.h
index 1f5ecf4..dca8c97 100644
--- a/ipc/ipc_channel_proxy.h
+++ b/ipc/ipc_channel_proxy.h
@@ -88,6 +88,13 @@ class IPC_EXPORT ChannelProxy : public Sender, public base::NonThreadSafe {
     // the message be handled in the default way.
     virtual bool OnMessageReceived(const Message& message);
 
+    // Called to query the Message classes supported by the filter.  Return
+    // false to indicate that all message types should reach the filter, or true
+    // if the resulting contents of |supported_message_classes| may be used to
+    // selectively offer messages of a particular class to the filter.
+    virtual bool GetSupportedMessageClasses(
+        std::vector<uint32>* supported_message_classes) const;
+
    protected:
     virtual ~MessageFilter();
 
@@ -230,6 +237,11 @@ class IPC_EXPORT ChannelProxy : public Sender, public base::NonThreadSafe {
     std::string channel_id_;
     bool channel_connected_called_;
 
+    // Routes a given message to a proper subset of |filters_|, depending
+    // on which message classes a filter might support.
+    class MessageFilterRouter;
+    scoped_ptr<MessageFilterRouter> message_filter_router_;
+
     // Holds filters between the AddFilter call on the listerner thread and the
     // IPC thread when they're added to filters_.
     std::vector<scoped_refptr<MessageFilter> > pending_filters_;
diff --git a/ipc/ipc_channel_proxy_unittest.cc b/ipc/ipc_channel_proxy_unittest.cc
new file mode 100644
index 0000000..431f410
--- /dev/null
+++ b/ipc/ipc_channel_proxy_unittest.cc
@@ -0,0 +1,270 @@
+// Copyright 2014 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 "build/build_config.h"
+
+#include "base/message_loop/message_loop.h"
+#include "base/pickle.h"
+#include "base/threading/thread.h"
+#include "ipc/ipc_message.h"
+#include "ipc/ipc_message_macros.h"
+#include "ipc/ipc_test_base.h"
+
+namespace {
+
+#if defined(IPC_MESSAGE_START)
+#undef IPC_MESSAGE_START
+#endif
+
+enum Command {
+  SEND,
+  QUIT
+};
+
+static void Send(IPC::Sender* sender,
+                 int message_class,
+                 Command command) {
+  const int IPC_MESSAGE_START = message_class;
+  IPC::Message* message = new IPC::Message(0,
+                                           IPC_MESSAGE_ID(),
+                                           IPC::Message::PRIORITY_NORMAL);
+  message->WriteInt(command);
+  sender->Send(message);
+}
+
+class QuitListener : public IPC::Listener {
+ public:
+  QuitListener() {}
+  virtual ~QuitListener() {}
+
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE {
+    PickleIterator iter(message);
+
+    int command = SEND;
+    EXPECT_TRUE(iter.ReadInt(&command));
+    if (command == QUIT)
+      base::MessageLoop::current()->QuitWhenIdle();
+
+    return true;
+  }
+};
+
+class ChannelReflectorListener : public IPC::Listener {
+ public:
+  ChannelReflectorListener() : channel_(NULL) {}
+  virtual ~ChannelReflectorListener() {}
+
+  void Init(IPC::Channel* channel) {
+    DCHECK(!channel_);
+    channel_ = channel;
+  }
+
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE {
+    CHECK(channel_);
+
+    PickleIterator iter(message);
+
+    int command = SEND;
+    EXPECT_TRUE(iter.ReadInt(&command));
+    if (command == QUIT) {
+      channel_->Send(new IPC::Message(message));
+      base::MessageLoop::current()->QuitWhenIdle();
+      return true;
+    }
+
+    channel_->Send(new IPC::Message(message));
+    return true;
+  }
+
+ private:
+  IPC::Channel* channel_;
+};
+
+class MessageCountFilter : public IPC::ChannelProxy::MessageFilter {
+ public:
+  MessageCountFilter()
+      : messages_received_(0),
+        supported_message_class_(0),
+        is_global_filter_(true),
+        filter_removed_(false),
+        message_filtering_enabled_(false) {}
+
+  MessageCountFilter(uint32 supported_message_class)
+      : messages_received_(0),
+        supported_message_class_(supported_message_class),
+        is_global_filter_(false),
+        filter_removed_(false),
+        message_filtering_enabled_(false) {}
+
+  virtual void OnFilterRemoved() OVERRIDE {
+    EXPECT_FALSE(filter_removed_);
+    filter_removed_ = true;
+  }
+
+  virtual bool OnMessageReceived(const IPC::Message& message) OVERRIDE {
+    if (!is_global_filter_) {
+      EXPECT_EQ(supported_message_class_, IPC_MESSAGE_CLASS(message));
+    }
+    ++messages_received_;
+    return message_filtering_enabled_;
+  }
+
+  virtual bool GetSupportedMessageClasses(
+      std::vector<uint32>* supported_message_classes) const OVERRIDE {
+    if (is_global_filter_)
+      return false;
+    supported_message_classes->push_back(supported_message_class_);
+    return true;
+  }
+
+  void set_message_filtering_enabled(bool enabled) {
+    message_filtering_enabled_ = enabled;
+  }
+
+  size_t messages_received() const { return messages_received_; }
+  bool filter_removed() const { return filter_removed_; }
+
+ private:
+  virtual ~MessageCountFilter() {}
+
+  size_t messages_received_;
+  uint32 supported_message_class_;
+  bool is_global_filter_;
+  bool filter_removed_;
+  bool message_filtering_enabled_;
+};
+
+class IPCChannelProxyTest : public IPCTestBase {
+ public:
+  IPCChannelProxyTest() {}
+  virtual ~IPCChannelProxyTest() {}
+
+  virtual void SetUp() OVERRIDE {
+    IPCTestBase::SetUp();
+
+    Init("ChannelProxyClient");
+
+    thread_.reset(new base::Thread("ChannelProxyTestServerThread"));
+    base::Thread::Options options;
+    options.message_loop_type = base::MessageLoop::TYPE_IO;
+    thread_->StartWithOptions(options);
+
+    listener_.reset(new QuitListener());
+    CreateChannelProxy(listener_.get(), thread_->message_loop_proxy().get());
+
+    ASSERT_TRUE(StartClient());
+  }
+
+  virtual void TearDown() {
+    DestroyChannelProxy();
+    thread_.reset();
+    listener_.reset();
+    IPCTestBase::TearDown();
+  }
+
+  void SendQuitMessageAndWaitForIdle() {
+    Send(sender(), -1, QUIT);
+    base::MessageLoop::current()->Run();
+    EXPECT_TRUE(WaitForClientShutdown());
+  }
+
+ private:
+  scoped_ptr<base::Thread> thread_;
+  scoped_ptr<QuitListener> listener_;
+};
+
+TEST_F(IPCChannelProxyTest, MessageClassFilters) {
+  // Construct a filter per message class.
+  std::vector<scoped_refptr<MessageCountFilter> > class_filters;
+  for (uint32 i = 0; i < LastIPCMsgStart; ++i) {
+    class_filters.push_back(make_scoped_refptr(
+        new MessageCountFilter(i)));
+    channel_proxy()->AddFilter(class_filters.back().get());
+  }
+
+  // Send a message for each class; each filter should receive just one message.
+  for (uint32 i = 0; i < LastIPCMsgStart; ++i)
+    Send(sender(), i, SEND);
+
+  // Send some messages not assigned to a specific or valid message class.
+  Send(sender(), -1, SEND);
+  Send(sender(), LastIPCMsgStart, SEND);
+  Send(sender(), LastIPCMsgStart + 1, SEND);
+
+  // Each filter should have received just the one sent message of the
+  // corresponding class.
+  SendQuitMessageAndWaitForIdle();
+  for (size_t i = 0; i < class_filters.size(); ++i)
+    EXPECT_EQ(1U, class_filters[i]->messages_received());
+}
+
+TEST_F(IPCChannelProxyTest, GlobalAndMessageClassFilters) {
+  // Add a class and global filter.
+  const int kMessageClass = 7;
+  scoped_refptr<MessageCountFilter> class_filter(
+      new MessageCountFilter(kMessageClass));
+  class_filter->set_message_filtering_enabled(false);
+  channel_proxy()->AddFilter(class_filter.get());
+
+  scoped_refptr<MessageCountFilter> global_filter(new MessageCountFilter());
+  global_filter->set_message_filtering_enabled(false);
+  channel_proxy()->AddFilter(global_filter.get());
+
+  // A message  of class |kMessageClass| should be seen by both the global
+  // filter and |kMessageClass|-specific filter.
+  Send(sender(), kMessageClass, SEND);
+
+  // A message of a different class should be seen only by the global filter.
+  Send(sender(), kMessageClass + 1, SEND);
+
+  // Flush all messages.
+  SendQuitMessageAndWaitForIdle();
+
+  // The class filter should have received only the class-specific message.
+  EXPECT_EQ(1U, class_filter->messages_received());
+
+  // The global filter should have received both SEND messages, as well as the
+  // final QUIT message.
+  EXPECT_EQ(3U, global_filter->messages_received());
+}
+
+TEST_F(IPCChannelProxyTest, FilterRemoval) {
+  // Add a class and global filter.
+  const int kMessageClass = 7;
+  scoped_refptr<MessageCountFilter> class_filter(
+      new MessageCountFilter(kMessageClass));
+  scoped_refptr<MessageCountFilter> global_filter(new MessageCountFilter());
+
+  // Add and remove both types of filters.
+  channel_proxy()->AddFilter(class_filter.get());
+  channel_proxy()->AddFilter(global_filter.get());
+  channel_proxy()->RemoveFilter(global_filter.get());
+  channel_proxy()->RemoveFilter(class_filter.get());
+
+  // Send some messages; they should not be seen by either filter.
+  Send(sender(), 0, SEND);
+  Send(sender(), kMessageClass, SEND);
+
+  // Ensure that the filters were removed and did not receive any messages.
+  SendQuitMessageAndWaitForIdle();
+  EXPECT_TRUE(global_filter->filter_removed());
+  EXPECT_TRUE(class_filter->filter_removed());
+  EXPECT_EQ(0U, class_filter->messages_received());
+  EXPECT_EQ(0U, global_filter->messages_received());
+}
+
+MULTIPROCESS_IPC_TEST_CLIENT_MAIN(ChannelProxyClient) {
+  base::MessageLoopForIO main_message_loop;
+  ChannelReflectorListener listener;
+  IPC::Channel channel(IPCTestBase::GetChannelName("ChannelProxyClient"),
+                       IPC::Channel::MODE_CLIENT,
+                       &listener);
+  CHECK(channel.Connect());
+  listener.Init(&channel);
+
+  base::MessageLoop::current()->Run();
+  return 0;
+}
+
+}  // namespace