bundles/org.simantics.db.impl/src/org/simantics/db/impl/TransientGraph.java

   1 /*******************************************************************************\r
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
   3  * in Industry THTH ry.\r
   4  * All rights reserved. This program and the accompanying materials\r
   5  * are made available under the terms of the Eclipse Public License v1.0\r
   6  * which accompanies this distribution, and is available at\r
   7  * http://www.eclipse.org/legal/epl-v10.html\r
   8  *\r
   9  * Contributors:\r
  10  *     VTT Technical Research Centre of Finland - initial API and implementation\r
  11  *******************************************************************************/\r
  12 package org.simantics.db.impl;\r
  13 \r
  14 import java.io.Closeable;\r
  15 import java.io.File;\r
  16 import java.io.IOException;\r
  17 import java.util.ArrayList;\r
  18 import java.util.Arrays;\r
  19 import java.util.Collection;\r
  20 import java.util.HashSet;\r
  21 import java.util.LinkedList;\r
  22 \r
  23 import org.simantics.databoard.Bindings;\r
  24 import org.simantics.databoard.binding.Binding;\r
  25 import org.simantics.databoard.serialization.SerializationException;\r
  26 import org.simantics.databoard.serialization.Serializer;\r
  27 import org.simantics.databoard.serialization.SerializerConstructionException;\r
  28 import org.simantics.db.AsyncReadGraph;\r
  29 import org.simantics.db.AsyncRequestProcessor;\r
  30 import org.simantics.db.RequestProcessor;\r
  31 import org.simantics.db.Resource;\r
  32 import org.simantics.db.Statement;\r
  33 import org.simantics.db.VirtualGraphContext;\r
  34 import org.simantics.db.VirtualGraphSource;\r
  35 import org.simantics.db.WriteGraph;\r
  36 import org.simantics.db.WriteOnlyGraph;\r
  37 import org.simantics.db.common.ByteFileReader;\r
  38 import org.simantics.db.common.ByteFileWriter;\r
  39 import org.simantics.db.common.StandardStatement;\r
  40 import org.simantics.db.common.request.WriteOnlyRequest;\r
  41 import org.simantics.db.common.request.WriteRequest;\r
  42 import org.simantics.db.common.utils.Logger;\r
  43 import org.simantics.db.exception.DatabaseException;\r
  44 import org.simantics.db.impl.graph.ReadGraphImpl;\r
  45 import org.simantics.db.impl.support.ResourceSupport;\r
  46 import org.simantics.db.impl.support.VirtualGraphServerSupport;\r
  47 import org.simantics.db.procedure.AsyncProcedure;\r
  48 import org.simantics.db.request.Write;\r
  49 import org.simantics.db.request.WriteOnly;\r
  50 import org.simantics.db.service.SerialisationSupport;\r
  51 import org.simantics.utils.datastructures.Callback;\r
  52 \r
  53 import gnu.trove.list.array.TIntArrayList;\r
  54 import gnu.trove.map.hash.TIntObjectHashMap;\r
  55 import gnu.trove.procedure.TIntObjectProcedure;\r
  56 import gnu.trove.procedure.TIntProcedure;\r
  57 import gnu.trove.set.hash.TIntHashSet;\r
  58 \r
  59 class VirtualCluster {\r
  60     static final boolean DEBUG = false;\r
  61     final static int[] EMPTY = new int[0];\r
  62 \r
  63     private final ArrayList<TIntArrayList> statements = new ArrayList<TIntArrayList>();\r
  64     private final TIntHashSet lazy = new TIntHashSet();\r
  65     private final TIntObjectHashMap<byte[]> values = new TIntObjectHashMap<byte[]>();\r
  66     private final int clusterId;\r
  67 \r
  68     public VirtualCluster(int clusterId) {\r
  69         this.clusterId = clusterId;\r
  70     }\r
  71     \r
  72     public int clusterId() {\r
  73         return clusterId;\r
  74     }\r
  75     \r
  76     public void trim() {\r
  77     }\r
  78     \r
  79     private TIntArrayList getPredicateMap(int subject) {\r
  80         \r
  81         int rId = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
  82         if(rId >= statements.size()) return null;\r
  83         return statements.get(rId);\r
  84         \r
  85     }\r
  86     \r
  87     public boolean isPending(int subject) {\r
  88         return lazy.contains(subject);\r
  89     }\r
  90 \r
  91     boolean containsPredicate(TIntArrayList statements, int predicate) {\r
  92         for(int i=0;i<statements.size();i+=2) if(statements.getQuick(i) == predicate) return true;\r
  93         return false;\r
  94     }\r
  95 \r
  96     int containsStatement(TIntArrayList statements, int predicate, int object) {\r
  97         for(int i=0;i<statements.size();i+=2) if(statements.getQuick(i) == predicate && statements.getQuick(i+1) == object) return i;\r
  98         return -1;\r
  99     }\r
 100     \r
 101     public boolean isPending(int subject, int predicate) {\r
 102         \r
 103         if(!lazy.contains(subject)) return false;\r
 104         TIntArrayList predicateMap = getPredicateMap(subject);\r
 105         if(predicateMap == null) return true;\r
 106         return !containsPredicate(predicateMap, predicate);\r
 107 //        return !predicateMap.contains(predicate);\r
 108         \r
 109     }\r
 110     \r
 111     public void resetLazy(int subject) {\r
 112 \r
 113         lazy.remove(subject);\r
 114         // Query all data from scratch\r
 115         int ri = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
 116         statements.set(ri, new TIntArrayList());\r
 117 \r
 118     }\r
 119     \r
 120     public void setLazy(int subject) {\r
 121 \r
 122         lazy.add(subject);\r
 123         \r
 124     }\r
 125 \r
 126     public void finish(int subject) {\r
 127 \r
 128         lazy.remove(subject);\r
 129         \r
 130     }\r
 131     \r
 132     public void setValue(int subject, byte[] data, int length) {\r
 133 \r
 134         values.put(subject, Arrays.copyOf(data, length));\r
 135         \r
 136     }\r
 137 \r
 138     public void denyValue(int subject) {\r
 139         \r
 140         values.remove(subject);\r
 141         \r
 142     }\r
 143 \r
 144     public void addStatements(int subject, int[] data) {\r
 145         \r
 146         TIntArrayList result = getPredicateMap(subject);\r
 147         if(result == null) {\r
 148             result = new TIntArrayList();\r
 149             int rId = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
 150             statements.ensureCapacity(rId+1);\r
 151             for(int i=statements.size();i<rId+1;i++) statements.add(null);\r
 152             statements.set(rId, result);\r
 153         }\r
 154         \r
 155         for(int i=0;i<data.length;i+=2) {\r
 156             int predicate =  data[i];\r
 157             int object =  data[i+1];\r
 158             if(containsStatement(result, predicate, object) < 0) {\r
 159                 result.add(predicate);\r
 160                 result.add(object);\r
 161             }\r
 162         }\r
 163         \r
 164     }\r
 165     \r
 166     public void claim(int subject, int predicate, int object) {\r
 167         \r
 168         TIntArrayList predicates = getPredicateMap(subject);\r
 169         if(predicates == null) {\r
 170             predicates = new TIntArrayList();\r
 171             int rId = ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(subject);\r
 172             statements.ensureCapacity(rId+1);\r
 173             for(int i=statements.size();i<rId+1;i++) statements.add(null);\r
 174             statements.set(rId, predicates);\r
 175         }\r
 176         \r
 177         if(containsStatement(predicates, predicate, object) < 0) {\r
 178             predicates.add(predicate);\r
 179             predicates.add(object);\r
 180         }\r
 181         \r
 182     }\r
 183     \r
 184     public void deny(int subject, int predicate, int object) {\r
 185     \r
 186         TIntArrayList predicates = getPredicateMap(subject);\r
 187         if(predicates == null) return;\r
 188         int index = containsStatement(predicates, predicate, object);\r
 189         if(index < 0) return;\r
 190         predicates.remove(index, 2);\r
 191         \r
 192     }\r
 193     \r
 194     int[] getObjects(int subject, int predicate) {\r
 195         \r
 196         TIntArrayList predicates = getPredicateMap(subject);\r
 197         if(predicates == null) return EMPTY;\r
 198         TIntArrayList result = new TIntArrayList();\r
 199         for(int i=0;i<predicates.size();i+=2) {\r
 200             if(predicates.getQuick(i) == predicate) {\r
 201                 result.add(predicates.getQuick(i+1));\r
 202             }\r
 203         }\r
 204         return result.toArray();\r
 205         \r
 206     }\r
 207     \r
 208     int[] getPredicates(int subject) {\r
 209 \r
 210         TIntArrayList predicates = getPredicateMap(subject);\r
 211         if(predicates == null) return EMPTY;\r
 212         TIntHashSet result = new TIntHashSet();\r
 213         for(int i=0;i<predicates.size();i+=2) {\r
 214             result.add(predicates.getQuick(i));\r
 215         }\r
 216         return result.toArray();\r
 217         \r
 218     }\r
 219     \r
 220     byte[] getValue(int subject) {\r
 221         return values.get(subject);\r
 222     }\r
 223 \r
 224     public int getTransientClusterKey() {\r
 225         int clusterBits = ClusterTraitsBase.getClusterBits(clusterId>>>1);\r
 226         if((clusterId & 1) == 0) // Virtual subjects\r
 227             return clusterBits | 0x80000000; // We are assuming that MSB means virtual resource.\r
 228         else // Database subjects\r
 229             return clusterBits;\r
 230     }\r
 231 \r
 232     public int getTransientId(int subject) {\r
 233         if((clusterId & 1) == 0) // Virtual subjects\r
 234             return ClusterTraitsBase.createResourceKeyNoThrow(clusterId>>1, subject) | 0x80000000;\r
 235         else // Database subjects\r
 236             return ClusterTraitsBase.createResourceKeyNoThrow(clusterId>>1, subject);\r
 237     }\r
 238     \r
 239     /*\r
 240      * Creates a persistent identifier for given transient graph cluster identifier.\r
 241      * \r
 242      * \r
 243      * For persistent clusters this obtains the random access id for (invalid)\r
 244      * resource at index 0 of given cluster. LSB in given id indicates persistence\r
 245      * \r
 246      * \r
 247      */\r
 248     \r
 249     public static long getClusterIdentifier(SerialisationSupport ss, int clusterKey) { \r
 250         if((clusterKey & 1) == 0)// Virtual subjects\r
 251             return clusterKey;\r
 252         else { // Database subjects\r
 253             int rk = ClusterTraitsBase.createResourceKeyNoThrow(clusterKey>>1, 1);\r
 254             try {\r
 255                 // Assuming that cluster's first resource is created first and not deleted.\r
 256                 // Assuming that resource index is in LSB bits.\r
 257                 return ss.getRandomAccessId(rk);\r
 258             } catch (DatabaseException e) {\r
 259                 Logger.defaultLogError("Could not get cluster id for virtual cluster key=" + clusterKey, e);\r
 260                 return -1;\r
 261             }\r
 262         }\r
 263     }\r
 264 \r
 265     public void saveImpl(final File file, SerialisationSupport ss) throws IOException {\r
 266         if (DEBUG)\r
 267             System.out.println("DEBUG: Saving virtual cluster " + clusterId + " to " + file.getAbsolutePath());\r
 268          \r
 269         final ByteFileWriter writer = new ByteFileWriter(file);\r
 270         \r
 271         try {\r
 272 \r
 273             int stms = 0;\r
 274             for(TIntArrayList list : statements) {\r
 275                 if(list != null)\r
 276                     stms += list.size();\r
 277             }\r
 278 \r
 279             writer.write((int)(1.5*stms));\r
 280             @SuppressWarnings("unused")\r
 281             int count = 0;\r
 282             for(int i=0;i<statements.size();i++) {\r
 283                 TIntArrayList list = statements.get(i);\r
 284                 if(list != null) {\r
 285                     for(int j=0;j<list.size();j+=2) {\r
 286                         try {\r
 287                             writer.write((short)i);\r
 288                             int rk = list.getQuick(j);\r
 289                             long rid = ss.getRandomAccessId(rk);\r
 290                             writer.write(rid);\r
 291                             rk = list.getQuick(j+1);\r
 292                             rid = ss.getRandomAccessId(rk);\r
 293                             writer.write(rid);\r
 294                             count++;\r
 295                         } catch (DatabaseException e) {\r
 296                             e.printStackTrace();\r
 297                         }\r
 298                     }\r
 299                 }\r
 300             }\r
 301 \r
 302             writer.write(values.size());\r
 303 \r
 304             values.forEachEntry(new TIntObjectProcedure<byte[]>() {\r
 305 \r
 306                 @Override\r
 307                 public boolean execute(int a, byte[] b) {\r
 308                     writer.write(a);\r
 309                     writer.write(b.length);\r
 310                     writer.write(b);\r
 311                     return true;\r
 312                 }\r
 313             });\r
 314             if (DEBUG)\r
 315                 System.out.println("TransientGraph[" + file.getAbsolutePath() + "] wrote " + count + " statements and " + values.size() + " values to disk.");\r
 316 \r
 317         } finally {\r
 318             writer.commit();\r
 319 //            FileUtils.uncheckedClose(_os);\r
 320         }\r
 321     }\r
 322     \r
 323     public void load(File file, SerialisationSupport serialization, VirtualGraphServerSupport vgss) throws DatabaseException {\r
 324         if (DEBUG)\r
 325             System.out.println("DEBUG: Loading virtual cluster " + clusterId + " from " + file.getAbsolutePath() + " " + file.length());\r
 326         \r
 327         ByteFileReader reader = null;\r
 328         try {\r
 329             \r
 330             if (!file.exists())\r
 331                 return;\r
 332 \r
 333             reader = new ByteFileReader(file);\r
 334             int clusterInt = ClusterTraitsBase.getClusterBits(clusterId()>>1);\r
 335             int stms = reader.readInt();\r
 336             for (int i = 0; i < stms; i += 3) {\r
 337 //                int rId = reader.readShort();\r
 338 //                if(vgss != null) vgss.addVirtual(clusterInt + rId);\r
 339 //                claim(rId,\r
 340 //                        serialization.getTransientId(reader.readLong()),\r
 341 //                        serialization.getTransientId(reader.readLong()));\r
 342                 int rId = reader.readShort();\r
 343                 long sId = reader.readLong();\r
 344                 long oId = reader.readLong();\r
 345                 int sTransientId = serialization.getTransientId(sId);\r
 346                 int oTransientId = serialization.getTransientId(oId);\r
 347                 \r
 348                 if(vgss != null)\r
 349                     vgss.addVirtual(clusterInt + rId);\r
 350                 \r
 351                 claim(rId, sTransientId, oTransientId);\r
 352             }\r
 353 \r
 354             int values = reader.readInt();\r
 355             for (int i = 0; i < values; i++) {\r
 356                 int subject = reader.readInt();\r
 357                 int length = reader.readInt();\r
 358                 setValue(subject, reader.readBytes(length), length);\r
 359             }\r
 360             if (DEBUG)\r
 361                 System.out.println("DEBUG: TransientGraph[" + file.getAbsolutePath() + "] loaded " + stms / 3 + " statements and " + values + " values from disk.");\r
 362             \r
 363         } catch (IOException e) {\r
 364             throw new DatabaseException(e);\r
 365         } finally {\r
 366             if (reader != null)\r
 367                 reader.close();\r
 368         }\r
 369     }\r
 370     \r
 371     void listStatements(SerialisationSupport ss, ArrayList<Statement> result) {\r
 372 \r
 373         int clusterKey = getTransientClusterKey();\r
 374         \r
 375         try {\r
 376             for(int i=0;i<statements.size();i++) {\r
 377                 TIntArrayList list = statements.get(i);\r
 378                 if(list != null) {\r
 379                     Resource subject = ss.getResource(clusterKey | i);\r
 380                     for(int j=0;j<list.size();j+=2) {\r
 381                         Resource p = ss.getResource(list.getQuick(j));\r
 382                         Resource o = ss.getResource(list.getQuick(j+1));\r
 383                         result.add(new StandardStatement(subject, p, o));\r
 384                     }\r
 385 \r
 386                 }\r
 387             }\r
 388         } catch (DatabaseException e) {\r
 389             e.printStackTrace();\r
 390         }\r
 391         \r
 392     }\r
 393 \r
 394     void listValues(final SerialisationSupport ss, final ArrayList<Resource> result) {\r
 395 \r
 396         values.forEachKey(new TIntProcedure() {\r
 397 \r
 398             @Override\r
 399             public boolean execute(int value) {\r
 400                 try {\r
 401                     result.add(ss.getResource(getTransientId(value)));\r
 402                 } catch (DatabaseException e) {\r
 403                     e.printStackTrace();\r
 404                 }\r
 405                 return true;\r
 406             }\r
 407 \r
 408         });\r
 409         \r
 410     }\r
 411 \r
 412 }\r
 413 \r
 414 public class TransientGraph implements VirtualGraphImpl, VirtualGraphContext {\r
 415     private static final boolean DEBUG = VirtualCluster.DEBUG;\r
 416     final private static int SWAP_LIMIT = 30;\r
 417     \r
 418 //    final private static byte[] NO_VALUE = new byte[0];\r
 419     final private static VirtualCluster NO_CLUSTER = new VirtualCluster(-1);\r
 420     \r
 421     final private Persistency persistency;\r
 422     \r
 423     final private SerialisationSupport serialization;\r
 424     final private ResourceSupport resourceSupport;\r
 425     final private VirtualGraphServerSupport virtualGraphServerSupport;\r
 426     final private RequestProcessor sessionRequestProcessor;\r
 427     \r
 428     /*\r
 429      * Cluster array by index.\r
 430      * -NO_CLUSTER value means that there is no such virtual cluster\r
 431      * -null value means that such virtual cluster could be on disk\r
 432      */\r
 433     final private ArrayList<VirtualCluster> clusters = new ArrayList<VirtualCluster>();\r
 434 \r
 435     /*\r
 436      * A list of resident clusters\r
 437      */\r
 438     final private LinkedList<VirtualCluster> memoryClusters = new LinkedList<VirtualCluster>();\r
 439     \r
 440     private final HashSet<VirtualGraphSource> sources = new HashSet<VirtualGraphSource>();\r
 441     \r
 442     final String identifier;\r
 443     final String databaseId;\r
 444 \r
 445     TIntObjectHashMap<TIntHashSet> NO_STATEMENTS = new TIntObjectHashMap<TIntHashSet>();\r
 446 \r
 447     int[] EMPTY = new int[0];\r
 448 \r
 449     public static TransientGraph workspacePersistent(SerialisationSupport ss, VirtualGraphServerSupport vgss, ResourceSupport rs, RequestProcessor srp, String databaseId, String identifier) throws DatabaseException {\r
 450         TransientGraph graph = new TransientGraph(ss, vgss, rs, srp, databaseId, identifier, Persistency.WORKSPACE);\r
 451         graph.load();\r
 452         return graph;\r
 453     }\r
 454     \r
 455     public static TransientGraph memoryPersistent(SerialisationSupport ss, VirtualGraphServerSupport vgss, ResourceSupport rs, RequestProcessor srp, String databaseId, String identifier) {\r
 456         return new TransientGraph(ss, vgss, rs, srp, databaseId, identifier, Persistency.MEMORY);\r
 457     }\r
 458     \r
 459     private TransientGraph(SerialisationSupport ss, VirtualGraphServerSupport vgss, ResourceSupport rs, RequestProcessor srp, String databaseId, String identifier, Persistency persistency) {\r
 460         this.serialization = ss;\r
 461         this.virtualGraphServerSupport = vgss;\r
 462         this.sessionRequestProcessor = srp;\r
 463         this.resourceSupport = rs;\r
 464         this.identifier = identifier;\r
 465         this.databaseId = databaseId;\r
 466         \r
 467         this.persistency = persistency;\r
 468     }\r
 469 \r
 470     public String getIdentifier()  {\r
 471         return identifier;\r
 472     }\r
 473     \r
 474     private int transientClusterId(long id) throws DatabaseException {\r
 475         if (DEBUG)\r
 476             System.out.println("DEBUG: transientClusterId=" + id);\r
 477         if ((id & 1) == 0) // Cluster of virtual subjects\r
 478             return (int)id;\r
 479         // Corresponds to persistent cluster\r
 480         int rk = serialization.getTransientId(id);\r
 481         return ClusterTraitsBase.getClusterKeyFromResourceKey(rk) << 1 | 1;\r
 482     }\r
 483 \r
 484     private String getPrefix() {\r
 485         return identifier + "." + persistency.identifier() + "." + databaseId;\r
 486     }\r
 487     \r
 488     private void load() throws DatabaseException {\r
 489 \r
 490         String prefix = getPrefix();\r
 491         for(String file : virtualGraphServerSupport.storagePath().list()) {\r
 492             try {\r
 493                 if(file.startsWith(prefix)) {\r
 494                     long clusterLong = Long.parseLong(file.substring(prefix.length()+4));\r
 495                     int clusterId = transientClusterId(clusterLong);\r
 496                     VirtualCluster cluster = new VirtualCluster(clusterId);\r
 497                     cluster.load(new File(virtualGraphServerSupport.storagePath(), file), serialization, (clusterId & 1) > 0 ? virtualGraphServerSupport : null);\r
 498                     clusters.ensureCapacity(clusterId+1);\r
 499                     for(int i=clusters.size(); i<clusterId+1; i++) clusters.add(null);\r
 500                     clusters.set(clusterId, cluster);\r
 501                     memoryClusters.addLast(cluster);\r
 502                 }\r
 503             } catch (DatabaseException t) {\r
 504                 // file is assumably broken, delete it\r
 505                 File filee = new File(virtualGraphServerSupport.storagePath(), file);\r
 506                 if (!filee.delete()) {\r
 507                     System.err.println("Could not delete file " + filee.getAbsolutePath());\r
 508                 }\r
 509                 throw t;\r
 510             }\r
 511         }\r
 512 \r
 513     }\r
 514 \r
 515     public void dispose() {\r
 516         try {\r
 517             saveImpl(serialization);\r
 518         } catch (IOException e) {\r
 519             e.printStackTrace();\r
 520         }\r
 521     }\r
 522 \r
 523     public void save() {\r
 524         \r
 525         try {\r
 526             saveImpl(serialization);\r
 527         } catch (IOException e) {\r
 528             e.printStackTrace();\r
 529         }\r
 530         \r
 531     }\r
 532 \r
 533     public void saveImpl(final SerialisationSupport ss) throws IOException {\r
 534         \r
 535         for(VirtualCluster cluster : memoryClusters) {\r
 536                 String prefix = getPrefix();\r
 537             File file = new File(virtualGraphServerSupport.storagePath(), prefix + ".vg." + VirtualCluster.getClusterIdentifier(ss, cluster.clusterId()));\r
 538             cluster.saveImpl(file, ss);\r
 539         }\r
 540         \r
 541     }\r
 542 \r
 543     /**\r
 544      * Closes a stream and ignores any resulting exception. This is useful\r
 545      * when doing stream cleanup in a finally block where secondary exceptions\r
 546      * are not worth logging.\r
 547      */\r
 548     static void uncheckedClose(Closeable closeable) {\r
 549         try {\r
 550             if (closeable != null)\r
 551                 closeable.close();\r
 552         } catch (IOException e) {\r
 553             //ignore\r
 554         }\r
 555     }\r
 556 \r
 557     private void trimClusters() {\r
 558         for(VirtualCluster cluster : memoryClusters) {\r
 559             cluster.trim();\r
 560         }\r
 561     }\r
 562     \r
 563     /*\r
 564      * Returns a transient cluster index\r
 565      * -Transient clusters for persistent resources have index with LSB=1\r
 566      * -Transient clusters for virtual resources have index with LSB=0\r
 567      * \r
 568      * @param subject is a DB client transient id\r
 569      * \r
 570      * For persistent resources cluster id is 2*clusterKey+1\r
 571      * For virtual resources transient ids are persistent and are directly chunked into 14-bit clusters.\r
 572      * \r
 573      */\r
 574     public static int getVirtualClusterKey(int subject) {\r
 575         if (subject > 0) {\r
 576             int ck = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(subject);\r
 577             return (ck << 1) | 1;\r
 578         }\r
 579         int ck = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(~subject);\r
 580         return ck << 1;\r
 581     }\r
 582     \r
 583     private VirtualCluster getOrLoad(int virtualClusterKey) {\r
 584 \r
 585         if(virtualClusterKey < clusters.size()) {\r
 586             VirtualCluster cluster = clusters.get(virtualClusterKey);\r
 587             if(NO_CLUSTER == cluster) return null;\r
 588             if(cluster != null) return cluster;\r
 589         }\r
 590         if (DEBUG)\r
 591             System.out.println("Loading virtual cluster " + virtualClusterKey + " for " + identifier);\r
 592 \r
 593         clusters.ensureCapacity(virtualClusterKey+1);\r
 594         for(int i=clusters.size(); i<virtualClusterKey+1; i++) clusters.add(null);\r
 595 \r
 596         //if(!VirtualCluster.isValidVirtualClusterKey(serialization, virtualClusterKey)) return null;\r
 597 \r
 598         long clusterIdentifier = VirtualCluster.getClusterIdentifier(serialization, virtualClusterKey);\r
 599         File file = new File(virtualGraphServerSupport.storagePath(), getPrefix() + ".vg." + clusterIdentifier);\r
 600         if(file.exists()) {\r
 601             VirtualCluster cluster = new VirtualCluster(virtualClusterKey);\r
 602 //      System.out.println("Loading virtual cluster2 " + virtualClusterKey + " for " + identifier);\r
 603             try {\r
 604                 cluster.load(file, serialization, (virtualClusterKey & 1) > 0 ? virtualGraphServerSupport : null);\r
 605                 clusters.set(virtualClusterKey, cluster);\r
 606                 memoryClusters.addFirst(cluster);\r
 607                 swap();\r
 608                 return cluster;\r
 609             } catch (DatabaseException e) {\r
 610                 e.printStackTrace();\r
 611                 // File must be corrupt, lets delete it so we wont load it next time in future\r
 612                 file.delete();\r
 613                 \r
 614                 clusters.set(virtualClusterKey, NO_CLUSTER);\r
 615                 return null;\r
 616             }\r
 617 \r
 618         } else {\r
 619             clusters.set(virtualClusterKey, NO_CLUSTER);\r
 620             return null;\r
 621         }\r
 622     }\r
 623 \r
 624     private void swap() {\r
 625         \r
 626         trimClusters();\r
 627         while(memoryClusters.size() > SWAP_LIMIT) {\r
 628             VirtualCluster remo = memoryClusters.removeLast();\r
 629             File file = new File(virtualGraphServerSupport.storagePath(), getPrefix() + ".vg." + VirtualCluster.getClusterIdentifier(serialization, remo.clusterId()));\r
 630             try {\r
 631                 remo.saveImpl(file, serialization);\r
 632             } catch (IOException e) {\r
 633                 e.printStackTrace();\r
 634             }\r
 635             clusters.set(remo.clusterId(), null);\r
 636         }\r
 637         \r
 638     }\r
 639     \r
 640     private synchronized VirtualCluster getCluster(int subject, boolean create) {\r
 641 \r
 642         int clusterId = getVirtualClusterKey(subject);\r
 643         \r
 644         VirtualCluster cluster = getOrLoad(clusterId);\r
 645         if(cluster != null) return cluster;\r
 646         \r
 647         if(create) {\r
 648 \r
 649             clusters.ensureCapacity(clusterId+1);\r
 650             for(int i=clusters.size(); i<clusterId+1; i++) clusters.add(null);\r
 651             cluster = new VirtualCluster(clusterId);\r
 652             clusters.set(clusterId, cluster);\r
 653             memoryClusters.addFirst(cluster);\r
 654             swap();\r
 655             return cluster;\r
 656 \r
 657         } else {\r
 658             return null;\r
 659         }\r
 660         \r
 661     }\r
 662     \r
 663     private synchronized void applyValue(int subject, Object value, Binding binding) {\r
 664 \r
 665         try {\r
 666             Serializer serializer = Bindings.getSerializer( binding );\r
 667             byte[] serialized = serializer.serialize(value);\r
 668             VirtualCluster cluster = getCluster(subject, true);\r
 669             cluster.setValue(subject, serialized, serialized.length);\r
 670         } catch (SerializationException e) {\r
 671             e.printStackTrace();\r
 672         } catch (SerializerConstructionException e) {\r
 673             e.printStackTrace();\r
 674         } catch (IOException e) {\r
 675             // TODO Auto-generated catch block\r
 676             e.printStackTrace();\r
 677         }\r
 678 \r
 679     }\r
 680 \r
 681     private synchronized void applyStatements(int subject, int[] statements) {\r
 682 \r
 683         VirtualCluster cluster = getCluster(subject, true);\r
 684         cluster.addStatements(subject, statements);\r
 685         \r
 686         \r
 687         if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
 688 \r
 689     }\r
 690 \r
 691     private synchronized void produceAllStatements(ReadGraphImpl graph, final int subject, final AsyncProcedure<Object> procedure) throws DatabaseException {\r
 692 \r
 693         VirtualCluster cluster = getCluster(subject, true);\r
 694 \r
 695         // This resource becomes a normal resource, all data is requeried\r
 696         cluster.resetLazy(subject);\r
 697         \r
 698         for(VirtualGraphSource source : sources) {\r
 699             source.getStatements(graph, this, subject);\r
 700         }\r
 701         \r
 702         if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
 703         \r
 704     }\r
 705 \r
 706     private synchronized void producePartialStatements(ReadGraphImpl graph, final int subject, final int predicate, final AsyncProcedure<Object> procedure) throws DatabaseException {\r
 707 \r
 708         for(VirtualGraphSource source : sources) {\r
 709             source.getStatements(graph, this, subject, predicate);\r
 710         }\r
 711         \r
 712         if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
 713         \r
 714     }\r
 715 \r
 716     @Override\r
 717     public int getIndex(Resource resource) {\r
 718         try {\r
 719             return serialization.getTransientId(resource);\r
 720         } catch (DatabaseException e) {\r
 721             e.printStackTrace();\r
 722         }\r
 723         return 0;\r
 724     }\r
 725 \r
 726     @Override\r
 727     public Resource getResource(int index) {\r
 728         return new ResourceImpl(resourceSupport, index);\r
 729     }\r
 730 \r
 731     @Override\r
 732     public void register(final VirtualGraphSource source) {\r
 733         if(sources.add(source)) {\r
 734             source.attach(TransientGraph.this);\r
 735         }\r
 736     }\r
 737 \r
 738     @Override\r
 739     public void claim(int subject, int predicate, int object) {\r
 740 \r
 741         VirtualCluster cluster = getCluster(subject, true);\r
 742         cluster.claim(subject, predicate, object);\r
 743         if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
 744 \r
 745     }\r
 746 \r
 747     @Override\r
 748     public synchronized int[] getObjects(int subject, int predicate) {\r
 749         VirtualCluster cluster = getCluster(subject, false);\r
 750         if(cluster == null) return EMPTY;\r
 751         return cluster.getObjects(subject, predicate);\r
 752     }\r
 753 \r
 754     @Override\r
 755     public synchronized int[] getPredicates(int subject) {\r
 756         VirtualCluster cluster = getCluster(subject, false);\r
 757         if(cluster == null) return EMPTY;\r
 758         return cluster.getPredicates(subject);\r
 759     }\r
 760 \r
 761     @Override\r
 762     public synchronized byte[] getValue(int subject) {\r
 763         VirtualCluster cluster = getCluster(subject, false);\r
 764         if(cluster == null) return null;\r
 765         return cluster.getValue(subject);\r
 766     }\r
 767 \r
 768     @Override\r
 769     public int newResource(boolean isLazy) {\r
 770         \r
 771         int id = virtualGraphServerSupport.createVirtual();\r
 772         VirtualCluster cluster = getCluster(id, true);\r
 773         if(isLazy) cluster.setLazy(id);\r
 774         return id;\r
 775 \r
 776     }\r
 777     \r
 778     @Override\r
 779     public void finish(int subject) {\r
 780         VirtualCluster cluster = getCluster(subject, false);\r
 781         cluster.finish(subject);\r
 782     }\r
 783 \r
 784     @Override\r
 785     public void deny(int subject, int predicate, int object) {\r
 786 \r
 787         VirtualCluster cluster = getCluster(subject, true);\r
 788         cluster.deny(subject, predicate, object);\r
 789 \r
 790     }\r
 791 \r
 792     @Override\r
 793     public void claimValue(int subject, byte[] data, int length) {\r
 794 \r
 795         VirtualCluster cluster = getCluster(subject, true);\r
 796         cluster.setValue(subject, data, length);\r
 797         if(subject > 0) virtualGraphServerSupport.addVirtual(subject);\r
 798 \r
 799     }\r
 800 \r
 801     @Override\r
 802     public void denyValue(int subject) {\r
 803         // FIXME: this implementation is probably not proper, Antti needs to work on this.\r
 804         VirtualCluster cluster = getCluster(subject, true);\r
 805         cluster.denyValue(subject);\r
 806         if(subject > 0) virtualGraphServerSupport.removeVirtual(subject);\r
 807     }\r
 808 \r
 809     @Override\r
 810     public void initialise(final Write write) {\r
 811         try {\r
 812             sessionRequestProcessor.syncRequest(new WriteRequest(this) {\r
 813 \r
 814                 @Override\r
 815                 public void perform(WriteGraph graph) throws DatabaseException {\r
 816                     write.perform(graph);\r
 817                 }\r
 818 \r
 819             });\r
 820         } catch (DatabaseException e) {\r
 821             e.printStackTrace();\r
 822         }\r
 823     }\r
 824 \r
 825     @Override\r
 826     public void postModification(AsyncRequestProcessor processor, final WriteOnly request) {\r
 827 \r
 828         if(processor == null) processor = sessionRequestProcessor;\r
 829 \r
 830         processor.asyncRequest(new WriteOnlyRequest(this) {\r
 831 \r
 832             @Override\r
 833             public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 834                 request.perform(graph);\r
 835             }\r
 836 \r
 837         });\r
 838 \r
 839     }\r
 840 \r
 841     @Override\r
 842     public void updateStatements(int resource, int[] statements) {\r
 843         applyStatements(resource, statements);\r
 844     }\r
 845 \r
 846     @Override\r
 847     public void updateValue(int resource, Object value, Binding binding) {\r
 848         applyValue(resource, value, binding);\r
 849     }\r
 850 \r
 851     @Override\r
 852     public boolean isPending(int subject) {\r
 853         \r
 854         VirtualCluster cluster = getCluster(subject, false);\r
 855         if(cluster == null) return false;\r
 856         else return cluster.isPending(subject);\r
 857         \r
 858     }\r
 859 \r
 860     @Override\r
 861     public boolean isPending(int subject, int predicate) {\r
 862 \r
 863         VirtualCluster cluster = getCluster(subject, false);\r
 864         if(cluster == null) return false;\r
 865         else return cluster.isPending(subject, predicate);\r
 866 \r
 867     }\r
 868     \r
 869     @Override\r
 870     public void load(ReadGraphImpl graph, int resource, int predicate, final Callback<ReadGraphImpl> callback) throws DatabaseException {\r
 871         producePartialStatements(graph, resource, predicate, new AsyncProcedure<Object>() {\r
 872 \r
 873             @Override\r
 874             public void execute(AsyncReadGraph graph, Object result) {\r
 875                 callback.run((ReadGraphImpl)graph);\r
 876             }\r
 877 \r
 878             @Override\r
 879             public void exception(AsyncReadGraph graph, Throwable throwable) {\r
 880                 callback.run((ReadGraphImpl)graph);\r
 881             }\r
 882             \r
 883         });\r
 884     }\r
 885 \r
 886     @Override\r
 887     public void load(ReadGraphImpl graph, int resource, final Callback<ReadGraphImpl> callback) throws DatabaseException {\r
 888         produceAllStatements(graph, resource, new AsyncProcedure<Object>() {\r
 889 \r
 890             @Override\r
 891             public void execute(AsyncReadGraph graph, Object result) {\r
 892                 callback.run((ReadGraphImpl)graph);\r
 893             }\r
 894 \r
 895             @Override\r
 896             public void exception(AsyncReadGraph graph, Throwable throwable) {\r
 897                 callback.run((ReadGraphImpl)graph);\r
 898             }\r
 899             \r
 900         });\r
 901     }\r
 902     \r
 903     public Collection<Statement> listStatements() {\r
 904         ArrayList<Statement> result = new ArrayList<Statement>();\r
 905         for(int i=0;i<clusters.size();i++) {\r
 906             VirtualCluster cluster = getOrLoad(i);\r
 907             if(cluster != null) {\r
 908                 cluster.listStatements(serialization, result);\r
 909             }\r
 910         }\r
 911         return result;\r
 912     }\r
 913 \r
 914     public Collection<Resource> listValues() {\r
 915         ArrayList<Resource> result = new ArrayList<Resource>();\r
 916         for(int i=0;i<clusters.size();i++) {\r
 917             VirtualCluster cluster = getOrLoad(i);\r
 918             if(cluster != null) {\r
 919                 cluster.listValues(serialization, result);\r
 920             }\r
 921         }\r
 922         return result;\r
 923     }\r
 924 \r
 925     @Override\r
 926     public Persistency getPersistency() {\r
 927         return persistency;\r
 928     }\r
 929     \r
 930     @Override\r
 931     public String toString() {\r
 932         String result = "'" + identifier + "'";\r
 933         if(Persistency.WORKSPACE == persistency) result += " (W)";\r
 934         else if(Persistency.MEMORY == persistency) result += " (M)";\r
 935         return result;\r
 936     }\r
 937 \r
 938 }\r
 939 \r