redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2025-07-29 17:41:15 +03:00
Code Activity

fix(txpipeline): keyless commands should take the slot of the keyed (#3411)

Browse Source 
* fix(txpipeline): keyless commands should take the slot of the keyed commands

* fix(txpipeline): extract only keyed cmds from all cmds

* chore(test): Add tests for keyless cmds and txpipeline

* fix(cmdSlot): Add preferred random slot

* fix(cmdSlot): Add shortlist of keyless cmds

* chore(test): Fix ring test

* fix(keylessCommands): Add list of keyless commands

Add list of keyless Commands based on the Commands output
for redis 8

* chore(txPipeline): refactor slottedCommands impl

* fix(osscluster): typo
This commit is contained in:
Nedyalko Dyakov
2025-06-24 10:34:23 +03:00
committed by GitHub
parent 884f9970c0
commit 05f42e2327
5 changed files with 157 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

116
osscluster.go
View File

@ -998,7 +998,7 @@ func (c *ClusterClient) Process(ctx context.Context, cmd Cmder) error {
}
func (c *ClusterClient) process(ctx context.Context, cmd Cmder) error {
slot := c.cmdSlot(cmd)
slot := c.cmdSlot(cmd, -1)
var node *clusterNode
var moved bool
var ask bool
@ -1344,9 +1344,13 @@ func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmd
return err
}
preferredRandomSlot := -1
if c.opt.ReadOnly && c.cmdsAreReadOnly(ctx, cmds) {
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
slot := c.cmdSlot(cmd, preferredRandomSlot)
if preferredRandomSlot == -1 {
preferredRandomSlot = slot
}
node, err := c.slotReadOnlyNode(state, slot)
if err != nil {
return err
@ -1357,7 +1361,10 @@ func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmd
}
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
slot := c.cmdSlot(cmd, preferredRandomSlot)
if preferredRandomSlot == -1 {
preferredRandomSlot = slot
}
node, err := state.slotMasterNode(slot)
if err != nil {
return err
@ -1519,58 +1526,78 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err
return err
}
cmdsMap := c.mapCmdsBySlot(cmds)
// TxPipeline does not support cross slot transaction.
if len(cmdsMap) > 1 {
keyedCmdsBySlot := c.slottedKeyedCommands(cmds)
slot := -1
switch len(keyedCmdsBySlot) {
case 0:
slot = hashtag.RandomSlot()
case 1:
for sl := range keyedCmdsBySlot {
slot = sl
break
}
default:
// TxPipeline does not support cross slot transaction.
setCmdsErr(cmds, ErrCrossSlot)
return ErrCrossSlot
}
for slot, cmds := range cmdsMap {
node, err := state.slotMasterNode(slot)
if err != nil {
setCmdsErr(cmds, err)
continue
node, err := state.slotMasterNode(slot)
if err != nil {
setCmdsErr(cmds, err)
return err
}
cmdsMap := map[*clusterNode][]Cmder{node: cmds}
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
setCmdsErr(cmds, err)
return err
}
}
cmdsMap := map[*clusterNode][]Cmder{node: cmds}
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
setCmdsErr(cmds, err)
return err
}
}
failedCmds := newCmdsMap()
var wg sync.WaitGroup
failedCmds := newCmdsMap()
var wg sync.WaitGroup
for node, cmds := range cmdsMap {
wg.Add(1)
go func(node *clusterNode, cmds []Cmder) {
defer wg.Done()
c.processTxPipelineNode(ctx, node, cmds, failedCmds)
}(node, cmds)
}
wg.Wait()
if len(failedCmds.m) == 0 {
break
}
cmdsMap = failedCmds.m
for node, cmds := range cmdsMap {
wg.Add(1)
go func(node *clusterNode, cmds []Cmder) {
defer wg.Done()
c.processTxPipelineNode(ctx, node, cmds, failedCmds)
}(node, cmds)
}
wg.Wait()
if len(failedCmds.m) == 0 {
break
}
cmdsMap = failedCmds.m
}
return cmdsFirstErr(cmds)
}
func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
cmdsMap := make(map[int][]Cmder)
// slottedKeyedCommands returns a map of slot to commands taking into account
// only commands that have keys.
func (c *ClusterClient) slottedKeyedCommands(cmds []Cmder) map[int][]Cmder {
cmdsSlots := map[int][]Cmder{}
preferredRandomSlot := -1
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
cmdsMap[slot] = append(cmdsMap[slot], cmd)
if cmdFirstKeyPos(cmd) == 0 {
continue
}
slot := c.cmdSlot(cmd, preferredRandomSlot)
if preferredRandomSlot == -1 {
preferredRandomSlot = slot
}
cmdsSlots[slot] = append(cmdsSlots[slot], cmd)
}
return cmdsMap
return cmdsSlots
}
func (c *ClusterClient) processTxPipelineNode(
@ -1885,17 +1912,20 @@ func (c *ClusterClient) cmdInfo(ctx context.Context, name string) *CommandInfo {
return info
}
func (c *ClusterClient) cmdSlot(cmd Cmder) int {
func (c *ClusterClient) cmdSlot(cmd Cmder, preferredRandomSlot int) int {
args := cmd.Args()
if args[0] == "cluster" && (args[1] == "getkeysinslot" || args[1] == "countkeysinslot") {
return args[2].(int)
}
return cmdSlot(cmd, cmdFirstKeyPos(cmd))
return cmdSlot(cmd, cmdFirstKeyPos(cmd), preferredRandomSlot)
}
func cmdSlot(cmd Cmder, pos int) int {
func cmdSlot(cmd Cmder, pos int, preferredRandomSlot int) int {
if pos == 0 {
if preferredRandomSlot != -1 {
return preferredRandomSlot
}
return hashtag.RandomSlot()
}
firstKey := cmd.stringArg(pos)