Neo4j Java并发v2.0M3:在其他线程并发创建新关系时遍历关系时的异常

Question

我在这里试图实现的是获取特定节点的关系数量，而其他线程则同时向它添加新的关系。我使用TestGraphDatabaseFactory().newImpermanentDatabase()图形服务在单元测试中运行我的代码。

我的代码由~50个线程执行，它看起来如下所示：

int numOfRels = 0;
try {
    Iterable<Relationship> rels = parentNode.getRelationships(RelTypes.RUNS, Direction.OUTGOING);
    while (rels.iterator().hasNext()) {
        numOfRels++;
        rels.iterator().next();
    }
}
catch(Exception e) {
    throw e;
}

// Enforce relationship limit
if (numOfRels > 10) {
    // do something
}

Transaction tx = graph.beginTx();
try {
    Node node = createMyNodeAndConnectToParentNode(...);

    tx.success();

    return node;
}
catch (Exception e) {
    tx.failure();
}
finally {
    tx.finish();
}

问题是，我偶尔会在上面的try-catch块(围绕着ArrayIndexOutOfBoundsException())中得到一个“getRelationships: 1”。如果我正确理解，Iterable并不是线程安全的，并导致了这个问题。

我的问题是，使用Neo4j的Java迭代不断变化的关系和节点的最佳方法是什么？

我收到以下错误：

Exception in thread "Thread-14" org.neo4j.helpers.ThisShouldNotHappenError: Developer: Stefan/Jake claims that: A property key id disappeared under our feet
    at org.neo4j.kernel.impl.core.NodeProxy.setProperty(NodeProxy.java:188)
    at com.inbiza.connio.neo4j.server.extensions.graph.AppEntity.createMyNodeAndConnectToParentNode(AppEntity.java:546)
    at com.inbiza.connio.neo4j.server.extensions.graph.AppEntity.create(AppEntity.java:305)
    at com.inbiza.connio.neo4j.server.extensions.TestEmbeddedConnioGraph$appCreatorThread.run(TestEmbeddedConnioGraph.java:61)
    at java.lang.Thread.run(Thread.java:722)
Exception in thread "Thread-92" java.lang.ArrayIndexOutOfBoundsException: 1
    at org.neo4j.kernel.impl.core.RelationshipIterator.fetchNextOrNull(RelationshipIterator.java:72)
    at org.neo4j.kernel.impl.core.RelationshipIterator.fetchNextOrNull(RelationshipIterator.java:36)
    at org.neo4j.helpers.collection.PrefetchingIterator.hasNext(PrefetchingIterator.java:55)
    at com.inbiza.connio.neo4j.server.extensions.graph.AppEntity.create(AppEntity.java:243)
    at com.inbiza.connio.neo4j.server.extensions.TestEmbeddedConnioGraph$appCreatorThread.run(TestEmbeddedConnioGraph.java:61)
    at java.lang.Thread.run(Thread.java:722)
Exception in thread "Thread-12" java.lang.ArrayIndexOutOfBoundsException: 1
    at org.neo4j.kernel.impl.core.RelationshipIterator.fetchNextOrNull(RelationshipIterator.java:72)
    at org.neo4j.kernel.impl.core.RelationshipIterator.fetchNextOrNull(RelationshipIterator.java:36)
    at org.neo4j.helpers.collection.PrefetchingIterator.hasNext(PrefetchingIterator.java:55)
    at com.inbiza.connio.neo4j.server.extensions.graph.AppEntity.create(AppEntity.java:243)
    at com.inbiza.connio.neo4j.server.extensions.TestEmbeddedConnioGraph$appCreatorThread.run(TestEmbeddedConnioGraph.java:61)
    at java.lang.Thread.run(Thread.java:722)
Exception in thread "Thread-93" java.lang.ArrayIndexOutOfBoundsException
Exception in thread "Thread-90" java.lang.ArrayIndexOutOfBoundsException

以下是负责创建节点的方法：

