forked from kubernetes/kubernetes
-
Notifications
You must be signed in to change notification settings - Fork 0
/
goroutinemap.go
115 lines (95 loc) · 3.18 KB
/
goroutinemap.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
/*
Copyright 2016 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
/*
Package goroutinemap implements a data structure for managing go routines
by name. It prevents the creation of new go routines if an existing go routine
with the same name exists.
*/
package goroutinemap
import (
"fmt"
"sync"
"k8s.io/kubernetes/pkg/util/runtime"
)
// GoRoutineMap defines the supported set of operations.
type GoRoutineMap interface {
// Run adds operationName to the list of running operations and spawns a new
// go routine to execute the operation. If an operation with the same name
// already exists, an error is returned. Once the operation is complete, the
// go routine is terminated and the operationName is removed from the list
// of executing operations allowing a new operation to be started with the
// same name without error.
Run(operationName string, operation func() error) error
// Wait blocks until all operations are completed. This is typically
// necessary during tests - the test should wait until all operations finish
// and evaluate results after that.
Wait()
}
// NewGoRoutineMap returns a new instance of GoRoutineMap.
func NewGoRoutineMap() GoRoutineMap {
return &goRoutineMap{
operations: make(map[string]bool),
}
}
type goRoutineMap struct {
operations map[string]bool
sync.Mutex
wg sync.WaitGroup
}
func (grm *goRoutineMap) Run(operationName string, operation func() error) error {
grm.Lock()
defer grm.Unlock()
if grm.operations[operationName] {
// Operation with name exists
return newAlreadyExistsError(operationName)
}
grm.operations[operationName] = true
grm.wg.Add(1)
go func() {
defer grm.operationComplete(operationName)
defer runtime.HandleCrash()
operation()
}()
return nil
}
func (grm *goRoutineMap) operationComplete(operationName string) {
defer grm.wg.Done()
grm.Lock()
defer grm.Unlock()
delete(grm.operations, operationName)
}
func (grm *goRoutineMap) Wait() {
grm.wg.Wait()
}
// alreadyExistsError is specific error returned when NewGoRoutine()
// detects that operation with given name is already running.
type alreadyExistsError struct {
operationName string
}
var _ error = alreadyExistsError{}
func (err alreadyExistsError) Error() string {
return fmt.Sprintf("Failed to create operation with name %q. An operation with that name already exists", err.operationName)
}
func newAlreadyExistsError(operationName string) error {
return alreadyExistsError{operationName}
}
// IsAlreadyExists returns true if an error returned from NewGoRoutine indicates
// that operation with the same name already exists.
func IsAlreadyExists(err error) bool {
switch err.(type) {
case alreadyExistsError:
return true
default:
return false
}
}