This commit is contained in:
Ben Sarmiento
2024-05-23 22:20:19 +02:00
parent 8636a0569d
commit beba993364
3 changed files with 78 additions and 36 deletions

View File

@@ -41,10 +41,12 @@ type TorrentManager struct {
latestState *LibraryState latestState *LibraryState
repairTrigger chan *Torrent repairTrigger chan *Torrent
repairSet mapset.Set[*Torrent] repairQueue mapset.Set[*Torrent]
repairRunning bool repairRunning bool
repairRunningMu sync.Mutex repairRunningMu sync.Mutex
trashBin mapset.Set[string] trashBin mapset.Set[string]
repairBin mapset.Set[string] // same as trash bin, but only if the torrent has been downloaded
} }
// NewTorrentManager creates a new torrent manager // NewTorrentManager creates a new torrent manager

View File

@@ -3,6 +3,7 @@ package torrent
import ( import (
"context" "context"
"fmt" "fmt"
"os"
"path/filepath" "path/filepath"
"strings" "strings"
"sync" "sync"
@@ -33,8 +34,6 @@ func (t *TorrentManager) refreshTorrents() []string {
freshIDs := mapset.NewSet[string]() freshIDs := mapset.NewSet[string]()
freshAccessKeys := mapset.NewSet[string]() freshAccessKeys := mapset.NewSet[string]()
t.log.Infof("Getting info of %d torrents", len(instances))
for i := range instances { for i := range instances {
wg.Add(1) wg.Add(1)
idx := i idx := i
@@ -156,7 +155,7 @@ func (t *TorrentManager) refreshTorrents() []string {
// StartRefreshJob periodically refreshes the torrents // StartRefreshJob periodically refreshes the torrents
func (t *TorrentManager) StartRefreshJob() { func (t *TorrentManager) StartRefreshJob() {
go func() { go func() {
t.log.Info("Starting periodic refresh job") t.log.Debug("Starting periodic refresh job")
refreshTicker := time.NewTicker(time.Duration(t.Config.GetRefreshEverySecs()) * time.Second) refreshTicker := time.NewTicker(time.Duration(t.Config.GetRefreshEverySecs()) * time.Second)
defer refreshTicker.Stop() defer refreshTicker.Stop()
@@ -417,3 +416,35 @@ func (t *TorrentManager) trash(torrentId string) {
t.log.Debugf("Trash: %s", torrentId) t.log.Debugf("Trash: %s", torrentId)
t.trashBin.Add(torrentId) t.trashBin.Add(torrentId)
} }
func (t *TorrentManager) trashOnceCompleted(torrentId string) {
t.log.Debugf("Trash once completed: %s", torrentId)
t.repairBin.Add(torrentId)
}
func (t *TorrentManager) saveToBinFile() {
data := map[string]interface{}{
"trash_bin": t.trashBin.ToSlice(), // Assuming trashBin is a mapset.Set[string]
"repair_bin": t.repairBin.ToSlice(), // Assuming repairBin is a mapset.Set[string]
}
jsonData, err := json.Marshal(data)
if err != nil {
t.log.Errorf("Failed to marshal bin data: %v", err)
return
}
file, err := os.Create("trash_bin")
if err != nil {
t.log.Errorf("Failed to create trash_bin file: %v", err)
return
}
defer file.Close()
_, err = file.Write(jsonData)
if err != nil {
t.log.Errorf("Failed to write to trash_bin file: %v", err)
} else {
t.log.Debug("Successfully saved bins to file")
}
}

View File

@@ -24,10 +24,10 @@ func (t *TorrentManager) StartRepairJob() {
return return
} }
t.repairTrigger = make(chan *Torrent) t.repairTrigger = make(chan *Torrent)
t.repairSet = mapset.NewSet[*Torrent]() t.repairQueue = mapset.NewSet[*Torrent]()
// there is 1 repair worker, with max 1 blocking task // there is 1 repair worker, with max 1 blocking task
go func() { go func() {
t.log.Info("Starting periodic repair job") t.log.Debug("Starting periodic repair job")
repairTicker := time.NewTicker(time.Duration(t.Config.GetRepairEveryMins()) * time.Minute) repairTicker := time.NewTicker(time.Duration(t.Config.GetRepairEveryMins()) * time.Minute)
defer repairTicker.Stop() defer repairTicker.Stop()
@@ -50,7 +50,7 @@ func (t *TorrentManager) invokeRepair(torrent *Torrent) {
t.repairRunningMu.Lock() t.repairRunningMu.Lock()
if t.repairRunning { if t.repairRunning {
t.repairRunningMu.Unlock() t.repairRunningMu.Unlock()
t.repairSet.Add(torrent) t.repairQueue.Add(torrent)
// don't do anything if repair is already running // don't do anything if repair is already running
return return
} }
@@ -66,9 +66,9 @@ func (t *TorrentManager) invokeRepair(torrent *Torrent) {
t.repairRunning = false t.repairRunning = false
t.repairRunningMu.Unlock() t.repairRunningMu.Unlock()
// before we let go, let's check repairSet // before we let go, let's check repairQueue
t.workerPool.Submit(func() { t.workerPool.Submit(func() {
queuedTorrent, exists := t.repairSet.Pop() queuedTorrent, exists := t.repairQueue.Pop()
if exists { if exists {
t.TriggerRepair(queuedTorrent) t.TriggerRepair(queuedTorrent)
} }
@@ -85,7 +85,7 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
var haystack cmap.ConcurrentMap[string, *Torrent] var haystack cmap.ConcurrentMap[string, *Torrent]
if torrent == nil { if torrent == nil {
haystack, _ = t.DirectoryMap.Get(INT_ALL) haystack, _ = t.DirectoryMap.Get(INT_ALL)
t.log.Info("Periodic repair started; searching for broken torrents") t.log.Debug("Periodic repair started; searching for broken torrents")
} else { } else {
haystack = cmap.New[*Torrent]() haystack = cmap.New[*Torrent]()
haystack.Set("", torrent) haystack.Set("", torrent)
@@ -94,7 +94,11 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
// collect all torrents that need to be repaired // collect all torrents that need to be repaired
toRepair := mapset.NewSet[*Torrent]() toRepair := mapset.NewSet[*Torrent]()
var wg sync.WaitGroup
haystack.IterCb(func(_ string, torrent *Torrent) { haystack.IterCb(func(_ string, torrent *Torrent) {
wg.Add(1)
_ = t.workerPool.Submit(func() {
defer wg.Done()
if torrent.UnrepairableReason != "" { if torrent.UnrepairableReason != "" {
return return
} }
@@ -106,7 +110,7 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
} }
}) })
if brokenFileCount > 0 { if brokenFileCount > 0 {
t.log.Debugf("Torrent %s has %d broken files, adding to repair list", t.GetKey(torrent), brokenFileCount) t.log.Debugf("Torrent %s has %d/%d broken files, adding to repair list", t.GetKey(torrent), brokenFileCount, torrent.SelectedFiles.Count())
toRepair.Add(torrent) toRepair.Add(torrent)
return return
} }
@@ -118,8 +122,10 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
return return
} }
}) })
})
wg.Wait()
var wg sync.WaitGroup
toRepair.Each(func(torrent *Torrent) bool { toRepair.Each(func(torrent *Torrent) bool {
wg.Add(1) wg.Add(1)
t.Repair(torrent, &wg) t.Repair(torrent, &wg)
@@ -127,7 +133,7 @@ func (t *TorrentManager) repairAll(torrent *Torrent) {
}) })
wg.Wait() wg.Wait()
t.log.Infof("Finished repair sequence for %d broken torrents", toRepair.Cardinality()) t.log.Infof("Finished periodic repair sequence for %d broken torrent(s)", toRepair.Cardinality())
} }
func (t *TorrentManager) Repair(torrent *Torrent, wg *sync.WaitGroup) { func (t *TorrentManager) Repair(torrent *Torrent, wg *sync.WaitGroup) {
@@ -206,7 +212,7 @@ func (t *TorrentManager) repair(torrent *Torrent) {
return return
} }
t.log.Infof("Repairing by downloading %d batches of the %d broken files of torrent %s", int(math.Ceil(float64(len(brokenFiles))/100)), len(brokenFiles), t.GetKey(torrent)) t.log.Infof("Torrent %s will be repaired by downloading %d batches of the %d broken files", int(math.Ceil(float64(len(brokenFiles))/100)), len(brokenFiles), t.GetKey(torrent))
newlyDownloadedIds := make([]string, 0) newlyDownloadedIds := make([]string, 0)
batchNum := 1 batchNum := 1
@@ -233,7 +239,7 @@ func (t *TorrentManager) repair(torrent *Torrent) {
t.log.Debugf("Downloading last batch of broken files of torrent %s", t.GetKey(torrent)) t.log.Debugf("Downloading last batch of broken files of torrent %s", t.GetKey(torrent))
if len(group) > 0 { if len(group) > 0 {
_, err := t.redownloadTorrent(torrent, group) redownloadedInfo, err := t.redownloadTorrent(torrent, group)
if err != nil { if err != nil {
t.log.Warnf("Cannot repair torrent %s by downloading broken files (error=%s) giving up", t.GetKey(torrent), err.Error()) t.log.Warnf("Cannot repair torrent %s by downloading broken files (error=%s) giving up", t.GetKey(torrent), err.Error())
for _, newId := range newlyDownloadedIds { for _, newId := range newlyDownloadedIds {
@@ -241,9 +247,12 @@ func (t *TorrentManager) repair(torrent *Torrent) {
} }
return return
} }
newlyDownloadedIds = append(newlyDownloadedIds, redownloadedInfo.ID)
} }
/// TODO: should we delete the old torrents that were replaced? for _, newId := range newlyDownloadedIds {
t.trashOnceCompleted(newId)
}
} }
func (t *TorrentManager) assignUnassignedLinks(torrent *Torrent) bool { func (t *TorrentManager) assignUnassignedLinks(torrent *Torrent) bool {