Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

RoundRobinLoadBalancer reduce copy/resize for connection add/remove #1514

Merged
merged 2 commits into from
Apr 23, 2021
Merged

RoundRobinLoadBalancer reduce copy/resize for connection add/remove #1514

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
eb19c43
RoundRobinLoadBalancer reduce copy/resize for connection add/remove
Scottmitch Apr 23, 2021
d329745
review comments
Scottmitch Apr 23, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions servicetalk-loadbalancer/gradle/spotbugs/main-exclusions.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,22 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Copyright © 2021 Apple Inc. and the ServiceTalk project authors
~
~ Licensed under the Apache License, Version 2.0 (the "License");
~ you may not use this file except in compliance with the License.
~ You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<FindBugsFilter>
<Match>
<Class name="io.servicetalk.loadbalancer.RoundRobinLoadBalancer$Host"/>
<Bug pattern="VO_VOLATILE_REFERENCE_TO_ARRAY"/>
</Match>
</FindBugsFilter>
84 changes: 54 additions & 30 deletions ...cetalk-loadbalancer/src/main/java/io/servicetalk/loadbalancer/RoundRobinLoadBalancer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright © 2018-2020 Apple Inc. and the ServiceTalk project authors
* Copyright © 2018-2021 Apple Inc. and the ServiceTalk project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
Expand Down Expand Up @@ -39,20 +39,23 @@

import java.util.AbstractMap.SimpleImmutableEntry;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Stream;

import static io.servicetalk.client.api.LoadBalancerReadyEvent.LOAD_BALANCER_NOT_READY_EVENT;
import static io.servicetalk.client.api.LoadBalancerReadyEvent.LOAD_BALANCER_READY_EVENT;
import static io.servicetalk.concurrent.api.AsyncCloseables.newCompositeCloseable;
import static io.servicetalk.concurrent.api.AsyncCloseables.toAsyncCloseable;
import static io.servicetalk.concurrent.api.Completable.mergeAllDelayError;
import static io.servicetalk.concurrent.api.Completable.completed;
import static io.servicetalk.concurrent.api.Processors.newPublisherProcessorDropHeadOnOverflow;
import static io.servicetalk.concurrent.api.Publisher.from;
import static io.servicetalk.concurrent.api.Single.defer;
import static io.servicetalk.concurrent.api.Single.failed;
import static io.servicetalk.concurrent.api.Single.succeeded;
Expand Down Expand Up @@ -86,6 +89,8 @@ public final class RoundRobinLoadBalancer<ResolvedAddress, C extends LoadBalance

private static final Logger LOGGER = LoggerFactory.getLogger(RoundRobinLoadBalancer.class);
private static final List<?> CLOSED_LIST = new ArrayList<>(0);
private static final Object[] CLOSED_ARRAY = new Object[0];
private static final Object[] EMPTY_ARRAY = new Object[0];

