下面列出了java.util.concurrent.atomic.AtomicReference#lazySet() 实例代码,或者点击链接到github查看源代码,也可以在右侧发表评论。
/** Drains the read buffer up to an amortized threshold. */
// @GuardedBy("evictionLock")
void drainReadBuffer(int bufferIndex) {
final long writeCount = readBufferWriteCount[bufferIndex].get();
for (int i = 0; i < READ_BUFFER_DRAIN_THRESHOLD; i++) {
final int index = (int) (readBufferReadCount[bufferIndex] & READ_BUFFER_INDEX_MASK);
final AtomicReference<Node<K, V>> slot = readBuffers[bufferIndex][index];
final Node<K, V> node = slot.get();
if (node == null) {
break;
}
slot.lazySet(null);
applyRead(node);
readBufferReadCount[bufferIndex]++;
}
readBufferDrainAtWriteCount[bufferIndex].lazySet(writeCount);
}
/** Drains the read buffer up to an amortized threshold. */
@GuardedBy("evictionLock")
void drainReadBuffer(int bufferIndex) {
final long writeCount = readBufferWriteCount[bufferIndex].get();
for (int i = 0; i < READ_BUFFER_DRAIN_THRESHOLD; i++) {
final int index = (int) (readBufferReadCount[bufferIndex] & READ_BUFFER_INDEX_MASK);
final AtomicReference<Node<K, V>> slot = readBuffers[bufferIndex][index];
final Node<K, V> node = slot.get();
if (node == null) {
break;
}
slot.lazySet(null);
applyRead(node);
readBufferReadCount[bufferIndex]++;
}
readBufferDrainAtWriteCount[bufferIndex].lazySet(writeCount);
}
private static ByteBuffer allocateDirectBufferIfNeeded(final SequentialFileFactory fileFactory,
final int requiredCapacity,
final AtomicReference<ByteBuffer> bufferRef) {
ByteBuffer buffer = bufferRef != null ? bufferRef.get() : null;
if (buffer != null && buffer.capacity() < requiredCapacity) {
fileFactory.releaseDirectBuffer(buffer);
buffer = null;
}
if (buffer == null) {
buffer = fileFactory.allocateDirectBuffer(requiredCapacity);
} else {
buffer.clear().limit(requiredCapacity);
}
if (bufferRef != null) {
bufferRef.lazySet(buffer);
}
return buffer;
}
private synchronized JournalLoadInformation load(final LoaderCallback loadManager,
final boolean changeData,
final JournalState replicationSync) throws Exception {
final JournalState state = this.state;
if (state == JournalState.STOPPED || state == JournalState.LOADED) {
throw new IllegalStateException("Journal " + this + " must be in " + JournalState.STARTED + " state, was " +
state);
}
if (state == replicationSync) {
throw new IllegalStateException("Journal cannot be in state " + JournalState.STARTED);
}
// AtomicReference is used only as a reference, not as an Atomic value
final AtomicReference<ByteBuffer> wholeFileBufferRef = new AtomicReference<>();
try {
return load(loadManager, changeData, replicationSync, wholeFileBufferRef);
} finally {
final ByteBuffer wholeFileBuffer = wholeFileBufferRef.get();
if (wholeFileBuffer != null) {
fileFactory.releaseDirectBuffer(wholeFileBuffer);
wholeFileBufferRef.lazySet(null);
}
}
}
/** Drains the read buffer up to an amortized threshold. */
@GuardedBy("evictionLock")
void drainReadBuffer(int bufferIndex) {
final long writeCount = readBufferWriteCount[bufferIndex].get();
for (int i = 0; i < READ_BUFFER_DRAIN_THRESHOLD; i++) {
final int index = (int) (readBufferReadCount[bufferIndex] & READ_BUFFER_INDEX_MASK);
final AtomicReference<Node<K, V>> slot = readBuffers[bufferIndex][index];
final Node<K, V> node = slot.get();
if (node == null) {
break;
}
slot.lazySet(null);
applyRead(node);
readBufferReadCount[bufferIndex]++;
}
readBufferDrainAtWriteCount[bufferIndex].lazySet(writeCount);
}
@Override
public void drainTo(Consumer<E> consumer) {
long head = readCounter;
long tail = relaxedWriteCounter();
long size = (tail - head);
if (size == 0) {
return;
}
do {
int index = (int) (head & BUFFER_MASK);
AtomicReference<E> slot = buffer[index];
E e = slot.get();
if (e == null) {
// not published yet
break;
}
slot.lazySet(null);
consumer.accept(e);
head++;
} while (head != tail);
lazySetReadCounter(head);
}
@Override
public void drainTo(Consumer<E> consumer) {
long head = readCounter;
long tail = relaxedWriteCounter();
long size = (tail - head);
if (size == 0) {
return;
}
do {
int index = (int) (head & BUFFER_MASK);
AtomicReference<E> slot = buffer[index];
E e = slot.get();
if (e == null) {
// not published yet
break;
}
slot.lazySet(null);
consumer.accept(e);
head++;
} while (head != tail);
lazySetReadCounter(head);
}
/** Drains the read buffer up to an amortized threshold. */
// @GuardedBy("evictionLock")
private void drainReadBuffer(int bufferIndex) {
final long writeCount = readBufferWriteCount[bufferIndex].get();
for (int i = 0; i < READ_BUFFER_DRAIN_THRESHOLD; i++) {
final int index = (int) (readBufferReadCount[bufferIndex] & READ_BUFFER_INDEX_MASK);
final AtomicReference<Node<K, V>> slot = readBuffers[bufferIndex][index];
final Node<K, V> node = slot.get();
if (node == null) {
break;
}
slot.lazySet(null);
applyRead(node);
readBufferReadCount[bufferIndex]++;
}
readBufferDrainAtWriteCount[bufferIndex].lazySet(writeCount);
}
/** Drains the read buffer up to an amortized threshold. */
// @GuardedBy("evictionLock")
private void drainReadBuffer(int bufferIndex) {
final long writeCount = readBufferWriteCount[bufferIndex].get();
for (int i = 0; i < ConcurrentLinkedHashMap.READ_BUFFER_DRAIN_THRESHOLD; i++) {
final int index = (int) (readBufferReadCount[bufferIndex] & ConcurrentLinkedHashMap.READ_BUFFER_INDEX_MASK);
final AtomicReference<Node<V>> slot = readBuffers[bufferIndex][index];
final Node<V> node = slot.get();
if (node == null) {
break;
}
slot.lazySet(null);
applyRead(node);
readBufferReadCount[bufferIndex]++;
}
readBufferDrainAtWriteCount[bufferIndex].lazySet(writeCount);
}
@Override
public void clear() {
evictionLock.lock();
try {
// Discard all entries
Node<K, V> node;
while ((node = evictionDeque.poll()) != null) {
data.remove(node.key, node);
makeDead(node);
}
// Discard all pending reads
for (AtomicReference<Node<K, V>>[] buffer : readBuffers) {
for (AtomicReference<Node<K, V>> slot : buffer) {
slot.lazySet(null);
}
}
// Apply all pending writes
Runnable task;
while ((task = writeBuffer.poll()) != null) {
task.run();
}
} finally {
evictionLock.unlock();
}
}
/**
* get returns the last value lazySet in same thread
*/
public void testGetLazySet() {
AtomicReference ai = new AtomicReference(one);
assertSame(one, ai.get());
ai.lazySet(two);
assertSame(two, ai.get());
ai.lazySet(m3);
assertSame(m3, ai.get());
}
@Override
public void clear() {
evictionLock.lock();
try {
// Discard all entries
Node<K, V> node;
while ((node = evictionDeque.poll()) != null) {
data.remove(node.key, node);
makeDead(node);
}
// Discard all pending reads
for (AtomicReference<Node<K, V>>[] buffer : readBuffers) {
for (AtomicReference<Node<K, V>> slot : buffer) {
slot.lazySet(null);
}
}
// Apply all pending writes
Runnable task;
while ((task = writeBuffer.poll()) != null) {
task.run();
}
} finally {
evictionLock.unlock();
}
}
@Override
public void clear() {
evictionLock.lock();
try {
// Discard all entries
Node<K, V> node;
while ((node = evictionDeque.poll()) != null) {
data.remove(node.key, node);
makeDead(node);
}
// Discard all pending reads
for (AtomicReference<Node<K, V>>[] buffer : readBuffers) {
for (AtomicReference<Node<K, V>> slot : buffer) {
slot.lazySet(null);
}
}
// Apply all pending writes
Runnable task;
while ((task = writeBuffer.poll()) != null) {
task.run();
}
} finally {
evictionLock.unlock();
}
}
@Override
public void drainTo(Consumer<E> consumer) {
for (int i = 0; i < BUFFER_SIZE; i++) {
final int index = (int) (readCounter & BUFFER_MASK);
final AtomicReference<Object> slot = buffer[index];
if (slot.get() instanceof Turn) {
break;
}
long next = readCounter + BUFFER_SIZE;
slot.lazySet(new Turn(next));
readCounter++;
}
}
/**
* get returns the last value lazySet in same thread
*/
public void testGetLazySet() {
AtomicReference ai = new AtomicReference(one);
assertSame(one, ai.get());
ai.lazySet(two);
assertSame(two, ai.get());
ai.lazySet(m3);
assertSame(m3, ai.get());
}
@Override
public void clear() {
evictionLock.lock();
try {
// Discard all entries
Node<K, V> node;
while ((node = evictionDeque.poll()) != null) {
data.remove(node.key, node);
makeDead(node);
}
// Discard all pending reads
for (AtomicReference<Node<K, V>>[] buffer : readBuffers) {
for (AtomicReference<Node<K, V>> slot : buffer) {
slot.lazySet(null);
}
}
// Apply all pending writes
Runnable task;
while ((task = writeBuffer.poll()) != null) {
task.run();
}
} finally {
evictionLock.unlock();
}
}
@Override
public void clear() {
evictionLock.lock();
try {
// Discard all entries
Node<V> node;
while ((node = evictionDeque.poll()) != null) {
data.remove(node.key, node);
makeDead(node);
}
// Discard all pending reads
for (AtomicReference<Node<V>>[] buffer : readBuffers) {
for (AtomicReference<Node<V>> slot : buffer) {
slot.lazySet(null);
}
}
// Apply all pending writes
Runnable task;
while ((task = writeBuffer.poll()) != null) {
task.run();
}
} finally {
evictionLock.unlock();
}
}