1 /**
2  *  Copyright Terracotta, Inc.
3  *
4  *  Licensed under the Apache License, Version 2.0 (the "License");
5  *  you may not use this file except in compliance with the License.
6  *  You may obtain a copy of the License at
7  *
8  *      http://www.apache.org/licenses/LICENSE-2.0
9  *
10  *  Unless required by applicable law or agreed to in writing, software
11  *  distributed under the License is distributed on an "AS IS" BASIS,
12  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  *  See the License for the specific language governing permissions and
14  *  limitations under the License.
15  */
16
17 package net.sf.ehcache.store.disk;
18
19 import net.sf.ehcache.Cache;
20 import net.sf.ehcache.CacheEntry;
21 import net.sf.ehcache.CacheException;
22 import net.sf.ehcache.Ehcache;
23 import net.sf.ehcache.Element;
24 import net.sf.ehcache.Status;
25 import net.sf.ehcache.concurrent.CacheLockProvider;
26 import net.sf.ehcache.concurrent.LockType;
27 import net.sf.ehcache.concurrent.StripedReadWriteLock;
28 import net.sf.ehcache.concurrent.Sync;
29 import net.sf.ehcache.config.CacheConfiguration;
30 import net.sf.ehcache.config.CacheConfigurationListener;
31 import net.sf.ehcache.config.PinningConfiguration;
32 import net.sf.ehcache.config.SizeOfPolicyConfiguration;
33 import net.sf.ehcache.pool.Pool;
34 import net.sf.ehcache.pool.PoolAccessor;
35 import net.sf.ehcache.pool.PoolableStore;
36 import net.sf.ehcache.pool.impl.UnboundedPool;
37 import net.sf.ehcache.store.AbstractStore;
38 import net.sf.ehcache.store.ElementValueComparator;
39 import net.sf.ehcache.store.Policy;
40 import net.sf.ehcache.store.StripedReadWriteLockProvider;
41 import net.sf.ehcache.store.TierableStore;
42 import net.sf.ehcache.store.disk.DiskStorageFactory.DiskMarker;
43 import net.sf.ehcache.store.disk.DiskStorageFactory.DiskSubstitute;
44 import net.sf.ehcache.store.disk.DiskStorageFactory.Placeholder;
45 import net.sf.ehcache.writer.CacheWriterManager;
46
47 import java.io.File;
48 import java.io.IOException;
49 import java.io.Serializable;
50 import java.util.AbstractSet;
51 import java.util.ArrayList;
52 import java.util.Collection;
53 import java.util.Collections;
54 import java.util.Iterator;
55 import java.util.List;
56 import java.util.Random;
57 import java.util.Set;
58 import java.util.concurrent.TimeUnit;
59 import java.util.concurrent.atomic.AtomicReference;
60 import java.util.concurrent.locks.ReadWriteLock;
61 import java.util.concurrent.locks.ReentrantReadWriteLock;
62
63 /**
64  * Implements a persistent-to-disk store.
65  * <p>
66  * All new elements are automatically scheduled for writing to disk.
67  *
68  * @author Chris Dennis
69  * @author Ludovic Orban
70  */
71 public final class DiskStore extends AbstractStore implements TierableStore, PoolableStore, StripedReadWriteLockProvider {
72
73     private static final int FFFFCD7D = 0xffffcd7d;
74     private static final int FIFTEEN = 15;
75     private static final int TEN = 10;
76     private static final int THREE = 3;
77     private static final int SIX = 6;
78     private static final int FOURTEEN = 14;
79     private static final int SIXTEEN = 16;
80
81     private static final int RETRIES_BEFORE_LOCK = 2;
82     private static final int DEFAULT_INITIAL_CAPACITY = 16;
83     private static final int DEFAULT_SEGMENT_COUNT = 64;
84     private static final float DEFAULT_LOAD_FACTOR = 0.75f;
85     private static final int SLEEP_INTERVAL_MS = 10;
86
87     private final DiskStorageFactory disk;
88     private final Random rndm = new Random();
89     private final Segment[] segments;
90     private final int segmentShift;
91     private final AtomicReference<Status> status = new AtomicReference<Status>(Status.STATUS_UNINITIALISED);
92     private final boolean tierPinned;
93     private final boolean persistent;
94
95     private volatile CacheLockProvider lockProvider;
96     private volatile Set<Object> keySet;
97     private volatile PoolAccessor onHeapPoolAccessor;
98     private volatile PoolAccessor onDiskPoolAccessor;
99
100
101     private DiskStore(DiskStorageFactory disk, Ehcache cache, Pool onHeapPool, Pool onDiskPool) {
102         this.segments = new Segment[DEFAULT_SEGMENT_COUNT];
103         this.segmentShift = Integer.numberOfLeadingZeros(segments.length - 1);
104         this.onHeapPoolAccessor = onHeapPool.createPoolAccessor(this,
105             SizeOfPolicyConfiguration.resolveMaxDepth(cache),
106             SizeOfPolicyConfiguration.resolveBehavior(cache).equals(SizeOfPolicyConfiguration.MaxDepthExceededBehavior.ABORT));
107         this.onDiskPoolAccessor = onDiskPool.createPoolAccessor(this, new DiskSizeOfEngine());
108
109         for (int i = 0; i < this.segments.length; ++i) {
110             this.segments[i] = new Segment(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR,
111                     disk, cache.getCacheConfiguration(), onHeapPoolAccessor, onDiskPoolAccessor, cache.getCacheEventNotificationService());
112         }
113
114         this.disk = disk;
115         this.disk.bind(this);
116         this.status.set(Status.STATUS_ALIVE);
117         this.tierPinned = cache.getCacheConfiguration().getPinningConfiguration() != null &&
118                      cache.getCacheConfiguration().getPinningConfiguration().getStore() == PinningConfiguration.Store.INCACHE;
119         this.persistent = cache.getCacheConfiguration().isDiskPersistent();
120     }
121
122     /**
123      * Creates a persitent-to-disk store for the given cache, using the given disk path.
124      *
125      * @param cache cache that fronts this store
126      * @param onHeapPool pool to track heap usage
127      * @param onDiskPool pool to track disk usage
128      * @return a fully initialized store
129      */
130     public static DiskStore create(Ehcache cache, Pool onHeapPool, Pool onDiskPool) {
131         if (cache.getCacheManager() == null) {
132             throw new CacheException("Can't create diskstore without a cache manager");
133         }
134         DiskStorageFactory disk = new DiskStorageFactory(cache, cache.getCacheEventNotificationService());
135         DiskStore store = new DiskStore(disk, cache, onHeapPool, onDiskPool);
136         cache.getCacheConfiguration().addConfigurationListener(new CacheConfigurationListenerAdapter(disk, onDiskPool));
137         return store;
138     }
139
140     /**
141      * Creates a persitent-to-disk store for the given cache, using the given disk path.
142      * Heap and disk usage are not tracked by the returned store.
143      *
144      * @param cache cache that fronts this store
145      * @return a fully initialized store
146      */
147     public static DiskStore create(Cache cache) {
148         return create(cache, new UnboundedPool(), new UnboundedPool());
149     }
150
151     /**
152      * {@inheritDoc}
153      */
154     public void unpinAll() {
155         // no-op
156     }
157
158     /**
159      * {@inheritDoc}
160      */
161     public boolean isPinned(Object key) {
162         return false;
163     }
164
165     /**
166      * {@inheritDoc}
167      */
168     public void setPinned(Object key, boolean pinned) {
169         // no-op
170     }
171
172
173     /**
174      * Will check whether a Placeholder that failed to flush to disk is lying around
175      * If so, it'll try to evict it
176      * @param key the key
177      * @return true if a failed marker was or is still there, false otherwise
178      */
179     public boolean cleanUpFailedMarker(final Serializable key) {
180         int hash = hash(key.hashCode());
181         return segmentFor(hash).cleanUpFailedMarker(key, hash);
182     }
183
184     /**
185      * {@inheritDoc}
186      */
187     public StripedReadWriteLock createStripedReadWriteLock() {
188         return new DiskStoreStripedReadWriteLock();
189     }
190
191     /**
192      *
193      */
194     private static final class CacheConfigurationListenerAdapter implements CacheConfigurationListener {
195
196         private final DiskStorageFactory disk;
197         private final Pool diskPool;
198
199         private CacheConfigurationListenerAdapter(DiskStorageFactory disk, Pool diskPool) {
200             this.disk = disk;
201             this.diskPool = diskPool;
202         }
203
204         /**
205          * {@inheritDoc}
206          */
207         public void timeToIdleChanged(long oldTimeToIdle, long newTimeToIdle) {
208             // no-op
209         }
210
211         /**
212          * {@inheritDoc}
213          */
214         public void timeToLiveChanged(long oldTimeToLive, long newTimeToLive) {
215             // no-op
216         }
217
218         /**
219          * {@inheritDoc}
220          */
221         public void diskCapacityChanged(int oldCapacity, int newCapacity) {
222             disk.setOnDiskCapacity(newCapacity);
223         }
224
225         /**
226          * {@inheritDoc}
227          */
228         public void memoryCapacityChanged(int oldCapacity, int newCapacity) {
229             // no-op
230         }
231
232         /**
233          * {@inheritDoc}
234          */
235         public void loggingChanged(boolean oldValue, boolean newValue) {
236             // no-op
237         }
238
239         /**
240          * {@inheritDoc}
241          */
242         public void registered(CacheConfiguration config) {
243             // no-op
244         }
245
246         /**
247          * {@inheritDoc}
248          */
249         public void deregistered(CacheConfiguration config) {
250             // no-op
251         }
252
253         /**
254          * {@inheritDoc}
255          */
256         public void maxBytesLocalHeapChanged(final long oldValue, final long newValue) {
257             // no-op
258         }
259
260         /**
261          * {@inheritDoc}
262          */
263         public void maxBytesLocalDiskChanged(final long oldValue, final long newValue) {
264             diskPool.setMaxSize(newValue);
265         }
266     }
267
268     /**
269      * Change the disk capacity, in number of elements
270      * @param newCapacity the new max elements on disk
271      */
272     public void changeDiskCapacity(int newCapacity) {
273         disk.setOnDiskCapacity(newCapacity);
274     }
275
276     /**
277      * {@inheritDoc}
278      */
279     public boolean bufferFull() {
280         return disk.bufferFull();
281     }
282
283     /**
284      * {@inheritDoc}
285      */
286     public boolean containsKeyInMemory(Object key) {
287         return false;
288     }
289
290     /**
291      * {@inheritDoc}
292      */
293     public boolean containsKeyOffHeap(Object key) {
294         return false;
295     }
296
297     /**
298      * {@inheritDoc}
299      */
300     public boolean containsKeyOnDisk(Object key) {
301         return containsKey(key);
302     }
303
304     /**
305      * {@inheritDoc}
306      */
307     public void expireElements() {
308         disk.expireElements();
309     }
310
311     /**
312      * {@inheritDoc}
313      */
314     public void flush() throws IOException {
315         disk.flush();
316     }
317
318     /**
319      * {@inheritDoc}
320      */
321     public Policy getInMemoryEvictionPolicy() {
322         return null;
323     }
324
325     /**
326      * {@inheritDoc}
327      */
328     public int getInMemorySize() {
329         return 0;
330     }
331
332     /**
333      * {@inheritDoc}
334      */
335     public long getInMemorySizeInBytes() {
336         long size = onHeapPoolAccessor.getSize();
337         if (size < 0) {
338             return 0;
339         } else {
340             return size;
341         }
342     }
343
344     /**
345      * {@inheritDoc}
346      */
347     public int getOffHeapSize() {
348         return 0;
349     }
350
351     /**
352      * {@inheritDoc}
353      */
354     public long getOffHeapSizeInBytes() {
355         return 0;
356     }
357
358     /**
359      * {@inheritDoc}
360      */
361     public int getOnDiskSize() {
362         return disk.getOnDiskSize();
363     }
364
365     /**
366      * {@inheritDoc}
367      */
368     public long getOnDiskSizeInBytes() {
369         long size = onDiskPoolAccessor.getSize();
370         if (size < 0) {
371             return disk.getOnDiskSizeInBytes();
372         } else {
373             return size;
374         }
375     }
376
377     /**
378      * {@inheritDoc}
379      */
380     public int getTerracottaClusteredSize() {
381         return 0;
382     }
383
384     /**
385      * {@inheritDoc}
386      */
387     public void setInMemoryEvictionPolicy(Policy policy) {
388     }
389
390     /**
391      * Return a reference to the data file backing this store.
392      *
393      * @return a reference to the data file backing this store.
394      */
395     public File getDataFile() {
396         return disk.getDataFile();
397     }
398
399     /**
400      * Return a reference to the index file for this store.
401      *
402      * @return a reference to the index file for this store.
403      */
404     public File getIndexFile() {
405         return disk.getIndexFile();
406     }
407
408     /**
409      * {@inheritDoc}
410      */
411     public Object getMBean() {
412         return null;
413     }
414
415     /**
416      * {@inheritDoc}
417      */
418     public void fill(Element e) {
419         put(e);
420     }
421
422     /**
423      * {@inheritDoc}
424      */
425     public boolean removeIfNotPinned(final Object key) {
426         return !tierPinned && remove(key) != null;
427     }
428
429     /**
430      * {@inheritDoc}
431      */
432     public boolean put(Element element) {
433         if (element == null) {
434             return false;
435         } else {
436             Object key = element.getObjectKey();
437             int hash = hash(key.hashCode());
438             Element oldElement = segmentFor(hash).put(key, hash, element, false);
439             return oldElement == null;
440         }
441     }
442
443     /**
444      * {@inheritDoc}
445      */
446     public boolean putWithWriter(Element element, CacheWriterManager writerManager) {
447         boolean newPut = put(element);
448         if (writerManager != null) {
449             try {
450                 writerManager.put(element);
451             } catch (RuntimeException e) {
452                 throw new StoreUpdateException(e, !newPut);
453             }
454         }
455         return newPut;
456     }
457
458     /**
459      * {@inheritDoc}
460      */
461     public Element get(Object key) {
462         if (key == null) {
463             return null;
464         }
465
466         int hash = hash(key.hashCode());
467         return segmentFor(hash).get(key, hash);
468     }
469
470     /**
471      * {@inheritDoc}
472      */
473     public Element getQuiet(Object key) {
474         return get(key);
475     }
476
477     /**
478      * Return the unretrieved (undecoded) value for this key
479      *
480      * @param key key to lookup
481      * @return Element or ElementSubstitute
482      */
483     public Object unretrievedGet(Object key) {
484         if (key == null) {
485             return null;
486         }
487
488         int hash = hash(key.hashCode());
489         return segmentFor(hash).unretrievedGet(key, hash);
490     }
491
492     /**
493      * Put the given encoded element directly into the store
494      *
495      * @param key the key of the element
496      * @param encoded the encoded element
497      * @return true if the encoded element was installed
498      * @throws IllegalArgumentException if the supplied key is already present
499      */
500     public boolean putRawIfAbsent(Object key, DiskMarker encoded) throws IllegalArgumentException {
501         int hash = hash(key.hashCode());
502         return segmentFor(hash).putRawIfAbsent(key, hash, encoded);
503     }
504
505     /**
506      * {@inheritDoc}
507      */
508     public List getKeys() {
509         return new ArrayList(keySet());
510     }
511
512     /**
513      * Get a set view of the keys in this store
514      *
515      * @return a set view of the keys in this store
516      */
517     public Set<Object> keySet() {
518         if (keySet != null) {
519             return keySet;
520         } else {
521             keySet = new KeySet();
522             return keySet;
523         }
524     }
525
526     /**
527      * {@inheritDoc}
528      */
529     public Element remove(Object key) {
530         if (key == null) {
531             return null;
532         }
533
534         int hash = hash(key.hashCode());
535         return segmentFor(hash).remove(key, hash, null, null);
536     }
537
538     /**
539      * {@inheritDoc}
540      */
541     public void removeNoReturn(Object key) {
542         if (key != null) {
543             int hash = hash(key.hashCode());
544             segmentFor(hash).removeNoReturn(key, hash);
545         }
546     }
547
548     /**
549      * {@inheritDoc}
550      */
551     public boolean isTierPinned() {
552         return tierPinned;
553     }
554
555     /**
556      * {@inheritDoc}
557      */
558     public Set getPresentPinnedKeys() {
559         return Collections.emptySet();
560     }
561
562     /**
563      * {@inheritDoc}
564      */
565     public boolean isPersistent() {
566         return persistent;
567     }
568
569     /**
570      * {@inheritDoc}
571      */
572     public Element removeWithWriter(Object key, CacheWriterManager writerManager) {
573         Element removed = remove(key);
574         if (writerManager != null) {
575             writerManager.remove(new CacheEntry(key, removed));
576         }
577         return removed;
578     }
579
580     /**
581      * {@inheritDoc}
582      */
583     public void removeAll() {
584         for (Segment s : segments) {
585             s.clear();
586         }
587     }
588
589     /**
590      * {@inheritDoc}
591      */
592     public void dispose() {
593         if (status.compareAndSet(Status.STATUS_ALIVE, Status.STATUS_SHUTDOWN)) {
594             disk.unbind();
595             onHeapPoolAccessor.unlink();
596             onDiskPoolAccessor.unlink();
597         }
598     }
599
600     /**
601      * {@inheritDoc}
602      */
603     public int getSize() {
604         final Segment[] segs = this.segments;
605         long size = -1;
606         // Try a few times to get accurate count. On failure due to
607         // continuous async changes in table, resort to locking.
608         for (int k = 0; k < RETRIES_BEFORE_LOCK; ++k) {
609             size = volatileSize(segs);
610             if (size >= 0) {
611                 break;
612             }
613         }
614         if (size < 0) {
615             // Resort to locking all segments
616             size = lockedSize(segs);
617         }
618         if (size > Integer.MAX_VALUE) {
619             return Integer.MAX_VALUE;
620         } else {
621             return (int) size;
622         }
623     }
624
625     private static long volatileSize(Segment[] segs) {
626         int[] mc = new int[segs.length];
627         long check = 0;
628         long sum = 0;
629         int mcsum = 0;
630         for (int i = 0; i < segs.length; ++i) {
631             sum += segs[i].count;
632             mc[i] = segs[i].modCount;
633             mcsum += mc[i];
634         }
635         if (mcsum != 0) {
636             for (int i = 0; i < segs.length; ++i) {
637                 check += segs[i].count;
638                 if (mc[i] != segs[i].modCount) {
639                     return -1;
640                 }
641             }
642         }
643         if (check == sum) {
644             return sum;
645         } else {
646             return -1;
647         }
648     }
649
650     private static long lockedSize(Segment[] segs) {
651         long size = 0;
652         for (Segment seg : segs) {
653             seg.readLock().lock();
654         }
655         for (Segment seg : segs) {
656             size += seg.count;
657         }
658         for (Segment seg : segs) {
659             seg.readLock().unlock();
660         }
661
662         return size;
663     }
664
665     /**
666      * {@inheritDoc}
667      */
668     public Status getStatus() {
669         return status.get();
670     }
671
672     /**
673      * {@inheritDoc}
674      */
675     public boolean evictFromOnHeap(int count, long size) {
676         // evicting from disk also frees up heap
677         return disk.evict(count) == count;
678     }
679
680     /**
681      * {@inheritDoc}
682      */
683     public boolean evictFromOnDisk(int count, long size) {
684         return disk.evict(count) == count;
685     }
686
687     /**
688      * {@inheritDoc}
689      */
690     public float getApproximateDiskHitRate() {
691         float sum = 0;
692         for (Segment s : segments) {
693             sum += s.getDiskHitRate();
694         }
695         return sum;
696     }
697
698     /**
699      * {@inheritDoc}
700      */
701     public float getApproximateDiskMissRate() {
702         float sum = 0;
703         for (Segment s : segments) {
704             sum += s.getDiskMissRate();
705         }
706         return sum;
707     }
708
709     /**
710      * {@inheritDoc}
711      */
712     public long getApproximateDiskCountSize() {
713         return getOnDiskSize();
714     }
715
716     /**
717      * {@inheritDoc}
718      */
719     public long getApproximateDiskByteSize() {
720         return getOnDiskSizeInBytes();
721     }
722
723     /**
724      * {@inheritDoc}
725      */
726     public float getApproximateHeapHitRate() {
727         return 0;
728     }
729
730     /**
731      * {@inheritDoc}
732      */
733     public float getApproximateHeapMissRate() {
734         return 0;
735     }
736
737     /**
738      * {@inheritDoc}
739      */
740     public long getApproximateHeapCountSize() {
741         return getInMemorySize();
742     }
743
744     /**
745      * {@inheritDoc}
746      */
747     public long getApproximateHeapByteSize() {
748         return getInMemorySizeInBytes();
749     }
750
751     /**
752      * {@inheritDoc}
753      */
754     public boolean containsKey(Object key) {
755         int hash = hash(key.hashCode());
756         return segmentFor(hash).containsKey(key, hash);
757     }
758
759     /**
760      * {@inheritDoc}
761      */
762     public Object getInternalContext() {
763         if (lockProvider != null) {
764             return lockProvider;
765         } else {
766             lockProvider = new LockProvider();
767             return lockProvider;
768         }
769     }
770
771     /**
772      * {@inheritDoc}
773      */
774     public Element putIfAbsent(Element element) throws NullPointerException {
775         Object key = element.getObjectKey();
776         int hash = hash(key.hashCode());
777         return segmentFor(hash).put(key, hash, element, true);
778     }
779
780     /**
781      * {@inheritDoc}
782      */
783     public Element removeElement(Element element, ElementValueComparator comparator) throws NullPointerException {
784         Object key = element.getObjectKey();
785         int hash = hash(key.hashCode());
786         return segmentFor(hash).remove(key, hash, element, comparator);
787     }
788
789     /**
790      * {@inheritDoc}
791      */
792     public boolean replace(Element old, Element element, ElementValueComparator comparator)
793             throws NullPointerException, IllegalArgumentException {
794         Object key = element.getObjectKey();
795         int hash = hash(key.hashCode());
796         return segmentFor(hash).replace(key, hash, old, element, comparator);
797     }
798
799     /**
800      * {@inheritDoc}
801      */
802     public Element replace(Element element) throws NullPointerException {
803         Object key = element.getObjectKey();
804         int hash = hash(key.hashCode());
805         return segmentFor(hash).replace(key, hash, element);
806     }
807
808     /**
809      * Atomically switch (CAS) the <code>expect</code> representation of this element for the
810      * <code>fault</code> representation.
811      * <p>
812      * A successful switch will return <code>true</code>, and free the replaced element/element-proxy.
813      * A failed switch will return <code>false</code> and free the element/element-proxy which was not
814      * installed.
815      *
816      * @param key key to which this element (proxy) is mapped
817      * @param expect element (proxy) expected
818      * @param fault element (proxy) to install
819      * @return <code>true</code> if <code>fault</code> was installed
820      */
821     public boolean fault(Object key, Placeholder expect, DiskMarker fault) {
822         int hash = hash(key.hashCode());
823         return segmentFor(hash).fault(key, hash, expect, fault);
824     }
825
826     /**
827      * Remove the matching mapping. The evict method does referential comparison
828      * of the unretrieved substitute against the argument value.
829      *
830      * @param key key to match against
831      * @param substitute optional value to match against
832      * @return <code>true</code> on a successful remove
833      */
834     public boolean evict(Object key, DiskSubstitute substitute) {
835         return evictElement(key, substitute) != null;
836     }
837
838     /**
839      * Remove the matching mapping. The evict method does referential comparison
840      * of the unretrieved substitute against the argument value.
841      *
842      * @param key key to match against
843      * @param substitute optional value to match against
844      * @return the evicted element on a successful remove
845      */
846     public Element evictElement(Object key, DiskSubstitute substitute) {
847         int hash = hash(key.hashCode());
848         return segmentFor(hash).evict(key, hash, substitute);
849     }
850
851     /**
852      * Select a random sample of elements generated by the supplied factory.
853      *
854      * @param factory generator of the given type
855      * @param sampleSize minimum number of elements to return
856      * @param keyHint a key on which we are currently working
857      * @return list of sampled elements/element substitute
858      */
859     public List<DiskStorageFactory.DiskSubstitute> getRandomSample(ElementSubstituteFilter factory, int sampleSize, Object keyHint) {
860         ArrayList<DiskStorageFactory.DiskSubstitute> sampled = new ArrayList<DiskStorageFactory.DiskSubstitute>(sampleSize);
861
862         // pick a random starting point in the map
863         int randomHash = rndm.nextInt();
864
865         final int segmentStart;
866         if (keyHint == null) {
867             segmentStart = (randomHash >>> segmentShift);
868         } else {
869             segmentStart = (hash(keyHint.hashCode()) >>> segmentShift);
870         }
871
872         int segmentIndex = segmentStart;
873         do {
874             segments[segmentIndex].addRandomSample(factory, sampleSize, sampled, randomHash);
875             if (sampled.size() >= sampleSize) {
876                 break;
877             }
878
879             // move to next segment
880             segmentIndex = (segmentIndex + 1) & (segments.length - 1);
881         } while (segmentIndex != segmentStart);
882
883         return sampled;
884     }
885
886     private static int hash(int hash) {
887         int spread = hash;
888         spread += (spread << FIFTEEN ^ FFFFCD7D);
889         spread ^= spread >>> TEN;
890         spread += (spread << THREE);
891         spread ^= spread >>> SIX;
892         spread += (spread << 2) + (spread << FOURTEEN);
893         return (spread ^ spread >>> SIXTEEN);
894     }
895
896     private Segment segmentFor(int hash) {
897         return segments[hash >>> segmentShift];
898     }
899
900     /**
901      * Key set implementation for the DiskStore
902      */
903     final class KeySet extends AbstractSet<Object> {
904
905         /**
906          * {@inheritDoc}
907          */
908         @Override
909         public Iterator<Object> iterator() {
910             return new KeyIterator();
911         }
912
913         /**
914          * {@inheritDoc}
915          */
916         @Override
917         public int size() {
918             return DiskStore.this.getSize();
919         }
920
921         /**
922          * {@inheritDoc}
923          */
924         @Override
925         public boolean contains(Object o) {
926             return DiskStore.this.containsKey(o);
927         }
928
929         /**
930          * {@inheritDoc}
931          */
932         @Override
933         public boolean remove(Object o) {
934             return DiskStore.this.remove(o) != null;
935         }
936
937         /**
938          * {@inheritDoc}
939          */
940         @Override
941         public void clear() {
942             DiskStore.this.removeAll();
943         }
944
945         /**
946          * {@inheritDoc}
947          */
948         @Override
949         public Object[] toArray() {
950             Collection<Object> c = new ArrayList<Object>();
951             for (Object object : this) {
952                 c.add(object);
953             }
954             return c.toArray();
955         }
956
957         /**
958          * {@inheritDoc}
959          */
960         @Override
961         public <T> T[] toArray(T[] a) {
962             Collection<Object> c = new ArrayList<Object>();
963             for (Object object : this) {
964                 c.add(object);
965             }
966             return c.toArray(a);
967         }
968     }
969
970     /**
971      * LockProvider implementation that uses the segment locks.
972      */
973     private class LockProvider implements CacheLockProvider {
974
975         /**
976          * {@inheritDoc}
977          */
978         public Sync getSyncForKey(Object key) {
979             int hash = key == null ? 0 : hash(key.hashCode());
980             return new ReadWriteLockSync(segmentFor(hash));
981         }
982     }
983
984     /**
985      * Superclass for all store iterators.
986      */
987     abstract class HashIterator {
988         private int segmentIndex;
989         private Iterator<HashEntry> currentIterator;
990
991         /**
992          * Constructs a new HashIterator
993          */
994         HashIterator() {
995             segmentIndex = segments.length;
996
997             while (segmentIndex > 0) {
998                 segmentIndex--;
999                 currentIterator = segments[segmentIndex].hashIterator();
1000                 if (currentIterator.hasNext()) {
1001                     return;
1002                 }
1003             }
1004         }
1005
1006         /**
1007          * {@inheritDoc}
1008          */
1009         public boolean hasNext() {
1010             if (this.currentIterator == null) {
1011                 return false;
1012             }
1013
1014             if (this.currentIterator.hasNext()) {
1015                 return true;
1016             } else {
1017                 while (segmentIndex > 0) {
1018                     segmentIndex--;
1019                     currentIterator = segments[segmentIndex].hashIterator();
1020                     if (currentIterator.hasNext()) {
1021                         return true;
1022                     }
1023                 }
1024             }
1025             return false;
1026         }
1027
1028         /**
1029          * Returns the next hash-entry - called by subclasses
1030          *
1031          * @return next HashEntry
1032          */
1033         protected HashEntry nextEntry() {
1034             if (currentIterator == null) {
1035                 return null;
1036             }
1037
1038             if (currentIterator.hasNext()) {
1039                 return currentIterator.next();
1040             } else {
1041                 while (segmentIndex > 0) {
1042                     segmentIndex--;
1043                     currentIterator = segments[segmentIndex].hashIterator();
1044                     if (currentIterator.hasNext()) {
1045                         return currentIterator.next();
1046                     }
1047                 }
1048             }
1049             return null;
1050         }
1051
1052         /**
1053          * {@inheritDoc}
1054          */
1055         public void remove() {
1056             currentIterator.remove();
1057         }
1058
1059         /**
1060          * Get the current segment index of this iterator
1061          *
1062          * @return the current segment index
1063          */
1064         int getCurrentSegmentIndex() {
1065             return segmentIndex;
1066         }
1067     }
1068
1069     /**
1070      * Iterator over the store key set.
1071      */
1072     private final class KeyIterator extends HashIterator implements Iterator<Object> {
1073         /**
1074          * {@inheritDoc}
1075          */
1076         public Object next() {
1077             return super.nextEntry().key;
1078         }
1079     }
1080
1081     /**
1082      * Sync implementation that wraps the segment locks
1083      */
1084     private static final class ReadWriteLockSync implements Sync {
1085
1086         private final ReentrantReadWriteLock lock;
1087
1088         private ReadWriteLockSync(ReentrantReadWriteLock lock) {
1089             this.lock = lock;
1090         }
1091
1092         /**
1093          * {@inheritDoc}
1094          */
1095         public void lock(LockType type) {
1096             switch (type) {
1097             case READ:
1098                 lock.readLock().lock();
1099                 break;
1100             case WRITE:
1101                 lock.writeLock().lock();
1102                 break;
1103             default:
1104                 throw new IllegalArgumentException("We don't support any other lock type than READ or WRITE!");
1105             }
1106         }
1107
1108         /**
1109          * {@inheritDoc}
1110          */
1111         public boolean tryLock(LockType type, long msec) throws InterruptedException {
1112             switch (type) {
1113             case READ:
1114                 return lock.readLock().tryLock(msec, TimeUnit.MILLISECONDS);
1115             case WRITE:
1116                 return lock.writeLock().tryLock(msec, TimeUnit.MILLISECONDS);
1117             default:
1118                 throw new IllegalArgumentException("We don't support any other lock type than READ or WRITE!");
1119             }
1120         }
1121
1122         /**
1123          * {@inheritDoc}
1124          */
1125         public void unlock(LockType type) {
1126             switch (type) {
1127             case READ:
1128                 lock.readLock().unlock();
1129                 break;
1130             case WRITE:
1131                 lock.writeLock().unlock();
1132                 break;
1133             default:
1134                 throw new IllegalArgumentException("We don't support any other lock type than READ or WRITE!");
1135             }
1136         }
1137
1138         /**
1139          * {@inheritDoc}
1140          */
1141         public boolean isHeldByCurrentThread(LockType type) {
1142             switch (type) {
1143             case READ:
1144                 throw new UnsupportedOperationException("Querying of read lock is not supported.");
1145             case WRITE:
1146                 return lock.isWriteLockedByCurrentThread();
1147             default:
1148                 throw new IllegalArgumentException("We don't support any other lock type than READ or WRITE!");
1149             }
1150         }
1151
1152     }
1153
1154     /**
1155      * StripedReadWriteLock impl.
1156      */
1157     private final class DiskStoreStripedReadWriteLock implements StripedReadWriteLock {
1158
1159         private final net.sf.ehcache.concurrent.ReadWriteLockSync[] locks =
1160             new net.sf.ehcache.concurrent.ReadWriteLockSync[DEFAULT_SEGMENT_COUNT];
1161
1162         private DiskStoreStripedReadWriteLock() {
1163             for (int i = 0; i < locks.length; i++) {
1164                 locks[i] = new net.sf.ehcache.concurrent.ReadWriteLockSync();
1165             }
1166         }
1167
1168         /**
1169          * {@inheritDoc}
1170          */
1171         public ReadWriteLock getLockForKey(final Object key) {
1172             return getSyncForKey(key).getReadWriteLock();
1173         }
1174
1175         /**
1176          * {@inheritDoc}
1177          */
1178         public List<net.sf.ehcache.concurrent.ReadWriteLockSync> getAllSyncs() {
1179             ArrayList<net.sf.ehcache.concurrent.ReadWriteLockSync> syncs =
1180                 new ArrayList<net.sf.ehcache.concurrent.ReadWriteLockSync>(locks.length);
1181             Collections.addAll(syncs, locks);
1182             return syncs;
1183         }
1184
1185         /**
1186          * {@inheritDoc}
1187          */
1188         public net.sf.ehcache.concurrent.ReadWriteLockSync getSyncForKey(final Object key) {
1189             return locks[indexFor(key)];
1190         }
1191
1192         private int indexFor(final Object key) {
1193             return hash(key.hashCode()) >>> segmentShift;
1194         }
1195     }
1196 }
1197