bundles/org.simantics.db.procore/src/fi/vtt/simantics/procore/internal/ClusterTable.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   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
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package fi.vtt.simantics.procore.internal;
  13
  14 import java.io.File;
  15 import java.util.ArrayList;
  16 import java.util.Collection;
  17 import java.util.HashMap;
  18 import java.util.Map;
  19 import java.util.TreeMap;
  20 import java.util.concurrent.Semaphore;
  21 import java.util.concurrent.atomic.AtomicLong;
  22 import java.util.function.Consumer;
  23
  24 import org.simantics.databoard.Bindings;
  25 import org.simantics.db.ClusterCreator;
  26 import org.simantics.db.Database;
  27 import org.simantics.db.Database.Session.ClusterChanges;
  28 import org.simantics.db.DevelopmentKeys;
  29 import org.simantics.db.SessionVariables;
  30 import org.simantics.db.common.utils.Logger;
  31 import org.simantics.db.exception.ClusterDoesNotExistException;
  32 import org.simantics.db.exception.DatabaseException;
  33 import org.simantics.db.exception.ResourceNotFoundException;
  34 import org.simantics.db.exception.RuntimeDatabaseException;
  35 import org.simantics.db.impl.ClusterBase;
  36 import org.simantics.db.impl.ClusterI;
  37 import org.simantics.db.impl.ClusterSupport;
  38 import org.simantics.db.impl.ClusterTraitsBase;
  39 import org.simantics.db.impl.IClusterTable;
  40 import org.simantics.db.impl.graph.WriteGraphImpl;
  41 import org.simantics.db.impl.query.QueryProcessor;
  42 import org.simantics.db.procore.cluster.ClusterBig;
  43 import org.simantics.db.procore.cluster.ClusterImpl;
  44 import org.simantics.db.procore.cluster.ClusterSmall;
  45 import org.simantics.db.procore.cluster.ClusterTraits;
  46 import org.simantics.db.procore.protocol.Constants;
  47 import org.simantics.db.service.ClusterCollectorPolicy;
  48 import org.simantics.db.service.ClusterCollectorPolicy.CollectorCluster;
  49 import org.simantics.db.service.ClusterUID;
  50 import org.simantics.utils.Development;
  51
  52 import fi.vtt.simantics.procore.DebugPolicy;
  53 import fi.vtt.simantics.procore.internal.ClusterControlImpl.ClusterStateImpl;
  54 import fi.vtt.simantics.procore.internal.SessionImplSocket.TaskHelper;
  55 import gnu.trove.map.hash.TLongIntHashMap;
  56 import gnu.trove.map.hash.TLongObjectHashMap;
  57 import gnu.trove.procedure.TIntProcedure;
  58 import gnu.trove.procedure.TLongObjectProcedure;
  59 import gnu.trove.procedure.TObjectProcedure;
  60 import gnu.trove.set.hash.TIntHashSet;
  61
  62 public final class ClusterTable implements IClusterTable {
  63
  64     private static final boolean VALIDATE_SIZE = false;
  65
  66     int maximumBytes = 128 * 1024 * 1024;
  67     int limit = (int)(0.8*(double)maximumBytes);
  68
  69     private final AtomicLong timeCounter = new AtomicLong(1);
  70
  71     final private SessionImplSocket sessionImpl;
  72     final private ArrayList<ClusterI> writeOnlyClusters = new ArrayList<ClusterI>();
  73     final private TIntHashSet writeOnlyInvalidates = new TIntHashSet();
  74     final private static int ARRAY_SIZE = ClusterTraits.getClusterArraySize();
  75     final private ClusterImpl[] clusterArray = new ClusterImpl[ARRAY_SIZE];
  76     final private Boolean[] immutables = new Boolean[ARRAY_SIZE];
  77     final private boolean[] virtuals = new boolean[ARRAY_SIZE];
  78     static class ImportanceEntry implements CollectorCluster {
  79         final public long importance;
  80         final public long clusterId;
  81         public ImportanceEntry(ClusterImpl impl) {
  82             this(impl.getImportance(), impl.getClusterId());
  83         }
  84
  85         public ImportanceEntry(long importance, long clusterId) {
  86             this.importance = importance;
  87             this.clusterId = clusterId;
  88         }
  89         @Override
  90         public long getImportance() {
  91             return importance;
  92         }
  93         @Override
  94         public long getClusterId() {
  95             return clusterId;
  96         }
  97         @Override
  98         public String toString() {
  99             return "CID " + clusterId;
 100         }
 101     }
 102     private class Clusters {
 103         // This makes sure that non-null values from hashMap.get can be trusted (no unsynchronized rehashes occur)
 104         final public TLongObjectHashMap<ClusterImpl> hashMap = new TLongObjectHashMap<ClusterImpl>(2 * ARRAY_SIZE);
 105         //private final HashMap<ClusterUID, ClusterImpl> clusterU2I = new HashMap<ClusterUID, ClusterImpl>(); // Maps cluster UID to cluster.
 106         private Clusters() {
 107             clear();
 108         }
 109         private void clear() {
 110             hashMap.clear();
 111           //  clusterU2I.clear();
 112             hashMap.put(0, null); // reserved for null value
 113           //  clusterU2I.put(ClusterUID.make(0, 0), null);
 114         }
 115         private int size() {
 116             return hashMap.size();
 117         }
 118         private ClusterImpl getClusterByClusterId(long clusterId) {
 119             return hashMap.get(clusterId);
 120         }
 121         private ClusterImpl getClusterByClusterUID(ClusterUID clusterUID) {
 122                 return getClusterByClusterId(clusterUID.second);
 123             //return clusterU2I.get(clusterUID);
 124         }
 125         private ClusterImpl getClusterByClusterUID(long id1, long id2) {
 126                 return getClusterByClusterId(id2);
 127         }
 128         private ClusterImpl makeProxy(long clusterId) {
 129             return makeProxy(ClusterUID.make(0, clusterId));
 130         }
 131         private ClusterImpl makeProxy(ClusterUID clusterUID) {
 132             ClusterImpl proxy = hashMap.get(clusterUID.second);
 133             if (null != proxy)
 134                 return proxy;
 135             int clusterKey = hashMap.size();
 136             ClusterSmall sentinel = new ClusterSmall(clusterUID, clusterKey, ClusterTable.this, sessionImpl.clusterTranslator);
 137             if (sentinel.clusterId != sentinel.clusterUID.second)
 138                 throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 139             create(sentinel);
 140             return sentinel;
 141         }
 142         private void replace(ClusterImpl proxy) {
 143             ClusterImpl old = clusterArray[proxy.clusterKey];
 144             create(proxy);
 145             if (null != old) {
 146                 if (old.clusterKey != proxy.clusterKey)
 147                     throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 148                 if (old.clusterId != proxy.clusterId)
 149                     throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 150                 if (!old.clusterUID.equals(proxy.clusterUID))
 151                     throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 152             }
 153         }
 154         private ClusterSmall freeProxy(ClusterImpl proxy) {
 155             ClusterImpl clusterImpl = hashMap.get(proxy.clusterId);
 156             if (null == clusterImpl)
 157                 throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 158 //            ClusterUID clusterUID = ClusterUID.make(0, proxy.clusterId);
 159 //            ClusterImpl clusterImpl2 = clusterU2I.get(clusterUID );
 160 //            if (clusterImpl != clusterImpl2)
 161 //                throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 162             if (proxy.clusterId != clusterImpl.clusterId)
 163                 throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 164             if (proxy.clusterKey != clusterImpl.clusterKey)
 165                 throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 166             ClusterSmall sentinel = new ClusterSmall(makeClusterUID(proxy.clusterId) , proxy.clusterKey, ClusterTable.this, sessionImpl.clusterTranslator);
 167             return (ClusterSmall)create(sentinel);
 168         }
 169         private ClusterImpl create(ClusterImpl clusterImpl) {
 170             hashMap.put(clusterImpl.clusterId, clusterImpl);
 171 //            clusterU2I.put(clusterImpl.clusterUID, clusterImpl);
 172             clusterArray[clusterImpl.clusterKey] = clusterImpl;
 173             return clusterImpl;
 174         }
 175         private void forEachValue(TObjectProcedure<ClusterImpl> procedure) {
 176             hashMap.forEachValue(procedure);
 177         }
 178         private void forEachEntry(TLongObjectProcedure<ClusterImpl> procedure) {
 179             hashMap.forEachEntry(procedure);
 180         }
 181         private ClusterUID makeClusterUID(long clusterId) {
 182             return ClusterUID.make(0, clusterId);
 183         }
 184         private void removeProxy(ClusterImpl proxy) {
 185             hashMap.remove(proxy.clusterId);
 186             clusterArray[proxy.clusterKey] = null;
 187         }
 188     }
 189     final public TreeMap<Long, CollectorCluster> importanceMap = new TreeMap<Long, CollectorCluster>();
 190
 191     private ClusterCollectorPolicy collectorPolicy;
 192     private boolean dirtySizeInBytes = true;
 193     private long sizeInBytes = 0;
 194     private final Clusters clusters;
 195     ClusterUID makeClusterUID(long clusterId) {
 196         return clusters.makeClusterUID(clusterId);
 197     }
 198     public ClusterUID getClusterUIDByResourceKey(int resourceKey) throws DatabaseException {
 199         int clusterKey = ClusterTraits.getClusterKeyFromResourceKey(resourceKey);
 200         return clusterArray[clusterKey].clusterUID;
 201     }
 202     ClusterTable(SessionImplSocket sessionImpl, File folderPath) {
 203         this.sessionImpl = sessionImpl;
 204         clusters = new Clusters();
 205     }
 206
 207     void dispose() {
 208         clusters.clear();
 209         importanceMap.clear();
 210     }
 211
 212     public void setCollectorPolicy(ClusterCollectorPolicy policy) {
 213         this.collectorPolicy = policy;
 214     }
 215
 216     /**
 217      * Sets Cluster Table allocation size by percentage of maximum usable
 218      * memory. Allocation is limited between 5% and 50% of maximum memory.
 219      *
 220      * @param precentage
 221      */
 222     private void setAllocateSize(double percentage) {
 223
 224         percentage = Math.max(percentage, 0.05);
 225         percentage = Math.min(percentage, 0.5);
 226
 227         maximumBytes = (int) Math.floor(percentage * Runtime.getRuntime().maxMemory());
 228     }
 229
 230     private double estimateProperAllocation(ClusterCollectorSupport support) {
 231         long max = Runtime.getRuntime().maxMemory();
 232         long free = Runtime.getRuntime().freeMemory();
 233         long total = Runtime.getRuntime().totalMemory();
 234         long realFree = max - total + free; // amount of free memory
 235         long current = support.getCurrentSize(); // currently allocated cache
 236         long inUseTotal = max - realFree; // currently used memory
 237         long inUseOther = inUseTotal - current; // memory that the application uses (without the cache)
 238         double otherUsePercentage = (double) inUseOther / (double) max; // percentage of memory that the application uses
 239         double aimFree = 0.2; // percentage of maximum heap that we try to keep free
 240         double estimate = 1.0 - otherUsePercentage - aimFree;
 241         estimate = Math.min(estimate, 0.5);
 242         estimate = Math.max(estimate, 0.05);
 243         // System.out.println("Estimated allocation percentage " + estimate +
 244 // " memory stats: max: " + max + ", free: " + realFree + ", inUse: " +
 245 // inUseTotal + ", cached: " + current + ", cacheSize: " + maximumBytes);
 246         return estimate;
 247     }
 248
 249     void checkCollect() {
 250         collector.collect();
 251     }
 252
 253     synchronized ClusterImpl getClusterByClusterId(long clusterId) {
 254         return clusters.getClusterByClusterId(clusterId);
 255     }
 256
 257     ClusterBase getClusterByClusterKey(int clusterKey) {
 258         return clusterArray[clusterKey];
 259     }
 260
 261     synchronized ClusterImpl makeProxy(ClusterUID clusterUID, long clusterId) {
 262         if (clusterUID.second != clusterId)
 263             throw new RuntimeDatabaseException("Illegal id for cluster=" + clusterUID + " id=" + clusterId);
 264         return clusters.makeProxy(clusterUID);
 265     }
 266
 267     synchronized ClusterImpl makeCluster(long clusterId, boolean writeOnly) {
 268         checkCollect();
 269         ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
 270         if (null != proxy)
 271             return proxy;
 272         ClusterUID clusterUID = ClusterUID.make(0, clusterId);
 273         int clusterKey = clusters.size(); // new key
 274         if (writeOnly) {
 275             proxy = new ClusterWriteOnly(clusterUID, clusterKey, sessionImpl);
 276             writeOnlyClusters.add(proxy);
 277             return clusters.create(proxy);
 278         } else {
 279             //printMaps("makeCluster");
 280             ClusterImpl cluster = ClusterImpl.make(clusterUID, clusterKey, sessionImpl.clusterTranslator);
 281             clusters.create(cluster);
 282             if (!cluster.isLoaded())
 283                 Logger.defaultLogError(new Exception("Bug in ClusterTable.makeCluster(long, boolean), cluster not loaded"));
 284             importanceMap.put(cluster.getImportance(), new ImportanceEntry(cluster));
 285             if (VALIDATE_SIZE)
 286                 validateSize("makeCluster");
 287             if(collectorPolicy != null) collectorPolicy.added(cluster);
 288             return cluster;
 289         }
 290     }
 291
 292     synchronized void replaceCluster(ClusterI cluster_) {
 293         ClusterImpl cluster = (ClusterImpl) cluster_;
 294         checkCollect();
 295         int clusterKey = cluster.getClusterKey();
 296         ClusterImpl existing = (ClusterImpl) clusterArray[clusterKey];
 297         if (existing.hasVirtual())
 298             cluster.markVirtual();
 299
 300         if (existing.cc != null) {
 301             if (existing.isLoaded()) {
 302                 // This shall be promoted to actual exception in the future -
 303                 // for now, minimal changes
 304                 new Exception("Trying to replace cluster with pending changes " + existing.getClusterUID())
 305                         .printStackTrace();
 306             } else {
 307                 // Adopt changes to loaded cluster
 308                 cluster.cc = existing.cc;
 309                 cluster.cc.adopt(cluster);
 310                 cluster.foreignLookup = existing.foreignLookup;
 311                 cluster.change = existing.change;
 312             }
 313         }
 314
 315         importanceMap.remove(existing.getImportance());
 316         if (collectorPolicy != null)
 317             collectorPolicy.removed((ClusterImpl)existing);
 318
 319         //System.out.println("ClusterTable.replace(" + existing + " (I=" + existing.getImportance() + ") => " + cluster + " (I=" + cluster.getImportance() + ")");
 320
 321         clusters.replace((ClusterImpl)cluster);
 322         if (!cluster.isLoaded())
 323             Logger.defaultLogError(new Exception("Bug in ClusterTable.replaceCluster(ClusterI), cluster not loaded"));
 324
 325         importanceMap.put(cluster.getImportance(), new ImportanceEntry((ClusterImpl)cluster));
 326         if(collectorPolicy != null) collectorPolicy.added((ClusterImpl)cluster);
 327
 328         if (!dirtySizeInBytes) {
 329             if (existing != cluster) {
 330                 adjustCachedSize(-existing.getCachedSize(), existing);
 331             }
 332             // This will update sizeInBytes through adjustCachedSize
 333             cluster.getCachedSize();
 334         }
 335         if (VALIDATE_SIZE)
 336             validateSize("replaceCluster");
 337     }
 338
 339     synchronized void release(CollectorCluster cluster) {
 340         importanceMap.remove(cluster.getImportance());
 341         release(cluster.getClusterId());
 342     }
 343
 344     synchronized void release(long clusterId) {
 345         //System.out.println("ClusterTable.release(" + clusterId + "): " + sizeInBytes);
 346         //validateSize();
 347         ClusterImpl clusterImpl = clusters.getClusterByClusterId(clusterId);
 348         if (null == clusterImpl)
 349             return;
 350         if(!clusterImpl.isLoaded() || clusterImpl.isEmpty())
 351             return;
 352         //printMaps("release");
 353         clusters.freeProxy(clusterImpl);
 354         importanceMap.remove(clusterImpl.getImportance());
 355         if (collectorPolicy != null)
 356             collectorPolicy.removed(clusterImpl);
 357         if (sessionImpl.writeState != null)
 358             sessionImpl.clusterStream.flush(clusterImpl.clusterUID);
 359         if (!dirtySizeInBytes) {
 360             adjustCachedSize(-clusterImpl.getCachedSize(), clusterImpl);
 361         }
 362         if (VALIDATE_SIZE)
 363             validateSize("release");
 364     }
 365
 366     synchronized void compact(long id) {
 367         ClusterI impl = clusters.getClusterByClusterId(id);
 368         if (impl != null)
 369             impl.compact();
 370     }
 371
 372     void updateSize() {
 373 // cachedSize = getSizeInBytes();
 374     }
 375
 376     double getLoadProbability() {
 377         // This can currently cause stack overflow
 378         return 1.0;
 379 // if(cachedSize < SLOW_LIMIT) return 1.0;
 380 // if(cachedSize > HIGH_LIMIT) return 1e-2;
 381 //
 382 // double pos = (double)(cachedSize - SLOW_LIMIT) / (double)(HIGH_LIMIT -
 383 // SLOW_LIMIT);
 384 // double val = 0.1 * ((pos-1) * (pos-1)) + 1e-2;
 385 // return val;
 386     }
 387
 388     class SizeProcedure implements TObjectProcedure<ClusterImpl> {
 389         public long result = 0;
 390
 391         @Override
 392         public boolean execute(ClusterImpl cluster) {
 393             if (cluster != null) {
 394                 try {
 395                     if (cluster.isLoaded() && !cluster.isEmpty()) {
 396                         result += cluster.getCachedSize();
 397                     }
 398                 } catch (Throwable t) {
 399                     Logger.defaultLogError(t);
 400                 }
 401             }
 402             return true;
 403         }
 404
 405         public void clear() {
 406             result = 0;
 407         }
 408
 409     };
 410
 411     private SizeProcedure sizeProcedure = new SizeProcedure();
 412     long getSizeInBytes() {
 413         if (dirtySizeInBytes) {
 414             sizeProcedure.clear();
 415             clusters.forEachValue(sizeProcedure);
 416             sizeInBytes = sizeProcedure.result;
 417             // System.err.println("recomputed size of clusterTable => " + sizeInBytes);
 418             setDirtySizeInBytes(false);
 419         }
 420         return sizeInBytes;
 421     }
 422
 423     public void setDirtySizeInBytes(boolean value) {
 424         dirtySizeInBytes = value;
 425     }
 426
 427     ClusterStateImpl getState() {
 428         final ClusterStateImpl result = new ClusterStateImpl();
 429         clusters.forEachEntry(new TLongObjectProcedure<ClusterImpl>() {
 430             @Override
 431             public boolean execute(long arg0, ClusterImpl arg1) {
 432                 if (arg1 == null)
 433                     return true;
 434                 if (arg1.isLoaded() && !arg1.isEmpty()) {
 435                     result.ids.add(new ImportanceEntry(arg1));
 436                 }
 437                 return true;
 438             }
 439         });
 440         return result;
 441     }
 442
 443     void restoreState(ClusterStateImpl state) {
 444         ClusterStateImpl current = getState();
 445         for (CollectorCluster id : current.ids)
 446             if (!state.ids.contains(id))
 447                 collectorSupport.release(id);
 448     }
 449
 450     class ClusterCollectorImpl implements ClusterCollector {
 451
 452         final private ClusterCollectorSupport support;
 453         private ClusterCollectorPolicy policy;
 454         ClusterCollectorImpl(ClusterCollectorSupport support) {
 455             this.support = support;
 456         }
 457
 458         ClusterCollectorPolicy setPolicy(ClusterCollectorPolicy newPolicy) {
 459
 460             ClusterCollectorPolicy oldPolicy = policy;
 461             policy = newPolicy;
 462
 463             if(policy != null) {
 464                 for (CollectorCluster id : support.getResidentClusters()) {
 465                     policy.added(getClusterByClusterId(id.getClusterId()));
 466                 }
 467             }
 468
 469             support.setPolicy(policy);
 470
 471             return oldPolicy;
 472
 473         }
 474
 475         @Override
 476         public void collect() {
 477
 478             if(policy != null) {
 479
 480                 release(policy.select());
 481
 482             } else {
 483
 484                 int size = support.getCurrentSize();
 485                 boolean dynamicAllocation = useDynamicAllocation();
 486                 if (dynamicAllocation)
 487                     setAllocateSize(estimateProperAllocation(support));
 488                 if (DebugPolicy.CLUSTER_COLLECTION) {
 489                     System.out.println("Cluster collector activated, current size = " + size + " limit = " + maximumBytes);
 490                 }
 491                 if (dynamicAllocation) {
 492                     int collectSize = maximumBytes / 2;
 493                     collectSize = Math.min(collectSize, 32 * 1024 * 1024);
 494                     // try to keep allocated clusters below the maximum
 495                     if (maximumBytes - size > collectSize)
 496                         return;
 497                     collectSize += size - maximumBytes;
 498                     collect(collectSize);
 499                 } else {
 500                     // try to keep allocated clusters below the maximum
 501                     if (size < maximumBytes)
 502                         return;
 503                     // shave off 20%
 504                     collect(size-limit);
 505                 }
 506
 507             }
 508
 509         }
 510
 511         private boolean useDynamicAllocation() {
 512             return "true".equalsIgnoreCase(System.getProperty("org.simantics.db.cluster.dynamicAlloc"));
 513         }
 514
 515         @Override
 516         public void collect(int target) {
 517
 518             if(policy != null) {
 519
 520                 release(policy.select(target));
 521
 522             } else {
 523
 524                 ArrayList<CollectorCluster> toRelease = new ArrayList<CollectorCluster>();
 525
 526                 for (CollectorCluster cluster : support.getResidentClusters()) {
 527                     toRelease.add(cluster);
 528                     long clusterSize = support.getClusterSize(cluster);
 529                     //System.err.println("release cluster with " + (clusterSize/1024) + " kiB - " + cluster);
 530                     target -= clusterSize;
 531                     if (target <= 0) {
 532                         break;
 533                     }
 534                 }
 535
 536                 release(toRelease);
 537
 538                 if (DebugPolicy.CLUSTER_COLLECTION) {
 539                     System.out.println("Cluster collector finished, current size = " + support.getCurrentSize());
 540                 }
 541
 542             }
 543
 544         }
 545
 546         void release(Collection<CollectorCluster> toRelease) {
 547             for (CollectorCluster id : toRelease) {
 548                 support.release(id);
 549             }
 550         }
 551
 552     }
 553
 554     private ClusterCollectorSupport collectorSupport = new ClusterCollectorSupportImpl(this);
 555     ClusterCollectorImpl collector = new ClusterCollectorImpl(collectorSupport);
 556
 557     void gc() {
 558         collector.collect();
 559     }
 560
 561     private long newResourceClusterId = Constants.NewClusterId;
 562     public static final int CLUSTER_FILL_SIZE = ClusterTraitsBase.getMaxNumberOfResources();
 563
 564     /*
 565      * Uusi id varataan vasta, kun lisï¿½tï¿½ï¿½n resurssi => reservedIds sisï¿½ltï¿½ï¿½
 566      * vain jo kï¿½ytï¿½ssï¿½ olevia klustereita
 567      */
 568
 569     ClusterImpl getNewResourceCluster(ClusterSupport cs, GraphSession graphSession, boolean writeOnly)
 570     throws DatabaseException {
 571         ClusterImpl result = null;
 572         if (Constants.NewClusterId == newResourceClusterId) {
 573             newResourceClusterId = graphSession.newClusterId();
 574             result = getClusterByClusterIdOrMake(newResourceClusterId, writeOnly);
 575         } else {
 576             ClusterImpl cluster = getClusterByClusterIdOrThrow(newResourceClusterId);
 577             if (cluster.getNumberOfResources(cs) >= CLUSTER_FILL_SIZE) {
 578                 newResourceClusterId = graphSession.newClusterId();
 579                 cluster = getClusterByClusterIdOrMake(newResourceClusterId, writeOnly);
 580             }
 581             result = cluster;
 582         }
 583         return ensureLoaded(result);
 584     }
 585
 586     void flushCluster(GraphSession graphSession) {
 587 // We seem to disagree about this.
 588 // graphSession.newClusterId();
 589         newResourceClusterId = Constants.NewClusterId;
 590     }
 591
 592     void writeOnlyInvalidate(ClusterI impl) {
 593         writeOnlyInvalidates.add(impl.getClusterKey());
 594     }
 595
 596     void removeWriteOnlyClusters() {
 597         for (ClusterI proxy : writeOnlyClusters) {
 598             if (!(proxy instanceof ClusterImpl))
 599                 throw new RuntimeDatabaseException("ClusterTable corrupted. Contact application support.");
 600             clusters.freeProxy((ClusterImpl)proxy);
 601         }
 602         writeOnlyClusters.clear();
 603         writeOnlyInvalidates.forEach(new TIntProcedure() {
 604             @Override
 605             public boolean execute(int clusterKey) {
 606                 ClusterImpl proxy = clusterArray[clusterKey];
 607                 ClusterUID clusterUID = proxy.getClusterUID();
 608                 clusters.freeProxy(proxy);
 609                 return true;
 610             }
 611         });
 612         writeOnlyInvalidates.clear();
 613     }
 614
 615     public ClusterImpl getClusterByClusterUID(ClusterUID clusterUID) {
 616         synchronized (this) {
 617             return clusters.getClusterByClusterUID(clusterUID);
 618         }
 619     }
 620     public int getClusterKeyByClusterUIDOrMakeProxy(ClusterUID clusterUID) {
 621         return getClusterKeyByClusterUIDOrMakeProxy(0/*clusterUID.first*/, clusterUID.second);
 622     }
 623     public int getClusterKeyByClusterUIDOrMakeProxy(long id1, long id2) {
 624         return getClusterByClusterUIDOrMakeProxy(id1, id2).clusterKey;
 625     }
 626     public ClusterImpl getClusterByClusterUIDOrMakeProxy(ClusterUID clusterUID) {
 627         synchronized (this) {
 628             ClusterImpl clusterImpl = clusters.getClusterByClusterUID(clusterUID);
 629             if (null == clusterImpl)
 630                 clusterImpl = clusters.makeProxy(clusterUID);
 631             return clusterImpl;
 632         }
 633     }
 634     public ClusterImpl getClusterByClusterUIDOrMakeProxy(long id1, long id2) {
 635         synchronized (this) {
 636             ClusterImpl clusterImpl = clusters.getClusterByClusterUID(id1, id2);
 637             if (null == clusterImpl)
 638                 clusterImpl = clusters.makeProxy(id2);
 639             return clusterImpl;
 640         }
 641     }
 642     ClusterImpl getLoadOrThrow(long clusterId) throws DatabaseException {
 643         synchronized (this) {
 644             ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
 645             int clusterKey = 0;
 646             if (proxy != null)
 647                 if (proxy.isLoaded())
 648                     return proxy;
 649                 else
 650                     clusterKey = proxy.getClusterKey();
 651             try {
 652                 if (clusterKey == 0) {
 653                     proxy = clusters.makeProxy(clusterId);
 654                     clusterKey = proxy.getClusterKey();
 655                 }
 656                 ClusterImpl ci = tryLoad(clusterId, clusterKey);
 657                 if (null == ci)
 658                     throw new ResourceNotFoundException(clusterId);
 659                 return ci;
 660             } catch (ClusterDoesNotExistException t) {
 661                 clusters.removeProxy(proxy);
 662                 throw t;
 663             }
 664         }
 665     }
 666
 667     ClusterImpl getClusterByClusterIdOrMake(long clusterId, boolean writeOnly) {
 668         synchronized (this) {
 669             ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
 670             if (proxy != null)
 671                 return proxy;
 672             else
 673                 return makeCluster(clusterId, writeOnly);
 674         }
 675     }
 676     ClusterImpl getClusterByClusterIdOrThrow(long clusterId) {
 677         synchronized (this) {
 678             ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
 679             if (null == proxy)
 680                 throw new IllegalArgumentException("Cluster id=" + clusterId + " is not created.");
 681             return proxy;
 682         }
 683     }
 684     long getClusterIdOrCreate(ClusterUID clusterUID) {
 685         return clusterUID.second;
 686     }
 687     final long getClusterIdByResourceKey(final int resourceKey)
 688     throws DatabaseException {
 689         int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKey(resourceKey);
 690         if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
 691             throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
 692         ClusterI c = clusterArray[clusterKey];
 693         if (c == null)
 694             throw new RuntimeException("No cluster for key " + resourceKey);
 695         return c.getClusterId();
 696     }
 697     final long getClusterIdByResourceKeyNoThrow(final int resourceKey) {
 698         int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
 699         if (ClusterTraitsBase.isVirtualClusterKey(clusterKey)) {
 700             Logger.defaultLogError("Tried to get a persistent cluster for a virtual resource. key=" + resourceKey);
 701             return 0;
 702         }
 703         ClusterI c = clusterArray[clusterKey];
 704         if (c == null) {
 705             Logger.defaultLogError("No cluster for key " + resourceKey);
 706             return 0;
 707         }
 708         return c.getClusterId();
 709     }
 710
 711     int counter = 0;
 712
 713     void refresh(long csid, SessionImplSocket session, ClusterUID[] clusterUID) {
 714         synchronized (this) {
 715             session.flushCounter = 0;
 716             session.clusterStream.reallyFlush();
 717             ClientChangesImpl cs = new ClientChangesImpl(session);
 718             if (session.clientChanges == null)
 719                 session.clientChanges = cs;
 720             //printMaps("refresh");
 721             for (int i=0; i<clusterUID.length; ++i) {
 722                 try {
 723                     if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 724                         System.err.println("cluster=" + clusterUID[i] + " has changed. length=" + clusterUID.length);
 725                     ClusterImpl oldCluster = clusters.getClusterByClusterUID(clusterUID[i]);
 726                     if (null == oldCluster)
 727                         continue;
 728                     if (!oldCluster.isLoaded())
 729                         continue;
 730                     int clusterKey = oldCluster.getClusterKey();
 731                     if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
 732                         continue;
 733                     boolean big = oldCluster instanceof ClusterBig;
 734                     boolean small = oldCluster instanceof ClusterSmall;
 735                     if (!big && !small)
 736                         continue;
 737 //                    ClusterImpl newCluster = (ClusterImpl) sessionImpl.graphSession.getClusterImpl(clusterUID[i], clusterKey);
 738                     Database.Session dbSession = sessionImpl.graphSession.dbSession;
 739
 740                     ClusterImpl newCluster = dbSession.clone(clusterUID[i], new ClusterCreator() {
 741
 742                         @Override
 743                         public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
 744                             ClusterSupport support = sessionImpl.clusterTranslator;
 745                             try {
 746                                 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
 747                             } catch (DatabaseException e) {
 748                                 e.printStackTrace();
 749                                 return null;
 750                             }
 751                         }
 752
 753                     });
 754                     if (null == newCluster)
 755                         continue;
 756                     if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 757                         System.err.println("cluster=" + newCluster + "  updated.");
 758                     importanceMap.remove(oldCluster.getImportance());
 759                     if (collectorPolicy != null)
 760                         collectorPolicy.removed(oldCluster);
 761                     clusters.replace(newCluster);
 762                     if (!newCluster.isLoaded())
 763                         Logger.defaultLogError(new Exception("Bug in ClusterTable.refresh, cluster not loaded"));
 764                     importanceMap.put(newCluster.getImportance(), new ImportanceEntry(newCluster));
 765                     if (collectorPolicy != null)
 766                         collectorPolicy.added(newCluster);
 767                     // Now we have fetched the new cluster but to emulate effects of the changes in it we fetch the cluster changes from server.
 768                     refreshCluster(csid, session, oldCluster, clusterUID[i], newCluster.clusterId, clusterKey);
 769                 } catch (Throwable t) {
 770                     Logger.defaultLogError("Failed to load cluster in refresh.", t);
 771                 }
 772             }
 773             if (VALIDATE_SIZE)
 774                 validateSize("refresh");
 775             // Fake update of cluster changes.
 776             QueryProcessor queryProcessor = session.getQueryProvider2();
 777             WriteGraphImpl writer = WriteGraphImpl.create(queryProcessor, session.writeSupport, null);
 778             TaskHelper th = null;
 779             if (null == session.writeState) {
 780                 th = new TaskHelper("Refresh");
 781                 session.writeState = new WriteState<Object>(writer, th.writeTraits, th.sema, th.proc);
 782                 try {
 783                     session.getQueryProvider2().performDirtyUpdates(writer);
 784                     session.fireMetadataListeners(writer, cs);
 785                     session.getQueryProvider2().performScheduledUpdates(writer);
 786                     session.fireReactionsToSynchronize(cs);
 787                     session.fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
 788                     session.printDiagnostics();
 789                 } finally {
 790                     if (null != th)
 791                         session.writeState = null;
 792                     cs.dispose();
 793                 }
 794             }
 795         }
 796     }
 797     final void refreshCluster(long csid, SessionImplSocket session, ClusterImpl cluster, ClusterUID clusterUID, long clusterId, int clusterKey)
 798     throws DatabaseException {
 799         if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 800             System.err.println("cluster=" + clusterUID + " id=" + clusterId + " key=" + clusterKey + " resources will be updated.");
 801         // get cluster change sets
 802         QueryProcessor queryProcessor = session.getQueryProvider2();
 803         ClusterChanges cc;
 804         try {
 805             cc = session.graphSession.getClusterChanges(clusterUID, csid);
 806         } catch (Exception e) {
 807             Logger.defaultLogError("Could not get cluster changes. cluster=" + clusterUID, e);
 808             release(clusterId);
 809             return;
 810         }
 811         for (int i=0; i<cc.getResourceIndex().length; ++i) {
 812             int resource = ClusterTraits.createResourceKey(clusterKey, cc.getResourceIndex()[i]);
 813             ClusterUID pClusterUID = new ClusterUID(cc.getPredicateFirst()[i], cc.getPredicateSecond()[i]);
 814             ClusterImpl pCluster = clusters.getClusterByClusterUID(pClusterUID);
 815             if (null == pCluster)
 816                 continue;
 817             int pClusterKey = pCluster.getClusterKey();
 818             int predicate = ClusterTraits.createResourceKey(pClusterKey, cc.getPredicateIndex()[i]);
 819             queryProcessor.updateStatements(resource, predicate);
 820             if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 821                 System.err.println("resource " + cc.getResourceIndex()[i] + " relation " + cc.getPredicateIndex()[i] + " changed.");
 822         }
 823         for (int i=0; i<cc.getValueIndex().length; ++i) {
 824             int resource = ClusterTraits.createResourceKey(clusterKey, cc.getValueIndex()[i]);
 825             queryProcessor.updateValue(resource);
 826             if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 827                 System.err.println("value " + cc.getValueIndex()[i] + " changed.");
 828         }
 829     }
 830     final void refreshImportance(ClusterImpl c) {
 831
 832         if (c.isWriteOnly())
 833             return;
 834
 835         //printMaps("refreshImportance");
 836
 837         importanceMap.remove(c.getImportance());
 838         if(collectorPolicy != null) collectorPolicy.removed(c);
 839
 840         long newImportance = timeCounter();
 841 // System.err.println("refreshImportance " + c.getClusterId() + " => " + newImportance);
 842         c.setImportance(newImportance);
 843         if (!c.isLoaded())
 844             Logger.defaultLogError(new Exception("Bug in ClusterTable.refreshImportance(ClusterImpl), cluster not loaded"));
 845
 846         importanceMap.put(c.getImportance(), new ImportanceEntry(c));
 847         if(collectorPolicy != null) collectorPolicy.added(c);
 848         if (VALIDATE_SIZE)
 849             validateSize("refreshImportance");
 850
 851     }
 852
 853     static long loadTime = 0;
 854
 855     @SuppressWarnings("unchecked")
 856     public final <T extends ClusterI> T getClusterProxyByResourceKey(final int resourceKey) {
 857         int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
 858         if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
 859             throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
 860         ClusterI cluster = clusterArray[clusterKey];
 861         if (cluster == null)
 862             throw new RuntimeException("No proxy for existing cluster. Resource key = " + resourceKey);
 863         return (T)cluster;
 864     }
 865
 866     TLongIntHashMap clusterLoadHistogram = new TLongIntHashMap();
 867     int clusterLoadCounter = 0;
 868
 869     private <T extends ClusterI> T ensureLoaded(T c) {
 870         ClusterI cluster;
 871         ClusterImpl cs = (ClusterImpl) c;
 872         try {
 873             if(DebugPolicy.REPORT_CLUSTER_LOADING) {
 874                 long start = System.nanoTime();
 875                 cluster = load2(cs.getClusterId(), cs.getClusterKey());
 876                 long load = System.nanoTime()-start;
 877                 loadTime +=  load;
 878                 if(DebugPolicy.REPORT_CLUSTER_LOADING) {
 879                     int was = clusterLoadHistogram.get(cluster.getClusterId());
 880                     clusterLoadHistogram.put(cluster.getClusterId(), was+1);
 881                     clusterLoadCounter++;
 882                     String text = "Load2 " + cluster + " " + 1e-9*loadTime + "s. " + 1e-6*load + "ms. " + clusterLoadCounter + " " + was + " " + cluster.getUsedSpace() + " " + cluster.getImportance();
 883                     if(DebugPolicy.REPORT_CLUSTER_LOADING_STACKS) {
 884                         new Exception(text).printStackTrace();
 885                     } else {
 886                         System.err.println(text);
 887                     }
 888                 }
 889             } else {
 890                 cluster = load2(cs.getClusterId(), cs.getClusterKey());
 891             }
 892         } catch (DatabaseException e) {
 893             Logger.defaultLogError(e);
 894             if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 895                 e.printStackTrace();
 896             String msg = "Failed to load cluster " + cs.getClusterUID();// + " resourceId=" + (((cs.getClusterId() << 16 + (resourceKey & 65535))));
 897             // TODO: this jams the system => needs refactoring.
 898             throw new RuntimeDatabaseException(msg, e);
 899         }
 900         return (T) cluster;
 901     }
 902
 903     @SuppressWarnings("unchecked")
 904     public final <T extends ClusterI> T getClusterByResourceKey(final int resourceKey) {
 905         int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
 906         if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
 907             throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
 908         ClusterI c = clusterArray[clusterKey];
 909         if (c == null)
 910             return null;
 911         if (c.isLoaded()) {
 912             if ((counter++ & 4095) == 0)
 913                 refreshImportance((ClusterImpl) c);
 914             return (T) c;
 915         }
 916         if (!(c instanceof ClusterSmall)) {
 917             Logger.defaultLogError("Proxy must be instance of ClusterSmall");
 918             return null;
 919         }
 920         return ensureLoaded((T)c);
 921     }
 922
 923     @SuppressWarnings("unchecked")
 924     final <T extends ClusterI> T checkedGetClusterByResourceKey(final int resourceKey) {
 925         int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
 926         if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
 927             throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
 928         ClusterI c = clusterArray[clusterKey];
 929         if (c == null)
 930             throw new RuntimeException("No cluster for resource key " + resourceKey);
 931         if (c.isLoaded()) {
 932             if ((counter++ & 4095) == 0)
 933                 refreshImportance((ClusterImpl) c);
 934             return (T) c;
 935         }
 936         if (!(c instanceof ClusterSmall)) {
 937             Logger.defaultLogError("Proxy must be instance of ClusterSmall");
 938             return null;
 939         }
 940         ClusterI cluster;
 941         ClusterSmall cs = (ClusterSmall) c;
 942         try {
 943 //          System.err.println("Load2 " + resourceKey);
 944             cluster = load2(cs.getClusterId(), cs.getClusterKey());
 945         } catch (DatabaseException e) {
 946             if (DebugPolicy.REPORT_CLUSTER_EVENTS)
 947                 e.printStackTrace();
 948             int resourceIndex = resourceKey & ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(resourceKey);
 949             long resourceId = ClusterTraitsBase.createResourceIdNoThrow(cs.getClusterId(), resourceIndex);
 950             String msg = "Failed to load cluster " + cs.getClusterUID() + " for resource key " + resourceKey + " resourceIndex=" + resourceIndex + " resourceId=" + resourceId;
 951             Logger.defaultLogError(msg, e);
 952             c.setDeleted(true, null);
 953             return (T)c;
 954         }
 955         return (T) cluster;
 956     }
 957
 958     void printDebugInfo(ClusterSupport support)
 959             throws DatabaseException {
 960         final int SIZE = clusters.size();
 961         long sum = 0;
 962         for (int i = 1; i < SIZE; ++i) {
 963             ClusterI c = clusterArray[i];
 964             c.getNumberOfResources(support);
 965             long size = c.getUsedSpace();
 966             System.out.println("cluster=" + c.getClusterId() + " size=" + size);
 967             c.printDebugInfo("koss: ", support);
 968             sum += size;
 969         }
 970         System.out.println("Total number of clusters " + SIZE);
 971         System.out.println("Total cluster size " + sum);
 972     }
 973
 974     Map<Long,ClusterImpl> prefetch = new HashMap<Long,ClusterImpl>();
 975
 976     public synchronized ClusterImpl load2(long clusterId, int clusterKey) throws DatabaseException {
 977         ClusterImpl curr = (ClusterImpl) clusterArray[clusterKey];
 978         if (curr.isLoaded())
 979             return curr;
 980
 981 //        getPrefetched();
 982 //
 983 //        curr = (ClusterImpl) clusterArray[clusterKey];
 984 //        if (curr.isLoaded())
 985 //            return curr;
 986
 987         final Semaphore s = new Semaphore(0);
 988         final DatabaseException[] ex = new DatabaseException[1];
 989
 990         load2(clusterId, clusterKey, e -> {
 991             ex[0] = e;
 992             s.release();
 993         });
 994         try {
 995             s.acquire();
 996         } catch (InterruptedException e) {
 997             Logger.defaultLogError(e);
 998         }
 999         if (null != ex[0])
1000             throw ex[0];
1001         if (Development.DEVELOPMENT) {
1002             if (Development.<Boolean>getProperty(DevelopmentKeys.CLUSTERTABLE_VALIDATE_ON_LOAD, Bindings.BOOLEAN)) {
1003                 try {
1004                     ClusterImpl loaded = (ClusterImpl) clusterArray[clusterKey];
1005                     if (loaded instanceof ClusterSmall)
1006                         ((ClusterSmall) loaded).check();
1007                     else if (loaded instanceof ClusterBig)
1008                         ((ClusterBig) loaded).check();
1009                 } catch (Throwable t) {
1010                     t.printStackTrace();
1011                 }
1012             }
1013         }
1014
1015         validate(clusterKey);
1016
1017         return (ClusterImpl) clusterArray[clusterKey];
1018     }
1019
1020     void validate(int clusterKey) {
1021
1022 //      if(sessionImpl.graphSession.graphClient instanceof GraphClientImpl2) {
1023 //
1024 //        ClusterImpl server = (ClusterImpl)clusterArray[clusterKey];
1025 //        String sd = server.dump(sessionImpl.clusterTranslator);
1026 //        GraphClientImpl2 gci = ((GraphClientImpl2)sessionImpl.graphSession.graphClient);
1027 //        ClusterImpl memory = gci.clusters.get(server.clusterUID);
1028 //        if(memory == null)
1029 //            System.err.println("ad");
1030 //        String md = memory.dump(gci.support);
1031 //
1032 //        if(!sd.equals(md)) {
1033 //
1034 //            int diffPos = 0;
1035 //            int minLength = Math.min(sd.length(), md.length());
1036 //            for (int i = 0; i < minLength; i++) {
1037 //                if (sd.charAt(i) != md.charAt(i)) {
1038 //                    diffPos = i;
1039 //                    break;
1040 //                }
1041 //            }
1042 //
1043 //            int start = Math.max(diffPos-200, 0);
1044 //            int end = Math.min(minLength-1, diffPos+200);
1045 //
1046 //            System.err.println("== CLUSTER DIFFERENCE " + clusterArray[clusterKey].getClusterUID() +  " == ");
1047 //            System.err.println("== SESSION == ");
1048 //            System.err.println(sd.substring(start, end));
1049 //            System.err.println("== MEM == ");
1050 //            System.err.println(md.substring(start, end));
1051 //            System.err.println("== CLUSTER DIFFERENCE ENDS == ");
1052 //
1053 //            throw new IllegalStateException();
1054 //        }
1055 //
1056 //      }
1057     }
1058
1059     public synchronized ClusterImpl getPrefetched() {
1060         synchronized(prefetch) {
1061             return null;
1062         }
1063     }
1064
1065     public synchronized void load2(long clusterId, int clusterKey, final Consumer<DatabaseException> runnable) {
1066
1067         assert (Constants.ReservedClusterId != clusterId);
1068
1069         ClusterImpl cluster = null;
1070         DatabaseException e = null;
1071
1072         try {
1073
1074             ClusterUID clusterUID = clusters.makeClusterUID(clusterId);
1075 //            cluster = (ClusterImpl) sessionImpl.getGraphSession().getClusterImpl(clusterUID, clusterKey);
1076             Database.Session session = sessionImpl.graphSession.dbSession;
1077 //            cluster = (ClusterImpl)gci.getClusterByClusterKey(clusterKey).clone(sessionImpl.clusterTranslator);
1078 //            cluster = (ClusterImpl)((ClusterImpl)gci.getClusterByClusterUIDOrMakeProxy(clusterUID)).clone(sessionImpl.clusterTranslator);
1079
1080 //            cluster = gci.clone(clusterUID, sessionImpl.clusterTranslator);
1081
1082             cluster = session.clone(clusterUID, new ClusterCreator() {
1083
1084                 @Override
1085                 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
1086                     ClusterSupport support = sessionImpl.clusterTranslator;
1087                     try {
1088                         return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
1089                     } catch (DatabaseException e) {
1090                         e.printStackTrace();
1091                         return null;
1092                     }
1093                 }
1094
1095             });
1096
1097         } catch (Throwable t) {
1098             // It's totally legal to call for non-existing cluster.
1099             // Sometimes this indicates an error, though.
1100             Logger.getDefault().logInfo("Load cluster failed.", t);
1101             if (t instanceof DatabaseException)
1102                 e = (DatabaseException) t;
1103             else
1104                 e = new DatabaseException("Load cluster failed.", t);
1105         }
1106         if (null == cluster) {
1107             runnable.accept(e);
1108             return;
1109         }
1110
1111 //      new Exception("Load cluster " + cluster.getClusterId() + " " + cluster.getCachedSize() + " " + cluster.getImmutable()).printStackTrace();
1112
1113         // Can not be called with null argument.
1114         replaceCluster(cluster);
1115         sessionImpl.onClusterLoaded(clusterId);
1116         runnable.accept(null);
1117
1118     }
1119
1120     public synchronized ClusterImpl tryLoad(long clusterId, int clusterKey) throws DatabaseException {
1121
1122         assert (Constants.ReservedClusterId != clusterId);
1123
1124         ClusterUID clusterUID = clusters.makeClusterUID(clusterId);
1125
1126         Database.Session session = sessionImpl.graphSession.dbSession;
1127
1128         ClusterImpl cluster = session.clone(clusterUID, new ClusterCreator() {
1129
1130             @Override
1131             public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
1132                 ClusterSupport support = sessionImpl.clusterTranslator;
1133                 try {
1134                     return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
1135                 } catch (DatabaseException e) {
1136                     e.printStackTrace();
1137                     return null;
1138                 }
1139             }
1140
1141         });
1142         if (null == clusterArray[clusterKey])
1143             clusterArray[clusterKey] = cluster;
1144         else
1145             replaceCluster(cluster);
1146         sessionImpl.onClusterLoaded(clusterId);
1147
1148         validate(clusterKey);
1149
1150         return cluster;
1151     }
1152
1153     public Collection<ClusterI> getClusters() {
1154         ArrayList<ClusterI> result = new ArrayList<ClusterI>();
1155         for (int i = 0; i < clusterArray.length; i++) {
1156             ClusterI cluster = clusterArray[i];
1157             if (cluster != null)
1158                 result.add(cluster);
1159         }
1160         return result;
1161     }
1162
1163     public int size() {
1164         return clusters.size();
1165     }
1166
1167     public long timeCounter() {
1168         return timeCounter.getAndIncrement();
1169     }
1170
1171     public boolean hasVirtual(int index) {
1172         return virtuals[index];
1173     }
1174
1175     public void markVirtual(int index) {
1176         virtuals[index] = true;
1177     }
1178
1179     public ClusterImpl[] getClusterArray() {
1180         return clusterArray;
1181     }
1182
1183     public boolean isImmutable(int id) {
1184
1185         int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(id);
1186         Boolean exist = immutables[clusterKey];
1187         if(exist != null) return exist;
1188
1189         ClusterI cluster = getClusterByResourceKey(id);
1190         if(cluster == null) {
1191                 return false;
1192         } else {
1193                 boolean result = cluster.getImmutable();
1194                 markImmutable(cluster, result);
1195                 return result;
1196         }
1197
1198     }
1199
1200     public void markImmutable(ClusterI cluster, boolean value) {
1201         immutables[cluster.getClusterKey()] = value;
1202     }
1203     @Override
1204     public int getClusterKeyByUID(long first, long second) throws DatabaseException {
1205         return getClusterKeyByClusterUIDOrMakeProxy(ClusterUID.make(first, second));
1206     }
1207
1208     public void adjustCachedSize(long l, ClusterI cluster) {
1209         if (l != 0) {
1210             //System.out.println("ClusterTable: adjusting cluster table cached size by " + l + ": " + sizeInBytes + " -> "
1211             //        + (sizeInBytes + l) + ", for cluster " + cluster.getClusterId() + " (" + cluster + ")");
1212             sizeInBytes += l;
1213         }
1214     }
1215
1216     private void validateSize(String place) {
1217         if (!VALIDATE_SIZE)
1218             return;
1219
1220         int ims = importanceMap.size();
1221         int ihms = countImportantClusters();
1222
1223 //        System.out.format("[ClusterTable.%s] Validating: byteSize=%d (%d MB), hashMap=%d/%d, importanceMap=%d%n",
1224 //                place, sizeInBytes, sizeInBytes / (1024*1024), ihms, clusters.hashMap.size(), ims);
1225
1226         int i = clusterArray.length;
1227         long size = 0;
1228         for (int j = 0; j < i; ++j) {
1229             ClusterI c = clusterArray[j];
1230             if (c == null)
1231                 continue;
1232             size += c.getCachedSize();
1233         }
1234         if (sizeInBytes != size) {
1235             if (!dirtySizeInBytes)
1236                 System.out.println("BUG: CACHED CLUSTER SIZE DIFFERS FROM CALCULATED: " + sizeInBytes + " != " + size + ", delta = " + (sizeInBytes - size));
1237             //else System.out.println("\"BUG?\": SIZES DIFFER: " + sizeInBytes + " != " + size + ", delta = " + (sizeInBytes - size));
1238         }
1239         if (ims != ihms) {
1240             System.out.println("BUG2: hashmap and importanceMap sizes differ: " + ihms + " != " + ims + ", delta=" + (ihms - ims));
1241             printMaps("validateSize");
1242         }
1243         //System.out.println("[" + place + "] VALIDATED");
1244     }
1245
1246     private void printMaps(String place) {
1247         int ihms = countImportantClusters();
1248         System.out.println("## printMaps(" + place + ") - " + importanceMap.size() + " - (" + ihms + "/" + clusters.hashMap.size() + ")");
1249         System.out.println("importanceMap (" + importanceMap.size() + "):");
1250         importanceMap.forEach((importance, ie) -> {
1251             System.out.format("\t%d: %s%n", importance, ie.toString());
1252         });
1253         System.out.println("clusters.hashMap (" + ihms + "/" + clusters.hashMap.size() + "):");
1254         clusters.hashMap.forEachEntry((cid, c) -> {
1255             boolean important = importantCluster(c);
1256             boolean wo = c != null && c.isWriteOnly();
1257             boolean empty = c != null && c.isEmpty();
1258             boolean loaded = c != null && c.isLoaded();
1259             System.out.format("\t%s: %d - %s (writeOnly=%b, empty=%b, loaded=%b)%n", important ? " I" : "NI", cid, c, wo, empty, loaded);
1260             return true;
1261         });
1262     }
1263
1264     private int countImportantClusters() {
1265         int[] result = { 0 };
1266         clusters.hashMap.forEachEntry((cid, c) -> {
1267             if (importantCluster(c))
1268                 result[0]++;
1269             return true;
1270         });
1271         return result[0];
1272     }
1273
1274     private static boolean importantCluster(ClusterImpl c) {
1275         return c != null && !c.isWriteOnly() && c.isLoaded();// && !c.isEmpty();
1276     }
1277
1278 }