You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by vrozov <gi...@git.apache.org> on 2015/11/11 03:15:41 UTC
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
GitHub user vrozov opened a pull request:
https://github.com/apache/incubator-apex-core/pull/162
Introduce Abstract and Forwarding Reservoir classes - review
@243826, @tweise, @PramodSSImmaneni please review
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/vrozov/incubator-apex-core APEX-254
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-core/pull/162.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #162
----
commit 8134affdd4bf509b49971a16112169b807fdbc95
Author: Vlad Rozov <v....@datatorrent.com>
Date: 2015-11-11T01:42:11Z
APEX-254 - Introduce Abstract and Forwarding Reservoir classes.
commit ce05aa2803b6ff6e8f9bdd0e5322ebe33a25a982
Author: Vlad Rozov <v....@datatorrent.com>
Date: 2015-11-11T01:43:06Z
APEX-254 - Introduce Abstract and Forwarding Reservoir classes.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by 243826 <gi...@git.apache.org>.
Github user 243826 commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/162#issuecomment-155655936
not able to get infer the cause to do this, where are the details?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov closed the pull request at:
https://github.com/apache/incubator-apex-core/pull/162
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44896738
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
@tweise : This is used during dynamic partitioning based on backlog. User needs to set the Annotation on the Operator to use this expensive flavor
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by 243826 <gi...@git.apache.org>.
Github user 243826 commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/162#issuecomment-155927418
Vlad, I have a bunch of questions and comments. Let's sync offline.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44946066
--- Diff: engine/src/main/java/com/datatorrent/stram/engine/AbstractReservoir.java ---
@@ -0,0 +1,340 @@
+/**
+ * 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 com.datatorrent.stram.engine;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datatorrent.api.Sink;
+import com.datatorrent.netlet.util.CircularBuffer;
+import com.datatorrent.netlet.util.UnsafeBlockingQueue;
+import com.datatorrent.stram.tuple.Tuple;
+
+public abstract class AbstractReservoir implements SweepableReservoir, BlockingQueue<Object>
--- End diff --
Will do.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44896500
--- Diff: engine/src/main/java/com/datatorrent/stram/engine/Reservoir.java ---
@@ -23,20 +23,26 @@
*
* @since 0.3.2
*/
-public interface Reservoir
+public interface Reservoir<T>
{
/**
* the count of elements in this SweepableReservoir.
*
* @return the count
*/
- public int size();
+ int size(final boolean dataTupleAware);
--- End diff --
@vrozov dataTupleAware: This Boolean tells that give the count of only data tuples and not control tuples.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44895568
--- Diff: engine/src/main/java/com/datatorrent/stram/engine/Reservoir.java ---
@@ -23,20 +23,26 @@
*
* @since 0.3.2
*/
-public interface Reservoir
+public interface Reservoir<T>
{
/**
* the count of elements in this SweepableReservoir.
*
* @return the count
*/
- public int size();
+ int size(final boolean dataTupleAware);
--- End diff --
Please document the parameter.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44947533
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
@gauravgopi123 How critical is to filter out control tuples for dynamic partitioning? Is there a check against zero data tuple aware queue size?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44946039
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java ---
@@ -90,7 +91,7 @@ public void put(Object tuple)
@Override
public String toString()
{
- return "InlineStream{" + super.toString() + '}';
+ return getClass().getName() + '@' + Integer.toHexString(hashCode()) + "{reservoir=" + getReservoir().toString() + '}';
--- End diff --
@tweise verbose code or verbose output? The verbose output is used for logging and debugging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/162#issuecomment-156951850
+1 for abstracting the queue. Did you run the benchmark with this change? As discussed, the extra level of indirection in DefaultReservoir should not add significant cost as the compiler can optimize it?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44896265
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
In which situation is this expensive flavor of size() used?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/162#issuecomment-155829466
Added more details to APEX-254 JIRA - it is a refactoring of existing functionality so there is a point in git that may be used to compare and verify functionality and performance before JCTools Spsc queue is introduced.
Is there a way to preserve renaming history in git? I introduced the second commit to preserve DefaultReservoir->AbstractReservoir renaming history, but it does not seem to work.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44970614
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
@gauravgopi123 I don't see how control tuples may significantly affect queue size. Assuming any reasonable queue capacity there will be only few control tuples in the queue compared to payload tuples and they should not make much difference on backlog. Is there a test case that demonstrates significance of excluding control tuples from the queue size?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44895894
--- Diff: engine/src/main/java/com/datatorrent/stram/engine/AbstractReservoir.java ---
@@ -0,0 +1,340 @@
+/**
+ * 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 com.datatorrent.stram.engine;
+
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.datatorrent.api.Sink;
+import com.datatorrent.netlet.util.CircularBuffer;
+import com.datatorrent.netlet.util.UnsafeBlockingQueue;
+import com.datatorrent.stram.tuple.Tuple;
+
+public abstract class AbstractReservoir implements SweepableReservoir, BlockingQueue<Object>
--- End diff --
Add documentation.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on the pull request:
https://github.com/apache/incubator-apex-core/pull/162#issuecomment-157626756
Obsoleted by #173
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by tweise <gi...@git.apache.org>.
Github user tweise commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44895766
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/InlineStream.java ---
@@ -90,7 +91,7 @@ public void put(Object tuple)
@Override
public String toString()
{
- return "InlineStream{" + super.toString() + '}';
+ return getClass().getName() + '@' + Integer.toHexString(hashCode()) + "{reservoir=" + getReservoir().toString() + '}';
--- End diff --
Why so verbose?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44938812
--- Diff: engine/src/main/java/com/datatorrent/stram/engine/Reservoir.java ---
@@ -23,20 +23,26 @@
*
* @since 0.3.2
*/
-public interface Reservoir
+public interface Reservoir<T>
{
/**
* the count of elements in this SweepableReservoir.
*
* @return the count
*/
- public int size();
+ int size(final boolean dataTupleAware);
--- End diff --
@gauravgopi123 Other than for CircularBufferReservoir this parameter will not be supported. Getting **exact** queue size is way too expensive, so SpscArrayQueue gives an upper estimate on the queue size. Filtering out control tuples is even more costly.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44948463
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
@vrozov : It is very critical for certain use cases and it is applied only when the StatsListener on the operator is annotation with StatsListener.DataQueueSize
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44958755
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
@vrozov : Use case when you want to dynamically partition an operator based on queue size of backlog. For that you don't want to count control tuples, you want to calculate only data tuples..
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44956255
--- Diff: engine/src/main/java/com/datatorrent/stram/stream/BufferServerSubscriber.java ---
@@ -242,6 +243,21 @@ public long getByteCount(boolean reset)
}
@Override
+ public int size(final boolean dataTupleAware)
+ {
+ int size = size();
+ if (dataTupleAware) {
--- End diff --
@gauravgopi123 Can you please describe use case? As for SpscArrayQueue size() is an estimate, accounting for control tuples does not provide any better estimate.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-apex-core pull request: Introduce Abstract and Forwardin...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-core/pull/162#discussion_r44943593
--- Diff: engine/src/main/java/com/datatorrent/stram/engine/Reservoir.java ---
@@ -23,20 +23,26 @@
*
* @since 0.3.2
*/
-public interface Reservoir
+public interface Reservoir<T>
{
/**
* the count of elements in this SweepableReservoir.
*
* @return the count
*/
- public int size();
+ int size(final boolean dataTupleAware);
--- End diff --
I agree.. I was just giving documentation of the parameter as it was me who has added this check
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---