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