Author: manik.surtani(a)jboss.com Date: 2007-10-16 10:19:21 -0400 (Tue, 16 Oct 2007) New Revision: 4623 Modified: core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java Log: JBCACHE-1170 - Leakage of marshalling region stored in thread local Modified: core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java =================================================================== --- core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java 2007-10-16 14:17:46 UTC (rev 4622) +++ core/trunk/src/main/java/org/jboss/cache/marshall/CacheMarshaller200.java 2007-10-16 14:19:21 UTC (rev 4623) @@ -80,7 +80,7 @@ init(manager, defaultInactive, useRegionBasedMarshalling); if (useRegionBasedMarshalling) { - log.debug("Using region based marshalling logic : marshalling Fqn as a String first for every call."); + log.debug("Using region based marshalling logic : marshalling Fqn first for every call."); } } @@ -106,6 +106,8 @@ { // if the return value we're trying to marshall is null we're easy ... region = null; + // we still need to clear the thread local though. + regionForCall.remove(); } else if (isReturnValue(o)) { @@ -113,6 +115,8 @@ // let's see if an incoming unmarshalling call for this exists, in the same thread stack and had registered // a Fqn region. region = regionForCall.get(); + regionForCall.remove(); + if (log.isTraceEnabled()) log.trace("Suspect this is a return value. Extract region from ThreadLocal as " + region); // otherwise, we need to marshall the retval. } @@ -188,6 +192,7 @@ } if (region == null) { + if (log.isDebugEnabled()) log.debug("Region does not exist for Fqn " + regionFqn + " - not using a context classloader."); retValue = unmarshallObject(in, refMap); } else @@ -655,7 +660,7 @@ return gtx; } - private Fqn unmarshallFqn(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception + Fqn unmarshallFqn(ObjectInputStream in, Map<Integer, Object> refMap) throws Exception { boolean isRoot = in.readBoolean(); Modified: core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java 2007-10-16 14:17:46 UTC (rev 4622) +++ core/trunk/src/test/java/org/jboss/cache/marshall/CacheMarshaller200Test.java 2007-10-16 14:19:21 UTC (rev 4623) @@ -6,11 +6,23 @@ */ package org.jboss.cache.marshall; +import org.jboss.cache.Fqn; +import org.jboss.cache.Region; +import org.jboss.cache.RegionManager; import static org.testng.AssertJUnit.assertEquals; - -import org.jboss.cache.RegionManager; import org.testng.annotations.Test; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.HashMap; +import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + @Test(groups = {"functional"}) public class CacheMarshaller200Test extends CacheMarshallerTestBase { @@ -33,4 +45,165 @@ assertEquals("hello", unmarshalled); } + public void testRegionalisedStream() throws Exception + { + // need to test what's going on with + CacheMarshaller200 cm200 = new CacheMarshaller200(new RegionManager(), false, true); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + cm200.objectToObjectStream("Hello World", oos, Fqn.fromString("/hello")); + oos.close(); + + ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray())); + + // test that the first thing on the stream is the fqn! + byte magic = ois.readByte(); + short ref = ois.readShort(); + System.out.println("Magic number is " + magic); + + assert magic == CacheMarshaller200.MAGICNUMBER_FQN; + + // now the chunks of an Fqn + Fqn f = cm200.unmarshallFqn(ois, new HashMap<Integer, Object>()); + + assert f.equals(Fqn.fromString("/hello")); + } + + /** + * This test emulates the behaviour observed when an incoming method call is unmarshalled (using objectFromOutputStream), + * region based marshalling is used, and it is expected that we would need to marshall the response using the same + * region as well. To deal with this, the region (as an Fqn) is read off the stream when unmarshalling and then + * stored in a ThreadLocal so it can be accessed for when the response needs to be marshalled again. + * + * The problem here - registered as JBCACHE-1170 - is that this has a tendency to leak scope and affect more than the + * required method call. + * + */ + public void testLeakageOfFqn() throws Throwable + { + // Use a thread pool so that we know threads will be reused. + // You don't need any concurrency here to demonstrate the failure - a single thread is enough. + ExecutorService e = Executors.newFixedThreadPool(1); + // to capture throwables + final List<Throwable> throwables = new CopyOnWriteArrayList<Throwable>(); + + // each thread will perform 1 of 3 tasks: + // 1. unmarshall a stream, and marshall a response - typical case such as a clustered get + // 2. unmarshall a stream, and marshall a 'void' response. + // 3. marshall a (primitive response) - case such as a state transfer sending out boolean status + + // first create a stream to unmarshall. + RegionManager rm = new RegionManager(); + final CacheMarshaller200 cm200 = new CacheMarshaller200(rm, false, true); + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos); + final Fqn region = Fqn.fromString("/hello"); + Region r = rm.getRegion(region, true); + r.registerContextClassLoader(this.getClass().getClassLoader()); + cm200.objectToObjectStream(MethodCallFactory.create(MethodDeclarations.clusteredGetMethod, null, null), oos, region); + oos.close(); + + final byte[] stream = baos.toByteArray(); + // so now the stream starts with the Fqn "/hello". + + // repeat 100 times + for (int i=0; i<100; i++) + { + if (i % 3 == 0) + { + // task 1 above + e.execute(new Runnable() + { + public void run() + { + try + { + cm200.objectFromObjectStream(new ObjectInputStream(new ByteArrayInputStream(stream))); + ByteArrayOutputStream out = new ByteArrayOutputStream(); + ObjectOutputStream outStream = new ObjectOutputStream(out); + cm200.objectToObjectStream("A result", outStream); + outStream.close(); + out.close(); + // test that the output stream has got "/hello" as it's region Fqn. + ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(out.toByteArray())); + assert in.readByte() == CacheMarshaller200.MAGICNUMBER_FQN : "The stream should start with an Fqn"; + // discard the nest refId short + in.readShort(); + Fqn f = cm200.unmarshallFqn(in, new HashMap<Integer, Object>()); + assert region.equals(f) : "Should use the same region for the response as was used for the request!"; + + } + catch (Throwable t) + { + throwables.add(t); + } + } + }); + } + else if (i % 3 == 1) + { + // task 2 above + e.execute(new Runnable() + { + public void run() + { + try + { + cm200.objectFromObjectStream(new ObjectInputStream(new ByteArrayInputStream(stream))); + // and now just send back a 'void' return type (In JGroups this is treated as a null) + cm200.objectToObjectStream(null, new ObjectOutputStream(new ByteArrayOutputStream())); + } + catch (Throwable t) + { + throwables.add(t); + } + } + }); + } + else if (i % 3 == 2) + { + // task 3 above + e.execute(new Runnable() + { + public void run() + { + try + { + + // and now don't bother with any umarshalling + // directly marshall a boolean. + ByteArrayOutputStream out = new ByteArrayOutputStream(); + ObjectOutputStream outStream = new ObjectOutputStream(out); + cm200.objectToObjectStream(true, outStream); + outStream.close(); + out.close(); + // test that the output stream has got "/hello" as it's region Fqn. + ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(out.toByteArray())); + byte magic = in.readByte(); + + assert magic != CacheMarshaller200.MAGICNUMBER_FQN : "The stream should NOT start with an Fqn!"; + assert magic == CacheMarshaller200.MAGICNUMBER_NULL : "Should start with a NULL"; + assert in.readByte() == CacheMarshaller200.MAGICNUMBER_BOOLEAN : "Should have a boolean magic number before the boolean value"; + assert in.readBoolean() : "The boolean written to the stream should be true"; + + } + catch (Throwable t) + { + throwables.add(t); + } + } + }); + } + } + + e.shutdown(); + e.awaitTermination(60, TimeUnit.SECONDS); + + for (Throwable t : throwables) + { + t.printStackTrace(); + } + + assert throwables.size() == 0 : "Should not have caught any exceptions!"; + } }