Skip to content

Write outlier messages through the ByteBuf API instead of staging in a heap array #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
merlimat wants to merge 1 commit into from
Closed

Write outlier messages through the ByteBuf API instead of staging in a heap array #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoBytesField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,18 @@ public void serialize(PrintWriter w) {
w.format("_i += _%sLen;\n", ccName);
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("%s;\n", writeTagToBufExpr(tagName()));
w.format("LightProtoCodec.writeVarInt(_b, _%sLen);\n", ccName);
w.format("if (_%sIdx == -1) {\n", ccName);
// Absolute-indexed copy for the same reason as the array path.
w.format(" %s.getBytes(%s.readerIndex(), _b, _%sLen);\n", ccName, ccName, ccName);
w.format("} else {\n");
w.format(" _parsedBuffer.getBytes(_%sIdx, _b, _%sLen);\n", ccName, ccName);
w.format("}\n");
}


@Override
public void materialize(PrintWriter w) {
Expand Down
15 changes: 15 additions & 0 deletions code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,6 +155,13 @@ public void fieldClear(PrintWriter w, String enclosingType) {

abstract public void serialize(PrintWriter w);

/**
* Emit code that writes this field through the ByteBuf API ({@code _b}): the
* allocation-free path used for messages too large to stage in a scratch array.
* Must produce bytes identical to {@link #serialize(PrintWriter)}.
*/
abstract public void serializeToBuf(PrintWriter w);

abstract public void serializeJson(PrintWriter w);

abstract public void parseJson(PrintWriter w);
Expand Down Expand Up @@ -220,6 +227,14 @@ protected String writeTagExpr(String tag) {
}
}

protected String writeTagToBufExpr(String tag) {
if (field.getNumber() <= 15) {
return String.format("_b.writeByte(%s)", tag);
} else {
return String.format("LightProtoCodec.writeVarInt(_b, %s)", tag);
}
}

protected String tagName() {
return "_" + Util.upperCase(field.getName(), "tag");
}
Expand Down
70 changes: 70 additions & 0 deletions code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMapField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -858,6 +858,36 @@ private void generateValueDataSize(PrintWriter w, String idxVar) {
}
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("for (int _entryIdx = 0; _entryIdx < _%sCount; _entryIdx++) {\n", ccName);

// Compute entry size
w.format(" int _entrySize = 0;\n");

// Key size: 1 (tag) + data size
w.format(" _entrySize += 1;\n"); // key tag is always 1 byte
generateKeyDataSize(w, "_entryIdx");

// Value size: 1 (tag) + data size
w.format(" _entrySize += 1;\n"); // value tag is always 1 byte
generateValueDataSize(w, "_entryIdx");

// Write outer tag + entry size
w.format(" %s;\n", writeTagToBufExpr(tagName()));
w.format(" LightProtoCodec.writeVarInt(_b, _entrySize);\n");

// Write key tag + key data
w.format(" _b.writeByte(%s);\n", keyTagConstant());
generateSerializeKeyDataToBuf(w, "_entryIdx");

// Write value tag + value data
w.format(" _b.writeByte(%s);\n", valueTagConstant());
generateSerializeValueDataToBuf(w, "_entryIdx");

w.format("}\n");
}

