Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:27
📄 semaphore.go // 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) } }	4.26 KB	2025-05-25 12:27
📄 semaphore_bench_test.go // 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_test import ( "context" "fmt" "testing" "golang.org/x/sync/semaphore" ) // weighted is an interface matching a subset of Weighted. It allows // alternate implementations for testing and benchmarking. type weighted interface { Acquire(context.Context, int64) error TryAcquire(int64) bool Release(int64) } // semChan implements Weighted using a channel for // comparing against the condition variable-based implementation. type semChan chan struct{} func newSemChan(n int64) semChan { return semChan(make(chan struct{}, n)) } func (s semChan) Acquire(_ context.Context, n int64) error { for i := int64(0); i < n; i++ { s <- struct{}{} } return nil } func (s semChan) TryAcquire(n int64) bool { if int64(len(s))+n > int64(cap(s)) { return false } for i := int64(0); i < n; i++ { s <- struct{}{} } return true } func (s semChan) Release(n int64) { for i := int64(0); i < n; i++ { <-s } } // acquireN calls Acquire(size) on sem N times and then calls Release(size) N times. func acquireN(b testing.B, sem weighted, size int64, N int) { b.ResetTimer() for i := 0; i < b.N; i++ { for j := 0; j < N; j++ { sem.Acquire(context.Background(), size) } for j := 0; j < N; j++ { sem.Release(size) } } } // tryAcquireN calls TryAcquire(size) on sem N times and then calls Release(size) N times. func tryAcquireN(b testing.B, sem weighted, size int64, N int) { b.ResetTimer() for i := 0; i < b.N; i++ { for j := 0; j < N; j++ { if !sem.TryAcquire(size) { b.Fatalf("TryAcquire(%v) = false, want true", size) } } for j := 0; j < N; j++ { sem.Release(size) } } } func BenchmarkNewSeq(b testing.B) { for _, cap := range []int64{1, 128} { b.Run(fmt.Sprintf("Weighted-%d", cap), func(b testing.B) { for i := 0; i < b.N; i++ { _ = semaphore.NewWeighted(cap) } }) b.Run(fmt.Sprintf("semChan-%d", cap), func(b testing.B) { for i := 0; i < b.N; i++ { _ = newSemChan(cap) } }) } } func BenchmarkAcquireSeq(b testing.B) { for _, c := range []struct { cap, size int64 N int }{ {1, 1, 1}, {2, 1, 1}, {16, 1, 1}, {128, 1, 1}, {2, 2, 1}, {16, 2, 8}, {128, 2, 64}, {2, 1, 2}, {16, 8, 2}, {128, 64, 2}, } { for _, w := range []struct { name string w weighted }{ {"Weighted", semaphore.NewWeighted(c.cap)}, {"semChan", newSemChan(c.cap)}, } { b.Run(fmt.Sprintf("%s-acquire-%d-%d-%d", w.name, c.cap, c.size, c.N), func(b testing.B) { acquireN(b, w.w, c.size, c.N) }) b.Run(fmt.Sprintf("%s-tryAcquire-%d-%d-%d", w.name, c.cap, c.size, c.N), func(b *testing.B) { tryAcquireN(b, w.w, c.size, c.N) }) } } }	2.78 KB	2025-05-25 12:27
📄 semaphore_example_test.go // 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_test import ( "context" "fmt" "log" "runtime" "golang.org/x/sync/semaphore" ) // Example_workerPool demonstrates how to use a semaphore to limit the number of // goroutines working on parallel tasks. // // This use of a semaphore mimics a typical “worker pool” pattern, but without // the need to explicitly shut down idle workers when the work is done. func Example_workerPool() { ctx := context.TODO() var ( maxWorkers = runtime.GOMAXPROCS(0) sem = semaphore.NewWeighted(int64(maxWorkers)) out = make([]int, 32) ) // Compute the output using up to maxWorkers goroutines at a time. for i := range out { // When maxWorkers goroutines are in flight, Acquire blocks until one of the // workers finishes. if err := sem.Acquire(ctx, 1); err != nil { log.Printf("Failed to acquire semaphore: %v", err) break } go func(i int) { defer sem.Release(1) out[i] = collatzSteps(i + 1) }(i) } // Acquire all of the tokens to wait for any remaining workers to finish. // // If you are already waiting for the workers by some other means (such as an // errgroup.Group), you can omit this final Acquire call. if err := sem.Acquire(ctx, int64(maxWorkers)); err != nil { log.Printf("Failed to acquire semaphore: %v", err) } fmt.Println(out) // Output: // [0 1 7 2 5 8 16 3 19 6 14 9 9 17 17 4 12 20 20 7 7 15 15 10 23 10 111 18 18 18 106 5] } // collatzSteps computes the number of steps to reach 1 under the Collatz // conjecture. (See https://en.wikipedia.org/wiki/Collatz_conjecture.) func collatzSteps(n int) (steps int) { if n <= 0 { panic("nonpositive input") } for ; n > 1; steps++ { if steps < 0 { panic("too many steps") } if n%2 == 0 { n /= 2 continue } const maxInt = int(^uint(0) >> 1) if n > (maxInt-1)/3 { panic("overflow") } n = 3*n + 1 } return steps }	2 KB	2025-05-25 12:27
📄 semaphore_test.go // 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_test import ( "context" "math/rand" "runtime" "sync" "testing" "time" "golang.org/x/sync/errgroup" "golang.org/x/sync/semaphore" ) const maxSleep = 1 * time.Millisecond func HammerWeighted(sem semaphore.Weighted, n int64, loops int) { for i := 0; i < loops; i++ { sem.Acquire(context.Background(), n) time.Sleep(time.Duration(rand.Int63n(int64(maxSleep/time.Nanosecond))) time.Nanosecond) sem.Release(n) } } func TestWeighted(t testing.T) { t.Parallel() n := runtime.GOMAXPROCS(0) loops := 10000 / n sem := semaphore.NewWeighted(int64(n)) var wg sync.WaitGroup wg.Add(n) for i := 0; i < n; i++ { i := i go func() { defer wg.Done() HammerWeighted(sem, int64(i), loops) }() } wg.Wait() } func TestWeightedPanic(t testing.T) { t.Parallel() defer func() { if recover() == nil { t.Fatal("release of an unacquired weighted semaphore did not panic") } }() w := semaphore.NewWeighted(1) w.Release(1) } func TestWeightedTryAcquire(t testing.T) { t.Parallel() ctx := context.Background() sem := semaphore.NewWeighted(2) tries := []bool{} sem.Acquire(ctx, 1) tries = append(tries, sem.TryAcquire(1)) tries = append(tries, sem.TryAcquire(1)) sem.Release(2) tries = append(tries, sem.TryAcquire(1)) sem.Acquire(ctx, 1) tries = append(tries, sem.TryAcquire(1)) want := []bool{true, false, true, false} for i := range tries { if tries[i] != want[i] { t.Errorf("tries[%d]: got %t, want %t", i, tries[i], want[i]) } } } func TestWeightedAcquire(t testing.T) { t.Parallel() ctx := context.Background() sem := semaphore.NewWeighted(2) tryAcquire := func(n int64) bool { ctx, cancel := context.WithTimeout(ctx, 10time.Millisecond) defer cancel() return sem.Acquire(ctx, n) == nil } tries := []bool{} sem.Acquire(ctx, 1) tries = append(tries, tryAcquire(1)) tries = append(tries, tryAcquire(1)) sem.Release(2) tries = append(tries, tryAcquire(1)) sem.Acquire(ctx, 1) tries = append(tries, tryAcquire(1)) want := []bool{true, false, true, false} for i := range tries { if tries[i] != want[i] { t.Errorf("tries[%d]: got %t, want %t", i, tries[i], want[i]) } } } func TestWeightedDoesntBlockIfTooBig(t testing.T) { t.Parallel() const n = 2 sem := semaphore.NewWeighted(n) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go sem.Acquire(ctx, n+1) } g, ctx := errgroup.WithContext(context.Background()) for i := n * 3; i > 0; i-- { g.Go(func() error { err := sem.Acquire(ctx, 1) if err == nil { time.Sleep(1 * time.Millisecond) sem.Release(1) } return err }) } if err := g.Wait(); err != nil { t.Errorf("semaphore.NewWeighted(%v) failed to AcquireCtx(_, 1) with AcquireCtx(_, %v) pending", n, n+1) } } // TestLargeAcquireDoesntStarve times out if a large call to Acquire starves. // Merely returning from the test function indicates success. func TestLargeAcquireDoesntStarve(t testing.T) { t.Parallel() ctx := context.Background() n := int64(runtime.GOMAXPROCS(0)) sem := semaphore.NewWeighted(n) running := true var wg sync.WaitGroup wg.Add(int(n)) for i := n; i > 0; i-- { sem.Acquire(ctx, 1) go func() { defer func() { sem.Release(1) wg.Done() }() for running { time.Sleep(1 time.Millisecond) sem.Release(1) sem.Acquire(ctx, 1) } }() } sem.Acquire(ctx, n) running = false sem.Release(n) wg.Wait() } // translated from https://github.com/zhiqiangxu/util/blob/master/mutex/crwmutex_test.go#L43 func TestAllocCancelDoesntStarve(t testing.T) { sem := semaphore.NewWeighted(10) // Block off a portion of the semaphore so that Acquire(_, 10) can eventually succeed. sem.Acquire(context.Background(), 1) // In the background, Acquire(_, 10). ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { sem.Acquire(ctx, 10) }() // Wait until the Acquire(_, 10) call blocks. for sem.TryAcquire(1) { sem.Release(1) runtime.Gosched() } // Now try to grab a read lock, and simultaneously unblock the Acquire(_, 10) call. // Both Acquire calls should unblock and return, in either order. go cancel() err := sem.Acquire(context.Background(), 1) if err != nil { t.Fatalf("Acquire(_, 1) failed unexpectedly: %v", err) } sem.Release(1) } func TestWeightedAcquireCanceled(t testing.T) { // https://go.dev/issue/63615 sem := semaphore.NewWeighted(2) ctx, cancel := context.WithCancel(context.Background()) sem.Acquire(context.Background(), 1) ch := make(chan struct{}) go func() { // Synchronize with the Acquire(2) below. for sem.TryAcquire(1) { sem.Release(1) } // Now cancel ctx, and then release the token. cancel() sem.Release(1) close(ch) }() // Since the context closing happens before enough tokens become available, // this Acquire must fail. if err := sem.Acquire(ctx, 2); err != context.Canceled { t.Errorf("Acquire with canceled context returned wrong error: want context.Canceled, got %v", err) } // There must always be two tokens in the semaphore after the other // goroutine releases the one we held at the start. <-ch if !sem.TryAcquire(2) { t.Fatal("TryAcquire after canceled Acquire failed") } // Additionally verify that we don't acquire with a done context even when // we wouldn't need to block to do so. sem.Release(2) if err := sem.Acquire(ctx, 1); err != context.Canceled { t.Errorf("Acquire with canceled context returned wrong error: want context.Canceled, got %v", err) } }	5.56 KB	2025-05-25 12:27

Terminal