javatests/com/google/gerrit/acceptance/server/util/TaskListenerIT.java - gerrit.git - Git at Google

 // Copyright (C) 2022 The Android Open Source Project
 //
 // Licensed 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 com.google.gerrit.acceptance.server.util;

 import static com.google.common.truth.Truth.assertThat;

 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import com.google.gerrit.acceptance.AbstractDaemonTest;
 import com.google.gerrit.extensions.annotations.Exports;
 import com.google.gerrit.server.git.WorkQueue;
 import com.google.gerrit.server.git.WorkQueue.Task;
 import com.google.gerrit.server.git.WorkQueue.TaskListener;
 import com.google.inject.AbstractModule;
 import com.google.inject.Inject;
 import com.google.inject.Module;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import org.junit.Before;
 import org.junit.Test;

 public class TaskListenerIT extends AbstractDaemonTest {
   /**
    * Use a LatchedMethod in a method to allow another thread to await the method's call. Once
    * called, the call() method will block until another thread calls the complete() method or until
    * a preset timeout is reached.
    */
   public static class LatchedMethod<T> {
     private volatile T value;

     private final CountDownLatch called = new CountDownLatch(1);
     private final CountDownLatch complete = new CountDownLatch(1);

     /** Assert that the call() method has not yet been called */
     public void assertUncalled() {
       assertThat(called.getCount()).isEqualTo(1);
     }

     /**
      * Assert that a timeout does not occur while awaiting the call() to be called. Fails if the
      * waiting time elapses before the call() method is called, otherwise passes.
      */
     public void assertCalledEventually() {
       assertThat(await(called)).isEqualTo(true);
     }

     public T call() {
       called.countDown();
       await(complete);
       return getValue();
     }

     public T call(T val) {
       set(val);
       return call();
     }

     public T callAndAwaitComplete() throws InterruptedException {
       called.countDown();
       complete.await();
       return getValue();
     }

     public void complete() {
       complete.countDown();
     }

     public void set(T val) {
       value = val;
     }

     public void complete(T val) {
       set(val);
       complete();
     }

     public void assertCalledEventuallyThenComplete(T val) {
       assertCalledEventually();
       complete(val);
     }

     protected T getValue() {
       return value;
     }
   }

   public static class LatchedRunnable implements Runnable {
     public LatchedMethod<?> run = new LatchedMethod<>();
     public String name = "latched-runnable";

     public LatchedRunnable(String name) {
       this.name = name;
     }

     public LatchedRunnable() {}

     @Override
     public void run() {
       run.call();
     }

     @Override
     public String toString() {
       return name;
     }
   }

   public static class ForwardingListener<T extends TaskListener> implements TaskListener {
     public volatile T delegate;
     public volatile Task<?> task;

     public void resetDelegate(T listener) {
       delegate = listener;
       task = null;
     }

     @Override
     public void onStart(Task<?> task) {
       if (isDelegatable(task)) {
         delegate.onStart(task);
       }
     }

     @Override
     public void onStop(Task<?> task) {
       if (isDelegatable(task)) {
         delegate.onStop(task);
       }
     }

     protected boolean isDelegatable(Task<?> task) {
       if (delegate != null) {
         if (this.task == task) {
           return true;
         }
         if (this.task == null) {
           this.task = task;
           return true;
         }
       }
       return false;
     }
   }

   public static class LatchedListener implements TaskListener {
     public LatchedMethod<?> onStart = new LatchedMethod<>();
     public LatchedMethod<?> onStop = new LatchedMethod<>();

     @Override
     public void onStart(Task<?> task) {
       onStart.call();
     }

     @Override
     public void onStop(Task<?> task) {
       onStop.call();
     }
   }

   private static final int AWAIT_TIMEOUT = 20;
   private static final TimeUnit AWAIT_TIMEUNIT = TimeUnit.MILLISECONDS;
   private static final long MS_EMPTY_QUEUE =
       TimeUnit.MILLISECONDS.convert(50, TimeUnit.MILLISECONDS);

   private static ForwardingListener<TaskListener> forwarder;

   @Inject private WorkQueue workQueue;
   private ScheduledExecutorService executor;

   private final LatchedListener listener = new LatchedListener();
   private final LatchedRunnable runnable = new LatchedRunnable();

   @Override
   public Module createModule() {
     return new AbstractModule() {
       @Override
       public void configure() {
         // Forwarder.delegate is empty on start to protect test listener from non-test tasks (such
         // as the "Log File Manager") interference
         forwarder = new ForwardingListener<>(); // Only gets bound once for all tests
         bind(TaskListener.class).annotatedWith(Exports.named("listener")).toInstance(forwarder);
       }
     };
   }

   @Before
   public void setupExecutorAndForwarder() throws InterruptedException {
     executor = workQueue.createQueue(1, "TaskListeners");

     // "Log File Manager"s are likely running and will interfere with tests
     while (0 != workQueue.getTasks().size()) {
       for (Task<?> t : workQueue.getTasks()) {
         @SuppressWarnings("unused")
         boolean unused = t.cancel(true);
       }
       TimeUnit.MILLISECONDS.sleep(1);
     }

     forwarder.resetDelegate(listener);

     assertQueueSize(0);
     assertThat(forwarder.task).isEqualTo(null);
     listener.onStart.assertUncalled();
     runnable.run.assertUncalled();
     listener.onStop.assertUncalled();
   }

   @Test
   public void onStartThenRunThenOnStopAreCalled() throws Exception {
     int size = assertQueueBlockedOnExecution(runnable);

     // onStartThenRunThenOnStopAreCalled -> onStart...Called
     listener.onStart.assertCalledEventually();
     assertQueueSize(size);
     runnable.run.assertUncalled();
     listener.onStop.assertUncalled();

     listener.onStart.complete();
     // onStartThenRunThenOnStopAreCalled -> ...ThenRun...Called
     runnable.run.assertCalledEventually();
     listener.onStop.assertUncalled();

     runnable.run.complete();
     // onStartThenRunThenOnStopAreCalled -> ...ThenOnStop...Called
     listener.onStop.assertCalledEventually();
     assertQueueSize(size);

     listener.onStop.complete();
     assertTaskCountIsEventually(--size);
   }

   @Test
   public void firstBlocksSecond() throws Exception {
     int size = assertQueueBlockedOnExecution(runnable);

     // firstBlocksSecond -> first...
     listener.onStart.assertCalledEventually();
     assertQueueSize(size);

     LatchedRunnable runnable2 = new LatchedRunnable();
     size = assertQueueBlockedOnExecution(runnable2);

     // firstBlocksSecond -> ...BlocksSecond
     runnable2.run.assertUncalled();
     assertQueueSize(size); // waiting on first

     listener.onStart.complete();
     runnable.run.assertCalledEventually();
     assertQueueSize(size); // waiting on first
     runnable2.run.assertUncalled();

     runnable.run.complete();
     listener.onStop.assertCalledEventually();
     assertQueueSize(size); // waiting on first
     runnable2.run.assertUncalled();

     listener.onStop.complete();
     runnable2.run.assertCalledEventually();
     assertQueueSize(--size);

     runnable2.run.complete();
     assertTaskCountIsEventually(--size);
   }

   @Test
   public void states() throws Exception {
     executor.execute(runnable);
     listener.onStart.assertCalledEventually();
     assertStateIs(Task.State.STARTING);

     listener.onStart.complete();
     runnable.run.assertCalledEventually();
     assertStateIs(Task.State.RUNNING);

     runnable.run.complete();
     listener.onStop.assertCalledEventually();
     assertStateIs(Task.State.STOPPING);

     listener.onStop.complete();
     assertAwaitQueueIsEmpty();
     assertStateIs(Task.State.DONE);
   }

   /** Fails if the waiting time elapses before the count is reached, otherwise passes */
   public static void assertTaskCountIsEventually(WorkQueue workQueue, int count)
       throws InterruptedException {
     long ms = 0;
     while (count != workQueue.getTasks().size()) {
       assertThat(ms++).isLessThan(MS_EMPTY_QUEUE);
       TimeUnit.MILLISECONDS.sleep(1);
     }
   }

   public static void assertQueueSize(WorkQueue workQueue, int size) {
     assertThat(workQueue.getTasks().size()).isEqualTo(size);
   }

   @CanIgnoreReturnValue
   public static boolean await(CountDownLatch latch) {
     try {
       return latch.await(AWAIT_TIMEOUT, AWAIT_TIMEUNIT);
     } catch (InterruptedException e) {
       return false;
     }
   }

   public void assertTaskCountIsEventually(int count) throws InterruptedException {
     TaskListenerIT.assertTaskCountIsEventually(workQueue, count);
   }

   public static void assertStateIs(Task<?> task, Task.State state) {
     assertThat(task).isNotNull();
     assertThat(task.getState()).isEqualTo(state);
   }

   private void assertStateIs(Task.State state) {
     assertStateIs(forwarder.task, state);
   }

   private int assertQueueBlockedOnExecution(Runnable runnable) {
     int expectedSize = workQueue.getTasks().size() + 1;
     executor.execute(runnable);
     assertQueueSize(expectedSize);
     return expectedSize;
   }

   private void assertQueueSize(int size) {
     assertQueueSize(workQueue, size);
   }

   private void assertAwaitQueueIsEmpty() throws InterruptedException {
     assertTaskCountIsEventually(0);
   }
 }
	// Copyright (C) 2022 The Android Open Source Project
	//
	// Licensed 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 com.google.gerrit.acceptance.server.util;

	import static com.google.common.truth.Truth.assertThat;

	import com.google.errorprone.annotations.CanIgnoreReturnValue;
	import com.google.gerrit.acceptance.AbstractDaemonTest;
	import com.google.gerrit.extensions.annotations.Exports;
	import com.google.gerrit.server.git.WorkQueue;
	import com.google.gerrit.server.git.WorkQueue.Task;
	import com.google.gerrit.server.git.WorkQueue.TaskListener;
	import com.google.inject.AbstractModule;
	import com.google.inject.Inject;
	import com.google.inject.Module;
	import java.util.concurrent.CountDownLatch;
	import java.util.concurrent.ScheduledExecutorService;
	import java.util.concurrent.TimeUnit;
	import org.junit.Before;
	import org.junit.Test;

	public class TaskListenerIT extends AbstractDaemonTest {
	/**
	* Use a LatchedMethod in a method to allow another thread to await the method's call. Once
	* called, the call() method will block until another thread calls the complete() method or until
	* a preset timeout is reached.
	*/
	public static class LatchedMethod<T> {
	private volatile T value;

	private final CountDownLatch called = new CountDownLatch(1);
	private final CountDownLatch complete = new CountDownLatch(1);

	/** Assert that the call() method has not yet been called */
	public void assertUncalled() {
	assertThat(called.getCount()).isEqualTo(1);
	}

	/**
	* Assert that a timeout does not occur while awaiting the call() to be called. Fails if the
	* waiting time elapses before the call() method is called, otherwise passes.
	*/
	public void assertCalledEventually() {
	assertThat(await(called)).isEqualTo(true);
	}

	public T call() {
	called.countDown();
	await(complete);
	return getValue();
	}

	public T call(T val) {
	set(val);
	return call();
	}

	public T callAndAwaitComplete() throws InterruptedException {
	called.countDown();
	complete.await();
	return getValue();
	}

	public void complete() {
	complete.countDown();
	}

	public void set(T val) {
	value = val;
	}

	public void complete(T val) {
	set(val);
	complete();
	}

	public void assertCalledEventuallyThenComplete(T val) {
	assertCalledEventually();
	complete(val);
	}

	protected T getValue() {
	return value;
	}
	}

	public static class LatchedRunnable implements Runnable {
	public LatchedMethod<?> run = new LatchedMethod<>();
	public String name = "latched-runnable";

	public LatchedRunnable(String name) {
	this.name = name;
	}

	public LatchedRunnable() {}

	@Override
	public void run() {
	run.call();
	}

	@Override
	public String toString() {
	return name;
	}
	}

	public static class ForwardingListener<T extends TaskListener> implements TaskListener {
	public volatile T delegate;
	public volatile Task<?> task;

	public void resetDelegate(T listener) {
	delegate = listener;
	task = null;
	}

	@Override
	public void onStart(Task<?> task) {
	if (isDelegatable(task)) {
	delegate.onStart(task);
	}
	}

	@Override
	public void onStop(Task<?> task) {
	if (isDelegatable(task)) {
	delegate.onStop(task);
	}
	}

	protected boolean isDelegatable(Task<?> task) {
	if (delegate != null) {
	if (this.task == task) {
	return true;
	}
	if (this.task == null) {
	this.task = task;
	return true;
	}
	}
	return false;
	}
	}

	public static class LatchedListener implements TaskListener {
	public LatchedMethod<?> onStart = new LatchedMethod<>();
	public LatchedMethod<?> onStop = new LatchedMethod<>();

	@Override
	public void onStart(Task<?> task) {
	onStart.call();
	}

	@Override
	public void onStop(Task<?> task) {
	onStop.call();
	}
	}

	private static final int AWAIT_TIMEOUT = 20;
	private static final TimeUnit AWAIT_TIMEUNIT = TimeUnit.MILLISECONDS;
	private static final long MS_EMPTY_QUEUE =
	TimeUnit.MILLISECONDS.convert(50, TimeUnit.MILLISECONDS);

	private static ForwardingListener<TaskListener> forwarder;

	@Inject private WorkQueue workQueue;
	private ScheduledExecutorService executor;

	private final LatchedListener listener = new LatchedListener();
	private final LatchedRunnable runnable = new LatchedRunnable();

	@Override
	public Module createModule() {
	return new AbstractModule() {
	@Override
	public void configure() {
	// Forwarder.delegate is empty on start to protect test listener from non-test tasks (such
	// as the "Log File Manager") interference
	forwarder = new ForwardingListener<>(); // Only gets bound once for all tests
	bind(TaskListener.class).annotatedWith(Exports.named("listener")).toInstance(forwarder);
	}
	};
	}

	@Before
	public void setupExecutorAndForwarder() throws InterruptedException {
	executor = workQueue.createQueue(1, "TaskListeners");

	// "Log File Manager"s are likely running and will interfere with tests
	while (0 != workQueue.getTasks().size()) {
	for (Task<?> t : workQueue.getTasks()) {
	@SuppressWarnings("unused")
	boolean unused = t.cancel(true);
	}
	TimeUnit.MILLISECONDS.sleep(1);
	}

	forwarder.resetDelegate(listener);

	assertQueueSize(0);
	assertThat(forwarder.task).isEqualTo(null);
	listener.onStart.assertUncalled();
	runnable.run.assertUncalled();
	listener.onStop.assertUncalled();
	}

	@Test
	public void onStartThenRunThenOnStopAreCalled() throws Exception {
	int size = assertQueueBlockedOnExecution(runnable);

	// onStartThenRunThenOnStopAreCalled -> onStart...Called
	listener.onStart.assertCalledEventually();
	assertQueueSize(size);
	runnable.run.assertUncalled();
	listener.onStop.assertUncalled();

	listener.onStart.complete();
	// onStartThenRunThenOnStopAreCalled -> ...ThenRun...Called
	runnable.run.assertCalledEventually();
	listener.onStop.assertUncalled();

	runnable.run.complete();
	// onStartThenRunThenOnStopAreCalled -> ...ThenOnStop...Called
	listener.onStop.assertCalledEventually();
	assertQueueSize(size);

	listener.onStop.complete();
	assertTaskCountIsEventually(--size);
	}

	@Test
	public void firstBlocksSecond() throws Exception {
	int size = assertQueueBlockedOnExecution(runnable);

	// firstBlocksSecond -> first...
	listener.onStart.assertCalledEventually();
	assertQueueSize(size);

	LatchedRunnable runnable2 = new LatchedRunnable();
	size = assertQueueBlockedOnExecution(runnable2);

	// firstBlocksSecond -> ...BlocksSecond
	runnable2.run.assertUncalled();
	assertQueueSize(size); // waiting on first

	listener.onStart.complete();
	runnable.run.assertCalledEventually();
	assertQueueSize(size); // waiting on first
	runnable2.run.assertUncalled();

	runnable.run.complete();
	listener.onStop.assertCalledEventually();
	assertQueueSize(size); // waiting on first
	runnable2.run.assertUncalled();

	listener.onStop.complete();
	runnable2.run.assertCalledEventually();
	assertQueueSize(--size);

	runnable2.run.complete();
	assertTaskCountIsEventually(--size);
	}

	@Test
	public void states() throws Exception {
	executor.execute(runnable);
	listener.onStart.assertCalledEventually();
	assertStateIs(Task.State.STARTING);

	listener.onStart.complete();
	runnable.run.assertCalledEventually();
	assertStateIs(Task.State.RUNNING);

	runnable.run.complete();
	listener.onStop.assertCalledEventually();
	assertStateIs(Task.State.STOPPING);

	listener.onStop.complete();
	assertAwaitQueueIsEmpty();
	assertStateIs(Task.State.DONE);
	}

	/** Fails if the waiting time elapses before the count is reached, otherwise passes */
	public static void assertTaskCountIsEventually(WorkQueue workQueue, int count)
	throws InterruptedException {
	long ms = 0;
	while (count != workQueue.getTasks().size()) {
	assertThat(ms++).isLessThan(MS_EMPTY_QUEUE);
	TimeUnit.MILLISECONDS.sleep(1);
	}
	}

	public static void assertQueueSize(WorkQueue workQueue, int size) {
	assertThat(workQueue.getTasks().size()).isEqualTo(size);
	}

	@CanIgnoreReturnValue
	public static boolean await(CountDownLatch latch) {
	try {
	return latch.await(AWAIT_TIMEOUT, AWAIT_TIMEUNIT);
	} catch (InterruptedException e) {
	return false;
	}
	}

	public void assertTaskCountIsEventually(int count) throws InterruptedException {
	TaskListenerIT.assertTaskCountIsEventually(workQueue, count);
	}

	public static void assertStateIs(Task<?> task, Task.State state) {
	assertThat(task).isNotNull();
	assertThat(task.getState()).isEqualTo(state);
	}

	private void assertStateIs(Task.State state) {
	assertStateIs(forwarder.task, state);
	}

	private int assertQueueBlockedOnExecution(Runnable runnable) {
	int expectedSize = workQueue.getTasks().size() + 1;
	executor.execute(runnable);
	assertQueueSize(expectedSize);
	return expectedSize;
	}

	private void assertQueueSize(int size) {
	assertQueueSize(workQueue, size);
	}

	private void assertAwaitQueueIsEmpty() throws InterruptedException {
	assertTaskCountIsEventually(0);
	}
	}