panic on pipeline execution

[caevv]

Code:

import "github.com/redis/go-redis/v9"

rdb := redis.NewClient(&redis.Options{
	Addr:         configs.Settings.RedisHost,
	Password:     configs.Settings.RedisPassword, // no password set
	DB:           configs.Settings.RedisDatabase, // use default DB
	DialTimeout:  time.Minute,
	ReadTimeout:  time.Minute,
	WriteTimeout: time.Minute,
})

pipeline := rdb.Pipeline()

// performance tasks on around 1.5 million txs in around 100 goroutines with:
func() error {
    return pipeline.Set(ctx, "key", "value", 0).Err()
}

pipeline.Exec(ctx)

Panic occours on pipeline.Exec().

Stack-trace:

bitbucket.org/company/myservice/internal/myservice.(*Service).Start.func1
	/Users/user/projects/myservice/internal/myservice/bootstrap.go:246
runtime.gopanic
	/opt/homebrew/Cellar/go/1.19.4/libexec/src/runtime/panic.go:884
runtime.panicmem
	/opt/homebrew/Cellar/go/1.19.4/libexec/src/runtime/panic.go:260
runtime.sigpanic
	/opt/homebrew/Cellar/go/1.19.4/libexec/src/runtime/signal_unix.go:835
github.com/redis/go-redis/v9.writeCmd
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/command.go:59
github.com/redis/go-redis/v9.writeCmds
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/command.go:51
github.com/redis/go-redis/v9.(*baseClient).txPipelineProcessCmds.func1
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:522
github.com/redis/go-redis/v9/internal/pool.(*Conn).WithWriter
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/internal/pool/conn.go:90
github.com/redis/go-redis/v9.(*baseClient).txPipelineProcessCmds
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:521
github.com/redis/go-redis/v9.(*baseClient).generalProcessPipeline.func1
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:476
github.com/redis/go-redis/v9.(*baseClient).withConn
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:350
github.com/redis/go-redis/v9.(*baseClient).generalProcessPipeline
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:474
github.com/redis/go-redis/v9.(*baseClient).processTxPipeline
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:452
github.com/redis/go-redis/v9.(*hooksMixin).processTxPipelineHook
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:181
github.com/redis/go-redis/v9.(*Client).TxPipeline.func1
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/redis.go:678
github.com/redis/go-redis/v9.(*Pipeline).Exec
	/Users/user/go/pkg/mod/github.com/redis/go-redis/v9@v9.0.2/pipeline.go:85
bitbucket.org/company/myservice/internal/myservice/repository/store.(*TxHeaderBatchBadger).Flush
	/Users/user/projects/myservice/internal/myservice/repository/store/tx_header_location.go:71
bitbucket.org/company/myservice/internal/myservice/processor.OnFinishProcessingBlock
	/Users/user/projects/myservice/internal/myservice/processor/block.go:100
bitbucket.org/company/myservice/internal/myservice.New.func3
	/Users/user/projects/myservice/internal/myservice/bootstrap.go:122
bitbucket.org/company/common/processor.(*Processor).ProcessBlock
	/Users/user/projects/common/processor/processor.go:196
bitbucket.org/company/myservice/internal/myservice.New.func7
	/Users/user/projects/myservice/internal/myservice/bootstrap.go:158
bitbucket.org/company/common/processor.(*CatchupWithNode).Start
	/Users/user/projects/common/processor/catchup.go:310
bitbucket.org/company/myservice/internal/myservice.(*Service).Start
	/Users/user/projects/myservice/internal/myservice/bootstrap.go:269
main.main
	/Users/user/projects/myservice/main.go:70
runtime.main
	/opt/homebrew/Cellar/go/1.19.4/libexec/src/runtime/proc.go:250
Program exited.

[vmihailenco]

In v9, we Changed Pipelines to not be thread-safe any more. Can that be an issue?

[caevv]

Thanks. That's absolutely that, rollbacked to v8 and all working now. Bit of a shame that this has been changed.

[monkey92t]

bad news....many users don't notice this change.

[monkey92t]

@caevv You should use like this:

type Pipeline struct {
	sync.Mutex
	redis.Pipeliner
}

p := Pipeline{
	Pipeliner: rdb.Pipeline(),
}

for i := 0; i < 100; i++ {
	go func() {
		for i := 0; i < 10000; i++ {
			p.Lock()
			p.Set(ctx, "key", "value", 0).Err()
			p.Unlock()
		}
	}()
}

p.Lock()
_, err := p.Exec(ctx)
p.Unlock()

[monkey92t]

If too many users report this issue, we consider restoring pipeline.mutex.

[caevv]

Thanks @monkey92t.

[keepchen]

I strongly agree with @ash2k 's proposal (#2351) and thank the contributors (@vmihailenco @monkey92t ) for their implementation.

Because in a multi-copy or distributed environment, the use of mutexes is meaningless. Instead, concurrency safety should be improved to the cluster level rather than the component itself.