You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2016/04/18 20:40:38 UTC
[1/2] incubator-apex-core git commit: APEXCORE-413 - Collision
between Sink.getCount() and SweepableReservoir.getCount()
Repository: incubator-apex-core
Updated Branches:
refs/heads/master 170eb4706 -> c2008f27a
APEXCORE-413 - Collision between Sink.getCount() and SweepableReservoir.getCount()
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/676ab410
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/676ab410
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/676ab410
Branch: refs/heads/master
Commit: 676ab4101f091bea0cdffde125e64ddcaf7d991f
Parents: 750d90d
Author: Vlad Rozov <v....@datatorrent.com>
Authored: Thu Mar 31 20:41:05 2016 -0700
Committer: Vlad Rozov <v....@datatorrent.com>
Committed: Fri Apr 15 08:44:03 2016 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/engine/OiONode.java | 11 +-
.../stram/engine/StreamingContainer.java | 9 +-
.../datatorrent/stram/stream/InlineStream.java | 33 +++++-
.../com/datatorrent/stram/stream/OiOStream.java | 105 ++++++++++++-------
.../stram/stream/InlineStreamTest.java | 8 +-
5 files changed, 112 insertions(+), 54 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/676ab410/engine/src/main/java/com/datatorrent/stram/engine/OiONode.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/engine/OiONode.java b/engine/src/main/java/com/datatorrent/stram/engine/OiONode.java
index 485651a..f968b4e 100644
--- a/engine/src/main/java/com/datatorrent/stram/engine/OiONode.java
+++ b/engine/src/main/java/com/datatorrent/stram/engine/OiONode.java
@@ -29,7 +29,6 @@ import com.datatorrent.api.Operator;
import com.datatorrent.api.Operator.InputPort;
import com.datatorrent.api.Sink;
import com.datatorrent.stram.plan.logical.Operators.PortContextPair;
-import com.datatorrent.stram.stream.OiOStream;
import com.datatorrent.stram.tuple.Tuple;
/**
@@ -51,11 +50,11 @@ public class OiONode extends GenericNode
super(operator, context);
}
- class ControlSink implements Sink<Tuple>
+ private class ControlSink implements Sink<Tuple>
{
final SweepableReservoir reservoir;
- ControlSink(SweepableReservoir sr)
+ private ControlSink(SweepableReservoir sr)
{
reservoir = sr;
}
@@ -159,11 +158,9 @@ public class OiONode extends GenericNode
}
- @Override
- public void connectInputPort(String port, SweepableReservoir reservoir)
+ public Sink<Tuple> getControlSink(SweepableReservoir reservoir)
{
- ((OiOStream)reservoir).setControlSink(new ControlSink(reservoir));
- super.connectInputPort(port, reservoir);
+ return new ControlSink(reservoir);
}
@SuppressWarnings("unused")
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/676ab410/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java b/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java
index ddde53c..2436776 100644
--- a/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java
+++ b/engine/src/main/java/com/datatorrent/stram/engine/StreamingContainer.java
@@ -1144,6 +1144,7 @@ public class StreamingContainer extends YarnContainerMain
inlineContext.setSinkId(sinkIdentifier);
Stream stream;
+ SweepableReservoir reservoir;
switch (nidi.locality) {
case CONTAINER_LOCAL:
int outputQueueCapacity = getOutputQueueCapacity(operatorList, nidi.sourceNodeId, nidi.sourcePortName);
@@ -1152,13 +1153,17 @@ public class StreamingContainer extends YarnContainerMain
}
stream = new InlineStream(queueCapacity);
+ reservoir = ((InlineStream)stream).getReservoir();
if (checkpoint.windowId >= 0) {
- node.connectInputPort(nidi.portName, new WindowIdActivatedReservoir(sinkIdentifier, (SweepableReservoir)stream, checkpoint.windowId));
+ node.connectInputPort(nidi.portName, new WindowIdActivatedReservoir(sinkIdentifier, reservoir,
+ checkpoint.windowId));
}
break;
case THREAD_LOCAL:
stream = new OiOStream();
+ reservoir = ((OiOStream)stream).getReservoir();
+ ((OiOStream.OiOReservoir)reservoir).setControlSink(((OiONode)node).getControlSink(reservoir));
oioNodes.put(ndi.id, nidi.sourceNodeId);
break;
@@ -1166,7 +1171,7 @@ public class StreamingContainer extends YarnContainerMain
throw new IllegalStateException("Locality can be either ContainerLocal or ThreadLocal");
}
- node.connectInputPort(nidi.portName, (SweepableReservoir)stream);
+ node.connectInputPort(nidi.portName, reservoir);
newStreams.put(sinkIdentifier, new ComponentContextPair<>(stream, inlineContext));
if (!(pair.component instanceof Stream.MultiSinkCapableStream)) {
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/676ab410/engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java b/engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java
index e909b80..ec9660b 100644
--- a/engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java
+++ b/engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java
@@ -22,10 +22,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.datatorrent.stram.engine.AbstractReservoir;
-import com.datatorrent.stram.engine.ForwardingReservoir;
import com.datatorrent.stram.engine.Stream;
import com.datatorrent.stram.engine.StreamContext;
import com.datatorrent.stram.engine.SweepableReservoir;
+import com.datatorrent.stram.tuple.Tuple;
/**
*
@@ -34,11 +34,19 @@ import com.datatorrent.stram.engine.SweepableReservoir;
*
* @since 0.3.2
*/
-public class InlineStream extends ForwardingReservoir implements Stream, SweepableReservoir
+public class InlineStream implements Stream
{
+ private int count;
+ private AbstractReservoir reservoir;
+
public InlineStream(int capacity)
{
- super(AbstractReservoir.newReservoir("InlineStream", capacity));
+ reservoir = AbstractReservoir.newReservoir("InlineStream", capacity);
+ }
+
+ public SweepableReservoir getReservoir()
+ {
+ return reservoir;
}
/**
@@ -48,7 +56,7 @@ public class InlineStream extends ForwardingReservoir implements Stream, Sweepab
@Override
public void setup(StreamContext context)
{
- setId(context.getId());
+ reservoir.setId(context.getId());
}
/**
@@ -80,7 +88,10 @@ public class InlineStream extends ForwardingReservoir implements Stream, Sweepab
public void put(Object tuple)
{
try {
- super.put(tuple);
+ reservoir.put(tuple);
+ if (!(tuple instanceof Tuple)) {
+ count++;
+ }
} catch (InterruptedException ie) {
logger.debug("Interrupted", ie);
throw new RuntimeException(ie);
@@ -88,6 +99,18 @@ public class InlineStream extends ForwardingReservoir implements Stream, Sweepab
}
@Override
+ public int getCount(boolean reset)
+ {
+ try {
+ return count;
+ } finally {
+ if (reset) {
+ count = 0;
+ }
+ }
+ }
+
+ @Override
public String toString()
{
return getClass().getName() + '@' + Integer.toHexString(hashCode()) + "{reservoir=" + getReservoir().toString() + '}';
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/676ab410/engine/src/main/java/com/datatorrent/stram/stream/OiOStream.java
----------------------------------------------------------------------
diff --git a/engine/src/main/java/com/datatorrent/stram/stream/OiOStream.java b/engine/src/main/java/com/datatorrent/stram/stream/OiOStream.java
index 7732ffc..61ed0e6 100644
--- a/engine/src/main/java/com/datatorrent/stram/stream/OiOStream.java
+++ b/engine/src/main/java/com/datatorrent/stram/stream/OiOStream.java
@@ -25,15 +25,24 @@ import com.datatorrent.stram.engine.SweepableReservoir;
import com.datatorrent.stram.tuple.Tuple;
/**
- * A non buffering stream which facilitates the ThreadLocal implementation of an operator.
+ * A non buffering stream which facilitates the ThreadLocal implementation of an operator.</p>
+ *
+ * {@link OiOStream} can not implement both {@link Stream} and {@link SweepableReservoir} interfaces as
+ * {@code getCount(boolean)} should reset the proper count. In case, OioStream implements both {@link Stream} and
+ * {@link SweepableReservoir}, it will not be possible to distinguish which count should be reset to 0, so either
+ * {@link Sink#getCount(boolean)} or {@link SweepableReservoir#getCount(boolean)} would return a wrong result: let's say
+ * upstream operator puts n tuples into {@link OiOStream}, both {@link Sink#getCount(boolean) Sink.getCount(true)} and
+ * {@link SweepableReservoir#getCount(boolean) SweepableReservoir.getCount(true)} should return n and reset count to 0.
+ * If there is only one count, only one call (whichever comes first) would return n, the second one would return 0.
*
* @since 0.3.5
*/
-public class OiOStream implements Stream, SweepableReservoir
+public class OiOStream implements Stream
{
private Sink<Object> sink;
private Sink<Tuple> control;
private int count;
+ private OiOReservoir reservoir = new OiOReservoir();
@Override
public void setup(StreamContext context)
@@ -45,11 +54,6 @@ public class OiOStream implements Stream, SweepableReservoir
{
}
- public void setControlSink(Sink<Tuple> control)
- {
- this.control = control;
- }
-
@Override
public void activate(StreamContext cntxt)
{
@@ -67,6 +71,7 @@ public class OiOStream implements Stream, SweepableReservoir
control.put((Tuple)t);
} else {
count++;
+ reservoir.count++;
sink.put(t);
}
}
@@ -83,42 +88,68 @@ public class OiOStream implements Stream, SweepableReservoir
}
}
- @Override
- public Sink<Object> setSink(Sink<Object> sink)
+ public SweepableReservoir getReservoir()
{
- try {
- return this.sink;
- } finally {
- this.sink = sink;
- }
+ return reservoir;
}
- @Override
- public Tuple sweep()
+ public class OiOReservoir implements SweepableReservoir
{
- throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
- }
+ private int count;
- /**
- * OiOStream is active when there is exactly one tuple present.
- * It's an error to have more than one tuple active on OiO.
- */
- @Override
- public int size(final boolean dataTupleAware)
- {
- return 1;
- }
+ public void setControlSink(Sink<Tuple> control)
+ {
+ OiOStream.this.control = control;
+ }
- @Override
- public boolean isEmpty()
- {
- return false;
- }
+ @Override
+ public int getCount(boolean reset)
+ {
+ try {
+ return count;
+ } finally {
+ if (reset) {
+ count = 0;
+ }
+ }
+ }
- @Override
- public Object remove()
- {
- throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
- }
+ @Override
+ public Sink<Object> setSink(Sink<Object> sink)
+ {
+ try {
+ return OiOStream.this.sink;
+ } finally {
+ OiOStream.this.sink = sink;
+ }
+ }
+ @Override
+ public Tuple sweep()
+ {
+ throw new UnsupportedOperationException("Not supported.");
+ }
+
+ /**
+ * OiOStream is active when there is exactly one tuple present.
+ * It's an error to have more than one tuple active on OiO.
+ */
+ @Override
+ public int size(boolean dataTupleAware)
+ {
+ return 1;
+ }
+
+ @Override
+ public boolean isEmpty()
+ {
+ return false;
+ }
+
+ @Override
+ public Object remove()
+ {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-apex-core/blob/676ab410/engine/src/test/java/com/datatorrent/stram/stream/InlineStreamTest.java
----------------------------------------------------------------------
diff --git a/engine/src/test/java/com/datatorrent/stram/stream/InlineStreamTest.java b/engine/src/test/java/com/datatorrent/stram/stream/InlineStreamTest.java
index 69d50ce..ef63a62 100644
--- a/engine/src/test/java/com/datatorrent/stram/stream/InlineStreamTest.java
+++ b/engine/src/test/java/com/datatorrent/stram/stream/InlineStreamTest.java
@@ -37,6 +37,7 @@ import com.datatorrent.stram.engine.GenericNode;
import com.datatorrent.stram.engine.Node;
import com.datatorrent.stram.engine.OperatorContext;
import com.datatorrent.stram.engine.StreamContext;
+import com.datatorrent.stram.engine.SweepableReservoir;
import com.datatorrent.stram.support.StramTestSupport;
import com.datatorrent.stram.support.StramTestSupport.WaitCondition;
import com.datatorrent.stram.tuple.Tuple;
@@ -69,7 +70,7 @@ public class InlineStreamTest
stream.setup(streamContext);
node1.connectOutputPort("output", stream);
- node2.connectInputPort("input", stream);
+ node2.connectInputPort("input", stream.getReservoir());
prev = null;
Sink<Object> sink = new Sink<Object>()
@@ -137,8 +138,9 @@ public class InlineStreamTest
@Override
public boolean isComplete()
{
- logger.debug("stream {} empty {}, size {}", stream, stream.isEmpty(), stream.size(false));
- return stream.isEmpty();
+ final SweepableReservoir reservoir = stream.getReservoir();
+ logger.debug("stream {} empty {}, size {}", stream, reservoir.isEmpty(), reservoir.size(false));
+ return reservoir.isEmpty();
}
};
[2/2] incubator-apex-core git commit: Merge branch 'APEXCORE-413' of
https://github.com/vrozov/incubator-apex-core
Posted by th...@apache.org.
Merge branch 'APEXCORE-413' of https://github.com/vrozov/incubator-apex-core
Project: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/commit/c2008f27
Tree: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/tree/c2008f27
Diff: http://git-wip-us.apache.org/repos/asf/incubator-apex-core/diff/c2008f27
Branch: refs/heads/master
Commit: c2008f27a18d4324974823d7fcde7be2820365d5
Parents: 170eb47 676ab41
Author: Thomas Weise <th...@datatorrent.com>
Authored: Mon Apr 18 11:27:21 2016 -0700
Committer: Thomas Weise <th...@datatorrent.com>
Committed: Mon Apr 18 11:27:21 2016 -0700
----------------------------------------------------------------------
.../com/datatorrent/stram/engine/OiONode.java | 11 +-
.../stram/engine/StreamingContainer.java | 9 +-
.../datatorrent/stram/stream/InlineStream.java | 33 +++++-
.../com/datatorrent/stram/stream/OiOStream.java | 105 ++++++++++++-------
.../stram/stream/InlineStreamTest.java | 8 +-
5 files changed, 112 insertions(+), 54 deletions(-)
----------------------------------------------------------------------