Java 类java.util.concurrent.atomic.LongAdder 实例源码

/**
 * forEachSequentially traverses all mappings
 */
public void testForEachSequentially() {
    LongAdder adder = new LongAdder();
    ConcurrentHashMap<Long, Long> m = longMap();
    m.forEach(Long.MAX_VALUE, (Long x, Long y) -> adder.add(x.longValue() + y.longValue()));
    assertEquals(adder.sum(), 3 * SIZE * (SIZE - 1) / 2);
}

/**
 * forEachInParallel traverses all mappings
 */
public void testForEachInParallel() {
    LongAdder adder = new LongAdder();
    ConcurrentHashMap<Long, Long> m = longMap();
    m.forEach(1L, (Long x, Long y) -> adder.add(x.longValue() + y.longValue()));
    assertEquals(adder.sum(), 3 * SIZE * (SIZE - 1) / 2);
}

/**
 * Purges a Bundle from the system.
 * @param bundleid - Bundle Identifier
 */
public static void purgeBundle(Long bundleid) {
    if (bundleid == null) {
        return;
    }
    Map.Entry<LongAdder,List<Transaction>> bundleInfo =  bundles.get(bundleid);
    if (bundleInfo == null) {
        return;
    }
    long ts = System.currentTimeMillis();
    for(Transaction t : bundleInfo.getValue()) {
        t.setStatusTs(OperationStatus.FAILED, ts);
        purgeOperation(t.getClientId().toString() + "/" + bundleid + "/" + t.getOpId().toString());
    }
    bundles.remove(bundleid);
}

@Test
public void LongConsumer()
{
    LongAdder adder = new LongAdder();
    // TODO - Convert the anonymous inner class to a lambda
    LongConsumer consumer = new LongConsumer()
    {
        @Override
        public void accept(long value)
        {
            adder.add(value);
        }
    };
    LongStream.rangeClosed(1, 5).forEach(consumer);
    Assert.assertEquals(15, adder.longValue());
}

@Test
public void test() {
    if(nodeCount > 1) {
        final long connCountSum = nodeFreq.values().stream().mapToLong(LongAdder::sum).sum();
        final long avgConnCountPerNode = connCountSum / nodeCount;
        for(final String nodeAddr: nodeFreq.keySet()) {
            assertTrue(nodeFreq.get(nodeAddr).sum() > 0);
            assertEquals(
                "Node count: " + nodeCount + ", node: \"" + nodeAddr
                    + "\", expected connection count: " + avgConnCountPerNode + ", actual: "
                    + nodeFreq.get(nodeAddr).sum(),
                avgConnCountPerNode, nodeFreq.get(nodeAddr).sum(), 1.5 * avgConnCountPerNode
            );
        }
    } else {
        assertTrue(true);
    }
}

@Test
public void testReloadCalledPeriodically() {
    LongAdder numOfReloadCalls = new LongAdder();
    PeriodicalReloadStrategy strategy = PeriodicalReloadStrategy.builder()
            .withInterval(Duration.ofMillis(50))
            .build();

    try {
        strategy.start(numOfReloadCalls::increment);

        await("Reload called more then once")
                .atMost(5, TimeUnit.SECONDS)
                .until(() -> numOfReloadCalls.longValue() > 1);
    } finally {
        strategy.stop();
    }
}

@Benchmark
@Warmup(iterations = 10)
@Measurement(iterations = 10)
@Fork(1)
@Threads(4)
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
public List measureRecommendationTraversal() throws IOException {
    Set<String> itemsYouLike = db.getOutgoingRelationshipNodeIds("LIKES", "person" + rand.nextInt(personCount));
    Map<String, LongAdder> occurrences = new HashMap<>();
    for (String item : itemsYouLike) {
        for (String person : db.getIncomingRelationshipNodeIds("LIKES", item)) {
            Set<String> itemsYouMightLike = db.getOutgoingRelationshipNodeIds("LIKES", person);
            itemsYouMightLike.removeAll(itemsYouLike);
            for (String unlikeditem : itemsYouMightLike) {
                occurrences.computeIfAbsent(unlikeditem, (t) -> new LongAdder()).increment();
            }
        }
    }
    List<Map.Entry<String, LongAdder>> itemList = new ArrayList<>(occurrences.entrySet());
    Collections.sort(itemList, (a, b) -> ( b.getValue().intValue() - a.getValue().intValue() ));
    return itemList.subList(0, Math.min(itemList.size(), 10));
}

