... more on this.<div><br></div><div>1.  ignore the FIXME in ensureWritableBytes ... or just delete that line if you compile it.</div><div><br></div><div>2.  this is about 2x slower than a direct heap buffer.  The overhead comes from the composite buffer. <br>

<br><div class="gmail_quote">On Wed, Oct 19, 2011 at 11:31 AM, Kevin Burton <span dir="ltr">&lt;<a href="mailto:burtonator@gmail.com">burtonator@gmail.com</a>&gt;</span> wrote:<br><blockquote class="gmail_quote" style="margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex;">

<div><br></div><div>Hey gang. </div><div><br></div><div>I&#39;ve been working on a map reduce implementation (which I&#39;m about to OSS) and I&#39;ve started to use ChannelBuffers extensively in the framework.</div><div>

<br>
</div><div>Anyway... some of the buffers I&#39;m dealing with are hundreds of MBs... I need extent based extension on some of them because going from 128MB to 256MB very quick isn&#39;t fun when I&#39;m trying to conserve memory.</div>


<div><br></div><div>This makes a SMALL change to DynamicChannelBuffer.</div><div><br></div><div>Basically,  I give it an initial capacity of say 128MB and then give it extents of 2MB .  When it&#39;s out of memory it will increase capacity by 2MB.  ... so worse case scenario I waste 2MB of memory.</div>


<div><br></div><div>I could re-implement this VERY easily with say 30 lines of code if we could change &#39;buffer&#39; in DynamicChannelBuffer to &#39;protected&#39; ....</div><div><br></div><div>If you would accept this change I&#39;ll just submit a new patch changing this to protected and then the new ExtendedChannelBuffer ...</div>


<div><br></div><div>Thoughts?</div><div><br></div><div><br></div><div><br></div><div>package peregrine.util;</div><div><br></div><div>import java.io.*;</div><div>import java.util.*;</div><div><br></div><div>import org.jboss.netty.buffer.*;</div>


<div><br></div><div>import java.io.IOException;</div><div>import java.io.InputStream;</div><div>import java.io.OutputStream;</div><div>import java.util.ArrayList;</div><div>import java.util.List;</div><div>import java.nio.ByteBuffer;</div>


