--- /dev/null
+package org.simantics.db.layer0.util;\r
+\r
+import java.io.IOException;\r
+import java.io.InputStream;\r
+import java.util.Collection;\r
+import java.util.HashMap;\r
+import java.util.Iterator;\r
+import java.util.Map;\r
+import java.util.Set;\r
+import java.util.TreeSet;\r
+\r
+import org.simantics.databoard.Bindings;\r
+import org.simantics.databoard.Databoard;\r
+import org.simantics.databoard.binding.Binding;\r
+import org.simantics.databoard.serialization.Serializer;\r
+import org.simantics.databoard.type.Datatype;\r
+import org.simantics.db.DirectStatements;\r
+import org.simantics.db.ReadGraph;\r
+import org.simantics.db.Resource;\r
+import org.simantics.db.ResourceMap;\r
+import org.simantics.db.Statement;\r
+import org.simantics.db.common.primitiverequest.Value;\r
+import org.simantics.db.common.procedure.adapter.TransientCacheListener;\r
+import org.simantics.db.common.utils.NameUtils;\r
+import org.simantics.db.exception.CancelTransactionException;\r
+import org.simantics.db.exception.DatabaseException;\r
+import org.simantics.db.layer0.adapter.SubgraphExtent.ExtentStatus;\r
+import org.simantics.db.layer0.util.ModelTransferableGraphSourceRequest.Expansion3;\r
+import org.simantics.db.service.CollectionSupport;\r
+import org.simantics.db.service.SerialisationSupport;\r
+import org.simantics.db.service.TransferableGraphSupport;\r
+import org.simantics.graph.db.TransferableGraphSource;\r
+import org.simantics.layer0.Layer0;\r
+import org.simantics.scl.runtime.function.Function1;\r
+\r
+import gnu.trove.list.array.TIntArrayList;\r
+import gnu.trove.map.hash.TIntIntHashMap;\r
+\r
+public class DomainProcessor3 {\r
+ \r
+ public enum ExclusionDecision {\r
+ INCLUDE, EXCLUDE_OBJECT\r
+ }\r
+\r
+ final static private boolean PROFILE = false;\r
+\r
+ Serializer variantSerializer;\r
+ Serializer datatypeSerializer;\r
+ Binding datatypeBinding;\r
+ boolean ignoreVirtual;\r
+\r
+ int id = 0;\r
+\r
+ Set<Resource> fringe = null;\r
+ Set<Resource> exclusions = null;\r
+ Function1<Statement,ExclusionDecision> exclusionFunction = null;\r
+ Set<Resource> predicates = null;\r
+ Map<Resource,Boolean> isRelatedToPredicates = null;\r
+ Set<Resource> deadPredicates = null;\r
+ Set<Resource> strongInverseSet = null;\r
+\r
+ TIntIntHashMap ids = null;\r
+ ResourceMap<ExtentStatus> status = null;\r
+ Map<Datatype, byte[]> bindings = new HashMap<Datatype, byte[]>();\r
+ final SerialisationSupport support;\r
+ final TransferableGraphConfiguration2 conf;\r
+ final TransferableGraphSupport tgs;\r
+\r
+ private Layer0 L0;\r
+\r
+ private long composedObjectCounter = 0;\r
+ private long fastInternalCounter = 0;\r
+ private long parentExternalCounter = 0;\r
+ private long fullInternalCounter = 0;\r
+ private long fullExternalCounter = 0;\r
+\r
+ long startupTime = 0;\r
+ long expandTime = 0;\r
+ long fullResolveTime = 0;\r
+ long fastResolveTime = 0;\r
+ long otherStatementTime = 0;\r
+ long parentResolveTime = 0;\r
+ long extentSeedTime = 0;\r
+ long classifyPredicateTime = 0;\r
+ long processFringeTime = 0;\r
+ long valueOutputTime = 0;\r
+ long statementOutputTime = 0;\r
+\r
+ public DomainProcessor3(ReadGraph graph, TransferableGraphConfiguration2 conf, DomainProcessorState state, boolean ignoreVirtual) throws DatabaseException {\r
+\r
+ this.L0 = Layer0.getInstance(graph);\r
+ this.tgs = graph.getService(TransferableGraphSupport.class);\r
+\r
+ this.support = graph.getService(SerialisationSupport.class);\r
+ this.ignoreVirtual = ignoreVirtual;\r
+ this.conf = conf;\r
+\r
+ if(PROFILE)\r
+ startupTime -= System.nanoTime();\r
+\r
+ CollectionSupport cs = graph.getService(CollectionSupport.class);\r
+\r
+ ids = state.ids;\r
+ status = cs.createMap(ExtentStatus.class);\r
+ predicates = cs.createSet();\r
+ exclusions = cs.createSet();\r
+ exclusionFunction = conf.exclusionFunction;\r
+ fringe = new TreeSet<Resource>();\r
+ isRelatedToPredicates = cs.createMap(Boolean.class);\r
+ deadPredicates = cs.createSet();\r
+ strongInverseSet = cs.createSet();\r
+\r
+ for(Map.Entry<Resource, ExtentStatus> entry : conf.preStatus.entrySet()) {\r
+ status.put(entry.getKey(), entry.getValue());\r
+ if(ExtentStatus.EXCLUDED.equals(entry.getValue())) exclusions.add(entry.getKey());\r
+ if(ExtentStatus.INTERNAL.equals(entry.getValue())) fringe.add(entry.getKey());\r
+ }\r
+\r
+ if(PROFILE)\r
+ startupTime += System.nanoTime();\r
+ \r
+// for(RootSpec p : conf.roots) {\r
+// if(p.internal)\r
+// fringe.add(p.resource);\r
+// }\r
+\r
+ }\r
+ \r
+ public ResourceMap<ExtentStatus> getStatus() {\r
+ return status;\r
+ }\r
+\r
+ public Collection<DirectStatements> extractFromFringe(ReadGraph graph, int maxAmount) throws DatabaseException {\r
+\r
+ CollectionSupport cs = graph.getService(CollectionSupport.class);\r
+ Collection<Resource> list = cs.createList();\r
+ Iterator<Resource> it = fringe.iterator();\r
+ for(int i=0;i<maxAmount;i++) {\r
+ if(!it.hasNext()) break;\r
+ list.add(it.next());\r
+ it.remove();\r
+ }\r
+\r
+ return graph.syncRequest(new Expansion3(list, ignoreVirtual));\r
+\r
+ }\r
+\r
+ public Collection<DirectStatements> expand(ReadGraph graph) throws DatabaseException {\r
+\r
+ if(PROFILE)\r
+ expandTime -= System.nanoTime();\r
+\r
+ Collection<DirectStatements> result = extractFromFringe(graph, 2<<12);\r
+\r
+ if(PROFILE)\r
+ expandTime += System.nanoTime();\r
+\r
+ return result;\r
+\r
+ }\r
+\r
+ public void classifyPredicates(ReadGraph graph, final Set<Resource> schedule, DomainProcessorState state) throws DatabaseException {\r
+\r
+ for(Resource predicate : schedule) {\r
+ \r
+ Boolean isRelatedTo = Boolean.FALSE;\r
+ \r
+ Resource single = graph.getPossibleSuperrelation(predicate);\r
+ if(single != null) {\r
+ \r
+ Boolean singleIsRelatedTo = isRelatedToPredicates.get(single);\r
+ if(singleIsRelatedTo == null) {\r
+ singleIsRelatedTo = graph.isSubrelationOf(single, L0.IsRelatedTo);\r
+ isRelatedToPredicates.put(single, singleIsRelatedTo);\r
+ }\r
+\r
+ isRelatedTo = singleIsRelatedTo;\r
+ \r
+ } else {\r
+ \r
+ if(graph.isSubrelationOf(predicate, L0.IsRelatedTo)) {\r
+ isRelatedTo = Boolean.TRUE;\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("isRelatedToPredicates4 => " + NameUtils.getSafeName(graph, predicate));\r
+ } else {\r
+ if (!graph.hasStatement(predicate)) {\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("FOUND DEAD PREDICATE (no statements): " + predicate);\r
+ deadPredicates.add(predicate);\r
+ // Prevents ModelTransferableGraphSource from\r
+ // trying to export these statements.\r
+ state.inverses.remove(support.getTransientId(predicate));\r
+ }\r
+ }\r
+ \r
+ }\r
+ \r
+ isRelatedToPredicates.put(predicate, isRelatedTo);\r
+\r
+ }\r
+\r
+ }\r
+\r
+ public void classifyPredicates(ReadGraph graph, DomainProcessorState state, final Collection<DirectStatements> expansion) throws DatabaseException {\r
+\r
+ if(PROFILE)\r
+ classifyPredicateTime -= System.nanoTime();\r
+\r
+ CollectionSupport cs = graph.getService(CollectionSupport.class);\r
+ final Set<Resource> schedule = cs.createSet();\r
+ final Map<Resource, Resource> newPredicates = cs.createMap(Resource.class);\r
+\r
+ for (DirectStatements stms : expansion) {\r
+ for(Statement stm : stms) {\r
+\r
+ Resource predicate = stm.getPredicate();\r
+ Resource object = stm.getObject();\r
+\r
+ if (exclusions.contains(object) || exclusions.contains(predicate))\r
+ continue;\r
+ \r
+ if (exclusionFunction != null) {\r
+ ExclusionDecision decision = exclusionFunction.apply(stm);\r
+ if(ExclusionDecision.EXCLUDE_OBJECT.equals(decision)) {\r
+ status.put(object, ExtentStatus.EXCLUDED);\r
+ exclusions.add(object);\r
+ continue;\r
+ }\r
+ }\r
+\r
+ if(predicates.add(predicate)) {\r
+ Resource inverse = graph.getPossibleInverse(predicate);\r
+ schedule.add(predicate);\r
+ if(inverse != null) {\r
+ newPredicates.put(predicate, inverse);\r
+ if(predicates.add(inverse)) schedule.add(inverse);\r
+ state.inverses.put(support.getTransientId(predicate), support.getTransientId(inverse));\r
+ state.inverses.put(support.getTransientId(inverse), support.getTransientId(predicate));\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("INVERSE FOR " + graph.getPossibleURI(predicate) + " => " + graph.getPossibleURI(inverse));\r
+ } else {\r
+ state.inverses.put(support.getTransientId(predicate), 0);\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("NO INVERSE FOR " + graph.getPossibleURI(predicate));\r
+ }\r
+\r
+ }\r
+\r
+ }\r
+ }\r
+\r
+ classifyPredicates(graph, schedule, state);\r
+\r
+ for(Map.Entry<Resource, Resource> entry : newPredicates.entrySet()) {\r
+ // Inverse is strong => this has strong inverse\r
+ Boolean isRelatedToValue = isRelatedToPredicates.get(entry.getValue());\r
+ if(isRelatedToValue) strongInverseSet.add(entry.getKey());\r
+ Boolean isRelatedToKey = isRelatedToPredicates.get(entry.getKey());\r
+ if(isRelatedToKey) strongInverseSet.add(entry.getValue());\r
+ }\r
+\r
+ if(PROFILE)\r
+ classifyPredicateTime += System.nanoTime();\r
+\r
+ }\r
+\r
+ /*\r
+ * Composed objects are internal. Mark them for expansion.\r
+ */\r
+\r
+ private Datatype getDatatype(ReadGraph graph, Resource subject) throws DatabaseException {\r
+ Resource object = graph.getSingleObject(subject, L0.HasDataType);\r
+ return graph.syncRequest(new Value<Datatype>(object, datatypeBinding), TransientCacheListener.<Datatype>instance());\r
+ }\r
+\r
+ public void processValue(ReadGraph graph, Resource subject, int sId, final DomainProcessorState state) throws DatabaseException, IOException {\r
+ final InputStream valueStream = tgs.getValueStream(graph, subject);\r
+ if (valueStream != null) {\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("[VALUE] " + NameUtils.getSafeName(graph, subject, true));\r
+ state.valueOutput.writeInt(sId);\r
+\r
+ if (conf.values) {\r
+ Datatype dt = getDatatype(graph, subject);\r
+\r
+ boolean canWriteRawVariant = !state.valueModifier.mayNeedModification(dt);\r
+ long rawVariantSizePos = 0;\r
+ state.valueOutput.writeByte(canWriteRawVariant\r
+ ? TransferableGraphSource.TAG_RAW_COPY_VARIANT_VALUE\r
+ : TransferableGraphSource.TAG_POTENTIALLY_MODIFIED_VARIANT_VALUE);\r
+ if (canWriteRawVariant) {\r
+ // Add space for raw variant byte size before the data\r
+ rawVariantSizePos = state.valueOutput.position();\r
+ state.valueOutput.writeInt(0);\r
+ }\r
+\r
+ byte[] typeBytes = bindings.get(dt);\r
+ if (typeBytes == null) {\r
+ typeBytes = datatypeSerializer.serialize(dt);\r
+ bindings.put(dt, typeBytes);\r
+ }\r
+\r
+ state.valueOutput.write(typeBytes);\r
+ Serializer s = Bindings.getSerializerUnchecked(Bindings.getBinding(dt));\r
+ s.skip(new InputStream() {\r
+ @Override\r
+ public int read() throws IOException {\r
+ int value = valueStream.read();\r
+ state.valueOutput.write(value);\r
+ return value;\r
+ }\r
+ });\r
+\r
+ if (canWriteRawVariant) {\r
+ long currentPos = state.valueOutput.position();\r
+ int variantSize = (int)(currentPos - rawVariantSizePos - 4);\r
+ state.valueOutput.position(rawVariantSizePos);\r
+ state.valueOutput.writeInt(variantSize);\r
+ state.valueOutput.position(currentPos);\r
+ }\r
+ }\r
+\r
+ state.valueCount++;\r
+ }\r
+ }\r
+\r
+ private TIntArrayList stream = new TIntArrayList();\r
+\r
+ public void addToStream(Resource predicate, Resource object) throws DatabaseException {\r
+ stream.add(support.getTransientId(predicate));\r
+ stream.add(support.getTransientId(object));\r
+ }\r
+\r
+ public void processStatement(ReadGraph graph, Resource subject, Statement stm) throws DatabaseException, IOException {\r
+\r
+ Resource predicate = stm.getPredicate();\r
+\r
+ Resource object = stm.getObject();\r
+\r
+ ExtentStatus objectStatus = status.get(object);\r
+\r
+ // Strong predicate\r
+ Boolean isRelatedTo = isRelatedToPredicates.get(predicate);\r
+ if ((objectStatus != ExtentStatus.EXCLUDED) && isRelatedTo) {\r
+\r
+ if(ModelTransferableGraphSourceRequest.LOG) logStatementWithExtent(graph, "related", objectStatus, subject, predicate, object);\r
+\r
+ addToStream(predicate, object);\r
+\r
+ if(objectStatus == null || objectStatus == ExtentStatus.PENDING) {\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("[ADDED TO FRINGE] " + NameUtils.getSafeName(graph, object));\r
+ fringe.add(object);\r
+ }\r
+\r
+ } else {\r
+\r
+ // Dead predicate\r
+ if (deadPredicates.contains(predicate)) {\r
+ if(ModelTransferableGraphSourceRequest.LOG) logStatementWithExtent(graph, "excluding statement with dead predicate ", objectStatus, subject, predicate, object);\r
+ return;\r
+ }\r
+\r
+ // Weak predicate\r
+ if(objectStatus == ExtentStatus.EXCLUDED) {\r
+\r
+ if(ModelTransferableGraphSourceRequest.LOG) logStatementWithExtent(graph, "weak reference to excluded object ", objectStatus, subject, predicate, object);\r
+\r
+ } else {\r
+\r
+ // The inverse is also weak (or there is no inverse)\r
+ if(!strongInverseSet.contains(predicate)) {\r
+\r
+ addToStream(predicate, object);\r
+\r
+ if(objectStatus == null) {\r
+ status.put(object, ExtentStatus.PENDING);\r
+ }\r
+\r
+ if(ModelTransferableGraphSourceRequest.LOG) logStatementWithExtent(graph, "fully weak internal", objectStatus, subject, predicate, object);\r
+\r
+ } else {\r
+\r
+ if(ModelTransferableGraphSourceRequest.LOG) logStatementWithExtent(graph, "strong inverse internal ", objectStatus, subject, predicate, object);\r
+\r
+ }\r
+\r
+ }\r
+\r
+ }\r
+ \r
+ }\r
+ \r
+ public void flushStatementStream(int sId, DomainProcessorState state) throws IOException {\r
+ if(!stream.isEmpty()) {\r
+ state.statementsOutput.writeInt(sId);\r
+ int streamSize = stream.size();\r
+ int statementCount = stream.size() / 2;\r
+ state.statementsOutput.writeInt(statementCount);\r
+ for (int i = 0; i < streamSize; i++)\r
+ state.statementsOutput.writeInt(stream.getQuick(i));\r
+ state.statementCount += 2*streamSize;\r
+ stream.resetQuick();\r
+ }\r
+ }\r
+\r
+ // For progress monitor book-keeping\r
+ private long internalResourceNumber = 0;\r
+ private long startTime = 0;\r
+ private long lastUpdateTime = 0;\r
+\r
+ public void processInternal(ReadGraph graph, Resource subject, DirectStatements stms, DomainProcessorState state) throws DatabaseException, IOException {\r
+\r
+ internalResourceNumber++;\r
+\r
+ // Update progress monitor with controlled frequency\r
+ long t = System.nanoTime();\r
+ long dt = t - lastUpdateTime;\r
+ if (dt > 200_000_000L) {\r
+ if (startTime == 0)\r
+ startTime = t;\r
+ lastUpdateTime = t;\r
+ double totalTime = (t - startTime) * 1e-9;\r
+ if (totalTime > 0) {\r
+ long speed = Math.round((double)internalResourceNumber / totalTime);\r
+ state.monitor.subTask("Included " + internalResourceNumber + " resources (" + speed + " resources/s)");\r
+ }\r
+ }\r
+\r
+ status.put(subject, ExtentStatus.INTERNAL);\r
+ if(ModelTransferableGraphSourceRequest.LOG) ModelTransferableGraphSourceRequest.log("[INTERNAL] " + NameUtils.getSafeName(graph, subject, true));\r
+\r
+ int sId = support.getTransientId(subject);\r
+\r
+ if(PROFILE)\r
+ valueOutputTime -= System.nanoTime();\r
+\r
+ processValue(graph, subject, sId, state);\r
+\r
+ if(PROFILE)\r
+ valueOutputTime += System.nanoTime();\r
+\r
+ if(PROFILE)\r
+ statementOutputTime -= System.nanoTime();\r
+\r
+ for(Statement stm : stms) {\r
+ processStatement(graph, subject, stm);\r
+ }\r
+\r
+ flushStatementStream(sId, state);\r
+\r
+ if(PROFILE)\r
+ statementOutputTime += System.nanoTime();\r
+\r
+ // Logarithmic progress monitor for unknown amount of work.\r
+ state.monitor.setWorkRemaining(100000);\r
+ state.monitor.worked(1);\r
+ }\r
+\r
+ public void processFringe(ReadGraph graph, Collection<DirectStatements> expansion, final DomainProcessorState state) throws DatabaseException, IOException {\r
+\r
+ if(PROFILE)\r
+ processFringeTime -= System.nanoTime();\r
+\r
+ for (DirectStatements stms : expansion) {\r
+\r
+ Resource subject = stms.getSubject();\r
+\r
+ boolean partOf = false;\r
+ for(Statement stm : stms) {\r
+ Resource predicate = stm.getPredicate();\r
+ if(L0.PartOf.equals(predicate)) {\r
+ partOf = true;\r
+ break;\r
+ }\r
+ }\r
+\r
+ ExtentStatus subjectStatus = status.get(subject);\r
+ if(ModelTransferableGraphSourceRequest.LOG && subjectStatus != null) ModelTransferableGraphSourceRequest.log("EXISTING STATUS FOR " + graph.getPossibleURI(subject) + " - " + subjectStatus);\r
+ if(subjectStatus == ExtentStatus.EXTERNAL || subjectStatus == ExtentStatus.EXCLUDED) continue;\r
+ if(partOf && (subjectStatus == null || ExtentStatus.PENDING == subjectStatus) && graph.getPossibleURI(subject) != null) {\r
+\r
+ status.put(subject, ExtentStatus.EXTERNAL);\r
+ if(ModelTransferableGraphSourceRequest.LOG) {\r
+ String uri = graph.getPossibleURI(subject);\r
+ if(uri == null) ModelTransferableGraphSourceRequest.log("[EXTERNAL]: No URI for " + subject);\r
+ else ModelTransferableGraphSourceRequest.log("[EXTERNAL] " + uri);\r
+ }\r
+\r
+ } else {\r
+\r
+ processInternal(graph, subject, stms, state);\r
+ \r
+ }\r
+\r
+ }\r
+\r
+ if(PROFILE)\r
+ processFringeTime += System.nanoTime();\r
+\r
+ }\r
+\r
+ public void process(ReadGraph graph, DomainProcessorState state) throws DatabaseException {\r
+\r
+ try {\r
+\r
+ this.variantSerializer = graph.getService(Databoard.class).getSerializerUnchecked(Bindings.VARIANT);\r
+ this.datatypeBinding = Bindings.getBindingUnchecked(Datatype.class);\r
+ this.datatypeSerializer = graph.getService(Databoard.class).getSerializerUnchecked(this.datatypeBinding);\r
+\r
+ for(Resource r : ConsistsOfProcess.walk(graph, fringe, exclusions, ignoreVirtual)) {\r
+ if (status.put(r, ExtentStatus.INTERNAL) == null) {\r
+ if(ModelTransferableGraphSourceRequest.LOG) {\r
+ String URI = graph.getPossibleURI(r);\r
+ if(URI != null) ModelTransferableGraphSourceRequest.log("URI INTERNAL " + URI);\r
+ else ModelTransferableGraphSourceRequest.log("URI has no URI for " + r);\r
+ }\r
+ fringe.add(r);\r
+ }\r
+ }\r
+\r
+ if (state.monitor.isCanceled())\r
+ throw new CancelTransactionException();\r
+\r
+ while(!fringe.isEmpty()) {\r
+\r
+ Collection<DirectStatements> expansion = expand(graph);\r
+ classifyPredicates(graph, state, expansion);\r
+ processFringe(graph, expansion, state);\r
+\r
+ if (state.monitor.isCanceled())\r
+ throw new CancelTransactionException();\r
+ }\r
+\r
+ if (ModelTransferableGraphSourceRequest.PROFILE) {\r
+ System.out.println(composedObjectCounter + " " + fastInternalCounter\r
+ + " " + parentExternalCounter + " "\r
+ + fullExternalCounter + " " + fullInternalCounter);\r
+ }\r
+\r
+ } catch (IOException e) {\r
+ throw new DatabaseException(e);\r
+ }\r
+\r
+ }\r
+\r
+ void logStatementWithExtent(ReadGraph graph, String header, ExtentStatus status, int sId, int pId, int oId) throws DatabaseException {\r
+ if(ModelTransferableGraphSourceRequest.LOG) {\r
+ SerialisationSupport support = graph.getService(SerialisationSupport.class);\r
+ String s = NameUtils.getURIOrSafeNameInternal(graph, support.getResource(sId));\r
+ String p = NameUtils.getURIOrSafeNameInternal(graph, support.getResource(pId));\r
+ String o = NameUtils.getURIOrSafeNameInternal(graph, support.getResource(oId));\r
+ ModelTransferableGraphSourceRequest.log(header + " [" + status + "] " + s + " - " + p + " - " + o);\r
+ }\r
+ }\r
+\r
+ void logStatementWithExtent(ReadGraph graph, String header, ExtentStatus status, Resource sId, Resource pId, Resource oId) throws DatabaseException {\r
+ if(ModelTransferableGraphSourceRequest.LOG) {\r
+ String s = NameUtils.getURIOrSafeNameInternal(graph, sId);\r
+ String p = NameUtils.getURIOrSafeNameInternal(graph, pId);\r
+ String o = NameUtils.getURIOrSafeNameInternal(graph, oId);\r
+ ModelTransferableGraphSourceRequest.log(header + " [" + status + "] " + s + " - " + p + " - " + o);\r
+ }\r
+ }\r
+\r
+}
\ No newline at end of file
Code Review / simantics / platform.git / blobdiff