diff --git a/go.mod b/go.mod index c7e97d2f5..c3de2a855 100644 --- a/go.mod +++ b/go.mod @@ -38,6 +38,7 @@ require ( github.com/twitchtv/twirp v8.1.0+incompatible github.com/urfave/cli/v2 v2.3.0 github.com/urfave/negroni v1.0.0 + go.uber.org/atomic v1.7.0 go.uber.org/zap v1.19.1 google.golang.org/protobuf v1.27.1 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b @@ -70,7 +71,6 @@ require ( github.com/prometheus/common v0.26.0 // indirect github.com/prometheus/procfs v0.6.0 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect - go.uber.org/atomic v1.7.0 // indirect go.uber.org/multierr v1.6.0 // indirect golang.org/x/crypto v0.0.0-20211215153901-e495a2d5b3d3 // indirect golang.org/x/mod v0.5.1 // indirect diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index da402868d..e7c82a3bd 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -4,12 +4,12 @@ import ( "math" "sort" "sync" - "sync/atomic" "time" "github.com/go-logr/logr" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" + "go.uber.org/atomic" "google.golang.org/protobuf/proto" "github.com/livekit/livekit-server/pkg/config" @@ -42,9 +42,10 @@ type Room struct { bufferFactory *buffer.Factory // time the first participant joined the room - joinedAt atomic.Value + joinedAt atomic.Int64 + holds atomic.Int32 // time that the last participant left the room - leftAt atomic.Value + leftAt atomic.Int64 closed chan struct{} closeOnce sync.Once @@ -145,30 +146,38 @@ func (r *Room) GetBufferFactory() *buffer.Factory { } func (r *Room) FirstJoinedAt() int64 { - j := r.joinedAt.Load() - if t, ok := j.(int64); ok { - return t - } - return 0 + return r.joinedAt.Load() } func (r *Room) LastLeftAt() int64 { - l := r.leftAt.Load() - if t, ok := l.(int64); ok { - return t + return r.leftAt.Load() +} + +func (r *Room) Hold() bool { + r.lock.Lock() + defer r.lock.Unlock() + + if r.IsClosed() { + return false } - return 0 + + r.holds.Inc() + return true +} + +func (r *Room) Release() { + r.holds.Dec() } func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions, iceServers []*livekit.ICEServer) error { + r.lock.Lock() + defer r.lock.Unlock() + if r.IsClosed() { prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "room_closed").Add(1) return ErrRoomClosed } - r.lock.Lock() - defer r.lock.Unlock() - if r.participants[participant.Identity()] != nil { prometheus.ServiceOperationCounter.WithLabelValues("participant_join", "error", "already_joined").Add(1) return ErrAlreadyJoined @@ -416,22 +425,18 @@ func (r *Room) IsClosed() bool { // CloseIfEmpty closes the room if all participants had left, or it's still empty past timeout func (r *Room) CloseIfEmpty() { - if r.IsClosed() { + r.lock.Lock() + defer r.lock.Unlock() + + if r.IsClosed() || r.holds.Load() > 0 { return } - r.lock.RLock() - visibleParticipants := 0 for _, p := range r.participants { if !p.Hidden() { - visibleParticipants++ + return } } - r.lock.RUnlock() - - if visibleParticipants > 0 { - return - } timeout := r.Room.EmptyTimeout var elapsed int64 @@ -446,17 +451,24 @@ func (r *Room) CloseIfEmpty() { } if elapsed >= int64(timeout) { - r.Close() + r.closeLocked() } } func (r *Room) Close() { + r.lock.Lock() + defer r.lock.Unlock() + + r.closeLocked() +} + +func (r *Room) closeLocked() { r.closeOnce.Do(func() { - close(r.closed) r.Logger.Infow("closing room") if r.onClose != nil { r.onClose() } + close(r.closed) }) } diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 2ac24fc34..fedefd594 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -182,6 +182,7 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam logger.Errorw("could not create room", err, "room", roomName) return } + defer room.Release() participant := room.GetParticipant(pi.Identity) if participant != nil { @@ -271,26 +272,27 @@ func (r *RoomManager) StartSession(ctx context.Context, roomName livekit.RoomNam if err = r.roomStore.StoreParticipant(ctx, roomName, participant.ToProto()); err != nil { pLogger.Errorw("could not store participant", err) } - // update roomstore with new numParticipants - if !participant.Hidden() { - err = r.roomStore.StoreRoom(ctx, room.Room) - if err != nil { - logger.Errorw("could not store room", err) - } - } - r.telemetry.ParticipantJoined(ctx, room.Room, participant.ToProto(), pi.Client) - participant.OnClose(func(p types.LocalParticipant, disallowedSubscriptions map[livekit.TrackID]livekit.ParticipantID) { - if err := r.roomStore.DeleteParticipant(ctx, roomName, p.Identity()); err != nil { - pLogger.Errorw("could not delete participant", err) - } - // update roomstore with new numParticipants + updateParticipantCount := func() { if !participant.Hidden() { err = r.roomStore.StoreRoom(ctx, room.Room) if err != nil { logger.Errorw("could not store room", err) } } + } + + // update room store with new numParticipants + updateParticipantCount() + + r.telemetry.ParticipantJoined(ctx, room.Room, participant.ToProto(), pi.Client) + participant.OnClose(func(p types.LocalParticipant, disallowedSubscriptions map[livekit.TrackID]livekit.ParticipantID) { + if err := r.roomStore.DeleteParticipant(ctx, roomName, p.Identity()); err != nil { + pLogger.Errorw("could not delete participant", err) + } + + // update room store with new numParticipants + updateParticipantCount() r.telemetry.ParticipantLeft(ctx, room.Room, p.ToProto()) room.RemoveDisallowedSubscriptions(p, disallowedSubscriptions) @@ -311,7 +313,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room room := r.rooms[roomName] r.lock.RUnlock() - if room != nil { + if room != nil && room.Hold() { return room, nil } @@ -323,6 +325,8 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room // construct ice servers room = rtc.NewRoom(ri, *r.rtcConfig, &r.config.Audio, r.telemetry) + room.Hold() + r.telemetry.RoomStarted(ctx, room.Room) room.OnClose(func() { @@ -333,11 +337,13 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room logger.Infow("room closed") }) + room.OnMetadataUpdate(func(metadata string) { if err := r.roomStore.StoreRoom(ctx, room.Room); err != nil { logger.Errorw("could not handle metadata update", err) } }) + room.OnParticipantChanged(func(p types.LocalParticipant) { if p.State() != livekit.ParticipantInfo_DISCONNECTED { if err := r.roomStore.StoreParticipant(ctx, roomName, p.ToProto()); err != nil { @@ -345,6 +351,7 @@ func (r *RoomManager) getOrCreateRoom(ctx context.Context, roomName livekit.Room } } }) + r.lock.Lock() r.rooms[roomName] = room r.lock.Unlock()