Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion build.sbt
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
ThisBuild / organization := "io.appthreat"
ThisBuild / version := "3.1.0"
ThisBuild / version := "3.1.1"
ThisBuild / scalaVersion := "3.10.0"

// sbt 2 applies bare settings to every subproject, so root-only settings live on the root project
Expand Down
38 changes: 38 additions & 0 deletions core-tests/src/test/java/overflowdb/util/NodesListTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,11 @@
import overflowdb.Graph;

import java.util.ArrayList;
import java.util.List;
import java.util.Vector;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
Expand Down Expand Up @@ -183,6 +187,40 @@ public void compactsAutomatically() {



/** A node that has been added is found by id while another thread keeps adding nodes, as when a pass writer adds
* nodes while the pass's parts traverse the graph. */
@Test
public void lookupsDuringConcurrentAdds() throws Exception {
final int count = 300_000;
final NodesList nl = new NodesList(16);
final NodeRef[] refs = new NodeRef[count];
for (int i = 0; i < count; i++) refs[i] = createDummyRef(i, "A");
final AtomicInteger added = new AtomicInteger();
final AtomicReference<String> failure = new AtomicReference<>();
final List<Thread> readers = new ArrayList<>();
for (int r = 0; r < 3; r++) {
readers.add(new Thread(() -> {
final ThreadLocalRandom random = ThreadLocalRandom.current();
while (added.get() < count && failure.get() == null) {
final int known = added.get();
if (known == 0) continue;
final int id = random.nextInt(known);
if (nl.nodeById(id) != refs[id] || !nl.contains(id)) {
failure.compareAndSet(null, "node " + id + " not found after " + known + " adds");
}
}
}));
}
readers.forEach(Thread::start);
for (int i = 0; i < count && failure.get() == null; i++) {
nl.add(refs[i]);
added.set(i + 1);
}
added.set(count);
for (Thread t : readers) t.join();
assertNull(failure.get());
}

private NodeRef createDummyRef(long id, String label) {
return new NodeRef(dummyGraph, id) {
public String label() {
Expand Down
25 changes: 15 additions & 10 deletions core/src/main/java/overflowdb/util/NodesList.java
Original file line number Diff line number Diff line change
Expand Up @@ -67,16 +67,19 @@ private int tryClaimEmptySlot() {
return nextEmptySlot;
}

public boolean contains(long id) {
/* Lookups take the same monitor as add/remove/compact. The trove map rehashes in place while a node is added, and
* a lookup racing that rehash can miss a node that is in the list. A pass writer adds nodes while the pass's parts
* traverse the graph, so a NodeRef built from an id (e.g. the OpNodes wrappers) could come out empty and then fail
* with "unable to read node from disk" for a node that was never written to disk. */
public synchronized boolean contains(long id) {
return nodeIndexByNodeId.containsKey(id);
}

public Node nodeById(long id) {
if (nodeIndexByNodeId.containsKey(id)) {
return nodes[nodeIndexByNodeId.get(id)];
} else {
return null;
}
public synchronized Node nodeById(long id) {
final int index = nodeIndexByNodeId.get(id);
// the map's no-entry value (0) is also a valid index
if (index == nodeIndexByNodeId.getNoEntryValue() && !nodeIndexByNodeId.containsKey(id)) return null;
return nodes[index];
}

public synchronized void remove(Node node) {
Expand Down Expand Up @@ -122,12 +125,14 @@ private synchronized void initialiseNodesByLabel(){
}
}

public ArrayList<Node> nodesByLabel(String label) {
/* The label map is mutated by add/remove and by computeIfAbsent below, so access to it is synchronized too. The
* returned list itself is live: iterating it while another thread adds nodes with that label is still unsafe. */
public synchronized ArrayList<Node> nodesByLabel(String label) {
TMap<String, ArrayList<Node>> nodesByLabel = getNodesByLabel();
return nodesByLabel.computeIfAbsent(label, k -> new ArrayList<>());
}

public Set<String> nodeLabels() {
public synchronized Set<String> nodeLabels() {
TMap<String, ArrayList<Node>> nodesByLabel = getNodesByLabel();
Set<String> ret = new HashSet<>(nodesByLabel.size());
nodesByLabel.forEach((key, value) -> {
Expand Down Expand Up @@ -201,7 +206,7 @@ protected int _elementDataSize() {
return nodes.length;
}

public int cardinality(String label) {
public synchronized int cardinality(String label) {
TMap<String, ArrayList<Node>> nodesByLabel = getNodesByLabel();
if (nodesByLabel.containsKey(label))
return nodesByLabel.get(label).size();
Expand Down
Loading