mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-29 01:33:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Review comments

Browse source
This commit is contained in:
Neil Alexander 2021-11-04 16:28:20 +00:00
parent 900adb6ad2
commit e7f7050163
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
1 changed files with 15 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
federationapi/routing/send.go
View file

@ -148,7 +148,7 @@ type inputWorker struct {
input *sendFIFOQueue input *sendFIFOQueue
} }
var txnIDs sync.Map // transaction ID -> chan util.JSONResponse var inFlightTxnsPerOrigin sync.Map // transaction ID -> chan util.JSONResponse
var inputWorkers sync.Map // room ID -> *inputWorker var inputWorkers sync.Map // room ID -> *inputWorker
// Send implements /_matrix/federation/v1/send/{txnID} // Send implements /_matrix/federation/v1/send/{txnID}
@ -169,14 +169,21 @@ func Send(
// txn ID to us. If they have and the txnIDs map contains an entry, // txn ID to us. If they have and the txnIDs map contains an entry,
// the transaction is still being worked on. The new client can wait // the transaction is still being worked on. The new client can wait
// for it to complete rather than creating more work. // for it to complete rather than creating more work.
index := string(request.Origin()) + string(txnID) index := string(request.Origin()) + "\000" + string(txnID)
if v, ok := txnIDs.Load(index); ok { ch, ok := inFlightTxnsPerOrigin.LoadOrStore(index, make(chan util.JSONResponse, 1))
if ok {
// This origin already submitted this txn ID to us, and the work // This origin already submitted this txn ID to us, and the work
// is still taking place, so we'll just wait for it to finish. // is still taking place, so we'll just wait for it to finish.
ctx, cancel := context.WithTimeout(httpReq.Context(), time.Minute*5)
defer cancel()
select { select {
case <-httpReq.Context().Done(): case <-ctx.Done():
// If the caller gives up then return straight away. We don't
// want to attempt to process what they sent us any further.
return util.JSONResponse{Code: http.StatusRequestTimeout} return util.JSONResponse{Code: http.StatusRequestTimeout}
case res := <-v.(chan util.JSONResponse): case res := <-ch.(chan util.JSONResponse):
// The original task just finished processing so let's return
// the result of it.
if res.Code == 0 { if res.Code == 0 {
return util.JSONResponse{Code: http.StatusAccepted} return util.JSONResponse{Code: http.StatusAccepted}
} }
@ -185,10 +192,8 @@ func Send(
} }
// Otherwise, store that we're currently working on this txn from // Otherwise, store that we're currently working on this txn from
// this origin. When we're done processing, close the channel. // this origin. When we're done processing, close the channel.
ch := make(chan util.JSONResponse, 1) defer close(ch.(chan util.JSONResponse))
txnIDs.Store(index, ch) defer inFlightTxnsPerOrigin.Delete(index)
defer close(ch)
defer txnIDs.Delete(index)
t := txnReq{ t := txnReq{
rsAPI: rsAPI, rsAPI: rsAPI,