<div>import java.nio.ByteOrder;</div><div>import java.nio.channels.GatheringByteChannel;</div><div>import java.nio.channels.ScatteringByteChannel;</div><div><br></div><div>/**</div><div> * A dynamic capacity buffer which increases its capacity as needed.  It is</div>


<div> * recommended to use {@link ChannelBuffers#dynamicBuffer(int)} instead of</div><div> * calling the constructor explicitly.</div><div> *</div><div> * @author &lt;a href=&quot;<a href="http://www.jboss.org/netty/" target="_blank">http://www.jboss.org/netty/</a>&quot;&gt;The Netty Project&lt;/a&gt;</div>


<div> * @author &lt;a href=&quot;<a href="http://gleamynode.net/" target="_blank">http://gleamynode.net/</a>&quot;&gt;Trustin Lee&lt;/a&gt;</div><div> *</div><div> * @version $Rev: 2206M $, $Date: 2010-11-09 15:04:10 +0900 (Tue, 09 Nov 2010) $</div>


<div> *</div><div> */</div><div>public class ExtendedChannelBuffer extends AbstractChannelBuffer {</div><div><br></div><div>    private final ChannelBufferFactory factory;</div><div>    private final ByteOrder endianness;</div>


<div>    private ChannelBuffer buffer;</div><div><br></div><div>    private List&lt;ChannelBuffer&gt; extents = new ArrayList();</div><div><br></div><div>    private int extentLength;</div><div><br></div><div>    private int initialCapacity;</div>


<div>    </div><div>    public ExtendedChannelBuffer(int initialCapacity, int extentLength) {</div><div>        this(ByteOrder.BIG_ENDIAN, initialCapacity, extentLength);</div><div>    }</div><div><br></div><div>    public ExtendedChannelBuffer(ByteOrder endianness, int initialCapacity, int extentLength) {</div>


<div>        this(endianness, initialCapacity, extentLength, HeapChannelBufferFactory.getInstance(endianness));</div><div>    }</div><div><br></div><div>    public ExtendedChannelBuffer(ByteOrder endianness, int initialCapacity, int extentLength, ChannelBufferFactory factory) {</div>


<div>        if (extentLength &lt; 0) {</div><div>            throw new IllegalArgumentException(&quot;extentLength: &quot; + extentLength);</div><div>        }</div><div>        if (endianness == null) {</div><div>            throw new NullPointerException(&quot;endianness&quot;);</div>


<div>        }</div><div>        if (factory == null) {</div><div>            throw new NullPointerException(&quot;factory&quot;);</div><div>        }</div><div>        this.factory = factory;</div><div>        this.endianness = endianness;</div>


<div>        this.extentLength = extentLength;</div><div><br></div><div>        extend( initialCapacity );</div><div>        </div><div>    }</div><div><br></div><div>    @Override</div><div>    public void ensureWritableBytes(int minWritableBytes) {</div>


<div><br></div><div>        if (minWritableBytes &lt;= writableBytes()) {</div><div>            return;</div><div>        }</div><div><br></div><div>        extend( extentLength );</div><div><br></div><div>    }</div><div>


<br></div><div>    private void extend( int newExtentLength ) {</div><div><br></div><div>        ChannelBuffer newExtent = factory().getBuffer(order(), newExtentLength);</div><div>        extents.add( newExtent );</div><div>


<br></div><div>        System.out.printf( &quot;FIXME: allocating extent: %s\n&quot;, newExtentLength );</div><div>        </div><div>        buffer = new CompositeChannelBuffer( endianness, extents );</div><div>        </div>


<div>    }</div><div>    </div><div>    public ChannelBufferFactory factory() {</div><div>        return factory;</div><div>    }</div><div><br></div><div>    public ByteOrder order() {</div><div>        return endianness;</div>


<div>    }</div><div><br></div><div>    public boolean isDirect() {</div><div>        return buffer.isDirect();</div><div>    }</div><div><br></div><div>    public int capacity() {</div><div>        return buffer.capacity();</div>


<div>    }</div><div><br></div><div>    public boolean hasArray() {</div><div>        return buffer.hasArray();</div><div>    }</div><div><br></div><div>    public byte[] array() {</div><div>        return buffer.array();</div>


<div>    }</div><div><br></div><div>    public int arrayOffset() {</div><div>        return buffer.arrayOffset();</div><div>    }</div><div><br></div><div>    public byte getByte(int index) {</div><div>        return buffer.getByte(index);</div>


<div>    }</div><div><br></div><div>    public short getShort(int index) {</div><div>        return buffer.getShort(index);</div><div>    }</div><div><br></div><div>    public int getUnsignedMedium(int index) {</div><div>


        return buffer.getUnsignedMedium(index);</div><div>    }</div><div><br></div><div>    public int getInt(int index) {</div><div>        return buffer.getInt(index);</div><div>    }</div><div><br></div><div>    public long getLong(int index) {</div>


<div>        return buffer.getLong(index);</div><div>    }</div><div><br></div><div>    public void getBytes(int index, byte[] dst, int dstIndex, int length) {</div><div>        buffer.getBytes(index, dst, dstIndex, length);</div>


<div>    }</div><div><br></div><div>    public void getBytes(int index, ChannelBuffer dst, int dstIndex, int length) {</div><div>        buffer.getBytes(index, dst, dstIndex, length);</div><div>    }</div><div><br></div>

<div>
    public void getBytes(int index, ByteBuffer dst) {</div><div>        buffer.getBytes(index, dst);</div><div>    }</div><div><br></div><div>    public int getBytes(int index, GatheringByteChannel out, int length)</div>

<div>
            throws IOException {</div><div>        return buffer.getBytes(index, out, length);</div><div>    }</div><div><br></div><div>    public void getBytes(int index, OutputStream out, int length)</div><div>            throws IOException {</div>


<div>        buffer.getBytes(index, out, length);</div><div>    }</div><div><br></div><div>    public void setByte(int index, int value) {</div><div>        buffer.setByte(index, value);</div><div>    }</div><div><br></div>


<div>    public void setShort(int index, int value) {</div><div>        buffer.setShort(index, value);</div><div>    }</div><div><br></div><div>    public void setMedium(int index, int value) {</div><div>        buffer.setMedium(index, value);</div>


<div>    }</div><div><br></div><div>    public void setInt(int index, int value) {</div><div>        buffer.setInt(index, value);</div><div>    }</div><div><br></div><div>    public void setLong(int index, long value) {</div>


<div>        buffer.setLong(index, value);</div><div>    }</div><div><br></div><div>    public void setBytes(int index, byte[] src, int srcIndex, int length) {</div><div>        buffer.setBytes(index, src, srcIndex, length);</div>


<div>    }</div><div><br></div><div>    public void setBytes(int index, ChannelBuffer src, int srcIndex, int length) {</div><div>        buffer.setBytes(index, src, srcIndex, length);</div><div>    }</div><div><br></div>

<div>
    public void setBytes(int index, ByteBuffer src) {</div><div>        buffer.setBytes(index, src);</div><div>    }</div><div><br></div><div>    public int setBytes(int index, InputStream in, int length)</div><div>            throws IOException {</div>


<div>        return buffer.setBytes(index, in, length);</div><div>    }</div><div><br></div><div>    public int setBytes(int index, ScatteringByteChannel in, int length)</div><div>            throws IOException {</div><div>


        return buffer.setBytes(index, in, length);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public void writeByte(int value) {</div><div>        ensureWritableBytes(1);</div><div>        super.writeByte(value);</div>


<div>    }</div><div><br></div><div>    @Override</div><div>    public void writeShort(int value) {</div><div>        ensureWritableBytes(2);</div><div>        super.writeShort(value);</div><div>    }</div><div><br></div>


<div>    @Override</div><div>    public void writeMedium(int value) {</div><div>        ensureWritableBytes(3);</div><div>        super.writeMedium(value);</div><div>    }</div><div><br></div><div>    @Override</div><div>


    public void writeInt(int value) {</div><div>        ensureWritableBytes(4);</div><div>        super.writeInt(value);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public void writeLong(long value) {</div>


<div>        ensureWritableBytes(8);</div><div>        super.writeLong(value);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public void writeBytes(byte[] src, int srcIndex, int length) {</div><div>


        ensureWritableBytes(length);</div><div>        super.writeBytes(src, srcIndex, length);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public void writeBytes(ChannelBuffer src, int srcIndex, int length) {</div>


<div>        ensureWritableBytes(length);</div><div>        super.writeBytes(src, srcIndex, length);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public void writeBytes(ByteBuffer src) {</div><div>


        ensureWritableBytes(src.remaining());</div><div>        super.writeBytes(src);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public int writeBytes(InputStream in, int length) throws IOException {</div>


<div>        ensureWritableBytes(length);</div><div>        return super.writeBytes(in, length);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public int writeBytes(ScatteringByteChannel in, int length)</div>


<div>            throws IOException {</div><div>        ensureWritableBytes(length);</div><div>        return super.writeBytes(in, length);</div><div>    }</div><div><br></div><div>    @Override</div><div>    public void writeZero(int length) {</div>


<div>        ensureWritableBytes(length);</div><div>        super.writeZero(length);</div><div>    }</div><div><br></div><div>    public ChannelBuffer duplicate() {</div><div>        return new DuplicatedChannelBuffer(this);</div>


<div>    }</div><div><br></div><div>    public ChannelBuffer copy(int index, int length) {</div><div>        ExtendedChannelBuffer copiedBuffer = new ExtendedChannelBuffer(order(), Math.max(length, 64), extentLength, factory());</div>


<div>        copiedBuffer.buffer = buffer.copy(index, length);</div><div>        copiedBuffer.setIndex(0, length);</div><div>        return copiedBuffer;</div><div>    }</div><div><br></div><div>    public ChannelBuffer slice(int index, int length) {</div>


<div>        if (index == 0) {</div><div>            if (length == 0) {</div><div>                return ChannelBuffers.EMPTY_BUFFER;</div><div>            }</div><div>            return new TruncatedChannelBuffer(this, length);</div>


<div>        } else {</div><div>            if (length == 0) {</div><div>                return ChannelBuffers.EMPTY_BUFFER;</div><div>            }</div><div>            return new SlicedChannelBuffer(this, index, length);</div>


<div>        }</div><div>    }</div><div><br></div><div>    public ByteBuffer toByteBuffer(int index, int length) {</div><div>        return buffer.toByteBuffer(index, length);</div><div>    }</div><div>}</div><div><br></div>


</blockquote></div><br></div>