RangeStoreDb extends AbstractLmdbDb<RangeStoreKey, ValueStoreKey> { public Optional<ValueStoreKey> get(final Txn<ByteBuffer> txn, final UID mapDefinitionUid, final long key) { return getAsBytes(txn, mapDefinitionUid, key) .map(valueSerde::deserialize); } @Inject  RangeStoreDb(@Assisted final Env<ByteBuffer> lmdbEnvironment,
                        final ByteBufferPool byteBufferPool,
                        final RangeStoreKeySerde keySerde,
                        final ValueStoreKeySerde valueSerde); Optional<ValueStoreKey> get(final Txn<ByteBuffer> txn, final UID mapDefinitionUid, final long key); Optional<ByteBuffer> getAsBytes(final Txn<ByteBuffer> txn, final UID mapDefinitionUid, final long key); boolean containsMapDefinition(final Txn<ByteBuffer> txn, final UID mapDefinitionUid); void deleteMapEntries(final Txn<ByteBuffer> writeTxn,
                                 final UID mapUid,
                                 final EntryConsumer entryConsumer); static final String DB_NAME; }
@Test void testGet() { final List<UID> uids = Arrays.asList(uid1, uid2, uid3); for (int i = 0; i < uids.size(); i++) { LOGGER.debug("Iteration {}, loading for UID {}", i, uids.get(i)); rangeStoreDb.put(key(uids.get(i), 1, 11), val((i * 10) + 1), false); rangeStoreDb.put(key(uids.get(i), 11, 13), val((i * 10) + 2), false); rangeStoreDb.put(key(uids.get(i), 13, 21), val((i * 10) + 3), false); rangeStoreDb.put(key(uids.get(i), 101, 201), val((i * 10) + 4), false); rangeStoreDb.put(key(uids.get(i), 201, 301), val((i * 10) + 5), false); } rangeStoreDb.logRawDatabaseContents(); rangeStoreDb.logDatabaseContents(); LmdbUtils.doWithReadTxn(lmdbEnv, txn -> { for (int i = 0; i < uids.size(); i++) { LOGGER.debug("Iteration {}, testing with UID {}", i, uids.get(i)); getAndAssert(txn, uids.get(i), 1, (i * 10) + 1); getAndAssert(txn, uids.get(i), 5, (i * 10) + 1); getAndAssert(txn, uids.get(i), 10, (i * 10) + 1); getAndAssert(txn, uids.get(i), 11, (i * 10) + 2); getAndAssert(txn, uids.get(i), 300, (i * 10) + 5); getAndAssertNotFound(txn, uids.get(i), 0); getAndAssertNotFound(txn, uids.get(i), 50); getAndAssertNotFound(txn, uids.get(i), 301); } }); }
@Disabled @Test void testGetWithNegativeNumbers() { final List<UID> uids = Arrays.asList(uid1); for (int i = 0; i < uids.size(); i++) { LOGGER.debug("Iteration {}, loading for UID {}", i, uids.get(i)); rangeStoreDb.put(key(uids.get(i), -20, -10), val((i * 10) + 1), false); rangeStoreDb.put(key(uids.get(i), -10, 0), val((i * 10) + 2), false); rangeStoreDb.put(key(uids.get(i), 0, 1), val((i * 10) + 3), false); rangeStoreDb.put(key(uids.get(i), 1, 11), val((i * 10) + 4), false); rangeStoreDb.put(key(uids.get(i), 12, 13), val((i * 10) + 5), false); rangeStoreDb.put(key(uids.get(i), 13, 21), val((i * 10) + 6), false); rangeStoreDb.put(key(uids.get(i), 101, 201), val((i * 10) + 4), false); rangeStoreDb.put(key(uids.get(i), 201, 301), val((i * 10) + 8), false); } rangeStoreDb.logRawDatabaseContents(); rangeStoreDb.logDatabaseContents(); LmdbUtils.doWithReadTxn(lmdbEnv, txn -> { Optional<ValueStoreKey> optValueStoreKey = rangeStoreDb.get(txn, uid2, 5); assertThat(optValueStoreKey).isNotEmpty(); assertThat(optValueStoreKey.get()).isEqualTo(val(11)); }); }
@Disabled @Test void testStripedSemaphore() throws InterruptedException { final List<String> keys = Arrays.asList( "one", "two", "three", "four"); final Map<String, AtomicInteger> map = keys.stream() .map(k -> Tuple.of(k, new AtomicInteger()) ) .collect(Collectors.toMap(Tuple2::_1, Tuple2::_2)); final Striped<Semaphore> stripedSemaphore = Striped.semaphore(10, 1); final ExecutorService executorService = Executors.newFixedThreadPool(6); IntStream.rangeClosed(1, 1000000) .forEach(i -> { executorService.submit(() -> { String key = keys.get(new Random().nextInt(keys.size())); LOGGER.debug("Using key {} on thread {}", key, Thread.currentThread().getName()); Semaphore semaphore = stripedSemaphore.get(key); try { try { semaphore.acquire(); } catch (InterruptedException e) { e.printStackTrace(); } AtomicInteger atomicInteger = map.get(key); boolean success = atomicInteger.compareAndSet(0, 1); if (!success) { throw new RuntimeException("compare and set failed"); } ThreadUtil.sleepAtLeastIgnoreInterrupts(500); success = atomicInteger.compareAndSet(1, 0); if (!success) { throw new RuntimeException("compare and set failed"); } } finally { semaphore.release(); } }); }); executorService.awaitTermination(20, TimeUnit.SECONDS); }