responsemanager.go 8.78 KB
Newer Older
1 2 3 4 5 6
package responsemanager

import (
	"context"
	"time"

7
	"github.com/ipfs/go-graphsync"
8 9
	"github.com/ipfs/go-graphsync/ipldbridge"
	gsmsg "github.com/ipfs/go-graphsync/message"
10
	"github.com/ipfs/go-graphsync/responsemanager/loader"
11
	"github.com/ipfs/go-graphsync/responsemanager/peerresponsemanager"
12
	"github.com/ipfs/go-graphsync/responsemanager/selectorvalidator"
13
	"github.com/ipfs/go-peertaskqueue/peertask"
14 15
	ipld "github.com/ipld/go-ipld-prime"
	cidlink "github.com/ipld/go-ipld-prime/linking/cid"
16
	"github.com/ipld/go-ipld-prime/traversal/selector"
17
	"github.com/libp2p/go-libp2p-core/peer"
18 19 20 21
)

const (
	maxInProcessRequests = 6
22
	maxRecursionDepth    = 100
23 24 25 26 27 28
	thawSpeed            = time.Millisecond * 100
)

type inProgressResponseStatus struct {
	ctx      context.Context
	cancelFn func()
29
	request  gsmsg.GraphSyncRequest
30 31 32 33
}

type responseKey struct {
	p         peer.ID
34
	requestID graphsync.RequestID
35 36 37
}

type responseTaskData struct {
38 39
	ctx     context.Context
	request gsmsg.GraphSyncRequest
40 41
}

42 43 44 45 46
type requestHook struct {
	overrideDefaultValidation bool
	hook                      graphsync.OnRequestReceivedHook
}

47 48 49
// QueryQueue is an interface that can receive new selector query tasks
// and prioritize them as needed, and pop them off later
type QueryQueue interface {
50 51 52
	PushBlock(to peer.ID, tasks ...peertask.Task)
	PopBlock() *peertask.TaskBlock
	Remove(identifier peertask.Identifier, p peer.ID)
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
	ThawRound()
}

// PeerManager is an interface that returns sender interfaces for peer responses.
type PeerManager interface {
	SenderForPeer(p peer.ID) peerresponsemanager.PeerResponseSender
}

type responseManagerMessage interface {
	handle(rm *ResponseManager)
}

// ResponseManager handles incoming requests from the network, initiates selector
// traversals, and transmits responses
type ResponseManager struct {
	ctx         context.Context
	cancelFn    context.CancelFunc
	loader      ipldbridge.Loader
	ipldBridge  ipldbridge.IPLDBridge
	peerManager PeerManager
	queryQueue  QueryQueue

	messages            chan responseManagerMessage
	workSignal          chan struct{}
	ticker              *time.Ticker
	inProgressResponses map[responseKey]inProgressResponseStatus
79
	requestHooks        []requestHook
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
}

// New creates a new response manager from the given context, loader,
// bridge to IPLD interface, peerManager, and queryQueue.
func New(ctx context.Context,
	loader ipldbridge.Loader,
	ipldBridge ipldbridge.IPLDBridge,
	peerManager PeerManager,
	queryQueue QueryQueue) *ResponseManager {
	ctx, cancelFn := context.WithCancel(ctx)
	return &ResponseManager{
		ctx:                 ctx,
		cancelFn:            cancelFn,
		loader:              loader,
		ipldBridge:          ipldBridge,
		peerManager:         peerManager,
		queryQueue:          queryQueue,
		messages:            make(chan responseManagerMessage, 16),
		workSignal:          make(chan struct{}, 1),
		ticker:              time.NewTicker(thawSpeed),
		inProgressResponses: make(map[responseKey]inProgressResponseStatus),
	}
}

type processRequestMessage struct {
	p        peer.ID
	requests []gsmsg.GraphSyncRequest
}

// ProcessRequests processes incoming requests for the given peer
110
func (rm *ResponseManager) ProcessRequests(ctx context.Context, p peer.ID, requests []gsmsg.GraphSyncRequest) {
111 112 113
	select {
	case rm.messages <- &processRequestMessage{p, requests}:
	case <-rm.ctx.Done():
114
	case <-ctx.Done():
115 116 117
	}
}

118
// RegisterHook registers an extension to process new incoming requests
119 120 121 122 123 124 125 126 127
func (rm *ResponseManager) RegisterHook(
	overrideDefaultValidation bool,
	hook graphsync.OnRequestReceivedHook) {
	select {
	case rm.messages <- &requestHook{overrideDefaultValidation, hook}:
	case <-rm.ctx.Done():
	}
}

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
type synchronizeMessage struct {
	sync chan struct{}
}

// this is a test utility method to force all messages to get processed
func (rm *ResponseManager) synchronize() {
	sync := make(chan struct{})
	select {
	case rm.messages <- &synchronizeMessage{sync}:
	case <-rm.ctx.Done():
	}
	select {
	case <-sync:
	case <-rm.ctx.Done():
	}
}

type responseDataRequest struct {
	key          responseKey
	taskDataChan chan *responseTaskData
}

type finishResponseRequest struct {
	key responseKey
}

