Skip to content

Commit

Permalink
1. clean up unused codes
Browse files Browse the repository at this point in the history 
2. optimize range/replica lookup logic in DistCallScheduler
  • Loading branch information
@popduke
popduke committed Sep 13, 2024
1 parent 6bdc85d commit 5e76092
Show file tree
Hide file tree
Showing 26 changed files with 699 additions and 1,587 deletions.
34 changes: 27 additions & 7 deletions ...kv/base-kv-store-client/src/main/java/com/baidu/bifromq/basekv/client/KVRangeSetting.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,9 @@
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ThreadLocalRandom;
import lombok.EqualsAndHashCode;
import lombok.ToString;
Expand All @@ -51,12 +51,12 @@ public KVRangeSetting(String clusterId, String leaderStoreId, KVRangeDescriptor
ver = desc.getVer();
boundary = desc.getBoundary();
leader = leaderStoreId;
Set<String> voters = new HashSet<>();
Set<String> inProcVoters = new HashSet<>();
Set<String> followers = new HashSet<>();
Set<String> inProcFollowers = new HashSet<>();
Set<String> allReplicas = new HashSet<>();
Set<String> inProcReplicas = new HashSet<>();
Set<String> voters = new TreeSet<>();
Set<String> inProcVoters = new TreeSet<>();
Set<String> followers = new TreeSet<>();
Set<String> inProcFollowers = new TreeSet<>();
Set<String> allReplicas = new TreeSet<>();
Set<String> inProcReplicas = new TreeSet<>();

Set<String> allVoters =
Sets.newHashSet(Iterables.concat(desc.getConfig().getVotersList(), desc.getConfig().getNextVotersList()));
Expand Down Expand Up @@ -97,19 +97,39 @@ public KVRangeSetting(String clusterId, String leaderStoreId, KVRangeDescriptor
this.inProcReplicas = Collections.unmodifiableList(Lists.newArrayList(inProcReplicas));
}

public boolean hasInProcVoter() {
return !inProcVoters.isEmpty();
}

public boolean hasInProcReplica() {
return !inProcReplicas.isEmpty();
}

public String randomReplica() {
if (!inProcReplicas.isEmpty()) {
if (inProcReplicas.size() == 1) {
return inProcReplicas.get(0);
}
return inProcReplicas.get(ThreadLocalRandom.current().nextInt(inProcReplicas.size()));
}
if (allReplicas.size() == 1) {
return allReplicas.get(0);
}
return allReplicas.get(ThreadLocalRandom.current().nextInt(allReplicas.size()));
}

public String randomVoters() {
if (getInProcStores(clusterId).contains(leader)) {
return leader;
} else if (!inProcVoters.isEmpty()) {
if (inProcVoters.size() == 1) {
return inProcVoters.get(0);
}
return inProcVoters.get(ThreadLocalRandom.current().nextInt(inProcVoters.size()));
}
if (voters.size() == 1) {
return voters.get(0);
}
return voters.get(ThreadLocalRandom.current().nextInt(voters.size()));
}
}
33 changes: 28 additions & 5 deletions ...t/bifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/EntityUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@

import static com.baidu.bifromq.dist.util.TopicUtil.escape;
import static com.baidu.bifromq.dist.util.TopicUtil.isNormalTopicFilter;
import static com.baidu.bifromq.dist.util.TopicUtil.parseSharedTopic;
import static com.baidu.bifromq.dist.util.TopicUtil.unescape;
import static com.baidu.bifromq.util.TopicConst.DELIMITER_CHAR;
import static com.baidu.bifromq.util.TopicConst.NUL;
Expand All @@ -25,7 +24,6 @@
import static com.google.protobuf.ByteString.copyFromUtf8;

