Skip to content

Commit cd80a9b

Browse files
rvaggrvagg
committed
fixup! feat(graphsync): unify req & resp Pause, Unpause & Cancel by RequestID
1 parent f5e9fa1 commit cd80a9b

File tree

5 files changed

+36
-36
lines changed

5 files changed

+36
-36
lines changed

‎responsemanager/client.go

+4-4
Original file line numberDiff line numberDiff line change
@@ -200,8 +200,8 @@ func (rm *ResponseManager) synchronize() {
200200
}
201201

202202
// StartTask starts the given task from the peer task queue
203-
func (rm *ResponseManager) StartTask(task *peertask.Task, responseTaskChan chan<- queryexecutor.ResponseTask) {
204-
rm.send(&startTaskRequest{task, responseTaskChan}, nil)
203+
func (rm *ResponseManager) StartTask(task *peertask.Task, p peer.ID, responseTaskChan chan<- queryexecutor.ResponseTask) {
204+
rm.send(&startTaskRequest{task, p, responseTaskChan}, nil)
205205
}
206206

207207
// GetUpdates is called to read pending updates for a task and clear them
@@ -210,9 +210,9 @@ func (rm *ResponseManager) GetUpdates(requestID graphsync.RequestID, updatesChan
210210
}
211211

212212
// FinishTask marks a task from the task queue as done
213-
func (rm *ResponseManager) FinishTask(task *peertask.Task, err error) {
213+
func (rm *ResponseManager) FinishTask(task *peertask.Task, p peer.ID, err error) {
214214
done := make(chan struct{}, 1)
215-
rm.send(&finishTaskRequest{task, err, done}, nil)
215+
rm.send(&finishTaskRequest{task, p, err, done}, nil)
216216
select {
217217
case <-rm.ctx.Done():
218218
case <-done:

‎responsemanager/messages.go

+4-2
Original file line numberDiff line numberDiff line change
@@ -86,12 +86,13 @@ func (rur *responseUpdateRequest) handle(rm *ResponseManager) {
8686

8787
type finishTaskRequest struct {
8888
task *peertask.Task
89+
p peer.ID
8990
err error
9091
done chan struct{}
9192
}
9293

9394
func (ftr *finishTaskRequest) handle(rm *ResponseManager) {
94-
rm.finishTask(ftr.task, ftr.err)
95+
rm.finishTask(ftr.task, ftr.p, ftr.err)
9596
select {
9697
case <-rm.ctx.Done():
9798
case ftr.done <- struct{}{}:
@@ -100,11 +101,12 @@ func (ftr *finishTaskRequest) handle(rm *ResponseManager) {
100101

101102
type startTaskRequest struct {
102103
task *peertask.Task
104+
p peer.ID
103105
taskDataChan chan<- queryexecutor.ResponseTask
104106
}
105107

106108
func (str *startTaskRequest) handle(rm *ResponseManager) {
107-
taskData := rm.startTask(str.task)
109+
taskData := rm.startTask(str.task, str.p)
108110

109111
select {
110112
case <-rm.ctx.Done():

‎responsemanager/queryexecutor/queryexecutor.go

+4-4
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,7 @@ func (qe *QueryExecutor) ExecuteTask(_ context.Context, pid peer.ID, task *peert
8787
// StartTask lets us block until this task is at the top of the execution stack
8888
responseTaskChan := make(chan ResponseTask)
8989
var rt ResponseTask
90-
qe.manager.StartTask(task, responseTaskChan)
90+
qe.manager.StartTask(task, pid, responseTaskChan)
9191
select {
9292
case rt = <-responseTaskChan:
9393
case <-qe.ctx.Done():
@@ -109,7 +109,7 @@ func (qe *QueryExecutor) ExecuteTask(_ context.Context, pid peer.ID, task *peert
109109
span.SetStatus(codes.Error, err.Error())
110110
}
111111
}
112-
qe.manager.FinishTask(task, err)
112+
qe.manager.FinishTask(task, pid, err)
113113
log.Debugw("finishing response execution", "id", rt.Request.ID(), "peer", pid.String(), "root_cid", rt.Request.Root().String())
114114
return false
115115
}
@@ -279,9 +279,9 @@ func (qe *QueryExecutor) sendResponse(ctx context.Context, p peer.ID, taskData R
279279

280280
// Manager providers an interface to the response manager
281281
type Manager interface {
282-
StartTask(task *peertask.Task, responseTaskChan chan<- ResponseTask)
282+
StartTask(task *peertask.Task, p peer.ID, responseTaskChan chan<- ResponseTask)
283283
GetUpdates(requestID graphsync.RequestID, updatesChan chan<- []gsmsg.GraphSyncRequest)
284-
FinishTask(task *peertask.Task, err error)
284+
FinishTask(task *peertask.Task, p peer.ID, err error)
285285
}
286286

287287
// BlockHooks is an interface for processing block hooks

‎responsemanager/queryexecutor/queryexecutor_test.go

+7-4
Original file line numberDiff line numberDiff line change
@@ -268,10 +268,11 @@ func newTestData(t *testing.T, blockCount int, expectedTraverse int) (*testData,
268268
td := &testData{}
269269
td.t = t
270270
td.ctx, td.cancel = context.WithTimeout(ctx, 10*time.Second)
271+
td.peer = testutil.GeneratePeers(1)[0]
271272
td.blockStore = make(map[ipld.Link][]byte)
272273
td.persistence = testutil.NewTestStore(td.blockStore)
273274
td.task = &peertask.Task{}
274-
td.manager = &fauxManager{ctx: ctx, t: t, expectedStartTask: td.task}
275+
td.manager = &fauxManager{ctx: ctx, t: t, expectedStartTask: td.task, expectedPeer: td.peer}
275276
td.blockHooks = hooks.NewBlockHooks()
276277
td.updateHooks = hooks.NewUpdateHooks()
277278
td.requestID = graphsync.NewRequestID()
@@ -280,7 +281,6 @@ func newTestData(t *testing.T, blockCount int, expectedTraverse int) (*testData,
280281
td.extensionData = basicnode.NewBytes(testutil.RandomBytes(100))
281282
td.extensionName = graphsync.ExtensionName("AppleSauce/McGee")
282283
td.responseCode = graphsync.ResponseStatusCode(101)
283-
td.peer = testutil.GeneratePeers(1)[0]
284284

285285
td.extension = graphsync.ExtensionData{
286286
Name: td.extensionName,
@@ -367,10 +367,12 @@ type fauxManager struct {
367367
t *testing.T
368368
responseTask ResponseTask
369369
expectedStartTask *peertask.Task
370+
expectedPeer peer.ID
370371
}
371372

372-
func (fm *fauxManager) StartTask(task *peertask.Task, responseTaskChan chan<- ResponseTask) {
373+
func (fm *fauxManager) StartTask(task *peertask.Task, p peer.ID, responseTaskChan chan<- ResponseTask) {
373374
require.Same(fm.t, fm.expectedStartTask, task)
375+
require.Equal(fm.t, fm.expectedPeer, p)
374376
go func() {
375377
select {
376378
case <-fm.ctx.Done():
@@ -382,7 +384,8 @@ func (fm *fauxManager) StartTask(task *peertask.Task, responseTaskChan chan<- Re
382384
func (fm *fauxManager) GetUpdates(requestID graphsync.RequestID, updatesChan chan<- []gsmsg.GraphSyncRequest) {
383385
}
384386

385-
func (fm *fauxManager) FinishTask(task *peertask.Task, err error) {
387+
func (fm *fauxManager) FinishTask(task *peertask.Task, p peer.ID, err error) {
388+
require.Equal(fm.t, fm.expectedPeer, p)
386389
}
387390

388391
type fauxResponseStream struct {

‎responsemanager/server.go

+17-22
Original file line numberDiff line numberDiff line change
@@ -26,11 +26,6 @@ import (
2626
// The code in this file implements the internal thread for the response manager.
2727
// These functions can modify the internal state of the ResponseManager
2828

29-
type queueTopic struct {
30-
p peer.ID
31-
requestID graphsync.RequestID
32-
}
33-
3429
func (rm *ResponseManager) cleanupInProcessResponses() {
3530
for _, response := range rm.inProgressResponses {
3631
response.cancelFn()
@@ -137,14 +132,14 @@ func (rm *ResponseManager) unpauseRequest(requestID graphsync.RequestID, extensi
137132
return nil
138133
})
139134
}
140-
rm.responseQueue.PushTask(inProgressResponse.peer, peertask.Task{Topic: queueTopic{inProgressResponse.peer, requestID}, Priority: math.MaxInt32, Work: 1})
135+
rm.responseQueue.PushTask(inProgressResponse.peer, peertask.Task{Topic: requestID, Priority: math.MaxInt32, Work: 1})
141136
return nil
142137
}
143138

144139
func (rm *ResponseManager) abortRequest(ctx context.Context, requestID graphsync.RequestID, err error) error {
145140
response, ok := rm.inProgressResponses[requestID]
146141
if ok {
147-
rm.responseQueue.Remove(queueTopic{response.peer, requestID}, response.peer)
142+
rm.responseQueue.Remove(requestID, response.peer)
148143
}
149144
if !ok || response.state == graphsync.CompletingSend {
150145
return graphsync.RequestNotFoundErr{}
@@ -258,7 +253,7 @@ func (rm *ResponseManager) processRequests(p peer.ID, requests []gsmsg.GraphSync
258253
}
259254
// TODO: Use a better work estimation metric.
260255

261-
rm.responseQueue.PushTask(p, peertask.Task{Topic: queueTopic{p, request.ID()}, Priority: int(request.Priority()), Work: 1})
256+
rm.responseQueue.PushTask(p, peertask.Task{Topic: request.ID(), Priority: int(request.Priority()), Work: 1})
262257
}
263258
}
264259

@@ -297,28 +292,28 @@ func (rm *ResponseManager) taskDataForKey(requestID graphsync.RequestID) queryex
297292
}
298293
}
299294

300-
func (rm *ResponseManager) startTask(task *peertask.Task) queryexecutor.ResponseTask {
301-
key := task.Topic.(queueTopic)
302-
taskData := rm.taskDataForKey(key.requestID)
295+
func (rm *ResponseManager) startTask(task *peertask.Task, p peer.ID) queryexecutor.ResponseTask {
296+
requestID := task.Topic.(graphsync.RequestID)
297+
taskData := rm.taskDataForKey(requestID)
303298
if taskData.Empty {
304-
rm.responseQueue.TaskDone(key.p, task)
299+
rm.responseQueue.TaskDone(p, task)
305300
}
306301

307302
return taskData
308303
}
309304

310-
func (rm *ResponseManager) finishTask(task *peertask.Task, err error) {
311-
key := task.Topic.(queueTopic)
312-
rm.responseQueue.TaskDone(key.p, task)
313-
response, ok := rm.inProgressResponses[key.requestID]
305+
func (rm *ResponseManager) finishTask(task *peertask.Task, p peer.ID, err error) {
306+
requestID := task.Topic.(graphsync.RequestID)
307+
rm.responseQueue.TaskDone(p, task)
308+
response, ok := rm.inProgressResponses[requestID]
314309
if !ok {
315310
return
316311
}
317312
if _, ok := err.(hooks.ErrPaused); ok {
318313
response.state = graphsync.Paused
319314
return
320315
}
321-
log.Infow("graphsync response processing complete (messages stil sending)", "request id", key.requestID.String(), "peer", key.p, "total time", time.Since(response.startTime))
316+
log.Infow("graphsync response processing complete (messages stil sending)", "request id", requestID.String(), "peer", p, "total time", time.Since(response.startTime))
322317

323318
if err != nil {
324319
response.span.RecordError(err)
@@ -327,13 +322,13 @@ func (rm *ResponseManager) finishTask(task *peertask.Task, err error) {
327322
}
328323

329324
if ipldutil.IsContextCancelErr(err) {
330-
rm.cancelledListeners.NotifyCancelledListeners(key.p, response.request)
331-
rm.terminateRequest(key.requestID)
325+
rm.cancelledListeners.NotifyCancelledListeners(p, response.request)
326+
rm.terminateRequest(requestID)
332327
return
333328
}
334329

335330
if err == queryexecutor.ErrNetworkError {
336-
rm.terminateRequest(key.requestID)
331+
rm.terminateRequest(requestID)
337332
return
338333
}
339334

@@ -385,11 +380,11 @@ func fromPeerTopics(pt *peertracker.PeerTrackerTopics) peerstate.TaskQueueState
385380
}
386381
active := make([]graphsync.RequestID, 0, len(pt.Active))
387382
for _, topic := range pt.Active {
388-
active = append(active, topic.(queueTopic).requestID)
383+
active = append(active, topic.(graphsync.RequestID))
389384
}
390385
pending := make([]graphsync.RequestID, 0, len(pt.Pending))
391386
for _, topic := range pt.Pending {
392-
pending = append(pending, topic.(queueTopic).requestID)
387+
pending = append(pending, topic.(graphsync.RequestID))
393388
}
394389
return peerstate.TaskQueueState{
395390
Active: active,

0 commit comments

Comments
 (0)