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.exception.ClusterDoesNotExistException;
31 import org.simantics.db.exception.DatabaseException;
32 import org.simantics.db.exception.ResourceNotFoundException;
33 import org.simantics.db.exception.RuntimeDatabaseException;
34 import org.simantics.db.impl.ClusterBase;
35 import org.simantics.db.impl.ClusterI;
36 import org.simantics.db.impl.ClusterSupport;
37 import org.simantics.db.impl.ClusterTraitsBase;
38 import org.simantics.db.impl.IClusterTable;
39 import org.simantics.db.impl.graph.WriteGraphImpl;
40 import org.simantics.db.impl.query.QueryProcessor;
41 import org.simantics.db.procore.cluster.ClusterBig;
42 import org.simantics.db.procore.cluster.ClusterImpl;
43 import org.simantics.db.procore.cluster.ClusterSmall;
44 import org.simantics.db.procore.cluster.ClusterTraits;
45 import org.simantics.db.procore.protocol.Constants;
46 import org.simantics.db.service.ClusterCollectorPolicy;
47 import org.simantics.db.service.ClusterCollectorPolicy.CollectorCluster;
48 import org.simantics.db.service.ClusterUID;
49 import org.simantics.utils.Development;
50 import org.slf4j.LoggerFactory;
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 org.slf4j.Logger LOGGER = LoggerFactory.getLogger(ClusterTable.class);
66 private static final boolean VALIDATE_SIZE = false;
68 int maximumBytes = 128 * 1024 * 1024;
69 int limit = (int)(0.8*(double)maximumBytes);
71 private final AtomicLong timeCounter = new AtomicLong(1);
73 final private SessionImplSocket sessionImpl;
74 final private ArrayList<ClusterI> writeOnlyClusters = new ArrayList<ClusterI>();
75 final private TIntHashSet writeOnlyInvalidates = new TIntHashSet();
76 final private static int ARRAY_SIZE = ClusterTraits.getClusterArraySize();
77 final private ClusterImpl[] clusterArray = new ClusterImpl[ARRAY_SIZE];
78 final private Boolean[] immutables = new Boolean[ARRAY_SIZE];
79 final private boolean[] virtuals = new boolean[ARRAY_SIZE];
80 static class ImportanceEntry implements CollectorCluster {
81 final public long importance;
82 final public long clusterId;
83 public ImportanceEntry(ClusterImpl impl) {
84 this(impl.getImportance(), impl.getClusterId());
87 public ImportanceEntry(long importance, long clusterId) {
88 this.importance = importance;
89 this.clusterId = clusterId;
92 public long getImportance() {
96 public long getClusterId() {
100 public String toString() {
101 return "CID " + clusterId;
104 private class Clusters {
105 // This makes sure that non-null values from hashMap.get can be trusted (no unsynchronized rehashes occur)
106 final public TLongObjectHashMap<ClusterImpl> hashMap = new TLongObjectHashMap<ClusterImpl>(2 * ARRAY_SIZE);
107 //private final HashMap<ClusterUID, ClusterImpl> clusterU2I = new HashMap<ClusterUID, ClusterImpl>(); // Maps cluster UID to cluster.
111 private void clear() {
113 // clusterU2I.clear();
114 hashMap.put(0, null); // reserved for null value
115 // clusterU2I.put(ClusterUID.make(0, 0), null);
118 return hashMap.size();
120 private ClusterImpl getClusterByClusterId(long clusterId) {
121 return hashMap.get(clusterId);
123 private ClusterImpl getClusterByClusterUID(ClusterUID clusterUID) {
124 return getClusterByClusterId(clusterUID.second);
125 //return clusterU2I.get(clusterUID);
127 private ClusterImpl getClusterByClusterUID(long id1, long id2) {
128 return getClusterByClusterId(id2);
130 private ClusterImpl makeProxy(long clusterId) {
131 return makeProxy(ClusterUID.make(0, clusterId));
133 private ClusterImpl makeProxy(ClusterUID clusterUID) {
134 ClusterImpl proxy = hashMap.get(clusterUID.second);
137 int clusterKey = hashMap.size();
138 ClusterSmall sentinel = new ClusterSmall(clusterUID, clusterKey, ClusterTable.this, sessionImpl.clusterTranslator);
139 if (sentinel.clusterId != sentinel.clusterUID.second)
140 throw new RuntimeDatabaseException("ClusterTable corrupted.");
144 private void replace(ClusterImpl proxy) {
145 ClusterImpl old = clusterArray[proxy.clusterKey];
148 if (old.clusterKey != proxy.clusterKey)
149 throw new RuntimeDatabaseException("ClusterTable corrupted.");
150 if (old.clusterId != proxy.clusterId)
151 throw new RuntimeDatabaseException("ClusterTable corrupted.");
152 if (!old.clusterUID.equals(proxy.clusterUID))
153 throw new RuntimeDatabaseException("ClusterTable corrupted.");
156 private ClusterSmall freeProxy(ClusterImpl proxy) {
157 ClusterImpl clusterImpl = hashMap.get(proxy.clusterId);
158 if (null == clusterImpl)
159 throw new RuntimeDatabaseException("ClusterTable corrupted.");
160 // ClusterUID clusterUID = ClusterUID.make(0, proxy.clusterId);
161 // ClusterImpl clusterImpl2 = clusterU2I.get(clusterUID );
162 // if (clusterImpl != clusterImpl2)
163 // throw new RuntimeDatabaseException("ClusterTable corrupted.");
164 if (proxy.clusterId != clusterImpl.clusterId)
165 throw new RuntimeDatabaseException("ClusterTable corrupted.");
166 if (proxy.clusterKey != clusterImpl.clusterKey)
167 throw new RuntimeDatabaseException("ClusterTable corrupted.");
168 ClusterSmall sentinel = new ClusterSmall(makeClusterUID(proxy.clusterId) , proxy.clusterKey, ClusterTable.this, sessionImpl.clusterTranslator);
169 return (ClusterSmall)create(sentinel);
171 private ClusterImpl create(ClusterImpl clusterImpl) {
172 hashMap.put(clusterImpl.clusterId, clusterImpl);
173 // clusterU2I.put(clusterImpl.clusterUID, clusterImpl);
174 clusterArray[clusterImpl.clusterKey] = clusterImpl;
177 private void forEachValue(TObjectProcedure<ClusterImpl> procedure) {
178 hashMap.forEachValue(procedure);
180 private void forEachEntry(TLongObjectProcedure<ClusterImpl> procedure) {
181 hashMap.forEachEntry(procedure);
183 private ClusterUID makeClusterUID(long clusterId) {
184 return ClusterUID.make(0, clusterId);
186 private void removeProxy(ClusterImpl proxy) {
187 hashMap.remove(proxy.clusterId);
188 clusterArray[proxy.clusterKey] = null;
191 final public TreeMap<Long, CollectorCluster> importanceMap = new TreeMap<Long, CollectorCluster>();
193 private ClusterCollectorPolicy collectorPolicy;
194 private boolean dirtySizeInBytes = true;
195 private long sizeInBytes = 0;
196 private final Clusters clusters;
197 ClusterUID makeClusterUID(long clusterId) {
198 return clusters.makeClusterUID(clusterId);
200 public ClusterUID getClusterUIDByResourceKey(int resourceKey) throws DatabaseException {
201 int clusterKey = ClusterTraits.getClusterKeyFromResourceKey(resourceKey);
202 return clusterArray[clusterKey].clusterUID;
204 ClusterTable(SessionImplSocket sessionImpl, File folderPath) {
205 this.sessionImpl = sessionImpl;
206 clusters = new Clusters();
211 importanceMap.clear();
214 public void setCollectorPolicy(ClusterCollectorPolicy policy) {
215 this.collectorPolicy = policy;
219 * Sets Cluster Table allocation size by percentage of maximum usable
220 * memory. Allocation is limited between 5% and 50% of maximum memory.
224 private void setAllocateSize(double percentage) {
226 percentage = Math.max(percentage, 0.05);
227 percentage = Math.min(percentage, 0.5);
229 maximumBytes = (int) Math.floor(percentage * Runtime.getRuntime().maxMemory());
232 private double estimateProperAllocation(ClusterCollectorSupport support) {
233 long max = Runtime.getRuntime().maxMemory();
234 long free = Runtime.getRuntime().freeMemory();
235 long total = Runtime.getRuntime().totalMemory();
236 long realFree = max - total + free; // amount of free memory
237 long current = support.getCurrentSize(); // currently allocated cache
238 long inUseTotal = max - realFree; // currently used memory
239 long inUseOther = inUseTotal - current; // memory that the application uses (without the cache)
240 double otherUsePercentage = (double) inUseOther / (double) max; // percentage of memory that the application uses
241 double aimFree = 0.2; // percentage of maximum heap that we try to keep free
242 double estimate = 1.0 - otherUsePercentage - aimFree;
243 estimate = Math.min(estimate, 0.5);
244 estimate = Math.max(estimate, 0.05);
245 // System.out.println("Estimated allocation percentage " + estimate +
246 // " memory stats: max: " + max + ", free: " + realFree + ", inUse: " +
247 // inUseTotal + ", cached: " + current + ", cacheSize: " + maximumBytes);
251 void checkCollect() {
255 synchronized ClusterImpl getClusterByClusterId(long clusterId) {
256 return clusters.getClusterByClusterId(clusterId);
259 ClusterBase getClusterByClusterKey(int clusterKey) {
260 return clusterArray[clusterKey];
263 synchronized ClusterImpl makeProxy(ClusterUID clusterUID, long clusterId) {
264 if (clusterUID.second != clusterId)
265 throw new RuntimeDatabaseException("Illegal id for cluster=" + clusterUID + " id=" + clusterId);
266 return clusters.makeProxy(clusterUID);
269 synchronized ClusterImpl makeCluster(long clusterId, boolean writeOnly) {
271 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
274 ClusterUID clusterUID = ClusterUID.make(0, clusterId);
275 int clusterKey = clusters.size(); // new key
277 proxy = new ClusterWriteOnly(clusterUID, clusterKey, sessionImpl);
278 writeOnlyClusters.add(proxy);
279 return clusters.create(proxy);
281 //printMaps("makeCluster");
282 ClusterImpl cluster = ClusterImpl.make(clusterUID, clusterKey, sessionImpl.clusterTranslator);
283 clusters.create(cluster);
284 if (!cluster.isLoaded())
285 LOGGER.error("", new Exception("Bug in ClusterTable.makeCluster(long, boolean), cluster not loaded"));
286 importanceMap.put(cluster.getImportance(), new ImportanceEntry(cluster));
288 validateSize("makeCluster");
289 if(collectorPolicy != null) collectorPolicy.added(cluster);
294 synchronized void replaceCluster(ClusterI cluster_) {
295 ClusterImpl cluster = (ClusterImpl) cluster_;
297 int clusterKey = cluster.getClusterKey();
298 ClusterImpl existing = (ClusterImpl) clusterArray[clusterKey];
299 if (existing.hasVirtual())
300 cluster.markVirtual();
302 if (existing.cc != null) {
303 if (existing.isLoaded()) {
304 // This shall be promoted to actual exception in the future -
305 // for now, minimal changes
306 new Exception("Trying to replace cluster with pending changes " + existing.getClusterUID())
309 // Adopt changes to loaded cluster
310 cluster.cc = existing.cc;
311 cluster.cc.adopt(cluster);
312 cluster.foreignLookup = existing.foreignLookup;
313 cluster.change = existing.change;
317 importanceMap.remove(existing.getImportance());
318 if (collectorPolicy != null)
319 collectorPolicy.removed((ClusterImpl)existing);
321 //System.out.println("ClusterTable.replace(" + existing + " (I=" + existing.getImportance() + ") => " + cluster + " (I=" + cluster.getImportance() + ")");
323 clusters.replace((ClusterImpl)cluster);
324 if (!cluster.isLoaded())
325 LOGGER.error("", new Exception("Bug in ClusterTable.replaceCluster(ClusterI), cluster not loaded"));
327 importanceMap.put(cluster.getImportance(), new ImportanceEntry((ClusterImpl)cluster));
328 if(collectorPolicy != null) collectorPolicy.added((ClusterImpl)cluster);
330 if (!dirtySizeInBytes) {
331 if (existing != cluster) {
332 adjustCachedSize(-existing.getCachedSize(), existing);
334 // This will update sizeInBytes through adjustCachedSize
335 cluster.getCachedSize();
338 validateSize("replaceCluster");
341 synchronized void release(CollectorCluster cluster) {
342 importanceMap.remove(cluster.getImportance());
343 release(cluster.getClusterId());
346 synchronized void release(long clusterId) {
347 //System.out.println("ClusterTable.release(" + clusterId + "): " + sizeInBytes);
349 ClusterImpl clusterImpl = clusters.getClusterByClusterId(clusterId);
350 if (null == clusterImpl)
352 if(!clusterImpl.isLoaded() || clusterImpl.isEmpty())
354 //printMaps("release");
355 clusters.freeProxy(clusterImpl);
356 importanceMap.remove(clusterImpl.getImportance());
357 if (collectorPolicy != null)
358 collectorPolicy.removed(clusterImpl);
359 if (sessionImpl.writeState != null)
360 sessionImpl.clusterStream.flush(clusterImpl.clusterUID);
361 if (!dirtySizeInBytes) {
362 adjustCachedSize(-clusterImpl.getCachedSize(), clusterImpl);
365 validateSize("release");
368 synchronized void compact(long id) {
369 ClusterI impl = clusters.getClusterByClusterId(id);
375 // cachedSize = getSizeInBytes();
378 double getLoadProbability() {
379 // This can currently cause stack overflow
381 // if(cachedSize < SLOW_LIMIT) return 1.0;
382 // if(cachedSize > HIGH_LIMIT) return 1e-2;
384 // double pos = (double)(cachedSize - SLOW_LIMIT) / (double)(HIGH_LIMIT -
386 // double val = 0.1 * ((pos-1) * (pos-1)) + 1e-2;
390 class SizeProcedure implements TObjectProcedure<ClusterImpl> {
391 public long result = 0;
394 public boolean execute(ClusterImpl cluster) {
395 if (cluster != null) {
397 if (cluster.isLoaded() && !cluster.isEmpty()) {
398 result += cluster.getCachedSize();
400 } catch (Throwable t) {
401 LOGGER.error("Could not calculate size", t);
407 public void clear() {
413 private SizeProcedure sizeProcedure = new SizeProcedure();
414 long getSizeInBytes() {
415 if (dirtySizeInBytes) {
416 sizeProcedure.clear();
417 clusters.forEachValue(sizeProcedure);
418 sizeInBytes = sizeProcedure.result;
419 // System.err.println("recomputed size of clusterTable => " + sizeInBytes);
420 setDirtySizeInBytes(false);
425 public void setDirtySizeInBytes(boolean value) {
426 dirtySizeInBytes = value;
429 ClusterStateImpl getState() {
430 final ClusterStateImpl result = new ClusterStateImpl();
431 clusters.forEachEntry(new TLongObjectProcedure<ClusterImpl>() {
433 public boolean execute(long arg0, ClusterImpl arg1) {
436 if (arg1.isLoaded() && !arg1.isEmpty()) {
437 result.ids.add(new ImportanceEntry(arg1));
445 void restoreState(ClusterStateImpl state) {
446 ClusterStateImpl current = getState();
447 for (CollectorCluster id : current.ids)
448 if (!state.ids.contains(id))
449 collectorSupport.release(id);
452 class ClusterCollectorImpl implements ClusterCollector {
454 final private ClusterCollectorSupport support;
455 private ClusterCollectorPolicy policy;
456 ClusterCollectorImpl(ClusterCollectorSupport support) {
457 this.support = support;
460 ClusterCollectorPolicy setPolicy(ClusterCollectorPolicy newPolicy) {
462 ClusterCollectorPolicy oldPolicy = policy;
466 for (CollectorCluster id : support.getResidentClusters()) {
467 policy.added(getClusterByClusterId(id.getClusterId()));
471 support.setPolicy(policy);
478 public void collect() {
482 release(policy.select());
486 int size = support.getCurrentSize();
487 boolean dynamicAllocation = useDynamicAllocation();
488 if (dynamicAllocation)
489 setAllocateSize(estimateProperAllocation(support));
490 if (DebugPolicy.CLUSTER_COLLECTION) {
491 System.out.println("Cluster collector activated, current size = " + size + " limit = " + maximumBytes);
493 if (dynamicAllocation) {
494 int collectSize = maximumBytes / 2;
495 collectSize = Math.min(collectSize, 32 * 1024 * 1024);
496 // try to keep allocated clusters below the maximum
497 if (maximumBytes - size > collectSize)
499 collectSize += size - maximumBytes;
500 collect(collectSize);
502 // try to keep allocated clusters below the maximum
503 if (size < maximumBytes)
513 private boolean useDynamicAllocation() {
514 return "true".equalsIgnoreCase(System.getProperty("org.simantics.db.cluster.dynamicAlloc"));
518 public void collect(int target) {
522 release(policy.select(target));
526 ArrayList<CollectorCluster> toRelease = new ArrayList<CollectorCluster>();
528 for (CollectorCluster cluster : support.getResidentClusters()) {
529 target -= support.getClusterSize(cluster);
531 toRelease.add(cluster);
539 if (DebugPolicy.CLUSTER_COLLECTION) {
540 System.out.println("Cluster collector finished, current size = " + support.getCurrentSize());
547 void release(Collection<CollectorCluster> toRelease) {
548 for (CollectorCluster id : toRelease) {
555 private ClusterCollectorSupport collectorSupport = new ClusterCollectorSupportImpl(this);
556 ClusterCollectorImpl collector = new ClusterCollectorImpl(collectorSupport);
562 private long newResourceClusterId = Constants.NewClusterId;
563 public static final int CLUSTER_FILL_SIZE = ClusterTraitsBase.getMaxNumberOfResources();
566 * Uusi id varataan vasta, kun lis�t��n resurssi => reservedIds sis�lt��
567 * vain jo k�yt�ss� olevia klustereita
570 ClusterImpl getNewResourceCluster(ClusterSupport cs, GraphSession graphSession, boolean writeOnly)
571 throws DatabaseException {
572 ClusterImpl result = null;
573 if (Constants.NewClusterId == newResourceClusterId) {
574 newResourceClusterId = graphSession.newClusterId();
575 result = getClusterByClusterIdOrMake(newResourceClusterId, writeOnly);
577 ClusterImpl cluster = getClusterByClusterIdOrThrow(newResourceClusterId);
578 if (cluster.getNumberOfResources(cs) >= CLUSTER_FILL_SIZE) {
579 newResourceClusterId = graphSession.newClusterId();
580 cluster = getClusterByClusterIdOrMake(newResourceClusterId, writeOnly);
584 return ensureLoaded(result);
587 void flushCluster(GraphSession graphSession) {
588 // We seem to disagree about this.
589 // graphSession.newClusterId();
590 newResourceClusterId = Constants.NewClusterId;
593 void writeOnlyInvalidate(ClusterI impl) {
594 writeOnlyInvalidates.add(impl.getClusterKey());
597 void removeWriteOnlyClusters() {
598 for (ClusterI proxy : writeOnlyClusters) {
599 if (!(proxy instanceof ClusterImpl))
600 throw new RuntimeDatabaseException("ClusterTable corrupted.");
601 clusters.freeProxy((ClusterImpl)proxy);
603 writeOnlyClusters.clear();
604 writeOnlyInvalidates.forEach(new TIntProcedure() {
606 public boolean execute(int clusterKey) {
607 ClusterImpl proxy = clusterArray[clusterKey];
608 ClusterUID clusterUID = proxy.getClusterUID();
609 clusters.freeProxy(proxy);
613 writeOnlyInvalidates.clear();
616 public ClusterImpl getClusterByClusterUID(ClusterUID clusterUID) {
617 synchronized (this) {
618 return clusters.getClusterByClusterUID(clusterUID);
621 public int getClusterKeyByClusterUIDOrMakeProxy(ClusterUID clusterUID) {
622 return getClusterKeyByClusterUIDOrMakeProxy(0/*clusterUID.first*/, clusterUID.second);
624 public int getClusterKeyByClusterUIDOrMakeProxy(long id1, long id2) {
625 return getClusterByClusterUIDOrMakeProxy(id1, id2).clusterKey;
627 public ClusterImpl getClusterByClusterUIDOrMakeProxy(ClusterUID clusterUID) {
628 synchronized (this) {
629 ClusterImpl clusterImpl = clusters.getClusterByClusterUID(clusterUID);
630 if (null == clusterImpl)
631 clusterImpl = clusters.makeProxy(clusterUID);
635 public ClusterImpl getClusterByClusterUIDOrMakeProxy(long id1, long id2) {
636 synchronized (this) {
637 ClusterImpl clusterImpl = clusters.getClusterByClusterUID(id1, id2);
638 if (null == clusterImpl)
639 clusterImpl = clusters.makeProxy(id2);
643 ClusterImpl getLoadOrThrow(long clusterId) throws DatabaseException {
644 synchronized (this) {
645 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
648 if (proxy.isLoaded())
651 clusterKey = proxy.getClusterKey();
653 if (clusterKey == 0) {
654 proxy = clusters.makeProxy(clusterId);
655 clusterKey = proxy.getClusterKey();
657 ClusterImpl ci = tryLoad(clusterId, clusterKey);
659 throw new ResourceNotFoundException(clusterId);
661 } catch (ClusterDoesNotExistException t) {
662 clusters.removeProxy(proxy);
668 ClusterImpl getClusterByClusterIdOrMake(long clusterId, boolean writeOnly) {
669 synchronized (this) {
670 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
674 return makeCluster(clusterId, writeOnly);
677 ClusterImpl getClusterByClusterIdOrThrow(long clusterId) {
678 synchronized (this) {
679 ClusterImpl proxy = clusters.getClusterByClusterId(clusterId);
681 throw new IllegalArgumentException("Cluster id=" + clusterId + " is not created.");
685 long getClusterIdOrCreate(ClusterUID clusterUID) {
686 return clusterUID.second;
688 final long getClusterIdByResourceKey(final int resourceKey)
689 throws DatabaseException {
690 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKey(resourceKey);
691 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
692 throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
693 ClusterI c = clusterArray[clusterKey];
695 throw new RuntimeException("No cluster for key " + resourceKey);
696 return c.getClusterId();
698 final long getClusterIdByResourceKeyNoThrow(final int resourceKey) {
699 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
700 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey)) {
701 LOGGER.error("Tried to get a persistent cluster for a virtual resource. key=" + resourceKey);
704 ClusterI c = clusterArray[clusterKey];
706 LOGGER.error("No cluster for key " + resourceKey);
709 return c.getClusterId();
714 void refresh(long csid, SessionImplSocket session, ClusterUID[] clusterUID) {
715 synchronized (this) {
716 session.flushCounter = 0;
717 session.clusterStream.reallyFlush();
718 ClientChangesImpl cs = new ClientChangesImpl(session);
719 if (session.clientChanges == null)
720 session.clientChanges = cs;
721 //printMaps("refresh");
722 for (int i=0; i<clusterUID.length; ++i) {
724 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
725 System.err.println("cluster=" + clusterUID[i] + " has changed. length=" + clusterUID.length);
726 ClusterImpl oldCluster = clusters.getClusterByClusterUID(clusterUID[i]);
727 if (null == oldCluster)
729 if (!oldCluster.isLoaded())
731 int clusterKey = oldCluster.getClusterKey();
732 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
734 boolean big = oldCluster instanceof ClusterBig;
735 boolean small = oldCluster instanceof ClusterSmall;
738 // ClusterImpl newCluster = (ClusterImpl) sessionImpl.graphSession.getClusterImpl(clusterUID[i], clusterKey);
739 Database.Session dbSession = sessionImpl.graphSession.dbSession;
741 ClusterImpl newCluster = dbSession.clone(clusterUID[i], new ClusterCreator() {
744 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
745 ClusterSupport support = sessionImpl.clusterTranslator;
747 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
748 } catch (DatabaseException e) {
755 if (null == newCluster)
757 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
758 System.err.println("cluster=" + newCluster + " updated.");
759 importanceMap.remove(oldCluster.getImportance());
760 if (collectorPolicy != null)
761 collectorPolicy.removed(oldCluster);
762 clusters.replace(newCluster);
763 if (!newCluster.isLoaded())
764 LOGGER.error("", new Exception("Bug in ClusterTable.refresh, cluster not loaded"));
765 importanceMap.put(newCluster.getImportance(), new ImportanceEntry(newCluster));
766 if (collectorPolicy != null)
767 collectorPolicy.added(newCluster);
768 // Now we have fetched the new cluster but to emulate effects of the changes in it we fetch the cluster changes from server.
769 refreshCluster(csid, session, oldCluster, clusterUID[i], newCluster.clusterId, clusterKey);
770 } catch (Throwable t) {
771 LOGGER.error("Failed to load cluster in refresh.", t);
775 validateSize("refresh");
776 // Fake update of cluster changes.
777 QueryProcessor queryProcessor = session.getQueryProvider2();
778 WriteGraphImpl writer = WriteGraphImpl.create(queryProcessor, session.writeSupport, null);
779 TaskHelper th = null;
780 if (null == session.writeState) {
781 th = new TaskHelper("Refresh");
782 session.writeState = new WriteState<Object>(writer, th.writeTraits, th.sema, th.proc);
784 session.getQueryProvider2().propagateChangesInQueryCache(writer);
785 session.fireMetadataListeners(writer, cs);
786 session.getQueryProvider2().listening.fireListeners(writer);
787 session.fireReactionsToSynchronize(cs);
788 session.fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
789 session.printDiagnostics();
792 session.writeState = null;
798 final void refreshCluster(long csid, SessionImplSocket session, ClusterImpl cluster, ClusterUID clusterUID, long clusterId, int clusterKey)
799 throws DatabaseException {
800 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
801 System.err.println("cluster=" + clusterUID + " id=" + clusterId + " key=" + clusterKey + " resources will be updated.");
802 // get cluster change sets
803 QueryProcessor queryProcessor = session.getQueryProvider2();
806 cc = session.graphSession.getClusterChanges(clusterUID, csid);
807 } catch (Exception e) {
808 LOGGER.error("Could not get cluster changes. cluster=" + clusterUID, e);
812 for (int i=0; i<cc.getResourceIndex().length; ++i) {
813 int resource = ClusterTraits.createResourceKey(clusterKey, cc.getResourceIndex()[i]);
814 ClusterUID pClusterUID = new ClusterUID(cc.getPredicateFirst()[i], cc.getPredicateSecond()[i]);
815 ClusterImpl pCluster = clusters.getClusterByClusterUID(pClusterUID);
816 if (null == pCluster)
818 int pClusterKey = pCluster.getClusterKey();
819 int predicate = ClusterTraits.createResourceKey(pClusterKey, cc.getPredicateIndex()[i]);
820 queryProcessor.updateStatements(resource, predicate);
821 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
822 System.err.println("resource " + cc.getResourceIndex()[i] + " relation " + cc.getPredicateIndex()[i] + " changed.");
824 for (int i=0; i<cc.getValueIndex().length; ++i) {
825 int resource = ClusterTraits.createResourceKey(clusterKey, cc.getValueIndex()[i]);
826 queryProcessor.updateValue(resource);
827 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
828 System.err.println("value " + cc.getValueIndex()[i] + " changed.");
831 final synchronized void refreshImportance(ClusterImpl c) {
836 //printMaps("refreshImportance");
838 importanceMap.remove(c.getImportance());
839 if(collectorPolicy != null) collectorPolicy.removed(c);
841 long newImportance = timeCounter();
842 // System.err.println("refreshImportance " + c.getClusterId() + " => " + newImportance);
843 c.setImportance(newImportance);
845 LOGGER.error("", new Exception("Bug in ClusterTable.refreshImportance(ClusterImpl), cluster not loaded"));
847 importanceMap.put(c.getImportance(), new ImportanceEntry(c));
848 if(collectorPolicy != null) collectorPolicy.added(c);
850 validateSize("refreshImportance");
854 static long loadTime = 0;
856 @SuppressWarnings("unchecked")
857 public final <T extends ClusterI> T getClusterProxyByResourceKey(final int resourceKey) {
858 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
859 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
860 throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
861 ClusterI cluster = clusterArray[clusterKey];
863 throw new RuntimeException("No proxy for existing cluster. Resource key = " + resourceKey);
867 TLongIntHashMap clusterLoadHistogram = new TLongIntHashMap();
868 int clusterLoadCounter = 0;
870 private <T extends ClusterI> T ensureLoaded(T c) {
872 ClusterImpl cs = (ClusterImpl) c;
874 if(DebugPolicy.REPORT_CLUSTER_LOADING) {
875 long start = System.nanoTime();
876 cluster = load2(cs.getClusterId(), cs.getClusterKey());
877 long load = System.nanoTime()-start;
879 if(DebugPolicy.REPORT_CLUSTER_LOADING) {
880 int was = clusterLoadHistogram.get(cluster.getClusterId());
881 clusterLoadHistogram.put(cluster.getClusterId(), was+1);
882 clusterLoadCounter++;
883 String text = "Load2 " + cluster + " " + 1e-9*loadTime + "s. " + 1e-6*load + "ms. " + clusterLoadCounter + " " + was + " " + cluster.getUsedSpace() + " " + cluster.getImportance();
884 if(DebugPolicy.REPORT_CLUSTER_LOADING_STACKS) {
885 new Exception(text).printStackTrace();
887 System.err.println(text);
891 cluster = load2(cs.getClusterId(), cs.getClusterKey());
893 } catch (DatabaseException e) {
894 LOGGER.error("Could not load cluster", e);
895 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
897 String msg = "Failed to load cluster " + cs.getClusterUID();// + " resourceId=" + (((cs.getClusterId() << 16 + (resourceKey & 65535))));
898 // TODO: this jams the system => needs refactoring.
899 throw new RuntimeDatabaseException(msg, e);
904 @SuppressWarnings("unchecked")
905 public synchronized final <T extends ClusterI> T getClusterByResourceKey(final int resourceKey) {
906 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
907 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
908 throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
909 ClusterI c = clusterArray[clusterKey];
913 if ((counter++ & 4095) == 0)
914 refreshImportance((ClusterImpl) c);
917 if (!(c instanceof ClusterSmall)) {
918 LOGGER.error("Proxy must be instance of ClusterSmall");
921 return ensureLoaded((T)c);
924 @SuppressWarnings("unchecked")
925 final <T extends ClusterI> T checkedGetClusterByResourceKey(final int resourceKey) {
926 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(resourceKey);
927 if (ClusterTraitsBase.isVirtualClusterKey(clusterKey))
928 throw new RuntimeException("Tried to get a persistent cluster for a virtual resource.");
929 ClusterI c = clusterArray[clusterKey];
931 throw new RuntimeException("No cluster for resource key " + resourceKey);
933 if ((counter++ & 4095) == 0)
934 refreshImportance((ClusterImpl) c);
937 if (!(c instanceof ClusterSmall)) {
938 LOGGER.error("Proxy must be instance of ClusterSmall");
942 ClusterSmall cs = (ClusterSmall) c;
944 // System.err.println("Load2 " + resourceKey);
945 cluster = load2(cs.getClusterId(), cs.getClusterKey());
946 } catch (DatabaseException e) {
947 if (DebugPolicy.REPORT_CLUSTER_EVENTS)
949 int resourceIndex = resourceKey & ClusterTraitsBase.getResourceIndexFromResourceKeyNoThrow(resourceKey);
950 long resourceId = ClusterTraitsBase.createResourceIdNoThrow(cs.getClusterId(), resourceIndex);
951 String msg = "Failed to load cluster " + cs.getClusterUID() + " for resource key " + resourceKey + " resourceIndex=" + resourceIndex + " resourceId=" + resourceId;
952 LOGGER.error(msg, e);
953 c.setDeleted(true, null);
959 void printDebugInfo(ClusterSupport support)
960 throws DatabaseException {
961 final int SIZE = clusters.size();
963 for (int i = 1; i < SIZE; ++i) {
964 ClusterI c = clusterArray[i];
965 c.getNumberOfResources(support);
966 long size = c.getUsedSpace();
967 System.out.println("cluster=" + c.getClusterId() + " size=" + size);
968 c.printDebugInfo("koss: ", support);
971 System.out.println("Total number of clusters " + SIZE);
972 System.out.println("Total cluster size " + sum);
975 Map<Long,ClusterImpl> prefetch = new HashMap<Long,ClusterImpl>();
977 public synchronized ClusterImpl load2(long clusterId, int clusterKey) throws DatabaseException {
978 ClusterImpl curr = (ClusterImpl) clusterArray[clusterKey];
984 // curr = (ClusterImpl) clusterArray[clusterKey];
985 // if (curr.isLoaded())
988 final Semaphore s = new Semaphore(0);
989 final DatabaseException[] ex = new DatabaseException[1];
991 load2(clusterId, clusterKey, e -> {
997 } catch (InterruptedException e) {
998 LOGGER.error("unable to acquire", e);
1002 if (Development.DEVELOPMENT) {
1003 if (Development.<Boolean>getProperty(DevelopmentKeys.CLUSTERTABLE_VALIDATE_ON_LOAD, Bindings.BOOLEAN)) {
1005 ClusterImpl loaded = (ClusterImpl) clusterArray[clusterKey];
1006 if (loaded instanceof ClusterSmall)
1007 ((ClusterSmall) loaded).check();
1008 else if (loaded instanceof ClusterBig)
1009 ((ClusterBig) loaded).check();
1010 } catch (Throwable t) {
1011 t.printStackTrace();
1016 validate(clusterKey);
1018 return (ClusterImpl) clusterArray[clusterKey];
1021 void validate(int clusterKey) {
1023 // if(sessionImpl.graphSession.graphClient instanceof GraphClientImpl2) {
1025 // ClusterImpl server = (ClusterImpl)clusterArray[clusterKey];
1026 // String sd = server.dump(sessionImpl.clusterTranslator);
1027 // GraphClientImpl2 gci = ((GraphClientImpl2)sessionImpl.graphSession.graphClient);
1028 // ClusterImpl memory = gci.clusters.get(server.clusterUID);
1029 // if(memory == null)
1030 // System.err.println("ad");
1031 // String md = memory.dump(gci.support);
1033 // if(!sd.equals(md)) {
1036 // int minLength = Math.min(sd.length(), md.length());
1037 // for (int i = 0; i < minLength; i++) {
1038 // if (sd.charAt(i) != md.charAt(i)) {
1044 // int start = Math.max(diffPos-200, 0);
1045 // int end = Math.min(minLength-1, diffPos+200);
1047 // System.err.println("== CLUSTER DIFFERENCE " + clusterArray[clusterKey].getClusterUID() + " == ");
1048 // System.err.println("== SESSION == ");
1049 // System.err.println(sd.substring(start, end));
1050 // System.err.println("== MEM == ");
1051 // System.err.println(md.substring(start, end));
1052 // System.err.println("== CLUSTER DIFFERENCE ENDS == ");
1054 // throw new IllegalStateException();
1060 public synchronized ClusterImpl getPrefetched() {
1061 synchronized(prefetch) {
1066 public synchronized void load2(long clusterId, int clusterKey, final Consumer<DatabaseException> runnable) {
1068 assert (Constants.ReservedClusterId != clusterId);
1070 ClusterImpl cluster = null;
1071 DatabaseException e = null;
1075 ClusterUID clusterUID = clusters.makeClusterUID(clusterId);
1076 // cluster = (ClusterImpl) sessionImpl.getGraphSession().getClusterImpl(clusterUID, clusterKey);
1077 Database.Session session = sessionImpl.graphSession.dbSession;
1078 // cluster = (ClusterImpl)gci.getClusterByClusterKey(clusterKey).clone(sessionImpl.clusterTranslator);
1079 // cluster = (ClusterImpl)((ClusterImpl)gci.getClusterByClusterUIDOrMakeProxy(clusterUID)).clone(sessionImpl.clusterTranslator);
1081 // cluster = gci.clone(clusterUID, sessionImpl.clusterTranslator);
1083 cluster = session.clone(clusterUID, new ClusterCreator() {
1086 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
1087 ClusterSupport support = sessionImpl.clusterTranslator;
1089 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
1090 } catch (DatabaseException e) {
1091 e.printStackTrace();
1098 } catch (Throwable t) {
1099 // It's totally legal to call for non-existing cluster.
1100 // Sometimes this indicates an error, though.
1101 LOGGER.error("Load cluster failed", t);
1102 if (t instanceof DatabaseException)
1103 e = (DatabaseException) t;
1105 e = new DatabaseException("Load cluster failed.", t);
1107 if (null == cluster) {
1112 // new Exception("Load cluster " + cluster.getClusterId() + " " + cluster.getCachedSize() + " " + cluster.getImmutable()).printStackTrace();
1114 // Can not be called with null argument.
1115 replaceCluster(cluster);
1116 sessionImpl.onClusterLoaded(clusterId);
1117 runnable.accept(null);
1121 public synchronized ClusterImpl tryLoad(long clusterId, int clusterKey) throws DatabaseException {
1123 assert (Constants.ReservedClusterId != clusterId);
1125 ClusterUID clusterUID = clusters.makeClusterUID(clusterId);
1127 Database.Session session = sessionImpl.graphSession.dbSession;
1129 ClusterImpl cluster = session.clone(clusterUID, new ClusterCreator() {
1132 public <T> T create(ClusterUID uid, byte[] bytes, int[] ints, long[] longs) {
1133 ClusterSupport support = sessionImpl.clusterTranslator;
1135 return (T)ClusterImpl.make(longs, ints, bytes, support, support.getClusterKeyByClusterUIDOrMake(uid));
1136 } catch (DatabaseException e) {
1137 e.printStackTrace();
1143 if (null == clusterArray[clusterKey])
1144 clusterArray[clusterKey] = cluster;
1146 replaceCluster(cluster);
1147 sessionImpl.onClusterLoaded(clusterId);
1149 validate(clusterKey);
1154 public Collection<ClusterI> getClusters() {
1155 ArrayList<ClusterI> result = new ArrayList<ClusterI>();
1156 for (int i = 0; i < clusterArray.length; i++) {
1157 ClusterI cluster = clusterArray[i];
1158 if (cluster != null)
1159 result.add(cluster);
1165 return clusters.size();
1168 public long timeCounter() {
1169 return timeCounter.getAndIncrement();
1172 public boolean hasVirtual(int index) {
1173 return virtuals[index];
1176 public void markVirtual(int index) {
1177 virtuals[index] = true;
1180 public ClusterImpl[] getClusterArray() {
1181 return clusterArray;
1184 public boolean isImmutable(int id) {
1186 int clusterKey = ClusterTraitsBase.getClusterKeyFromResourceKeyNoThrow(id);
1187 Boolean exist = immutables[clusterKey];
1188 if(exist != null) return exist;
1190 ClusterI cluster = getClusterByResourceKey(id);
1191 if(cluster == null) {
1194 boolean result = cluster.getImmutable();
1195 markImmutable(cluster, result);
1201 public void markImmutable(ClusterI cluster, boolean value) {
1202 immutables[cluster.getClusterKey()] = value;
1205 public int getClusterKeyByUID(long first, long second) throws DatabaseException {
1206 return getClusterKeyByClusterUIDOrMakeProxy(ClusterUID.make(first, second));
1209 public void adjustCachedSize(long l, ClusterI cluster) {
1211 //System.out.println("ClusterTable: adjusting cluster table cached size by " + l + ": " + sizeInBytes + " -> "
1212 // + (sizeInBytes + l) + ", for cluster " + cluster.getClusterId() + " (" + cluster + ")");
1217 private void validateSize(String place) {
1221 int ims = importanceMap.size();
1222 int ihms = countImportantClusters();
1224 // System.out.format("[ClusterTable.%s] Validating: byteSize=%d (%d MB), hashMap=%d/%d, importanceMap=%d%n",
1225 // place, sizeInBytes, sizeInBytes / (1024*1024), ihms, clusters.hashMap.size(), ims);
1227 int i = clusterArray.length;
1229 for (int j = 0; j < i; ++j) {
1230 ClusterI c = clusterArray[j];
1233 size += c.getCachedSize();
1235 if (sizeInBytes != size) {
1236 if (!dirtySizeInBytes)
1237 System.out.println("BUG: CACHED CLUSTER SIZE DIFFERS FROM CALCULATED: " + sizeInBytes + " != " + size + ", delta = " + (sizeInBytes - size));
1238 //else System.out.println("\"BUG?\": SIZES DIFFER: " + sizeInBytes + " != " + size + ", delta = " + (sizeInBytes - size));
1241 System.out.println("BUG2: hashmap and importanceMap sizes differ: " + ihms + " != " + ims + ", delta=" + (ihms - ims));
1242 printMaps("validateSize");
1244 //System.out.println("[" + place + "] VALIDATED");
1247 private void printMaps(String place) {
1248 int ihms = countImportantClusters();
1249 System.out.println("## printMaps(" + place + ") - " + importanceMap.size() + " - (" + ihms + "/" + clusters.hashMap.size() + ")");
1250 System.out.println("importanceMap (" + importanceMap.size() + "):");
1251 importanceMap.forEach((importance, ie) -> {
1252 System.out.format("\t%d: %s%n", importance, ie.toString());
1254 System.out.println("clusters.hashMap (" + ihms + "/" + clusters.hashMap.size() + "):");
1255 clusters.hashMap.forEachEntry((cid, c) -> {
1256 boolean important = importantCluster(c);
1257 boolean wo = c != null && c.isWriteOnly();
1258 boolean empty = c != null && c.isEmpty();
1259 boolean loaded = c != null && c.isLoaded();
1260 System.out.format("\t%s: %d - %s (writeOnly=%b, empty=%b, loaded=%b)%n", important ? " I" : "NI", cid, c, wo, empty, loaded);
1265 private int countImportantClusters() {
1266 int[] result = { 0 };
1267 clusters.hashMap.forEachEntry((cid, c) -> {
1268 if (importantCluster(c))
1275 private static boolean importantCluster(ClusterImpl c) {
1276 return c != null && !c.isWriteOnly() && c.isLoaded();// && !c.isEmpty();