LBRYCommunity/lbry.go
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases 46 Packages Wiki Activity

remove redisDB dependency #25

Merged
nikooo777 merged 12 commits from use_video_statuses into master 2018-08-20 13:56:26 +02:00
Conversation 1 Commits 12 Files changed 5 +40 -26
2 changed files with 40 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit b3f7f93f5f - Show all commits

38
ytsync/manager.go
View file

@ -90,19 +90,19 @@ func (s SyncManager) fetchChannels(status string) ([]apiYoutubeChannel, error) {
return response.Data, nil
}
type apiSyncUpdateResponse struct {
Success bool `json:"success"`
Error null.String `json:"error"`
Data null.String `json:"data"`
type apiChannelStatusResponse struct {
Success bool `json:"success"`
Error null.String `json:"error"`
Data []syncedVideo `json:"data"`
}
type syncedVideo struct {
VideoID string
Published bool
FailureReason string
VideoID string `json:"video_id"`
Published bool `json:"published"`
FailureReason string `json:"failure_reason"`
}
func (s SyncManager) setChannelSyncStatus(channelID string, status string) (map[string]syncedVideo, error) {
func (s SyncManager) setChannelStatus(channelID string, status string) (map[string]syncedVideo, error) {
endpoint := s.ApiURL + "/yt/channel_status"
res, _ := http.PostForm(endpoint, url.Values{
@ -113,7 +113,7 @@ func (s SyncManager) setChannelSyncStatus(channelID string, status string) (map[
})
defer res.Body.Close()
body, _ := ioutil.ReadAll(res.Body)
var response apiSyncUpdateResponse
var response apiChannelStatusResponse
err := json.Unmarshal(body, &response)
if err != nil {
return nil, err
@ -121,18 +121,12 @@ func (s SyncManager) setChannelSyncStatus(channelID string, status string) (map[
if !response.Error.IsNull() {
return nil, errors.Err(response.Error.String)
}
if !response.Data.IsNull() {
if response.Data.String == "ok" {
if response.Data != nil {
if len(response.Data) == 0 {
return nil, nil
}
var sv []syncedVideo
err := json.Unmarshal([]byte(response.Data.String), &sv)
if err != nil {
return nil, errors.Err("could not parse synced videos")
}
svs := make(map[string]syncedVideo)
for _, v := range sv {
for _, v := range response.Data {
svs[v.VideoID] = v
}
return svs, nil
@ -145,6 +139,12 @@ const (
VideoStatusFailed = "failed"
)
type apiVideoStatusResponse struct {
Success bool `json:"success"`
Error null.String `json:"error"`
Data null.String `json:"data"`
}
func (s SyncManager) MarkVideoStatus(channelID string, videoID string, status string, claimID string, claimName string, failureReason string) error {
endpoint := s.ApiURL + "/yt/video_status"
@ -168,7 +168,7 @@ func (s SyncManager) MarkVideoStatus(channelID string, videoID string, status st
res, _ := http.PostForm(endpoint, vals)
defer res.Body.Close()
body, _ := ioutil.ReadAll(res.Body)
var response apiSyncUpdateResponse
var response apiVideoStatusResponse
err := json.Unmarshal(body, &response)
if err != nil {
return err

28
ytsync/ytsync.go
View file

@ -21,6 +21,7 @@ import (
"github.com/lbryio/lbry.go/jsonrpc"
"github.com/lbryio/lbry.go/stop"
"github.com/lbryio/lbry.go/util"
"github.com/lbryio/lbry.go/ytsync/redisdb"
"github.com/lbryio/lbry.go/ytsync/sources"
"github.com/mitchellh/go-ps"
log "github.com/sirupsen/logrus"
@ -63,9 +64,9 @@ type Sync struct {
daemon *jsonrpc.Client
lyoshenka commented 2018-08-09 16:31:56 +02:00 (Migrated from github.com)
Review
Copy link

if you have multiple mutexes, you can't call one of them just mux. its not clear what that's for. it could be something like walletSetupMux. though once you name it that, it becomes clear that something might be wrong with wrapping the whole wallet setup in a single mutex. does the whole thing actually need to be locked?

the mutexes should be pointers, so new copies are not created if the Sync struct is copied

its clearer if you put the mutex right above the thing the mutex is locking, and leave newlines on either side. so in this example, upt videosMapMux right above syncedVideos. and then maybe rename it to syncedVideosMux, like so:

...other vars...

syncedVideosMux *sync.Mutex
syncedVideos    map[string]syncedVideo

...other vars...
if you have multiple mutexes, you can't call one of them just `mux`. its not clear what that's for. it could be something like `walletSetupMux`. though once you name it that, it becomes clear that something might be wrong with wrapping the whole wallet setup in a single mutex. does the whole thing *actually* need to be locked? the mutexes should be pointers, so new copies are not created if the Sync struct is copied its clearer if you put the mutex right above the thing the mutex is locking, and leave newlines on either side. so in this example, upt `videosMapMux` right above `syncedVideos`. and then maybe rename it to `syncedVideosMux`, like so: ``` ...other vars... syncedVideosMux *sync.Mutex syncedVideos map[string]syncedVideo ...other vars... ```
lyoshenka commented 2018-08-09 16:32:30 +02:00 (Migrated from github.com)
Review
Copy link

this should also be a pointer. but more importantly, you don't need this if you already have a stop.Group. stop.Group works as a combo WaitGroup + channel that will be closed to indicate stopping (and can be closed safely multiple times). so use grp.Add() and grp.Wait() instead

this should also be a pointer. but more importantly, you don't need this if you already have a stop.Group. stop.Group works as a combo WaitGroup + channel that will be closed to indicate stopping (and can be closed safely multiple times). so use `grp.Add()` and `grp.Wait()` instead
lyoshenka commented 2018-08-09 16:38:35 +02:00 (Migrated from github.com)
Review
Copy link

it won't let me comment below, so I'm commenting here:

you call Add() and Done() inside startWorker(workerNum int), but the correct pattern is to make those calls outside the function. startWorker() doesn't know if its being run asynchronously or not. you only need a waitgroup if it is. there are also subtle concurrency issues with calling it inside the function. so the right way to go is to remove Add and Done from inside startWorker, and do this:

	for i := 0; i < s.ConcurrentVideos; i++ {
		s.wg.Add(1)
		go func() {
			defer s.wg.Done()
			s.startWorker(i)
		}()
	}
it won't let me comment below, so I'm commenting here: you call `Add()` and `Done()` inside `startWorker(workerNum int)`, but the correct pattern is to make those calls outside the function. startWorker() doesn't know if its being run asynchronously or not. you only need a waitgroup if it is. there are also subtle concurrency issues with calling it inside the function. so the right way to go is to remove Add and Done from inside startWorker, and do this: ``` for i := 0; i < s.ConcurrentVideos; i++ { s.wg.Add(1) go func() { defer s.wg.Done() s.startWorker(i) }() } ```
lyoshenka commented 2018-08-09 16:45:11 +02:00 (Migrated from github.com)
Review
Copy link

you can use a read/write lock to lock the map for reading when you read it, and writing when you write to it. this lets multiple threads read the data at once, which is safe and blocks less. use *sync.RWMutex instead, and call mux.RLock() and mux.RUnlock() when you're only reading from the variable. leave mux.Lock() for writing

you can use a read/write lock to lock the map for reading when you read it, and writing when you write to it. this lets multiple threads read the data at once, which is safe and blocks less. use `*sync.RWMutex` instead, and call `mux.RLock()` and `mux.RUnlock()` when you're only reading from the variable. leave `mux.Lock()` for writing
nikooo777 commented 2018-08-09 18:30:21 +02:00 (Migrated from github.com)
Review
Copy link

I need to look into that, this is from your original code and I don't remember ever changing it. Thanks for the pointers there.

I need to look into that, this is from your original code and I don't remember ever changing it. Thanks for the pointers there.
nikooo777 commented 2018-08-09 18:32:46 +02:00 (Migrated from github.com)
Review
Copy link

Will rename and move the mutexes.
The reason I didn't use read/write locks is that i don't want the application to read when it's being written to, plus the locks are held for a very very short time so there would be no noticeable improvement, only a higher risk of race conditions happening during wallet refills.

Will rename and move the mutexes. The reason I didn't use read/write locks is that i don't want the application to read when it's being written to, plus the locks are held for a very very short time so there would be no noticeable improvement, only a higher risk of race conditions happening during wallet refills.
nikooo777 commented 2018-08-14 16:37:24 +02:00 (Migrated from github.com)
Review
Copy link

I looked more into why we need to lock the whole walletSetup function. The reason I'm doing that is to avoid multiple threads from refilling the wallet concurrently. I don't think I can easily break up the the function to lock fewer lines of code. I think it's fair to leave it like that.

I removed the wait group in favor of the stop group

I looked more into why we need to lock the whole walletSetup function. The reason I'm doing that is to avoid multiple threads from refilling the wallet concurrently. I don't think I can easily break up the the function to lock fewer lines of code. I think it's fair to leave it like that. I removed the wait group in favor of the stop group
lyoshenka commented 2018-08-18 15:28:20 +02:00 (Migrated from github.com)
Review
Copy link

a read/write lock does not allow reading during writing. thats the point of every lock. what it does allow is multiple concurrent reads. when you do RLock, others can RLock at the same time. when you Lock (for writing), no one's allowed to Lock or RLock at the same time.

a read/write lock does not allow reading during writing. thats the point of every lock. what it does allow is multiple concurrent reads. when you do RLock, others can RLock at the same time. when you Lock (for writing), no one's allowed to Lock or RLock at the same time.
nikooo777 commented 2018-08-20 13:03:09 +02:00 (Migrated from github.com)
Review
Copy link

Oh yes, you're right, not sure what was going in my mind. I'll swap that

Oh yes, you're right, not sure what was going in my mind. I'll swap that
claimAddress string
videoDirectory string
//db *redisdb.DB
syncedVideos map[string]syncedVideo
grp *stop.Group
db *redisdb.DB
syncedVideos map[string]syncedVideo
grp *stop.Group
mux sync.Mutex
wg sync.WaitGroup
@ -109,10 +110,11 @@ func (s *Sync) FullCycle() (e error) {
if s.YoutubeChannelID == "" {
return errors.Err("channel ID not provided")
}
syncedVideos, err := s.Manager.setChannelSyncStatus(s.YoutubeChannelID, StatusSyncing)
syncedVideos, err := s.Manager.setChannelStatus(s.YoutubeChannelID, StatusSyncing)
if err != nil {
return err
}
defer func() {
if e != nil {
//conditions for which a channel shouldn't be marked as failed
@ -123,14 +125,14 @@ func (s *Sync) FullCycle() (e error) {
if util.SubstringInSlice(e.Error(), noFailConditions) {
return
}
_, err := s.Manager.setChannelSyncStatus(s.YoutubeChannelID, StatusFailed)
_, err := s.Manager.setChannelStatus(s.YoutubeChannelID, StatusFailed)
if err != nil {
msg := fmt.Sprintf("Failed setting failed state for channel %s.", s.LbryChannelName)
err = errors.Prefix(msg, err)
e = errors.Prefix(err.Error(), e)
}
} else if !s.IsInterrupted() {
_, err := s.Manager.setChannelSyncStatus(s.YoutubeChannelID, StatusSynced)
_, err := s.Manager.setChannelStatus(s.YoutubeChannelID, StatusSynced)
if err != nil {
e = err
}
@ -181,7 +183,7 @@ func (s *Sync) FullCycle() (e error) {
return errors.Wrap(err, 0)
}
//s.db = redisdb.New()
s.db = redisdb.New()
s.syncedVideos = syncedVideos
s.grp = stop.New()
s.queue = make(chan video)
@ -502,6 +504,18 @@ func (s *Sync) processVideo(v video) (err error) {
return nil
}
alreadyPublishedOld, err := s.db.IsPublished(v.ID())
if err != nil {
return err
}
//TODO: remove this after a few runs
if alreadyPublishedOld && !alreadyPublished {
//seems like something in the migration of blobs didn't go perfectly right so warn about it!
SendInfoToSlack("A video that was previously published is on the local database but isn't on the remote db! fix it @Nikooo777! \nchannelID: %s, videoID: %s",
s.YoutubeChannelID, v.ID())
return nil
}
if alreadyPublished {
log.Println(v.ID() + " already published")
return nil