1 /*******************************************************************************
2 * Copyright (c) 2007, 2010 Association for Decentralized Information Management
4 * All rights reserved. This program and the accompanying materials
5 * are made available under the terms of the Eclipse Public License v1.0
6 * which accompanies this distribution, and is available at
7 * http://www.eclipse.org/legal/epl-v10.html
10 * VTT Technical Research Centre of Finland - initial API and implementation
11 *******************************************************************************/
12 package org.simantics.db.layer0.util;
14 import java.io.BufferedOutputStream;
15 import java.io.DataOutput;
16 import java.io.DataOutputStream;
18 import java.io.FileNotFoundException;
19 import java.io.FileOutputStream;
20 import java.io.IOException;
21 import java.lang.management.ManagementFactory;
22 import java.lang.reflect.InvocationTargetException;
23 import java.lang.reflect.Method;
24 import java.util.ArrayList;
25 import java.util.Collection;
26 import java.util.Collections;
27 import java.util.List;
28 import java.util.UUID;
30 import org.apache.commons.io.output.DeferredFileOutputStream;
31 import org.eclipse.core.runtime.IProgressMonitor;
32 import org.eclipse.core.runtime.OperationCanceledException;
33 import org.eclipse.core.runtime.SubMonitor;
34 import org.simantics.databoard.Bindings;
35 import org.simantics.databoard.Datatypes;
36 import org.simantics.databoard.accessor.error.AccessorException;
37 import org.simantics.databoard.binding.Binding;
38 import org.simantics.databoard.binding.error.BindingException;
39 import org.simantics.databoard.binding.mutable.Variant;
40 import org.simantics.databoard.parser.repository.DataTypeSyntaxError;
41 import org.simantics.databoard.serialization.RuntimeSerializerConstructionException;
42 import org.simantics.databoard.type.Datatype;
43 import org.simantics.databoard.util.binary.BinaryFile;
44 import org.simantics.databoard.util.binary.BinaryMemory;
45 import org.simantics.databoard.util.binary.DeferredBinaryFile;
46 import org.simantics.databoard.util.binary.NullRandomAccessBinary;
47 import org.simantics.databoard.util.binary.RandomAccessBinary;
48 import org.simantics.db.DirectStatements;
49 import org.simantics.db.ReadGraph;
50 import org.simantics.db.RequestProcessor;
51 import org.simantics.db.Resource;
52 import org.simantics.db.Statement;
53 import org.simantics.db.common.request.UniqueRead;
54 import org.simantics.db.common.utils.NameUtils;
55 import org.simantics.db.exception.CancelTransactionException;
56 import org.simantics.db.exception.DatabaseException;
57 import org.simantics.db.exception.ValidationException;
58 import org.simantics.db.layer0.adapter.SubgraphExtent.ExtentStatus;
59 import org.simantics.db.layer0.internal.SimanticsInternal;
60 import org.simantics.db.service.ClusterControl;
61 import org.simantics.db.service.ClusterControl.ClusterState;
62 import org.simantics.db.service.ClusteringSupport;
63 import org.simantics.db.service.CollectionSupport;
64 import org.simantics.db.service.DirectQuerySupport;
65 import org.simantics.db.service.SerialisationSupport;
66 import org.simantics.graph.representation.Extensions;
67 import org.simantics.graph.utils.TGResourceUtil;
68 import org.simantics.graph.utils.TGResourceUtil.LongAdapter;
69 import org.simantics.layer0.Layer0;
70 import org.simantics.utils.threads.logger.ITask;
71 import org.simantics.utils.threads.logger.ThreadLogger;
73 import gnu.trove.list.array.TIntArrayList;
74 import gnu.trove.map.hash.TIntIntHashMap;
75 import gnu.trove.map.hash.TLongObjectHashMap;
76 import gnu.trove.procedure.TIntProcedure;
77 import gnu.trove.procedure.TLongObjectProcedure;
78 import gnu.trove.set.hash.TIntHashSet;
80 public class ModelTransferableGraphSourceRequest extends UniqueRead<ModelTransferableGraphSource> {
82 public static String LOG_FILE = "transferableGraph.log";
83 final static boolean LOG = false;
84 final static private boolean DEBUG = false;
85 final static boolean PROFILE = false;
87 private TransferableGraphConfiguration2 configuration;
88 private SubMonitor monitor;
90 static DataOutput log;
96 FileOutputStream stream = new FileOutputStream(LOG_FILE);
97 log = new DataOutputStream(stream);
98 } catch (FileNotFoundException e) {
105 static void log(String line) {
108 log.writeUTF(line + "\n");
109 } catch (IOException e) {
115 public ModelTransferableGraphSourceRequest(TransferableGraphConfiguration2 conf) {
119 public ModelTransferableGraphSourceRequest(IProgressMonitor monitor, TransferableGraphConfiguration2 conf) {
120 this.monitor = SubMonitor.convert(monitor);
121 this.configuration = conf;
127 int statementIndex = 0;
129 TIntArrayList externalParents = new TIntArrayList();
130 ArrayList<String> externalNames = new ArrayList<String>();
135 private SerialisationSupport support;
137 private Resource getResource(int r) throws DatabaseException {
138 return support.getResource(r);
141 public int getInternalId(int r) {
145 public boolean validateExternal(Resource ext) {
146 ExtentStatus status = configuration.preStatus.get(ext);
148 if(ExtentStatus.INTERNAL.equals(status)) return false;
149 else if(ExtentStatus.EXCLUDED.equals(status)) return false;
156 * @return -2 if r is not really external and the statement should be excluded
159 public int getId(ReadGraph graph, int r) throws DatabaseException {
160 if(ids.containsKey(r)) {
161 int ret = ids.get(r);
163 for(int i=0;i<=indent;++i)
164 System.out.print(" ");
165 System.out.println("Cycle!!!"); // with " + GraphUtils.getReadableName(g, r));
170 if(!validateExternal(getResource(r))) return -2;
171 Collection<Resource> parents = graph.getObjects(getResource(r), L0.PartOf);
172 if(parents.size() != 1) {
173 throw new ValidationException("Reference to external resource "
174 + NameUtils.getSafeName(graph, getResource(r), true) + " without unique uri (" + parents.size() + " parents).");
176 for(Resource p : parents) {
178 int pid = getId(graph, support.getTransientId(p));
179 if(pid == -2) return -2;
180 externalParents.add(pid);
183 externalNames.add((String)graph.getRelatedValue(getResource(r), L0.HasName));
190 public ModelTransferableGraphSource perform(ReadGraph graph) throws DatabaseException {
192 support = graph.getService(SerialisationSupport.class);
194 this.L0 = Layer0.getInstance(graph);
196 long total = System.nanoTime();
197 long startupTime = System.nanoTime();
198 long startupTimeEnd = System.nanoTime();
199 long domainTime = System.nanoTime();
201 String otherStatements = "other" + UUID.randomUUID().toString();
202 String valueFileName = "value" + UUID.randomUUID().toString();
204 File base_ = SimanticsInternal.getTemporaryDirectory();
205 File base = new File(base_, "exports");
208 File otherStatementsFile = new File(base, otherStatements);
209 File valueFile = new File(base, valueFileName);
211 // System.err.println("f: " + otherStatementsFile.getAbsolutePath());
214 DeferredFileOutputStream otherStatementsStream = new DeferredFileOutputStream(1024*1024, otherStatementsFile);
216 DataOutputStream otherStatementsOutput = new DataOutputStream(new BufferedOutputStream(otherStatementsStream, 1024*1024));
217 DeferredBinaryFile valueOutput = new DeferredBinaryFile(valueFile, 1024*1024, 128*1024);
219 ClusterControl cc = graph.getService(ClusterControl.class);
220 ClusterState clusterState = cc.getClusterState();
222 TIntHashSet excludedShared = new TIntHashSet();
224 ids = new TIntIntHashMap(1000, 0.75f);
226 DomainProcessorState state = new DomainProcessorState();
227 state.extensions.putAll(configuration.baseExtensions);
229 state.statementsOutput = otherStatementsOutput;
230 state.valueOutput = valueOutput;
231 state.valueCount = 0;
232 state.excludedShared = excludedShared;
233 state.monitor = monitor;
234 state.valueModifier = composeTGValueModifier(configuration.valueModifiers);
236 getDomain2(graph, configuration, state, configuration.ignoreVirtualResources);
240 cc.restoreClusterState(clusterState);
242 otherStatementsOutput.flush();
243 otherStatementsOutput.close();
245 // Do not close valueOutput, just flush it and reuse it in
246 // ModelTransferableGraphSource for reading.
249 long domainDuration = System.nanoTime() - domainTime;
252 state.otherStatementsInput = toRandomAccessBinary(otherStatementsStream, 128*1024);
253 state.valueInput = toRandomAccessBinary(valueOutput);
254 state.statementsOutput = null;
255 state.valueOutput = null;
257 long totalEnd = System.nanoTime();
260 System.out.println("startup in " + 1e-9*(startupTimeEnd - startupTime) + "s.");
261 System.out.println("domain was found in " + 1e-9*(domainDuration) + "s.");
262 System.out.println("total time for building subgraph was " + 1e-9*(totalEnd-total) + "s.");
265 return getSource(graph, configuration, state, otherStatementsFile, valueFile);
267 } catch (DatabaseException e) {
269 } catch (IOException e) {
270 throw new DatabaseException(e.getMessage(), e);
271 } catch (Throwable e) {
272 dumpHeap("crash.hprof");
273 throw new DatabaseException(e.getMessage(), e);
278 protected ModelTransferableGraphSource getSource(ReadGraph graph, TransferableGraphConfiguration2 configuration, DomainProcessorState state, File otherStatementsFile, File valueFile) throws DatabaseException {
279 return new ModelTransferableGraphSource(graph, configuration, state, otherStatementsFile, valueFile);
282 private TGValueModifier composeTGValueModifier(Collection<TGValueModifier> configuredModifiers) {
283 List<TGValueModifier> valueModifiers = configuredModifiers == null ? new ArrayList<>(2) : new ArrayList<>(configuredModifiers.size() + 2);
284 valueModifiers.add(new ResourceTGValueModifier(support));
285 valueModifiers.add(RevisionTGValueModifier.INSTANCE);
286 return new ComposedTGValueModifier(valueModifiers.toArray(new TGValueModifier[valueModifiers.size()]));
289 private static RandomAccessBinary toRandomAccessBinary(DeferredFileOutputStream stream, int bufferSize) throws IOException {
290 if (stream.isInMemory())
291 return new BinaryMemory(stream.getData());
292 return new BinaryFile(stream.getFile(), bufferSize);
295 private static RandomAccessBinary toRandomAccessBinary(DeferredBinaryFile file) throws IOException {
296 RandomAccessBinary b = file.getBackend();
297 long size = b.position();
299 if (b instanceof BinaryMemory) {
305 public static DomainOnlyProcessor getDomainOnly(RequestProcessor processor, IProgressMonitor monitor, final Resource resource) throws DatabaseException {
306 return getDomainOnly(processor, monitor, Collections.singletonList(resource));
309 public static DomainOnlyProcessor getDomainOnly(RequestProcessor processor, final IProgressMonitor monitor, final Collection<Resource> resources) throws DatabaseException {
311 return processor.syncRequest(new UniqueRead<DomainOnlyProcessor>() {
314 public DomainOnlyProcessor perform(ReadGraph graph) throws DatabaseException {
318 TransferableGraphConfiguration2 conf = TransferableGraphConfiguration2.createWithResources(graph, resources, Collections.<Resource>emptyList());
320 DomainProcessorState state = new DomainProcessorState();
321 state.extensions.putAll(conf.baseExtensions);
322 state.ids = new TIntIntHashMap(1000, 0.75f);
323 state.statementsOutput = new DataOutputStream(new NullOutputStream());
324 state.valueOutput = new NullRandomAccessBinary();
325 state.valueCount = 0;
326 state.excludedShared = new TIntHashSet();
327 state.monitor = SubMonitor.convert(monitor);
329 return getDomainOnly(graph, conf, state, conf.ignoreVirtualResources);
331 } catch (OperationCanceledException e) {
343 public static class DomainOnlyProcessor extends DomainProcessor3 {
345 final Resource instanceOf;
346 final public List<Resource> internals;
347 final public List<Resource> internalTypes;
349 private int counter = 0;
351 public DomainOnlyProcessor(ReadGraph graph, TransferableGraphConfiguration2 conf, DomainProcessorState state, boolean ignoreVirtual) throws DatabaseException {
352 super(graph, conf, state, ignoreVirtual);
353 CollectionSupport cs = graph.getService(CollectionSupport.class);
354 internals = cs.createList();
355 internalTypes = cs.createList();
356 instanceOf = Layer0.getInstance(graph).InstanceOf;
360 final public void addToStream(Resource predicate, Resource object) throws DatabaseException {
364 public void flushStatementStream(int sId, DomainProcessorState state) throws IOException {
368 public void processValue(ReadGraph graph, Resource subject, int sId, DomainProcessorState state) throws DatabaseException, IOException {
372 public void processInternal(ReadGraph graph, Resource subject, DirectStatements stms, DomainProcessorState state) throws DatabaseException, IOException {
374 if((counter++ & 1023) == 0)
375 if(state.monitor != null)
376 if(state.monitor.isCanceled())
377 throw new CancelTransactionException();
379 super.processInternal(graph, subject, stms, state);
381 internals.add(subject);
383 Resource singleType = null;
384 for(Statement s : stms) {
385 if(instanceOf.equals(s.getPredicate())) {
386 if(singleType != null) {
387 internalTypes.add(null);
390 singleType = s.getObject();
395 internalTypes.add(singleType);
402 public static DomainOnlyProcessor getDomainOnly(final ReadGraph graph , final TransferableGraphConfiguration2 conf, DomainProcessorState state, boolean ignoreVirtual) throws DatabaseException {
404 final DomainOnlyProcessor processor = new DomainOnlyProcessor(graph, conf, state, ignoreVirtual);
405 getDomain2(graph, state, processor);
410 public static DomainProcessor3 getDomain2(final ReadGraph graph , final TransferableGraphConfiguration2 conf, DomainProcessorState state, boolean ignoreVirtual) throws DatabaseException {
412 ITask task = ThreadLogger.getInstance().begin("getDomain2");
414 final DomainProcessor3 processor = new DomainProcessor3(graph, conf, state, ignoreVirtual);
416 getDomain2(graph, state, processor);
418 final SerialisationSupport support = graph.getService(SerialisationSupport.class);
419 final ClusteringSupport cls = graph.getService(ClusteringSupport.class);
420 final Resource indexRoot = processor.conf.indexRoot;
422 if (state.monitor.isCanceled())
423 throw new CancelTransactionException();
425 TLongObjectHashMap<TIntArrayList> clusterMap = new TLongObjectHashMap<TIntArrayList>();
426 for(Resource r : processor.status.keySet()) {
427 ExtentStatus status = processor.status.get(r);
428 int transientId = support.getTransientId(r);
429 if(ExtentStatus.INTERNAL == status) {
430 long cluster = cls.getCluster(r);
431 TIntArrayList list = clusterMap.get(cluster);
433 list = new TIntArrayList();
434 clusterMap.put(cluster, list);
436 list.add(transientId);
437 } else if(ExtentStatus.EXTERNAL == status) {
438 state.externals.add(transientId);
439 } else if(ExtentStatus.PENDING == status) {
440 String uri = graph.getPossibleURI(r);
442 // All internal resources with uri have been discovered already => this must then be external
443 //state.externals.add(transientId);
444 // Pending resources are found through weak links - if they are still pending at this stage do not add an external
445 processor.status.put(r, ExtentStatus.EXTERNAL);
448 state.pending.add(transientId);
449 System.err.println("Pending status in export: " + NameUtils.getSafeName(graph, r, true) + " (" + graph.getPossibleURI(r) + ")");
454 // Now that we know the status of the resources lets process weak statements
455 for(Statement stm : processor.unresolvedWeakLinks) {
456 ExtentStatus status = processor.status.get(stm.getObject());
457 if(ExtentStatus.INTERNAL == status) {
458 // Weak links between internals are exported
459 int transientId = support.getTransientId(stm.getSubject());
460 processor.addToStream(stm.getPredicate(), stm.getObject());
462 processor.flushStatementStream(transientId, state);
463 } catch (IOException e) {
464 throw new DatabaseException(e);
469 if (state.monitor.isCanceled())
470 throw new CancelTransactionException();
472 final TIntArrayList clustering = new TIntArrayList();
473 clusterMap.forEachEntry(new TLongObjectProcedure<TIntArrayList>() {
476 public boolean execute(long cluster, TIntArrayList b) {
477 clustering.add(b.size());
478 b.forEach(new TIntProcedure() {
481 public boolean execute(int rId) {
482 processor.ids.put(rId, processor.id++);
492 if (state.monitor.isCanceled())
493 throw new CancelTransactionException();
495 final TIntArrayList clusterSets = new TIntArrayList();
496 clusterMap.forEachEntry(new TLongObjectProcedure<TIntArrayList>() {
499 public boolean execute(long cluster, TIntArrayList b) {
501 Resource clusterSet = cls.getClusterSetOfCluster(cluster);
502 if(clusterSet != null) {
503 int transientId = support.getTransientId(clusterSet);
504 if(processor.ids.containsKey(transientId)) {
505 clusterSets.add(processor.ids.get(transientId));
508 if(graph.getRootLibrary().equals(clusterSet)) {
509 clusterSets.add(Extensions.ROOT_LIBRARY_CLUSTER_SET);
511 } else if (clusterSet.equals(indexRoot)) {
512 clusterSets.add(Extensions.INDEX_ROOT_CLUSTER_SET);
517 } catch (DatabaseException e) {
519 clusterSets.add(Extensions.NO_CLUSTER_SET);
526 state.extensions.put(Extensions.CLUSTERING, new Variant(Bindings.INT_ARRAY, clustering.toArray()));
527 state.extensions.put(Extensions.CLUSTER_SETS, new Variant(Bindings.INT_ARRAY, clusterSets.toArray()));
529 long total = processor.startupTime + processor.expandTime
530 + processor.classifyPredicateTime
531 + processor.processFringeTime + processor.extentSeedTime
532 + processor.fullResolveTime + processor.fastResolveTime +
533 + processor.parentResolveTime + processor.otherStatementTime;
536 System.out.println("startup took " + 1e-9 * processor.startupTime + "s.");
537 System.out.println("expand took " + 1e-9 * processor.expandTime + "s.");
538 System.out.println("classifyPredicates took " + 1e-9 * processor.classifyPredicateTime + "s.");
539 System.out.println("processFringe took " + 1e-9 * processor.processFringeTime + "s.");
540 System.out.println("extentSeeding took " + 1e-9 * processor.extentSeedTime + "s.");
541 System.out.println("fullResolve took " + 1e-9 * processor.fullResolveTime + "s.");
542 System.out.println("fastResolve took " + 1e-9 * processor.fastResolveTime + "s.");
543 System.out.println("parentResolve took " + 1e-9 * processor.parentResolveTime + "s.");
544 System.out.println("otherStatements took " + 1e-9 * processor.otherStatementTime + "s.");
545 System.out.println("value output took " + 1e-9 * processor.valueOutputTime + "s.");
546 System.out.println("statement output took " + 1e-9 * processor.statementOutputTime + "s.");
547 System.out.println("total " + 1e-9 * total + "s.");
556 public static DomainProcessor3 getDomain2(final ReadGraph graph , DomainProcessorState state, final DomainProcessor3 processor) throws DatabaseException {
557 processor.process(graph, state);
561 static class Expansion3 extends UniqueRead<Collection<DirectStatements>> {
563 final private Collection<Resource> roots;
564 final boolean ignoreVirtual;
566 public Expansion3(Collection<Resource> roots, boolean ignoreVirtual) {
568 this.ignoreVirtual = ignoreVirtual;
572 public Collection<DirectStatements> perform(ReadGraph graph) {
574 ArrayList<DirectStatements> result = new ArrayList<DirectStatements>();
576 final DirectQuerySupport dqs = graph.getService(DirectQuerySupport.class);
578 final DomainStatementProcedure3 proc = new DomainStatementProcedure3(result);
581 for(Resource r : roots) {
582 dqs.forEachDirectPersistentStatement(graph, r, proc);
585 for(Resource r : roots) {
586 dqs.forEachDirectStatement(graph, r, proc);
595 private static void dumpHeap(String path) {
598 Object bean = getBean();
602 Method m = bean.getClass().getMethod("dumpHeap", String.class, boolean.class);
603 m.invoke(bean, path, true);
605 } catch (IllegalArgumentException e) {
606 } catch (IllegalAccessException e) {
607 } catch (SecurityException e) {
608 } catch (NoSuchMethodException e) {
609 } catch (InvocationTargetException e) {
615 private static Object getBean() {
616 Class<?> beanClass = getBeanClass();
617 if (beanClass == null)
620 Object bean = ManagementFactory.newPlatformMXBeanProxy(
621 ManagementFactory.getPlatformMBeanServer(),
622 "com.sun.management:type=HotSpotDiagnostic",
625 } catch (IOException e) {
630 private static Class<?> getBeanClass() {
632 Class<?> clazz = Class.forName("com.sun.management.HotSpotDiagnosticMXBean");
634 } catch (ClassNotFoundException e) {
639 public static void main(String[] args) {
643 Datatype dt = Datatypes.translate("{ parts : ( | ResourceRVIPart { role : |CHILD|PROPERTY, resource : Long(unit=\"resource\") } | StringRVIPart { role : |CHILD|PROPERTY, string : String } ) [] }");
644 Binding b = Bindings.getBinding(dt);
645 Object value = b.createDefault();
646 Variant variant = new Variant(b, value);
647 TGResourceUtil util = new TGResourceUtil();
648 LongAdapter la = new LongAdapter() {
650 public long adapt(long in) {
654 util.adaptValue( variant.getBinding(), variant.getValue(), la );
656 } catch (DataTypeSyntaxError e) {
658 } catch (BindingException e) {
660 } catch (RuntimeSerializerConstructionException e) {
662 } catch (AccessorException e) {