hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyInputStream.java - hadoop - Git at Google

 /*
  * 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 org.apache.hadoop.fs.FSExceptionMessages;
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.storage.BlockInputStream;
 import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;

 /**
  * Maintaining a list of BlockInputStream. Read based on offset.
  */
 public class KeyInputStream extends InputStream implements Seekable {

   private static final Logger LOG =
       LoggerFactory.getLogger(KeyInputStream.class);

   private static final int EOF = -1;

   private String key;
   private long length = 0;
   private boolean closed = false;

   // List of BlockInputStreams, one for each block in the key
   private final List<BlockInputStream> blockStreams;

   // blockOffsets[i] stores the index of the first data byte in
   // blockStream w.r.t the key data.
   // For example, let’s say the block size is 200 bytes and block[0] stores
   // data from indices 0 - 199, block[1] from indices 200 - 399 and so on.
   // Then, blockOffset[0] = 0 (the offset of the first byte of data in
   // block[0]), blockOffset[1] = 200 and so on.
   private long[] blockOffsets = null;

   // Index of the blockStream corresponding to the current position of the
   // KeyInputStream i.e. offset of the data to be read next
   private int blockIndex;

   // Tracks the blockIndex corresponding to the last seeked position so that it
   // can be reset if a new position is seeked.
   private int blockIndexOfPrevPosition;

   public KeyInputStream() {
     blockStreams = new ArrayList<>();
     blockIndex = 0;
   }

   /**
    * For each block in keyInfo, add a BlockInputStream to blockStreams.
    */
   public static LengthInputStream getFromOmKeyInfo(OmKeyInfo keyInfo,
       XceiverClientManager xceiverClientManager,
       boolean verifyChecksum) {
     List<OmKeyLocationInfo> keyLocationInfos = keyInfo
         .getLatestVersionLocations().getBlocksLatestVersionOnly();

     KeyInputStream keyInputStream = new KeyInputStream();
     keyInputStream.initialize(keyInfo.getKeyName(), keyLocationInfos,
         xceiverClientManager, verifyChecksum);

     return new LengthInputStream(keyInputStream, keyInputStream.length);
   }

   private synchronized void initialize(String keyName,
       List<OmKeyLocationInfo> blockInfos,
       XceiverClientManager xceiverClientManager,
       boolean verifyChecksum) {
     this.key = keyName;
     this.blockOffsets = new long[blockInfos.size()];
     long keyLength = 0;
     for (int i = 0; i < blockInfos.size(); i++) {
       OmKeyLocationInfo omKeyLocationInfo = blockInfos.get(i);
       if (LOG.isDebugEnabled()) {
         LOG.debug("Adding stream for accessing {}. The stream will be " +
             "initialized later.", omKeyLocationInfo);
       }

       addStream(omKeyLocationInfo, xceiverClientManager,
           verifyChecksum);

       this.blockOffsets[i] = keyLength;
       keyLength += omKeyLocationInfo.getLength();
     }
     this.length = keyLength;
   }

   /**
    * Append another BlockInputStream to the end of the list. Note that the
    * BlockInputStream is only created here and not initialized. The
    * BlockInputStream is initialized when a read operation is performed on
    * the block for the first time.
    */
   private synchronized void addStream(OmKeyLocationInfo blockInfo,
       XceiverClientManager xceiverClientMngr,
       boolean verifyChecksum) {
     blockStreams.add(new BlockInputStream(blockInfo.getBlockID(),
         blockInfo.getLength(), blockInfo.getPipeline(), blockInfo.getToken(),
         verifyChecksum, xceiverClientMngr));
   }

   @VisibleForTesting
   public void addStream(BlockInputStream blockInputStream) {
     blockStreams.add(blockInputStream);
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public synchronized int read() throws IOException {
     byte[] buf = new byte[1];
     if (read(buf, 0, 1) == EOF) {
       return EOF;
     }
     return Byte.toUnsignedInt(buf[0]);
   }

   /**
    * {@inheritDoc}
    */
   @Override
   public synchronized int read(byte[] b, int off, int len) throws IOException {
     checkOpen();
     if (b == null) {
       throw new NullPointerException();
     }
     if (off < 0 || len < 0 || len > b.length - off) {
       throw new IndexOutOfBoundsException();
     }
     if (len == 0) {
       return 0;
     }
     int totalReadLen = 0;
     while (len > 0) {
       // if we are at the last block and have read the entire block, return
       if (blockStreams.size() == 0 ||
           (blockStreams.size() - 1 <= blockIndex &&
               blockStreams.get(blockIndex)
                   .getRemaining() == 0)) {
         return totalReadLen == 0 ? EOF : totalReadLen;
       }

       // Get the current blockStream and read data from it
       BlockInputStream current = blockStreams.get(blockIndex);
       int numBytesToRead = Math.min(len, (int)current.getRemaining());
       int numBytesRead = current.read(b, off, numBytesToRead);
       if (numBytesRead != numBytesToRead) {
         // This implies that there is either data loss or corruption in the
         // chunk entries. Even EOF in the current stream would be covered in
         // this case.
         throw new IOException(String.format(
             "Inconsistent read for blockID=%s length=%d numBytesRead=%d",
             current.getBlockID(), current.getLength(), numBytesRead));
       }
       totalReadLen += numBytesRead;
       off += numBytesRead;
       len -= numBytesRead;
       if (current.getRemaining() <= 0 &&
           ((blockIndex + 1) < blockStreams.size())) {
         blockIndex += 1;
       }
     }
     return totalReadLen;
   }

   /**
    * Seeks the KeyInputStream to the specified position. This involves 2 steps:
    *    1. Updating the blockIndex to the blockStream corresponding to the
    *    seeked position.
    *    2. Seeking the corresponding blockStream to the adjusted position.
    *
    * For example, let’s say the block size is 200 bytes and block[0] stores
    * data from indices 0 - 199, block[1] from indices 200 - 399 and so on.
    * Let’s say we seek to position 240. In the first step, the blockIndex
    * would be updated to 1 as indices 200 - 399 reside in blockStream[1]. In
    * the second step, the blockStream[1] would be seeked to position 40 (=
    * 240 - blockOffset[1] (= 200)).
    */
   @Override
   public synchronized void seek(long pos) throws IOException {
     checkOpen();
     if (pos < 0 || pos >= length) {
       if (pos == 0) {
         // It is possible for length and pos to be zero in which case
         // seek should return instead of throwing exception
         return;
       }
       throw new EOFException(
           "EOF encountered at pos: " + pos + " for key: " + key);
     }

     // 1. Update the blockIndex
     if (blockIndex >= blockStreams.size()) {
       blockIndex = Arrays.binarySearch(blockOffsets, pos);
     } else if (pos < blockOffsets[blockIndex]) {
       blockIndex =
           Arrays.binarySearch(blockOffsets, 0, blockIndex, pos);
     } else if (pos >= blockOffsets[blockIndex] + blockStreams
         .get(blockIndex).getLength()) {
       blockIndex = Arrays
           .binarySearch(blockOffsets, blockIndex + 1,
               blockStreams.size(), pos);
     }
     if (blockIndex < 0) {
       // Binary search returns -insertionPoint - 1  if element is not present
       // in the array. insertionPoint is the point at which element would be
       // inserted in the sorted array. We need to adjust the blockIndex
       // accordingly so that blockIndex = insertionPoint - 1
       blockIndex = -blockIndex - 2;
     }

     // Reset the previous blockStream's position
     blockStreams.get(blockIndexOfPrevPosition).resetPosition();

     // 2. Seek the blockStream to the adjusted position
     blockStreams.get(blockIndex).seek(pos - blockOffsets[blockIndex]);
     blockIndexOfPrevPosition = blockIndex;
   }

   @Override
   public synchronized long getPos() throws IOException {
     return length == 0 ? 0 : blockOffsets[blockIndex] +
         blockStreams.get(blockIndex).getPos();
   }

   @Override
   public boolean seekToNewSource(long targetPos) throws IOException {
     return false;
   }

   @Override
   public int available() throws IOException {
     checkOpen();
     long remaining = length - getPos();
     return remaining <= Integer.MAX_VALUE ? (int) remaining : Integer.MAX_VALUE;
   }

   @Override
   public void close() throws IOException {
     closed = true;
     for (BlockInputStream blockStream : blockStreams) {
       blockStream.close();
     }
   }

   /**
    * Verify that the input 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 checkOpen() throws IOException {
     if (closed) {
       throw new IOException(
           ": " + FSExceptionMessages.STREAM_IS_CLOSED + " Key: " + key);
     }
   }

   @VisibleForTesting
   public synchronized int getCurrentStreamIndex() {
     return blockIndex;
   }

   @VisibleForTesting
   public long getRemainingOfIndex(int index) throws IOException {
     return blockStreams.get(index).getRemaining();
   }
 }
	/*
	* 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 org.apache.hadoop.fs.FSExceptionMessages;
	import org.apache.hadoop.fs.Seekable;
	import org.apache.hadoop.hdds.scm.XceiverClientManager;
	import org.apache.hadoop.hdds.scm.storage.BlockInputStream;
	import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
	import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.EOFException;
	import java.io.IOException;
	import java.io.InputStream;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.List;

	/**
	* Maintaining a list of BlockInputStream. Read based on offset.
	*/
	public class KeyInputStream extends InputStream implements Seekable {

	private static final Logger LOG =
	LoggerFactory.getLogger(KeyInputStream.class);

	private static final int EOF = -1;

	private String key;
	private long length = 0;
	private boolean closed = false;

	// List of BlockInputStreams, one for each block in the key
	private final List<BlockInputStream> blockStreams;

	// blockOffsets[i] stores the index of the first data byte in
	// blockStream w.r.t the key data.
	// For example, let’s say the block size is 200 bytes and block[0] stores
	// data from indices 0 - 199, block[1] from indices 200 - 399 and so on.
	// Then, blockOffset[0] = 0 (the offset of the first byte of data in
	// block[0]), blockOffset[1] = 200 and so on.
	private long[] blockOffsets = null;

	// Index of the blockStream corresponding to the current position of the
	// KeyInputStream i.e. offset of the data to be read next
	private int blockIndex;

	// Tracks the blockIndex corresponding to the last seeked position so that it
	// can be reset if a new position is seeked.
	private int blockIndexOfPrevPosition;

	public KeyInputStream() {
	blockStreams = new ArrayList<>();
	blockIndex = 0;
	}

	/**
	* For each block in keyInfo, add a BlockInputStream to blockStreams.
	*/
	public static LengthInputStream getFromOmKeyInfo(OmKeyInfo keyInfo,
	XceiverClientManager xceiverClientManager,
	boolean verifyChecksum) {
	List<OmKeyLocationInfo> keyLocationInfos = keyInfo
	.getLatestVersionLocations().getBlocksLatestVersionOnly();

	KeyInputStream keyInputStream = new KeyInputStream();
	keyInputStream.initialize(keyInfo.getKeyName(), keyLocationInfos,
	xceiverClientManager, verifyChecksum);

	return new LengthInputStream(keyInputStream, keyInputStream.length);
	}

	private synchronized void initialize(String keyName,
	List<OmKeyLocationInfo> blockInfos,
	XceiverClientManager xceiverClientManager,
	boolean verifyChecksum) {
	this.key = keyName;
	this.blockOffsets = new long[blockInfos.size()];
	long keyLength = 0;
	for (int i = 0; i < blockInfos.size(); i++) {
	OmKeyLocationInfo omKeyLocationInfo = blockInfos.get(i);
	if (LOG.isDebugEnabled()) {
	LOG.debug("Adding stream for accessing {}. The stream will be " +
	"initialized later.", omKeyLocationInfo);
	}

	addStream(omKeyLocationInfo, xceiverClientManager,
	verifyChecksum);

	this.blockOffsets[i] = keyLength;
	keyLength += omKeyLocationInfo.getLength();
	}
	this.length = keyLength;
	}

	/**
	* Append another BlockInputStream to the end of the list. Note that the
	* BlockInputStream is only created here and not initialized. The
	* BlockInputStream is initialized when a read operation is performed on
	* the block for the first time.
	*/
	private synchronized void addStream(OmKeyLocationInfo blockInfo,
	XceiverClientManager xceiverClientMngr,
	boolean verifyChecksum) {
	blockStreams.add(new BlockInputStream(blockInfo.getBlockID(),
	blockInfo.getLength(), blockInfo.getPipeline(), blockInfo.getToken(),
	verifyChecksum, xceiverClientMngr));
	}

	@VisibleForTesting
	public void addStream(BlockInputStream blockInputStream) {
	blockStreams.add(blockInputStream);
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public synchronized int read() throws IOException {
	byte[] buf = new byte[1];
	if (read(buf, 0, 1) == EOF) {
	return EOF;
	}
	return Byte.toUnsignedInt(buf[0]);
	}

	/**
	* {@inheritDoc}
	*/
	@Override
	public synchronized int read(byte[] b, int off, int len) throws IOException {
	checkOpen();
	if (b == null) {
	throw new NullPointerException();
	}
	if (off < 0 \|\| len < 0 \|\| len > b.length - off) {
	throw new IndexOutOfBoundsException();
	}
	if (len == 0) {
	return 0;
	}
	int totalReadLen = 0;
	while (len > 0) {
	// if we are at the last block and have read the entire block, return
	if (blockStreams.size() == 0 \|\|
	(blockStreams.size() - 1 <= blockIndex &&
	blockStreams.get(blockIndex)
	.getRemaining() == 0)) {
	return totalReadLen == 0 ? EOF : totalReadLen;
	}

	// Get the current blockStream and read data from it
	BlockInputStream current = blockStreams.get(blockIndex);
	int numBytesToRead = Math.min(len, (int)current.getRemaining());
	int numBytesRead = current.read(b, off, numBytesToRead);
	if (numBytesRead != numBytesToRead) {
	// This implies that there is either data loss or corruption in the
	// chunk entries. Even EOF in the current stream would be covered in
	// this case.
	throw new IOException(String.format(
	"Inconsistent read for blockID=%s length=%d numBytesRead=%d",
	current.getBlockID(), current.getLength(), numBytesRead));
	}
	totalReadLen += numBytesRead;
	off += numBytesRead;
	len -= numBytesRead;
	if (current.getRemaining() <= 0 &&
	((blockIndex + 1) < blockStreams.size())) {
	blockIndex += 1;
	}
	}
	return totalReadLen;
	}

	/**
	* Seeks the KeyInputStream to the specified position. This involves 2 steps:
	* 1. Updating the blockIndex to the blockStream corresponding to the
	* seeked position.
	* 2. Seeking the corresponding blockStream to the adjusted position.
	*
	* For example, let’s say the block size is 200 bytes and block[0] stores
	* data from indices 0 - 199, block[1] from indices 200 - 399 and so on.
	* Let’s say we seek to position 240. In the first step, the blockIndex
	* would be updated to 1 as indices 200 - 399 reside in blockStream[1]. In
	* the second step, the blockStream[1] would be seeked to position 40 (=
	* 240 - blockOffset[1] (= 200)).
	*/
	@Override
	public synchronized void seek(long pos) throws IOException {
	checkOpen();
	if (pos < 0 \|\| pos >= length) {
	if (pos == 0) {
	// It is possible for length and pos to be zero in which case
	// seek should return instead of throwing exception
	return;
	}
	throw new EOFException(
	"EOF encountered at pos: " + pos + " for key: " + key);
	}

	// 1. Update the blockIndex
	if (blockIndex >= blockStreams.size()) {
	blockIndex = Arrays.binarySearch(blockOffsets, pos);
	} else if (pos < blockOffsets[blockIndex]) {
	blockIndex =
	Arrays.binarySearch(blockOffsets, 0, blockIndex, pos);
	} else if (pos >= blockOffsets[blockIndex] + blockStreams
	.get(blockIndex).getLength()) {
	blockIndex = Arrays
	.binarySearch(blockOffsets, blockIndex + 1,
	blockStreams.size(), pos);
	}
	if (blockIndex < 0) {
	// Binary search returns -insertionPoint - 1 if element is not present
	// in the array. insertionPoint is the point at which element would be
	// inserted in the sorted array. We need to adjust the blockIndex
	// accordingly so that blockIndex = insertionPoint - 1
	blockIndex = -blockIndex - 2;
	}

	// Reset the previous blockStream's position
	blockStreams.get(blockIndexOfPrevPosition).resetPosition();

	// 2. Seek the blockStream to the adjusted position
	blockStreams.get(blockIndex).seek(pos - blockOffsets[blockIndex]);
	blockIndexOfPrevPosition = blockIndex;
	}

	@Override
	public synchronized long getPos() throws IOException {
	return length == 0 ? 0 : blockOffsets[blockIndex] +
	blockStreams.get(blockIndex).getPos();
	}

	@Override
	public boolean seekToNewSource(long targetPos) throws IOException {
	return false;
	}

	@Override
	public int available() throws IOException {
	checkOpen();
	long remaining = length - getPos();
	return remaining <= Integer.MAX_VALUE ? (int) remaining : Integer.MAX_VALUE;
	}

	@Override
	public void close() throws IOException {
	closed = true;
	for (BlockInputStream blockStream : blockStreams) {
	blockStream.close();
	}
	}

	/**
	* Verify that the input 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 checkOpen() throws IOException {
	if (closed) {
	throw new IOException(
	": " + FSExceptionMessages.STREAM_IS_CLOSED + " Key: " + key);
	}
	}

	@VisibleForTesting
	public synchronized int getCurrentStreamIndex() {
	return blockIndex;
	}

	@VisibleForTesting
	public long getRemainingOfIndex(int index) throws IOException {
	return blockStreams.get(index).getRemaining();
	}
	}