Bug 1155494 - Add a 'compressall' message flag. f=handyman r=billm (parts of this patch were written by handyman with r=billm)
authorBlake Kaplan <mrbkap@gmail.com>
Tue, 05 May 2015 16:20:42 -0700
changeset 272339 640dd49d323607f9514024039e66ab03d3c8daa8
parent 272338 e89d092c5f473fa9ab4eba57244da1a59d314a55
child 272340 44317690f61d7ba92a15aeb04e06b0dc71714363
push id4830
push userjlund@mozilla.com
push dateMon, 29 Jun 2015 20:18:48 +0000
treeherdermozilla-beta@4c2175bb0420 [default view] [failures only]
perfherder[talos] [build metrics] [platform microbench] (compared to previous push)
reviewersbillm, billm
bugs1155494
milestone40.0a1
first release with
nightly linux32
nightly linux64
nightly mac
nightly win32
nightly win64
last release without
nightly linux32
nightly linux64
nightly mac
nightly win32
nightly win64
Bug 1155494 - Add a 'compressall' message flag. f=handyman r=billm (parts of this patch were written by handyman with r=billm)
dom/ipc/PBrowser.ipdl
ipc/chromium/src/chrome/common/ipc_message.cc
ipc/chromium/src/chrome/common/ipc_message.h
ipc/glue/MessageChannel.cpp
ipc/ipdl/ipdl/lower.py
ipc/ipdl/ipdl/parser.py
ipc/ipdl/ipdl/type.py
--- a/dom/ipc/PBrowser.ipdl
+++ b/dom/ipc/PBrowser.ipdl
@@ -540,17 +540,17 @@ child:
          nullable PRenderFrame renderFrame,
          bool parentIsActive);
 
     LoadURL(nsCString uri, BrowserConfiguration config);
 
     CacheFileDescriptor(nsString path, FileDescriptor fd);
 
     UpdateDimensions(IntRect rect, ScreenIntSize size, ScreenOrientation orientation,
-                     LayoutDeviceIntPoint chromeDisp) compress;
+                     LayoutDeviceIntPoint chromeDisp) compressall;
 
     UpdateFrame(FrameMetrics frame);
 
     // The following methods correspond to functions on the GeckoContentController
     // interface in gfx/layers/apz/public/GeckoContentController.h. Refer to documentation
     // in that file for these functions.
     RequestFlingSnap(ViewID aScrollID, CSSPoint aDestination);
     AcknowledgeScrollUpdate(ViewID aScrollId, uint32_t aScrollGeneration);
