Source code

001package com.hammurapi.store.local;
002
003import java.util.Collection;
004import java.util.Comparator;
005import java.util.Iterator;
006import java.util.LinkedList;
007import java.util.Map;
008import java.util.WeakHashMap;
009import java.util.concurrent.ConcurrentHashMap;
010import java.util.concurrent.ExecutorService;
011import java.util.concurrent.atomic.AtomicLong;
012import java.util.concurrent.locks.Lock;
013import java.util.concurrent.locks.ReadWriteLock;
014import java.util.concurrent.locks.ReentrantReadWriteLock;
015
016import com.hammurapi.common.ObservableConverter;
017import com.hammurapi.common.concurrent.CallerThreadExecutorService;
018import com.hammurapi.extract.Extractor;
019import com.hammurapi.extract.Predicate;
020import com.hammurapi.store.AbstractIndex;
021import com.hammurapi.store.AbstractStore;
022import com.hammurapi.store.DeputyReadWriteLock;
023import com.hammurapi.store.Index.Type;
024import com.hammurapi.store.Store;
025import com.hammurapi.store.StoreException;
026import com.hammurapi.store.TrackingLock;
027import com.hammurapi.store.local.LocalHandle.HandleStrength;
028
029/**
030 * In-memory object store.
031 * @author Pavel Vlasov
032 *
033 * @param <T>
034 */
035public abstract class LocalStoreBase<T, PK, S extends Store<T,PK,S>> extends AbstractStore<T, PK, S> {
036        
037        /**
038         * Store configuration.
039         * @author Pavel Vlasov
040         *
041         */
042        public static class Config<T,PK,S extends Store<T,PK,S>> implements Cloneable {
043                
044                private final HandleFactory<T,PK,S,LocalHandle<T,PK,S>> DEFAULT_HANDLE_FACTORY = new HandleFactory<T, PK, S, LocalHandle<T,PK,S>>() {
045
046                        @Override
047                        public LocalHandle<T, PK, S> createHandle(
048                                        S store,
049                                        T obj,
050                                        PK primaryKey,
051                                        Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache,
052                                        Predicate<T, S>[] validators,
053                                        HandleStrength handleStrength,
054                                        boolean cacheExtracted) {
055                                return new LocalHandle<T,PK,S>((LocalStoreBase<T, PK, S>) store, obj, primaryKey, cache, validators, handleStrength, cacheExtracted);
056                        }
057                        
058                };
059                
060                private long cleanupTaskThreshold = CLEANUP_TASK_THRESHOLD;             
061                private ExecutorService executorService;
062                private Extractor<T, PK, S> primaryKeyExtractor;
063                private boolean cacheExtracted = true;
064                private HandleStrength handleStrength = HandleStrength.STRONG;
065                private boolean weakPrimaryKey;
066                private Map<PK, Handle<T,PK,S>> pkStore;
067                private Collection<Handle<T,PK,S>> noPkStore;
068                private HandleFactory<T, PK, S, LocalHandle<T,PK,S>> handleFactory = DEFAULT_HANDLE_FACTORY;
069                private ObservableConverter<T> observableConverter;
070//              private TimeUnit statsTimeUnit;
071//              private boolean collectStats;
072                
073                // TODO Asynch index exception listener.
074//              BackingStore<T, PK> backingStore; 
075//              int writeBehindBuffer;
076//              long writeBehindDelay;
077//              TimeUnit writeBehindTimeUnit;
078                
079//              /**
080//               * If time unit is set, store collects execution time statistics.
081//               */
082//              public void setStatsTimeUnit(TimeUnit statsTimeUnit) {
083//                      this.statsTimeUnit = statsTimeUnit;
084//              }
085//              
086//              public TimeUnit getStatsTimeUnit() {
087//                      return statsTimeUnit;
088//              }
089//              
090//              /**
091//               * If set to true, store collects execution statistics.
092//               * @param collectStats
093//               */
094//              public void setCollectStats(boolean collectStats) {
095//                      this.collectStats = collectStats;
096//              }
097//              
098//              public boolean isCollectStats() {
099//                      return collectStats;
100//              }
101                
102                /**
103                 * @param cleanupTaskThreshold When number of invalidated (removed) handles exceeds this
104                 * threshold, the store performs cleanup of internal collections.
105                 */
106                public void setCleanupTaskThreshold(long cleanupTaskThreshold) {
107                        this.cleanupTaskThreshold = cleanupTaskThreshold;
108                }
109                
110                /**
111                 * @return Cleanup task threshold.
112                 */
113                public long getCleanupTaskThreshold() {
114                        return cleanupTaskThreshold;
115                }
116                
117                @Override
118                public Object clone() throws CloneNotSupportedException {
119                        return super.clone();
120                }
121
122                /**
123                 * @return Executor service. If null, all operations are performed in the caller thread.
124                 */
125                public ExecutorService getExecutorService() {
126                        return executorService;
127                }
128
129                /**
130                 * 
131                 * @param executorService Executor service. If null, all operations are performed in the caller thread.
132                 */
133                public void setExecutorService(ExecutorService executorService) {
134                        this.executorService = executorService;
135                }
136
137                
138                public Extractor<T, PK, S> getPrimaryKeyExtractor() {
139                        return primaryKeyExtractor;
140                }
141
142                /**
143                 * @param primaryKeyExtractor Primary key extractor. Can be null.
144                 */
145                public void setPrimaryKeyExtractor(
146                                Extractor<T, PK, S> primaryKeyExtractor) {
147                        this.primaryKeyExtractor = primaryKeyExtractor;
148                }
149
150                public boolean isCacheExtracted() {
151                        return cacheExtracted;
152                }
153
154                /**
155                 * @param cacheExtracted If true, values extracted from stored objects are cached until object is updated 
156                 * through update() or put() methods. Default value is true.
157                 */
158                public void setCacheExtracted(boolean cacheExtracted) {
159                        this.cacheExtracted = cacheExtracted;
160                }
161                
162                public HandleStrength getHandleStrength() {
163                        return handleStrength;
164                }
165
166                /**
167                 * @param handleStrength Handle strength. Default handle strength is HARD.
168                 */
169                public void setHandleStrength(HandleStrength handleStrength) {
170                        this.handleStrength = handleStrength;
171                }
172
173                public boolean isWeakPrimaryKey() {
174                        return weakPrimaryKey;
175                }
176
177                /**
178                 * @param weakStore If true, the store uses WeakHashMap if primary key is set and
179                 * pkStore is not set explicitly.
180                 */
181                public void setWeakPrimaryKey(boolean weakPrimaryKey) {
182                        this.weakPrimaryKey = weakPrimaryKey;
183                }
184
185                public Map<PK, Handle<T, PK, S>> getPkStore() {
186                        return pkStore;
187                }
188
189                /**
190                 * @param pkStore Explicitly provided pkStore instance.
191                 */
192                public void setPkStore(Map<PK, Handle<T, PK, S>> pkStore) {
193                        this.pkStore = pkStore;
194                }
195
196                public Collection<Handle<T, PK, S>> getNoPkStore() {
197                        return noPkStore;
198                }
199
200                /**
201                 * @param noPkStore Explicitly provided noPkStore instance.
202                 */
203                public void setNoPkStore(Collection<Handle<T, PK, S>> noPkStore) {
204                        this.noPkStore = noPkStore;
205                }
206                
207                public void setHandleFactory(HandleFactory<T, PK, S, LocalHandle<T, PK, S>> handleFactory) {
208                        this.handleFactory = handleFactory;
209                }
210                
211                public HandleFactory<T, PK, S, LocalHandle<T, PK, S>> getHandleFactory() {
212                        return handleFactory;
213                }
214                
215                public void setObservableConverter(ObservableConverter<T> observableConverter) {
216                        this.observableConverter = observableConverter;
217                }
218                
219                public ObservableConverter<T> getObservableConverter() {
220                        return observableConverter;
221                }
222        }
223
224        private static final long CLEANUP_TASK_THRESHOLD = 1000;
225        
226        protected Config<T,PK,S> config;
227        
228        @SuppressWarnings("unchecked")
229        protected LocalStoreBase(Config<T,PK,S> config) {
230                try {
231                        this.config = (Config<T, PK, S>) config.clone();
232                } catch (CloneNotSupportedException e) {
233                        throw new StoreException(e);
234                }
235                
236                if (this.config.getExecutorService()==null) {
237                        this.config.setExecutorService(CallerThreadExecutorService.INSTANCE);
238                }
239
240                if (this.config.getPrimaryKeyExtractor()==null) {
241                        if (this.config.getNoPkStore()==null) {
242                                this.config.setNoPkStore(new LinkedList<Handle<T,PK,S>>() { // TODO - Address uniqueness
243                                        
244                                        // Synchronized for situations when updaters perform inserts.
245                                        public synchronized boolean add(Handle<T,PK,S> e) {
246                                                return super.add(e);
247                                        }
248                                });                             
249                        }
250                } else {
251                        if (this.config.getPkStore()==null) {
252                                if (this.config.isWeakPrimaryKey()) {
253                                        this.config.setPkStore(new WeakHashMap<PK, Handle<T,PK,S>>() {
254                                                
255                                                // Synchronized for situations when updaters perform inserts.
256                                                public synchronized Handle<T,PK,S> put(PK key, Handle<T,PK,S> value) {
257                                                        return super.put(key, value);
258                                                };
259                                        
260                                        });
261                                } else {
262                                        this.config.setPkStore(new ConcurrentHashMap<PK, Handle<T,PK,S>>());
263                                }
264                        }
265                }
266                
267        }
268        
269        private ReentrantReadWriteLock storeLock = new ReentrantReadWriteLock();
270        private TrackingLock readLock = new LocalTrackingLock(storeLock.readLock());
271        private TrackingLock writeLock = new LocalTrackingLock(storeLock.writeLock());
272
273        @Override
274        public Lock readLock() {
275                return readLock;
276        }
277
278        @Override
279        public Lock writeLock() {
280                return writeLock;
281        }
282
283        @Override
284        protected Map<PK, Handle<T, PK, S>> getPkStore() {
285                return config.getPkStore();
286        }
287        
288        @Override
289        protected Collection<Handle<T, PK, S>> getNoPkStore() {
290                return getPrimaryKeyExtractor()==null ? config.getNoPkStore() : getPkStore().values();
291        }
292
293        @Override
294        protected ExecutorService getExecutorService() {
295                return config.getExecutorService();
296        }
297        
298
299        @SuppressWarnings("unchecked")
300        @Override
301        protected Handle<T, PK, S> createHandle(
302                        T obj,
303                        PK primaryKey,
304                        Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> cache, 
305                        Predicate<T, S>[] validators) {
306                return config.getHandleFactory().createHandle((S) this, obj, primaryKey, cache, validators, config.getHandleStrength(), config.isCacheExtracted());
307        }
308
309        @Override
310        protected Collection<Handle<T, PK, S>> createIndexCollection(
311                        boolean unique, 
312                        boolean ordered, 
313                        Comparator<T> comparator) {
314                
315                throw new UnsupportedOperationException();
316        }
317
318        @Override
319        protected Map<S, Map<Extractor<T, ? super PK, S>, ? super PK>> createCache() {
320                if (config.isCacheExtracted()) {
321                        return new ConcurrentHashMap<S, Map<Extractor<T, ? super PK, S>, ? super PK>>();
322                }
323                return null;
324        }
325
326        @Override
327        protected ReadWriteLock createMasterLock() {            
328                ReadWriteLock master = new ReadWriteLock() {
329                        
330                        @Override
331                        public Lock writeLock() {
332                                return writeLock;
333                        }
334                        
335                        @Override
336                        public Lock readLock() {
337                                return readLock;
338                        }
339                };
340                return new DeputyReadWriteLock(master, readLock.isLocked() || writeLock.isLocked(), writeLock.isLocked());
341        }
342        
343        private AtomicLong removedCounter = new AtomicLong(0);
344
345        /**
346         * If there is a lot of cleared handles this method posts a task to compact store collections..
347         */
348        @Override
349        protected void onRemoved(int removed) {
350                if (removedCounter.addAndGet(removed)>config.getCleanupTaskThreshold()) {
351                        getExecutorService().submit(new Runnable() {
352
353                                @Override
354                                public void run() {
355                                        writeLock().lock();
356                                        int cleanedUp = 0;
357                                        try {
358                                                Iterator<Handle<T, PK, S>> it = getNoPkStore().iterator(); // Takes care of both map and collection.
359                                                while (it.hasNext()) {
360                                                        Handle<T, PK, S> h = it.next();
361                                                        if (!h.isValid()) {
362                                                                it.remove();
363                                                                ++cleanedUp;
364                                                        }
365                                                }
366                                                
367                                                for (AbstractIndex<T,? extends T,PK,?,S> index: getIndices()) {
368                                                        index.cleanup();
369                                                }
370                                        } finally {
371                                                writeLock().unlock();
372                                                if (cleanedUp>0) {
373                                                        removedCounter.addAndGet(-cleanedUp);
374                                                }
375                                        }
376                                }
377                                
378                        });
379                }
380        }
381
382        @Override
383        public Extractor<T, PK, S> getPrimaryKeyExtractor() {
384                return config.getPrimaryKeyExtractor();
385        }
386        
387        private Collection<AbstractIndex<T, ? extends T, PK, ?, S>> indices = new LinkedList<AbstractIndex<T,? extends T,PK,?,S>>();
388
389        @Override
390        protected Collection<AbstractIndex<T, ? extends T, PK, ?, S>> getIndices() {
391                return indices;
392        }
393        
394        @Override
395        protected <V, ST extends T> AbstractIndex<T, ST, PK, V, S> createIndex(
396                        Predicate<T, S> predicate,
397                        Extractor<ST, V, S> extractor, 
398                        Type type,
399                        boolean ordered, 
400                        Comparator<V> comparator) {
401                
402                return new LocalIndex<T, ST, PK, V, S>(predicate, extractor, type, ordered, comparator, this);
403        }
404 
405        protected void handleIndexAsynchException(Exception e) {
406                e.printStackTrace();            
407        }
408        
409        /**
410         * To make it visible to LocalHandle.
411         */
412        @Override
413        protected void updateIndices(Handle<T, PK, S> handle, boolean isNew) {
414                super.updateIndices(handle, isNew);
415        }
416        
417        @Override
418        protected abstract S createDeputy();
419}