remove locks on series

Author: anyongjin

core.go

@@ -94,12 +94,10 @@ func (e *BarEnv) NewSeries(data []float64) *Series {
 	xlogs := make(map[int]*CrossLog)
 	res := e.newSeries(data, nil, nil, nil, subs, xlogs)
 	e.VNum += 1
-	e.LockItems.Lock()
 	if e.Items == nil {
 		e.Items = make(map[int]*Series)
 	}
 	e.Items[res.ID] = res
-	e.LockItems.Unlock()
 	return res
 }
 
@@ -112,19 +110,15 @@ func (e *BarEnv) newSeries(data []float64, cols []*Series, more interface{}, dup
 		xlogs = make(map[int]*CrossLog)
 	}
 	res := &Series{
-		ID:         e.VNum,
-		Env:        e,
-		Data:       data,
-		Cols:       cols,
-		Time:       e.TimeStart,
-		More:       more,
-		DupMore:    dupMore,
-		Subs:       subs,
-		XLogs:      xlogs,
-		LockSubMap: make(map[string]*sync.Mutex),
-	}
-	for fn := range res.Subs {
-		res.LockSubMap[fn] = &sync.Mutex{}
+		ID:      e.VNum,
+		Env:     e,
+		Data:    data,
+		Cols:    cols,
+		Time:    e.TimeStart,
+		More:    more,
+		DupMore: dupMore,
+		Subs:    subs,
+		XLogs:   xlogs,
 	}
 	return res
 }
