@@ -67,15 +67,16 @@ type GraphSync struct {
6767}
6868
6969type graphsyncConfigOptions struct {
70- totalMaxMemoryResponder uint64
71- maxMemoryPerPeerResponder uint64
72- totalMaxMemoryRequestor uint64
73- maxMemoryPerPeerRequestor uint64
74- maxInProgressIncomingRequests uint64
75- maxInProgressOutgoingRequests uint64
76- registerDefaultValidator bool
77- maxLinksPerOutgoingRequest uint64
78- maxLinksPerIncomingRequest uint64
70+ totalMaxMemoryResponder uint64
71+ maxMemoryPerPeerResponder uint64
72+ totalMaxMemoryRequestor uint64
73+ maxMemoryPerPeerRequestor uint64
74+ maxInProgressIncomingRequests uint64
75+ maxInProgressIncomingRequestsPerPeer uint64
76+ maxInProgressOutgoingRequests uint64
77+ registerDefaultValidator bool
78+ maxLinksPerOutgoingRequest uint64
79+ maxLinksPerIncomingRequest uint64
7980}
8081
8182// Option defines the functional option type that can be used to configure
@@ -123,15 +124,29 @@ func MaxMemoryPerPeerRequestor(maxMemoryPerPeer uint64) Option {
123124}
124125
125126// MaxInProgressIncomingRequests changes the maximum number of
126- // graphsync requests that are processed in parallel (default 6)
127+ // incoming graphsync requests that are processed in parallel (default 6)
127128func MaxInProgressIncomingRequests (maxInProgressIncomingRequests uint64 ) Option {
128129 return func (gs * graphsyncConfigOptions ) {
129130 gs .maxInProgressIncomingRequests = maxInProgressIncomingRequests
130131 }
131132}
132133
134+ // MaxInProgressIncomingRequestsPerPeer changes the maximum number of
135+ // incoming graphsync requests that are processed in parallel on a per-peer basis.
136+ // The value is not set by default.
137+ // Useful in an environment for very high bandwidth graphsync responders serving
138+ // many peers
139+ // Note: if for some reason this is set higher than MaxInProgressIncomingRequests
140+ // it will simply have no effect.
141+ // Note: setting a value of zero will have no effect
142+ func MaxInProgressIncomingRequestsPerPeer (maxInProgressIncomingRequestsPerPeer uint64 ) Option {
143+ return func (gs * graphsyncConfigOptions ) {
144+ gs .maxInProgressIncomingRequestsPerPeer = maxInProgressIncomingRequestsPerPeer
145+ }
146+ }
147+
133148// MaxInProgressOutgoingRequests changes the maximum number of
134- // graphsync requests that are processed in parallel (default 6)
149+ // outgoing graphsync requests that are processed in parallel (default 6)
135150func MaxInProgressOutgoingRequests (maxInProgressOutgoingRequests uint64 ) Option {
136151 return func (gs * graphsyncConfigOptions ) {
137152 gs .maxInProgressOutgoingRequests = maxInProgressOutgoingRequests
@@ -202,7 +217,11 @@ func New(parent context.Context, network gsnet.GraphSyncNetwork,
202217 requestManager := requestmanager .New (ctx , asyncLoader , linkSystem , outgoingRequestHooks , incomingResponseHooks , networkErrorListeners , requestQueue , network .ConnectionManager (), gsConfig .maxLinksPerOutgoingRequest )
203218 requestExecutor := executor .NewExecutor (requestManager , incomingBlockHooks , asyncLoader .AsyncLoad )
204219 responseAssembler := responseassembler .New (ctx , peerManager )
205- peerTaskQueue := peertaskqueue .New ()
220+ var ptqopts []peertaskqueue.Option
221+ if gsConfig .maxInProgressIncomingRequestsPerPeer > 0 {
222+ ptqopts = append (ptqopts , peertaskqueue .MaxOutstandingWorkPerPeer (int (gsConfig .maxInProgressIncomingRequestsPerPeer )))
223+ }
224+ peerTaskQueue := peertaskqueue .New (ptqopts ... )
206225 responseManager := responsemanager .New (ctx , linkSystem , responseAssembler , peerTaskQueue , requestQueuedHooks , incomingRequestHooks , outgoingBlockHooks , requestUpdatedHooks , completedResponseListeners , requestorCancelledListeners , blockSentListeners , networkErrorListeners , gsConfig .maxInProgressIncomingRequests , network .ConnectionManager (), gsConfig .maxLinksPerIncomingRequest )
207226 graphSync := & GraphSync {
208227 network : network ,
0 commit comments