fix: use goid to identify differnt goroutine (#22740)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

pkg/cloudcommon/app/app.go

@@ -30,7 +30,7 @@ import (
 func InitApp(options *common_options.BaseOptions, dbAccess bool) *appsrv.Application {
 	// cache := appsrv.NewCache(options.AuthTokenCacheSize)
 	log.Infof("RequestWorkerCount: %d", options.RequestWorkerCount)
-	app := appsrv.NewApplication(options.ApplicationID, options.RequestWorkerCount, dbAccess)
+	app := appsrv.NewApplication(options.ApplicationID, options.RequestWorkerCount, options.RequestWorkerQueueSize, dbAccess)
 	app.CORSAllowHosts(options.CorsHosts)
 	app.SetDefaultTimeout(time.Duration(options.DefaultProcessTimeoutSeconds) * time.Second)
 	// app.SetContext(appsrv.APP_CONTEXT_KEY_CACHE, cache)

pkg/cloudcommon/db/lockman/inmemory.go

@@ -19,10 +19,12 @@ import (
 	"runtime/debug"
 	"sync"
 
+	"github.com/petermattis/goid"
+
 	"yunion.io/x/log"
 )
 
-const (
+var (
 	debug_log = false
 )
 
@@ -34,15 +36,15 @@ type SInMemoryLockRecord struct {
 	key    string
 	lock   *sync.Mutex
 	cond   *sync.Cond
-	holder context.Context
+	holder int64
 	depth  int
 	waiter *FIFO
 }
 
-func newInMemoryLockRecord(ctx context.Context) *SInMemoryLockRecord {
+func newInMemoryLockRecord(ctxDummy context.Context) *SInMemoryLockRecord {
 	lock := &sync.Mutex{}
 	cond := sync.NewCond(lock)
-	rec := SInMemoryLockRecord{lock: lock, cond: cond, holder: ctx, depth: 0, waiter: NewFIFO()}
+	rec := SInMemoryLockRecord{lock: lock, cond: cond, holder: -1, depth: 0, waiter: NewFIFO()}
 	return &rec
 }
 
@@ -51,24 +53,29 @@ func (rec *SInMemoryLockRecord) fatalf(fmtStr string, args ...interface{}) {
 	log.Fatalf(fmtStr, args...)
 }
 
-func (rec *SInMemoryLockRecord) lockContext(ctx context.Context) {
+func (rec *SInMemoryLockRecord) lockContext(ctxDummy context.Context) {
 	rec.lock.Lock()
 	defer rec.lock.Unlock()
 
-	if rec.holder == nil {
-		rec.holder = ctx
+	curGoid := goid.Get()
+
+	if rec.holder < 0 {
+		if debug_log {
+			log.Debugf("lockContext: curGoid=[%d] key=[%s] create new record", curGoid, rec.key)
+		}
+		rec.holder = curGoid
 		rec.depth = 1
 		return
 	}
 
 	if debug_log {
-		log.Debugf("rec.hold=[%p] ctx=[%p] %v key=[%s]", rec.holder, ctx, rec.holder == ctx, rec.key)
+		log.Debugf("rec.hold=[%d] ctx=[%d] %v key=[%s]", rec.holder, curGoid, rec.holder == curGoid, rec.key)
 	}
 
-	if rec.holder == ctx {
+	if rec.holder == curGoid {
 		rec.depth += 1
 		if debug_log {
-			log.Infof("lockContext: same ctx, depth: %d [%p] key=[%s]", rec.depth, rec.holder, rec.key)
+			log.Infof("lockContext: same ctx, depth: %d holder=[%d] ctx=[%d] key=[%s]", rec.depth, rec.holder, curGoid, rec.key)
 		}
 		if rec.depth > 32 {
 			// XXX MUST BE BUG ???
@@ -79,57 +86,59 @@ func (rec *SInMemoryLockRecord) lockContext(ctx context.Context) {
 
 	// check
 	rec.waiter.Enum(func(ele interface{}) {
-		electx := ele.(context.Context)
-		if electx == ctx {
-			rec.fatalf("try to lock from a waiter context???? key=[%s]", rec.key)
+		electx := ele.(int64)
+		if electx == curGoid {
+			rec.fatalf("try to lock from a waiter context???? curGoid=[%d] waiterGoid=[%d] key=[%s]", curGoid, electx, rec.key)
 		}
 	})
 
-	rec.waiter.Push(ctx)
+	rec.waiter.Push(curGoid)
 
 	if debug_log {
-		log.Debugf("waiter size %d after push", rec.waiter.Len())
-		log.Debugf("Start to wait ... [%p] key=[%s]", ctx, rec.key)
+		log.Debugf("waiter size %d after push curGoid=[%d]", rec.waiter.Len(), curGoid)
+		log.Debugf("Start to wait ... holder=[%d] curGoid [%d] key=[%s]", rec.holder, curGoid, rec.key)
 	}
 
-	for rec.holder != nil {
+	for rec.holder >= 0 {
 		rec.cond.Wait()
 	}
 
 	if debug_log {
-		log.Debugf("End of wait ... [%p] key=[%s]", ctx, rec.key)
+		log.Debugf("End of wait ... holder=[%d] curGoid [%d] key=[%s]", rec.holder, curGoid, rec.key)
 	}
 
-	rec.waiter.Pop(ctx)
+	rec.waiter.Pop(curGoid)
 
 	if debug_log {
-		log.Debugf("waiter size %d after pop key=[%s]", rec.waiter.Len(), rec.key)
+		log.Debugf("waiter size %d after pop curGoid=[%d] key=[%s]", rec.waiter.Len(), curGoid, rec.key)
 	}
 
-	rec.holder = ctx
+	rec.holder = curGoid
 	rec.depth = 1
 }
 
-func (rec *SInMemoryLockRecord) unlockContext(ctx context.Context) (needClean bool) {
+func (rec *SInMemoryLockRecord) unlockContext(ctxDummy context.Context) (needClean bool) {
 	rec.lock.Lock()
 	defer rec.lock.Unlock()
 
-	if rec.holder != ctx {
-		rec.fatalf("try to unlock a wait context??? key=[%s]", rec.key)
+	curGoid := goid.Get()
+
+	if rec.holder != curGoid {
+		rec.fatalf("try to unlock a wait context??? key=[%s] holder=[%d] curGoid=[%d]", rec.key, rec.holder, curGoid)
 	}
 
 	if debug_log {
-		log.Debugf("unlockContext depth %d [%p] key=[%s]", rec.depth, ctx, rec.key)
+		log.Debugf("unlockContext depth %d curGoid=[%d] key=[%s]", rec.depth, curGoid, rec.key)
 	}
 
 	rec.depth -= 1
 
 	if rec.depth <= 0 {
 		if debug_log {
-			log.Debugf("depth 0, to release lock for context [%p] key=[%s]", ctx, rec.key)
+			log.Debugf("depth 0, to release lock for context curGoid=[%d] key=[%s]", curGoid, rec.key)
 		}
 
-		rec.holder = nil
+		rec.holder = -1
 		if rec.waiter.Len() == 0 {
 			return true
 		}
@@ -154,11 +163,11 @@ func NewInMemoryLockManager() ILockManager {
 	return &lockMan
 }
 
-func (lockman *SInMemoryLockManager) getRecordWithLock(ctx context.Context, key string) *SInMemoryLockRecord {
+func (lockman *SInMemoryLockManager) getRecordWithLock(ctx context.Context, key string, new bool) *SInMemoryLockRecord {
 	lockman.tableLock.Lock()
 	defer lockman.tableLock.Unlock()
 
-	return lockman.getRecord(ctx, key, true)
+	return lockman.getRecord(ctx, key, new)
 }
 
 func (lockman *SInMemoryLockManager) getRecord(ctx context.Context, key string, new bool) *SInMemoryLockRecord {
@@ -175,23 +184,16 @@ func (lockman *SInMemoryLockManager) getRecord(ctx context.Context, key string,
 }
 
 func (lockman *SInMemoryLockManager) LockKey(ctx context.Context, key string) {
-	record := lockman.getRecordWithLock(ctx, key)
-
+	record := lockman.getRecordWithLock(ctx, key, true)
 	record.lockContext(ctx)
 }
 
 func (lockman *SInMemoryLockManager) UnlockKey(ctx context.Context, key string) {
-	lockman.tableLock.Lock()
-	defer lockman.tableLock.Unlock()
-
-	record := lockman.getRecord(ctx, key, false)
+	record := lockman.getRecordWithLock(ctx, key, false)
 	if record == nil {
-		log.Errorf("BUG: unlock an non-existent lock\n%s", debug.Stack())
+		log.Errorf("BUG: unlock an non-existent lock ctx: %p key: %s\n%s", ctx, key, debug.Stack())
 		return
 	}
 
-	needClean := record.unlockContext(ctx)
-	if needClean {
-		delete(lockman.lockTable, key)
-	}
+	record.unlockContext(ctx)
 }

pkg/cloudcommon/db/lockman/inmemory_test.go

@@ -16,9 +16,17 @@ package lockman
 
 import (
 	"context"
+	"fmt"
 	"math/rand"
+	"os"
+	"sync"
 	"testing"
 	"time"
+
+	"github.com/petermattis/goid"
+
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/utils"
 )
 
 func TestInMemoryLockManager(t *testing.T) {
@@ -56,3 +64,74 @@ func TestRunManu(t *testing.T) {
 	}
 	t.Logf("complete")
 }
+
+type app struct {
+	ctx     context.Context
+	key     string
+	lockman ILockManager
+}
+
+func (app *app) run() {
+	app.lockman.LockKey(app.ctx, app.key)
+	defer app.lockman.UnlockKey(app.ctx, app.key)
+
+	fmt.Printf("run for goid: %d key %s\n", goid.Get(), app.key)
+}
+
+type emptyKey struct{}
+
+func TestRunManu3(t *testing.T) {
+	for i := 0; i < 100; i++ {
+		TestRunManu2(t)
+	}
+}
+
+func TestRunManu2(t *testing.T) {
+	rand.Seed(100)
+	lockman := NewInMemoryLockManager()
+
+	debug_log = true
+
+	// 使用 WaitGroup 来等待所有 goroutine 完成
+	var wg sync.WaitGroup
+	MAX_GOROUTINE := 4096
+	MAX_KEY := 4
+
+	complete := make(chan struct{})
+
+	go func() {
+		for {
+			select {
+			case <-time.After(time.Second):
+				t.Logf("timeout")
+				utils.DumpAllGoroutineStack(os.Stdout)
+			case <-complete:
+				return
+			}
+		}
+	}()
+
+	bgCtx := context.Background()
+
+	// 为每个 goroutine 创建独立的 context
+	for i := 0; i < MAX_GOROUTINE; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			// 为每个 goroutine 创建新的 context
+			ctx := context.WithValue(bgCtx, emptyKey{}, id)
+			log.Infof("ctx for id %d: %p", id, ctx)
+			app := &app{
+				ctx:     ctx,
+				key:     fmt.Sprintf("test-%d", id%MAX_KEY),
+				lockman: lockman,
+			}
+			app.run()
+		}(i)
+	}
+
+	// 等待所有 goroutine 完成
+	wg.Wait()
+	close(complete)
+	t.Logf("complete")
+}

pkg/cloudcommon/options/options.go

@@ -71,6 +71,8 @@ type BaseOptions struct {
 	ApplicationID      string `help:"Application ID"`
 	RequestWorkerCount int    `default:"8" help:"Request worker thread count, default is 8"`
 
+	RequestWorkerQueueSize int `default:"10" help:"Request worker queue size, default is 10"`
+
 	TaskWorkerCount      int `default:"4" help:"Task manager worker thread count, default is 4"`
 	LocalTaskWorkerCount int `default:"4" help:"Worker thread count that runs local tasks, default is 4"`
 

pkg/cloudcommon/policy/policy.go

@@ -31,12 +31,12 @@ import (
 
 	"yunion.io/x/onecloud/pkg/apis"
 	identity_api "yunion.io/x/onecloud/pkg/apis/identity"
-	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/cloudcommon/consts"
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/auth"
 	"yunion.io/x/onecloud/pkg/mcclient/modules/identity"
+	"yunion.io/x/onecloud/pkg/util/ctx"
 	"yunion.io/x/onecloud/pkg/util/hashcache"
 	"yunion.io/x/onecloud/pkg/util/rbacutils"
 	"yunion.io/x/onecloud/pkg/util/tagutils"
@@ -76,7 +76,7 @@ type SPolicyManager struct {
 	policyCache     *hashcache.Cache // policy cache
 	permissionCache *hashcache.Cache // permission cache
 
-	fetchWorker *appsrv.SWorkerManager
+	// fetchWorker *appsrv.SWorkerManager
 
 	lock *sync.Mutex
 }
@@ -146,19 +146,19 @@ func (manager *SPolicyManager) init(refreshInterval time.Duration, workerCount i
 	defaultFetcherFuncAddr := reflect.ValueOf(DefaultPolicyFetcher).Pointer()
 	remoteFetcherFuncAddr := reflect.ValueOf(auth.FetchMatchPolicies).Pointer()
 	log.Debugf("DefaultPolicyFetcher: %x RemotePolicyFetcher: %x", defaultFetcherFuncAddr, remoteFetcherFuncAddr)
-	var isDB bool
-	if defaultFetcherFuncAddr == remoteFetcherFuncAddr {
-		// remote fetcher, so start watcher
-		isDB = false
-	} else {
-		isDB = true
-	}
+	// var isDB bool
+	// if defaultFetcherFuncAddr == remoteFetcherFuncAddr {
+	// remote fetcher, so start watcher
+	// isDB = false
+	// } else {
+	// 	isDB = true
+	// }
 
-	if workerCount <= 0 {
-		workerCount = 1
-	}
-	log.Infof("policy fetch worker count %d", workerCount)
-	manager.fetchWorker = appsrv.NewWorkerManager("policyFetchWorker", workerCount, 2048, isDB)
+	// if workerCount <= 0 {
+	//	workerCount = 1
+	// }
+	// log.Infof("policy fetch worker count %d", workerCount)
+	// manager.fetchWorker = appsrv.NewWorkerManager("policyFetchWorker", workerCount, 2048, isDB)
 }
 
 func getMaskedLoginIp(userCred mcclient.TokenCredential) string {
@@ -255,53 +255,22 @@ func (manager *SPolicyManager) Allow(targetScope rbacscope.TRbacScope, userCred
 	return rbacutils.PolicyDeny
 }
 
-type fetchResult struct {
-	output *mcclient.SFetchMatchPoliciesOutput
-	err    error
-}
-
-type policyTask struct {
-	manager  *SPolicyManager
-	key      string
-	userCred mcclient.TokenCredential
-	resChan  chan fetchResult
-}
-
-func (t *policyTask) Run() {
-	val := t.manager.policyCache.AtomicGet(t.key)
-	result := fetchResult{}
-	if gotypes.IsNil(val) {
-		pg, err := DefaultPolicyFetcher(context.Background(), t.userCred)
-		if err != nil {
-			result.err = errors.Wrap(err, "DefaultPolicyFetcher")
-		} else {
-			t.manager.policyCache.AtomicSet(t.key, pg)
-			result.output = pg
-		}
-	} else {
-		result.output = val.(*mcclient.SFetchMatchPoliciesOutput)
-	}
-	t.resChan <- result
-}
-
-func (t *policyTask) Dump() string {
-	return ""
-}
-
 func (manager *SPolicyManager) fetchMatchedPolicies(userCred mcclient.TokenCredential) (*mcclient.SFetchMatchPoliciesOutput, error) {
 	key := policyKey(userCred)
 
-	task := policyTask{
-		manager:  manager,
-		key:      key,
-		userCred: userCred,
+	val := manager.policyCache.AtomicGet(key)
+	if !gotypes.IsNil(val) {
+		// cache hit
+		return val.(*mcclient.SFetchMatchPoliciesOutput), nil
 	}
-	task.resChan = make(chan fetchResult)
 
-	manager.fetchWorker.Run(&task, nil, nil)
+	pg, err := DefaultPolicyFetcher(ctx.CtxWithTime(), userCred)
+	if err != nil {
+		return nil, errors.Wrap(err, "DefaultPolicyFetcher")
+	}
 
-	res := <-task.resChan
-	return res.output, res.err
+	manager.policyCache.AtomicSet(key, pg)
+	return pg, nil
 }
 
 func (manager *SPolicyManager) allow(scope rbacscope.TRbacScope, userCred mcclient.TokenCredential, service string, resource string, action string, extra ...string) rbacutils.SPolicyResult {