import com.baidu.bifromq.dist.rpc.proto.GroupMatchRecord;
import com.baidu.bifromq.dist.util.TopicUtil;
import com.google.protobuf.ByteString;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
Expand Down Expand Up @@ -160,7 +158,7 @@ public static ByteString toNormalMatchRecordKey(String tenantId, String topicFil

public static ByteString toGroupMatchRecordKey(String tenantId, String topicFilter) {
assert !isNormalTopicFilter(topicFilter);
TopicUtil.SharedTopicFilter stf = parseSharedTopic(topicFilter);
SharedTopicFilter stf = parseSharedTopic(topicFilter);
return matchRecordKeyPrefix(tenantId, stf.topicFilter)
.concat(stf.ordered ? FLAG_ORDERD_SHARE : FLAG_UNORDERD_SHARE)
.concat(copyFromUtf8(stf.shareGroup));
Expand All @@ -172,7 +170,7 @@ public static ByteString toMatchRecordKey(String tenantId, String topicFilter, S
.concat(FLAG_NORMAL)
.concat(copyFromUtf8(qInboxId));
} else {
TopicUtil.SharedTopicFilter stf = parseSharedTopic(topicFilter);
SharedTopicFilter stf = parseSharedTopic(topicFilter);
return matchRecordKeyPrefix(tenantId, stf.topicFilter)
.concat(stf.ordered ? FLAG_ORDERD_SHARE : FLAG_UNORDERD_SHARE)
.concat(copyFromUtf8(stf.shareGroup));
Expand All @@ -187,11 +185,22 @@ public static ByteString toMatchRecordKeyPrefix(String tenantId, String topicFil
if (isNormalTopicFilter(topicFilter)) {
return matchRecordKeyPrefix(tenantId, topicFilter);
} else {
TopicUtil.SharedTopicFilter stf = parseSharedTopic(topicFilter);
SharedTopicFilter stf = parseSharedTopic(topicFilter);
return matchRecordKeyPrefix(tenantId, stf.topicFilter);
}
}

private static SharedTopicFilter parseSharedTopic(String topicFilter) {
assert !isNormalTopicFilter(topicFilter);
String sharePrefix = topicFilter.startsWith(UNORDERED_SHARE) ? UNORDERED_SHARE : ORDERED_SHARE;
boolean ordered = !topicFilter.startsWith(UNORDERED_SHARE);
String rest = topicFilter.substring((sharePrefix + DELIMITER_CHAR).length());
int firstTopicSeparatorIndex = rest.indexOf(DELIMITER_CHAR);
String shareGroup = rest.substring(0, firstTopicSeparatorIndex);
return new SharedTopicFilter(topicFilter, ordered, shareGroup,
rest.substring(firstTopicSeparatorIndex + 1));
}

public static String parseTopicFilter(String matchRecordKeyStr) {
// <tenantId><NUL><1><ESCAPED_TOPIC_FILTER><NUL><FLAG><SCOPED_INBOX|SHARE_GROUP>
int firstSplit = matchRecordKeyStr.indexOf(NUL_CHAR);
Expand Down Expand Up @@ -238,4 +247,18 @@ public static String parseOriginalTopicFilter(String matchRecordKeyStr) {
}
}
}

public static TenantAndEscapedTopicFilter parseTenantAndEscapedTopicFilter(ByteString matchRecordKey) {
String matchRecordKeyStr = matchRecordKey.toStringUtf8();
int firstSplit = matchRecordKeyStr.indexOf(NUL_CHAR);
int lastSplit = matchRecordKeyStr.lastIndexOf(NUL_CHAR);
return new TenantAndEscapedTopicFilter(matchRecordKeyStr.substring(0, firstSplit),
matchRecordKeyStr.substring(firstSplit + 2, lastSplit));
}

public record TenantAndEscapedTopicFilter(String tenantId, String escapedTopicFilter) {
public String toGlobalTopicFilter() {
return tenantId + NUL + escapedTopicFilter;
}
}
}
28 changes: 28 additions & 0 deletions ...mq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/entity/SharedTopicFilter.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
/*
* Copyright (c) 2024. The BifroMQ Authors. All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
* http://www.apache.org/licenses/LICENSE-2.0
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and limitations under the License.
*/

package com.baidu.bifromq.dist.entity;

public class SharedTopicFilter {
public String originTopicFilter;
public boolean ordered;
public String shareGroup;
public String topicFilter;

public SharedTopicFilter(String originTopicFilter, boolean ordered, String shareName, String filter) {
this.originTopicFilter = originTopicFilter;
this.ordered = ordered;
this.shareGroup = shareName;
this.topicFilter = filter;
}
}
2 changes: 1 addition & 1 deletion .../bifromq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/trie/TopicTrieNode.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,7 @@ TopicTrieNode<V> child(String levelName) {
}

/**
* The builder for building a TopicTrie.
* The builder for building a TopicTrieNode.
*
* @param <V> the value associated with the topic
*/
Expand Down
148 changes: 0 additions & 148 deletions ...omq-dist-rpc-definition/src/main/java/com/baidu/bifromq/dist/util/TopicFilterMatcher.java
View file

This file was deleted.

Loading

0 comments on commit 5e76092

Please sign in to comment.