diff --git a/go.mod b/go.mod index 53d652423..474e13851 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/jxskiss/base62 v1.1.0 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 github.com/livekit/mediatransportutil v0.0.0-20230130133657-96cfb115473a - github.com/livekit/protocol v1.4.3-0.20230218193429-26f188cb8404 + github.com/livekit/protocol v1.4.3-0.20230222030027-e946da680c7b github.com/livekit/psrpc v0.2.7 github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 github.com/mackerelio/go-osstat v0.2.3 diff --git a/go.sum b/go.sum index 45e6e7e27..09d90a4fc 100644 --- a/go.sum +++ b/go.sum @@ -232,8 +232,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ= github.com/livekit/mediatransportutil v0.0.0-20230130133657-96cfb115473a h1:5UkGQpskXp7HcBmyrCwWtO7ygDWbqtjN09Yva4l/nyE= github.com/livekit/mediatransportutil v0.0.0-20230130133657-96cfb115473a/go.mod h1:1Dlx20JPoIKGP45eo+yuj0HjeE25zmyeX/EWHiPCjFw= -github.com/livekit/protocol v1.4.3-0.20230218193429-26f188cb8404 h1:kVrJYB9o8Cu5MpzIf4+e22auWMSlLhhdDVG3TZw1QQw= -github.com/livekit/protocol v1.4.3-0.20230218193429-26f188cb8404/go.mod h1:mVzmVesPCIgk2gg/jMr6PWtHu8dfRdhaAJ6okFs5nQw= +github.com/livekit/protocol v1.4.3-0.20230222030027-e946da680c7b h1:2toBhQKTJSqoizjUUOPapRX9prKpNcLxSuvhZ9NKeGI= +github.com/livekit/protocol v1.4.3-0.20230222030027-e946da680c7b/go.mod h1:mVzmVesPCIgk2gg/jMr6PWtHu8dfRdhaAJ6okFs5nQw= github.com/livekit/psrpc v0.2.7 h1:j8ns7+t/7LJxTH/jnD9Ds7qn3VkNOV/qCUbt2nbirfU= github.com/livekit/psrpc v0.2.7/go.mod h1:2wtOo1F03vub2qIjx0rAPpVplg873670/LN08o/yopM= github.com/livekit/rtcscore-go v0.0.0-20220815072451-20ee10ae1995 h1:vOaY2qvfLihDyeZtnGGN1Law9wRrw8BMGCr1TygTvMw= diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 77020fbe3..a3f009e60 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -5,7 +5,6 @@ import ( "errors" "io" "math" - "runtime" "sort" "sync" "time" @@ -16,6 +15,7 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/routing" @@ -1169,44 +1169,10 @@ func BroadcastDataPacketForRoom(r types.Room, source types.LocalParticipant, dp destParticpants = append(destParticpants, op) } - if len(destParticpants) < dataForwardLoadBalanceThreshold { - for _, op := range destParticpants { - err := op.SendDataPacket(dp, dpData) - if err != nil && !errors.Is(err, io.ErrClosedPipe) { - logger.Infow("send data packet error", "error", err, "participant", op.Identity()) - } + utils.ParallelExec(destParticpants, dataForwardLoadBalanceThreshold, 1, func(op types.LocalParticipant) { + err := op.SendDataPacket(dp, dpData) + if err != nil && !errors.Is(err, io.ErrClosedPipe) { + op.GetLogger().Infow("send data packet error", "error", err) } - return - } - - // parallel - enables much more efficient multi-core utilization - start := atomic.NewUint64(0) - end := uint64(len(destParticpants)) - - step := uint64(1) - - var wg sync.WaitGroup - numCPU := runtime.NumCPU() - wg.Add(numCPU) - for p := 0; p < numCPU; p++ { - go func() { - defer wg.Done() - for { - n := start.Add(step) - if n >= end+step { - return - } - - for i := n - step; i < n && i < end; i++ { - op := destParticpants[i] - err := op.SendDataPacket(dp, dpData) - if err != nil && !errors.Is(err, io.ErrClosedPipe) { - logger.Infow("send data packet error", "error", err, "participant", op.Identity()) - } - } - } - }() - } - wg.Wait() - + }) } diff --git a/pkg/sfu/downtrackspreader.go b/pkg/sfu/downtrackspreader.go index 7121c2df4..04463eb8d 100644 --- a/pkg/sfu/downtrackspreader.go +++ b/pkg/sfu/downtrackspreader.go @@ -1,13 +1,11 @@ package sfu import ( - "runtime" "sync" - "go.uber.org/atomic" - "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + "github.com/livekit/protocol/utils" ) type DownTrackSpreaderParams struct { @@ -21,18 +19,12 @@ type DownTrackSpreader struct { downTrackMu sync.RWMutex downTracks map[livekit.ParticipantID]TrackSender downTracksShadow []TrackSender - numProcs int } func NewDownTrackSpreader(params DownTrackSpreaderParams) *DownTrackSpreader { d := &DownTrackSpreader{ params: params, downTracks: make(map[livekit.ParticipantID]TrackSender), - numProcs: runtime.NumCPU(), - } - - if runtime.GOMAXPROCS(0) < d.numProcs { - d.numProcs = runtime.GOMAXPROCS(0) } return d @@ -83,39 +75,17 @@ func (d *DownTrackSpreader) HasDownTrack(subscriberID livekit.ParticipantID) boo func (d *DownTrackSpreader) Broadcast(writer func(TrackSender)) { downTracks := d.GetDownTracks() - if d.params.Threshold == 0 || (len(downTracks)) < d.params.Threshold { - // serial - not enough down tracks for parallelization to outweigh overhead - for _, dt := range downTracks { - writer(dt) - } - } else { - // parallel - enables much more efficient multi-core utilization - start := atomic.NewUint64(0) - end := uint64(len(downTracks)) - - // 100µs is enough to amortize the overhead and provide sufficient load balancing. - // WriteRTP takes about 50µs on average, so we write to 2 down tracks per loop. - step := uint64(2) - - var wg sync.WaitGroup - wg.Add(d.numProcs) - for p := 0; p < d.numProcs; p++ { - go func() { - defer wg.Done() - for { - n := start.Add(step) - if n >= end+step { - return - } - - for i := n - step; i < n && i < end; i++ { - writer(downTracks[i]) - } - } - }() - } - wg.Wait() + threshold := uint64(d.params.Threshold) + if threshold == 0 { + threshold = 1000000 } + + // 100µs is enough to amortize the overhead and provide sufficient load balancing. + // WriteRTP takes about 50µs on average, so we write to 2 down tracks per loop. + step := uint64(2) + utils.ParallelExec(downTracks, threshold, step, func(dt TrackSender) { + writer(dt) + }) } func (d *DownTrackSpreader) DownTrackCount() int {