@@ -170,9 +164,7 @@ func (e *BarEnv) Clone() *BarEnv {
 	if e.Info != nil {
 		res.Info = e.Info.CopyTo(res)
 	}
-	e.LockItems.RLock()
 	itemList := maps.Values(e.Items)
-	e.LockItems.RUnlock()
 	for v := range itemList {
 		v.CopyTo(res)
 	}
@@ -189,7 +181,6 @@ func (e *BarEnv) ResetTo(env *BarEnv) {
 		env.Volume.ID: true,
 		env.Info.ID:   true,
 	}
-	env.LockItems.Lock()
 	var items = make([]*Series, 0, len(env.Items))
 	for id, s := range env.Items {
 		if _, ok := rootIds[id]; ok {
@@ -198,7 +189,6 @@ func (e *BarEnv) ResetTo(env *BarEnv) {
 		delete(e.Items, id)
 		items = append(items, s)
 	}
-	env.LockItems.Unlock()
 	for _, s := range items {
 		s.CopyTo(e)
 	}
@@ -214,11 +204,9 @@ func (s *Series) Set(obj interface{}) *Series {
 	if s.Cached() {
 		return s
 	}
-	s.LockData.Lock()
 	if !s.Cached() {
 		s.Append(obj)
 	}
-	s.LockData.Unlock()
 	return s
 }
 
@@ -335,11 +323,9 @@ func (s *Series) Add(obj interface{}) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(s.Get(0) + val)
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -348,11 +334,9 @@ func (s *Series) Sub(obj interface{}) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(s.Get(0) - val)
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -361,11 +345,9 @@ func (s *Series) Mul(obj interface{}) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(s.Get(0) * val)
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -374,11 +356,9 @@ func (s *Series) Div(obj interface{}) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(s.Get(0) / val)
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -387,11 +367,9 @@ func (s *Series) Min(obj interface{}) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(math.Min(s.Get(0), val))
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -400,11 +378,9 @@ func (s *Series) Max(obj interface{}) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(math.Max(s.Get(0), val))
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -413,11 +389,9 @@ func (s *Series) Abs() *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		res.Append(math.Abs(s.Get(0)))
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -429,23 +403,21 @@ func (s *Series) Len() int {
 }
 
 func (s *Series) Cut(keepNum int) {
-	s.LockSub.Lock()
+	curLen := len(s.Data)
+	if curLen <= keepNum {
+		return
+	}
 	for _, dv := range s.Subs {
 		for _, v := range dv {
 			v.Cut(keepNum)
 		}
 	}
-	s.LockSub.Unlock()
 	if len(s.Cols) > 0 {
 		for _, col := range s.Cols {
 			col.Cut(keepNum)
 		}
 		return
 	}
-	curLen := len(s.Data)
-	if curLen <= keepNum {
-		return
-	}
 	s.Data = s.Data[curLen-keepNum:]
 }
 
@@ -454,7 +426,6 @@ func (s *Series) Back(num int) *Series {
 	if res.Cached() {
 		return res
 	}
-	res.LockData.Lock()
 	if !res.Cached() {
 		endPos := len(s.Data) - num
 		if endPos > 0 {
@@ -464,7 +435,6 @@ func (s *Series) Back(num int) *Series {
 		}
 		res.Time = s.Env.TimeStop
 	}
-	res.LockData.Unlock()
 	return res
 }
 
@@ -488,58 +458,43 @@ func (s *Series) objVal(rel string, obj interface{}) (*Series, float64) {
 }
 
 func (s *Series) To(k string, v int) *Series {
-	s.LockSub.Lock()
 	sub, _ := s.Subs[k]
 	if sub == nil {
 		sub = make(map[int]*Series)
 		s.Subs[k] = sub
 	}
-	lock, ok := s.LockSubMap[k]
-	if !ok {
-		lock = &sync.Mutex{}
-		s.LockSubMap[k] = lock
-	}
-	s.LockSub.Unlock()
-	lock.Lock()
 	old, _ := sub[v]
 	if old == nil {
 		old = s.Env.NewSeries(nil)
 		sub[v] = old
 	}
-	lock.Unlock()
 	return old
 }
 
 func (s *Series) CopyTo(e *BarEnv) *Series {
 	if e == nil {
 		e = s.Env
 	}
-	e.LockItems.Lock()
 	if e.Items == nil {
 		e.Items = make(map[int]*Series)
 	}
 	old, ok := e.Items[s.ID]
-	e.LockItems.Unlock()
 	if ok {
 		return old
 	}
 	cols := make([]*Series, len(s.Cols))
 	for i, v := range s.Cols {
 		cols[i] = v.CopyTo(e)
 	}
-	s.LockSub.Lock()
 	subs := maps.Clone(s.Subs)
-	s.LockSub.Unlock()
 	for fn, idMap := range subs {
 		sub := make(map[int]*Series)
 		for id, v := range idMap {
 			sub[id] = v.CopyTo(e)
 		}
 		subs[fn] = sub
 	}
-	s.LockXLogs.Lock()
 	xlogs := maps.Clone(s.XLogs)
-	s.LockXLogs.Unlock()
 	for id, v := range xlogs {
 		xlogs[id] = v.Clone()
 	}
@@ -548,16 +503,12 @@ func (s *Series) CopyTo(e *BarEnv) *Series {
 	if s.DupMore != nil && s.More != nil {
 		res.More = s.DupMore(s.More)
 	}
-	e.LockItems.Lock()
 	e.Items[s.ID] = res
-	e.LockItems.Unlock()
 	return res
 }
 
 func (s *Series) loadEnvSubs() {
-	s.Env.LockItems.RLock()
 	envItems := maps.Clone(s.Env.Items)
-	s.Env.LockItems.RUnlock()
 	for _, idMap := range s.Subs {
 		for id := range idMap {
 			dup, ok := envItems[id]
@@ -594,7 +545,6 @@ func (s *Series) Cross(obj2 interface{}) int {
 	}
 	var newData = false
 	var log *CrossLog
-	s.LockXLogs.Lock()
 	if val, ok := s.XLogs[key]; ok {
 		log = val
 		if env.TimeStart > log.Time {
@@ -606,7 +556,6 @@ func (s *Series) Cross(obj2 interface{}) int {
 		log = &CrossLog{env.TimeStart, math.NaN(), []*XState{}}
 		s.XLogs[key] = log
 	}
-	s.LockXLogs.Unlock()
 	if newData {
 		diffVal := s.Get(0) - v2
 		if diffVal != 0 && !math.IsNaN(diffVal) {

core_test.go

@@ -152,21 +152,23 @@ func TestConcurrent(t *testing.T) {
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			// 测试基本读取操作
+			// 测试基本读取操作（读取不需要锁）
 			_ = testEnv.Close.Get(0)
 			_ = testEnv.High.Get(1)
 			_ = testEnv.Low.Get(2)
 
-			// 测试计算操作
+			// 并发计算操作需要手动加锁
+			testEnv.Lock.Lock()
 			_ = testEnv.Close.Add(100).Get(0)
 			_ = testEnv.Close.Sub(50).Get(0)
 			_ = testEnv.Close.Mul(1.1).Get(0)
 			_ = testEnv.High.Sub(testEnv.Low).Abs().Get(0)
 
 			// 测试交叉计算
 			_ = testEnv.Close.Cross(30000)
+			testEnv.Lock.Unlock()
 
-			// 测试范围操作
+			// 测试范围操作（只读不需要锁）
 			_ = testEnv.Close.Range(0, 5)
 		}()
 	}

develop.md

@@ -1,10 +1,21 @@
 
-### 支持并发操作BarEnv和Series
-为避免额外冗余创建BarEnv和Series，多策略和多账户会复用，多策略和多账户需支持并发调用，所以banta也需支持并发。
-测试给Series的Subs和XLogs，BarEnv的Items和Data改为sync.Map后，性能相比RWMutex+map大幅下降；故保留RWMutex方案。
+### 并发操作BarEnv和Series
+为避免额外冗余创建BarEnv和Series，多策略和多账户会复用，多策略和多账户需支持并发调用。
+
+原方案使用RWMutex保护Series各方法，但会带来43%的性能损失。现改为在BarEnv中只保留一个通用锁`Lock`，由用户手动调用。
+
+原因：banta的相关函数调用通常是密集且相邻的，所以一个锁就足矣，需要并发时可加锁保护banta代码块。
+
+使用示例：
+```go
+env.Lock.Lock()
+// 进行banta运算
+result := banta.EMA(env.Close, 20).Get(0)
+env.Lock.Unlock()
+```
 
 | 方案 | 耗时 | 内存 | 分配次数 | 性能变化 |
-|------|------|------|---------|---------|
+|------|------|------|---------|---------| 
 | 无锁 | 21ms | 10.8MB | 327K | 基准 |
 | RWMutex | 30ms | 10.8MB | 327K | +43% |
 | sync.Map | 59ms | 23MB | 708K | +181% |