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:59 UTC
[incubator-nemo] 19/40: merge as 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 3e127274fd50c700f8a670193627379387075e1a
Merge: d9f7485 3943919
Author: sanha <sa...@naver.com>
AuthorDate: Tue Mar 13 19:20:20 2018 +0900
merge as bytes
.../java/edu/snu/nemo/common/coder/BytesCoder.java | 50 +++--
...rtitionerProperty.java => AsBytesProperty.java} | 28 ++-
.../executionproperty/CompressionProperty.java | 8 +-
.../executionproperty/PartitionerProperty.java | 3 +-
.../ir/executionproperty/ExecutionProperty.java | 1 +
.../compiler/frontend/beam/BeamKeyExtractor.java | 2 +-
.../compiletime/annotating/CompressionPass.java | 7 +-
.../annotating/SailfishEdgeAsBytesPass.java | 61 ++++++
.../annotating/SailfishEdgePartitionerPass.java | 52 +++++
.../pass/compiletime/composite/SailfishPass.java | 6 +-
.../executor/bytetransfer/ByteInputContext.java | 12 +-
.../executor/bytetransfer/ByteOutputContext.java | 18 +-
.../executor/bytetransfer/FrameDecoder.java | 4 +-
.../runtime/executor/data/BlockManagerWorker.java | 142 +++++++------
.../snu/nemo/runtime/executor/data/DataUtil.java | 75 ++++---
.../runtime/executor/data/LimitedInputStream.java | 37 +++-
.../runtime/executor/data/SerializerManager.java | 12 +-
.../nemo/runtime/executor/data/block/Block.java | 55 +++--
.../runtime/executor/data/block/FileBlock.java | 227 +++++++++++++--------
.../data/block/NonSerializedMemoryBlock.java | 81 +++++---
.../executor/data/block/SerializedMemoryBlock.java | 146 ++++++++-----
.../executor/data/metadata/FileMetadata.java | 37 +++-
.../executor/data/metadata/LocalFileMetadata.java | 9 +-
.../executor/data/metadata/RemoteFileMetadata.java | 35 +++-
.../IncrementPartitioner.java} | 28 +--
.../runtime/executor/data/stores/BlockStore.java | 120 +++--------
.../executor/data/stores/GlusterFileStore.java | 169 +++------------
.../executor/data/stores/LocalBlockStore.java | 120 +----------
.../executor/data/stores/LocalFileStore.java | 61 +++---
.../runtime/executor/data/stores/MemoryStore.java | 38 +++-
.../executor/data/stores/RemoteFileStore.java | 2 +-
.../data/stores/SerializedMemoryStore.java | 33 ++-
.../streamchainer/CompressionStreamChainer.java | 4 +-
.../runtime/executor/datatransfer/InputReader.java | 31 ++-
.../executor/datatransfer/OutputWriter.java | 57 +++++-
.../compiletime/composite/SailfishPassTest.java | 11 +-
.../snu/nemo/tests/runtime/RuntimeTestUtil.java | 2 +-
.../runtime/executor/data/BlockStoreTest.java | 59 +++---
.../executor/datatransfer/DataTransferTest.java | 2 +
39 files changed, 1053 insertions(+), 792 deletions(-)
diff --cc runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java
index 067ec7d,3751df3..449bd64
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/InputReader.java
@@@ -45,8 -47,8 +47,9 @@@ import org.slf4j.LoggerFactory
* TODO #492: Modularize the data communication pattern.
*/
public final class InputReader extends DataTransfer {
+ private static final Logger LOG = LoggerFactory.getLogger(InputReader.class.getName());
private final int dstTaskIndex;
+ private final boolean readAsBytes;
private final BlockManagerWorker blockManagerWorker;
/**
@@@ -56,8 -59,10 +59,8 @@@
private final RuntimeEdge runtimeEdge;
public InputReader(final int dstTaskIndex,
- // TODO #717: Remove nullable.
- // (If the source is not an IR vertex, do not make InputReader.)
- @Nullable final IRVertex srcVertex, // null if the source vertex is not an IR vertex.
+ final IRVertex srcVertex,
- final RuntimeEdge runtimeEdge,
+ final RuntimeEdge<?> runtimeEdge,
final BlockManagerWorker blockManagerWorker) {
super(runtimeEdge.getId());
this.dstTaskIndex = dstTaskIndex;
diff --cc runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java
index 1cb96d2,73cdc4c..3313ef2
--- a/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java
+++ b/runtime/executor/src/main/java/edu/snu/nemo/runtime/executor/datatransfer/OutputWriter.java
@@@ -23,8 -23,10 +23,9 @@@ import edu.snu.nemo.common.ir.execution
import edu.snu.nemo.runtime.common.RuntimeIdGenerator;
import edu.snu.nemo.runtime.common.plan.RuntimeEdge;
import edu.snu.nemo.runtime.executor.data.BlockManagerWorker;
+ import edu.snu.nemo.runtime.executor.data.block.Block;
import edu.snu.nemo.runtime.executor.data.partitioner.*;
-import javax.annotation.Nullable;
import java.util.*;
/**
@@@ -34,12 -36,12 +35,13 @@@ public final class OutputWriter extend
private final String blockId;
private final RuntimeEdge<?> runtimeEdge;
private final String srcVertexId;
- @Nullable private final IRVertex dstIrVertex;
+ private final IRVertex dstIrVertex;
private final DataStoreProperty.Value blockStoreValue;
- private Optional<Long> writtenBytes;
+ private long writtenBytes;
private final BlockManagerWorker blockManagerWorker;
private final Partitioner partitioner;
+ private final boolean writable;
+ private final Block blockToWrite;
/**
* Constructor.
@@@ -85,19 -90,43 +89,38 @@@
throw new UnsupportedPartitionerException(
new Throwable("Partitioner " + partitionerPropertyValue + " is not supported."));
}
- blockManagerWorker.createBlock(blockId, blockStoreValue);
+
+ final AsBytesProperty.Value asBytesPropertyValue = runtimeEdge.getProperty(ExecutionProperty.Key.AsBytes);
+ if (asBytesPropertyValue == null) {
+ blockToWrite = blockManagerWorker.createBlock(blockId, blockStoreValue);
+ } else {
+ switch (asBytesPropertyValue) {
+ case ReadAsBytes:
+ blockToWrite = blockManagerWorker.createBlock(blockId, blockStoreValue, true, false);
+ break;
+ case WriteAsBytes:
+ blockToWrite = blockManagerWorker.createBlock(blockId, blockStoreValue, false, true);
+ break;
+ case ReadWriteAsBytes:
+ blockToWrite = blockManagerWorker.createBlock(blockId, blockStoreValue, true, true);
+ break;
+ default:
+ throw new UnsupportedPartitionerException(
+ new Throwable("AsBytes " + asBytesPropertyValue + " is not supported."));
+ }
+ }
- }
- /**
- * Writes output data depending on the communication pattern of the edge.
- *
- * @param dataToWrite An iterable for the elements to be written.
- */
- public void write(final Iterable dataToWrite) {
final DuplicateEdgeGroupPropertyValue duplicateDataProperty =
runtimeEdge.getProperty(ExecutionProperty.Key.DuplicateEdgeGroup);
- if (duplicateDataProperty == null
+ writable = duplicateDataProperty == null
|| duplicateDataProperty.getRepresentativeEdgeId().equals(runtimeEdge.getId())
- || duplicateDataProperty.getGroupSize() <= 1) {
- dataToWrite.forEach(element -> {
- blockToWrite.write(partitioner.partition(element), element);
- });
+ || duplicateDataProperty.getGroupSize() <= 1;
+ }
+
+ public void writeElement(final Object element) {
+ if (writable) {
- blockManagerWorker.write(blockId, partitioner.partition(element), element, blockStoreValue);
- }
++ blockToWrite.write(partitioner.partition(element), element);
+ } // If else, does not need to write because the data is duplicated.
}
/**
--
To stop receiving notification emails like this one, please contact
jeongyoon@apache.org.