static Node createMyNodeAndConnectToParentNode(GraphDatabaseService graph, final Node ownerAccountNode, final String suggestedName, Map properties) {

  final String accountId = checkNotNull((String)ownerAccountNode.getProperty("account_id"));

  Node appNode = graph.createNode();
  appNode.setProperty("urn_name", App.composeUrnName(accountId, suggestedName.toLowerCase().trim()));

  int nextId = nodeId.addAndGet(1); // I normally use getOrCreate idiom but to simplify I replaced it with an atomic int - that would do for testing 

  String urn = App.composeUrnUid(accountId,  nextId);
  appNode.setProperty("urn_uid", urn);
  appNode.setProperty("id", nextId);
  appNode.setProperty("name", suggestedName);

  Index<Node> indexUid =  graph.index().forNodes("EntityUrnUid");
  indexUid.add(appNode, "urn_uid", urn);

  appNode.addLabel(LabelTypes.App);

  appNode.setProperty("version", properties.get("version"));
  appNode.setProperty("description", properties.get("description"));

  Relationship rel = ownerAccountNode.createRelationshipTo(appNode, RelTypes.RUNS);
  rel.setProperty("date_created", fmt.print(new DateTime()));

  return appNode;
}

我在看org.neo4j.kernel.impl.core.RelationshipIterator.fetchNextOrNull()

看起来，我的测试生成了一个条件，如果( (status = fromNode.getMoreRelationships( nodeManager )).loaded() AC.26 lastTimeILookedThereWasMoreToLoad )没有被执行，并且在中间更改currentTypeIterator状态。

RelIdIterator currentTypeIterator = rels[currentTypeIndex];  //<-- this is where is crashes
do
{
  if ( currentTypeIterator.hasNext() )
  ...
  ... 

  while ( !currentTypeIterator.hasNext() )
  {
    if ( ++currentTypeIndex < rels.length )
    {
        currentTypeIterator = rels[currentTypeIndex];
    }
    else if ( (status = fromNode.getMoreRelationships( nodeManager )).loaded()
            // This is here to guard for that someone else might have loaded
            // stuff in this relationship chain (and exhausted it) while I
            // iterated over my batch of relationships. It will only happen
            // for nodes which have more than <grab size> relationships and
            // isn't fully loaded when starting iterating.
            || lastTimeILookedThereWasMoreToLoad )
    {
        ....
    }
  }
} while ( currentTypeIterator.hasNext() );

我还测试了几个锁定场景。下面的一个解决了这个问题。不确定每次在基于此的关系中迭代时是否应该使用锁。

Transaction txRead = graph.beginTx();
try {
  txRead.acquireReadLock(parentNode);

  long numOfRels = 0L;
  Iterable<Relationship> rels = parentNode.getRelationships(RelTypes.RUNS, Direction.OUTGOING);
  while (rels.iterator().hasNext()) {
    numOfRels++;
    rels.iterator().next();
  }

  txRead.success();
}
finally {
  txRead.finish();
}

我对Neo4j及其源代码库非常陌生；只是作为我们产品的潜在数据存储进行测试。如果有人知道Neo4j的内部和外部是怎么回事，我会很感激。

Answer 1

我觉得这是个虫子。由Iterable返回的getRelationships()应该是不可变的。当调用此方法时，迭代器中将提供到该时刻为止的所有可用Nodes。(您可以从org.neo4j.kernel.IntArrayIterator验证这一点)

我尝试通过让250个线程从节点插入关系到其他节点来复制它。以及在第一个节点的迭代器上循环一个主线程。仔细分析，迭代器只包含上次调用getRelationship()时添加的关系。我从来没有想到过这个问题。

请你把你的完整代码，海事组织可能会有一些愚蠢的错误。之所以不能发生这种情况，是因为在添加关系时，写锁已经就位，因此读取是同步的。

Answer 2

这是个窃听器。修补程序在以下拉请求中捕获：https://github.com/neo4j/neo4j/pull/1011