diff --git a/dump/src/util/dump/ExternalizableBean.java b/dump/src/util/dump/ExternalizableBean.java index 536646a..6224d4a 100644 --- a/dump/src/util/dump/ExternalizableBean.java +++ b/dump/src/util/dump/ExternalizableBean.java @@ -815,6 +815,7 @@ default int readExternalFieldNumber( ObjectInput in ) throws IOException { @Override default void writeExternal( ObjectOutput out ) throws IOException { + StreamCache streamCache = null; try { ClassConfig _config = getConfig(getClass()); @@ -826,7 +827,6 @@ default void writeExternal( ObjectOutput out ) throws IOException { } ObjectOutput originalOut = out; - StreamCache streamCache = null; BytesCache bytesCache = null; ObjectOutput cachingOut = null; @@ -1207,6 +1207,7 @@ default void writeExternal( ObjectOutput out ) throws IOException { originalOut.writeInt(bytesCache.size()); bytesCache.writeTo(originalOut); out = originalOut; + bytesCache.reset(); streamCache._inUse = false; } } @@ -1214,6 +1215,16 @@ default void writeExternal( ObjectOutput out ) throws IOException { catch ( Exception e ) { throw new RuntimeException("Failed to externalize class " + getClass().getName(), e); } + finally { + if ( streamCache != null && streamCache._inUse ) { + // a field failed to write: the cached ObjectOutput may hold partial state, so never reuse this StreamCache. + // Replace it in the thread-local (if it is the one from there), which also drops its buffer. + streamCache._bytesCache.reset(); + if ( STREAM_CACHE.get() == streamCache ) { + STREAM_CACHE.set(new StreamCache()); + } + } + } } /** diff --git a/dump/src/util/dump/ExternalizationHelper.java b/dump/src/util/dump/ExternalizationHelper.java index 5edafe7..41f4a0d 100644 --- a/dump/src/util/dump/ExternalizationHelper.java +++ b/dump/src/util/dump/ExternalizationHelper.java @@ -58,6 +58,10 @@ class ExternalizationHelper { static final Map CLASS_CHANGED_INCOMPATIBLY = new ConcurrentHashMap<>(); static final ThreadLocal STREAM_CACHE = ThreadLocal.withInitial(StreamCache::new); + /** Buffers larger than this are not retained by a {@link BytesCache} after a write finished; they are replaced by a small one. */ + static final int BYTES_CACHE_MAX_RETAINED_CAPACITY = 1024 * 1024; + static final int BYTES_CACHE_INITIAL_CAPACITY = 1024; + static { try { boolean config = Boolean.parseBoolean(System.getProperty("ExternalizableBean.USE_UNSAFE_FIELD_ACCESSORS", "true")); @@ -1075,7 +1079,7 @@ static class BytesCache extends OutputStream { protected int _count; public BytesCache() { - this(1024); + this(BYTES_CACHE_INITIAL_CAPACITY); } public BytesCache( int size ) { @@ -1085,11 +1089,14 @@ public BytesCache( int size ) { _buffer = new byte[size]; } + /** + * Marks the buffer as empty and drops an oversized backing array, so that it is not retained (e.g. by a long-living thread). + */ public void reset() { _count = 0; - if ( _buffer.length > 1048576 ) { + if ( _buffer.length > BYTES_CACHE_MAX_RETAINED_CAPACITY ) { // let it shrink - _buffer = new byte[1024]; + _buffer = new byte[BYTES_CACHE_INITIAL_CAPACITY]; } } diff --git a/dump/test/util/dump/ExternalizationBufferRetentionTest.java b/dump/test/util/dump/ExternalizationBufferRetentionTest.java new file mode 100644 index 0000000..006dd69 --- /dev/null +++ b/dump/test/util/dump/ExternalizationBufferRetentionTest.java @@ -0,0 +1,147 @@ +package util.dump; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.junit.Test; + +import util.dump.ExternalizableBean.externalize; +import util.dump.ExternalizationHelper.BytesCache; +import util.dump.ExternalizationHelper.StreamCache; + +public class ExternalizationBufferRetentionTest { + + public static class LargeBean implements ExternalizableBean { + + // List fields have a dynamic length and therefore go through the thread-local buffer. + // The large field must be written last: a following dynamic field would reset() (and thereby shrink) the buffer anyway. + @externalize(1) + public List _nested; + + @externalize(2) + public List _data; + + public LargeBean() {} + + public LargeBean( byte[] data, LargeBean nested ) { + _data = new ArrayList<>(Collections.singletonList(data)); + _nested = nested == null ? new ArrayList<>() : new ArrayList<>(Collections.singletonList(nested)); + } + } + + private static byte[] write( LargeBean bean ) throws Exception { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try ( ObjectOutputStream out = new ObjectOutputStream(bytes) ) { + bean.writeExternal(out); + } + return bytes.toByteArray(); + } + + @Test + public void largeWriteDoesNotRetainLargeBuffer() throws Exception { + int size = ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY * 4; + byte[] serialized = write(new LargeBean(new byte[size], null)); + assertThat(serialized.length).isGreaterThan(size); + + StreamCache cache = ExternalizationHelper.STREAM_CACHE.get(); + assertThat(cache._inUse).isFalse(); + assertThat(cache._bytesCache._buffer.length).isLessThanOrEqualTo(ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY); + } + + @Test + public void smallWriteKeepsBufferForReuse() throws Exception { + write(new LargeBean(new byte[10_000], null)); + StreamCache cache = ExternalizationHelper.STREAM_CACHE.get(); + byte[] retained = cache._bytesCache._buffer; + assertThat(retained.length).isGreaterThanOrEqualTo(10_000); + + write(new LargeBean(new byte[10_000], null)); + assertThat(cache._bytesCache._buffer).isSameAs(retained); + } + + @Test + public void nestedLargeWriteRoundtripsAndDoesNotRetain() throws Exception { + int size = ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY * 2; + LargeBean inner = new LargeBean(new byte[size], null); + LargeBean outer = new LargeBean(new byte[size], inner); + byte[] serialized = write(outer); + + LargeBean read = new LargeBean(); + try ( ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(serialized)) ) { + read.readExternal(in); + } + assertThat(read._data.get(0)).hasSize(size); + assertThat(read._nested.get(0)._data.get(0)).hasSize(size); + + StreamCache cache = ExternalizationHelper.STREAM_CACHE.get(); + assertThat(cache._inUse).isFalse(); + assertThat(cache._bytesCache._buffer.length).isLessThanOrEqualTo(ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY); + } + + @Test + public void failedWriteDoesNotPoisonNextWrite() throws Exception { + LargeBean broken = new LargeBean(new byte[ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY * 2], null); + broken._data.add(null); + broken._nested.add(new LargeBean(null, null) { + @Override + public void writeExternal( java.io.ObjectOutput out ) { + throw new IllegalStateException("boom"); + } + }); + try { + write(broken); + org.junit.Assert.fail("expected failure"); + } + catch ( RuntimeException expected ) { + // expected + } + + StreamCache cache = ExternalizationHelper.STREAM_CACHE.get(); + assertThat(cache._inUse).isFalse(); + assertThat(cache._bytesCache._buffer.length).isLessThanOrEqualTo(ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY); + + LargeBean good = new LargeBean(new byte[] { 1, 2, 3 }, null); + LargeBean read = new LargeBean(); + try ( ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(write(good))) ) { + read.readExternal(in); + } + assertThat(read._data.get(0)).containsExactly(1, 2, 3); + } + + @Test + public void largeWriteOnWorkerThreadDoesNotRetainLargeBufferOnThatThread() throws Exception { + // the buffer is retained per thread, so check the thread-local of the long-living thread that wrote + java.util.concurrent.ExecutorService executor = java.util.concurrent.Executors.newSingleThreadExecutor(); + try { + int size = ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY * 4; + executor.submit(() -> write(new LargeBean(new byte[size], null))).get(); + int retainedOnWorker = executor.submit(() -> ExternalizationHelper.STREAM_CACHE.get()._bytesCache._buffer.length).get(); + assertThat(retainedOnWorker).isLessThanOrEqualTo(ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY); + } + finally { + executor.shutdownNow(); + } + } + + @Test + public void bytesCacheKeepsBufferAtCapAndDropsAboveCap() { + BytesCache atCap = new BytesCache(ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY); + byte[] kept = atCap._buffer; + atCap.reset(); + assertThat(atCap._buffer).isSameAs(kept); + assertThat(atCap.size()).isZero(); + + BytesCache aboveCap = new BytesCache(ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY + 1); + aboveCap.reset(); + assertThat(aboveCap._buffer.length).isEqualTo(ExternalizationHelper.BYTES_CACHE_INITIAL_CAPACITY); + assertThat(aboveCap.size()).isZero(); + } +} diff --git a/dump/test/util/dump/ExternalizationBufferStressTester.java b/dump/test/util/dump/ExternalizationBufferStressTester.java new file mode 100644 index 0000000..545e7e1 --- /dev/null +++ b/dump/test/util/dump/ExternalizationBufferStressTester.java @@ -0,0 +1,212 @@ +package util.dump; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutput; +import java.io.ObjectOutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Random; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; + +import util.dump.ExternalizableBean.externalize; + + +/** + * Manual stress test (not run by the build) for the thread-local buffer in {@link ExternalizableBean#writeExternal}. + *

+ * Many threads serialize beans of mixed sizes (mostly small, occasionally huge, sometimes nested, sometimes failing) and verify the + * roundtrip. After every write each thread checks that the buffer it keeps is not larger than the cap, i.e. that a large write did + * not leave a large array behind on a long-living thread. + *

+ * Run: {@code java -Xmx512m -cp target/classes:target/test-classes: util.dump.ExternalizationBufferStressTester [threads] [seconds]} + *

+ * Exit code 0 means no corruption and no retained oversized buffer was observed. + */ +public class ExternalizationBufferStressTester { + + public static class Bean implements ExternalizableBean { + + @externalize(1) + public List _nested = new ArrayList<>(); + + @externalize(2) + public List _strings = new ArrayList<>(); + + // must be the LAST dynamic field: a following dynamic field would reset() (and thereby shrink) the buffer, hiding the leak + @externalize(3) + public List _data = new ArrayList<>(); + + // fails on demand, to exercise the failure path + public boolean _failOnWrite; + + @Override + public void writeExternal( ObjectOutput out ) throws IOException { + if ( _failOnWrite ) { + throw new IllegalStateException("simulated failure"); + } + ExternalizableBean.super.writeExternal(out); + } + + @Override + public boolean equals( Object o ) { + if ( !(o instanceof Bean) ) { + return false; + } + Bean b = (Bean)o; + if ( _data.size() != b._data.size() || !_strings.equals(b._strings) || !_nested.equals(b._nested) ) { + return false; + } + for ( int i = 0; i < _data.size(); i++ ) { + if ( !Arrays.equals(_data.get(i), b._data.get(i)) ) { + return false; + } + } + return true; + } + + @Override + public int hashCode() { + return _data.size() + 31 * _strings.hashCode(); + } + } + + + private static final int CAP = ExternalizationHelper.BYTES_CACHE_MAX_RETAINED_CAPACITY; + + private static final AtomicLong WRITES = new AtomicLong(); + private static final AtomicLong HUGE_WRITES = new AtomicLong(); + private static final AtomicLong FAILED_WRITES = new AtomicLong(); + private static final AtomicLong CORRUPTIONS = new AtomicLong(); + private static final AtomicLong RETAINED_VIOLATIONS = new AtomicLong(); + private static final AtomicLong MAX_RETAINED_SEEN = new AtomicLong(); + + + public static void main( String[] args ) throws Exception { + int threads = args.length > 0 ? Integer.parseInt(args[0]) : 16; + int seconds = args.length > 1 ? Integer.parseInt(args[1]) : 30; + System.out.println("threads=" + threads + ", seconds=" + seconds + ", cap=" + CAP + " B, maxHeap=" + Runtime.getRuntime().maxMemory() / (1024 * 1024) + " MB"); + + long end = System.nanoTime() + TimeUnit.SECONDS.toNanos(seconds); + ExecutorService executor = Executors.newFixedThreadPool(threads); + List> futures = new ArrayList<>(); + for ( int t = 0; t < threads; t++ ) { + long seed = t * 7919L + System.nanoTime(); + futures.add(executor.submit(() -> worker(new Random(seed), end))); + } + + long nextReport = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while ( System.nanoTime() < end ) { + Thread.sleep(200); + if ( System.nanoTime() >= nextReport ) { + nextReport += TimeUnit.SECONDS.toNanos(5); + report(); + } + } + for ( java.util.concurrent.Future f : futures ) { + f.get(); + } + executor.shutdown(); + report(); + + boolean ok = CORRUPTIONS.get() == 0 && RETAINED_VIOLATIONS.get() == 0; + System.out.println(ok ? "OK" : "FAILED"); + System.exit(ok ? 0 : 1); + } + + private static void report() { + Runtime rt = Runtime.getRuntime(); + System.out.printf("writes=%d (huge=%d, failed=%d) corruptions=%d retainedViolations=%d maxRetainedAfterWrite=%d B usedHeap=%d MB%n", // + WRITES.get(), HUGE_WRITES.get(), FAILED_WRITES.get(), CORRUPTIONS.get(), RETAINED_VIOLATIONS.get(), MAX_RETAINED_SEEN.get(), + (rt.totalMemory() - rt.freeMemory()) / (1024 * 1024)); + } + + private static void worker( Random random, long end ) { + while ( System.nanoTime() < end ) { + try { + boolean fail = random.nextInt(200) == 0; + Bean bean = newBean(random, 0); + bean._failOnWrite = fail; + + byte[] bytes; + try { + bytes = write(bean); + if ( fail ) { + CORRUPTIONS.incrementAndGet(); + System.err.println("expected failure did not happen"); + } + } + catch ( RuntimeException e ) { + if ( !fail ) { + throw e; + } + FAILED_WRITES.incrementAndGet(); + checkRetained(); + continue; + } + + Bean read = new Bean(); + try ( ObjectInputStream in = new ObjectInputStream(new ByteArrayInputStream(bytes)) ) { + read.readExternal(in); + } + if ( !bean.equals(read) ) { + CORRUPTIONS.incrementAndGet(); + System.err.println("roundtrip mismatch"); + } + WRITES.incrementAndGet(); + checkRetained(); + } + catch ( Exception e ) { + CORRUPTIONS.incrementAndGet(); + e.printStackTrace(); + } + } + } + + /** After a write has finished, the buffer kept by this thread must be small, and the cache must be free for reuse. */ + private static void checkRetained() { + ExternalizationHelper.StreamCache cache = ExternalizationHelper.STREAM_CACHE.get(); + int retained = cache._bytesCache._buffer.length; + MAX_RETAINED_SEEN.accumulateAndGet(retained, Math::max); + if ( retained > CAP || cache._inUse ) { + RETAINED_VIOLATIONS.incrementAndGet(); + System.err.println("retained buffer " + retained + " B, inUse=" + cache._inUse + " on " + Thread.currentThread().getName()); + } + } + + private static Bean newBean( Random random, int depth ) { + Bean b = new Bean(); + int kind = random.nextInt(1000); + int dataBytes = kind < 3 ? CAP * (2 + random.nextInt(6)) // huge: 2 to 7 MB + : kind < 50 ? 10_000 + random.nextInt(200_000) // medium + : random.nextInt(200); // small + if ( dataBytes >= CAP ) { + HUGE_WRITES.incrementAndGet(); + } + byte[] data = new byte[dataBytes]; + random.nextBytes(data); + b._data.add(data); + for ( int i = random.nextInt(3); i > 0; i-- ) { + b._strings.add("s" + random.nextInt()); + } + if ( depth < 3 && random.nextInt(4) == 0 ) { + b._nested.add(newBean(random, depth + 1)); + } + return b; + } + + private static byte[] write( Bean bean ) throws IOException { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try ( ObjectOutputStream out = new ObjectOutputStream(bytes) ) { + bean.writeExternal(out); + } + return bytes.toByteArray(); + } +}