You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by chinmaykolhatkar <gi...@git.apache.org> on 2016/01/04 12:45:49 UTC
[GitHub] incubator-apex-malhar pull request: [Review Only] AbstractAsyncPro...
GitHub user chinmaykolhatkar opened a pull request:
https://github.com/apache/incubator-apex-malhar/pull/157
[Review Only] AbstractAsyncProcessor processes tuples async
Processes tuples async/parallel threads.
Also takes care of order and timeout of processing.
Also take care of checkpointing.
@gauravgopi123 Please review.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/chinmaykolhatkar/incubator-apex-malhar APEXMALHAR-1963_AsyncProcessor
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-apex-malhar/pull/157.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 #157
----
commit 1c9461f625f59a4c78f7480af24ae410a06a0448
Author: chinmaykolhatkar <ch...@datatorrent.com>
Date: 2016-01-04T11:43:19Z
Added AbstractAsyncProcessor.
----
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48948897
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
+ private boolean maintainTupleOrder = false;
+ private long processTimeoutMs = 0;
+
+ /**
+ * This Queue will hold all the tuples that are in process, has completed processing but yet to be notified for
+ * processing completion.
+ */
+ private Queue<ProcessTuple> waitingTuples = Lists.newLinkedList();
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
+ {
+ @Override public void process(INPUT input)
+ {
+ processTuple(input);
+ }
+
+ @Override public void setup(Context.PortContext context)
+ {
+ setupInputPort(context);
+ }
+ };
+
+ /**
+ * This method is exposed for implementing class so as to take care of input port based initialization.
+ * For eg. Retrieve TUPLE_CLASS attribute from input port.
+ *
+ * @param context PortContext of input port
+ */
+ protected void setupInputPort(Context.PortContext context)
+ {
+ // Do nothing. Concrete class may choose to do override this. For eg. reading TUPLE_CLASS attribute.
+ }
+
+ /**
+ * Setup method of this operator will initialize the various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ if (numProcessors == 1) {
+ this.executor = Executors.newSingleThreadExecutor();
+ } else if (numProcessors == 0) {
+ this.executor = Executors.newCachedThreadPool();
+ } else if (numProcessors > 0) {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
--- End diff --
Do we need to distinguish between Executors.newSingleThreadExecutor() and Executors.newFixedThreadPool(numProcessors) with numProcessors = 1 ? Would it be functionally different ?
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r51079084
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,341 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default 32 threads will be created.
+ * These worker threads, spawned by operator thread, can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * Minimum value is 1 and max allowed number of threads are 32.
+ *
+ * Default number of threads is 32.
+ */
+ @Min(1)
+ @Max(32)
+ private int numProcessors = 32;
+
+ /**
+ * Tells whether the <i>handleProcessedTuple</i> should be called for tuples in the same order in which
+ * they get enqueued using <i>enqueueTupleForProcessing</i>.
+ */
+ private boolean maintainTupleOrder = false;
+
+ /**
+ * Timeout in milliseconds after which processing of tuple may be terminated.
+ */
+ private long processTimeoutMs = 0;
+
+ /**
+ * The queue holds following types of tuples:
+ * <ul>
+ * <li>Yet to be processed</li>
+ * <li>In Process</li>
+ * <li>Completed processing but yet to be notified for process completion</li>
+ * </ul>
+ */
+ private Queue<WorkItem> waitingTuples = Lists.newLinkedList();
+
+ /**
+ * Setup method of this operator will initialize various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
+ replayTuplesIfRequired();
+ }
+
+ /**
+ * This method is called from setup of the operator. If <i>waitingTuples</i> contains anything, the method enqueues
+ * all those tuples for processing to worker threads.
+ *
+ * Dependening on condition the method will behave as follows:
+ * <ul>
+ * <li>During fresh start of operator, this method will not have any effect as there are no tuples in
+ * <i>waitingTuples</i>.</li>
+ * <li>During restore of operator to previous checkpointed state, if there are any tuples already present
+ * in <i>waitingTuples</i> that means they're yet to be notified, hence enqueued again.</li>
+ * </ul>
+ */
+ private void replayTuplesIfRequired()
+ {
+ // Enqueue everything that is still left.
+ // Its left because its either not completed processing OR its not emitted because of ordering restrictions.
+ for (WorkItem item : waitingTuples) {
+ item.taskHandle = executor.submit(new Processor(item));
+ item.processState = State.SUBMITTED;
--- End diff --
This possibly overwrites state of processed item.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49882703
--- Diff: library/src/test/java/com/datatorrent/lib/async/AbstractAsyncProcessorTest.java ---
@@ -0,0 +1,145 @@
+/**
+ * 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.lib.async;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+
+public class AbstractAsyncProcessorTest
+{
+ public static class TestAsyncProcessor extends AbstractAsyncProcessor<String, String>
+ {
+ public transient final DefaultInputPort<String> input = new DefaultInputPort<String>()
+ {
+ @Override public void process(String s)
+ {
+ enqueueTupleForProcessing(s);
+ }
+ };
+ public transient final DefaultOutputPort<String> output = new DefaultOutputPort<>();
+ public transient final DefaultOutputPort<String> error = new DefaultOutputPort<>();
+
+ @Override protected void handleProcessedTuple(String inpTuple, String resultTuple, State processState)
+ {
+ if (processState == State.SUCCESS) {
+ output.emit(resultTuple);
+ } else {
+ error.emit(inpTuple);
+ }
+ }
+
+ @Override protected String processTupleAsync(String tuple)
+ {
+ int i = Integer.parseInt(tuple) % 3;
+ switch (i % 3) {
+ case 1:
+ return tuple + ";FIRSTPASS";
+ case 2:
+ return tuple + ";SECONDPASS";
+ case 0:
+ default:
+ throw new RuntimeException("Expected");
+ }
+ }
+ }
+
+ @Test public void maintainOrderTest() throws InterruptedException
+ {
+ testOperator(1, true, 5000);
+ testOperator(1, true, 15000);
+ }
+
+ @Test public void noOrderMaintainedTest() throws InterruptedException
+ {
+ testOperator(1, false, 5000);
+ testOperator(2, false, 15000);
+ }
+
+ private void testOperator(int numThreads, boolean maintainOrder, long waitInterval) throws InterruptedException
+ {
+ CollectorTestSink<Object> sinkOut = new CollectorTestSink<>();
+ CollectorTestSink<Object> sinkErr = new CollectorTestSink<>();
+
+ TestAsyncProcessor async = new TestAsyncProcessor();
+ async.setNumProcessors(numThreads);
+ async.setMaintainTupleOrder(maintainOrder);
+ async.output.setSink(sinkOut);
+ async.error.setSink(sinkErr);
+ async.setup(null);
+
+ for (int windowId = 0; windowId < 10; windowId++) {
+ async.beginWindow(windowId);
+ for (int i = 0; i < 100; i++) {
+ async.input.put(Integer.toString(windowId * 100 + i));
+ }
+ async.endWindow();
+ }
+
+ for (int i = 0; i < 10; i++) {
+ Thread.sleep(waitInterval / 10);
+ async.endWindow();
+ }
--- End diff --
what is this loop doing? endWindow() is called for every beginWindow and that has been done in previous loop (line 91-97)
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49928282
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,344 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default
+ * threads will be created on need basis with no upper limit. These worker threads, spawned by operator thread,
+ * can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * 0 - Threads will be created as per need.
--- End diff --
So here is what I've done:
Min = 1 & Max = 32.
32 is the cap on number of threads enforced by javax Max annotation.
Default is set to 32.
Let me know if some other cap on number of threads need to be there.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48937769
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
--- End diff --
at *the* end of
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059426
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
+ private boolean maintainTupleOrder = false;
+ private long processTimeoutMs = 0;
+
+ /**
+ * This Queue will hold all the tuples that are in process, has completed processing but yet to be notified for
+ * processing completion.
+ */
+ private Queue<ProcessTuple> waitingTuples = Lists.newLinkedList();
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
+ {
+ @Override public void process(INPUT input)
+ {
+ processTuple(input);
+ }
+
+ @Override public void setup(Context.PortContext context)
+ {
+ setupInputPort(context);
+ }
+ };
+
+ /**
+ * This method is exposed for implementing class so as to take care of input port based initialization.
+ * For eg. Retrieve TUPLE_CLASS attribute from input port.
+ *
+ * @param context PortContext of input port
+ */
+ protected void setupInputPort(Context.PortContext context)
+ {
+ // Do nothing. Concrete class may choose to do override this. For eg. reading TUPLE_CLASS attribute.
+ }
+
+ /**
+ * Setup method of this operator will initialize the various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ if (numProcessors == 1) {
+ this.executor = Executors.newSingleThreadExecutor();
+ } else if (numProcessors == 0) {
+ this.executor = Executors.newCachedThreadPool();
+ } else if (numProcessors > 0) {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
+ } else {
--- End diff --
Done.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by vrozov <gi...@git.apache.org>.
Github user vrozov commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r51079142
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,341 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default 32 threads will be created.
+ * These worker threads, spawned by operator thread, can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * Minimum value is 1 and max allowed number of threads are 32.
+ *
+ * Default number of threads is 32.
+ */
+ @Min(1)
+ @Max(32)
+ private int numProcessors = 32;
+
+ /**
+ * Tells whether the <i>handleProcessedTuple</i> should be called for tuples in the same order in which
+ * they get enqueued using <i>enqueueTupleForProcessing</i>.
+ */
+ private boolean maintainTupleOrder = false;
+
+ /**
+ * Timeout in milliseconds after which processing of tuple may be terminated.
+ */
+ private long processTimeoutMs = 0;
+
+ /**
+ * The queue holds following types of tuples:
+ * <ul>
+ * <li>Yet to be processed</li>
+ * <li>In Process</li>
+ * <li>Completed processing but yet to be notified for process completion</li>
+ * </ul>
+ */
+ private Queue<WorkItem> waitingTuples = Lists.newLinkedList();
+
+ /**
+ * Setup method of this operator will initialize various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
+ replayTuplesIfRequired();
+ }
+
+ /**
+ * This method is called from setup of the operator. If <i>waitingTuples</i> contains anything, the method enqueues
+ * all those tuples for processing to worker threads.
+ *
+ * Dependening on condition the method will behave as follows:
+ * <ul>
+ * <li>During fresh start of operator, this method will not have any effect as there are no tuples in
+ * <i>waitingTuples</i>.</li>
+ * <li>During restore of operator to previous checkpointed state, if there are any tuples already present
+ * in <i>waitingTuples</i> that means they're yet to be notified, hence enqueued again.</li>
+ * </ul>
+ */
+ private void replayTuplesIfRequired()
+ {
+ // Enqueue everything that is still left.
+ // Its left because its either not completed processing OR its not emitted because of ordering restrictions.
+ for (WorkItem item : waitingTuples) {
+ item.taskHandle = executor.submit(new Processor(item));
+ item.processState = State.SUBMITTED;
+ item.processStartTime = 0;
+ item.outTuple = null;
+ }
+ }
+
+ /**
+ * Teardown of operator which stops executor service.
+ */
+ @Override public void teardown()
+ {
+ this.executor.shutdownNow();
+ }
+
+ /**
+ * Enqueues tuple of type QUEUETUPLE for processing to worker threads.
+ * This method should be called only from operator thread.
+ *
+ * @param QUEUETUPLE Input tuple of type QUEUETUPLE
+ */
+ protected void enqueueTupleForProcessing(QUEUETUPLE QUEUETUPLE)
+ {
+ WorkItem item = new WorkItem(QUEUETUPLE);
+ waitingTuples.add(item);
+ item.taskHandle = executor.submit(new Processor(item));
+ item.processState = State.SUBMITTED;
--- End diff --
This possibly overwrites state of processed item.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49882090
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,344 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default
+ * threads will be created on need basis with no upper limit. These worker threads, spawned by operator thread,
+ * can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * 0 - Threads will be created as per need.
--- End diff --
I don't think creating threads on need basis is good idea.. You need to have control on the number of threads
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r51114146
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,341 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default 32 threads will be created.
+ * These worker threads, spawned by operator thread, can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * Minimum value is 1 and max allowed number of threads are 32.
+ *
+ * Default number of threads is 32.
+ */
+ @Min(1)
+ @Max(32)
+ private int numProcessors = 32;
+
+ /**
+ * Tells whether the <i>handleProcessedTuple</i> should be called for tuples in the same order in which
+ * they get enqueued using <i>enqueueTupleForProcessing</i>.
+ */
+ private boolean maintainTupleOrder = false;
+
+ /**
+ * Timeout in milliseconds after which processing of tuple may be terminated.
+ */
+ private long processTimeoutMs = 0;
+
+ /**
+ * The queue holds following types of tuples:
+ * <ul>
+ * <li>Yet to be processed</li>
+ * <li>In Process</li>
+ * <li>Completed processing but yet to be notified for process completion</li>
+ * </ul>
+ */
+ private Queue<WorkItem> waitingTuples = Lists.newLinkedList();
+
+ /**
+ * Setup method of this operator will initialize various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
+ replayTuplesIfRequired();
+ }
+
+ /**
+ * This method is called from setup of the operator. If <i>waitingTuples</i> contains anything, the method enqueues
+ * all those tuples for processing to worker threads.
+ *
+ * Dependening on condition the method will behave as follows:
+ * <ul>
+ * <li>During fresh start of operator, this method will not have any effect as there are no tuples in
+ * <i>waitingTuples</i>.</li>
+ * <li>During restore of operator to previous checkpointed state, if there are any tuples already present
+ * in <i>waitingTuples</i> that means they're yet to be notified, hence enqueued again.</li>
+ * </ul>
+ */
+ private void replayTuplesIfRequired()
+ {
+ // Enqueue everything that is still left.
+ // Its left because its either not completed processing OR its not emitted because of ordering restrictions.
+ for (WorkItem item : waitingTuples) {
+ item.taskHandle = executor.submit(new Processor(item));
+ item.processState = State.SUBMITTED;
--- End diff --
calling executor.submit after variables are set. This should take care of overrides.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48939626
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
--- End diff --
Default of 1 for this purpose is not recommended. That will almost make it serial for each tuple. Suggest to increase it to a higher number. There is no perfect number for this. It will certainly depend on the system you are interacting with.
Suggested number is 16 at least.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar closed the pull request at:
https://github.com/apache/incubator-apex-malhar/pull/157
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059185
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
--- End diff --
Done
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059306
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
--- End diff --
Done.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r51114181
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,341 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Max;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default 32 threads will be created.
+ * These worker threads, spawned by operator thread, can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * Minimum value is 1 and max allowed number of threads are 32.
+ *
+ * Default number of threads is 32.
+ */
+ @Min(1)
+ @Max(32)
+ private int numProcessors = 32;
+
+ /**
+ * Tells whether the <i>handleProcessedTuple</i> should be called for tuples in the same order in which
+ * they get enqueued using <i>enqueueTupleForProcessing</i>.
+ */
+ private boolean maintainTupleOrder = false;
+
+ /**
+ * Timeout in milliseconds after which processing of tuple may be terminated.
+ */
+ private long processTimeoutMs = 0;
+
+ /**
+ * The queue holds following types of tuples:
+ * <ul>
+ * <li>Yet to be processed</li>
+ * <li>In Process</li>
+ * <li>Completed processing but yet to be notified for process completion</li>
+ * </ul>
+ */
+ private Queue<WorkItem> waitingTuples = Lists.newLinkedList();
+
+ /**
+ * Setup method of this operator will initialize various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
+ replayTuplesIfRequired();
+ }
+
+ /**
+ * This method is called from setup of the operator. If <i>waitingTuples</i> contains anything, the method enqueues
+ * all those tuples for processing to worker threads.
+ *
+ * Dependening on condition the method will behave as follows:
+ * <ul>
+ * <li>During fresh start of operator, this method will not have any effect as there are no tuples in
+ * <i>waitingTuples</i>.</li>
+ * <li>During restore of operator to previous checkpointed state, if there are any tuples already present
+ * in <i>waitingTuples</i> that means they're yet to be notified, hence enqueued again.</li>
+ * </ul>
+ */
+ private void replayTuplesIfRequired()
+ {
+ // Enqueue everything that is still left.
+ // Its left because its either not completed processing OR its not emitted because of ordering restrictions.
+ for (WorkItem item : waitingTuples) {
+ item.taskHandle = executor.submit(new Processor(item));
+ item.processState = State.SUBMITTED;
+ item.processStartTime = 0;
+ item.outTuple = null;
+ }
+ }
+
+ /**
+ * Teardown of operator which stops executor service.
+ */
+ @Override public void teardown()
+ {
+ this.executor.shutdownNow();
+ }
+
+ /**
+ * Enqueues tuple of type QUEUETUPLE for processing to worker threads.
+ * This method should be called only from operator thread.
+ *
+ * @param QUEUETUPLE Input tuple of type QUEUETUPLE
+ */
+ protected void enqueueTupleForProcessing(QUEUETUPLE QUEUETUPLE)
+ {
+ WorkItem item = new WorkItem(QUEUETUPLE);
+ waitingTuples.add(item);
+ item.taskHandle = executor.submit(new Processor(item));
+ item.processState = State.SUBMITTED;
--- End diff --
calling executor.submit after variables are set. This should take care of overrides.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48937866
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
--- End diff --
-> When a tuple
Give some name to the other thread.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48937830
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
--- End diff --
Once -> One
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48948739
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
+ private boolean maintainTupleOrder = false;
+ private long processTimeoutMs = 0;
+
+ /**
+ * This Queue will hold all the tuples that are in process, has completed processing but yet to be notified for
+ * processing completion.
+ */
+ private Queue<ProcessTuple> waitingTuples = Lists.newLinkedList();
+
+ @InputPortFieldAnnotation(optional = true)
--- End diff --
why optional ?
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059730
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
+ private boolean maintainTupleOrder = false;
+ private long processTimeoutMs = 0;
+
+ /**
+ * This Queue will hold all the tuples that are in process, has completed processing but yet to be notified for
+ * processing completion.
+ */
+ private Queue<ProcessTuple> waitingTuples = Lists.newLinkedList();
+
+ @InputPortFieldAnnotation(optional = true)
--- End diff --
I'm actually considering to remove the input port form this.
I just need to ensure that this abstract operator is bound to operator lifecycle, so that the checkpointing can be handled properly.
The abstract has provided enqueue method which should take care of queing the tuples for processing.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059547
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
+ private boolean maintainTupleOrder = false;
+ private long processTimeoutMs = 0;
+
+ /**
+ * This Queue will hold all the tuples that are in process, has completed processing but yet to be notified for
+ * processing completion.
+ */
+ private Queue<ProcessTuple> waitingTuples = Lists.newLinkedList();
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
+ {
+ @Override public void process(INPUT input)
+ {
+ processTuple(input);
+ }
+
+ @Override public void setup(Context.PortContext context)
+ {
+ setupInputPort(context);
+ }
+ };
+
+ /**
+ * This method is exposed for implementing class so as to take care of input port based initialization.
+ * For eg. Retrieve TUPLE_CLASS attribute from input port.
+ *
+ * @param context PortContext of input port
+ */
+ protected void setupInputPort(Context.PortContext context)
+ {
+ // Do nothing. Concrete class may choose to do override this. For eg. reading TUPLE_CLASS attribute.
+ }
+
+ /**
+ * Setup method of this operator will initialize the various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ if (numProcessors == 1) {
+ this.executor = Executors.newSingleThreadExecutor();
+ } else if (numProcessors == 0) {
+ this.executor = Executors.newCachedThreadPool();
+ } else if (numProcessors > 0) {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
--- End diff --
Done.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by gauravgopi123 <gi...@git.apache.org>.
Github user gauravgopi123 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49882952
--- Diff: library/src/test/java/com/datatorrent/lib/async/AbstractAsyncProcessorTest.java ---
@@ -0,0 +1,145 @@
+/**
+ * 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.lib.async;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+
+public class AbstractAsyncProcessorTest
+{
+ public static class TestAsyncProcessor extends AbstractAsyncProcessor<String, String>
+ {
+ public transient final DefaultInputPort<String> input = new DefaultInputPort<String>()
+ {
+ @Override public void process(String s)
+ {
+ enqueueTupleForProcessing(s);
+ }
+ };
+ public transient final DefaultOutputPort<String> output = new DefaultOutputPort<>();
+ public transient final DefaultOutputPort<String> error = new DefaultOutputPort<>();
+
+ @Override protected void handleProcessedTuple(String inpTuple, String resultTuple, State processState)
+ {
+ if (processState == State.SUCCESS) {
+ output.emit(resultTuple);
+ } else {
+ error.emit(inpTuple);
+ }
+ }
+
+ @Override protected String processTupleAsync(String tuple)
+ {
+ int i = Integer.parseInt(tuple) % 3;
+ switch (i % 3) {
+ case 1:
+ return tuple + ";FIRSTPASS";
+ case 2:
+ return tuple + ";SECONDPASS";
+ case 0:
+ default:
+ throw new RuntimeException("Expected");
+ }
+ }
+ }
+
+ @Test public void maintainOrderTest() throws InterruptedException
+ {
+ testOperator(1, true, 5000);
+ testOperator(1, true, 15000);
+ }
+
+ @Test public void noOrderMaintainedTest() throws InterruptedException
+ {
+ testOperator(1, false, 5000);
+ testOperator(2, false, 15000);
+ }
+
+ private void testOperator(int numThreads, boolean maintainOrder, long waitInterval) throws InterruptedException
+ {
+ CollectorTestSink<Object> sinkOut = new CollectorTestSink<>();
+ CollectorTestSink<Object> sinkErr = new CollectorTestSink<>();
+
+ TestAsyncProcessor async = new TestAsyncProcessor();
+ async.setNumProcessors(numThreads);
+ async.setMaintainTupleOrder(maintainOrder);
+ async.output.setSink(sinkOut);
+ async.error.setSink(sinkErr);
+ async.setup(null);
+
+ for (int windowId = 0; windowId < 10; windowId++) {
+ async.beginWindow(windowId);
+ for (int i = 0; i < 100; i++) {
+ async.input.put(Integer.toString(windowId * 100 + i));
+ }
+ async.endWindow();
+ }
+
+ for (int i = 0; i < 10; i++) {
+ Thread.sleep(waitInterval / 10);
+ async.endWindow();
+ }
+ async.teardown();
+
+ List<Object> outTuples = sinkOut.collectedTuples;
+ List<Object> errTuples = sinkErr.collectedTuples;
+ Assert.assertEquals(666, outTuples.size());
+ Assert.assertEquals(334, errTuples.size());
+
+ int previous = -1;
+ for (Object outTuple : outTuples) {
+ Assert.assertTrue(outTuple instanceof String);
+ String outTuple1 = (String)outTuple;
+ String[] split = outTuple1.split(";");
+ Assert.assertEquals(2, split.length);
+ int current = Integer.parseInt(split[0]);
--- End diff --
why not use swtich case?
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by sandeepdeshmukh <gi...@git.apache.org>.
Github user sandeepdeshmukh commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48940518
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
+ private boolean maintainTupleOrder = false;
+ private long processTimeoutMs = 0;
+
+ /**
+ * This Queue will hold all the tuples that are in process, has completed processing but yet to be notified for
+ * processing completion.
+ */
+ private Queue<ProcessTuple> waitingTuples = Lists.newLinkedList();
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<INPUT> input = new DefaultInputPort<INPUT>()
+ {
+ @Override public void process(INPUT input)
+ {
+ processTuple(input);
+ }
+
+ @Override public void setup(Context.PortContext context)
+ {
+ setupInputPort(context);
+ }
+ };
+
+ /**
+ * This method is exposed for implementing class so as to take care of input port based initialization.
+ * For eg. Retrieve TUPLE_CLASS attribute from input port.
+ *
+ * @param context PortContext of input port
+ */
+ protected void setupInputPort(Context.PortContext context)
+ {
+ // Do nothing. Concrete class may choose to do override this. For eg. reading TUPLE_CLASS attribute.
+ }
+
+ /**
+ * Setup method of this operator will initialize the various types of executors possible.
+ *
+ * @param context OperatorContext
+ */
+ @Override public void setup(Context.OperatorContext context)
+ {
+ if (numProcessors == 1) {
+ this.executor = Executors.newSingleThreadExecutor();
+ } else if (numProcessors == 0) {
+ this.executor = Executors.newCachedThreadPool();
+ } else if (numProcessors > 0) {
+ this.executor = Executors.newFixedThreadPool(numProcessors);
+ } else {
--- End diff --
@Min = 0 for numProcessors? Will save this else case.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by shubham-pathak22 <gi...@git.apache.org>.
Github user shubham-pathak22 commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r48949023
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
--- End diff --
Do we need to have max limit ?
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49928172
--- Diff: library/src/test/java/com/datatorrent/lib/async/AbstractAsyncProcessorTest.java ---
@@ -0,0 +1,145 @@
+/**
+ * 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.lib.async;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+
+public class AbstractAsyncProcessorTest
+{
+ public static class TestAsyncProcessor extends AbstractAsyncProcessor<String, String>
+ {
+ public transient final DefaultInputPort<String> input = new DefaultInputPort<String>()
+ {
+ @Override public void process(String s)
+ {
+ enqueueTupleForProcessing(s);
+ }
+ };
+ public transient final DefaultOutputPort<String> output = new DefaultOutputPort<>();
+ public transient final DefaultOutputPort<String> error = new DefaultOutputPort<>();
+
+ @Override protected void handleProcessedTuple(String inpTuple, String resultTuple, State processState)
+ {
+ if (processState == State.SUCCESS) {
+ output.emit(resultTuple);
+ } else {
+ error.emit(inpTuple);
+ }
+ }
+
+ @Override protected String processTupleAsync(String tuple)
+ {
+ int i = Integer.parseInt(tuple) % 3;
+ switch (i % 3) {
+ case 1:
+ return tuple + ";FIRSTPASS";
+ case 2:
+ return tuple + ";SECONDPASS";
+ case 0:
+ default:
+ throw new RuntimeException("Expected");
+ }
+ }
+ }
+
+ @Test public void maintainOrderTest() throws InterruptedException
+ {
+ testOperator(1, true, 5000);
+ testOperator(1, true, 15000);
+ }
+
+ @Test public void noOrderMaintainedTest() throws InterruptedException
+ {
+ testOperator(1, false, 5000);
+ testOperator(2, false, 15000);
+ }
+
+ private void testOperator(int numThreads, boolean maintainOrder, long waitInterval) throws InterruptedException
+ {
+ CollectorTestSink<Object> sinkOut = new CollectorTestSink<>();
+ CollectorTestSink<Object> sinkErr = new CollectorTestSink<>();
+
+ TestAsyncProcessor async = new TestAsyncProcessor();
+ async.setNumProcessors(numThreads);
+ async.setMaintainTupleOrder(maintainOrder);
+ async.output.setSink(sinkOut);
+ async.error.setSink(sinkErr);
+ async.setup(null);
+
+ for (int windowId = 0; windowId < 10; windowId++) {
+ async.beginWindow(windowId);
+ for (int i = 0; i < 100; i++) {
+ async.input.put(Integer.toString(windowId * 100 + i));
+ }
+ async.endWindow();
+ }
+
+ for (int i = 0; i < 10; i++) {
+ Thread.sleep(waitInterval / 10);
+ async.endWindow();
+ }
--- End diff --
Got it. I should have added the beginWindow also here. Will update.
This loop is just to make sure to wait for all the items to finish processing.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059172
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
--- End diff --
Done
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49928222
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,344 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import javax.validation.constraints.Min;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.common.util.BaseOperator;
+
+
+/**
+ * This base operator can be used to enqueue tuples for asynchronous and parallel processing by separate worker threads.
+ *
+ * User can configure number of threads required for async processing. By default
+ * threads will be created on need basis with no upper limit. These worker threads, spawned by operator thread,
+ * can also be configured to processing timeout.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Implementing class can enqueue the tuples of type QUEUETUPLE using <i>enqueueTupleForProcessing</i>
+ * method.</li>
+ * <li>When a tuple gets its turn for execution, <i>processTupleAsync</i> gets called
+ * by worker thread. This is asynchronous call</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at the end of every window and respective
+ * call to <i>handleProcessedTuple</i> will be made. This callback is in operator thread.</li>
+ * </ol>
+ *
+ *
+ * <b>Use case examples:</b>
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Operations which are long running tasks per tuple but DAG i/o should not be blocked.</li>
+ * </ul>
+ *
+ * @param <QUEUETUPLE> input type of tuple
+ * @param <RESULTTUPLE> Result of async processing of tuple.
+ */
+public abstract class AbstractAsyncProcessor<QUEUETUPLE, RESULTTUPLE> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ /**
+ * This is an executor service which holds thread pool and enqueues task for processing.
+ */
+ private transient ExecutorService executor;
+
+ /**
+ * Number of worker threads for processing
+ * 0 - Threads will be created as per need.
--- End diff --
I agree.What would you suggest a good ma number will be. I'll remove the cached pool all together.
And put a limit on that.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49928190
--- Diff: library/src/test/java/com/datatorrent/lib/async/AbstractAsyncProcessorTest.java ---
@@ -0,0 +1,145 @@
+/**
+ * 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.lib.async;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.lib.testbench.CollectorTestSink;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+
+public class AbstractAsyncProcessorTest
+{
+ public static class TestAsyncProcessor extends AbstractAsyncProcessor<String, String>
+ {
+ public transient final DefaultInputPort<String> input = new DefaultInputPort<String>()
+ {
+ @Override public void process(String s)
+ {
+ enqueueTupleForProcessing(s);
+ }
+ };
+ public transient final DefaultOutputPort<String> output = new DefaultOutputPort<>();
+ public transient final DefaultOutputPort<String> error = new DefaultOutputPort<>();
+
+ @Override protected void handleProcessedTuple(String inpTuple, String resultTuple, State processState)
+ {
+ if (processState == State.SUCCESS) {
+ output.emit(resultTuple);
+ } else {
+ error.emit(inpTuple);
+ }
+ }
+
+ @Override protected String processTupleAsync(String tuple)
+ {
+ int i = Integer.parseInt(tuple) % 3;
+ switch (i % 3) {
+ case 1:
+ return tuple + ";FIRSTPASS";
+ case 2:
+ return tuple + ";SECONDPASS";
+ case 0:
+ default:
+ throw new RuntimeException("Expected");
+ }
+ }
+ }
+
+ @Test public void maintainOrderTest() throws InterruptedException
+ {
+ testOperator(1, true, 5000);
+ testOperator(1, true, 15000);
+ }
+
+ @Test public void noOrderMaintainedTest() throws InterruptedException
+ {
+ testOperator(1, false, 5000);
+ testOperator(2, false, 15000);
+ }
+
+ private void testOperator(int numThreads, boolean maintainOrder, long waitInterval) throws InterruptedException
+ {
+ CollectorTestSink<Object> sinkOut = new CollectorTestSink<>();
+ CollectorTestSink<Object> sinkErr = new CollectorTestSink<>();
+
+ TestAsyncProcessor async = new TestAsyncProcessor();
+ async.setNumProcessors(numThreads);
+ async.setMaintainTupleOrder(maintainOrder);
+ async.output.setSink(sinkOut);
+ async.error.setSink(sinkErr);
+ async.setup(null);
+
+ for (int windowId = 0; windowId < 10; windowId++) {
+ async.beginWindow(windowId);
+ for (int i = 0; i < 100; i++) {
+ async.input.put(Integer.toString(windowId * 100 + i));
+ }
+ async.endWindow();
+ }
+
+ for (int i = 0; i < 10; i++) {
+ Thread.sleep(waitInterval / 10);
+ async.endWindow();
+ }
+ async.teardown();
+
+ List<Object> outTuples = sinkOut.collectedTuples;
+ List<Object> errTuples = sinkErr.collectedTuples;
+ Assert.assertEquals(666, outTuples.size());
+ Assert.assertEquals(334, errTuples.size());
+
+ int previous = -1;
+ for (Object outTuple : outTuples) {
+ Assert.assertTrue(outTuple instanceof String);
+ String outTuple1 = (String)outTuple;
+ String[] split = outTuple1.split(";");
+ Assert.assertEquals(2, split.length);
+ int current = Integer.parseInt(split[0]);
--- End diff --
Done.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49059361
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
--- End diff --
Made it 16.
---
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-malhar pull request: APEXMALHAR-1963 AbstractAsyncP...
Posted by chinmaykolhatkar <gi...@git.apache.org>.
Github user chinmaykolhatkar commented on a diff in the pull request:
https://github.com/apache/incubator-apex-malhar/pull/157#discussion_r49061423
--- Diff: library/src/main/java/com/datatorrent/lib/async/AbstractAsyncProcessor.java ---
@@ -0,0 +1,357 @@
+/**
+ * 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.lib.async;
+
+import java.util.Iterator;
+import java.util.Queue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.Operator.IdleTimeHandler;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+/**
+ * This base operator having a single input and tuple coming from input port are enqueued for processing asynchronously
+ * by Thread Executors.
+ *
+ * User can configure number of of threads that are required for processing. Defaulting to single thread which is
+ * different than operator thread.
+ * The user will also configure the timeout for the operations to happen.
+ *
+ * <b>Workflow is as follows:</b>
+ * <ol>
+ * <li>Tuple of type INPUT is received on input port.</li>
+ * <li>Tuple gets enqueued into Executors by calling <i>enqueueTupleForProcessing</i>.
+ * Once can override <i>processTuple</i> method in which case, its users responsibility to enqueue tuple for
+ * processing by calling <i>enqueueTupleForProcessing</i>.</li>
+ * <li>When tuple gets its turn of execution, <i>processTupleAsync</i> gets called from thread other than
+ * operator thread.</li>
+ * <li>Based on maintainTupleOrder parameter, the results are consolidated at end of every window and respective call
+ * to <i>handleProcessedTuple</i> will be made. This call will happen in operator thread.</li>
+ * </ol>
+ *
+ * This operator can be implemented to do async and parallel operations.
+ *
+ * Use case examples:
+ * <ul>
+ * <li>Parallel reads from external datastore/database system.</li>
+ * <li>Doing any operations which are long running tasks per tuple but DAG io should not be blocked.</li>
+ * </ul>
+ *
+ * @param <INPUT> input type of tuple
+ * @param <RESULT> Result of async processing of tuple.
+ * @since 3.3.0
+ */
+public abstract class AbstractAsyncProcessor<INPUT, RESULT> extends BaseOperator implements IdleTimeHandler
+{
+ private static final Logger logger = LoggerFactory.getLogger(AbstractAsyncProcessor.class);
+
+ private transient ExecutorService executor;
+ private int numProcessors = 1;
--- End diff --
Actually let me keep this to 0. This means, the threads will be created as per need.
---
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.
---