[improve][broker] Don't call ManagedLedger#asyncAddEntry in Netty I/O thread

[BewareMyPower]

Motivation

#23940 brings a behavior change that the core logic of ManagedLedger#asyncAddEntry now won't switch threads, which means it will be executed directly in Netty I/O thread via PersistentTopic#asyncAddEntry.

The beforeAddEntry method calls theintercept and interceptWithNumberOfMessages methods for all broker entry interceptors and prepends a new broker entry metadata buffer on the original buffer (though it's just a composite buffer).

There is a risk that when many producers send messages to the same managed ledger concurrently, the process of asyncAddEntry might block the Netty I/O thread for some time and cause the performance regression.

Modifications

In PersistentTopic#publishMessage, expose the getExecutor() method for ManagedLedger and execute ManagedLedger#asyncAddEntry in that executor. The change of #12606 is moved to PersistentTopic as well that the buffer is retained before switching to another thread.

After that, only synchronize afterAddEntryToQueue with other synchronized methods of ManagedLedgerImpl. P.S. actually I don't think synchronized is needed here but the logic is not trivial like beforeAddEntryToQueue and beforeAddEntry, so I still retain it as synchronized.

ManagedLedgerImpl#asyncAddEntry still doesn't switch the thread, so it would still be possible for the downstream application to synchronize asyncAddEntry, either by adding a lock (e.g. synchronized) or executing this method is a single thread.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: BewareMyPower#40

[merlimat]

An extra context switch for each entry is costly, especially when you have many small entries and little or no batching. That's why we put it on the same thread.

If the interceptor needs to do expensive work, maybe only the interceptor part should be done in a different thread, though it shouldn't affect it when we don't use interceptor.

[lhotari]

An extra context switch for each entry is costly, especially when you have many small entries and little or no batching. That's why we put it on the same thread.

@merlimat The thread switching was added in PR #9039, already in December 2020. The reason to make this change is related to a performance concern of #23940 changes which removed the thread switching.

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

Lines 796 to 826 in ee5b13a

	public void asyncAddEntry(ByteBuf buffer, int numberOfMessages, AddEntryCallback callback, Object ctx) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] asyncAddEntry size={} state={}", name, buffer.readableBytes(), state);
	}
	// retain buffer in this thread
	buffer.retain();
	// Jump to specific thread to avoid contention from writers writing from different threads
	final var addOperation = OpAddEntry.createNoRetainBuffer(this, buffer, numberOfMessages, callback, ctx,
	currentLedgerTimeoutTriggered);
	var added = false;
	try {
	// Use synchronized to ensure if `addOperation` is added to queue and fails later, it will be the first
	// element in `pendingAddEntries`.
	synchronized (this) {
	if (managedLedgerInterceptor != null) {
	managedLedgerInterceptor.beforeAddEntry(addOperation, addOperation.getNumberOfMessages());
	}
	final var state = STATE_UPDATER.get(this);
	beforeAddEntryToQueue(state);
	pendingAddEntries.add(addOperation);
	added = true;
	afterAddEntryToQueue(state, addOperation);
	}
	} catch (Throwable throwable) {
	if (!added) {
	addOperation.failed(ManagedLedgerException.getManagedLedgerException(throwable));
	} // else: all elements of `pendingAddEntries` will fail in another thread
	}
	}

In Pulsar use cases, synchronization on CPU intensive operations (or blocking IO operations) in Netty IO threads could cause performance regressions. In this case, it would impact use cases where there's a large number of producers producing to a single topic.
Blocking IO threads will have a broader impact since it will impact Netty IO of all connections sharing the same IO thread.

Before #23940, the code looks like this:

pulsar/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

Lines 796 to 810 in 7a79c78

	public void asyncAddEntry(ByteBuf buffer, int numberOfMessages, AddEntryCallback callback, Object ctx) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] asyncAddEntry size={} state={}", name, buffer.readableBytes(), state);
	}
	// retain buffer in this thread
	buffer.retain();
	// Jump to specific thread to avoid contention from writers writing from different threads
	executor.execute(() -> {
	OpAddEntry addOperation = OpAddEntry.createNoRetainBuffer(this, buffer, numberOfMessages, callback, ctx,
	currentLedgerTimeoutTriggered);
	internalAsyncAddEntry(addOperation);
	});
	}

btw. In the Pulsar code base, we have a problem in how IO threads are used. IO threads are used to process work that shouldn't be handled with IO threads at all. I have created an issue #23865. There should be a separate thread pool for running blocking operations and CPU intensive synchronized operations.

[lhotari]

