src/main/java/com/google/devtools/build/lib/concurrent/TaskFifo.java - bazel - Git at Google

 // Copyright 2023 The Bazel Authors. All rights reserved.
 //
 // 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.devtools.build.lib.concurrent;

 import static com.google.common.base.MoreObjects.toStringHelper;
 import static com.google.devtools.build.lib.concurrent.PriorityWorkerPool.TASKS_MAX_VALUE;

 import com.google.common.annotations.VisibleForTesting;
 import com.google.devtools.build.lib.unsafe.UnsafeProvider;
 import java.util.Arrays;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import sun.misc.Unsafe;

 /**
  * A fixed-capacity concurrent FIFO for tasks used by {@link PriorityWorkerPool}.
  *
  * <p>This class is a higher performance, nearly garbage-free, but less flexible substitute for
  * {@link ConcurrentLinkedQueue},
  *
  * <ul>
  *   <li>The queue capacity is fixed.
  *   <li>The client must guarantee not to take more than it has added.
  *   <li>The client must have an fallback if {@link TaskFifo#tryAppend} fails.
  * </ul>
  *
  * <p>This class is inspired by Morrison, Adam, and Yehuda Afek. "Fast concurrent queues for x86
  * processors." Proceedings of the 18th ACM SIGPLAN symposium on Principles and practice of parallel
  * programming. 2013.
  */
 final class TaskFifo {
   private static final Integer SKIP_SLOW_APPENDER = 1;

   /**
    * The power of 2 backing array capacity.
    *
    * <p>This is one more than the number of elements the queue can contain at one time. This helps
    * the efficiency of bits used to represent the number of elements enqueued. For example, the
    * number of bits needed to represent the element count for a queue of size 256 is 9, but only 8
    * bits are needed for a queue of size 255.
    */
   @VisibleForTesting static final int CAPACITY = TASKS_MAX_VALUE + 1;

   /** AND with this mask performs modulo {@link #CAPACITY}. */
   private static final int CAPACITY_MASK = CAPACITY - 1;

   /**
    * Circular buffer containing tasks and skip metadata.
    *
    * <p>The algorithm assigns to each caller of {@link #tryAppend} or {@link #take} a monotonically
    * increasing index (not including {@link #tryAppend} calls that would exceed capacity). {@link
    * #take} cannot be called more times than successful {@link tryAppend} calls by client
    * restriction. Thus each taker is assigned to a previous appender by matching index.
    *
    * <p>The naive algorithm based on the above would not be lock-free due to slow or descheduled
    * threads. For example, consider a taker assigned to an index where the corresponding appender's
    * thread has been descheduled before writing its task to the queue. When the taker observes its
    * assigned queue position, it does not see a value. The converse scenario is also possible. An
    * appender could see an occupied queue position while expecting {@code null} when a taker on the
    * same cyclic offset from a previous epoch is slow.
    *
    * <p>To resolve these situations, the threads that encounter them actively place a skip marker
    * into the queue that needs to be consumed by its counterpart. A taker that observes a null value
    * places a {@link Integer} {@code 1} to mark it, then skips to the next index. On seeing the
    * marker, the appender decrements it (with {@code 1} transitioning back to {@code null}). The
    * taker, when skipping to the next index, can expect to find a value there because an incomplete
    * append does not count as a successful one so there should be an extra complete append at a
    * subsequent index. Likewise, appenders that have looped all the way around to the same offset,
    * should expect to find an empty queue position due to capacity constraints.
    *
    * <p>Appenders that observe a value when expecting an empty position wrap the value with {@link
    * TaskWithSkippedAppends} then skip to the next available index. Slow takers decrement the counts
    * on the wrappers then skip to the next available index.
    *
    * <p>The skip marker has a count because the number of threads that could potentially be
    * descheduled at a particular index is only limited by the queue capacity, though more than one
    * should be extremely rare.
    *
    * <p>Each queue position contains one of the following.
    *
    * <ul>
    *   <li>{@code null} is an empty position.
    *   <li>{@link Runnable} is a position containing a task.
    *   <li>{@link Integer} is a count of takers that skipped the position because they observed a
    *       null value. The count corresponds to slow appenders at the position.
    *   <li>{@link TaskWithSkippedAppends} is a task with a count of appenders that skipped the
    *       position due to it being still occupied with a task. The count corresponds to slow takers
    *       assigned to the position.
    * </ul>
    *
    * <p>Correctness of the algorithm is straightforward. Anytime a taker skips a position, it adds a
    * count so that the same number of appenders skip that position and vice versa. Therefore the
    * number of takers and appenders skipping any given position stays balanced so the take and
    * append indices stay synchronized.
    */
   private final Object[] queue = new Object[CAPACITY];

   /**
    * Address of index for appending; incremented by appending.
    *
    * <p>The actual array offset is the value modulo {@link #CAPACITY}.
    */
   private final long appendIndexAddress;

   /**
    * Address of index for taking; incremented by taking.
    *
    * <p>The actual array offset is the value modulo {@link #CAPACITY}.
    */
   private final long takeIndexAddress;

   /**
    * The queue contains no more than this many tasks.
    *
    * <p>This is incremented before appending and decremented after taking.
    */
   private final long sizeAddress;

   /**
    * Constructor.
    *
    * <p>The caller owns the memory associated with the provided addresses.
    *
    * @param sizeAddress padded location of the {@code int} size of this queue.
    * @param takeIndexAddress padded location of the {@code int} take index.
    * @param appendIndexAddress padded location of the {@code int} append index.
    */
   TaskFifo(long sizeAddress, long appendIndexAddress, long takeIndexAddress) {
     this.sizeAddress = sizeAddress;
     this.appendIndexAddress = appendIndexAddress;
     this.takeIndexAddress = takeIndexAddress;

     // Explicitly initializes the provided addresses.
     UNSAFE.putInt(null, sizeAddress, 0);
     UNSAFE.putInt(null, appendIndexAddress, 0);
     UNSAFE.putInt(null, takeIndexAddress, 0);
   }

   /**
    * Tries to insert a task into the queue.
    *
    * @return true if successful, false if it would have exceeded the capacity.
    */
   boolean tryAppend(Runnable task) {
     // Optimistically increases size, and rolls back if it exceeds capacity.
     if (UNSAFE.getAndAddInt(null, sizeAddress, 1) >= TASKS_MAX_VALUE) {
       UNSAFE.getAndAddInt(null, sizeAddress, -1);
       return false;
     }

     do {
       int offset = getQueueOffset(UNSAFE.getAndAddInt(null, appendIndexAddress, 1));
       // In the common case, we can avoid an extra read.
       if (UNSAFE.compareAndSwapObject(queue, offset, null, task)) {
         return true;
       }
       do {
         // A plain read is sufficient here because this is always preceded by a failed CAS of the
         // same memory location.
         Object snapshot = UNSAFE.getObject(queue, offset);
         // It's possible that the taker outraced the snapshot above.
         if (snapshot == null) {
           if (UNSAFE.compareAndSwapObject(queue, offset, null, task)) {
             return true;
           }
           continue; // Refreshes the snapshot.
         }
         // There's some slowness that has to be resolved.
         Object target;
         if (snapshot instanceof Integer) {
           // There were previous takes without corresponding appends. Acknowledges a taker that
           // skipped this offset.
           int newCount = ((Integer) snapshot) - 1;
           target = newCount == 0 ? null : newCount;
         } else if (snapshot instanceof Runnable) {
           // A taker was slow.
           target = new TaskWithSkippedAppends((Runnable) snapshot, /* skippedAppendCount= */ 1);
         } else {
           // Multiple takers are slow. This should be very rare. Increments the skip count.
           target = ((TaskWithSkippedAppends) snapshot).incrementSkips();
         }
         if (UNSAFE.compareAndSwapObject(queue, offset, snapshot, target)) {
           break; // Success, skips to next.
         } // Otherwise refreshes the snapshot.
       } while (true);
     } while (true);
   }

   /**
    * Takes an available task.
    *
    * <p>This must not be called more times than {@link #tryAppend} has succeeded.
    */
   Runnable take() {
     do {
       int offset = getQueueOffset(UNSAFE.getAndAddInt(null, takeIndexAddress, 1));
       do {
         // A plain read is sufficient here.
         // 1. The initial read is supported by the client. In most cases, the client establishes the
         //    necessary happens-before relationship in honoring the constraint of no more takes than
         //    successful appends.
         // 2. On subsequent reads, this immediately follows a failed CAS of the same memory
         //    location, which refreshes the memory.
         Object snapshot = UNSAFE.getObject(queue, offset);
         if (snapshot instanceof Runnable) {
           // Attempts to take ownership of the task.
           if (UNSAFE.compareAndSwapObject(queue, offset, snapshot, null)) {
             UNSAFE.getAndAddInt(null, sizeAddress, -1);
             return (Runnable) snapshot;
           }
         } else {
           Object target;
           if (snapshot == null) {
             target = SKIP_SLOW_APPENDER;
           } else if (snapshot instanceof Integer) {
             // Increments the count due to multiple slow appenders, which should be very rare.
             target = ((Integer) snapshot).intValue() + 1;
           } else {
             // There have been appends without corresponding takes. Acknowledges one skip.
             target = ((TaskWithSkippedAppends) snapshot).decrementSkips();
           }
           if (UNSAFE.compareAndSwapObject(queue, offset, snapshot, target)) {
             break; // Success, skips to next.
           } // Otherwise refreshes the snapshot.
         }
       } while (true);
     } while (true);
   }

   int size() {
     return UNSAFE.getIntVolatile(null, sizeAddress);
   }

   void clear() {
     UNSAFE.putInt(null, sizeAddress, 0);
     UNSAFE.putInt(null, appendIndexAddress, 0);
     UNSAFE.putInt(null, takeIndexAddress, 0);
     Arrays.fill(queue, null);
   }

   @Override
   public String toString() {
     int appendIndex = UNSAFE.getIntVolatile(null, appendIndexAddress);
     int takeIndex = UNSAFE.getIntVolatile(null, takeIndexAddress);
     var helper =
         toStringHelper(this)
             .add("size", UNSAFE.getIntVolatile(null, sizeAddress))
             .add("appendIndex", String.format("%d (%d)", appendIndex, appendIndex & CAPACITY_MASK))
             .add("takeIndex", String.format("%d (%d)", takeIndex, takeIndex & CAPACITY_MASK));
     StringBuilder buf = new StringBuilder("[");
     for (int i = 0; i < CAPACITY; ++i) {
       if (i > 0) {
         buf.append(',');
       }
       if (i % 10 == 0) {
         buf.append(i).append(':');
       }
       var elt = queue[i];
       if (elt == null) {
         buf.append('0');
       } else if (elt instanceof Runnable) {
         buf.append('1');
       } else if (elt instanceof Integer) {
         buf.append('S').append(elt);
       } else {
         buf.append('T').append(((TaskWithSkippedAppends) elt).skippedAppendCount);
       }
     }
     helper.add("queue", buf.append(']').toString());
     return helper.toString();
   }

   @VisibleForTesting
   Object[] getQueueForTesting() {
     return queue;
   }

   private static int getQueueOffset(int index) {
     return TASKS_BASE + TASKS_SCALE * (index & CAPACITY_MASK);
   }

   @VisibleForTesting
   static class TaskWithSkippedAppends {
     private final Runnable task;
     private final int skippedAppendCount;

     private TaskWithSkippedAppends(Runnable task, int skippedAppendCount) {
       this.task = task;
       this.skippedAppendCount = skippedAppendCount;
     }

     private Object decrementSkips() {
       if (skippedAppendCount <= 1) {
         return task;
       }
       return new TaskWithSkippedAppends(task, skippedAppendCount - 1);
     }

     private TaskWithSkippedAppends incrementSkips() {
       return new TaskWithSkippedAppends(task, skippedAppendCount + 1);
     }

     @VisibleForTesting
     Runnable taskForTesting() {
       return task;
     }

     @VisibleForTesting
     int skippedAppendCountForTesting() {
       return skippedAppendCount;
     }
   }

   private static final Unsafe UNSAFE = UnsafeProvider.unsafe();

   private static final int TASKS_BASE = Unsafe.ARRAY_OBJECT_BASE_OFFSET;
   private static final int TASKS_SCALE = Unsafe.ARRAY_OBJECT_INDEX_SCALE;
 }
	// Copyright 2023 The Bazel Authors. All rights reserved.
	//
	// 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.devtools.build.lib.concurrent;

	import static com.google.common.base.MoreObjects.toStringHelper;
	import static com.google.devtools.build.lib.concurrent.PriorityWorkerPool.TASKS_MAX_VALUE;

	import com.google.common.annotations.VisibleForTesting;
	import com.google.devtools.build.lib.unsafe.UnsafeProvider;
	import java.util.Arrays;
	import java.util.concurrent.ConcurrentLinkedQueue;
	import sun.misc.Unsafe;

	/**
	* A fixed-capacity concurrent FIFO for tasks used by {@link PriorityWorkerPool}.
	*
	* <p>This class is a higher performance, nearly garbage-free, but less flexible substitute for
	* {@link ConcurrentLinkedQueue},
	*
	* <ul>
	* <li>The queue capacity is fixed.
	* <li>The client must guarantee not to take more than it has added.
	* <li>The client must have an fallback if {@link TaskFifo#tryAppend} fails.
	* </ul>
	*
	* <p>This class is inspired by Morrison, Adam, and Yehuda Afek. "Fast concurrent queues for x86
	* processors." Proceedings of the 18th ACM SIGPLAN symposium on Principles and practice of parallel
	* programming. 2013.
	*/
	final class TaskFifo {
	private static final Integer SKIP_SLOW_APPENDER = 1;

	/**
	* The power of 2 backing array capacity.
	*
	* <p>This is one more than the number of elements the queue can contain at one time. This helps
	* the efficiency of bits used to represent the number of elements enqueued. For example, the
	* number of bits needed to represent the element count for a queue of size 256 is 9, but only 8
	* bits are needed for a queue of size 255.
	*/
	@VisibleForTesting static final int CAPACITY = TASKS_MAX_VALUE + 1;

	/** AND with this mask performs modulo {@link #CAPACITY}. */
	private static final int CAPACITY_MASK = CAPACITY - 1;

	/**
	* Circular buffer containing tasks and skip metadata.
	*
	* <p>The algorithm assigns to each caller of {@link #tryAppend} or {@link #take} a monotonically
	* increasing index (not including {@link #tryAppend} calls that would exceed capacity). {@link
	* #take} cannot be called more times than successful {@link tryAppend} calls by client
	* restriction. Thus each taker is assigned to a previous appender by matching index.
	*
	* <p>The naive algorithm based on the above would not be lock-free due to slow or descheduled
	* threads. For example, consider a taker assigned to an index where the corresponding appender's
	* thread has been descheduled before writing its task to the queue. When the taker observes its
	* assigned queue position, it does not see a value. The converse scenario is also possible. An
	* appender could see an occupied queue position while expecting {@code null} when a taker on the
	* same cyclic offset from a previous epoch is slow.
	*
	* <p>To resolve these situations, the threads that encounter them actively place a skip marker
	* into the queue that needs to be consumed by its counterpart. A taker that observes a null value
	* places a {@link Integer} {@code 1} to mark it, then skips to the next index. On seeing the
	* marker, the appender decrements it (with {@code 1} transitioning back to {@code null}). The
	* taker, when skipping to the next index, can expect to find a value there because an incomplete
	* append does not count as a successful one so there should be an extra complete append at a
	* subsequent index. Likewise, appenders that have looped all the way around to the same offset,
	* should expect to find an empty queue position due to capacity constraints.
	*
	* <p>Appenders that observe a value when expecting an empty position wrap the value with {@link
	* TaskWithSkippedAppends} then skip to the next available index. Slow takers decrement the counts
	* on the wrappers then skip to the next available index.
	*
	* <p>The skip marker has a count because the number of threads that could potentially be
	* descheduled at a particular index is only limited by the queue capacity, though more than one
	* should be extremely rare.
	*
	* <p>Each queue position contains one of the following.
	*
	* <ul>
	* <li>{@code null} is an empty position.
	* <li>{@link Runnable} is a position containing a task.
	* <li>{@link Integer} is a count of takers that skipped the position because they observed a
	* null value. The count corresponds to slow appenders at the position.
	* <li>{@link TaskWithSkippedAppends} is a task with a count of appenders that skipped the
	* position due to it being still occupied with a task. The count corresponds to slow takers
	* assigned to the position.
	* </ul>
	*
	* <p>Correctness of the algorithm is straightforward. Anytime a taker skips a position, it adds a
	* count so that the same number of appenders skip that position and vice versa. Therefore the
	* number of takers and appenders skipping any given position stays balanced so the take and
	* append indices stay synchronized.
	*/
	private final Object[] queue = new Object[CAPACITY];

	/**
	* Address of index for appending; incremented by appending.
	*
	* <p>The actual array offset is the value modulo {@link #CAPACITY}.
	*/
	private final long appendIndexAddress;

	/**
	* Address of index for taking; incremented by taking.
	*
	* <p>The actual array offset is the value modulo {@link #CAPACITY}.
	*/
	private final long takeIndexAddress;

	/**
	* The queue contains no more than this many tasks.
	*
	* <p>This is incremented before appending and decremented after taking.
	*/
	private final long sizeAddress;

	/**
	* Constructor.
	*
	* <p>The caller owns the memory associated with the provided addresses.
	*
	* @param sizeAddress padded location of the {@code int} size of this queue.
	* @param takeIndexAddress padded location of the {@code int} take index.
	* @param appendIndexAddress padded location of the {@code int} append index.
	*/
	TaskFifo(long sizeAddress, long appendIndexAddress, long takeIndexAddress) {
	this.sizeAddress = sizeAddress;
	this.appendIndexAddress = appendIndexAddress;
	this.takeIndexAddress = takeIndexAddress;

	// Explicitly initializes the provided addresses.
	UNSAFE.putInt(null, sizeAddress, 0);
	UNSAFE.putInt(null, appendIndexAddress, 0);
	UNSAFE.putInt(null, takeIndexAddress, 0);
	}

	/**
	* Tries to insert a task into the queue.
	*
	* @return true if successful, false if it would have exceeded the capacity.
	*/
	boolean tryAppend(Runnable task) {
	// Optimistically increases size, and rolls back if it exceeds capacity.
	if (UNSAFE.getAndAddInt(null, sizeAddress, 1) >= TASKS_MAX_VALUE) {
	UNSAFE.getAndAddInt(null, sizeAddress, -1);
	return false;
	}

	do {
	int offset = getQueueOffset(UNSAFE.getAndAddInt(null, appendIndexAddress, 1));
	// In the common case, we can avoid an extra read.
	if (UNSAFE.compareAndSwapObject(queue, offset, null, task)) {
	return true;
	}
	do {
	// A plain read is sufficient here because this is always preceded by a failed CAS of the
	// same memory location.
	Object snapshot = UNSAFE.getObject(queue, offset);
	// It's possible that the taker outraced the snapshot above.
	if (snapshot == null) {
	if (UNSAFE.compareAndSwapObject(queue, offset, null, task)) {
	return true;
	}
	continue; // Refreshes the snapshot.
	}
	// There's some slowness that has to be resolved.
	Object target;
	if (snapshot instanceof Integer) {
	// There were previous takes without corresponding appends. Acknowledges a taker that
	// skipped this offset.
	int newCount = ((Integer) snapshot) - 1;
	target = newCount == 0 ? null : newCount;
	} else if (snapshot instanceof Runnable) {
	// A taker was slow.
	target = new TaskWithSkippedAppends((Runnable) snapshot, /* skippedAppendCount= */ 1);
	} else {
	// Multiple takers are slow. This should be very rare. Increments the skip count.
	target = ((TaskWithSkippedAppends) snapshot).incrementSkips();
	}
	if (UNSAFE.compareAndSwapObject(queue, offset, snapshot, target)) {
	break; // Success, skips to next.
	} // Otherwise refreshes the snapshot.
	} while (true);
	} while (true);
	}

	/**
	* Takes an available task.
	*
	* <p>This must not be called more times than {@link #tryAppend} has succeeded.
	*/
	Runnable take() {
	do {
	int offset = getQueueOffset(UNSAFE.getAndAddInt(null, takeIndexAddress, 1));
	do {
	// A plain read is sufficient here.
	// 1. The initial read is supported by the client. In most cases, the client establishes the
	// necessary happens-before relationship in honoring the constraint of no more takes than
	// successful appends.
	// 2. On subsequent reads, this immediately follows a failed CAS of the same memory
	// location, which refreshes the memory.
	Object snapshot = UNSAFE.getObject(queue, offset);
	if (snapshot instanceof Runnable) {
	// Attempts to take ownership of the task.
	if (UNSAFE.compareAndSwapObject(queue, offset, snapshot, null)) {
	UNSAFE.getAndAddInt(null, sizeAddress, -1);
	return (Runnable) snapshot;
	}
	} else {
	Object target;
	if (snapshot == null) {
	target = SKIP_SLOW_APPENDER;
	} else if (snapshot instanceof Integer) {
	// Increments the count due to multiple slow appenders, which should be very rare.
	target = ((Integer) snapshot).intValue() + 1;
	} else {
	// There have been appends without corresponding takes. Acknowledges one skip.
	target = ((TaskWithSkippedAppends) snapshot).decrementSkips();
	}
	if (UNSAFE.compareAndSwapObject(queue, offset, snapshot, target)) {
	break; // Success, skips to next.
	} // Otherwise refreshes the snapshot.
	}
	} while (true);
	} while (true);
	}

	int size() {
	return UNSAFE.getIntVolatile(null, sizeAddress);
	}

	void clear() {
	UNSAFE.putInt(null, sizeAddress, 0);
	UNSAFE.putInt(null, appendIndexAddress, 0);
	UNSAFE.putInt(null, takeIndexAddress, 0);
	Arrays.fill(queue, null);
	}

	@Override
	public String toString() {
	int appendIndex = UNSAFE.getIntVolatile(null, appendIndexAddress);
	int takeIndex = UNSAFE.getIntVolatile(null, takeIndexAddress);
	var helper =
	toStringHelper(this)
	.add("size", UNSAFE.getIntVolatile(null, sizeAddress))
	.add("appendIndex", String.format("%d (%d)", appendIndex, appendIndex & CAPACITY_MASK))
	.add("takeIndex", String.format("%d (%d)", takeIndex, takeIndex & CAPACITY_MASK));
	StringBuilder buf = new StringBuilder("[");
	for (int i = 0; i < CAPACITY; ++i) {
	if (i > 0) {
	buf.append(',');
	}
	if (i % 10 == 0) {
	buf.append(i).append(':');
	}
	var elt = queue[i];
	if (elt == null) {
	buf.append('0');
	} else if (elt instanceof Runnable) {
	buf.append('1');
	} else if (elt instanceof Integer) {
	buf.append('S').append(elt);
	} else {
	buf.append('T').append(((TaskWithSkippedAppends) elt).skippedAppendCount);
	}
	}
	helper.add("queue", buf.append(']').toString());
	return helper.toString();
	}

	@VisibleForTesting
	Object[] getQueueForTesting() {
	return queue;
	}

	private static int getQueueOffset(int index) {
	return TASKS_BASE + TASKS_SCALE * (index & CAPACITY_MASK);
	}

	@VisibleForTesting
	static class TaskWithSkippedAppends {
	private final Runnable task;
	private final int skippedAppendCount;

	private TaskWithSkippedAppends(Runnable task, int skippedAppendCount) {
	this.task = task;
	this.skippedAppendCount = skippedAppendCount;
	}

	private Object decrementSkips() {
	if (skippedAppendCount <= 1) {
	return task;
	}
	return new TaskWithSkippedAppends(task, skippedAppendCount - 1);
	}

	private TaskWithSkippedAppends incrementSkips() {
	return new TaskWithSkippedAppends(task, skippedAppendCount + 1);
	}

	@VisibleForTesting
	Runnable taskForTesting() {
	return task;
	}

	@VisibleForTesting
	int skippedAppendCountForTesting() {
	return skippedAppendCount;
	}
	}

	private static final Unsafe UNSAFE = UnsafeProvider.unsafe();

	private static final int TASKS_BASE = Unsafe.ARRAY_OBJECT_BASE_OFFSET;
	private static final int TASKS_SCALE = Unsafe.ARRAY_OBJECT_INDEX_SCALE;
	}