Skip to content

Fix queuing in AsyncLucenePersistedState #50958

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

Merged
merged 3 commits into from
Jan 14, 2020
Merged

Fix queuing in AsyncLucenePersistedState #50958

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
05fabd4
Fix queuing in AsyncLucenePersistedState
ywelsch Jan 14, 2020
b17d3f2
ok
ywelsch Jan 14, 2020
37205d8
checkstyle
ywelsch Jan 14, 2020
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
81 changes: 42 additions & 39 deletions server/src/main/java/org/elasticsearch/gateway/GatewayMetaState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,6 @@
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.AbstractRunnable;
import org.elasticsearch.common.util.concurrent.EsExecutors;
import org.elasticsearch.common.util.concurrent.EsRejectedExecutionException;
import org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor;
import org.elasticsearch.core.internal.io.IOUtils;
import org.elasticsearch.env.NodeMetaData;
Expand Down Expand Up @@ -291,7 +290,9 @@ public void setCurrentTerm(long currentTerm) {
} else {
logger.trace("queuing term update (setting term to {})", currentTerm);
newCurrentTermQueued = true;
scheduleUpdate();
if (newStateQueued == false) {
scheduleUpdate();
}
}
}
}
Expand All @@ -305,55 +306,57 @@ public void setLastAcceptedState(ClusterState clusterState) {
} else {
logger.trace("queuing cluster state update (setting cluster state to {})", clusterState.version());
newStateQueued = true;
scheduleUpdate();
if (newCurrentTermQueued == false) {
scheduleUpdate();
}
}
}
}

private void scheduleUpdate() {
assert Thread.holdsLock(mutex);
try {
threadPoolExecutor.execute(new AbstractRunnable() {
assert threadPoolExecutor.getQueue().isEmpty() : "threadPoolExecutor queue not empty";
threadPoolExecutor.execute(new AbstractRunnable() {

@Override
public void onFailure(Exception e) {
logger.error("Exception occurred when storing new meta data", e);
}
@Override
public void onFailure(Exception e) {
logger.error("Exception occurred when storing new meta data", e);
}

@Override
protected void doRun() {
final Long term;
final ClusterState clusterState;
synchronized (mutex) {
if (newCurrentTermQueued) {
term = getCurrentTerm();
newCurrentTermQueued = false;
} else {
term = null;
}
if (newStateQueued) {
clusterState = getLastAcceptedState();
newStateQueued = false;
} else {
clusterState = null;
}
}
// write current term before last accepted state so that it is never below term in last accepted state
if (term != null) {
persistedState.setCurrentTerm(term);
@Override
public void onRejection(Exception e) {
assert threadPoolExecutor.isShutdown() : "only expect rejections when shutting down";
}

@Override
protected void doRun() {
final Long term;
final ClusterState clusterState;
synchronized (mutex) {
if (newCurrentTermQueued) {
term = getCurrentTerm();
logger.trace("resetting newCurrentTermQueued");
newCurrentTermQueued = false;
} else {
term = null;
}
if (clusterState != null) {
persistedState.setLastAcceptedState(resetVotingConfiguration(clusterState));
if (newStateQueued) {
clusterState = getLastAcceptedState();
logger.trace("resetting newStateQueued");
newStateQueued = false;
} else {
clusterState = null;
}
}
});
} catch (EsRejectedExecutionException e) {
// ignore cases where we are shutting down..., there is really nothing interesting to be done here...
if (threadPoolExecutor.isShutdown() == false) {
assert false : "only expect rejections when shutting down";
throw e;
// write current term before last accepted state so that it is never below term in last accepted state
if (term != null) {
persistedState.setCurrentTerm(term);
}
if (clusterState != null) {
persistedState.setLastAcceptedState(resetVotingConfiguration(clusterState));
}
}
}
});
}

static final CoordinationMetaData.VotingConfiguration staleStateConfiguration =
Expand Down