Cutlery/Kyoo
1
0
Fork 0
mirror of https://github.com/zoriya/Kyoo.git synced 2025-06-03 13:44:33 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix tracker

Browse Source
This commit is contained in:
Zoe Roux 2024-08-08 15:43:54 +02:00
parent 70147e4fe8
commit e623818585
2 changed files with 39 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
transcoder/src/tracker.go
View File

@ -7,6 +7,7 @@ import (
type ClientInfo struct { type ClientInfo struct {
client string client string
sha string
path string path string
video *VideoKey video *VideoKey
audio *uint32 audio *uint32
@ -19,7 +20,7 @@ type Tracker struct {
clients map[string]ClientInfo clients map[string]ClientInfo
// key: client_id // key: client_id
visitDate map[string]time.Time visitDate map[string]time.Time
// key: path // key: sha
lastUsage map[string]time.Time lastUsage map[string]time.Time
transcoder *Transcoder transcoder *Transcoder
deletedStream chan string deletedStream chan string
@ -56,7 +57,7 @@ func (t *Tracker) start() {
old, ok := t.clients[info.client] old, ok := t.clients[info.client]
// First fixup the info. Most routes ruturn partial infos // First fixup the info. Most routes ruturn partial infos
if ok && old.path == info.path { if ok && old.sha == info.sha {
if info.video == nil { if info.video == nil {
info.video = old.video info.video = old.video
} }
@ -73,24 +74,24 @@ func (t *Tracker) start() {
t.clients[info.client] = info t.clients[info.client] = info
t.visitDate[info.client] = time.Now() t.visitDate[info.client] = time.Now()
t.lastUsage[info.path] = time.Now() t.lastUsage[info.sha] = time.Now()
// now that the new info is stored and fixed, kill old streams // now that the new info is stored and fixed, kill old streams
if ok && old.path == info.path { if ok && old.sha == info.sha {
if old.audio != info.audio && old.audio != nil { if old.audio != nil && (info.audio == nil || *info.audio != *old.audio) {
t.KillAudioIfDead(old.path, *old.audio) t.KillAudioIfDead(old.sha, old.path, *old.audio)
} }
if old.video != info.video && old.video != nil { if old.video != nil && (info.video == nil || *info.video != *old.video) {
t.KillVideoIfDead(old.path, *old.video) t.KillVideoIfDead(old.sha, old.path, *old.video)
} }
if old.vhead != -1 && Abs(info.vhead-old.vhead) > 100 { if old.vhead != -1 && Abs(info.vhead-old.vhead) > 100 {
t.KillOrphanedHeads(old.path, old.video, nil) t.KillOrphanedHeads(old.sha, old.video, nil)
} }
if old.ahead != -1 && Abs(info.ahead-old.ahead) > 100 { if old.ahead != -1 && Abs(info.ahead-old.ahead) > 100 {
t.KillOrphanedHeads(old.path, nil, old.audio) t.KillOrphanedHeads(old.sha, nil, old.audio)
} }
} else if ok { } else if ok {
t.KillStreamIfDead(old.path) t.KillStreamIfDead(old.sha, old.path)
} }
case <-timer: case <-timer:
@ -105,11 +106,11 @@ func (t *Tracker) start() {
delete(t.clients, client) delete(t.clients, client)
delete(t.visitDate, client) delete(t.visitDate, client)
if !t.KillStreamIfDead(info.path) { if !t.KillStreamIfDead(info.sha, info.path) {
audio_cleanup := info.audio != nil && t.KillAudioIfDead(info.path, *info.audio) audio_cleanup := info.audio != nil && t.KillAudioIfDead(info.sha, info.path, *info.audio)
video_cleanup := info.video != nil && t.KillVideoIfDead(info.path, *info.video) video_cleanup := info.video != nil && t.KillVideoIfDead(info.sha, info.path, *info.video)
if !audio_cleanup || !video_cleanup { if !audio_cleanup || !video_cleanup {
t.KillOrphanedHeads(info.path, info.video, info.audio) t.KillOrphanedHeads(info.sha, info.video, info.audio)
} }
} }
} }
@ -119,46 +120,46 @@ func (t *Tracker) start() {
} }
} }
func (t *Tracker) KillStreamIfDead(path string) bool { func (t *Tracker) KillStreamIfDead(sha string, path string) bool {
for _, stream := range t.clients { for _, stream := range t.clients {
if stream.path == path { if stream.sha == sha {
return false return false
} }
} }
log.Printf("Nobody is watching %s. Killing it", path) log.Printf("Nobody is watching %s. Killing it", path)
stream, ok := t.transcoder.streams.Get(path) stream, ok := t.transcoder.streams.Get(sha)
if !ok { if !ok {
return false return false
} }
stream.Kill() stream.Kill()
go func() { go func() {
time.Sleep(4 * time.Hour) time.Sleep(4 * time.Hour)
t.deletedStream <- path t.deletedStream <- sha
}() }()
return true return true
} }
func (t *Tracker) DestroyStreamIfOld(path string) { func (t *Tracker) DestroyStreamIfOld(sha string) {
if time.Since(t.lastUsage[path]) < 4*time.Hour { if time.Since(t.lastUsage[sha]) < 4*time.Hour {
return return
} }
stream, ok := t.transcoder.streams.GetAndRemove(path) stream, ok := t.transcoder.streams.GetAndRemove(sha)
if !ok { if !ok {
return return
} }
stream.Destroy() stream.Destroy()
} }
func (t *Tracker) KillAudioIfDead(path string, audio uint32) bool { func (t *Tracker) KillAudioIfDead(sha string, path string, audio uint32) bool {
for _, stream := range t.clients { for _, stream := range t.clients {
if stream.path == path && stream.audio != nil && *stream.audio == audio { if stream.sha == sha && stream.audio != nil && *stream.audio == audio {
return false return false
} }
} }
log.Printf("Nobody is listening audio %d of %s. Killing it", audio, path) log.Printf("Nobody is listening audio %d of %s. Killing it", audio, path)
stream, ok := t.transcoder.streams.Get(path) stream, ok := t.transcoder.streams.Get(sha)
if !ok { if !ok {
return false return false
} }
@ -170,15 +171,15 @@ func (t *Tracker) KillAudioIfDead(path string, audio uint32) bool {
return true return true
} }
func (t *Tracker) KillVideoIfDead(path string, video VideoKey) bool { func (t *Tracker) KillVideoIfDead(sha string, path string, video VideoKey) bool {
for _, stream := range t.clients { for _, stream := range t.clients {
if stream.path == path && stream.video != nil && *stream.video == video { if stream.sha == sha && stream.video != nil && *stream.video == video {
return false return false
} }
} }
log.Printf("Nobody is watching %s video %d quality %s. Killing it", path, video.idx, video.quality) log.Printf("Nobody is watching %s video %d quality %s. Killing it", path, video.idx, video.quality)
stream, ok := t.transcoder.streams.Get(path) stream, ok := t.transcoder.streams.Get(sha)
if !ok { if !ok {
return false return false
} }
@ -190,8 +191,8 @@ func (t *Tracker) KillVideoIfDead(path string, video VideoKey) bool {
return true return true
} }
func (t *Tracker) KillOrphanedHeads(path string, video *VideoKey, audio *uint32) { func (t *Tracker) KillOrphanedHeads(sha string, video *VideoKey, audio *uint32) {
stream, ok := t.transcoder.streams.Get(path) stream, ok := t.transcoder.streams.Get(sha)
if !ok { if !ok {
return return
} }
@ -221,9 +222,9 @@ func (t *Tracker) killOrphanedeheads(stream *Stream, is_video bool) {
distance := int32(99999) distance := int32(99999)
for _, info := range t.clients { for _, info := range t.clients {
ihead := info.vhead ihead := info.ahead
if is_video { if is_video {
ihead = info.ahead ihead = info.vhead
} }
distance = min(Abs(ihead-head.segment), distance) distance = min(Abs(ihead-head.segment), distance)
} }

7
transcoder/src/transcoder.go
View File

@ -6,7 +6,7 @@ import (
) )
type Transcoder struct { type Transcoder struct {
// All file streams currently running, index is file path // All file streams currently running, index is sha
streams CMap[string, *FileStream] streams CMap[string, *FileStream]
clientChan chan ClientInfo clientChan chan ClientInfo
tracker *Tracker tracker *Tracker
@ -54,6 +54,7 @@ func (t *Transcoder) GetMaster(path string, client string, sha string) (string,
} }
t.clientChan <- ClientInfo{ t.clientChan <- ClientInfo{
client: client, client: client,
sha: sha,
path: path, path: path,
video: nil, video: nil,
audio: nil, audio: nil,
@ -76,6 +77,7 @@ func (t *Transcoder) GetVideoIndex(
} }
t.clientChan <- ClientInfo{ t.clientChan <- ClientInfo{
client: client, client: client,
sha: sha,
path: path, path: path,
video: &VideoKey{video, quality}, video: &VideoKey{video, quality},
audio: nil, audio: nil,
@ -97,6 +99,7 @@ func (t *Transcoder) GetAudioIndex(
} }
t.clientChan <- ClientInfo{ t.clientChan <- ClientInfo{
client: client, client: client,
sha: sha,
path: path, path: path,
audio: &audio, audio: &audio,
vhead: -1, vhead: -1,
@ -119,6 +122,7 @@ func (t *Transcoder) GetVideoSegment(
} }
t.clientChan <- ClientInfo{ t.clientChan <- ClientInfo{
client: client, client: client,
sha: sha,
path: path, path: path,
video: &VideoKey{video, quality}, video: &VideoKey{video, quality},
vhead: segment, vhead: segment,
@ -141,6 +145,7 @@ func (t *Transcoder) GetAudioSegment(
} }
t.clientChan <- ClientInfo{ t.clientChan <- ClientInfo{
client: client, client: client,
sha: sha,
path: path, path: path,
audio: &audio, audio: &audio,
ahead: segment, ahead: segment,