Great work @BewareMyPower. Some comments added in this first pass.

[lhotari]

this should be an internal concern of the ManagedLedger implementation's asyncAddEntry method. One reason for this is that there's multiple asyncAddEntry signatures and we'd like to add them all in order.

[BewareMyPower]

I think we should restore the synchronized keyword to the asyncAddEntry method to make it thread safe as it is before

Instead of that, I still think the synchronization should be performed from the caller. asyncAddEntry only needs to synchronize it with other asyncAddEntry or addEntry method calls. It does not need to synchronize with other managed ledger's synchronized methods. Let me improve the apiNotes parts

[lhotari]

There are 2 different aspects to consider: thread safety and ordering.

Regarding "I still think the synchronization should be performed from the caller":
In Java, synchronization is not only about performing operations one by one under a mutually exclusive lock. "Visibility" is an important aspect of Java thread safety. That's why it doesn't make sense for callers to synchronize calls to asyncAddEntry since all callers would need to use the same lock for both ordering and thread safety.

Snippet from "Java Concurrency in Practice", Chapter 2 "Thread safety":

It is a common mistake to assume that synchronization needs to be used only when writing to shared variables; this is simply not true.
For each mutable state variable that may be accessed by more than one thread, all accesses to that variable must be performed with the same lock held. In this case, we say that the variable is guarded by that lock.

How a ManagedLedger implementation achieves ordering guarantees and thread safety is an internal implementation detail. In the case of ManagedLedger, it doesn't make sense to delegate the responsibility of thread safety to the caller.

Another downside of the ledger.getExecutor().execute solution is that it exposes internal implementation details that callers of the API must be aware of. This is not great API design when such implementation details are exposed.

I think we need to consider a different solution. I can see some code examples in #23940 of what needs to be solved. To me, it seems this could be solved with the Object ctx parameter by passing a ctx that is also understood by the interceptor. @BewareMyPower, would you be able to research that type of solution instead?

[BewareMyPower]

@merlimat The thread switching was added in PR #9039, already in December 2020.

@merlimat @lhotari to correct it, this is the very early behavior introduced in #1521.

This PR intends to decouple ManagedLedger#asyncAddEntry and PersistentTopic#asyncAddEntry so that the managed ledger interface can be more flexible for the downstream protocol handlers to use.

After that, all write operations from Pulsar client will still keep the original behavior that switches to managed ledger's executor to call ManagedLedger#asyncAddEntry.

However, regarding the downstream, for example, in my Kafka protocol handler implementation, PersistentTopic#publishMessage is not called in an I/O thread. Instead, it's called in an independent worker thread. Then I can choose to call persistentTopic.getManagedLedger().asyncAddEntry(/* ... */) in order, which can be achieved by adding the synchronized keyword or using the same worker thread for the same topic.

The comment here makes sense to a certain extent, but it might be a new topic (e.g. thread switching vs. synchronized) to discuss, which is beyond the scope of this PR. At least, the existing thread switching approach can already achieve high publish performance, which is verified by many benchmarks.

[lhotari]

Please check the review comments

[lhotari]

using the STATE_UPDATER doesn't make any difference for plain reads of the field value.

[lhotari]

using the STATE_UPDATER doesn't make any difference for plain reads of the field value.

[lhotari]

There are 2 different aspects to consider: thread safety and ordering.

Regarding "I still think the synchronization should be performed from the caller":
In Java, synchronization is not only about performing operations one by one under a mutually exclusive lock. "Visibility" is an important aspect of Java thread safety. That's why it doesn't make sense for callers to synchronize calls to asyncAddEntry since all callers would need to use the same lock for both ordering and thread safety.

Snippet from "Java Concurrency in Practice", Chapter 2 "Thread safety":

It is a common mistake to assume that synchronization needs to be used only when writing to shared variables; this is simply not true.
For each mutable state variable that may be accessed by more than one thread, all accesses to that variable must be performed with the same lock held. In this case, we say that the variable is guarded by that lock.

How a ManagedLedger implementation achieves ordering guarantees and thread safety is an internal implementation detail. In the case of ManagedLedger, it doesn't make sense to delegate the responsibility of thread safety to the caller.

Another downside of the ledger.getExecutor().execute solution is that it exposes internal implementation details that callers of the API must be aware of. This is not great API design when such implementation details are exposed.

I think we need to consider a different solution. I can see some code examples in #23940 of what needs to be solved. To me, it seems this could be solved with the Object ctx parameter by passing a ctx that is also understood by the interceptor. @BewareMyPower, would you be able to research that type of solution instead?