Removed non-daemon dataflow parallel group

Author: vaclav

build.gradle

@@ -277,6 +277,8 @@ task demo(type: DemoTask, dependsOn: 'compileGroovy') {
             'DemoSieveEratosthenesTheGoWay',              //Never stops
             'DemoSieveEratosthenesTheGoWayWithOperators', //Never stops
             'DemoThreading',                              //Never stops
+            'DemoProducerConsumer1',                      //Never stops
+            'DemoPhysicalCalculations',                   //Needs user input
             'DemoFibonacci1',                             //Needs classes from its source folder
             'DemoFibonacci2',                             //Needs classes from its source folder
             'DemoNumbers',                                //Needs classes from its source folder

grails-doc/src/guide/2.6 What's new.gdoc

@@ -41,6 +41,7 @@ h3. Dataflow
 * Fixed issues with reusing the same parallel group in the DataFlowVariable whenBound() handlers
 * Tasks can return values and can be joined
 * DataFlowQueue's _whenBound_ renamed to _wheneverBound_ and _whenNextBound_ renamed to _whenBound_ to obey the _DataFlowChannel_ interface contract
+* DataFlow default parallel group has been made daemon
 
 h3. Agent
 

grails-doc/src/guide/7. Dataflow Concurrency.gdoc

@@ -12,7 +12,6 @@ final def z = new DataFlowVariable()
 
 task {
     z << x.val + y.val
-    println "Result: ${z.val}"
 }
 
 task {
@@ -22,6 +21,8 @@ task {
 task {
     y << 5
 }
+
+println "Result: ${z.val}"
 {code}
 
 Or the same algorithm rewritten using the _DataFlows_ class.
@@ -34,7 +35,6 @@ final def df = new DataFlows()
 
 task {
     df.z = df.x + df.y
-    println "Result: ${df.z}"
 }
 
 task {
@@ -44,6 +44,9 @@ task {
 task {
     df.y = 5
 }
+
+println "Result: ${df.z}"
+
 {code}
 
 We start three logical tasks, which can run in parallel and perform their particular activities. The tasks need to exchange data and they do so using *Dataflow Variables*.

grails-doc/src/guide/7.1 Tasks.gdoc

@@ -47,8 +47,7 @@ task {
                     it.val.toUpperCase().contains 'GROOVY'
                 }).size()
     }
-    System.exit 0
-}
+}.join()
 {code}
 
 h3. Grouping tasks
@@ -73,7 +72,7 @@ group.with {
 {code}
 
 {note:Title=Custom thread pools for dataflow}
-The default thread pool for dataflow tasks contains non-daemon threads, which means your application will not exit before all tasks complete.
+The default thread pool for dataflow tasks contains daemon threads, which means your application will exit as soon as the main thread finishes and won't wait for all tasks to complete.
 When grouping tasks, make sure that your custom thread pools either use daemon threads, too, which can be achieved by
 using DefaultPGroup or by providing your own thread factory to a thread pool constructor,
 or in case your thread pools use non-daemon threads, such as when using the NonDaemonPGroup group class, make sure you shutdown the group or the thread pool explicitly by calling its shutdown() method,
@@ -104,8 +103,7 @@ task {
                     it.val.toUpperCase().contains 'GROOVY'
                 }).size()
     }
