From 76b0dc62a73cc1d7adbd7d94727f341cc90742c6 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 31 Jan 2021 14:19:16 -0800 Subject: [PATCH] close rooms after delay instead of immediately after participant departure, to allow reconnects --- pkg/rtc/participant_internal_test.go | 2 + pkg/rtc/room.go | 58 +++++++++++++++++++++++----- pkg/rtc/room_test.go | 47 +++++++++++++++++++--- pkg/service/roommanager.go | 15 ++++++- pkg/service/server.go | 35 +++++++++++++---- test/multinode_test.go | 31 +++++++-------- 6 files changed, 149 insertions(+), 39 deletions(-) diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index f9f2a6566..84e177356 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -12,6 +12,8 @@ import ( "github.com/livekit/livekit-server/proto/livekit" ) +//func Test + func TestIsReady(t *testing.T) { tests := []struct { state livekit.ParticipantInfo_State diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 82fab5558..8fc2ca30f 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -2,6 +2,7 @@ package rtc import ( "sync" + "sync/atomic" "time" "github.com/thoas/go-funk" @@ -12,24 +13,38 @@ import ( "github.com/livekit/livekit-server/proto/livekit" ) +const ( + DefaultEmptyTimeout = 5 * 60 // 5 mins +) + type Room struct { livekit.Room config WebRTCConfig lock sync.RWMutex // map of identity -> Participant participants map[string]types.Participant - hasJoined bool - isClosed utils.AtomicFlag - onClose func() + // time the first participant joined the room + joinedAt atomic.Value + // time that the last participant left the room + leftAt atomic.Value + isClosed utils.AtomicFlag + onClose func() } func NewRoom(room *livekit.Room, config WebRTCConfig) *Room { - return &Room{ + r := &Room{ Room: *room, config: config, lock: sync.RWMutex{}, participants: make(map[string]types.Participant), } + if r.EmptyTimeout == 0 { + r.EmptyTimeout = DefaultEmptyTimeout + } + if r.CreationTime == 0 { + r.CreationTime = time.Now().Unix() + } + return r } func (r *Room) GetParticipant(identity string) types.Participant { @@ -44,6 +59,22 @@ func (r *Room) GetParticipants() []types.Participant { return funk.Values(r.participants).([]types.Participant) } +func (r *Room) FirstJoinedAt() int64 { + j := r.joinedAt.Load() + if t, ok := j.(int64); ok { + return t + } + return 0 +} + +func (r *Room) LastLeftAt() int64 { + l := r.leftAt.Load() + if t, ok := l.(int64); ok { + return t + } + return 0 +} + func (r *Room) Join(participant types.Participant) error { if r.isClosed.Get() { return ErrRoomClosed @@ -54,14 +85,15 @@ func (r *Room) Join(participant types.Participant) error { if r.participants[participant.Identity()] != nil { return ErrAlreadyJoined - } if r.MaxParticipants > 0 && int(r.MaxParticipants) == len(r.participants) { return ErrMaxParticipantsExceeded } - r.hasJoined = true + if r.FirstJoinedAt() == 0 { + r.joinedAt.Store(time.Now().Unix()) + } // it's important to set this before connection, we don't want to miss out on any publishedTracks participant.OnTrackPublished(r.onTrackAdded) @@ -126,7 +158,9 @@ func (r *Room) RemoveParticipant(identity string) { delete(r.participants, identity) - go r.CloseIfEmpty() + if len(r.participants) == 0 { + r.leftAt.Store(time.Now().Unix()) + } } // Close the room if all participants had left, or it's still empty past timeout @@ -143,8 +177,14 @@ func (r *Room) CloseIfEmpty() { return } - elapsed := uint32(time.Now().Unix() - r.CreationTime) - if r.hasJoined || (r.EmptyTimeout > 0 && elapsed >= r.EmptyTimeout) { + var elapsed int64 + if r.FirstJoinedAt() > 0 { + // compute elasped from last departure + elapsed = time.Now().Unix() - r.LastLeftAt() + } else { + elapsed = time.Now().Unix() - r.CreationTime + } + if elapsed >= int64(r.EmptyTimeout) { r.Close() } } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index 7951526c5..bab8119a5 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/assert" + "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/rtc" "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" "github.com/livekit/livekit-server/proto/livekit" @@ -17,15 +18,49 @@ const ( defaultDelay = 10 * time.Millisecond ) +func init() { + logger.InitDevelopment("") +} + +func TestJoinedState(t *testing.T) { + t.Run("new room should return joinedAt 0", func(t *testing.T) { + rm := newRoomWithParticipants(t, 0) + assert.Equal(t, int64(0), rm.FirstJoinedAt()) + assert.Equal(t, int64(0), rm.LastLeftAt()) + }) + + t.Run("should be current time when a participant joins", func(t *testing.T) { + s := time.Now().Unix() + rm := newRoomWithParticipants(t, 1) + assert.Equal(t, s, rm.FirstJoinedAt()) + assert.Equal(t, int64(0), rm.LastLeftAt()) + }) + + t.Run("should be set when a participant leaves", func(t *testing.T) { + rm := newRoomWithParticipants(t, 1) + p0 := rm.GetParticipants()[0] + s := time.Now().Unix() + rm.RemoveParticipant(p0.Identity()) + assert.Equal(t, s, rm.LastLeftAt()) + }) + + t.Run("LastLeftAt should not be set when there are still participants in the room", func(t *testing.T) { + rm := newRoomWithParticipants(t, 2) + p0 := rm.GetParticipants()[0] + rm.RemoveParticipant(p0.Identity()) + assert.EqualValues(t, 0, rm.LastLeftAt()) + }) +} + func TestRoomJoin(t *testing.T) { t.Run("joining returns existing participant data", func(t *testing.T) { rm := newRoomWithParticipants(t, numParticipants) - p := newMockParticipant("new") + pNew := newMockParticipant("new") - rm.Join(p) + rm.Join(pNew) // expect new participant to get a JoinReply - info, participants := p.SendJoinResponseArgsForCall(0) + info, participants := pNew.SendJoinResponseArgsForCall(0) assert.Equal(t, info.Sid, rm.Sid) assert.Len(t, participants, numParticipants) assert.Len(t, rm.GetParticipants(), numParticipants+1) @@ -98,10 +133,13 @@ func TestRoomClosure(t *testing.T) { isClosed = true }) p := rm.GetParticipants()[0] + // allows immediate close after + rm.EmptyTimeout = 0 rm.RemoveParticipant(p.Identity()) time.Sleep(defaultDelay) + rm.CloseIfEmpty() assert.Len(t, rm.GetParticipants(), 0) assert.True(t, isClosed) @@ -114,7 +152,7 @@ func TestRoomClosure(t *testing.T) { rm.OnClose(func() { isClosed = true }) - + assert.NotZero(t, rm.EmptyTimeout) rm.CloseIfEmpty() assert.False(t, isClosed) }) @@ -156,7 +194,6 @@ func TestNewTrack(t *testing.T) { } func newRoomWithParticipants(t *testing.T, num int) *rtc.Room { - rm := rtc.NewRoom( &livekit.Room{Name: "room"}, rtc.WebRTCConfig{}, diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index d22e607e5..7bc22b2ba 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -4,6 +4,8 @@ import ( "sync" "time" + "github.com/thoas/go-funk" + "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/rtc" @@ -98,6 +100,7 @@ func (r *RoomManager) DeleteRoom(roomName string) error { err2 = r.roomStore.DeleteRoom(roomName) }() + wg.Wait() if err != nil { err = err2 } @@ -106,7 +109,7 @@ func (r *RoomManager) DeleteRoom(roomName string) error { } // clean up after old rooms that have been around for awhile -func (r *RoomManager) Cleanup() error { +func (r *RoomManager) CleanupRooms() error { // cleanup rooms that have been left for over a day rooms, err := r.roomStore.ListRooms() if err != nil { @@ -124,6 +127,16 @@ func (r *RoomManager) Cleanup() error { return nil } +func (r *RoomManager) CloseIdleRooms() { + r.lock.RLock() + rooms := funk.Values(r.rooms).([]*rtc.Room) + r.lock.RUnlock() + + for _, room := range rooms { + room.CloseIfEmpty() + } +} + // starts WebRTC session when a new participant is connected, takes place on RTC node func (r *RoomManager) StartSession(roomName, identity string, reconnect bool, requestSource routing.MessageSource, responseSink routing.MessageSink) { room, err := r.getOrCreateRoom(roomName) diff --git a/pkg/service/server.go b/pkg/service/server.go index dd6a79ffe..1c5ca4aa3 100644 --- a/pkg/service/server.go +++ b/pkg/service/server.go @@ -14,6 +14,7 @@ import ( "github.com/livekit/livekit-server/pkg/config" "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/routing" + "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/proto/livekit" "github.com/livekit/livekit-server/version" ) @@ -24,8 +25,9 @@ type LivekitServer struct { rtcService *RTCService httpServer *http.Server router routing.Router + roomManager *RoomManager currentNode routing.LocalNode - running bool + running utils.AtomicFlag doneChan chan bool } @@ -42,6 +44,7 @@ func NewLivekitServer(conf *config.Config, roomServer: livekit.NewRoomServiceServer(roomService), rtcService: rtcService, router: router, + roomManager: roomManager, currentNode: currentNode, } @@ -66,17 +69,17 @@ func NewLivekitServer(conf *config.Config, router.OnNewParticipantRTC(roomManager.StartSession) // clean up old rooms on startup - err = roomManager.Cleanup() + err = roomManager.CleanupRooms() return } func (s *LivekitServer) IsRunning() bool { - return s.running + return s.running.Get() } func (s *LivekitServer) Start() error { - if s.running { + if s.running.Get() { return errors.New("already running") } @@ -104,7 +107,9 @@ func (s *LivekitServer) Start() error { s.httpServer.Serve(ln) }() - s.running = true + go s.backgroundWorker() + + s.running.TrySet(true) <-s.doneChan @@ -120,9 +125,23 @@ func (s *LivekitServer) Start() error { } func (s *LivekitServer) Stop() { - s.running = false - s.router.Stop() - s.doneChan <- true + if s.running.TrySet(false) { + s.router.Stop() + close(s.doneChan) + } +} + +// worker to perform periodic tasks per node +func (s *LivekitServer) backgroundWorker() { + roomTicker := time.NewTicker(30 * time.Second) + for { + select { + case <-s.doneChan: + return + case <-roomTicker.C: + s.roomManager.CloseIdleRooms() + } + } } func configureMiddlewares(handler http.Handler, middlewares ...negroni.Handler) *negroni.Negroni { diff --git a/test/multinode_test.go b/test/multinode_test.go index faef3e4f7..03b5ca094 100644 --- a/test/multinode_test.go +++ b/test/multinode_test.go @@ -1,14 +1,10 @@ package test import ( - "context" "testing" - "github.com/go-redis/redis/v8" "github.com/stretchr/testify/assert" - "github.com/livekit/livekit-server/pkg/routing" - "github.com/livekit/livekit-server/pkg/service" "github.com/livekit/livekit-server/proto/livekit" ) @@ -58,18 +54,21 @@ func TestMultiNodeRouting(t *testing.T) { c1.Stop() c2.Stop() - // ensure that room is closed - rc := redisClient() - ctx := context.Background() - withTimeout(t, "room should be closed", func() bool { - if rc.HGet(ctx, service.RoomsKey, testRoom).Err() == nil { - return false - } - return true - }) - - assert.Equal(t, redis.Nil, rc.HGet(ctx, routing.NodeRoomKey, testRoom).Err()) - assert.Equal(t, redis.Nil, rc.HGet(ctx, service.RoomIdMap, testRoom).Err()) + // TODO: delete room explicitly and ensure it's closed + // + //// ensure that room is closed + // + //rc := redisClient() + //ctx := context.Background() + //withTimeout(t, "room should be closed", func() bool { + // if rc.HGet(ctx, service.RoomsKey, testRoom).Err() == nil { + // return false + // } + // return true + //}) + // + //assert.Equal(t, redis.Nil, rc.HGet(ctx, routing.NodeRoomKey, testRoom).Err()) + //assert.Equal(t, redis.Nil, rc.HGet(ctx, service.RoomIdMap, testRoom).Err()) } func TestConnectWithoutCreation(t *testing.T) {