private void doRunTest(int serverInstances, int threadsPerServer, int invocationsPerClient, int clientThreads, int delayPerRequest, int clientMaxWait) throws Exception {
  when(testService.getString(any())).thenAnswer(createAnswer(delayPerRequest));

  int totalServerThreads = serverInstances * threadsPerServer;
  long targetTime = delayPerRequest * clientThreads * invocationsPerClient / serverInstances / threadsPerServer;

  System.out.println(String.format("Running %d server instances with %d threads (threadsPerServer=%d delayPerRequest=%d)", totalServerThreads, serverInstances, threadsPerServer, delayPerRequest));
  System.out.println(String.format("Executing %d clients with %d requests/client (total %d requests)", clientThreads, invocationsPerClient, clientThreads * invocationsPerClient));
  System.out.println(String.format("Target time %dms", targetTime));

  for (int i = 0; i < serverInstances; i++) {
    setupServer(threadsPerServer);
  }


  LongAdder timer = new LongAdder();
  try (TimerContext ignored = TimerContext.timerMillis(timer::add)) {
    runParallelClients(invocationsPerClient, clientThreads, clientMaxWait);
  }

  System.out.println(String.format("Target time %dms - Time used %dms", targetTime, timer.longValue()));
}

/**
 * A parallel sized stream of doubles generates the given number of values
 */
public void testDoublesCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 0;
    for (int reps = 0; reps < REPS; ++reps) {
        counter.reset();
        r.doubles(size).parallel().forEach(x -> {
            counter.increment();
        });
        assertEquals(counter.sum(), size);
        size += 524959;
    }
}

/**
 * forEachEntryInParallel traverses all entries
 */
public void testForEachEntryInParallel() {
    LongAdder adder = new LongAdder();
    ConcurrentHashMap<Long, Long> m = longMap();
    m.forEachEntry(1L, (Map.Entry<Long,Long> e) -> adder.add(e.getKey().longValue() + e.getValue().longValue()));
    assertEquals(adder.sum(), 3 * SIZE * (SIZE - 1) / 2);
}

/**
 * Mapped forEachEntrySequentially traverses the given
 * transformations of all entries
 */
public void testMappedForEachEntrySequentially() {
    LongAdder adder = new LongAdder();
    ConcurrentHashMap<Long, Long> m = longMap();
    m.forEachEntry(Long.MAX_VALUE, (Map.Entry<Long,Long> e) -> Long.valueOf(e.getKey().longValue() + e.getValue().longValue()),
                               (Long x) -> adder.add(x.longValue()));
    assertEquals(adder.sum(), 3 * SIZE * (SIZE - 1) / 2);
}

/**
 * forEachEntrySequentially traverses all entries
 */
public void testForEachEntrySequentially() {
    LongAdder adder = new LongAdder();
    ConcurrentHashMap<Long, Long> m = longMap();
    m.forEachEntry(Long.MAX_VALUE, (Map.Entry<Long,Long> e) -> adder.add(e.getKey().longValue() + e.getValue().longValue()));
    assertEquals(adder.sum(), 3 * SIZE * (SIZE - 1) / 2);
}

/**
 * A parallel unsized stream of ints generates at least 100 values
 */
public void testUnsizedIntsCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 100;
    r.ints().limit(size).parallel().forEach(x -> counter.increment());
    assertEquals(size, counter.sum());
}

/**
 * A parallel unsized stream of doubles generates at least 100 values
 */
public void testUnsizedDoublesCount() {
    LongAdder counter = new LongAdder();
    Random r = new Random();
    long size = 100;
    r.doubles().limit(size).parallel().forEach(x -> {
        counter.increment();
    });
    assertEquals(counter.sum(), size);
}

/**
 * Creates an operation that is part of a bundle.
 * @param input - Operation Input
 * @param bundleLink - Bundle Identifier
 * @param startTime - The start time of the transaction
 * @return Transaction that is part of a Bundle Operation
 * @throws EmptyBodyException - when input is null
 */
static public Transaction newTransaction(OpInput input, Long bundleLink, long startTime)
        throws EmptyBodyException {
    Map.Entry<LongAdder,List<Transaction>> bundleInfo =  (bundleLink != null) ? bundles.get(bundleLink) : null;
    if (bundleInfo == null) {
        bundleInfo = new AbstractMap.SimpleEntry<LongAdder,List<Transaction>>(new LongAdder(),
                new ArrayList<Transaction>());
        bundles.put(bundleLink, bundleInfo);
    }
    return (bundleLink != null) ? new Transaction(input, bundleLink, startTime, bundleInfo) :
                new Transaction(input, startTime);
}

/**
 * add adds given value to current, and sum returns current value
 */
public void testAddAndSum() {
    LongAdder ai = new LongAdder();
    ai.add(2);
    assertEquals(2, ai.sum());
    ai.add(-4);
    assertEquals(-2, ai.sum());
}

