Skip to content

Commit 46c99a0

Browse files
bladehan1bladehan1
committed
feat(metrics): attribute business read waits
1 parent c2cd699 commit 46c99a0

5 files changed

Lines changed: 194 additions & 13 deletions

File tree

‎chainbase/src/main/java/org/tron/common/storage/rocksdb/RocksDbDataSourceImpl.java‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import org.tron.core.db.common.iterator.RockStoreIterator;
5151
import org.tron.core.db2.common.Instance;
5252
import org.tron.core.db2.common.WrappedByteArray;
53+
import org.tron.core.db2.core.ExecutionAttribution;
5354
import org.tron.core.exception.TronError;
5455

5556

@@ -348,11 +349,18 @@ public void putData(byte[] key, byte[] value) {
348349

349350
@Override
350351
public byte[] getData(byte[] key) {
352+
long lockStarted = ExecutionAttribution.sample(dataBaseName, "database_lock");
351353
resetDbLock.readLock().lock();
352354
try {
355+
ExecutionAttribution.sampled(dataBaseName, "database_lock", lockStarted, 0, false);
353356
throwIfNotAlive();
354357
checkArgNotNull(key, "key");
355-
return database.get(key);
358+
long getStarted = ExecutionAttribution.sample(dataBaseName, "database_get");
359+
try {
360+
return database.get(key);
361+
} finally {
362+
ExecutionAttribution.sampled(dataBaseName, "database_get", getStarted, 0, false);
363+
}
356364
} catch (RocksDBException e) {
357365
throw new RuntimeException(dataBaseName, e);
358366
} finally {

‎chainbase/src/main/java/org/tron/core/db/BandwidthProcessor.java‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@
1919
import org.tron.core.capsule.AccountCapsule;
2020
import org.tron.core.capsule.AssetIssueCapsule;
2121
import org.tron.core.capsule.TransactionCapsule;
22+
import org.tron.core.db2.core.ExecutionAttribution;
23+
import org.tron.core.db2.core.ExecutionAttribution.ReadScope;
24+
import org.tron.core.db2.core.ExecutionAttribution.ReadSite;
2225
import org.tron.core.exception.AccountResourceInsufficientException;
2326
import org.tron.core.exception.ContractValidateException;
2427
import org.tron.core.exception.TooBigTransactionException;
@@ -130,7 +133,10 @@ public void consume(TransactionCapsule trx, TransactionTrace trace)
130133
logger.debug("TxId {}, bandwidth cost: {}.", trx.getTransactionId(), bytesSize);
131134
trace.setNetBill(bytesSize, 0);
132135
byte[] address = TransactionCapsule.getOwner(contract);
133-
AccountCapsule accountCapsule = chainBaseManager.getAccountStore().get(address);
136+
AccountCapsule accountCapsule;
137+
try (ReadScope ignored = ExecutionAttribution.readSite(ReadSite.BANDWIDTH_OWNER)) {
138+
accountCapsule = chainBaseManager.getAccountStore().get(address);
139+
}
134140
if (accountCapsule == null) {
135141
throw new ContractValidateException(String.format("account [%s] does not exist",
136142
StringUtil.encode58Check(address)));
@@ -268,8 +274,10 @@ public boolean contractCreateNewAccount(Contract contract) {
268274
} catch (Exception ex) {
269275
throw new RuntimeException(ex.getMessage());
270276
}
271-
toAccount =
272-
chainBaseManager.getAccountStore().get(transferContract.getToAddress().toByteArray());
277+
try (ReadScope ignored = ExecutionAttribution.readSite(ReadSite.BANDWIDTH_RECEIVER)) {
278+
toAccount =
279+
chainBaseManager.getAccountStore().get(transferContract.getToAddress().toByteArray());
280+
}
273281
return toAccount == null;
274282
case TransferAssetContract:
275283
TransferAssetContract transferAssetContract;
@@ -278,8 +286,10 @@ public boolean contractCreateNewAccount(Contract contract) {
278286
} catch (Exception ex) {
279287
throw new RuntimeException(ex.getMessage());
280288
}
281-
toAccount = chainBaseManager.getAccountStore()
282-
.get(transferAssetContract.getToAddress().toByteArray());
289+
try (ReadScope ignored = ExecutionAttribution.readSite(ReadSite.BANDWIDTH_RECEIVER)) {
290+
toAccount = chainBaseManager.getAccountStore()
291+
.get(transferAssetContract.getToAddress().toByteArray());
292+
}
283293
return toAccount == null;
284294
default:
285295
return false;

‎chainbase/src/main/java/org/tron/core/db2/core/ExecutionAttribution.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ public final class ExecutionAttribution implements AutoCloseable {
1717
private final Map<String, long[]> stages = new LinkedHashMap<>();
1818
private final Map<String, long[]> reads = new LinkedHashMap<>();
1919
private final Map<Phase, Map<String, long[]>> phaseReads = new EnumMap<>(Phase.class);
20+
private final Map<String, long[]> siteReads = new LinkedHashMap<>();
21+
private ReadSite site = ReadSite.OTHER;
2022
private Phase phase = Phase.OTHER;
2123
private boolean published;
2224

@@ -26,6 +28,38 @@ public enum Phase {
2628
private final String label = name().toLowerCase(Locale.ROOT);
2729
}
2830

31+
public enum ReadSite {
32+
OTHER, BANDWIDTH_OWNER, BANDWIDTH_RECEIVER, BALANCE_HISTORY;
33+
34+
private final String label = name().toLowerCase(Locale.ROOT);
35+
}
36+
37+
/** Narrow caller scope; restores the enclosing site even when a read throws. */
38+
public static ReadScope readSite(ReadSite site) {
39+
ExecutionAttribution current = CURRENT.get();
40+
return current == null ? null : new ReadScope(current, site);
41+
}
42+
43+
public static final class ReadScope implements AutoCloseable {
44+
private final ExecutionAttribution owner;
45+
private final ReadSite previous;
46+
47+
private ReadScope(ExecutionAttribution owner, ReadSite site) {
48+
this.owner = owner;
49+
this.previous = owner.site;
50+
owner.site = site;
51+
}
52+
53+
@Override
54+
public void close() {
55+
owner.site = previous;
56+
}
57+
}
58+
59+
private String siteKey(String key) {
60+
return phase.label + ":" + site.label + ":" + key;
61+
}
62+
2963
public static long start(Phase phase) {
3064
ExecutionAttribution current = CURRENT.get();
3165
if (current == null) {
@@ -82,6 +116,7 @@ public static long sample(String store, String operation) {
82116
long[] row = current.reads.computeIfAbsent(key, ignored -> new long[5]);
83117
current.phaseReads.computeIfAbsent(current.phase, ignored -> new LinkedHashMap<>())
84118
.computeIfAbsent(key, ignored -> new long[5])[0]++;
119+
current.siteReads.computeIfAbsent(current.siteKey(key), ignored -> new long[5])[0]++;
85120
return (row[0]++ & 63) == 0 ? System.nanoTime() : 0;
86121
}
87122

@@ -95,6 +130,7 @@ public static void sampled(String store, String operation, long start, int layer
95130
String key = store + ":" + operation;
96131
long elapsed = System.nanoTime() - start;
97132
finish(current.reads.get(key), elapsed, layers, hit);
133+
finish(current.siteReads.get(current.siteKey(key)), elapsed, layers, hit);
98134
finish(current.phaseReads.get(current.phase).get(key), elapsed, layers, hit);
99135
}
100136
}
@@ -137,6 +173,19 @@ public void publish() {
137173
}
138174
}
139175
});
176+
siteReads.forEach((name, row) -> {
177+
String[] parts = name.split(":");
178+
logger.info("Chainbase execution site read: head={}, blockHash={}, status=executed, "
179+
+ "phase={}, site={}, store={}, operation={}, calls={}, samples={}, nanos={}, "
180+
+ "layers={}, hits={}", head, hash, parts[0], parts[1], parts[2], parts[3],
181+
row[0], row[1], row[2], row[3], row[4]);
182+
if (Metrics.enabled()) {
183+
String[] kinds = {"calls", "samples", "nanos", "layers", "hits"};
184+
for (int i = 0; i < kinds.length; i++) {
185+
Export.SITE_READS.labels(parts[0], parts[1], parts[2], parts[3], kinds[i]).inc(row[i]);
186+
}
187+
}
188+
});
140189
phaseReads.forEach((phase, operations) -> operations.forEach((name, row) -> {
141190
String[] parts = name.split(":");
142191
logger.info("Chainbase execution phase read: head={}, blockHash={}, status=executed, "
@@ -182,6 +231,10 @@ private static final class Export {
182231
.name("tron_chainbase_execution_phase_read_total")
183232
.help("Partition of loop read counters by caller phase; sampled time is not total IO.")
184233
.labelNames("phase", "store", "operation", "kind").register();
234+
private static final Counter SITE_READS = Counter.build()
235+
.name("tron_chainbase_execution_site_read_total")
236+
.help("Caller site partition of loop reads; sampled time is not total IO.")
237+
.labelNames("phase", "site", "store", "operation", "kind").register();
185238
private static final Counter READS = Counter.build()
186239
.name("tron_chainbase_execution_read_total")
187240
.help("Loop-thread reads; sampled nanoseconds/layers/hits, not physical disk IO.")

‎chainbase/src/main/java/org/tron/core/store/AccountStore.java‎

Lines changed: 15 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -2,25 +2,27 @@
22

33
import com.google.protobuf.ByteString;
44
import com.typesafe.config.ConfigObject;
5+
import java.util.HashMap;
6+
import java.util.List;
7+
import java.util.Map;
8+
import java.util.OptionalLong;
59
import org.apache.commons.lang3.ArrayUtils;
610
import org.springframework.beans.factory.annotation.Autowired;
711
import org.springframework.beans.factory.annotation.Value;
812
import org.springframework.stereotype.Component;
913
import org.tron.common.parameter.CommonParameter;
1014
import org.tron.common.utils.Commons;
11-
import org.tron.core.db2.core.ExecutionAttribution;
1215
import org.tron.core.capsule.AccountCapsule;
1316
import org.tron.core.capsule.BlockCapsule;
1417
import org.tron.core.db.TronStoreWithRevoking;
1518
import org.tron.core.db.accountstate.AccountStateCallBackUtils;
19+
import org.tron.core.db2.core.ExecutionAttribution;
20+
import org.tron.core.db2.core.ExecutionAttribution.ReadScope;
21+
import org.tron.core.db2.core.ExecutionAttribution.ReadSite;
1622
import org.tron.core.exception.TronError;
1723
import org.tron.protos.contract.BalanceContract.TransactionBalanceTrace;
1824
import org.tron.protos.contract.BalanceContract.TransactionBalanceTrace.Operation;
1925

20-
import java.util.HashMap;
21-
import java.util.List;
22-
import java.util.Map;
23-
import java.util.OptionalLong;
2426

2527
@Component
2628
public class AccountStore extends TronStoreWithRevoking<AccountCapsule> {
@@ -71,7 +73,10 @@ public AccountCapsule get(byte[] key) {
7173
@Override
7274
public void put(byte[] key, AccountCapsule item) {
7375
if (CommonParameter.getInstance().isHistoryBalanceLookup()) {
74-
AccountCapsule old = super.getUnchecked(key);
76+
AccountCapsule old;
77+
try (ReadScope ignored = ExecutionAttribution.readSite(ReadSite.BALANCE_HISTORY)) {
78+
old = super.getUnchecked(key);
79+
}
7580
if (old == null) {
7681
if (item.getBalance() != 0) {
7782
recordBalance(item, item.getBalance());
@@ -95,7 +100,10 @@ public void put(byte[] key, AccountCapsule item) {
95100
@Override
96101
public void delete(byte[] key) {
97102
if (CommonParameter.getInstance().isHistoryBalanceLookup()) {
98-
AccountCapsule old = super.getUnchecked(key);
103+
AccountCapsule old;
104+
try (ReadScope ignored = ExecutionAttribution.readSite(ReadSite.BALANCE_HISTORY)) {
105+
old = super.getUnchecked(key);
106+
}
99107
if (old != null) {
100108
recordBalance(old, -old.getBalance());
101109
}

‎framework/src/test/java/org/tron/core/db2/core/ExecutionAttributionTest.java‎

Lines changed: 102 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,23 @@
88
import static org.mockito.Mockito.when;
99

1010
import io.prometheus.client.CollectorRegistry;
11+
import java.lang.reflect.Field;
12+
import java.util.concurrent.FutureTask;
13+
import java.util.concurrent.TimeUnit;
14+
import java.util.concurrent.locks.ReentrantReadWriteLock;
1115
import org.junit.After;
1216
import org.junit.Before;
17+
import org.junit.Rule;
1318
import org.junit.Test;
19+
import org.junit.rules.TemporaryFolder;
1420
import org.tron.common.parameter.CommonParameter;
21+
import org.tron.common.storage.rocksdb.RocksDbDataSourceImpl;
1522
import org.tron.core.db2.common.DB;
1623

1724
public class ExecutionAttributionTest {
25+
@Rule
26+
public TemporaryFolder temporaryFolder = new TemporaryFolder();
27+
1828
private String previous;
1929
private boolean metrics;
2030
private org.tron.core.config.args.Storage storage;
@@ -40,6 +50,98 @@ public void cleanup() {
4050
CommonParameter.getInstance().setMetricsPrometheusEnable(metrics);
4151
}
4252

53+
@Test
54+
public void databaseTimingPreservesLockingValuesAndFailureRelease() throws Exception {
55+
RocksDbDataSourceImpl db = new RocksDbDataSourceImpl(
56+
temporaryFolder.newFolder().toString(), "account");
57+
byte[] key = {1};
58+
byte[] value = {2};
59+
db.putData(key, value);
60+
Field field = RocksDbDataSourceImpl.class.getDeclaredField("resetDbLock");
61+
field.setAccessible(true);
62+
ReentrantReadWriteLock lock = (ReentrantReadWriteLock) field.get(db);
63+
double lockCalls = read("database_lock", "calls");
64+
double getCalls = read("database_get", "calls");
65+
FutureTask<byte[]> task = new FutureTask<>(() -> {
66+
try (ExecutionAttribution attempt = ExecutionAttribution.open(4, "locked")) {
67+
byte[] result = db.getData(key);
68+
attempt.publish();
69+
return result;
70+
}
71+
});
72+
Thread reader = new Thread(task);
73+
try {
74+
lock.writeLock().lock();
75+
try {
76+
reader.start();
77+
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
78+
while (!lock.hasQueuedThread(reader) && System.nanoTime() < deadline) {
79+
Thread.yield();
80+
}
81+
assertTrue("reader must wait for reset lock", lock.hasQueuedThread(reader));
82+
assertEquals(getCalls, read("database_get", "calls"), 0);
83+
} finally {
84+
lock.writeLock().unlock();
85+
}
86+
assertArrayEquals(value, task.get(5, TimeUnit.SECONDS));
87+
assertEquals(lockCalls + 1, read("database_lock", "calls"), 0);
88+
assertEquals(getCalls + 1, read("database_get", "calls"), 0);
89+
try (ExecutionAttribution attempt = ExecutionAttribution.open(5, "invalid")) {
90+
org.junit.Assert.assertThrows(IllegalArgumentException.class, () -> db.getData(null));
91+
assertEquals(0, lock.getReadLockCount());
92+
assertNull(db.getData(new byte[] {3}));
93+
assertArrayEquals(value, db.getData(key));
94+
attempt.publish();
95+
}
96+
assertEquals(lockCalls + 4, read("database_lock", "calls"), 0);
97+
assertEquals(getCalls + 3, read("database_get", "calls"), 0);
98+
} finally {
99+
reader.join(5000);
100+
db.closeDB();
101+
}
102+
}
103+
104+
@Test
105+
public void callerScopesPartitionReadsAndRestoreAfterExceptions() {
106+
double owner = siteCalls("bandwidth_owner");
107+
double history = siteCalls("balance_history");
108+
double other = siteCalls("other");
109+
double total = read("snapshot", "calls");
110+
try (ExecutionAttribution attempt = ExecutionAttribution.open(3, "sites")) {
111+
try (ExecutionAttribution.ReadScope ignored = ExecutionAttribution.readSite(
112+
ExecutionAttribution.ReadSite.BANDWIDTH_OWNER)) {
113+
sampledAccountRead();
114+
try (ExecutionAttribution.ReadScope nested = ExecutionAttribution.readSite(
115+
ExecutionAttribution.ReadSite.BALANCE_HISTORY)) {
116+
sampledAccountRead();
117+
throw new IllegalStateException("read failed");
118+
} catch (IllegalStateException expected) {
119+
sampledAccountRead();
120+
}
121+
}
122+
sampledAccountRead();
123+
attempt.publish();
124+
}
125+
assertEquals(owner + 2, siteCalls("bandwidth_owner"), 0);
126+
assertEquals(history + 1, siteCalls("balance_history"), 0);
127+
assertEquals(other + 1, siteCalls("other"), 0);
128+
assertEquals(total + 4, read("snapshot", "calls"), 0);
129+
assertNull(ExecutionAttribution.readSite(ExecutionAttribution.ReadSite.BANDWIDTH_OWNER));
130+
}
131+
132+
private static void sampledAccountRead() {
133+
long started = ExecutionAttribution.sample("account", "snapshot");
134+
ExecutionAttribution.sampled("account", "snapshot", started, 0, false);
135+
}
136+
137+
private static double siteCalls(String site) {
138+
Double value = CollectorRegistry.defaultRegistry.getSampleValue(
139+
"tron_chainbase_execution_site_read_total",
140+
new String[] {"phase", "site", "store", "operation", "kind"},
141+
new String[] {"other", site, "account", "snapshot", "calls"});
142+
return value == null ? 0 : value;
143+
}
144+
43145
@Test
44146
public void disabledAndAbandonedAttemptsDoNotExportOrLeak() {
45147
System.clearProperty("tron.chainbase.executionAttribution");

0 commit comments

Comments
 (0)