You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by GitBox <gi...@apache.org> on 2021/08/06 15:01:22 UTC
[GitHub] [ozone] kaijchen commented on a change in pull request #2495: HDDS-5488. [Ozone-Streaming] Add a new BlockOutputStream/KeyOutputStream to support streaming api
kaijchen commented on a change in pull request #2495:
URL: https://github.com/apache/ozone/pull/2495#discussion_r683894813
##########
File path: hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java
##########
@@ -0,0 +1,767 @@
+/*
+ * 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.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.KeyValue;
+import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.XceiverClientFactory;
+import org.apache.hadoop.hdds.scm.XceiverClientRatis;
+import org.apache.hadoop.hdds.scm.XceiverClientReply;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.ozone.common.Checksum;
+import org.apache.hadoop.ozone.common.ChecksumData;
+import org.apache.hadoop.ozone.common.ChunkBuffer;
+import org.apache.hadoop.ozone.common.OzoneChecksumException;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.security.token.TokenIdentifier;
+import org.apache.ratis.client.api.DataStreamOutput;
+import org.apache.ratis.io.StandardWriteOption;
+import org.apache.ratis.protocol.DataStreamReply;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.putBlockAsync;
+
+/**
+ * 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 BlockDataStreamOutput extends OutputStream
+ implements org.apache.hadoop.hdds.scm.storage.DataStreamOutput {
+ public static final Logger LOG =
+ LoggerFactory.getLogger(BlockDataStreamOutput.class);
+ public static final String EXCEPTION_MSG =
+ "Unexpected Storage Container Exception: ";
+ private static final CompletableFuture[] EMPTY_FUTURE_ARRAY = {};
+
+ private AtomicReference<BlockID> blockID;
+
+ private final BlockData.Builder containerBlockData;
+ private XceiverClientFactory xceiverClientFactory;
+ private XceiverClientRatis xceiverClient;
+ private OzoneClientConfig config;
+
+ private int chunkIndex;
+ private final AtomicLong chunkOffset = new AtomicLong();
+ private final BufferPool bufferPool;
+ // 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 final AtomicReference<IOException> ioException;
+ private final 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;
+
+ // List containing buffers for which the putBlock call will
+ // update the length in the datanodes. This list will just maintain
+ // references to the buffers in the BufferPool which will be cleared
+ // when the watchForCommit acknowledges a putBlock logIndex has been
+ // committed on all datanodes. This list will be a place holder for buffers
+ // which got written between successive putBlock calls.
+ private List<ChunkBuffer> bufferList;
+
+ // This object will maintain the commitIndexes and byteBufferList in order
+ // Also, corresponding to the logIndex, the corresponding list of buffers will
+ // be released from the buffer pool.
+ private final CommitWatcher commitWatcher;
+
+ private final List<DatanodeDetails> failedServers;
+ private final Checksum checksum;
+
+ //number of buffers used before doing a flush/putBlock.
+ private int flushPeriod;
+ //bytes remaining to write in the current buffer.
+ private int currentBufferRemaining;
+ //current buffer allocated to write
+ private ChunkBuffer currentBuffer;
+ private final Token<? extends TokenIdentifier> token;
+ private final DataStreamOutput out;
+ private CompletableFuture<DataStreamReply> dataStreamCloseReply;
+ private List<CompletableFuture<DataStreamReply>> futures = new ArrayList<>();
+ private final long syncSize = 0; // TODO: disk sync is disabled for now
+ private long syncPosition = 0;
+
+ /**
+ * Creates a new BlockOutputStream.
+ *
+ * @param blockID block ID
+ * @param xceiverClientManager client manager that controls client
+ * @param pipeline pipeline where block will be written
+ * @param bufferPool pool of buffers
+ */
+ public BlockDataStreamOutput(
+ BlockID blockID,
+ XceiverClientFactory xceiverClientManager,
+ Pipeline pipeline,
+ BufferPool bufferPool,
+ OzoneClientConfig config,
+ Token<? extends TokenIdentifier> token
+ ) throws IOException {
+ this.xceiverClientFactory = xceiverClientManager;
+ this.config = config;
+ this.blockID = new AtomicReference<>(blockID);
+ KeyValue keyValue =
+ KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build();
+ this.containerBlockData =
+ BlockData.newBuilder().setBlockID(blockID.getDatanodeBlockIDProtobuf())
+ .addMetadata(keyValue);
+ this.xceiverClient =
+ (XceiverClientRatis)xceiverClientManager.acquireClient(pipeline);
+ // Alternatively, stream setup can be delayed till the first chunk write.
+ this.out = setupStream();
+ this.bufferPool = bufferPool;
+ this.token = token;
+
+ //number of buffers used before doing a flush
+ refreshCurrentBuffer(bufferPool);
+ flushPeriod = (int) (config.getStreamBufferFlushSize() / config
+ .getStreamBufferSize());
+
+ Preconditions
+ .checkArgument(
+ (long) flushPeriod * config.getStreamBufferSize() == config
+ .getStreamBufferFlushSize());
+
+ // A single thread executor handle the responses of async requests
+ responseExecutor = Executors.newSingleThreadExecutor();
+ commitWatcher = new CommitWatcher(bufferPool, xceiverClient);
+ bufferList = null;
+ totalDataFlushedLength = 0;
+ writtenDataLength = 0;
+ failedServers = new ArrayList<>(0);
+ ioException = new AtomicReference<>(null);
+ checksum = new Checksum(config.getChecksumType(),
+ config.getBytesPerChecksum());
+ }
+
+ private DataStreamOutput setupStream() throws IOException {
+ // Execute a dummy WriteChunk request to get the path of the target file,
+ // but does NOT write any data to it.
+ ContainerProtos.WriteChunkRequestProto.Builder writeChunkRequest =
+ ContainerProtos.WriteChunkRequestProto.newBuilder()
+ .setBlockID(blockID.get().getDatanodeBlockIDProtobuf());
+
+ String id = xceiverClient.getPipeline().getFirstNode().getUuidString();
+ ContainerProtos.ContainerCommandRequestProto.Builder builder =
+ ContainerProtos.ContainerCommandRequestProto.newBuilder()
+ .setCmdType(ContainerProtos.Type.StreamInit)
+ .setContainerID(blockID.get().getContainerID())
+ .setDatanodeUuid(id).setWriteChunk(writeChunkRequest);
+
+ ContainerCommandRequestMessage message =
+ ContainerCommandRequestMessage.toMessage(builder.build(), null);
+
+ return Preconditions.checkNotNull(xceiverClient.getDataStreamApi())
+ .stream(message.getContent().asReadOnlyByteBuffer());
+ }
+
+ private void refreshCurrentBuffer(BufferPool pool) {
+ currentBuffer = pool.getCurrentBuffer();
+ currentBufferRemaining =
+ currentBuffer != null ? currentBuffer.remaining() : 0;
+ }
+
+ public BlockID getBlockID() {
+ return blockID.get();
+ }
+
+ public long getTotalAckDataLength() {
+ return commitWatcher.getTotalAckDataLength();
+ }
+
+ public long getWrittenDataLength() {
+ return writtenDataLength;
+ }
+
+ public List<DatanodeDetails> getFailedServers() {
+ return failedServers;
+ }
+
+ @VisibleForTesting
+ public XceiverClientRatis getXceiverClient() {
+ return xceiverClient;
+ }
+
+ @VisibleForTesting
+ public long getTotalDataFlushedLength() {
+ return totalDataFlushedLength;
+ }
+
+ @VisibleForTesting
+ public BufferPool getBufferPool() {
+ return bufferPool;
+ }
+
+ public IOException getIoException() {
+ return ioException.get();
+ }
+
+ @VisibleForTesting
+ public Map<Long, List<ChunkBuffer>> getCommitIndex2flushedDataMap() {
+ return commitWatcher.getCommitIndex2flushedDataMap();
+ }
+
+ @Override
+ public void write(ByteBuf b) throws IOException {
+ checkOpen();
+ if (b == null) {
+ throw new NullPointerException();
+ }
+ final int len = b.readableBytes();
+ if (len == 0) {
+ return;
+ }
+ ChunkBuffer chunk = ChunkBuffer.wrap(b.nioBuffer());
+ writeChunk(chunk);
+ writtenDataLength += len;
+ updateFlushLength();
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ byte[] buf = new byte[1];
+ buf[0] = (byte) b;
+ write(buf);
+ }
+
+ private void writeChunkIfNeeded() throws IOException {
+ if (currentBufferRemaining == 0) {
+ writeChunk(currentBuffer);
+ }
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ write(Unpooled.wrappedBuffer(b), off, len);
+ }
+
+ private void doFlushOrWatchIfNeeded() throws IOException {
+ if (currentBufferRemaining == 0) {
+ if (bufferPool.getNumberOfUsedBuffers() % flushPeriod == 0) {
+ updateFlushLength();
+ executePutBlock(false, false);
+ }
+ // Data in the bufferPool can not exceed streamBufferMaxSize
+ if (bufferPool.getNumberOfUsedBuffers() == bufferPool.getCapacity()) {
+ handleFullBuffer();
+ }
+ }
+ }
+
+ private void allocateNewBufferIfNeeded() {
+ if (currentBufferRemaining == 0) {
+ currentBuffer = bufferPool.allocateBuffer(config.getBufferIncrement());
+ currentBufferRemaining = currentBuffer.remaining();
+ }
+ }
+
+ private void updateFlushLength() {
+ totalDataFlushedLength = writtenDataLength;
+ }
+
+ private boolean isBufferPoolFull() {
+ return bufferPool.computeBufferData() == config.getStreamBufferMaxSize();
+ }
+
+ /**
+ * 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;
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Retrying write length {} for blockID {}", len, blockID);
+ }
+ Preconditions.checkArgument(len <= config.getStreamBufferMaxSize());
+ int count = 0;
+ while (len > 0) {
+ ChunkBuffer buffer = bufferPool.getBuffer(count);
+ long writeLen = Math.min(buffer.position(), len);
+ if (!buffer.hasRemaining()) {
+ writeChunk(buffer);
+ }
+ len -= writeLen;
+ count++;
+ writtenDataLength += writeLen;
+ // we should not call isBufferFull/shouldFlush here.
+ // The buffer might already be full as whole data is already cached in
+ // the buffer. We should just validate
+ // if we wrote data of size streamBufferMaxSize/streamBufferFlushSize to
+ // call for handling full buffer/flush buffer condition.
+ if (writtenDataLength % config.getStreamBufferFlushSize() == 0) {
+ // reset the position to zero as now we will be reading the
+ // next buffer in the list
+ updateFlushLength();
+ executePutBlock(false, false);
+ }
+ if (writtenDataLength == config.getStreamBufferMaxSize()) {
+ handleFullBuffer();
+ }
+ }
+ }
+
+ /**
+ * 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 (!commitWatcher.getFutureMap().isEmpty()) {
+ waitOnFlushFutures();
+ }
+ } catch (ExecutionException e) {
+ handleExecutionException(e);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ handleInterruptedException(ex, true);
+ }
+ watchForCommit(true);
+ }
+
+
+ // 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() {
+ commitWatcher.releaseBuffersOnException();
+ refreshCurrentBuffer(bufferPool);
+ }
+
+ /**
+ * calls watchForCommit API of the Ratis Client. For Standalone client,
+ * it is a no op.
+ * @param bufferFull flag indicating whether bufferFull condition is hit or
+ * its called as part flush/close
+ * @return minimum commit index replicated to all nodes
+ * @throws IOException IOException in case watch gets timed out
+ */
+ private void watchForCommit(boolean bufferFull) throws IOException {
+ checkOpen();
+ try {
+ XceiverClientReply reply = bufferFull ?
+ commitWatcher.watchOnFirstIndex() : commitWatcher.watchOnLastIndex();
+ if (reply != null) {
+ List<DatanodeDetails> dnList = reply.getDatanodes();
+ if (!dnList.isEmpty()) {
+ Pipeline pipe = xceiverClient.getPipeline();
+
+ LOG.warn("Failed to commit BlockId {} on {}. Failed nodes: {}",
+ blockID, pipe, dnList);
+ failedServers.addAll(dnList);
+ }
+ }
+ } catch (IOException ioe) {
+ setIoException(ioe);
+ throw getIoException();
+ }
+ refreshCurrentBuffer(bufferPool);
+
+ }
+
+ /**
+ * @param close whether putBlock is happening as part of closing the stream
+ * @param force true if no data was written since most recent putBlock and
+ * stream is being closed
+ */
+ private CompletableFuture<ContainerProtos.
+ ContainerCommandResponseProto> executePutBlock(boolean close,
+ boolean force) throws IOException {
+ checkOpen();
+ long flushPos = totalDataFlushedLength;
+ final List<ChunkBuffer> byteBufferList;
+ if (!force) {
+ Preconditions.checkNotNull(bufferList);
+ byteBufferList = bufferList;
+ bufferList = null;
+ Preconditions.checkNotNull(byteBufferList);
+ } else {
+ byteBufferList = null;
+ }
+
+ try {
+ CompletableFuture.allOf(futures.toArray(EMPTY_FUTURE_ARRAY)).get();
+ } catch (Exception e) {
+ LOG.warn("Failed to write all chunks through stream: " + e);
+ throw new IOException(e);
+ }
+ if (close) {
+ dataStreamCloseReply = out.closeAsync();
+ }
+
+ CompletableFuture<ContainerProtos.
+ ContainerCommandResponseProto> flushFuture = null;
+ try {
+ BlockData blockData = containerBlockData.build();
+ XceiverClientReply asyncReply =
+ putBlockAsync(xceiverClient, blockData, close, token);
+ CompletableFuture<ContainerProtos.ContainerCommandResponseProto> future =
+ asyncReply.getResponse();
+ flushFuture = future.thenApplyAsync(e -> {
+ try {
+ validateResponse(e);
+ } catch (IOException sce) {
+ throw new CompletionException(sce);
+ }
+ // if the ioException is not set, putBlock is successful
+ if (getIoException() == null && !force) {
+ BlockID responseBlockID = BlockID.getFromProtobuf(
+ e.getPutBlock().getCommittedBlockLength().getBlockID());
+ Preconditions.checkState(blockID.get().getContainerBlockID()
+ .equals(responseBlockID.getContainerBlockID()));
+ // updates the bcsId of the block
+ blockID.set(responseBlockID);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Adding index " + asyncReply.getLogIndex() + " commitMap size "
+ + commitWatcher.getCommitInfoMapSize() + " flushLength "
+ + flushPos + " numBuffers " + byteBufferList.size()
+ + " blockID " + blockID + " bufferPool size" + bufferPool
+ .getSize() + " currentBufferIndex " + bufferPool
+ .getCurrentBufferIndex());
+ }
+ // for standalone protocol, logIndex will always be 0.
+ commitWatcher
+ .updateCommitInfoMap(asyncReply.getLogIndex(), byteBufferList);
+ }
+ return e;
+ }, responseExecutor).exceptionally(e -> {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("putBlock failed for blockID {} with exception {}",
+ blockID, e.getLocalizedMessage());
+ }
+ CompletionException ce = new CompletionException(e);
+ setIoException(ce);
+ throw ce;
+ });
+ } catch (IOException | ExecutionException e) {
+ throw new IOException(EXCEPTION_MSG + e.toString(), e);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ handleInterruptedException(ex, false);
+ }
+ commitWatcher.getFutureMap().put(flushPos, flushFuture);
+ return flushFuture;
+ }
+
+ @Override
+ public void flush() throws IOException {
+ if (xceiverClientFactory != null && xceiverClient != null
+ && bufferPool != null && bufferPool.getSize() > 0
+ && (!config.isStreamBufferFlushDelay() ||
+ writtenDataLength - totalDataFlushedLength
+ >= config.getStreamBufferSize())) {
+ try {
+ handleFlush(false);
+ } catch (ExecutionException e) {
+ // just set the exception here as well in order to maintain sanctity of
+ // ioException field
+ handleExecutionException(e);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ handleInterruptedException(ex, true);
+ }
+ }
+ }
+
+ private void writeChunk(ChunkBuffer buffer)
+ throws IOException {
+ // This data in the buffer will be pushed to datanode and a reference will
+ // be added to the bufferList. Once putBlock gets executed, this list will
+ // be marked null. Hence, during first writeChunk call after every putBlock
+ // call or during the first call to writeChunk here, the list will be null.
+
+ if (bufferList == null) {
+ bufferList = new ArrayList<>();
+ }
+ bufferList.add(buffer);
+ writeChunkToContainer(buffer.duplicate(0, buffer.position()));
Review comment:
What's the difference between `duplicate` and `slice`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org