RefDataOffHeapStore extends AbstractRefDataStore implements RefDataStore, HasSystemInfo { @Override public Optional<RefDataValue> getValue(final MapDefinition mapDefinition, final String key) { try (PooledByteBuffer valueStoreKeyPooledBufferClone = valueStore.getPooledKeyBuffer()) { Optional<RefDataValue> optionalRefDataValue = LmdbUtils.getWithReadTxn(lmdbEnvironment, readTxn -> getValueStoreKey(readTxn, mapDefinition, key) .flatMap(valueStoreKeyBuffer -> { final ByteBuffer valueStoreKeyBufferClone = valueStoreKeyPooledBufferClone.getByteBuffer(); ByteBufferUtils.copy(valueStoreKeyBuffer, valueStoreKeyBufferClone); return Optional.of(valueStoreKeyBufferClone); }) .flatMap(valueStoreKeyBuffer -> valueStore.get(readTxn, valueStoreKeyBuffer))); LOGGER.trace("getValue({}, {}) - {}", mapDefinition, key, optionalRefDataValue); return optionalRefDataValue; } } @Inject  RefDataOffHeapStore(
            final TempDirProvider tempDirProvider,
            final PathCreator pathCreator,
            final ReferenceDataConfig referenceDataConfig,
            final ByteBufferPool byteBufferPool,
            final Factory keyValueStoreDbFactory,
            final ValueStoreDb.Factory valueStoreDbFactory,
            final ValueStoreMetaDb.Factory valueStoreMetaDbFactory,
            final RangeStoreDb.Factory rangeStoreDbFactory,
            final MapUidForwardDb.Factory mapUidForwardDbFactory,
            final MapUidReverseDb.Factory mapUidReverseDbFactory,
            final RefDataValueConverter refDataValueConverter,
            final ProcessingInfoDb.Factory processingInfoDbFactory); @Override StorageType getStorageType(); @Override Optional<RefDataProcessingInfo> getAndTouchProcessingInfo(final RefStreamDefinition refStreamDefinition); @Override boolean isDataLoaded(final RefStreamDefinition refStreamDefinition); @Override boolean exists(final MapDefinition mapDefinition); @Override Optional<RefDataValue> getValue(final MapDefinition mapDefinition,
                                           final String key); @Override boolean consumeValueBytes(final MapDefinition mapDefinition,
                                     final String key,
                                     final Consumer<TypedByteBuffer> valueBytesConsumer); @Override void purgeOldData(); @Override long getKeyValueEntryCount(); @Override long getKeyRangeValueEntryCount(); @Override long getProcessingInfoEntryCount(); @Override void logAllContents(); @Override void logAllContents(Consumer<String> logEntryConsumer); @Override List<RefStoreEntry> list(final int limit); List<RefStoreEntry> list(final int limit,
                                    final Predicate<RefStoreEntry> filter); @Override SystemInfoResult getSystemInfo();  }
