You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by wh...@apache.org on 2015/08/22 22:30:51 UTC
[1/4] hadoop git commit: HDFS-8934. Move ShortCircuitShm to
hdfs-client. Contributed by Mingliang Liu.
Repository: hadoop
Updated Branches:
refs/heads/branch-2 8c205267c -> 8e4afa3a6
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java
deleted file mode 100644
index 17365fb..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java
+++ /dev/null
@@ -1,125 +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.hdfs.util;
-
-import java.io.EOFException;
-import java.io.FilterInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-
-import com.google.common.base.Preconditions;
-
-/**
- * An InputStream implementations which reads from some other InputStream
- * but expects an exact number of bytes. Any attempts to read past the
- * specified number of bytes will return as if the end of the stream
- * was reached. If the end of the underlying stream is reached prior to
- * the specified number of bytes, an EOFException is thrown.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class ExactSizeInputStream extends FilterInputStream {
- private int remaining;
-
- /**
- * Construct an input stream that will read no more than
- * 'numBytes' bytes.
- *
- * If an EOF occurs on the underlying stream before numBytes
- * bytes have been read, an EOFException will be thrown.
- *
- * @param in the inputstream to wrap
- * @param numBytes the number of bytes to read
- */
- public ExactSizeInputStream(InputStream in, int numBytes) {
- super(in);
- Preconditions.checkArgument(numBytes >= 0,
- "Negative expected bytes: ", numBytes);
- this.remaining = numBytes;
- }
-
- @Override
- public int available() throws IOException {
- return Math.min(super.available(), remaining);
- }
-
- @Override
- public int read() throws IOException {
- // EOF if we reached our limit
- if (remaining <= 0) {
- return -1;
- }
- final int result = super.read();
- if (result >= 0) {
- --remaining;
- } else if (remaining > 0) {
- // Underlying stream reached EOF but we haven't read the expected
- // number of bytes.
- throw new EOFException(
- "Premature EOF. Expected " + remaining + "more bytes");
- }
- return result;
- }
-
- @Override
- public int read(final byte[] b, final int off, int len)
- throws IOException {
- if (remaining <= 0) {
- return -1;
- }
- len = Math.min(len, remaining);
- final int result = super.read(b, off, len);
- if (result >= 0) {
- remaining -= result;
- } else if (remaining > 0) {
- // Underlying stream reached EOF but we haven't read the expected
- // number of bytes.
- throw new EOFException(
- "Premature EOF. Expected " + remaining + "more bytes");
- }
- return result;
- }
-
- @Override
- public long skip(final long n) throws IOException {
- final long result = super.skip(Math.min(n, remaining));
- if (result > 0) {
- remaining -= result;
- } else if (remaining > 0) {
- // Underlying stream reached EOF but we haven't read the expected
- // number of bytes.
- throw new EOFException(
- "Premature EOF. Expected " + remaining + "more bytes");
- }
- return result;
- }
-
- @Override
- public boolean markSupported() {
- return false;
- }
-
- @Override
- public void mark(int readlimit) {
- throw new UnsupportedOperationException();
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocolPB/TestPBHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocolPB/TestPBHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocolPB/TestPBHelper.java
index c7233bd..f25fb1b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocolPB/TestPBHelper.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/protocolPB/TestPBHelper.java
@@ -153,7 +153,7 @@ public class TestPBHelper {
@Test
public void testConvertDatanodeID() {
DatanodeID dn = DFSTestUtil.getLocalDatanodeID();
- DatanodeIDProto dnProto = PBHelper.convert(dn);
+ DatanodeIDProto dnProto = PBHelperClient.convert(dn);
DatanodeID dn2 = PBHelper.convert(dnProto);
compare(dn, dn2);
}
@@ -332,12 +332,12 @@ public class TestPBHelper {
@Test
public void testConvertExtendedBlock() {
ExtendedBlock b = getExtendedBlock();
- ExtendedBlockProto bProto = PBHelper.convert(b);
+ ExtendedBlockProto bProto = PBHelperClient.convert(b);
ExtendedBlock b1 = PBHelper.convert(bProto);
assertEquals(b, b1);
b.setBlockId(-1);
- bProto = PBHelper.convert(b);
+ bProto = PBHelperClient.convert(b);
b1 = PBHelper.convert(bProto);
assertEquals(b, b1);
}
@@ -398,7 +398,7 @@ public class TestPBHelper {
Token<BlockTokenIdentifier> token = new Token<BlockTokenIdentifier>(
"identifier".getBytes(), "password".getBytes(), new Text("kind"),
new Text("service"));
- TokenProto tokenProto = PBHelper.convert(token);
+ TokenProto tokenProto = PBHelperClient.convert(token);
Token<BlockTokenIdentifier> token2 = PBHelper.convert(tokenProto);
compare(token, token2);
}
@@ -592,16 +592,16 @@ public class TestPBHelper {
@Test
public void testChecksumTypeProto() {
assertEquals(DataChecksum.Type.NULL,
- PBHelper.convert(HdfsProtos.ChecksumTypeProto.CHECKSUM_NULL));
+ PBHelperClient.convert(HdfsProtos.ChecksumTypeProto.CHECKSUM_NULL));
assertEquals(DataChecksum.Type.CRC32,
- PBHelper.convert(HdfsProtos.ChecksumTypeProto.CHECKSUM_CRC32));
+ PBHelperClient.convert(HdfsProtos.ChecksumTypeProto.CHECKSUM_CRC32));
assertEquals(DataChecksum.Type.CRC32C,
- PBHelper.convert(HdfsProtos.ChecksumTypeProto.CHECKSUM_CRC32C));
- assertEquals(PBHelper.convert(DataChecksum.Type.NULL),
+ PBHelperClient.convert(HdfsProtos.ChecksumTypeProto.CHECKSUM_CRC32C));
+ assertEquals(PBHelperClient.convert(DataChecksum.Type.NULL),
HdfsProtos.ChecksumTypeProto.CHECKSUM_NULL);
- assertEquals(PBHelper.convert(DataChecksum.Type.CRC32),
+ assertEquals(PBHelperClient.convert(DataChecksum.Type.CRC32),
HdfsProtos.ChecksumTypeProto.CHECKSUM_CRC32);
- assertEquals(PBHelper.convert(DataChecksum.Type.CRC32C),
+ assertEquals(PBHelperClient.convert(DataChecksum.Type.CRC32C),
HdfsProtos.ChecksumTypeProto.CHECKSUM_CRC32C);
}
[3/4] hadoop git commit: HDFS-8934. Move ShortCircuitShm to
hdfs-client. Contributed by Mingliang Liu.
Posted by wh...@apache.org.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
new file mode 100644
index 0000000..78325a3
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
@@ -0,0 +1,647 @@
+/**
+ * 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.hdfs.shortcircuit;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.BitSet;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.Random;
+
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.hadoop.fs.InvalidRequestException;
+import org.apache.hadoop.hdfs.ExtendedBlockId;
+import org.apache.hadoop.io.nativeio.NativeIO;
+import org.apache.hadoop.io.nativeio.NativeIO.POSIX;
+import org.apache.hadoop.util.Shell;
+import org.apache.hadoop.util.StringUtils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import sun.misc.Unsafe;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ComparisonChain;
+import com.google.common.primitives.Ints;
+
+/**
+ * A shared memory segment used to implement short-circuit reads.
+ */
+public class ShortCircuitShm {
+ private static final Logger LOG = LoggerFactory.getLogger(ShortCircuitShm.class);
+
+ protected static final int BYTES_PER_SLOT = 64;
+
+ private static final Unsafe unsafe = safetyDance();
+
+ private static Unsafe safetyDance() {
+ try {
+ Field f = Unsafe.class.getDeclaredField("theUnsafe");
+ f.setAccessible(true);
+ return (Unsafe)f.get(null);
+ } catch (Throwable e) {
+ LOG.error("failed to load misc.Unsafe", e);
+ }
+ return null;
+ }
+
+ /**
+ * Calculate the usable size of a shared memory segment.
+ * We round down to a multiple of the slot size and do some validation.
+ *
+ * @param stream The stream we're using.
+ * @return The usable size of the shared memory segment.
+ */
+ private static int getUsableLength(FileInputStream stream)
+ throws IOException {
+ int intSize = Ints.checkedCast(stream.getChannel().size());
+ int slots = intSize / BYTES_PER_SLOT;
+ if (slots == 0) {
+ throw new IOException("size of shared memory segment was " +
+ intSize + ", but that is not enough to hold even one slot.");
+ }
+ return slots * BYTES_PER_SLOT;
+ }
+
+ /**
+ * Identifies a DfsClientShm.
+ */
+ public static class ShmId implements Comparable<ShmId> {
+ private static final Random random = new Random();
+ private final long hi;
+ private final long lo;
+
+ /**
+ * Generate a random ShmId.
+ *
+ * We generate ShmIds randomly to prevent a malicious client from
+ * successfully guessing one and using that to interfere with another
+ * client.
+ */
+ public static ShmId createRandom() {
+ return new ShmId(random.nextLong(), random.nextLong());
+ }
+
+ public ShmId(long hi, long lo) {
+ this.hi = hi;
+ this.lo = lo;
+ }
+
+ public long getHi() {
+ return hi;
+ }
+
+ public long getLo() {
+ return lo;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if ((o == null) || (o.getClass() != this.getClass())) {
+ return false;
+ }
+ ShmId other = (ShmId)o;
+ return new EqualsBuilder().
+ append(hi, other.hi).
+ append(lo, other.lo).
+ isEquals();
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder().
+ append(this.hi).
+ append(this.lo).
+ toHashCode();
+ }
+
+ @Override
+ public String toString() {
+ return String.format("%016x%016x", hi, lo);
+ }
+
+ @Override
+ public int compareTo(ShmId other) {
+ return ComparisonChain.start().
+ compare(hi, other.hi).
+ compare(lo, other.lo).
+ result();
+ }
+ };
+
+ /**
+ * Uniquely identifies a slot.
+ */
+ public static class SlotId {
+ private final ShmId shmId;
+ private final int slotIdx;
+
+ public SlotId(ShmId shmId, int slotIdx) {
+ this.shmId = shmId;
+ this.slotIdx = slotIdx;
+ }
+
+ public ShmId getShmId() {
+ return shmId;
+ }
+
+ public int getSlotIdx() {
+ return slotIdx;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if ((o == null) || (o.getClass() != this.getClass())) {
+ return false;
+ }
+ SlotId other = (SlotId)o;
+ return new EqualsBuilder().
+ append(shmId, other.shmId).
+ append(slotIdx, other.slotIdx).
+ isEquals();
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder().
+ append(this.shmId).
+ append(this.slotIdx).
+ toHashCode();
+ }
+
+ @Override
+ public String toString() {
+ return String.format("SlotId(%s:%d)", shmId.toString(), slotIdx);
+ }
+ }
+
+ public class SlotIterator implements Iterator<Slot> {
+ int slotIdx = -1;
+
+ @Override
+ public boolean hasNext() {
+ synchronized (ShortCircuitShm.this) {
+ return allocatedSlots.nextSetBit(slotIdx + 1) != -1;
+ }
+ }
+
+ @Override
+ public Slot next() {
+ synchronized (ShortCircuitShm.this) {
+ int nextSlotIdx = allocatedSlots.nextSetBit(slotIdx + 1);
+ if (nextSlotIdx == -1) {
+ throw new NoSuchElementException();
+ }
+ slotIdx = nextSlotIdx;
+ return slots[nextSlotIdx];
+ }
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException("SlotIterator " +
+ "doesn't support removal");
+ }
+ }
+
+ /**
+ * A slot containing information about a replica.
+ *
+ * The format is:
+ * word 0
+ * bit 0:32 Slot flags (see below).
+ * bit 33:63 Anchor count.
+ * word 1:7
+ * Reserved for future use, such as statistics.
+ * Padding is also useful for avoiding false sharing.
+ *
+ * Little-endian versus big-endian is not relevant here since both the client
+ * and the server reside on the same computer and use the same orientation.
+ */
+ public class Slot {
+ /**
+ * Flag indicating that the slot is valid.
+ *
+ * The DFSClient sets this flag when it allocates a new slot within one of
+ * its shared memory regions.
+ *
+ * The DataNode clears this flag when the replica associated with this slot
+ * is no longer valid. The client itself also clears this flag when it
+ * believes that the DataNode is no longer using this slot to communicate.
+ */
+ private static final long VALID_FLAG = 1L<<63;
+
+ /**
+ * Flag indicating that the slot can be anchored.
+ */
+ private static final long ANCHORABLE_FLAG = 1L<<62;
+
+ /**
+ * The slot address in memory.
+ */
+ private final long slotAddress;
+
+ /**
+ * BlockId of the block this slot is used for.
+ */
+ private final ExtendedBlockId blockId;
+
+ Slot(long slotAddress, ExtendedBlockId blockId) {
+ this.slotAddress = slotAddress;
+ this.blockId = blockId;
+ }
+
+ /**
+ * Get the short-circuit memory segment associated with this Slot.
+ *
+ * @return The enclosing short-circuit memory segment.
+ */
+ public ShortCircuitShm getShm() {
+ return ShortCircuitShm.this;
+ }
+
+ /**
+ * Get the ExtendedBlockId associated with this slot.
+ *
+ * @return The ExtendedBlockId of this slot.
+ */
+ public ExtendedBlockId getBlockId() {
+ return blockId;
+ }
+
+ /**
+ * Get the SlotId of this slot, containing both shmId and slotIdx.
+ *
+ * @return The SlotId of this slot.
+ */
+ public SlotId getSlotId() {
+ return new SlotId(getShmId(), getSlotIdx());
+ }
+
+ /**
+ * Get the Slot index.
+ *
+ * @return The index of this slot.
+ */
+ public int getSlotIdx() {
+ return Ints.checkedCast(
+ (slotAddress - baseAddress) / BYTES_PER_SLOT);
+ }
+
+ /**
+ * Clear the slot.
+ */
+ void clear() {
+ unsafe.putLongVolatile(null, this.slotAddress, 0);
+ }
+
+ private boolean isSet(long flag) {
+ long prev = unsafe.getLongVolatile(null, this.slotAddress);
+ return (prev & flag) != 0;
+ }
+
+ private void setFlag(long flag) {
+ long prev;
+ do {
+ prev = unsafe.getLongVolatile(null, this.slotAddress);
+ if ((prev & flag) != 0) {
+ return;
+ }
+ } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
+ prev, prev | flag));
+ }
+
+ private void clearFlag(long flag) {
+ long prev;
+ do {
+ prev = unsafe.getLongVolatile(null, this.slotAddress);
+ if ((prev & flag) == 0) {
+ return;
+ }
+ } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
+ prev, prev & (~flag)));
+ }
+
+ public boolean isValid() {
+ return isSet(VALID_FLAG);
+ }
+
+ public void makeValid() {
+ setFlag(VALID_FLAG);
+ }
+
+ public void makeInvalid() {
+ clearFlag(VALID_FLAG);
+ }
+
+ public boolean isAnchorable() {
+ return isSet(ANCHORABLE_FLAG);
+ }
+
+ public void makeAnchorable() {
+ setFlag(ANCHORABLE_FLAG);
+ }
+
+ public void makeUnanchorable() {
+ clearFlag(ANCHORABLE_FLAG);
+ }
+
+ public boolean isAnchored() {
+ long prev = unsafe.getLongVolatile(null, this.slotAddress);
+ if ((prev & VALID_FLAG) == 0) {
+ // Slot is no longer valid.
+ return false;
+ }
+ return ((prev & 0x7fffffff) != 0);
+ }
+
+ /**
+ * Try to add an anchor for a given slot.
+ *
+ * When a slot is anchored, we know that the block it refers to is resident
+ * in memory.
+ *
+ * @return True if the slot is anchored.
+ */
+ public boolean addAnchor() {
+ long prev;
+ do {
+ prev = unsafe.getLongVolatile(null, this.slotAddress);
+ if ((prev & VALID_FLAG) == 0) {
+ // Slot is no longer valid.
+ return false;
+ }
+ if ((prev & ANCHORABLE_FLAG) == 0) {
+ // Slot can't be anchored right now.
+ return false;
+ }
+ if ((prev & 0x7fffffff) == 0x7fffffff) {
+ // Too many other threads have anchored the slot (2 billion?)
+ return false;
+ }
+ } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
+ prev, prev + 1));
+ return true;
+ }
+
+ /**
+ * Remove an anchor for a given slot.
+ */
+ public void removeAnchor() {
+ long prev;
+ do {
+ prev = unsafe.getLongVolatile(null, this.slotAddress);
+ Preconditions.checkState((prev & 0x7fffffff) != 0,
+ "Tried to remove anchor for slot " + slotAddress +", which was " +
+ "not anchored.");
+ } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
+ prev, prev - 1));
+ }
+
+ @Override
+ public String toString() {
+ return "Slot(slotIdx=" + getSlotIdx() + ", shm=" + getShm() + ")";
+ }
+ }
+
+ /**
+ * ID for this SharedMemorySegment.
+ */
+ private final ShmId shmId;
+
+ /**
+ * The base address of the memory-mapped file.
+ */
+ private final long baseAddress;
+
+ /**
+ * The mmapped length of the shared memory segment
+ */
+ private final int mmappedLength;
+
+ /**
+ * The slots associated with this shared memory segment.
+ * slot[i] contains the slot at offset i * BYTES_PER_SLOT,
+ * or null if that slot is not allocated.
+ */
+ private final Slot slots[];
+
+ /**
+ * A bitset where each bit represents a slot which is in use.
+ */
+ private final BitSet allocatedSlots;
+
+ /**
+ * Create the ShortCircuitShm.
+ *
+ * @param shmId The ID to use.
+ * @param stream The stream that we're going to use to create this
+ * shared memory segment.
+ *
+ * Although this is a FileInputStream, we are going to
+ * assume that the underlying file descriptor is writable
+ * as well as readable. It would be more appropriate to use
+ * a RandomAccessFile here, but that class does not have
+ * any public accessor which returns a FileDescriptor,
+ * unlike FileInputStream.
+ */
+ public ShortCircuitShm(ShmId shmId, FileInputStream stream)
+ throws IOException {
+ if (!NativeIO.isAvailable()) {
+ throw new UnsupportedOperationException("NativeIO is not available.");
+ }
+ if (Shell.WINDOWS) {
+ throw new UnsupportedOperationException(
+ "DfsClientShm is not yet implemented for Windows.");
+ }
+ if (unsafe == null) {
+ throw new UnsupportedOperationException(
+ "can't use DfsClientShm because we failed to " +
+ "load misc.Unsafe.");
+ }
+ this.shmId = shmId;
+ this.mmappedLength = getUsableLength(stream);
+ this.baseAddress = POSIX.mmap(stream.getFD(),
+ POSIX.MMAP_PROT_READ | POSIX.MMAP_PROT_WRITE, true, mmappedLength);
+ this.slots = new Slot[mmappedLength / BYTES_PER_SLOT];
+ this.allocatedSlots = new BitSet(slots.length);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("creating " + this.getClass().getSimpleName() +
+ "(shmId=" + shmId +
+ ", mmappedLength=" + mmappedLength +
+ ", baseAddress=" + String.format("%x", baseAddress) +
+ ", slots.length=" + slots.length + ")");
+ }
+ }
+
+ public final ShmId getShmId() {
+ return shmId;
+ }
+
+ /**
+ * Determine if this shared memory object is empty.
+ *
+ * @return True if the shared memory object is empty.
+ */
+ synchronized final public boolean isEmpty() {
+ return allocatedSlots.nextSetBit(0) == -1;
+ }
+
+ /**
+ * Determine if this shared memory object is full.
+ *
+ * @return True if the shared memory object is full.
+ */
+ synchronized final public boolean isFull() {
+ return allocatedSlots.nextClearBit(0) >= slots.length;
+ }
+
+ /**
+ * Calculate the base address of a slot.
+ *
+ * @param slotIdx Index of the slot.
+ * @return The base address of the slot.
+ */
+ private final long calculateSlotAddress(int slotIdx) {
+ long offset = slotIdx;
+ offset *= BYTES_PER_SLOT;
+ return this.baseAddress + offset;
+ }
+
+ /**
+ * Allocate a new slot and register it.
+ *
+ * This function chooses an empty slot, initializes it, and then returns
+ * the relevant Slot object.
+ *
+ * @return The new slot.
+ */
+ synchronized public final Slot allocAndRegisterSlot(
+ ExtendedBlockId blockId) {
+ int idx = allocatedSlots.nextClearBit(0);
+ if (idx >= slots.length) {
+ throw new RuntimeException(this + ": no more slots are available.");
+ }
+ allocatedSlots.set(idx, true);
+ Slot slot = new Slot(calculateSlotAddress(idx), blockId);
+ slot.clear();
+ slot.makeValid();
+ slots[idx] = slot;
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": allocAndRegisterSlot " + idx + ": allocatedSlots=" + allocatedSlots +
+ StringUtils.getStackTrace(Thread.currentThread()));
+ }
+ return slot;
+ }
+
+ synchronized public final Slot getSlot(int slotIdx)
+ throws InvalidRequestException {
+ if (!allocatedSlots.get(slotIdx)) {
+ throw new InvalidRequestException(this + ": slot " + slotIdx +
+ " does not exist.");
+ }
+ return slots[slotIdx];
+ }
+
+ /**
+ * Register a slot.
+ *
+ * This function looks at a slot which has already been initialized (by
+ * another process), and registers it with us. Then, it returns the
+ * relevant Slot object.
+ *
+ * @return The slot.
+ *
+ * @throws InvalidRequestException
+ * If the slot index we're trying to allocate has not been
+ * initialized, or is already in use.
+ */
+ synchronized public final Slot registerSlot(int slotIdx,
+ ExtendedBlockId blockId) throws InvalidRequestException {
+ if (slotIdx < 0) {
+ throw new InvalidRequestException(this + ": invalid negative slot " +
+ "index " + slotIdx);
+ }
+ if (slotIdx >= slots.length) {
+ throw new InvalidRequestException(this + ": invalid slot " +
+ "index " + slotIdx);
+ }
+ if (allocatedSlots.get(slotIdx)) {
+ throw new InvalidRequestException(this + ": slot " + slotIdx +
+ " is already in use.");
+ }
+ Slot slot = new Slot(calculateSlotAddress(slotIdx), blockId);
+ if (!slot.isValid()) {
+ throw new InvalidRequestException(this + ": slot " + slotIdx +
+ " is not marked as valid.");
+ }
+ slots[slotIdx] = slot;
+ allocatedSlots.set(slotIdx, true);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": registerSlot " + slotIdx + ": allocatedSlots=" + allocatedSlots +
+ StringUtils.getStackTrace(Thread.currentThread()));
+ }
+ return slot;
+ }
+
+ /**
+ * Unregisters a slot.
+ *
+ * This doesn't alter the contents of the slot. It just means
+ *
+ * @param slotIdx Index of the slot to unregister.
+ */
+ synchronized public final void unregisterSlot(int slotIdx) {
+ Preconditions.checkState(allocatedSlots.get(slotIdx),
+ "tried to unregister slot " + slotIdx + ", which was not registered.");
+ allocatedSlots.set(slotIdx, false);
+ slots[slotIdx] = null;
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": unregisterSlot " + slotIdx);
+ }
+ }
+
+ /**
+ * Iterate over all allocated slots.
+ *
+ * Note that this method isn't safe if
+ *
+ * @return The slot iterator.
+ */
+ public SlotIterator slotIterator() {
+ return new SlotIterator();
+ }
+
+ public void free() {
+ try {
+ POSIX.munmap(baseAddress, mmappedLength);
+ } catch (IOException e) {
+ LOG.warn(this + ": failed to munmap", e);
+ }
+ LOG.trace(this + ": freed");
+ }
+
+ @Override
+ public String toString() {
+ return this.getClass().getSimpleName() + "(" + shmId + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java
new file mode 100644
index 0000000..17365fb
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/util/ExactSizeInputStream.java
@@ -0,0 +1,125 @@
+/**
+ * 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.hdfs.util;
+
+import java.io.EOFException;
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * An InputStream implementations which reads from some other InputStream
+ * but expects an exact number of bytes. Any attempts to read past the
+ * specified number of bytes will return as if the end of the stream
+ * was reached. If the end of the underlying stream is reached prior to
+ * the specified number of bytes, an EOFException is thrown.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class ExactSizeInputStream extends FilterInputStream {
+ private int remaining;
+
+ /**
+ * Construct an input stream that will read no more than
+ * 'numBytes' bytes.
+ *
+ * If an EOF occurs on the underlying stream before numBytes
+ * bytes have been read, an EOFException will be thrown.
+ *
+ * @param in the inputstream to wrap
+ * @param numBytes the number of bytes to read
+ */
+ public ExactSizeInputStream(InputStream in, int numBytes) {
+ super(in);
+ Preconditions.checkArgument(numBytes >= 0,
+ "Negative expected bytes: ", numBytes);
+ this.remaining = numBytes;
+ }
+
+ @Override
+ public int available() throws IOException {
+ return Math.min(super.available(), remaining);
+ }
+
+ @Override
+ public int read() throws IOException {
+ // EOF if we reached our limit
+ if (remaining <= 0) {
+ return -1;
+ }
+ final int result = super.read();
+ if (result >= 0) {
+ --remaining;
+ } else if (remaining > 0) {
+ // Underlying stream reached EOF but we haven't read the expected
+ // number of bytes.
+ throw new EOFException(
+ "Premature EOF. Expected " + remaining + "more bytes");
+ }
+ return result;
+ }
+
+ @Override
+ public int read(final byte[] b, final int off, int len)
+ throws IOException {
+ if (remaining <= 0) {
+ return -1;
+ }
+ len = Math.min(len, remaining);
+ final int result = super.read(b, off, len);
+ if (result >= 0) {
+ remaining -= result;
+ } else if (remaining > 0) {
+ // Underlying stream reached EOF but we haven't read the expected
+ // number of bytes.
+ throw new EOFException(
+ "Premature EOF. Expected " + remaining + "more bytes");
+ }
+ return result;
+ }
+
+ @Override
+ public long skip(final long n) throws IOException {
+ final long result = super.skip(Math.min(n, remaining));
+ if (result > 0) {
+ remaining -= result;
+ } else if (remaining > 0) {
+ // Underlying stream reached EOF but we haven't read the expected
+ // number of bytes.
+ throw new EOFException(
+ "Premature EOF. Expected " + remaining + "more bytes");
+ }
+ return result;
+ }
+
+ @Override
+ public boolean markSupported() {
+ return false;
+ }
+
+ @Override
+ public void mark(int readlimit) {
+ throw new UnsupportedOperationException();
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
index dc5bff1..3246919 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
@@ -482,6 +482,8 @@ Release 2.8.0 - UNRELEASED
HDFS-8823. Move replication factor into individual blocks. (wheat9)
+ HDFS-8934. Move ShortCircuitShm to hdfs-client. (Mingliang Liu via wheat9)
+
OPTIMIZATIONS
HDFS-8026. Trace FSOutputSummer#writeChecksumChunks rather than
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderFactory.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderFactory.java
index 8517173..fec6b85 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderFactory.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/BlockReaderFactory.java
@@ -45,7 +45,7 @@ import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.datatransfer.InvalidEncryptionKeyException;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
@@ -592,7 +592,7 @@ public class BlockReaderFactory implements ShortCircuitReplicaCreator {
failureInjector.getSupportsReceiptVerification());
DataInputStream in = new DataInputStream(peer.getInputStream());
BlockOpResponseProto resp = BlockOpResponseProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
DomainSocket sock = peer.getDomainSocket();
failureInjector.injectRequestFileDescriptorsFailure();
switch (resp.getStatus()) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
index a7b518e..12646b5 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSClient.java
@@ -149,7 +149,7 @@ import org.apache.hadoop.hdfs.protocol.datatransfer.sasl.DataTransferSaslUtil;
import org.apache.hadoop.hdfs.protocol.datatransfer.sasl.SaslDataTransferClient;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumResponseProto;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.security.token.block.DataEncryptionKey;
import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
@@ -1928,7 +1928,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
new Sender(out).blockChecksum(block, lb.getBlockToken());
final BlockOpResponseProto reply =
- BlockOpResponseProto.parseFrom(PBHelper.vintPrefixed(in));
+ BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(in));
String logInfo = "for block " + block + " from datanode " + datanodes[j];
DataTransferProtoUtil.checkBlockOpStatus(reply, logInfo);
@@ -1960,7 +1960,7 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
// read crc-type
final DataChecksum.Type ct;
if (checksumData.hasCrcType()) {
- ct = PBHelper.convert(checksumData
+ ct = PBHelperClient.convert(checksumData
.getCrcType());
} else {
LOG.debug("Retrieving checksum from an earlier-version DataNode: " +
@@ -2088,11 +2088,11 @@ public class DFSClient implements java.io.Closeable, RemotePeerFactory,
new Sender(out).readBlock(lb.getBlock(), lb.getBlockToken(), clientName,
0, 1, true, CachingStrategy.newDefaultStrategy());
final BlockOpResponseProto reply =
- BlockOpResponseProto.parseFrom(PBHelper.vintPrefixed(in));
+ BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(in));
String logInfo = "trying to read " + lb.getBlock() + " from datanode " + dn;
DataTransferProtoUtil.checkBlockOpStatus(reply, logInfo);
- return PBHelper.convert(reply.getReadOpChecksumInfo().getChecksum().getType());
+ return PBHelperClient.convert(reply.getReadOpChecksumInfo().getChecksum().getType());
} finally {
IOUtils.cleanup(null, pair.in, pair.out);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java
index 8dd85b7..a975312 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DataStreamer.java
@@ -67,7 +67,7 @@ import org.apache.hadoop.hdfs.protocol.datatransfer.PipelineAck;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockStoragePolicySuite;
import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
@@ -1245,7 +1245,7 @@ class DataStreamer extends Daemon {
//ack
BlockOpResponseProto response =
- BlockOpResponseProto.parseFrom(PBHelper.vintPrefixed(in));
+ BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(in));
if (SUCCESS != response.getStatus()) {
throw new IOException("Failed to add a datanode");
}
@@ -1524,7 +1524,7 @@ class DataStreamer extends Daemon {
// receive ack for connect
BlockOpResponseProto resp = BlockOpResponseProto.parseFrom(
- PBHelper.vintPrefixed(blockReplyStream));
+ PBHelperClient.vintPrefixed(blockReplyStream));
pipelineStatus = resp.getStatus();
firstBadLink = resp.getFirstBadLink();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java
deleted file mode 100644
index 7b9e8e3..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java
+++ /dev/null
@@ -1,82 +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.hdfs;
-
-import org.apache.commons.lang.builder.EqualsBuilder;
-import org.apache.commons.lang.builder.HashCodeBuilder;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-
-/**
- * An immutable key which identifies a block.
- */
-@InterfaceAudience.Private
-final public class ExtendedBlockId {
- /**
- * The block ID for this block.
- */
- private final long blockId;
-
- /**
- * The block pool ID for this block.
- */
- private final String bpId;
-
- public static ExtendedBlockId fromExtendedBlock(ExtendedBlock block) {
- return new ExtendedBlockId(block.getBlockId(), block.getBlockPoolId());
- }
-
- public ExtendedBlockId(long blockId, String bpId) {
- this.blockId = blockId;
- this.bpId = bpId;
- }
-
- public long getBlockId() {
- return this.blockId;
- }
-
- public String getBlockPoolId() {
- return this.bpId;
- }
-
- @Override
- public boolean equals(Object o) {
- if ((o == null) || (o.getClass() != this.getClass())) {
- return false;
- }
- ExtendedBlockId other = (ExtendedBlockId)o;
- return new EqualsBuilder().
- append(blockId, other.blockId).
- append(bpId, other.bpId).
- isEquals();
- }
-
- @Override
- public int hashCode() {
- return new HashCodeBuilder().
- append(this.blockId).
- append(this.bpId).
- toHashCode();
- }
-
- @Override
- public String toString() {
- return new StringBuilder().append(blockId).
- append("_").append(bpId).toString();
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader.java
index d70f419..05a9f2c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader.java
@@ -39,7 +39,7 @@ import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReadOpChecksumInfoProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
import org.apache.hadoop.hdfs.shortcircuit.ClientMmap;
@@ -414,7 +414,7 @@ public class RemoteBlockReader extends FSInputChecker implements BlockReader {
new BufferedInputStream(peer.getInputStream(), bufferSize));
BlockOpResponseProto status = BlockOpResponseProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
RemoteBlockReader2.checkSuccess(status, peer, block, file);
ReadOpChecksumInfoProto checksumInfo =
status.getReadOpChecksumInfo();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader2.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader2.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader2.java
index c368d65..4c23d36 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader2.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/RemoteBlockReader2.java
@@ -43,7 +43,7 @@ import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseP
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientReadStatusProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReadOpChecksumInfoProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
import org.apache.hadoop.hdfs.shortcircuit.ClientMmap;
@@ -417,7 +417,7 @@ public class RemoteBlockReader2 implements BlockReader {
DataInputStream in = new DataInputStream(peer.getInputStream());
BlockOpResponseProto status = BlockOpResponseProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
checkSuccess(status, peer, block, file);
ReadOpChecksumInfoProto checksumInfo =
status.getReadOpChecksumInfo();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java
deleted file mode 100644
index 4792b0e..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java
+++ /dev/null
@@ -1,132 +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.hdfs.net;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.channels.ReadableByteChannel;
-
-import org.apache.hadoop.net.unix.DomainSocket;
-import org.apache.hadoop.classification.InterfaceAudience;
-
-/**
- * Represents a peer that we communicate with by using blocking I/O
- * on a UNIX domain socket.
- */
-@InterfaceAudience.Private
-public class DomainPeer implements Peer {
- private final DomainSocket socket;
- private final OutputStream out;
- private final InputStream in;
- private final ReadableByteChannel channel;
-
- public DomainPeer(DomainSocket socket) {
- this.socket = socket;
- this.out = socket.getOutputStream();
- this.in = socket.getInputStream();
- this.channel = socket.getChannel();
- }
-
- @Override
- public ReadableByteChannel getInputStreamChannel() {
- return channel;
- }
-
- @Override
- public void setReadTimeout(int timeoutMs) throws IOException {
- socket.setAttribute(DomainSocket.RECEIVE_TIMEOUT, timeoutMs);
- }
-
- @Override
- public int getReceiveBufferSize() throws IOException {
- return socket.getAttribute(DomainSocket.RECEIVE_BUFFER_SIZE);
- }
-
- @Override
- public boolean getTcpNoDelay() throws IOException {
- /* No TCP, no TCP_NODELAY. */
- return false;
- }
-
- @Override
- public void setWriteTimeout(int timeoutMs) throws IOException {
- socket.setAttribute(DomainSocket.SEND_TIMEOUT, timeoutMs);
- }
-
- @Override
- public boolean isClosed() {
- return !socket.isOpen();
- }
-
- @Override
- public void close() throws IOException {
- socket.close();
- }
-
- @Override
- public String getRemoteAddressString() {
- return "unix:" + socket.getPath();
- }
-
- @Override
- public String getLocalAddressString() {
- return "<local>";
- }
-
- @Override
- public InputStream getInputStream() throws IOException {
- return in;
- }
-
- @Override
- public OutputStream getOutputStream() throws IOException {
- return out;
- }
-
- @Override
- public boolean isLocal() {
- /* UNIX domain sockets can only be used for local communication. */
- return true;
- }
-
- @Override
- public String toString() {
- return "DomainPeer(" + getRemoteAddressString() + ")";
- }
-
- @Override
- public DomainSocket getDomainSocket() {
- return socket;
- }
-
- @Override
- public boolean hasSecureChannel() {
- //
- // Communication over domain sockets is assumed to be secure, since it
- // doesn't pass over any network. We also carefully control the privileges
- // that can be used on the domain socket inode and its parent directories.
- // See #{java.org.apache.hadoop.net.unix.DomainSocket#validateSocketPathSecurity0}
- // for details.
- //
- // So unless you are running as root or the hdfs superuser, you cannot
- // launch a man-in-the-middle attach on UNIX domain socket traffic.
- //
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/Peer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/Peer.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/Peer.java
deleted file mode 100644
index 42cf287..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/net/Peer.java
+++ /dev/null
@@ -1,123 +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.hdfs.net;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.nio.channels.ReadableByteChannel;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.net.unix.DomainSocket;
-
-/**
- * Represents a connection to a peer.
- */
-@InterfaceAudience.Private
-public interface Peer extends Closeable {
- /**
- * @return The input stream channel associated with this
- * peer, or null if it has none.
- */
- public ReadableByteChannel getInputStreamChannel();
-
- /**
- * Set the read timeout on this peer.
- *
- * @param timeoutMs The timeout in milliseconds.
- */
- public void setReadTimeout(int timeoutMs) throws IOException;
-
- /**
- * @return The receive buffer size.
- */
- public int getReceiveBufferSize() throws IOException;
-
- /**
- * @return True if TCP_NODELAY is turned on.
- */
- public boolean getTcpNoDelay() throws IOException;
-
- /**
- * Set the write timeout on this peer.
- *
- * Note: this is not honored for BasicInetPeer.
- * See {@link BasicSocketPeer#setWriteTimeout} for details.
- *
- * @param timeoutMs The timeout in milliseconds.
- */
- public void setWriteTimeout(int timeoutMs) throws IOException;
-
- /**
- * @return true only if the peer is closed.
- */
- public boolean isClosed();
-
- /**
- * Close the peer.
- *
- * It's safe to re-close a Peer that is already closed.
- */
- public void close() throws IOException;
-
- /**
- * @return A string representing the remote end of our
- * connection to the peer.
- */
- public String getRemoteAddressString();
-
- /**
- * @return A string representing the local end of our
- * connection to the peer.
- */
- public String getLocalAddressString();
-
- /**
- * @return An InputStream associated with the Peer.
- * This InputStream will be valid until you close
- * this peer with Peer#close.
- */
- public InputStream getInputStream() throws IOException;
-
- /**
- * @return An OutputStream associated with the Peer.
- * This OutputStream will be valid until you close
- * this peer with Peer#close.
- */
- public OutputStream getOutputStream() throws IOException;
-
- /**
- * @return True if the peer resides on the same
- * computer as we.
- */
- public boolean isLocal();
-
- /**
- * @return The DomainSocket associated with the current
- * peer, or null if there is none.
- */
- public DomainSocket getDomainSocket();
-
- /**
- * Return true if the channel is secure.
- *
- * @return True if our channel to this peer is not
- * susceptible to man-in-the-middle attacks.
- */
- public boolean hasSecureChannel();
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java
deleted file mode 100644
index 5f86e52..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java
+++ /dev/null
@@ -1,62 +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.hdfs.protocol.datatransfer;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-
-/** Block Construction Stage */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public enum BlockConstructionStage {
- /** The enumerates are always listed as regular stage followed by the
- * recovery stage.
- * Changing this order will make getRecoveryStage not working.
- */
- // pipeline set up for block append
- PIPELINE_SETUP_APPEND,
- // pipeline set up for failed PIPELINE_SETUP_APPEND recovery
- PIPELINE_SETUP_APPEND_RECOVERY,
- // data streaming
- DATA_STREAMING,
- // pipeline setup for failed data streaming recovery
- PIPELINE_SETUP_STREAMING_RECOVERY,
- // close the block and pipeline
- PIPELINE_CLOSE,
- // Recover a failed PIPELINE_CLOSE
- PIPELINE_CLOSE_RECOVERY,
- // pipeline set up for block creation
- PIPELINE_SETUP_CREATE,
- // transfer RBW for adding datanodes
- TRANSFER_RBW,
- // transfer Finalized for adding datanodes
- TRANSFER_FINALIZED;
-
- final static private byte RECOVERY_BIT = (byte)1;
-
- /**
- * get the recovery stage of this stage
- */
- public BlockConstructionStage getRecoveryStage() {
- if (this == PIPELINE_SETUP_CREATE) {
- throw new IllegalArgumentException( "Unexpected blockStage " + this);
- } else {
- return values()[ordinal()|RECOVERY_BIT];
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java
deleted file mode 100644
index 284281a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java
+++ /dev/null
@@ -1,148 +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.hdfs.protocol.datatransfer;
-
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.hdfs.net.Peer;
-import org.apache.hadoop.hdfs.protocol.DatanodeID;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BaseHeaderProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ChecksumProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientOperationHeaderProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpWriteBlockProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.DataTransferTraceInfoProto;
-import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.ChecksumTypeProto;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
-import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
-import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.util.DataChecksum;
-import org.apache.htrace.Span;
-import org.apache.htrace.Trace;
-import org.apache.htrace.TraceInfo;
-import org.apache.htrace.TraceScope;
-
-/**
- * Static utilities for dealing with the protocol buffers used by the
- * Data Transfer Protocol.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public abstract class DataTransferProtoUtil {
- static BlockConstructionStage fromProto(
- OpWriteBlockProto.BlockConstructionStage stage) {
- return BlockConstructionStage.valueOf(stage.name());
- }
-
- static OpWriteBlockProto.BlockConstructionStage toProto(
- BlockConstructionStage stage) {
- return OpWriteBlockProto.BlockConstructionStage.valueOf(stage.name());
- }
-
- public static ChecksumProto toProto(DataChecksum checksum) {
- ChecksumTypeProto type = PBHelper.convert(checksum.getChecksumType());
- // ChecksumType#valueOf never returns null
- return ChecksumProto.newBuilder()
- .setBytesPerChecksum(checksum.getBytesPerChecksum())
- .setType(type)
- .build();
- }
-
- public static DataChecksum fromProto(ChecksumProto proto) {
- if (proto == null) return null;
-
- int bytesPerChecksum = proto.getBytesPerChecksum();
- DataChecksum.Type type = PBHelper.convert(proto.getType());
- return DataChecksum.newDataChecksum(type, bytesPerChecksum);
- }
-
- static ClientOperationHeaderProto buildClientHeader(ExtendedBlock blk,
- String client, Token<BlockTokenIdentifier> blockToken) {
- ClientOperationHeaderProto header =
- ClientOperationHeaderProto.newBuilder()
- .setBaseHeader(buildBaseHeader(blk, blockToken))
- .setClientName(client)
- .build();
- return header;
- }
-
- static BaseHeaderProto buildBaseHeader(ExtendedBlock blk,
- Token<BlockTokenIdentifier> blockToken) {
- BaseHeaderProto.Builder builder = BaseHeaderProto.newBuilder()
- .setBlock(PBHelper.convert(blk))
- .setToken(PBHelper.convert(blockToken));
- if (Trace.isTracing()) {
- Span s = Trace.currentSpan();
- builder.setTraceInfo(DataTransferTraceInfoProto.newBuilder()
- .setTraceId(s.getTraceId())
- .setParentId(s.getSpanId()));
- }
- return builder.build();
- }
-
- public static TraceInfo fromProto(DataTransferTraceInfoProto proto) {
- if (proto == null) return null;
- if (!proto.hasTraceId()) return null;
- return new TraceInfo(proto.getTraceId(), proto.getParentId());
- }
-
- public static TraceScope continueTraceSpan(ClientOperationHeaderProto header,
- String description) {
- return continueTraceSpan(header.getBaseHeader(), description);
- }
-
- public static TraceScope continueTraceSpan(BaseHeaderProto header,
- String description) {
- return continueTraceSpan(header.getTraceInfo(), description);
- }
-
- public static TraceScope continueTraceSpan(DataTransferTraceInfoProto proto,
- String description) {
- TraceScope scope = null;
- TraceInfo info = fromProto(proto);
- if (info != null) {
- scope = Trace.startSpan(description, info);
- }
- return scope;
- }
-
- public static void checkBlockOpStatus(
- BlockOpResponseProto response,
- String logInfo) throws IOException {
- if (response.getStatus() != Status.SUCCESS) {
- if (response.getStatus() == Status.ERROR_ACCESS_TOKEN) {
- throw new InvalidBlockTokenException(
- "Got access token error"
- + ", status message " + response.getMessage()
- + ", " + logInfo
- );
- } else {
- throw new IOException(
- "Got error"
- + ", status message " + response.getMessage()
- + ", " + logInfo
- );
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java
deleted file mode 100644
index 48e931d..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java
+++ /dev/null
@@ -1,201 +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.hdfs.protocol.datatransfer;
-
-import java.io.IOException;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
-import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
-import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.SlotId;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.util.DataChecksum;
-
-/**
- * Transfer data to/from datanode using a streaming protocol.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public interface DataTransferProtocol {
- public static final Log LOG = LogFactory.getLog(DataTransferProtocol.class);
-
- /** Version for data transfers between clients and datanodes
- * This should change when serialization of DatanodeInfo, not just
- * when protocol changes. It is not very obvious.
- */
- /*
- * Version 28:
- * Declare methods in DataTransferProtocol interface.
- */
- public static final int DATA_TRANSFER_VERSION = 28;
-
- /**
- * Read a block.
- *
- * @param blk the block being read.
- * @param blockToken security token for accessing the block.
- * @param clientName client's name.
- * @param blockOffset offset of the block.
- * @param length maximum number of bytes for this read.
- * @param sendChecksum if false, the DN should skip reading and sending
- * checksums
- * @param cachingStrategy The caching strategy to use.
- */
- public void readBlock(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken,
- final String clientName,
- final long blockOffset,
- final long length,
- final boolean sendChecksum,
- final CachingStrategy cachingStrategy) throws IOException;
-
- /**
- * Write a block to a datanode pipeline.
- * The receiver datanode of this call is the next datanode in the pipeline.
- * The other downstream datanodes are specified by the targets parameter.
- * Note that the receiver {@link DatanodeInfo} is not required in the
- * parameter list since the receiver datanode knows its info. However, the
- * {@link StorageType} for storing the replica in the receiver datanode is a
- * parameter since the receiver datanode may support multiple storage types.
- *
- * @param blk the block being written.
- * @param storageType for storing the replica in the receiver datanode.
- * @param blockToken security token for accessing the block.
- * @param clientName client's name.
- * @param targets other downstream datanodes in the pipeline.
- * @param targetStorageTypes target {@link StorageType}s corresponding
- * to the target datanodes.
- * @param source source datanode.
- * @param stage pipeline stage.
- * @param pipelineSize the size of the pipeline.
- * @param minBytesRcvd minimum number of bytes received.
- * @param maxBytesRcvd maximum number of bytes received.
- * @param latestGenerationStamp the latest generation stamp of the block.
- * @param pinning whether to pin the block, so Balancer won't move it.
- * @param targetPinnings whether to pin the block on target datanode
- */
- public void writeBlock(final ExtendedBlock blk,
- final StorageType storageType,
- final Token<BlockTokenIdentifier> blockToken,
- final String clientName,
- final DatanodeInfo[] targets,
- final StorageType[] targetStorageTypes,
- final DatanodeInfo source,
- final BlockConstructionStage stage,
- final int pipelineSize,
- final long minBytesRcvd,
- final long maxBytesRcvd,
- final long latestGenerationStamp,
- final DataChecksum requestedChecksum,
- final CachingStrategy cachingStrategy,
- final boolean allowLazyPersist,
- final boolean pinning,
- final boolean[] targetPinnings) throws IOException;
- /**
- * Transfer a block to another datanode.
- * The block stage must be
- * either {@link BlockConstructionStage#TRANSFER_RBW}
- * or {@link BlockConstructionStage#TRANSFER_FINALIZED}.
- *
- * @param blk the block being transferred.
- * @param blockToken security token for accessing the block.
- * @param clientName client's name.
- * @param targets target datanodes.
- */
- public void transferBlock(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken,
- final String clientName,
- final DatanodeInfo[] targets,
- final StorageType[] targetStorageTypes) throws IOException;
-
- /**
- * Request short circuit access file descriptors from a DataNode.
- *
- * @param blk The block to get file descriptors for.
- * @param blockToken Security token for accessing the block.
- * @param slotId The shared memory slot id to use, or null
- * to use no slot id.
- * @param maxVersion Maximum version of the block data the client
- * can understand.
- * @param supportsReceiptVerification True if the client supports
- * receipt verification.
- */
- public void requestShortCircuitFds(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken,
- SlotId slotId, int maxVersion, boolean supportsReceiptVerification)
- throws IOException;
-
- /**
- * Release a pair of short-circuit FDs requested earlier.
- *
- * @param slotId SlotID used by the earlier file descriptors.
- */
- public void releaseShortCircuitFds(final SlotId slotId) throws IOException;
-
- /**
- * Request a short circuit shared memory area from a DataNode.
- *
- * @param clientName The name of the client.
- */
- public void requestShortCircuitShm(String clientName) throws IOException;
-
- /**
- * Receive a block from a source datanode
- * and then notifies the namenode
- * to remove the copy from the original datanode.
- * Note that the source datanode and the original datanode can be different.
- * It is used for balancing purpose.
- *
- * @param blk the block being replaced.
- * @param storageType the {@link StorageType} for storing the block.
- * @param blockToken security token for accessing the block.
- * @param delHint the hint for deleting the block in the original datanode.
- * @param source the source datanode for receiving the block.
- */
- public void replaceBlock(final ExtendedBlock blk,
- final StorageType storageType,
- final Token<BlockTokenIdentifier> blockToken,
- final String delHint,
- final DatanodeInfo source) throws IOException;
-
- /**
- * Copy a block.
- * It is used for balancing purpose.
- *
- * @param blk the block being copied.
- * @param blockToken security token for accessing the block.
- */
- public void copyBlock(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken) throws IOException;
-
- /**
- * Get block checksum (MD5 of CRC32).
- *
- * @param blk a block.
- * @param blockToken security token for accessing the block.
- * @throws IOException
- */
- public void blockChecksum(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken) throws IOException;
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java
deleted file mode 100644
index 3077498..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java
+++ /dev/null
@@ -1,66 +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.hdfs.protocol.datatransfer;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-
-/** Operation */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public enum Op {
- WRITE_BLOCK((byte)80),
- READ_BLOCK((byte)81),
- READ_METADATA((byte)82),
- REPLACE_BLOCK((byte)83),
- COPY_BLOCK((byte)84),
- BLOCK_CHECKSUM((byte)85),
- TRANSFER_BLOCK((byte)86),
- REQUEST_SHORT_CIRCUIT_FDS((byte)87),
- RELEASE_SHORT_CIRCUIT_FDS((byte)88),
- REQUEST_SHORT_CIRCUIT_SHM((byte)89),
- CUSTOM((byte)127);
-
- /** The code for this operation. */
- public final byte code;
-
- private Op(byte code) {
- this.code = code;
- }
-
- private static final int FIRST_CODE = values()[0].code;
- /** Return the object represented by the code. */
- private static Op valueOf(byte code) {
- final int i = (code & 0xff) - FIRST_CODE;
- return i < 0 || i >= values().length? null: values()[i];
- }
-
- /** Read from in */
- public static Op read(DataInput in) throws IOException {
- return valueOf(in.readByte());
- }
-
- /** Write to out */
- public void write(DataOutput out) throws IOException {
- out.write(code);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/PipelineAck.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/PipelineAck.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/PipelineAck.java
index a811f39..44f38c6 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/PipelineAck.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/PipelineAck.java
@@ -17,7 +17,7 @@
*/
package org.apache.hadoop.hdfs.protocol.datatransfer;
-import static org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
+import static org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed;
import java.io.IOException;
import java.io.InputStream;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Receiver.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Receiver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Receiver.java
index d435543..694f521 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Receiver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Receiver.java
@@ -19,7 +19,7 @@ package org.apache.hadoop.hdfs.protocol.datatransfer;
import static org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtoUtil.fromProto;
import static org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtoUtil.continueTraceSpan;
-import static org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
+import static org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed;
import java.io.DataInputStream;
import java.io.IOException;
@@ -39,6 +39,7 @@ import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpWriteBlockProt
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessRequestProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmRequestProto;
import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.SlotId;
import org.apache.htrace.TraceScope;
@@ -136,7 +137,7 @@ public abstract class Receiver implements DataTransferProtocol {
proto.getClass().getSimpleName());
try {
writeBlock(PBHelper.convert(proto.getHeader().getBaseHeader().getBlock()),
- PBHelper.convertStorageType(proto.getStorageType()),
+ PBHelperClient.convertStorageType(proto.getStorageType()),
PBHelper.convert(proto.getHeader().getBaseHeader().getToken()),
proto.getHeader().getClientName(),
targets,
@@ -228,7 +229,7 @@ public abstract class Receiver implements DataTransferProtocol {
proto.getClass().getSimpleName());
try {
replaceBlock(PBHelper.convert(proto.getHeader().getBlock()),
- PBHelper.convertStorageType(proto.getStorageType()),
+ PBHelperClient.convertStorageType(proto.getStorageType()),
PBHelper.convert(proto.getHeader().getToken()),
proto.getDelHint(),
PBHelper.convert(proto.getSource()));
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java
deleted file mode 100644
index df69125..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java
+++ /dev/null
@@ -1,261 +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.hdfs.protocol.datatransfer;
-
-import static org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtoUtil.toProto;
-
-import java.io.DataOutput;
-import java.io.DataOutputStream;
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.fs.StorageType;
-import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
-import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.CachingStrategyProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ChecksumProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientOperationHeaderProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.DataTransferTraceInfoProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpCopyBlockProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpReadBlockProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpReplaceBlockProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpRequestShortCircuitAccessProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpTransferBlockProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpWriteBlockProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessRequestProto;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmRequestProto;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
-import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
-import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
-import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.SlotId;
-import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.util.DataChecksum;
-
-import org.apache.htrace.Trace;
-import org.apache.htrace.Span;
-
-import com.google.protobuf.Message;
-
-/** Sender */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class Sender implements DataTransferProtocol {
- private final DataOutputStream out;
-
- /** Create a sender for DataTransferProtocol with a output stream. */
- public Sender(final DataOutputStream out) {
- this.out = out;
- }
-
- /** Initialize a operation. */
- private static void op(final DataOutput out, final Op op
- ) throws IOException {
- out.writeShort(DataTransferProtocol.DATA_TRANSFER_VERSION);
- op.write(out);
- }
-
- private static void send(final DataOutputStream out, final Op opcode,
- final Message proto) throws IOException {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Sending DataTransferOp " + proto.getClass().getSimpleName()
- + ": " + proto);
- }
- op(out, opcode);
- proto.writeDelimitedTo(out);
- out.flush();
- }
-
- static private CachingStrategyProto getCachingStrategy(CachingStrategy cachingStrategy) {
- CachingStrategyProto.Builder builder = CachingStrategyProto.newBuilder();
- if (cachingStrategy.getReadahead() != null) {
- builder.setReadahead(cachingStrategy.getReadahead().longValue());
- }
- if (cachingStrategy.getDropBehind() != null) {
- builder.setDropBehind(cachingStrategy.getDropBehind().booleanValue());
- }
- return builder.build();
- }
-
- @Override
- public void readBlock(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken,
- final String clientName,
- final long blockOffset,
- final long length,
- final boolean sendChecksum,
- final CachingStrategy cachingStrategy) throws IOException {
-
- OpReadBlockProto proto = OpReadBlockProto.newBuilder()
- .setHeader(DataTransferProtoUtil.buildClientHeader(blk, clientName, blockToken))
- .setOffset(blockOffset)
- .setLen(length)
- .setSendChecksums(sendChecksum)
- .setCachingStrategy(getCachingStrategy(cachingStrategy))
- .build();
-
- send(out, Op.READ_BLOCK, proto);
- }
-
-
- @Override
- public void writeBlock(final ExtendedBlock blk,
- final StorageType storageType,
- final Token<BlockTokenIdentifier> blockToken,
- final String clientName,
- final DatanodeInfo[] targets,
- final StorageType[] targetStorageTypes,
- final DatanodeInfo source,
- final BlockConstructionStage stage,
- final int pipelineSize,
- final long minBytesRcvd,
- final long maxBytesRcvd,
- final long latestGenerationStamp,
- DataChecksum requestedChecksum,
- final CachingStrategy cachingStrategy,
- final boolean allowLazyPersist,
- final boolean pinning,
- final boolean[] targetPinnings) throws IOException {
- ClientOperationHeaderProto header = DataTransferProtoUtil.buildClientHeader(
- blk, clientName, blockToken);
-
- ChecksumProto checksumProto =
- DataTransferProtoUtil.toProto(requestedChecksum);
-
- OpWriteBlockProto.Builder proto = OpWriteBlockProto.newBuilder()
- .setHeader(header)
- .setStorageType(PBHelper.convertStorageType(storageType))
- .addAllTargets(PBHelper.convert(targets, 1))
- .addAllTargetStorageTypes(PBHelper.convertStorageTypes(targetStorageTypes, 1))
- .setStage(toProto(stage))
- .setPipelineSize(pipelineSize)
- .setMinBytesRcvd(minBytesRcvd)
- .setMaxBytesRcvd(maxBytesRcvd)
- .setLatestGenerationStamp(latestGenerationStamp)
- .setRequestedChecksum(checksumProto)
- .setCachingStrategy(getCachingStrategy(cachingStrategy))
- .setAllowLazyPersist(allowLazyPersist)
- .setPinning(pinning)
- .addAllTargetPinnings(PBHelper.convert(targetPinnings, 1));
-
- if (source != null) {
- proto.setSource(PBHelper.convertDatanodeInfo(source));
- }
-
- send(out, Op.WRITE_BLOCK, proto.build());
- }
-
- @Override
- public void transferBlock(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken,
- final String clientName,
- final DatanodeInfo[] targets,
- final StorageType[] targetStorageTypes) throws IOException {
-
- OpTransferBlockProto proto = OpTransferBlockProto.newBuilder()
- .setHeader(DataTransferProtoUtil.buildClientHeader(
- blk, clientName, blockToken))
- .addAllTargets(PBHelper.convert(targets))
- .addAllTargetStorageTypes(PBHelper.convertStorageTypes(targetStorageTypes))
- .build();
-
- send(out, Op.TRANSFER_BLOCK, proto);
- }
-
- @Override
- public void requestShortCircuitFds(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken,
- SlotId slotId, int maxVersion, boolean supportsReceiptVerification)
- throws IOException {
- OpRequestShortCircuitAccessProto.Builder builder =
- OpRequestShortCircuitAccessProto.newBuilder()
- .setHeader(DataTransferProtoUtil.buildBaseHeader(
- blk, blockToken)).setMaxVersion(maxVersion);
- if (slotId != null) {
- builder.setSlotId(PBHelper.convert(slotId));
- }
- builder.setSupportsReceiptVerification(supportsReceiptVerification);
- OpRequestShortCircuitAccessProto proto = builder.build();
- send(out, Op.REQUEST_SHORT_CIRCUIT_FDS, proto);
- }
-
- @Override
- public void releaseShortCircuitFds(SlotId slotId) throws IOException {
- ReleaseShortCircuitAccessRequestProto.Builder builder =
- ReleaseShortCircuitAccessRequestProto.newBuilder().
- setSlotId(PBHelper.convert(slotId));
- if (Trace.isTracing()) {
- Span s = Trace.currentSpan();
- builder.setTraceInfo(DataTransferTraceInfoProto.newBuilder()
- .setTraceId(s.getTraceId()).setParentId(s.getSpanId()));
- }
- ReleaseShortCircuitAccessRequestProto proto = builder.build();
- send(out, Op.RELEASE_SHORT_CIRCUIT_FDS, proto);
- }
-
- @Override
- public void requestShortCircuitShm(String clientName) throws IOException {
- ShortCircuitShmRequestProto.Builder builder =
- ShortCircuitShmRequestProto.newBuilder().
- setClientName(clientName);
- if (Trace.isTracing()) {
- Span s = Trace.currentSpan();
- builder.setTraceInfo(DataTransferTraceInfoProto.newBuilder()
- .setTraceId(s.getTraceId()).setParentId(s.getSpanId()));
- }
- ShortCircuitShmRequestProto proto = builder.build();
- send(out, Op.REQUEST_SHORT_CIRCUIT_SHM, proto);
- }
-
- @Override
- public void replaceBlock(final ExtendedBlock blk,
- final StorageType storageType,
- final Token<BlockTokenIdentifier> blockToken,
- final String delHint,
- final DatanodeInfo source) throws IOException {
- OpReplaceBlockProto proto = OpReplaceBlockProto.newBuilder()
- .setHeader(DataTransferProtoUtil.buildBaseHeader(blk, blockToken))
- .setStorageType(PBHelper.convertStorageType(storageType))
- .setDelHint(delHint)
- .setSource(PBHelper.convertDatanodeInfo(source))
- .build();
-
- send(out, Op.REPLACE_BLOCK, proto);
- }
-
- @Override
- public void copyBlock(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken) throws IOException {
- OpCopyBlockProto proto = OpCopyBlockProto.newBuilder()
- .setHeader(DataTransferProtoUtil.buildBaseHeader(blk, blockToken))
- .build();
-
- send(out, Op.COPY_BLOCK, proto);
- }
-
- @Override
- public void blockChecksum(final ExtendedBlock blk,
- final Token<BlockTokenIdentifier> blockToken) throws IOException {
- OpBlockChecksumProto proto = OpBlockChecksumProto.newBuilder()
- .setHeader(DataTransferProtoUtil.buildBaseHeader(blk, blockToken))
- .build();
-
- send(out, Op.BLOCK_CHECKSUM, proto);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java
index 398d44c..852819f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/sasl/DataTransferSaslUtil.java
@@ -24,7 +24,7 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_DATA_TRANSFER_SASL_PROPS_
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ENCRYPT_DATA_TRANSFER_CIPHER_KEY_BITLENGTH_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ENCRYPT_DATA_TRANSFER_CIPHER_KEY_BITLENGTH_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_ENCRYPT_DATA_TRANSFER_CIPHER_SUITES_KEY;
-import static org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
+import static org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed;
import java.io.IOException;
import java.io.InputStream;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolServerSideTranslatorPB.java
index 2bc6a18..a628287 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolServerSideTranslatorPB.java
@@ -137,7 +137,7 @@ public class ClientDatanodeProtocolServerSideTranslatorPB implements
throw new ServiceException(e);
}
return GetBlockLocalPathInfoResponseProto.newBuilder()
- .setBlock(PBHelper.convert(resp.getBlock()))
+ .setBlock(PBHelperClient.convert(resp.getBlock()))
.setLocalPath(resp.getBlockPath()).setLocalMetaPath(resp.getMetaPath())
.build();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolTranslatorPB.java
index 9d6375b..53ca147 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientDatanodeProtocolTranslatorPB.java
@@ -185,7 +185,7 @@ public class ClientDatanodeProtocolTranslatorPB implements
@Override
public long getReplicaVisibleLength(ExtendedBlock b) throws IOException {
GetReplicaVisibleLengthRequestProto req = GetReplicaVisibleLengthRequestProto
- .newBuilder().setBlock(PBHelper.convert(b)).build();
+ .newBuilder().setBlock(PBHelperClient.convert(b)).build();
try {
return rpcProxy.getReplicaVisibleLength(NULL_CONTROLLER, req).getLength();
} catch (ServiceException e) {
@@ -218,8 +218,8 @@ public class ClientDatanodeProtocolTranslatorPB implements
Token<BlockTokenIdentifier> token) throws IOException {
GetBlockLocalPathInfoRequestProto req =
GetBlockLocalPathInfoRequestProto.newBuilder()
- .setBlock(PBHelper.convert(block))
- .setToken(PBHelper.convert(token)).build();
+ .setBlock(PBHelperClient.convert(block))
+ .setToken(PBHelperClient.convert(token)).build();
GetBlockLocalPathInfoResponseProto resp;
try {
resp = rpcProxy.getBlockLocalPathInfo(NULL_CONTROLLER, req);
[2/4] hadoop git commit: HDFS-8934. Move ShortCircuitShm to
hdfs-client. Contributed by Mingliang Liu.
Posted by wh...@apache.org.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
index 02b20d6..a95f397 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolServerSideTranslatorPB.java
@@ -698,7 +698,7 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
RpcController controller, GetDatanodeReportRequestProto req)
throws ServiceException {
try {
- List<? extends DatanodeInfoProto> result = PBHelper.convert(server
+ List<? extends DatanodeInfoProto> result = PBHelperClient.convert(server
.getDatanodeReport(PBHelper.convert(req.getType())));
return GetDatanodeReportResponseProto.newBuilder()
.addAllDi(result).build();
@@ -892,7 +892,7 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
server.setQuota(req.getPath(), req.getNamespaceQuota(),
req.getStoragespaceQuota(),
req.hasStorageType() ?
- PBHelper.convertStorageType(req.getStorageType()): null);
+ PBHelperClient.convertStorageType(req.getStorageType()): null);
return VOID_SETQUOTA_RESPONSE;
} catch (IOException e) {
throw new ServiceException(e);
@@ -992,7 +992,7 @@ public class ClientNamenodeProtocolServerSideTranslatorPB implements
GetDelegationTokenResponseProto.Builder rspBuilder =
GetDelegationTokenResponseProto.newBuilder();
if (token != null) {
- rspBuilder.setToken(PBHelper.convert(token));
+ rspBuilder.setToken(PBHelperClient.convert(token));
}
return rspBuilder.build();
} catch (IOException e) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
index 7e57b97..a0431b1 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java
@@ -390,7 +390,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
String holder) throws AccessControlException, FileNotFoundException,
UnresolvedLinkException, IOException {
AbandonBlockRequestProto req = AbandonBlockRequestProto.newBuilder()
- .setB(PBHelper.convert(b)).setSrc(src).setHolder(holder)
+ .setB(PBHelperClient.convert(b)).setSrc(src).setHolder(holder)
.setFileId(fileId).build();
try {
rpcProxy.abandonBlock(null, req);
@@ -409,9 +409,9 @@ public class ClientNamenodeProtocolTranslatorPB implements
AddBlockRequestProto.Builder req = AddBlockRequestProto.newBuilder()
.setSrc(src).setClientName(clientName).setFileId(fileId);
if (previous != null)
- req.setPrevious(PBHelper.convert(previous));
- if (excludeNodes != null)
- req.addAllExcludeNodes(PBHelper.convert(excludeNodes));
+ req.setPrevious(PBHelperClient.convert(previous));
+ if (excludeNodes != null)
+ req.addAllExcludeNodes(PBHelperClient.convert(excludeNodes));
if (favoredNodes != null) {
req.addAllFavoredNodes(Arrays.asList(favoredNodes));
}
@@ -433,10 +433,10 @@ public class ClientNamenodeProtocolTranslatorPB implements
.newBuilder()
.setSrc(src)
.setFileId(fileId)
- .setBlk(PBHelper.convert(blk))
- .addAllExistings(PBHelper.convert(existings))
+ .setBlk(PBHelperClient.convert(blk))
+ .addAllExistings(PBHelperClient.convert(existings))
.addAllExistingStorageUuids(Arrays.asList(existingStorageIDs))
- .addAllExcludes(PBHelper.convert(excludes))
+ .addAllExcludes(PBHelperClient.convert(excludes))
.setNumAdditionalNodes(numAdditionalNodes)
.setClientName(clientName)
.build();
@@ -458,7 +458,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
.setClientName(clientName)
.setFileId(fileId);
if (last != null)
- req.setLast(PBHelper.convert(last));
+ req.setLast(PBHelperClient.convert(last));
try {
return rpcProxy.complete(null, req.build()).getResult();
} catch (ServiceException e) {
@@ -817,7 +817,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
.setNamespaceQuota(namespaceQuota)
.setStoragespaceQuota(storagespaceQuota);
if (type != null) {
- builder.setStorageType(PBHelper.convertStorageType(type));
+ builder.setStorageType(PBHelperClient.convertStorageType(type));
}
final SetQuotaRequestProto req = builder.build();
try {
@@ -895,7 +895,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
String clientName) throws IOException {
UpdateBlockForPipelineRequestProto req = UpdateBlockForPipelineRequestProto
.newBuilder()
- .setBlock(PBHelper.convert(block))
+ .setBlock(PBHelperClient.convert(block))
.setClientName(clientName)
.build();
try {
@@ -911,8 +911,8 @@ public class ClientNamenodeProtocolTranslatorPB implements
ExtendedBlock newBlock, DatanodeID[] newNodes, String[] storageIDs) throws IOException {
UpdatePipelineRequestProto req = UpdatePipelineRequestProto.newBuilder()
.setClientName(clientName)
- .setOldBlock(PBHelper.convert(oldBlock))
- .setNewBlock(PBHelper.convert(newBlock))
+ .setOldBlock(PBHelperClient.convert(oldBlock))
+ .setNewBlock(PBHelperClient.convert(newBlock))
.addAllNewNodes(Arrays.asList(PBHelper.convert(newNodes)))
.addAllStorageIDs(storageIDs == null ? null : Arrays.asList(storageIDs))
.build();
@@ -943,7 +943,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
public long renewDelegationToken(Token<DelegationTokenIdentifier> token)
throws IOException {
RenewDelegationTokenRequestProto req = RenewDelegationTokenRequestProto.newBuilder().
- setToken(PBHelper.convert(token)).
+ setToken(PBHelperClient.convert(token)).
build();
try {
return rpcProxy.renewDelegationToken(null, req).getNewExpiryTime();
@@ -957,7 +957,7 @@ public class ClientNamenodeProtocolTranslatorPB implements
throws IOException {
CancelDelegationTokenRequestProto req = CancelDelegationTokenRequestProto
.newBuilder()
- .setToken(PBHelper.convert(token))
+ .setToken(PBHelperClient.convert(token))
.build();
try {
rpcProxy.cancelDelegationToken(null, req);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
index 94028a2..0b46927 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/DatanodeProtocolClientSideTranslatorPB.java
@@ -298,11 +298,11 @@ public class DatanodeProtocolClientSideTranslatorPB implements
) throws IOException {
CommitBlockSynchronizationRequestProto.Builder builder =
CommitBlockSynchronizationRequestProto.newBuilder()
- .setBlock(PBHelper.convert(block)).setNewGenStamp(newgenerationstamp)
+ .setBlock(PBHelperClient.convert(block)).setNewGenStamp(newgenerationstamp)
.setNewLength(newlength).setCloseFile(closeFile)
.setDeleteBlock(deleteblock);
for (int i = 0; i < newtargets.length; i++) {
- builder.addNewTaragets(PBHelper.convert(newtargets[i]));
+ builder.addNewTaragets(PBHelperClient.convert(newtargets[i]));
builder.addNewTargetStorages(newtargetstorages[i]);
}
CommitBlockSynchronizationRequestProto req = builder.build();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
index fee62a4..17ba196 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/InterDatanodeProtocolTranslatorPB.java
@@ -105,7 +105,7 @@ public class InterDatanodeProtocolTranslatorPB implements
long recoveryId, long newBlockId, long newLength) throws IOException {
UpdateReplicaUnderRecoveryRequestProto req =
UpdateReplicaUnderRecoveryRequestProto.newBuilder()
- .setBlock(PBHelper.convert(oldBlock))
+ .setBlock(PBHelperClient.convert(oldBlock))
.setNewLength(newLength).setNewBlockId(newBlockId)
.setRecoveryId(recoveryId).build();
try {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
index 82c5c4c..bcb96ba 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/NamenodeProtocolTranslatorPB.java
@@ -101,7 +101,7 @@ public class NamenodeProtocolTranslatorPB implements NamenodeProtocol,
public BlocksWithLocations getBlocks(DatanodeInfo datanode, long size)
throws IOException {
GetBlocksRequestProto req = GetBlocksRequestProto.newBuilder()
- .setDatanode(PBHelper.convert((DatanodeID)datanode)).setSize(size)
+ .setDatanode(PBHelperClient.convert((DatanodeID)datanode)).setSize(size)
.build();
try {
return PBHelper.convert(rpcProxy.getBlocks(NULL_CONTROLLER, req)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
index a252262..dbb1861 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java
@@ -347,7 +347,7 @@ public class PBHelper {
if (types == null || types.length == 0) {
return null;
}
- List<StorageTypeProto> list = convertStorageTypes(types);
+ List<StorageTypeProto> list = PBHelperClient.convertStorageTypes(types);
return StorageTypesProto.newBuilder().addAllStorageTypes(list).build();
}
@@ -382,20 +382,6 @@ public class PBHelper {
.getInfoSecurePort() : 0, dn.getIpcPort());
}
- public static DatanodeIDProto convert(DatanodeID dn) {
- // For wire compatibility with older versions we transmit the StorageID
- // which is the same as the DatanodeUuid. Since StorageID is a required
- // field we pass the empty string if the DatanodeUuid is not yet known.
- return DatanodeIDProto.newBuilder()
- .setIpAddr(dn.getIpAddr())
- .setHostName(dn.getHostName())
- .setXferPort(dn.getXferPort())
- .setDatanodeUuid(dn.getDatanodeUuid() != null ? dn.getDatanodeUuid() : "")
- .setInfoPort(dn.getInfoPort())
- .setInfoSecurePort(dn.getInfoSecurePort())
- .setIpcPort(dn.getIpcPort()).build();
- }
-
// Arrays of DatanodeId
public static DatanodeIDProto[] convert(DatanodeID[] did) {
if (did == null)
@@ -403,7 +389,7 @@ public class PBHelper {
final int len = did.length;
DatanodeIDProto[] result = new DatanodeIDProto[len];
for (int i = 0; i < len; ++i) {
- result[i] = convert(did[i]);
+ result[i] = PBHelperClient.convert(did[i]);
}
return result;
}
@@ -434,7 +420,7 @@ public class PBHelper {
.setBlock(convert(blk.getBlock()))
.addAllDatanodeUuids(Arrays.asList(blk.getDatanodeUuids()))
.addAllStorageUuids(Arrays.asList(blk.getStorageIDs()))
- .addAllStorageTypes(convertStorageTypes(blk.getStorageTypes()))
+ .addAllStorageTypes(PBHelperClient.convertStorageTypes(blk.getStorageTypes()))
.build();
}
@@ -596,16 +582,6 @@ public class PBHelper {
eb.getGenerationStamp());
}
- public static ExtendedBlockProto convert(final ExtendedBlock b) {
- if (b == null) return null;
- return ExtendedBlockProto.newBuilder().
- setPoolId(b.getBlockPoolId()).
- setBlockId(b.getBlockId()).
- setNumBytes(b.getNumBytes()).
- setGenerationStamp(b.getGenerationStamp()).
- build();
- }
-
public static RecoveringBlockProto convert(RecoveringBlock b) {
if (b == null) {
return null;
@@ -626,17 +602,6 @@ public class PBHelper {
new RecoveringBlock(block, locs, b.getNewGenStamp());
}
- public static DatanodeInfoProto.AdminState convert(
- final DatanodeInfo.AdminStates inAs) {
- switch (inAs) {
- case NORMAL: return DatanodeInfoProto.AdminState.NORMAL;
- case DECOMMISSION_INPROGRESS:
- return DatanodeInfoProto.AdminState.DECOMMISSION_INPROGRESS;
- case DECOMMISSIONED: return DatanodeInfoProto.AdminState.DECOMMISSIONED;
- default: return DatanodeInfoProto.AdminState.NORMAL;
- }
- }
-
static public DatanodeInfo convert(DatanodeInfoProto di) {
if (di == null) return null;
return new DatanodeInfo(
@@ -648,12 +613,6 @@ public class PBHelper {
di.getXceiverCount(), PBHelper.convert(di.getAdminState()));
}
- static public DatanodeInfoProto convertDatanodeInfo(DatanodeInfo di) {
- if (di == null) return null;
- return convert(di);
- }
-
-
static public DatanodeInfo[] convert(DatanodeInfoProto di[]) {
if (di == null) return null;
DatanodeInfo[] result = new DatanodeInfo[di.length];
@@ -663,27 +622,6 @@ public class PBHelper {
return result;
}
- public static List<? extends HdfsProtos.DatanodeInfoProto> convert(
- DatanodeInfo[] dnInfos) {
- return convert(dnInfos, 0);
- }
-
- /**
- * Copy from {@code dnInfos} to a target of list of same size starting at
- * {@code startIdx}.
- */
- public static List<? extends HdfsProtos.DatanodeInfoProto> convert(
- DatanodeInfo[] dnInfos, int startIdx) {
- if (dnInfos == null)
- return null;
- ArrayList<HdfsProtos.DatanodeInfoProto> protos = Lists
- .newArrayListWithCapacity(dnInfos.length);
- for (int i = startIdx; i < dnInfos.length; i++) {
- protos.add(convert(dnInfos[i]));
- }
- return protos;
- }
-
public static DatanodeInfo[] convert(List<DatanodeInfoProto> list) {
DatanodeInfo[] info = new DatanodeInfo[list.size()];
for (int i = 0; i < info.length; i++) {
@@ -691,32 +629,11 @@ public class PBHelper {
}
return info;
}
-
- public static DatanodeInfoProto convert(DatanodeInfo info) {
- DatanodeInfoProto.Builder builder = DatanodeInfoProto.newBuilder();
- if (info.getNetworkLocation() != null) {
- builder.setLocation(info.getNetworkLocation());
- }
- builder
- .setId(PBHelper.convert((DatanodeID)info))
- .setCapacity(info.getCapacity())
- .setDfsUsed(info.getDfsUsed())
- .setRemaining(info.getRemaining())
- .setBlockPoolUsed(info.getBlockPoolUsed())
- .setCacheCapacity(info.getCacheCapacity())
- .setCacheUsed(info.getCacheUsed())
- .setLastUpdate(info.getLastUpdate())
- .setLastUpdateMonotonic(info.getLastUpdateMonotonic())
- .setXceiverCount(info.getXceiverCount())
- .setAdminState(PBHelper.convert(info.getAdminState()))
- .build();
- return builder.build();
- }
public static DatanodeStorageReportProto convertDatanodeStorageReport(
DatanodeStorageReport report) {
return DatanodeStorageReportProto.newBuilder()
- .setDatanodeInfo(convert(report.getDatanodeInfo()))
+ .setDatanodeInfo(PBHelperClient.convert(report.getDatanodeInfo()))
.addAllStorageReports(convertStorageReports(report.getStorageReports()))
.build();
}
@@ -768,7 +685,7 @@ public class PBHelper {
Lists.newLinkedList(Arrays.asList(b.getCachedLocations()));
for (int i = 0; i < locs.length; i++) {
DatanodeInfo loc = locs[i];
- builder.addLocs(i, PBHelper.convert(loc));
+ builder.addLocs(i, PBHelperClient.convert(loc));
boolean locIsCached = cachedLocs.contains(loc);
builder.addIsCached(locIsCached);
if (locIsCached) {
@@ -782,7 +699,7 @@ public class PBHelper {
StorageType[] storageTypes = b.getStorageTypes();
if (storageTypes != null) {
for (int i = 0; i < storageTypes.length; ++i) {
- builder.addStorageTypes(PBHelper.convertStorageType(storageTypes[i]));
+ builder.addStorageTypes(PBHelperClient.convertStorageType(storageTypes[i]));
}
}
final String[] storageIDs = b.getStorageIDs();
@@ -790,8 +707,8 @@ public class PBHelper {
builder.addAllStorageIDs(Arrays.asList(storageIDs));
}
- return builder.setB(PBHelper.convert(b.getBlock()))
- .setBlockToken(PBHelper.convert(b.getBlockToken()))
+ return builder.setB(PBHelperClient.convert(b.getBlock()))
+ .setBlockToken(PBHelperClient.convert(b.getBlockToken()))
.setCorrupt(b.isCorrupt()).setOffset(b.getStartOffset()).build();
}
@@ -832,14 +749,6 @@ public class PBHelper {
return lb;
}
- public static TokenProto convert(Token<?> tok) {
- return TokenProto.newBuilder().
- setIdentifier(ByteString.copyFrom(tok.getIdentifier())).
- setPassword(ByteString.copyFrom(tok.getPassword())).
- setKind(tok.getKind().toString()).
- setService(tok.getService().toString()).build();
- }
-
public static Token<BlockTokenIdentifier> convert(
TokenProto blockToken) {
return new Token<BlockTokenIdentifier>(blockToken.getIdentifier()
@@ -891,7 +800,7 @@ public class PBHelper {
DatanodeRegistration registration) {
DatanodeRegistrationProto.Builder builder = DatanodeRegistrationProto
.newBuilder();
- return builder.setDatanodeID(PBHelper.convert((DatanodeID) registration))
+ return builder.setDatanodeID(PBHelperClient.convert((DatanodeID) registration))
.setStorageInfo(PBHelper.convert(registration.getStorageInfo()))
.setKeys(PBHelper.convert(registration.getExportedKeys()))
.setSoftwareVersion(registration.getSoftwareVersion()).build();
@@ -983,7 +892,7 @@ public class PBHelper {
if (types != null) {
for (StorageType[] ts : types) {
StorageTypesProto.Builder builder = StorageTypesProto.newBuilder();
- builder.addAllStorageTypes(convertStorageTypes(ts));
+ builder.addAllStorageTypes(PBHelperClient.convertStorageTypes(ts));
list.add(builder.build());
}
}
@@ -1014,7 +923,7 @@ public class PBHelper {
DatanodeInfosProto[] ret = new DatanodeInfosProto[targets.length];
for (int i = 0; i < targets.length; i++) {
ret[i] = DatanodeInfosProto.newBuilder()
- .addAllDatanodes(PBHelper.convert(targets[i])).build();
+ .addAllDatanodes(PBHelperClient.convert(targets[i])).build();
}
return Arrays.asList(ret);
}
@@ -1338,7 +1247,7 @@ public class PBHelper {
fs.getFileBufferSize(),
fs.getEncryptDataTransfer(),
fs.getTrashInterval(),
- PBHelper.convert(fs.getChecksumType()));
+ PBHelperClient.convert(fs.getChecksumType()));
}
public static FsServerDefaultsProto convert(FsServerDefaults fs) {
@@ -1351,7 +1260,7 @@ public class PBHelper {
.setFileBufferSize(fs.getFileBufferSize())
.setEncryptDataTransfer(fs.getEncryptDataTransfer())
.setTrashInterval(fs.getTrashInterval())
- .setChecksumType(PBHelper.convert(fs.getChecksumType()))
+ .setChecksumType(PBHelperClient.convert(fs.getChecksumType()))
.build();
}
@@ -1739,7 +1648,7 @@ public class PBHelper {
if (cs.hasTypeQuotaInfos()) {
for (HdfsProtos.StorageTypeQuotaInfoProto info :
cs.getTypeQuotaInfos().getTypeQuotaInfoList()) {
- StorageType type = PBHelper.convertStorageType(info.getType());
+ StorageType type = PBHelperClient.convertStorageType(info.getType());
builder.typeConsumed(type, info.getConsumed());
builder.typeQuota(type, info.getQuota());
}
@@ -1763,7 +1672,7 @@ public class PBHelper {
for (StorageType t: StorageType.getTypesSupportingQuota()) {
HdfsProtos.StorageTypeQuotaInfoProto info =
HdfsProtos.StorageTypeQuotaInfoProto.newBuilder().
- setType(convertStorageType(t)).
+ setType(PBHelperClient.convertStorageType(t)).
setConsumed(cs.getTypeConsumed(t)).
setQuota(cs.getTypeQuota(t)).
build();
@@ -1808,7 +1717,7 @@ public class PBHelper {
public static DatanodeStorageProto convert(DatanodeStorage s) {
return DatanodeStorageProto.newBuilder()
.setState(PBHelper.convertState(s.getState()))
- .setStorageType(PBHelper.convertStorageType(s.getStorageType()))
+ .setStorageType(PBHelperClient.convertStorageType(s.getStorageType()))
.setStorageUuid(s.getStorageID()).build();
}
@@ -1822,44 +1731,10 @@ public class PBHelper {
}
}
- public static List<StorageTypeProto> convertStorageTypes(
- StorageType[] types) {
- return convertStorageTypes(types, 0);
- }
-
- public static List<StorageTypeProto> convertStorageTypes(
- StorageType[] types, int startIdx) {
- if (types == null) {
- return null;
- }
- final List<StorageTypeProto> protos = new ArrayList<StorageTypeProto>(
- types.length);
- for (int i = startIdx; i < types.length; ++i) {
- protos.add(convertStorageType(types[i]));
- }
- return protos;
- }
-
- public static StorageTypeProto convertStorageType(StorageType type) {
- switch(type) {
- case DISK:
- return StorageTypeProto.DISK;
- case SSD:
- return StorageTypeProto.SSD;
- case ARCHIVE:
- return StorageTypeProto.ARCHIVE;
- case RAM_DISK:
- return StorageTypeProto.RAM_DISK;
- default:
- throw new IllegalStateException(
- "BUG: StorageType not found, type=" + type);
- }
- }
-
public static DatanodeStorage convert(DatanodeStorageProto s) {
return new DatanodeStorage(s.getStorageUuid(),
PBHelper.convertState(s.getState()),
- PBHelper.convertStorageType(s.getStorageType()));
+ PBHelperClient.convertStorageType(s.getStorageType()));
}
private static State convertState(StorageState state) {
@@ -1872,22 +1747,6 @@ public class PBHelper {
}
}
- public static StorageType convertStorageType(StorageTypeProto type) {
- switch(type) {
- case DISK:
- return StorageType.DISK;
- case SSD:
- return StorageType.SSD;
- case ARCHIVE:
- return StorageType.ARCHIVE;
- case RAM_DISK:
- return StorageType.RAM_DISK;
- default:
- throw new IllegalStateException(
- "BUG: StorageTypeProto not found, type=" + type);
- }
- }
-
public static StorageType[] convertStorageTypes(
List<StorageTypeProto> storageTypesList, int expectedSize) {
final StorageType[] storageTypes = new StorageType[expectedSize];
@@ -1896,7 +1755,7 @@ public class PBHelper {
Arrays.fill(storageTypes, StorageType.DEFAULT);
} else {
for (int i = 0; i < storageTypes.length; ++i) {
- storageTypes[i] = convertStorageType(storageTypesList.get(i));
+ storageTypes[i] = PBHelperClient.convertStorageType(storageTypesList.get(i));
}
}
return storageTypes;
@@ -2080,10 +1939,6 @@ public class PBHelper {
return reportProto;
}
- public static DataChecksum.Type convert(HdfsProtos.ChecksumTypeProto type) {
- return DataChecksum.Type.valueOf(type.getNumber());
- }
-
public static CacheDirectiveInfoProto convert
(CacheDirectiveInfo info) {
CacheDirectiveInfoProto.Builder builder =
@@ -2256,9 +2111,6 @@ public class PBHelper {
return new CachePoolEntry(info, stats);
}
- public static HdfsProtos.ChecksumTypeProto convert(DataChecksum.Type type) {
- return HdfsProtos.ChecksumTypeProto.valueOf(type.id);
- }
public static DatanodeLocalInfoProto convert(DatanodeLocalInfo info) {
DatanodeLocalInfoProto.Builder builder = DatanodeLocalInfoProto.newBuilder();
@@ -2273,17 +2125,6 @@ public class PBHelper {
proto.getConfigVersion(), proto.getUptime());
}
- public static InputStream vintPrefixed(final InputStream input)
- throws IOException {
- final int firstByte = input.read();
- if (firstByte == -1) {
- throw new EOFException("Premature EOF: no length prefix available");
- }
-
- int size = CodedInputStream.readRawVarint32(firstByte, input);
- assert size >= 0;
- return new ExactSizeInputStream(input, size);
- }
private static AclEntryScopeProto convert(AclEntryScope v) {
return AclEntryScopeProto.valueOf(v.ordinal());
@@ -2507,30 +2348,11 @@ public class PBHelper {
proto.getKeyName());
}
- public static ShortCircuitShmSlotProto convert(SlotId slotId) {
- return ShortCircuitShmSlotProto.newBuilder().
- setShmId(convert(slotId.getShmId())).
- setSlotIdx(slotId.getSlotIdx()).
- build();
- }
-
- public static ShortCircuitShmIdProto convert(ShmId shmId) {
- return ShortCircuitShmIdProto.newBuilder().
- setHi(shmId.getHi()).
- setLo(shmId.getLo()).
- build();
-
- }
-
public static SlotId convert(ShortCircuitShmSlotProto slotId) {
- return new SlotId(PBHelper.convert(slotId.getShmId()),
+ return new SlotId(PBHelperClient.convert(slotId.getShmId()),
slotId.getSlotIdx());
}
- public static ShmId convert(ShortCircuitShmIdProto shmId) {
- return new ShmId(shmId.getHi(), shmId.getLo());
- }
-
private static Event.CreateEvent.INodeType createTypeConvert(InotifyProtos.INodeType
type) {
switch (type) {
@@ -3037,18 +2859,6 @@ public class PBHelper {
ezKeyVersionName);
}
- public static List<Boolean> convert(boolean[] targetPinnings, int idx) {
- List<Boolean> pinnings = new ArrayList<Boolean>();
- if (targetPinnings == null) {
- pinnings.add(Boolean.FALSE);
- } else {
- for (; idx < targetPinnings.length; ++idx) {
- pinnings.add(Boolean.valueOf(targetPinnings[idx]));
- }
- }
- return pinnings;
- }
-
public static boolean[] convertBooleanList(
List<Boolean> targetPinningsList) {
final boolean[] targetPinnings = new boolean[targetPinningsList.size()];
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
deleted file mode 100644
index 2fa86fa..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
+++ /dev/null
@@ -1,41 +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.hdfs.security.token.block;
-
-import java.io.IOException;
-
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.classification.InterfaceStability;
-
-/**
- * Access token verification failed.
- */
-@InterfaceAudience.Private
-@InterfaceStability.Evolving
-public class InvalidBlockTokenException extends IOException {
- private static final long serialVersionUID = 168L;
-
- public InvalidBlockTokenException() {
- super();
- }
-
- public InvalidBlockTokenException(String msg) {
- super(msg);
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
index 40564df..dc967ff 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Dispatcher.java
@@ -17,7 +17,7 @@
*/
package org.apache.hadoop.hdfs.server.balancer;
-import static org.apache.hadoop.hdfs.protocolPB.PBHelper.vintPrefixed;
+import static org.apache.hadoop.hdfs.protocolPB.PBHelperClient.vintPrefixed;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
deleted file mode 100644
index 215df13..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
+++ /dev/null
@@ -1,76 +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.hdfs.server.datanode;
-
-/**
- * The caching strategy we should use for an HDFS read or write operation.
- */
-public class CachingStrategy {
- private final Boolean dropBehind; // null = use server defaults
- private final Long readahead; // null = use server defaults
-
- public static CachingStrategy newDefaultStrategy() {
- return new CachingStrategy(null, null);
- }
-
- public static CachingStrategy newDropBehind() {
- return new CachingStrategy(true, null);
- }
-
- public static class Builder {
- private Boolean dropBehind;
- private Long readahead;
-
- public Builder(CachingStrategy prev) {
- this.dropBehind = prev.dropBehind;
- this.readahead = prev.readahead;
- }
-
- public Builder setDropBehind(Boolean dropBehind) {
- this.dropBehind = dropBehind;
- return this;
- }
-
- public Builder setReadahead(Long readahead) {
- this.readahead = readahead;
- return this;
- }
-
- public CachingStrategy build() {
- return new CachingStrategy(dropBehind, readahead);
- }
- }
-
- public CachingStrategy(Boolean dropBehind, Long readahead) {
- this.dropBehind = dropBehind;
- this.readahead = readahead;
- }
-
- public Boolean getDropBehind() {
- return dropBehind;
- }
-
- public Long getReadahead() {
- return readahead;
- }
-
- public String toString() {
- return "CachingStrategy(dropBehind=" + dropBehind +
- ", readahead=" + readahead + ")";
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
index fa3b78c..df03808 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataNode.java
@@ -137,7 +137,7 @@ import org.apache.hadoop.hdfs.protocolPB.DatanodeProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdfs.protocolPB.InterDatanodeProtocolPB;
import org.apache.hadoop.hdfs.protocolPB.InterDatanodeProtocolServerSideTranslatorPB;
import org.apache.hadoop.hdfs.protocolPB.InterDatanodeProtocolTranslatorPB;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockPoolTokenSecretManager;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier.AccessMode;
@@ -2172,7 +2172,7 @@ public class DataNode extends ReconfigurableBase
// read ack
if (isClient) {
DNTransferAckProto closeAck = DNTransferAckProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
if (LOG.isDebugEnabled()) {
LOG.debug(getClass().getSimpleName() + ": close-ack=" + closeAck);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
index e9cf436..dfaa525 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataXceiver.java
@@ -70,7 +70,7 @@ import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReadOpChecksumIn
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
import org.apache.hadoop.hdfs.server.datanode.DataNode.ShortCircuitFdsUnsupportedException;
import org.apache.hadoop.hdfs.server.datanode.DataNode.ShortCircuitFdsVersionException;
@@ -427,7 +427,7 @@ class DataXceiver extends Receiver implements Runnable {
throws IOException {
DataNodeFaultInjector.get().sendShortCircuitShmResponse();
ShortCircuitShmResponseProto.newBuilder().setStatus(SUCCESS).
- setId(PBHelper.convert(shmInfo.shmId)).build().
+ setId(PBHelperClient.convert(shmInfo.shmId)).build().
writeDelimitedTo(socketOut);
// Send the file descriptor for the shared memory segment.
byte buf[] = new byte[] { (byte)0 };
@@ -559,7 +559,7 @@ class DataXceiver extends Receiver implements Runnable {
// to respond with a Status enum.
try {
ClientReadStatusProto stat = ClientReadStatusProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
if (!stat.hasStatus()) {
LOG.warn("Client " + peer.getRemoteAddressString() +
" did not send a valid status code after reading. " +
@@ -745,7 +745,7 @@ class DataXceiver extends Receiver implements Runnable {
// read connect ack (only for clients, not for replication req)
if (isClient) {
BlockOpResponseProto connectAck =
- BlockOpResponseProto.parseFrom(PBHelper.vintPrefixed(mirrorIn));
+ BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(mirrorIn));
mirrorInStatus = connectAck.getStatus();
firstBadLink = connectAck.getFirstBadLink();
if (LOG.isDebugEnabled() || mirrorInStatus != SUCCESS) {
@@ -962,7 +962,7 @@ class DataXceiver extends Receiver implements Runnable {
.setBytesPerCrc(bytesPerCRC)
.setCrcPerBlock(crcPerBlock)
.setMd5(ByteString.copyFrom(md5.getDigest()))
- .setCrcType(PBHelper.convert(checksum.getChecksumType())))
+ .setCrcType(PBHelperClient.convert(checksum.getChecksumType())))
.build()
.writeDelimitedTo(out);
out.flush();
@@ -1147,8 +1147,8 @@ class DataXceiver extends Receiver implements Runnable {
// receive the response from the proxy
BlockOpResponseProto copyResponse = BlockOpResponseProto.parseFrom(
- PBHelper.vintPrefixed(proxyReply));
-
+ PBHelperClient.vintPrefixed(proxyReply));
+
String logInfo = "copy block " + block + " from "
+ proxySock.getRemoteSocketAddress();
DataTransferProtoUtil.checkBlockOpStatus(copyResponse, logInfo);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
index 9df1713..0afb06c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSImageFormatPBINode.java
@@ -42,6 +42,7 @@ import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.HdfsConstants;
import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.BlockProto;
import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfoContiguous;
import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
@@ -154,7 +155,7 @@ public final class FSImageFormatPBINode {
QuotaByStorageTypeFeatureProto proto) {
ImmutableList.Builder<QuotaByStorageTypeEntry> b = ImmutableList.builder();
for (QuotaByStorageTypeEntryProto quotaEntry : proto.getQuotasList()) {
- StorageType type = PBHelper.convertStorageType(quotaEntry.getStorageType());
+ StorageType type = PBHelperClient.convertStorageType(quotaEntry.getStorageType());
long quota = quotaEntry.getQuota();
b.add(new QuotaByStorageTypeEntry.Builder().setStorageType(type)
.setQuota(quota).build());
@@ -459,7 +460,7 @@ public final class FSImageFormatPBINode {
if (q.getTypeSpace(t) >= 0) {
QuotaByStorageTypeEntryProto.Builder eb =
QuotaByStorageTypeEntryProto.newBuilder().
- setStorageType(PBHelper.convertStorageType(t)).
+ setStorageType(PBHelperClient.convertStorageType(t)).
setQuota(q.getTypeSpace(t));
b.addQuotas(eb);
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
deleted file mode 100644
index 81cc68d..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
+++ /dev/null
@@ -1,119 +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.hdfs.shortcircuit;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.Iterator;
-
-import org.apache.hadoop.hdfs.net.DomainPeer;
-import org.apache.hadoop.hdfs.shortcircuit.DfsClientShmManager.EndpointShmManager;
-import org.apache.hadoop.net.unix.DomainSocket;
-import org.apache.hadoop.net.unix.DomainSocketWatcher;
-
-import com.google.common.base.Preconditions;
-
-/**
- * DfsClientShm is a subclass of ShortCircuitShm which is used by the
- * DfsClient.
- * When the UNIX domain socket associated with this shared memory segment
- * closes unexpectedly, we mark the slots inside this segment as disconnected.
- * ShortCircuitReplica objects that contain disconnected slots are stale,
- * and will not be used to service new reads or mmap operations.
- * However, in-progress read or mmap operations will continue to proceed.
- * Once the last slot is deallocated, the segment can be safely munmapped.
- *
- * Slots may also become stale because the associated replica has been deleted
- * on the DataNode. In this case, the DataNode will clear the 'valid' bit.
- * The client will then see these slots as stale (see
- * #{ShortCircuitReplica#isStale}).
- */
-public class DfsClientShm extends ShortCircuitShm
- implements DomainSocketWatcher.Handler {
- /**
- * The EndpointShmManager associated with this shared memory segment.
- */
- private final EndpointShmManager manager;
-
- /**
- * The UNIX domain socket associated with this DfsClientShm.
- * We rely on the DomainSocketWatcher to close the socket associated with
- * this DomainPeer when necessary.
- */
- private final DomainPeer peer;
-
- /**
- * True if this shared memory segment has lost its connection to the
- * DataNode.
- *
- * {@link DfsClientShm#handle} sets this to true.
- */
- private boolean disconnected = false;
-
- DfsClientShm(ShmId shmId, FileInputStream stream, EndpointShmManager manager,
- DomainPeer peer) throws IOException {
- super(shmId, stream);
- this.manager = manager;
- this.peer = peer;
- }
-
- public EndpointShmManager getEndpointShmManager() {
- return manager;
- }
-
- public DomainPeer getPeer() {
- return peer;
- }
-
- /**
- * Determine if the shared memory segment is disconnected from the DataNode.
- *
- * This must be called with the DfsClientShmManager lock held.
- *
- * @return True if the shared memory segment is stale.
- */
- public synchronized boolean isDisconnected() {
- return disconnected;
- }
-
- /**
- * Handle the closure of the UNIX domain socket associated with this shared
- * memory segment by marking this segment as stale.
- *
- * If there are no slots associated with this shared memory segment, it will
- * be freed immediately in this function.
- */
- @Override
- public boolean handle(DomainSocket sock) {
- manager.unregisterShm(getShmId());
- synchronized (this) {
- Preconditions.checkState(!disconnected);
- disconnected = true;
- boolean hadSlots = false;
- for (Iterator<Slot> iter = slotIterator(); iter.hasNext(); ) {
- Slot slot = iter.next();
- slot.makeInvalid();
- hadSlots = true;
- }
- if (!hadSlots) {
- free();
- }
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
deleted file mode 100644
index 062539a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
+++ /dev/null
@@ -1,514 +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.hdfs.shortcircuit;
-
-import java.io.BufferedOutputStream;
-import java.io.Closeable;
-import java.io.DataOutputStream;
-import java.io.EOFException;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map.Entry;
-import java.util.TreeMap;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
-
-import org.apache.commons.lang.mutable.MutableBoolean;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.classification.InterfaceAudience;
-import org.apache.hadoop.hdfs.ExtendedBlockId;
-import org.apache.hadoop.hdfs.net.DomainPeer;
-import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
-import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol;
-import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
-import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmResponseProto;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
-import org.apache.hadoop.hdfs.server.datanode.ShortCircuitRegistry;
-import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.ShmId;
-import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
-import org.apache.hadoop.io.IOUtils;
-import org.apache.hadoop.net.unix.DomainSocket;
-import org.apache.hadoop.net.unix.DomainSocketWatcher;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-
-/**
- * Manages short-circuit memory segments for an HDFS client.
- *
- * Clients are responsible for requesting and releasing shared memory segments used
- * for communicating with the DataNode. The client will try to allocate new slots
- * in the set of existing segments, falling back to getting a new segment from the
- * DataNode via {@link DataTransferProtocol#requestShortCircuitFds}.
- *
- * The counterpart to this class on the DataNode is {@link ShortCircuitRegistry}.
- * See {@link ShortCircuitRegistry} for more information on the communication protocol.
- */
-@InterfaceAudience.Private
-public class DfsClientShmManager implements Closeable {
- private static final Log LOG = LogFactory.getLog(DfsClientShmManager.class);
-
- /**
- * Manages short-circuit memory segments that pertain to a given DataNode.
- */
- class EndpointShmManager {
- /**
- * The datanode we're managing.
- */
- private final DatanodeInfo datanode;
-
- /**
- * Shared memory segments which have no empty slots.
- *
- * Protected by the manager lock.
- */
- private final TreeMap<ShmId, DfsClientShm> full =
- new TreeMap<ShmId, DfsClientShm>();
-
- /**
- * Shared memory segments which have at least one empty slot.
- *
- * Protected by the manager lock.
- */
- private final TreeMap<ShmId, DfsClientShm> notFull =
- new TreeMap<ShmId, DfsClientShm>();
-
- /**
- * True if this datanode doesn't support short-circuit shared memory
- * segments.
- *
- * Protected by the manager lock.
- */
- private boolean disabled = false;
-
- /**
- * True if we're in the process of loading a shared memory segment from
- * this DataNode.
- *
- * Protected by the manager lock.
- */
- private boolean loading = false;
-
- EndpointShmManager (DatanodeInfo datanode) {
- this.datanode = datanode;
- }
-
- /**
- * Pull a slot out of a preexisting shared memory segment.
- *
- * Must be called with the manager lock held.
- *
- * @param blockId The blockId to put inside the Slot object.
- *
- * @return null if none of our shared memory segments contain a
- * free slot; the slot object otherwise.
- */
- private Slot allocSlotFromExistingShm(ExtendedBlockId blockId) {
- if (notFull.isEmpty()) {
- return null;
- }
- Entry<ShmId, DfsClientShm> entry = notFull.firstEntry();
- DfsClientShm shm = entry.getValue();
- ShmId shmId = shm.getShmId();
- Slot slot = shm.allocAndRegisterSlot(blockId);
- if (shm.isFull()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": pulled the last slot " + slot.getSlotIdx() +
- " out of " + shm);
- }
- DfsClientShm removedShm = notFull.remove(shmId);
- Preconditions.checkState(removedShm == shm);
- full.put(shmId, shm);
- } else {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": pulled slot " + slot.getSlotIdx() +
- " out of " + shm);
- }
- }
- return slot;
- }
-
- /**
- * Ask the DataNode for a new shared memory segment. This function must be
- * called with the manager lock held. We will release the lock while
- * communicating with the DataNode.
- *
- * @param clientName The current client name.
- * @param peer The peer to use to talk to the DataNode.
- *
- * @return Null if the DataNode does not support shared memory
- * segments, or experienced an error creating the
- * shm. The shared memory segment itself on success.
- * @throws IOException If there was an error communicating over the socket.
- * We will not throw an IOException unless the socket
- * itself (or the network) is the problem.
- */
- private DfsClientShm requestNewShm(String clientName, DomainPeer peer)
- throws IOException {
- final DataOutputStream out =
- new DataOutputStream(
- new BufferedOutputStream(peer.getOutputStream()));
- new Sender(out).requestShortCircuitShm(clientName);
- ShortCircuitShmResponseProto resp =
- ShortCircuitShmResponseProto.parseFrom(
- PBHelper.vintPrefixed(peer.getInputStream()));
- String error = resp.hasError() ? resp.getError() : "(unknown)";
- switch (resp.getStatus()) {
- case SUCCESS:
- DomainSocket sock = peer.getDomainSocket();
- byte buf[] = new byte[1];
- FileInputStream fis[] = new FileInputStream[1];
- if (sock.recvFileInputStreams(fis, buf, 0, buf.length) < 0) {
- throw new EOFException("got EOF while trying to transfer the " +
- "file descriptor for the shared memory segment.");
- }
- if (fis[0] == null) {
- throw new IOException("the datanode " + datanode + " failed to " +
- "pass a file descriptor for the shared memory segment.");
- }
- try {
- DfsClientShm shm =
- new DfsClientShm(PBHelper.convert(resp.getId()),
- fis[0], this, peer);
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": createNewShm: created " + shm);
- }
- return shm;
- } finally {
- IOUtils.cleanup(LOG, fis[0]);
- }
- case ERROR_UNSUPPORTED:
- // The DataNode just does not support short-circuit shared memory
- // access, and we should stop asking.
- LOG.info(this + ": datanode does not support short-circuit " +
- "shared memory access: " + error);
- disabled = true;
- return null;
- default:
- // The datanode experienced some kind of unexpected error when trying to
- // create the short-circuit shared memory segment.
- LOG.warn(this + ": error requesting short-circuit shared memory " +
- "access: " + error);
- return null;
- }
- }
-
- /**
- * Allocate a new shared memory slot connected to this datanode.
- *
- * Must be called with the EndpointShmManager lock held.
- *
- * @param peer The peer to use to talk to the DataNode.
- * @param usedPeer (out param) Will be set to true if we used the peer.
- * When a peer is used
- *
- * @param clientName The client name.
- * @param blockId The block ID to use.
- * @return null if the DataNode does not support shared memory
- * segments, or experienced an error creating the
- * shm. The shared memory segment itself on success.
- * @throws IOException If there was an error communicating over the socket.
- */
- Slot allocSlot(DomainPeer peer, MutableBoolean usedPeer,
- String clientName, ExtendedBlockId blockId) throws IOException {
- while (true) {
- if (closed) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": the DfsClientShmManager has been closed.");
- }
- return null;
- }
- if (disabled) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": shared memory segment access is disabled.");
- }
- return null;
- }
- // Try to use an existing slot.
- Slot slot = allocSlotFromExistingShm(blockId);
- if (slot != null) {
- return slot;
- }
- // There are no free slots. If someone is loading more slots, wait
- // for that to finish.
- if (loading) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": waiting for loading to finish...");
- }
- finishedLoading.awaitUninterruptibly();
- } else {
- // Otherwise, load the slot ourselves.
- loading = true;
- lock.unlock();
- DfsClientShm shm;
- try {
- shm = requestNewShm(clientName, peer);
- if (shm == null) continue;
- // See #{DfsClientShmManager#domainSocketWatcher} for details
- // about why we do this before retaking the manager lock.
- domainSocketWatcher.add(peer.getDomainSocket(), shm);
- // The DomainPeer is now our responsibility, and should not be
- // closed by the caller.
- usedPeer.setValue(true);
- } finally {
- lock.lock();
- loading = false;
- finishedLoading.signalAll();
- }
- if (shm.isDisconnected()) {
- // If the peer closed immediately after the shared memory segment
- // was created, the DomainSocketWatcher callback might already have
- // fired and marked the shm as disconnected. In this case, we
- // obviously don't want to add the SharedMemorySegment to our list
- // of valid not-full segments.
- if (LOG.isDebugEnabled()) {
- LOG.debug(this + ": the UNIX domain socket associated with " +
- "this short-circuit memory closed before we could make " +
- "use of the shm.");
- }
- } else {
- notFull.put(shm.getShmId(), shm);
- }
- }
- }
- }
-
- /**
- * Stop tracking a slot.
- *
- * Must be called with the EndpointShmManager lock held.
- *
- * @param slot The slot to release.
- */
- void freeSlot(Slot slot) {
- DfsClientShm shm = (DfsClientShm)slot.getShm();
- shm.unregisterSlot(slot.getSlotIdx());
- if (shm.isDisconnected()) {
- // Stale shared memory segments should not be tracked here.
- Preconditions.checkState(!full.containsKey(shm.getShmId()));
- Preconditions.checkState(!notFull.containsKey(shm.getShmId()));
- if (shm.isEmpty()) {
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": freeing empty stale " + shm);
- }
- shm.free();
- }
- } else {
- ShmId shmId = shm.getShmId();
- full.remove(shmId); // The shm can't be full if we just freed a slot.
- if (shm.isEmpty()) {
- notFull.remove(shmId);
-
- // If the shared memory segment is now empty, we call shutdown(2) on
- // the UNIX domain socket associated with it. The DomainSocketWatcher,
- // which is watching this socket, will call DfsClientShm#handle,
- // cleaning up this shared memory segment.
- //
- // See #{DfsClientShmManager#domainSocketWatcher} for details about why
- // we don't want to call DomainSocketWatcher#remove directly here.
- //
- // Note that we could experience 'fragmentation' here, where the
- // DFSClient allocates a bunch of slots in different shared memory
- // segments, and then frees most of them, but never fully empties out
- // any segment. We make some attempt to avoid this fragmentation by
- // always allocating new slots out of the shared memory segment with the
- // lowest ID, but it could still occur. In most workloads,
- // fragmentation should not be a major concern, since it doesn't impact
- // peak file descriptor usage or the speed of allocation.
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": shutting down UNIX domain socket for " +
- "empty " + shm);
- }
- shutdown(shm);
- } else {
- notFull.put(shmId, shm);
- }
- }
- }
-
- /**
- * Unregister a shared memory segment.
- *
- * Once a segment is unregistered, we will not allocate any more slots
- * inside that segment.
- *
- * The DomainSocketWatcher calls this while holding the DomainSocketWatcher
- * lock.
- *
- * @param shmId The ID of the shared memory segment to unregister.
- */
- void unregisterShm(ShmId shmId) {
- lock.lock();
- try {
- full.remove(shmId);
- notFull.remove(shmId);
- } finally {
- lock.unlock();
- }
- }
-
- @Override
- public String toString() {
- return String.format("EndpointShmManager(%s, parent=%s)",
- datanode, DfsClientShmManager.this);
- }
-
- PerDatanodeVisitorInfo getVisitorInfo() {
- return new PerDatanodeVisitorInfo(full, notFull, disabled);
- }
-
- final void shutdown(DfsClientShm shm) {
- try {
- shm.getPeer().getDomainSocket().shutdown();
- } catch (IOException e) {
- LOG.warn(this + ": error shutting down shm: got IOException calling " +
- "shutdown(SHUT_RDWR)", e);
- }
- }
- }
-
- private boolean closed = false;
-
- private final ReentrantLock lock = new ReentrantLock();
-
- /**
- * A condition variable which is signalled when we finish loading a segment
- * from the Datanode.
- */
- private final Condition finishedLoading = lock.newCondition();
-
- /**
- * Information about each Datanode.
- */
- private final HashMap<DatanodeInfo, EndpointShmManager> datanodes =
- new HashMap<DatanodeInfo, EndpointShmManager>(1);
-
- /**
- * The DomainSocketWatcher which keeps track of the UNIX domain socket
- * associated with each shared memory segment.
- *
- * Note: because the DomainSocketWatcher makes callbacks into this
- * DfsClientShmManager object, you must MUST NOT attempt to take the
- * DomainSocketWatcher lock while holding the DfsClientShmManager lock,
- * or else deadlock might result. This means that most DomainSocketWatcher
- * methods are off-limits unless you release the manager lock first.
- */
- private final DomainSocketWatcher domainSocketWatcher;
-
- DfsClientShmManager(int interruptCheckPeriodMs) throws IOException {
- this.domainSocketWatcher = new DomainSocketWatcher(interruptCheckPeriodMs,
- "client");
- }
-
- public Slot allocSlot(DatanodeInfo datanode, DomainPeer peer,
- MutableBoolean usedPeer, ExtendedBlockId blockId,
- String clientName) throws IOException {
- lock.lock();
- try {
- if (closed) {
- LOG.trace(this + ": the DfsClientShmManager isclosed.");
- return null;
- }
- EndpointShmManager shmManager = datanodes.get(datanode);
- if (shmManager == null) {
- shmManager = new EndpointShmManager(datanode);
- datanodes.put(datanode, shmManager);
- }
- return shmManager.allocSlot(peer, usedPeer, clientName, blockId);
- } finally {
- lock.unlock();
- }
- }
-
- public void freeSlot(Slot slot) {
- lock.lock();
- try {
- DfsClientShm shm = (DfsClientShm)slot.getShm();
- shm.getEndpointShmManager().freeSlot(slot);
- } finally {
- lock.unlock();
- }
- }
-
- @VisibleForTesting
- public static class PerDatanodeVisitorInfo {
- public final TreeMap<ShmId, DfsClientShm> full;
- public final TreeMap<ShmId, DfsClientShm> notFull;
- public final boolean disabled;
-
- PerDatanodeVisitorInfo(TreeMap<ShmId, DfsClientShm> full,
- TreeMap<ShmId, DfsClientShm> notFull, boolean disabled) {
- this.full = full;
- this.notFull = notFull;
- this.disabled = disabled;
- }
- }
-
- @VisibleForTesting
- public interface Visitor {
- void visit(HashMap<DatanodeInfo, PerDatanodeVisitorInfo> info)
- throws IOException;
- }
-
- @VisibleForTesting
- public void visit(Visitor visitor) throws IOException {
- lock.lock();
- try {
- HashMap<DatanodeInfo, PerDatanodeVisitorInfo> info =
- new HashMap<DatanodeInfo, PerDatanodeVisitorInfo>();
- for (Entry<DatanodeInfo, EndpointShmManager> entry :
- datanodes.entrySet()) {
- info.put(entry.getKey(), entry.getValue().getVisitorInfo());
- }
- visitor.visit(info);
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Close the DfsClientShmManager.
- */
- @Override
- public void close() throws IOException {
- lock.lock();
- try {
- if (closed) return;
- closed = true;
- } finally {
- lock.unlock();
- }
- // When closed, the domainSocketWatcher will issue callbacks that mark
- // all the outstanding DfsClientShm segments as stale.
- IOUtils.cleanup(LOG, domainSocketWatcher);
- }
-
-
- @Override
- public String toString() {
- return String.format("ShortCircuitShmManager(%08x)",
- System.identityHashCode(this));
- }
-
- @VisibleForTesting
- public DomainSocketWatcher getDomainSocketWatcher() {
- return domainSocketWatcher;
- }
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
index db4cbe2..15b8dea 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitCache.java
@@ -44,7 +44,7 @@ import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessResponseProto;
import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
-import org.apache.hadoop.hdfs.protocolPB.PBHelper;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ipc.RetriableException;
@@ -201,7 +201,7 @@ public class ShortCircuitCache implements Closeable {
DataInputStream in = new DataInputStream(sock.getInputStream());
ReleaseShortCircuitAccessResponseProto resp =
ReleaseShortCircuitAccessResponseProto.parseFrom(
- PBHelper.vintPrefixed(in));
+ PBHelperClient.vintPrefixed(in));
if (resp.getStatus() != Status.SUCCESS) {
String error = resp.hasError() ? resp.getError() : "(unknown)";
throw new IOException(resp.getStatus().toString() + ": " + error);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
deleted file mode 100644
index 7b89d0a..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/shortcircuit/ShortCircuitShm.java
+++ /dev/null
@@ -1,646 +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.hdfs.shortcircuit;
-
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.lang.reflect.Field;
-import java.util.BitSet;
-import java.util.Iterator;
-import java.util.NoSuchElementException;
-import java.util.Random;
-
-import org.apache.commons.lang.builder.EqualsBuilder;
-import org.apache.commons.lang.builder.HashCodeBuilder;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.fs.InvalidRequestException;
-import org.apache.hadoop.hdfs.ExtendedBlockId;
-import org.apache.hadoop.io.nativeio.NativeIO;
-import org.apache.hadoop.io.nativeio.NativeIO.POSIX;
-import org.apache.hadoop.util.Shell;
-import org.apache.hadoop.util.StringUtils;
-
-import sun.misc.Unsafe;
-
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ComparisonChain;
-import com.google.common.primitives.Ints;
-
-/**
- * A shared memory segment used to implement short-circuit reads.
- */
-public class ShortCircuitShm {
- private static final Log LOG = LogFactory.getLog(ShortCircuitShm.class);
-
- protected static final int BYTES_PER_SLOT = 64;
-
- private static final Unsafe unsafe = safetyDance();
-
- private static Unsafe safetyDance() {
- try {
- Field f = Unsafe.class.getDeclaredField("theUnsafe");
- f.setAccessible(true);
- return (Unsafe)f.get(null);
- } catch (Throwable e) {
- LOG.error("failed to load misc.Unsafe", e);
- }
- return null;
- }
-
- /**
- * Calculate the usable size of a shared memory segment.
- * We round down to a multiple of the slot size and do some validation.
- *
- * @param stream The stream we're using.
- * @return The usable size of the shared memory segment.
- */
- private static int getUsableLength(FileInputStream stream)
- throws IOException {
- int intSize = Ints.checkedCast(stream.getChannel().size());
- int slots = intSize / BYTES_PER_SLOT;
- if (slots == 0) {
- throw new IOException("size of shared memory segment was " +
- intSize + ", but that is not enough to hold even one slot.");
- }
- return slots * BYTES_PER_SLOT;
- }
-
- /**
- * Identifies a DfsClientShm.
- */
- public static class ShmId implements Comparable<ShmId> {
- private static final Random random = new Random();
- private final long hi;
- private final long lo;
-
- /**
- * Generate a random ShmId.
- *
- * We generate ShmIds randomly to prevent a malicious client from
- * successfully guessing one and using that to interfere with another
- * client.
- */
- public static ShmId createRandom() {
- return new ShmId(random.nextLong(), random.nextLong());
- }
-
- public ShmId(long hi, long lo) {
- this.hi = hi;
- this.lo = lo;
- }
-
- public long getHi() {
- return hi;
- }
-
- public long getLo() {
- return lo;
- }
-
- @Override
- public boolean equals(Object o) {
- if ((o == null) || (o.getClass() != this.getClass())) {
- return false;
- }
- ShmId other = (ShmId)o;
- return new EqualsBuilder().
- append(hi, other.hi).
- append(lo, other.lo).
- isEquals();
- }
-
- @Override
- public int hashCode() {
- return new HashCodeBuilder().
- append(this.hi).
- append(this.lo).
- toHashCode();
- }
-
- @Override
- public String toString() {
- return String.format("%016x%016x", hi, lo);
- }
-
- @Override
- public int compareTo(ShmId other) {
- return ComparisonChain.start().
- compare(hi, other.hi).
- compare(lo, other.lo).
- result();
- }
- };
-
- /**
- * Uniquely identifies a slot.
- */
- public static class SlotId {
- private final ShmId shmId;
- private final int slotIdx;
-
- public SlotId(ShmId shmId, int slotIdx) {
- this.shmId = shmId;
- this.slotIdx = slotIdx;
- }
-
- public ShmId getShmId() {
- return shmId;
- }
-
- public int getSlotIdx() {
- return slotIdx;
- }
-
- @Override
- public boolean equals(Object o) {
- if ((o == null) || (o.getClass() != this.getClass())) {
- return false;
- }
- SlotId other = (SlotId)o;
- return new EqualsBuilder().
- append(shmId, other.shmId).
- append(slotIdx, other.slotIdx).
- isEquals();
- }
-
- @Override
- public int hashCode() {
- return new HashCodeBuilder().
- append(this.shmId).
- append(this.slotIdx).
- toHashCode();
- }
-
- @Override
- public String toString() {
- return String.format("SlotId(%s:%d)", shmId.toString(), slotIdx);
- }
- }
-
- public class SlotIterator implements Iterator<Slot> {
- int slotIdx = -1;
-
- @Override
- public boolean hasNext() {
- synchronized (ShortCircuitShm.this) {
- return allocatedSlots.nextSetBit(slotIdx + 1) != -1;
- }
- }
-
- @Override
- public Slot next() {
- synchronized (ShortCircuitShm.this) {
- int nextSlotIdx = allocatedSlots.nextSetBit(slotIdx + 1);
- if (nextSlotIdx == -1) {
- throw new NoSuchElementException();
- }
- slotIdx = nextSlotIdx;
- return slots[nextSlotIdx];
- }
- }
-
- @Override
- public void remove() {
- throw new UnsupportedOperationException("SlotIterator " +
- "doesn't support removal");
- }
- }
-
- /**
- * A slot containing information about a replica.
- *
- * The format is:
- * word 0
- * bit 0:32 Slot flags (see below).
- * bit 33:63 Anchor count.
- * word 1:7
- * Reserved for future use, such as statistics.
- * Padding is also useful for avoiding false sharing.
- *
- * Little-endian versus big-endian is not relevant here since both the client
- * and the server reside on the same computer and use the same orientation.
- */
- public class Slot {
- /**
- * Flag indicating that the slot is valid.
- *
- * The DFSClient sets this flag when it allocates a new slot within one of
- * its shared memory regions.
- *
- * The DataNode clears this flag when the replica associated with this slot
- * is no longer valid. The client itself also clears this flag when it
- * believes that the DataNode is no longer using this slot to communicate.
- */
- private static final long VALID_FLAG = 1L<<63;
-
- /**
- * Flag indicating that the slot can be anchored.
- */
- private static final long ANCHORABLE_FLAG = 1L<<62;
-
- /**
- * The slot address in memory.
- */
- private final long slotAddress;
-
- /**
- * BlockId of the block this slot is used for.
- */
- private final ExtendedBlockId blockId;
-
- Slot(long slotAddress, ExtendedBlockId blockId) {
- this.slotAddress = slotAddress;
- this.blockId = blockId;
- }
-
- /**
- * Get the short-circuit memory segment associated with this Slot.
- *
- * @return The enclosing short-circuit memory segment.
- */
- public ShortCircuitShm getShm() {
- return ShortCircuitShm.this;
- }
-
- /**
- * Get the ExtendedBlockId associated with this slot.
- *
- * @return The ExtendedBlockId of this slot.
- */
- public ExtendedBlockId getBlockId() {
- return blockId;
- }
-
- /**
- * Get the SlotId of this slot, containing both shmId and slotIdx.
- *
- * @return The SlotId of this slot.
- */
- public SlotId getSlotId() {
- return new SlotId(getShmId(), getSlotIdx());
- }
-
- /**
- * Get the Slot index.
- *
- * @return The index of this slot.
- */
- public int getSlotIdx() {
- return Ints.checkedCast(
- (slotAddress - baseAddress) / BYTES_PER_SLOT);
- }
-
- /**
- * Clear the slot.
- */
- void clear() {
- unsafe.putLongVolatile(null, this.slotAddress, 0);
- }
-
- private boolean isSet(long flag) {
- long prev = unsafe.getLongVolatile(null, this.slotAddress);
- return (prev & flag) != 0;
- }
-
- private void setFlag(long flag) {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & flag) != 0) {
- return;
- }
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev | flag));
- }
-
- private void clearFlag(long flag) {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & flag) == 0) {
- return;
- }
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev & (~flag)));
- }
-
- public boolean isValid() {
- return isSet(VALID_FLAG);
- }
-
- public void makeValid() {
- setFlag(VALID_FLAG);
- }
-
- public void makeInvalid() {
- clearFlag(VALID_FLAG);
- }
-
- public boolean isAnchorable() {
- return isSet(ANCHORABLE_FLAG);
- }
-
- public void makeAnchorable() {
- setFlag(ANCHORABLE_FLAG);
- }
-
- public void makeUnanchorable() {
- clearFlag(ANCHORABLE_FLAG);
- }
-
- public boolean isAnchored() {
- long prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & VALID_FLAG) == 0) {
- // Slot is no longer valid.
- return false;
- }
- return ((prev & 0x7fffffff) != 0);
- }
-
- /**
- * Try to add an anchor for a given slot.
- *
- * When a slot is anchored, we know that the block it refers to is resident
- * in memory.
- *
- * @return True if the slot is anchored.
- */
- public boolean addAnchor() {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- if ((prev & VALID_FLAG) == 0) {
- // Slot is no longer valid.
- return false;
- }
- if ((prev & ANCHORABLE_FLAG) == 0) {
- // Slot can't be anchored right now.
- return false;
- }
- if ((prev & 0x7fffffff) == 0x7fffffff) {
- // Too many other threads have anchored the slot (2 billion?)
- return false;
- }
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev + 1));
- return true;
- }
-
- /**
- * Remove an anchor for a given slot.
- */
- public void removeAnchor() {
- long prev;
- do {
- prev = unsafe.getLongVolatile(null, this.slotAddress);
- Preconditions.checkState((prev & 0x7fffffff) != 0,
- "Tried to remove anchor for slot " + slotAddress +", which was " +
- "not anchored.");
- } while (!unsafe.compareAndSwapLong(null, this.slotAddress,
- prev, prev - 1));
- }
-
- @Override
- public String toString() {
- return "Slot(slotIdx=" + getSlotIdx() + ", shm=" + getShm() + ")";
- }
- }
-
- /**
- * ID for this SharedMemorySegment.
- */
- private final ShmId shmId;
-
- /**
- * The base address of the memory-mapped file.
- */
- private final long baseAddress;
-
- /**
- * The mmapped length of the shared memory segment
- */
- private final int mmappedLength;
-
- /**
- * The slots associated with this shared memory segment.
- * slot[i] contains the slot at offset i * BYTES_PER_SLOT,
- * or null if that slot is not allocated.
- */
- private final Slot slots[];
-
- /**
- * A bitset where each bit represents a slot which is in use.
- */
- private final BitSet allocatedSlots;
-
- /**
- * Create the ShortCircuitShm.
- *
- * @param shmId The ID to use.
- * @param stream The stream that we're going to use to create this
- * shared memory segment.
- *
- * Although this is a FileInputStream, we are going to
- * assume that the underlying file descriptor is writable
- * as well as readable. It would be more appropriate to use
- * a RandomAccessFile here, but that class does not have
- * any public accessor which returns a FileDescriptor,
- * unlike FileInputStream.
- */
- public ShortCircuitShm(ShmId shmId, FileInputStream stream)
- throws IOException {
- if (!NativeIO.isAvailable()) {
- throw new UnsupportedOperationException("NativeIO is not available.");
- }
- if (Shell.WINDOWS) {
- throw new UnsupportedOperationException(
- "DfsClientShm is not yet implemented for Windows.");
- }
- if (unsafe == null) {
- throw new UnsupportedOperationException(
- "can't use DfsClientShm because we failed to " +
- "load misc.Unsafe.");
- }
- this.shmId = shmId;
- this.mmappedLength = getUsableLength(stream);
- this.baseAddress = POSIX.mmap(stream.getFD(),
- POSIX.MMAP_PROT_READ | POSIX.MMAP_PROT_WRITE, true, mmappedLength);
- this.slots = new Slot[mmappedLength / BYTES_PER_SLOT];
- this.allocatedSlots = new BitSet(slots.length);
- if (LOG.isTraceEnabled()) {
- LOG.trace("creating " + this.getClass().getSimpleName() +
- "(shmId=" + shmId +
- ", mmappedLength=" + mmappedLength +
- ", baseAddress=" + String.format("%x", baseAddress) +
- ", slots.length=" + slots.length + ")");
- }
- }
-
- public final ShmId getShmId() {
- return shmId;
- }
-
- /**
- * Determine if this shared memory object is empty.
- *
- * @return True if the shared memory object is empty.
- */
- synchronized final public boolean isEmpty() {
- return allocatedSlots.nextSetBit(0) == -1;
- }
-
- /**
- * Determine if this shared memory object is full.
- *
- * @return True if the shared memory object is full.
- */
- synchronized final public boolean isFull() {
- return allocatedSlots.nextClearBit(0) >= slots.length;
- }
-
- /**
- * Calculate the base address of a slot.
- *
- * @param slotIdx Index of the slot.
- * @return The base address of the slot.
- */
- private final long calculateSlotAddress(int slotIdx) {
- long offset = slotIdx;
- offset *= BYTES_PER_SLOT;
- return this.baseAddress + offset;
- }
-
- /**
- * Allocate a new slot and register it.
- *
- * This function chooses an empty slot, initializes it, and then returns
- * the relevant Slot object.
- *
- * @return The new slot.
- */
- synchronized public final Slot allocAndRegisterSlot(
- ExtendedBlockId blockId) {
- int idx = allocatedSlots.nextClearBit(0);
- if (idx >= slots.length) {
- throw new RuntimeException(this + ": no more slots are available.");
- }
- allocatedSlots.set(idx, true);
- Slot slot = new Slot(calculateSlotAddress(idx), blockId);
- slot.clear();
- slot.makeValid();
- slots[idx] = slot;
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": allocAndRegisterSlot " + idx + ": allocatedSlots=" + allocatedSlots +
- StringUtils.getStackTrace(Thread.currentThread()));
- }
- return slot;
- }
-
- synchronized public final Slot getSlot(int slotIdx)
- throws InvalidRequestException {
- if (!allocatedSlots.get(slotIdx)) {
- throw new InvalidRequestException(this + ": slot " + slotIdx +
- " does not exist.");
- }
- return slots[slotIdx];
- }
-
- /**
- * Register a slot.
- *
- * This function looks at a slot which has already been initialized (by
- * another process), and registers it with us. Then, it returns the
- * relevant Slot object.
- *
- * @return The slot.
- *
- * @throws InvalidRequestException
- * If the slot index we're trying to allocate has not been
- * initialized, or is already in use.
- */
- synchronized public final Slot registerSlot(int slotIdx,
- ExtendedBlockId blockId) throws InvalidRequestException {
- if (slotIdx < 0) {
- throw new InvalidRequestException(this + ": invalid negative slot " +
- "index " + slotIdx);
- }
- if (slotIdx >= slots.length) {
- throw new InvalidRequestException(this + ": invalid slot " +
- "index " + slotIdx);
- }
- if (allocatedSlots.get(slotIdx)) {
- throw new InvalidRequestException(this + ": slot " + slotIdx +
- " is already in use.");
- }
- Slot slot = new Slot(calculateSlotAddress(slotIdx), blockId);
- if (!slot.isValid()) {
- throw new InvalidRequestException(this + ": slot " + slotIdx +
- " is not marked as valid.");
- }
- slots[slotIdx] = slot;
- allocatedSlots.set(slotIdx, true);
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": registerSlot " + slotIdx + ": allocatedSlots=" + allocatedSlots +
- StringUtils.getStackTrace(Thread.currentThread()));
- }
- return slot;
- }
-
- /**
- * Unregisters a slot.
- *
- * This doesn't alter the contents of the slot. It just means
- *
- * @param slotIdx Index of the slot to unregister.
- */
- synchronized public final void unregisterSlot(int slotIdx) {
- Preconditions.checkState(allocatedSlots.get(slotIdx),
- "tried to unregister slot " + slotIdx + ", which was not registered.");
- allocatedSlots.set(slotIdx, false);
- slots[slotIdx] = null;
- if (LOG.isTraceEnabled()) {
- LOG.trace(this + ": unregisterSlot " + slotIdx);
- }
- }
-
- /**
- * Iterate over all allocated slots.
- *
- * Note that this method isn't safe if
- *
- * @return The slot iterator.
- */
- public SlotIterator slotIterator() {
- return new SlotIterator();
- }
-
- public void free() {
- try {
- POSIX.munmap(baseAddress, mmappedLength);
- } catch (IOException e) {
- LOG.warn(this + ": failed to munmap", e);
- }
- LOG.trace(this + ": freed");
- }
-
- @Override
- public String toString() {
- return this.getClass().getSimpleName() + "(" + shmId + ")";
- }
-}
[4/4] hadoop git commit: HDFS-8934. Move ShortCircuitShm to
hdfs-client. Contributed by Mingliang Liu.
Posted by wh...@apache.org.
HDFS-8934. Move ShortCircuitShm to hdfs-client. Contributed by Mingliang Liu.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/8e4afa3a
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/8e4afa3a
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/8e4afa3a
Branch: refs/heads/branch-2
Commit: 8e4afa3a671583c95263218b85cf6bfbc1e43635
Parents: 8c20526
Author: Haohui Mai <wh...@apache.org>
Authored: Sat Aug 22 13:30:19 2015 -0700
Committer: Haohui Mai <wh...@apache.org>
Committed: Sat Aug 22 13:30:48 2015 -0700
----------------------------------------------------------------------
.../org/apache/hadoop/hdfs/ExtendedBlockId.java | 82 +++
.../org/apache/hadoop/hdfs/net/DomainPeer.java | 132 ++++
.../java/org/apache/hadoop/hdfs/net/Peer.java | 123 ++++
.../datatransfer/BlockConstructionStage.java | 62 ++
.../datatransfer/DataTransferProtoUtil.java | 146 +++++
.../datatransfer/DataTransferProtocol.java | 202 ++++++
.../hadoop/hdfs/protocol/datatransfer/Op.java | 66 ++
.../hdfs/protocol/datatransfer/Sender.java | 261 ++++++++
.../hadoop/hdfs/protocolPB/PBHelperClient.java | 254 ++++++++
.../token/block/InvalidBlockTokenException.java | 41 ++
.../hdfs/server/datanode/CachingStrategy.java | 76 +++
.../hadoop/hdfs/shortcircuit/DfsClientShm.java | 119 ++++
.../hdfs/shortcircuit/DfsClientShmManager.java | 522 +++++++++++++++
.../hdfs/shortcircuit/ShortCircuitShm.java | 647 +++++++++++++++++++
.../hadoop/hdfs/util/ExactSizeInputStream.java | 125 ++++
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt | 2 +
.../apache/hadoop/hdfs/BlockReaderFactory.java | 4 +-
.../java/org/apache/hadoop/hdfs/DFSClient.java | 10 +-
.../org/apache/hadoop/hdfs/DataStreamer.java | 6 +-
.../org/apache/hadoop/hdfs/ExtendedBlockId.java | 82 ---
.../apache/hadoop/hdfs/RemoteBlockReader.java | 4 +-
.../apache/hadoop/hdfs/RemoteBlockReader2.java | 4 +-
.../org/apache/hadoop/hdfs/net/DomainPeer.java | 132 ----
.../java/org/apache/hadoop/hdfs/net/Peer.java | 123 ----
.../datatransfer/BlockConstructionStage.java | 62 --
.../datatransfer/DataTransferProtoUtil.java | 148 -----
.../datatransfer/DataTransferProtocol.java | 201 ------
.../hadoop/hdfs/protocol/datatransfer/Op.java | 66 --
.../hdfs/protocol/datatransfer/PipelineAck.java | 2 +-
.../hdfs/protocol/datatransfer/Receiver.java | 7 +-
.../hdfs/protocol/datatransfer/Sender.java | 261 --------
.../datatransfer/sasl/DataTransferSaslUtil.java | 2 +-
...tDatanodeProtocolServerSideTranslatorPB.java | 2 +-
.../ClientDatanodeProtocolTranslatorPB.java | 6 +-
...tNamenodeProtocolServerSideTranslatorPB.java | 6 +-
.../ClientNamenodeProtocolTranslatorPB.java | 28 +-
.../DatanodeProtocolClientSideTranslatorPB.java | 4 +-
.../InterDatanodeProtocolTranslatorPB.java | 2 +-
.../NamenodeProtocolTranslatorPB.java | 2 +-
.../apache/hadoop/hdfs/protocolPB/PBHelper.java | 228 +------
.../token/block/InvalidBlockTokenException.java | 41 --
.../hadoop/hdfs/server/balancer/Dispatcher.java | 2 +-
.../hdfs/server/datanode/CachingStrategy.java | 76 ---
.../hadoop/hdfs/server/datanode/DataNode.java | 4 +-
.../hdfs/server/datanode/DataXceiver.java | 14 +-
.../server/namenode/FSImageFormatPBINode.java | 5 +-
.../hadoop/hdfs/shortcircuit/DfsClientShm.java | 119 ----
.../hdfs/shortcircuit/DfsClientShmManager.java | 514 ---------------
.../hdfs/shortcircuit/ShortCircuitCache.java | 4 +-
.../hdfs/shortcircuit/ShortCircuitShm.java | 646 ------------------
.../hadoop/hdfs/util/ExactSizeInputStream.java | 125 ----
.../hadoop/hdfs/protocolPB/TestPBHelper.java | 20 +-
52 files changed, 2949 insertions(+), 2873 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java
new file mode 100644
index 0000000..7b9e8e3
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/ExtendedBlockId.java
@@ -0,0 +1,82 @@
+/**
+ * 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.hdfs;
+
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+
+/**
+ * An immutable key which identifies a block.
+ */
+@InterfaceAudience.Private
+final public class ExtendedBlockId {
+ /**
+ * The block ID for this block.
+ */
+ private final long blockId;
+
+ /**
+ * The block pool ID for this block.
+ */
+ private final String bpId;
+
+ public static ExtendedBlockId fromExtendedBlock(ExtendedBlock block) {
+ return new ExtendedBlockId(block.getBlockId(), block.getBlockPoolId());
+ }
+
+ public ExtendedBlockId(long blockId, String bpId) {
+ this.blockId = blockId;
+ this.bpId = bpId;
+ }
+
+ public long getBlockId() {
+ return this.blockId;
+ }
+
+ public String getBlockPoolId() {
+ return this.bpId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if ((o == null) || (o.getClass() != this.getClass())) {
+ return false;
+ }
+ ExtendedBlockId other = (ExtendedBlockId)o;
+ return new EqualsBuilder().
+ append(blockId, other.blockId).
+ append(bpId, other.bpId).
+ isEquals();
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder().
+ append(this.blockId).
+ append(this.bpId).
+ toHashCode();
+ }
+
+ @Override
+ public String toString() {
+ return new StringBuilder().append(blockId).
+ append("_").append(bpId).toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java
new file mode 100644
index 0000000..4792b0e
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/DomainPeer.java
@@ -0,0 +1,132 @@
+/**
+ * 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.hdfs.net;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.ReadableByteChannel;
+
+import org.apache.hadoop.net.unix.DomainSocket;
+import org.apache.hadoop.classification.InterfaceAudience;
+
+/**
+ * Represents a peer that we communicate with by using blocking I/O
+ * on a UNIX domain socket.
+ */
+@InterfaceAudience.Private
+public class DomainPeer implements Peer {
+ private final DomainSocket socket;
+ private final OutputStream out;
+ private final InputStream in;
+ private final ReadableByteChannel channel;
+
+ public DomainPeer(DomainSocket socket) {
+ this.socket = socket;
+ this.out = socket.getOutputStream();
+ this.in = socket.getInputStream();
+ this.channel = socket.getChannel();
+ }
+
+ @Override
+ public ReadableByteChannel getInputStreamChannel() {
+ return channel;
+ }
+
+ @Override
+ public void setReadTimeout(int timeoutMs) throws IOException {
+ socket.setAttribute(DomainSocket.RECEIVE_TIMEOUT, timeoutMs);
+ }
+
+ @Override
+ public int getReceiveBufferSize() throws IOException {
+ return socket.getAttribute(DomainSocket.RECEIVE_BUFFER_SIZE);
+ }
+
+ @Override
+ public boolean getTcpNoDelay() throws IOException {
+ /* No TCP, no TCP_NODELAY. */
+ return false;
+ }
+
+ @Override
+ public void setWriteTimeout(int timeoutMs) throws IOException {
+ socket.setAttribute(DomainSocket.SEND_TIMEOUT, timeoutMs);
+ }
+
+ @Override
+ public boolean isClosed() {
+ return !socket.isOpen();
+ }
+
+ @Override
+ public void close() throws IOException {
+ socket.close();
+ }
+
+ @Override
+ public String getRemoteAddressString() {
+ return "unix:" + socket.getPath();
+ }
+
+ @Override
+ public String getLocalAddressString() {
+ return "<local>";
+ }
+
+ @Override
+ public InputStream getInputStream() throws IOException {
+ return in;
+ }
+
+ @Override
+ public OutputStream getOutputStream() throws IOException {
+ return out;
+ }
+
+ @Override
+ public boolean isLocal() {
+ /* UNIX domain sockets can only be used for local communication. */
+ return true;
+ }
+
+ @Override
+ public String toString() {
+ return "DomainPeer(" + getRemoteAddressString() + ")";
+ }
+
+ @Override
+ public DomainSocket getDomainSocket() {
+ return socket;
+ }
+
+ @Override
+ public boolean hasSecureChannel() {
+ //
+ // Communication over domain sockets is assumed to be secure, since it
+ // doesn't pass over any network. We also carefully control the privileges
+ // that can be used on the domain socket inode and its parent directories.
+ // See #{java.org.apache.hadoop.net.unix.DomainSocket#validateSocketPathSecurity0}
+ // for details.
+ //
+ // So unless you are running as root or the hdfs superuser, you cannot
+ // launch a man-in-the-middle attach on UNIX domain socket traffic.
+ //
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/Peer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/Peer.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/Peer.java
new file mode 100644
index 0000000..42cf287
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/net/Peer.java
@@ -0,0 +1,123 @@
+/**
+ * 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.hdfs.net;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.channels.ReadableByteChannel;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.net.unix.DomainSocket;
+
+/**
+ * Represents a connection to a peer.
+ */
+@InterfaceAudience.Private
+public interface Peer extends Closeable {
+ /**
+ * @return The input stream channel associated with this
+ * peer, or null if it has none.
+ */
+ public ReadableByteChannel getInputStreamChannel();
+
+ /**
+ * Set the read timeout on this peer.
+ *
+ * @param timeoutMs The timeout in milliseconds.
+ */
+ public void setReadTimeout(int timeoutMs) throws IOException;
+
+ /**
+ * @return The receive buffer size.
+ */
+ public int getReceiveBufferSize() throws IOException;
+
+ /**
+ * @return True if TCP_NODELAY is turned on.
+ */
+ public boolean getTcpNoDelay() throws IOException;
+
+ /**
+ * Set the write timeout on this peer.
+ *
+ * Note: this is not honored for BasicInetPeer.
+ * See {@link BasicSocketPeer#setWriteTimeout} for details.
+ *
+ * @param timeoutMs The timeout in milliseconds.
+ */
+ public void setWriteTimeout(int timeoutMs) throws IOException;
+
+ /**
+ * @return true only if the peer is closed.
+ */
+ public boolean isClosed();
+
+ /**
+ * Close the peer.
+ *
+ * It's safe to re-close a Peer that is already closed.
+ */
+ public void close() throws IOException;
+
+ /**
+ * @return A string representing the remote end of our
+ * connection to the peer.
+ */
+ public String getRemoteAddressString();
+
+ /**
+ * @return A string representing the local end of our
+ * connection to the peer.
+ */
+ public String getLocalAddressString();
+
+ /**
+ * @return An InputStream associated with the Peer.
+ * This InputStream will be valid until you close
+ * this peer with Peer#close.
+ */
+ public InputStream getInputStream() throws IOException;
+
+ /**
+ * @return An OutputStream associated with the Peer.
+ * This OutputStream will be valid until you close
+ * this peer with Peer#close.
+ */
+ public OutputStream getOutputStream() throws IOException;
+
+ /**
+ * @return True if the peer resides on the same
+ * computer as we.
+ */
+ public boolean isLocal();
+
+ /**
+ * @return The DomainSocket associated with the current
+ * peer, or null if there is none.
+ */
+ public DomainSocket getDomainSocket();
+
+ /**
+ * Return true if the channel is secure.
+ *
+ * @return True if our channel to this peer is not
+ * susceptible to man-in-the-middle attacks.
+ */
+ public boolean hasSecureChannel();
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java
new file mode 100644
index 0000000..5f86e52
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/BlockConstructionStage.java
@@ -0,0 +1,62 @@
+/**
+ * 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.hdfs.protocol.datatransfer;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/** Block Construction Stage */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public enum BlockConstructionStage {
+ /** The enumerates are always listed as regular stage followed by the
+ * recovery stage.
+ * Changing this order will make getRecoveryStage not working.
+ */
+ // pipeline set up for block append
+ PIPELINE_SETUP_APPEND,
+ // pipeline set up for failed PIPELINE_SETUP_APPEND recovery
+ PIPELINE_SETUP_APPEND_RECOVERY,
+ // data streaming
+ DATA_STREAMING,
+ // pipeline setup for failed data streaming recovery
+ PIPELINE_SETUP_STREAMING_RECOVERY,
+ // close the block and pipeline
+ PIPELINE_CLOSE,
+ // Recover a failed PIPELINE_CLOSE
+ PIPELINE_CLOSE_RECOVERY,
+ // pipeline set up for block creation
+ PIPELINE_SETUP_CREATE,
+ // transfer RBW for adding datanodes
+ TRANSFER_RBW,
+ // transfer Finalized for adding datanodes
+ TRANSFER_FINALIZED;
+
+ final static private byte RECOVERY_BIT = (byte)1;
+
+ /**
+ * get the recovery stage of this stage
+ */
+ public BlockConstructionStage getRecoveryStage() {
+ if (this == PIPELINE_SETUP_CREATE) {
+ throw new IllegalArgumentException( "Unexpected blockStage " + this);
+ } else {
+ return values()[ordinal()|RECOVERY_BIT];
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java
new file mode 100644
index 0000000..28097ab
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtoUtil.java
@@ -0,0 +1,146 @@
+/**
+ * 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.hdfs.protocol.datatransfer;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BaseHeaderProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.BlockOpResponseProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.Status;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ChecksumProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientOperationHeaderProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpWriteBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.DataTransferTraceInfoProto;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.ChecksumTypeProto;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
+import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
+import org.apache.hadoop.hdfs.security.token.block.InvalidBlockTokenException;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.DataChecksum;
+import org.apache.htrace.Span;
+import org.apache.htrace.Trace;
+import org.apache.htrace.TraceInfo;
+import org.apache.htrace.TraceScope;
+
+/**
+ * Static utilities for dealing with the protocol buffers used by the
+ * Data Transfer Protocol.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public abstract class DataTransferProtoUtil {
+ static BlockConstructionStage fromProto(
+ OpWriteBlockProto.BlockConstructionStage stage) {
+ return BlockConstructionStage.valueOf(stage.name());
+ }
+
+ static OpWriteBlockProto.BlockConstructionStage toProto(
+ BlockConstructionStage stage) {
+ return OpWriteBlockProto.BlockConstructionStage.valueOf(stage.name());
+ }
+
+ public static ChecksumProto toProto(DataChecksum checksum) {
+ ChecksumTypeProto type = PBHelperClient.convert(checksum.getChecksumType());
+ // ChecksumType#valueOf never returns null
+ return ChecksumProto.newBuilder()
+ .setBytesPerChecksum(checksum.getBytesPerChecksum())
+ .setType(type)
+ .build();
+ }
+
+ public static DataChecksum fromProto(ChecksumProto proto) {
+ if (proto == null) return null;
+
+ int bytesPerChecksum = proto.getBytesPerChecksum();
+ DataChecksum.Type type = PBHelperClient.convert(proto.getType());
+ return DataChecksum.newDataChecksum(type, bytesPerChecksum);
+ }
+
+ static ClientOperationHeaderProto buildClientHeader(ExtendedBlock blk,
+ String client, Token<BlockTokenIdentifier> blockToken) {
+ ClientOperationHeaderProto header =
+ ClientOperationHeaderProto.newBuilder()
+ .setBaseHeader(buildBaseHeader(blk, blockToken))
+ .setClientName(client)
+ .build();
+ return header;
+ }
+
+ static BaseHeaderProto buildBaseHeader(ExtendedBlock blk,
+ Token<BlockTokenIdentifier> blockToken) {
+ BaseHeaderProto.Builder builder = BaseHeaderProto.newBuilder()
+ .setBlock(PBHelperClient.convert(blk))
+ .setToken(PBHelperClient.convert(blockToken));
+ if (Trace.isTracing()) {
+ Span s = Trace.currentSpan();
+ builder.setTraceInfo(DataTransferTraceInfoProto.newBuilder()
+ .setTraceId(s.getTraceId())
+ .setParentId(s.getSpanId()));
+ }
+ return builder.build();
+ }
+
+ public static TraceInfo fromProto(DataTransferTraceInfoProto proto) {
+ if (proto == null) return null;
+ if (!proto.hasTraceId()) return null;
+ return new TraceInfo(proto.getTraceId(), proto.getParentId());
+ }
+
+ public static TraceScope continueTraceSpan(ClientOperationHeaderProto header,
+ String description) {
+ return continueTraceSpan(header.getBaseHeader(), description);
+ }
+
+ public static TraceScope continueTraceSpan(BaseHeaderProto header,
+ String description) {
+ return continueTraceSpan(header.getTraceInfo(), description);
+ }
+
+ public static TraceScope continueTraceSpan(DataTransferTraceInfoProto proto,
+ String description) {
+ TraceScope scope = null;
+ TraceInfo info = fromProto(proto);
+ if (info != null) {
+ scope = Trace.startSpan(description, info);
+ }
+ return scope;
+ }
+
+ public static void checkBlockOpStatus(
+ BlockOpResponseProto response,
+ String logInfo) throws IOException {
+ if (response.getStatus() != Status.SUCCESS) {
+ if (response.getStatus() == Status.ERROR_ACCESS_TOKEN) {
+ throw new InvalidBlockTokenException(
+ "Got access token error"
+ + ", status message " + response.getMessage()
+ + ", " + logInfo
+ );
+ } else {
+ throw new IOException(
+ "Got error"
+ + ", status message " + response.getMessage()
+ + ", " + logInfo
+ );
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java
new file mode 100644
index 0000000..1f7e378
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/DataTransferProtocol.java
@@ -0,0 +1,202 @@
+/**
+ * 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.hdfs.protocol.datatransfer;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
+import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
+import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.SlotId;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.DataChecksum;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Transfer data to/from datanode using a streaming protocol.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public interface DataTransferProtocol {
+ public static final Logger LOG = LoggerFactory.getLogger(DataTransferProtocol.class);
+
+ /** Version for data transfers between clients and datanodes
+ * This should change when serialization of DatanodeInfo, not just
+ * when protocol changes. It is not very obvious.
+ */
+ /*
+ * Version 28:
+ * Declare methods in DataTransferProtocol interface.
+ */
+ public static final int DATA_TRANSFER_VERSION = 28;
+
+ /**
+ * Read a block.
+ *
+ * @param blk the block being read.
+ * @param blockToken security token for accessing the block.
+ * @param clientName client's name.
+ * @param blockOffset offset of the block.
+ * @param length maximum number of bytes for this read.
+ * @param sendChecksum if false, the DN should skip reading and sending
+ * checksums
+ * @param cachingStrategy The caching strategy to use.
+ */
+ public void readBlock(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String clientName,
+ final long blockOffset,
+ final long length,
+ final boolean sendChecksum,
+ final CachingStrategy cachingStrategy) throws IOException;
+
+ /**
+ * Write a block to a datanode pipeline.
+ * The receiver datanode of this call is the next datanode in the pipeline.
+ * The other downstream datanodes are specified by the targets parameter.
+ * Note that the receiver {@link DatanodeInfo} is not required in the
+ * parameter list since the receiver datanode knows its info. However, the
+ * {@link StorageType} for storing the replica in the receiver datanode is a
+ * parameter since the receiver datanode may support multiple storage types.
+ *
+ * @param blk the block being written.
+ * @param storageType for storing the replica in the receiver datanode.
+ * @param blockToken security token for accessing the block.
+ * @param clientName client's name.
+ * @param targets other downstream datanodes in the pipeline.
+ * @param targetStorageTypes target {@link StorageType}s corresponding
+ * to the target datanodes.
+ * @param source source datanode.
+ * @param stage pipeline stage.
+ * @param pipelineSize the size of the pipeline.
+ * @param minBytesRcvd minimum number of bytes received.
+ * @param maxBytesRcvd maximum number of bytes received.
+ * @param latestGenerationStamp the latest generation stamp of the block.
+ * @param pinning whether to pin the block, so Balancer won't move it.
+ * @param targetPinnings whether to pin the block on target datanode
+ */
+ public void writeBlock(final ExtendedBlock blk,
+ final StorageType storageType,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String clientName,
+ final DatanodeInfo[] targets,
+ final StorageType[] targetStorageTypes,
+ final DatanodeInfo source,
+ final BlockConstructionStage stage,
+ final int pipelineSize,
+ final long minBytesRcvd,
+ final long maxBytesRcvd,
+ final long latestGenerationStamp,
+ final DataChecksum requestedChecksum,
+ final CachingStrategy cachingStrategy,
+ final boolean allowLazyPersist,
+ final boolean pinning,
+ final boolean[] targetPinnings) throws IOException;
+ /**
+ * Transfer a block to another datanode.
+ * The block stage must be
+ * either {@link BlockConstructionStage#TRANSFER_RBW}
+ * or {@link BlockConstructionStage#TRANSFER_FINALIZED}.
+ *
+ * @param blk the block being transferred.
+ * @param blockToken security token for accessing the block.
+ * @param clientName client's name.
+ * @param targets target datanodes.
+ */
+ public void transferBlock(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String clientName,
+ final DatanodeInfo[] targets,
+ final StorageType[] targetStorageTypes) throws IOException;
+
+ /**
+ * Request short circuit access file descriptors from a DataNode.
+ *
+ * @param blk The block to get file descriptors for.
+ * @param blockToken Security token for accessing the block.
+ * @param slotId The shared memory slot id to use, or null
+ * to use no slot id.
+ * @param maxVersion Maximum version of the block data the client
+ * can understand.
+ * @param supportsReceiptVerification True if the client supports
+ * receipt verification.
+ */
+ public void requestShortCircuitFds(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken,
+ SlotId slotId, int maxVersion, boolean supportsReceiptVerification)
+ throws IOException;
+
+ /**
+ * Release a pair of short-circuit FDs requested earlier.
+ *
+ * @param slotId SlotID used by the earlier file descriptors.
+ */
+ public void releaseShortCircuitFds(final SlotId slotId) throws IOException;
+
+ /**
+ * Request a short circuit shared memory area from a DataNode.
+ *
+ * @param clientName The name of the client.
+ */
+ public void requestShortCircuitShm(String clientName) throws IOException;
+
+ /**
+ * Receive a block from a source datanode
+ * and then notifies the namenode
+ * to remove the copy from the original datanode.
+ * Note that the source datanode and the original datanode can be different.
+ * It is used for balancing purpose.
+ *
+ * @param blk the block being replaced.
+ * @param storageType the {@link StorageType} for storing the block.
+ * @param blockToken security token for accessing the block.
+ * @param delHint the hint for deleting the block in the original datanode.
+ * @param source the source datanode for receiving the block.
+ */
+ public void replaceBlock(final ExtendedBlock blk,
+ final StorageType storageType,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String delHint,
+ final DatanodeInfo source) throws IOException;
+
+ /**
+ * Copy a block.
+ * It is used for balancing purpose.
+ *
+ * @param blk the block being copied.
+ * @param blockToken security token for accessing the block.
+ */
+ public void copyBlock(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken) throws IOException;
+
+ /**
+ * Get block checksum (MD5 of CRC32).
+ *
+ * @param blk a block.
+ * @param blockToken security token for accessing the block.
+ * @throws IOException
+ */
+ public void blockChecksum(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken) throws IOException;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java
new file mode 100644
index 0000000..3077498
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Op.java
@@ -0,0 +1,66 @@
+/**
+ * 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.hdfs.protocol.datatransfer;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/** Operation */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public enum Op {
+ WRITE_BLOCK((byte)80),
+ READ_BLOCK((byte)81),
+ READ_METADATA((byte)82),
+ REPLACE_BLOCK((byte)83),
+ COPY_BLOCK((byte)84),
+ BLOCK_CHECKSUM((byte)85),
+ TRANSFER_BLOCK((byte)86),
+ REQUEST_SHORT_CIRCUIT_FDS((byte)87),
+ RELEASE_SHORT_CIRCUIT_FDS((byte)88),
+ REQUEST_SHORT_CIRCUIT_SHM((byte)89),
+ CUSTOM((byte)127);
+
+ /** The code for this operation. */
+ public final byte code;
+
+ private Op(byte code) {
+ this.code = code;
+ }
+
+ private static final int FIRST_CODE = values()[0].code;
+ /** Return the object represented by the code. */
+ private static Op valueOf(byte code) {
+ final int i = (code & 0xff) - FIRST_CODE;
+ return i < 0 || i >= values().length? null: values()[i];
+ }
+
+ /** Read from in */
+ public static Op read(DataInput in) throws IOException {
+ return valueOf(in.readByte());
+ }
+
+ /** Write to out */
+ public void write(DataOutput out) throws IOException {
+ out.write(code);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java
new file mode 100644
index 0000000..2d11dc2
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/datatransfer/Sender.java
@@ -0,0 +1,261 @@
+/**
+ * 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.hdfs.protocol.datatransfer;
+
+import static org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtoUtil.toProto;
+
+import java.io.DataOutput;
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.CachingStrategyProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ChecksumProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ClientOperationHeaderProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.DataTransferTraceInfoProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpBlockChecksumProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpCopyBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpReadBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpReplaceBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpRequestShortCircuitAccessProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpTransferBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.OpWriteBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ReleaseShortCircuitAccessRequestProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmRequestProto;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
+import org.apache.hadoop.hdfs.security.token.block.BlockTokenIdentifier;
+import org.apache.hadoop.hdfs.server.datanode.CachingStrategy;
+import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.SlotId;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.DataChecksum;
+
+import org.apache.htrace.Trace;
+import org.apache.htrace.Span;
+
+import com.google.protobuf.Message;
+
+/** Sender */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class Sender implements DataTransferProtocol {
+ private final DataOutputStream out;
+
+ /** Create a sender for DataTransferProtocol with a output stream. */
+ public Sender(final DataOutputStream out) {
+ this.out = out;
+ }
+
+ /** Initialize a operation. */
+ private static void op(final DataOutput out, final Op op
+ ) throws IOException {
+ out.writeShort(DataTransferProtocol.DATA_TRANSFER_VERSION);
+ op.write(out);
+ }
+
+ private static void send(final DataOutputStream out, final Op opcode,
+ final Message proto) throws IOException {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Sending DataTransferOp " + proto.getClass().getSimpleName()
+ + ": " + proto);
+ }
+ op(out, opcode);
+ proto.writeDelimitedTo(out);
+ out.flush();
+ }
+
+ static private CachingStrategyProto getCachingStrategy(CachingStrategy cachingStrategy) {
+ CachingStrategyProto.Builder builder = CachingStrategyProto.newBuilder();
+ if (cachingStrategy.getReadahead() != null) {
+ builder.setReadahead(cachingStrategy.getReadahead().longValue());
+ }
+ if (cachingStrategy.getDropBehind() != null) {
+ builder.setDropBehind(cachingStrategy.getDropBehind().booleanValue());
+ }
+ return builder.build();
+ }
+
+ @Override
+ public void readBlock(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String clientName,
+ final long blockOffset,
+ final long length,
+ final boolean sendChecksum,
+ final CachingStrategy cachingStrategy) throws IOException {
+
+ OpReadBlockProto proto = OpReadBlockProto.newBuilder()
+ .setHeader(DataTransferProtoUtil.buildClientHeader(blk, clientName, blockToken))
+ .setOffset(blockOffset)
+ .setLen(length)
+ .setSendChecksums(sendChecksum)
+ .setCachingStrategy(getCachingStrategy(cachingStrategy))
+ .build();
+
+ send(out, Op.READ_BLOCK, proto);
+ }
+
+
+ @Override
+ public void writeBlock(final ExtendedBlock blk,
+ final StorageType storageType,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String clientName,
+ final DatanodeInfo[] targets,
+ final StorageType[] targetStorageTypes,
+ final DatanodeInfo source,
+ final BlockConstructionStage stage,
+ final int pipelineSize,
+ final long minBytesRcvd,
+ final long maxBytesRcvd,
+ final long latestGenerationStamp,
+ DataChecksum requestedChecksum,
+ final CachingStrategy cachingStrategy,
+ final boolean allowLazyPersist,
+ final boolean pinning,
+ final boolean[] targetPinnings) throws IOException {
+ ClientOperationHeaderProto header = DataTransferProtoUtil.buildClientHeader(
+ blk, clientName, blockToken);
+
+ ChecksumProto checksumProto =
+ DataTransferProtoUtil.toProto(requestedChecksum);
+
+ OpWriteBlockProto.Builder proto = OpWriteBlockProto.newBuilder()
+ .setHeader(header)
+ .setStorageType(PBHelperClient.convertStorageType(storageType))
+ .addAllTargets(PBHelperClient.convert(targets, 1))
+ .addAllTargetStorageTypes(PBHelperClient.convertStorageTypes(targetStorageTypes, 1))
+ .setStage(toProto(stage))
+ .setPipelineSize(pipelineSize)
+ .setMinBytesRcvd(minBytesRcvd)
+ .setMaxBytesRcvd(maxBytesRcvd)
+ .setLatestGenerationStamp(latestGenerationStamp)
+ .setRequestedChecksum(checksumProto)
+ .setCachingStrategy(getCachingStrategy(cachingStrategy))
+ .setAllowLazyPersist(allowLazyPersist)
+ .setPinning(pinning)
+ .addAllTargetPinnings(PBHelperClient.convert(targetPinnings, 1));
+
+ if (source != null) {
+ proto.setSource(PBHelperClient.convertDatanodeInfo(source));
+ }
+
+ send(out, Op.WRITE_BLOCK, proto.build());
+ }
+
+ @Override
+ public void transferBlock(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String clientName,
+ final DatanodeInfo[] targets,
+ final StorageType[] targetStorageTypes) throws IOException {
+
+ OpTransferBlockProto proto = OpTransferBlockProto.newBuilder()
+ .setHeader(DataTransferProtoUtil.buildClientHeader(
+ blk, clientName, blockToken))
+ .addAllTargets(PBHelperClient.convert(targets))
+ .addAllTargetStorageTypes(PBHelperClient.convertStorageTypes(targetStorageTypes))
+ .build();
+
+ send(out, Op.TRANSFER_BLOCK, proto);
+ }
+
+ @Override
+ public void requestShortCircuitFds(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken,
+ SlotId slotId, int maxVersion, boolean supportsReceiptVerification)
+ throws IOException {
+ OpRequestShortCircuitAccessProto.Builder builder =
+ OpRequestShortCircuitAccessProto.newBuilder()
+ .setHeader(DataTransferProtoUtil.buildBaseHeader(
+ blk, blockToken)).setMaxVersion(maxVersion);
+ if (slotId != null) {
+ builder.setSlotId(PBHelperClient.convert(slotId));
+ }
+ builder.setSupportsReceiptVerification(supportsReceiptVerification);
+ OpRequestShortCircuitAccessProto proto = builder.build();
+ send(out, Op.REQUEST_SHORT_CIRCUIT_FDS, proto);
+ }
+
+ @Override
+ public void releaseShortCircuitFds(SlotId slotId) throws IOException {
+ ReleaseShortCircuitAccessRequestProto.Builder builder =
+ ReleaseShortCircuitAccessRequestProto.newBuilder().
+ setSlotId(PBHelperClient.convert(slotId));
+ if (Trace.isTracing()) {
+ Span s = Trace.currentSpan();
+ builder.setTraceInfo(DataTransferTraceInfoProto.newBuilder()
+ .setTraceId(s.getTraceId()).setParentId(s.getSpanId()));
+ }
+ ReleaseShortCircuitAccessRequestProto proto = builder.build();
+ send(out, Op.RELEASE_SHORT_CIRCUIT_FDS, proto);
+ }
+
+ @Override
+ public void requestShortCircuitShm(String clientName) throws IOException {
+ ShortCircuitShmRequestProto.Builder builder =
+ ShortCircuitShmRequestProto.newBuilder().
+ setClientName(clientName);
+ if (Trace.isTracing()) {
+ Span s = Trace.currentSpan();
+ builder.setTraceInfo(DataTransferTraceInfoProto.newBuilder()
+ .setTraceId(s.getTraceId()).setParentId(s.getSpanId()));
+ }
+ ShortCircuitShmRequestProto proto = builder.build();
+ send(out, Op.REQUEST_SHORT_CIRCUIT_SHM, proto);
+ }
+
+ @Override
+ public void replaceBlock(final ExtendedBlock blk,
+ final StorageType storageType,
+ final Token<BlockTokenIdentifier> blockToken,
+ final String delHint,
+ final DatanodeInfo source) throws IOException {
+ OpReplaceBlockProto proto = OpReplaceBlockProto.newBuilder()
+ .setHeader(DataTransferProtoUtil.buildBaseHeader(blk, blockToken))
+ .setStorageType(PBHelperClient.convertStorageType(storageType))
+ .setDelHint(delHint)
+ .setSource(PBHelperClient.convertDatanodeInfo(source))
+ .build();
+
+ send(out, Op.REPLACE_BLOCK, proto);
+ }
+
+ @Override
+ public void copyBlock(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken) throws IOException {
+ OpCopyBlockProto proto = OpCopyBlockProto.newBuilder()
+ .setHeader(DataTransferProtoUtil.buildBaseHeader(blk, blockToken))
+ .build();
+
+ send(out, Op.COPY_BLOCK, proto);
+ }
+
+ @Override
+ public void blockChecksum(final ExtendedBlock blk,
+ final Token<BlockTokenIdentifier> blockToken) throws IOException {
+ OpBlockChecksumProto proto = OpBlockChecksumProto.newBuilder()
+ .setHeader(DataTransferProtoUtil.buildBaseHeader(blk, blockToken))
+ .build();
+
+ send(out, Op.BLOCK_CHECKSUM, proto);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java
new file mode 100644
index 0000000..edf658a
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelperClient.java
@@ -0,0 +1,254 @@
+/**
+ * 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.hdfs.protocolPB;
+
+import com.google.common.collect.Lists;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.CodedInputStream;
+import org.apache.hadoop.fs.StorageType;
+import org.apache.hadoop.hdfs.protocol.DatanodeID;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmIdProto;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmSlotProto;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.DatanodeIDProto;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.DatanodeInfoProto;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.ExtendedBlockProto;
+import org.apache.hadoop.hdfs.protocol.proto.HdfsProtos.StorageTypeProto;
+import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.ShmId;
+import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.SlotId;
+import org.apache.hadoop.hdfs.util.ExactSizeInputStream;
+import org.apache.hadoop.security.proto.SecurityProtos.TokenProto;
+import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.DataChecksum;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Utilities for converting protobuf classes to and from implementation classes
+ * and other helper utilities to help in dealing with protobuf.
+ *
+ * Note that when converting from an internal type to protobuf type, the
+ * converter never return null for protobuf type. The check for internal type
+ * being null must be done before calling the convert() method.
+ */
+public class PBHelperClient {
+ private PBHelperClient() {
+ /** Hidden constructor */
+ }
+
+ public static ByteString getByteString(byte[] bytes) {
+ return ByteString.copyFrom(bytes);
+ }
+
+ public static ShmId convert(ShortCircuitShmIdProto shmId) {
+ return new ShmId(shmId.getHi(), shmId.getLo());
+ }
+
+ public static DataChecksum.Type convert(HdfsProtos.ChecksumTypeProto type) {
+ return DataChecksum.Type.valueOf(type.getNumber());
+ }
+
+ public static HdfsProtos.ChecksumTypeProto convert(DataChecksum.Type type) {
+ return HdfsProtos.ChecksumTypeProto.valueOf(type.id);
+ }
+
+ public static ExtendedBlockProto convert(final ExtendedBlock b) {
+ if (b == null) return null;
+ return ExtendedBlockProto.newBuilder().
+ setPoolId(b.getBlockPoolId()).
+ setBlockId(b.getBlockId()).
+ setNumBytes(b.getNumBytes()).
+ setGenerationStamp(b.getGenerationStamp()).
+ build();
+ }
+
+ public static TokenProto convert(Token<?> tok) {
+ return TokenProto.newBuilder().
+ setIdentifier(ByteString.copyFrom(tok.getIdentifier())).
+ setPassword(ByteString.copyFrom(tok.getPassword())).
+ setKind(tok.getKind().toString()).
+ setService(tok.getService().toString()).build();
+ }
+
+ public static ShortCircuitShmIdProto convert(ShmId shmId) {
+ return ShortCircuitShmIdProto.newBuilder().
+ setHi(shmId.getHi()).
+ setLo(shmId.getLo()).
+ build();
+
+ }
+
+ public static ShortCircuitShmSlotProto convert(SlotId slotId) {
+ return ShortCircuitShmSlotProto.newBuilder().
+ setShmId(convert(slotId.getShmId())).
+ setSlotIdx(slotId.getSlotIdx()).
+ build();
+ }
+
+ public static DatanodeIDProto convert(DatanodeID dn) {
+ // For wire compatibility with older versions we transmit the StorageID
+ // which is the same as the DatanodeUuid. Since StorageID is a required
+ // field we pass the empty string if the DatanodeUuid is not yet known.
+ return DatanodeIDProto.newBuilder()
+ .setIpAddr(dn.getIpAddr())
+ .setHostName(dn.getHostName())
+ .setXferPort(dn.getXferPort())
+ .setDatanodeUuid(dn.getDatanodeUuid() != null ? dn.getDatanodeUuid() : "")
+ .setInfoPort(dn.getInfoPort())
+ .setInfoSecurePort(dn.getInfoSecurePort())
+ .setIpcPort(dn.getIpcPort()).build();
+ }
+
+ public static DatanodeInfoProto.AdminState convert(
+ final DatanodeInfo.AdminStates inAs) {
+ switch (inAs) {
+ case NORMAL: return DatanodeInfoProto.AdminState.NORMAL;
+ case DECOMMISSION_INPROGRESS:
+ return DatanodeInfoProto.AdminState.DECOMMISSION_INPROGRESS;
+ case DECOMMISSIONED: return DatanodeInfoProto.AdminState.DECOMMISSIONED;
+ default: return DatanodeInfoProto.AdminState.NORMAL;
+ }
+ }
+
+ public static DatanodeInfoProto convert(DatanodeInfo info) {
+ DatanodeInfoProto.Builder builder = DatanodeInfoProto.newBuilder();
+ if (info.getNetworkLocation() != null) {
+ builder.setLocation(info.getNetworkLocation());
+ }
+ builder
+ .setId(convert((DatanodeID) info))
+ .setCapacity(info.getCapacity())
+ .setDfsUsed(info.getDfsUsed())
+ .setRemaining(info.getRemaining())
+ .setBlockPoolUsed(info.getBlockPoolUsed())
+ .setCacheCapacity(info.getCacheCapacity())
+ .setCacheUsed(info.getCacheUsed())
+ .setLastUpdate(info.getLastUpdate())
+ .setLastUpdateMonotonic(info.getLastUpdateMonotonic())
+ .setXceiverCount(info.getXceiverCount())
+ .setAdminState(convert(info.getAdminState()))
+ .build();
+ return builder.build();
+ }
+
+ public static List<? extends HdfsProtos.DatanodeInfoProto> convert(
+ DatanodeInfo[] dnInfos) {
+ return convert(dnInfos, 0);
+ }
+
+ /**
+ * Copy from {@code dnInfos} to a target of list of same size starting at
+ * {@code startIdx}.
+ */
+ public static List<? extends HdfsProtos.DatanodeInfoProto> convert(
+ DatanodeInfo[] dnInfos, int startIdx) {
+ if (dnInfos == null)
+ return null;
+ ArrayList<HdfsProtos.DatanodeInfoProto> protos = Lists
+ .newArrayListWithCapacity(dnInfos.length);
+ for (int i = startIdx; i < dnInfos.length; i++) {
+ protos.add(convert(dnInfos[i]));
+ }
+ return protos;
+ }
+
+ public static List<Boolean> convert(boolean[] targetPinnings, int idx) {
+ List<Boolean> pinnings = new ArrayList<>();
+ if (targetPinnings == null) {
+ pinnings.add(Boolean.FALSE);
+ } else {
+ for (; idx < targetPinnings.length; ++idx) {
+ pinnings.add(targetPinnings[idx]);
+ }
+ }
+ return pinnings;
+ }
+
+ static public DatanodeInfoProto convertDatanodeInfo(DatanodeInfo di) {
+ if (di == null) return null;
+ return convert(di);
+ }
+
+ public static StorageTypeProto convertStorageType(StorageType type) {
+ switch(type) {
+ case DISK:
+ return StorageTypeProto.DISK;
+ case SSD:
+ return StorageTypeProto.SSD;
+ case ARCHIVE:
+ return StorageTypeProto.ARCHIVE;
+ case RAM_DISK:
+ return StorageTypeProto.RAM_DISK;
+ default:
+ throw new IllegalStateException(
+ "BUG: StorageType not found, type=" + type);
+ }
+ }
+
+ public static StorageType convertStorageType(StorageTypeProto type) {
+ switch(type) {
+ case DISK:
+ return StorageType.DISK;
+ case SSD:
+ return StorageType.SSD;
+ case ARCHIVE:
+ return StorageType.ARCHIVE;
+ case RAM_DISK:
+ return StorageType.RAM_DISK;
+ default:
+ throw new IllegalStateException(
+ "BUG: StorageTypeProto not found, type=" + type);
+ }
+ }
+
+ public static List<StorageTypeProto> convertStorageTypes(
+ StorageType[] types) {
+ return convertStorageTypes(types, 0);
+ }
+
+ public static List<StorageTypeProto> convertStorageTypes(
+ StorageType[] types, int startIdx) {
+ if (types == null) {
+ return null;
+ }
+ final List<StorageTypeProto> protos = new ArrayList<>(
+ types.length);
+ for (int i = startIdx; i < types.length; ++i) {
+ protos.add(PBHelperClient.convertStorageType(types[i]));
+ }
+ return protos;
+ }
+
+ public static InputStream vintPrefixed(final InputStream input)
+ throws IOException {
+ final int firstByte = input.read();
+ if (firstByte == -1) {
+ throw new EOFException("Premature EOF: no length prefix available");
+ }
+
+ int size = CodedInputStream.readRawVarint32(firstByte, input);
+ assert size >= 0;
+ return new ExactSizeInputStream(input, size);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
new file mode 100644
index 0000000..2fa86fa
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/security/token/block/InvalidBlockTokenException.java
@@ -0,0 +1,41 @@
+/**
+ * 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.hdfs.security.token.block;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Access token verification failed.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class InvalidBlockTokenException extends IOException {
+ private static final long serialVersionUID = 168L;
+
+ public InvalidBlockTokenException() {
+ super();
+ }
+
+ public InvalidBlockTokenException(String msg) {
+ super(msg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
new file mode 100644
index 0000000..215df13
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/server/datanode/CachingStrategy.java
@@ -0,0 +1,76 @@
+/**
+ * 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.hdfs.server.datanode;
+
+/**
+ * The caching strategy we should use for an HDFS read or write operation.
+ */
+public class CachingStrategy {
+ private final Boolean dropBehind; // null = use server defaults
+ private final Long readahead; // null = use server defaults
+
+ public static CachingStrategy newDefaultStrategy() {
+ return new CachingStrategy(null, null);
+ }
+
+ public static CachingStrategy newDropBehind() {
+ return new CachingStrategy(true, null);
+ }
+
+ public static class Builder {
+ private Boolean dropBehind;
+ private Long readahead;
+
+ public Builder(CachingStrategy prev) {
+ this.dropBehind = prev.dropBehind;
+ this.readahead = prev.readahead;
+ }
+
+ public Builder setDropBehind(Boolean dropBehind) {
+ this.dropBehind = dropBehind;
+ return this;
+ }
+
+ public Builder setReadahead(Long readahead) {
+ this.readahead = readahead;
+ return this;
+ }
+
+ public CachingStrategy build() {
+ return new CachingStrategy(dropBehind, readahead);
+ }
+ }
+
+ public CachingStrategy(Boolean dropBehind, Long readahead) {
+ this.dropBehind = dropBehind;
+ this.readahead = readahead;
+ }
+
+ public Boolean getDropBehind() {
+ return dropBehind;
+ }
+
+ public Long getReadahead() {
+ return readahead;
+ }
+
+ public String toString() {
+ return "CachingStrategy(dropBehind=" + dropBehind +
+ ", readahead=" + readahead + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
new file mode 100644
index 0000000..81cc68d
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShm.java
@@ -0,0 +1,119 @@
+/**
+ * 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.hdfs.shortcircuit;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.Iterator;
+
+import org.apache.hadoop.hdfs.net.DomainPeer;
+import org.apache.hadoop.hdfs.shortcircuit.DfsClientShmManager.EndpointShmManager;
+import org.apache.hadoop.net.unix.DomainSocket;
+import org.apache.hadoop.net.unix.DomainSocketWatcher;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * DfsClientShm is a subclass of ShortCircuitShm which is used by the
+ * DfsClient.
+ * When the UNIX domain socket associated with this shared memory segment
+ * closes unexpectedly, we mark the slots inside this segment as disconnected.
+ * ShortCircuitReplica objects that contain disconnected slots are stale,
+ * and will not be used to service new reads or mmap operations.
+ * However, in-progress read or mmap operations will continue to proceed.
+ * Once the last slot is deallocated, the segment can be safely munmapped.
+ *
+ * Slots may also become stale because the associated replica has been deleted
+ * on the DataNode. In this case, the DataNode will clear the 'valid' bit.
+ * The client will then see these slots as stale (see
+ * #{ShortCircuitReplica#isStale}).
+ */
+public class DfsClientShm extends ShortCircuitShm
+ implements DomainSocketWatcher.Handler {
+ /**
+ * The EndpointShmManager associated with this shared memory segment.
+ */
+ private final EndpointShmManager manager;
+
+ /**
+ * The UNIX domain socket associated with this DfsClientShm.
+ * We rely on the DomainSocketWatcher to close the socket associated with
+ * this DomainPeer when necessary.
+ */
+ private final DomainPeer peer;
+
+ /**
+ * True if this shared memory segment has lost its connection to the
+ * DataNode.
+ *
+ * {@link DfsClientShm#handle} sets this to true.
+ */
+ private boolean disconnected = false;
+
+ DfsClientShm(ShmId shmId, FileInputStream stream, EndpointShmManager manager,
+ DomainPeer peer) throws IOException {
+ super(shmId, stream);
+ this.manager = manager;
+ this.peer = peer;
+ }
+
+ public EndpointShmManager getEndpointShmManager() {
+ return manager;
+ }
+
+ public DomainPeer getPeer() {
+ return peer;
+ }
+
+ /**
+ * Determine if the shared memory segment is disconnected from the DataNode.
+ *
+ * This must be called with the DfsClientShmManager lock held.
+ *
+ * @return True if the shared memory segment is stale.
+ */
+ public synchronized boolean isDisconnected() {
+ return disconnected;
+ }
+
+ /**
+ * Handle the closure of the UNIX domain socket associated with this shared
+ * memory segment by marking this segment as stale.
+ *
+ * If there are no slots associated with this shared memory segment, it will
+ * be freed immediately in this function.
+ */
+ @Override
+ public boolean handle(DomainSocket sock) {
+ manager.unregisterShm(getShmId());
+ synchronized (this) {
+ Preconditions.checkState(!disconnected);
+ disconnected = true;
+ boolean hadSlots = false;
+ for (Iterator<Slot> iter = slotIterator(); iter.hasNext(); ) {
+ Slot slot = iter.next();
+ slot.makeInvalid();
+ hadSlots = true;
+ }
+ if (!hadSlots) {
+ free();
+ }
+ }
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/8e4afa3a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
new file mode 100644
index 0000000..f70398a
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/shortcircuit/DfsClientShmManager.java
@@ -0,0 +1,522 @@
+/**
+ * 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.hdfs.shortcircuit;
+
+import java.io.BufferedOutputStream;
+import java.io.Closeable;
+import java.io.DataOutputStream;
+import java.io.EOFException;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map.Entry;
+import java.util.TreeMap;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.commons.lang.mutable.MutableBoolean;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.ExtendedBlockId;
+import org.apache.hadoop.hdfs.net.DomainPeer;
+import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
+import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol;
+import org.apache.hadoop.hdfs.protocol.datatransfer.Sender;
+import org.apache.hadoop.hdfs.protocol.proto.DataTransferProtos.ShortCircuitShmResponseProto;
+import org.apache.hadoop.hdfs.protocolPB.PBHelperClient;
+import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.ShmId;
+import org.apache.hadoop.hdfs.shortcircuit.ShortCircuitShm.Slot;
+import org.apache.hadoop.net.unix.DomainSocket;
+import org.apache.hadoop.net.unix.DomainSocketWatcher;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manages short-circuit memory segments for an HDFS client.
+ *
+ * Clients are responsible for requesting and releasing shared memory segments used
+ * for communicating with the DataNode. The client will try to allocate new slots
+ * in the set of existing segments, falling back to getting a new segment from the
+ * DataNode via {@link DataTransferProtocol#requestShortCircuitFds}.
+ *
+ * The counterpart to this class on the DataNode is {@link ShortCircuitRegistry}.
+ * See {@link ShortCircuitRegistry} for more information on the communication protocol.
+ */
+@InterfaceAudience.Private
+public class DfsClientShmManager implements Closeable {
+ private static final Logger LOG = LoggerFactory.getLogger(
+ DfsClientShmManager.class);
+
+ /**
+ * Manages short-circuit memory segments that pertain to a given DataNode.
+ */
+ class EndpointShmManager {
+ /**
+ * The datanode we're managing.
+ */
+ private final DatanodeInfo datanode;
+
+ /**
+ * Shared memory segments which have no empty slots.
+ *
+ * Protected by the manager lock.
+ */
+ private final TreeMap<ShmId, DfsClientShm> full =
+ new TreeMap<ShmId, DfsClientShm>();
+
+ /**
+ * Shared memory segments which have at least one empty slot.
+ *
+ * Protected by the manager lock.
+ */
+ private final TreeMap<ShmId, DfsClientShm> notFull =
+ new TreeMap<ShmId, DfsClientShm>();
+
+ /**
+ * True if this datanode doesn't support short-circuit shared memory
+ * segments.
+ *
+ * Protected by the manager lock.
+ */
+ private boolean disabled = false;
+
+ /**
+ * True if we're in the process of loading a shared memory segment from
+ * this DataNode.
+ *
+ * Protected by the manager lock.
+ */
+ private boolean loading = false;
+
+ EndpointShmManager (DatanodeInfo datanode) {
+ this.datanode = datanode;
+ }
+
+ /**
+ * Pull a slot out of a preexisting shared memory segment.
+ *
+ * Must be called with the manager lock held.
+ *
+ * @param blockId The blockId to put inside the Slot object.
+ *
+ * @return null if none of our shared memory segments contain a
+ * free slot; the slot object otherwise.
+ */
+ private Slot allocSlotFromExistingShm(ExtendedBlockId blockId) {
+ if (notFull.isEmpty()) {
+ return null;
+ }
+ Entry<ShmId, DfsClientShm> entry = notFull.firstEntry();
+ DfsClientShm shm = entry.getValue();
+ ShmId shmId = shm.getShmId();
+ Slot slot = shm.allocAndRegisterSlot(blockId);
+ if (shm.isFull()) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": pulled the last slot " + slot.getSlotIdx() +
+ " out of " + shm);
+ }
+ DfsClientShm removedShm = notFull.remove(shmId);
+ Preconditions.checkState(removedShm == shm);
+ full.put(shmId, shm);
+ } else {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": pulled slot " + slot.getSlotIdx() +
+ " out of " + shm);
+ }
+ }
+ return slot;
+ }
+
+ /**
+ * Ask the DataNode for a new shared memory segment. This function must be
+ * called with the manager lock held. We will release the lock while
+ * communicating with the DataNode.
+ *
+ * @param clientName The current client name.
+ * @param peer The peer to use to talk to the DataNode.
+ *
+ * @return Null if the DataNode does not support shared memory
+ * segments, or experienced an error creating the
+ * shm. The shared memory segment itself on success.
+ * @throws IOException If there was an error communicating over the socket.
+ * We will not throw an IOException unless the socket
+ * itself (or the network) is the problem.
+ */
+ private DfsClientShm requestNewShm(String clientName, DomainPeer peer)
+ throws IOException {
+ final DataOutputStream out =
+ new DataOutputStream(
+ new BufferedOutputStream(peer.getOutputStream()));
+ new Sender(out).requestShortCircuitShm(clientName);
+ ShortCircuitShmResponseProto resp =
+ ShortCircuitShmResponseProto.parseFrom(
+ PBHelperClient.vintPrefixed(peer.getInputStream()));
+ String error = resp.hasError() ? resp.getError() : "(unknown)";
+ switch (resp.getStatus()) {
+ case SUCCESS:
+ DomainSocket sock = peer.getDomainSocket();
+ byte buf[] = new byte[1];
+ FileInputStream fis[] = new FileInputStream[1];
+ if (sock.recvFileInputStreams(fis, buf, 0, buf.length) < 0) {
+ throw new EOFException("got EOF while trying to transfer the " +
+ "file descriptor for the shared memory segment.");
+ }
+ if (fis[0] == null) {
+ throw new IOException("the datanode " + datanode + " failed to " +
+ "pass a file descriptor for the shared memory segment.");
+ }
+ try {
+ DfsClientShm shm =
+ new DfsClientShm(PBHelperClient.convert(resp.getId()),
+ fis[0], this, peer);
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": createNewShm: created " + shm);
+ }
+ return shm;
+ } finally {
+ try {
+ fis[0].close();
+ } catch (Throwable e) {
+ LOG.debug("Exception in closing " + fis[0], e);
+ }
+ }
+ case ERROR_UNSUPPORTED:
+ // The DataNode just does not support short-circuit shared memory
+ // access, and we should stop asking.
+ LOG.info(this + ": datanode does not support short-circuit " +
+ "shared memory access: " + error);
+ disabled = true;
+ return null;
+ default:
+ // The datanode experienced some kind of unexpected error when trying to
+ // create the short-circuit shared memory segment.
+ LOG.warn(this + ": error requesting short-circuit shared memory " +
+ "access: " + error);
+ return null;
+ }
+ }
+
+ /**
+ * Allocate a new shared memory slot connected to this datanode.
+ *
+ * Must be called with the EndpointShmManager lock held.
+ *
+ * @param peer The peer to use to talk to the DataNode.
+ * @param usedPeer (out param) Will be set to true if we used the peer.
+ * When a peer is used
+ *
+ * @param clientName The client name.
+ * @param blockId The block ID to use.
+ * @return null if the DataNode does not support shared memory
+ * segments, or experienced an error creating the
+ * shm. The shared memory segment itself on success.
+ * @throws IOException If there was an error communicating over the socket.
+ */
+ Slot allocSlot(DomainPeer peer, MutableBoolean usedPeer,
+ String clientName, ExtendedBlockId blockId) throws IOException {
+ while (true) {
+ if (closed) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": the DfsClientShmManager has been closed.");
+ }
+ return null;
+ }
+ if (disabled) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": shared memory segment access is disabled.");
+ }
+ return null;
+ }
+ // Try to use an existing slot.
+ Slot slot = allocSlotFromExistingShm(blockId);
+ if (slot != null) {
+ return slot;
+ }
+ // There are no free slots. If someone is loading more slots, wait
+ // for that to finish.
+ if (loading) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": waiting for loading to finish...");
+ }
+ finishedLoading.awaitUninterruptibly();
+ } else {
+ // Otherwise, load the slot ourselves.
+ loading = true;
+ lock.unlock();
+ DfsClientShm shm;
+ try {
+ shm = requestNewShm(clientName, peer);
+ if (shm == null) continue;
+ // See #{DfsClientShmManager#domainSocketWatcher} for details
+ // about why we do this before retaking the manager lock.
+ domainSocketWatcher.add(peer.getDomainSocket(), shm);
+ // The DomainPeer is now our responsibility, and should not be
+ // closed by the caller.
+ usedPeer.setValue(true);
+ } finally {
+ lock.lock();
+ loading = false;
+ finishedLoading.signalAll();
+ }
+ if (shm.isDisconnected()) {
+ // If the peer closed immediately after the shared memory segment
+ // was created, the DomainSocketWatcher callback might already have
+ // fired and marked the shm as disconnected. In this case, we
+ // obviously don't want to add the SharedMemorySegment to our list
+ // of valid not-full segments.
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(this + ": the UNIX domain socket associated with " +
+ "this short-circuit memory closed before we could make " +
+ "use of the shm.");
+ }
+ } else {
+ notFull.put(shm.getShmId(), shm);
+ }
+ }
+ }
+ }
+
+ /**
+ * Stop tracking a slot.
+ *
+ * Must be called with the EndpointShmManager lock held.
+ *
+ * @param slot The slot to release.
+ */
+ void freeSlot(Slot slot) {
+ DfsClientShm shm = (DfsClientShm)slot.getShm();
+ shm.unregisterSlot(slot.getSlotIdx());
+ if (shm.isDisconnected()) {
+ // Stale shared memory segments should not be tracked here.
+ Preconditions.checkState(!full.containsKey(shm.getShmId()));
+ Preconditions.checkState(!notFull.containsKey(shm.getShmId()));
+ if (shm.isEmpty()) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": freeing empty stale " + shm);
+ }
+ shm.free();
+ }
+ } else {
+ ShmId shmId = shm.getShmId();
+ full.remove(shmId); // The shm can't be full if we just freed a slot.
+ if (shm.isEmpty()) {
+ notFull.remove(shmId);
+
+ // If the shared memory segment is now empty, we call shutdown(2) on
+ // the UNIX domain socket associated with it. The DomainSocketWatcher,
+ // which is watching this socket, will call DfsClientShm#handle,
+ // cleaning up this shared memory segment.
+ //
+ // See #{DfsClientShmManager#domainSocketWatcher} for details about why
+ // we don't want to call DomainSocketWatcher#remove directly here.
+ //
+ // Note that we could experience 'fragmentation' here, where the
+ // DFSClient allocates a bunch of slots in different shared memory
+ // segments, and then frees most of them, but never fully empties out
+ // any segment. We make some attempt to avoid this fragmentation by
+ // always allocating new slots out of the shared memory segment with the
+ // lowest ID, but it could still occur. In most workloads,
+ // fragmentation should not be a major concern, since it doesn't impact
+ // peak file descriptor usage or the speed of allocation.
+ if (LOG.isTraceEnabled()) {
+ LOG.trace(this + ": shutting down UNIX domain socket for " +
+ "empty " + shm);
+ }
+ shutdown(shm);
+ } else {
+ notFull.put(shmId, shm);
+ }
+ }
+ }
+
+ /**
+ * Unregister a shared memory segment.
+ *
+ * Once a segment is unregistered, we will not allocate any more slots
+ * inside that segment.
+ *
+ * The DomainSocketWatcher calls this while holding the DomainSocketWatcher
+ * lock.
+ *
+ * @param shmId The ID of the shared memory segment to unregister.
+ */
+ void unregisterShm(ShmId shmId) {
+ lock.lock();
+ try {
+ full.remove(shmId);
+ notFull.remove(shmId);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format("EndpointShmManager(%s, parent=%s)",
+ datanode, DfsClientShmManager.this);
+ }
+
+ PerDatanodeVisitorInfo getVisitorInfo() {
+ return new PerDatanodeVisitorInfo(full, notFull, disabled);
+ }
+
+ final void shutdown(DfsClientShm shm) {
+ try {
+ shm.getPeer().getDomainSocket().shutdown();
+ } catch (IOException e) {
+ LOG.warn(this + ": error shutting down shm: got IOException calling " +
+ "shutdown(SHUT_RDWR)", e);
+ }
+ }
+ }
+
+ private boolean closed = false;
+
+ private final ReentrantLock lock = new ReentrantLock();
+
+ /**
+ * A condition variable which is signalled when we finish loading a segment
+ * from the Datanode.
+ */
+ private final Condition finishedLoading = lock.newCondition();
+
+ /**
+ * Information about each Datanode.
+ */
+ private final HashMap<DatanodeInfo, EndpointShmManager> datanodes =
+ new HashMap<DatanodeInfo, EndpointShmManager>(1);
+
+ /**
+ * The DomainSocketWatcher which keeps track of the UNIX domain socket
+ * associated with each shared memory segment.
+ *
+ * Note: because the DomainSocketWatcher makes callbacks into this
+ * DfsClientShmManager object, you must MUST NOT attempt to take the
+ * DomainSocketWatcher lock while holding the DfsClientShmManager lock,
+ * or else deadlock might result. This means that most DomainSocketWatcher
+ * methods are off-limits unless you release the manager lock first.
+ */
+ private final DomainSocketWatcher domainSocketWatcher;
+
+ DfsClientShmManager(int interruptCheckPeriodMs) throws IOException {
+ this.domainSocketWatcher = new DomainSocketWatcher(interruptCheckPeriodMs,
+ "client");
+ }
+
+ public Slot allocSlot(DatanodeInfo datanode, DomainPeer peer,
+ MutableBoolean usedPeer, ExtendedBlockId blockId,
+ String clientName) throws IOException {
+ lock.lock();
+ try {
+ if (closed) {
+ LOG.trace(this + ": the DfsClientShmManager isclosed.");
+ return null;
+ }
+ EndpointShmManager shmManager = datanodes.get(datanode);
+ if (shmManager == null) {
+ shmManager = new EndpointShmManager(datanode);
+ datanodes.put(datanode, shmManager);
+ }
+ return shmManager.allocSlot(peer, usedPeer, clientName, blockId);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ public void freeSlot(Slot slot) {
+ lock.lock();
+ try {
+ DfsClientShm shm = (DfsClientShm)slot.getShm();
+ shm.getEndpointShmManager().freeSlot(slot);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @VisibleForTesting
+ public static class PerDatanodeVisitorInfo {
+ public final TreeMap<ShmId, DfsClientShm> full;
+ public final TreeMap<ShmId, DfsClientShm> notFull;
+ public final boolean disabled;
+
+ PerDatanodeVisitorInfo(TreeMap<ShmId, DfsClientShm> full,
+ TreeMap<ShmId, DfsClientShm> notFull, boolean disabled) {
+ this.full = full;
+ this.notFull = notFull;
+ this.disabled = disabled;
+ }
+ }
+
+ @VisibleForTesting
+ public interface Visitor {
+ void visit(HashMap<DatanodeInfo, PerDatanodeVisitorInfo> info)
+ throws IOException;
+ }
+
+ @VisibleForTesting
+ public void visit(Visitor visitor) throws IOException {
+ lock.lock();
+ try {
+ HashMap<DatanodeInfo, PerDatanodeVisitorInfo> info =
+ new HashMap<DatanodeInfo, PerDatanodeVisitorInfo>();
+ for (Entry<DatanodeInfo, EndpointShmManager> entry :
+ datanodes.entrySet()) {
+ info.put(entry.getKey(), entry.getValue().getVisitorInfo());
+ }
+ visitor.visit(info);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Close the DfsClientShmManager.
+ */
+ @Override
+ public void close() throws IOException {
+ lock.lock();
+ try {
+ if (closed) return;
+ closed = true;
+ } finally {
+ lock.unlock();
+ }
+ // When closed, the domainSocketWatcher will issue callbacks that mark
+ // all the outstanding DfsClientShm segments as stale.
+ try {
+ domainSocketWatcher.close();
+ } catch (Throwable e) {
+ LOG.debug("Exception in closing " + domainSocketWatcher, e);
+ }
+ }
+
+
+ @Override
+ public String toString() {
+ return String.format("ShortCircuitShmManager(%08x)",
+ System.identityHashCode(this));
+ }
+
+ @VisibleForTesting
+ public DomainSocketWatcher getDomainSocketWatcher() {
+ return domainSocketWatcher;
+ }
+}