Source code

001package com.hammurapi.store;
002
003import java.lang.reflect.InvocationHandler;
004import java.lang.reflect.Method;
005import java.lang.reflect.Proxy;
006import java.util.ArrayList;
007import java.util.Collection;
008import java.util.Collections;
009import java.util.Comparator;
010import java.util.Iterator;
011import java.util.LinkedList;
012import java.util.List;
013import java.util.Map;
014import java.util.concurrent.ExecutorService;
015import java.util.concurrent.atomic.AtomicBoolean;
016import java.util.concurrent.atomic.AtomicInteger;
017import java.util.concurrent.locks.Lock;
018
019import com.hammurapi.extract.Extractor;
020import com.hammurapi.extract.Predicate;
021import com.hammurapi.store.Store.Handle;
022
023public abstract class AbstractIndex<ST, T extends ST, PK, V, S extends Store<ST,PK,S>> implements OrderedIndex<ST, T, PK, V, S> {
024
025        private Predicate<ST, S> predicate;
026        private Extractor<T, V, S> extractor; 
027        protected Index.Type type; 
028        protected boolean ordered;
029        protected Comparator<V> comparator;
030        protected AbstractStore<ST, PK, S> store;
031        protected AtomicBoolean isCorrupt = new AtomicBoolean(false);
032        
033        protected AbstractIndex(
034                        Predicate<ST, S> predicate,
035                        Extractor<T, V, S> extractor, 
036                        Index.Type type, 
037                        boolean ordered, 
038                        Comparator<V> comparator,
039                        AbstractStore<ST,PK,S> store) {
040                this.predicate = predicate;
041                this.extractor = extractor;
042                this.type = type;
043                this.ordered = ordered;
044                this.comparator = comparator;
045                this.store = store;
046        }
047        
048        /**
049         * @return Index unique store. If index is ordered, the index store
050         * shall be of type SortedMap.
051         */
052        protected abstract Map<V, Handle<ST,PK,S>> getIndexUniqueStore();
053        
054        /**
055         * @return Index multi-store. If index is ordered, the index store
056         * shall be of type SortedMap.
057         */
058        protected abstract Map<V, Collection<Handle<ST,PK,S>>> getIndexMultiStore();
059        
060        protected abstract Lock readLock();
061        
062        protected abstract Lock writeLock();
063        
064        /**
065         * Upgrades index type.
066         * @param type Asynchronous and lazy indices can be 
067         * upgraded to synchronous. All index types can be upgraded
068         * to unique if current index data does not contain duplicates.
069         * @param ordered unordered index can be upgraded to ordered.
070         */
071        public abstract void upgrade(Index.Type type, boolean ordered, Comparator<V> comparator);
072        
073        Collection<Handle<ST,PK,S>> getHandles() {
074                checkCorrupt();
075                applyUpdates();
076                try { 
077                        Collection<Handle<ST,PK,S>> ret = new LinkedList<Handle<ST,PK,S>>();
078                        if (isUnique()) {
079                                for (Handle<ST,PK,S> h: getIndexUniqueStore().values()) {
080                                        if (h.isValid()) {
081                                                ret.add(h);
082                                        }
083                                }
084                        } else {
085                                for (Collection<Handle<ST,PK,S>> ch: getIndexMultiStore().values()) {
086                                        for (Handle<ST,PK,S> h: ch) {
087                                                if (h.isValid()) {
088                                                        ret.add(h);
089                                                }
090                                        }
091                                }
092                        }
093                        return ret;
094                } finally {
095                        readLock().unlock();
096                }
097        }
098
099        @SuppressWarnings("unchecked")
100        @Override
101        public Iterator<T> iterator() {
102                checkCorrupt();
103                getStore().readLock().lock();
104                try {
105                        applyUpdates();
106                        try { 
107                                Collection<T> ret = new LinkedList<T>();
108                                if (isUnique()) {
109                                        for (Handle<ST,PK,S> h: getIndexUniqueStore().values()) {
110                                                if (h.isValid()) {
111                                                        ret.add((T) h.get());
112                                                }
113                                        }
114                                } else {
115                                        for (Collection<Handle<ST,PK,S>> ch: getIndexMultiStore().values()) {
116                                                for (Handle<ST,PK,S> h: ch) {
117                                                        if (h.isValid()) {
118                                                                ret.add((T) h.get());
119                                                        }
120                                                }
121                                        }
122                                }
123                                return ret.iterator();
124                        } finally {
125                                readLock().unlock();
126                        }
127                } finally {
128                        getStore().readLock().unlock();
129                }
130        }
131
132        @SuppressWarnings("unchecked")
133        @Override
134        public S getStore() {
135                return (S) store;
136        }
137
138        @Override
139        public Predicate<ST, S> getPredicate() {
140                return predicate;
141        }
142
143        @Override
144        public Extractor<T, V, S> getExtractor() {
145                return extractor;
146        }
147
148        @Override
149        public boolean isUnique() {
150                return Type.UNIQUE.equals(type);
151        }
152
153        @Override
154        public boolean isOrdered() {
155                return ordered;
156        }               
157
158        @Override
159        public Comparator<V> getComparator() {
160                return comparator;
161        }
162        
163        private Comparator<V> naturalComparator = new NaturalComparator<V>();
164        
165        protected abstract Collection<UpdateEntry> getPendingUpdates();
166
167        /**
168         * Updates index
169         * @param changed Changed/added handle.
170         * @param isNew If true, then handle was added, updated otherwise.
171         */
172        public void update(Handle<ST,PK,S> changed, boolean isNew) {
173                checkCorrupt();
174                switch (type) {
175                case UNIQUE:
176                case SYNCHRONOUS:
177                        writeLock().lock();
178                        try {
179                                applyUpdate(changed, isNew);
180                        } finally {
181                                writeLock().unlock();
182                        }
183                        break;
184                case ASYNCHRONOUS:
185                        writeLock().lock();
186                        try {
187                                getPendingUpdates().add(new UpdateEntry(changed, isNew));
188                                getExecutorService().submit(applyUpdatesTask);
189                        } finally {
190                                writeLock().unlock();
191                        }
192                        break;
193                case LAZY:
194                        writeLock().lock();
195                        try {
196                                getPendingUpdates().add(new UpdateEntry(changed, isNew));
197                        } finally {
198                                writeLock().unlock();
199                        }
200                        break;
201                }
202        }
203        
204        protected class UpdateEntry {
205                Handle<ST,PK,S> handle;
206                boolean isNew;
207                
208                public UpdateEntry(Handle<ST, PK, S> handle, boolean isNew) {
209                        super();
210                        this.handle = handle;
211                        this.isNew = isNew;
212                }
213
214                @Override
215                public int hashCode() {
216                        final int prime = 31;
217                        int result = 1;
218                        result = prime * result + ((handle == null) ? 0 : handle.hashCode());
219                        result = prime * result + (isNew ? 1231 : 1237);
220                        return result;
221                }
222
223                @Override
224                public boolean equals(Object obj) {
225                        if (this == obj)
226                                return true;
227                        if (obj == null)
228                                return false;
229                        if (getClass() != obj.getClass())
230                                return false;
231                        UpdateEntry other = (UpdateEntry) obj;
232                        if (handle == null) {
233                                if (other.handle != null)
234                                        return false;
235                        } else if (!handle.equals(other.handle))
236                                return false;
237                        if (isNew != other.isNew)
238                                return false;
239                        return true;
240                }
241
242        }
243        
244        protected abstract ExecutorService getExecutorService();
245        
246        protected void checkCorrupt() {
247                if (isCorrupt.get()) {
248                        throw new StoreException("Index is corrupt");
249                }
250        }
251        
252        boolean isCorrupt() {
253                return isCorrupt.get();
254        }
255        
256        protected Runnable applyUpdatesTask = new Runnable() {
257                
258                public void run() {
259                        writeLock().lock();
260                        try {
261                                Iterator<UpdateEntry> hit = getPendingUpdates().iterator();
262                                while (hit.hasNext()) {
263                                        UpdateEntry updateEntry = hit.next();
264                                        applyUpdate(updateEntry.handle, updateEntry.isNew);
265                                        hit.remove();
266                                }
267                        } catch (Exception e) {
268                                isCorrupt.set(true);
269                                handleAsynchException(e);
270                        } finally {
271                                writeLock().unlock();
272                        }
273                };
274        };
275        
276        
277        /**
278         * Applies pending updates, if any. This method acquires
279         * index write lock. If update operation is successful,
280         * then this method downgrades the lock to read lock.
281         * In the case of exception, write lock is released and
282         * read lock is not acquired.   
283         */
284        private void applyUpdates() {
285                writeLock().lock();
286                try {
287                        Iterator<UpdateEntry> hit = getPendingUpdates().iterator();
288                        while (hit.hasNext()) {
289                                UpdateEntry updateEntry = hit.next();
290                                applyUpdate(updateEntry.handle, updateEntry.isNew);
291                                hit.remove();
292                        }
293                        readLock().lock();
294                } finally {
295                        writeLock().unlock();
296                }
297        }
298        
299        private AtomicInteger granularity = new AtomicInteger();
300        
301        /**
302         * This method is invoked within write lock, no need
303         * to acquire own lock.
304         * @param changed
305         */
306        @SuppressWarnings("unchecked")
307        protected void applyUpdate(Handle<ST,PK,S> changed, boolean isNew) {              
308                if (isUnique()) {
309                        Map<V, Handle<ST, PK, S>> is = getIndexUniqueStore();
310                        if (!isNew) {
311                                is.values().remove(changed);
312                        }
313                        if (predicate==null || changed.extract(predicate)) {
314                                V key = extractor==null ? (V) changed.get() : changed.extract((Extractor<ST, V, S>) extractor);
315                                if (is.containsKey(key)) {
316                                        throw new StoreException("Violation of unique index: "+this);
317                                }
318                                is.put(key, changed);                           
319                        }
320                        granularity.set(is.size());
321                } else {
322                        Map<V, Collection<Handle<ST, PK, S>>> is = getIndexMultiStore();
323                        if (!isNew) {
324                                Iterator<Collection<Handle<ST, PK, S>>> vit = is.values().iterator();
325                                while (vit.hasNext()) {
326                                        Collection<Handle<ST, PK, S>> nextBucket = vit.next();
327                                        if (nextBucket.remove(changed)) {;
328                                                if (nextBucket.isEmpty()) {
329                                                        vit.remove();
330                                                }
331                                        }
332                                }
333                        }
334                        
335                        if (predicate==null || changed.extract(predicate)) {
336                                V key = extractor==null ? (V) changed.get() : changed.extract((Extractor<ST, V, S>) extractor);
337                                Collection<Handle<ST, PK, S>> bucket = is.get(key);
338                                if (bucket == null) {
339                                        bucket = new LinkedList<Store.Handle<ST,PK,S>>();
340                                        is.put(key, bucket);
341                                }
342                                bucket.add(changed);                            
343                        }                       
344                        
345                        granularity.set(getIndexMultiStore().size());                   
346                }
347                
348        }
349        
350        public int getGranularity() {
351                return granularity.get();
352        }
353        
354        /**
355         * Handles exceptions thrown in asynchronous applyUpdates().
356         * @param e
357         */
358        protected abstract void handleAsynchException(Exception e);
359        
360        /**
361         * Removes invalid handles from index store. This method
362         * operates under store's write lock, so it is not necessary
363         * to acquire own lock. 
364         */
365        // TODO - invoke from store cleanup asynchronously.
366        public void cleanup() {
367                if (isUnique()) {
368                        Iterator<Handle<ST, PK, S>> hit = getIndexUniqueStore().values().iterator();
369                        while (hit.hasNext()) {
370                                if (!hit.next().isValid()) {
371                                        hit.remove();
372                                }
373                        }
374                } else {
375                        Iterator<Collection<Handle<ST, PK, S>>> chit = getIndexMultiStore().values().iterator();
376                        while (chit.hasNext()) {
377                                Collection<Handle<ST, PK, S>> ch = chit.next();
378                                Iterator<Handle<ST, PK, S>> hit = ch.iterator();
379                                while (hit.hasNext()) {
380                                        if (!hit.next().isValid()) {
381                                                hit.remove();
382                                        }
383                                }
384                                if (ch.isEmpty()) {
385                                        chit.remove();
386                                }
387                        }
388                }
389        }
390        
391        private Iterable<ST> findMulti(V value) {
392                checkCorrupt();
393                applyUpdates();
394                try {
395                        if (isUnique()) { // For upgraded
396                                Handle<ST, PK, S> ret = getIndexUniqueStore().get(value);
397                                if (ret==null || !ret.isValid()) {
398                                        return Collections.emptyList();
399                                }
400                                return Collections.singleton(ret.get());
401                        }
402                        
403                        Collection<Handle<ST, PK, S>> bucket = getIndexMultiStore().get(value);
404                        if (bucket==null) {
405                                return Collections.emptyList();
406                        }
407                        Collection<ST> ret = new ArrayList<ST>();
408                        for (Handle<ST,PK,S> h: bucket) {
409                                if (h.isValid()) {
410                                        ret.add(h.get());
411                                }
412                        }
413                        return ret;
414                } finally {
415                        readLock().unlock();
416                }
417        }
418        
419        private ST findUnique(V value) {
420                checkCorrupt();
421                applyUpdates();
422                try {
423                        Handle<ST, PK, S> ret = getIndexUniqueStore().get(value);
424                        return ret!=null && ret.isValid() ? ret.get() : null;
425                } finally {
426                        readLock().unlock();
427                }
428        }
429        
430        @SuppressWarnings("unchecked")
431        public Iterable<T> find(V from, V to, boolean fromInclusive, boolean toInclusive) {
432                checkCorrupt();
433                applyUpdates();
434                try {
435                        Collection<T> ret = new LinkedList<T>();
436                        // Step 1 - key set to array.
437                        Collection<V> keys = isUnique() ? getIndexUniqueStore().keySet() : getIndexMultiStore().keySet();
438                        List<V> keyList = keys instanceof List ? (List<V>) keys : new ArrayList<V>(keys);
439                        Comparator<V> theComparator = getComparator()==null ? naturalComparator : getComparator();
440                        int fromIdx = Collections.binarySearch(keyList, from, theComparator);
441                        if (fromIdx<0) {
442                                // -Insertion point - 1
443                                // Inclusiveness doesn't matter
444                                fromIdx = -fromIdx - 1; 
445                        } else {
446                                if (!fromInclusive) {
447                                        ++fromIdx;
448                                }
449                        }
450                        
451                        int toIdx = Collections.binarySearch(keyList, to, theComparator);
452                        if (toIdx<0) {
453                                // -Insertion point - 1
454                                // Inclusiveness doesn't matter
455                                toIdx = -toIdx - 1;                             
456                        } else {
457                                if (!toInclusive) {
458                                        --toIdx;
459                                }
460                        }
461                        
462                        for (int i=fromIdx; i<=toIdx; ++i) {
463                                V currentKey = keyList.get(i);
464                                
465                                if (isUnique()) {
466                                        Handle<ST, PK, S> handle = getIndexUniqueStore().get(currentKey);
467                                        if (handle.isValid()) {
468                                                ret.add((T) handle.get());
469                                        }
470                                } else {
471                                        for (Handle<ST,PK,S> h: getIndexMultiStore().get(currentKey)) {
472                                                if (h.isValid()) {
473                                                        ret.add((T) h.get());
474                                                }
475                                        }
476                                }                               
477                        }
478                        return ret;
479                } finally {
480                        readLock().unlock();
481                }
482        }
483        
484        private InvocationHandler invocationHandler = new InvocationHandler() {
485                
486                @SuppressWarnings("unchecked")
487                @Override
488                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
489                        if (method.getName().equals("find") && method.getParameterTypes().length==1) {
490                                if (UniqueIndex.class.equals(method.getDeclaringClass())) {
491                                        return findUnique((V) args[0]);
492                                } 
493                                return findMulti((V) args[0]);
494                        }
495                        return method.invoke(AbstractIndex.this, args);
496                }
497        };
498        
499        /**
500         * @return proxy which implements interfaces appropriate for the index type.
501         */
502        @SuppressWarnings("unchecked")
503        public Index<ST, T,PK,V,S> createProxy() {
504                if (isUnique()) {
505                        if (isOrdered()) {
506                                return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {OrderedIndex.class, UniqueIndex.class}, invocationHandler);
507                        } else {
508                                return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {UniqueIndex.class}, invocationHandler);                             
509                        }
510                } else {
511                        if (isOrdered()) {
512                                return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {OrderedIndex.class, MultiIndex.class}, invocationHandler);
513                        } else {
514                                return (Index<ST, T, PK, V, S>) Proxy.newProxyInstance(this.getClass().getClassLoader(), new Class[] {MultiIndex.class}, invocationHandler);                              
515                        }
516                }
517        }
518
519        @Override
520        public String toString() {
521                return "AbstractIndex [predicate=" + predicate + ", extractor="
522                                + extractor + ", type=" + type + ", ordered=" + ordered
523                                + ", comparator=" + comparator + ", isCorrupt=" + isCorrupt
524                                + "]";
525        }
526
527
528        
529}