4
4
"context"
5
5
"errors"
6
6
"math"
7
+ "time"
7
8
8
9
"github.com/ipfs/go-peertaskqueue/peertask"
9
10
"github.com/libp2p/go-libp2p-core/peer"
@@ -155,6 +156,11 @@ func (rm *ResponseManager) processRequests(p peer.ID, requests []gsmsg.GraphSync
155
156
networkErrorListeners : rm .networkErrorListeners ,
156
157
connManager : rm .connManager ,
157
158
})
159
+ log .Infow ("graphsync request initiated" , "request id" , request .ID (), "peer" , p , "root" , request .Root ())
160
+ ipr , ok := rm .inProgressResponses [key ]
161
+ if ok && ipr .state == graphsync .Running {
162
+ log .Warnf ("there is an identical request already in progress" , "request id" , request .ID (), "peer" , p )
163
+ }
158
164
159
165
rm .inProgressResponses [key ] =
160
166
& inProgressResponseStatus {
@@ -167,7 +173,8 @@ func (rm *ResponseManager) processRequests(p peer.ID, requests []gsmsg.GraphSync
167
173
UpdateSignal : make (chan struct {}, 1 ),
168
174
ErrSignal : make (chan error , 1 ),
169
175
},
170
- state : graphsync .Queued ,
176
+ state : graphsync .Queued ,
177
+ startTime : time .Now (),
171
178
}
172
179
// TODO: Use a better work estimation metric.
173
180
@@ -180,6 +187,8 @@ func (rm *ResponseManager) taskDataForKey(key responseKey) queryexecutor.Respons
180
187
if ! hasResponse {
181
188
return queryexecutor.ResponseTask {Empty : true }
182
189
}
190
+ log .Infow ("graphsync response processing begins" , "request id" , key .requestID , "peer" , key .p , "total time" , time .Since (response .startTime ))
191
+
183
192
if response .loader == nil || response .traverser == nil {
184
193
loader , traverser , isPaused , err := (& queryPreparer {rm .requestHooks , rm .responseAssembler , rm .linkSystem , rm .maxLinksPerRequest }).prepareQuery (response .ctx , key .p , response .request , response .signals , response .subscriber )
185
194
if err != nil {
@@ -212,6 +221,7 @@ func (rm *ResponseManager) startTask(task *peertask.Task) queryexecutor.Response
212
221
if taskData .Empty {
213
222
rm .responseQueue .TaskDone (key .p , task )
214
223
}
224
+
215
225
return taskData
216
226
}
217
227
@@ -226,6 +236,8 @@ func (rm *ResponseManager) finishTask(task *peertask.Task, err error) {
226
236
response .state = graphsync .Paused
227
237
return
228
238
}
239
+ log .Infow ("graphsync response processing complete (messages stil sending)" , "request id" , key .requestID , "peer" , key .p , "total time" , time .Since (response .startTime ))
240
+
229
241
if err != nil {
230
242
log .Infof ("response failed: %w" , err )
231
243
}
0 commit comments