From f519f803842203aa9b0c46e38b7ea3d284c821d6 Mon Sep 17 00:00:00 2001 From: "duanyi.aster" Date: Sun, 11 Jul 2021 14:16:11 +0800 Subject: [PATCH] fix: truely copy _ProgramMap when RCU write --- decoder/pools.go | 11 ++-- internal/caching/pcache.go | 16 +++++- internal/caching/pcache_test.go | 55 ++++++++++++++++++++ issue45_test.go | 90 +++++++++++++++++++++++++++++++++ 4 files changed, 166 insertions(+), 6 deletions(-) create mode 100644 internal/caching/pcache_test.go create mode 100644 issue45_test.go diff --git a/decoder/pools.go b/decoder/pools.go index f054b68..c53ad40 100644 --- a/decoder/pools.go +++ b/decoder/pools.go @@ -36,10 +36,11 @@ const ( ) var ( - stackPool = sync.Pool{} - valueCache = []unsafe.Pointer(nil) - fieldCache = []*caching.FieldMap(nil) - programCache = caching.CreateProgramCache() + stackPool = sync.Pool{} + valueCache = []unsafe.Pointer(nil) + fieldCache = []*caching.FieldMap(nil) + fieldCacheMux = sync.Mutex{} + programCache = caching.CreateProgramCache() ) type _Stack struct { @@ -75,7 +76,9 @@ func freezeValue(v unsafe.Pointer) uintptr { } func freezeFields(v *caching.FieldMap) int64 { + fieldCacheMux.Lock() fieldCache = append(fieldCache, v) + fieldCacheMux.Unlock() return referenceFields(v) } diff --git a/internal/caching/pcache.go b/internal/caching/pcache.go index a9ec4e7..a42ec05 100644 --- a/internal/caching/pcache.go +++ b/internal/caching/pcache.go @@ -50,6 +50,18 @@ func newProgramMap() *_ProgramMap { } } +func (self *_ProgramMap) copy() *_ProgramMap { + fork := &_ProgramMap{ + n: self.n, + m: self.m, + b: make([]_ProgramEntry, len(self.b)), + } + for i, f := range self.b { + fork.b[i] = f + } + return fork +} + func (self *_ProgramMap) get(vt *rt.GoType) interface{} { i := self.m + 1 p := vt.Hash & self.m @@ -70,12 +82,12 @@ func (self *_ProgramMap) get(vt *rt.GoType) interface{} { } func (self *_ProgramMap) add(vt *rt.GoType, fn interface{}) *_ProgramMap { - p := self + p := self.copy() f := float64(atomic.LoadUint64(&p.n) + 1) / float64(p.m + 1) /* check for load factor */ if f > _LoadFactor { - p = self.rehash() + p = p.rehash() } /* insert the value */ diff --git a/internal/caching/pcache_test.go b/internal/caching/pcache_test.go new file mode 100644 index 0000000..688b8a4 --- /dev/null +++ b/internal/caching/pcache_test.go @@ -0,0 +1,55 @@ +/* + * Copyright 2021 ByteDance Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package caching + +import ( + `sync` + `testing` + + `github.com/bytedance/sonic/internal/rt` +) + +func TestPcacheRace(t *testing.T) { + t.Parallel() + + pc := CreateProgramCache() + wg := sync.WaitGroup{} + wg.Add(2) + start := make(chan struct{}, 2) + + go func(){ + defer wg.Done() + var k = map[string]interface{}{} + <- start + for i:=0; i<100; i++ { + pc.Put(rt.UnpackEface(k).Type, k) + } + }() + + go func(){ + defer wg.Done() + var k = map[string]interface{}{} + <- start + for i:=0; i<100; i++ { + pc.Get(rt.UnpackEface(k).Type) + } + }() + + start <- struct{}{} + start <- struct{}{} + wg.Wait() +} \ No newline at end of file diff --git a/issue45_test.go b/issue45_test.go new file mode 100644 index 0000000..51c56ea --- /dev/null +++ b/issue45_test.go @@ -0,0 +1,90 @@ +/* + * Copyright 2021 ByteDance Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package sonic + +import ( + `fmt` + `sync` + `testing` +) + +func ExtractJson(idx int, body interface{}) { + j := []byte(exampleJson[idx]) + + // REPLACE sonic WITH json + err := Unmarshal(j, &body) + if err != nil { + fmt.Println(err.Error()) + return + } +} + +var exampleJson1 = `{}` + +var exampleJson2 = `{}` + +var exampleJson = []string{exampleJson1, exampleJson2} + +type raceTestStruct struct { + f1 *[]raceTestStruct2 `json:"f1"` + f2 *int `json:"f2"` + f3 *string `json:"f3"` + f4 *string `json:"f4"` +} +type raceTestStruct2 struct { + g1 *string `json:"g1"` + g2 *string `json:"g2"` + g3 *string `json:"g3"` + g4 []raceTestStruct3 `json:"g4"` +} +type raceTestStruct3 struct { + e1 *string `json:"e1"` + e2 *string `json:"e2"` + e3 *float64 `json:"e3"` + e4 *float64 `json:"e4"` +} + +func TestExtracJson(t *testing.T) { + + wg := sync.WaitGroup{} + + resultChan := make(chan raceTestStruct, 2) + + wg.Add(1) + go func() { + defer wg.Done() + var model raceTestStruct + ExtractJson(0, &model) + resultChan <- model + }() + + wg.Add(1) + go func() { + defer wg.Done() + var model raceTestStruct + ExtractJson(1, &model) + resultChan <- model + }() + + var results []raceTestStruct + for i := 0; i < 2; i++ { + results = append(results, <-resultChan) + } + + wg.Wait() + close(resultChan) +} \ No newline at end of file