-    System.exit 0
-}
+}.join()
 
 def downloadPage(def url) {
     def page = new DataFlowVariable()
@@ -138,7 +136,7 @@ final def decelerationForce = new DataFlowVariable()
 final def deceleration = new DataFlowVariable()
 final def distance = new DataFlowVariable()
 
-task {
+def t = task {
     println """
 Calculating distance required to stop a moving ball.
 ====================================================
@@ -190,6 +188,9 @@ task {
 task {
     distance << deceleration.val * ((velocity.val/deceleration.val) ** 2) * 0.5
 }
+
+t.join()
+
 {code}Note: I did my best to make all the physical calculations right. Feel free to change the values and see how long distance you need to stop the rolling ball.
 
 h2. Deterministic deadlocks

grails-doc/src/guide/7.3 Operators.gdoc

@@ -235,7 +235,7 @@ group.with {
 {code}
 
 {note:Title=Custom thread pools for dataflow}
-The default thread pool for dataflow operators contains non-daemon threads, which means your application will not exit before all operators are stopped.
+The default thread pool for dataflow operators contains daemon threads, which means your application will exit as soon as the main thread finishes and won't wait for all tasks to complete.
 When grouping operators, make sure that your custom thread pools either use daemon threads, too, which can be achieved by
 using DefaultPGroup or by providing your own thread factory to a thread pool constructor,
 or in case your thread pools use non-daemon threads, such as when using the NonDaemonPGroup group class, make sure you shutdown the group or the thread pool explicitly by calling its shutdown() method,

src/main/groovy/groovyx/gpars/dataflow/DataFlow.java

@@ -19,7 +19,9 @@
 import groovy.lang.Closure;
 import groovyx.gpars.actor.Actor;
 import groovyx.gpars.dataflow.operator.DataFlowProcessor;
+import groovyx.gpars.group.DefaultPGroup;
 import groovyx.gpars.group.PGroup;
+import groovyx.gpars.scheduler.ResizeablePool;
 
 import java.util.List;
 import java.util.Map;
@@ -37,7 +39,7 @@ public abstract class DataFlow {
     /**
      * The parallel group used by all Dataflow Concurrency actors by default.
      */
-    public static final DataFlowPGroup DATA_FLOW_GROUP = new DataFlowPGroup(1);
+    public static final PGroup DATA_FLOW_GROUP = new DefaultPGroup(new ResizeablePool(true, 1));
 
     /**
      * Maps threads/tasks to parallel groups they belong to

src/main/groovy/groovyx/gpars/dataflow/DataFlowPGroup.java

@@ -19,7 +19,7 @@ package groovyx.gpars.samples.dataflow
 import groovyx.gpars.dataflow.DataFlowVariable
 import java.util.concurrent.Executors
 
-final many = 1..(limit)
+final many = 1..(100)
 
 List dfs = many.collect { new DataFlowVariable() }
 def result = new DataFlowVariable()

src/test/groovy/groovyx/gpars/samples/dataflow/BenchmarkManyDataFlowVariables.groovy

@@ -41,4 +41,4 @@ def y = new DataFlowVariable<List<Integer>>()
 
 task { x << ints(0, 500) }
 task { y << sum(0, x.val) }
-task { println("List of sums: " + y.val); System.exit(0) }
+task { println("List of sums: " + y.val); System.exit(0) }.join()

src/test/groovy/groovyx/gpars/samples/dataflow/DataFlowDemo2.groovy

@@ -40,5 +40,4 @@ task {
 task {
     Thread.sleep(1000)
     println "Sum: ${producer.collect {it * it}.inject(0) {sum, x -> sum + x}}"
-    System.exit 0
-}
+}.join()

src/test/groovy/groovyx/gpars/samples/dataflow/DataFlowDemo3.groovy

@@ -39,4 +39,4 @@ task {
 
 task {
     y << 5
-}
+}.join()

src/test/groovy/groovyx/gpars/samples/dataflow/DataFlowDemo6.groovy

@@ -27,7 +27,7 @@ import groovyx.gpars.scheduler.ResizeablePool
  * @author Vaclav Pech
  */
 
-final def group = new DefaultPGroup(new ResizeablePool(false))
+final def group = new DefaultPGroup(new ResizeablePool(true))
 
 final def a = new DataFlowVariable()
 final def b = new DataFlowVariable()
@@ -39,14 +39,14 @@ group.with {
 
     task {
         println "Result: ${b.val}"
-        System.exit 0
+        group.shutdown()
     }
 
     Thread.sleep 2000
 
     task {
         a << 10
     }
+    b.join()
 }
 
-group.shutdown()

src/test/groovy/groovyx/gpars/samples/dataflow/DemoAvoidPotentialDeadlock1.groovy

@@ -1,58 +1,58 @@
-// GPars - Groovy Parallel Systems
-//
-// Copyright © 2008-10  The original author or authors
-//
-// Licensed 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 groovyx.gpars.samples.dataflow
-
-import groovyx.gpars.dataflow.DataFlowVariable
-import groovyx.gpars.group.DefaultPGroup
-import groovyx.gpars.group.PGroup
-import groovyx.gpars.scheduler.DefaultPool
-
-/**
- * Demonstrates deadlock prevention using the whenBound (>>) dataflow construct. Instead of blocking the task
- * together with its underlying thread on a read call to DataFlowVariable.val, the whenBound handler will be invoked only
- * after the value of the DataFlowVariable is set and so the reader doesn't consume a thread while waiting for a value
- * to arrive.
- * The code would end up deadlocked if we blocked the threads during reads, since the first two tasks
- * wait for each other to bind values to a and b. Only the third thread can unlock the two threads by setting value of a.
- *
- * @author Vaclav Pech
- */
-
-final PGroup group = new DefaultPGroup(new DefaultPool(false, 1))
-
-final def a = new DataFlowVariable()
-final def b = new DataFlowVariable()
-
-group.with {
-    task {
-        a >> {b << 20 + it}
-    }
-
-    task {
-        b >> {
-            println "Result: ${it}"
-            System.exit 0
-        }
-    }
-
-    Thread.sleep 2000
-
-    task {
-        a << 10
-    }
-
+// GPars - Groovy Parallel Systems
+//
+// Copyright © 2008-10  The original author or authors
+//
+// Licensed 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 groovyx.gpars.samples.dataflow
+
+import groovyx.gpars.dataflow.DataFlowVariable
+import groovyx.gpars.group.DefaultPGroup
+import groovyx.gpars.group.PGroup
+import groovyx.gpars.scheduler.DefaultPool
+
+/**
+ * Demonstrates deadlock prevention using the whenBound (>>) dataflow construct. Instead of blocking the task
+ * together with its underlying thread on a read call to DataFlowVariable.val, the whenBound handler will be invoked only
+ * after the value of the DataFlowVariable is set and so the reader doesn't consume a thread while waiting for a value
+ * to arrive.
+ * The code would end up deadlocked if we blocked the threads during reads, since the first two tasks
+ * wait for each other to bind values to a and b. Only the third thread can unlock the two threads by setting value of a.
+ *
+ * @author Vaclav Pech
+ */
+
+final PGroup group = new DefaultPGroup(new DefaultPool(true, 1))
+
+final def a = new DataFlowVariable()
+final def b = new DataFlowVariable()
+
+group.with {
+    task {
+        a >> {b << 20 + it}
+    }
+
+    task {
+        b >> {
+            println "Result: ${it}"
+            group.shutdown()
+        }
+    }
+
+    Thread.sleep 2000
+
+    task {
+        a << 10
+    }
+    b.join()
 }

src/test/groovy/groovyx/gpars/samples/dataflow/DemoAvoidingPotentialDeadlock2.groovy

@@ -50,4 +50,3 @@ println 'Reading from the stream'
 println ''
 println "The stream is now empty. Length =  ${stream.length()}"
 
-System.exit 0