From 4219640fbb584a9347a96cdb092066ade11c990e Mon Sep 17 00:00:00 2001 From: alanb Date: Thu, 12 Aug 2010 19:53:25 +0100 Subject: [PATCH] 6971825: (so) improve scatter/gather implementation Reviewed-by: chegar, sherman --- .../sun/nio/ch/DatagramChannelImpl.java | 38 +-- .../classes/sun/nio/ch/FileChannelImpl.java | 35 +- src/share/classes/sun/nio/ch/IOUtil.java | 304 ++++++++---------- .../classes/sun/nio/ch/IOVecWrapper.java | 98 +++++- .../classes/sun/nio/ch/SocketChannelImpl.java | 38 +-- src/share/classes/sun/nio/ch/Util.java | 201 +++++++++--- 6 files changed, 409 insertions(+), 305 deletions(-) diff --git a/src/share/classes/sun/nio/ch/DatagramChannelImpl.java b/src/share/classes/sun/nio/ch/DatagramChannelImpl.java index 41b37c9e9..1d7d6758d 100644 --- a/src/share/classes/sun/nio/ch/DatagramChannelImpl.java +++ b/src/share/classes/sun/nio/ch/DatagramChannelImpl.java @@ -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); } diff --git a/src/share/classes/sun/nio/ch/FileChannelImpl.java b/src/share/classes/sun/nio/ch/FileChannelImpl.java index 1bbd5e025..af732f5b3 100644 --- a/src/share/classes/sun/nio/ch/FileChannelImpl.java +++ b/src/share/classes/sun/nio/ch/FileChannelImpl.java @@ -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 { diff --git a/src/share/classes/sun/nio/ch/IOUtil.java b/src/share/classes/sun/nio/ch/IOUtil.java index 5b1dd95cd..86acff635 100644 --- a/src/share/classes/sun/nio/ch/IOUtil.java +++ b/src/share/classes/sun/nio/ch/IOUtil.java @@ -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 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 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= 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 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 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 0) { + vec.setBuffer(iov_len, buf, pos, rem); - // Notify the buffers how many bytes were read - for (int i=0; i= 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 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 cached = + new ThreadLocal(); + + 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; diff --git a/src/share/classes/sun/nio/ch/SocketChannelImpl.java b/src/share/classes/sun/nio/ch/SocketChannelImpl.java index ee84a0e62..7d42d7d00 100644 --- a/src/share/classes/sun/nio/ch/SocketChannelImpl.java +++ b/src/share/classes/sun/nio/ch/SocketChannelImpl.java @@ -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) { diff --git a/src/share/classes/sun/nio/ch/Util.java b/src/share/classes/sun/nio/ch/Util.java index b17801cbe..50d280865 100644 --- a/src/share/classes/sun/nio/ch/Util.java +++ b/src/share/classes/sun/nio/ch/Util.java @@ -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>[] bufferPool; + // Per-thread cache of temporary direct buffers + private static ThreadLocal bufferCache = + new ThreadLocal() + { + @Override + protected BufferCache initialValue() { + return new BufferCache(); + } + }; - @SuppressWarnings("unchecked") - static ThreadLocal>[] 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>(); + /** + * 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 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 ref = bufferPool[i].get(); - if ((ref == null) || (ref.get() == null)) { - bufferPool[i].set(new SoftReference(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 ref = bufferPool[i].get(); - ByteBuffer inCacheBuf = ref.get(); - if ((inCacheBuf == null) || (buf.capacity() > inCacheBuf.capacity())) { - bufferPool[i].set(new SoftReference(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 { -- GitLab