Try to process rooms concurrently in FS /send

federationapi/routing/send.go

@@ -193,6 +193,7 @@ type txnFederationClient interface {
 
 func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.RespSend, *util.JSONResponse) {
 	results := make(map[string]gomatrixserverlib.PDUResult)
+	var resultsMutex sync.Mutex
 
 	pdus := []*gomatrixserverlib.HeaderedEvent{}
 	for _, pdu := range t.PDUs {
@@ -248,6 +249,72 @@ func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.Res
 		pdus = append(pdus, event.Headered(verRes.RoomVersion))
 	}
 
+	perRoom := map[string]chan *gomatrixserverlib.Event{}
+	perCount := map[string]int{}
+	for _, e := range pdus {
+		perCount[e.RoomID()]++
+	}
+	for s, c := range perCount {
+		perRoom[s] = make(chan *gomatrixserverlib.Event, c)
+	}
+	for _, e := range pdus {
+		perRoom[e.RoomID()] <- e.Unwrap()
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(len(perRoom))
+
+	for _, q := range perRoom {
+		go func(q chan *gomatrixserverlib.Event) {
+			defer wg.Done()
+			for e := range q {
+				evStart := time.Now()
+				if err := t.processEvent(ctx, e); err != nil {
+					if isProcessingErrorFatal(err) {
+						sentry.CaptureException(err)
+						// Any other error should be the result of a temporary error in
+						// our server so we should bail processing the transaction entirely.
+						util.GetLogger(ctx).Warnf("Processing %s failed fatally: %s", e.EventID(), err)
+						processEventSummary.WithLabelValues(t.work, MetricsOutcomeFatal).Observe(
+							float64(time.Since(evStart).Nanoseconds()) / 1000.,
+						)
+					} else {
+						// Auth errors mean the event is 'rejected' which have to be silent to appease sytest
+						errMsg := ""
+						outcome := MetricsOutcomeRejected
+						_, rejected := err.(*gomatrixserverlib.NotAllowed)
+						if !rejected {
+							errMsg = err.Error()
+							outcome = MetricsOutcomeFail
+						}
+						util.GetLogger(ctx).WithError(err).WithField("event_id", e.EventID()).WithField("rejected", rejected).Warn(
+							"Failed to process incoming federation event, skipping",
+						)
+						processEventSummary.WithLabelValues(t.work, outcome).Observe(
+							float64(time.Since(evStart).Nanoseconds()) / 1000.,
+						)
+						resultsMutex.Lock()
+						results[e.EventID()] = gomatrixserverlib.PDUResult{
+							Error: errMsg,
+						}
+						resultsMutex.Unlock()
+					}
+				} else {
+					resultsMutex.Lock()
+					results[e.EventID()] = gomatrixserverlib.PDUResult{}
+					resultsMutex.Unlock()
+					pduCountTotal.WithLabelValues("success").Inc()
+					processEventSummary.WithLabelValues(t.work, MetricsOutcomeOK).Observe(
+						float64(time.Since(evStart).Nanoseconds()) / 1000.,
+					)
+				}
+			}
+		}(q)
+	}
+
+	wg.Wait()
+
+	/*
 		// Process the events.
 		for _, e := range pdus {
 			evStart := time.Now()
@@ -304,6 +371,7 @@ func (t *txnReq) processTransaction(ctx context.Context) (*gomatrixserverlib.Res
 				)
 			}
 		}
+	*/
 
 	t.processEDUs(ctx)
 	if c := len(results); c > 0 {