--- a/ipc/chromium/src/chrome/common/ipc_message.cc
+++ b/ipc/chromium/src/chrome/common/ipc_message.cc
@@ -44,16 +44,18 @@ Message::Message()
 Message::Message(int32_t routing_id, msgid_t type, PriorityValue priority,
                  MessageCompression compression, const char* const name)
     : Pickle(sizeof(Header)) {
   header()->routing = routing_id;
   header()->type = type;
   header()->flags = priority;
   if (compression == COMPRESSION_ENABLED)
     header()->flags |= COMPRESS_BIT;
+  else if (compression == COMPRESSION_ALL)
+    header()->flags |= COMPRESSALL_BIT;
 #if defined(OS_POSIX)
   header()->num_fds = 0;
 #endif
   header()->interrupt_remote_stack_depth_guess = static_cast<uint32_t>(-1);
   header()->interrupt_local_stack_depth = static_cast<uint32_t>(-1);
   header()->seqno = 0;
 #if defined(OS_MACOSX)
   header()->cookie = 0;
--- a/ipc/chromium/src/chrome/common/ipc_message.h
+++ b/ipc/chromium/src/chrome/common/ipc_message.h
@@ -51,17 +51,18 @@ class Message : public Pickle {
   enum PriorityValue {
     PRIORITY_NORMAL = 1,
     PRIORITY_HIGH = 2,
     PRIORITY_URGENT = 3
   };
 
   enum MessageCompression {
     COMPRESSION_NONE,
-    COMPRESSION_ENABLED
+    COMPRESSION_ENABLED,
+    COMPRESSION_ALL
   };
 
   virtual ~Message();
 
   Message();
 
   // Initialize a message with a user-defined type, priority value, and
   // destination WebView ID.
@@ -94,18 +95,22 @@ class Message : public Pickle {
   }
 
   // True if this is a synchronous message.
   bool is_interrupt() const {
     return (header()->flags & INTERRUPT_BIT) != 0;
   }
 
   // True if compression is enabled for this message.
-  bool compress() const {
-    return (header()->flags & COMPRESS_BIT) != 0;
+  MessageCompression compress_type() const {
+    return (header()->flags & COMPRESS_BIT) ?
+               COMPRESSION_ENABLED :
+               (header()->flags & COMPRESSALL_BIT) ?
+                   COMPRESSION_ALL :
+                   COMPRESSION_NONE;
   }
 
   // Set this on a reply to a synchronous message.
   void set_reply() {
     header()->flags |= REPLY_BIT;
   }
 
   bool is_reply() const {
@@ -280,16 +285,17 @@ class Message : public Pickle {
     SYNC_BIT        = 0x0004,
     REPLY_BIT       = 0x0008,
     REPLY_ERROR_BIT = 0x0010,
     UNBLOCK_BIT     = 0x0020,
     PUMPING_MSGS_BIT= 0x0040,
     HAS_SENT_TIME_BIT = 0x0080,
     INTERRUPT_BIT   = 0x0100,
     COMPRESS_BIT    = 0x0200,
+    COMPRESSALL_BIT = 0x0400,
   };
 
   struct Header : Pickle::Header {
     int32_t routing;  // ID of the view that this message is destined for
     msgid_t type;   // specifies the user-defined message type
     uint32_t flags;   // specifies control flags for the message
 #if defined(OS_POSIX)
     uint32_t num_fds; // the number of descriptors included with this message
--- a/ipc/glue/MessageChannel.cpp
+++ b/ipc/glue/MessageChannel.cpp
@@ -607,16 +607,29 @@ MessageChannel::ShouldDeferMessage(const
     //
     // Deferring in the parent only sort of breaks message ordering. When the
     // child's message comes in, we can pretend the child hasn't quite
     // finished sending it yet. Since the message is sync, we know that the
     // child hasn't moved on yet.
     return mSide == ParentSide && aMsg.transaction_id() != mCurrentTransaction;
 }
 
+// Predicate that is true for messages that should be consolidated if 'compress' is set.
+class MatchingKinds {
+    typedef IPC::Message Message;
+    Message::msgid_t mType;
+    int32_t mRoutingId;
+public:
+    MatchingKinds(Message::msgid_t aType, int32_t aRoutingId) :
+        mType(aType), mRoutingId(aRoutingId) {}
+    bool operator()(const Message &msg) {
+        return msg.type() == mType && msg.routing_id() == mRoutingId;
+    }
+};
+
 void
 MessageChannel::OnMessageReceivedFromLink(const Message& aMsg)
 {
     AssertLinkThread();
     mMonitor->AssertCurrentThreadOwns();
 
     if (MaybeInterceptSpecialIOMessage(aMsg))
         return;
@@ -648,27 +661,45 @@ MessageChannel::OnMessageReceivedFromLin
         }
 
         mRecvd = new Message(aMsg);
         NotifyWorkerThread();
         return;
     }
 
     // Prioritized messages cannot be compressed.
-    MOZ_ASSERT(!aMsg.compress() || aMsg.priority() == IPC::Message::PRIORITY_NORMAL);
+    MOZ_ASSERT_IF(aMsg.compress_type() != IPC::Message::COMPRESSION_NONE,
+                  aMsg.priority() == IPC::Message::PRIORITY_NORMAL);
 
-    bool compress = (aMsg.compress() && !mPending.empty() &&
-                     mPending.back().type() == aMsg.type() &&
-                     mPending.back().routing_id() == aMsg.routing_id());
-    if (compress) {
-        // This message type has compression enabled, and the back of the
-        // queue was the same message type and routed to the same destination.
-        // Replace it with the newer message.
-        MOZ_ASSERT(mPending.back().compress());
-        mPending.pop_back();
+    bool compress = false;
+    if (aMsg.compress_type() == IPC::Message::COMPRESSION_ENABLED) {
+        compress = (!mPending.empty() &&
+                    mPending.back().type() == aMsg.type() &&
+                    mPending.back().routing_id() == aMsg.routing_id());
+        if (compress) {
+            // This message type has compression enabled, and the back of the
+            // queue was the same message type and routed to the same destination.
+            // Replace it with the newer message.
+            MOZ_ASSERT(mPending.back().compress_type() ==
+                       IPC::Message::COMPRESSION_ENABLED);
+            mPending.pop_back();
+        }
+    } else if (aMsg.compress_type() == IPC::Message::COMPRESSION_ALL) {
+        // Check the message queue for another message with this type/destination.
+        auto it = std::find_if(mPending.rbegin(), mPending.rend(),
+                               MatchingKinds(aMsg.type(), aMsg.routing_id()));
+        if (it != mPending.rend()) {
+            // This message type has compression enabled, and the queue holds
+            // a message with the same message type and routed to the same destination.
+            // Erase it.  Note that, since we always compress these redundancies, There Can
+            // Be Only One.
+            compress = true;
+            MOZ_ASSERT((*it).compress_type() == IPC::Message::COMPRESSION_ALL);
+            mPending.erase((++it).base());
+        }
     }
 
     bool shouldWakeUp = AwaitingInterruptReply() ||
                         (AwaitingSyncReply() && !ShouldDeferMessage(aMsg)) ||
                         AwaitingIncomingMessage();
 
     // There are three cases we're concerned about, relating to the state of the
     // main thread:
--- a/ipc/ipdl/ipdl/lower.py
+++ b/ipc/ipdl/ipdl/lower.py
@@ -1855,18 +1855,21 @@ def _generateMessageClass(clsname, msgid
     cls = Class(name=clsname, inherits=[ Inherit(Type('IPC::Message')) ])
     cls.addstmt(Label.PUBLIC)
 
     idenum = TypeEnum()
     idenum.addId('ID', msgid)
     cls.addstmt(StmtDecl(Decl(idenum, '')))
 
     # make the message constructor
-    if compress:
+    if compress == 'compress':
         compression = ExprVar('COMPRESSION_ENABLED')
+    elif compress:
+        assert compress == 'compressall'
+        compression = ExprVar('COMPRESSION_ALL')
     else:
         compression = ExprVar('COMPRESSION_NONE')
     if priority == ipdl.ast.NORMAL_PRIORITY:
         priorityEnum = 'IPC::Message::PRIORITY_NORMAL'
     elif priority == ipdl.ast.HIGH_PRIORITY:
         priorityEnum = 'IPC::Message::PRIORITY_HIGH'
     else:
         assert priority == ipdl.ast.URGENT_PRIORITY
--- a/ipc/ipdl/ipdl/parser.py
+++ b/ipc/ipdl/ipdl/parser.py
@@ -118,16 +118,17 @@ reserved = set((
         'as',
         'async',
         'both',
         'bridges',
         'call',
         'child',
         'class',
         'compress',
+        'compressall',
         '__delete__',
         'delete',                       # reserve 'delete' to prevent its use
         'from',
         'goto',
         'high',
         'include',
         'intr',
         'manager',
@@ -541,22 +542,27 @@ def p_MessageOutParams(p):
     """MessageOutParams : RETURNS '(' ParamList ')'
                         | """
     if 1 == len(p):
         p[0] = [ ]
     else:
         p[0] = p[3]
 
 def p_OptionalMessageCompress(p):
-    """OptionalMessageCompress : COMPRESS
+    """OptionalMessageCompress : MessageCompress
                                | """
     if 1 == len(p):
         p[0] = ''
     else:
-        p[0] = 'compress'
+        p[0] = p[1]
+
+def p_MessageCompress(p):
+    """MessageCompress : COMPRESS
+                       | COMPRESSALL"""
+    p[0] = p[1]
 
 ##--------------------
 ## State machine
 
 def p_TransitionStmtsOpt(p):
     """TransitionStmtsOpt : TransitionStmt TransitionStmtsOpt
                           |"""
     if 1 == len(p):
--- a/ipc/ipdl/ipdl/type.py
+++ b/ipc/ipdl/ipdl/type.py
@@ -1096,17 +1096,17 @@ class GatherDecls(TcheckVisitor):
             cdtype = self.currentProtocolDecl.type
 
 
         # enter message scope
         self.symtab.enterScope(md)
 
         msgtype = MessageType(md.priority, md.sendSemantics, md.direction,
                               ctor=isctor, dtor=isdtor, cdtype=cdtype,
-                              compress=(md.compress == 'compress'))
+                              compress=md.compress)
 
         # replace inparam Param nodes with proper Decls
         def paramToDecl(param):
             ptname = param.typespec.basename()
             ploc = param.typespec.loc
 
             ptdecl = self.symtab.lookup(ptname)
             if ptdecl is None: