LBRYCommunity/ytsync
1
0
Fork 0
Code Issues 39 Pull requests 8 Projects Releases 197 Packages Wiki Activity

update synced videos map on progress

Browse source 
add locking
fix bugs
This commit is contained in:
Niko Storni 2018-08-03 13:21:42 -04:00
parent e15476b4f8
commit 24aca302ff
3 changed files with 23 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
manager.go
View file

@ -270,7 +270,7 @@ func (s SyncManager) Start() error {
}
SendInfoToSlack("A non fatal error was reported by the sync process. %s\nContinuing...", err.Error())
}
SendInfoToSlack("Syncing %s reached an end. (Iteration %d/%d - total session iterations: %d))", sync.LbryChannelName, i, len(syncs), syncCount)
SendInfoToSlack("Syncing %s reached an end. (Iteration %d/%d - total session iterations: %d))", sync.LbryChannelName, i+1, len(syncs), syncCount)
syncCount++
if sync.IsInterrupted() || (s.Limit != 0 && syncCount >= s.Limit) {
shouldInterruptLoop = true

2
setup.go
View file

@ -44,7 +44,9 @@ func (s *Sync) walletSetup() error {
return nil
}
s.videosMapMux.Lock()
numPublished := uint64(len(s.syncedVideos)) //should we only count published videos? Credits are allocated even for failed ones...
s.videosMapMux.Unlock()
log.Debugf("We already published %d videos", numPublished)
if float64(numOnSource)-float64(numPublished) > float64(s.Manager.VideosLimit) {

20
ytsync.go
View file

@ -68,11 +68,22 @@ type Sync struct {
syncedVideos map[string]syncedVideo
grp *stop.Group
videosMapMux sync.Mutex
mux sync.Mutex
wg sync.WaitGroup
queue chan video
}
func (s *Sync) AppendSyncedVideo(videoID string, published bool, failureReason string) {
s.videosMapMux.Lock()
defer s.videosMapMux.Unlock()
s.syncedVideos[videoID] = syncedVideo{
VideoID: videoID,
Published: published,
FailureReason: failureReason,
}
}
// SendErrorToSlack Sends an error message to the default channel and to the process log.
func SendErrorToSlack(format string, a ...interface{}) error {
message := format
@ -184,7 +195,9 @@ func (s *Sync) FullCycle() (e error) {
}
s.db = redisdb.New()
s.videosMapMux.Lock()
s.syncedVideos = syncedVideos
s.videosMapMux.Unlock()
s.grp = stop.New()
s.queue = make(chan video)
@ -343,6 +356,7 @@ func (s *Sync) startWorker(workerNum int) {
if err != nil {
SendErrorToSlack("Failed to mark video on the database: %s", err.Error())
}
s.AppendSyncedVideo(v.ID(), false, err.Error())
}
break
}
@ -496,7 +510,9 @@ func (s *Sync) processVideo(v video) (err error) {
log.Println(v.ID() + " took " + time.Since(start).String())
}(time.Now())
s.videosMapMux.Lock()
sv, ok := s.syncedVideos[v.ID()]
s.videosMapMux.Unlock()
alreadyPublished := ok && sv.Published
neverRetryFailures := []string{
@ -535,12 +551,10 @@ func (s *Sync) processVideo(v video) (err error) {
return err
}
err = s.Manager.MarkVideoStatus(s.YoutubeChannelID, v.ID(), VideoStatusPublished, summary.ClaimID, summary.ClaimName, "")
if err != nil {
SendErrorToSlack("Failed to mark video on the database: %s", err.Error())
}
if err != nil {
return err
}
s.AppendSyncedVideo(v.ID(), true, "")
return nil
}