@SuppressWarnings("rawtypes")
private static final AtomicReferenceFieldUpdater<RoundRobinLoadBalancer, List> activeHostsUpdater =
Expand Down Expand Up @@ -260,13 +265,14 @@ private Single<C> selectConnection0(Predicate<C> selector) {
final ThreadLocalRandom rnd = ThreadLocalRandom.current();

// Try first to see if an existing connection can be used
final List<C> connections = host.connections;
final int size = connections.size();
final Object[] connections = host.connections;
// With small enough search space, attempt all connections.
// Back off after exploring most of the search space, it gives diminishing returns.
final int attempts = size < MIN_SEARCH_SPACE ? size : (int) (size * SEARCH_FACTOR);
final int attempts = connections.length < MIN_SEARCH_SPACE ?
connections.length : (int) (connections.length * SEARCH_FACTOR);
for (int i = 0; i < attempts; i++) {
final C connection = connections.get(rnd.nextInt(size));
@SuppressWarnings("unchecked")
final C connection = (C) connections[rnd.nextInt(connections.length)];
if (selector.test(connection)) {
return succeeded(connection);
}
Expand Down Expand Up @@ -331,64 +337,77 @@ List<Entry<ResolvedAddress, List<C>>> activeAddresses() {
return activeHosts.stream().map(Host::asEntry).collect(toList());
}

private static class Host<Addr, C extends ListenableAsyncCloseable> implements AsyncCloseable {
private static final class Host<Addr, C extends ListenableAsyncCloseable> implements AsyncCloseable {
@SuppressWarnings("rawtypes")
private static final AtomicReferenceFieldUpdater<Host, List> connectionsUpdater =
AtomicReferenceFieldUpdater.newUpdater(Host.class, List.class, "connections");
private static final AtomicReferenceFieldUpdater<Host, Object[]> connectionsUpdater =
AtomicReferenceFieldUpdater.newUpdater(Host.class, Object[].class, "connections");

final Addr address;
private volatile List<C> connections = emptyList();
private volatile Object[] connections = EMPTY_ARRAY;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This could be ListenableAsyncCloseable

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How big is connections likely to be and how much contention is there likely to be? Could perhaps just use or extend CopyOnWriteArraySet

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Its internal CopyOnWriteArrayList uses ReentrantLock that we prefer to avoid as it may content on the event-loop threads.


Host(Addr address) {
this.address = requireNonNull(address);
}

void markInactive() {
@SuppressWarnings("unchecked")
final List<C> toRemove = connectionsUpdater.getAndSet(this, CLOSED_LIST);
LOGGER.debug("Closing {} connection(s) gracefully to inactive address: {}", toRemove.size(), address);
for (C conn : toRemove) {
conn.closeAsyncGracefully().subscribe();
final Object[] toRemove = connectionsUpdater.getAndSet(this, CLOSED_ARRAY);
LOGGER.debug("Closing {} connection(s) gracefully to inactive address: {}", toRemove.length, address);
for (Object conn : toRemove) {
@SuppressWarnings("unchecked")
final C cConn = (C) conn;
cConn.closeAsyncGracefully().subscribe();
}
}

boolean isInactive() {
return connections == CLOSED_LIST;
return connections == CLOSED_ARRAY;
}

boolean addConnection(C connection) {
for (;;) {
List<C> existing = this.connections;
if (existing == CLOSED_LIST) {
final Object[] existing = this.connections;
if (existing == CLOSED_ARRAY) {
return false;
}
ArrayList<C> connectionAdded = new ArrayList<>(existing);
connectionAdded.add(connection);
if (connectionsUpdater.compareAndSet(this, existing, connectionAdded)) {
Object[] newList = Arrays.copyOf(existing, existing.length + 1);
newList[existing.length] = connection;
if (connectionsUpdater.compareAndSet(this, existing, newList)) {
break;
}
}
Comment on lines 367 to 377
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do {} while(cU.compareAndSet()) would avoid the break


// Instrument the new connection so we prune it on close
connection.onClose().beforeFinally(() -> {
for (;;) {
final List<C> existing = connections;
if (existing == CLOSED_LIST) {
final Object[] existing = this.connections;
if (existing == CLOSED_ARRAY) {
break;
}
Comment on lines +383 to 385
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not really necessary as CLOSED_ARRAY is always empty.

ArrayList<C> connectionRemoved = new ArrayList<>(existing);
if (!connectionRemoved.remove(connection) ||
connectionsUpdater.compareAndSet(this, existing, connectionRemoved)) {
int i = 0;
for (; i < existing.length; ++i) {
if (existing[i].equals(connection)) {
break;
}
}
if (i == existing.length) {
break;
} else {
Object[] newList = new Object[existing.length - 1];
System.arraycopy(existing, 0, newList, 0, i);
System.arraycopy(existing, i + 1, newList, i, newList.length - i);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Consider adding a case for i == 0, then you can do a single arraycopy:

 System.arraycopy(existing, 1, newList, 0, newList.length);

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

WDYT about this one?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm not sure it is worth it bcz arraycopy already does bounds checking. we could also specialize other cases (e.g. i == existing.length - 1) but lets save for followup investigation.

if (connectionsUpdater.compareAndSet(this, existing, newList)) {
break;
}
}
}
}).subscribe();
Comment on lines 380 to 403
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Perhaps make this a standalone removeConnectionMethod() and call that in beforeFinally?

return true;
}

// Used for testing only
@SuppressWarnings("unchecked")
Entry<Addr, List<C>> asEntry() {
return new SimpleImmutableEntry<>(address, new ArrayList<>(connections));
return new SimpleImmutableEntry<>(address, Stream.of(connections).map(conn -> (C) conn).collect(toList()));
}

@Override
Expand All @@ -403,20 +422,25 @@ public Completable closeAsyncGracefully() {

@SuppressWarnings("unchecked")
private Completable doClose(final Function<? super C, Completable> closeFunction) {
return defer(() -> succeeded((List<C>) connectionsUpdater.getAndSet(this, CLOSED_LIST)))
.flatMapCompletable(list -> mergeAllDelayError(list.stream().map(closeFunction)::iterator));
return Completable.defer(() -> {
final Object[] connections = connectionsUpdater.getAndSet(this, CLOSED_ARRAY);
return connections == CLOSED_ARRAY ? completed() :
from(connections).flatMapCompletableDelayError(conn -> closeFunction.apply((C) conn));
});
}

@Override
public String toString() {
return "Host{" +
"address=" + address +
", removed=" + (connections == CLOSED_LIST) +
", removed=" + (connections == CLOSED_ARRAY) +
'}';
}
}

private static final class StacklessNoAvailableHostException extends NoAvailableHostException {
private static final long serialVersionUID = 5942960040738091793L;

private StacklessNoAvailableHostException(final String message) {
super(message);
}
Expand Down