You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@nemo.apache.org by GitBox <gi...@apache.org> on 2018/12/03 04:37:57 UTC
[GitHub] seojangho closed pull request #164: [NEMO-293] OOM exception in
streaming
seojangho closed pull request #164: [NEMO-293] OOM exception in streaming
URL: https://github.com/apache/incubator-nemo/pull/164
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ByteOutputContext.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ByteOutputContext.java
index 315760c99..12761b27f 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ByteOutputContext.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/bytetransfer/ByteOutputContext.java
@@ -18,10 +18,14 @@
*/
package org.apache.nemo.runtime.executor.bytetransfer;
+import io.netty.buffer.ByteBufOutputStream;
+import org.apache.nemo.common.coder.EncoderFactory;
+import org.apache.nemo.runtime.executor.data.DataUtil;
import org.apache.nemo.runtime.executor.data.FileArea;
import org.apache.nemo.runtime.executor.data.partition.SerializedPartition;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
+import org.apache.nemo.runtime.executor.data.streamchainer.Serializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -94,7 +98,7 @@ public void close() throws IOException {
currentByteOutputStream.close();
}
channel.writeAndFlush(DataFrameEncoder.DataFrame.newInstance(getContextId()))
- .addListener(getChannelWriteListener());
+ .addListener(getChannelWriteListener());
deregister();
closed = true;
}
@@ -150,7 +154,7 @@ public void write(final byte[] bytes, final int offset, final int length) throws
* @throws IOException when an exception has been set or this stream was closed
*/
public ByteOutputStream writeSerializedPartition(final SerializedPartition serializedPartition)
- throws IOException {
+ throws IOException {
write(serializedPartition.getData(), 0, serializedPartition.getLength());
return this;
}
@@ -177,7 +181,7 @@ public ByteOutputStream writeFileArea(final FileArea fileArea) throws IOExceptio
}
@Override
- public synchronized void close() throws IOException {
+ public void close() throws IOException {
if (closed) {
return;
}
@@ -198,19 +202,41 @@ private void writeByteBuf(final ByteBuf byteBuf) throws IOException {
}
}
+ /**
+ * Write an element to the channel.
+ * @param element element
+ * @param serializer serializer
+ */
+ public void writeElement(final Object element,
+ final Serializer serializer) {
+ final ByteBuf byteBuf = channel.alloc().ioBuffer();
+ final ByteBufOutputStream byteBufOutputStream = new ByteBufOutputStream(byteBuf);
+ try {
+ final OutputStream wrapped =
+ DataUtil.buildOutputStream(byteBufOutputStream, serializer.getEncodeStreamChainers());
+ final EncoderFactory.Encoder encoder = serializer.getEncoderFactory().create(wrapped);
+ encoder.encode(element);
+ wrapped.close();
+
+ writeByteBuf(byteBuf);
+ } catch (final IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
/**
* Writes a data frame.
* @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 {
+ private void writeDataFrame(final Object body, final long length) throws IOException {
ensureNoException();
if (closed) {
throw new IOException("Stream already closed.");
}
channel.writeAndFlush(DataFrameEncoder.DataFrame.newInstance(getContextId(), body, length, newSubStream))
- .addListener(getChannelWriteListener());
+ .addListener(getChannelWriteListener());
newSubStream = false;
}
}
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/DataFetcherOutputCollector.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/DataFetcherOutputCollector.java
index d50ad82fc..9995e6a92 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/DataFetcherOutputCollector.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/DataFetcherOutputCollector.java
@@ -37,6 +37,8 @@
/**
* It forwards output to the next operator.
* @param nextOperatorVertex next operator to emit data and watermark
+ * @param edgeIndex edge index
+ * @param watermarkManager watermark manager
*/
public DataFetcherOutputCollector(final OperatorVertex nextOperatorVertex,
final int edgeIndex,
diff --git a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java
index 03d7470f0..f93797519 100644
--- a/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java
+++ b/runtime/executor/src/main/java/org/apache/nemo/runtime/executor/datatransfer/PipeOutputWriter.java
@@ -18,14 +18,11 @@
*/
package org.apache.nemo.runtime.executor.datatransfer;
-import org.apache.nemo.common.DirectByteArrayOutputStream;
-import org.apache.nemo.common.coder.EncoderFactory;
import org.apache.nemo.common.ir.edge.executionproperty.CommunicationPatternProperty;
import org.apache.nemo.common.punctuation.Watermark;
import org.apache.nemo.runtime.common.RuntimeIdManager;
import org.apache.nemo.runtime.common.plan.RuntimeEdge;
import org.apache.nemo.runtime.executor.bytetransfer.ByteOutputContext;
-import org.apache.nemo.runtime.executor.data.DataUtil;
import org.apache.nemo.runtime.executor.data.PipeManagerWorker;
import org.apache.nemo.runtime.executor.data.partitioner.Partitioner;
import org.apache.nemo.runtime.executor.data.streamchainer.Serializer;
@@ -33,7 +30,6 @@
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.io.OutputStream;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
@@ -79,16 +75,7 @@
private void writeData(final Object element, final List<ByteOutputContext> pipeList) {
pipeList.forEach(pipe -> {
try (final ByteOutputContext.ByteOutputStream pipeToWriteTo = pipe.newOutputStream()) {
- // Serialize (Do not compress)
- final DirectByteArrayOutputStream bytesOutputStream = new DirectByteArrayOutputStream();
- final OutputStream wrapped =
- DataUtil.buildOutputStream(bytesOutputStream, serializer.getEncodeStreamChainers());
- final EncoderFactory.Encoder encoder = serializer.getEncoderFactory().create(wrapped);
- encoder.encode(element);
- wrapped.close();
-
- // Write
- pipeToWriteTo.write(bytesOutputStream.getBufDirectly());
+ pipeToWriteTo.writeElement(element, serializer);
} catch (IOException e) {
throw new RuntimeException(e); // For now we crash the executor on IOException
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services