Skip to content

D
dragonwell8_jdk

openanolis / dragonwell8_jdk

通知 3
Star 2
Fork 0
D
dragonwell8_jdk

前往新版Gitcode，体验更适合开发者的 AI 搜索 >>

提交 4219640f 编写于 作者: A alanb
浏览文件
操作

6971825: (so) improve scatter/gather implementation 

Reviewed-by: chegar, sherman
上级 ee7b3a88
隐藏空白更改
内联 并排
Showing with 409 addition and 305 deletion
src/share/classes/sun/nio/ch/DatagramChannelImpl.java
浏览文件 @ 4219640f
......@@ -536,9 +536,11 @@ class DatagramChannelImpl
}
}
private long read0(ByteBuffer[] bufs) throws IOException {
if (bufs == null)
throw new NullPointerException();
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
throw new IndexOutOfBoundsException();
synchronized (readLock) {
synchronized (stateLock) {
ensureOpen();
......@@ -552,7 +554,7 @@ class DatagramChannelImpl
return 0;
readerThread = NativeThread.current();
do {
n = IOUtil.read(fd, bufs, nd);
n = IOUtil.read(fd, dsts, offset, length, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
return IOStatus.normalize(n);
} finally {
......@@ -563,15 +565,6 @@ class DatagramChannelImpl
}
}
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
throw new IndexOutOfBoundsException();
// ## Fix IOUtil.write so that we can avoid this array copy
return read0(Util.subsequence(dsts, offset, length));
}
public int write(ByteBuffer buf) throws IOException {
if (buf == null)
throw new NullPointerException();
......@@ -599,9 +592,11 @@ class DatagramChannelImpl
}
}
private long write0(ByteBuffer[] bufs) throws IOException {
if (bufs == null)
throw new NullPointerException();
public long write(ByteBuffer[] srcs, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
throw new IndexOutOfBoundsException();
synchronized (writeLock) {
synchronized (stateLock) {
ensureOpen();
......@@ -615,7 +610,7 @@ class DatagramChannelImpl
return 0;
writerThread = NativeThread.current();
do {
n = IOUtil.write(fd, bufs, nd);
n = IOUtil.write(fd, srcs, offset, length, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
return IOStatus.normalize(n);
} finally {
......@@ -626,15 +621,6 @@ class DatagramChannelImpl
}
}
public long write(ByteBuffer[] srcs, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
throw new IndexOutOfBoundsException();
// ## Fix IOUtil.write so that we can avoid this array copy
return write0(Util.subsequence(srcs, offset, length));
}
protected void implConfigureBlocking(boolean block) throws IOException {
IOUtil.configureBlocking(fd, block);
}
......
src/share/classes/sun/nio/ch/FileChannelImpl.java
浏览文件 @ 4219640f
......@@ -143,7 +143,11 @@ public class FileChannelImpl
}
}
private long read0(ByteBuffer[] dsts) throws IOException {
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
throw new IndexOutOfBoundsException();
ensureOpen();
if (!readable)
throw new NonReadableChannelException();
......@@ -156,7 +160,7 @@ public class FileChannelImpl
if (!isOpen())
return 0;
do {
n = IOUtil.read(fd, dsts, nd);
n = IOUtil.read(fd, dsts, offset, length, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
return IOStatus.normalize(n);
} finally {
......@@ -167,15 +171,6 @@ public class FileChannelImpl
}
}
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
throw new IndexOutOfBoundsException();
// ## Fix IOUtil.write so that we can avoid this array copy
return read0(Util.subsequence(dsts, offset, length));
}
public int write(ByteBuffer src) throws IOException {
ensureOpen();
if (!writable)
......@@ -200,7 +195,11 @@ public class FileChannelImpl
}
}
private long write0(ByteBuffer[] srcs) throws IOException {
public long write(ByteBuffer[] srcs, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
throw new IndexOutOfBoundsException();
ensureOpen();
if (!writable)
throw new NonWritableChannelException();
......@@ -213,7 +212,7 @@ public class FileChannelImpl
if (!isOpen())
return 0;
do {
n = IOUtil.write(fd, srcs, nd);
n = IOUtil.write(fd, srcs, offset, length, nd);
} while ((n == IOStatus.INTERRUPTED) && isOpen());
return IOStatus.normalize(n);
} finally {
......@@ -224,16 +223,6 @@ public class FileChannelImpl
}
}
public long write(ByteBuffer[] srcs, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
throw new IndexOutOfBoundsException();
// ## Fix IOUtil.write so that we can avoid this array copy
return write0(Util.subsequence(srcs, offset, length));
}
// -- Other operations --
public long position() throws IOException {
......
src/share/classes/sun/nio/ch/IOUtil.java
浏览文件 @ 4219640f
......@@ -38,34 +38,6 @@ class IOUtil {
private IOUtil() { } // No instantiation
/*
* Returns the index of first buffer in bufs with remaining,
* or -1 if there is nothing left
*/
private static int remaining(ByteBuffer[] bufs) {
int numBufs = bufs.length;
for (int i=0; i<numBufs; i++) {
if (bufs[i].hasRemaining()) {
return i;
}
}
return -1;
}
/*
* Returns a new ByteBuffer array with only unfinished buffers in it
*/
private static ByteBuffer[] skipBufs(ByteBuffer[] bufs,
int nextWithRemaining)
{
int newSize = bufs.length - nextWithRemaining;
ByteBuffer[] temp = new ByteBuffer[newSize];
for (int i=0; i<newSize; i++) {
temp[i] = bufs[i + nextWithRemaining];
}
return temp;
}
static int write(FileDescriptor fd, ByteBuffer src, long position,
NativeDispatcher nd, Object lock)
throws IOException
......@@ -93,7 +65,7 @@ class IOUtil {
}
return n;
} finally {
Util.releaseTemporaryDirectBuffer(bb);
Util.offerFirstTemporaryDirectBuffer(bb);
}
}
......@@ -125,88 +97,81 @@ class IOUtil {
static long write(FileDescriptor fd, ByteBuffer[] bufs, NativeDispatcher nd)
throws IOException
{
int nextWithRemaining = remaining(bufs);
// if all bufs are empty we should return immediately
if (nextWithRemaining < 0)
return 0;
// If some bufs are empty we should skip them
if (nextWithRemaining > 0)
bufs = skipBufs(bufs, nextWithRemaining);
return write(fd, bufs, 0, bufs.length, nd);
}
int numBufs = bufs.length;
static long write(FileDescriptor fd, ByteBuffer[] bufs, int offset, int length,
NativeDispatcher nd)
throws IOException
{
IOVecWrapper vec = IOVecWrapper.get(length);
// Create shadow to ensure DirectByteBuffers are used
ByteBuffer[] shadow = new ByteBuffer[numBufs];
boolean completed = false;
int iov_len = 0;
try {
for (int i=0; i<numBufs; i++) {
if (!(bufs[i] instanceof DirectBuffer)) {
int pos = bufs[i].position();
int lim = bufs[i].limit();
assert (pos <= lim);
int rem = (pos <= lim ? lim - pos : 0);
ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
shadow[i] = bb;
// Leave slow buffer position untouched; it will be updated
// after we see how many bytes were really written out
bb.put(bufs[i]);
bufs[i].position(pos);
bb.flip();
} else {
shadow[i] = bufs[i];
}
}
IOVecWrapper vec = null;
long bytesWritten = 0;
try {
// Create a native iovec array
vec= new IOVecWrapper(numBufs);
// Fill in the iovec array with appropriate data
for (int i=0; i<numBufs; i++) {
ByteBuffer nextBuffer = shadow[i];
// put in the buffer addresses
long pos = nextBuffer.position();
long len = nextBuffer.limit() - pos;
vec.putBase(i, ((DirectBuffer)nextBuffer).address() + pos);
vec.putLen(i, len);
}
// Iterate over buffers to populate native iovec array.
int count = offset + length;
for (int i=offset; i<count; i++) {
ByteBuffer buf = bufs[i];
int pos = buf.position();
int lim = buf.limit();
assert (pos <= lim);
int rem = (pos <= lim ? lim - pos : 0);
if (rem > 0) {
vec.setBuffer(iov_len, buf, pos, rem);
// allocate shadow buffer to ensure I/O is done with direct buffer
if (!(buf instanceof DirectBuffer)) {
ByteBuffer shadow = Util.getTemporaryDirectBuffer(rem);
shadow.put(buf);
shadow.flip();
vec.setShadow(iov_len, shadow);
buf.position(pos); // temporarily restore position in user buffer
buf = shadow;
pos = shadow.position();
}
// Invoke native call to fill the buffers
bytesWritten = nd.writev(fd, vec.address, numBufs);
} finally {
vec.free();
vec.putBase(iov_len, ((DirectBuffer)buf).address() + pos);
vec.putLen(iov_len, rem);
iov_len++;
}
}
long returnVal = bytesWritten;
if (iov_len == 0)
return 0L;
long bytesWritten = nd.writev(fd, vec.address, iov_len);
// Notify the buffers how many bytes were taken
for (int i=0; i<numBufs; i++) {
ByteBuffer nextBuffer = bufs[i];
int pos = nextBuffer.position();
int lim = nextBuffer.limit();
assert (pos <= lim);
int len = (pos <= lim ? lim - pos : lim);
if (bytesWritten >= len) {
bytesWritten -= len;
int newPosition = pos + len;
nextBuffer.position(newPosition);
} else { // Buffers not completely filled
if (bytesWritten > 0) {
assert(pos + bytesWritten < (long)Integer.MAX_VALUE);
int newPosition = (int)(pos + bytesWritten);
nextBuffer.position(newPosition);
}
break;
long left = bytesWritten;
for (int j=0; j<iov_len; j++) {
if (left > 0) {
ByteBuffer buf = vec.getBuffer(j);
int pos = vec.getPosition(j);
int rem = vec.getRemaining(j);
int n = (left > rem) ? rem : (int)left;
buf.position(pos + n);
left -= n;
}
// return shadow buffers to buffer pool
ByteBuffer shadow = vec.getShadow(j);
if (shadow != null)
Util.offerLastTemporaryDirectBuffer(shadow);
vec.clearRefs(j);
}
return returnVal;
completed = true;
return bytesWritten;
} finally {
// return any substituted buffers to cache
for (int i=0; i<numBufs; i++) {
ByteBuffer bb = shadow[i];
if (bb != null && bb != bufs[i]) {
Util.releaseTemporaryDirectBuffer(bb);
// if an error occurred then clear refs to buffers and return any shadow
// buffers to cache
if (!completed) {
for (int j=0; j<iov_len; j++) {
ByteBuffer shadow = vec.getShadow(j);
if (shadow != null)
Util.offerLastTemporaryDirectBuffer(shadow);
vec.clearRefs(j);
}
}
}
......@@ -231,7 +196,7 @@ class IOUtil {
dst.put(bb);
return n;
} finally {
Util.releaseTemporaryDirectBuffer(bb);
Util.offerFirstTemporaryDirectBuffer(bb);
}
}
......@@ -262,92 +227,85 @@ class IOUtil {
static long read(FileDescriptor fd, ByteBuffer[] bufs, NativeDispatcher nd)
throws IOException
{
int nextWithRemaining = remaining(bufs);
// if all bufs are empty we should return immediately
if (nextWithRemaining < 0)
return 0;
// If some bufs are empty we should skip them
if (nextWithRemaining > 0)
bufs = skipBufs(bufs, nextWithRemaining);
return read(fd, bufs, 0, bufs.length, nd);
}
int numBufs = bufs.length;
static long read(FileDescriptor fd, ByteBuffer[] bufs, int offset, int length,
NativeDispatcher nd)
throws IOException
{
IOVecWrapper vec = IOVecWrapper.get(length);
// Read into the shadow to ensure DirectByteBuffers are used
ByteBuffer[] shadow = new ByteBuffer[numBufs];
boolean usingSlowBuffers = false;
boolean completed = false;
int iov_len = 0;
try {
for (int i=0; i<numBufs; i++) {
if (bufs[i].isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
if (!(bufs[i] instanceof DirectBuffer)) {
shadow[i] = Util.getTemporaryDirectBuffer(bufs[i].remaining());
usingSlowBuffers = true;
} else {
shadow[i] = bufs[i];
}
}
IOVecWrapper vec = null;
long bytesRead = 0;
try {
// Create a native iovec array
vec = new IOVecWrapper(numBufs);
// Fill in the iovec array with appropriate data
for (int i=0; i<numBufs; i++) {
ByteBuffer nextBuffer = shadow[i];
// put in the buffer addresses
long pos = nextBuffer.position();
long len = nextBuffer.remaining();
vec.putBase(i, ((DirectBuffer)nextBuffer).address() + pos);
vec.putLen(i, len);
}
// Iterate over buffers to populate native iovec array.
int count = offset + length;
for (int i=offset; i<count; i++) {
ByteBuffer buf = bufs[i];
if (buf.isReadOnly())
throw new IllegalArgumentException("Read-only buffer");
int pos = buf.position();
int lim = buf.limit();
assert (pos <= lim);
int rem = (pos <= lim ? lim - pos : 0);
// Invoke native call to fill the buffers
bytesRead = nd.readv(fd, vec.address, numBufs);
} finally {
vec.free();
}
long returnVal = bytesRead;
if (rem > 0) {
vec.setBuffer(iov_len, buf, pos, rem);
// Notify the buffers how many bytes were read
for (int i=0; i<numBufs; i++) {
ByteBuffer nextBuffer = shadow[i];
// Note: should this have been cached from above?
int pos = nextBuffer.position();
int len = nextBuffer.remaining();
if (bytesRead >= len) {
bytesRead -= len;
int newPosition = pos + len;
nextBuffer.position(newPosition);
} else { // Buffers not completely filled
if (bytesRead > 0) {
assert(pos + bytesRead < (long)Integer.MAX_VALUE);
int newPosition = (int)(pos + bytesRead);
nextBuffer.position(newPosition);
// allocate shadow buffer to ensure I/O is done with direct buffer
if (!(buf instanceof DirectBuffer)) {
ByteBuffer shadow = Util.getTemporaryDirectBuffer(rem);
vec.setShadow(iov_len, shadow);
buf = shadow;
pos = shadow.position();
}
break;
vec.putBase(iov_len, ((DirectBuffer)buf).address() + pos);
vec.putLen(iov_len, rem);
iov_len++;
}
}
if (iov_len == 0)
return 0L;
// Put results from shadow into the slow buffers
if (usingSlowBuffers) {
for (int i=0; i<numBufs; i++) {
if (!(bufs[i] instanceof DirectBuffer)) {
shadow[i].flip();
bufs[i].put(shadow[i]);
long bytesRead = nd.readv(fd, vec.address, iov_len);
// Notify the buffers how many bytes were read
long left = bytesRead;
for (int j=0; j<iov_len; j++) {
ByteBuffer shadow = vec.getShadow(j);
if (left > 0) {
ByteBuffer buf = vec.getBuffer(j);
int rem = vec.getRemaining(j);
int n = (left > rem) ? rem : (int)left;
if (shadow == null) {
int pos = vec.getPosition(j);
buf.position(pos + n);
} else {
shadow.limit(shadow.position() + n);
buf.put(shadow);
}
left -= n;
}
if (shadow != null)
Util.offerLastTemporaryDirectBuffer(shadow);
vec.clearRefs(j);
}
return returnVal;
completed = true;
return bytesRead;
} finally {
// return any substituted buffers to cache
if (usingSlowBuffers) {
for (int i=0; i<numBufs; i++) {
ByteBuffer bb = shadow[i];
if (bb != null && bb != bufs[i]) {
Util.releaseTemporaryDirectBuffer(bb);
}
// if an error occurred then clear refs to buffers and return any shadow
// buffers to cache
if (!completed) {
for (int j=0; j<iov_len; j++) {
ByteBuffer shadow = vec.getShadow(j);
if (shadow != null)
Util.offerLastTemporaryDirectBuffer(shadow);
vec.clearRefs(j);
}
}
}
......
src/share/classes/sun/nio/ch/IOVecWrapper.java
浏览文件 @ 4219640f
......@@ -25,6 +25,7 @@
package sun.nio.ch;
import java.nio.ByteBuffer;
import sun.misc.*;
......@@ -43,23 +44,98 @@ import sun.misc.*;
class IOVecWrapper {
// Miscellaneous constants
static int BASE_OFFSET = 0;
static int LEN_OFFSET;
static int SIZE_IOVEC;
private static final int BASE_OFFSET = 0;
private static final int LEN_OFFSET;
private static final int SIZE_IOVEC;
// The iovec array
private AllocatedNativeObject vecArray;
private final AllocatedNativeObject vecArray;
// Number of elements in iovec array
private final int size;
// Buffers and position/remaining corresponding to elements in iovec array
private final ByteBuffer[] buf;
private final int[] position;
private final int[] remaining;
// Shadow buffers for cases when original buffer is substituted
private final ByteBuffer[] shadow;
// Base address of this array
long address;
final long address;
// Address size in bytes
static int addressSize;
IOVecWrapper(int newSize) {
newSize = (newSize + 1) * SIZE_IOVEC;
vecArray = new AllocatedNativeObject(newSize, false);
address = vecArray.address();
private static class Deallocator implements Runnable {
private final AllocatedNativeObject obj;
Deallocator(AllocatedNativeObject obj) {
this.obj = obj;
}
public void run() {
obj.free();
}
}
// per thread IOVecWrapper
private static final ThreadLocal<IOVecWrapper> cached =
new ThreadLocal<IOVecWrapper>();
private IOVecWrapper(int size) {
this.size = size;
this.buf = new ByteBuffer[size];
this.position = new int[size];
this.remaining = new int[size];
this.shadow = new ByteBuffer[size];
this.vecArray = new AllocatedNativeObject(size * SIZE_IOVEC, false);
this.address = vecArray.address();
}
static IOVecWrapper get(int size) {
IOVecWrapper wrapper = cached.get();
if (wrapper != null && wrapper.size < size) {
// not big enough; eagerly release memory
wrapper.vecArray.free();
wrapper = null;
}
if (wrapper == null) {
wrapper = new IOVecWrapper(size);
Cleaner.create(wrapper, new Deallocator(wrapper.vecArray));
cached.set(wrapper);
}
return wrapper;
}
void setBuffer(int i, ByteBuffer buf, int pos, int rem) {
this.buf[i] = buf;
this.position[i] = pos;
this.remaining[i] = rem;
}
void setShadow(int i, ByteBuffer buf) {
shadow[i] = buf;
}
ByteBuffer getBuffer(int i) {
return buf[i];
}
int getPosition(int i) {
return position[i];
}
int getRemaining(int i) {
return remaining[i];
}
ByteBuffer getShadow(int i) {
return shadow[i];
}
void clearRefs(int i) {
buf[i] = null;
shadow[i] = null;
}
void putBase(int i, long base) {
......@@ -78,10 +154,6 @@ class IOVecWrapper {
vecArray.putLong(offset, len);
}
void free() {
vecArray.free();
}
static {
addressSize = Util.unsafe().addressSize();
LEN_OFFSET = addressSize;
......
src/share/classes/sun/nio/ch/SocketChannelImpl.java
浏览文件 @ 4219640f
......@@ -385,9 +385,11 @@ class SocketChannelImpl
}
}
private long read0(ByteBuffer[] bufs) throws IOException {
if (bufs == null)
throw new NullPointerException();
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
throw new IndexOutOfBoundsException();
synchronized (readLock) {
if (!ensureReadOpen())
return -1;
......@@ -401,7 +403,7 @@ class SocketChannelImpl
}
for (;;) {
n = IOUtil.read(fd, bufs, nd);
n = IOUtil.read(fd, dsts, offset, length, nd);
if ((n == IOStatus.INTERRUPTED) && isOpen())
continue;
return IOStatus.normalize(n);
......@@ -418,15 +420,6 @@ class SocketChannelImpl
}
}
public long read(ByteBuffer[] dsts, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
throw new IndexOutOfBoundsException();
// ## Fix IOUtil.write so that we can avoid this array copy
return read0(Util.subsequence(dsts, offset, length));
}
public int write(ByteBuffer buf) throws IOException {
if (buf == null)
throw new NullPointerException();
......@@ -458,9 +451,11 @@ class SocketChannelImpl
}
}
public long write0(ByteBuffer[] bufs) throws IOException {
if (bufs == null)
throw new NullPointerException();
public long write(ByteBuffer[] srcs, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
throw new IndexOutOfBoundsException();
synchronized (writeLock) {
ensureWriteOpen();
long n = 0;
......@@ -472,7 +467,7 @@ class SocketChannelImpl
writerThread = NativeThread.current();
}
for (;;) {
n = IOUtil.write(fd, bufs, nd);
n = IOUtil.write(fd, srcs, offset, length, nd);
if ((n == IOStatus.INTERRUPTED) && isOpen())
continue;
return IOStatus.normalize(n);
......@@ -489,15 +484,6 @@ class SocketChannelImpl
}
}
public long write(ByteBuffer[] srcs, int offset, int length)
throws IOException
{
if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
throw new IndexOutOfBoundsException();
// ## Fix IOUtil.write so that we can avoid this array copy
return write0(Util.subsequence(srcs, offset, length));
}
// package-private
int sendOutOfBandData(byte b) throws IOException {
synchronized (writeLock) {
......
src/share/classes/sun/nio/ch/Util.java
浏览文件 @ 4219640f
......@@ -41,67 +41,180 @@ import sun.security.action.GetPropertyAction;
class Util {
// -- Caches --
// The number of temp buffers in our pool
private static final int TEMP_BUF_POOL_SIZE = 3;
private static final int TEMP_BUF_POOL_SIZE = 8;
// Per-thread soft cache of the last temporary direct buffer
private static ThreadLocal<SoftReference<ByteBuffer>>[] bufferPool;
// Per-thread cache of temporary direct buffers
private static ThreadLocal<BufferCache> bufferCache =
new ThreadLocal<BufferCache>()
{
@Override
protected BufferCache initialValue() {
return new BufferCache();
}
};
@SuppressWarnings("unchecked")
static ThreadLocal<SoftReference<ByteBuffer>>[] createThreadLocalBufferPool() {
return new ThreadLocal[TEMP_BUF_POOL_SIZE];
}
/**
* A simple cache of direct buffers.
*/
private static class BufferCache {
// the array of buffers
private ByteBuffer[] buffers;
// the number of buffers in the cache
private int count;
// the index of the first valid buffer (undefined if count == 0)
private int start;
private int next(int i) {
return (i + 1) % TEMP_BUF_POOL_SIZE;
}
BufferCache() {
buffers = new ByteBuffer[TEMP_BUF_POOL_SIZE];
}
static {
bufferPool = createThreadLocalBufferPool();
for (int i=0; i<TEMP_BUF_POOL_SIZE; i++)
bufferPool[i] = new ThreadLocal<SoftReference<ByteBuffer>>();
/**
* Removes and returns a buffer from the cache of at least the given
* size (or null if no suitable buffer is found).
*/
ByteBuffer get(int size) {
if (count == 0)
return null; // cache is empty
ByteBuffer[] buffers = this.buffers;
// search for suitable buffer (often the first buffer will do)
ByteBuffer buf = buffers[start];
if (buf.capacity() < size) {
buf = null;
int i = start;
while ((i = next(i)) != start) {
ByteBuffer bb = buffers[i];
if (bb == null)
break;
if (bb.capacity() >= size) {
buf = bb;
break;
}
}
if (buf == null)
return null;
// move first element to here to avoid re-packing
buffers[i] = buffers[start];
}
// remove first element
buffers[start] = null;
start = next(start);
count--;
// prepare the buffer and return it
buf.rewind();
buf.limit(size);
return buf;
}
boolean offerFirst(ByteBuffer buf) {
if (count >= TEMP_BUF_POOL_SIZE) {
return false;
} else {
start = (start + TEMP_BUF_POOL_SIZE - 1) % TEMP_BUF_POOL_SIZE;
buffers[start] = buf;
count++;
return true;
}
}
boolean offerLast(ByteBuffer buf) {
if (count >= TEMP_BUF_POOL_SIZE) {
return false;
} else {
int next = (start + count) % TEMP_BUF_POOL_SIZE;
buffers[next] = buf;
count++;
return true;
}
}
boolean isEmpty() {
return count == 0;
}
ByteBuffer removeFirst() {
assert count > 0;
ByteBuffer buf = buffers[start];
buffers[start] = null;
start = next(start);
count--;
return buf;
}
}
/**
* Returns a temporary buffer of at least the given size
*/
static ByteBuffer getTemporaryDirectBuffer(int size) {
ByteBuffer buf = null;
// Grab a buffer if available
for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {
SoftReference<ByteBuffer> ref = bufferPool[i].get();
if ((ref != null) && ((buf = ref.get()) != null) &&
(buf.capacity() >= size)) {
buf.rewind();
buf.limit(size);
bufferPool[i].set(null);
return buf;
BufferCache cache = bufferCache.get();
ByteBuffer buf = cache.get(size);
if (buf != null) {
return buf;
} else {
// No suitable buffer in the cache so we need to allocate a new
// one. To avoid the cache growing then we remove the first
// buffer from the cache and free it.
if (!cache.isEmpty()) {
buf = cache.removeFirst();
free(buf);
}
return ByteBuffer.allocateDirect(size);
}
// Make a new one
return ByteBuffer.allocateDirect(size);
}
/**
* Releases a temporary buffer by returning to the cache or freeing it.
*/
static void releaseTemporaryDirectBuffer(ByteBuffer buf) {
if (buf == null)
return;
// Put it in an empty slot if such exists
for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {
SoftReference<ByteBuffer> ref = bufferPool[i].get();
if ((ref == null) || (ref.get() == null)) {
bufferPool[i].set(new SoftReference<ByteBuffer>(buf));
return;
}
offerFirstTemporaryDirectBuffer(buf);
}
/**
* Releases a temporary buffer by returning to the cache or freeing it. If
* returning to the cache then insert it at the start so that it is
* likely to be returned by a subsequent call to getTemporaryDirectBuffer.
*/
static void offerFirstTemporaryDirectBuffer(ByteBuffer buf) {
assert buf != null;
BufferCache cache = bufferCache.get();
if (!cache.offerFirst(buf)) {
// cache is full
free(buf);
}
// Otherwise replace a smaller one in the cache if such exists
for (int i=0; i<TEMP_BUF_POOL_SIZE; i++) {
SoftReference<ByteBuffer> ref = bufferPool[i].get();
ByteBuffer inCacheBuf = ref.get();
if ((inCacheBuf == null) || (buf.capacity() > inCacheBuf.capacity())) {
bufferPool[i].set(new SoftReference<ByteBuffer>(buf));
return;
}
}
/**
* Releases a temporary buffer by returning to the cache or freeing it. If
* returning to the cache then insert it at the end. This makes it
* suitable for scatter/gather operations where the buffers are returned to
* cache in same order that they were obtained.
*/
static void offerLastTemporaryDirectBuffer(ByteBuffer buf) {
assert buf != null;
BufferCache cache = bufferCache.get();
if (!cache.offerLast(buf)) {
// cache is full
free(buf);
}
}
// release memory
((DirectBuffer)buf).cleaner().clean();
/**
* Frees the memory for the given direct buffer
*/
private static void free(ByteBuffer buf) {
((DirectBuffer)buf).cleaner().clean();
}
private static class SelectorWrapper {
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或