/**
 * sumThenReset() returns sum; subsequent sum() returns zero
 */
public void testSumThenReset() {
    LongAdder ai = new LongAdder();
    ai.add(2);
    assertEquals(2, ai.sum());
    assertEquals(2, ai.sumThenReset());
    assertEquals(0, ai.sum());
}

/**
 * A parallel unsized stream of longs generates at least 100 values
 */
public void testUnsizedLongsCount() {
    LongAdder counter = new LongAdder();
    Random r = new Random();
    long size = 100;
    r.longs().limit(size).parallel().forEach(x -> {
        counter.increment();
    });
    assertEquals(counter.sum(), size);
}

/**
 * longValue returns current value.
 */
public void testLongValue() {
    LongAdder ai = new LongAdder();
    assertEquals(0, ai.longValue());
    ai.increment();
    assertEquals(1, ai.longValue());
}

GeneralScheduler(TaskQueue taskQueue
        , LazyTaskQueue lazyTaskQueue
        , DuplicateRemover duplicateRemover
        , TaskFilter filter) {
    this.taskQueue = taskQueue;
    this.lazyTaskQueue = lazyTaskQueue;
    this.duplicateRemover = duplicateRemover;
    this.taskFilter = filter;
    this.totalTaskInput = new LongAdder();
    this.totalTaskOutput = new LongAdder();
    this.isSuspend = new AtomicBoolean();
}

@Test
public void breachTest() {
    System.out.println("BREACH NOTIFICATION TEST");
    G10Monitor monitor = new G10Monitor();
    final LongAdder count = new LongAdder();
    monitor.registerBreachNotificationHandler((breachNotification) -> {
        if (printToConcole) {
            System.out.println("breach notification:" + breachNotification.toString());
        }
        count.increment();
    });
    PriceOrderHelper helper = new PriceOrderHelper(monitor);
    //send orders - within bias limit for 1 minute bucket, 
    //but breaches for 10 minute bucket
    helper.setWallClock(1 * 1000);
    helper.rejectOrderWithBias(CcyPair.EURUSD, 0.0012, true);
    helper.setWallClock(50 * 1000);
    helper.rejectOrderWithBias(CcyPair.EURUSD, 0.0014, true);
    //should be no breaches below 0.0015 for 1 minute
    helper.setWallClock(70 * 1000);
    assertEquals(0, count.intValue());
    //tick to 10 minutes + and should see a breach
    helper.setWallClock(100 + 10 * 60 * 1000);
    assertEquals(1, count.intValue());

    //fire out of limit for non-monitored ccys - no effect
    helper.rejectOrderWithBias(CcyPair.GBPCAD, 0.14, true);
    helper.rejectOrderWithBias(CcyPair.GBPCAD, 0.34, true);
    helper.setWallClock(100 + 11 * 60 * 1000);
    assertEquals(1, count.intValue());
    helper.setWallClock(100 + 21 * 60 * 1000);

    //lets test 1 moinute breach but no 10 minute notifications
    helper.rejectOrderWithBias(CcyPair.EURUSD, 0.0032, true);
    helper.setWallClock(100 + 22 * 60 * 1000);
    assertEquals(2, count.intValue());
}

/**
 * Cleanups queue and deallocate directBuffers manually
 */
@TearDown(Level.Iteration)
public void tearDown() {
    LongAdder counter = new LongAdder();
    // cleanup
    meter.retain((meterId, timestamp, value) -> counter.increment());

    // cheat cleanup to initial size
    long cnt = counter.sum();
    for (int i = 0; i < cnt / 4096; i++)
        meter.retain((meterId, timestamp, value) -> counter.increment());

    System.out.println("processed: " + counter.sum() + " measures");
}

/**
 * Cleanups queue and deallocate directBuffers manually
 */
@TearDown(Level.Iteration)
public void tearDown() {
    LongAdder counter = new LongAdder();
    // cleanup
    meter.retain((meterId, timestamp, value) -> counter.increment());

    // cheat cleanup to initial size
    long cnt = counter.sum();
    for (int i = 0; i < cnt / 4096; i++)
        meter.retain((meterId, timestamp, value) -> counter.increment());

    System.out.println("processed: " + counter.sum() + " measures");
}

