1 /*******************************************************************************
2 * Copyright (c) 2007, 2010 Association for Decentralized Information Management
4 * All rights reserved. This program and the accompanying materials
5 * are made available under the terms of the Eclipse Public License v1.0
6 * which accompanies this distribution, and is available at
7 * http://www.eclipse.org/legal/epl-v10.html
10 * VTT Technical Research Centre of Finland - initial API and implementation
11 *******************************************************************************/
12 package fi.vtt.simantics.procore.internal;
15 import java.util.ArrayList;
16 import java.util.Collection;
17 import java.util.HashMap;
19 import java.util.TreeMap;
20 import java.util.concurrent.Semaphore;
21 import java.util.concurrent.atomic.AtomicLong;
22 import java.util.function.Consumer;
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;
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;
62 public final class ClusterTable implements IClusterTable {
64 private static final boolean VALIDATE_SIZE = false;
66 int maximumBytes = 128 * 1024 * 1024;
67 int limit = (int)(0.8*(double)maximumBytes);
69 private final AtomicLong timeCounter = new AtomicLong(1);
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());
85 public ImportanceEntry(long importance, long clusterId) {
86 this.importance = importance;
87 this.clusterId = clusterId;
90 public long getImportance() {
94 public long getClusterId() {
98 public String toString() {
99 return "CID " + clusterId;
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.
109 private void clear() {
111 // clusterU2I.clear();
112 hashMap.put(0, null); // reserved for null value
113 // clusterU2I.put(ClusterUID.make(0, 0), null);
116 return hashMap.size();
118 private ClusterImpl getClusterByClusterId(long clusterId) {
119 return hashMap.get(clusterId);
121 private ClusterImpl getClusterByClusterUID(ClusterUID clusterUID) {
122 return getClusterByClusterId(clusterUID.second);
123 //return clusterU2I.get(clusterUID);
125 private ClusterImpl getClusterByClusterUID(long id1, long id2) {
126 return getClusterByClusterId(id2);
128 private ClusterImpl makeProxy(long clusterId) {
129 return makeProxy(ClusterUID.make(0, clusterId));
131 private ClusterImpl makeProxy(ClusterUID clusterUID) {
132 ClusterImpl proxy = hashMap.get(clusterUID.second);
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.");
142 private void replace(ClusterImpl proxy) {
143 ClusterImpl old = clusterArray[proxy.clusterKey];
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.");
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);
169 private ClusterImpl create(ClusterImpl clusterImpl) {
170 hashMap.put(clusterImpl.clusterId, clusterImpl);
171 // clusterU2I.put(clusterImpl.clusterUID, clusterImpl);
172 clusterArray[clusterImpl.clusterKey] = clusterImpl;
175 private void forEachValue(TObjectProcedure<ClusterImpl> procedure) {
176 hashMap.forEachValue(procedure);
178 private void forEachEntry(TLongObjectProcedure<ClusterImpl> procedure) {
179 hashMap.forEachEntry(procedure);
181 private ClusterUID makeClusterUID(long clusterId) {
182 return ClusterUID.make(0, clusterId);
184 private void removeProxy(ClusterImpl proxy) {
185 hashMap.remove(proxy.clusterId);
186 clusterArray[proxy.clusterKey] = null;
189 final public TreeMap<Long, CollectorCluster> importanceMap = new TreeMap<Long, CollectorCluster>();
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);
198 public ClusterUID getClusterUIDByResourceKey(int resourceKey) throws DatabaseException {
199 int clusterKey = ClusterTraits.getClusterKeyFromResourceKey(resourceKey);
200 return clusterArray[clusterKey].clusterUID;
202 ClusterTable(SessionImplSocket sessionImpl, File folderPath) {
203 this.sessionImpl = sessionImpl;
204 clusters = new Clusters();
209 importanceMap.clear();
212 public void setCollectorPolicy(ClusterCollectorPolicy policy) {
213 this.collectorPolicy = policy;
217 * Sets Cluster Table allocation size by percentage of maximum usable
218 * memory. Allocation is limited between 5% and 50% of maximum memory.
222 private void setAllocateSize(double percentage) {
224 percentage = Math.max(percentage, 0.05);
225 percentage = Math.min(percentage, 0.5);
227 maximumBytes = (int) Math.floor(percentage * Runtime.getRuntime().maxMemory());
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);
249 void checkCollect() {
253 synchronized ClusterImpl getClusterByClusterId(long clusterId) {
254 return clusters.getClusterByClusterId(clusterId);
257 ClusterBase getClusterByClusterKey(int clusterKey) {
258 return clusterArray[clusterKey];
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);
267 synchronized ClusterImpl makeCluster(long clusterId, boolean writeOnly) {
269 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
272 ClusterUID clusterUID = ClusterUID.make(0, clusterId);
273 int clusterKey = clusters.size(); // new key
275 proxy = new ClusterWriteOnly(clusterUID, clusterKey, sessionImpl);
276 writeOnlyClusters.add(proxy);
277 return clusters.create(proxy);
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));
286 validateSize("makeCluster");
287 if(collectorPolicy != null) collectorPolicy.added(cluster);
292 synchronized void replaceCluster(ClusterI cluster_) {
293 ClusterImpl cluster = (ClusterImpl) cluster_;
295 int clusterKey = cluster.getClusterKey();
296 ClusterImpl existing = (ClusterImpl) clusterArray[clusterKey];
297 if (existing.hasVirtual())
298 cluster.markVirtual();
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())
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;
315 importanceMap.remove(existing.getImportance());
316 if (collectorPolicy != null)
317 collectorPolicy.removed((ClusterImpl)existing);
319 //System.out.println("ClusterTable.replace(" + existing + " (I=" + existing.getImportance() + ") => " + cluster + " (I=" + cluster.getImportance() + ")");
321 clusters.replace((ClusterImpl)cluster);
322 if (!cluster.isLoaded())
323 Logger.defaultLogError(new Exception("Bug in ClusterTable.replaceCluster(ClusterI), cluster not loaded"));
325 importanceMap.put(cluster.getImportance(), new ImportanceEntry((ClusterImpl)cluster));
326 if(collectorPolicy != null) collectorPolicy.added((ClusterImpl)cluster);
328 if (!dirtySizeInBytes) {
329 if (existing != cluster) {
330 adjustCachedSize(-existing.getCachedSize(), existing);
332 // This will update sizeInBytes through adjustCachedSize
333 cluster.getCachedSize();
336 validateSize("replaceCluster");
339 synchronized void release(CollectorCluster cluster) {
340 importanceMap.remove(cluster.getImportance());
341 release(cluster.getClusterId());
344 synchronized void release(long clusterId) {
345 //System.out.println("ClusterTable.release(" + clusterId + "): " + sizeInBytes);
347 ClusterImpl clusterImpl = clusters.getClusterByClusterId(clusterId);
348 if (null == clusterImpl)
350 if(!clusterImpl.isLoaded() || clusterImpl.isEmpty())
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);
363 validateSize("release");
366 synchronized void compact(long id) {
367 ClusterI impl = clusters.getClusterByClusterId(id);
373 // cachedSize = getSizeInBytes();
376 double getLoadProbability() {
377 // This can currently cause stack overflow
379 // if(cachedSize < SLOW_LIMIT) return 1.0;
380 // if(cachedSize > HIGH_LIMIT) return 1e-2;
382 // double pos = (double)(cachedSize - SLOW_LIMIT) / (double)(HIGH_LIMIT -
384 // double val = 0.1 * ((pos-1) * (pos-1)) + 1e-2;
388 class SizeProcedure implements TObjectProcedure<ClusterImpl> {
389 public long result = 0;
392 public boolean execute(ClusterImpl cluster) {
393 if (cluster != null) {
395 if (cluster.isLoaded() && !cluster.isEmpty()) {
396 result += cluster.getCachedSize();
398 } catch (Throwable t) {
399 Logger.defaultLogError(t);
405 public void clear() {
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);
423 public void setDirtySizeInBytes(boolean value) {
424 dirtySizeInBytes = value;
427 ClusterStateImpl getState() {
428 final ClusterStateImpl result = new ClusterStateImpl();
429 clusters.forEachEntry(new TLongObjectProcedure<ClusterImpl>() {
431 public boolean execute(long arg0, ClusterImpl arg1) {
434 if (arg1.isLoaded() && !arg1.isEmpty()) {
435 result.ids.add(new ImportanceEntry(arg1));
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);
450 class ClusterCollectorImpl implements ClusterCollector {
452 final private ClusterCollectorSupport support;
453 private ClusterCollectorPolicy policy;
454 ClusterCollectorImpl(ClusterCollectorSupport support) {
455 this.support = support;
458 ClusterCollectorPolicy setPolicy(ClusterCollectorPolicy newPolicy) {
460 ClusterCollectorPolicy oldPolicy = policy;
464 for (CollectorCluster id : support.getResidentClusters()) {
465 policy.added(getClusterByClusterId(id.getClusterId()));
469 support.setPolicy(policy);
476 public void collect() {
480 release(policy.select());
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);
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)
497 collectSize += size - maximumBytes;
498 collect(collectSize);
500 // try to keep allocated clusters below the maximum
501 if (size < maximumBytes)
511 private boolean useDynamicAllocation() {
512 return "true".equalsIgnoreCase(System.getProperty("org.simantics.db.cluster.dynamicAlloc"));
516 public void collect(int target) {
520 release(policy.select(target));
524 ArrayList<CollectorCluster> toRelease = new ArrayList<CollectorCluster>();
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;
538 if (DebugPolicy.CLUSTER_COLLECTION) {
539 System.out.println("Cluster collector finished, current size = " + support.getCurrentSize());
546 void release(Collection<CollectorCluster> toRelease) {
547 for (CollectorCluster id : toRelease) {
554 private ClusterCollectorSupport collectorSupport = new ClusterCollectorSupportImpl(this);
555 ClusterCollectorImpl collector = new ClusterCollectorImpl(collectorSupport);
561 private long newResourceClusterId = Constants.NewClusterId;
562 public static final int CLUSTER_FILL_SIZE = ClusterTraitsBase.getMaxNumberOfResources();
565 * Uusi id varataan vasta, kun lis�t��n resurssi => reservedIds sis�lt��
566 * vain jo k�yt�ss� olevia klustereita
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);
576 ClusterImpl cluster = getClusterByClusterIdOrThrow(newResourceClusterId);
577 if (cluster.getNumberOfResources(cs) >= CLUSTER_FILL_SIZE) {
578 newResourceClusterId = graphSession.newClusterId();
579 cluster = getClusterByClusterIdOrMake(newResourceClusterId, writeOnly);
583 return ensureLoaded(result);
586 void flushCluster(GraphSession graphSession) {
587 // We seem to disagree about this.
588 // graphSession.newClusterId();
589 newResourceClusterId = Constants.NewClusterId;
592 void writeOnlyInvalidate(ClusterI impl) {
593 writeOnlyInvalidates.add(impl.getClusterKey());
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);
602 writeOnlyClusters.clear();
603 writeOnlyInvalidates.forEach(new TIntProcedure() {
605 public boolean execute(int clusterKey) {
606 ClusterImpl proxy = clusterArray[clusterKey];
607 ClusterUID clusterUID = proxy.getClusterUID();
608 clusters.freeProxy(proxy);
612 writeOnlyInvalidates.clear();
615 public ClusterImpl getClusterByClusterUID(ClusterUID clusterUID) {
616 synchronized (this) {
617 return clusters.getClusterByClusterUID(clusterUID);
620 public int getClusterKeyByClusterUIDOrMakeProxy(ClusterUID clusterUID) {
621 return getClusterKeyByClusterUIDOrMakeProxy(0/*clusterUID.first*/, clusterUID.second);
623 public int getClusterKeyByClusterUIDOrMakeProxy(long id1, long id2) {
624 return getClusterByClusterUIDOrMakeProxy(id1, id2).clusterKey;
626 public ClusterImpl getClusterByClusterUIDOrMakeProxy(ClusterUID clusterUID) {
627 synchronized (this) {
628 ClusterImpl clusterImpl = clusters.getClusterByClusterUID(clusterUID);
629 if (null == clusterImpl)
630 clusterImpl = clusters.makeProxy(clusterUID);
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);
642 ClusterImpl getLoadOrThrow(long clusterId) throws DatabaseException {
643 synchronized (this) {
644 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
647 if (proxy.isLoaded())
650 clusterKey = proxy.getClusterKey();
652 if (clusterKey == 0) {
653 proxy = clusters.makeProxy(clusterId);
654 clusterKey = proxy.getClusterKey();
656 ClusterImpl ci = tryLoad(clusterId, clusterKey);
658 throw new ResourceNotFoundException(clusterId);
660 } catch (ClusterDoesNotExistException t) {
661 clusters.removeProxy(proxy);
667 ClusterImpl getClusterByClusterIdOrMake(long clusterId, boolean writeOnly) {
668 synchronized (this) {
669 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
673 return makeCluster(clusterId, writeOnly);
676 ClusterImpl getClusterByClusterIdOrThrow(long clusterId) {
677 synchronized (this) {
678 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
680 throw new IllegalArgumentException("Cluster id=" + clusterId + " is not created.");
684 long getClusterIdOrCreate(ClusterUID clusterUID) {
685 return clusterUID.second;
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];
694 throw new RuntimeException("No cluster for key " + resourceKey);
695 return c.getClusterId();
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);
703 ClusterI c = clusterArray[clusterKey];
705 Logger.defaultLogError("No cluster for key " + resourceKey);
708 return c.getClusterId();
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) {
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)
728 if (!oldCluster.isLoaded())
730 int clusterKey = oldCluster.getClusterKey();
731 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
733 boolean big = oldCluster instanceof ClusterBig;
734 boolean small = oldCluster instanceof ClusterSmall;
737 // ClusterImpl newCluster = (ClusterImpl) sessionImpl.graphSession.getClusterImpl(clusterUID[i], clusterKey);
738 Database.Session dbSession = sessionImpl.graphSession.dbSession;
740 ClusterImpl newCluster = dbSession.clone(clusterUID[i], new ClusterCreator() {
743 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
744 ClusterSupport support = sessionImpl.clusterTranslator;
746 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
747 } catch (DatabaseException e) {
754 if (null == newCluster)
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);
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);
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();
791 session.writeState = null;
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();
805 cc = session.graphSession.getClusterChanges(clusterUID, csid);
806 } catch (Exception e) {
807 Logger.defaultLogError("Could not get cluster changes. cluster=" + clusterUID, e);
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)
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.");
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.");
830 final void refreshImportance(ClusterImpl c) {
835 //printMaps("refreshImportance");
837 importanceMap.remove(c.getImportance());
838 if(collectorPolicy != null) collectorPolicy.removed(c);
840 long newImportance = timeCounter();
841 // System.err.println("refreshImportance " + c.getClusterId() + " => " + newImportance);
842 c.setImportance(newImportance);
844 Logger.defaultLogError(new Exception("Bug in ClusterTable.refreshImportance(ClusterImpl), cluster not loaded"));
846 importanceMap.put(c.getImportance(), new ImportanceEntry(c));
847 if(collectorPolicy != null) collectorPolicy.added(c);
849 validateSize("refreshImportance");
853 static long loadTime = 0;
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];
862 throw new RuntimeException("No proxy for existing cluster. Resource key = " + resourceKey);
866 TLongIntHashMap clusterLoadHistogram = new TLongIntHashMap();
867 int clusterLoadCounter = 0;
869 private <T extends ClusterI> T ensureLoaded(T c) {
871 ClusterImpl cs = (ClusterImpl) c;
873 if(DebugPolicy.REPORT_CLUSTER_LOADING) {
874 long start = System.nanoTime();
875 cluster = load2(cs.getClusterId(), cs.getClusterKey());
876 long load = System.nanoTime()-start;
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();
886 System.err.println(text);
890 cluster = load2(cs.getClusterId(), cs.getClusterKey());
892 } catch (DatabaseException e) {
893 Logger.defaultLogError(e);
894 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
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);
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];
912 if ((counter++ & 4095) == 0)
913 refreshImportance((ClusterImpl) c);
916 if (!(c instanceof ClusterSmall)) {
917 Logger.defaultLogError("Proxy must be instance of ClusterSmall");
920 return ensureLoaded((T)c);
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];
930 throw new RuntimeException("No cluster for resource key " + resourceKey);
932 if ((counter++ & 4095) == 0)
933 refreshImportance((ClusterImpl) c);
936 if (!(c instanceof ClusterSmall)) {
937 Logger.defaultLogError("Proxy must be instance of ClusterSmall");
941 ClusterSmall cs = (ClusterSmall) c;
943 // System.err.println("Load2 " + resourceKey);
944 cluster = load2(cs.getClusterId(), cs.getClusterKey());
945 } catch (DatabaseException e) {
946 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
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);
958 void printDebugInfo(ClusterSupport support)
959 throws DatabaseException {
960 final int SIZE = clusters.size();
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);
970 System.out.println("Total number of clusters " + SIZE);
971 System.out.println("Total cluster size " + sum);
974 Map<Long,ClusterImpl> prefetch = new HashMap<Long,ClusterImpl>();
976 public synchronized ClusterImpl load2(long clusterId, int clusterKey) throws DatabaseException {
977 ClusterImpl curr = (ClusterImpl) clusterArray[clusterKey];
983 // curr = (ClusterImpl) clusterArray[clusterKey];
984 // if (curr.isLoaded())
987 final Semaphore s = new Semaphore(0);
988 final DatabaseException[] ex = new DatabaseException[1];
990 load2(clusterId, clusterKey, e -> {
996 } catch (InterruptedException e) {
997 Logger.defaultLogError(e);
1001 if (Development.DEVELOPMENT) {
1002 if (Development.<Boolean>getProperty(DevelopmentKeys.CLUSTERTABLE_VALIDATE_ON_LOAD, Bindings.BOOLEAN)) {
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();
1015 validate(clusterKey);
1017 return (ClusterImpl) clusterArray[clusterKey];
1020 void validate(int clusterKey) {
1022 // if(sessionImpl.graphSession.graphClient instanceof GraphClientImpl2) {
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);
1032 // if(!sd.equals(md)) {
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)) {
1043 // int start = Math.max(diffPos-200, 0);
1044 // int end = Math.min(minLength-1, diffPos+200);
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 == ");
1053 // throw new IllegalStateException();
1059 public synchronized ClusterImpl getPrefetched() {
1060 synchronized(prefetch) {
1065 public synchronized void load2(long clusterId, int clusterKey, final Consumer<DatabaseException> runnable) {
1067 assert (Constants.ReservedClusterId != clusterId);
1069 ClusterImpl cluster = null;
1070 DatabaseException e = null;
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);
1080 // cluster = gci.clone(clusterUID, sessionImpl.clusterTranslator);
1082 cluster = session.clone(clusterUID, new ClusterCreator() {
1085 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
1086 ClusterSupport support = sessionImpl.clusterTranslator;
1088 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
1089 } catch (DatabaseException e) {
1090 e.printStackTrace();
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;
1104 e = new DatabaseException("Load cluster failed.", t);
1106 if (null == cluster) {
1111 // new Exception("Load cluster " + cluster.getClusterId() + " " + cluster.getCachedSize() + " " + cluster.getImmutable()).printStackTrace();
1113 // Can not be called with null argument.
1114 replaceCluster(cluster);
1115 sessionImpl.onClusterLoaded(clusterId);
1116 runnable.accept(null);
1120 public synchronized ClusterImpl tryLoad(long clusterId, int clusterKey) throws DatabaseException {
1122 assert (Constants.ReservedClusterId != clusterId);
1124 ClusterUID clusterUID = clusters.makeClusterUID(clusterId);
1126 Database.Session session = sessionImpl.graphSession.dbSession;
1128 ClusterImpl cluster = session.clone(clusterUID, new ClusterCreator() {
1131 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
1132 ClusterSupport support = sessionImpl.clusterTranslator;
1134 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
1135 } catch (DatabaseException e) {
1136 e.printStackTrace();
1142 if (null == clusterArray[clusterKey])
1143 clusterArray[clusterKey] = cluster;
1145 replaceCluster(cluster);
1146 sessionImpl.onClusterLoaded(clusterId);
1148 validate(clusterKey);
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);
1164 return clusters.size();
1167 public long timeCounter() {
1168 return timeCounter.getAndIncrement();
1171 public boolean hasVirtual(int index) {
1172 return virtuals[index];
1175 public void markVirtual(int index) {
1176 virtuals[index] = true;
1179 public ClusterImpl[] getClusterArray() {
1180 return clusterArray;
1183 public boolean isImmutable(int id) {
1185 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(id);
1186 Boolean exist = immutables[clusterKey];
1187 if(exist != null) return exist;
1189 ClusterI cluster = getClusterByResourceKey(id);
1190 if(cluster == null) {
1193 boolean result = cluster.getImmutable();
1194 markImmutable(cluster, result);
1200 public void markImmutable(ClusterI cluster, boolean value) {
1201 immutables[cluster.getClusterKey()] = value;
1204 public int getClusterKeyByUID(long first, long second) throws DatabaseException {
1205 return getClusterKeyByClusterUIDOrMakeProxy(ClusterUID.make(first, second));
1208 public void adjustCachedSize(long l, ClusterI cluster) {
1210 //System.out.println("ClusterTable: adjusting cluster table cached size by " + l + ": " + sizeInBytes + " -> "
1211 // + (sizeInBytes + l) + ", for cluster " + cluster.getClusterId() + " (" + cluster + ")");
1216 private void validateSize(String place) {
1220 int ims = importanceMap.size();
1221 int ihms = countImportantClusters();
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);
1226 int i = clusterArray.length;
1228 for (int j = 0; j < i; ++j) {
1229 ClusterI c = clusterArray[j];
1232 size += c.getCachedSize();
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));
1240 System.out.println("BUG2: hashmap and importanceMap sizes differ: " + ihms + " != " + ims + ", delta=" + (ihms - ims));
1241 printMaps("validateSize");
1243 //System.out.println("[" + place + "] VALIDATED");
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());
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);
1264 private int countImportantClusters() {
1265 int[] result = { 0 };
1266 clusters.hashMap.forEachEntry((cid, c) -> {
1267 if (importantCluster(c))
1274 private static boolean importantCluster(ClusterImpl c) {
1275 return c != null && !c.isWriteOnly() && c.isLoaded();// && !c.isEmpty();