Skip to content

Commit dda7a6d

Browse files
cowtowncodercowtowncoder
committed
Rewrite encoder/decoder recycling to work reliably (2.7 has recycling just disabled)
1 parent c735002 commit dda7a6d

File tree

4 files changed

+125
-82
lines changed

4 files changed

+125
-82
lines changed

‎avro/src/main/java/com/fasterxml/jackson/dataformat/avro/AvroGenerator.java

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -146,6 +146,7 @@ public AvroGenerator(IOContext ctxt, int jsonFeatures, int avroFeatures,
146146
_formatFeatures = avroFeatures;
147147
_output = output;
148148
_avroContext = AvroWriteContext.createNullContext();
149+
_encoder = CodecRecycler.encoder(_output, isEnabled(Feature.AVRO_BUFFERING));
149150
}
150151

151152
public void setSchema(AvroSchema schema)
@@ -155,9 +156,6 @@ public void setSchema(AvroSchema schema)
155156
}
156157
_rootSchema = schema;
157158
// start with temporary root...
158-
if (_encoder == null) {
159-
_encoder = AvroSchema.encoder(_output, isEnabled(Feature.AVRO_BUFFERING));
160-
}
161159
_avroContext = _rootContext = AvroWriteContext.createRootContext(this,
162160
schema.getAvroSchema(), _encoder);
163161
}
@@ -579,7 +577,12 @@ protected final void _verifyValueWrite(String typeMsg) throws IOException {
579577

580578
@Override
581579
protected void _releaseBuffers() {
582-
// nothing special to do...
580+
// no super implementation to call
581+
BinaryEncoder e = _encoder;
582+
if (e != null) {
583+
_encoder = null;
584+
CodecRecycler.release(e);
585+
}
583586
}
584587

585588
/*

‎avro/src/main/java/com/fasterxml/jackson/dataformat/avro/AvroSchema.java

Lines changed: 6 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -1,19 +1,14 @@
11
package com.fasterxml.jackson.dataformat.avro;
22

3-
import java.io.*;
4-
import java.lang.ref.SoftReference;
53
import java.util.concurrent.atomic.AtomicReference;
64

5+
import org.apache.avro.Schema;
6+
77
import com.fasterxml.jackson.core.FormatSchema;
8+
89
import com.fasterxml.jackson.dataformat.avro.deser.AvroReaderFactory;
910
import com.fasterxml.jackson.dataformat.avro.deser.AvroStructureReader;
1011

11-
import org.apache.avro.Schema;
12-
import org.apache.avro.io.BinaryDecoder;
13-
import org.apache.avro.io.BinaryEncoder;
14-
import org.apache.avro.io.DecoderFactory;
15-
import org.apache.avro.io.EncoderFactory;
16-
1712
/**
1813
* Wrapper for Schema information needed to encode and decode Avro-format
1914
* data.
@@ -22,18 +17,11 @@ public class AvroSchema implements FormatSchema
2217
{
2318
public final static String TYPE_ID = "avro";
2419

25-
protected final static DecoderFactory DECODER_FACTORY = DecoderFactory.get();
26-
27-
protected final static EncoderFactory ENCODER_FACTORY = EncoderFactory.get();
28-
29-
protected final static ThreadLocal<SoftReference<BinaryDecoder>> decoderRecycler
30-
= new ThreadLocal<SoftReference<BinaryDecoder>>();
31-
32-
protected final static ThreadLocal<SoftReference<BinaryEncoder>> encoderRecycler
33-
= new ThreadLocal<SoftReference<BinaryEncoder>>();
34-
3520
protected final Schema _avroSchema;
3621

22+
/**
23+
* Lazily instantiated value reader for this schema.
24+
*/
3725
protected final AtomicReference<AvroStructureReader> _reader = new AtomicReference<AvroStructureReader>();
3826

3927
public AvroSchema(Schema asch)
@@ -48,61 +36,6 @@ public String getSchemaType() {
4836

4937
public Schema getAvroSchema() { return _avroSchema; }
5038

51-
public static BinaryDecoder decoder(InputStream in, boolean buffering)
52-
{
53-
/*
54-
SoftReference<BinaryDecoder> ref = decoderRecycler.get();
55-
BinaryDecoder prev = (ref == null) ? null : ref.get();
56-
// Factory will check if the decoder has a matching type for reuse.
57-
// If not, it will drop the instance being reused and will return a new, proper one.
58-
BinaryDecoder next = buffering
59-
? DECODER_FACTORY.binaryDecoder(in, prev)
60-
: DECODER_FACTORY.directBinaryDecoder(in, prev);
61-
62-
decoderRecycler.set(new SoftReference<BinaryDecoder>(next));
63-
return next;
64-
*/
65-
return buffering
66-
? DECODER_FACTORY.binaryDecoder(in, null)
67-
: DECODER_FACTORY.directBinaryDecoder(in, null);
68-
}
69-
70-
public static BinaryDecoder decoder(byte[] buffer, int offset, int len)
71-
{
72-
/*
73-
SoftReference<BinaryDecoder> ref = decoderRecycler.get();
74-
BinaryDecoder prev = (ref == null) ? null : ref.get();
75-
76-
if (prev != null) {
77-
return DECODER_FACTORY.binaryDecoder(buffer, offset, len, prev);
78-
}
79-
prev = DECODER_FACTORY.binaryDecoder(buffer, offset, len, null);
80-
decoderRecycler.set(new SoftReference<BinaryDecoder>(prev));
81-
return prev;
82-
*/
83-
return DECODER_FACTORY.binaryDecoder(buffer, offset, len, null);
84-
}
85-
86-
public static BinaryEncoder encoder(OutputStream out, boolean buffering)
87-
{
88-
/*
89-
SoftReference<BinaryEncoder> ref = encoderRecycler.get();
90-
BinaryEncoder prev = (ref == null) ? null : ref.get();
91-
// Factory will check if the decoder has a matching type for reuse.
92-
// If not, it will drop the instance being reused and will return a new, proper one.
93-
BinaryEncoder next =
94-
buffering
95-
? ENCODER_FACTORY.binaryEncoder(out, prev)
96-
: ENCODER_FACTORY.directBinaryEncoder(out, prev);
97-
98-
encoderRecycler.set(new SoftReference<BinaryEncoder>(next));
99-
return next;
100-
*/
101-
return buffering
102-
? ENCODER_FACTORY.binaryEncoder(out, null)
103-
: ENCODER_FACTORY.directBinaryEncoder(out, null);
104-
}
105-
10639
public AvroStructureReader getReader()
10740
{
10841
AvroStructureReader r = _reader.get();

‎avro/src/main/java/com/fasterxml/jackson/dataformat/avro/CodecRecycler.java

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
package com.fasterxml.jackson.dataformat.avro;
2+
3+
import java.io.InputStream;
4+
import java.io.OutputStream;
5+
import java.lang.ref.SoftReference;
6+
7+
import org.apache.avro.io.BinaryDecoder;
8+
import org.apache.avro.io.BinaryEncoder;
9+
import org.apache.avro.io.DecoderFactory;
10+
import org.apache.avro.io.EncoderFactory;
11+
12+
/**
13+
* Simple helper class that contains extracted functionality for
14+
* simple encoder/decoder recycling.
15+
*
16+
* @since 2.8.7
17+
*/
18+
public final class CodecRecycler
19+
{
20+
protected final static DecoderFactory DECODER_FACTORY = DecoderFactory.get();
21+
22+
protected final static EncoderFactory ENCODER_FACTORY = EncoderFactory.get();
23+
24+
protected final static ThreadLocal<SoftReference<CodecRecycler>> _recycler
25+
= new ThreadLocal<SoftReference<CodecRecycler>>();
26+
27+
private BinaryDecoder decoder;
28+
private BinaryEncoder encoder;
29+
30+
private CodecRecycler() { }
31+
32+
/*
33+
/**********************************************************
34+
/* Public API
35+
/**********************************************************
36+
*/
37+
38+
public static BinaryDecoder decoder(InputStream in, boolean buffering)
39+
{
40+
BinaryDecoder prev = _recycler().claimDecoder();
41+
return buffering
42+
? DECODER_FACTORY.binaryDecoder(in, prev)
43+
: DECODER_FACTORY.directBinaryDecoder(in, prev);
44+
}
45+
46+
public static BinaryDecoder decoder(byte[] buffer, int offset, int len)
47+
{
48+
BinaryDecoder prev = _recycler().claimDecoder();
49+
return DECODER_FACTORY.binaryDecoder(buffer, offset, len, prev);
50+
}
51+
52+
public static BinaryEncoder encoder(OutputStream out, boolean buffering)
53+
{
54+
BinaryEncoder prev = _recycler().claimEncoder();
55+
return buffering
56+
? ENCODER_FACTORY.binaryEncoder(out, prev)
57+
: ENCODER_FACTORY.directBinaryEncoder(out, prev);
58+
}
59+
60+
public static void release(BinaryDecoder dec) {
61+
_recycler().decoder = dec;
62+
}
63+
64+
public static void release(BinaryEncoder enc) {
65+
_recycler().encoder = enc;
66+
}
67+
68+
/*
69+
/**********************************************************
70+
/* Internal per-instance methods
71+
/**********************************************************
72+
*/
73+
74+
private static CodecRecycler _recycler() {
75+
SoftReference<CodecRecycler> ref = _recycler.get();
76+
CodecRecycler r = (ref == null) ? null : ref.get();
77+
78+
if (r == null) {
79+
r = new CodecRecycler();
80+
_recycler.set(new SoftReference<CodecRecycler>(r));
81+
}
82+
return r;
83+
}
84+
85+
private BinaryDecoder claimDecoder() {
86+
BinaryDecoder d = decoder;
87+
decoder = null;
88+
return d;
89+
}
90+
91+
private BinaryEncoder claimEncoder() {
92+
BinaryEncoder e = encoder;
93+
encoder = null;
94+
return e;
95+
}
96+
}

‎avro/src/main/java/com/fasterxml/jackson/dataformat/avro/deser/AvroParserImpl.java

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
import com.fasterxml.jackson.core.io.IOContext;
1111
import com.fasterxml.jackson.dataformat.avro.AvroParser;
1212
import com.fasterxml.jackson.dataformat.avro.AvroSchema;
13+
import com.fasterxml.jackson.dataformat.avro.CodecRecycler;
1314

1415
/**
1516
* Implementation class that exposes additional internal API
@@ -18,16 +19,16 @@
1819
public class AvroParserImpl extends AvroParser
1920
{
2021
protected final static byte[] NO_BYTES = new byte[0];
21-
22-
protected final BinaryDecoder _decoder;
22+
23+
protected BinaryDecoder _decoder;
2324

2425
protected ByteBuffer _byteBuffer;
2526

2627
public AvroParserImpl(IOContext ctxt, int parserFeatures, int avroFeatures,
2728
ObjectCodec codec, InputStream in)
2829
{
2930
super(ctxt, parserFeatures, avroFeatures, codec, in);
30-
_decoder = AvroSchema.decoder(in, isEnabled(Feature.AVRO_BUFFERING));
31+
_decoder = CodecRecycler.decoder(in, isEnabled(Feature.AVRO_BUFFERING));
3132
}
3233

3334
public AvroParserImpl(IOContext ctxt, int parserFeatures, int avroFeatures,
@@ -36,7 +37,17 @@ public AvroParserImpl(IOContext ctxt, int parserFeatures, int avroFeatures,
3637
{
3738
super(ctxt, parserFeatures, avroFeatures, codec,
3839
data, offset, len);
39-
_decoder = AvroSchema.decoder(data, offset, len);
40+
_decoder = CodecRecycler.decoder(data, offset, len);
41+
}
42+
43+
@Override
44+
protected void _releaseBuffers() throws IOException {
45+
super._releaseBuffers();
46+
BinaryDecoder d = _decoder;
47+
if (d != null) {
48+
_decoder = null;
49+
CodecRecycler.release(d);
50+
}
4051
}
4152

4253
@Override
@@ -51,7 +62,7 @@ public JsonParser overrideFormatFeatures(int values, int mask) {
5162
}
5263
return this;
5364
}
54-
65+
5566
/*
5667
/**********************************************************
5768
/* Abstract method impls

0 commit comments

Comments
 (0)