This is an automated email from the ASF dual-hosted git repository.
andy pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/jena.git
The following commit(s) were added to refs/heads/master by this push:
new d2e2316 JENA-1758: Use a semaphore to notify thread has finished
new 9a398d6 Merge pull request #695 from afs/jena1758-semaphore
d2e2316 is described below
commit d2e23164a08082f3e2c52c43b99b0bf0e66dcc91
Author: Andy Seaborne <[email protected]>
AuthorDate: Fri Feb 21 13:05:58 2020 +0000
JENA-1758: Use a semaphore to notify thread has finished
---
.../apache/jena/tdb2/loader/main/DataToTuples.java | 13 ++++++++-----
.../org/apache/jena/tdb2/loader/main/Indexer.java | 13 ++-----------
.../org/apache/jena/tdb2/loader/main/PhasedOps.java | 19 +++++++++++++++++++
3 files changed, 29 insertions(+), 16 deletions(-)
diff --git
a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
index a0886a9..24b30bb 100644
---
a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
+++
b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/DataToTuples.java
@@ -18,10 +18,13 @@
package org.apache.jena.tdb2.loader.main;
+import static org.apache.jena.tdb2.loader.main.PhasedOps.acquire;
+
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Semaphore;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -101,6 +104,9 @@ public class DataToTuples implements BulkStartFinish {
}
}
+ /** Semaphore for the other thread to indicate it has finished. */
+ private final Semaphore termination = new Semaphore(0);
+
@Override
public void startBulk() {
thread = new Thread(()->action());
@@ -109,11 +115,7 @@ public class DataToTuples implements BulkStartFinish {
@Override
public void finishBulk() {
- try {
- thread.join();
- } catch (InterruptedException e) {
- throw new BulkLoaderException("InterruptedException", e);
- }
+ acquire(termination);
}
// Triples.
@@ -155,6 +157,7 @@ public class DataToTuples implements BulkStartFinish {
}
transaction.end();
CoLib.finish(coordinator);
+ termination.release();
}
//@Override
diff --git
a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
index 02cd56f..54aef8f 100644
---
a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
+++
b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/Indexer.java
@@ -18,6 +18,8 @@
package org.apache.jena.tdb2.loader.main;
+import static org.apache.jena.tdb2.loader.main.PhasedOps.acquire;
+
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
@@ -25,7 +27,6 @@ import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Semaphore;
import org.apache.jena.atlas.lib.ArrayUtils;
-import org.apache.jena.atlas.lib.Timer;
import org.apache.jena.atlas.lib.tuple.Tuple;
import org.apache.jena.atlas.logging.Log;
import org.apache.jena.dboe.transaction.txn.Transaction;
@@ -65,16 +66,6 @@ public class Indexer implements BulkStartFinish {
}
}
- private static long acquire(Semaphore semaphore, int numPermits) {
- return Timer.time(()->{
- try { semaphore.acquire(numPermits); }
- catch (InterruptedException e) {
- Log.error(Indexer.class, "Interrupted", e);
- throw new RuntimeException(e);
- }
- });
- }
-
/** Return a function that delivers multiple {@code List<Tuple<NodeId>>>}
to this indexer */
public Destination<Tuple<NodeId>> index() {
return this::index;
diff --git
a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
index 7ee8d3e..852127e 100644
---
a/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
+++
b/jena-db/jena-tdb2/src/main/java/org/apache/jena/tdb2/loader/main/PhasedOps.java
@@ -19,11 +19,14 @@
package org.apache.jena.tdb2.loader.main;
import java.util.*;
+import java.util.concurrent.Semaphore;
import java.util.function.Consumer;
import java.util.function.IntFunction;
import java.util.stream.Stream;
+import org.apache.jena.atlas.lib.Timer;
import org.apache.jena.atlas.lib.tuple.Tuple;
+import org.apache.jena.atlas.logging.Log;
import org.apache.jena.tdb2.loader.BulkLoaderException;
import org.apache.jena.tdb2.loader.base.MonitorOutput;
import org.apache.jena.tdb2.loader.base.ProgressMonitor;
@@ -37,6 +40,22 @@ import org.apache.jena.tdb2.store.tupletable.TupleIndex;
*/
class PhasedOps {
+ /** Acquire one permit from a semaphore. Return the time spent waiting. */
+ /* package */ static long acquire(Semaphore termination) {
+ return acquire(termination, 1);
+ }
+
+ /** Acquire permits from a semaphore. Return the time spent waiting. */
+ /* package */ static long acquire(Semaphore semaphore, int numPermits) {
+ return Timer.time(()->{
+ try { semaphore.acquire(numPermits); }
+ catch (InterruptedException e) {
+ Log.error(Indexer.class, "Interrupted", e);
+ throw new RuntimeException(e);
+ }
+ });
+ }
+
static Map<String, TupleIndex> indexMap(DatasetGraphTDB dsgtdb) {
Map<String, TupleIndex> indexMap = new HashMap<>();
// All triple/quad indexes.
