This commit is contained in:
Niko Storni 2021-07-24 00:08:13 +02:00
parent 63a574ec2f
commit b4913ecedf
19 changed files with 24 additions and 65 deletions

View file

@ -7,7 +7,7 @@ import (
"github.com/lbryio/lbry.go/v2/schema/stake" "github.com/lbryio/lbry.go/v2/schema/stake"
"github.com/davecgh/go-spew/spew" "github.com/davecgh/go-spew/spew"
"github.com/golang/protobuf/jsonpb" "github.com/gogo/protobuf/jsonpb"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/spf13/cobra" "github.com/spf13/cobra"
) )

View file

@ -41,6 +41,9 @@ func populateDbCmd(cmd *cobra.Command, args []string) {
log.Fatal(err) log.Fatal(err)
} }
blobs, err := speedwalk.AllFiles(diskStorePath, true) blobs, err := speedwalk.AllFiles(diskStorePath, true)
if err != nil {
log.Fatal(err)
}
err = localDb.AddBlobs(blobs) err = localDb.AddBlobs(blobs)
if err != nil { if err != nil {
log.Errorf("error while storing to db: %s", errors.FullTrace(err)) log.Errorf("error while storing to db: %s", errors.FullTrace(err))

View file

@ -165,8 +165,8 @@ func mustGetFlagInt64(cmd *cobra.Command, name string) int64 {
return v return v
} }
func mustGetFlagBool(cmd *cobra.Command, name string) bool { //func mustGetFlagBool(cmd *cobra.Command, name string) bool {
v, err := cmd.Flags().GetBool(name) // v, err := cmd.Flags().GetBool(name)
checkErr(err) // checkErr(err)
return v // return v
} //}

View file

@ -266,9 +266,9 @@ func (s *SQL) HasBlobs(hashes []string, touch bool) (map[string]bool, error) {
if touch { if touch {
if s.TrackAccess == TrackAccessBlobs { if s.TrackAccess == TrackAccessBlobs {
s.touchBlobs(idsNeedingTouch) _ = s.touchBlobs(idsNeedingTouch)
} else if s.TrackAccess == TrackAccessStreams { } else if s.TrackAccess == TrackAccessStreams {
s.touchStreams(idsNeedingTouch) _ = s.touchStreams(idsNeedingTouch)
} }
} }

2
go.mod
View file

@ -14,6 +14,7 @@ require (
github.com/ekyoung/gin-nice-recovery v0.0.0-20160510022553-1654dca486db github.com/ekyoung/gin-nice-recovery v0.0.0-20160510022553-1654dca486db
github.com/gin-gonic/gin v1.7.1 github.com/gin-gonic/gin v1.7.1
github.com/go-sql-driver/mysql v1.6.0 github.com/go-sql-driver/mysql v1.6.0
github.com/gogo/protobuf v1.2.1
github.com/golang/protobuf v1.5.2 github.com/golang/protobuf v1.5.2
github.com/google/go-cmp v0.5.6 // indirect github.com/google/go-cmp v0.5.6 // indirect
github.com/google/gops v0.3.18 github.com/google/gops v0.3.18
@ -44,6 +45,7 @@ require (
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect
golang.org/x/text v0.3.6 // indirect golang.org/x/text v0.3.6 // indirect
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
google.golang.org/protobuf v1.27.1 // indirect
) )
go 1.16 go 1.16

4
go.sum
View file

@ -171,6 +171,7 @@ github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRx
github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s= github.com/gogo/googleapis v1.1.0/go.mod h1:gf4bu3Q80BeJ6H1S1vYPm8/ELATdvryBaNFGgqEef3s=
github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
github.com/gogo/protobuf v1.2.1 h1:/s5zKNz0uPFCZ5hddgPdo2TK2TVrUNMn0OOX8/aZMTE=
github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4= github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc= github.com/golang/groupcache v0.0.0-20160516000752-02826c3e7903/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@ -877,8 +878,9 @@ google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miE
google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
google.golang.org/protobuf v1.26.0 h1:bxAC2xTBsZGibn2RTntX0oH50xLsqy1OxA9tTL3p/lk=
google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
google.golang.org/protobuf v1.27.1 h1:SnqbnDw1V7RiZcXPx5MEeqPv2s79L9i7BJUlG/+RurQ=
google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

View file

@ -33,7 +33,6 @@ func InitWorkers(server *Server, workers int) {
} }
}(i) }(i)
} }
return
} }
func enqueue(b *blobRequest) { func enqueue(b *blobRequest) {

View file

@ -112,7 +112,6 @@ func (p *Store) Delete(hash string) error {
return errors.Err(shared.ErrNotImplemented) return errors.Err(shared.ErrNotImplemented)
} }
// Delete is not supported // Shutdown is not supported
func (p *Store) Shutdown() { func (p *Store) Shutdown() {
return
} }

View file

@ -33,7 +33,6 @@ func InitWorkers(server *Server, workers int) {
} }
}(i) }(i)
} }
return
} }
func enqueue(b *blobRequest) { func enqueue(b *blobRequest) {

View file

@ -77,7 +77,6 @@ func (p *Store) Delete(hash string) error {
return errors.Err(shared.ErrNotImplemented) return errors.Err(shared.ErrNotImplemented)
} }
// Delete is not supported // Shutdown is not supported
func (p *Store) Shutdown() { func (p *Store) Shutdown() {
return
} }

View file

@ -104,5 +104,4 @@ func (c *CachingStore) Delete(hash string) error {
func (c *CachingStore) Shutdown() { func (c *CachingStore) Shutdown() {
c.origin.Shutdown() c.origin.Shutdown()
c.cache.Shutdown() c.cache.Shutdown()
return
} }

View file

@ -107,5 +107,4 @@ func (c *CloudFrontROStore) Delete(_ string) error {
// Shutdown shuts down the store gracefully // Shutdown shuts down the store gracefully
func (c *CloudFrontROStore) Shutdown() { func (c *CloudFrontROStore) Shutdown() {
return
} }

View file

@ -59,5 +59,4 @@ func (c *CloudFrontRWStore) Delete(hash string) error {
func (c *CloudFrontRWStore) Shutdown() { func (c *CloudFrontRWStore) Shutdown() {
c.s3.Shutdown() c.s3.Shutdown()
c.cf.Shutdown() c.cf.Shutdown()
return
} }

View file

@ -198,5 +198,4 @@ func (d *DBBackedStore) initBlocked() error {
// Shutdown shuts down the store gracefully // Shutdown shuts down the store gracefully
func (d *DBBackedStore) Shutdown() { func (d *DBBackedStore) Shutdown() {
d.blobs.Shutdown() d.blobs.Shutdown()
return
} }

View file

@ -9,7 +9,6 @@ import (
"io/ioutil" "io/ioutil"
"os" "os"
"path" "path"
"runtime"
"time" "time"
"github.com/lbryio/reflector.go/shared" "github.com/lbryio/reflector.go/shared"
@ -23,25 +22,6 @@ import (
"go.uber.org/atomic" "go.uber.org/atomic"
) )
func init() {
writeCh = make(chan writeRequest)
for i := 0; i < runtime.NumCPU(); i++ {
go func() {
for {
select {
case r := <-writeCh:
err := ioutil.WriteFile(r.filename, r.data, r.perm)
if err != nil {
log.Errorf("could not write file %s to disk, failed with error: %s", r.filename, err.Error())
}
}
}
}()
}
}
var writeCh chan writeRequest
// DiskStore stores blobs on a local disk // DiskStore stores blobs on a local disk
type DiskStore struct { type DiskStore struct {
// the location of blobs on disk // the location of blobs on disk
@ -228,21 +208,6 @@ func (d *DiskStore) initOnce() error {
return nil return nil
} }
type writeRequest struct {
filename string
data []byte
perm os.FileMode
}
// Shutdown shuts down the store gracefully // Shutdown shuts down the store gracefully
func (d *DiskStore) Shutdown() { func (d *DiskStore) Shutdown() {
return
}
func writeFile(filename string, data []byte, perm os.FileMode) {
writeCh <- writeRequest{
filename: filename,
data: data,
perm: perm,
}
} }

View file

@ -96,7 +96,7 @@ func (l *GcacheStore) Get(hash string) (stream.Blob, shared.BlobTrace, error) {
// Put stores the blob. Following LFUDA rules it's not guaranteed that a SET will store the value!!! // Put stores the blob. Following LFUDA rules it's not guaranteed that a SET will store the value!!!
func (l *GcacheStore) Put(hash string, blob stream.Blob) error { func (l *GcacheStore) Put(hash string, blob stream.Blob) error {
l.cache.Set(hash, true) _ = l.cache.Set(hash, true)
has, _ := l.Has(hash) has, _ := l.Has(hash)
if has { if has {
err := l.store.Put(hash, blob) err := l.store.Put(hash, blob)
@ -109,7 +109,7 @@ func (l *GcacheStore) Put(hash string, blob stream.Blob) error {
// PutSD stores the sd blob. Following LFUDA rules it's not guaranteed that a SET will store the value!!! // PutSD stores the sd blob. Following LFUDA rules it's not guaranteed that a SET will store the value!!!
func (l *GcacheStore) PutSD(hash string, blob stream.Blob) error { func (l *GcacheStore) PutSD(hash string, blob stream.Blob) error {
l.cache.Set(hash, true) _ = l.cache.Set(hash, true)
has, _ := l.Has(hash) has, _ := l.Has(hash)
if has { if has {
err := l.store.PutSD(hash, blob) err := l.store.PutSD(hash, blob)
@ -145,7 +145,7 @@ func (l *GcacheStore) loadExisting(store lister, maxItems int) error {
added := 0 added := 0
for _, h := range existing { for _, h := range existing {
l.cache.Set(h, true) _ = l.cache.Set(h, true)
added++ added++
if maxItems > 0 && added >= maxItems { // underlying cache is bigger than the cache if maxItems > 0 && added >= maxItems { // underlying cache is bigger than the cache
break break
@ -156,5 +156,4 @@ func (l *GcacheStore) loadExisting(store lister, maxItems int) error {
// Shutdown shuts down the store gracefully // Shutdown shuts down the store gracefully
func (l *GcacheStore) Shutdown() { func (l *GcacheStore) Shutdown() {
return
} }

View file

@ -115,7 +115,7 @@ func (n *HttpStore) PutSD(string, stream.Blob) error {
func (n *HttpStore) Delete(string) error { func (n *HttpStore) Delete(string) error {
return shared.ErrNotImplemented return shared.ErrNotImplemented
} }
func (n *HttpStore) Shutdown() { return } func (n *HttpStore) Shutdown() {}
// buffer pool to reduce GC // buffer pool to reduce GC
// https://www.captaincodeman.com/2017/06/02/golang-buffer-pool-gotcha // https://www.captaincodeman.com/2017/06/02/golang-buffer-pool-gotcha

View file

@ -70,6 +70,4 @@ func (c *ITTTStore) Delete(hash string) error {
} }
// Shutdown shuts down the store gracefully // Shutdown shuts down the store gracefully
func (c *ITTTStore) Shutdown() { func (c *ITTTStore) Shutdown() {}
return
}

View file

@ -77,6 +77,4 @@ func (m *MemStore) Debug() map[string]stream.Blob {
} }
// Shutdown shuts down the store gracefully // Shutdown shuts down the store gracefully
func (m *MemStore) Shutdown() { func (m *MemStore) Shutdown() {}
return
}