diff --git a/platform/util/src/com/intellij/util/containers/ContainerUtil.java b/platform/util/src/com/intellij/util/containers/ContainerUtil.java index a5d5bd886531..7019b0e47a12 100644 --- a/platform/util/src/com/intellij/util/containers/ContainerUtil.java +++ b/platform/util/src/com/intellij/util/containers/ContainerUtil.java @@ -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 > int compareLexicographically(List o1, List 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; + } } diff --git a/platform/util/src/com/intellij/util/messages/impl/MessageBusConnectionImpl.java b/platform/util/src/com/intellij/util/messages/impl/MessageBusConnectionImpl.java index ce2ca630e40f..a3051241c4ef 100644 --- a/platform/util/src/com/intellij/util/messages/impl/MessageBusConnectionImpl.java +++ b/platform/util/src/com/intellij/util/messages/impl/MessageBusConnectionImpl.java @@ -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> 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; + } } diff --git a/platform/util/src/com/intellij/util/messages/impl/MessageBusImpl.java b/platform/util/src/com/intellij/util/messages/impl/MessageBusImpl.java index b20e3c6fb0b4..5040467dc1b2 100644 --- a/platform/util/src/com/intellij/util/messages/impl/MessageBusImpl.java +++ b/platform/util/src/com/intellij/util/messages/impl/MessageBusImpl.java @@ -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 MESSAGE_BUS_COMPARATOR = new Comparator() { + @Override + public int compare(MessageBusImpl bus1, MessageBusImpl bus2) { + return ContainerUtil.compareLexicographically(bus1.myOrder, bus2.myOrder); + } + }; private final ThreadLocal> 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> 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 myOrder; + private final ConcurrentMap mySyncPublishers = new ConcurrentHashMap(); private final ConcurrentMap myAsyncPublishers = new ConcurrentHashMap(); + + /** + * This bus's subscribers + */ private final ConcurrentMap> mySubscribers = new ConcurrentHashMap>(); + + /** + * Caches subscribers for this bus and its children or parent, depending on the topic's broadcast policy + */ + private final ConcurrentMap> mySubscriberCache = + new ConcurrentHashMap>(); private final List 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>(); } } @@ -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 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 childOrder = new ArrayList(myOrder.size() + 1); + childOrder.addAll(myOrder); + childOrder.add(lastChildIndex + 1); + return childOrder; } private void notifyChildBusDisposed(final MessageBusImpl childBus) { boolean removed = myChildBuses.remove(childBus); + Map 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 result) { final List topicSubscribers = mySubscribers.get(topic); if (topicSubscribers != null) { - Queue 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 topicSubscribers = mySubscriberCache.get(topic); + if (topicSubscribers == null) { + topicSubscribers = new SmartList(); + 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> ref = getRootBus().myWaitingBuses; + SortedMap map = ref.get(); + if (map == null) { + ref.set(map = new TreeMap(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 map = myWaitingBuses.get(); + if (map != null) { + Set buses = map.keySet(); + if (!buses.isEmpty()) { + for (MessageBusImpl bus : new ArrayList(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 topicSubscribers : mySubscribers.values()) { topicSubscribers.remove(connection); } + getRootBus().clearSubscriberCache(); if (myDisposed) return; final Iterator 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); } diff --git a/platform/util/testSrc/com/intellij/util/messages/MessageBusTest.java b/platform/util/testSrc/com/intellij/util/messages/MessageBusTest.java index a016060a1605..e27dbc606d5d 100644 --- a/platform/util/testSrc/com/intellij/util/messages/MessageBusTest.java +++ b/platform/util/testSrc/com/intellij/util/messages/MessageBusTest.java @@ -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"); diff --git a/platform/util/util.iml b/platform/util/util.iml index d0155fd9b0b0..4bb188ce67a6 100644 --- a/platform/util/util.iml +++ b/platform/util/util.iml @@ -23,6 +23,7 @@ +