@Test void testLoaderConcurrency() { final RefStreamDefinition refStreamDefinition = buildUniqueRefStreamDefinition(); final long effectiveTimeMs = System.currentTimeMillis(); final MapDefinition mapDefinitionKey = new MapDefinition(refStreamDefinition, "MyKeyMap"); final MapDefinition mapDefinitionRange = new MapDefinition(refStreamDefinition, "MyRangeMap"); final int recCount = 1_000; Runnable loadTask = () -> { LOGGER.debug("Running loadTask on thread {}", Thread.currentThread().getName()); try { refDataStore.doWithLoaderUnlessComplete(refStreamDefinition, effectiveTimeMs, refDataLoader -> { refDataLoader.setCommitInterval(200); refDataLoader.initialise(false); long rangeStartInc = 0; long rangeEndExc; for (int i = 0; i < recCount; i++) { refDataLoader.put(mapDefinitionKey, "key" + i, StringValue.of("Value" + i)); rangeEndExc = rangeStartInc + 10; Range<Long> range = new Range<>(rangeStartInc, rangeEndExc); rangeStartInc = rangeEndExc; refDataLoader.put(mapDefinitionRange, range, StringValue.of("Value" + i)); } refDataLoader.completeProcessing(); LOGGER.debug("Finished loading data"); }); LOGGER.debug("Getting values"); LAMBDA_LOGGER.logDurationIfDebugEnabled(() -> { IntStream.range(0, recCount) .boxed() .sorted(Comparator.reverseOrder()) .forEach(i -> { Optional<RefDataValue> optValue = refDataStore.getValue(mapDefinitionKey, "key" + i); assertThat(optValue.isPresent()); optValue = refDataStore.getValue(mapDefinitionKey, Integer.toString((i * 10) + 5)); assertThat(optValue.isPresent()); }); }, () -> LogUtil.message("Getting {} entries, twice", recCount)); } catch (Exception e) { throw new RuntimeException(e); } LOGGER.debug("Finished running loadTask on thread {}", Thread.currentThread().getName()); }; ExecutorService executorService = Executors.newFixedThreadPool(6); List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10) .boxed() .map(i -> { LOGGER.debug("Running async task on thread {}", Thread.currentThread().getName()); final CompletableFuture<Void> future = CompletableFuture.runAsync(loadTask, executorService); LOGGER.debug("Got future"); return future; }) .collect(Collectors.toList()); futures.forEach(voidCompletableFuture -> { try { voidCompletableFuture.get(); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } }); LOGGER.debug("Finished all"); }
@Test void testLoaderConcurrency_multipleStreamDefs() { final long effectiveTimeMs = System.currentTimeMillis(); final int recCount = 1_000; Consumer<RefStreamDefinition> loadTask = refStreamDefinition -> { final MapDefinition mapDefinitionKey = new MapDefinition(refStreamDefinition, "MyKeyMap"); final MapDefinition mapDefinitionRange = new MapDefinition(refStreamDefinition, "MyRangeMap"); LOGGER.debug("Running loadTask on thread {}", Thread.currentThread().getName()); try { refDataStore.doWithLoaderUnlessComplete(refStreamDefinition, effectiveTimeMs, refDataLoader -> { refDataLoader.setCommitInterval(200); refDataLoader.initialise(false); long rangeStartInc = 0; long rangeEndExc; for (int i = 0; i < recCount; i++) { refDataLoader.put(mapDefinitionKey, "key" + i, StringValue.of("Value" + i)); rangeEndExc = rangeStartInc + 10; Range<Long> range = new Range<>(rangeStartInc, rangeEndExc); rangeStartInc = rangeEndExc; refDataLoader.put(mapDefinitionRange, range, StringValue.of("Value" + i)); } refDataLoader.completeProcessing(); LOGGER.debug("Finished loading data"); }); LOGGER.debug("Getting values"); LAMBDA_LOGGER.logDurationIfDebugEnabled(() -> { IntStream.range(0, recCount) .boxed() .sorted(Comparator.reverseOrder()) .forEach(i -> { Optional<RefDataValue> optValue = refDataStore.getValue(mapDefinitionKey, "key" + i); assertThat(optValue.isPresent()); optValue = refDataStore.getValue(mapDefinitionKey, Integer.toString((i * 10) + 5)); assertThat(optValue.isPresent()); }); }, () -> LogUtil.message("Getting {} entries, twice", recCount)); } catch (Exception e) { throw new RuntimeException(e); } LOGGER.debug("Finished running loadTask on thread {}", Thread.currentThread().getName()); }; ExecutorService executorService = Executors.newFixedThreadPool(6); List<CompletableFuture<Void>> futures = IntStream.rangeClosed(1, 10) .boxed() .map(i -> { LOGGER.debug("Running async task on thread {}", Thread.currentThread().getName()); RefStreamDefinition refStreamDefinition = buildUniqueRefStreamDefinition(); final CompletableFuture<Void> future = CompletableFuture.runAsync( () -> loadTask.accept(refStreamDefinition), executorService); LOGGER.debug("Got future"); return future; }) .collect(Collectors.toList()); futures.forEach(voidCompletableFuture -> { try { voidCompletableFuture.get(); } catch (InterruptedException | ExecutionException e) { throw new RuntimeException(e); } }); LOGGER.debug("Finished all"); }