You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by je...@apache.org on 2018/04/06 02:35:57 UTC
[incubator-nemo] 17/40: limit in bytes
This is an automated email from the ASF dual-hosted git repository.
jeongyoon pushed a commit to branch skew_exp
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
commit 5c34d0dedda76f14e16a926e8ca01d1bdbb81c20
Author: sanha <sa...@naver.com>
AuthorDate: Mon Mar 12 01:25:42 2018 +0900
limit in bytes
---
.../java/edu/snu/nemo/common/coder/BytesCoder.java | 55 +++++++++++++++--
.../executor/bytetransfer/ByteOutputContext.java | 18 +++---
.../snu/nemo/runtime/executor/data/DataUtil.java | 68 ++++++++++++----------
.../runtime/executor/data/SerializerManager.java | 4 +-
.../runtime/executor/data/block/FileBlock.java | 18 +++---
.../executor/data/metadata/FileMetadata.java | 4 +-
.../executor/data/stores/AbstractBlockStore.java | 2 +-
.../runtime/executor/data/stores/MemoryStore.java | 2 +-
.../data/stores/SerializedMemoryStore.java | 2 +-
9 files changed, 113 insertions(+), 60 deletions(-)
diff --git a/common/src/main/java/edu/snu/nemo/common/coder/BytesCoder.java b/common/src/main/java/edu/snu/nemo/common/coder/BytesCoder.java
index 3a6a356..48d275f 100644
--- a/common/src/main/java/edu/snu/nemo/common/coder/BytesCoder.java
+++ b/common/src/main/java/edu/snu/nemo/common/coder/BytesCoder.java
@@ -20,7 +20,7 @@ import java.io.*;
/**
* A {@link Coder} which is used for an array of bytes.
*/
-public final class BytesCoder implements Coder<byte[]> {
+public final class BytesCoder implements Coder<BytesCoder.BytesWrapper> {
/**
* Constructor.
@@ -29,25 +29,68 @@ public final class BytesCoder implements Coder<byte[]> {
}
@Override
- public void encode(final byte[] value, final OutputStream outStream) throws IOException {
+ public void encode(final BytesWrapper value, final OutputStream outStream) throws IOException {
try (final DataOutputStream dataOutputStream = new DataOutputStream(outStream)) {
- dataOutputStream.writeInt(value.length); // Write the size of this byte array.
- dataOutputStream.write(value);
+ dataOutputStream.writeInt(value.getLength()); // Write the size of this byte array.
+ dataOutputStream.writeLong(value.getElementsTotal()); // Write the number of elements in this byte array.
+ dataOutputStream.write(value.getBytes());
}
}
@Override
- public byte[] decode(final InputStream inStream) throws IOException {
+ public BytesWrapper decode(final InputStream inStream) throws IOException {
// If the inStream is closed well in upper level, it is okay to not close this stream
// because the DataInputStream itself will not contain any extra information.
// (when we close this stream, the inStream will be closed together.)
final DataInputStream dataInputStream = new DataInputStream(inStream);
final int bytesToRead = dataInputStream.readInt();
+ final long elementsTotal = dataInputStream.readLong();
final byte[] bytes = new byte[bytesToRead]; // Read the size of this byte array.
final int readBytes = dataInputStream.read(bytes);
if (bytesToRead != readBytes) {
throw new IOException("Have to read " + bytesToRead + " but read only " + readBytes + " bytes.");
}
- return bytes;
+ return new BytesWrapper(elementsTotal, bytes);
+ }
+
+ /**
+ * A wrapper class for an array of bytes.
+ */
+ public final class BytesWrapper {
+ private long elementsTotal;
+ private byte[] bytes;
+
+ /**
+ * Constructor.
+ *
+ * @param elementsTotal the total number of elements in this array of bytes.
+ * @param bytes the array of bytes.
+ */
+ public BytesWrapper(final long elementsTotal,
+ final byte[] bytes) {
+ this.elementsTotal = elementsTotal;
+ this.bytes = bytes;
+ }
+
+ /**
+ * @return the length of the byte array.
+ */
+ public int getLength() {
+ return bytes.length;
+ }
+
+ /**
+ * @return the total number of elements in the byte array.
+ */
+ public long getElementsTotal() {
+ return elementsTotal;
+ }
+
+ /**
+ * @return the wrapped byte array.
+ */
+ public byte[] getBytes() {
+ return bytes;
+ }
}
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/bytetransfer/ByteOutputContext.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/bytetransfer/ByteOutputContext.java
index ebe9d4f..097bd4b 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/bytetransfer/ByteOutputContext.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/bytetransfer/ByteOutputContext.java
@@ -43,10 +43,10 @@ public final class ByteOutputContext extends ByteTransferContext implements Auto
/**
* Creates a output context.
*
- * @param remoteExecutorId id of the remote executor
- * @param contextId identifier for this context
- * @param contextDescriptor user-provided context descriptor
- * @param contextManager {@link ContextManager} for the channel
+ * @param remoteExecutorId id of the remote executor
+ * @param contextId identifier for this context
+ * @param contextDescriptor user-provided context descriptor
+ * @param contextManager {@link ContextManager} for the channel
*/
ByteOutputContext(final String remoteExecutorId,
final ContextId contextId,
@@ -58,6 +58,7 @@ public final class ByteOutputContext extends ByteTransferContext implements Auto
/**
* Closes existing sub-stream (if any) and create a new sub-stream.
+ *
* @return new {@link ByteOutputStream}
* @throws IOException if an exception was set or this context was closed.
*/
@@ -114,7 +115,7 @@ public final class ByteOutputContext extends ByteTransferContext implements Auto
/**
* An {@link OutputStream} implementation which buffers data to {@link ByteBuf}s.
- *
+ * <p>
* <p>Public methods are thread safe,
* although the execution order may not be linearized if they were called from different threads.</p>
*/
@@ -139,6 +140,7 @@ public final class ByteOutputContext extends ByteTransferContext implements Auto
/**
* Writes {@link SerializedPartition}.
+ *
* @param serializedPartition {@link SerializedPartition} to write.
* @return {@code this}
* @throws IOException when an exception has been set or this stream was closed
@@ -184,6 +186,7 @@ public final class ByteOutputContext extends ByteTransferContext implements Auto
/**
* Writes a data frame, from {@link ByteBuf}.
+ *
* @param byteBuf {@link ByteBuf} to write.
*/
private void writeByteBuf(final ByteBuf byteBuf) throws IOException {
@@ -194,8 +197,9 @@ public final class ByteOutputContext extends ByteTransferContext implements Auto
/**
* Writes a data frame.
- * @param body the body or {@code null}
- * @param length the length of the body, in bytes
+ *
+ * @param body the body or {@code null}
+ * @param length the length of the body, in bytes
* @throws IOException when an exception has been set or this stream was closed
*/
private synchronized void writeDataFrame(final Object body, final long length) throws IOException {
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/DataUtil.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/DataUtil.java
index 861451a..87aeeef 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/DataUtil.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/DataUtil.java
@@ -64,21 +64,22 @@ public final class DataUtil {
/**
* Reads the data of a partition from an input stream and deserializes it.
*
- * @param elementsInPartition the number of elements in this partition.
- * @param serializer the serializer to decode the bytes.
- * @param key the key value of the result partition.
- * @param inputStream the input stream which will return the data in the partition as bytes.
- * @param <K> the key type of the partitions.
+ * @param length the length of this partition in bytes.
+ * @param serializer the serializer to decode the bytes.
+ * @param key the key value of the result partition.
+ * @param inputStream the input stream which will return the data in the partition as bytes.
+ * @param <K> the key type of the partitions.
* @return the list of deserialized elements.
* @throws IOException if fail to deserialize.
*/
- public static <K extends Serializable> NonSerializedPartition deserializePartition(final long elementsInPartition,
- final Serializer serializer,
- final K key,
- final InputStream inputStream) throws IOException {
+ public static <K extends Serializable>
+ NonSerializedPartition deserializePartition(final int length,
+ final Serializer serializer,
+ final K key,
+ final InputStream inputStream) throws IOException {
final List deserializedData = new ArrayList();
final InputStreamIterator iterator = new InputStreamIterator(Collections.singletonList(inputStream).iterator(),
- serializer, elementsInPartition);
+ serializer, length);
iterator.forEachRemaining(deserializedData::add);
return new NonSerializedPartition(key, deserializedData, iterator.getNumSerializedBytes(),
iterator.getNumEncodedBytes());
@@ -135,7 +136,7 @@ public final class DataUtil {
try (final ByteArrayInputStream byteArrayInputStream =
new ByteArrayInputStream(partitionToConvert.getData())) {
final NonSerializedPartition<K> deserializePartition = deserializePartition(
- partitionToConvert.getElementsCount(), serializer, key, byteArrayInputStream);
+ partitionToConvert.getLength(), serializer, key, byteArrayInputStream);
nonSerializedPartitions.add(deserializePartition);
}
}
@@ -187,30 +188,31 @@ public final class DataUtil {
/**
* An iterator that emits objects from {@link InputStream} using the corresponding {@link Coder}.
+ *
* @param <T> The type of elements.
*/
public static final class InputStreamIterator<T> implements IteratorWithNumBytes<T> {
private final Iterator<InputStream> inputStreams;
private final Serializer<T> serializer;
- private final long limit;
+ private final int limit;
private volatile CountingInputStream serializedCountingStream = null;
private volatile CountingInputStream encodedCountingStream = null;
private volatile boolean hasNext = false;
private volatile T next;
private volatile boolean cannotContinueDecoding = false;
- private volatile long elementsDecoded = 0;
private volatile long numSerializedBytes = 0;
private volatile long numEncodedBytes = 0;
/**
* Construct {@link Iterator} from {@link InputStream} and {@link Coder}.
*
- * @param inputStreams The streams to read data from.
- * @param serializer The serializer.
+ * @param inputStreams the streams to read data from.
+ * @param serializer the serializer.
*/
- public InputStreamIterator(final Iterator<InputStream> inputStreams, final Serializer<T> serializer) {
+ public InputStreamIterator(final Iterator<InputStream> inputStreams,
+ final Serializer<T> serializer) {
this.inputStreams = inputStreams;
this.serializer = serializer;
// -1 means no limit.
@@ -220,14 +222,14 @@ public final class DataUtil {
/**
* Construct {@link Iterator} from {@link InputStream} and {@link Coder}.
*
- * @param inputStreams The streams to read data from.
- * @param serializer The serializer.
- * @param limit The number of elements from the {@link InputStream}.
+ * @param inputStreams the streams to read data from.
+ * @param serializer the serializer.
+ * @param limit the maximum bytes from the {@link InputStream}.
*/
public InputStreamIterator(
final Iterator<InputStream> inputStreams,
final Serializer<T> serializer,
- final long limit) {
+ final int limit) {
if (limit < 0) {
throw new IllegalArgumentException("Negative limit not allowed.");
}
@@ -244,7 +246,7 @@ public final class DataUtil {
if (cannotContinueDecoding) {
return false;
}
- if (limit != -1 && limit == elementsDecoded) {
+ if (limit != -1 && limit == numSerializedBytes) {
cannotContinueDecoding = true;
return false;
}
@@ -267,7 +269,6 @@ public final class DataUtil {
try {
next = serializer.getCoder().decode(encodedCountingStream);
hasNext = true;
- elementsDecoded++;
return true;
} catch (final IOException e) {
// IOException from decoder indicates EOF event.
@@ -314,10 +315,10 @@ public final class DataUtil {
* @param in the {@link InputStream}.
* @param streamChainers the list of {@link StreamChainer} to be applied on the stream.
* @return chained {@link InputStream}.
- * @throws IOException if fail to create new stream.
+ * @throws IOException if fail to create new stream.
*/
public static InputStream buildInputStream(final InputStream in, final List<StreamChainer> streamChainers)
- throws IOException {
+ throws IOException {
InputStream chained = in;
for (final StreamChainer streamChainer : streamChainers) {
chained = streamChainer.chainInput(chained);
@@ -331,10 +332,10 @@ public final class DataUtil {
* @param out the {@link OutputStream}.
* @param streamChainers the list of {@link StreamChainer} to be applied on the stream.
* @return chained {@link OutputStream}.
- * @throws IOException if fail to create new stream.
+ * @throws IOException if fail to create new stream.
*/
public static OutputStream buildOutputStream(final OutputStream out, final List<StreamChainer> streamChainers)
- throws IOException {
+ throws IOException {
OutputStream chained = out;
final List<StreamChainer> temporaryStreamChainerList = new ArrayList<>(streamChainers);
Collections.reverse(temporaryStreamChainerList);
@@ -346,13 +347,15 @@ public final class DataUtil {
/**
* {@link Iterator} with interface to access to the number of bytes.
+ *
* @param <T> the type of decoded object
*/
public interface IteratorWithNumBytes<T> extends Iterator<T> {
/**
* Create an {@link IteratorWithNumBytes}, with no information about the number of bytes.
+ *
* @param innerIterator {@link Iterator} to wrap
- * @param <E> the type of decoded object
+ * @param <E> the type of decoded object
* @return an {@link IteratorWithNumBytes}, with no information about the number of bytes
*/
static <E> IteratorWithNumBytes<E> of(final Iterator<E> innerIterator) {
@@ -381,10 +384,11 @@ public final class DataUtil {
/**
* Create an {@link IteratorWithNumBytes}, with the number of bytes in decoded and serialized form.
- * @param innerIterator {@link Iterator} to wrap
+ *
+ * @param innerIterator {@link Iterator} to wrap
* @param numSerializedBytes the number of bytes in serialized form
- * @param numEncodedBytes the number of bytes in encoded form
- * @param <E> the type of decoded object
+ * @param numEncodedBytes the number of bytes in encoded form
+ * @param <E> the type of decoded object
* @return an {@link IteratorWithNumBytes}, with the information about the number of bytes
*/
static <E> IteratorWithNumBytes<E> of(final Iterator<E> innerIterator,
@@ -428,14 +432,14 @@ public final class DataUtil {
/**
* @return the number of bytes in serialized form (which is, for example, encoded and compressed)
* @throws NumBytesNotSupportedException when the operation is not supported
- * @throws IllegalStateException when the information is not ready
+ * @throws IllegalStateException when the information is not ready
*/
long getNumSerializedBytes() throws NumBytesNotSupportedException;
/**
* @return the number of bytes in encoded form (which is ready to be decoded)
* @throws NumBytesNotSupportedException when the operation is not supported
- * @throws IllegalStateException when the information is not ready
+ * @throws IllegalStateException when the information is not ready
*/
long getNumEncodedBytes() throws NumBytesNotSupportedException;
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java
index 0cf2ceb..3138278 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/SerializerManager.java
@@ -38,7 +38,7 @@ import java.util.concurrent.ConcurrentMap;
*/
public final class SerializerManager {
private static final Logger LOG = LoggerFactory.getLogger(SerializerManager.class.getName());
- private static final Serializer<byte[]> AS_BYTES_SERIALIZER =
+ private static final Serializer<BytesCoder.BytesWrapper> AS_BYTES_SERIALIZER =
new Serializer<>(new BytesCoder(), Collections.emptyList());
private final ConcurrentMap<String, Serializer> runtimeEdgeIdToSerializer = new ConcurrentHashMap<>();
@@ -93,7 +93,7 @@ public final class SerializerManager {
/**
* @return a serializer which just converts between ByteStream and array of bytes.
*/
- public static Serializer<byte[]> getAsBytesSerializer() {
+ public static Serializer<BytesCoder.BytesWrapper> getAsBytesSerializer() {
return AS_BYTES_SERIALIZER;
}
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/block/FileBlock.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/block/FileBlock.java
index d140c0e..4e64245 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/block/FileBlock.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/block/FileBlock.java
@@ -101,10 +101,12 @@ public final class FileBlock<K extends Serializable> implements Block<K> {
if (metadata.isCommitted()) {
throw new BlockWriteException(new Throwable("The partition is already committed!"));
} else {
+ final Serializer serializerToUse = metadata.isWriteAsBytes()
+ ? SerializerManager.getAsBytesSerializer() : serializer;
try {
SerializedPartition<K> partition = nonCommittedPartitionsMap.get(key);
if (partition == null) {
- partition = new SerializedPartition<>(key, serializer);
+ partition = new SerializedPartition<>(key, serializerToUse);
nonCommittedPartitionsMap.put(key, partition);
}
partition.write(element);
@@ -127,8 +129,8 @@ public final class FileBlock<K extends Serializable> implements Block<K> {
throw new BlockWriteException(new Throwable("The partition is already committed!"));
} else {
try {
- final Serializer serializerToUse = metadata.isWriteAsBytes() ?
- SerializerManager.getAsBytesSerializer() : serializer; // TODO: just use byte[]
+ final Serializer serializerToUse = metadata.isWriteAsBytes()
+ ? SerializerManager.getAsBytesSerializer() : serializer;
final Iterable<SerializedPartition<K>> convertedPartitions =
DataUtil.convertToSerPartitions(serializerToUse, partitions);
writeSerializedPartitions(convertedPartitions);
@@ -173,8 +175,8 @@ public final class FileBlock<K extends Serializable> implements Block<K> {
} else {
// Deserialize the data
final List<NonSerializedPartition<K>> deserializedPartitions = new ArrayList<>();
- final Serializer serializerToUse = metadata.isReadAsBytes() ?
- SerializerManager.getAsBytesSerializer() : serializer;
+ final Serializer serializerToUse = metadata.isReadAsBytes()
+ ? SerializerManager.getAsBytesSerializer() : serializer;
try {
try (final FileInputStream fileStream = new FileInputStream(filePath)) {
for (final PartitionMetadata<K> partitionMetadata : metadata.getPartitionMetadataList()) {
@@ -186,11 +188,11 @@ public final class FileBlock<K extends Serializable> implements Block<K> {
// compression stream. We recommend to wrap with LimitedInputStream once more when
// reading input from chained compression InputStream.
// Plus, this stream must be not closed to prevent to close the filtered file partition.
+ final int length = partitionMetadata.getPartitionSize();
final LimitedInputStream limitedInputStream =
- new LimitedInputStream(fileStream, partitionMetadata.getPartitionSize());
+ new LimitedInputStream(fileStream, length);
final NonSerializedPartition<K> deserializePartition =
- DataUtil.deserializePartition(
- partitionMetadata.getElementsTotal(), serializerToUse, key, limitedInputStream); // TODO: byte[]
+ DataUtil.deserializePartition(length, serializerToUse, key, limitedInputStream);
deserializedPartitions.add(deserializePartition);
// rearrange file pointer
final long toSkip = partitionMetadata.getPartitionSize() - availableBefore + fileStream.available();
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/metadata/FileMetadata.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/metadata/FileMetadata.java
index cae38b2..818c546 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/metadata/FileMetadata.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/metadata/FileMetadata.java
@@ -127,14 +127,14 @@ public abstract class FileMetadata<K extends Serializable> {
/**
* @return whether read data from this file as arrays of bytes or not.
*/
- public boolean isReadAsBytes() {
+ public final boolean isReadAsBytes() {
return readAsBytes;
}
/**
* @return whether write to this file as arrays of bytes or not.
*/
- public boolean isWriteAsBytes() {
+ public final boolean isWriteAsBytes() {
return writeAsBytes;
}
/**
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/AbstractBlockStore.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/AbstractBlockStore.java
index ec71b25..4c73ee3 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/AbstractBlockStore.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/AbstractBlockStore.java
@@ -43,7 +43,7 @@ public abstract class AbstractBlockStore implements BlockStore {
* @see BlockStore#createBlock(String).
*/
@Override
- public Block createBlock(final String blockId) {
+ public final Block createBlock(final String blockId) {
return createBlock(blockId, false, false);
}
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/MemoryStore.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/MemoryStore.java
index d60e79e..bd608d3 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/MemoryStore.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/MemoryStore.java
@@ -49,7 +49,7 @@ public final class MemoryStore extends LocalBlockStore {
final boolean readAsBytes,
final boolean writeAsBytes) {
final Serializer serializer = getSerializerFromWorker(blockId);
- return new NonSerializedMemoryBlock(serializer, readAsBytes, writeAsBytes);
+ return new NonSerializedMemoryBlock(blockId, serializer, readAsBytes, writeAsBytes);
}
/**
diff --git a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/SerializedMemoryStore.java b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/SerializedMemoryStore.java
index b125096..83c16e8 100644
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/SerializedMemoryStore.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/data/stores/SerializedMemoryStore.java
@@ -48,7 +48,7 @@ public final class SerializedMemoryStore extends LocalBlockStore {
final boolean readAsBytes,
final boolean writeAsBytes) {
final Serializer serializer = getSerializerFromWorker(blockId);
- return new SerializedMemoryBlock(serializer, readAsBytes, writeAsBytes);
+ return new SerializedMemoryBlock(blockId, serializer, readAsBytes, writeAsBytes);
}
/**
--
To stop receiving notification emails like this one, please contact
jeongyoon@apache.org.