funky-eyes commented on code in PR #8208:
URL: https://github.com/apache/incubator-seata/pull/8208#discussion_r3878686558
##########
common/src/main/java/org/apache/seata/common/store/LockMode.java:
##########
@@ -32,7 +32,11 @@ public enum LockMode {
/**
* raft store
*/
- RAFT("raft");
+ RAFT("raft"),
+ /**
+ * RocksDB lock mode for file store engine.
+ */
+ ROCKSDB("rocksdb");
Review Comment:
Why does lockMode have a separate RocksDB implementation? I think the better
approach would be to abstract the underlying storage operations of the existing
file implementation into an SPI, with the current file-based implementation
serving as the default engine and RocksDB provided as an alternative
implementation.
In other words, we could introduce a configuration such as:
store.file.engine=default/rocksdb
This way, RocksDB would be treated as a pluggable storage engine for the
existing file storage mode, rather than introducing a separate storage mode
specifically for RocksDB.
##########
server/src/main/java/org/apache/seata/server/coordinator/DefaultCoordinator.java:
##########
@@ -574,14 +635,164 @@ private boolean isRetryTimeout(long now, long timeout,
long beginTime) {
return timeout >= ALWAYS_RETRY_BOUNDARY && now - beginTime > timeout;
}
+ private List<GlobalSession> findBackgroundSessions(GlobalStatus status,
boolean lazyLoadBranch) {
+ return findBackgroundSessions(new GlobalStatus[] {status},
lazyLoadBranch);
+ }
+
+ private List<GlobalSession> findBackgroundSessions(GlobalStatus[]
statuses, boolean lazyLoadBranch) {
+ if (statuses == null || statuses.length == 0) {
+ return Collections.emptyList();
+ }
+ if (statuses.length == 1) {
+ return findBackgroundSessionsBySingleStatus(statuses[0],
lazyLoadBranch);
+ }
+ SessionManager sessionManager = SessionHolder.getRootSessionManager();
+ if (sessionManager instanceof RocksDBSessionManager &&
backgroundSessionQueryLimit >= statuses.length) {
+ return
findBackgroundSessionsByMultipleStatusesForRocksDB(statuses, lazyLoadBranch,
sessionManager);
+ }
+ List<GlobalSession> sessions = new ArrayList<>();
+ if (backgroundSessionQueryLimit < statuses.length) {
+ int startIndex = nextBackgroundSessionStatusStart(statuses,
backgroundSessionQueryLimit);
+ for (int i = 0; i < backgroundSessionQueryLimit; i++) {
+ GlobalStatus status = statuses[(startIndex + i) %
statuses.length];
+ sessions.addAll(findBackgroundSessionsBySingleStatus(status,
lazyLoadBranch, 1));
+ }
+
sessions.sort(Comparator.comparingLong(GlobalSession::getBeginTime));
+ return sessions;
+ }
+ int budgetPerStatus = backgroundSessionQueryLimit / statuses.length;
+ int remainder = backgroundSessionQueryLimit % statuses.length;
+ for (int i = 0; i < statuses.length; i++) {
+ int statusBudget = budgetPerStatus + (i < remainder ? 1 : 0);
+ if (statusBudget == 0) {
+ break;
+ }
+ sessions.addAll(findBackgroundSessionsBySingleStatus(statuses[i],
lazyLoadBranch, statusBudget));
+ }
+ sessions.sort(Comparator.comparingLong(GlobalSession::getBeginTime));
+ return sessions;
+ }
+
+ private int nextBackgroundSessionStatusStart(GlobalStatus[] statuses, int
statusesPerRound) {
+ List<GlobalStatus> statusGroup =
+
Collections.unmodifiableList(Arrays.asList(Arrays.copyOf(statuses,
statuses.length)));
+ AtomicInteger nextStart =
+
backgroundSessionStatusGroupOffsets.computeIfAbsent(statusGroup, ignored -> new
AtomicInteger());
+ return nextStart.getAndUpdate(current -> (current + statusesPerRound)
% statuses.length);
+ }
+
+ private synchronized List<GlobalSession>
findBackgroundSessionsByMultipleStatusesForRocksDB(
+ GlobalStatus[] statuses, boolean lazyLoadBranch, SessionManager
sessionManager) {
+ SessionCondition sessionCondition = new SessionCondition(statuses);
+ sessionCondition.setLazyLoadBranch(lazyLoadBranch);
+ sessionCondition.setLimit(backgroundSessionQueryLimit);
+ sessionCondition.setScanLimit(backgroundSessionQueryLimit);
+ Map<GlobalStatus, byte[]> statusScanCursors = new
EnumMap<>(GlobalStatus.class);
+ for (GlobalStatus status : statuses) {
+ byte[] cursor = backgroundSessionStatusCursors.get(status);
+ if (cursor != null) {
+ statusScanCursors.put(status, cursor);
+ }
+ }
+ sessionCondition.setStatusScanCursors(statusScanCursors);
+ long startedAt = System.nanoTime();
+ List<GlobalSession> sessions =
sessionManager.findGlobalSessions(sessionCondition);
+ updateBackgroundSessionCursors(statuses, sessionCondition);
+ logBackgroundSessionScan(Arrays.toString(statuses), sessions,
sessionCondition.getScanStats(), startedAt);
+ if (CollectionUtils.isEmpty(sessions)) {
+ return Collections.emptyList();
+ }
+ return limitBackgroundSessions(sessions);
+ }
+
+ private List<GlobalSession>
findBackgroundSessionsBySingleStatus(GlobalStatus status, boolean
lazyLoadBranch) {
+ return limitBackgroundSessions(
+ findBackgroundSessionsBySingleStatus(status, lazyLoadBranch,
backgroundSessionQueryLimit));
+ }
+
+ private List<GlobalSession> findBackgroundSessionsBySingleStatus(
Review Comment:
The coordination layer should not contain so much implementation-specific
logic. Instead, those details should be encapsulated within the respective
storage implementations.
##########
console/src/main/resources/static/console-fe/src/pages/TransactionInfo/TransactionInfo.tsx:
##########
Review Comment:
I think the frontend changes should be handled in a separate PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