@Test
public void test()
throws Exception {
    final LongAdder connCounter = new LongAdder();
    final ExecutorService executor = Executors.newFixedThreadPool(CONCURRENCY);
    for(int i = 0; i < CONCURRENCY; i ++) {
        executor.submit(
            () -> {
                Channel conn;
                for(int j = 0; j < CONN_ATTEMPTS; j ++) {
                    try {
                        while(null == (conn = connPool.lease())) {
                            Thread.sleep(1);
                        }
                        conn.writeAndFlush(PAYLOAD.retain()).sync();
                        connPool.release(conn);
                        connCounter.increment();
                    } catch(final InterruptedException e) {
                        break;
                    } catch(final Throwable cause) {
                        cause.printStackTrace(System.err);
                    }
                }
            }
        );
    }
    executor.shutdown();
    executor.awaitTermination(TEST_TIME_SECONDS, TimeUnit.SECONDS);
    assertTrue(executor.isTerminated());
    assertEquals(
        CONCURRENCY * CONN_ATTEMPTS, connCounter.sum(),
        2 * CONCURRENCY * CONN_ATTEMPTS / FAIL_EVERY_CONN_ATTEMPT
    );
}

@Test
public void test()
throws Exception {
    final LongAdder connCounter = new LongAdder();
    final ExecutorService executor = Executors.newFixedThreadPool(CONCURRENCY);
    for(int i = 0; i < CONCURRENCY; i ++) {
        executor.submit(
            () -> {
                Channel conn;
                for(int j = 0; j < CONN_ATTEMPTS; j ++) {
                    try {
                        while(null == (conn = connPool.lease())) {
                            Thread.sleep(1);
                        }
                        conn.writeAndFlush(PAYLOAD.retain()).sync();
                        connPool.release(conn);
                        connCounter.increment();
                    } catch(final InterruptedException e) {
                        break;
                    } catch(final Throwable cause) {
                        cause.printStackTrace(System.err);
                    }
                }
            }
        );
    }
    executor.shutdown();
    executor.awaitTermination(TEST_TIME_SECONDS, TimeUnit.SECONDS);
    assertTrue(executor.isTerminated());
    assertEquals(
        CONCURRENCY * CONN_ATTEMPTS, connCounter.sum(),
        2 * CONCURRENCY * CONN_ATTEMPTS / FAIL_EVERY_CONN_ATTEMPT
    );
}

/**
 * A sequential sized stream of doubles generates the given number of values
 */
public void testDoublesCount() {
    LongAdder counter = new LongAdder();
    Random r = new Random();
    long size = 0;
    for (int reps = 0; reps < REPS; ++reps) {
        counter.reset();
        r.doubles(size).forEach(x -> {
            counter.increment();
        });
        assertEquals(counter.sum(), size);
        size += 524959;
    }
}

@Test
public void testReloadCalledOnChangeInConsul() {
    String filename = getTestFilename();
    String fieldName = RandomStringUtils.randomAlphanumeric(12);

    putConfigInConsul(filename, fieldName, RandomStringUtils.randomAlphanumeric(12));

    ConsulFileConfigurationSource configurationSource = createConfigurationSource(filename);
    ConsulWatchReloadStrategy reloadStrategy = ConsulWatchReloadStrategy.builder()
            .withConsulConfigurationSource(configurationSource)
            .build();

    LongAdder numberOfReloads = new LongAdder();

    try {
        reloadStrategy.start(numberOfReloads::increment);

        // KVCache calls his listeners after it starts
        assertThat(numberOfReloads.longValue()).isLessThanOrEqualTo(1);

        putConfigInConsul(filename, fieldName, RandomStringUtils.randomAlphanumeric(12));

        await("Reload called after change in consul")
                .atMost(5, TimeUnit.SECONDS)
                .until(() -> numberOfReloads.longValue() > 1);
    } finally {
        reloadStrategy.stop();
    }
}

@Test
public void testLongadder(){


    LongAdder adder = new LongAdder();
    adder.add(1);
    adder.add(2);
    // fixme
    System.out.println(adder.longValue());
}