private void generateSerializeKeyData(PrintWriter w, String idxVar) {
if (isStringKey()) {
w.format(" LightProtoCodec.StringHolder _ksh = _%sKeys[%s];\n", ccName, idxVar);
Expand Down Expand Up @@ -901,6 +931,46 @@ private void generateSerializeValueData(PrintWriter w, String idxVar) {
}
}

private void generateSerializeKeyDataToBuf(PrintWriter w, String idxVar) {
if (isStringKey()) {
w.format(" LightProtoCodec.StringHolder _ksh = _%sKeys[%s];\n", ccName, idxVar);
w.format(" LightProtoCodec.writeVarInt(_b, _ksh.len);\n");
w.format(" if (_ksh.idx == -1) {\n");
w.format(" LightProtoCodec.writeString(_b, _ksh.s, _ksh.len);\n");
w.format(" } else {\n");
w.format(" _parsedBuffer.getBytes(_ksh.idx, _b, _ksh.len);\n");
w.format(" }\n");
} else {
LightProtoNumberField.serializeNumberToBuf(w, keyField, String.format("_%sKeys[%s]", ccName, idxVar));
}
}

private void generateSerializeValueDataToBuf(PrintWriter w, String idxVar) {
if (isStringValue()) {
w.format(" LightProtoCodec.StringHolder _vsh = _%sValues[%s];\n", ccName, idxVar);
w.format(" LightProtoCodec.writeVarInt(_b, _vsh.len);\n");
w.format(" if (_vsh.idx == -1) {\n");
w.format(" LightProtoCodec.writeString(_b, _vsh.s, _vsh.len);\n");
w.format(" } else {\n");
w.format(" _parsedBuffer.getBytes(_vsh.idx, _b, _vsh.len);\n");
w.format(" }\n");
} else if (isBytesValue()) {
w.format(" LightProtoCodec.BytesHolder _vbh = _%sValues[%s];\n", ccName, idxVar);
w.format(" LightProtoCodec.writeVarInt(_b, _vbh.len);\n");
w.format(" if (_vbh.idx == -1) {\n");
w.format(" _vbh.b.getBytes(_vbh.b.readerIndex(), _b, _vbh.len);\n");
w.format(" } else {\n");
w.format(" _parsedBuffer.getBytes(_vbh.idx, _b, _vbh.len);\n");
w.format(" }\n");
} else if (isMessageValue()) {
w.format(" LightProtoCodec.writeVarInt(_b, _%sValues[%s].getSerializedSize());\n",
ccName, idxVar);
w.format(" _%sValues[%s]._writeTo(_b);\n", ccName, idxVar);
} else {
LightProtoNumberField.serializeNumberToBuf(w, valueField, String.format("_%sValues[%s]", ccName, idxVar));
}
}

@Override
public void serializedSize(PrintWriter w) {
w.format("for (int _i = 0; _i < _%sCount; _i++) {\n", ccName);
Expand Down
46 changes: 42 additions & 4 deletions code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -376,16 +376,23 @@ private void generateSerialize(PrintWriter w) {
w.format(" int _writeIdx = _b.writerIndex();\n");
w.format(" _writeTo(_b.array(), _b.arrayOffset() + _writeIdx);\n");
w.format(" _b.writerIndex(_writeIdx + _serializedSize);\n");
w.format(" } else if (_serializedSize > LightProtoCodec.SCRATCH_RETAIN_MAX) {\n");
// Messages too large for a retained scratch array write through the
// ByteBuf API field by field: slower per element, but allocation-free.
// Staging them would allocate a fresh full-size heap array on every
// write (multi-MB arrays are G1 humongous allocations), invisible to
// any direct-memory accounting sized to the target buffer.
w.format(" _b.ensureWritable(_serializedSize);\n");
w.format(" _writeTo(_b);\n");
w.format(" } else {\n");
// Direct, composite and other buffers: compose in a scratch array cached
// on this (typically pooled) instance and transfer with a single bulk
// write. Plain byte[] stores compile to raw memory accesses on every JDK,
// unlike sun.misc.Unsafe accesses which carry a per-call deprecation
// check since JDK 24.
// check since JDK 24. The dispatch above bounds _serializedSize by
// SCRATCH_RETAIN_MAX, so the scratch is always retainable.
w.format(" byte[] _s = LightProtoCodec.scratchFor(this._scratch, _serializedSize);\n");
w.format(" if (_s.length <= LightProtoCodec.SCRATCH_RETAIN_MAX) {\n");
w.format(" this._scratch = _s;\n");
w.format(" }\n");
w.format(" this._scratch = _s;\n");
w.format(" _writeTo(_s, 0);\n");
w.format(" _b.writeBytes(_s, 0, _serializedSize);\n");
w.format(" }\n");
Expand Down Expand Up @@ -423,6 +430,37 @@ private void generateSerialize(PrintWriter w) {

w.format(" return _i;\n");
w.format(" }\n");

w.println(" /**");
w.println(" * Internal: serialize this message field by field through the ByteBuf");
w.println(" * API. The allocation-free path for messages larger than");
w.println(" * {@code SCRATCH_RETAIN_MAX}; nested messages write through as well, so");
w.println(" * no element of the tree stages in a scratch array. Public only so that");
w.println(" * generated messages in other packages can serialize nested fields of");
w.println(" * this type into the same buffer.");
w.println(" */");
w.format(" public void _writeTo(io.netty.buffer.ByteBuf _b) {\n");
if (hasRequiredFields()) {
w.format(" checkRequiredFields();\n");
}
if (useBitDrivenTraversal()) {
// Same set-bit traversal as the array path: field order — and bytes —
// must match it exactly.
emitBitDrivenTraversal(w, f -> f.serializeToBuf(w));
} else {
for (LightProtoField f : fields) {
String condition = f.serializeCondition();
if (condition != null) {
w.format(" if (%s) {\n", condition);
f.serializeToBuf(w);
w.format(" }\n");
} else {
f.serializeToBuf(w);
}
}
}

w.format(" }\n");
}

private void generateGetSerializedSize(PrintWriter w) {
Expand Down
9 changes: 9 additions & 0 deletions ...-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoMessageField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,15 @@ public void serialize(PrintWriter w) {
w.format("_i = %s._writeTo(_a, _i);\n", ccName);
}

@Override
public void serializeToBuf(PrintWriter w) {
// Nested messages write through as well, so no element of the tree
// stages in a scratch array.
w.format("%s;\n", writeTagToBufExpr(tagName()));
w.format("LightProtoCodec.writeVarInt(_b, %s.getSerializedSize());\n", ccName);
w.format("%s._writeTo(_b);\n", ccName);
}

@Override
public void clear(PrintWriter w) {
w.format("if (%s()){\n", Util.camelCase("has", ccName));
Expand Down
40 changes: 40 additions & 0 deletions code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoNumberField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,40 @@ static void serializeNumber(PrintWriter w, ProtoFieldDescriptor field, String na
}
}

static void serializeNumberToBuf(PrintWriter w, ProtoFieldDescriptor field, String name) {
if (field.isEnumField()) {
w.format(" LightProtoCodec.writeVarInt(_b, %s.getValue());\n", name);
} else if (field.getProtoType().equals("bool")) {
w.format(" _b.writeByte(%s ? 1 : 0);\n", name);
} else if (field.getProtoType().equals("int32")) {
w.format(" LightProtoCodec.writeVarInt(_b, %s);\n", name);
} else if (field.getProtoType().equals("uint32")) {
w.format(" LightProtoCodec.writeVarInt(_b, %s);\n", name);
} else if (field.getProtoType().equals("sint32")) {
w.format(" LightProtoCodec.writeSignedVarInt(_b, %s);\n", name);
} else if (field.getProtoType().equals("sint64")) {
w.format(" LightProtoCodec.writeSignedVarInt64(_b, %s);\n", name);
} else if (field.getProtoType().equals("int64")) {
w.format(" LightProtoCodec.writeVarInt64(_b, %s);\n", name);
} else if (field.getProtoType().equals("uint64")) {
w.format(" LightProtoCodec.writeVarInt64(_b, %s);\n", name);
} else if (field.getProtoType().equals("fixed32")) {
w.format(" LightProtoCodec.writeFixedInt32(_b, %s);\n", name);
} else if (field.getProtoType().equals("fixed64")) {
w.format(" LightProtoCodec.writeFixedInt64(_b, %s);\n", name);
} else if (field.getProtoType().equals("sfixed32")) {
w.format(" LightProtoCodec.writeFixedInt32(_b, %s);\n", name);
} else if (field.getProtoType().equals("sfixed64")) {
w.format(" LightProtoCodec.writeFixedInt64(_b, %s);\n", name);
} else if (field.getProtoType().equals("double")) {
w.format(" LightProtoCodec.writeDouble(_b, %s);\n", name);
} else if (field.getProtoType().equals("float")) {
w.format(" LightProtoCodec.writeFloat(_b, %s);\n", name);
} else {
throw new IllegalArgumentException("Failed to write serializer for field: " + field.getProtoType());
}
}

static String parseNumber(ProtoFieldDescriptor field) {
if (field.isEnumField()) {
return String.format("%s.valueOf(LightProtoCodec.readVarInt(_buffer))", field.getJavaType());
Expand Down Expand Up @@ -193,6 +227,12 @@ public void serialize(PrintWriter w) {
serializeNumber(w, field, ccName);
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("%s;\n", writeTagToBufExpr(tagName()));
serializeNumberToBuf(w, field, ccName);
}

@Override
public void serializeJson(PrintWriter w) {
String type = field.getProtoType();
Expand Down
14 changes: 14 additions & 0 deletions ...ator/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedBytesField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,20 @@ public void serialize(PrintWriter w) {
w.format("}\n");
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName);
w.format(" LightProtoCodec.BytesHolder _bh = %s[i];\n", pluralName);
w.format(" %s;\n", writeTagToBufExpr(tagName()));
w.format(" LightProtoCodec.writeVarInt(_b, _bh.len);\n");
w.format(" if (_bh.idx == -1) {\n");
w.format(" _bh.b.getBytes(_bh.b.readerIndex(), _b, _bh.len);\n");
w.format(" } else {\n");
w.format(" _parsedBuffer.getBytes(_bh.idx, _b, _bh.len);\n");
w.format(" }\n");
w.format("}\n");
}

@Override
public void serializeJson(PrintWriter w) {
w.format("_b.writeByte('[');\n");
Expand Down
10 changes: 10 additions & 0 deletions ...or/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedMessageField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,16 @@ public void serialize(PrintWriter w) {
w.format("}\n");
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName);
w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName);
w.format(" %s;\n", writeTagToBufExpr(tagName()));
w.format(" LightProtoCodec.writeVarInt(_b, _item.getSerializedSize());\n");
w.format(" _item._writeTo(_b);\n");
w.format("}\n");
}

@Override
public void serializeJson(PrintWriter w) {
w.format("_b.writeByte('[');\n");
Expand Down
28 changes: 28 additions & 0 deletions ...tor/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedNumberField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,34 @@ public void serialize(PrintWriter w) {
}
}

@Override
public void serializeToBuf(PrintWriter w) {
int fixedSize = LightProtoNumberField.fixedDataSize(field);
if (field.isPacked()) {
w.format(" %s;\n", writeTagToBufExpr(tagName() + "_PACKED"));
if (fixedSize >= 0) {
w.format(" LightProtoCodec.writeVarInt(_b, _%sCount * %d);\n", pluralName, fixedSize);
} else {
w.format(" int _%sSize = 0;\n", pluralName);
w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName);
w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName);
w.format(" _%sSize += %s;\n", pluralName, LightProtoNumberField.serializedSizeOfNumber(field, "_item"));
w.format("}\n");
w.format(" LightProtoCodec.writeVarInt(_b, _%sSize);\n", pluralName);
}
w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName);
w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName);
LightProtoNumberField.serializeNumberToBuf(w, field, "_item");
w.format("}\n");
} else {
w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName);
w.format(" %s _item = %s[i];\n", field.getJavaType(), pluralName);
w.format(" %s;\n", writeTagToBufExpr(tagName()));
LightProtoNumberField.serializeNumberToBuf(w, field, "_item");
w.format("}\n");
}
}

