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
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
// Copyright (C) 2019-2023 Algorand, Inc.
// This file is part of go-algorand
//
// go-algorand is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// go-algorand is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with go-algorand. If not, see <https://www.gnu.org/licenses/>.
package agreement
import (
"context"
"errors"
"sync"
"github.com/algorand/go-algorand/util/execpool"
)
type asyncVerifyVoteRequest struct {
ctx context.Context
l LedgerReader
uv *unauthenticatedVote
uev *unauthenticatedEquivocationVote
index uint64
message message
// a channel that holds the response
out chan<- asyncVerifyVoteResponse
}
type asyncVerifyVoteResponse struct {
v vote
ev equivocationVote
index uint64
message message
err error
cancelled bool
// a pointer to the request
req *asyncVerifyVoteRequest
}
// AsyncVoteVerifier uses workers to verify agreement protocol votes and writes the results on an output channel specified by the user.
type AsyncVoteVerifier struct {
done chan struct{}
wg sync.WaitGroup
workerWaitCh chan struct{}
backlogExecPool execpool.BacklogPool
execpoolOut chan interface{}
ctx context.Context
ctxCancel context.CancelFunc
}
// MakeAsyncVoteVerifier creates an AsyncVoteVerifier with workers as the number of CPUs
func MakeAsyncVoteVerifier(verificationPool execpool.BacklogPool) *AsyncVoteVerifier {
verifier := &AsyncVoteVerifier{
done: make(chan struct{}),
}
if verificationPool == nil {
// The MakeBacklog would internall allocate an execution pool if none was provided.
verificationPool = execpool.MakeBacklog(nil, 0, execpool.HighPriority, verifier)
}
verifier.backlogExecPool = verificationPool
// The backlog execution pool is going to have 2*GetParallelism() items in the input channel.
// Since we want our output channel to be sufficitly large, we're going to allocate the size of the
// input channel, plus all the content of the currently-executing tasks. That would prevent the
// pool from getting stuck by client enqueuing messages, as long as these clients keep pulling from the
// output queue at the same rate.
verifier.execpoolOut = make(chan interface{}, 3*verificationPool.GetParallelism())
verifier.ctx, verifier.ctxCancel = context.WithCancel(context.Background())
verifier.workerWaitCh = make(chan struct{})
go verifier.worker()
return verifier
}
func (avv *AsyncVoteVerifier) worker() {
defer close(avv.workerWaitCh)
for res := range avv.execpoolOut {
asyncResponse := res.(*asyncVerifyVoteResponse)
if asyncResponse != nil {
asyncResponse.req.out <- *asyncResponse
}
avv.wg.Done()
}
}
func (avv *AsyncVoteVerifier) executeVoteVerification(task interface{}) interface{} {
req := task.(asyncVerifyVoteRequest)
select {
case <-req.ctx.Done():
// request cancelled, return an error response on the channel
return &asyncVerifyVoteResponse{err: req.ctx.Err(), cancelled: true, req: &req, index: req.index}
default:
// request was not cancelled, so we verify it here and return the result on the channel
v, err := req.uv.verify(req.l)
req.message.Vote = v
var e *LedgerDroppedRoundError
cancelled := errors.As(err, &e)
return &asyncVerifyVoteResponse{v: v, index: req.index, message: req.message, err: err, cancelled: cancelled, req: &req}
}
}
func (avv *AsyncVoteVerifier) executeEqVoteVerification(task interface{}) interface{} {
req := task.(asyncVerifyVoteRequest)
select {
case <-req.ctx.Done():
// request cancelled, return an error response on the channel
return &asyncVerifyVoteResponse{err: req.ctx.Err(), cancelled: true, req: &req, index: req.index}
default:
// request was not cancelled, so we verify it here and return the result on the channel
ev, err := req.uev.verify(req.l)
var e *LedgerDroppedRoundError
cancelled := errors.As(err, &e)
return &asyncVerifyVoteResponse{ev: ev, index: req.index, message: req.message, err: err, cancelled: cancelled, req: &req}
}
}
func (avv *AsyncVoteVerifier) verifyVote(verctx context.Context, l LedgerReader, uv unauthenticatedVote, index uint64, message message, out chan<- asyncVerifyVoteResponse) error {
select {
case <-avv.ctx.Done(): // if we're quitting, don't enqueue the request
// case <-verctx.Done(): DO NOT DO THIS! otherwise we will lose the vote (and forget to clean up)!
// instead, enqueue so the worker will set the error value and return the cancelled vote properly.
default:
// if we're done while waiting for room in the requests channel, don't queue the request
req := asyncVerifyVoteRequest{ctx: verctx, l: l, uv: &uv, index: index, message: message, out: out}
avv.wg.Add(1)
if err := avv.backlogExecPool.EnqueueBacklog(avv.ctx, avv.executeVoteVerification, req, avv.execpoolOut); err != nil {
// we want to call "wg.Done()" here to "fix" the accounting of the number of pending tasks.
// if we got a non-nil, it means that our context has expired, which means that we won't see this task
// getting to the verification function.
avv.wg.Done()
return err
}
}
return nil
}
func (avv *AsyncVoteVerifier) verifyEqVote(verctx context.Context, l LedgerReader, uev unauthenticatedEquivocationVote, index uint64, message message, out chan<- asyncVerifyVoteResponse) error {
select {
case <-avv.ctx.Done(): // if we're quitting, don't enqueue the request
// case <-verctx.Done(): DO NOT DO THIS! otherwise we will lose the vote (and forget to clean up)!
// instead, enqueue so the worker will set the error value and return the cancelled vote properly.
default:
// if we're done while waiting for room in the requests channel, don't queue the request
req := asyncVerifyVoteRequest{ctx: verctx, l: l, uev: &uev, index: index, message: message, out: out}
avv.wg.Add(1)
if err := avv.backlogExecPool.EnqueueBacklog(avv.ctx, avv.executeEqVoteVerification, req, avv.execpoolOut); err != nil {
// we want to call "wg.Done()" here to "fix" the accounting of the number of pending tasks.
// if we got a non-nil, it means that our context has expired, which means that we won't see this task
// getting to the verification function.
avv.wg.Done()
return err
}
}
return nil
}
// Quit tells the AsyncVoteVerifier to shutdown and waits until all workers terminate.
func (avv *AsyncVoteVerifier) Quit() {
// indicate we're done and wait for all workers to finish
avv.ctxCancel()
// wait until all the tasks we've given the pool are done.
avv.wg.Wait()
if avv.backlogExecPool.GetOwner() == avv {
avv.backlogExecPool.Shutdown()
}
// since no more tasks are coming, we can safely close the output pool channel.
close(avv.execpoolOut)
// wait until the worker function exists.
<-avv.workerWaitCh
}
// Parallelism gives the maximum parallelism of the vote verifier.
func (avv *AsyncVoteVerifier) Parallelism() int {
return avv.backlogExecPool.GetParallelism()
}
|
