serving2137.go

     1  // Copyright 2025 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a MIT
     3  // license that can be found in the LICENSE file.
     4  
     5  package main
     6  
     7  import (
     8  	"os"
     9  	"runtime"
    10  	"runtime/pprof"
    11  	"sync"
    12  	"time"
    13  )
    14  
    15  func init() {
    16  	register("Serving2137", Serving2137)
    17  }
    18  
    19  type token_serving2137 struct{}
    20  
    21  type request_serving2137 struct {
    22  	lock     *sync.Mutex
    23  	accepted chan bool
    24  }
    25  
    26  type Breaker_serving2137 struct {
    27  	pendingRequests chan token_serving2137
    28  	activeRequests  chan token_serving2137
    29  }
    30  
    31  func (b *Breaker_serving2137) Maybe(thunk func()) bool {
    32  	var t token_serving2137
    33  	select {
    34  	default:
    35  		// Pending request queue is full.  Report failure.
    36  		return false
    37  	case b.pendingRequests <- t:
    38  		// Pending request has capacity.
    39  		// Wait for capacity in the active queue.
    40  		b.activeRequests <- t
    41  		// Defer releasing capacity in the active and pending request queue.
    42  		defer func() {
    43  			<-b.activeRequests
    44  			runtime.Gosched()
    45  			<-b.pendingRequests
    46  		}()
    47  		// Do the thing.
    48  		thunk()
    49  		// Report success
    50  		return true
    51  	}
    52  }
    53  
    54  func (b *Breaker_serving2137) concurrentRequest() request_serving2137 {
    55  	r := request_serving2137{lock: &sync.Mutex{}, accepted: make(chan bool, 1)}
    56  	r.lock.Lock()
    57  	var start sync.WaitGroup
    58  	start.Add(1)
    59  	go func() { // G2, G3
    60  		start.Done()
    61  		runtime.Gosched()
    62  		ok := b.Maybe(func() {
    63  			// Will block on locked mutex.
    64  			r.lock.Lock()
    65  			runtime.Gosched()
    66  			r.lock.Unlock()
    67  		})
    68  		r.accepted <- ok
    69  	}()
    70  	start.Wait() // Ensure that the go func has had a chance to execute.
    71  	return r
    72  }
    73  
    74  // Perform n requests against the breaker, returning mutexes for each
    75  // request which succeeded, and a slice of bools for all requests.
    76  func (b *Breaker_serving2137) concurrentRequests(n int) []request_serving2137 {
    77  	requests := make([]request_serving2137, n)
    78  	for i := range requests {
    79  		requests[i] = b.concurrentRequest()
    80  	}
    81  	return requests
    82  }
    83  
    84  func NewBreaker_serving2137(queueDepth, maxConcurrency int32) *Breaker_serving2137 {
    85  	return &Breaker_serving2137{
    86  		pendingRequests: make(chan token_serving2137, queueDepth+maxConcurrency),
    87  		activeRequests:  make(chan token_serving2137, maxConcurrency),
    88  	}
    89  }
    90  
    91  func unlock_serving2137(req request_serving2137) {
    92  	req.lock.Unlock()
    93  	runtime.Gosched()
    94  	// Verify that function has completed
    95  	ok := <-req.accepted
    96  	runtime.Gosched()
    97  	// Requeue for next usage
    98  	req.accepted <- ok
    99  }
   100  
   101  func unlockAll_serving2137(requests []request_serving2137) {
   102  	for _, lc := range requests {
   103  		unlock_serving2137(lc)
   104  	}
   105  }
   106  
   107  func Serving2137() {
   108  	prof := pprof.Lookup("goroutineleak")
   109  	defer func() {
   110  		time.Sleep(100 * time.Millisecond)
   111  		prof.WriteTo(os.Stdout, 2)
   112  	}()
   113  
   114  	for i := 0; i < 1000; i++ {
   115  		go func() {
   116  			b := NewBreaker_serving2137(1, 1)
   117  
   118  			locks := b.concurrentRequests(2) // G1
   119  			unlockAll_serving2137(locks)
   120  		}()
   121  	}
   122  }
   123
View as plain text