func (rm *ResponseManager) processQueriesWorker() {
	taskDataChan := make(chan *responseTaskData)
	var taskData *responseTaskData
	for {
158 159
		nextTaskBlock := rm.queryQueue.PopBlock()
		for nextTaskBlock == nil {
160 161 162 163
			select {
			case <-rm.ctx.Done():
				return
			case <-rm.workSignal:
164
				nextTaskBlock = rm.queryQueue.PopBlock()
165 166
			case <-rm.ticker.C:
				rm.queryQueue.ThawRound()
167
				nextTaskBlock = rm.queryQueue.PopBlock()
168 169
			}
		}
170 171
		for _, task := range nextTaskBlock.Tasks {
			key := task.Identifier.(responseKey)
172 173 174 175 176 177 178 179 180 181
			select {
			case rm.messages <- &responseDataRequest{key, taskDataChan}:
			case <-rm.ctx.Done():
				return
			}
			select {
			case taskData = <-taskDataChan:
			case <-rm.ctx.Done():
				return
			}
182
			rm.executeQuery(taskData.ctx, key.p, taskData.request)
183 184 185 186 187
			select {
			case rm.messages <- &finishResponseRequest{key}:
			case <-rm.ctx.Done():
			}
		}
188 189
		nextTaskBlock.Done(nextTaskBlock.Tasks)

190 191 192 193
	}

}

194
func noopVisitor(tp ipldbridge.TraversalProgress, n ipld.Node, tr ipldbridge.TraversalReason) error {
195 196 197 198 199
	return nil
}

func (rm *ResponseManager) executeQuery(ctx context.Context,
	p peer.ID,
200
	request gsmsg.GraphSyncRequest) {
201
	peerResponseSender := rm.peerManager.SenderForPeer(p)
202 203 204 205
	extensionData, selector, err := rm.validateRequest(p, request)
	for _, datum := range extensionData {
		peerResponseSender.SendExtensionData(request.ID(), datum)
	}
206 207 208 209
	if err != nil {
		peerResponseSender.FinishWithError(request.ID(), graphsync.RequestFailedUnknown)
		return
	}
210 211 212 213 214 215 216 217 218 219 220 221 222 223 224
	rootLink := cidlink.Link{Cid: request.Root()}
	wrappedLoader := loader.WrapLoader(rm.loader, request.ID(), peerResponseSender)
	err = rm.ipldBridge.Traverse(ctx, wrappedLoader, rootLink, selector, noopVisitor)
	if err != nil {
		peerResponseSender.FinishWithError(request.ID(), graphsync.RequestFailedUnknown)
		return
	}
	peerResponseSender.FinishRequest(request.ID())
}

func (rm *ResponseManager) validateRequest(p peer.ID, request graphsync.RequestData) ([]graphsync.ExtensionData, selector.Selector, error) {
	selectorSpec, err := rm.ipldBridge.DecodeNode(request.Selector())
	if err != nil {
		return nil, nil, err
	}
225
	var isValidated bool
226
	var allExtensionData []graphsync.ExtensionData
227 228
	for _, requestHook := range rm.requestHooks {
		extensionData, err := requestHook.hook(p, request)
229
		allExtensionData = append(allExtensionData, extensionData...)
230
		if err != nil {
231
			return allExtensionData, nil, err
232 233 234 235 236 237 238 239
		}
		if requestHook.overrideDefaultValidation {
			isValidated = true
		}
	}
	if !isValidated {
		err = selectorvalidator.ValidateSelector(rm.ipldBridge, selectorSpec, maxRecursionDepth)
		if err != nil {
240
			return allExtensionData, nil, err
241
		}
242
	}
243
	selector, err := rm.ipldBridge.ParseSelector(selectorSpec)
244
	return allExtensionData, selector, err
245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287
}

// Startup starts processing for the WantManager.
func (rm *ResponseManager) Startup() {
	go rm.run()
}

// Shutdown ends processing for the want manager.
func (rm *ResponseManager) Shutdown() {
	rm.cancelFn()
}

func (rm *ResponseManager) cleanupInProcessResponses() {
	for _, response := range rm.inProgressResponses {
		response.cancelFn()
	}
}

func (rm *ResponseManager) run() {
	defer rm.cleanupInProcessResponses()
	for i := 0; i < maxInProcessRequests; i++ {
		go rm.processQueriesWorker()
	}

	for {
		select {
		case <-rm.ctx.Done():
			return
		case message := <-rm.messages:
			message.handle(rm)
		}
	}
}

func (prm *processRequestMessage) handle(rm *ResponseManager) {
	for _, request := range prm.requests {
		key := responseKey{p: prm.p, requestID: request.ID()}
		if !request.IsCancel() {
			ctx, cancelFn := context.WithCancel(rm.ctx)
			rm.inProgressResponses[key] =
				inProgressResponseStatus{
					ctx:      ctx,
					cancelFn: cancelFn,
288
					request:  request,
289
				}
290
			rm.queryQueue.PushBlock(prm.p, peertask.Task{Identifier: key, Priority: int(request.Priority())})
291 292 293 294 295 296 297 298 299 300 301 302 303 304
			select {
			case rm.workSignal <- struct{}{}:
			default:
			}
		} else {
			rm.queryQueue.Remove(key, key.p)
			response, ok := rm.inProgressResponses[key]
			if ok {
				response.cancelFn()
			}
		}
	}
}

305 306 307 308
func (rh *requestHook) handle(rm *ResponseManager) {
	rm.requestHooks = append(rm.requestHooks, *rh)
}

309 310 311 312
func (rdr *responseDataRequest) handle(rm *ResponseManager) {
	response, ok := rm.inProgressResponses[rdr.key]
	var taskData *responseTaskData
	if ok {
313
		taskData = &responseTaskData{response.ctx, response.request}
314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337
	} else {
		taskData = nil
	}
	select {
	case <-rm.ctx.Done():
	case rdr.taskDataChan <- taskData:
	}
}

func (frr *finishResponseRequest) handle(rm *ResponseManager) {
	response, ok := rm.inProgressResponses[frr.key]
	if !ok {
		return
	}
	delete(rm.inProgressResponses, frr.key)
	response.cancelFn()
}

func (sm *synchronizeMessage) handle(rm *ResponseManager) {
	select {
	case <-rm.ctx.Done():
	case sm.sync <- struct{}{}:
	}
}