optimize: lockfree

cloudwego · Nov 7, 2023 · 54bef26 · 54bef26
1 parent 92ff0af
commit 54bef26
Show file tree

Hide file tree

Showing 8 changed files with 201 additions and 35 deletions.
diff --git a/internal/loader/funcdata.go b/internal/loader/funcdata.go
@@ -17,8 +17,10 @@
 package loader
 
 import (
-    `sync`
-    _ `unsafe`
+    `sync/atomic`
+    `unsafe`
+
+    `github.com/cloudwego/frugal/internal/utils`
 )
 
 const (
@@ -35,8 +37,8 @@ var lastmoduledatap *_ModuleData
 func moduledataverify1(_ *_ModuleData)
 
 var (
-    modLock sync.Mutex
-    modList []*_ModuleData
+    /* retains local reference of all modules to bypass gc */
+    modList = utils.ListNode{}
 )
 
 func toZigzag(v int) int {
@@ -73,9 +75,32 @@ func encodeVariant(v int) []byte {
 }
 
 func registerModule(mod *_ModuleData) {
-    modLock.Lock()
-    modList = append(modList, mod)
-    lastmoduledatap.next = mod
-    lastmoduledatap = mod
-    modLock.Unlock()
+    modList.Prepend(unsafe.Pointer(mod))
+    registerModuleLockFree(&lastmoduledatap, mod)
+}
+
+func registerModuleLockFree(tail **_ModuleData, mod *_ModuleData) {
+    for {
+        oldTail := loadModule(tail)
+        if casModule(tail, oldTail, mod) {
+            storeModule(&oldTail.next, mod)
+            break
+        }
+    }
+}
+
+func loadModule(p **_ModuleData) *_ModuleData {
+    return (*_ModuleData)(atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(p))))
+}
+
+func storeModule(p **_ModuleData, value *_ModuleData) {
+    atomic.StorePointer((*unsafe.Pointer)(unsafe.Pointer(p)), unsafe.Pointer(value))
+}
+
+func casModule(p **_ModuleData, oldValue *_ModuleData, newValue *_ModuleData) bool {
+    return atomic.CompareAndSwapPointer(
+        (*unsafe.Pointer)(unsafe.Pointer(p)),
+        unsafe.Pointer(oldValue),
+        unsafe.Pointer(newValue),
+    )
 }
