test/unit/org/apache/cassandra/concurrent/AbstractExecutorPlusTest.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.cassandra.concurrent;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;

import org.junit.Assert;
import org.junit.Ignore;

import org.apache.cassandra.utils.Closeable;
import org.apache.cassandra.utils.WithResources;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.Semaphore;

import static org.apache.cassandra.utils.concurrent.Semaphore.newSemaphore;

@Ignore
public abstract class AbstractExecutorPlusTest
{
    interface Verify<V>
    {
        void test(V test) throws Throwable;
    }

    static <V> Verify<V> ignoreNull(Verify<V> verify)
    {
        return test -> { if (test != null) verify.test(test); };
    }

    public <E extends ExecutorPlus> void testPooled(Supplier<ExecutorBuilder<? extends E>> builders) throws Throwable
    {
        testSuccess(builders);
        testFailure(builders);
    }

    public <E extends SequentialExecutorPlus> void testSequential(Supplier<ExecutorBuilder<? extends E>> builders) throws Throwable
    {
        testSuccess(builders);
        testFailure(builders);
        testAtLeastOnce(builders);
    }

    Runnable wrapSubmit(Runnable submit)
    {
        return submit;
    }

    public <E extends ExecutorPlus> void testSuccess(Supplier<ExecutorBuilder<? extends E>> builders) throws Throwable
    {
        testExecution(builders.get().build(), wrapSubmit(() -> {}), ignoreNull(Future::get));
        testExecution(builders.get().build(), WithResources.none(), wrapSubmit(() -> {}), ignoreNull(Future::get));
    }

    public <E extends ExecutorPlus> void testFailure(Supplier<ExecutorBuilder<? extends E>> builders) throws Throwable
    {
        ExecutorBuilder<? extends E> builder = builders.get();
        AtomicReference<Throwable> failure = new AtomicReference<>();
        Thread.UncaughtExceptionHandler ueh = (thread, f) -> failure.set(f);
        builder.withUncaughtExceptionHandler(ueh);
        Verify<Future<?>> verify = f -> {
            int c = 0;
            while (f == null && failure.get() == null && c++ < 100000)
                Thread.yield();
            Assert.assertTrue(failure.get() instanceof OutOfMemoryError);
            if (f != null)
                Assert.assertTrue(f.cause() instanceof OutOfMemoryError);
            failure.set(null);
        };
        Runnable submit = wrapSubmit(() -> { throw new OutOfMemoryError(); });
        testExecution(builder.build(), submit, verify);
        testExecution(builder.build(), WithResources.none(), submit, verify);
        testFailGetWithResources(builder.build(), () -> { throw new OutOfMemoryError(); }, verify);
        testFailCloseWithResources(builder.build(), () -> () -> { throw new OutOfMemoryError(); }, verify);
    }

    public <E extends SequentialExecutorPlus> void testAtLeastOnce(Supplier<ExecutorBuilder<? extends E>> builders) throws Throwable
    {
        ExecutorBuilder<? extends E> builder = builders.get();
        AtomicReference<Throwable> failure = new AtomicReference<>();
        Thread.UncaughtExceptionHandler ueh = (thread, f) -> failure.set(f);
        builder.withUncaughtExceptionHandler(ueh);

        SequentialExecutorPlus exec = builder.build();

        Semaphore enter = newSemaphore(0);
        Semaphore exit = newSemaphore(0);
        Semaphore runAfter = newSemaphore(0);
        SequentialExecutorPlus.AtLeastOnceTrigger trigger;
        trigger = exec.atLeastOnceTrigger(() -> { enter.release(1); exit.acquireThrowUncheckedOnInterrupt(1); });

        // check runAfter runs immediately
        trigger.runAfter(() -> runAfter.release(1));
        Assert.assertTrue(runAfter.tryAcquire(1, 1L, TimeUnit.SECONDS));

        Assert.assertTrue(trigger.trigger());
        enter.acquire(1);
        Assert.assertTrue(trigger.trigger());
        Assert.assertFalse(trigger.trigger());
        trigger.runAfter(() -> runAfter.release(1));
        Assert.assertFalse(runAfter.tryAcquire(1, 10L, TimeUnit.MILLISECONDS));
        exit.release(1);
        enter.acquire(1);
        Assert.assertFalse(runAfter.tryAcquire(1, 10L, TimeUnit.MILLISECONDS));
        Assert.assertTrue(trigger.trigger());
        Assert.assertFalse(trigger.trigger());
        exit.release(1);
        Assert.assertTrue(runAfter.tryAcquire(1, 1L, TimeUnit.SECONDS));
        exit.release(1);

        trigger = exec.atLeastOnceTrigger(() -> { throw new OutOfMemoryError(); });
        trigger.trigger();
        trigger.sync();
        Assert.assertTrue(failure.get() instanceof OutOfMemoryError);
    }

