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

Jump to bottom

Fix multiple concurrency bugs in Master.gatherTableInformation() #546

Merged
merged 7 commits into from Jul 10, 2018
Merged

Fix multiple concurrency bugs in Master.gatherTableInformation() #546

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
a1c875a
Fix multiple concurrency bugs in Master.gatherTableInformation()
keith-turner Jul 2, 2018
91127f4
deprecated unused prop
keith-turner Jul 2, 2018
8e29faa
use general rpc timeout
keith-turner Jul 2, 2018
6fa3f76
code review update
keith-turner Jul 3, 2018
7182b1a
code review updates
keith-turner Jul 5, 2018
476db84
make num threads configurable again
keith-turner Jul 9, 2018
34dce7b
code review update
keith-turner Jul 10, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 4 additions & 0 deletions core/src/main/java/org/apache/accumulo/core/conf/Property.java
View file
Open in desktop
Expand Up @@ -340,6 +340,10 @@ public enum Property {
MASTER_REPLICATION_COORDINATOR_THREADCHECK("master.replication.coordinator.threadcheck.time",
"5s", PropertyType.TIMEDURATION,
"The time between adjustments of the coordinator thread pool"),
/**
* @deprecated since 1.9.1
Copy link
Member

Choose a reason for hiding this comment

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

I actually meant in the text in the property description. The javadoc comment here won't really be useful. Also, it should be 1.9.2.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

oh yeah, should be be 1.9.2. Not exactly sure what you are looking for, can you give a code example?

Copy link
Member

@ctubbsii ctubbsii Jul 4, 2018 •
edited

Choose a reason for hiding this comment

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

See line 348 below. s/The number of threads/Deprecated since 1.9.2. The number of threads/
The @Deprecated annotation is sufficient for devs, but these String descriptions get published in our docs for users, where they are useful for both users and devs.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

I updated the docs. I looked into the the code that generates prop docs and it generates documentation for deprecated props. So users looking at the docs will know which props are deprecated, but they will not know when. If ACCUMULO-4592 were done, could also add a deprecated version.

*/
@Deprecated
Copy link
Member

Choose a reason for hiding this comment

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

Description should be updated to reflect when it was deprecated. I noticed while reviewing deprecated items that we hadn't been doing this for configuration properties, and it's very useful.

MASTER_STATUS_THREAD_POOL_SIZE("master.status.threadpool.size", "1", PropertyType.COUNT,
"The number of threads to use when fetching the tablet server status for balancing."),
MASTER_METADATA_SUSPENDABLE("master.metadata.suspendable", "false", PropertyType.BOOLEAN,
Expand Down
32 changes: 22 additions & 10 deletions server/master/src/main/java/org/apache/accumulo/master/Master.java
View file
Open in desktop
Expand Up @@ -32,6 +32,7 @@
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
Expand Down Expand Up @@ -163,6 +164,7 @@
import org.slf4j.LoggerFactory;

import com.google.common.base.Optional;
import com.google.common.collect.ImmutableSortedMap;
import com.google.common.collect.Iterables;

/**
Expand Down Expand Up @@ -1065,7 +1067,7 @@ public void run() {
private long updateStatus()
throws AccumuloException, AccumuloSecurityException, TableNotFoundException {
Set<TServerInstance> currentServers = tserverSet.getCurrentServers();
tserverStatus = Collections.synchronizedSortedMap(gatherTableInformation(currentServers));
tserverStatus = gatherTableInformation(currentServers);
checkForHeldServer(tserverStatus);

if (!badServers.isEmpty()) {
Expand Down Expand Up @@ -1146,12 +1148,16 @@ private long balanceTablets() {

private SortedMap<TServerInstance,TabletServerStatus> gatherTableInformation(
Set<TServerInstance> currentServers) {
final long rpcTimeout = getConfiguration().getTimeInMillis(Property.GENERAL_RPC_TIMEOUT);
long start = System.currentTimeMillis();
int threads = Math.max(getConfiguration().getCount(Property.MASTER_STATUS_THREAD_POOL_SIZE), 1);
ExecutorService tp = Executors.newFixedThreadPool(threads);
final SortedMap<TServerInstance,TabletServerStatus> result = new TreeMap<>();
ExecutorService tp = Executors.newCachedThreadPool();
final SortedMap<TServerInstance,TabletServerStatus> result = new ConcurrentSkipListMap<>();
for (TServerInstance serverInstance : currentServers) {
final TServerInstance server = serverInstance;
// Since an unbounded thread pool is being used, rate limit how fast task are added to the
// executor. This prevents the threads from growing large unless there are lots of
// unresponsive tservers.
sleepUninterruptibly(Math.max(5, rpcTimeout / 24000), TimeUnit.MILLISECONDS);
tp.submit(new Runnable() {
@Override
public void run() {
Expand Down Expand Up @@ -1191,18 +1197,24 @@ public void run() {
}
tp.shutdown();
try {
tp.awaitTermination(getConfiguration().getTimeInMillis(Property.TSERV_CLIENT_TIMEOUT) * 2,
TimeUnit.MILLISECONDS);
tp.awaitTermination(Math.max(10000, rpcTimeout / 3), TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
log.debug("Interrupted while fetching status");
}

tp.shutdownNow();

// Threads may still modify map after shutdownNow is called, so create an immutable snapshot.
SortedMap<TServerInstance,TabletServerStatus> info = ImmutableSortedMap.copyOf(result);
Copy link
Member

Choose a reason for hiding this comment

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

If something isn't in this map by the time we get here, we could just add the remaining directly to the badServers.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The bad server processing needs some attention, I will open up a follow on issue.


synchronized (badServers) {
badServers.keySet().retainAll(currentServers);
badServers.keySet().removeAll(result.keySet());
badServers.keySet().removeAll(info.keySet());
}
log.debug(String.format("Finished gathering information from %d servers in %.2f seconds",
result.size(), (System.currentTimeMillis() - start) / 1000.));
return result;
log.debug(String.format("Finished gathering information from %d of %d servers in %.2f seconds",
info.size(), currentServers.size(), (System.currentTimeMillis() - start) / 1000.));

return info;
}

public void run() throws IOException, InterruptedException, KeeperException {
Expand Down