IDEA-124556 MessageBus event processing should be faster with many child buses and (almost) no listeners

2026-09-27 10:03:11 +07:00 · 2014-05-01 19:15:25 +02:00
parent 77c7af8c09
commit db58c9b881
5 changed files with 171 additions and 24 deletions
@@ -2122,5 +2122,15 @@ public class ContainerUtil extends ContainerUtilRt {
    // IBM JDK provides correct version in java.version property, but not in java.runtime.version property
    return StringUtil.compareVersionNumbers(SystemInfo.JAVA_VERSION, "1.7") >= 0;
  }
+
+  public static <T extends Comparable<T>> int compareLexicographically(List<T> o1, List<T> o2) {
+    for (int i = 0; i < Math.min(o1.size(), o2.size()); i++) {
+      int result = o1.get(i).compareTo(o2.get(i));
+      if (result != 0) {
+        return result;
+      }
+    }
+    return o1.size() < o2.size() ? -1 : o1.size() == o2.size() ? 0 : 1;
+  }
 }

@@ -33,7 +33,7 @@ public class MessageBusConnectionImpl implements MessageBusConnection {
  private static final Logger LOG = Logger.getInstance("#com.intellij.util.messages.impl.MessageBusConnectionImpl");

  private final MessageBusImpl myBus;
-  @SuppressWarnings("SSBasedInspection") 
+  @SuppressWarnings("SSBasedInspection")
  private final ThreadLocal<Queue<Message>> myPendingMessages = MessageBusImpl.createThreadLocalQueue();

  private MessageHandler myDefaultHandler;
@@ -63,7 +63,7 @@ public class MessageBusConnectionImpl implements MessageBusConnection {
    }
    if (topic.getListenerClass().isInstance(myDefaultHandler)) {
      throw new IllegalStateException("Can't subscribe to the topic '" + topic +"'. Default handler has incompatible type - expected: '" +
-        topic.getListenerClass() + "', actual: '" + myDefaultHandler.getClass() + "'");
+                                      topic.getListenerClass() + "', actual: '" + myDefaultHandler.getClass() + "'");
    }

    subscribe(topic, (L)myDefaultHandler);
@@ -129,4 +129,8 @@ public class MessageBusConnectionImpl implements MessageBusConnection {
  public String toString() {
    return mySubscriptions.toString();
  }
+
+  MessageBusImpl getBus() {
+    return myBus;
+  }
 }
@@ -23,6 +23,7 @@ import com.intellij.openapi.Disposable;
 import com.intellij.openapi.diagnostic.Logger;
 import com.intellij.openapi.util.Disposer;
 import com.intellij.util.ConcurrencyUtil;
+import com.intellij.util.SmartList;
 import com.intellij.util.containers.ContainerUtil;
 import com.intellij.util.messages.MessageBus;
 import com.intellij.util.messages.MessageBusConnection;
