--- /dev/null
+/*******************************************************************************\r
+ * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
+ * in Industry THTH ry.\r
+ * All rights reserved. This program and the accompanying materials\r
+ * are made available under the terms of the Eclipse Public License v1.0\r
+ * which accompanies this distribution, and is available at\r
+ * http://www.eclipse.org/legal/epl-v10.html\r
+ *\r
+ * Contributors:\r
+ * VTT Technical Research Centre of Finland - initial API and implementation\r
+ *******************************************************************************/\r
+package org.simantics.db.impl;\r
+\r
+import java.io.Closeable;\r
+import java.io.File;\r
+import java.io.IOException;\r
+import java.util.ArrayList;\r
+import java.util.Arrays;\r
+import java.util.Collection;\r
+import java.util.HashSet;\r
+import java.util.LinkedList;\r
+\r
+import org.simantics.databoard.Bindings;\r
+import org.simantics.databoard.binding.Binding;\r
+import org.simantics.databoard.serialization.SerializationException;\r
+import org.simantics.databoard.serialization.Serializer;\r
+import org.simantics.databoard.serialization.SerializerConstructionException;\r
+import org.simantics.db.AsyncReadGraph;\r
+import org.simantics.db.AsyncRequestProcessor;\r
+import org.simantics.db.RequestProcessor;\r
+import org.simantics.db.Resource;\r
+import org.simantics.db.Statement;\r
+import org.simantics.db.VirtualGraphContext;\r
+import org.simantics.db.VirtualGraphSource;\r
+import org.simantics.db.WriteGraph;\r
+import org.simantics.db.WriteOnlyGraph;\r
+import org.simantics.db.common.ByteFileReader;\r
+import org.simantics.db.common.ByteFileWriter;\r
+import org.simantics.db.common.StandardStatement;\r
+import org.simantics.db.common.request.WriteOnlyRequest;\r
+import org.simantics.db.common.request.WriteRequest;\r
+import org.simantics.db.common.utils.Logger;\r
+import org.simantics.db.exception.DatabaseException;\r
+import org.simantics.db.impl.graph.ReadGraphImpl;\r
+import org.simantics.db.impl.support.ResourceSupport;\r
+import org.simantics.db.impl.support.VirtualGraphServerSupport;\r
+import org.simantics.db.procedure.AsyncProcedure;\r
+import org.simantics.db.request.Write;\r
+import org.simantics.db.request.WriteOnly;\r
+import org.simantics.db.service.SerialisationSupport;\r
+import org.simantics.utils.datastructures.Callback;\r
+\r
+import gnu.trove.list.array.TIntArrayList;\r
+import gnu.trove.map.hash.TIntObjectHashMap;\r
+import gnu.trove.procedure.TIntObjectProcedure;\r
+import gnu.trove.procedure.TIntProcedure;\r
+import gnu.trove.set.hash.TIntHashSet;\r
+\r
+class VirtualCluster {\r
+ static final boolean DEBUG = false;\r
+ final static int[] EMPTY = new int[0];\r
+\r
+ private final ArrayList<TIntArrayList> statements = new ArrayList<TIntArrayList>();\r
+ private final TIntHashSet lazy = new TIntHashSet();\r
+ private final TIntObjectHashMap<byte[]> values = new TIntObjectHashMap<byte[]>();\r
+ private final int clusterId;\r
+\r
+ public VirtualCluster(int clusterId) {\r
+ this.clusterId = clusterId;\r
+ }\r
+ \r
+ public int clusterId() {\r
+ return clusterId;\r
+ }\r
+ \r
+ public void trim() {\r
+ }\r
+ \r
+ private TIntArrayList getPredicateMap(int subject) {\r
+ \r
+ int rId = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
+ if(rId >= statements.size()) return null;\r
+ return statements.get(rId);\r
+ \r
+ }\r
+ \r
+ public boolean isPending(int subject) {\r
+ return lazy.contains(subject);\r
+ }\r
+\r
+ boolean containsPredicate(TIntArrayList statements, int predicate) {\r
+ for(int i=0;i<statements.size();i+=2) if(statements.getQuick(i) == predicate) return true;\r
+ return false;\r
+ }\r
+\r
+ int containsStatement(TIntArrayList statements, int predicate, int object) {\r
+ for(int i=0;i<statements.size();i+=2) if(statements.getQuick(i) == predicate && statements.getQuick(i+1) == object) return i;\r
+ return -1;\r
+ }\r
+ \r
+ public boolean isPending(int subject, int predicate) {\r
+ \r
+ if(!lazy.contains(subject)) return false;\r
+ TIntArrayList predicateMap = getPredicateMap(subject);\r
+ if(predicateMap == null) return true;\r
+ return !containsPredicate(predicateMap, predicate);\r
+// return !predicateMap.contains(predicate);\r
+ \r
+ }\r
+ \r
+ public void resetLazy(int subject) {\r
+\r
+ lazy.remove(subject);\r
+ // Query all data from scratch\r
+ int ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
+ statements.set(ri, new TIntArrayList());\r
+\r
+ }\r
+ \r
+ public void setLazy(int subject) {\r
+\r
+ lazy.add(subject);\r
+ \r
+ }\r
+\r
+ public void finish(int subject) {\r
+\r
+ lazy.remove(subject);\r
+ \r
+ }\r
+ \r
+ public void setValue(int subject, byte[] data, int length) {\r
+\r
+ values.put(subject, Arrays.copyOf(data, length));\r
+ \r
+ }\r
+\r
+ public void denyValue(int subject) {\r
+ \r
+ values.remove(subject);\r
+ \r
+ }\r
+\r
+ public void addStatements(int subject, int[] data) {\r
+ \r
+ TIntArrayList result = getPredicateMap(subject);\r
+ if(result == null) {\r
+ result = new TIntArrayList();\r
+ int rId = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
+ statements.ensureCapacity(rId+1);\r
+ for(int i=statements.size();i<rId+1;i++) statements.add(null);\r
+ statements.set(rId, result);\r
+ }\r
+ \r
+ for(int i=0;i<data.length;i+=2) {\r
+ int predicate = data[i];\r
+ int object = data[i+1];\r
+ if(containsStatement(result, predicate, object) < 0) {\r
+ result.add(predicate);\r
+ result.add(object);\r
+ }\r
+ }\r
+ \r
+ }\r
+ \r
+ public void claim(int subject, int predicate, int object) {\r
+ \r
+ TIntArrayList predicates = getPredicateMap(subject);\r
+ if(predicates == null) {\r
+ predicates = new TIntArrayList();\r
+ int rId = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
+ statements.ensureCapacity(rId+1);\r
+ for(int i=statements.size();i<rId+1;i++) statements.add(null);\r
+ statements.set(rId, predicates);\r
+ }\r
+ \r
+ if(containsStatement(predicates, predicate, object) < 0) {\r
+ predicates.add(predicate);\r
+ predicates.add(object);\r
+ }\r
+ \r
+ }\r
+ \r
+ public void deny(int subject, int predicate, int object) {\r
+ \r
+ TIntArrayList predicates = getPredicateMap(subject);\r
+ if(predicates == null) return;\r
+ int index = containsStatement(predicates, predicate, object);\r
+ if(index < 0) return;\r
+ predicates.remove(index, 2);\r
+ \r
+ }\r
+ \r
+ int[] getObjects(int subject, int predicate) {\r
+ \r
+ TIntArrayList predicates = getPredicateMap(subject);\r
+ if(predicates == null) return EMPTY;\r
+ TIntArrayList result = new TIntArrayList();\r
+ for(int i=0;i<predicates.size();i+=2) {\r
+ if(predicates.getQuick(i) == predicate) {\r
+ result.add(predicates.getQuick(i+1));\r
+ }\r
+ }\r
+ return result.toArray();\r
+ \r
+ }\r
+ \r
+ int[] getPredicates(int subject) {\r
+\r
+ TIntArrayList predicates = getPredicateMap(subject);\r
+ if(predicates == null) return EMPTY;\r
+ TIntHashSet result = new TIntHashSet();\r
+ for(int i=0;i<predicates.size();i+=2) {\r
+ result.add(predicates.getQuick(i));\r
+ }\r
+ return result.toArray();\r
+ \r
+ }\r
+ \r
+ byte[] getValue(int subject) {\r
+ return values.get(subject);\r
+ }\r
+\r
+ public int getTransientClusterKey() {\r
+ int clusterBits = ClusterTraitsBase.getClusterBits(clusterId>>>1);\r
+ if((clusterId & 1) == 0) // Virtual subjects\r
+ return clusterBits | 0x80000000; // We are assuming that MSB means virtual resource.\r
+ else // Database subjects\r
+ return clusterBits;\r
+ }\r
+\r
+ public int getTransientId(int subject) {\r
+ if((clusterId & 1) == 0) // Virtual subjects\r
+ return ClusterTraitsBase.createResourceKeyNoThrow(clusterId>>1, subject) | 0x80000000;\r
+ else // Database subjects\r
+ return ClusterTraitsBase.createResourceKeyNoThrow(clusterId>>1, subject);\r
+ }\r
+ \r
+ /*\r
+ * Creates a persistent identifier for given transient graph cluster identifier.\r
+ * \r
+ * \r
+ * For persistent clusters this obtains the random access id for (invalid)\r
+ * resource at index 0 of given cluster. LSB in given id indicates persistence\r
+ * \r
+ * \r
+ */\r
+ \r
+ public static long getClusterIdentifier(SerialisationSupport ss, int clusterKey) { \r
+ if((clusterKey & 1) == 0)// Virtual subjects\r
+ return clusterKey;\r
+ else { // Database subjects\r
+ int rk = ClusterTraitsBase.createResourceKeyNoThrow(clusterKey>>1, 1);\r
+ try {\r
+ // Assuming that cluster's first resource is created first and not deleted.\r
+ // Assuming that resource index is in LSB bits.\r
+ return ss.getRandomAccessId(rk);\r
+ } catch (DatabaseException e) {\r
+ Logger.defaultLogError("Could not get cluster id for virtual cluster key=" + clusterKey, e);\r
+ return -1;\r
+ }\r
+ }\r
+ }\r
+\r
+ public void saveImpl(final File file, SerialisationSupport ss) throws IOException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Saving virtual cluster " + clusterId + " to " + file.getAbsolutePath());\r
+ \r
+ final ByteFileWriter writer = new ByteFileWriter(file);\r
+ \r
+ try {\r
+\r
+ int stms = 0;\r
+ for(TIntArrayList list : statements) {\r
+ if(list != null)\r
+ stms += list.size();\r
+ }\r
+\r
+ writer.write((int)(1.5*stms));\r
+ @SuppressWarnings("unused")\r
+ int count = 0;\r
+ for(int i=0;i<statements.size();i++) {\r
+ TIntArrayList list = statements.get(i);\r
+ if(list != null) {\r
+ for(int j=0;j<list.size();j+=2) {\r
+ try {\r
+ writer.write((short)i);\r
+ int rk = list.getQuick(j);\r
+ long rid = ss.getRandomAccessId(rk);\r
+ writer.write(rid);\r
+ rk = list.getQuick(j+1);\r
+ rid = ss.getRandomAccessId(rk);\r
+ writer.write(rid);\r
+ count++;\r
+ } catch (DatabaseException e) {\r
+ e.printStackTrace();\r
+ }\r
+ }\r
+ }\r
+ }\r
+\r
+ writer.write(values.size());\r
+\r
+ values.forEachEntry(new TIntObjectProcedure<byte[]>() {\r
+\r
+ @Override\r
+ public boolean execute(int a, byte[] b) {\r
+ writer.write(a);\r
+ writer.write(b.length);\r
+ writer.write(b);\r
+ return true;\r
+ }\r
+ });\r
+ if (DEBUG)\r
+ System.out.println("TransientGraph[" + file.getAbsolutePath() + "] wrote " + count + " statements and " + values.size() + " values to disk.");\r
+\r
+ } finally {\r
+ writer.commit();\r
+// FileUtils.uncheckedClose(_os);\r
+ }\r
+ }\r
+ \r
+ public void load(File file, SerialisationSupport serialization, VirtualGraphServerSupport vgss) throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: Loading virtual cluster " + clusterId + " from " + file.getAbsolutePath() + " " + file.length());\r
+ \r
+ ByteFileReader reader = null;\r
+ try {\r
+ \r
+ if (!file.exists())\r
+ return;\r
+\r
+ reader = new ByteFileReader(file);\r
+ int clusterInt = ClusterTraitsBase.getClusterBits(clusterId()>>1);\r
+ int stms = reader.readInt();\r
+ for (int i = 0; i < stms; i += 3) {\r
+// int rId = reader.readShort();\r
+// if(vgss != null) vgss.addVirtual(clusterInt + rId);\r
+// claim(rId,\r
+// serialization.getTransientId(reader.readLong()),\r
+// serialization.getTransientId(reader.readLong()));\r
+ int rId = reader.readShort();\r
+ long sId = reader.readLong();\r
+ long oId = reader.readLong();\r
+ int sTransientId = serialization.getTransientId(sId);\r
+ int oTransientId = serialization.getTransientId(oId);\r
+ \r
+ if(vgss != null)\r
+ vgss.addVirtual(clusterInt + rId);\r
+ \r
+ claim(rId, sTransientId, oTransientId);\r
+ }\r
+\r
+ int values = reader.readInt();\r
+ for (int i = 0; i < values; i++) {\r
+ int subject = reader.readInt();\r
+ int length = reader.readInt();\r
+ setValue(subject, reader.readBytes(length), length);\r
+ }\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: TransientGraph[" + file.getAbsolutePath() + "] loaded " + stms / 3 + " statements and " + values + " values from disk.");\r
+ \r
+ } catch (IOException e) {\r
+ throw new DatabaseException(e);\r
+ } finally {\r
+ if (reader != null)\r
+ reader.close();\r
+ }\r
+ }\r
+ \r
+ void listStatements(SerialisationSupport ss, ArrayList<Statement> result) {\r
+\r
+ int clusterKey = getTransientClusterKey();\r
+ \r
+ try {\r
+ for(int i=0;i<statements.size();i++) {\r
+ TIntArrayList list = statements.get(i);\r
+ if(list != null) {\r
+ Resource subject = ss.getResource(clusterKey | i);\r
+ for(int j=0;j<list.size();j+=2) {\r
+ Resource p = ss.getResource(list.getQuick(j));\r
+ Resource o = ss.getResource(list.getQuick(j+1));\r
+ result.add(new StandardStatement(subject, p, o));\r
+ }\r
+\r
+ }\r
+ }\r
+ } catch (DatabaseException e) {\r
+ e.printStackTrace();\r
+ }\r
+ \r
+ }\r
+\r
+ void listValues(final SerialisationSupport ss, final ArrayList<Resource> result) {\r
+\r
+ values.forEachKey(new TIntProcedure() {\r
+\r
+ @Override\r
+ public boolean execute(int value) {\r
+ try {\r
+ result.add(ss.getResource(getTransientId(value)));\r
+ } catch (DatabaseException e) {\r
+ e.printStackTrace();\r
+ }\r
+ return true;\r
+ }\r
+\r
+ });\r
+ \r
+ }\r
+\r
+}\r
+\r
+public class TransientGraph implements VirtualGraphImpl, VirtualGraphContext {\r
+ private static final boolean DEBUG = VirtualCluster.DEBUG;\r
+ final private static int SWAP_LIMIT = 30;\r
+ \r
+// final private static byte[] NO_VALUE = new byte[0];\r
+ final private static VirtualCluster NO_CLUSTER = new VirtualCluster(-1);\r
+ \r
+ final private Persistency persistency;\r
+ \r
+ final private SerialisationSupport serialization;\r
+ final private ResourceSupport resourceSupport;\r
+ final private VirtualGraphServerSupport virtualGraphServerSupport;\r
+ final private RequestProcessor sessionRequestProcessor;\r
+ \r
+ /*\r
+ * Cluster array by index.\r
+ * -NO_CLUSTER value means that there is no such virtual cluster\r
+ * -null value means that such virtual cluster could be on disk\r
+ */\r
+ final private ArrayList<VirtualCluster> clusters = new ArrayList<VirtualCluster>();\r
+\r
+ /*\r
+ * A list of resident clusters\r
+ */\r
+ final private LinkedList<VirtualCluster> memoryClusters = new LinkedList<VirtualCluster>();\r
+ \r
+ private final HashSet<VirtualGraphSource> sources = new HashSet<VirtualGraphSource>();\r
+ \r
+ final String identifier;\r
+ final String databaseId;\r
+\r
+ TIntObjectHashMap<TIntHashSet> NO_STATEMENTS = new TIntObjectHashMap<TIntHashSet>();\r
+\r
+ int[] EMPTY = new int[0];\r
+\r
+ public static TransientGraph workspacePersistent(SerialisationSupport ss, VirtualGraphServerSupport vgss, ResourceSupport rs, RequestProcessor srp, String databaseId, String identifier) throws DatabaseException {\r
+ TransientGraph graph = new TransientGraph(ss, vgss, rs, srp, databaseId, identifier, Persistency.WORKSPACE);\r
+ graph.load();\r
+ return graph;\r
+ }\r
+ \r
+ public static TransientGraph memoryPersistent(SerialisationSupport ss, VirtualGraphServerSupport vgss, ResourceSupport rs, RequestProcessor srp, String databaseId, String identifier) {\r
+ return new TransientGraph(ss, vgss, rs, srp, databaseId, identifier, Persistency.MEMORY);\r
+ }\r
+ \r
+ private TransientGraph(SerialisationSupport ss, VirtualGraphServerSupport vgss, ResourceSupport rs, RequestProcessor srp, String databaseId, String identifier, Persistency persistency) {\r
+ this.serialization = ss;\r
+ this.virtualGraphServerSupport = vgss;\r
+ this.sessionRequestProcessor = srp;\r
+ this.resourceSupport = rs;\r
+ this.identifier = identifier;\r
+ this.databaseId = databaseId;\r
+ \r
+ this.persistency = persistency;\r
+ }\r
+\r
+ public String getIdentifier() {\r
+ return identifier;\r
+ }\r
+ \r
+ private int transientClusterId(long id) throws DatabaseException {\r
+ if (DEBUG)\r
+ System.out.println("DEBUG: transientClusterId=" + id);\r
+ if ((id & 1) == 0) // Cluster of virtual subjects\r
+ return (int)id;\r
+ // Corresponds to persistent cluster\r
+ int rk = serialization.getTransientId(id);\r
+ return ClusterTraitsBase.getClusterKeyFromResourceKey(rk) << 1 | 1;\r
+ }\r
+\r
+ private String getPrefix() {\r
+ return identifier + "." + persistency.identifier() + "." + databaseId;\r
+ }\r
+ \r
+ private void load() throws DatabaseException {\r
+\r
+ String prefix = getPrefix();\r
+ for(String file : virtualGraphServerSupport.storagePath().list()) {\r
+ try {\r
+ if(file.startsWith(prefix)) {\r
+ long clusterLong = Long.parseLong(file.substring(prefix.length()+4));\r
+ int clusterId = transientClusterId(clusterLong);\r
+ VirtualCluster cluster = new VirtualCluster(clusterId);\r
+ cluster.load(new File(virtualGraphServerSupport.storagePath(), file), serialization, (clusterId & 1) > 0 ? virtualGraphServerSupport : null);\r
+ clusters.ensureCapacity(clusterId+1);\r
+ for(int i=clusters.size(); i<clusterId+1; i++) clusters.add(null);\r
+ clusters.set(clusterId, cluster);\r
+ memoryClusters.addLast(cluster);\r
+ }\r
+ } catch (DatabaseException t) {\r
+ // file is assumably broken, delete it\r
+ File filee = new File(virtualGraphServerSupport.storagePath(), file);\r
+ if (!filee.delete()) {\r
+ System.err.println("Could not delete file " + filee.getAbsolutePath());\r
+ }\r
+ throw t;\r
+ }\r
+ }\r
+\r
+ }\r
+\r
+ public void dispose() {\r
+ try {\r
+ saveImpl(serialization);\r
+ } catch (IOException e) {\r
+ e.printStackTrace();\r
+ }\r
+ }\r
+\r
+ public void save() {\r
+ \r
+ try {\r
+ saveImpl(serialization);\r
+ } catch (IOException e) {\r
+ e.printStackTrace();\r
+ }\r
+ \r
+ }\r
+\r
+ public void saveImpl(final SerialisationSupport ss) throws IOException {\r
+ \r
+ for(VirtualCluster cluster : memoryClusters) {\r
+ String prefix = getPrefix();\r
+ File file = new File(virtualGraphServerSupport.storagePath(), prefix + ".vg." + VirtualCluster.getClusterIdentifier(ss, cluster.clusterId()));\r
+ cluster.saveImpl(file, ss);\r
+ }\r
+ \r
+ }\r
+\r
+ /**\r
+ * Closes a stream and ignores any resulting exception. This is useful\r
+ * when doing stream cleanup in a finally block where secondary exceptions\r
+ * are not worth logging.\r
+ */\r
+ static void uncheckedClose(Closeable closeable) {\r
+ try {\r
+ if (closeable != null)\r
+ closeable.close();\r
+ } catch (IOException e) {\r
+ //ignore\r
+ }\r
+ }\r
+\r
+ private void trimClusters() {\r
+ for(VirtualCluster cluster : memoryClusters) {\r
+ cluster.trim();\r
+ }\r
+ }\r
+ \r
+ /*\r
+ * Returns a transient cluster index\r
+ * -Transient clusters for persistent resources have index with LSB=1\r
+ * -Transient clusters for virtual resources have index with LSB=0\r
+ * \r
+ * @param subject is a DB client transient id\r
+ * \r
+ * For persistent resources cluster id is 2*clusterKey+1\r
+ * For virtual resources transient ids are persistent and are directly chunked into 14-bit clusters.\r
+ * \r
+ */\r
+ public static int getVirtualClusterKey(int subject) {\r
+ if (subject > 0) {\r
+ int ck = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(subject);\r
+ return (ck << 1) | 1;\r
+ }\r
+ int ck = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(~subject);\r
+ return ck << 1;\r
+ }\r
+ \r
+ private VirtualCluster getOrLoad(int virtualClusterKey) {\r
+\r
+ if(virtualClusterKey < clusters.size()) {\r
+ VirtualCluster cluster = clusters.get(virtualClusterKey);\r
+ if(NO_CLUSTER == cluster) return null;\r
+ if(cluster != null) return cluster;\r
+ }\r
+ if (DEBUG)\r
+ System.out.println("Loading virtual cluster " + virtualClusterKey + " for " + identifier);\r
+\r
+ clusters.ensureCapacity(virtualClusterKey+1);\r
+ for(int i=clusters.size(); i<virtualClusterKey+1; i++) clusters.add(null);\r
+\r
+ //if(!VirtualCluster.isValidVirtualClusterKey(serialization, virtualClusterKey)) return null;\r
+\r
+ long clusterIdentifier = VirtualCluster.getClusterIdentifier(serialization, virtualClusterKey);\r
+ File file = new File(virtualGraphServerSupport.storagePath(), getPrefix() + ".vg." + clusterIdentifier);\r
+ if(file.exists()) {\r
+ VirtualCluster cluster = new VirtualCluster(virtualClusterKey);\r
+// System.out.println("Loading virtual cluster2 " + virtualClusterKey + " for " + identifier);\r
+ try {\r
+ cluster.load(file, serialization, (virtualClusterKey & 1) > 0 ? virtualGraphServerSupport : null);\r
+ clusters.set(virtualClusterKey, cluster);\r
+ memoryClusters.addFirst(cluster);\r
+ swap();\r
+ return cluster;\r
+ } catch (DatabaseException e) {\r
+ e.printStackTrace();\r
+ // File must be corrupt, lets delete it so we wont load it next time in future\r
+ file.delete();\r
+ \r
+ clusters.set(virtualClusterKey, NO_CLUSTER);\r
+ return null;\r
+ }\r
+\r
+ } else {\r
+ clusters.set(virtualClusterKey, NO_CLUSTER);\r
+ return null;\r
+ }\r
+ }\r
+\r
+ private void swap() {\r
+ \r
+ trimClusters();\r
+ while(memoryClusters.size() > SWAP_LIMIT) {\r
+ VirtualCluster remo = memoryClusters.removeLast();\r
+ File file = new File(virtualGraphServerSupport.storagePath(), getPrefix() + ".vg." + VirtualCluster.getClusterIdentifier(serialization, remo.clusterId()));\r
+ try {\r
+ remo.saveImpl(file, serialization);\r
+ } catch (IOException e) {\r
+ e.printStackTrace();\r
+ }\r
+ clusters.set(remo.clusterId(), null);\r
+ }\r
+ \r
+ }\r
+ \r
+ private synchronized VirtualCluster getCluster(int subject, boolean create) {\r
+\r
+ int clusterId = getVirtualClusterKey(subject);\r
+ \r
+ VirtualCluster cluster = getOrLoad(clusterId);\r
+ if(cluster != null) return cluster;\r
+ \r
+ if(create) {\r
+\r
+ clusters.ensureCapacity(clusterId+1);\r
+ for(int i=clusters.size(); i<clusterId+1; i++) clusters.add(null);\r
+ cluster = new VirtualCluster(clusterId);\r
+ clusters.set(clusterId, cluster);\r
+ memoryClusters.addFirst(cluster);\r
+ swap();\r
+ return cluster;\r
+\r
+ } else {\r
+ return null;\r
+ }\r
+ \r
+ }\r
+ \r
+ private synchronized void applyValue(int subject, Object value, Binding binding) {\r
+\r
+ try {\r
+ Serializer serializer = Bindings.getSerializer( binding );\r
+ byte[] serialized = serializer.serialize(value);\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+ cluster.setValue(subject, serialized, serialized.length);\r
+ } catch (SerializationException e) {\r
+ e.printStackTrace();\r
+ } catch (SerializerConstructionException e) {\r
+ e.printStackTrace();\r
+ } catch (IOException e) {\r
+ // TODO Auto-generated catch block\r
+ e.printStackTrace();\r
+ }\r
+\r
+ }\r
+\r
+ private synchronized void applyStatements(int subject, int[] statements) {\r
+\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+ cluster.addStatements(subject, statements);\r
+ \r
+ \r
+ if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
+\r
+ }\r
+\r
+ private synchronized void produceAllStatements(ReadGraphImpl graph, final int subject, final AsyncProcedure<Object> procedure) throws DatabaseException {\r
+\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+\r
+ // This resource becomes a normal resource, all data is requeried\r
+ cluster.resetLazy(subject);\r
+ \r
+ for(VirtualGraphSource source : sources) {\r
+ source.getStatements(graph, this, subject);\r
+ }\r
+ \r
+ if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
+ \r
+ }\r
+\r
+ private synchronized void producePartialStatements(ReadGraphImpl graph, final int subject, final int predicate, final AsyncProcedure<Object> procedure) throws DatabaseException {\r
+\r
+ for(VirtualGraphSource source : sources) {\r
+ source.getStatements(graph, this, subject, predicate);\r
+ }\r
+ \r
+ if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
+ \r
+ }\r
+\r
+ @Override\r
+ public int getIndex(Resource resource) {\r
+ try {\r
+ return serialization.getTransientId(resource);\r
+ } catch (DatabaseException e) {\r
+ e.printStackTrace();\r
+ }\r
+ return 0;\r
+ }\r
+\r
+ @Override\r
+ public Resource getResource(int index) {\r
+ return new ResourceImpl(resourceSupport, index);\r
+ }\r
+\r
+ @Override\r
+ public void register(final VirtualGraphSource source) {\r
+ if(sources.add(source)) {\r
+ source.attach(TransientGraph.this);\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public void claim(int subject, int predicate, int object) {\r
+\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+ cluster.claim(subject, predicate, object);\r
+ if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
+\r
+ }\r
+\r
+ @Override\r
+ public synchronized int[] getObjects(int subject, int predicate) {\r
+ VirtualCluster cluster = getCluster(subject, false);\r
+ if(cluster == null) return EMPTY;\r
+ return cluster.getObjects(subject, predicate);\r
+ }\r
+\r
+ @Override\r
+ public synchronized int[] getPredicates(int subject) {\r
+ VirtualCluster cluster = getCluster(subject, false);\r
+ if(cluster == null) return EMPTY;\r
+ return cluster.getPredicates(subject);\r
+ }\r
+\r
+ @Override\r
+ public synchronized byte[] getValue(int subject) {\r
+ VirtualCluster cluster = getCluster(subject, false);\r
+ if(cluster == null) return null;\r
+ return cluster.getValue(subject);\r
+ }\r
+\r
+ @Override\r
+ public int newResource(boolean isLazy) {\r
+ \r
+ int id = virtualGraphServerSupport.createVirtual();\r
+ VirtualCluster cluster = getCluster(id, true);\r
+ if(isLazy) cluster.setLazy(id);\r
+ return id;\r
+\r
+ }\r
+ \r
+ @Override\r
+ public void finish(int subject) {\r
+ VirtualCluster cluster = getCluster(subject, false);\r
+ cluster.finish(subject);\r
+ }\r
+\r
+ @Override\r
+ public void deny(int subject, int predicate, int object) {\r
+\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+ cluster.deny(subject, predicate, object);\r
+\r
+ }\r
+\r
+ @Override\r
+ public void claimValue(int subject, byte[] data, int length) {\r
+\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+ cluster.setValue(subject, data, length);\r
+ if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
+\r
+ }\r
+\r
+ @Override\r
+ public void denyValue(int subject) {\r
+ // FIXME: this implementation is probably not proper, Antti needs to work on this.\r
+ VirtualCluster cluster = getCluster(subject, true);\r
+ cluster.denyValue(subject);\r
+ if(subject > 0) virtualGraphServerSupport.removeVirtual(subject);\r
+ }\r
+\r
+ @Override\r
+ public void initialise(final Write write) {\r
+ try {\r
+ sessionRequestProcessor.syncRequest(new WriteRequest(this) {\r
+\r
+ @Override\r
+ public void perform(WriteGraph graph) throws DatabaseException {\r
+ write.perform(graph);\r
+ }\r
+\r
+ });\r
+ } catch (DatabaseException e) {\r
+ e.printStackTrace();\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public void postModification(AsyncRequestProcessor processor, final WriteOnly request) {\r
+\r
+ if(processor == null) processor = sessionRequestProcessor;\r
+\r
+ processor.asyncRequest(new WriteOnlyRequest(this) {\r
+\r
+ @Override\r
+ public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
+ request.perform(graph);\r
+ }\r
+\r
+ });\r
+\r
+ }\r
+\r
+ @Override\r
+ public void updateStatements(int resource, int[] statements) {\r
+ applyStatements(resource, statements);\r
+ }\r
+\r
+ @Override\r
+ public void updateValue(int resource, Object value, Binding binding) {\r
+ applyValue(resource, value, binding);\r
+ }\r
+\r
+ @Override\r
+ public boolean isPending(int subject) {\r
+ \r
+ VirtualCluster cluster = getCluster(subject, false);\r
+ if(cluster == null) return false;\r
+ else return cluster.isPending(subject);\r
+ \r
+ }\r
+\r
+ @Override\r
+ public boolean isPending(int subject, int predicate) {\r
+\r
+ VirtualCluster cluster = getCluster(subject, false);\r
+ if(cluster == null) return false;\r
+ else return cluster.isPending(subject, predicate);\r
+\r
+ }\r
+ \r
+ @Override\r
+ public void load(ReadGraphImpl graph, int resource, int predicate, final Callback<ReadGraphImpl> callback) throws DatabaseException {\r
+ producePartialStatements(graph, resource, predicate, new AsyncProcedure<Object>() {\r
+\r
+ @Override\r
+ public void execute(AsyncReadGraph graph, Object result) {\r
+ callback.run((ReadGraphImpl)graph);\r
+ }\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable throwable) {\r
+ callback.run((ReadGraphImpl)graph);\r
+ }\r
+ \r
+ });\r
+ }\r
+\r
+ @Override\r
+ public void load(ReadGraphImpl graph, int resource, final Callback<ReadGraphImpl> callback) throws DatabaseException {\r
+ produceAllStatements(graph, resource, new AsyncProcedure<Object>() {\r
+\r
+ @Override\r
+ public void execute(AsyncReadGraph graph, Object result) {\r
+ callback.run((ReadGraphImpl)graph);\r
+ }\r
+\r
+ @Override\r
+ public void exception(AsyncReadGraph graph, Throwable throwable) {\r
+ callback.run((ReadGraphImpl)graph);\r
+ }\r
+ \r
+ });\r
+ }\r
+ \r
+ public Collection<Statement> listStatements() {\r
+ ArrayList<Statement> result = new ArrayList<Statement>();\r
+ for(int i=0;i<clusters.size();i++) {\r
+ VirtualCluster cluster = getOrLoad(i);\r
+ if(cluster != null) {\r
+ cluster.listStatements(serialization, result);\r
+ }\r
+ }\r
+ return result;\r
+ }\r
+\r
+ public Collection<Resource> listValues() {\r
+ ArrayList<Resource> result = new ArrayList<Resource>();\r
+ for(int i=0;i<clusters.size();i++) {\r
+ VirtualCluster cluster = getOrLoad(i);\r
+ if(cluster != null) {\r
+ cluster.listValues(serialization, result);\r
+ }\r
+ }\r
+ return result;\r
+ }\r
+\r
+ @Override\r
+ public Persistency getPersistency() {\r
+ return persistency;\r
+ }\r
+ \r
+ @Override\r
+ public String toString() {\r
+ String result = "'" + identifier + "'";\r
+ if(Persistency.WORKSPACE == persistency) result += " (W)";\r
+ else if(Persistency.MEMORY == persistency) result += " (M)";\r
+ return result;\r
+ }\r
+\r
+}\r
+\r
Code Review / simantics / platform.git / blobdiff