From 730a9322813c4df8bedc469d9d19c3e1a944d581 Mon Sep 17 00:00:00 2001 From: Evan Date: Thu, 14 May 2026 11:17:36 +0800 Subject: [PATCH 1/2] Limit PKCS11 session creation The PKCS#11 BCCSP previously fell back to OpenSession unbounded whenever the session cache was empty, allowing the number of concurrently checked out sessions to grow with the number of concurrent callers. Under high sign concurrency this surfaces from HSM/PKCS#11 libraries as CKR_SESSION_COUNT on OpenSession and CKR_DEVICE_ERROR on subsequent operations (see issue #50). Gate concurrent OpenSession calls behind a semaphore.Weighted whose weight equals sessionCacheSize, so the number of outstanding sessions is bounded. Cached sessions in sessPool intentionally do not occupy a slot: getSession reacquires its own slot when pulling one out, and returnSession releases the slot when caching, so a caller blocked on the bound is unblocked the moment another session is returned or closed. Expose the bound via a new public PKCS11Opts.SessionCacheSize field (with json/yaml/mapstructure tags) so it can be configured through the existing BCCSP factory configuration alongside Library/Label/Pin etc., with no environment variable required. A value of 0 selects the default (30); a negative value disables the bound and preserves the original unbounded code path for callers that want it. Update closeSession to release the slot only for sessions that were previously registered, guarding against double-release when an open succeeds but the subsequent Login fails and the partial session is closed before being tracked. Tests in pkcs11_test.go (TestPKCS11GetSession, TestSessionHandleCaching) exercise the bound: they fill the cache, spawn an extra getSession in a goroutine, and assert it blocks until a session is returned. Without this change those tests would fail because the goroutine would proceed straight through to createSession instead of blocking. Signed-off-by: Evan --- bccsp/pkcs11/conf.go | 20 +- bccsp/pkcs11/pkcs11.go | 69 +++-- bccsp/pkcs11/pkcs11_test.go | 244 ++++++++++++------ go.mod | 2 +- .../golang.org/x/sync/semaphore/semaphore.go | 160 ++++++++++++ vendor/modules.txt | 1 + 6 files changed, 384 insertions(+), 112 deletions(-) create mode 100644 vendor/golang.org/x/sync/semaphore/semaphore.go diff --git a/bccsp/pkcs11/conf.go b/bccsp/pkcs11/conf.go index a6604eb..8787238 100644 --- a/bccsp/pkcs11/conf.go +++ b/bccsp/pkcs11/conf.go @@ -21,15 +21,17 @@ type PKCS11Opts struct { Hash string `json:"hash"` // PKCS11 options - Library string `json:"library"` - Label string `json:"label"` - Pin string `json:"pin"` - SoftwareVerify bool `json:"softwareverify,omitempty"` - Immutable bool `json:"immutable,omitempty"` - AltID string `json:"altid,omitempty"` - KeyIDs []KeyIDMapping `json:"keyids,omitempty" mapstructure:"keyids"` - - sessionCacheSize int + Library string `json:"library"` + Label string `json:"label"` + Pin string `json:"pin"` + SoftwareVerify bool `json:"softwareverify,omitempty"` + Immutable bool `json:"immutable,omitempty"` + AltID string `json:"altid,omitempty"` + KeyIDs []KeyIDMapping `json:"keyids,omitempty" mapstructure:"keyids"` + SessionCacheSize uint `json:"session_cache_size,omitempty"` + + // createSessionRetries / createSessionRetryDelay are internal overrides + // intended for tests. createSessionRetries int createSessionRetryDelay time.Duration } diff --git a/bccsp/pkcs11/pkcs11.go b/bccsp/pkcs11/pkcs11.go index a55f63c..3487e97 100644 --- a/bccsp/pkcs11/pkcs11.go +++ b/bccsp/pkcs11/pkcs11.go @@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0 package pkcs11 import ( + "context" "crypto/ecdsa" "crypto/elliptic" "crypto/sha256" @@ -26,6 +27,7 @@ import ( "github.com/miekg/pkcs11" "github.com/pkg/errors" "go.uber.org/zap/zapcore" + "golang.org/x/sync/semaphore" ) var ( @@ -49,6 +51,11 @@ type Provider struct { sessLock sync.Mutex sessPool chan pkcs11.SessionHandle + // sessSem bounds the number of concurrently outstanding (checked-out) + // sessions. Cached sessions in sessPool do not hold a slot; the slot is + // released on returnSession (when the session is cached) or on + // closeSession. + sessSem *semaphore.Weighted sessions map[pkcs11.SessionHandle]struct{} cacheLock sync.RWMutex @@ -90,8 +97,9 @@ func New(opts PKCS11Opts, keyStore bccsp.KeyStore, options ...Option) (*Provider return nil, errors.Wrapf(err, "Failed initializing fallback SW BCCSP") } - if opts.sessionCacheSize == 0 { - opts.sessionCacheSize = defaultSessionCacheSize + cacheSize := int(opts.SessionCacheSize) + if cacheSize == 0 { + cacheSize = defaultSessionCacheSize } if opts.createSessionRetries == 0 { opts.createSessionRetries = defaultCreateSessionRetries @@ -100,18 +108,14 @@ func New(opts PKCS11Opts, keyStore bccsp.KeyStore, options ...Option) (*Provider opts.createSessionRetryDelay = defaultCreateSessionRetryDelay } - var sessPool chan pkcs11.SessionHandle - if opts.sessionCacheSize > 0 { - sessPool = make(chan pkcs11.SessionHandle, opts.sessionCacheSize) - } - csp := &Provider{ BCCSP: swCSP, curve: curve, getKeyIDForSKI: func(ski []byte) []byte { return ski }, createSessionRetries: opts.createSessionRetries, createSessionRetryDelay: opts.createSessionRetryDelay, - sessPool: sessPool, + sessPool: make(chan pkcs11.SessionHandle, cacheSize), + sessSem: semaphore.NewWeighted(int64(cacheSize)), sessions: map[pkcs11.SessionHandle]struct{}{}, handleCache: map[string]pkcs11.ObjectHandle{}, keyCache: map[string]bccsp.Key{}, @@ -161,7 +165,7 @@ func (csp *Provider) initialize(opts PKCS11Opts) (*Provider, error) { csp.ctx = ctx csp.pin = opts.Pin - session, err := csp.createSession() + session, err := csp.getSession() if err != nil { return nil, err } @@ -329,16 +333,37 @@ func (csp *Provider) verifyECDSA(k ecdsaPublicKey, signature, digest []byte) (bo return csp.verifyP11ECDSA(k.ski, digest, r, s, k.pub.Curve.Params().BitSize/8) } +// getSession returns a session for the caller to use. If a cached session is +// available it is returned; otherwise a new session is opened, gated by the +// sessSem semaphore so the number of concurrently outstanding sessions never +// exceeds SessionCacheSize. +// +// Slot accounting: +// - Acquire one slot up front, before either reusing a cached session or +// opening a new one. The slot represents the resulting outstanding session. +// - returnSession releases the slot when the session is successfully cached. +// - closeSession releases the slot when a known session is closed. +// +// This intentionally keeps cached sessions out of the slot count: a session +// sitting in sessPool is not in-flight and should not block a caller waiting +// for a slot. A caller that subsequently pulls that cached session out will +// reacquire its own slot. func (csp *Provider) getSession() (session pkcs11.SessionHandle, err error) { - for { - select { - case session = <-csp.sessPool: - return - default: - // cache is empty (or completely in use), create a new session - return csp.createSession() - } + if err = csp.sessSem.Acquire(context.Background(), 1); err != nil { + return 0, errors.Wrap(err, "acquire session slot") + } + + select { + case session = <-csp.sessPool: + return session, nil + default: } + + session, err = csp.createSession() + if err != nil { + csp.sessSem.Release(1) + } + return session, err } func (csp *Provider) createSession() (pkcs11.SessionHandle, error) { @@ -379,21 +404,21 @@ func (csp *Provider) closeSession(session pkcs11.SessionHandle) { } csp.sessLock.Lock() - defer csp.sessLock.Unlock() - - // purge the handle cache if the last session closes delete(csp.sessions, session) + // purge the handle cache if the last session closes if len(csp.sessions) == 0 { csp.clearCaches() } + csp.sessLock.Unlock() + + csp.sessSem.Release(1) } func (csp *Provider) returnSession(session pkcs11.SessionHandle) { select { case csp.sessPool <- session: - // returned session back to session cache + csp.sessSem.Release(1) default: - // have plenty of sessions in cache, dropping csp.closeSession(session) } } diff --git a/bccsp/pkcs11/pkcs11_test.go b/bccsp/pkcs11/pkcs11_test.go index 1480386..1a18e10 100644 --- a/bccsp/pkcs11/pkcs11_test.go +++ b/bccsp/pkcs11/pkcs11_test.go @@ -16,6 +16,8 @@ import ( "encoding/asn1" "strconv" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -89,13 +91,14 @@ func TestNew(t *testing.T) { require.Equal(t, defaultCreateSessionRetries, csp.createSessionRetries) require.Equal(t, defaultCreateSessionRetryDelay, csp.createSessionRetryDelay) require.Equal(t, defaultSessionCacheSize, cap(csp.sessPool)) + require.NotNil(t, csp.sessSem) }) t.Run("ConditionalOverride", func(t *testing.T) { opts := defaultOptions() opts.createSessionRetries = 3 opts.createSessionRetryDelay = time.Second - opts.sessionCacheSize = -1 + opts.SessionCacheSize = 7 csp, err := New(opts, ks) require.NoError(t, err) @@ -103,7 +106,8 @@ func TestNew(t *testing.T) { require.Equal(t, 3, csp.createSessionRetries) require.Equal(t, time.Second, csp.createSessionRetryDelay) - require.Nil(t, csp.sessPool) + require.Equal(t, 7, cap(csp.sessPool)) + require.NotNil(t, csp.sessSem) }) } @@ -564,7 +568,12 @@ func TestInitialize(t *testing.T) { }) t.Run("MissingPin", func(t *testing.T) { - _, err := (&Provider{}).initialize(PKCS11Opts{Library: lib, Pin: "", Label: label}) + opts := defaultOptions() + opts.Library = lib + opts.Label = label + opts.Pin = "" + + _, err := New(opts, newKeyStore(t)) require.Error(t, err) require.Contains(t, err.Error(), "Login failed: pkcs11") }) @@ -614,35 +623,56 @@ func TestCurveForSecurityLevel(t *testing.T) { func TestPKCS11GetSession(t *testing.T) { opts := defaultOptions() - opts.sessionCacheSize = 5 + opts.SessionCacheSize = 5 csp, cleanup := newProvider(t, opts) defer cleanup() - sessionCacheSize := opts.sessionCacheSize + sessionCacheSize := int(opts.SessionCacheSize) var sessions []pkcs11.SessionHandle - for i := 0; i < 3*sessionCacheSize; i++ { + for i := 0; i < sessionCacheSize; i++ { session, err := csp.getSession() require.NoError(t, err) sessions = append(sessions, session) } + require.Len(t, csp.sessions, sessionCacheSize) - // Return all sessions, should leave sessionCacheSize cached - for _, session := range sessions { + blockedSession := make(chan pkcs11.SessionHandle, 1) + blockedErr := make(chan error, 1) + go func() { + session, err := csp.getSession() + if err != nil { + blockedErr <- err + return + } + blockedSession <- session + }() + + select { + case session := <-blockedSession: csp.returnSession(session) + t.Fatal("getSession should block when all cached sessions are checked out") + case err := <-blockedErr: + require.NoError(t, err) + case <-time.After(100 * time.Millisecond): } - // Should be able to get sessionCacheSize cached sessions - sessions = nil - for i := 0; i < sessionCacheSize; i++ { - session, err := csp.getSession() + csp.returnSession(sessions[0]) + select { + case sessions[0] = <-blockedSession: + case err := <-blockedErr: require.NoError(t, err) - sessions = append(sessions, session) + case <-time.After(time.Second): + t.Fatal("getSession did not resume after a session was returned") } - // Cleanup + // Return all sessions, should leave sessionCacheSize cached. for _, session := range sessions { csp.returnSession(session) } + require.Len(t, csp.sessions, sessionCacheSize) + require.Len(t, csp.sessPool, sessionCacheSize) + + // Cleanup } func TestSessionHandleCaching(t *testing.T) { @@ -660,92 +690,72 @@ func TestSessionHandleCaching(t *testing.T) { require.Equal(t, h, privHandle) } - t.Run("SessionCacheDisabled", func(t *testing.T) { - opts := defaultOptions() - opts.sessionCacheSize = -1 - - csp, cleanup := newProvider(t, opts) - defer cleanup() - - require.Nil(t, csp.sessPool, "sessPool channel should be nil") - require.Empty(t, csp.sessions, "sessions set should be empty") - require.Empty(t, csp.handleCache, "handleCache should be empty") - - sess1, err := csp.getSession() - require.NoError(t, err) - require.Len(t, csp.sessions, 1, "expected one open session") - - sess2, err := csp.getSession() - require.NoError(t, err) - require.Len(t, csp.sessions, 2, "expected two open sessions") - - // Generate a key - k, err := csp.KeyGen(&bccsp.ECDSAP256KeyGenOpts{Temporary: false}) - require.NoError(t, err) - verifyHandleCache(t, csp, sess1, k) - require.Len(t, csp.handleCache, 2, "expected two handles in handle cache") - - csp.returnSession(sess1) - require.Len(t, csp.sessions, 1, "expected one open session") - verifyHandleCache(t, csp, sess1, k) - require.Len(t, csp.handleCache, 2, "expected two handles in handle cache") - - csp.returnSession(sess2) - require.Empty(t, csp.sessions, "expected sessions to be empty") - require.Empty(t, csp.handleCache, "expected handles to be cleared") - }) - t.Run("SessionCacheEnabled", func(t *testing.T) { opts := defaultOptions() - opts.sessionCacheSize = 1 + opts.SessionCacheSize = 1 csp, cleanup := newProvider(t, opts) defer cleanup() require.NotNil(t, csp.sessPool, "sessPool channel should not be nil") require.Equal(t, 1, cap(csp.sessPool)) + require.NotNil(t, csp.sessSem, "sessSem should not be nil") require.Len(t, csp.sessions, 1, "sessions should contain login session") require.Len(t, csp.sessPool, 1, "sessionPool should hold login session") require.Empty(t, csp.handleCache, "handleCache should be empty") - sess1, err := csp.getSession() + k, err := csp.KeyGen(&bccsp.ECDSAP256KeyGenOpts{Temporary: false}) require.NoError(t, err) - require.Len(t, csp.sessions, 1, "expected one open session (sess1 from login)") - require.Len(t, csp.sessPool, 0, "sessionPool should be empty") + require.Len(t, csp.sessions, 1, "expected one open session") + require.Len(t, csp.sessPool, 1, "sessionPool should hold returned session") - sess2, err := csp.getSession() + sess1, err := csp.getSession() require.NoError(t, err) - require.Len(t, csp.sessions, 2, "expected two open sessions (sess1 and sess2)") + require.Len(t, csp.sessions, 1, "expected one open session (sess1 from login)") require.Len(t, csp.sessPool, 0, "sessionPool should be empty") - - // Generate a key - k, err := csp.KeyGen(&bccsp.ECDSAP256KeyGenOpts{Temporary: false}) - require.NoError(t, err) verifyHandleCache(t, csp, sess1, k) require.Len(t, csp.handleCache, 2, "expected two handles in handle cache") - csp.returnSession(sess1) - require.Len(t, csp.sessions, 2, "expected two open sessions (sess2 in-use, sess1 cached)") - require.Len(t, csp.sessPool, 1, "sessionPool should have one handle (sess1)") - verifyHandleCache(t, csp, sess1, k) - require.Len(t, csp.handleCache, 2, "expected two handles in handle cache") - - csp.returnSession(sess2) - require.Len(t, csp.sessions, 1, "expected one cached session (sess1)") - require.Len(t, csp.sessPool, 1, "sessionPool should have one handle (sess1)") - require.Len(t, csp.handleCache, 2, "expected two handles in handle cache") + blockedSession := make(chan pkcs11.SessionHandle, 1) + blockedErr := make(chan error, 1) + go func() { + session, err := csp.getSession() + if err != nil { + blockedErr <- err + return + } + blockedSession <- session + }() + select { + case session := <-blockedSession: + csp.returnSession(session) + t.Fatal("getSession should block when the session cache size limit is reached") + case err := <-blockedErr: + require.NoError(t, err) + case <-time.After(100 * time.Millisecond): + } - _, err = csp.getSession() - require.NoError(t, err) + csp.returnSession(sess1) + select { + case sess1 = <-blockedSession: + case err := <-blockedErr: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("getSession did not resume after a session was returned") + } require.Len(t, csp.sessions, 1, "expected one open session (sess1)") require.Len(t, csp.sessPool, 0, "sessionPool should be empty") require.Len(t, csp.handleCache, 2, "expected two handles in handle cache") + + csp.returnSession(sess1) + require.Len(t, csp.sessions, 1, "expected one cached session") + require.Len(t, csp.sessPool, 1, "sessionPool should have one handle") }) } func TestKeyCache(t *testing.T) { opts := defaultOptions() - opts.sessionCacheSize = 1 + opts.SessionCacheSize = 1 csp, cleanup := newProvider(t, opts) defer cleanup() @@ -854,21 +864,95 @@ func TestDelegation(t *testing.T) { func TestHandleSessionReturn(t *testing.T) { opts := defaultOptions() - opts.sessionCacheSize = 5 + opts.SessionCacheSize = 5 csp, cleanup := newProvider(t, opts) defer cleanup() - // Retrieve and destroy default session created during initialization + // Pull the login session, invalidate it via a raw PKCS#11 CloseSession + // so the handle is still tracked in csp.sessions but unusable, then + // push it back through returnSession to cache the invalid handle. session, err := csp.getSession() require.NoError(t, err) - csp.closeSession(session) - - // Verify session pool is empty and place invalid session in pool - require.Empty(t, csp.sessPool, "sessionPool should be empty") - csp.returnSession(pkcs11.SessionHandle(^uint(0))) + require.NoError(t, csp.ctx.CloseSession(session)) + csp.returnSession(session) + require.Len(t, csp.sessPool, 1, "invalidated session should be cached") - // Attempt to generate key with invalid session + // KeyGen pulls the invalid handle from the pool; handleSessionReturn + // must remove it via closeSession so the pool ends up empty. _, err = csp.KeyGen(&bccsp.ECDSAP256KeyGenOpts{Temporary: false}) require.EqualError(t, err, "Failed generating ECDSA P256 key: P11: keypair generate failed [pkcs11: 0xB3: CKR_SESSION_HANDLE_INVALID]") require.Empty(t, csp.sessPool, "sessionPool should be empty") } + +// TestPKCS11SessionLimit recreates the unbounded-session-creation problem +// reported in issue #50. The Provider is configured with a small session +// cache, then many concurrent callers each acquire a session, hold it +// briefly, and return it. The test records the peak number of sessions +// outstanding (tracked in csp.sessions) over the run. +// +// Without a bound on concurrent OpenSession calls, every caller whose +// arrival finds an empty sessPool falls through to createSession() and +// opens a brand new PKCS#11 session, so the peak grows with the number +// of concurrent callers regardless of sessionCacheSize. Under high sign +// concurrency that causes the PKCS#11 token to return CKR_SESSION_COUNT +// on OpenSession and subsequent CKR_DEVICE_ERROR on operations. +// +// On the current upstream main this test fails with peak == callers +// (e.g. 25 outstanding for a sessionCacheSize=5 cap). The PR limits +// concurrent OpenSession via a semaphore.Weighted, so the same test +// passes with peak == sessionCacheSize. +func TestPKCS11SessionLimit(t *testing.T) { + // Exercise the package default so the limit under test matches what + // production callers will actually see, not a test-only override. + const ( + cacheSize = defaultSessionCacheSize + callers = 5 * defaultSessionCacheSize + holdFor = 50 * time.Millisecond + ) + + opts := defaultOptions() + csp, cleanup := newProvider(t, opts) + defer cleanup() + + countOutstanding := func() int32 { + csp.sessLock.Lock() + defer csp.sessLock.Unlock() + return int32(len(csp.sessions)) + } + + var peak int32 + recordPeak := func() { + cur := countOutstanding() + for { + p := atomic.LoadInt32(&peak) + if cur <= p || atomic.CompareAndSwapInt32(&peak, p, cur) { + return + } + } + } + + var wg sync.WaitGroup + start := make(chan struct{}) + for i := 0; i < callers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + sess, err := csp.getSession() + if err != nil { + t.Errorf("getSession: %v", err) + return + } + recordPeak() + time.Sleep(holdFor) + csp.returnSession(sess) + }() + } + close(start) + wg.Wait() + + require.LessOrEqualf(t, peak, int32(cacheSize), + "peak concurrent open PKCS#11 sessions %d exceeded sessionCacheSize %d "+ + "(unbounded createSession fall-through; see issue #50)", + peak, cacheSize) +} diff --git a/go.mod b/go.mod index 74cf6d3..cf646fc 100644 --- a/go.mod +++ b/go.mod @@ -14,6 +14,7 @@ require ( github.com/stretchr/testify v1.9.0 github.com/sykesm/zap-logfmt v0.0.4 go.uber.org/zap v1.27.0 + golang.org/x/sync v0.19.0 golang.org/x/tools v0.39.0 google.golang.org/grpc v1.79.3 ) @@ -46,7 +47,6 @@ require ( go.uber.org/multierr v1.11.0 // indirect golang.org/x/mod v0.30.0 // indirect golang.org/x/net v0.48.0 // indirect - golang.org/x/sync v0.19.0 // indirect golang.org/x/sys v0.39.0 // indirect golang.org/x/text v0.32.0 // indirect google.golang.org/protobuf v1.36.10 // indirect diff --git a/vendor/golang.org/x/sync/semaphore/semaphore.go b/vendor/golang.org/x/sync/semaphore/semaphore.go new file mode 100644 index 0000000..b618162 --- /dev/null +++ b/vendor/golang.org/x/sync/semaphore/semaphore.go @@ -0,0 +1,160 @@ +// Copyright 2017 The Go Authors. All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// Package semaphore provides a weighted semaphore implementation. +package semaphore // import "golang.org/x/sync/semaphore" + +import ( + "container/list" + "context" + "sync" +) + +type waiter struct { + n int64 + ready chan<- struct{} // Closed when semaphore acquired. +} + +// NewWeighted creates a new weighted semaphore with the given +// maximum combined weight for concurrent access. +func NewWeighted(n int64) *Weighted { + w := &Weighted{size: n} + return w +} + +// Weighted provides a way to bound concurrent access to a resource. +// The callers can request access with a given weight. +type Weighted struct { + size int64 + cur int64 + mu sync.Mutex + waiters list.List +} + +// Acquire acquires the semaphore with a weight of n, blocking until resources +// are available or ctx is done. On success, returns nil. On failure, returns +// ctx.Err() and leaves the semaphore unchanged. +func (s *Weighted) Acquire(ctx context.Context, n int64) error { + done := ctx.Done() + + s.mu.Lock() + select { + case <-done: + // ctx becoming done has "happened before" acquiring the semaphore, + // whether it became done before the call began or while we were + // waiting for the mutex. We prefer to fail even if we could acquire + // the mutex without blocking. + s.mu.Unlock() + return ctx.Err() + default: + } + if s.size-s.cur >= n && s.waiters.Len() == 0 { + // Since we hold s.mu and haven't synchronized since checking done, if + // ctx becomes done before we return here, it becoming done must have + // "happened concurrently" with this call - it cannot "happen before" + // we return in this branch. So, we're ok to always acquire here. + s.cur += n + s.mu.Unlock() + return nil + } + + if n > s.size { + // Don't make other Acquire calls block on one that's doomed to fail. + s.mu.Unlock() + <-done + return ctx.Err() + } + + ready := make(chan struct{}) + w := waiter{n: n, ready: ready} + elem := s.waiters.PushBack(w) + s.mu.Unlock() + + select { + case <-done: + s.mu.Lock() + select { + case <-ready: + // Acquired the semaphore after we were canceled. + // Pretend we didn't and put the tokens back. + s.cur -= n + s.notifyWaiters() + default: + isFront := s.waiters.Front() == elem + s.waiters.Remove(elem) + // If we're at the front and there're extra tokens left, notify other waiters. + if isFront && s.size > s.cur { + s.notifyWaiters() + } + } + s.mu.Unlock() + return ctx.Err() + + case <-ready: + // Acquired the semaphore. Check that ctx isn't already done. + // We check the done channel instead of calling ctx.Err because we + // already have the channel, and ctx.Err is O(n) with the nesting + // depth of ctx. + select { + case <-done: + s.Release(n) + return ctx.Err() + default: + } + return nil + } +} + +// TryAcquire acquires the semaphore with a weight of n without blocking. +// On success, returns true. On failure, returns false and leaves the semaphore unchanged. +func (s *Weighted) TryAcquire(n int64) bool { + s.mu.Lock() + success := s.size-s.cur >= n && s.waiters.Len() == 0 + if success { + s.cur += n + } + s.mu.Unlock() + return success +} + +// Release releases the semaphore with a weight of n. +func (s *Weighted) Release(n int64) { + s.mu.Lock() + s.cur -= n + if s.cur < 0 { + s.mu.Unlock() + panic("semaphore: released more than held") + } + s.notifyWaiters() + s.mu.Unlock() +} + +func (s *Weighted) notifyWaiters() { + for { + next := s.waiters.Front() + if next == nil { + break // No more waiters blocked. + } + + w := next.Value.(waiter) + if s.size-s.cur < w.n { + // Not enough tokens for the next waiter. We could keep going (to try to + // find a waiter with a smaller request), but under load that could cause + // starvation for large requests; instead, we leave all remaining waiters + // blocked. + // + // Consider a semaphore used as a read-write lock, with N tokens, N + // readers, and one writer. Each reader can Acquire(1) to obtain a read + // lock. The writer can Acquire(N) to obtain a write lock, excluding all + // of the readers. If we allow the readers to jump ahead in the queue, + // the writer will starve — there is always one token available for every + // reader. + break + } + + s.cur += w.n + s.waiters.Remove(next) + close(w.ready) + } +} diff --git a/vendor/modules.txt b/vendor/modules.txt index 49b7fd3..f36fd8d 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -192,6 +192,7 @@ golang.org/x/net/html/charset # golang.org/x/sync v0.19.0 ## explicit; go 1.24.0 golang.org/x/sync/errgroup +golang.org/x/sync/semaphore # golang.org/x/sys v0.39.0 ## explicit; go 1.24.0 golang.org/x/sys/unix From 65b4b8f68200bb731defcac496ba9441f6beb2a7 Mon Sep 17 00:00:00 2001 From: Evan Date: Mon, 18 May 2026 14:43:04 +0800 Subject: [PATCH 2/2] pkcs11: simplify returnSession to a direct send sessPool and sessSem now share cacheSize as capacity, so the channel send in returnSession can never block: the caller still holds an outstanding sem.Acquire(1) from its prior getSession(), which bounds acquired-but-not-returned sessions A >= 1, leaving cached L <= cacheSize-1. The select+default branch was therefore unreachable; remove it so the control flow reflects the actual invariant. Signed-off-by: Evan --- bccsp/pkcs11/pkcs11.go | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/bccsp/pkcs11/pkcs11.go b/bccsp/pkcs11/pkcs11.go index 3487e97..8aafe77 100644 --- a/bccsp/pkcs11/pkcs11.go +++ b/bccsp/pkcs11/pkcs11.go @@ -415,12 +415,8 @@ func (csp *Provider) closeSession(session pkcs11.SessionHandle) { } func (csp *Provider) returnSession(session pkcs11.SessionHandle) { - select { - case csp.sessPool <- session: - csp.sessSem.Release(1) - default: - csp.closeSession(session) - } + csp.sessPool <- session + csp.sessSem.Release(1) } // Look for an EC key by SKI, stored in CKA_ID