Room closed fix (#370)

* Room closed fix

* switch to uber/atomic

* release after err check

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

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)
 	})
 }
 

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()