Initial matchmaker hook integration. (43699992) · Commits · public-projects / Nakama

server/matchmaker.go

+7 −321

Original line number	Diff line number	Diff line
		@@ -17,8 +17,6 @@ package server
		import (
		"context"
		"fmt"
		"math"
		"sort"
		"sync"
		"time"

		@@ -286,9 +284,9 @@ func (m LocalMatchmaker) OnMatchedEntries(fn func(entries [][]MatchmakerEntry)
		m.matchedEntriesFn = fn
		}

		func (m *LocalMatchmaker) Process() {
		matchedEntries := make([][]*MatchmakerEntry, 0, 5)
		var customMatchingFn func([][]MatchmakerEntry) [][]MatchmakerEntry

		func (m *LocalMatchmaker) Process() {
		startTime := time.Now()

		m.Lock()
		@@ -306,323 +304,11 @@ func (m *LocalMatchmaker) Process() {
		return
		}

		var threshold bool
		var timer *time.Timer
		if m.revThresholdFn != nil {
		timer = m.revThresholdFn()
		defer timer.Stop()
		}

		for ticket, index := range m.activeIndexes {
		if !threshold && timer != nil {
		select {
		case <-timer.C:
		threshold = true
		default:
		}
		}

		index.Intervals++
		lastInterval := index.Intervals >= m.config.GetMatchmaker().MaxIntervals \|\| index.MinCount == index.MaxCount
		if lastInterval {
		// Drop from active indexes if it has reached its max intervals, or if its min/max counts are equal. In the
		// latter case keeping it active would have the same result as leaving it in the pool, so this saves work.
		delete(m.activeIndexes, ticket)
		}

		if m.active.Load() != 1 {
		continue
		}

		indexQuery := bluge.NewBooleanQuery()

		// Results must match the query string.
		indexQuery.AddMust(index.ParsedQuery)

		// Results must also have compatible min/max ranges, for example 2-4 must not match with 6-8.
		minCountRange := bluge.NewNumericRangeInclusiveQuery(
		float64(index.MinCount), math.Inf(1), true, true).
		SetField("min_count")
		indexQuery.AddMust(minCountRange)
		maxCountRange := bluge.NewNumericRangeInclusiveQuery(
		math.Inf(-1), float64(index.MaxCount), true, true).
		SetField("max_count")
		indexQuery.AddMust(maxCountRange)

		// Results must not include the current party, if any.
		if index.PartyId != "" {
		partyIdQuery := bluge.NewTermQuery(index.PartyId)
		partyIdQuery.SetField("party_id")
		indexQuery.AddMustNot(partyIdQuery)
		}

		searchRequest := bluge.NewTopNSearch(len(m.indexes), indexQuery)
		// Sort results to try and select the best match, or if the
		// matches are equivalent, the longest waiting tickets first.
		searchRequest.SortBy([]string{"-_score", "created_at"})

		indexReader, err := m.indexWriter.Reader()
		if err != nil {
		m.logger.Error("error accessing index reader", zap.Error(err))
		continue
		}

		result, err := indexReader.Search(m.ctx, searchRequest)
		if err != nil {
		_ = indexReader.Close()
		m.logger.Error("error searching index", zap.Error(err))
		continue
		}

		blugeMatches, err := IterateBlugeMatches(result, map[string]struct{}{}, m.logger)
		if err != nil {
		_ = indexReader.Close()
		m.logger.Error("error iterating search results", zap.Error(err))
		continue
		}

		err = indexReader.Close()
		if err != nil {
		m.logger.Error("error closing index reader", zap.Error(err))
		continue
		}

		for idx, hit := range blugeMatches.Hits {
		if hit.ID == ticket {
		// Remove the current ticket.
		blugeMatches.Hits = append(blugeMatches.Hits[:idx], blugeMatches.Hits[idx+1:]...)
		break
		}
		}

		// Form possible combinations, in case multiple matches might be suitable.
		entryCombos := make([][]*MatchmakerEntry, 0, 5)
		lastHitCounter := len(blugeMatches.Hits) - 1
		for hitCounter, hit := range blugeMatches.Hits {
		hitIndex, ok := m.indexes[hit.ID]
		if !ok {
		// Ticket did not exist, should not happen.
		m.logger.Warn("matchmaker process missing index", zap.String("ticket", hit.ID))
		continue
		}

		if !threshold && m.config.GetMatchmaker().RevPrecision {
		outerMutualMatch, err := validateMatch(m, indexReader, hitIndex.ParsedQuery, hit.ID, ticket)
		if err != nil {
		m.logger.Error("error validating mutual match", zap.Error(err))
		continue
		} else if !outerMutualMatch {
		// This search hit is not a mutual match with the outer ticket.
		continue
		}
		}

		if index.MaxCount < hitIndex.MaxCount && hitIndex.Intervals <= m.config.GetMatchmaker().MaxIntervals {
		// This match would be less than the search hit's preferred max, and they can still wait. Let them wait more.
		continue
		}

		// Check if there are overlapping session IDs, and if so these tickets are ineligible to match together.
		var sessionIdConflict bool
		for sessionID := range index.SessionIDs {
		if _, found := hitIndex.SessionIDs[sessionID]; found {
		sessionIdConflict = true
		break
		}
		}
		if sessionIdConflict {
		continue
		}

		entries, ok := m.entries[hit.ID]
		if !ok {
		// Ticket did not exist, should not happen.
		m.logger.Warn("matchmaker process missing entries", zap.String("ticket", hit.ID))
		continue
		}

		var foundComboIdx int
		var foundCombo []*MatchmakerEntry
		var mutualMatchConflict bool
		for entryComboIdx, entryCombo := range entryCombos {
		if len(entryCombo)+len(entries)+index.Count <= index.MaxCount {
		// There is room in this combo for these entries. Check if there are session ID conflicts with current combo.
		for _, entry := range entryCombo {
		if _, found := hitIndex.SessionIDs[entry.Presence.SessionId]; found {
		sessionIdConflict = true
		break
		}
		if !threshold && m.config.GetMatchmaker().RevPrecision {
		entryMatchesSearchHitQuery, err := validateMatch(m, indexReader, hitIndex.ParsedQuery, hit.ID, entry.Ticket)
		if err != nil {
		mutualMatchConflict = true
		m.logger.Error("error validating mutual match", zap.Error(err))
		break
		} else if !entryMatchesSearchHitQuery {
		mutualMatchConflict = true
		// This search hit is not a mutual match with the outer ticket.
		break
		}
		// MatchmakerEntry does not have the query, read it out of indexes.
		if entriesIndexEntry, ok := m.indexes[entry.Ticket]; ok {
		searchHitMatchesEntryQuery, err := validateMatch(m, indexReader, entriesIndexEntry.ParsedQuery, entry.Ticket, hit.ID)
		if err != nil {
		mutualMatchConflict = true
		m.logger.Error("error validating mutual match", zap.Error(err))
		break
		} else if !searchHitMatchesEntryQuery {
		mutualMatchConflict = true
		// This search hit is not a mutual match with the outer ticket.
		break
		}
		var matchedEntries [][]*MatchmakerEntry
		if customMatchingFn != nil {
		matchedEntries = m.processCustom(customMatchingFn)
		} else {
		m.logger.Warn("matchmaker missing index entry for entry combo")
		}
		}
		}
		if sessionIdConflict \|\| mutualMatchConflict {
		continue
		}

		entryCombo = append(entryCombo, entries...)
		entryCombos[entryComboIdx] = entryCombo

		foundCombo = entryCombo
		foundComboIdx = entryComboIdx
		break
		}
		}
		// Either processing first hit, or current hit entries combined with previous hits may tip over index.MaxCount.
		if foundCombo == nil {
		entryCombo := make([]*MatchmakerEntry, len(entries))
		copy(entryCombo, entries)
		entryCombos = append(entryCombos, entryCombo)

		foundCombo = entryCombo
		foundComboIdx = len(entryCombos) - 1
		}

		// The combo is considered match-worthy if either the max count has been satisfied, or ALL of these conditions are met:
		// * It is the last interval for this active index.
		// * The combo at least satisfies the min count.
		// * The combo does not exceed the max count.
		// * There are no more hits that may further fill the found combo, so we get as close as possible to the max count.
		if l := len(foundCombo) + index.Count; l == index.MaxCount \|\| (lastInterval && l >= index.MinCount && l <= index.MaxCount && hitCounter >= lastHitCounter) {
		if rem := l % index.CountMultiple; rem != 0 {
		// The size of the combination being considered does not satisfy the count multiple.
		// Attempt to adjust the combo by removing the smallest possible number of entries.
		// Prefer keeping entries that have been in the matchmaker the longest, if possible.
		eligibleIndexesUniq := make(map[*MatchmakerIndex]struct{}, len(foundCombo))
		for _, e := range foundCombo {
		// Only tickets individually less <= the removable size are considered.
		// For example removing a party of 3 when we're only looking to remove 2 is not allowed.
		if foundIndex, ok := m.indexes[e.Ticket]; ok && foundIndex.Count <= rem {
		eligibleIndexesUniq[foundIndex] = struct{}{}
		}
		}

		eligibleIndexes := make([]*MatchmakerIndex, 0, len(eligibleIndexesUniq))
		for idx := range eligibleIndexesUniq {
		eligibleIndexes = append(eligibleIndexes, idx)
		}

		eligibleGroups := groupIndexes(eligibleIndexes, rem)
		if len(eligibleGroups) <= 0 {
		// No possible combination to remove, unlikely but guard.
		continue
		}
		// Sort to ensure we keep as many of the longest-waiting tickets as possible.
		sort.Slice(eligibleGroups, func(i, j int) bool {
		return eligibleGroups[i].avgCreatedAt < eligibleGroups[j].avgCreatedAt
		})
		// The most eligible group is removed from the combo.
		for _, egIndex := range eligibleGroups[0].indexes {
		for i := 0; i < len(foundCombo); i++ {
		if egIndex.Ticket == foundCombo[i].Ticket {
		foundCombo[i] = foundCombo[len(foundCombo)-1]
		foundCombo[len(foundCombo)-1] = nil
		foundCombo = foundCombo[:len(foundCombo)-1]
		i--
		}
		}
		}

		// We've removed something, update the known size of the currently considered combo.
		l = len(foundCombo) + index.Count

		if l%index.CountMultiple != 0 {
		// Removal was insufficient, the combo is still not valid for the required multiple.
		continue
		}
		}

		// Check that ALL of these conditions are true for ALL matched entries:
		// * The found combo size satisfies the minimum count.
		// * The found combo size satisfies the maximum count.
		// * The found combo size satisfies the count multiple.
		// For any condition failures it does not matter which specific condition is not met.
		var conditionFailed bool
		for _, e := range foundCombo {
		if foundIndex, ok := m.indexes[e.Ticket]; ok && (foundIndex.MinCount > l \|\| foundIndex.MaxCount < l \|\| l%foundIndex.CountMultiple != 0) {
		conditionFailed = true
		break
		}
		}
		if conditionFailed {
		continue
		}

		// Found a suitable match.
		entries, ok := m.entries[ticket]
		if !ok {
		// Ticket did not exist, should not happen.
		m.logger.Warn("matchmaker process missing entries", zap.String("ticket", hit.ID))
		break
		}
		currentMatchedEntries := append(foundCombo, entries...)

		// Remove the found combos from currently tracked list.
		entryCombos = append(entryCombos[:foundComboIdx], entryCombos[foundComboIdx+1:]...)

		matchedEntries = append(matchedEntries, currentMatchedEntries)

		// Remove all entries/indexes that have just matched. It must be done here so any following process iterations
		// cannot pick up the same tickets to match against.
		ticketsToDelete := make(map[string]struct{}, len(currentMatchedEntries))
		for _, entry := range currentMatchedEntries {
		if _, ok := ticketsToDelete[entry.Ticket]; !ok {
		m.batch.Delete(bluge.Identifier(entry.Ticket))
		ticketsToDelete[entry.Ticket] = struct{}{}
		}
		delete(m.entries, entry.Ticket)
		delete(m.indexes, entry.Ticket)
		delete(m.activeIndexes, entry.Ticket)
		delete(m.revCache, entry.Ticket)
		if sessionTickets, ok := m.sessionTickets[entry.Presence.SessionId]; ok {
		if l := len(sessionTickets); l <= 1 {
		delete(m.sessionTickets, entry.Presence.SessionId)
		} else {
		delete(sessionTickets, entry.Ticket)
		}
		}
		if entry.PartyId != "" {
		if partyTickets, ok := m.partyTickets[entry.PartyId]; ok {
		if l := len(partyTickets); l <= 1 {
		delete(m.partyTickets, entry.PartyId)
		} else {
		delete(partyTickets, entry.Ticket)
		}
		}
		}
		}
		if err := m.indexWriter.Batch(m.batch); err != nil {
		m.logger.Error("error deleting matchmaker process entries batch", zap.Error(err))
		}
		m.batch.Reset()

		break
		}
		}
		matchedEntries = m.processDefault()
		}

		m.Unlock()

server/matchmaker_process.go

0 → 100644

+653 −0

File added.

Preview size limit exceeded, changes collapsed.