@Override
public void serializeJson(PrintWriter w) {
w.format("_b.writeByte('[');\n");
Expand Down
14 changes: 14 additions & 0 deletions ...tor/src/main/java/io/streamnative/lightproto/generator/LightProtoRepeatedStringField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,20 @@ public void serialize(PrintWriter w) {
w.format("}\n");
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("for (int i = 0; i < _%sCount; i++) {\n", pluralName);
w.format(" LightProtoCodec.StringHolder _sh = %s[i];\n", pluralName);
w.format(" %s;\n", writeTagToBufExpr(tagName()));
w.format(" LightProtoCodec.writeVarInt(_b, _sh.len);\n");
w.format(" if (_sh.idx == -1) {\n");
w.format(" LightProtoCodec.writeString(_b, _sh.s, _sh.len);\n");
w.format(" } else {\n");
w.format(" _parsedBuffer.getBytes(_sh.idx, _b, _sh.len);\n");
w.format(" }\n");
w.format("}\n");
}

@Override
public void serializeJson(PrintWriter w) {
w.format("_b.writeByte('[');\n");
Expand Down
11 changes: 11 additions & 0 deletions code-generator/src/main/java/io/streamnative/lightproto/generator/LightProtoStringField.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,17 @@ public void serialize(PrintWriter w) {
w.format("}\n");
}

@Override
public void serializeToBuf(PrintWriter w) {
w.format("%s;\n", writeTagToBufExpr(tagName()));
w.format("LightProtoCodec.writeVarInt(_b, _%sBufferLen);\n", ccName);
w.format("if (_%sBufferIdx == -1) {\n", ccName);
w.format(" LightProtoCodec.writeString(_b, %s, _%sBufferLen);\n", ccName, ccName);
w.format("} else {\n");
w.format(" _parsedBuffer.getBytes(_%sBufferIdx, _b, _%sBufferLen);\n", ccName, ccName);
w.format("}\n");
}

@Override
public void serializeJson(PrintWriter w) {
w.format("LightProtoCodec.writeJsonString(_b, %s());\n", Util.camelCase("get", field.getName()));
Expand Down
Loading
Loading