Deadlock when using helper.Duplicate function in the indicator package

[chentiangang]

Hello,

I encountered a deadlock issue when using the helper.Duplicate function in the indicator package. Here is the code I used to reproduce the issue:


package main

import (
    "fmt"
    "github.com/cinar/indicator/v2/helper"
)

func main() {
    input := helper.SliceToChan([]float64{-10, 20, -4, -5})
    outputs := helper.Duplicate(input, 2)
    fmt.Println(helper.ChanToSlice(outputs[0]))
    fmt.Println(helper.ChanToSlice(outputs[1]))
}

Problem:

When running this code, it results in a deadlock. I expected the helper.Duplicate function to allow me to duplicate the input channel without causing a deadlock.

fatal error: all goroutines are asleep - deadlock!
goroutine 1 [chan receive]:
github.com/cinar/indicator/v2/helper.ChanToSlice[...](...)
        D:/atop/vendor/github.com/cinar/indicator/v2/helper/chan_to_slice.go:22
main.main()
        D:/atop/ahot/select/atr/main.go:22 +0xe7

goroutine 7 [chan send]:
github.com/cinar/indicator/v2/helper.SliceToChan[...].func1()
        D:/atop/vendor/github.com/cinar/indicator/v2/helper/slice_to_chan.go:24 +0x77
created by github.com/cinar/indicator/v2/helper.SliceToChan[...] in goroutine 1
        D:/atop/vendor/github.com/cinar/indicator/v2/helper/slice_to_chan.go:20 +0xab

goroutine 8 [chan send]:
github.com/cinar/indicator/v2/helper.Duplicate[...].func1()
        D:/atop/vendor/github.com/cinar/indicator/v2/helper/duplicate.go:34 +0x10c
created by github.com/cinar/indicator/v2/helper.Duplicate[...] in goroutine 1
        D:/atop/vendor/github.com/cinar/indicator/v2/helper/duplicate.go:27 +0x146

Environment:

Go version: go version go1.23.0 windows/amd64
Indicator version: github.com/cinar/indicator/v2 v2.1.7
Could you please take a look? I’d appreciate any guidance or suggestions to resolve this.

Thank you!

[cinar]

Hi!

Thank you for filling this issue. Yes, looking at your example code, by design, it will get into a deadlock, let me explain why, and how to get around that:

Imagine helper.Duplicate() creates two copies of an input channel. It expects these copies to be used in a strict order:

• outputs[0] is read first.
• outputs[1] is read only after outputs[0] is read.
• outputs[0] can be read again, but only after outputs[1] is read.

If your code doesn't follow this order (e.g., trying to use outputs[1] before outputs[0]), it creates a "deadlock".

From duplicate.go:

		for n := range input {
			for _, output := range outputs {
				output <- n
			}
		}

In your example, helper.ChanToSlice(outputs[0]) will block until the entire channel is read into a slice. As it is a blocking call, helper.ChanToSlice(outputs[1]) won't start. As outputs[1] does not get read simultaneously, it will block outputs[0], and you'll get this problem.

One way to get around this is through the helper.Buffered(). It will allow you have a set a buffer for the channel, so that it won't block immediately.

For example, in your case, if you use the helper.Buffered(), it will no longer get into a deadlock:

package main

import (
        "fmt"

        "github.com/cinar/indicator/v2/helper"
)

func main() {
        input := helper.SliceToChan([]float64{-10, 20, -4, -5})
        outputs := helper.Duplicate(input, 2)

        outputs[0] = helper.Buffered(outputs[0], 4)
        outputs[1] = helper.Buffered(outputs[1], 4)

        fmt.Println(helper.ChanToSlice(outputs[0]))
        fmt.Println(helper.ChanToSlice(outputs[1]))
}

I am very much interested in learning more about your use case.

[chentiangang]

I noticed in your rsi_strategy.go code:


// Report processes the provided asset snapshots and generates a report annotated with the recommended actions.
func (r *RsiStrategy) Report(c <-chan *asset.Snapshot) *helper.Report {
	//
	// snapshots[0] -> dates
	// snapshots[1] -> Compute     -> actions -> annotations
	// snapshots[2] -> closings[0] -> close
	//              -> closings[1] -> Rsi.Compute -> rsi
	//
	snapshots := helper.Duplicate(c, 3)
	
	dates := asset.SnapshotsAsDates(snapshots[0])
	closings := helper.Duplicate(asset.SnapshotsAsClosings(snapshots[2]), 2)
	rsi := helper.Shift(r.Rsi.Compute(closings[1]), r.Rsi.IdlePeriod(), 0)
	
	actions, outcomes := strategy.ComputeWithOutcome(r, snapshots[1])
	annotations := strategy.ActionsToAnnotations(actions)
	outcomes = helper.MultiplyBy(outcomes, 100)
	
	report := helper.NewReport(r.Name(), dates)
	report.AddChart()
	report.AddChart()
	
	report.AddColumn(helper.NewNumericReportColumn("Close", closings[0]))
	report.AddColumn(helper.NewNumericReportColumn("RSI", rsi), 1)
	report.AddColumn(helper.NewAnnotationReportColumn(annotations), 0, 1)
	
	report.AddColumn(helper.NewNumericReportColumn("Outcome", outcomes), 2)
	
	return report
}

I’m curious about why it’s possible to use snapshots without strictly following a sequential order here. Additionally, if I don’t know the exact buffer size needed, what would be a recommended approach to avoid potential deadlocks?

Thank you for your insights!