kubernetes1321.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  /*
     6   * Project: kubernetes
     7   * Issue or PR  : https://github.com/kubernetes/kubernetes/pull/1321
     8   * Buggy version: 9cd0fc70f1ca852c903b18b0933991036b3b2fa1
     9   * fix commit-id: 435e0b73bb99862f9dedf56a50260ff3dfef14ff
    10   * Flaky: 1/100
    11   */
    12  package main
    13  
    14  import (
    15  	"os"
    16  	"runtime"
    17  	"runtime/pprof"
    18  	"sync"
    19  	"time"
    20  )
    21  
    22  func init() {
    23  	register("Kubernetes1321", Kubernetes1321)
    24  }
    25  
    26  type muxWatcher_kubernetes1321 struct {
    27  	result chan struct{}
    28  	m      *Mux_kubernetes1321
    29  	id     int64
    30  }
    31  
    32  func (mw *muxWatcher_kubernetes1321) Stop() {
    33  	mw.m.stopWatching(mw.id)
    34  }
    35  
    36  type Mux_kubernetes1321 struct {
    37  	lock     sync.Mutex
    38  	watchers map[int64]*muxWatcher_kubernetes1321
    39  }
    40  
    41  func NewMux_kubernetes1321() *Mux_kubernetes1321 {
    42  	m := &Mux_kubernetes1321{
    43  		watchers: map[int64]*muxWatcher_kubernetes1321{},
    44  	}
    45  	go m.loop() // G2
    46  	return m
    47  }
    48  
    49  func (m *Mux_kubernetes1321) Watch() *muxWatcher_kubernetes1321 {
    50  	mw := &muxWatcher_kubernetes1321{
    51  		result: make(chan struct{}),
    52  		m:      m,
    53  		id:     int64(len(m.watchers)),
    54  	}
    55  	m.watchers[mw.id] = mw
    56  	runtime.Gosched()
    57  	return mw
    58  }
    59  
    60  func (m *Mux_kubernetes1321) loop() {
    61  	for i := 0; i < 100; i++ {
    62  		m.distribute()
    63  	}
    64  }
    65  
    66  func (m *Mux_kubernetes1321) distribute() {
    67  	m.lock.Lock()
    68  	defer m.lock.Unlock()
    69  	for _, w := range m.watchers {
    70  		w.result <- struct{}{}
    71  		runtime.Gosched()
    72  	}
    73  }
    74  
    75  func (m *Mux_kubernetes1321) stopWatching(id int64) {
    76  	m.lock.Lock()
    77  	defer m.lock.Unlock()
    78  	w, ok := m.watchers[id]
    79  	if !ok {
    80  		return
    81  	}
    82  	delete(m.watchers, id)
    83  	close(w.result)
    84  }
    85  
    86  func testMuxWatcherClose_kubernetes1321() {
    87  	m := NewMux_kubernetes1321()
    88  	m.watchers[m.Watch().id].Stop()
    89  }
    90  
    91  func Kubernetes1321() {
    92  	prof := pprof.Lookup("goroutineleak")
    93  	defer func() {
    94  		time.Sleep(100 * time.Millisecond)
    95  		prof.WriteTo(os.Stdout, 2)
    96  	}()
    97  	for i := 0; i < 1000; i++ {
    98  		go testMuxWatcherClose_kubernetes1321() // G1
    99  	}
   100  }
   101
View as plain text