-
Notifications
You must be signed in to change notification settings - Fork 8.8k
/
custodian.go
126 lines (110 loc) · 3.35 KB
/
custodian.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
116
117
118
119
120
121
122
123
124
125
126
/*
Copyright IBM Corp. All Rights Reserved.
SPDX-License-Identifier: Apache-2.0
*/
package lifecycle
import (
"sync"
"github.com/hyperledger/fabric/core/container"
)
//go:generate counterfeiter -o mock/chaincode_launcher.go --fake-name ChaincodeLauncher . ChaincodeLauncher
type ChaincodeLauncher interface {
Launch(ccid string) error
Stop(ccid string) error
}
// ChaincodeCustodian is responsible for enqueuing builds and launches
// of chaincodes as they become available and stops when chaincodes
// are no longer referenced by an active chaincode definition.
type ChaincodeCustodian struct {
cond *sync.Cond
mutex sync.Mutex
choreQueue []*chaincodeChore
halt bool
}
// chaincodeChore represents a unit of work to be performed by the worker
// routine. It identifies the chaincode the work is associated with. If
// the work is to launch, then runnable is true (and stoppable is false).
// If the work is to stop, then stoppable is true (and runnable is false).
// If the work is simply to build, then runnable and stoppable are false.
type chaincodeChore struct {
chaincodeID string
runnable bool
stoppable bool
}
// NewChaincodeCustodian creates an instance of a chaincode custodian. It is the
// instantiator's responsibility to spawn a go routine to service the Work routine
// along with the appropriate dependencies.
func NewChaincodeCustodian() *ChaincodeCustodian {
cc := &ChaincodeCustodian{}
cc.cond = sync.NewCond(&cc.mutex)
return cc
}
func (cc *ChaincodeCustodian) NotifyInstalled(chaincodeID string) {
cc.mutex.Lock()
defer cc.mutex.Unlock()
cc.choreQueue = append(cc.choreQueue, &chaincodeChore{
chaincodeID: chaincodeID,
})
cc.cond.Signal()
}
func (cc *ChaincodeCustodian) NotifyInstalledAndRunnable(chaincodeID string) {
cc.mutex.Lock()
defer cc.mutex.Unlock()
cc.choreQueue = append(cc.choreQueue, &chaincodeChore{
chaincodeID: chaincodeID,
runnable: true,
})
cc.cond.Signal()
}
func (cc *ChaincodeCustodian) NotifyStoppable(chaincodeID string) {
cc.mutex.Lock()
defer cc.mutex.Unlock()
cc.choreQueue = append(cc.choreQueue, &chaincodeChore{
chaincodeID: chaincodeID,
stoppable: true,
})
cc.cond.Signal()
}
func (cc *ChaincodeCustodian) Close() {
cc.mutex.Lock()
defer cc.mutex.Unlock()
cc.halt = true
cc.cond.Signal()
}
func (cc *ChaincodeCustodian) Work(buildRegistry *container.BuildRegistry, builder ChaincodeBuilder, launcher ChaincodeLauncher) {
for {
cc.mutex.Lock()
if len(cc.choreQueue) == 0 && !cc.halt {
cc.cond.Wait()
}
if cc.halt {
cc.mutex.Unlock()
return
}
chore := cc.choreQueue[0]
cc.choreQueue = cc.choreQueue[1:]
cc.mutex.Unlock()
if chore.runnable {
if err := launcher.Launch(chore.chaincodeID); err != nil {
logger.Warningf("could not launch chaincode '%s': %s", chore.chaincodeID, err)
}
continue
}
if chore.stoppable {
if err := launcher.Stop(chore.chaincodeID); err != nil {
logger.Warningf("could not stop chaincode '%s': %s", chore.chaincodeID, err)
}
continue
}
buildStatus, ok := buildRegistry.BuildStatus(chore.chaincodeID)
if ok {
logger.Debugf("skipping build of chaincode '%s' as it is already in progress", chore.chaincodeID)
continue
}
err := builder.Build(chore.chaincodeID)
if err != nil {
logger.Warningf("could not build chaincode '%s': %s", chore.chaincodeID, err)
}
buildStatus.Notify(err)
}
}