@Test
@DisplayName( "Connection Reap" )
public void basicReapTest() throws SQLException {
    int MIN_POOL_SIZE = 40, MAX_POOL_SIZE = 100, CAllS = 1000, REAP_TIMEOUT_MS = 1000;

    AgroalDataSourceConfigurationSupplier configurationSupplier = new AgroalDataSourceConfigurationSupplier()
            .connectionPoolConfiguration( cp -> cp
                    .initialSize( MAX_POOL_SIZE )
                    .minSize( MIN_POOL_SIZE )
                    .maxSize( MAX_POOL_SIZE )
                    .reapTimeout( ofMillis( REAP_TIMEOUT_MS ) )
            );

    CountDownLatch allLatch = new CountDownLatch( MAX_POOL_SIZE );
    CountDownLatch destroyLatch = new CountDownLatch( MAX_POOL_SIZE - MIN_POOL_SIZE );
    LongAdder reapCount = new LongAdder();

    AgroalDataSourceListener listener = new ReapListener( allLatch, reapCount, destroyLatch );

    try ( AgroalDataSource dataSource = AgroalDataSource.from( configurationSupplier, listener ) ) {
        for ( int i = 0; i < CAllS; i++ ) {
            Connection connection = dataSource.getConnection();
            assertNotNull( connection.getSchema(), "Expected non null value" );
            connection.close();
        }
        try {
            logger.info( format( "Awaiting test of all the {0} connections on the pool", MAX_POOL_SIZE ) );
            if ( !allLatch.await( 3L * REAP_TIMEOUT_MS, MILLISECONDS ) ) {
                fail( format( "{0} connections not tested for reap", allLatch.getCount() ) );
            }
            logger.info( format( "Waiting for reaping of {0} connections ", MAX_POOL_SIZE - MIN_POOL_SIZE ) );
            if ( !destroyLatch.await( 2L * REAP_TIMEOUT_MS, MILLISECONDS ) ) {
                fail( format( "{0} idle connections not sent for destruction", destroyLatch.getCount() ) );
            }
            assertEquals( MAX_POOL_SIZE - MIN_POOL_SIZE, reapCount.longValue(), "Unexpected number of idle connections " );
        } catch ( InterruptedException e ) {
            fail( "Test fail due to interrupt" );
        }
    }
}

/**
 * toString returns current value.
 */
public void testToString() {
    LongAdder ai = new LongAdder();
    assertEquals("0", ai.toString());
    ai.increment();
    assertEquals(Long.toString(1), ai.toString());
}

/**
 * A parallel sized stream of ints generates the given number of values
 */
public void testIntsCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 0;
    for (int reps = 0; reps < REPS; ++reps) {
        counter.reset();
        r.ints(size).parallel().forEach(x -> {
            counter.increment();
        });
        assertEquals(counter.sum(), size);
        size += 524959;
    }
}

/**
 * A parallel sized stream of longs generates the given number of values
 */
public void testLongsCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 0;
    for (int reps = 0; reps < REPS; ++reps) {
        counter.reset();
        r.longs(size).parallel().forEach(x -> {
            counter.increment();
        });
        assertEquals(counter.sum(), size);
        size += 524959;
    }
}

/**
 * A parallel unsized stream of ints generates at least 100 values
 */
public void testUnsizedIntsCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 100;
    r.ints().limit(size).parallel().forEach(x -> {
        counter.increment();
    });
    assertEquals(counter.sum(), size);
}

/**
 * A parallel unsized stream of longs generates at least 100 values
 */
public void testUnsizedLongsCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 100;
    r.longs().limit(size).parallel().forEach(x -> {
        counter.increment();
    });
    assertEquals(counter.sum(), size);
}

/**
 * A parallel unsized stream of doubles generates at least 100 values
 */
public void testUnsizedDoublesCount() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 100;
    r.doubles().limit(size).parallel().forEach(x -> {
        counter.increment();
    });
    assertEquals(counter.sum(), size);
}

public void run() {
    try {
        barrier.await();
        LongAdder a = adder;
        for (int i = 0; i < incs; ++i)
            a.add(1L);
        result = a.sum();
        barrier.await();
    } catch (Throwable t) { throw new Error(t); }
}

/**
 * A sequential unsized stream of longs generates at least 100 values
 */
public void testUnsizedLongsCountSeq() {
    LongAdder counter = new LongAdder();
    ThreadLocalRandom r = ThreadLocalRandom.current();
    long size = 100;
    r.longs().limit(size).forEach(x -> {
        counter.increment();
    });
    assertEquals(counter.sum(), size);
}

/**
 * floatValue returns current value.
 */
public void testFloatValue() {
    LongAdder ai = new LongAdder();
    assertEquals(0.0f, ai.floatValue());
    ai.increment();
    assertEquals(1.0f, ai.floatValue());
}

public void run() {
    phaser.arriveAndAwaitAdvance();
    phaser.arriveAndAwaitAdvance();
    LongAdder a = adder;
    for (int i = 0; i < incs; ++i)
        a.increment();
    result = a.sum();
    phaser.arrive();
}

/**
 * A parallel sized stream of ints generates the given number of values
 */
public void testIntsCount() {
    LongAdder counter = new LongAdder();
    SplittableRandom r = new SplittableRandom();
    long size = 0;
    for (int reps = 0; reps < REPS; ++reps) {
        counter.reset();
        r.ints(size).parallel().forEach(x -> {counter.increment();});
        assertEquals(counter.sum(), size);
        size += 524959;
    }
}