You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by tillrohrmann <gi...@git.apache.org> on 2017/04/26 14:45:28 UTC
[GitHub] flink pull request #3536: [FLINK-6048] Asynchronous snapshots for heap-based...
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/3536#discussion_r113465977
--- Diff: flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultOperatorStateBackend.java ---
@@ -131,59 +169,134 @@ public void dispose() {
@Override
public RunnableFuture<OperatorStateHandle> snapshot(
- long checkpointId,
- long timestamp,
- CheckpointStreamFactory streamFactory,
- CheckpointOptions checkpointOptions) throws Exception {
+ final long checkpointId,
+ final long timestamp,
+ final CheckpointStreamFactory streamFactory,
+ final CheckpointOptions checkpointOptions) throws Exception {
+
+ final long syncStartTime = System.currentTimeMillis();
if (registeredStates.isEmpty()) {
return DoneFuture.nullValue();
}
- List<OperatorBackendSerializationProxy.StateMetaInfo<?>> metaInfoList =
- new ArrayList<>(registeredStates.size());
-
- for (Map.Entry<String, PartitionableListState<?>> entry : registeredStates.entrySet()) {
- PartitionableListState<?> state = entry.getValue();
- OperatorBackendSerializationProxy.StateMetaInfo<?> metaInfo =
- new OperatorBackendSerializationProxy.StateMetaInfo<>(
- state.getName(),
- state.getPartitionStateSerializer(),
- state.getAssignmentMode());
- metaInfoList.add(metaInfo);
+ final Map<String, PartitionableListState<?>> registeredStatesDeepCopies =
+ new HashMap<>(registeredStates.size());
+
+ // eagerly create deep copies of the list states in the sync phase, so that we can use them in the async writing
+ for (Map.Entry<String, PartitionableListState<?>> entry : this.registeredStates.entrySet()) {
+
+ PartitionableListState<?> listState = entry.getValue();
+ if (null != listState) {
+ listState = listState.deepCopy();
+ }
+ registeredStatesDeepCopies.put(entry.getKey(), listState);
}
- Map<String, OperatorStateHandle.StateMetaInfo> writtenStatesMetaData = new HashMap<>(registeredStates.size());
+ // implementation of the async IO operation, based on FutureTask
+ final AbstractAsyncIOCallable<OperatorStateHandle, CheckpointStreamFactory.CheckpointStateOutputStream> ioCallable =
+ new AbstractAsyncIOCallable<OperatorStateHandle, CheckpointStreamFactory.CheckpointStateOutputStream>() {
+
+ AtomicBoolean open = new AtomicBoolean(false);
+
+ @Override
+ public CheckpointStreamFactory.CheckpointStateOutputStream openIOHandle() throws Exception {
+ if (open.compareAndSet(false, true)) {
+ CheckpointStreamFactory.CheckpointStateOutputStream stream =
+ streamFactory.createCheckpointStateOutputStream(checkpointId, timestamp);
+ try {
+ closeStreamOnCancelRegistry.registerClosable(stream);
+ return stream;
+ } catch (Exception ex) {
+ open.set(false);
+ throw ex;
+ }
+ } else {
+ throw new IOException("Operation already opened.");
+ }
+ }
- CheckpointStreamFactory.CheckpointStateOutputStream out = streamFactory.
- createCheckpointStateOutputStream(checkpointId, timestamp);
+ @Override
+ public OperatorStateHandle performOperation() throws Exception {
+ long asyncStartTime = System.currentTimeMillis();
- try {
- closeStreamOnCancelRegistry.registerClosable(out);
+ final Map<String, OperatorStateHandle.StateMetaInfo> writtenStatesMetaData =
+ new HashMap<>(registeredStatesDeepCopies.size());
- DataOutputView dov = new DataOutputViewStreamWrapper(out);
+ List<OperatorBackendSerializationProxy.StateMetaInfo<?>> metaInfoList =
+ new ArrayList<>(registeredStatesDeepCopies.size());
- OperatorBackendSerializationProxy backendSerializationProxy =
- new OperatorBackendSerializationProxy(metaInfoList);
+ for (Map.Entry<String, PartitionableListState<?>> entry :
+ registeredStatesDeepCopies.entrySet()) {
- backendSerializationProxy.write(dov);
+ PartitionableListState<?> state = entry.getValue();
+ OperatorBackendSerializationProxy.StateMetaInfo<?> metaInfo =
+ new OperatorBackendSerializationProxy.StateMetaInfo<>(
+ state.getName(),
+ state.getPartitionStateSerializer(),
+ state.getAssignmentMode());
+ metaInfoList.add(metaInfo);
+ }
- dov.writeInt(registeredStates.size());
- for (Map.Entry<String, PartitionableListState<?>> entry : registeredStates.entrySet()) {
+ CheckpointStreamFactory.CheckpointStateOutputStream out = getIoHandle();
+ DataOutputView dov = new DataOutputViewStreamWrapper(out);
- PartitionableListState<?> value = entry.getValue();
- long[] partitionOffsets = value.write(out);
- OperatorStateHandle.Mode mode = value.getAssignmentMode();
- writtenStatesMetaData.put(entry.getKey(), new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
- }
+ OperatorBackendSerializationProxy backendSerializationProxy =
+ new OperatorBackendSerializationProxy(metaInfoList);
- OperatorStateHandle handle = new OperatorStateHandle(writtenStatesMetaData, out.closeAndGetHandle());
+ backendSerializationProxy.write(dov);
- return new DoneFuture<>(handle);
- } finally {
- closeStreamOnCancelRegistry.unregisterClosable(out);
- out.close();
+ dov.writeInt(registeredStatesDeepCopies.size());
+
+ for (Map.Entry<String, PartitionableListState<?>> entry :
+ registeredStatesDeepCopies.entrySet()) {
+
+ PartitionableListState<?> value = entry.getValue();
+ long[] partitionOffsets = value.write(out);
+ OperatorStateHandle.Mode mode = value.getAssignmentMode();
+ writtenStatesMetaData.put(
+ entry.getKey(),
+ new OperatorStateHandle.StateMetaInfo(partitionOffsets, mode));
+ }
+
+ if (open.compareAndSet(true, false)) {
+
+ OperatorStateHandle operatorStateHandle =
+ new OperatorStateHandle(writtenStatesMetaData, out.closeAndGetHandle());
+
+ if (asynchronousSnapshots) {
+ LOG.info("DefaultOperatorStateBackend snapshot ({}, asynchronous part) in thread {} took {} ms.",
+ streamFactory, Thread.currentThread(), (System.currentTimeMillis() - asyncStartTime));
+ }
+
+ return operatorStateHandle;
+ } else {
+ throw new IOException("Checkpoint stream already closed.");
+ }
+ }
+
+ @Override
+ public void done(boolean canceled) {
+ if (open.compareAndSet(true, false)) {
+ CheckpointStreamFactory.CheckpointStateOutputStream stream = getIoHandle();
+ if (null != stream) {
+ closeStreamOnCancelRegistry.unregisterClosable(stream);
--- End diff --
Shouldn't we always try to remove the stream from the `ClosableRegistry` after the callable has been executed?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---