    void testExecution(ExecutorPlus e, WithResources withResources, Runnable submit, Verify<Future<?>> verify) throws Throwable
    {
        AtomicInteger i = new AtomicInteger();
        e.execute(() -> { i.incrementAndGet(); return withResources.get(); } , () -> { i.incrementAndGet(); submit.run(); });
        while (i.get() < 2) Thread.yield();
        verify.test(null);
        verify.test(e.submit(() -> { i.incrementAndGet(); return withResources.get(); }, () -> { i.incrementAndGet(); submit.run(); return null; }).await());
        Assert.assertEquals(4, i.get());
        verify.test(e.submit(() -> { i.incrementAndGet(); return withResources.get(); }, () -> { i.incrementAndGet(); submit.run(); }).await());
        Assert.assertEquals(6, i.get());
        verify.test(e.submit(() -> { i.incrementAndGet(); return withResources.get(); }, () -> { i.incrementAndGet(); submit.run(); }, null).await());
        Assert.assertEquals(8, i.get());
    }

    void testExecution(ExecutorPlus e, Runnable submit, Verify<Future<?>> verify) throws Throwable
    {
        AtomicInteger i = new AtomicInteger();
        e.execute(() -> { i.incrementAndGet(); submit.run(); });
        while (i.get() < 1) Thread.yield();
        verify.test(null);
        e.maybeExecuteImmediately(() -> { i.incrementAndGet(); submit.run(); });
        while (i.get() < 2) Thread.yield();
        verify.test(null);
        verify.test(e.submit(() -> { i.incrementAndGet(); submit.run(); return null; }).await());
        Assert.assertEquals(3, i.get());
        verify.test(e.submit(() -> { i.incrementAndGet(); submit.run(); }).await());
        Assert.assertEquals(4, i.get());
        verify.test(e.submit(() -> { i.incrementAndGet(); submit.run(); }, null).await());
        Assert.assertEquals(5, i.get());
    }

    void testFailGetWithResources(ExecutorPlus e, WithResources withResources, Verify<Future<?>> verify) throws Throwable
    {
        AtomicInteger i = new AtomicInteger();
        WithResources countingOnGetResources = () -> { i.incrementAndGet(); return withResources.get(); };
        AtomicBoolean executed = new AtomicBoolean();
        e.execute(countingOnGetResources, () -> executed.set(true));
        while (i.get() < 1) Thread.yield();
        verify.test(null);
        Assert.assertFalse(executed.get());
        verify.test(e.submit(countingOnGetResources, () -> { executed.set(true); return null; } ).await());
        Assert.assertEquals(2, i.get());
        Assert.assertFalse(executed.get());
        verify.test(e.submit(countingOnGetResources, () -> { executed.set(true); }).await());
        Assert.assertEquals(3, i.get());
        Assert.assertFalse(executed.get());
        verify.test(e.submit(countingOnGetResources, () -> { executed.set(true); }, null).await());
        Assert.assertEquals(4, i.get());
        Assert.assertFalse(executed.get());
    }

    void testFailCloseWithResources(ExecutorPlus e, WithResources withResources, Verify<Future<?>> verify) throws Throwable
    {
        AtomicInteger i = new AtomicInteger();
        WithResources countingOnCloseResources = () -> { Closeable close = withResources.get(); return () -> { i.incrementAndGet(); close.close(); }; };
        e.execute(countingOnCloseResources, i::incrementAndGet);
        while (i.get() < 2) Thread.yield();
        verify.test(null);
        verify.test(e.submit(countingOnCloseResources, () -> { i.incrementAndGet(); return null; } ).await());
        Assert.assertEquals(4, i.get());
        verify.test(e.submit(countingOnCloseResources, () -> { i.incrementAndGet(); }).await());
        Assert.assertEquals(6, i.get());
        verify.test(e.submit(countingOnCloseResources, () -> { i.incrementAndGet(); }, null).await());
        Assert.assertEquals(8, i.get());
    }

}