Skip to content
This repository has been archived by the owner on Nov 25, 2024. It is now read-only.

Federation API fixes #1899

Merged
merged 5 commits into from
Jul 5, 2021
Merged

Federation API fixes #1899

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
da95342
Ensure worker has work before starting goroutine
neilalexander Jul 5, 2021
d72542f
Revert "Remove processEventWithMissingStateMutex"
neilalexander Jul 5, 2021
c165573
Use request context when processing transactions
neilalexander Jul 5, 2021
59b1df0
Keep goroutine count down by not starting work for things where the c…
neilalexander Jul 5, 2021
5e6fa55
Remove mutex, start workers at correct time
neilalexander Jul 5, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 9 additions & 16 deletions federationapi/routing/send.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -205,7 +205,7 @@ func Send(

util.GetLogger(httpReq.Context()).Infof("Received transaction %q from %q containing %d PDUs, %d EDUs", txnID, request.Origin(), len(t.PDUs), len(t.EDUs))

resp, jsonErr := t.processTransaction(context.Background())
resp, jsonErr := t.processTransaction(httpReq.Context())
if jsonErr != nil {
util.GetLogger(httpReq.Context()).WithField("jsonErr", jsonErr).Error("t.processTransaction failed")
return *jsonErr
Expand Down Expand Up @@ -253,11 +253,8 @@ type txnFederationClient interface {

func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.RespSend, *util.JSONResponse) {
results := make(map[string]gomatrixserverlib.PDUResult)
//var resultsMutex sync.Mutex

var wg sync.WaitGroup
var tasks []*inputTask
wg.Add(1) // for processEDUs

for _, pdu := range t.PDUs {
pduCountTotal.WithLabelValues("total").Inc()
Expand Down Expand Up @@ -313,9 +310,6 @@ func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.Res
input: newSendFIFOQueue(),
})
worker := v.(*inputWorker)
if !worker.running.Load() {
go worker.run()
}
wg.Add(1)
task := &inputTask{
ctx: ctx,
Expand All @@ -325,13 +319,12 @@ func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.Res
}
tasks = append(tasks, task)
worker.input.push(task)
if worker.running.CAS(false, true) {
go worker.run()
}
}

go func() {
defer wg.Done()
t.processEDUs(ctx)
}()

t.processEDUs(ctx)
wg.Wait()

for _, task := range tasks {
Expand All @@ -351,9 +344,6 @@ func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.Res
}

func (t *inputWorker) run() {
if !t.running.CAS(false, true) {
return
}
defer t.running.Store(false)
for {
task, ok := t.input.pop()
Expand All @@ -371,7 +361,10 @@ func (t *inputWorker) run() {
return
default:
evStart := time.Now()
task.err = task.t.processEvent(task.ctx, task.event)
// TODO: Is 5 minutes too long?
ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
task.err = task.t.processEvent(ctx, task.event)
cancel()
task.duration = time.Since(evStart)
if err := task.err; err != nil {
switch err.(type) {
Expand Down