You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@bookkeeper.apache.org by ch...@apache.org on 2023/02/27 08:00:20 UTC
[bookkeeper] branch master updated: Avoid thread-local state when computing CRCs (#3811)
This is an automated email from the ASF dual-hosted git repository.
chenhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/bookkeeper.git
The following commit(s) were added to refs/heads/master by this push:
new 1e02853485 Avoid thread-local state when computing CRCs (#3811)
1e02853485 is described below
commit 1e028534855a4dc83a74432f9de9a062c94c9984
Author: Matteo Merli <mm...@apache.org>
AuthorDate: Mon Feb 27 00:00:13 2023 -0800
Avoid thread-local state when computing CRCs (#3811)
### Motivation
In `DigestManager` there are several accesses to ThreadLocal variable per each entry processed.
The reason is the mainly due to `DigestManager` API which exposes a stateful `update()` method which can be invoked multiple times and keeps the current checksum as a thread-local variable.
If we exclude MAC digest which is 20 bytes, for other digests we can instead keep the current checksum in a local variable and pass it each time, avoiding all the thread-locals and also the need for writing the checksum result into a buffer.
### Benchmarks
#### Before #3810
```
Benchmark (entrySize) Mode Cnt Score Error Units
DigestManagerBenchmark.verifyDigest 64 thrpt 3 13.450 ± 3.634 ops/us
DigestManagerBenchmark.verifyDigest 1024 thrpt 3 7.908 ± 2.637 ops/us
DigestManagerBenchmark.verifyDigest 4086 thrpt 3 3.233 ± 0.882 ops/us
DigestManagerBenchmark.verifyDigest 8192 thrpt 3 1.846 ± 0.047 ops/us
```
#### After #3810
```
Benchmark (entrySize) Mode Cnt Score Error Units
DigestManagerBenchmark.verifyDigest 64 thrpt 3 46.312 ± 7.414 ops/us
DigestManagerBenchmark.verifyDigest 1024 thrpt 3 13.379 ± 1.069 ops/us
DigestManagerBenchmark.verifyDigest 4086 thrpt 3 3.787 ± 0.059 ops/us
DigestManagerBenchmark.verifyDigest 8192 thrpt 3 1.956 ± 0.052 ops/us
```
#### After this change
```
Benchmark (entrySize) Mode Cnt Score Error Units
DigestManagerBenchmark.verifyDigest 64 thrpt 3 130.108 ± 4.854 ops/us
DigestManagerBenchmark.verifyDigest 1024 thrpt 3 17.744 ± 0.238 ops/us
DigestManagerBenchmark.verifyDigest 4086 thrpt 3 4.104 ± 0.181 ops/us
DigestManagerBenchmark.verifyDigest 8192 thrpt 3 2.050 ± 0.066 ops/us
```
---
.../proto/checksum/CRC32CDigestManager.java | 26 +++-----
.../proto/checksum/CRC32DigestManager.java | 11 ++-
.../bookkeeper/proto/checksum/DigestManager.java | 78 +++++++++++++---------
.../proto/checksum/DummyDigestManager.java | 11 ++-
.../proto/checksum/MacDigestManager.java | 10 ++-
.../proto/checksum/DigestTypeBenchmark.java | 4 +-
6 files changed, 84 insertions(+), 56 deletions(-)
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32CDigestManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32CDigestManager.java
index 446e0b35ab..1fb3d47b3e 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32CDigestManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32CDigestManager.java
@@ -21,20 +21,11 @@ package org.apache.bookkeeper.proto.checksum;
import com.scurrilous.circe.checksum.Crc32cIntChecksum;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
-import io.netty.util.concurrent.FastThreadLocal;
import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.mutable.MutableInt;
@Slf4j
class CRC32CDigestManager extends DigestManager {
- private static final FastThreadLocal<MutableInt> currentCrc = new FastThreadLocal<MutableInt>() {
- @Override
- protected MutableInt initialValue() throws Exception {
- return new MutableInt(0);
- }
- };
-
public CRC32CDigestManager(long ledgerId, boolean useV2Protocol, ByteBufAllocator allocator) {
super(ledgerId, useV2Protocol, allocator);
}
@@ -45,16 +36,17 @@ class CRC32CDigestManager extends DigestManager {
}
@Override
- void populateValueAndReset(ByteBuf buf) {
- MutableInt current = currentCrc.get();
- buf.writeInt(current.intValue());
- current.setValue(0);
+ boolean isInt32Digest() {
+ return true;
+ }
+
+ @Override
+ void populateValueAndReset(int digest, ByteBuf buf) {
+ buf.writeInt(digest);
}
@Override
- void update(ByteBuf data, int offset, int len) {
- MutableInt current = currentCrc.get();
- final int lastCrc = current.intValue();
- current.setValue(Crc32cIntChecksum.resumeChecksum(lastCrc, data, offset, len));
+ int update(int digest, ByteBuf data, int offset, int len) {
+ return Crc32cIntChecksum.resumeChecksum(digest, data, offset, len);
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32DigestManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32DigestManager.java
index ea34f13069..21be2651a7 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32DigestManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/CRC32DigestManager.java
@@ -57,12 +57,19 @@ class CRC32DigestManager extends DigestManager {
}
@Override
- void populateValueAndReset(ByteBuf buf) {
+ void populateValueAndReset(int digest, ByteBuf buf) {
buf.writeLong(crc.get().getValueAndReset());
}
@Override
- void update(ByteBuf data, int offset, int len) {
+ int update(int digest, ByteBuf data, int offset, int len) {
crc.get().update(data, offset, len);
+ return 0;
+ }
+
+ @Override
+ boolean isInt32Digest() {
+ // This is stored as 8 bytes
+ return false;
}
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
index b142448aa4..87fd58eebf 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DigestManager.java
@@ -53,13 +53,11 @@ public abstract class DigestManager {
abstract int getMacCodeLength();
- void update(byte[] data) {
- update(Unpooled.wrappedBuffer(data), 0, data.length);
- }
+ abstract int update(int digest, ByteBuf buffer, int offset, int len);
- abstract void update(ByteBuf buffer, int offset, int len);
+ abstract void populateValueAndReset(int digest, ByteBuf buffer);
- abstract void populateValueAndReset(ByteBuf buffer);
+ abstract boolean isInt32Digest();
final int macCodeLength;
@@ -112,7 +110,7 @@ public abstract class DigestManager {
headersBuffer.writeLong(lastAddConfirmed);
headersBuffer.writeLong(length);
- update(headersBuffer, 0, METADATA_LENGTH);
+ int digest = update(0, headersBuffer, 0, METADATA_LENGTH);
// don't unwrap slices
final ByteBuf unwrapped = data.unwrap() != null && data.unwrap() instanceof CompositeByteBuf
@@ -121,11 +119,15 @@ public abstract class DigestManager {
ReferenceCountUtil.safeRelease(data);
if (unwrapped instanceof CompositeByteBuf) {
- ((CompositeByteBuf) unwrapped).forEach(b -> update(b, b.readerIndex(), b.readableBytes()));
+ CompositeByteBuf cbb = ((CompositeByteBuf) unwrapped);
+ for (int i = 0; i < cbb.numComponents(); i++) {
+ ByteBuf b = cbb.component(i);
+ digest = update(digest, b, b.readerIndex(), b.readableBytes());
+ }
} else {
- update(unwrapped, unwrapped.readerIndex(), unwrapped.readableBytes());
+ digest = update(digest, unwrapped, unwrapped.readerIndex(), unwrapped.readableBytes());
}
- populateValueAndReset(headersBuffer);
+ populateValueAndReset(digest, headersBuffer);
return ByteBufList.get(headersBuffer, unwrapped);
}
@@ -147,8 +149,8 @@ public abstract class DigestManager {
headersBuffer.writeLong(ledgerId);
headersBuffer.writeLong(lac);
- update(headersBuffer, 0, LAC_METADATA_LENGTH);
- populateValueAndReset(headersBuffer);
+ int digest = update(0, headersBuffer, 0, LAC_METADATA_LENGTH);
+ populateValueAndReset(digest, headersBuffer);
return ByteBufList.get(headersBuffer);
}
@@ -183,18 +185,26 @@ public abstract class DigestManager {
this.getClass().getName(), dataReceived.readableBytes());
throw new BKDigestMatchException();
}
- update(dataReceived, 0, METADATA_LENGTH);
+ int digest = update(0, dataReceived, 0, METADATA_LENGTH);
int offset = METADATA_LENGTH + macCodeLength;
- update(dataReceived, offset, dataReceived.readableBytes() - offset);
-
- ByteBuf digest = DIGEST_BUFFER.get();
- digest.clear();
- populateValueAndReset(digest);
-
- if (!ByteBufUtil.equals(digest, 0, dataReceived, METADATA_LENGTH, macCodeLength)) {
- logger.error("Mac mismatch for ledger-id: " + ledgerId + ", entry-id: " + entryId);
- throw new BKDigestMatchException();
+ digest = update(digest, dataReceived, offset, dataReceived.readableBytes() - offset);
+
+ if (isInt32Digest()) {
+ int receivedDigest = dataReceived.getInt(METADATA_LENGTH);
+ if (receivedDigest != digest) {
+ logger.error("Digest mismatch for ledger-id: " + ledgerId + ", entry-id: " + entryId);
+ throw new BKDigestMatchException();
+ }
+ } else {
+ ByteBuf digestBuf = DIGEST_BUFFER.get();
+ digestBuf.clear();
+ populateValueAndReset(digest, digestBuf);
+
+ if (!ByteBufUtil.equals(digestBuf, 0, dataReceived, METADATA_LENGTH, macCodeLength)) {
+ logger.error("Mac mismatch for ledger-id: " + ledgerId + ", entry-id: " + entryId);
+ throw new BKDigestMatchException();
+ }
}
long actualLedgerId = dataReceived.readLong();
@@ -223,17 +233,25 @@ public abstract class DigestManager {
throw new BKDigestMatchException();
}
- update(dataReceived, 0, LAC_METADATA_LENGTH);
+ int digest = update(0, dataReceived, 0, LAC_METADATA_LENGTH);
- ByteBuf digest = DIGEST_BUFFER.get();
- digest.clear();
-
- populateValueAndReset(digest);
-
- if (!ByteBufUtil.equals(digest, 0, dataReceived, LAC_METADATA_LENGTH, macCodeLength)) {
- logger.error("Mac mismatch for ledger-id LAC: " + ledgerId);
- throw new BKDigestMatchException();
+ if (isInt32Digest()) {
+ int receivedDigest = dataReceived.getInt(LAC_METADATA_LENGTH);
+ if (receivedDigest != digest) {
+ logger.error("Digest mismatch for ledger-id LAC: " + ledgerId);
+ throw new BKDigestMatchException();
+ }
+ } else {
+ ByteBuf digestBuf = DIGEST_BUFFER.get();
+ digestBuf.clear();
+ populateValueAndReset(digest, digestBuf);
+
+ if (!ByteBufUtil.equals(digestBuf, 0, dataReceived, LAC_METADATA_LENGTH, macCodeLength)) {
+ logger.error("Mac mismatch for ledger-id LAC: " + ledgerId);
+ throw new BKDigestMatchException();
+ }
}
+
long actualLedgerId = dataReceived.readLong();
long lac = dataReceived.readLong();
if (actualLedgerId != ledgerId) {
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DummyDigestManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DummyDigestManager.java
index 9414c3ee55..b15499f0cc 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DummyDigestManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/DummyDigestManager.java
@@ -38,8 +38,15 @@ public class DummyDigestManager extends DigestManager {
}
@Override
- void update(ByteBuf buffer, int offset, int len) {}
+ int update(int digest, ByteBuf buffer, int offset, int len) {
+ return 0;
+ }
@Override
- void populateValueAndReset(ByteBuf buffer) {}
+ void populateValueAndReset(int digest, ByteBuf buffer) {}
+
+ @Override
+ boolean isInt32Digest() {
+ return false;
+ }
}
diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/MacDigestManager.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/MacDigestManager.java
index 515e0f2484..c04c411c6c 100644
--- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/MacDigestManager.java
+++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/checksum/MacDigestManager.java
@@ -91,14 +91,18 @@ public class MacDigestManager extends DigestManager {
@Override
- void populateValueAndReset(ByteBuf buffer) {
+ void populateValueAndReset(int digest, ByteBuf buffer) {
buffer.writeBytes(mac.get().doFinal());
}
@Override
- void update(ByteBuf data, int offset, int len) {
+ int update(int digest, ByteBuf data, int offset, int len) {
mac.get().update(data.slice(offset, len).nioBuffer());
+ return 0;
}
-
+ @Override
+ boolean isInt32Digest() {
+ return false;
+ }
}
diff --git a/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java b/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
index 59bb6708bf..7201aba8cf 100644
--- a/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
+++ b/microbenchmarks/src/main/java/org/apache/bookkeeper/proto/checksum/DigestTypeBenchmark.java
@@ -172,9 +172,9 @@ public class DigestTypeBenchmark {
public void digestManager(MyState state) {
final ByteBuf buff = state.getByteBuff(state.bufferType);
final DigestManager dm = state.getDigestManager(state.digest);
- dm.update(buff, 0, buff.readableBytes());
+ int digest = dm.update(0, buff, 0, buff.readableBytes());
state.digestBuf.clear();
- dm.populateValueAndReset(state.digestBuf);
+ dm.populateValueAndReset(digest, state.digestBuf);
}
}