redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2025-04-19 07:22:17 +03:00
Code

Merge pull request #2358 from monkey92t/fifo

Browse Source 
feat: hook mode changed to FIFO
This commit is contained in:
Monkey 2023-01-21 18:44:14 +08:00 committed by GitHub
commit eeb49d3733
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 193 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
cluster.go
View File

@ -838,7 +838,7 @@ type ClusterClient struct {
state *clusterStateHolder
cmdsInfoCache *cmdsInfoCache
cmdable
hooks
hooksMixin
}
// NewClusterClient returns a Redis Cluster client as described in
@ -855,9 +855,12 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
c.cmdsInfoCache = newCmdsInfoCache(c.cmdsInfo)
c.cmdable = c.Process
c.hooks.setProcess(c.process)
c.hooks.setProcessPipeline(c.processPipeline)
c.hooks.setProcessTxPipeline(c.processTxPipeline)
c.initHooks(hooks{
dial: nil,
process: c.process,
pipeline: c.processPipeline,
txPipeline: c.processTxPipeline,
})
return c
}
@ -889,7 +892,7 @@ func (c *ClusterClient) Do(ctx context.Context, args ...interface{}) *Cmd {
}
func (c *ClusterClient) Process(ctx context.Context, cmd Cmder) error {
err := c.hooks.process(ctx, cmd)
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}
@ -1187,7 +1190,7 @@ func (c *ClusterClient) loadState(ctx context.Context) (*clusterState, error) {
func (c *ClusterClient) Pipeline() Pipeliner {
pipe := Pipeline{
exec: pipelineExecer(c.hooks.processPipeline),
exec: pipelineExecer(c.processPipelineHook),
}
pipe.init()
return &pipe
@ -1276,7 +1279,7 @@ func (c *ClusterClient) cmdsAreReadOnly(ctx context.Context, cmds []Cmder) bool
func (c *ClusterClient) processPipelineNode(
ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
) {
_ = node.Client.hooks.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
_ = node.Client.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
cn, err := node.Client.getConn(ctx)
if err != nil {
_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
@ -1380,7 +1383,7 @@ func (c *ClusterClient) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
cmds = wrapMultiExec(ctx, cmds)
return c.hooks.processTxPipeline(ctx, cmds)
return c.processTxPipelineHook(ctx, cmds)
},
}
pipe.init()
@ -1453,7 +1456,7 @@ func (c *ClusterClient) processTxPipelineNode(
ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
) {
cmds = wrapMultiExec(ctx, cmds)
_ = node.Client.hooks.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
_ = node.Client.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
cn, err := node.Client.getConn(ctx)
if err != nil {
_ = c.mapCmdsByNode(ctx, failedCmds, cmds)

10
cluster_commands.go
View File

@ -8,7 +8,7 @@ import (
func (c *ClusterClient) DBSize(ctx context.Context) *IntCmd {
cmd := NewIntCmd(ctx, "dbsize")
_ = c.hooks.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
_ = c.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
var size int64
err := c.ForEachMaster(ctx, func(ctx context.Context, master *Client) error {
n, err := master.DBSize(ctx).Result()
@ -30,8 +30,8 @@ func (c *ClusterClient) DBSize(ctx context.Context) *IntCmd {
func (c *ClusterClient) ScriptLoad(ctx context.Context, script string) *StringCmd {
cmd := NewStringCmd(ctx, "script", "load", script)
_ = c.hooks.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
mu := &sync.Mutex{}
_ = c.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
var mu sync.Mutex
err := c.ForEachShard(ctx, func(ctx context.Context, shard *Client) error {
val, err := shard.ScriptLoad(ctx, script).Result()
if err != nil {
@ -56,7 +56,7 @@ func (c *ClusterClient) ScriptLoad(ctx context.Context, script string) *StringCm
func (c *ClusterClient) ScriptFlush(ctx context.Context) *StatusCmd {
cmd := NewStatusCmd(ctx, "script", "flush")
_ = c.hooks.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
_ = c.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
err := c.ForEachShard(ctx, func(ctx context.Context, shard *Client) error {
return shard.ScriptFlush(ctx).Err()
})
@ -82,7 +82,7 @@ func (c *ClusterClient) ScriptExists(ctx context.Context, hashes ...string) *Boo
result[i] = true
}
_ = c.hooks.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
_ = c.withProcessHook(ctx, cmd, func(ctx context.Context, _ Cmder) error {
var mu sync.Mutex
err := c.ForEachShard(ctx, func(ctx context.Context, shard *Client) error {
val, err := shard.ScriptExists(ctx, hashes...).Result()

178
redis.go
View File

@ -40,18 +40,42 @@ type (
ProcessPipelineHook func(ctx context.Context, cmds []Cmder) error
)
type hooksMixin struct {
slice []Hook
initial hooks
current hooks
}
func (hs *hooksMixin) initHooks(hooks hooks) {
hs.initial = hooks
hs.chain()
}
type hooks struct {
slice []Hook
dialHook DialHook
processHook ProcessHook
processPipelineHook ProcessPipelineHook
processTxPipelineHook ProcessPipelineHook
dial DialHook
process ProcessHook
pipeline ProcessPipelineHook
txPipeline ProcessPipelineHook
}
func (h *hooks) setDefaults() {
if h.dial == nil {
h.dial = func(ctx context.Context, network, addr string) (net.Conn, error) { return nil, nil }
}
if h.process == nil {
h.process = func(ctx context.Context, cmd Cmder) error { return nil }
}
if h.pipeline == nil {
h.pipeline = func(ctx context.Context, cmds []Cmder) error { return nil }
}
if h.txPipeline == nil {
h.txPipeline = func(ctx context.Context, cmds []Cmder) error { return nil }
}
}
// AddHook is to add a hook to the queue.
// Hook is a function executed during network connection, command execution, and pipeline,
// it is a first-in-last-out stack queue (FILO).
// The first to be added to the queue is the execution function of the redis command (the last to be executed).
// it is a first-in-first-out stack queue (FIFO).
// You need to execute the next hook in each hook, unless you want to terminate the execution of the command.
// For example, you added hook-1, hook-2:
//
@ -81,95 +105,80 @@ type hooks struct {
//
// The execution sequence is:
//
// hook-2 start -> hook-1 start -> exec redis cmd -> hook-1 end -> hook-2 end
// hook-1 start -> hook-2 start -> exec redis cmd -> hook-2 end -> hook-1 end
//
// Please note: "next(ctx, cmd)" is very important, it will call the next hook,
// if "next(ctx, cmd)" is not executed in hook-1, the redis command will not be executed.
func (hs *hooks) AddHook(hook Hook) {
// if "next(ctx, cmd)" is not executed, the redis command will not be executed.
func (hs *hooksMixin) AddHook(hook Hook) {
hs.slice = append(hs.slice, hook)
hs.dialHook = hook.DialHook(hs.dialHook)
hs.processHook = hook.ProcessHook(hs.processHook)
hs.processPipelineHook = hook.ProcessPipelineHook(hs.processPipelineHook)
hs.processTxPipelineHook = hook.ProcessPipelineHook(hs.processTxPipelineHook)
hs.chain()
}
func (hs *hooks) clone() hooks {
func (hs *hooksMixin) chain() {
hs.initial.setDefaults()
hs.current.dial = hs.initial.dial
hs.current.process = hs.initial.process
hs.current.pipeline = hs.initial.pipeline
hs.current.txPipeline = hs.initial.txPipeline
for i := len(hs.slice) - 1; i >= 0; i-- {
if wrapped := hs.slice[i].DialHook(hs.current.dial); wrapped != nil {
hs.current.dial = wrapped
}
if wrapped := hs.slice[i].ProcessHook(hs.current.process); wrapped != nil {
hs.current.process = wrapped
}
if wrapped := hs.slice[i].ProcessPipelineHook(hs.current.pipeline); wrapped != nil {
hs.current.pipeline = wrapped
}
if wrapped := hs.slice[i].ProcessPipelineHook(hs.current.txPipeline); wrapped != nil {
hs.current.txPipeline = wrapped
}
}
}
func (hs *hooksMixin) clone() hooksMixin {
clone := *hs
l := len(clone.slice)
clone.slice = clone.slice[:l:l]
return clone
}
func (hs *hooks) setDial(dial DialHook) {
hs.dialHook = dial
for _, h := range hs.slice {
if wrapped := h.DialHook(hs.dialHook); wrapped != nil {
hs.dialHook = wrapped
}
}
}
func (hs *hooks) setProcess(process ProcessHook) {
hs.processHook = process
for _, h := range hs.slice {
if wrapped := h.ProcessHook(hs.processHook); wrapped != nil {
hs.processHook = wrapped
}
}
}
func (hs *hooks) setProcessPipeline(processPipeline ProcessPipelineHook) {
hs.processPipelineHook = processPipeline
for _, h := range hs.slice {
if wrapped := h.ProcessPipelineHook(hs.processPipelineHook); wrapped != nil {
hs.processPipelineHook = wrapped
}
}
}
func (hs *hooks) setProcessTxPipeline(processTxPipeline ProcessPipelineHook) {
hs.processTxPipelineHook = processTxPipeline
for _, h := range hs.slice {
if wrapped := h.ProcessPipelineHook(hs.processTxPipelineHook); wrapped != nil {
hs.processTxPipelineHook = wrapped
}
}
}
func (hs *hooks) withProcessHook(ctx context.Context, cmd Cmder, hook ProcessHook) error {
for _, h := range hs.slice {
if wrapped := h.ProcessHook(hook); wrapped != nil {
func (hs *hooksMixin) withProcessHook(ctx context.Context, cmd Cmder, hook ProcessHook) error {
for i := len(hs.slice) - 1; i >= 0; i-- {
if wrapped := hs.slice[i].ProcessHook(hook); wrapped != nil {
hook = wrapped
}
}
return hook(ctx, cmd)
}
func (hs *hooks) withProcessPipelineHook(
func (hs *hooksMixin) withProcessPipelineHook(
ctx context.Context, cmds []Cmder, hook ProcessPipelineHook,
) error {
for _, h := range hs.slice {
if wrapped := h.ProcessPipelineHook(hook); wrapped != nil {
for i := len(hs.slice) - 1; i >= 0; i-- {
if wrapped := hs.slice[i].ProcessPipelineHook(hook); wrapped != nil {
hook = wrapped
}
}
return hook(ctx, cmds)
}
func (hs *hooks) dial(ctx context.Context, network, addr string) (net.Conn, error) {
return hs.dialHook(ctx, network, addr)
func (hs *hooksMixin) dialHook(ctx context.Context, network, addr string) (net.Conn, error) {
return hs.current.dial(ctx, network, addr)
}
func (hs *hooks) process(ctx context.Context, cmd Cmder) error {
return hs.processHook(ctx, cmd)
func (hs *hooksMixin) processHook(ctx context.Context, cmd Cmder) error {
return hs.current.process(ctx, cmd)
}
func (hs *hooks) processPipeline(ctx context.Context, cmds []Cmder) error {
return hs.processPipelineHook(ctx, cmds)
func (hs *hooksMixin) processPipelineHook(ctx context.Context, cmds []Cmder) error {
return hs.current.pipeline(ctx, cmds)
}
func (hs *hooks) processTxPipeline(ctx context.Context, cmds []Cmder) error {
return hs.processTxPipelineHook(ctx, cmds)
func (hs *hooksMixin) processTxPipelineHook(ctx context.Context, cmds []Cmder) error {
return hs.current.txPipeline(ctx, cmds)
}
//------------------------------------------------------------------------------
@ -575,7 +584,7 @@ func (c *baseClient) context(ctx context.Context) context.Context {
type Client struct {
*baseClient
cmdable
hooks
hooksMixin
}
// NewClient returns a client to the Redis Server specified by Options.
@ -588,17 +597,19 @@ func NewClient(opt *Options) *Client {
},
}
c.init()
c.connPool = newConnPool(opt, c.hooks.dial)
c.connPool = newConnPool(opt, c.dialHook)
return &c
}
func (c *Client) init() {
c.cmdable = c.Process
c.hooks.setDial(c.baseClient.dial)
c.hooks.setProcess(c.baseClient.process)
c.hooks.setProcessPipeline(c.baseClient.processPipeline)
c.hooks.setProcessTxPipeline(c.baseClient.processTxPipeline)
c.initHooks(hooks{
dial: c.baseClient.dial,
process: c.baseClient.process,
pipeline: c.baseClient.processPipeline,
txPipeline: c.baseClient.processTxPipeline,
})
}
func (c *Client) WithTimeout(timeout time.Duration) *Client {
@ -620,7 +631,7 @@ func (c *Client) Do(ctx context.Context, args ...interface{}) *Cmd {
}
func (c *Client) Process(ctx context.Context, cmd Cmder) error {
err := c.hooks.process(ctx, cmd)
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}
@ -644,7 +655,7 @@ func (c *Client) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmd
func (c *Client) Pipeline() Pipeliner {
pipe := Pipeline{
exec: pipelineExecer(c.hooks.processPipeline),
exec: pipelineExecer(c.processPipelineHook),
}
pipe.init()
return &pipe
@ -659,7 +670,7 @@ func (c *Client) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
cmds = wrapMultiExec(ctx, cmds)
return c.hooks.processTxPipeline(ctx, cmds)
return c.processTxPipelineHook(ctx, cmds)
},
}
pipe.init()
@ -742,7 +753,7 @@ type Conn struct {
baseClient
cmdable
statefulCmdable
hooks
hooksMixin
}
func newConn(opt *Options, connPool pool.Pooler) *Conn {
@ -755,17 +766,18 @@ func newConn(opt *Options, connPool pool.Pooler) *Conn {
c.cmdable = c.Process
c.statefulCmdable = c.Process
c.hooks.setDial(c.baseClient.dial)
c.hooks.setProcess(c.baseClient.process)
c.hooks.setProcessPipeline(c.baseClient.processPipeline)
c.hooks.setProcessTxPipeline(c.baseClient.processTxPipeline)
c.initHooks(hooks{
dial: c.baseClient.dial,
process: c.baseClient.process,
pipeline: c.baseClient.processPipeline,
txPipeline: c.baseClient.processTxPipeline,
})
return &c
}
func (c *Conn) Process(ctx context.Context, cmd Cmder) error {
err := c.hooks.process(ctx, cmd)
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}
@ -776,7 +788,7 @@ func (c *Conn) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder
func (c *Conn) Pipeline() Pipeliner {
pipe := Pipeline{
exec: c.hooks.processPipeline,
exec: c.processPipelineHook,
}
pipe.init()
return &pipe
@ -791,7 +803,7 @@ func (c *Conn) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
cmds = wrapMultiExec(ctx, cmds)
return c.hooks.processTxPipeline(ctx, cmds)
return c.processTxPipelineHook(ctx, cmds)
},
}
pipe.init()

48
redis_test.go
View File

@ -483,3 +483,51 @@ var _ = Describe("Conn", func() {
Expect(err).NotTo(HaveOccurred())
})
})
var _ = Describe("Hook", func() {
var client *redis.Client
BeforeEach(func() {
client = redis.NewClient(redisOptions())
Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
})
AfterEach(func() {
err := client.Close()
Expect(err).NotTo(HaveOccurred())
})
It("fifo", func() {
var res []string
client.AddHook(&hook{
processHook: func(hook redis.ProcessHook) redis.ProcessHook {
return func(ctx context.Context, cmd redis.Cmder) error {
res = append(res, "hook-1-process-start")
err := hook(ctx, cmd)
res = append(res, "hook-1-process-end")
return err
}
},
})
client.AddHook(&hook{
processHook: func(hook redis.ProcessHook) redis.ProcessHook {
return func(ctx context.Context, cmd redis.Cmder) error {
res = append(res, "hook-2-process-start")
err := hook(ctx, cmd)
res = append(res, "hook-2-process-end")
return err
}
},
})
err := client.Ping(ctx).Err()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(Equal([]string{
"hook-1-process-start",
"hook-2-process-start",
"hook-2-process-end",
"hook-1-process-end",
}))
})
})

26
ring.go
View File

@ -487,7 +487,7 @@ func (c *ringSharding) Close() error {
// Otherwise you should use Redis Cluster.
type Ring struct {
cmdable
hooks
hooksMixin
opt *RingOptions
sharding *ringSharding
@ -509,12 +509,14 @@ func NewRing(opt *RingOptions) *Ring {
ring.cmdsInfoCache = newCmdsInfoCache(ring.cmdsInfo)
ring.cmdable = ring.Process
ring.hooks.setProcess(ring.process)
ring.hooks.setProcessPipeline(func(ctx context.Context, cmds []Cmder) error {
return ring.generalProcessPipeline(ctx, cmds, false)
})
ring.hooks.setProcessTxPipeline(func(ctx context.Context, cmds []Cmder) error {
return ring.generalProcessPipeline(ctx, cmds, true)
ring.initHooks(hooks{
process: ring.process,
pipeline: func(ctx context.Context, cmds []Cmder) error {
return ring.generalProcessPipeline(ctx, cmds, false)
},
txPipeline: func(ctx context.Context, cmds []Cmder) error {
return ring.generalProcessPipeline(ctx, cmds, true)
},
})
go ring.sharding.Heartbeat(hbCtx, opt.HeartbeatFrequency)
@ -534,7 +536,7 @@ func (c *Ring) Do(ctx context.Context, args ...interface{}) *Cmd {
}
func (c *Ring) Process(ctx context.Context, cmd Cmder) error {
err := c.hooks.process(ctx, cmd)
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}
@ -717,7 +719,7 @@ func (c *Ring) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder
func (c *Ring) Pipeline() Pipeliner {
pipe := Pipeline{
exec: pipelineExecer(c.hooks.processPipeline),
exec: pipelineExecer(c.processPipelineHook),
}
pipe.init()
return &pipe
@ -731,7 +733,7 @@ func (c *Ring) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
cmds = wrapMultiExec(ctx, cmds)
return c.hooks.processTxPipeline(ctx, cmds)
return c.processTxPipelineHook(ctx, cmds)
},
}
pipe.init()
@ -772,9 +774,9 @@ func (c *Ring) generalProcessPipeline(
if tx {
cmds = wrapMultiExec(ctx, cmds)
_ = shard.Client.hooks.processTxPipeline(ctx, cmds)
_ = shard.Client.processTxPipelineHook(ctx, cmds)
} else {
_ = shard.Client.hooks.processPipeline(ctx, cmds)
_ = shard.Client.processPipelineHook(ctx, cmds)
}
}(hash, cmds)
}

14
sentinel.go
View File

@ -214,7 +214,7 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
}
rdb.init()
connPool = newConnPool(opt, rdb.hooks.dial)
connPool = newConnPool(opt, rdb.dialHook)
rdb.connPool = connPool
rdb.onClose = failover.Close
@ -267,7 +267,7 @@ func masterReplicaDialer(
// SentinelClient is a client for a Redis Sentinel.
type SentinelClient struct {
*baseClient
hooks
hooksMixin
}
func NewSentinelClient(opt *Options) *SentinelClient {
@ -278,15 +278,17 @@ func NewSentinelClient(opt *Options) *SentinelClient {
},
}
c.hooks.setDial(c.baseClient.dial)
c.hooks.setProcess(c.baseClient.process)
c.connPool = newConnPool(opt, c.hooks.dial)
c.initHooks(hooks{
dial: c.baseClient.dial,
process: c.baseClient.process,
})
c.connPool = newConnPool(opt, c.dialHook)
return c
}
func (c *SentinelClient) Process(ctx context.Context, cmd Cmder) error {
err := c.hooks.process(ctx, cmd)
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}

20
tx.go
View File

@ -19,7 +19,7 @@ type Tx struct {
baseClient
cmdable
statefulCmdable
hooks
hooksMixin
}
func (c *Client) newTx() *Tx {
@ -28,7 +28,7 @@ func (c *Client) newTx() *Tx {
opt: c.opt,
connPool: pool.NewStickyConnPool(c.connPool),
},
hooks: c.hooks.clone(),
hooksMixin: c.hooksMixin.clone(),
}
tx.init()
return &tx
@ -38,14 +38,16 @@ func (c *Tx) init() {
c.cmdable = c.Process
c.statefulCmdable = c.Process
c.hooks.setDial(c.baseClient.dial)
c.hooks.setProcess(c.baseClient.process)
c.hooks.setProcessPipeline(c.baseClient.processPipeline)
c.hooks.setProcessTxPipeline(c.baseClient.processTxPipeline)
c.initHooks(hooks{
dial: c.baseClient.dial,
process: c.baseClient.process,
pipeline: c.baseClient.processPipeline,
txPipeline: c.baseClient.processTxPipeline,
})
}
func (c *Tx) Process(ctx context.Context, cmd Cmder) error {
err := c.hooks.process(ctx, cmd)
err := c.processHook(ctx, cmd)
cmd.SetErr(err)
return err
}
@ -100,7 +102,7 @@ func (c *Tx) Unwatch(ctx context.Context, keys ...string) *StatusCmd {
func (c *Tx) Pipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds)
return c.processPipelineHook(ctx, cmds)
},
}
pipe.init()
@ -130,7 +132,7 @@ func (c *Tx) TxPipeline() Pipeliner {
pipe := Pipeline{
exec: func(ctx context.Context, cmds []Cmder) error {
cmds = wrapMultiExec(ctx, cmds)
return c.hooks.processTxPipeline(ctx, cmds)
return c.processTxPipelineHook(ctx, cmds)
},
}
pipe.init()