Author: manik.surtani(a)jboss.com Date: 2008-10-14 05:23:28 -0400 (Tue, 14 Oct 2008) New Revision: 6927 Added: core/branches/2.2.X/src/main/java/org/jboss/cache/io/ core/branches/2.2.X/src/main/java/org/jboss/cache/io/ByteBuffer.java core/branches/2.2.X/src/main/java/org/jboss/cache/io/ExposedByteArrayOutputStream.java Removed: core/branches/2.2.X/src/main/java/org/jboss/cache/util/ExposedByteArrayOutputStream.java Modified: core/branches/2.2.X/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/Marshaller.java core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/VersionAwareMarshaller.java core/branches/2.2.X/src/main/java/org/jboss/cache/remoting/jgroups/ChannelMessageListener.java core/branches/2.2.X/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java Log: Ported JBCACHE-1382: Investigate use of ExposedBAOS instead of BAOS when marshalling objects Modified: core/branches/2.2.X/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/buddyreplication/BuddyManager.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -31,13 +31,13 @@ import org.jboss.cache.factories.annotations.Inject; import org.jboss.cache.factories.annotations.Start; import org.jboss.cache.factories.annotations.Stop; +import org.jboss.cache.io.ExposedByteArrayOutputStream; import org.jboss.cache.lock.TimeoutException; import org.jboss.cache.notifications.Notifier; import org.jboss.cache.notifications.annotation.CacheListener; import org.jboss.cache.notifications.annotation.ViewChanged; import org.jboss.cache.notifications.event.ViewChangedEvent; import org.jboss.cache.statetransfer.StateTransferManager; -import org.jboss.cache.util.ExposedByteArrayOutputStream; import org.jboss.cache.util.concurrent.ConcurrentHashSet; import org.jboss.cache.util.reflect.ReflectionUtil; import org.jboss.util.stream.MarshalledValueInputStream; @@ -48,7 +48,18 @@ import org.jgroups.util.Util; import java.io.ByteArrayInputStream; -import java.util.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.SortedSet; +import java.util.TreeSet; +import java.util.Vector; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; Added: core/branches/2.2.X/src/main/java/org/jboss/cache/io/ByteBuffer.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/io/ByteBuffer.java (rev 0) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/io/ByteBuffer.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -0,0 +1,46 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.cache.io; + +import org.jgroups.util.Buffer; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; + +/** + * A subclass of a JGroups Buffer + */ +public class ByteBuffer extends Buffer +{ + public ByteBuffer(byte[] bytes, int offset, int length) + { + super(bytes, offset, length); + } + + /** + * @return an input stream for the bytes in the buffer + */ + public InputStream getStream() + { + return new ByteArrayInputStream(getBuf(), getOffset(), getLength()); + } +} Copied: core/branches/2.2.X/src/main/java/org/jboss/cache/io/ExposedByteArrayOutputStream.java (from rev 6926, core/branches/2.2.X/src/main/java/org/jboss/cache/util/ExposedByteArrayOutputStream.java) =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/io/ExposedByteArrayOutputStream.java (rev 0) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/io/ExposedByteArrayOutputStream.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -0,0 +1,159 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jboss.cache.io; + +import net.jcip.annotations.NotThreadSafe; + +import java.io.ByteArrayOutputStream; + +/** + * Extends ByteArrayOutputStream, but exposes the internal buffer. + * Using this, callers don't need to call toByteArray() which copies the + * internal buffer. + * <p> + * Also overrides the superclass' behavior of always doubling the size of the + * internal buffer any time more capacity is needed. This class doubles the + * size until the internal buffer reaches a configurable max size (default is + * 4MB), after which it begins growing the buffer in 25% increments. This is + * intended to help prevent an OutOfMemoryError during a resize of a large + * buffer. + * </p> + * <p> + * A version of this class was originally created by Bela Ban as part of the + * JGroups library. + * </p> + * This class is not threadsafe as it will not support concurrent readers and writers. + * <p/> + * + * @author <a href="mailto://brian.stansberry@jboss.com">Brian Stansberry</a> + * @version $Id$ + */ +@NotThreadSafe +public class ExposedByteArrayOutputStream extends ByteArrayOutputStream +{ + /** + * Default buffer size after which if more buffer capacity + * is needed the buffer will grow by 25% rather than 100% + */ + public static final int DEFAULT_DOUBLING_SIZE = 4 * 1024 * 1024; // 4MB + + private int maxDoublingSize = DEFAULT_DOUBLING_SIZE; + + public ExposedByteArrayOutputStream() + { + super(); + } + + public ExposedByteArrayOutputStream(int size) + { + super(size); + } + + /** + * Creates a new byte array output stream, with a buffer capacity of + * the specified size, in bytes. + * + * @param size the initial size. + * @param maxDoublingSize the buffer size, after which if more capacity + * is needed the buffer will grow by 25% + * rather than 100% + * @throws IllegalArgumentException if size is negative. + */ + public ExposedByteArrayOutputStream(int size, int maxDoublingSize) + { + super(size); + this.maxDoublingSize = maxDoublingSize; + } + + /** + * Gets the internal buffer array. Note that the length of this array + * will almost certainly be longer than the data written to it; call + * <code>size()</code> to get the number of bytes of actual data. + */ + public final byte[] getRawBuffer() + { + return buf; + } + + @Override + public void write(byte[] b, int off, int len) + { + if ((off < 0) || (off > b.length) || (len < 0) || + ((off + len) > b.length) || ((off + len) < 0)) + { + throw new IndexOutOfBoundsException(); + } + else if (len == 0) + { + return; + } + + int newcount = count + len; + if (newcount > buf.length) + { + byte newbuf[] = new byte[getNewBufferSize(buf.length, newcount)]; + System.arraycopy(buf, 0, newbuf, 0, count); + buf = newbuf; + } + + System.arraycopy(b, off, buf, count, len); + count = newcount; + } + + @Override + public void write(int b) + { + int newcount = count + 1; + if (newcount > buf.length) + { + byte newbuf[] = new byte[getNewBufferSize(buf.length, newcount)]; + System.arraycopy(buf, 0, newbuf, 0, count); + buf = newbuf; + } + buf[count] = (byte) b; + count = newcount; + } + + /** + * Gets the highest internal buffer size after which if more capacity + * is needed the buffer will grow in 25% increments rather than 100%. + */ + public final int getMaxDoublingSize() + { + return maxDoublingSize; + } + + /** + * Gets the number of bytes to which the internal buffer should be resized. + * + * @param curSize the current number of bytes + * @param minNewSize the minimum number of bytes required + * @return the size to which the internal buffer should be resized + */ + public final int getNewBufferSize(int curSize, int minNewSize) + { + if (curSize <= maxDoublingSize) + return Math.max(curSize << 1, minNewSize); + else + return Math.max(curSize + (curSize >> 2), minNewSize); + } +} Property changes on: core/branches/2.2.X/src/main/java/org/jboss/cache/io/ExposedByteArrayOutputStream.java ___________________________________________________________________ Name: svn:eol-style + native Modified: core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/AbstractMarshaller.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -35,7 +35,9 @@ import org.jboss.cache.config.Configuration; import org.jboss.cache.factories.annotations.Inject; import org.jboss.cache.factories.annotations.Start; +import org.jboss.cache.io.ByteBuffer; import org.jboss.cache.transaction.GlobalTransaction; +import org.jgroups.util.Buffer; import java.io.InputStream; import java.io.ObjectInputStream; @@ -87,31 +89,44 @@ } // implement the basic contract set in RPCDispatcher.AbstractMarshaller - public byte[] objectToByteBuffer(Object obj) throws Exception + public Object objectFromStream(InputStream in) throws Exception { throw new RuntimeException("Needs to be overridden!"); } - public Object objectFromByteBuffer(byte[] bytes) throws Exception + public RegionalizedMethodCall regionalizedMethodCallFromByteBuffer(byte[] buffer) throws Exception { throw new RuntimeException("Needs to be overridden!"); } - public Object objectFromStream(InputStream in) throws Exception + public RegionalizedMethodCall regionalizedMethodCallFromObjectStream(ObjectInputStream in) throws Exception { throw new RuntimeException("Needs to be overridden!"); } - public RegionalizedMethodCall regionalizedMethodCallFromByteBuffer(byte[] buffer) throws Exception + public ByteBuffer objectToBuffer(Object o) throws Exception { throw new RuntimeException("Needs to be overridden!"); } - public RegionalizedMethodCall regionalizedMethodCallFromObjectStream(ObjectInputStream in) throws Exception + public Object objectFromByteBuffer(byte[] bytes) throws Exception { + return objectFromByteBuffer(bytes, 0, bytes.length); + } + + public Object objectFromByteBuffer(byte[] bytes, int offset, int len) throws Exception + { throw new RuntimeException("Needs to be overridden!"); } + public byte[] objectToByteBuffer(Object obj) throws Exception + { + Buffer b = objectToBuffer(obj); + byte[] bytes = new byte[b.getLength()]; + System.arraycopy(b.getBuf(), b.getOffset(), bytes, 0, b.getLength()); + return bytes; + } + protected Fqn extractFqn(ReplicableCommand cmd) { if (cmd == null) throw new NullPointerException("Command is null"); Modified: core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/Marshaller.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/Marshaller.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/Marshaller.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -31,7 +31,7 @@ * @author <a href="mailto://manik@jboss.org">Manik Surtani</a> * @since 2.0.0 */ -public interface Marshaller extends RpcDispatcher.Marshaller +public interface Marshaller extends RpcDispatcher.Marshaller2 { /** * Marshalls an object to a given {@link ObjectOutputStream} Modified: core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/VersionAwareMarshaller.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/VersionAwareMarshaller.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/marshall/VersionAwareMarshaller.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -12,11 +12,12 @@ import org.jboss.cache.factories.ComponentRegistry; import org.jboss.cache.factories.annotations.Inject; import org.jboss.cache.factories.annotations.Start; +import org.jboss.cache.io.ByteBuffer; +import org.jboss.cache.io.ExposedByteArrayOutputStream; import org.jboss.cache.util.Util; import org.jboss.util.stream.MarshalledValueInputStream; import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; import java.io.InputStream; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; @@ -150,9 +151,9 @@ } @Override - public byte[] objectToByteBuffer(Object obj) throws Exception + public ByteBuffer objectToBuffer(Object obj) throws Exception { - ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream(128); ObjectOutputStream out = new ObjectOutputStream(baos); out.writeShort(versionInt); @@ -163,15 +164,15 @@ out.close(); // and return bytes. - return baos.toByteArray(); + return new ByteBuffer(baos.getRawBuffer(), 0, baos.size()); } @Override - public Object objectFromByteBuffer(byte[] buf) throws Exception + public Object objectFromByteBuffer(byte[] bytes, int offset, int len) throws Exception { Marshaller marshaller; int versionId; - ObjectInputStream in = new MarshalledValueInputStream(new ByteArrayInputStream(buf)); + ObjectInputStream in = new MarshalledValueInputStream(new ByteArrayInputStream(bytes, offset, len)); try { Modified: core/branches/2.2.X/src/main/java/org/jboss/cache/remoting/jgroups/ChannelMessageListener.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/remoting/jgroups/ChannelMessageListener.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/remoting/jgroups/ChannelMessageListener.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -7,8 +7,8 @@ import org.jboss.cache.config.Configuration; import org.jboss.cache.factories.annotations.Inject; import org.jboss.cache.factories.annotations.NonVolatile; +import org.jboss.cache.io.ExposedByteArrayOutputStream; import org.jboss.cache.statetransfer.StateTransferManager; -import org.jboss.cache.util.ExposedByteArrayOutputStream; import org.jboss.util.stream.MarshalledValueInputStream; import org.jboss.util.stream.MarshalledValueOutputStream; import org.jgroups.ExtendedMessageListener; Deleted: core/branches/2.2.X/src/main/java/org/jboss/cache/util/ExposedByteArrayOutputStream.java =================================================================== --- core/branches/2.2.X/src/main/java/org/jboss/cache/util/ExposedByteArrayOutputStream.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/main/java/org/jboss/cache/util/ExposedByteArrayOutputStream.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -1,141 +0,0 @@ -/* - * JBoss, the OpenSource J2EE webOS - * - * Distributable under LGPL license. - * See terms of license at gnu.org. - */ - -package org.jboss.cache.util; - -import java.io.ByteArrayOutputStream; - -/** - * Extends ByteArrayOutputStream, but exposes the internal buffer. - * Using this, callers don't need to call toByteArray() which copies the - * internal buffer. - * <p> - * Also overrides the superclass' behavior of always doubling the size of the - * internal buffer any time more capacity is needed. This class doubles the - * size until the internal buffer reaches a configurable max size (default is - * 4MB), after which it begins growing the buffer in 25% increments. This is - * intended to help prevent an OutOfMemoryError during a resize of a large - * buffer. - * </p> - * <p> - * A version of this class was originally created by Bela Ban as part of the - * JGroups library. - * </p> - * - * @author <a href="mailto://brian.stansberry@jboss.com">Brian Stansberry</a> - * @version $Id$ - */ -public class ExposedByteArrayOutputStream extends ByteArrayOutputStream -{ - /** - * Default buffer size after which if more buffer capacity - * is needed the buffer will grow by 25% rather than 100% - */ - public static final int DEFAULT_DOUBLING_SIZE = 4 * 1024 * 1024; // 4MB - - private int maxDoublingSize = DEFAULT_DOUBLING_SIZE; - - public ExposedByteArrayOutputStream() - { - super(); - } - - public ExposedByteArrayOutputStream(int size) - { - super(size); - } - - /** - * Creates a new byte array output stream, with a buffer capacity of - * the specified size, in bytes. - * - * @param size the initial size. - * @param maxDoublingSize the buffer size, after which if more capacity - * is needed the buffer will grow by 25% - * rather than 100% - * @throws IllegalArgumentException if size is negative. - */ - public ExposedByteArrayOutputStream(int size, int maxDoublingSize) - { - super(size); - this.maxDoublingSize = maxDoublingSize; - } - - /** - * Gets the internal buffer array. Note that the length of this array - * will almost certainly be longer than the data written to it; call - * <code>size()</code> to get the number of bytes of actual data. - */ - public byte[] getRawBuffer() - { - return buf; - } - - @Override - public synchronized void write(byte[] b, int off, int len) - { - if ((off < 0) || (off > b.length) || (len < 0) || - ((off + len) > b.length) || ((off + len) < 0)) - { - throw new IndexOutOfBoundsException(); - } - else if (len == 0) - { - return; - } - - int newcount = count + len; - if (newcount > buf.length) - { - byte newbuf[] = new byte[getNewBufferSize(buf.length, newcount)]; - System.arraycopy(buf, 0, newbuf, 0, count); - buf = newbuf; - } - - System.arraycopy(b, off, buf, count, len); - count = newcount; - } - - @Override - public synchronized void write(int b) - { - int newcount = count + 1; - if (newcount > buf.length) - { - byte newbuf[] = new byte[getNewBufferSize(buf.length, newcount)]; - System.arraycopy(buf, 0, newbuf, 0, count); - buf = newbuf; - } - buf[count] = (byte) b; - count = newcount; - } - - /** - * Gets the highest internal buffer size after which if more capacity - * is needed the buffer will grow in 25% increments rather than 100%. - */ - public int getMaxDoublingSize() - { - return maxDoublingSize; - } - - /** - * Gets the number of bytes to which the internal buffer should be resized. - * - * @param curSize the current number of bytes - * @param minNewSize the minimum number of bytes required - * @return the size to which the internal buffer should be resized - */ - public int getNewBufferSize(int curSize, int minNewSize) - { - if (curSize <= maxDoublingSize) - return Math.max(curSize << 1, minNewSize); - else - return Math.max(curSize + (curSize >> 2), minNewSize); - } - -} Modified: core/branches/2.2.X/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java =================================================================== --- core/branches/2.2.X/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java 2008-10-13 22:20:04 UTC (rev 6926) +++ core/branches/2.2.X/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java 2008-10-14 09:23:28 UTC (rev 6927) @@ -9,10 +9,13 @@ import org.jboss.cache.commands.tx.PrepareCommand; import org.jboss.cache.factories.ComponentRegistry; import org.jboss.cache.marshall.CommandAwareRpcDispatcher; +import org.jboss.cache.marshall.InactiveRegionAwareRpcDispatcher; import org.jboss.cache.marshall.Marshaller; import org.jboss.cache.marshall.RegionalizedMethodCall; import org.jboss.cache.util.TestingUtil; import org.jgroups.blocks.RpcDispatcher; +import org.jgroups.blocks.RpcDispatcher.Marshaller2; +import org.jgroups.util.Buffer; import java.io.InputStream; import java.io.ObjectInputStream; @@ -57,8 +60,15 @@ ComponentRegistry componentRegistry = TestingUtil.extractComponentRegistry(cache); RPCManager rpcManager = componentRegistry.getComponent(RPCManager.class); CommandAwareRpcDispatcher realDispatcher = (CommandAwareRpcDispatcher) TestingUtil.extractField(rpcManager, "rpcDispatcher"); - RpcDispatcher.Marshaller realMarshaller = realDispatcher.getMarshaller(); - MarshallerDelegate delegate = new MarshallerDelegate(realMarshaller); + RpcDispatcher.Marshaller2 realMarshaller = (Marshaller2) realDispatcher.getMarshaller(); + + RpcDispatcher.Marshaller delegate = null; + + if (realDispatcher instanceof InactiveRegionAwareRpcDispatcher) + delegate = new RegionMarshallerDelegate((Marshaller) realMarshaller); + else + delegate = new MarshallerDelegate(realMarshaller); + realDispatcher.setMarshaller(delegate); realDispatcher.setRequestMarshaller(delegate); realDispatcher.setResponseMarshaller(delegate); @@ -194,6 +204,16 @@ { return realOne.objectFromByteBuffer(bytes); } + + public Buffer objectToBuffer(Object obj) throws Exception + { + return realOne.objectToBuffer(obj); + } + + public Object objectFromByteBuffer(byte[] buf, int offset, int length) throws Exception + { + return realOne.objectFromByteBuffer(buf, offset, length); + } } /**