/**
* Copyright 2024 Beijing Baolande Software Corporation
*
* 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.
*
* Runtime Library Exception to the Apache 2.0 License:
*
* As an exception, if you use this Software to compile your source code and
* portions of this Software are embedded into the binary product as a result,
* you may redistribute such product without providing attribution as would
* otherwise be required by Sections 4(a), 4(b) and 4(d) of the License.
*/
package redis_sdk.client
public open class ClusterInfoCache {
static let logger: Logger = LoggerFactory.getLogger("redis-sdk")
private static let initCacheLock = Mutex()
private static let idGenerator = AtomicInt64(0)
private let id: Int64
/**
* 缓存ClusterInfoCache
*/
private static let clusterInfoCaches = HashMap<ClusterInfoCacheKey, ClusterInfoCache>()
private let clusterConfig: ClusterRedisClientConfig
private let startNodes: HashSet<HostAndPort>
private let respVersionAware: RespVersionAware = DefaultRespVersionAware()
private let rwl = ReadWriteLock()
private let r = rwl.readLock
private let w = rwl.writeLock
private let slots = Array<?ClientTcpEndpoint>(CLUSTER_HASHSLOTS, repeat: None)
private let slotNodes = Array<?HostAndPort>(CLUSTER_HASHSLOTS, repeat: None)
private let nodes = HashMap<String, ClientTcpEndpoint>()
private let threadPool: ThreadPool
private let rediscoverLock = Mutex()
private let initialized = AtomicBool(false)
/**
* 处理Cache引用相关的锁
*/
private let refCacheMutex = Mutex()
private let clusterConnectionProviders = ArrayList<ClusterConnectionProvider>()
private var closeTaskRef: ?CloseClusterInfoCacheTask = None
private let refCount = AtomicUInt64(0)
private let closed = AtomicBool(false)
private var topologyRefreshExecutor: ?Timer = None
private static let MASTER_NODE_INDEX = 2
private static let CLUSTER_HASHSLOTS = 16384
private let subscribeMode: Bool
/**
* 从缓存中获取ClusterInfoCache
*/
static func getClusterInfoCache(clusterConfig: ClusterRedisClientConfig): ClusterInfoCache {
return getClusterInfoCache(clusterConfig, subscribeMode: false)
}
/**
* 从缓存中获取ClusterInfoCache
*
*/
static func getClusterInfoCache(clusterConfig: ClusterRedisClientConfig, subscribeMode!: Bool): ClusterInfoCache {
let clusterInfoCacheKey = ClusterInfoCacheKey(clusterConfig, subscribeMode: subscribeMode)
synchronized(initCacheLock) {
if (let Some(clusterInfoCache) <- clusterInfoCaches.get(clusterInfoCacheKey)) {
if (clusterInfoCache.retain()) {
return clusterInfoCache
} else {
clusterInfoCaches.remove(clusterInfoCacheKey)
}
}
let clusterInfoCache = ClusterInfoCache(clusterConfig, subscribeMode: subscribeMode)
clusterInfoCache.initialize()
clusterInfoCaches.add(clusterInfoCacheKey, clusterInfoCache)
// 新创建ClusterInfoCache的也需要调用retain，确保引用计数正确
clusterInfoCache.retain()
return clusterInfoCache
}
}
public init(clusterConfig: ClusterRedisClientConfig) {
this(clusterConfig, subscribeMode: false)
}
public init(clusterConfig: ClusterRedisClientConfig, subscribeMode!: Bool) {
this.id = idGenerator.fetchAdd(1)
this.clusterConfig = clusterConfig
this.startNodes = clusterConfig.getClusterHostAndPorts()
this.subscribeMode = subscribeMode
// 集群中多个实例的客户端共用一个线程池
this.threadPool = RedisClientBuilder.buildThreadPool(clusterConfig)
}
public func getClusterConfig() {
return this.clusterConfig
}
public func getRespVersion(): ?ProtocolVersion {
return respVersionAware.getRespVersion()
}
/**
* 添加使用该ClusterInfoCache的ClusterConnectionProvider
*
*/
func addClusterConnectionProvider(clusterConnectionProvider: ClusterConnectionProvider) {
checkOpen()
synchronized(refCacheMutex) {
if (!this.clusterConnectionProviders.contains(clusterConnectionProvider)) {
this.clusterConnectionProviders.add(clusterConnectionProvider)
}
}
if (logger.isDebugEnabled()) {
logger.debug(
"ClusterInfoCache::addClusterConnectionProvider, add ${clusterConnectionProvider} to ${getClusterInfoCacheStatus()}"
)
}
}
/**
* 移除使用该ClusterInfoCache的ClusterConnectionProvider
*
* 没有ClusterConnectionProvider使用该ClusterInfoCache时，则关闭ClusterInfoCache
*
*/
func removeClusterConnectionProvider(clusterConnectionProvider: ClusterConnectionProvider) {
func checkEquals(item: ClusterConnectionProvider) {
if (refEq(clusterConnectionProvider, item) || clusterConnectionProvider == item) {
if (refCount.load() > 0) {
refCount.fetchSub(1)
}
return true
}
return false
}
synchronized(refCacheMutex) {
this.clusterConnectionProviders.removeIf(checkEquals)
if (logger.isDebugEnabled()) {
logger.debug(
"ClusterInfoCache::removeClusterConnectionProvider, remove ${clusterConnectionProvider} from ${getClusterInfoCacheStatus()}"
)
}
if (refCount.load() == 0 && clusterConnectionProviders.isEmpty()) {
if (closeTaskRef.isNone()) {
let closeTask = CloseClusterInfoCacheTask(this)
closeTaskRef = closeTask
if (logger.isDebugEnabled()) {
logger.debug(
"ClusterInfoCache::removeClusterConnectionProvider, create CloseClusterInfoCacheTask for ${getClusterInfoCacheStatus()}"
)
}
spawn {
Thread.currentThread.name = "ClusterInfoCache-Destory"
closeTask.run()
}
}
}
}
}
/**
* 保留ClusterInfoCache，避免被关闭
*/
private func retain(): Bool {
synchronized(refCacheMutex) {
if (!closed.load()) {
refCount.fetchAdd(1)
return true
}
}
return false
}
func getClusterInfoCacheStatus(): String {
return "ClusterInfoCache{id: ${id}, start nodes: ${clusterConfig.getStartNodes()}, reference count: ${refCount.load()}}"
}
public open func initialize() {
if (initialized.compareAndSwap(false, true)) {
initializeSlotsCache(clusterConfig.getClusterHostAndPorts(), clusterConfig)
//init slot cache refresh timer
if (let Some(topologyRefreshPeriod) <- clusterConfig.topologyRefreshPeriod) {
logger.info(
"Cluster topology refresh start, period: {clusterConfig.topologyRefreshPeriod}, startNodes: ${getStartNodes()}"
)
topologyRefreshExecutor = Timer.repeat(
topologyRefreshPeriod,
topologyRefreshPeriod,
{
=>
if (logger.isDebugEnabled()) {
logger.debug("Cluster topology refresh run, old nodes: ${getNodesKeySet()}")
}
renewClusterSlots(None)
if (logger.isDebugEnabled()) {
logger.debug("Cluster topology refresh run, new nodes: ${getNodesKeySet()}")
}
}
)
}
}
}
public func initializeSlotsCache(startNodes: HashSet<HostAndPort>, clusterConfig: ClusterRedisClientConfig): Unit {
if (startNodes.isEmpty()) {
throw RedisClusterOperationException("No nodes to initialize cluster slots cache.")
}
var firstMessage = ""
var startNodeList = ArrayList<HostAndPort>(startNodes)
ClusterUtils.shuffle(startNodeList)
for (node in startNodeList) {
var tcpEndPoint: ?ClientTcpEndpoint = None
try {
let endPoint = this.buildClusterEndPoint(node, subscribeMode)
tcpEndPoint = endPoint
discoverClusterNodesAndSlots(endPoint)
return
} catch (ex: Exception) {
if (firstMessage == "") {
firstMessage = ex.message
}
} finally {
if (let Some(v) <- tcpEndPoint) {
v.stop()
}
}
}
var message = ""
if (firstMessage != "") {
message = "Could not initialize cluster slots cache, ${firstMessage}"
} else {
message = "Could not initialize cluster slots cache."
}
throw RedisClusterOperationException(message)
}
/**
* 创建连接Cluster实例的ClientTcpEndpoint
*/
private func buildClusterEndPoint(hostAndPort: HostAndPort, isSubscribedMode: Bool): ClientTcpEndpoint {
let config = ClientEndpointConfig()
config.host = hostAndPort.host
config.port = hostAndPort.port
clusterConfig.configTcpEndpointOptions(config)
let tcpEndpoint = ClientTcpEndpoint(config, this.threadPool)
if (!isSubscribedMode) {
// 添加Redis命令相关的编解码器
RedisClientBuilder.configEndpoint(tcpEndpoint)
} else {
RedisSubscriberBuilder.configEndpoint(tcpEndpoint, RedisSubscriberHandler.getInstance())
}
let handshakeInfo = RedisClientBuilder.getHandShakeInfo(clusterConfig)
let clientInitializer: RedisClientInitializer = RedisClientInitializer(
respVersionAware,
handshakeInfo,
clusterConfig.waitResponseTimeout
)
tcpEndpoint.setConnectionInitializer(clientInitializer)
tcpEndpoint.start()
// 设置ClientInitializer之后再验证连接
RedisClientBuilder.validClientEndPoint(tcpEndpoint)
return tcpEndpoint
}
public func getConnection(args: CommandArgs): ClientTcpEndpoint {
let slot = args.getKeySlot()
var clientTcpEndpoint: ClientTcpEndpoint
if (slot >= 0) {
clientTcpEndpoint = getConnectionFromSlot(slot)
} else {
clientTcpEndpoint = getConnection()
}
return clientTcpEndpoint
}
public func getConnection(node: HostAndPort): ClientTcpEndpoint {
let endpoint = setupNodeIfNotExist(node)
return endpoint
}
public func getConnection(): ClientTcpEndpoint {
var pools = getShuffledNodesPool()
var firstExceptionMessage = ""
for (pool in pools) {
try {
executePing(pool)
return pool
} catch (ex: Exception) {
if (firstExceptionMessage == "") {
firstExceptionMessage = ex.message
}
}
}
var message = ""
if (firstExceptionMessage != "") {
message = "No reachable node in cluster, ${firstExceptionMessage}"
} else {
message = "No reachable node in cluster."
}
throw RedisClusterOperationException(message)
}
public func executeCommand(clientTcpEndPoint: ClientTcpEndpoint, command: RedisCommand): RedisMessage {
checkOpen()
var needCloseSession: ?Session = None<Session>
try {
let session = clientTcpEndPoint.createSession()
needCloseSession = Some(session)
session.writeAndFlushMessage(command)
return command.waitForResponse(clusterConfig.waitResponseTimeout)
} catch (ex: Exception) {
throw ex
} finally {
if (let Some(session) <- needCloseSession) {
session.close()
}
}
}
public func getConnectionFromSlot(slot: Int64): ClientTcpEndpoint {
var poolVal = getSlotPool(slot)
if (let Some(pool) <- poolVal) {
return pool
} else {
renewClusterSlots(None)
poolVal = getSlotPool(slot)
if (let Some(pool) <- poolVal) {
return pool
} else {
return getConnection()
}
}
}
public func renewClusterSlots(clientTcpEndpointVal: ?ClientTcpEndpoint): Unit {
if (rediscoverLock.tryLock()) {
try {
//First, if pool is available, use pool renew.
if (let Some(clientTcpEndpoint) <- clientTcpEndpointVal) {
try {
discoverClusterSlots(clientTcpEndpoint)
return
} catch (ex: RedisException) {
// try nodes form all pools
}
}
// Then, we use startNodes to try, as long as startNodes is available,
// whether it is vip, domain, or physical ip, it will succeed.
if (startNodes.size > 0) {
for (node in startNodes) {
var tcpEndPoint: ?ClientTcpEndpoint = None
try {
let endPoint = this.buildClusterEndPoint(node, subscribeMode)
tcpEndPoint = endPoint
discoverClusterSlots(endPoint)
return
} catch (ex: Exception) {
// try next nodes
} finally {
if (let Some(v) <- tcpEndPoint) {
v.stop()
}
}
}
}
//Finally, we go back to the ShuffledNodesPool and try the remaining physical nodes.
for (pool in getShuffledNodesPool()) {
try {
if (startNodes.contains(HostAndPort.toHostAndPort(pool.getRemoteAddress()))) {
continue
}
discoverClusterSlots(pool)
return
} catch (ex: Exception) {
// try next nodes
}
}
} finally {
rediscoverLock.unlock()
}
}
}
public func getShuffledNodesPool(): Array<ClientTcpEndpoint> {
r.lock()
try {
let pools = nodes.values().toArray()
ClusterUtils.shuffle(pools)
return pools
} finally {
r.unlock()
}
}
public func discoverClusterSlots(clientTcpEndpoint: ClientTcpEndpoint): Unit {
let slotsInfo = executeClusterSlot(clientTcpEndpoint)
if (slotsInfo.size == 0) {
throw RedisClusterOperationException("Cluster slots list is empty.")
}
w.lock()
try {
ClusterUtils.fillNone(slots)
ClusterUtils.fillNone(slotNodes)
let hostAndPortKeys = HashSet<String>()
for (slotObj in slotsInfo) {
if (let Some(slotObj) <- (slotObj as ArrayList<Any>)) {
if (slotObj.size <= MASTER_NODE_INDEX) {
continue
}
var slotNums = getAssignedSlotArray(slotObj)
var size = slotObj.size
for (i in MASTER_NODE_INDEX..size : 1) {
if (let Some(items) <- (slotObj[i] as ArrayList<Any>)) {
if (items.isEmpty()) {
continue
}
let targetNode: HostAndPort = generateHostAndPort(items)
hostAndPortKeys.add(getNodeKey(targetNode))
setupNodeIfNotExist(targetNode)
if (i == MASTER_NODE_INDEX) {
assignSlotsToNode(slotNums, targetNode)
}
}
}
}
}
// Remove dead nodes according to the latest query
var entryIt = nodes.iterator()
while (let Some((key, pool)) <- entryIt.next()) {
if (!hostAndPortKeys.contains(key)) {
try {
pool.stop()
} catch (ex: Exception) {
//ignore
}
entryIt.remove()
}
}
} finally {
w.unlock()
}
}
public func discoverClusterNodesAndSlots(clientTcpEndpoint: ClientTcpEndpoint) {
let clusterSlots: ArrayList<Any> = executeClusterSlot(clientTcpEndpoint)
if (clusterSlots.size == 0) {
throw RedisClusterOperationException("Cluster slots list is empty.")
}
w.lock()
try {
reset()
let hostAndPortKeys = HashSet<String>()
for (slotObj in clusterSlots) {
if (let Some(slotObj) <- (slotObj as ArrayList<Any>)) {
if (slotObj.size <= MASTER_NODE_INDEX) {
continue
}
var slotNums = getAssignedSlotArray(slotObj)
var size = slotObj.size
for (i in MASTER_NODE_INDEX..size : 1) {
if (let Some(items) <- (slotObj[i] as ArrayList<Any>)) {
if (items.isEmpty()) {
continue
}
let targetNode: HostAndPort = generateHostAndPort(items)
setupNodeIfNotExist(targetNode)
if (i == MASTER_NODE_INDEX) {
assignSlotsToNode(slotNums, targetNode)
}
}
}
}
}
} finally {
w.unlock()
}
}
public func close(): Unit {
var realClose = false
synchronized(refCacheMutex) {
closeTaskRef = None
if (refCount.load() == 0 && clusterConnectionProviders.isEmpty()) {
// 确保不被ClusterConnectionProvider引用时，才关闭
if (closed.compareAndSwap(false, true)) {
if (logger.isDebugEnabled()) {
logger.debug("ClusterInfoCache::close, close ${getClusterInfoCacheStatus()}")
}
realClose = true
}
} else {
if (logger.isDebugEnabled()) {
logger.debug(
"ClusterInfoCache::close, abandon CloseClusterInfoCacheTask, because the ${getClusterInfoCacheStatus()} is reused"
)
}
}
}
if (realClose) {
synchronized(initCacheLock) {
// 从缓存中移除
let clusterInfoCacheKey = ClusterInfoCacheKey(clusterConfig, subscribeMode: subscribeMode)
clusterInfoCaches.remove(clusterInfoCacheKey)
}
// 会锁ReadWriteMutex，移到synchronized(refCacheMutex)之外
reset()
if (let Some(executor) <- topologyRefreshExecutor) {
logger.info("Cluster topology execute cancel method, startNodes: ${startNodes}");
executor.cancel()
}
}
}
public func reset(): Unit {
w.lock()
try {
for (pool in nodes.values()) {
try {
pool.stop()
} catch (e: RedisException) {
// pass
}
}
nodes.clear()
ClusterUtils.fillNone(slots)
ClusterUtils.fillNone(slotNodes)
} finally {
w.unlock()
}
}
public func assignSlotsToNode(targetSlots: ArrayList<Int64>, targetNode: HostAndPort) {
w.lock()
try {
let targetPool = setupNodeIfNotExist(targetNode)
for (slot in targetSlots) {
slots[slot] = targetPool
slotNodes[slot] = targetNode
}
} finally {
w.unlock()
}
}
public func getSlotPool(slot: Int64): ?ClientTcpEndpoint {
r.lock()
try {
return slots[slot]
} finally {
r.unlock()
}
}
public func setupNodeIfNotExist(node: HostAndPort): ClientTcpEndpoint {
w.lock()
try {
let nodeKey = getNodeKey(node)
if (let Some(v) <- nodes.get(nodeKey)) {
return v
}
let nodePool = this.buildClusterEndPoint(node, subscribeMode)
nodes.add(nodeKey, nodePool)
return nodePool
} finally {
w.unlock()
}
}
public func getNodesKeySet(): ArrayList<String> {
let keys = ArrayList<String>()
var entryIt = nodes.iterator()
while (let Some((key, pool)) <- entryIt.next()) {
keys.add(key)
}
return keys
}
public func getStartNodes(): HashSet<HostAndPort> {
return startNodes
}
public func getNodes() {
r.lock()
try {
return HashMap<String, ClientTcpEndpoint>(nodes)
} finally {
r.unlock()
}
}
private func generateHostAndPort(items: ArrayList<Any>): HostAndPort {
var host = getStringFromAny(items[0])
var port = getInt64FromAny(items[1])
return HostAndPort(host, UInt16(port))
}
private func getNodeKey(hnp: HostAndPort) {
return hnp.toString()
}
private func getAssignedSlotArray(slotInfo: ArrayList<Any>): ArrayList<Int64> {
let slotNums = ArrayList<Int64>()
let start = getInt64FromAny(slotInfo[0])
let end = getInt64FromAny(slotInfo[1])
var slot = start
while (slot <= end) {
slotNums.add(slot)
slot++
}
return slotNums
}
private func getStringFromAny(obj: Any): String {
if (let Some(v) <- (obj as String)) {
return v
}
throw RedisException("Failure to convert any type to string type")
}
private func getInt64FromAny(obj: Any): Int64 {
if (let Some(v) <- (obj as Int64)) {
return v
}
throw RedisException("Failure to convert any type to int64 type")
}
/**
返回数据格式
1) 1) (integer) 0
2) (integer) 5460
3) 1) "127.0.0.1"
2) (integer) 30001
3) "09dbe9720cda62f7865eabc5fd8857c5d2678366"
4) 1) hostname
2) "host-1.redis.example.com"
4) 1) "127.0.0.1"
2) (integer) 30004
3) "821d8ca00d7ccf931ed3ffc7e3db0599d2271abf"
4) 1) hostname
2) "host-2.redis.example.com"
2) 1) (integer) 5461
2) (integer) 10922
3) 1) "127.0.0.1"
2) (integer) 30002
3) "c9d93d9f2c0c524ff34cc11838c2003d8c29e013"
4) 1) hostname
2) "host-3.redis.example.com"
4) 1) "127.0.0.1"
2) (integer) 30005
3) "faadb3eb99009de4ab72ad6b6ed87634c7ee410f"
4) 1) hostname
2) "host-4.redis.example.com"
*/
private func executeClusterSlot(clientEndPoint: ClientTcpEndpoint): ArrayList<Any> {
let command = RedisCommandBuilder.clusterSlots()
let message = executeCommand(clientEndPoint, command)
return command.getBuilder().build(message)
}
private func executePing(clientEndPoint: ClientTcpEndpoint): String {
let command = RedisCommandBuilder.ping()
let message = executeCommand(clientEndPoint, command)
return command.getBuilder().build(message)
}
/**
* 检查客户端是否已经关闭
*/
private func checkOpen(): Unit {
if (isClosed()) {
throw RedisException("Client is already closed")
}
}
private func isClosed(): Bool {
return closed.load()
}
}
/**
* ClusterInfoCache的缓存中，用于关联ClusterInfoCache的key
*
* 普通模式和订阅模式，不能共享ClusterInfoCache
*/
class ClusterInfoCacheKey <: Hashable & Equatable<ClusterInfoCacheKey> {
private let subscribeMode: Bool
private let clusterConfig: ClusterRedisClientConfig
public init(clusterConfig: ClusterRedisClientConfig) {
this(clusterConfig, subscribeMode: false)
}
public init(clusterConfig: ClusterRedisClientConfig, subscribeMode!: Bool) {
this.clusterConfig = clusterConfig
this.subscribeMode = subscribeMode
}
@OverflowWrapping
public override func hashCode(): Int64 {
var hashCode = 53 * subscribeMode.hashCode()
hashCode = hashCode + clusterConfig.hashCode()
return hashCode
}
public operator override func ==(other: ClusterInfoCacheKey): Bool {
if (subscribeMode != other.subscribeMode) {
return false
}
if (clusterConfig != other.clusterConfig) {
return false
}
return true
}
public operator override func !=(other: ClusterInfoCacheKey): Bool {
if (subscribeMode != other.subscribeMode) {
return true
}
if (clusterConfig != other.clusterConfig) {
return true
}
return false
}
}
/**
* 延时关闭ClusterInfoCache的任务
*/
class CloseClusterInfoCacheTask <: Runnable {
private let clusterInfoCache: ClusterInfoCache
public init(clusterInfoCache: ClusterInfoCache) {
this.clusterInfoCache = clusterInfoCache
}
public func run(): Unit {
try {
// 延迟一段时间再关闭, 在该期间内ClusterInfoCache再次被引用则取消关闭
let delayDuraion = clusterInfoCache.getClusterConfig().closeClusterInfoCacheDelayDuration
if (delayDuraion > Duration.Zero) {
// 延迟一段时间再关闭, 在该期间内SentinelManager再次被引用则取消关闭
sleep(delayDuraion)
}
clusterInfoCache.close()
} catch (ex: Exception) {
SentinelManager
.logger
.warn(
"Failure to close ClusterInfoCache ${clusterInfoCache.getClusterInfoCacheStatus()}",
ex
)
}
}
}