diff --git a/internal/loader/funcdata_go116_117.go b/internal/loader/funcdata_go116_117.go
@@ -19,10 +19,10 @@
 package loader
 
 import (
-    `sync`
     `unsafe`
 
     `github.com/cloudwego/frugal/internal/rt`
+    `github.com/cloudwego/frugal/internal/utils`
 )
 
 type _Func struct {
@@ -115,16 +115,11 @@ var modHeader = &_PCHeader {
 }
 
 var (
-    emptyByte  byte
-    bucketList []*_FindFuncBucket
-    bucketLock sync.Mutex
-)
+    emptyByte byte
 
-func appendToBucketList(ftab *_FindFuncBucket) {
-    bucketLock.Lock()
-    bucketList = append(bucketList, ftab)
-    bucketLock.Unlock()
-}
+    /* retain local reference of all buckets to bypass gc */
+    bucketList = &utils.ListNode{}
+)
 
 func registerFunction(name string, pc uintptr, size uintptr, frame rt.Frame) {
     var pbase uintptr
@@ -171,7 +166,7 @@ func registerFunction(name string, pc uintptr, size uintptr, frame rt.Frame) {
     /* pin the find function bucket */
     ftab := &ffunc[0]
     pctab = append(pctab, 0)
-    appendToBucketList(ftab)
+    bucketList.Prepend(unsafe.Pointer(ftab))
 
     /* function entry */
     fn := _Func {

diff --git a/internal/loader/funcdata_go118_121.go b/internal/loader/funcdata_go118_121.go
@@ -20,10 +20,10 @@
 package loader
 
 import (
-    `sync`
     `unsafe`
 
     `github.com/cloudwego/frugal/internal/rt`
+    `github.com/cloudwego/frugal/internal/utils`
 )
 
 type _FuncTab struct {
@@ -60,16 +60,11 @@ const minfunc = 16
 const pcbucketsize = 256 * minfunc
 
 var (
-    emptyByte  byte
-    bucketList []*_FindFuncBucket
-    bucketLock sync.Mutex
-)
+    emptyByte byte
 
-func appendToBucketList(ftab *_FindFuncBucket) {
-    bucketLock.Lock()
-    bucketList = append(bucketList, ftab)
-    bucketLock.Unlock()
-}
+    /* retain local reference of all buckets to bypass gc */
+    bucketList = &utils.ListNode{}
+)
 
 func registerFunction(name string, pc uintptr, size uintptr, frame rt.Frame) {
     var pbase uintptr
@@ -111,7 +106,8 @@ func registerFunction(name string, pc uintptr, size uintptr, frame rt.Frame) {
     /* pin the find function bucket */
     ftab := &ffunc[0]
     pctab = append(pctab, 0)
-    appendToBucketList(ftab)
+    bucketList.Prepend(unsafe.Pointer(ftab))
+
 
     /* pin the pointer maps */
     argptrs := frame.ArgPtrs.Pin()

diff --git a/internal/loader/funcdata_test.go b/internal/loader/funcdata_test.go
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2023 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 loader
+
+import (
+    "testing"
+    "sync"
+)
+
+func Test_registerModuleLockFree(t *testing.T) {
+    n, parallel := 1000, 8
+    head := _ModuleData{}
+    tail := &head
+    wg := sync.WaitGroup{}
+    wg.Add(parallel)
+    filler := func(n int) {
+        defer wg.Done()
+        for i := 0; i < n; i++ {
+            m := &_ModuleData{}
+            registerModuleLockFree(&tail, m)
+        }
+    }
+    for i := 0; i < parallel; i++ {
+        go filler(n)
+    }
+    wg.Wait()
+    i := 0
+    for p := head.next; p != nil; p = p.next {
+        i += 1
+    }
+    if i != parallel * n {
+        t.Errorf("got %v, expected %v", i, parallel * n)
+    }
+}
diff --git a/internal/loader/loader_amd64_test.go b/internal/loader/loader_amd64_test.go
@@ -122,14 +122,18 @@ func TestLoader_StackMap(t *testing.T) {
     require.NoError(t, asm.Assemble(src))
     smb.AddField(true)
     cc := asm.Code()
+    smb1 := new(rt.StackMapBuilder)
+    smb1.AddField(false)
     fp := Loader(cc).Load("test_with_stackmap", rt.Frame {
         SpTab: []rt.Stack {
             { Sp:  0, Nb: 4 },
             { Sp: 24, Nb: uintptr(len(cc) - 5) },
             { Sp:  0, Nb: 0 },
         },
         ArgSize   : 0,
-        ArgPtrs   : new(rt.StackMapBuilder).Build(),
+        ArgPtrs   : smb1.Build(), // Build() should be called on non-empty Builder,
+                                  // otherwise mallocgc will allocate less than enough
+                                  // which leads to fatal error when run with -race
         LocalPtrs : smb.Build(),
     })
     dumpfunction(*(*func())(unsafe.Pointer(&fp)))

diff --git a/internal/utils/list.go b/internal/utils/list.go
@@ -0,0 +1,47 @@
+/*
+ * Copyright 2023 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 utils
+
+import (
+    `sync/atomic`
+    `unsafe`
+)
+
+/* ListNode can be used to save references */
+type ListNode struct{
+    value unsafe.Pointer
+    next  *ListNode
+}
+
+/* Prepend creates a new node with value=p and adds it at the beginning of this list */
+func (n *ListNode) Prepend(p unsafe.Pointer) {
+    for {
+        oldNext := atomic.LoadPointer((*unsafe.Pointer)(unsafe.Pointer(&n.next)))
+        newNode := &ListNode{
+            value: p,
+            next: (*ListNode)(oldNext),
+        }
+        success := atomic.CompareAndSwapPointer(
+            (*unsafe.Pointer)(unsafe.Pointer(&n.next)),
+            oldNext,
+            unsafe.Pointer(newNode),
+        )
+        if success {
+            break
+        }
+    }
+}
diff --git a/internal/utils/list_test.go b/internal/utils/list_test.go
@@ -0,0 +1,48 @@
+/*
+ * Copyright 2023 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 utils
+
+import (
+    "testing"
+    "sync"
+    "unsafe"
+)
+
+func Test_ListNode_Prepend(t *testing.T) {
+    n, parallel := 1000, 8
+    head := ListNode{}
+    wg := sync.WaitGroup{}
+    wg.Add(parallel)
+    filler := func(n int) {
+        defer wg.Done()
+        for i := 0; i < n; i++ {
+            v := i
+            head.Prepend((unsafe.Pointer)(&v))
+        }
+    }
+    for i := 0; i < parallel; i++ {
+        go filler(n)
+    }
+    wg.Wait()
+    i := 0
+    for p := head.next; p != nil; p = p.next {
+        i += 1
+    }
+    if i != parallel * n {
+        t.Errorf("got %v, expected %v", i, parallel * n)
+    }
+}
diff --git a/tests/allsize_test.go b/tests/allsize_test.go
@@ -81,8 +81,8 @@ var samples []Sample
 var (
 	bytesCount  = 16
 	stringCount = 16
-	listCount       = 8
-	mapCount        = 8
+	listCount   = 8
+	mapCount    = 8
 )
 
 func getSamples() []Sample {
@@ -575,7 +575,10 @@ func BenchmarkUnmarshalAllSize_Parallel_Frugal(b *testing.B) {
 				var v = reflect.New(rtype).Interface()
 				for pb.Next() {
 					objectmemclr(v)
-					_, err = frugal.DecodeObject(buf, v)
+					// new error object to avoid concurrent write to `err` which fails with -race
+					if _, errDecode := frugal.DecodeObject(buf, v); errDecode != nil {
+						b.Fatal(errDecode)
+					}
 				}
 			})
 		})