LBRYCommunity/reflector.go
1
0
Fork 0
Code Issues 7 Pull requests Projects Releases 10 Packages Wiki Activity

Merge branch 'public-lib'

Browse source
This commit is contained in:
Niko Storni 2019-08-28 15:07:52 +02:00
commit 949ea2f2d8
1 changed files with 22 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
reflector/uploader.go
View file

@ -24,6 +24,10 @@ const (
errInc
)
type Summary struct {
Total, AlreadyStored, Sd, Blob, Err int
}
type Uploader struct {
db *db.SQL
store *store.DBBackedS3Store // could just be store.BlobStore interface
@ -32,9 +36,7 @@ type Uploader struct {
stopper *stop.Group
countChan chan increment
count struct {
total, alreadyStored, sd, blob, err int
}
count Summary
}
func NewUploader(db *db.SQL, store *store.DBBackedS3Store, workers int, skipExistsCheck bool) *Uploader {
@ -59,7 +61,7 @@ func (u *Uploader) Upload(dirOrFilePath string) error {
return err
}
u.count.total = len(paths)
u.count.Total = len(paths)
hashes := make([]string, len(paths))
for i, p := range paths {
@ -74,10 +76,10 @@ func (u *Uploader) Upload(dirOrFilePath string) error {
if err != nil {
return err
}
u.count.alreadyStored = len(exists)
u.count.AlreadyStored = len(exists)
}
log.Infof("%d new blobs to upload", u.count.total-u.count.alreadyStored)
log.Infof("%d new blobs to upload", u.count.Total-u.count.AlreadyStored)
workerWG := sync.WaitGroup{}
pathChan := make(chan string)
@ -118,11 +120,11 @@ Upload:
u.stopper.Stop()
log.Infoln("SUMMARY")
log.Infof("%d blobs total", u.count.total)
log.Infof("%d blobs already stored", u.count.alreadyStored)
log.Infof("%d SD blobs uploaded", u.count.sd)
log.Infof("%d content blobs uploaded", u.count.blob)
log.Infof("%d errors encountered", u.count.err)
log.Infof("%d blobs total", u.count.Total)
log.Infof("%d blobs already stored", u.count.AlreadyStored)
log.Infof("%d SD blobs uploaded", u.count.Sd)
log.Infof("%d content blobs uploaded", u.count.Blob)
log.Infof("%d errors encountered", u.count.Err)
return nil
}
@ -198,19 +200,23 @@ func (u *Uploader) counter() {
}
switch incrementType {
case sdInc:
u.count.sd++
u.count.Sd++
case blobInc:
u.count.blob++
u.count.Blob++
case errInc:
u.count.err++
u.count.Err++
}
}
if (u.count.sd+u.count.blob)%50 == 0 {
log.Infof("%d of %d done (%s elapsed, %.3fs per blob)", u.count.sd+u.count.blob, u.count.total-u.count.alreadyStored, time.Since(start).String(), time.Since(start).Seconds()/float64(u.count.sd+u.count.blob))
if (u.count.Sd+u.count.Blob)%50 == 0 {
log.Infof("%d of %d done (%s elapsed, %.3fs per blob)", u.count.Sd+u.count.Blob, u.count.Total-u.count.AlreadyStored, time.Since(start).String(), time.Since(start).Seconds()/float64(u.count.Sd+u.count.Blob))
}
}
}
func (u *Uploader) GetSummary() Summary {
return u.count
}
func (u *Uploader) inc(t increment) {
select {
case u.countChan <- t: