You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by bo...@apache.org on 2017/10/10 18:48:25 UTC
[3/8] storm git commit: STORM-2438: added in rebalance changes to
support RAS
http://git-wip-us.apache.org/repos/asf/storm/blob/e94b735a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
index 6533d15..da62925 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/supervisor/Slot.java
@@ -19,7 +19,7 @@
package org.apache.storm.daemon.supervisor;
import com.codahale.metrics.Meter;
-
+import com.google.common.annotations.VisibleForTesting;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
@@ -27,12 +27,13 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
+import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
-
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.cluster.IStormClusterState;
@@ -44,6 +45,9 @@ import org.apache.storm.generated.LocalAssignment;
import org.apache.storm.generated.ProfileAction;
import org.apache.storm.generated.ProfileRequest;
import org.apache.storm.localizer.AsyncLocalizer;
+import org.apache.storm.localizer.BlobChangingCallback;
+import org.apache.storm.localizer.GoodToGo;
+import org.apache.storm.localizer.LocallyCachedBlob;
import org.apache.storm.metric.StormMetricsRegistry;
import org.apache.storm.scheduler.ISupervisor;
import org.apache.storm.utils.LocalState;
@@ -53,7 +57,7 @@ import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class Slot extends Thread implements AutoCloseable {
+public class Slot extends Thread implements AutoCloseable, BlobChangingCallback {
private static final Logger LOG = LoggerFactory.getLogger(Slot.class);
private static final Meter numWorkersLaunched =
StormMetricsRegistry.registerMeter("supervisor:num-workers-launched");
@@ -74,8 +78,9 @@ public class Slot extends Thread implements AutoCloseable {
WAITING_FOR_WORKER_START,
KILL_AND_RELAUNCH,
KILL,
- WAITING_FOR_BASIC_LOCALIZATION,
- WAITING_FOR_BLOB_LOCALIZATION;
+ KILL_BLOB_UPDATE,
+ WAITING_FOR_BLOB_LOCALIZATION,
+ WAITING_FOR_BLOB_UPDATE;
}
static class StaticState {
@@ -89,11 +94,13 @@ public class Slot extends Thread implements AutoCloseable {
public final String host;
public final ISupervisor iSupervisor;
public final LocalState localState;
+ public final BlobChangingCallback changingCallback;
StaticState(AsyncLocalizer localizer, long hbTimeoutMs, long firstHbTimeoutMs,
- long killSleepMs, long monitorFreqMs,
- ContainerLauncher containerLauncher, String host, int port,
- ISupervisor iSupervisor, LocalState localState) {
+ long killSleepMs, long monitorFreqMs,
+ ContainerLauncher containerLauncher, String host, int port,
+ ISupervisor iSupervisor, LocalState localState,
+ BlobChangingCallback changingCallback) {
this.localizer = localizer;
this.hbTimeoutMs = hbTimeoutMs;
this.firstHbTimeoutMs = firstHbTimeoutMs;
@@ -104,9 +111,12 @@ public class Slot extends Thread implements AutoCloseable {
this.port = port;
this.iSupervisor = iSupervisor;
this.localState = localState;
+ this.changingCallback = changingCallback;
}
}
-
+
+ //TODO go through all of the state transitions and make sure we handle changingBlobs
+ //TODO make sure to add in transition helpers that clean changingBlobs && pendingChangeingBlobs for not the current topology
static class DynamicState {
public final MachineState state;
public final LocalAssignment newAssignment;
@@ -116,6 +126,9 @@ public class Slot extends Thread implements AutoCloseable {
public final Future<Void> pendingDownload;
public final Set<TopoProfileAction> profileActions;
public final Set<TopoProfileAction> pendingStopProfileActions;
+ public final Set<BlobChangeing> changingBlobs;
+ public final LocalAssignment pendingChangingBlobsAssignment;
+ public final Set<Future<Void>> pendingChangeingBlobs;
/**
* The last time that WAITING_FOR_WORKER_START, KILL, or KILL_AND_RELAUNCH were entered into.
@@ -139,15 +152,22 @@ public class Slot extends Thread implements AutoCloseable {
this.newAssignment = newAssignment;
this.pendingLocalization = null;
this.pendingDownload = null;
- this.profileActions = new HashSet<>();
- this.pendingStopProfileActions = new HashSet<>();
+ this.profileActions = Collections.emptySet();
+ this.pendingStopProfileActions = Collections.emptySet();
+ this.changingBlobs = Collections.emptySet();
+ this.pendingChangingBlobsAssignment = null;
+ this.pendingChangeingBlobs = Collections.emptySet();
}
-
+
public DynamicState(final MachineState state, final LocalAssignment newAssignment,
- final Container container, final LocalAssignment currentAssignment,
- final LocalAssignment pendingLocalization, final long startTime,
- final Future<Void> pendingDownload, final Set<TopoProfileAction> profileActions,
- final Set<TopoProfileAction> pendingStopProfileActions) {
+ final Container container, final LocalAssignment currentAssignment,
+ final LocalAssignment pendingLocalization, final long startTime,
+ final Future<Void> pendingDownload, final Set<TopoProfileAction> profileActions,
+ final Set<TopoProfileAction> pendingStopProfileActions,
+ final Set<BlobChangeing> changingBlobs,
+ final Set<Future<Void>> pendingChangingBlobs, final LocalAssignment pendingChaningBlobsAssignment) {
+ assert pendingChangingBlobs != null;
+ assert !(pendingChangingBlobs.isEmpty() ^ (pendingChaningBlobsAssignment == null));
this.state = state;
this.newAssignment = newAssignment;
this.currentAssignment = currentAssignment;
@@ -157,6 +177,9 @@ public class Slot extends Thread implements AutoCloseable {
this.pendingDownload = pendingDownload;
this.profileActions = profileActions;
this.pendingStopProfileActions = pendingStopProfileActions;
+ this.changingBlobs = changingBlobs;
+ this.pendingChangeingBlobs = pendingChangingBlobs;
+ this.pendingChangingBlobsAssignment = pendingChaningBlobsAssignment;
}
public String toString() {
@@ -179,47 +202,75 @@ public class Slot extends Thread implements AutoCloseable {
*/
public DynamicState withNewAssignment(LocalAssignment newAssignment) {
return new DynamicState(this.state, newAssignment,
- this.container, this.currentAssignment,
- this.pendingLocalization, this.startTime,
- this.pendingDownload, this.profileActions,
- this.pendingStopProfileActions);
+ this.container, this.currentAssignment,
+ this.pendingLocalization, this.startTime,
+ this.pendingDownload, this.profileActions,
+ this.pendingStopProfileActions, this.changingBlobs,
+ this.pendingChangeingBlobs, this.pendingChangingBlobsAssignment);
}
-
+
public DynamicState withPendingLocalization(LocalAssignment pendingLocalization, Future<Void> pendingDownload) {
return new DynamicState(this.state, this.newAssignment,
- this.container, this.currentAssignment,
- pendingLocalization, this.startTime,
- pendingDownload, this.profileActions,
- this.pendingStopProfileActions);
+ this.container, this.currentAssignment,
+ pendingLocalization, this.startTime,
+ pendingDownload, this.profileActions,
+ this.pendingStopProfileActions, this.changingBlobs,
+ this.pendingChangeingBlobs, this.pendingChangingBlobsAssignment);
}
public DynamicState withPendingLocalization(Future<Void> pendingDownload) {
return withPendingLocalization(this.pendingLocalization, pendingDownload);
}
-
+
public DynamicState withState(final MachineState state) {
long newStartTime = Time.currentTimeMillis();
return new DynamicState(state, this.newAssignment,
- this.container, this.currentAssignment,
- this.pendingLocalization, newStartTime,
- this.pendingDownload, this.profileActions,
- this.pendingStopProfileActions);
+ this.container, this.currentAssignment,
+ this.pendingLocalization, newStartTime,
+ this.pendingDownload, this.profileActions,
+ this.pendingStopProfileActions, this.changingBlobs,
+ this.pendingChangeingBlobs, this.pendingChangingBlobsAssignment);
}
public DynamicState withCurrentAssignment(final Container container, final LocalAssignment currentAssignment) {
return new DynamicState(this.state, this.newAssignment,
- container, currentAssignment,
- this.pendingLocalization, this.startTime,
- this.pendingDownload, this.profileActions,
- this.pendingStopProfileActions);
+ container, currentAssignment,
+ this.pendingLocalization, this.startTime,
+ this.pendingDownload, this.profileActions,
+ this.pendingStopProfileActions, this.changingBlobs,
+ this.pendingChangeingBlobs, this.pendingChangingBlobsAssignment);
}
public DynamicState withProfileActions(Set<TopoProfileAction> profileActions, Set<TopoProfileAction> pendingStopProfileActions) {
return new DynamicState(this.state, this.newAssignment,
- this.container, this.currentAssignment,
- this.pendingLocalization, this.startTime,
- this.pendingDownload, profileActions,
- pendingStopProfileActions);
+ this.container, this.currentAssignment,
+ this.pendingLocalization, this.startTime,
+ this.pendingDownload, profileActions,
+ pendingStopProfileActions, this.changingBlobs,
+ this.pendingChangeingBlobs, this.pendingChangingBlobsAssignment);
+ }
+
+ public DynamicState withChangingBlobs(Set<BlobChangeing> changingBlobs) {
+ if (changingBlobs == this.changingBlobs) {
+ return this;
+ }
+ return new DynamicState(this.state, this.newAssignment,
+ this.container, this.currentAssignment,
+ this.pendingLocalization, this.startTime,
+ this.pendingDownload, profileActions,
+ this.pendingStopProfileActions, changingBlobs,
+ this.pendingChangeingBlobs, this.pendingChangingBlobsAssignment);
+ }
+
+ public DynamicState withPendingChangeingBlobs(Set<Future<Void>> pendingChangeingBlobs,
+ LocalAssignment pendingChangeingBlobsAssignment) {
+ return new DynamicState(this.state, this.newAssignment,
+ this.container, this.currentAssignment,
+ this.pendingLocalization, this.startTime,
+ this.pendingDownload, profileActions,
+ this.pendingStopProfileActions, this.changingBlobs,
+ pendingChangeingBlobs,
+ pendingChangeingBlobsAssignment);
}
};
@@ -263,7 +314,41 @@ public class Slot extends Thread implements AutoCloseable {
return "{ " + topoId + ": " + request + " }";
}
}
-
+
+ /**
+ * Holds the information about a blob that is changing.
+ */
+ static class BlobChangeing {
+ private final LocalAssignment assignment;
+ private final LocallyCachedBlob blob;
+ private final GoodToGo.GoodToGoLatch latch;
+
+ public BlobChangeing(LocalAssignment assignment, LocallyCachedBlob blob, GoodToGo.GoodToGoLatch latch) {
+ this.assignment = assignment;
+ this.blob = blob;
+ this.latch = latch;
+ }
+
+ @Override
+ public String toString() {
+ return "BLOB CHANGING " + blob + " " + assignment;
+ }
+ }
+
+ @VisibleForTesting
+ static boolean forSameTopology(LocalAssignment a, LocalAssignment b) {
+ if (a == null && b == null) {
+ return true;
+ }
+ if (a != null && b != null) {
+ if (a.get_topology_id().equals(b.get_topology_id())) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ @VisibleForTesting
static boolean equivalent(LocalAssignment a, LocalAssignment b) {
if (a == null && b == null) {
return true;
@@ -298,14 +383,16 @@ public class Slot extends Thread implements AutoCloseable {
return handleRunning(dynamicState, staticState);
case WAITING_FOR_WORKER_START:
return handleWaitingForWorkerStart(dynamicState, staticState);
+ case KILL_BLOB_UPDATE:
+ return handleKillBlobUpdate(dynamicState, staticState);
case KILL_AND_RELAUNCH:
return handleKillAndRelaunch(dynamicState, staticState);
case KILL:
return handleKill(dynamicState, staticState);
- case WAITING_FOR_BASIC_LOCALIZATION:
- return handleWaitingForBasicLocalization(dynamicState, staticState);
case WAITING_FOR_BLOB_LOCALIZATION:
return handleWaitingForBlobLocalization(dynamicState, staticState);
+ case WAITING_FOR_BLOB_UPDATE:
+ return handleWaitingForBlobUpdate(dynamicState, staticState);
default:
throw new IllegalStateException("Code not ready to handle a state of "+dynamicState.state);
}
@@ -325,12 +412,14 @@ public class Slot extends Thread implements AutoCloseable {
if (dynamicState.newAssignment == null) {
return dynamicState.withState(MachineState.EMPTY);
}
- Future<Void> pendingDownload = staticState.localizer.requestDownloadBaseTopologyBlobs(dynamicState.newAssignment, staticState.port);
- return dynamicState.withPendingLocalization(dynamicState.newAssignment, pendingDownload).withState(MachineState.WAITING_FOR_BASIC_LOCALIZATION);
+ Future<Void> pendingDownload = staticState.localizer.requestDownloadTopologyBlobs(dynamicState.newAssignment,
+ staticState.port, staticState.changingCallback);
+ return dynamicState.withPendingLocalization(dynamicState.newAssignment, pendingDownload)
+ .withState(MachineState.WAITING_FOR_BLOB_LOCALIZATION);
}
/**
- * Kill the current container and start downloading what the new assignment needs, if there is a new assignment
+ * Kill the current container and start downloading what the new assignment needs, if there is a new assignment.
* PRECONDITION: container != null
* @param dynamicState current state
* @param staticState static data
@@ -339,17 +428,36 @@ public class Slot extends Thread implements AutoCloseable {
*/
static DynamicState killContainerForChangedAssignment(DynamicState dynamicState, StaticState staticState) throws Exception {
assert(dynamicState.container != null);
-
+
staticState.iSupervisor.killedWorker(staticState.port);
dynamicState.container.kill();
Future<Void> pendingDownload = null;
if (dynamicState.newAssignment != null) {
- pendingDownload = staticState.localizer.requestDownloadBaseTopologyBlobs(dynamicState.newAssignment, staticState.port);
+ pendingDownload = staticState.localizer.requestDownloadTopologyBlobs(dynamicState.newAssignment, staticState.port,
+ staticState.changingCallback);
}
+ dynamicState = drainAllChangingBlobs(dynamicState);
Time.sleep(staticState.killSleepMs);
return dynamicState.withPendingLocalization(dynamicState.newAssignment, pendingDownload).withState(MachineState.KILL);
}
-
+
+ /**
+ * Kill the current container, and wait go to the state to inform the localizer that we are ready to go.
+ * PRECONDITION: container != null
+ * @param dynamicState current state
+ * @param staticState static data
+ * @return the next state
+ */
+ private static DynamicState killContainerForChangedBlobs(DynamicState dynamicState, StaticState staticState) throws Exception {
+ assert(dynamicState.container != null);
+
+ staticState.iSupervisor.killedWorker(staticState.port);
+ dynamicState.container.kill();
+
+ Time.sleep(staticState.killSleepMs);
+ return dynamicState.withState(MachineState.KILL_BLOB_UPDATE);
+ }
+
/**
* Kill the current container and relaunch it. (Something odd happened)
* PRECONDITION: container != null
@@ -367,11 +475,11 @@ public class Slot extends Thread implements AutoCloseable {
//any stop profile actions that hadn't timed out yet, we should restart after the worker is running again.
HashSet<TopoProfileAction> mod = new HashSet<>(dynamicState.profileActions);
mod.addAll(dynamicState.pendingStopProfileActions);
- return dynamicState.withState(MachineState.KILL_AND_RELAUNCH).withProfileActions(mod, Collections.<TopoProfileAction> emptySet());
+ return dynamicState.withState(MachineState.KILL_AND_RELAUNCH).withProfileActions(mod, Collections.emptySet());
}
/**
- * Clean up a container
+ * Clean up a container.
* PRECONDITION: All of the processes have died.
* @param dynamicState current state
* @param staticState static data
@@ -391,7 +499,79 @@ public class Slot extends Thread implements AutoCloseable {
}
return ret;
}
-
+
+ /**
+ * Drop all of the changingBlobs and pendingChangeingBlobs.
+ * @param dynamicState current state.
+ * @return the next state.
+ */
+ private static DynamicState drainAllChangingBlobs(DynamicState dynamicState) {
+ if (!dynamicState.changingBlobs.isEmpty()) {
+ for (BlobChangeing rc : dynamicState.changingBlobs) {
+ rc.latch.countDown();
+ }
+ dynamicState = dynamicState.withChangingBlobs(Collections.emptySet());
+ }
+
+ if (!dynamicState.pendingChangeingBlobs.isEmpty()) {
+ dynamicState = dynamicState.withPendingChangeingBlobs(Collections.emptySet(), null);
+ }
+
+ return dynamicState;
+ }
+
+ /**
+ * Informs the async localizer for all of blobs that the worker is dead.
+ *
+ * PRECONDITION: container is null
+ * PRECONDITION: changingBlobs should only be for the given assignment.
+ * @param dynamicState the current state
+ * @return the futures for the current assignment.
+ */
+ private static DynamicState informChangedBlobs(DynamicState dynamicState, LocalAssignment assignment) {
+ assert dynamicState.container == null;
+ assert dynamicState.changingBlobs.stream().allMatch((cr) -> forSameTopology(cr.assignment, assignment));
+
+ Set<Future<Void>> futures = new HashSet<>(dynamicState.changingBlobs.size());
+ if (forSameTopology(dynamicState.pendingChangingBlobsAssignment, assignment)) {
+ //We need to add the new futures to the existing ones
+ futures.addAll(dynamicState.pendingChangeingBlobs);
+ }
+ //Otherwise they will just be replaced
+
+ for (BlobChangeing rc: dynamicState.changingBlobs) {
+ futures.add(rc.latch.countDown());
+ }
+
+ LOG.debug("found changeing blobs {} moving them to pending...", dynamicState.changingBlobs);
+ return dynamicState.withChangingBlobs(Collections.emptySet())
+ .withPendingChangeingBlobs(futures, assignment);
+ }
+
+ /**
+ * Filter all of the changing blobs to just be for those compatible with the given assignment.
+ * All others will be released appropriately.
+ *
+ * @param dynamicState the current state
+ * @param assignment the assignment to look for
+ * @return the updated dynamicState
+ */
+ private static DynamicState filterChangingBlobsFor(DynamicState dynamicState, final LocalAssignment assignment) {
+ if (!dynamicState.changingBlobs.isEmpty()) {
+ return dynamicState;
+ }
+
+ HashSet<BlobChangeing> savedBlobs = new HashSet<>(dynamicState.changingBlobs.size());
+ for (BlobChangeing rc: dynamicState.changingBlobs) {
+ if (forSameTopology(assignment, rc.assignment)) {
+ savedBlobs.add(rc);
+ } else {
+ rc.latch.countDown();
+ }
+ }
+ return dynamicState.withChangingBlobs(savedBlobs);
+ }
+
/**
* State Transitions for WAITING_FOR_BLOB_LOCALIZATION state.
* PRECONDITION: neither pendingLocalization nor pendingDownload is null.
@@ -410,17 +590,35 @@ public class Slot extends Thread implements AutoCloseable {
// We don't support canceling the download through the future yet,
// so to keep everything in sync, just wait
try {
+ //Release things that don't need to wait for us to finish downloading.
+ dynamicState = filterChangingBlobsFor(dynamicState, dynamicState.pendingLocalization);
+ if (!dynamicState.changingBlobs.isEmpty()) {
+ //Unblock downloading by accepting the futures.
+ dynamicState = informChangedBlobs(dynamicState, dynamicState.pendingLocalization);
+ }
+
dynamicState.pendingDownload.get(1000, TimeUnit.MILLISECONDS);
//Downloading of all blobs finished.
if (!equivalent(dynamicState.newAssignment, dynamicState.pendingLocalization)) {
//Scheduling changed
staticState.localizer.releaseSlotFor(dynamicState.pendingLocalization, staticState.port);
- return prepareForNewAssignmentNoWorkersRunning(dynamicState, staticState);
+ return prepareForNewAssignmentNoWorkersRunning(dynamicState.withPendingChangeingBlobs(Collections.emptySet(), null),
+ staticState);
+ }
+
+ if (!dynamicState.pendingChangeingBlobs.isEmpty()) {
+ LOG.info("There are pending changes, waiting for them to finish before launching container...");
+ //We cannot launch the container yet the resources may still be updating
+ return dynamicState.withState(MachineState.WAITING_FOR_BLOB_UPDATE)
+ .withPendingLocalization(null, null);
}
+
dynamicState = updateAssignmentIfNeeded(dynamicState);
numWorkersLaunched.mark();
Container c = staticState.containerLauncher.launchContainer(staticState.port, dynamicState.pendingLocalization, staticState.localState);
- return dynamicState.withCurrentAssignment(c, dynamicState.pendingLocalization).withState(MachineState.WAITING_FOR_WORKER_START).withPendingLocalization(null, null);
+ return dynamicState
+ .withCurrentAssignment(c, dynamicState.pendingLocalization).withState(MachineState.WAITING_FOR_WORKER_START)
+ .withPendingLocalization(null, null);
} catch (TimeoutException e) {
//We waited for 1 second loop around and try again....
return dynamicState;
@@ -439,29 +637,64 @@ public class Slot extends Thread implements AutoCloseable {
return dynamicState.withState(MachineState.EMPTY);
}
}
-
+
+ private static final long ONE_SEC_IN_NANO = TimeUnit.NANOSECONDS.convert(1, TimeUnit.SECONDS);
+
/**
- * State Transitions for WAITING_FOR_BASIC_LOCALIZATION state.
- * PRECONDITION: neither pendingLocalization nor pendingDownload is null.
- * PRECONDITION: The slot should be empty
+ * State Transitions for WAITING_FOR_BLOB_UPDATE state.
+ *
+ * PRECONDITION: container is null
+ * PRECONDITION: pendingChangeingBlobs is not empty (otherwise why did we go to this state)
+ * PRECONDITION: pendingChangingBlobsAssignment is not null.
+ *
* @param dynamicState current state
* @param staticState static data
* @return the next state
* @throws Exception on any error
*/
- static DynamicState handleWaitingForBasicLocalization(DynamicState dynamicState, StaticState staticState) throws Exception {
- assert(dynamicState.pendingLocalization != null);
- assert(dynamicState.pendingDownload != null);
- assert(dynamicState.container == null);
-
- //Ignore changes to scheduling while downloading the topology code
- // We don't support canceling the download through the future yet,
- // so to keep everything in sync, just wait
+ private static DynamicState handleWaitingForBlobUpdate(DynamicState dynamicState, StaticState staticState)
+ throws Exception {
+ assert dynamicState.container == null;
+ assert dynamicState.pendingChangingBlobsAssignment != null;
+ assert !dynamicState.pendingChangeingBlobs.isEmpty();
+
+ if (!equivalent(dynamicState.newAssignment, dynamicState.currentAssignment)) {
+ //We were rescheduled while waiting for the resources to be updated,
+ // but the container is already not running.
+ LOG.info("SLOT {}: Assignment Changed from {} to {}", staticState.port,
+ dynamicState.currentAssignment, dynamicState.newAssignment);
+ if (dynamicState.currentAssignment != null) {
+ staticState.localizer.releaseSlotFor(dynamicState.currentAssignment, staticState.port);
+ }
+ staticState.localizer.releaseSlotFor(dynamicState.pendingChangingBlobsAssignment, staticState.port);
+ return prepareForNewAssignmentNoWorkersRunning(dynamicState.withCurrentAssignment(null, null)
+ .withPendingChangeingBlobs(Collections.emptySet(), null),
+ staticState);
+ }
+
+ dynamicState = filterChangingBlobsFor(dynamicState, dynamicState.pendingChangingBlobsAssignment);
+ if (!dynamicState.changingBlobs.isEmpty()) {
+ dynamicState = informChangedBlobs(dynamicState, dynamicState.pendingChangingBlobsAssignment);
+ }
+
+ //We only have a set amount of time we can wait for before looping around again
+ long start = Time.nanoTime();
try {
- dynamicState.pendingDownload.get(1000, TimeUnit.MILLISECONDS);
- Future<Void> pendingDownload = staticState.localizer.requestDownloadTopologyBlobs(dynamicState.pendingLocalization, staticState.port);
- return dynamicState.withPendingLocalization(pendingDownload).withState(MachineState.WAITING_FOR_BLOB_LOCALIZATION);
- } catch (TimeoutException e) {
+ for (Future<Void> pending: dynamicState.pendingChangeingBlobs) {
+ long now = Time.nanoTime();
+ long timeLeft = ONE_SEC_IN_NANO - (now - start);
+ if (timeLeft <= 0) {
+ throw new TimeoutException();
+ }
+ pending.get(timeLeft, TimeUnit.NANOSECONDS);
+ }
+ //All done we can launch the worker now
+ Container c = staticState.containerLauncher.launchContainer(staticState.port, dynamicState.pendingChangingBlobsAssignment,
+ staticState.localState);
+ return dynamicState
+ .withCurrentAssignment(c, dynamicState.pendingChangingBlobsAssignment).withState(MachineState.WAITING_FOR_WORKER_START)
+ .withPendingChangeingBlobs(Collections.emptySet(), null);
+ } catch (TimeoutException ex) {
return dynamicState;
}
}
@@ -478,11 +711,11 @@ public class Slot extends Thread implements AutoCloseable {
static DynamicState handleKill(DynamicState dynamicState, StaticState staticState) throws Exception {
assert(dynamicState.container != null);
assert(dynamicState.currentAssignment != null);
-
+
if (dynamicState.container.areAllProcessesDead()) {
- LOG.warn("SLOT {} all processes are dead...", staticState.port);
- return cleanupCurrentContainer(dynamicState, staticState,
- dynamicState.pendingLocalization == null ? MachineState.EMPTY : MachineState.WAITING_FOR_BASIC_LOCALIZATION);
+ LOG.info("SLOT {} all processes are dead...", staticState.port);
+ return cleanupCurrentContainer(dynamicState, staticState,
+ dynamicState.pendingLocalization == null ? MachineState.EMPTY : MachineState.WAITING_FOR_BLOB_LOCALIZATION);
}
LOG.warn("SLOT {} force kill and wait...", staticState.port);
@@ -525,6 +758,42 @@ public class Slot extends Thread implements AutoCloseable {
}
/**
+ * State Transitions for KILL_BLOB_UPDATE state.
+ * PRECONDITION: container.kill() was called
+ * PRECONDITION: container != null && currentAssignment != null
+ *
+ * @param dynamicState current state
+ * @param staticState static data
+ * @return the next state
+ * @throws Exception on any error
+ */
+ private static DynamicState handleKillBlobUpdate(DynamicState dynamicState, StaticState staticState) throws Exception {
+ assert(dynamicState.container != null);
+ assert(dynamicState.currentAssignment != null);
+
+ //Release things that don't need to wait for us
+ dynamicState = filterChangingBlobsFor(dynamicState, dynamicState.currentAssignment);
+
+ if (dynamicState.container.areAllProcessesDead()) {
+ if (equivalent(dynamicState.newAssignment, dynamicState.currentAssignment)) {
+ dynamicState.container.cleanUp();
+ dynamicState = dynamicState.withCurrentAssignment(null, dynamicState.currentAssignment);
+ return informChangedBlobs(dynamicState, dynamicState.currentAssignment)
+ .withState(MachineState.WAITING_FOR_BLOB_UPDATE);
+ }
+ //Scheduling changed after we killed all of the processes
+ return prepareForNewAssignmentNoWorkersRunning(cleanupCurrentContainer(dynamicState, staticState, null), staticState);
+ }
+ //The child processes typically exit in < 1 sec. If 2 mins later they are still around something is wrong
+ if ((Time.currentTimeMillis() - dynamicState.startTime) > 120_000) {
+ throw new RuntimeException("Not all processes in " + dynamicState.container + " exited after 120 seconds");
+ }
+ dynamicState.container.forceKill();
+ Time.sleep(staticState.killSleepMs);
+ return dynamicState;
+ }
+
+ /**
* State Transitions for WAITING_FOR_WORKER_START state.
* PRECONDITION: container != null && currentAssignment != null
* @param dynamicState current state
@@ -546,7 +815,8 @@ public class Slot extends Thread implements AutoCloseable {
if (!equivalent(dynamicState.newAssignment, dynamicState.currentAssignment)) {
//We were rescheduled while waiting for the worker to come up
- LOG.warn("SLOT {}: Assignment Changed from {} to {}", staticState.port, dynamicState.currentAssignment, dynamicState.newAssignment);
+ LOG.info("SLOT {}: Assignment Changed from {} to {}", staticState.port, dynamicState.currentAssignment,
+ dynamicState.newAssignment);
return killContainerForChangedAssignment(dynamicState, staticState);
}
dynamicState = updateAssignmentIfNeeded(dynamicState);
@@ -556,6 +826,12 @@ public class Slot extends Thread implements AutoCloseable {
LOG.warn("SLOT {}: Container {} failed to launch in {} ms.", staticState.port, dynamicState.container, staticState.firstHbTimeoutMs);
return killAndRelaunchContainer(dynamicState, staticState);
}
+
+ dynamicState = filterChangingBlobsFor(dynamicState, dynamicState.currentAssignment);
+ if (!dynamicState.changingBlobs.isEmpty()) {
+ //Kill the container and restart it
+ return killContainerForChangedBlobs(dynamicState, staticState);
+ }
Time.sleep(1000);
return dynamicState;
}
@@ -573,12 +849,19 @@ public class Slot extends Thread implements AutoCloseable {
assert(dynamicState.currentAssignment != null);
if (!equivalent(dynamicState.newAssignment, dynamicState.currentAssignment)) {
- LOG.warn("SLOT {}: Assignment Changed from {} to {}", staticState.port, dynamicState.currentAssignment, dynamicState.newAssignment);
+ LOG.info("SLOT {}: Assignment Changed from {} to {}", staticState.port, dynamicState.currentAssignment,
+ dynamicState.newAssignment);
//Scheduling changed while running...
return killContainerForChangedAssignment(dynamicState, staticState);
}
dynamicState = updateAssignmentIfNeeded(dynamicState);
+ dynamicState = filterChangingBlobsFor(dynamicState, dynamicState.currentAssignment);
+ if (!dynamicState.changingBlobs.isEmpty()) {
+ //Kill the container and restart it
+ return killContainerForChangedBlobs(dynamicState, staticState);
+ }
+
if (dynamicState.container.didMainProcessExit()) {
numWorkersKilledProcessExit.mark();
LOG.warn("SLOT {}: main process has exited", staticState.port);
@@ -660,6 +943,8 @@ public class Slot extends Thread implements AutoCloseable {
}
static DynamicState handleEmpty(DynamicState dynamicState, StaticState staticState) throws InterruptedException, IOException {
+ assert dynamicState.changingBlobs.isEmpty();
+ assert dynamicState.pendingChangingBlobsAssignment == null;
if (!equivalent(dynamicState.newAssignment, dynamicState.currentAssignment)) {
return prepareForNewAssignmentNoWorkersRunning(dynamicState, staticState);
}
@@ -669,15 +954,17 @@ public class Slot extends Thread implements AutoCloseable {
if (dynamicState.profileActions != null && !dynamicState.profileActions.isEmpty()) {
//Nothing is scheduled here so throw away all of the profileActions
LOG.warn("Dropping {} no topology is running", dynamicState.profileActions);
- dynamicState = dynamicState.withProfileActions(Collections.<TopoProfileAction> emptySet(), Collections.<TopoProfileAction> emptySet());
+ dynamicState = dynamicState.withProfileActions(Collections.emptySet(), Collections.emptySet());
}
+ //Drop the change notifications we are not running anything right now
+ dynamicState = drainAllChangingBlobs(dynamicState);
Time.sleep(1000);
return dynamicState;
}
private final AtomicReference<LocalAssignment> newAssignment = new AtomicReference<>();
- private final AtomicReference<Set<TopoProfileAction>> profiling =
- new AtomicReference<Set<TopoProfileAction>>(new HashSet<TopoProfileAction>());
+ private final AtomicReference<Set<TopoProfileAction>> profiling = new AtomicReference<>(new HashSet<>());
+ private final BlockingQueue<BlobChangeing> changingBlobs = new LinkedBlockingQueue<>();
private final StaticState staticState;
private final IStormClusterState clusterState;
private volatile boolean done = false;
@@ -685,11 +972,11 @@ public class Slot extends Thread implements AutoCloseable {
private final AtomicReference<Map<Long, LocalAssignment>> cachedCurrentAssignments;
public Slot(AsyncLocalizer localizer, Map<String, Object> conf,
- ContainerLauncher containerLauncher, String host,
- int port, LocalState localState,
- IStormClusterState clusterState,
- ISupervisor iSupervisor,
- AtomicReference<Map<Long, LocalAssignment>> cachedCurrentAssignments) throws Exception {
+ ContainerLauncher containerLauncher, String host,
+ int port, LocalState localState,
+ IStormClusterState clusterState,
+ ISupervisor iSupervisor,
+ AtomicReference<Map<Long, LocalAssignment>> cachedCurrentAssignments) throws Exception {
super("SLOT_"+port);
this.cachedCurrentAssignments = cachedCurrentAssignments;
@@ -715,23 +1002,25 @@ public class Slot extends Thread implements AutoCloseable {
}
dynamicState = new DynamicState(currentAssignment, container, newAssignment);
- staticState = new StaticState(localizer,
- ObjectReader.getInt(conf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS)) * 1000,
- ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_WORKER_START_TIMEOUT_SECS)) * 1000,
- ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS)) * 1000,
- ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_MONITOR_FREQUENCY_SECS)) * 1000,
- containerLauncher,
- host,
- port,
- iSupervisor,
- localState);
+ staticState = new StaticState(localizer,
+ ObjectReader.getInt(conf.get(Config.SUPERVISOR_WORKER_TIMEOUT_SECS)) * 1000,
+ ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_WORKER_START_TIMEOUT_SECS)) * 1000,
+ ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_WORKER_SHUTDOWN_SLEEP_SECS)) * 1000,
+ ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_MONITOR_FREQUENCY_SECS)) * 1000,
+ containerLauncher,
+ host,
+ port,
+ iSupervisor,
+ localState,
+ this);
this.newAssignment.set(dynamicState.newAssignment);
if (MachineState.RUNNING == dynamicState.state) {
//We are running so we should recover the blobs.
- staticState.localizer.recoverRunningTopology(currentAssignment, port);
+ staticState.localizer.recoverRunningTopology(currentAssignment, port, this);
saveNewAssignment(currentAssignment);
}
- LOG.warn("SLOT {}:{} Starting in state {} - assignment {}", staticState.host, staticState.port, dynamicState.state, dynamicState.currentAssignment);
+ LOG.info("SLOT {}:{} Starting in state {} - assignment {}", staticState.host, staticState.port, dynamicState.state,
+ dynamicState.currentAssignment);
}
public MachineState getMachineState() {
@@ -739,13 +1028,24 @@ public class Slot extends Thread implements AutoCloseable {
}
/**
- * Set a new assignment asynchronously
+ * Set a new assignment asynchronously.
* @param newAssignment the new assignment for this slot to run, null to run nothing
*/
public void setNewAssignment(LocalAssignment newAssignment) {
this.newAssignment.set(newAssignment);
}
-
+
+ @Override
+ public void blobChanging(LocalAssignment assignment, int port, LocallyCachedBlob blob, GoodToGo go) {
+ assert port == staticState.port : "got a callaback that is not for us " + port + " != " + staticState.port;
+ //This is called async so lets assume that it is something we care about
+ try {
+ changingBlobs.put(new BlobChangeing(assignment, blob, go.getLatch()));
+ } catch (InterruptedException e) {
+ throw new RuntimeException("This should not have happend, but it did (the queue is unbounded)", e);
+ }
+ }
+
public void addProfilerActions(Set<TopoProfileAction> actions) {
if (actions != null) {
while(true) {
@@ -800,10 +1100,28 @@ public class Slot extends Thread implements AutoCloseable {
while(!done) {
Set<TopoProfileAction> origProfileActions = new HashSet<>(profiling.get());
Set<TopoProfileAction> removed = new HashSet<>(origProfileActions);
-
- DynamicState nextState =
- stateMachineStep(dynamicState.withNewAssignment(newAssignment.get())
- .withProfileActions(origProfileActions, dynamicState.pendingStopProfileActions), staticState);
+
+ Set<BlobChangeing> changingResourcesToHandle = dynamicState.changingBlobs;
+ if (!changingBlobs.isEmpty()) {
+ changingResourcesToHandle = new HashSet<>(changingResourcesToHandle);
+ changingBlobs.drainTo(changingResourcesToHandle);
+ Iterator<BlobChangeing> it = changingResourcesToHandle.iterator();
+
+ //Remove/Clean up changed requests that are not for us
+ while(it.hasNext()) {
+ BlobChangeing rc = it.next();
+ if (!forSameTopology(rc.assignment, dynamicState.currentAssignment) &&
+ !forSameTopology(rc.assignment, dynamicState.newAssignment)) {
+ rc.latch.countDown(); //Ignore the future
+ it.remove();
+ }
+ }
+ }
+
+ DynamicState nextState =
+ stateMachineStep(dynamicState.withNewAssignment(newAssignment.get())
+ .withProfileActions(origProfileActions, dynamicState.pendingStopProfileActions)
+ .withChangingBlobs(changingResourcesToHandle), staticState);
if (LOG.isDebugEnabled() || dynamicState.state != nextState.state) {
LOG.info("STATE {} -> {}", dynamicState, nextState);
http://git-wip-us.apache.org/repos/asf/storm/blob/e94b735a/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java b/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
index 043de8a..4b47502 100644
--- a/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
+++ b/storm-server/src/main/java/org/apache/storm/localizer/AsyncLocalizer.java
@@ -24,13 +24,10 @@ import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileWriter;
-import java.io.FilenameFilter;
import java.io.IOException;
-import java.io.OutputStream;
import java.io.PrintWriter;
-import java.net.JarURLConnection;
-import java.net.URL;
import java.net.URLDecoder;
+import java.nio.file.DirectoryStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -58,7 +55,6 @@ import java.util.stream.Collectors;
import org.apache.commons.io.FileUtils;
import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
-import org.apache.storm.blobstore.BlobStore;
import org.apache.storm.blobstore.ClientBlobStore;
import org.apache.storm.blobstore.InputStreamWithMeta;
import org.apache.storm.daemon.supervisor.AdvancedFSOps;
@@ -71,7 +67,6 @@ import org.apache.storm.streams.Pair;
import org.apache.storm.utils.ConfigUtils;
import org.apache.storm.utils.NimbusLeaderNotFoundException;
import org.apache.storm.utils.ObjectReader;
-import org.apache.storm.utils.ServerConfigUtils;
import org.apache.storm.utils.ServerUtils;
import org.apache.storm.utils.ShellUtils;
import org.apache.storm.utils.Utils;
@@ -110,11 +105,12 @@ public class AsyncLocalizer implements AutoCloseable {
private final AtomicReference<Map<Long, LocalAssignment>> currAssignment;
private final boolean isLocalMode;
- private final Map<String, LocalDownloadedResource> basicPending;
private final Map<String, LocalDownloadedResource> blobPending;
private final Map<String, Object> conf;
private final AdvancedFSOps fsOps;
private final boolean symlinksDisabled;
+ private final ConcurrentHashMap<String, LocallyCachedBlob> topologyBlobs = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, CompletableFuture<Void>> topologyBasicDownloaded = new ConcurrentHashMap<>();
// track resources - user to resourceSet
private final ConcurrentMap<String, LocalizedResourceSet> userRsrc = new ConcurrentHashMap<>();
@@ -126,16 +122,10 @@ public class AsyncLocalizer implements AutoCloseable {
// cleanup
private long cacheTargetSize;
- private long cacheCleanupPeriod;
-
-
- public AsyncLocalizer(Map<String, Object> conf, AtomicReference<Map<Long, LocalAssignment>> currAssignment,
- Map<Integer, LocalAssignment> portToAssignments) throws IOException {
- this(conf, ConfigUtils.supervisorLocalDir(conf), AdvancedFSOps.make(conf), currAssignment, portToAssignments);
- }
+ private final long cacheCleanupPeriod;
@VisibleForTesting
- AsyncLocalizer(Map<String, Object> conf, String baseDir, AdvancedFSOps ops,
+ AsyncLocalizer(Map<String, Object> conf, AdvancedFSOps ops, String baseDir,
AtomicReference<Map<Long, LocalAssignment>> currAssignment,
Map<Integer, LocalAssignment> portToAssignments) throws IOException {
@@ -146,7 +136,7 @@ public class AsyncLocalizer implements AutoCloseable {
// default cache size 10GB, converted to Bytes
cacheTargetSize = ObjectReader.getInt(conf.get(DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_TARGET_SIZE_MB),
10 * 1024).longValue() << 20;
- // default 30 seconds.
+ // default 30 seconds. (we cache the size so it is cheap to do)
cacheCleanupPeriod = ObjectReader.getInt(conf.get(
DaemonConfig.SUPERVISOR_LOCALIZER_CACHE_CLEANUP_INTERVAL_MS), 30 * 1000).longValue();
@@ -156,17 +146,141 @@ public class AsyncLocalizer implements AutoCloseable {
DaemonConfig.SUPERVISOR_BLOBSTORE_DOWNLOAD_MAX_RETRIES), 3);
execService = Executors.newScheduledThreadPool(threadPoolSize,
- new ThreadFactoryBuilder().setNameFormat("AsyncLocalizer Executor").build());
+ new ThreadFactoryBuilder().setNameFormat("AsyncLocalizer Executor - %d").build());
reconstructLocalizedResources();
symlinksDisabled = (boolean)conf.getOrDefault(Config.DISABLE_SYMLINKS, false);
- basicPending = new HashMap<>();
blobPending = new HashMap<>();
this.currAssignment = currAssignment;
recoverBlobReferences(portToAssignments);
}
+ public AsyncLocalizer(Map<String, Object> conf, AtomicReference<Map<Long, LocalAssignment>> currAssignment,
+ Map<Integer, LocalAssignment> portToAssignments) throws IOException {
+ this(conf, AdvancedFSOps.make(conf), ConfigUtils.supervisorLocalDir(conf), currAssignment, portToAssignments);
+ }
+
+ @VisibleForTesting
+ LocallyCachedBlob getTopoJar(final String topologyId) throws IOException {
+ String topoJarKey = ConfigUtils.masterStormJarKey(topologyId);
+ LocallyCachedBlob topoJar = topologyBlobs.get(topoJarKey);
+ if (topoJar == null) {
+ topoJar = new LocallyCachedTopologyBlob(topologyId, isLocalMode, conf, fsOps,
+ LocallyCachedTopologyBlob.TopologyBlobType.TOPO_JAR);
+ topologyBlobs.put(topoJarKey, topoJar);
+ }
+ return topoJar;
+ }
+
+ @VisibleForTesting
+ LocallyCachedBlob getTopoCode(final String topologyId) throws IOException {
+ String topoCodeKey = ConfigUtils.masterStormCodeKey(topologyId);
+ LocallyCachedBlob topoCode = topologyBlobs.get(topoCodeKey);
+ if (topoCode == null) {
+ topoCode = new LocallyCachedTopologyBlob(topologyId, isLocalMode, conf, fsOps,
+ LocallyCachedTopologyBlob.TopologyBlobType.TOPO_CODE);
+ topologyBlobs.put(topoCodeKey, topoCode);
+ }
+ return topoCode;
+ }
+
+ @VisibleForTesting
+ LocallyCachedBlob getTopoConf(final String topologyId) throws IOException {
+ String topoConfKey = ConfigUtils.masterStormConfKey(topologyId);
+ LocallyCachedBlob topoConf = topologyBlobs.get(topoConfKey);
+ if (topoConf == null) {
+ topoConf = new LocallyCachedTopologyBlob(topologyId, isLocalMode, conf, fsOps,
+ LocallyCachedTopologyBlob.TopologyBlobType.TOPO_CONF);
+ topologyBlobs.put(topoConfKey, topoConf);
+ }
+ return topoConf;
+ }
+
+ public synchronized CompletableFuture<Void> requestDownloadTopologyBlobs(final LocalAssignment assignment, final int port,
+ final BlobChangingCallback cb) throws IOException {
+ final String topologyId = assignment.get_topology_id();
+
+ CompletableFuture<Void> baseBlobs = requestDownloadBaseTopologyBlobs(assignment, port, cb);
+ return baseBlobs.thenComposeAsync((v) -> {
+ LocalDownloadedResource localResource = blobPending.get(topologyId);
+ if (localResource == null) {
+ Supplier<Void> supplier = new DownloadBlobs(topologyId, assignment.get_owner());
+ localResource = new LocalDownloadedResource(CompletableFuture.supplyAsync(supplier, execService));
+ blobPending.put(topologyId, localResource);
+ }
+ CompletableFuture<Void> r = localResource.reserve(port, assignment);
+ LOG.debug("Reserved blobs {} {}", topologyId, localResource);
+ return r;
+ });
+ }
+
+ @VisibleForTesting
+ synchronized CompletableFuture<Void> requestDownloadBaseTopologyBlobs(final LocalAssignment assignment, final int port,
+ BlobChangingCallback cb) throws IOException {
+ PortAndAssignment pna = new PortAndAssignment(port, assignment);
+ final String topologyId = assignment.get_topology_id();
+
+ LocallyCachedBlob topoJar = getTopoJar(topologyId);
+ topoJar.addReference(pna, cb);
+
+ LocallyCachedBlob topoCode = getTopoCode(topologyId);
+ topoCode.addReference(pna, cb);
+
+ LocallyCachedBlob topoConf = getTopoConf(topologyId);
+ topoConf.addReference(pna, cb);
+
+ CompletableFuture<Void> ret = topologyBasicDownloaded.get(topologyId);
+ if (ret == null) {
+ ret = downloadOrUpdate(topoJar, topoCode, topoConf);
+ }
+ return ret;
+ }
+
+ private static final int ATTEMPTS_INTERVAL_TIME = 100;
+
+ private CompletableFuture<Void> downloadOrUpdate(LocallyCachedBlob ... blobs) {
+ CompletableFuture<Void> [] all = new CompletableFuture[blobs.length];
+ for (int i = 0; i < blobs.length; i++) {
+ final LocallyCachedBlob blob = blobs[i];
+ all[i] = CompletableFuture.runAsync(() -> {
+ LOG.debug("STARTING download of {}", blob);
+ try (ClientBlobStore blobStore = ServerUtils.getClientBlobStoreForSupervisor(conf)) {
+ boolean done = false;
+ long failures = 0;
+ while (!done) {
+ try {
+ synchronized (blob) {
+ long localVersion = blob.getLocalVersion();
+ long remoteVersion = blob.getRemoteVersion(blobStore);
+ if (localVersion != remoteVersion) {
+ try {
+ long newVersion = blob.downloadToTempLocation(blobStore);
+ blob.informAllOfChangeAndWaitForConsensus();
+ blob.commitNewVersion(newVersion);
+ blob.informAllChangeComplete();
+ } finally {
+ blob.cleanupOrphanedData();
+ }
+ }
+ }
+ done = true;
+ } catch (Exception e) {
+ failures++;
+ if (failures > blobDownloadRetries) {
+ throw new RuntimeException("Could not download...", e);
+ }
+ LOG.warn("Failed to download blob {} will try again in {} ms", blob, ATTEMPTS_INTERVAL_TIME, e);
+ Utils.sleep(ATTEMPTS_INTERVAL_TIME);
+ }
+ }
+ }
+ LOG.debug("FINISHED download of {}", blob);
+ }, execService);
+ }
+ return CompletableFuture.allOf(all);
+ }
+
/**
* For each of the downloaded topologies, adds references to the blobs that the topologies are using. This is used to reconstruct the
* cache on restart.
@@ -211,11 +325,20 @@ public class AsyncLocalizer implements AutoCloseable {
* with a suffix. The runnable is intended to be run periodically by a timer, created elsewhere.
*/
private void updateBlobs() {
+ for (LocallyCachedBlob blob : topologyBlobs.values()) {
+ if (blob.isUsed()) {
+ try {
+ downloadOrUpdate(blob);
+ } catch (Exception e) {
+ LOG.error("Could not update {}", blob, e);
+ }
+ }
+ }
try {
Map<String, String> topoIdToOwner = currAssignment.get().values().stream()
.map((la) -> Pair.of(la.get_topology_id(), la.get_owner()))
.distinct()
- .collect(Collectors.toMap((p) -> p.getFirst(), (p) -> p.getSecond()));
+ .collect(Collectors.toMap(Pair::getFirst, Pair::getSecond));
for (String topoId : readDownloadedTopologyIds(conf)) {
String owner = topoIdToOwner.get(topoId);
if (owner == null) {
@@ -272,129 +395,6 @@ public class AsyncLocalizer implements AutoCloseable {
}
//ILocalizer
- private class DownloadBaseBlobsDistributed implements Supplier<Void> {
- protected final String topologyId;
- protected final File stormRoot;
- protected final LocalAssignment assignment;
- protected final String owner;
-
- public DownloadBaseBlobsDistributed(String topologyId, LocalAssignment assignment) throws IOException {
- this.topologyId = topologyId;
- stormRoot = new File(ConfigUtils.supervisorStormDistRoot(conf, this.topologyId));
- this.assignment = assignment;
- owner = assignment.get_owner();
- }
-
- protected void downloadBaseBlobs(File tmproot) throws Exception {
- String stormJarKey = ConfigUtils.masterStormJarKey(topologyId);
- String stormCodeKey = ConfigUtils.masterStormCodeKey(topologyId);
- String topoConfKey = ConfigUtils.masterStormConfKey(topologyId);
- String jarPath = ConfigUtils.supervisorStormJarPath(tmproot.getAbsolutePath());
- String codePath = ConfigUtils.supervisorStormCodePath(tmproot.getAbsolutePath());
- String confPath = ConfigUtils.supervisorStormConfPath(tmproot.getAbsolutePath());
- fsOps.forceMkdir(tmproot);
- fsOps.restrictDirectoryPermissions(tmproot);
- ClientBlobStore blobStore = ServerUtils.getClientBlobStoreForSupervisor(conf);
- try {
- ServerUtils.downloadResourcesAsSupervisor(stormJarKey, jarPath, blobStore);
- ServerUtils.downloadResourcesAsSupervisor(stormCodeKey, codePath, blobStore);
- ServerUtils.downloadResourcesAsSupervisor(topoConfKey, confPath, blobStore);
- } finally {
- blobStore.shutdown();
- }
- ServerUtils.extractDirFromJar(jarPath, ServerConfigUtils.RESOURCES_SUBDIR, tmproot);
- }
-
- @Override
- public Void get() {
- try {
- if (fsOps.fileExists(stormRoot)) {
- if (!fsOps.supportsAtomicDirectoryMove()) {
- LOG.warn("{} may have partially downloaded blobs, recovering", topologyId);
- fsOps.deleteIfExists(stormRoot);
- } else {
- LOG.warn("{} already downloaded blobs, skipping", topologyId);
- return null;
- }
- }
- boolean deleteAll = true;
- String tmproot = ServerConfigUtils.supervisorTmpDir(conf) + File.separator + Utils.uuid();
- File tr = new File(tmproot);
- try {
- downloadBaseBlobs(tr);
- if (assignment.is_set_total_node_shared()) {
- File sharedMemoryDirTmpLocation = new File(tr, "shared_by_topology");
- //We need to create a directory for shared memory to write to (we should not encourage this though)
- Path path = sharedMemoryDirTmpLocation.toPath();
- Files.createDirectories(path);
- }
- fsOps.moveDirectoryPreferAtomic(tr, stormRoot);
- fsOps.setupStormCodeDir(owner, stormRoot);
- if (assignment.is_set_total_node_shared()) {
- File sharedMemoryDir = new File(stormRoot, "shared_by_topology");
- fsOps.setupWorkerArtifactsDir(owner, sharedMemoryDir);
- }
- deleteAll = false;
- } finally {
- if (deleteAll) {
- LOG.warn("Failed to download basic resources for topology-id {}", topologyId);
- fsOps.deleteIfExists(tr);
- fsOps.deleteIfExists(stormRoot);
- }
- }
- return null;
- } catch (Exception e) {
- LOG.warn("Caught Exception While Downloading (rethrowing)... ", e);
- throw new RuntimeException(e);
- }
- }
- }
-
- private class DownloadBaseBlobsLocal extends DownloadBaseBlobsDistributed {
-
- public DownloadBaseBlobsLocal(String topologyId, LocalAssignment assignment) throws IOException {
- super(topologyId, assignment);
- }
-
- @Override
- protected void downloadBaseBlobs(File tmproot) throws Exception {
- fsOps.forceMkdir(tmproot);
- String stormCodeKey = ConfigUtils.masterStormCodeKey(topologyId);
- String topoConfKey = ConfigUtils.masterStormConfKey(topologyId);
- File codePath = new File(ConfigUtils.supervisorStormCodePath(tmproot.getAbsolutePath()));
- File confPath = new File(ConfigUtils.supervisorStormConfPath(tmproot.getAbsolutePath()));
- BlobStore blobStore = ServerUtils.getNimbusBlobStore(conf, null);
- try {
- try (OutputStream codeOutStream = fsOps.getOutputStream(codePath)){
- blobStore.readBlobTo(stormCodeKey, codeOutStream, null);
- }
- try (OutputStream confOutStream = fsOps.getOutputStream(confPath)) {
- blobStore.readBlobTo(topoConfKey, confOutStream, null);
- }
- } finally {
- blobStore.shutdown();
- }
-
- ClassLoader classloader = Thread.currentThread().getContextClassLoader();
- String resourcesJar = resourcesJar();
- URL url = classloader.getResource(ServerConfigUtils.RESOURCES_SUBDIR);
-
- String targetDir = tmproot + File.separator;
- if (resourcesJar != null) {
- LOG.info("Extracting resources from jar at {} to {}", resourcesJar, targetDir);
- ServerUtils.extractDirFromJar(resourcesJar, ServerConfigUtils.RESOURCES_SUBDIR, new File(targetDir));
- } else if (url != null) {
- LOG.info("Copying resources at {} to {}", url, targetDir);
- if ("jar".equals(url.getProtocol())) {
- JarURLConnection urlConnection = (JarURLConnection) url.openConnection();
- ServerUtils.extractDirFromJar(urlConnection.getJarFileURL().getFile(), ServerConfigUtils.RESOURCES_SUBDIR, new File(targetDir));
- } else {
- fsOps.copyDirectory(new File(url.getFile()), new File(targetDir, ConfigUtils.RESOURCES_SUBDIR));
- }
- }
- }
- }
-
private class DownloadBlobs implements Supplier<Void> {
private final String topologyId;
private final String topoOwner;
@@ -472,51 +472,21 @@ public class AsyncLocalizer implements AutoCloseable {
}
}
- public synchronized CompletableFuture<Void> requestDownloadBaseTopologyBlobs(final LocalAssignment assignment, final int port) throws IOException {
+ public synchronized void recoverRunningTopology(final LocalAssignment assignment, final int port,
+ final BlobChangingCallback cb) throws IOException {
+ PortAndAssignment pna = new PortAndAssignment(port, assignment);
final String topologyId = assignment.get_topology_id();
- LocalDownloadedResource localResource = basicPending.get(topologyId);
- if (localResource == null) {
- Supplier<Void> supplier;
- if (isLocalMode) {
- supplier = new DownloadBaseBlobsLocal(topologyId, assignment);
- } else {
- supplier = new DownloadBaseBlobsDistributed(topologyId, assignment);
- }
- localResource = new LocalDownloadedResource(CompletableFuture.supplyAsync(supplier, execService));
- basicPending.put(topologyId, localResource);
- }
- CompletableFuture<Void> ret = localResource.reserve(port, assignment);
- LOG.debug("Reserved basic {} {}", topologyId, localResource);
- return ret;
- }
- private static String resourcesJar() throws IOException {
- String path = ServerUtils.currentClasspath();
- if (path == null) {
- return null;
- }
+ LocallyCachedBlob topoJar = getTopoJar(topologyId);
+ topoJar.addReference(pna, cb);
- for (String jpath : path.split(File.pathSeparator)) {
- if (jpath.endsWith(".jar")) {
- if (ServerUtils.zipDoesContainDir(jpath, ServerConfigUtils.RESOURCES_SUBDIR)) {
- return jpath;
- }
- }
- }
- return null;
- }
+ LocallyCachedBlob topoCode = getTopoCode(topologyId);
+ topoCode.addReference(pna, cb);
- public synchronized void recoverRunningTopology(LocalAssignment assignment, int port) {
- final String topologyId = assignment.get_topology_id();
- LocalDownloadedResource localResource = basicPending.get(topologyId);
- if (localResource == null) {
- localResource = new LocalDownloadedResource(ALL_DONE_FUTURE);
- basicPending.put(topologyId, localResource);
- }
- localResource.reserve(port, assignment);
- LOG.debug("Recovered basic {} {}", topologyId, localResource);
+ LocallyCachedBlob topoConf = getTopoConf(topologyId);
+ topoConf.addReference(pna, cb);
- localResource = blobPending.get(topologyId);
+ LocalDownloadedResource localResource = blobPending.get(topologyId);
if (localResource == null) {
localResource = new LocalDownloadedResource(ALL_DONE_FUTURE);
blobPending.put(topologyId, localResource);
@@ -525,19 +495,6 @@ public class AsyncLocalizer implements AutoCloseable {
LOG.debug("Recovered blobs {} {}", topologyId, localResource);
}
- public synchronized CompletableFuture<Void> requestDownloadTopologyBlobs(LocalAssignment assignment, int port) {
- final String topologyId = assignment.get_topology_id();
- LocalDownloadedResource localResource = blobPending.get(topologyId);
- if (localResource == null) {
- Supplier<Void> supplier = new DownloadBlobs(topologyId, assignment.get_owner());
- localResource = new LocalDownloadedResource(CompletableFuture.supplyAsync(supplier, execService));
- blobPending.put(topologyId, localResource);
- }
- CompletableFuture<Void> ret = localResource.reserve(port, assignment);
- LOG.debug("Reserved blobs {} {}", topologyId, localResource);
- return ret;
- }
-
/**
* Remove this assignment/port as blocking resources from being cleaned up.
*
@@ -546,8 +503,29 @@ public class AsyncLocalizer implements AutoCloseable {
* @throws IOException on any error
*/
public synchronized void releaseSlotFor(LocalAssignment assignment, int port) throws IOException {
+ PortAndAssignment pna = new PortAndAssignment(port, assignment);
final String topologyId = assignment.get_topology_id();
LOG.debug("Releasing slot for {} {}", topologyId, port);
+
+ String topoJarKey = ConfigUtils.masterStormJarKey(topologyId);
+ String topoCodeKey = ConfigUtils.masterStormCodeKey(topologyId);
+ String topoConfKey = ConfigUtils.masterStormConfKey(topologyId);
+
+ LocallyCachedBlob topoJar = topologyBlobs.get(topoJarKey);
+ if (topoJar != null) {
+ topoJar.removeReference(pna);
+ }
+
+ LocallyCachedBlob topoCode = topologyBlobs.get(topoCodeKey);
+ if (topoCode != null) {
+ topoCode.removeReference(pna);
+ }
+
+ LocallyCachedBlob topoConfBlob = topologyBlobs.get(topoConfKey);
+ if (topoConfBlob != null) {
+ topoConfBlob.removeReference(pna);
+ }
+
LocalDownloadedResource localResource = blobPending.get(topologyId);
if (localResource == null || !localResource.release(port, assignment)) {
LOG.warn("Released blob reference {} {} for something that we didn't have {}", topologyId, port, localResource);
@@ -574,38 +552,12 @@ public class AsyncLocalizer implements AutoCloseable {
} else {
LOG.debug("Released blob reference {} {} still waiting on {}", topologyId, port, localResource);
}
-
- localResource = basicPending.get(topologyId);
- if (localResource == null || !localResource.release(port, assignment)) {
- LOG.warn("Released basic reference {} {} for something that we didn't have {}", topologyId, port, localResource);
- } else if (localResource.isDone()){
- LOG.info("Released blob reference {} {} Cleaning up basic files...", topologyId, port);
- basicPending.remove(topologyId);
- String path = ConfigUtils.supervisorStormDistRoot(conf, topologyId);
- fsOps.deleteIfExists(new File(path), null, "rmr "+topologyId);
- } else {
- LOG.debug("Released basic reference {} {} still waiting on {}", topologyId, port, localResource);
- }
- }
-
- public synchronized void cleanupUnusedTopologies() throws IOException {
- File distRoot = new File(ConfigUtils.supervisorStormDistRoot(conf));
- LOG.info("Cleaning up unused topologies in {}", distRoot);
- File[] children = distRoot.listFiles();
- if (children != null) {
- for (File topoDir : children) {
- String topoId = URLDecoder.decode(topoDir.getName(), "UTF-8");
- if (basicPending.get(topoId) == null && blobPending.get(topoId) == null) {
- fsOps.deleteIfExists(topoDir, null, "rmr " + topoId);
- }
- }
- }
}
//From Localizer
- // For testing, it allows setting size in bytes
- protected void setTargetCacheSize(long size) {
+ @VisibleForTesting
+ void setTargetCacheSize(long size) {
cacheTargetSize = size;
}
@@ -615,33 +567,33 @@ public class AsyncLocalizer implements AutoCloseable {
}
// baseDir/supervisor/usercache/
- protected File getUserCacheDir() {
+ private File getUserCacheDir() {
return new File(localBaseDir, USERCACHE);
}
// baseDir/supervisor/usercache/user1/
- protected File getLocalUserDir(String userName) {
+ File getLocalUserDir(String userName) {
return new File(getUserCacheDir(), userName);
}
// baseDir/supervisor/usercache/user1/filecache
- public File getLocalUserFileCacheDir(String userName) {
+ File getLocalUserFileCacheDir(String userName) {
return new File(getLocalUserDir(userName), FILECACHE);
}
// baseDir/supervisor/usercache/user1/filecache/files
- protected File getCacheDirForFiles(File dir) {
+ private File getCacheDirForFiles(File dir) {
return new File(dir, FILESDIR);
}
// get the directory to put uncompressed archives in
// baseDir/supervisor/usercache/user1/filecache/archives
- protected File getCacheDirForArchives(File dir) {
+ private File getCacheDirForArchives(File dir) {
return new File(dir, ARCHIVESDIR);
}
- protected void addLocalizedResourceInDir(String dir, LocalizedResourceSet lrsrcSet,
- boolean uncompress) {
+ private void addLocalizedResourceInDir(String dir, LocalizedResourceSet lrsrcSet,
+ boolean uncompress) {
File[] lrsrcs = readCurrentBlobs(dir);
if (lrsrcs != null) {
@@ -662,22 +614,17 @@ public class AsyncLocalizer implements AutoCloseable {
}
// Looks for files in the directory with .current suffix
- protected File[] readCurrentBlobs(String location) {
+ private File[] readCurrentBlobs(String location) {
File dir = new File(location);
File[] files = null;
if (dir.exists()) {
- files = dir.listFiles(new FilenameFilter() {
- @Override
- public boolean accept(File dir, String name) {
- return name.toLowerCase().endsWith(ServerUtils.DEFAULT_CURRENT_BLOB_SUFFIX);
- }
- });
+ files = dir.listFiles((d, name) -> name.toLowerCase().endsWith(ServerUtils.DEFAULT_CURRENT_BLOB_SUFFIX));
}
return files;
}
// Check to see if there are any existing files already localized.
- protected void reconstructLocalizedResources() {
+ private void reconstructLocalizedResources() {
try {
LOG.info("Reconstruct localized resource: " + getUserCacheDir().getPath());
Collection<File> users = ConfigUtils.readDirFiles(getUserCacheDir().getPath());
@@ -705,8 +652,8 @@ public class AsyncLocalizer implements AutoCloseable {
}
// ignores invalid user/topo/key
- public synchronized void removeBlobReference(String key, String user, String topo,
- boolean uncompress) throws AuthorizationException, KeyNotFoundException {
+ synchronized void removeBlobReference(String key, String user, String topo,
+ boolean uncompress) throws AuthorizationException, KeyNotFoundException {
LocalizedResourceSet lrsrcSet = userRsrc.get(user);
if (lrsrcSet != null) {
LocalizedResource lrsrc = lrsrcSet.get(key, uncompress);
@@ -723,8 +670,8 @@ public class AsyncLocalizer implements AutoCloseable {
}
}
- public synchronized void addReferences(List<LocalResource> localresource, String user,
- String topo) {
+ synchronized void addReferences(List<LocalResource> localresource, String user,
+ String topo) {
LocalizedResourceSet lrsrcSet = userRsrc.get(user);
if (lrsrcSet != null) {
for (LocalResource blob : localresource) {
@@ -746,9 +693,9 @@ public class AsyncLocalizer implements AutoCloseable {
* This function either returns the blob in the existing cache or if it doesn't exist in the
* cache, it will download the blob and will block until the download is complete.
*/
- public LocalizedResource getBlob(LocalResource localResource, String user, String topo,
+ LocalizedResource getBlob(LocalResource localResource, String user, String topo,
File userFileDir) throws AuthorizationException, KeyNotFoundException, IOException {
- ArrayList<LocalResource> arr = new ArrayList<LocalResource>();
+ ArrayList<LocalResource> arr = new ArrayList<>();
arr.add(localResource);
List<LocalizedResource> results = getBlobs(arr, user, topo, userFileDir);
if (results.isEmpty() || results.size() != 1) {
@@ -758,15 +705,15 @@ public class AsyncLocalizer implements AutoCloseable {
return results.get(0);
}
- protected boolean isLocalizedResourceDownloaded(LocalizedResource lrsrc) {
+ private boolean isLocalizedResourceDownloaded(LocalizedResource lrsrc) {
File rsrcFileCurrent = new File(lrsrc.getCurrentSymlinkPath());
File rsrcFileWithVersion = new File(lrsrc.getFilePathWithVersion());
File versionFile = new File(lrsrc.getVersionFilePath());
return (rsrcFileWithVersion.exists() && rsrcFileCurrent.exists() && versionFile.exists());
}
- protected boolean isLocalizedResourceUpToDate(LocalizedResource lrsrc,
- ClientBlobStore blobstore) throws AuthorizationException, KeyNotFoundException {
+ private boolean isLocalizedResourceUpToDate(LocalizedResource lrsrc,
+ ClientBlobStore blobstore) throws AuthorizationException, KeyNotFoundException {
String localFile = lrsrc.getFilePath();
long nimbusBlobVersion = ServerUtils.nimbusVersionOfBlob(lrsrc.getKey(), blobstore);
long currentBlobVersion = ServerUtils.localVersionOfBlob(localFile);
@@ -781,8 +728,8 @@ public class AsyncLocalizer implements AutoCloseable {
* This function updates blobs on the supervisor. It uses a separate thread pool and runs
* asynchronously of the download and delete.
*/
- public List<LocalizedResource> updateBlobs(List<LocalResource> localResources,
- String user) throws AuthorizationException, KeyNotFoundException, IOException {
+ List<LocalizedResource> updateBlobs(List<LocalResource> localResources,
+ String user) throws AuthorizationException, KeyNotFoundException, IOException {
LocalizedResourceSet lrsrcSet = userRsrc.get(user);
ArrayList<LocalizedResource> results = new ArrayList<>();
ArrayList<Callable<LocalizedResource>> updates = new ArrayList<>();
@@ -799,10 +746,8 @@ public class AsyncLocalizer implements AutoCloseable {
LocalizedResource lrsrc = lrsrcSet.get(key, localResource.shouldUncompress());
if (lrsrc == null) {
LOG.warn("blob requested for update doesn't exist: {}", key);
- continue;
} else if ((boolean) conf.getOrDefault(Config.DISABLE_SYMLINKS, false)) {
LOG.warn("symlinks are disabled so blobs cannot be downloaded.");
- continue;
} else {
// update it if either the version isn't the latest or if any local blob files are missing
if (!isLocalizedResourceUpToDate(lrsrc, blobstore) ||
@@ -855,8 +800,8 @@ public class AsyncLocalizer implements AutoCloseable {
* cache, it downloads them in parallel (up to SUPERVISOR_BLOBSTORE_DOWNLOAD_THREAD_COUNT)
* and will block until all of them have been downloaded
*/
- public synchronized List<LocalizedResource> getBlobs(List<LocalResource> localResources,
- String user, String topo, File userFileDir)
+ synchronized List<LocalizedResource> getBlobs(List<LocalResource> localResources,
+ String user, String topo, File userFileDir)
throws AuthorizationException, KeyNotFoundException, IOException {
if ((boolean) conf.getOrDefault(Config.DISABLE_SYMLINKS, false)) {
throw new KeyNotFoundException("symlinks are disabled so blobs cannot be downloaded.");
@@ -942,8 +887,8 @@ public class AsyncLocalizer implements AutoCloseable {
private boolean uncompress;
private boolean isUpdate;
- public DownloadBlob(AsyncLocalizer localizer, Map<String, Object> conf, String key, File localFile,
- String user, boolean uncompress, boolean update) {
+ DownloadBlob(AsyncLocalizer localizer, Map<String, Object> conf, String key, File localFile,
+ String user, boolean uncompress, boolean update) {
this.localizer = localizer;
this.conf = conf;
this.key = key;
@@ -1098,7 +1043,7 @@ public class AsyncLocalizer implements AutoCloseable {
}
}
- public void setBlobPermissions(Map<String, Object> conf, String user, String path)
+ private void setBlobPermissions(Map<String, Object> conf, String user, String path)
throws IOException {
if (!ObjectReader.getBoolean(conf.get(Config.SUPERVISOR_RUN_WORKER_AS_USER), false)) {
@@ -1109,7 +1054,7 @@ public class AsyncLocalizer implements AutoCloseable {
String stormHome = System.getProperty("storm.home");
wlCommand = stormHome + "/bin/worker-launcher";
}
- List<String> command = new ArrayList<String>(Arrays.asList(wlCommand, user, "blob", path));
+ List<String> command = new ArrayList<>(Arrays.asList(wlCommand, user, "blob", path));
String[] commandArray = command.toArray(new String[command.size()]);
ShellUtils.ShellCommandExecutor shExec = new ShellUtils.ShellCommandExecutor(commandArray);
@@ -1127,15 +1072,54 @@ public class AsyncLocalizer implements AutoCloseable {
}
}
+ private interface ConsumePathAndId {
+ void accept(Path path, String topologyId) throws IOException;
+ }
- public synchronized void cleanup() {
+ private void forEachTopologyDistDir(ConsumePathAndId consumer) throws IOException {
+ Path stormCodeRoot = Paths.get(ConfigUtils.supervisorStormDistRoot(conf));
+ if (Files.exists(stormCodeRoot) && Files.isDirectory(stormCodeRoot)) {
+ try (DirectoryStream<Path> children = Files.newDirectoryStream(stormCodeRoot)) {
+ for (Path child : children) {
+ if (Files.isDirectory(child)) {
+ String topologyId = child.getFileName().toString();
+ consumer.accept(child, topologyId);
+ }
+ }
+ }
+ }
+ }
+
+ @VisibleForTesting
+ synchronized void cleanup() {
LocalizedResourceRetentionSet toClean = new LocalizedResourceRetentionSet(cacheTargetSize);
// need one large set of all and then clean via LRU
for (LocalizedResourceSet t : userRsrc.values()) {
toClean.addResources(t);
LOG.debug("Resources to be cleaned after adding {} : {}", t.getUser(), toClean);
}
+
+ toClean.addResources(topologyBlobs);
toClean.cleanup();
+
+ HashSet<String> safeTopologyIds = new HashSet<>();
+ for (String blobKey : topologyBlobs.keySet()) {
+ safeTopologyIds.add(ConfigUtils.getIdFromBlobKey(blobKey));
+ }
+
+ //Deleting this early does not hurt anything
+ topologyBasicDownloaded.keySet().removeIf(topoId -> !safeTopologyIds.contains(topoId));
+
+ try {
+ forEachTopologyDistDir((p, topologyId) -> {
+ if (!safeTopologyIds.contains(topologyId)) {
+ fsOps.deleteIfExists(p.toFile());
+ }
+ });
+ } catch (Exception e) {
+ LOG.error("Could not read topology directories for cleanup", e);
+ }
+
LOG.debug("Resource cleanup: {}", toClean);
for (LocalizedResourceSet t : userRsrc.values()) {
if (t.getSize() == 0) {
@@ -1154,5 +1138,4 @@ public class AsyncLocalizer implements AutoCloseable {
}
}
}
-
}
http://git-wip-us.apache.org/repos/asf/storm/blob/e94b735a/storm-server/src/main/java/org/apache/storm/localizer/BlobChangingCallback.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/localizer/BlobChangingCallback.java b/storm-server/src/main/java/org/apache/storm/localizer/BlobChangingCallback.java
new file mode 100644
index 0000000..e99f95b
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/localizer/BlobChangingCallback.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.storm.localizer;
+
+import org.apache.storm.generated.LocalAssignment;
+
+/**
+ * Callback for when a localized blob is going to change.
+ */
+public interface BlobChangingCallback {
+
+ /**
+ * Informs the listener that a blob has changed and is ready to update and replace a localized blob that has been marked as
+ * tied to the life cycle of the worker process.
+ *
+ * If `go.getLatch()` is never called before the method completes it is assumed that
+ * the listener is good with the blob changing.
+ * @param assignment the assignment this resource and callback are registered with.
+ * @param port the port that this resource and callback are registered with.
+ * @param blob the blob that is going to change.
+ * @param go a way to indicate if the listener is ready for the resource to change.
+ */
+ void blobChanging(LocalAssignment assignment, int port, LocallyCachedBlob blob, GoodToGo go);
+}
http://git-wip-us.apache.org/repos/asf/storm/blob/e94b735a/storm-server/src/main/java/org/apache/storm/localizer/GoodToGo.java
----------------------------------------------------------------------
diff --git a/storm-server/src/main/java/org/apache/storm/localizer/GoodToGo.java b/storm-server/src/main/java/org/apache/storm/localizer/GoodToGo.java
new file mode 100644
index 0000000..04c7a06
--- /dev/null
+++ b/storm-server/src/main/java/org/apache/storm/localizer/GoodToGo.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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.
+ */
+
+package org.apache.storm.localizer;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+
+/**
+ * Used as a way to give feedback that the listener is ready for the caller to change the blob.
+ * By calling @{link GoodToGo#getLatch()} the listener indicates that it wants to block
+ * changing the blob until the CountDownLatch is triggered with a call to @{link CountDownLatch#countDown()}.
+ */
+public class GoodToGo {
+ public static class GoodToGoLatch {
+ private final CountDownLatch latch;
+ private final Future<Void> doneChangeing;
+ private boolean wasCounted = false;
+
+ public GoodToGoLatch(CountDownLatch latch, Future<Void> doneChangeing) {
+ this.latch = latch;
+ this.doneChangeing = doneChangeing;
+ }
+
+ public synchronized Future<Void> countDown() {
+ if (!wasCounted) {
+ latch.countDown();
+ wasCounted = true;
+ }
+ return doneChangeing;
+ }
+ }
+
+ private final GoodToGoLatch latch;
+ private boolean gotLatch = false;
+
+ public GoodToGo(CountDownLatch latch, Future<Void> doneChangeing) {
+ this.latch = new GoodToGoLatch(latch, doneChangeing);
+ }
+
+ /**
+ * Get the latch and indicate that you want to block the blob being changed.
+ * @return the latch to use when you are ready.
+ */
+ public synchronized GoodToGoLatch getLatch() {
+ gotLatch = true;
+ return latch;
+ }
+
+ synchronized void countDownIfLatchWasNotGotten() {
+ if (!gotLatch) {
+ latch.countDown();
+ }
+ }
+}