Streamline AsyncShardFetch#getNumberOfInFlightFetches

[luyuncheng]

When we restart a ES 7.10 cluster, we found lead master cpu hot_threads much cost in org.elasticsearch.action.admin.cluster.health.TransportClusterHealthAction.getResponse -> org.elasticsearch.cluster.routing.allocation.AllocationService.getNumberOfInFlightFetches

we trace the function getNumberOfInFlightFetches

it shows that all time cost in GatewayAllocator.getNumberOfInFlightFetches ,

elasticsearch/server/src/main/java/org/elasticsearch/gateway/GatewayAllocator.java

Lines 85 to 93 in b29399e

	public int getNumberOfInFlightFetches() {
	int count = 0;
	for (AsyncShardFetch<NodeGatewayStartedShards> fetch : asyncFetchStarted.values()) {
	count += fetch.getNumberOfInFlightFetches();
	}
	for (AsyncShardFetch<NodeStoreFilesMetadata> fetch : asyncFetchStore.values()) {
	count += fetch.getNumberOfInFlightFetches();
	}
	return count;

In some case, it will scan all shards * ndoes every time.

elasticsearch/server/src/main/java/org/elasticsearch/gateway/AsyncShardFetch.java

Lines 77 to 84 in b29399e

	public synchronized int getNumberOfInFlightFetches() {
	int count = 0;
	for (NodeEntry<T> nodeEntry : cache.values()) {
	if (nodeEntry.isFetching()) {
	count++;
	}
	}
	return count;

BUT, in most cases, the fetching status is all false after scanned once.
there is no need to iterator nodes every time. we can add a fetching cache to ignore this at this PR

ISSUE: #93631

[elasticsearchmachine]

Pinging @elastic/es-distributed (Team:Distributed)

[DaveCTurner]

@elasticmachine generate changelog

[DaveCTurner]

Thanks @luyuncheng for spotting this and suggesting a fix. I think there's a better approach, however: we can make it so that getNumberOfInFlightFetches (and hasAnyNodeFetching) don't need to iterate at all. Instead, we can track the number of in-flight fetches directly in another field, adjusting the field when calling any of the methods that adjust NodeEntry#fetching and when adding or removing entries from AsyncShardFetch#cache. Would you try doing that?

[luyuncheng]

Thanks @luyuncheng for spotting this and suggesting a fix. I think there's a better approach, however: we can make it so that getNumberOfInFlightFetches (and hasAnyNodeFetching) don't need to iterate at all. Instead, we can track the number of in-flight fetches directly in another field, adjusting the field when calling any of the methods that adjust NodeEntry#fetching and when adding or removing entries from AsyncShardFetch#cache. Would you try doing that?

LGTM, Let me try.

[luyuncheng]

we can track the number of in-flight fetches directly in another field

@DaveCTurner at commit 6080f60
i try to add a field fetchingCount tracking the number of in-flight fetches, and it can remove the iterator from getNumberOfInFlightFetches and hasAnyNodeFetching

[DaveCTurner]

Nice. I left a few small comments.

I'd also like to introduce a method to verify that the count is consistent when not running in prod:

private boolean assertFetchingCountConsistent() {
    assert Thread.holdsLock(this);
    assert fetchingCount.get() == cache.values().stream().filter(NodeEntry::isFetching).count();
    return true;
}

and then we can say assert assertFetchingCountConsistent(); after we've changed the cache contents. That way we should pick up any mistakes in this area quickly.

[DaveCTurner]

Much nicer 😄

[DaveCTurner]

I think this could just be a volatile int because it's only ever updated within synchronized methods.

[DaveCTurner]

I'm ok with keeping this private, I don't think it needs testing.

[luyuncheng]

++ modified

[DaveCTurner]

I have pushed a commit (5beddb8) which adds the required changelog YAML file. You'll need to pull this branch and merge or rebase your changes on top of my commit.

[DaveCTurner]

@elasticmachine ok to test

[DaveCTurner]

LGTM, thanks @luyuncheng

[DaveCTurner]

(it's Friday evening here, I will not merge this before Monday)