dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-28 11:44:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

update to utils parallel execute (#1450)

Browse Source 
* log change

* remove identity field

* update to protocol parallel execute

* update protocol
This commit is contained in:
cnderrauber
2023-02-22 11:09:32 +08:00
committed by GitHub
parent 6f326be4f7
commit 1b3d6fad54
4 changed files with 20 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/room.go
+6 -40
View File
@@ -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()
})
}
pkg/sfu/downtrackspreader.go
+11 -41
View File
@@ -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 {