@Test(groups = "stress") public void testConcurrentWrite() throws InterruptedException { final int THREADS = 8; final AtomicBoolean run = new AtomicBoolean(true); final AtomicInteger written = new AtomicInteger(); final CountDownLatch started = new CountDownLatch(THREADS); final CountDownLatch finished = new CountDownLatch(THREADS); for (int i = 0; i < THREADS; ++i) { final int thread = i; fork(() -> { try { started.countDown(); int i1 = 0; while (run.get()) { InternalCacheEntry entry = TestInternalCacheEntryFactory.create("k" + i1, "v" + i1); MarshallableEntry me = MarshalledEntryUtil.create(entry, getMarshaller()); try { store.write(me); ++i1; int prev; do { prev = written.get(); if ((prev & (1 << thread)) != 0) break; } while (written.compareAndSet(prev, prev | (1 << thread))); } catch (PersistenceException e) { } } } catch (Exception e) { log.error("Failed", e); throw new RuntimeException(e); } finally { finished.countDown(); } }); } if (finished.await(1, TimeUnit.SECONDS)) { fail("Test shouldn't have finished yet"); } run.set(false); if (!finished.await(30, TimeUnit.SECONDS)) { fail("Test should have finished!"); } assertEquals(written.get(), (1 << THREADS) - 1, "pre"); }
@Override public CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry) { return handler.write(segment, entry); }
RocksDBStore implements NonBlockingStore<K, V> { @Override public CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry) { return handler.write(segment, entry); } }
RocksDBStore implements NonBlockingStore<K, V> { @Override public CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry) { return handler.write(segment, entry); }  }
RocksDBStore implements NonBlockingStore<K, V> { @Override public CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry) { return handler.write(segment, entry); }  @Override CompletionStage<Void> start(InitializationContext ctx); @Override CompletionStage<Void> stop(); @Override Set<Characteristic> characteristics(); @Override CompletionStage<Boolean> isAvailable(); @Override CompletionStage<Void> clear(); @Override CompletionStage<Long> size(IntSet segments); @Override CompletionStage<Long> approximateSize(IntSet segments); @Override CompletionStage<Boolean> containsKey(int segment, Object key); @Override Publisher<K> publishKeys(IntSet segments, Predicate<? super K> filter); @Override Publisher<MarshallableEntry<K, V>> publishEntries(IntSet segments, Predicate<? super K> filter, boolean includeValues); @Override CompletionStage<Boolean> delete(int segment, Object key); @Override CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry); @Override CompletionStage<MarshallableEntry<K, V>> load(int segment, Object key); @Override CompletionStage<Void> batch(int publisherCount, Publisher<SegmentedPublisher<Object>> removePublisher,
         Publisher<SegmentedPublisher<MarshallableEntry<K, V>>> writePublisher); @Override Publisher<MarshallableEntry<K, V>> purgeExpired(); @Override CompletionStage<Void> addSegments(IntSet segments); @Override CompletionStage<Void> removeSegments(IntSet segments); }
RocksDBStore implements NonBlockingStore<K, V> { @Override public CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry) { return handler.write(segment, entry); }  @Override CompletionStage<Void> start(InitializationContext ctx); @Override CompletionStage<Void> stop(); @Override Set<Characteristic> characteristics(); @Override CompletionStage<Boolean> isAvailable(); @Override CompletionStage<Void> clear(); @Override CompletionStage<Long> size(IntSet segments); @Override CompletionStage<Long> approximateSize(IntSet segments); @Override CompletionStage<Boolean> containsKey(int segment, Object key); @Override Publisher<K> publishKeys(IntSet segments, Predicate<? super K> filter); @Override Publisher<MarshallableEntry<K, V>> publishEntries(IntSet segments, Predicate<? super K> filter, boolean includeValues); @Override CompletionStage<Boolean> delete(int segment, Object key); @Override CompletionStage<Void> write(int segment, MarshallableEntry<? extends K, ? extends V> entry); @Override CompletionStage<MarshallableEntry<K, V>> load(int segment, Object key); @Override CompletionStage<Void> batch(int publisherCount, Publisher<SegmentedPublisher<Object>> removePublisher,
         Publisher<SegmentedPublisher<MarshallableEntry<K, V>>> writePublisher); @Override Publisher<MarshallableEntry<K, V>> purgeExpired(); @Override CompletionStage<Void> addSegments(IntSet segments); @Override CompletionStage<Void> removeSegments(IntSet segments);  }