HighContentionAllocator { public CompletableFuture<Long> allocate(final String valueToStore) { final byte[] valueBytes = Tuple.from(valueToStore).pack(); return initialWindow() .thenCompose(initialWindow -> chooseWindow(initialWindow, false)) .thenCompose(window -> chooseCandidate(window, valueBytes)); } HighContentionAllocator(@Nonnull FDBRecordContext context,
                                   @Nonnull KeySpacePath basePath);  HighContentionAllocator(@Nonnull FDBRecordContext context,
                                   @Nonnull KeySpacePath basePath,
                                   @Nonnull Function<Long, CompletableFuture<Boolean>> candidateCheck);  HighContentionAllocator(@Nonnull FDBRecordContext context,
                                   @Nonnull Subspace basePathSubspace,
                                   @Nonnull Function<Long, CompletableFuture<Boolean>> candidateCheck);  HighContentionAllocator(@Nonnull FDBRecordContext context,
                                   @Nonnull Subspace counterSubspace,
                                   @Nonnull Subspace allocationSubspace); protected  HighContentionAllocator(@Nonnull FDBRecordContext context,
                                      @Nonnull Subspace counterSubspace,
                                      @Nonnull Subspace allocationSubspace,
                                      @Nonnull Function<Long, CompletableFuture<Boolean>> candidateCheck); static HighContentionAllocator forRoot(@Nonnull FDBRecordContext context,
                                                  @Nonnull Subspace counterSubspace,
                                                  @Nonnull Subspace allocationSubspace); static HighContentionAllocator forRoot(@Nonnull FDBRecordContext context, @Nonnull KeySpacePath basePath); @VisibleForTesting Subspace getAllocationSubspace(); CompletableFuture<Long> allocate(final String valueToStore); void setWindow(long count); void forceAllocate(@Nonnull String key, @Nonnull Long value);  }
@Test void testAllocationsUnique() { final Map<Long, String> allocated = new HashMap<>(); try (FDBRecordContext context = database.openContext()) { HighContentionAllocator hca = new HighContentionAllocator(context, keySpace.path("test-path")); for (int i = 0; i < 50; i++) { String storedValue = "allocate-" + i; Long thisAllocation = hca.allocate(storedValue).join(); assertThat("allocations are unique within a transaction", allocated, not(hasKey(thisAllocation))); allocated.put(thisAllocation, storedValue); } validateAllocation(context, hca, allocated); context.commit(); } try (FDBRecordContext context = database.openContext()) { HighContentionAllocator hca = new HighContentionAllocator(context, keySpace.path("test-path")); validateAllocation(context, hca, allocated); for (int i = 0; i < 10; i++) { String storedValue = "new-allocate-" + i; Long newAllocation = hca.allocate(storedValue).join(); assertThat("allocations are unique across transactions", allocated, not(hasKey(newAllocation))); } } }
@Test void testAllocationsParallelSameTransaction() { final Map<Long, String> allocated; try (FDBRecordContext context = database.openContext()) { HighContentionAllocator hca = new HighContentionAllocator(context, keySpace.path("test-path")); List<CompletableFuture<Pair<Long, String>>> allocationOperations = new ArrayList<>(); for (int i = 0; i < 50; i++) { String storedValue = "allocate-" + i; allocationOperations.add(hca.allocate(storedValue).thenApply(id -> Pair.of(id, storedValue))); } allocated = AsyncUtil.getAll(allocationOperations).join().stream() .collect(Collectors.toMap(Pair::getLeft, Pair::getRight)); assertThat("every allocation operation has a distinct value", allocated.entrySet(), hasSize(50)); validateAllocation(context, hca, allocated); context.commit(); } }
@Test void testAllocationsParallelSeparateTransactions() { List<CompletableFuture<Pair<Long, String>>> allocationOperations = new ArrayList<>(); AtomicInteger count = new AtomicInteger(); for (int i = 0; i < 50; i++) { CompletableFuture<Pair<Long, String>> allocation = database.runAsync(context -> { HighContentionAllocator hca = new HighContentionAllocator(context, keySpace.path("test-path")); String storedValue = "allocate-" + count.getAndIncrement(); return hca.allocate(storedValue).thenApply(id -> Pair.of(id, storedValue)); }); allocationOperations.add(allocation); } Map<Long, String> allocated = AsyncUtil.getAll(allocationOperations).join().stream() .collect(Collectors.toMap(Pair::getLeft, Pair::getRight)); assertThat("all values are allocated", allocated.entrySet(), hasSize(50)); try (FDBRecordContext context = database.openContext()) { HighContentionAllocator hca = new HighContentionAllocator(context, keySpace.path("test-path")); validateAllocation(context, hca, allocated); } }