@@ -33,20 +34,51 @@ import org.jetbrains.annotations.NotNull;
 import java.lang.reflect.InvocationHandler;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Queue;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ConcurrentMap;

 public class MessageBusImpl implements MessageBus {
  private static final Logger LOG = Logger.getInstance("#com.intellij.util.messages.impl.MessageBusImpl");
+  private static final Comparator<MessageBusImpl> MESSAGE_BUS_COMPARATOR = new Comparator<MessageBusImpl>() {
+    @Override
+    public int compare(MessageBusImpl bus1, MessageBusImpl bus2) {
+      return ContainerUtil.compareLexicographically(bus1.myOrder, bus2.myOrder);
+    }
+  };
  private final ThreadLocal<Queue<DeliveryJob>> myMessageQueue = createThreadLocalQueue();
+
+  /**
+   * Holds the counts of pending messages for all message buses in the hierarchy
+   * This field is null for non-root buses
+   * The map's keys are sorted by {@link #myOrder}
+   *
+   * Used to avoid traversing the whole hierarchy when there are no messages to be sent in most of it
+   */
+  private final ThreadLocal<SortedMap<MessageBusImpl, Integer>> myWaitingBuses;
+
+  /**
+   * Root's order is empty
+   * Child bus's order is its parent order plus one more element, an int that's bigger than that of all sibling buses that come before
+   * Sorting by these vectors lexicographically gives DFS order
+   */
+  private final List<Integer> myOrder;
+
  private final ConcurrentMap<Topic, Object> mySyncPublishers = new ConcurrentHashMap<Topic, Object>();
  private final ConcurrentMap<Topic, Object> myAsyncPublishers = new ConcurrentHashMap<Topic, Object>();
+
+  /**
+   * This bus's subscribers
+   */
  private final ConcurrentMap<Topic, List<MessageBusConnectionImpl>> mySubscribers =
    new ConcurrentHashMap<Topic, List<MessageBusConnectionImpl>>();
+
+  /**
+   * Caches subscribers for this bus and its children or parent, depending on the topic's broadcast policy
+   */
+  private final ConcurrentMap<Topic, List<MessageBusConnectionImpl>> mySubscriberCache =
+    new ConcurrentHashMap<Topic, List<MessageBusConnectionImpl>>();
  private final List<MessageBusImpl> myChildBuses = ContainerUtil.createLockFreeCopyOnWriteList();

  private static final Object NA = new Object();
@@ -66,8 +98,12 @@ public class MessageBusImpl implements MessageBus {
    myOwner = owner.toString();
    myParentBus = (MessageBusImpl)parentBus;
    if (myParentBus != null) {
-      myParentBus.notifyChildBusCreated(this);
+      myOrder = myParentBus.notifyChildBusCreated(this);
      LOG.assertTrue(myParentBus.myChildBuses.contains(this));
+      myWaitingBuses = null;
+    } else {
+      myOrder = Collections.emptyList();
+      myWaitingBuses = new ThreadLocal<SortedMap<MessageBusImpl, Integer>>();
    }
  }

@@ -76,13 +112,33 @@ public class MessageBusImpl implements MessageBus {
    return myParentBus;
  }

-  private void notifyChildBusCreated(final MessageBusImpl childBus) {
-    myChildBuses.add(childBus);
+  @NotNull
+  private MessageBusImpl getRootBus() {
+    return myParentBus != null ? myParentBus.getRootBus() : this;
+  }
+
+  private List<Integer> notifyChildBusCreated(final MessageBusImpl childBus) {
    LOG.assertTrue(childBus.myParentBus == this);
+
+    MessageBusImpl lastChild = myChildBuses.isEmpty() ? null : myChildBuses.get(myChildBuses.size() - 1);
+    myChildBuses.add(childBus);
+    getRootBus().clearSubscriberCache();
+
+    int lastChildIndex = lastChild == null ? 0 : lastChild.myOrder.get(lastChild.myOrder.size() - 1);
+    if (lastChildIndex == Integer.MAX_VALUE) {
+      LOG.error("Too many child buses");
+    }
+    List<Integer> childOrder = new ArrayList<Integer>(myOrder.size() + 1);
+    childOrder.addAll(myOrder);
+    childOrder.add(lastChildIndex + 1);
+    return childOrder;
  }

  private void notifyChildBusDisposed(final MessageBusImpl childBus) {
    boolean removed = myChildBuses.remove(childBus);
+    Map<MessageBusImpl, Integer> map = getRootBus().myWaitingBuses.get();
+    if (map != null) map.remove(childBus);
+    getRootBus().clearSubscriberCache();
    LOG.assertTrue(removed);
  }

@@ -170,6 +226,8 @@ public class MessageBusImpl implements MessageBus {
    if (myParentBus != null) {
      myParentBus.notifyChildBusDisposed(this);
      myParentBus = null;
+    } else {
+      myWaitingBuses.remove();
    }
    myDisposed = true;
  }
@@ -178,28 +236,58 @@ public class MessageBusImpl implements MessageBus {
    LOG.assertTrue(!myDisposed, "Already disposed");
  }

-  private void postMessage(Message message) {
-    checkNotDisposed();
-    final Topic topic = message.getTopic();
+  private void calcSubscribers(Topic topic, List<MessageBusConnectionImpl> result) {
    final List<MessageBusConnectionImpl> topicSubscribers = mySubscribers.get(topic);
    if (topicSubscribers != null) {
-      Queue<DeliveryJob> queue = myMessageQueue.get();
-      for (MessageBusConnectionImpl subscriber : topicSubscribers) {
-        queue.offer(new DeliveryJob(subscriber, message));
-        subscriber.scheduleMessageDelivery(message);
-      }
+      result.addAll(topicSubscribers);
    }

    Topic.BroadcastDirection direction = topic.getBroadcastDirection();

    if (direction == Topic.BroadcastDirection.TO_CHILDREN) {
      for (MessageBusImpl childBus : myChildBuses) {
-        childBus.postMessage(message);
+        childBus.calcSubscribers(topic, result);
      }
    }

    if (direction == Topic.BroadcastDirection.TO_PARENT && myParentBus != null) {
-      myParentBus.postMessage(message);
+      myParentBus.calcSubscribers(topic, result);
+    }
+  }
+
+  private void postMessage(Message message) {
+    checkNotDisposed();
+    final Topic topic = message.getTopic();
+    List<MessageBusConnectionImpl> topicSubscribers = mySubscriberCache.get(topic);
+    if (topicSubscribers == null) {
+      topicSubscribers = new SmartList<MessageBusConnectionImpl>();
+      calcSubscribers(topic, topicSubscribers);
+      mySubscriberCache.put(topic, topicSubscribers);
+    }
+    if (!topicSubscribers.isEmpty()) {
+      for (MessageBusConnectionImpl subscriber : topicSubscribers) {
+        subscriber.getBus().myMessageQueue.get().offer(new DeliveryJob(subscriber, message));
+        subscriber.getBus().notifyPendingJobChange(1);
+        subscriber.scheduleMessageDelivery(message);
+      }
+    }
+  }
+
+  private void notifyPendingJobChange(int delta) {
+    ThreadLocal<SortedMap<MessageBusImpl, Integer>> ref = getRootBus().myWaitingBuses;
+    SortedMap<MessageBusImpl, Integer> map = ref.get();
+    if (map == null) {
+      ref.set(map = new TreeMap<MessageBusImpl, Integer>(MESSAGE_BUS_COMPARATOR));
+    }
+    Integer countObject = map.get(this);
+    int count = countObject == null ? 0 : countObject;
+    int newCount = count + delta;
+    if (newCount > 0) {
+      map.put(this, newCount);
+    } else if (newCount == 0) {
+      map.remove(this);
+    } else {
+      LOG.error("Negative job count: " + this);
    }
  }

@@ -216,7 +304,15 @@ public class MessageBusImpl implements MessageBus {
      myParentBus.pumpMessages();
    }
    else {
-      doPumpMessages();
+      Map<MessageBusImpl, Integer> map = myWaitingBuses.get();
+      if (map != null) {
+        Set<MessageBusImpl> buses = map.keySet();
+        if (!buses.isEmpty()) {
+          for (MessageBusImpl bus : new ArrayList<MessageBusImpl>(buses)) {
+            bus.doPumpMessages();
+          }
+        }
+      }
    }
  }

@@ -225,14 +321,10 @@ public class MessageBusImpl implements MessageBus {
    do {
      DeliveryJob job = queue.poll();
      if (job == null) break;
+      notifyPendingJobChange(-1);
      job.connection.deliverMessage(job.message);
    }
    while (true);
-
-    for (MessageBusImpl childBus : myChildBuses) {
-      LOG.assertTrue(childBus.myParentBus == this);
-      childBus.doPumpMessages();
-    }
  }

  void notifyOnSubscription(final MessageBusConnectionImpl connection, final Topic topic) {
@@ -244,12 +336,21 @@ public class MessageBusImpl implements MessageBus {
    }

    topicSubscribers.add(connection);
+    getRootBus().clearSubscriberCache();
+  }
+
+  private void clearSubscriberCache() {
+    mySubscriberCache.clear();
+    for (MessageBusImpl bus : myChildBuses) {
+      bus.clearSubscriberCache();
+    }
  }

  void notifyConnectionTerminated(final MessageBusConnectionImpl connection) {
    for (List<MessageBusConnectionImpl> topicSubscribers : mySubscribers.values()) {
      topicSubscribers.remove(connection);
    }
+    getRootBus().clearSubscriberCache();
    if (myDisposed) return;

    final Iterator<DeliveryJob> i = myMessageQueue.get().iterator();
@@ -257,6 +358,7 @@ public class MessageBusImpl implements MessageBus {
      final DeliveryJob job = i.next();
      if (job.connection == connection) {
        i.remove();
+        notifyPendingJobChange(-1);
      }
    }
  }
@@ -265,6 +367,7 @@ public class MessageBusImpl implements MessageBus {
    checkNotDisposed();
    final DeliveryJob job = myMessageQueue.get().poll();
    if (job == null) return;
+    notifyPendingJobChange(-1);
    job.connection.deliverMessage(job.message);
  }

@@ -20,6 +20,8 @@
 package com.intellij.util.messages;

 import com.intellij.openapi.util.text.StringUtil;
+import com.intellij.testFramework.PlatformTestUtil;
+import com.intellij.util.messages.impl.MessageBusImpl;
 import junit.framework.TestCase;

 import java.util.ArrayList;
@@ -196,6 +198,33 @@ public class MessageBusTest extends TestCase {
                 "inside:t11:done",
                 "C2T1Handler:t12");
  }
+
+  public void testPostingPerformanceWithLowListenerDensityInHierarchy() {
+    //simulating million fileWithNoDocumentChanged events on refresh in a thousand-module project
+    MessageBusImpl childBus = new MessageBusImpl(this, myBus);
+    childBus.connect().subscribe(TOPIC1, new T1Listener() {
+      @Override
+      public void t11() {
+      }
+
+      @Override
+      public void t12() {
+      }
+    });
+    for (int i = 0; i < 1000; i++) {
+      new MessageBusImpl(this, childBus);
+    }
+
+    PlatformTestUtil.assertTiming("Too long", 2000, new Runnable() {
+      @Override
+      public void run() {
+        T1Listener publisher = myBus.syncPublisher(TOPIC1);
+        for (int i = 0; i < 1000000; i++) {
+          publisher.t11();
+        }
+      }
+    });
+  }
  
  private void assertEvents(String... expected) {
    String joinExpected = StringUtil.join(expected, "\n");
@@ -23,6 +23,7 @@
    <orderEntry type="library" name="CGLIB" level="project" />
    <orderEntry type="library" name="asm" level="project" />
    <orderEntry type="library" exported="" name="ForkJoin" level="project" />
+    <orderEntry type="module" module-name="testFramework" scope="TEST" />
  </component>
  <component name="copyright">
    <Base>