Use AtomicIntegerFieldUpdater to mutate volatile int totalAvailablePermits in PersistentDispatcherMultipleConsumers class

[lhotari]

Motivation

AtomicIntegerFieldUpdater should be used to update volatile int fields when the value is incremented or decremented. This prevents "lost update" issues.

Modifications

Use existing TOTAL_AVAILABLE_PERMITS_UPDATER to mutate totalAvailablePermits field.

Co-authored-by: Devin Bost devinbost@users.noreply.github.com

[lhotari]

This PR was co-authored with @devinbost . Thanks for @devinbost for pinpointing the code location around totalAvailablePermits in PersistentDispatcherMultipleConsumers class. This fix wouldn't have happened without @devinbost 's persistent investigation work. This change might fix issues such as #6054.

@codelipenghui @merlimat @eolivelli @rdhabalia Please review this change.

[lhotari]

btw. This problem was only in the PersistentDispatcherMultipleConsumers class. Similar code in NonPersistentDispatcherMultipleConsumers didn't have the issue:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java

Lines 94 to 148 in 31f8315

	@Override
	public synchronized void removeConsumer(Consumer consumer) throws BrokerServiceException {
	if (consumerSet.removeAll(consumer) == 1) {
	consumerList.remove(consumer);
	log.info("Removed consumer {}", consumer);
	if (consumerList.isEmpty()) {
	if (closeFuture != null) {
	log.info("[{}] All consumers removed. Subscription is disconnected", name);
	closeFuture.complete(null);
	}
	TOTAL_AVAILABLE_PERMITS_UPDATER.set(this, 0);
	}
	} else {
	if (log.isDebugEnabled()) {
	log.debug("[{}] Trying to remove a non-connected consumer: {}", name, consumer);
	}
	TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, -consumer.getAvailablePermits());
	}
	}
	@Override
	public boolean isConsumerConnected() {
	return !consumerList.isEmpty();
	}
	@Override
	public CopyOnWriteArrayList<Consumer> getConsumers() {
	return consumerList;
	}
	@Override
	public synchronized boolean canUnsubscribe(Consumer consumer) {
	return consumerList.size() == 1 && consumerSet.contains(consumer);
	}
	@Override
	public CompletableFuture<Void> close() {
	IS_CLOSED_UPDATER.set(this, TRUE);
	return disconnectAllConsumers();
	}
	@Override
	public synchronized void consumerFlow(Consumer consumer, int additionalNumberOfMessages) {
	if (!consumerSet.contains(consumer)) {
	if (log.isDebugEnabled()) {
	log.debug("[{}] Ignoring flow control from disconnected consumer {}", name, consumer);
	}
	return;
	}
	TOTAL_AVAILABLE_PERMITS_UPDATER.addAndGet(this, additionalNumberOfMessages);
	if (log.isDebugEnabled()) {
	log.debug("[{}] Trigger new read after receiving flow control message", consumer);
	}
	}

[lhotari]

/pulsarbot run-failure-checks

[eolivelli]

LGTM

[lhotari]

/pulsarbot run-failure-checks

[merlimat]

This would be equivalent to totalAvailablePermits = 0

[lhotari]

true. I tried to make it consistent with the style used in NonPersistentDispatcherMultipleConsumers:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentDispatcherMultipleConsumers.java

Line 104 in 31f8315

TOTAL_AVAILABLE_PERMITS_UPDATER.set(this, 0);

[merlimat]

The increments/decrement on the totalAvailablePermits should already be safe because they're all done while holding the mutex on the dispatcher object.

[lhotari]

yes you are right, it seems that all updates are already happening withing synchronized methods.

[lhotari]

@merlimat I re-opened this PR. readMoreEntries method isn't synchronized and it's getting called from this location without synchronization:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Lines 2439 to 2451 in a1cebdb

	public void unblockDispatchersOnUnAckMessages(List<PersistentDispatcherMultipleConsumers> dispatcherList) {
	lock.writeLock().lock();
	try {
	dispatcherList.forEach(dispatcher -> {
	dispatcher.unBlockDispatcherOnUnackedMsgs();
	executor().execute(() -> dispatcher.readMoreEntries());
	log.info("[{}] Dispatcher is unblocked", dispatcher.getName());
	blockedDispatchers.remove(dispatcher);
	});
	} finally {
	lock.writeLock().unlock();
	}
	}

.

[devinbost]

/pulsarbot run-failure-checks

[lhotari]

I'm closing this PR since as @merlimat pointed out, all mutations are already safe since they are done in synchronized methods in PersistentDispatcherMultipleConsumers class.

[lhotari]

Currently PersistentStreamingDispatcherMultipleConsumers.readMoreEntries method isn't synchronized:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java

Lines 137 to 140 in f2d72c9

	@Override
	public void readMoreEntries() {
	// totalAvailablePermits may be updated by other threads
	int currentTotalAvailablePermits = totalAvailablePermits;

Is there a reason why it's not a synchronized method? In one location the method is called within a synchronized block like this:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java

Lines 191 to 205 in 31f8315

	@Override
	public void run(Timeout timeout) throws Exception {
	if (log.isDebugEnabled()) {
	log.debug("[{}] Timer triggered", dispatcher.getName());
	}
	if (timeout.isCancelled()) {
	return;
	}
	synchronized (dispatcher) {
	currentTimeoutTarget = -1;
	timeout = null;
	dispatcher.readMoreEntries();
	}
	}

Synchronization is missing in this location:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

Lines 2439 to 2451 in a1cebdb

	public void unblockDispatchersOnUnAckMessages(List<PersistentDispatcherMultipleConsumers> dispatcherList) {
	lock.writeLock().lock();
	try {
	dispatcherList.forEach(dispatcher -> {
	dispatcher.unBlockDispatcherOnUnackedMsgs();
	executor().execute(() -> dispatcher.readMoreEntries());
	log.info("[{}] Dispatcher is unblocked", dispatcher.getName());
	blockedDispatchers.remove(dispatcher);
	});
	} finally {
	lock.writeLock().unlock();
	}
	}

[lhotari]

Another location where the readEntries call is explicitly wrapped in a synchronized block

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

Lines 612 to 621 in 875262a

	topic.getBrokerService().executor().schedule(() -> {
	synchronized (PersistentDispatcherMultipleConsumers.this) {
	if (!havePendingRead) {
	log.info("[{}] Retrying read operation", name);
	readMoreEntries();
	} else {
	log.info("[{}] Skipping read retry: havePendingRead {}", name, havePendingRead, exception);
	}
	}
	}, waitTimeMillis, TimeUnit.MILLISECONDS);

I'm running an experiment to see if tests pass when readEntries method is made synchronized.

[lhotari]

I opened a separate PR #10413 to make the readMoreEntries method synchronized.

[lhotari]

This PR is not needed since #10413 was merged.