Skip to content

Commit ed1df63

Browse files
zzw4githubzzw4github
committed
CountedCompleter
1 parent b9016e5 commit ed1df63

25 files changed

Lines changed: 862 additions & 36 deletions

File tree

‎akka/src/main/java/actors/props/DemoActor.java‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@ public class DemoActor extends UntypedActor {
1414
public static Props props(final int magicNumber) {
1515
return Props.create(new Creator<DemoActor>() {
1616
private static final long serialVersionUID = 1L;
17-
1817
@Override
1918
public DemoActor create() {
2019
return new DemoActor(magicNumber);

‎commons-io/pom.xml‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<parent>
6+
<artifactId>tutorial</artifactId>
7+
<groupId>me.zzw.app</groupId>
8+
<version>1.0-SNAPSHOT</version>
9+
</parent>
10+
<modelVersion>4.0.0</modelVersion>
11+
12+
<artifactId>commons-io</artifactId>
13+
<dependencies>
14+
<dependency>
15+
<groupId>commons-io</groupId>
16+
<artifactId>commons-io</artifactId>
17+
<version>2.5</version>
18+
</dependency>
19+
20+
</dependencies>
21+
22+
</project>

‎concurrent/src/main/me.zzw.app.concurrent/TCountedCompleter.java‎

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,204 @@
1+
package main.me.zzw.app.concurrent;
2+
3+
import java.util.concurrent.CountedCompleter;
4+
import java.util.concurrent.atomic.AtomicReference;
5+
6+
/**
7+
* Created by infosea on 2016-09-09.
8+
*/
9+
class MyOperation<E> {
10+
void apply(E e) {
11+
System.out.println(e);
12+
}
13+
}
14+
15+
class ForEach<E> extends CountedCompleter<Void> {
16+
17+
public static <E> void forEach(E[] array, MyOperation<E> op) {
18+
new ForEach<E>(null, array, op, 0, array.length).invoke();
19+
}
20+
21+
final E[] array; final MyOperation<E> op; final int lo, hi;
22+
ForEach(CountedCompleter<?> p, E[] array, MyOperation<E> op, int lo, int hi){
23+
super(p);
24+
this.array = array;
25+
this.op = op;
26+
this.lo = lo;
27+
this.hi = hi;
28+
}
29+
30+
@Override
31+
public void compute() {
32+
if(hi -lo >=2 ){
33+
int mid = (lo + hi) >>> 1;
34+
setPendingCount(2); // must set pending count before fork
35+
new ForEach(this, array, op, mid, hi).fork(); // right child;
36+
new ForEach(this, array, op, lo, mid).fork(); // left child
37+
}else if (hi > lo)
38+
op.apply(array[lo]);
39+
tryComplete();
40+
}
41+
}
42+
43+
class ForEach2<E> extends CountedCompleter<Void>{
44+
45+
public static <E> void forEach(E[] array, MyOperation<E> op) {
46+
new ForEach<E>(null, array, op, 0, array.length).invoke();
47+
}
48+
49+
final E[] array; final MyOperation<E> op; final int lo, hi;
50+
ForEach2(CountedCompleter<?> p, E[] array, MyOperation<E> op, int lo, int hi){
51+
super(p);
52+
this.array = array;
53+
this.op = op;
54+
this.lo = lo;
55+
this.hi = hi;
56+
}
57+
58+
public void compute() { // version 2
59+
if (hi - lo >= 2) {
60+
int mid = (lo + hi) >>> 1;
61+
setPendingCount(1); // only one pending
62+
new ForEach2(this, array, op, mid, hi).fork(); // right child
63+
new ForEach2(this, array, op, lo, mid).compute(); // direct invoke
64+
} else {
65+
if (hi > lo)
66+
op.apply(array[lo]);
67+
tryComplete();
68+
}
69+
}
70+
}
71+
72+
class ForEach3<E> extends CountedCompleter<Void> {
73+
74+
public static <E> void forEach(E[] array, MyOperation<E> op) {
75+
new ForEach3<E>(null, array, op, 0, array.length).invoke();
76+
}
77+
78+
final E[] array;
79+
final MyOperation<E> op;
80+
final int lo, hi;
81+
82+
ForEach3(CountedCompleter<?> p, E[] array, MyOperation<E> op, int lo, int hi) {
83+
super(p);
84+
this.array = array;
85+
this.op = op;
86+
this.lo = lo;
87+
this.hi = hi;
88+
}
89+
90+
public void compute() { // version 3
91+
int l = lo, h = hi;
92+
while (h - l >= 2) {
93+
int mid = (l + h) >>> 1;
94+
addToPendingCount(1);
95+
new ForEach3(this, array, op, mid, h).fork(); // right child
96+
h = mid;
97+
}
98+
if (h > l)
99+
op.apply(array[l]);
100+
propagateCompletion();
101+
}
102+
}
103+
104+
class Searcher<E> extends CountedCompleter<E> {
105+
106+
final E[] array; final AtomicReference<E> result; final int lo, hi;
107+
108+
Searcher(CountedCompleter<?> p, E[] array, AtomicReference<E> result, int lo, int hi) {
109+
super(p);
110+
this.array = array; this.result = result; this.lo = lo; this.hi = hi;
111+
}
112+
113+
public E getRawResult() { return result.get(); }
114+
115+
public void compute() { // similar to ForEach version 3
116+
int l = lo, h = hi;
117+
while (result.get() == null && h >= l) {
118+
if (h - l >= 2) {
119+
int mid = (l + h) >>> 1;
120+
addToPendingCount(1);
121+
new Searcher(this, array, result, mid, h).fork();
122+
h = mid;
123+
}
124+
else {
125+
E x = array[l];
126+
if (matches(x) && result.compareAndSet(null, x))
127+
quietlyCompleteRoot(); // root task is now joinable
128+
break;
129+
}
130+
}
131+
tryComplete(); // normally complete whether or not found
132+
}
133+
134+
boolean matches(E e) {
135+
return true;
136+
} // return true if found
137+
138+
139+
public static <E> E search(E[] array) {
140+
return new Searcher<E>(null, array, new AtomicReference<E>(), 0, array.length).invoke();
141+
}
142+
143+
}
144+
145+
class MyMapper<E> { E apply(E v) { ... } }
146+
class MyReducer<E> { E apply(E x, E y) { ... } }
147+
class MapReducer<E> extends CountedCompleter<E> {
148+
final E[] array; final MyMapper<E> mapper;
149+
final MyReducer<E> reducer; final int lo, hi;
150+
MapReducer<E> sibling;
151+
E result;
152+
MapReducer(CountedCompleter<?> p, E[] array, MyMapper<E> mapper,
153+
MyReducer<E> reducer, int lo, int hi) {
154+
super(p);
155+
this.array = array; this.mapper = mapper;
156+
this.reducer = reducer; this.lo = lo; this.hi = hi;
157+
}
158+
public void compute() {
159+
if (hi - lo >= 2) {
160+
int mid = (lo + hi) >>> 1;
161+
MapReducer<E> left = new MapReducer(this, array, mapper, reducer, lo, mid);
162+
MapReducer<E> right = new MapReducer(this, array, mapper, reducer, mid, hi);
163+
left.sibling = right;
164+
right.sibling = left;
165+
setPendingCount(1); // only right is pending
166+
right.fork();
167+
left.compute(); // directly execute left
168+
}
169+
else {
170+
if (hi > lo)
171+
result = mapper.apply(array[lo]);
172+
tryComplete();
173+
}
174+
}
175+
public void onCompletion(CountedCompleter<?> caller) {
176+
if (caller != this) {
177+
MapReducer<E> child = (MapReducer<E>)caller;
178+
MapReducer<E> sib = child.sibling;
179+
if (sib == null || sib.result == null)
180+
result = child.result;
181+
else
182+
result = reducer.apply(child.result, sib.result);
183+
}
184+
}
185+
public E getRawResult() { return result; }
186+
187+
public static <E> E mapReduce(E[] array, MyMapper<E> mapper, MyReducer<E> reducer) {
188+
return new MapReducer<E>(null, array, mapper, reducer,
189+
0, array.length).invoke();
190+
}
191+
}
192+
193+
194+
195+
public class TCountedCompleter {
196+
public static void main(String[] args) {
197+
ForEach.forEach(new Integer[]{2,6,4,3,6,7,8,4,3,6,8,9,0,5,4,3,2}, new MyOperation<Integer>());
198+
System.out.println("-----------");
199+
ForEach2.forEach(new Integer[]{2,6,4,3,6,7,8,4,3,6,8,9,0,5,4,3,2}, new MyOperation<Integer>());
200+
System.out.println("-----------");
201+
ForEach3.forEach(new Integer[]{2,6,4,3,6,7,8,4,3,6,8,9,0,5,4,3,2}, new MyOperation<Integer>());
202+
203+
}
204+
}

‎concurrent/src/main/me.zzw.app.concurrent/TForkJoinTask.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package main.me.zzw.app.concurrent;
2+
3+
import java.util.concurrent.ExecutionException;
4+
import java.util.concurrent.ForkJoinTask;
5+
import java.util.concurrent.RecursiveTask;
6+
7+
/**
8+
* Created by infosea on 2016-09-09.
9+
*/
10+
public class TForkJoinTask {
11+
public static String splie(String s){
12+
return s;
13+
}
14+
public static void main(String[] args) throws ExecutionException, InterruptedException {
15+
String s = "abcdefg";
16+
ForkJoinTask<String> task = new RecursiveTask<String>() {
17+
@Override
18+
protected String compute() {
19+
if(s.length()>6){
20+
//split(s);
21+
}
22+
return s;
23+
}
24+
};
25+
task.invoke();
26+
String ss = task.get();
27+
System.out.println(ss);
28+
}
29+
}

‎jdk8/pom.xml‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,16 @@
1616
<artifactId>guava</artifactId>
1717
<version>19.0</version>
1818
</dependency>
19+
<dependency>
20+
<groupId>log4j</groupId>
21+
<artifactId>log4j</artifactId>
22+
<version>1.2.17</version>
23+
</dependency>
24+
<dependency>
25+
<groupId>junit</groupId>
26+
<artifactId>junit</artifactId>
27+
<version>4.12</version>
28+
</dependency>
1929
</dependencies>
2030

2131
</project>

‎jdk8/src/main/java/me/zzw/app/jdk8/api/countedComplete/CountedCompleterApi.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
package me.zzw.app.jdk8.api.countedComplete;
2+
3+
import java.util.concurrent.CountedCompleter;
4+
5+
/**
6+
* Created by infosea on 2016-07-28.
7+
*/
8+
class MyOperation<E> {
9+
void apply(E e) {
10+
System.out.println(e);
11+
}
12+
}
13+
14+
15+
class ForEach<E> extends CountedCompleter<Void> {
16+
public static <E> void forEach(E[] array, MyOperation<E> op) {
17+
new ForEach<E>(null, array, op, 0, array.length).invoke();
18+
}
19+
20+
final E[] array;
21+
final MyOperation<E> op;
22+
final int lo, hi;
23+
24+
ForEach(CountedCompleter<?> p, E[] array, MyOperation<E> op, int lo, int hi) {
25+
super(p);
26+
this.array = array;
27+
this.op = op;
28+
this.lo = lo;
29+
this.hi = hi;
30+
}
31+
32+
public void compute() { // version 1
33+
if (hi - lo >= 2) {
34+
int mid = (lo + hi) >>> 1;
35+
setPendingCount(2); // must set pending count before fork
36+
new ForEach(this, array, op, mid, hi).fork(); // right child
37+
new ForEach(this, array, op, lo, mid).fork(); // left child
38+
} else if (hi > lo)
39+
op.apply(array[lo]);
40+
tryComplete();
41+
}
42+
}
43+
44+
public class CountedCompleterApi {
45+
public static void main(String... args) {
46+
String[] strings = new String[]{"hello", "world"};
47+
ForEach.forEach(strings, new MyOperation<String>());
48+
49+
strings = new String[]{"hello", "world", "this", "is ", "a", "countedCompleter"};
50+
ForEach<String> forEach = new ForEach<String>(null, strings, new MyOperation<String>(),0, 6);
51+
forEach.invoke();
52+
}
53+
}

0 commit comments

Comments
 (0)