HDDS-925. Rename ChunkGroupOutputStream to KeyOutputStream and ChunkOutputStream to BlockOutputStream. Contributed by Shashikant Banerjee.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/4ff1c46d Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/4ff1c46d Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/4ff1c46d Branch: refs/heads/HDFS-12943 Commit: 4ff1c46d95c2b1bc645fe12004eb7c434b8f7b74 Parents: ee10ba2 Author: Shashikant Banerjee <[email protected]> Authored: Tue Dec 18 18:03:46 2018 +0530 Committer: Shashikant Banerjee <[email protected]> Committed: Tue Dec 18 18:03:46 2018 +0530 ---------------------------------------------------------------------- .../hdds/scm/storage/BlockOutputStream.java | 627 ++++++++++++++ .../hdds/scm/storage/ChunkOutputStream.java | 627 -------------- .../ozone/client/io/ChunkGroupOutputStream.java | 812 ------------------- .../hadoop/ozone/client/io/KeyOutputStream.java | 812 +++++++++++++++++++ .../ozone/client/io/OzoneOutputStream.java | 8 +- .../hadoop/ozone/client/rpc/RpcClient.java | 10 +- .../rpc/TestCloseContainerHandlingByClient.java | 54 +- .../rpc/TestContainerStateMachineFailures.java | 6 +- .../client/rpc/TestFailureHandlingByClient.java | 20 +- .../ozone/client/rpc/TestOzoneRpcClient.java | 6 +- .../web/storage/DistributedStorageHandler.java | 6 +- .../hadoop/ozone/om/TestChunkStreams.java | 2 +- .../hadoop/fs/ozone/OzoneFSOutputStream.java | 6 +- 13 files changed, 1498 insertions(+), 1498 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ff1c46d/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java new file mode 100644 index 0000000..32c6b6a --- /dev/null +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java @@ -0,0 +1,627 @@ +/* + * 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.hadoop.hdds.scm.storage; +import com.google.common.base.Preconditions; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; +import org.apache.hadoop.hdds.scm.XceiverClientAsyncReply; +import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; +import org.apache.hadoop.ozone.common.Checksum; +import org.apache.hadoop.ozone.common.ChecksumData; +import org.apache.hadoop.ozone.common.OzoneChecksumException; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; +import org.apache.commons.codec.digest.DigestUtils; +import org.apache.hadoop.hdds.scm.XceiverClientManager; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.KeyValue; +import org.apache.hadoop.hdds.client.BlockID; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.OutputStream; +import java.nio.Buffer; +import java.nio.ByteBuffer; +import java.util.UUID; +import java.util.List; +import java.util.ArrayList; +import java.util.concurrent.*; + +import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls + .putBlockAsync; +import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls + .writeChunkAsync; + +/** + * An {@link OutputStream} used by the REST service in combination with the + * SCMClient to write the value of a key to a sequence + * of container chunks. Writes are buffered locally and periodically written to + * the container as a new chunk. In order to preserve the semantics that + * replacement of a pre-existing key is atomic, each instance of the stream has + * an internal unique identifier. This unique identifier and a monotonically + * increasing chunk index form a composite key that is used as the chunk name. + * After all data is written, a putKey call creates or updates the corresponding + * container key, and this call includes the full list of chunks that make up + * the key data. The list of chunks is updated all at once. Therefore, a + * concurrent reader never can see an intermediate state in which different + * chunks of data from different versions of the key data are interleaved. + * This class encapsulates all state management for buffering and writing + * through to the container. + */ +public class BlockOutputStream extends OutputStream { + public static final Logger LOG = + LoggerFactory.getLogger(BlockOutputStream.class); + + private BlockID blockID; + private final String key; + private final String traceID; + private final BlockData.Builder containerBlockData; + private XceiverClientManager xceiverClientManager; + private XceiverClientSpi xceiverClient; + private final Checksum checksum; + private final String streamId; + private int chunkIndex; + private int chunkSize; + private final long streamBufferFlushSize; + private final long streamBufferMaxSize; + private final long watchTimeout; + private List<ByteBuffer> bufferList; + // The IOException will be set by response handling thread in case there is an + // exception received in the response. If the exception is set, the next + // request will fail upfront. + private IOException ioException; + private ExecutorService responseExecutor; + + // the effective length of data flushed so far + private long totalDataFlushedLength; + + // effective data write attempted so far for the block + private long writtenDataLength; + + // total data which has been successfully flushed and acknowledged + // by all servers + private long totalAckDataLength; + + // list to hold up all putBlock futures + private List<CompletableFuture<ContainerProtos.ContainerCommandResponseProto>> + futureList; + // map containing mapping for putBlock logIndex to to flushedDataLength Map. + private ConcurrentHashMap<Long, Long> commitIndex2flushedDataMap; + + private int currentBufferIndex; + + /** + * Creates a new BlockOutputStream. + * + * @param blockID block ID + * @param key chunk key + * @param xceiverClientManager client manager that controls client + * @param xceiverClient client to perform container calls + * @param traceID container protocol call args + * @param chunkSize chunk size + * @param bufferList list of byte buffers + * @param streamBufferFlushSize flush size + * @param streamBufferMaxSize max size of the currentBuffer + * @param watchTimeout watch timeout + * @param checksum checksum + */ + public BlockOutputStream(BlockID blockID, String key, + XceiverClientManager xceiverClientManager, XceiverClientSpi xceiverClient, + String traceID, int chunkSize, long streamBufferFlushSize, + long streamBufferMaxSize, long watchTimeout, + List<ByteBuffer> bufferList, Checksum checksum) { + this.blockID = blockID; + this.key = key; + this.traceID = traceID; + this.chunkSize = chunkSize; + KeyValue keyValue = + KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build(); + this.containerBlockData = + BlockData.newBuilder().setBlockID(blockID.getDatanodeBlockIDProtobuf()) + .addMetadata(keyValue); + this.xceiverClientManager = xceiverClientManager; + this.xceiverClient = xceiverClient; + this.streamId = UUID.randomUUID().toString(); + this.chunkIndex = 0; + this.streamBufferFlushSize = streamBufferFlushSize; + this.streamBufferMaxSize = streamBufferMaxSize; + this.watchTimeout = watchTimeout; + this.bufferList = bufferList; + this.checksum = checksum; + + // A single thread executor handle the responses of async requests + responseExecutor = Executors.newSingleThreadExecutor(); + commitIndex2flushedDataMap = new ConcurrentHashMap<>(); + totalAckDataLength = 0; + futureList = new ArrayList<>(); + totalDataFlushedLength = 0; + currentBufferIndex = 0; + writtenDataLength = 0; + } + + public BlockID getBlockID() { + return blockID; + } + + public long getTotalSuccessfulFlushedData() { + return totalAckDataLength; + } + + public long getWrittenDataLength() { + return writtenDataLength; + } + + private long computeBufferData() { + int dataLength = + bufferList.stream().mapToInt(Buffer::position).sum(); + Preconditions.checkState(dataLength <= streamBufferMaxSize); + return dataLength; + } + + + @Override + public void write(int b) throws IOException { + checkOpen(); + byte[] buf = new byte[1]; + buf[0] = (byte) b; + write(buf, 0, 1); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + if (b == null) { + throw new NullPointerException(); + } + if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) + || ((off + len) < 0)) { + throw new IndexOutOfBoundsException(); + } + if (len == 0) { + return; + } + while (len > 0) { + checkOpen(); + int writeLen; + allocateBuffer(); + ByteBuffer currentBuffer = getCurrentBuffer(); + writeLen = + Math.min(chunkSize - currentBuffer.position() % chunkSize, len); + currentBuffer.put(b, off, writeLen); + if (currentBuffer.position() % chunkSize == 0) { + int pos = currentBuffer.position() - chunkSize; + int limit = currentBuffer.position(); + writeChunk(pos, limit); + } + off += writeLen; + len -= writeLen; + writtenDataLength += writeLen; + if (currentBuffer.position() == streamBufferFlushSize) { + totalDataFlushedLength += streamBufferFlushSize; + handlePartialFlush(); + } + long bufferedData = computeBufferData(); + // Data in the bufferList can not exceed streamBufferMaxSize + if (bufferedData == streamBufferMaxSize) { + handleFullBuffer(); + } + } + } + + private ByteBuffer getCurrentBuffer() { + ByteBuffer buffer = bufferList.get(currentBufferIndex); + if (!buffer.hasRemaining()) { + currentBufferIndex = + currentBufferIndex < getMaxNumBuffers() - 1 ? ++currentBufferIndex : + 0; + } + return bufferList.get(currentBufferIndex); + } + + private int getMaxNumBuffers() { + return (int)(streamBufferMaxSize/streamBufferFlushSize); + } + + private void allocateBuffer() { + for (int i = bufferList.size(); i < getMaxNumBuffers(); i++) { + bufferList.add(ByteBuffer.allocate((int)streamBufferFlushSize)); + } + } + + /** + * Will be called on the retryPath in case closedContainerException/ + * TimeoutException. + * @param len length of data to write + * @throws IOException if error occurred + */ + + // In this case, the data is already cached in the currentBuffer. + public void writeOnRetry(long len) throws IOException { + if (len == 0) { + return; + } + int off = 0; + int pos = off; + while (len > 0) { + long writeLen; + writeLen = Math.min(chunkSize, len); + if (writeLen == chunkSize) { + int limit = pos + chunkSize; + writeChunk(pos, limit); + } + off += writeLen; + len -= writeLen; + writtenDataLength += writeLen; + if (off % streamBufferFlushSize == 0) { + // reset the position to zero as now we wll readng thhe next buffer in + // the list + pos = 0; + totalDataFlushedLength += streamBufferFlushSize; + handlePartialFlush(); + } + if (computeBufferData() % streamBufferMaxSize == 0) { + handleFullBuffer(); + } + } + } + + /** + * just update the totalAckDataLength. Since we have allocated + * the currentBuffer more than the streamBufferMaxSize, we can keep on writing + * to the currentBuffer. In case of failure, we will read the data starting + * from totalAckDataLength. + */ + private void updateFlushIndex(long index) { + if (!commitIndex2flushedDataMap.isEmpty()) { + Preconditions.checkState(commitIndex2flushedDataMap.containsKey(index)); + totalAckDataLength = commitIndex2flushedDataMap.remove(index); + LOG.debug("Total data successfully replicated: " + totalAckDataLength); + futureList.remove(0); + // Flush has been committed to required servers successful. + // just swap the bufferList head and tail after clearing. + ByteBuffer currentBuffer = bufferList.remove(0); + currentBuffer.clear(); + if (currentBufferIndex != 0) { + currentBufferIndex--; + } + bufferList.add(currentBuffer); + } + } + + /** + * This is a blocking call. It will wait for the flush till the commit index + * at the head of the commitIndex2flushedDataMap gets replicated to all or + * majority. + * @throws IOException + */ + private void handleFullBuffer() throws IOException { + try { + checkOpen(); + if (!futureList.isEmpty()) { + waitOnFlushFutures(); + } + } catch (InterruptedException | ExecutionException e) { + adjustBuffersOnException(); + throw new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } + if (!commitIndex2flushedDataMap.isEmpty()) { + watchForCommit( + commitIndex2flushedDataMap.keySet().stream().mapToLong(v -> v) + .min().getAsLong()); + } + } + + private void adjustBuffers(long commitIndex) { + commitIndex2flushedDataMap.keySet().stream().forEach(index -> { + if (index <= commitIndex) { + updateFlushIndex(index); + } else { + return; + } + }); + } + + // It may happen that once the exception is encountered , we still might + // have successfully flushed up to a certain index. Make sure the buffers + // only contain data which have not been sufficiently replicated + private void adjustBuffersOnException() { + adjustBuffers(xceiverClient.getReplicatedMinCommitIndex()); + } + + /** + * calls watchForCommit API of the Ratis Client. For Standalone client, + * it is a no op. + * @param commitIndex log index to watch for + * @return minimum commit index replicated to all nodes + * @throws IOException IOException in case watch gets timed out + */ + private void watchForCommit(long commitIndex) throws IOException { + checkOpen(); + Preconditions.checkState(!commitIndex2flushedDataMap.isEmpty()); + try { + long index = + xceiverClient.watchForCommit(commitIndex, watchTimeout); + adjustBuffers(index); + } catch (TimeoutException | InterruptedException | ExecutionException e) { + LOG.warn("watchForCommit failed for index " + commitIndex, e); + adjustBuffersOnException(); + throw new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } + } + + private CompletableFuture<ContainerProtos. + ContainerCommandResponseProto> handlePartialFlush() + throws IOException { + checkOpen(); + long flushPos = totalDataFlushedLength; + String requestId = + traceID + ContainerProtos.Type.PutBlock + chunkIndex + blockID; + CompletableFuture<ContainerProtos. + ContainerCommandResponseProto> flushFuture; + try { + XceiverClientAsyncReply asyncReply = + putBlockAsync(xceiverClient, containerBlockData.build(), requestId); + CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future = + asyncReply.getResponse(); + flushFuture = future.thenApplyAsync(e -> { + try { + validateResponse(e); + } catch (IOException sce) { + future.completeExceptionally(sce); + return e; + } + // if the ioException is not set, putBlock is successful + if (ioException == null) { + LOG.debug( + "Adding index " + asyncReply.getLogIndex() + " commitMap size " + + commitIndex2flushedDataMap.size()); + BlockID responseBlockID = BlockID.getFromProtobuf( + e.getPutBlock().getCommittedBlockLength().getBlockID()); + Preconditions.checkState(blockID.getContainerBlockID() + .equals(responseBlockID.getContainerBlockID())); + // updates the bcsId of the block + blockID = responseBlockID; + // for standalone protocol, logIndex will always be 0. + commitIndex2flushedDataMap.put(asyncReply.getLogIndex(), flushPos); + } + return e; + }, responseExecutor).exceptionally(e -> { + LOG.debug( + "putBlock failed for blockID " + blockID + " with exception " + e + .getLocalizedMessage()); + CompletionException ce = new CompletionException(e); + setIoException(ce); + throw ce; + }); + } catch (IOException | InterruptedException | ExecutionException e) { + throw new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } + futureList.add(flushFuture); + return flushFuture; + } + + @Override + public void flush() throws IOException { + if (xceiverClientManager != null && xceiverClient != null + && bufferList != null) { + checkOpen(); + int bufferSize = bufferList.size(); + if (bufferSize > 0) { + try { + // flush the last chunk data residing on the currentBuffer + if (totalDataFlushedLength < writtenDataLength) { + ByteBuffer currentBuffer = getCurrentBuffer(); + int pos = currentBuffer.position() - (currentBuffer.position() + % chunkSize); + int limit = currentBuffer.position() - pos; + writeChunk(pos, currentBuffer.position()); + totalDataFlushedLength += limit; + handlePartialFlush(); + } + waitOnFlushFutures(); + // just check again if the exception is hit while waiting for the + // futures to ensure flush has indeed succeeded + checkOpen(); + } catch (InterruptedException | ExecutionException e) { + adjustBuffersOnException(); + throw new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } + } + } + } + + private void writeChunk(int pos, int limit) throws IOException { + // Please note : We are not flipping the slice when we write since + // the slices are pointing the currentBuffer start and end as needed for + // the chunk write. Also please note, Duplicate does not create a + // copy of data, it only creates metadata that points to the data + // stream. + ByteBuffer chunk = bufferList.get(currentBufferIndex).duplicate(); + chunk.position(pos); + chunk.limit(limit); + writeChunkToContainer(chunk); + } + + @Override + public void close() throws IOException { + if (xceiverClientManager != null && xceiverClient != null + && bufferList != null) { + int bufferSize = bufferList.size(); + if (bufferSize > 0) { + try { + // flush the last chunk data residing on the currentBuffer + if (totalDataFlushedLength < writtenDataLength) { + ByteBuffer currentBuffer = getCurrentBuffer(); + int pos = currentBuffer.position() - (currentBuffer.position() + % chunkSize); + int limit = currentBuffer.position() - pos; + writeChunk(pos, currentBuffer.position()); + totalDataFlushedLength += limit; + handlePartialFlush(); + } + waitOnFlushFutures(); + // irrespective of whether the commitIndex2flushedDataMap is empty + // or not, ensure there is no exception set + checkOpen(); + if (!commitIndex2flushedDataMap.isEmpty()) { + // wait for the last commit index in the commitIndex2flushedDataMap + // to get committed to all or majority of nodes in case timeout + // happens. + long lastIndex = + commitIndex2flushedDataMap.keySet().stream() + .mapToLong(v -> v).max().getAsLong(); + LOG.debug( + "waiting for last flush Index " + lastIndex + " to catch up"); + watchForCommit(lastIndex); + } + } catch (InterruptedException | ExecutionException e) { + adjustBuffersOnException(); + throw new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } finally { + cleanup(); + } + } + // clear the currentBuffer + bufferList.stream().forEach(ByteBuffer::clear); + } + } + + private void waitOnFlushFutures() + throws InterruptedException, ExecutionException { + CompletableFuture<Void> combinedFuture = CompletableFuture + .allOf(futureList.toArray(new CompletableFuture[futureList.size()])); + // wait for all the transactions to complete + combinedFuture.get(); + } + + private void validateResponse( + ContainerProtos.ContainerCommandResponseProto responseProto) + throws IOException { + try { + // if the ioException is already set, it means a prev request has failed + // just throw the exception. The current operation will fail with the + // original error + if (ioException != null) { + throw ioException; + } + ContainerProtocolCalls.validateContainerResponse(responseProto); + } catch (StorageContainerException sce) { + LOG.error("Unexpected Storage Container Exception: ", sce); + setIoException(sce); + throw sce; + } + } + + private void setIoException(Exception e) { + if (ioException != null) { + ioException = new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } + } + + public void cleanup() { + if (xceiverClientManager != null) { + xceiverClientManager.releaseClient(xceiverClient); + } + xceiverClientManager = null; + xceiverClient = null; + if (futureList != null) { + futureList.clear(); + } + futureList = null; + if (commitIndex2flushedDataMap != null) { + commitIndex2flushedDataMap.clear(); + } + commitIndex2flushedDataMap = null; + responseExecutor.shutdown(); + } + + /** + * Checks if the stream is open or exception has occured. + * If not, throws an exception. + * + * @throws IOException if stream is closed + */ + private void checkOpen() throws IOException { + if (xceiverClient == null) { + throw new IOException("BlockOutputStream has been closed."); + } else if (ioException != null) { + adjustBuffersOnException(); + throw ioException; + } + } + + /** + * Writes buffered data as a new chunk to the container and saves chunk + * information to be used later in putKey call. + * + * @throws IOException if there is an I/O error while performing the call + * @throws OzoneChecksumException if there is an error while computing + * checksum + */ + private void writeChunkToContainer(ByteBuffer chunk) throws IOException { + int effectiveChunkSize = chunk.remaining(); + ByteString data = ByteString.copyFrom(chunk); + ChecksumData checksumData = checksum.computeChecksum(data); + ChunkInfo chunkInfo = ChunkInfo.newBuilder() + .setChunkName(DigestUtils.md5Hex(key) + "_stream_" + streamId + + "_chunk_" + ++chunkIndex) + .setOffset(0) + .setLen(effectiveChunkSize) + .setChecksumData(checksumData.getProtoBufMessage()) + .build(); + // generate a unique requestId + String requestId = + traceID + ContainerProtos.Type.WriteChunk + chunkIndex + chunkInfo + .getChunkName(); + try { + XceiverClientAsyncReply asyncReply = + writeChunkAsync(xceiverClient, chunkInfo, blockID, data, requestId); + CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future = + asyncReply.getResponse(); + future.thenApplyAsync(e -> { + try { + validateResponse(e); + } catch (IOException sce) { + future.completeExceptionally(sce); + } + return e; + }, responseExecutor).exceptionally(e -> { + LOG.debug( + "writing chunk failed " + chunkInfo.getChunkName() + " blockID " + + blockID + " with exception " + e.getLocalizedMessage()); + CompletionException ce = new CompletionException(e); + setIoException(ce); + throw ce; + }); + } catch (IOException | InterruptedException | ExecutionException e) { + throw new IOException( + "Unexpected Storage Container Exception: " + e.toString(), e); + } + LOG.debug( + "writing chunk " + chunkInfo.getChunkName() + " blockID " + blockID + + " length " + effectiveChunkSize); + containerBlockData.addChunks(chunkInfo); + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ff1c46d/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java deleted file mode 100644 index 6e2ca59..0000000 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkOutputStream.java +++ /dev/null @@ -1,627 +0,0 @@ -/* - * 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.hadoop.hdds.scm.storage; -import com.google.common.base.Preconditions; -import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; -import org.apache.hadoop.hdds.scm.XceiverClientAsyncReply; -import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; -import org.apache.hadoop.ozone.common.Checksum; -import org.apache.hadoop.ozone.common.ChecksumData; -import org.apache.hadoop.ozone.common.OzoneChecksumException; -import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; -import org.apache.commons.codec.digest.DigestUtils; -import org.apache.hadoop.hdds.scm.XceiverClientManager; -import org.apache.hadoop.hdds.scm.XceiverClientSpi; -import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; -import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData; -import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.KeyValue; -import org.apache.hadoop.hdds.client.BlockID; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.OutputStream; -import java.nio.Buffer; -import java.nio.ByteBuffer; -import java.util.UUID; -import java.util.List; -import java.util.ArrayList; -import java.util.concurrent.*; - -import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls - .putBlockAsync; -import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls - .writeChunkAsync; - -/** - * An {@link OutputStream} used by the REST service in combination with the - * SCMClient to write the value of a key to a sequence - * of container chunks. Writes are buffered locally and periodically written to - * the container as a new chunk. In order to preserve the semantics that - * replacement of a pre-existing key is atomic, each instance of the stream has - * an internal unique identifier. This unique identifier and a monotonically - * increasing chunk index form a composite key that is used as the chunk name. - * After all data is written, a putKey call creates or updates the corresponding - * container key, and this call includes the full list of chunks that make up - * the key data. The list of chunks is updated all at once. Therefore, a - * concurrent reader never can see an intermediate state in which different - * chunks of data from different versions of the key data are interleaved. - * This class encapsulates all state management for buffering and writing - * through to the container. - */ -public class ChunkOutputStream extends OutputStream { - public static final Logger LOG = - LoggerFactory.getLogger(ChunkOutputStream.class); - - private BlockID blockID; - private final String key; - private final String traceID; - private final BlockData.Builder containerBlockData; - private XceiverClientManager xceiverClientManager; - private XceiverClientSpi xceiverClient; - private final Checksum checksum; - private final String streamId; - private int chunkIndex; - private int chunkSize; - private final long streamBufferFlushSize; - private final long streamBufferMaxSize; - private final long watchTimeout; - private List<ByteBuffer> bufferList; - // The IOException will be set by response handling thread in case there is an - // exception received in the response. If the exception is set, the next - // request will fail upfront. - private IOException ioException; - private ExecutorService responseExecutor; - - // the effective length of data flushed so far - private long totalDataFlushedLength; - - // effective data write attempted so far for the block - private long writtenDataLength; - - // total data which has been successfully flushed and acknowledged - // by all servers - private long totalAckDataLength; - - // list to hold up all putBlock futures - private List<CompletableFuture<ContainerProtos.ContainerCommandResponseProto>> - futureList; - // map containing mapping for putBlock logIndex to to flushedDataLength Map. - private ConcurrentHashMap<Long, Long> commitIndex2flushedDataMap; - - private int currentBufferIndex; - - /** - * Creates a new ChunkOutputStream. - * - * @param blockID block ID - * @param key chunk key - * @param xceiverClientManager client manager that controls client - * @param xceiverClient client to perform container calls - * @param traceID container protocol call args - * @param chunkSize chunk size - * @param bufferList list of byte buffers - * @param streamBufferFlushSize flush size - * @param streamBufferMaxSize max size of the currentBuffer - * @param watchTimeout watch timeout - * @param checksum checksum - */ - public ChunkOutputStream(BlockID blockID, String key, - XceiverClientManager xceiverClientManager, XceiverClientSpi xceiverClient, - String traceID, int chunkSize, long streamBufferFlushSize, - long streamBufferMaxSize, long watchTimeout, - List<ByteBuffer> bufferList, Checksum checksum) { - this.blockID = blockID; - this.key = key; - this.traceID = traceID; - this.chunkSize = chunkSize; - KeyValue keyValue = - KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build(); - this.containerBlockData = - BlockData.newBuilder().setBlockID(blockID.getDatanodeBlockIDProtobuf()) - .addMetadata(keyValue); - this.xceiverClientManager = xceiverClientManager; - this.xceiverClient = xceiverClient; - this.streamId = UUID.randomUUID().toString(); - this.chunkIndex = 0; - this.streamBufferFlushSize = streamBufferFlushSize; - this.streamBufferMaxSize = streamBufferMaxSize; - this.watchTimeout = watchTimeout; - this.bufferList = bufferList; - this.checksum = checksum; - - // A single thread executor handle the responses of async requests - responseExecutor = Executors.newSingleThreadExecutor(); - commitIndex2flushedDataMap = new ConcurrentHashMap<>(); - totalAckDataLength = 0; - futureList = new ArrayList<>(); - totalDataFlushedLength = 0; - currentBufferIndex = 0; - writtenDataLength = 0; - } - - public BlockID getBlockID() { - return blockID; - } - - public long getTotalSuccessfulFlushedData() { - return totalAckDataLength; - } - - public long getWrittenDataLength() { - return writtenDataLength; - } - - private long computeBufferData() { - int dataLength = - bufferList.stream().mapToInt(Buffer::position).sum(); - Preconditions.checkState(dataLength <= streamBufferMaxSize); - return dataLength; - } - - - @Override - public void write(int b) throws IOException { - checkOpen(); - byte[] buf = new byte[1]; - buf[0] = (byte) b; - write(buf, 0, 1); - } - - @Override - public void write(byte[] b, int off, int len) throws IOException { - if (b == null) { - throw new NullPointerException(); - } - if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) - || ((off + len) < 0)) { - throw new IndexOutOfBoundsException(); - } - if (len == 0) { - return; - } - while (len > 0) { - checkOpen(); - int writeLen; - allocateBuffer(); - ByteBuffer currentBuffer = getCurrentBuffer(); - writeLen = - Math.min(chunkSize - currentBuffer.position() % chunkSize, len); - currentBuffer.put(b, off, writeLen); - if (currentBuffer.position() % chunkSize == 0) { - int pos = currentBuffer.position() - chunkSize; - int limit = currentBuffer.position(); - writeChunk(pos, limit); - } - off += writeLen; - len -= writeLen; - writtenDataLength += writeLen; - if (currentBuffer.position() == streamBufferFlushSize) { - totalDataFlushedLength += streamBufferFlushSize; - handlePartialFlush(); - } - long bufferedData = computeBufferData(); - // Data in the bufferList can not exceed streamBufferMaxSize - if (bufferedData == streamBufferMaxSize) { - handleFullBuffer(); - } - } - } - - private ByteBuffer getCurrentBuffer() { - ByteBuffer buffer = bufferList.get(currentBufferIndex); - if (!buffer.hasRemaining()) { - currentBufferIndex = - currentBufferIndex < getMaxNumBuffers() - 1 ? ++currentBufferIndex : - 0; - } - return bufferList.get(currentBufferIndex); - } - - private int getMaxNumBuffers() { - return (int)(streamBufferMaxSize/streamBufferFlushSize); - } - - private void allocateBuffer() { - for (int i = bufferList.size(); i < getMaxNumBuffers(); i++) { - bufferList.add(ByteBuffer.allocate((int)streamBufferFlushSize)); - } - } - - /** - * Will be called on the retryPath in case closedContainerException/ - * TimeoutException. - * @param len length of data to write - * @throws IOException if error occurred - */ - - // In this case, the data is already cached in the currentBuffer. - public void writeOnRetry(long len) throws IOException { - if (len == 0) { - return; - } - int off = 0; - int pos = off; - while (len > 0) { - long writeLen; - writeLen = Math.min(chunkSize, len); - if (writeLen == chunkSize) { - int limit = pos + chunkSize; - writeChunk(pos, limit); - } - off += writeLen; - len -= writeLen; - writtenDataLength += writeLen; - if (off % streamBufferFlushSize == 0) { - // reset the position to zero as now we wll readng thhe next buffer in - // the list - pos = 0; - totalDataFlushedLength += streamBufferFlushSize; - handlePartialFlush(); - } - if (computeBufferData() % streamBufferMaxSize == 0) { - handleFullBuffer(); - } - } - } - - /** - * just update the totalAckDataLength. Since we have allocated - * the currentBuffer more than the streamBufferMaxSize, we can keep on writing - * to the currentBuffer. In case of failure, we will read the data starting - * from totalAckDataLength. - */ - private void updateFlushIndex(long index) { - if (!commitIndex2flushedDataMap.isEmpty()) { - Preconditions.checkState(commitIndex2flushedDataMap.containsKey(index)); - totalAckDataLength = commitIndex2flushedDataMap.remove(index); - LOG.debug("Total data successfully replicated: " + totalAckDataLength); - futureList.remove(0); - // Flush has been committed to required servers successful. - // just swap the bufferList head and tail after clearing. - ByteBuffer currentBuffer = bufferList.remove(0); - currentBuffer.clear(); - if (currentBufferIndex != 0) { - currentBufferIndex--; - } - bufferList.add(currentBuffer); - } - } - - /** - * This is a blocking call. It will wait for the flush till the commit index - * at the head of the commitIndex2flushedDataMap gets replicated to all or - * majority. - * @throws IOException - */ - private void handleFullBuffer() throws IOException { - try { - checkOpen(); - if (!futureList.isEmpty()) { - waitOnFlushFutures(); - } - } catch (InterruptedException | ExecutionException e) { - adjustBuffersOnException(); - throw new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } - if (!commitIndex2flushedDataMap.isEmpty()) { - watchForCommit( - commitIndex2flushedDataMap.keySet().stream().mapToLong(v -> v) - .min().getAsLong()); - } - } - - private void adjustBuffers(long commitIndex) { - commitIndex2flushedDataMap.keySet().stream().forEach(index -> { - if (index <= commitIndex) { - updateFlushIndex(index); - } else { - return; - } - }); - } - - // It may happen that once the exception is encountered , we still might - // have successfully flushed up to a certain index. Make sure the buffers - // only contain data which have not been sufficiently replicated - private void adjustBuffersOnException() { - adjustBuffers(xceiverClient.getReplicatedMinCommitIndex()); - } - - /** - * calls watchForCommit API of the Ratis Client. For Standalone client, - * it is a no op. - * @param commitIndex log index to watch for - * @return minimum commit index replicated to all nodes - * @throws IOException IOException in case watch gets timed out - */ - private void watchForCommit(long commitIndex) throws IOException { - checkOpen(); - Preconditions.checkState(!commitIndex2flushedDataMap.isEmpty()); - try { - long index = - xceiverClient.watchForCommit(commitIndex, watchTimeout); - adjustBuffers(index); - } catch (TimeoutException | InterruptedException | ExecutionException e) { - LOG.warn("watchForCommit failed for index " + commitIndex, e); - adjustBuffersOnException(); - throw new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } - } - - private CompletableFuture<ContainerProtos. - ContainerCommandResponseProto> handlePartialFlush() - throws IOException { - checkOpen(); - long flushPos = totalDataFlushedLength; - String requestId = - traceID + ContainerProtos.Type.PutBlock + chunkIndex + blockID; - CompletableFuture<ContainerProtos. - ContainerCommandResponseProto> flushFuture; - try { - XceiverClientAsyncReply asyncReply = - putBlockAsync(xceiverClient, containerBlockData.build(), requestId); - CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future = - asyncReply.getResponse(); - flushFuture = future.thenApplyAsync(e -> { - try { - validateResponse(e); - } catch (IOException sce) { - future.completeExceptionally(sce); - return e; - } - // if the ioException is not set, putBlock is successful - if (ioException == null) { - LOG.debug( - "Adding index " + asyncReply.getLogIndex() + " commitMap size " - + commitIndex2flushedDataMap.size()); - BlockID responseBlockID = BlockID.getFromProtobuf( - e.getPutBlock().getCommittedBlockLength().getBlockID()); - Preconditions.checkState(blockID.getContainerBlockID() - .equals(responseBlockID.getContainerBlockID())); - // updates the bcsId of the block - blockID = responseBlockID; - // for standalone protocol, logIndex will always be 0. - commitIndex2flushedDataMap.put(asyncReply.getLogIndex(), flushPos); - } - return e; - }, responseExecutor).exceptionally(e -> { - LOG.debug( - "putBlock failed for blockID " + blockID + " with exception " + e - .getLocalizedMessage()); - CompletionException ce = new CompletionException(e); - setIoException(ce); - throw ce; - }); - } catch (IOException | InterruptedException | ExecutionException e) { - throw new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } - futureList.add(flushFuture); - return flushFuture; - } - - @Override - public void flush() throws IOException { - if (xceiverClientManager != null && xceiverClient != null - && bufferList != null) { - checkOpen(); - int bufferSize = bufferList.size(); - if (bufferSize > 0) { - try { - // flush the last chunk data residing on the currentBuffer - if (totalDataFlushedLength < writtenDataLength) { - ByteBuffer currentBuffer = getCurrentBuffer(); - int pos = currentBuffer.position() - (currentBuffer.position() - % chunkSize); - int limit = currentBuffer.position() - pos; - writeChunk(pos, currentBuffer.position()); - totalDataFlushedLength += limit; - handlePartialFlush(); - } - waitOnFlushFutures(); - // just check again if the exception is hit while waiting for the - // futures to ensure flush has indeed succeeded - checkOpen(); - } catch (InterruptedException | ExecutionException e) { - adjustBuffersOnException(); - throw new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } - } - } - } - - private void writeChunk(int pos, int limit) throws IOException { - // Please note : We are not flipping the slice when we write since - // the slices are pointing the currentBuffer start and end as needed for - // the chunk write. Also please note, Duplicate does not create a - // copy of data, it only creates metadata that points to the data - // stream. - ByteBuffer chunk = bufferList.get(currentBufferIndex).duplicate(); - chunk.position(pos); - chunk.limit(limit); - writeChunkToContainer(chunk); - } - - @Override - public void close() throws IOException { - if (xceiverClientManager != null && xceiverClient != null - && bufferList != null) { - int bufferSize = bufferList.size(); - if (bufferSize > 0) { - try { - // flush the last chunk data residing on the currentBuffer - if (totalDataFlushedLength < writtenDataLength) { - ByteBuffer currentBuffer = getCurrentBuffer(); - int pos = currentBuffer.position() - (currentBuffer.position() - % chunkSize); - int limit = currentBuffer.position() - pos; - writeChunk(pos, currentBuffer.position()); - totalDataFlushedLength += limit; - handlePartialFlush(); - } - waitOnFlushFutures(); - // irrespective of whether the commitIndex2flushedDataMap is empty - // or not, ensure there is no exception set - checkOpen(); - if (!commitIndex2flushedDataMap.isEmpty()) { - // wait for the last commit index in the commitIndex2flushedDataMap - // to get committed to all or majority of nodes in case timeout - // happens. - long lastIndex = - commitIndex2flushedDataMap.keySet().stream() - .mapToLong(v -> v).max().getAsLong(); - LOG.debug( - "waiting for last flush Index " + lastIndex + " to catch up"); - watchForCommit(lastIndex); - } - } catch (InterruptedException | ExecutionException e) { - adjustBuffersOnException(); - throw new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } finally { - cleanup(); - } - } - // clear the currentBuffer - bufferList.stream().forEach(ByteBuffer::clear); - } - } - - private void waitOnFlushFutures() - throws InterruptedException, ExecutionException { - CompletableFuture<Void> combinedFuture = CompletableFuture - .allOf(futureList.toArray(new CompletableFuture[futureList.size()])); - // wait for all the transactions to complete - combinedFuture.get(); - } - - private void validateResponse( - ContainerProtos.ContainerCommandResponseProto responseProto) - throws IOException { - try { - // if the ioException is already set, it means a prev request has failed - // just throw the exception. The current operation will fail with the - // original error - if (ioException != null) { - throw ioException; - } - ContainerProtocolCalls.validateContainerResponse(responseProto); - } catch (StorageContainerException sce) { - LOG.error("Unexpected Storage Container Exception: ", sce); - setIoException(sce); - throw sce; - } - } - - private void setIoException(Exception e) { - if (ioException != null) { - ioException = new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } - } - - public void cleanup() { - if (xceiverClientManager != null) { - xceiverClientManager.releaseClient(xceiverClient); - } - xceiverClientManager = null; - xceiverClient = null; - if (futureList != null) { - futureList.clear(); - } - futureList = null; - if (commitIndex2flushedDataMap != null) { - commitIndex2flushedDataMap.clear(); - } - commitIndex2flushedDataMap = null; - responseExecutor.shutdown(); - } - - /** - * Checks if the stream is open or exception has occured. - * If not, throws an exception. - * - * @throws IOException if stream is closed - */ - private void checkOpen() throws IOException { - if (xceiverClient == null) { - throw new IOException("ChunkOutputStream has been closed."); - } else if (ioException != null) { - adjustBuffersOnException(); - throw ioException; - } - } - - /** - * Writes buffered data as a new chunk to the container and saves chunk - * information to be used later in putKey call. - * - * @throws IOException if there is an I/O error while performing the call - * @throws OzoneChecksumException if there is an error while computing - * checksum - */ - private void writeChunkToContainer(ByteBuffer chunk) throws IOException { - int effectiveChunkSize = chunk.remaining(); - ByteString data = ByteString.copyFrom(chunk); - ChecksumData checksumData = checksum.computeChecksum(data); - ChunkInfo chunkInfo = ChunkInfo.newBuilder() - .setChunkName(DigestUtils.md5Hex(key) + "_stream_" + streamId + - "_chunk_" + ++chunkIndex) - .setOffset(0) - .setLen(effectiveChunkSize) - .setChecksumData(checksumData.getProtoBufMessage()) - .build(); - // generate a unique requestId - String requestId = - traceID + ContainerProtos.Type.WriteChunk + chunkIndex + chunkInfo - .getChunkName(); - try { - XceiverClientAsyncReply asyncReply = - writeChunkAsync(xceiverClient, chunkInfo, blockID, data, requestId); - CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future = - asyncReply.getResponse(); - future.thenApplyAsync(e -> { - try { - validateResponse(e); - } catch (IOException sce) { - future.completeExceptionally(sce); - } - return e; - }, responseExecutor).exceptionally(e -> { - LOG.debug( - "writing chunk failed " + chunkInfo.getChunkName() + " blockID " - + blockID + " with exception " + e.getLocalizedMessage()); - CompletionException ce = new CompletionException(e); - setIoException(ce); - throw ce; - }); - } catch (IOException | InterruptedException | ExecutionException e) { - throw new IOException( - "Unexpected Storage Container Exception: " + e.toString(), e); - } - LOG.debug( - "writing chunk " + chunkInfo.getChunkName() + " blockID " + blockID - + " length " + effectiveChunkSize); - containerBlockData.addChunks(chunkInfo); - } -} http://git-wip-us.apache.org/repos/asf/hadoop/blob/4ff1c46d/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java ---------------------------------------------------------------------- diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java deleted file mode 100644 index 49835cc..0000000 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/ChunkGroupOutputStream.java +++ /dev/null @@ -1,812 +0,0 @@ -/* - * 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.hadoop.ozone.client.io; - -import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; -import org.apache.hadoop.fs.FSExceptionMessages; -import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result; -import org.apache.hadoop.hdds.client.BlockID; -import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerNotOpenException; -import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline; -import org.apache.hadoop.ozone.common.Checksum; -import org.apache.hadoop.ozone.om.helpers.*; -import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType; -import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; -import org.apache.hadoop.ozone.om.protocolPB.OzoneManagerProtocolClientSideTranslatorPB; -import org.apache.hadoop.hdds.scm.XceiverClientManager; -import org.apache.hadoop.hdds.scm.XceiverClientSpi; -import org.apache.hadoop.hdds.scm.container.common.helpers - .StorageContainerException; -import org.apache.hadoop.hdds.scm.protocolPB - .StorageContainerLocationProtocolClientSideTranslatorPB; -import org.apache.hadoop.hdds.scm.storage.ChunkOutputStream; -import org.apache.ratis.protocol.RaftRetryFailureException; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.OutputStream; -import java.nio.ByteBuffer; -import java.util.ArrayList; -import java.util.List; -import java.util.Optional; -import java.util.ListIterator; -import java.util.concurrent.TimeoutException; - -/** - * Maintaining a list of ChunkInputStream. Write based on offset. - * - * Note that this may write to multiple containers in one write call. In case - * that first container succeeded but later ones failed, the succeeded writes - * are not rolled back. - * - * TODO : currently not support multi-thread access. - */ -public class ChunkGroupOutputStream extends OutputStream { - - public static final Logger LOG = - LoggerFactory.getLogger(ChunkGroupOutputStream.class); - - // array list's get(index) is O(1) - private final ArrayList<ChunkOutputStreamEntry> streamEntries; - private int currentStreamIndex; - private final OzoneManagerProtocolClientSideTranslatorPB omClient; - private final - StorageContainerLocationProtocolClientSideTranslatorPB scmClient; - private final OmKeyArgs keyArgs; - private final long openID; - private final XceiverClientManager xceiverClientManager; - private final int chunkSize; - private final String requestID; - private boolean closed; - private final long streamBufferFlushSize; - private final long streamBufferMaxSize; - private final long watchTimeout; - private final long blockSize; - private final Checksum checksum; - private List<ByteBuffer> bufferList; - private OmMultipartCommitUploadPartInfo commitUploadPartInfo; - /** - * A constructor for testing purpose only. - */ - @VisibleForTesting - public ChunkGroupOutputStream() { - streamEntries = new ArrayList<>(); - omClient = null; - scmClient = null; - keyArgs = null; - openID = -1; - xceiverClientManager = null; - chunkSize = 0; - requestID = null; - closed = false; - streamBufferFlushSize = 0; - streamBufferMaxSize = 0; - bufferList = new ArrayList<>(1); - ByteBuffer buffer = ByteBuffer.allocate(1); - bufferList.add(buffer); - watchTimeout = 0; - blockSize = 0; - this.checksum = new Checksum(); - } - - /** - * For testing purpose only. Not building output stream from blocks, but - * taking from externally. - * - * @param outputStream - * @param length - */ - @VisibleForTesting - public void addStream(OutputStream outputStream, long length) { - streamEntries.add( - new ChunkOutputStreamEntry(outputStream, length, checksum)); - } - - @VisibleForTesting - public List<ChunkOutputStreamEntry> getStreamEntries() { - return streamEntries; - } - @VisibleForTesting - public XceiverClientManager getXceiverClientManager() { - return xceiverClientManager; - } - - public List<OmKeyLocationInfo> getLocationInfoList() throws IOException { - List<OmKeyLocationInfo> locationInfoList = new ArrayList<>(); - for (ChunkOutputStreamEntry streamEntry : streamEntries) { - OmKeyLocationInfo info = - new OmKeyLocationInfo.Builder().setBlockID(streamEntry.blockID) - .setLength(streamEntry.currentPosition).setOffset(0) - .build(); - LOG.debug("block written " + streamEntry.blockID + ", length " - + streamEntry.currentPosition + " bcsID " + streamEntry.blockID - .getBlockCommitSequenceId()); - locationInfoList.add(info); - } - return locationInfoList; - } - - public ChunkGroupOutputStream(OpenKeySession handler, - XceiverClientManager xceiverClientManager, - StorageContainerLocationProtocolClientSideTranslatorPB scmClient, - OzoneManagerProtocolClientSideTranslatorPB omClient, int chunkSize, - String requestId, ReplicationFactor factor, ReplicationType type, - long bufferFlushSize, long bufferMaxSize, long size, long watchTimeout, - Checksum checksum, String uploadID, int partNumber, boolean isMultipart) { - this.streamEntries = new ArrayList<>(); - this.currentStreamIndex = 0; - this.omClient = omClient; - this.scmClient = scmClient; - OmKeyInfo info = handler.getKeyInfo(); - this.keyArgs = new OmKeyArgs.Builder().setVolumeName(info.getVolumeName()) - .setBucketName(info.getBucketName()).setKeyName(info.getKeyName()) - .setType(type).setFactor(factor).setDataSize(info.getDataSize()) - .setIsMultipartKey(isMultipart).setMultipartUploadID( - uploadID).setMultipartUploadPartNumber(partNumber) - .build(); - this.openID = handler.getId(); - this.xceiverClientManager = xceiverClientManager; - this.chunkSize = chunkSize; - this.requestID = requestId; - this.streamBufferFlushSize = bufferFlushSize; - this.streamBufferMaxSize = bufferMaxSize; - this.blockSize = size; - this.watchTimeout = watchTimeout; - this.checksum = checksum; - - Preconditions.checkState(chunkSize > 0); - Preconditions.checkState(streamBufferFlushSize > 0); - Preconditions.checkState(streamBufferMaxSize > 0); - Preconditions.checkState(blockSize > 0); - Preconditions.checkState(streamBufferFlushSize % chunkSize == 0); - Preconditions.checkState(streamBufferMaxSize % streamBufferFlushSize == 0); - Preconditions.checkState(blockSize % streamBufferMaxSize == 0); - this.bufferList = new ArrayList<>(); - } - - /** - * When a key is opened, it is possible that there are some blocks already - * allocated to it for this open session. In this case, to make use of these - * blocks, we need to add these blocks to stream entries. But, a key's version - * also includes blocks from previous versions, we need to avoid adding these - * old blocks to stream entries, because these old blocks should not be picked - * for write. To do this, the following method checks that, only those - * blocks created in this particular open version are added to stream entries. - * - * @param version the set of blocks that are pre-allocated. - * @param openVersion the version corresponding to the pre-allocation. - * @throws IOException - */ - public void addPreallocateBlocks(OmKeyLocationInfoGroup version, - long openVersion) throws IOException { - // server may return any number of blocks, (0 to any) - // only the blocks allocated in this open session (block createVersion - // equals to open session version) - for (OmKeyLocationInfo subKeyInfo : version.getLocationList()) { - if (subKeyInfo.getCreateVersion() == openVersion) { - addKeyLocationInfo(subKeyInfo); - } - } - } - - private void addKeyLocationInfo(OmKeyLocationInfo subKeyInfo) - throws IOException { - ContainerWithPipeline containerWithPipeline = scmClient - .getContainerWithPipeline(subKeyInfo.getContainerID()); - XceiverClientSpi xceiverClient = - xceiverClientManager.acquireClient(containerWithPipeline.getPipeline()); - streamEntries.add(new ChunkOutputStreamEntry(subKeyInfo.getBlockID(), - keyArgs.getKeyName(), xceiverClientManager, xceiverClient, requestID, - chunkSize, subKeyInfo.getLength(), streamBufferFlushSize, - streamBufferMaxSize, watchTimeout, bufferList, checksum)); - } - - - @Override - public void write(int b) throws IOException { - byte[] buf = new byte[1]; - buf[0] = (byte) b; - write(buf, 0, 1); - } - - /** - * Try to write the bytes sequence b[off:off+len) to streams. - * - * NOTE: Throws exception if the data could not fit into the remaining space. - * In which case nothing will be written. - * TODO:May need to revisit this behaviour. - * - * @param b byte data - * @param off starting offset - * @param len length to write - * @throws IOException - */ - @Override - public void write(byte[] b, int off, int len) - throws IOException { - checkNotClosed(); - if (b == null) { - throw new NullPointerException(); - } - if ((off < 0) || (off > b.length) || (len < 0) || ((off + len) > b.length) - || ((off + len) < 0)) { - throw new IndexOutOfBoundsException(); - } - if (len == 0) { - return; - } - handleWrite(b, off, len, false); - } - - private long computeBufferData() { - return bufferList.stream().mapToInt(value -> value.position()) - .sum(); - } - - private void handleWrite(byte[] b, int off, long len, boolean retry) - throws IOException { - int succeededAllocates = 0; - while (len > 0) { - if (streamEntries.size() <= currentStreamIndex) { - Preconditions.checkNotNull(omClient); - // allocate a new block, if a exception happens, log an error and - // throw exception to the caller directly, and the write fails. - try { - allocateNewBlock(currentStreamIndex); - succeededAllocates += 1; - } catch (IOException ioe) { - LOG.error("Try to allocate more blocks for write failed, already " - + "allocated " + succeededAllocates + " blocks for this write."); - throw ioe; - } - } - // in theory, this condition should never violate due the check above - // still do a sanity check. - Preconditions.checkArgument(currentStreamIndex < streamEntries.size()); - ChunkOutputStreamEntry current = streamEntries.get(currentStreamIndex); - - // length(len) will be in int range if the call is happening through - // write API of chunkOutputStream. Length can be in long range if it comes - // via Exception path. - int writeLen = Math.min((int)len, (int) current.getRemaining()); - long currentPos = current.getWrittenDataLength(); - try { - if (retry) { - current.writeOnRetry(len); - } else { - current.write(b, off, writeLen); - } - } catch (IOException ioe) { - if (checkIfContainerIsClosed(ioe) || checkIfTimeoutException(ioe)) { - // for the current iteration, totalDataWritten - currentPos gives the - // amount of data already written to the buffer - writeLen = (int) (current.getWrittenDataLength() - currentPos); - LOG.debug("writeLen {}, total len {}", writeLen, len); - handleException(current, currentStreamIndex); - } else { - throw ioe; - } - } - if (current.getRemaining() <= 0) { - // since the current block is already written close the stream. - handleFlushOrClose(true); - currentStreamIndex += 1; - } - len -= writeLen; - off += writeLen; - } - } - - /** - * Discards the subsequent pre allocated blocks and removes the streamEntries - * from the streamEntries list for the container which is closed. - * @param containerID id of the closed container - */ - private void discardPreallocatedBlocks(long containerID) { - // currentStreamIndex < streamEntries.size() signifies that, there are still - // pre allocated blocks available. - if (currentStreamIndex < streamEntries.size()) { - ListIterator<ChunkOutputStreamEntry> streamEntryIterator = - streamEntries.listIterator(currentStreamIndex); - while (streamEntryIterator.hasNext()) { - if (streamEntryIterator.next().blockID.getContainerID() - == containerID) { - streamEntryIterator.remove(); - } - } - } - } - - /** - * It might be possible that the blocks pre allocated might never get written - * while the stream gets closed normally. In such cases, it would be a good - * idea to trim down the locationInfoList by removing the unused blocks if any - * so as only the used block info gets updated on OzoneManager during close. - */ - private void removeEmptyBlocks() { - if (currentStreamIndex < streamEntries.size()) { - ListIterator<ChunkOutputStreamEntry> streamEntryIterator = - streamEntries.listIterator(currentStreamIndex); - while (streamEntryIterator.hasNext()) { - if (streamEntryIterator.next().currentPosition == 0) { - streamEntryIterator.remove(); - } - } - } - } - /** - * It performs following actions : - * a. Updates the committed length at datanode for the current stream in - * datanode. - * b. Reads the data from the underlying buffer and writes it the next stream. - * - * @param streamEntry StreamEntry - * @param streamIndex Index of the entry - * @throws IOException Throws IOException if Write fails - */ - private void handleException(ChunkOutputStreamEntry streamEntry, - int streamIndex) throws IOException { - long totalSuccessfulFlushedData = - streamEntry.getTotalSuccessfulFlushedData(); - //set the correct length for the current stream - streamEntry.currentPosition = totalSuccessfulFlushedData; - long bufferedDataLen = computeBufferData(); - // just clean up the current stream. - streamEntry.cleanup(); - if (bufferedDataLen > 0) { - // If the data is still cached in the underlying stream, we need to - // allocate new block and write this data in the datanode. - currentStreamIndex += 1; - handleWrite(null, 0, bufferedDataLen, true); - } - if (totalSuccessfulFlushedData == 0) { - streamEntries.remove(streamIndex); - currentStreamIndex -= 1; - } - // discard subsequent pre allocated blocks from the streamEntries list - // from the closed container - discardPreallocatedBlocks(streamEntry.blockID.getContainerID()); - } - - private boolean checkIfContainerIsClosed(IOException ioe) { - if (ioe.getCause() != null) { - return checkIfContainerNotOpenOrRaftRetryFailureException(ioe) || Optional - .of(ioe.getCause()) - .filter(e -> e instanceof StorageContainerException) - .map(e -> (StorageContainerException) e) - .filter(sce -> sce.getResult() == Result.CLOSED_CONTAINER_IO) - .isPresent(); - } - return false; - } - - private boolean checkIfContainerNotOpenOrRaftRetryFailureException( - IOException ioe) { - Throwable t = ioe.getCause(); - while (t != null) { - if (t instanceof ContainerNotOpenException - || t instanceof RaftRetryFailureException) { - return true; - } - t = t.getCause(); - } - return false; - } - - private boolean checkIfTimeoutException(IOException ioe) { - if (ioe.getCause() != null) { - return Optional.of(ioe.getCause()) - .filter(e -> e instanceof TimeoutException).isPresent(); - } else { - return false; - } - } - - private long getKeyLength() { - return streamEntries.stream().mapToLong(e -> e.currentPosition) - .sum(); - } - - /** - * Contact OM to get a new block. Set the new block with the index (e.g. - * first block has index = 0, second has index = 1 etc.) - * - * The returned block is made to new ChunkOutputStreamEntry to write. - * - * @param index the index of the block. - * @throws IOException - */ - private void allocateNewBlock(int index) throws IOException { - OmKeyLocationInfo subKeyInfo = omClient.allocateBlock(keyArgs, openID); - addKeyLocationInfo(subKeyInfo); - } - - @Override - public void flush() throws IOException { - checkNotClosed(); - handleFlushOrClose(false); - } - - /** - * Close or Flush the latest outputStream. - * @param close Flag which decides whether to call close or flush on the - * outputStream. - * @throws IOException In case, flush or close fails with exception. - */ - private void handleFlushOrClose(boolean close) throws IOException { - if (streamEntries.size() == 0) { - return; - } - int size = streamEntries.size(); - int streamIndex = - currentStreamIndex >= size ? size - 1 : currentStreamIndex; - ChunkOutputStreamEntry entry = streamEntries.get(streamIndex); - if (entry != null) { - try { - if (close) { - entry.close(); - } else { - entry.flush(); - } - } catch (IOException ioe) { - if (checkIfContainerIsClosed(ioe) || checkIfTimeoutException(ioe)) { - // This call will allocate a new streamEntry and write the Data. - // Close needs to be retried on the newly allocated streamEntry as - // as well. - handleException(entry, streamIndex); - handleFlushOrClose(close); - } else { - throw ioe; - } - } - } - } - - /** - * Commit the key to OM, this will add the blocks as the new key blocks. - * - * @throws IOException - */ - @Override - public void close() throws IOException { - if (closed) { - return; - } - closed = true; - try { - handleFlushOrClose(true); - if (keyArgs != null) { - // in test, this could be null - removeEmptyBlocks(); - keyArgs.setDataSize(getKeyLength()); - keyArgs.setLocationInfoList(getLocationInfoList()); - // When the key is multipart upload part file upload, we should not - // commit the key, as this is not an actual key, this is a just a - // partial key of a large file. - if (keyArgs.getIsMultipartKey()) { - commitUploadPartInfo = omClient.commitMultipartUploadPart(keyArgs, - openID); - } else { - omClient.commitKey(keyArgs, openID); - } - } else { - LOG.warn("Closing ChunkGroupOutputStream, but key args is null"); - } - } catch (IOException ioe) { - throw ioe; - } finally { - if (bufferList != null) { - bufferList.stream().forEach(e -> e.clear()); - } - bufferList = null; - } - } - - public OmMultipartCommitUploadPartInfo getCommitUploadPartInfo() { - return commitUploadPartInfo; - } - - /** - * Builder class of ChunkGroupOutputStream. - */ - public static class Builder { - private OpenKeySession openHandler; - private XceiverClientManager xceiverManager; - private StorageContainerLocationProtocolClientSideTranslatorPB scmClient; - private OzoneManagerProtocolClientSideTranslatorPB omClient; - private int chunkSize; - private String requestID; - private ReplicationType type; - private ReplicationFactor factor; - private long streamBufferFlushSize; - private long streamBufferMaxSize; - private long blockSize; - private long watchTimeout; - private Checksum checksum; - private String multipartUploadID; - private int multipartNumber; - private boolean isMultipartKey; - - - public Builder setMultipartUploadID(String uploadID) { - this.multipartUploadID = uploadID; - return this; - } - - public Builder setMultipartNumber(int partNumber) { - this.multipartNumber = partNumber; - return this; - } - - public Builder setHandler(OpenKeySession handler) { - this.openHandler = handler; - return this; - } - - public Builder setXceiverClientManager(XceiverClientManager manager) { - this.xceiverManager = manager; - return this; - } - - public Builder setScmClient( - StorageContainerLocationProtocolClientSideTranslatorPB client) { - this.scmClient = client; - return this; - } - - public Builder setOmClient( - OzoneManagerProtocolClientSideTranslatorPB client) { - this.omClient = client; - return this; - } - - public Builder setChunkSize(int size) { - this.chunkSize = size; - return this; - } - - public Builder setRequestID(String id) { - this.requestID = id; - return this; - } - - public Builder setType(ReplicationType replicationType) { - this.type = replicationType; - return this; - } - - public Builder setFactor(ReplicationFactor replicationFactor) { - this.factor = replicationFactor; - return this; - } - - public Builder setStreamBufferFlushSize(long size) { - this.streamBufferFlushSize = size; - return this; - } - - public Builder setStreamBufferMaxSize(long size) { - this.streamBufferMaxSize = size; - return this; - } - - public Builder setBlockSize(long size) { - this.blockSize = size; - return this; - } - - public Builder setWatchTimeout(long timeout) { - this.watchTimeout = timeout; - return this; - } - - public Builder setChecksum(Checksum checksumObj){ - this.checksum = checksumObj; - return this; - } - - public Builder setIsMultipartKey(boolean isMultipart) { - this.isMultipartKey = isMultipart; - return this; - } - - public ChunkGroupOutputStream build() throws IOException { - return new ChunkGroupOutputStream(openHandler, xceiverManager, scmClient, - omClient, chunkSize, requestID, factor, type, streamBufferFlushSize, - streamBufferMaxSize, blockSize, watchTimeout, checksum, - multipartUploadID, multipartNumber, isMultipartKey); - } - } - - private static class ChunkOutputStreamEntry extends OutputStream { - private OutputStream outputStream; - private BlockID blockID; - private final String key; - private final XceiverClientManager xceiverClientManager; - private final XceiverClientSpi xceiverClient; - private final Checksum checksum; - private final String requestId; - private final int chunkSize; - // total number of bytes that should be written to this stream - private final long length; - // the current position of this stream 0 <= currentPosition < length - private long currentPosition; - - private final long streamBufferFlushSize; - private final long streamBufferMaxSize; - private final long watchTimeout; - private List<ByteBuffer> bufferList; - - ChunkOutputStreamEntry(BlockID blockID, String key, - XceiverClientManager xceiverClientManager, - XceiverClientSpi xceiverClient, String requestId, int chunkSize, - long length, long streamBufferFlushSize, long streamBufferMaxSize, - long watchTimeout, List<ByteBuffer> bufferList, Checksum checksum) { - this.outputStream = null; - this.blockID = blockID; - this.key = key; - this.xceiverClientManager = xceiverClientManager; - this.xceiverClient = xceiverClient; - this.requestId = requestId; - this.chunkSize = chunkSize; - - this.length = length; - this.currentPosition = 0; - this.streamBufferFlushSize = streamBufferFlushSize; - this.streamBufferMaxSize = streamBufferMaxSize; - this.watchTimeout = watchTimeout; - this.checksum = checksum; - this.bufferList = bufferList; - } - - /** - * For testing purpose, taking a some random created stream instance. - * @param outputStream a existing writable output stream - * @param length the length of data to write to the stream - */ - ChunkOutputStreamEntry(OutputStream outputStream, long length, - Checksum checksum) { - this.outputStream = outputStream; - this.blockID = null; - this.key = null; - this.xceiverClientManager = null; - this.xceiverClient = null; - this.requestId = null; - this.chunkSize = -1; - - this.length = length; - this.currentPosition = 0; - streamBufferFlushSize = 0; - streamBufferMaxSize = 0; - bufferList = null; - watchTimeout = 0; - this.checksum = checksum; - } - - long getLength() { - return length; - } - - long getRemaining() { - return length - currentPosition; - } - - private void checkStream() { - if (this.outputStream == null) { - this.outputStream = - new ChunkOutputStream(blockID, key, xceiverClientManager, - xceiverClient, requestId, chunkSize, streamBufferFlushSize, - streamBufferMaxSize, watchTimeout, bufferList, checksum); - } - } - - @Override - public void write(int b) throws IOException { - checkStream(); - outputStream.write(b); - this.currentPosition += 1; - } - - @Override - public void write(byte[] b, int off, int len) throws IOException { - checkStream(); - outputStream.write(b, off, len); - this.currentPosition += len; - } - - @Override - public void flush() throws IOException { - if (this.outputStream != null) { - this.outputStream.flush(); - } - } - - @Override - public void close() throws IOException { - if (this.outputStream != null) { - this.outputStream.close(); - // after closing the chunkOutPutStream, blockId would have been - // reconstructed with updated bcsId - if (this.outputStream instanceof ChunkOutputStream) { - this.blockID = ((ChunkOutputStream) outputStream).getBlockID(); - } - } - } - - long getTotalSuccessfulFlushedData() throws IOException { - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - blockID = out.getBlockID(); - return out.getTotalSuccessfulFlushedData(); - } else if (outputStream == null) { - // For a pre allocated block for which no write has been initiated, - // the OutputStream will be null here. - // In such cases, the default blockCommitSequenceId will be 0 - return 0; - } - throw new IOException("Invalid Output Stream for Key: " + key); - } - - long getWrittenDataLength() throws IOException { - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - return out.getWrittenDataLength(); - } else if (outputStream == null) { - // For a pre allocated block for which no write has been initiated, - // the OutputStream will be null here. - // In such cases, the default blockCommitSequenceId will be 0 - return 0; - } - throw new IOException("Invalid Output Stream for Key: " + key); - } - - void cleanup() { - checkStream(); - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - out.cleanup(); - } - } - - void writeOnRetry(long len) throws IOException { - checkStream(); - if (this.outputStream instanceof ChunkOutputStream) { - ChunkOutputStream out = (ChunkOutputStream) this.outputStream; - out.writeOnRetry(len); - this.currentPosition += len; - } else { - throw new IOException("Invalid Output Stream for Key: " + key); - } - } - } - - /** - * Verify that the output stream is open. Non blocking; this gives - * the last state of the volatile {@link #closed} field. - * @throws IOException if the connection is closed. - */ - private void checkNotClosed() throws IOException { - if (closed) { - throw new IOException( - ": " + FSExceptionMessages.STREAM_IS_CLOSED + " Key: " + keyArgs - .getKeyName()); - } - } -} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
