Add config for authoritative match buffers. (#213) (e1252b30) · Commits · public-projects / Nakama

CHANGELOG.md

+7 −0

Original line number	Diff line number	Diff line
		@@ -4,6 +4,13 @@ All notable changes to this project are documented below.
		The format is based on [keep a changelog](http://keepachangelog.com) and this project uses [semantic versioning](http://semver.org).

		## [Unreleased]
		### Added
		- Config option to adjust authoritative match data input queue size.
		- Config option to adjust authoritative match call queue size.

		### Changed
		- Presence list in match join responses no longer contains the user's own presence.
		- Presence list in channel join responses no longer contains the user's own presence.

		## [2.0.1] - 2018-06-15
		### Added

server/config.go

+27 −0

Original line number	Diff line number	Diff line
		@@ -41,6 +41,7 @@ type Config interface {
		GetDatabase() *DatabaseConfig
		GetSocial() *SocialConfig
		GetRuntime() *RuntimeConfig
		GetMatch() *MatchConfig
		GetConsole() *ConsoleConfig
		}

		@@ -111,6 +112,12 @@ func ParseArgs(logger *zap.Logger, args []string) Config {
		if mainConfig.GetRuntime().RegistrySize < 128 {
		logger.Fatal("Runtime instance registry size must be >= 128", zap.Int("runtime.registry_size", mainConfig.GetRuntime().RegistrySize))
		}
		if mainConfig.GetMatch().InputQueueSize < 1 {
		logger.Fatal("Match input queue size must be >= 1", zap.Int("match.input_queue_size", mainConfig.GetMatch().InputQueueSize))
		}
		if mainConfig.GetMatch().CallQueueSize < 1 {
		logger.Fatal("Match call queue size must be >= 1", zap.Int("match.call_queue_size", mainConfig.GetMatch().CallQueueSize))
		}

		// If the runtime path is not overridden, set it to `datadir/modules`.
		if mainConfig.GetRuntime().Path == "" {
		@@ -188,6 +195,7 @@ type config struct {
		Database *DatabaseConfig `yaml:"database" json:"database" usage:"Database connection settings."`
		Social *SocialConfig `yaml:"social" json:"social" usage:"Properties for social provider integrations."`
		Runtime *RuntimeConfig `yaml:"runtime" json:"runtime" usage:"Script Runtime properties."`
		Match *MatchConfig `yaml:"match" json:"match" usage:"Authoritative realtime match properties."`
		Console *ConsoleConfig `yaml:"console" json:"console" usage:"Console settings."`
		}

		@@ -207,6 +215,7 @@ func NewConfig(logger zap.Logger) config {
		Database: NewDatabaseConfig(),
		Social: NewSocialConfig(),
		Runtime: NewRuntimeConfig(),
		Match: NewMatchConfig(),
		Console: NewConsoleConfig(),
		}
		}
		@@ -247,6 +256,10 @@ func (c config) GetRuntime() RuntimeConfig {
		return c.Runtime
		}

		func (c config) GetMatch() MatchConfig {
		return c.Match
		}

		func (c config) GetConsole() ConsoleConfig {
		return c.Console
		}
		@@ -399,6 +412,20 @@ func NewRuntimeConfig() *RuntimeConfig {
		}
		}

		// MatchConfig is configuration relevant to authoritative realtime multiplayer matches.
		type MatchConfig struct {
		InputQueueSize int `yaml:"input_queue_size" json:"input_queue_size" usage:"Size of the authoritative match buffer that stores client messages until they can be processed by the next tick. Default 128."`
		CallQueueSize int `yaml:"call_queue_size" json:"call_queue_size" usage:"Size of the authoritative match buffer that sequences calls to match handler callbacks to ensure no overlaps. Default 128."`
		}

		// NewMatchConfig creates a new MatchConfig struct.
		func NewMatchConfig() *MatchConfig {
		return &MatchConfig{
		InputQueueSize: 128,
		CallQueueSize: 128,
		}
		}

		// ConsoleConfig is configuration relevant to the embedded console.
		type ConsoleConfig struct {
		Port int `yaml:"port" json:"port" usage:"The port for accepting connections for the embedded console, listening on all interfaces."`

server/match_handler.go

+2 −7

Original line number	Diff line number	Diff line
		@@ -28,11 +28,6 @@ import (
		"go.uber.org/zap"
		)

		const (
		InputQueueSize = 128
		CallQueueSize = 128
		)

		type MatchDataMessage struct {
		UserID uuid.UUID
		SessionID uuid.UUID
		@@ -220,9 +215,9 @@ func NewMatchHandler(logger zap.Logger, db sql.DB, config Config, socialClient
		ctx: ctx,
		// Dispatcher below.

		inputCh: make(chan *MatchDataMessage, InputQueueSize),
		inputCh: make(chan *MatchDataMessage, config.GetMatch().InputQueueSize),
		// Ticker below.
		callCh: make(chan func(mh *MatchHandler), CallQueueSize),
		callCh: make(chan func(mh *MatchHandler), config.GetMatch().CallQueueSize),
		stopCh: make(chan struct{}),
		stopped: false,

server/pipeline_channel.go

+12 −7

Original line number	Diff line number	Diff line
		@@ -246,14 +246,19 @@ func (p Pipeline) channelJoin(logger zap.Logger, session Session, envelope *rt
		}
		}

		userPresences := make([]*rtapi.UserPresence, len(presences))
		for i := 0; i < len(presences); i++ {
		userPresences[i] = &rtapi.UserPresence{
		UserId: presences[i].UserID.String(),
		SessionId: presences[i].ID.SessionID.String(),
		Username: presences[i].Meta.Username,
		Persistence: presences[i].Meta.Persistence,
		}
		userPresences := make([]*rtapi.UserPresence, 0, len(presences))
		for _, presence := range presences {
		if isNew && presence.UserID == session.UserID() && presence.ID.SessionID == session.ID() {
		// Ensure the user themselves does not appear in the list of existing channel presences.
		// Only for new joins, not if the user is joining a channel they're already part of.
		continue
		}
		userPresences = append(userPresences, &rtapi.UserPresence{
		UserId: presence.UserID.String(),
		SessionId: presence.ID.SessionID.String(),
		Username: presence.Meta.Username,
		Persistence: presence.Meta.Persistence,
		})
		}

		session.Send(false, 0, &rtapi.Envelope{Cid: envelope.Cid, Message: &rtapi.Envelope_Channel{Channel: &rtapi.Channel{

server/pipeline_match.go

+19 −16

Original line number	Diff line number	Diff line
		@@ -45,19 +45,17 @@ func (p Pipeline) matchCreate(logger zap.Logger, session Session, envelope *rt
		return
		}

		self := &rtapi.UserPresence{
		UserId: session.UserID().String(),
		SessionId: session.ID().String(),
		Username: username,
		}

		session.Send(false, 0, &rtapi.Envelope{Cid: envelope.Cid, Message: &rtapi.Envelope_Match{Match: &rtapi.Match{
		MatchId: fmt.Sprintf("%v.", matchID.String()),
		Authoritative: false,
		// No label.
		Size: 1,
		Presences: []*rtapi.UserPresence{self},
		Self: self,
		// No presences.
		Self: &rtapi.UserPresence{
		UserId: session.UserID().String(),
		SessionId: session.ID().String(),
		Username: username,
		},
		}}})
		}

		@@ -165,7 +163,8 @@ func (p Pipeline) matchJoin(logger zap.Logger, session Session, envelope *rtap

		var label *wrappers.StringValue
		meta := p.tracker.GetLocalBySessionIDStreamUserID(session.ID(), stream, session.UserID())
		if meta == nil {
		isNew := meta == nil
		if isNew {
		username := session.Username()
		found := true
		allow := true
		@@ -215,17 +214,17 @@ func (p Pipeline) matchJoin(logger zap.Logger, session Session, envelope *rtap
		ps := p.tracker.ListByStream(stream, true)
		presences := make([]*rtapi.UserPresence, 0, len(ps))
		for _, p := range ps {
		if isNew && p.UserID == session.UserID() && p.ID.SessionID == session.ID() {
		// Ensure the user themselves does not appear in the list of existing match presences.
		// Only for new joins, not if the user is joining a match they're already part of.
		continue
		}
		presences = append(presences, &rtapi.UserPresence{
		UserId: p.UserID.String(),
		SessionId: p.ID.SessionID.String(),
		Username: p.Meta.Username,
		})
		}
		self := &rtapi.UserPresence{
		UserId: session.UserID().String(),
		SessionId: session.ID().String(),
		Username: meta.Username,
		}

		session.Send(false, 0, &rtapi.Envelope{Cid: envelope.Cid, Message: &rtapi.Envelope_Match{Match: &rtapi.Match{
		MatchId: matchIDString,
		@@ -233,7 +232,11 @@ func (p Pipeline) matchJoin(logger zap.Logger, session Session, envelope *rtap
		Label: label,
		Size: int32(len(presences)),
		Presences: presences,
		Self: self,
		Self: &rtapi.UserPresence{
		UserId: session.UserID().String(),
		SessionId: session.ID().String(),
		Username: meta.Username,
		},
		}}})
		}