Mirror/mochi
1
0
Fork 0
mirror of https://github.com/sot-tech/mochi.git synced 2026-07-25 09:08:09 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge commits 129aac230aa..828edb8fd8b from https://github.com/chihaya/chihaya

Browse Source
This commit is contained in:
Širhoe Biazhkovič
2022-04-12 15:58:14 +03:00
parent c858576f76
commit c7edbb52f2
51 changed files with 562 additions and 1327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
storage/memory/storage.go
+10 -9
View File
@@ -5,12 +5,13 @@ package memory
import (
"encoding/binary"
"fmt"
"math"
"reflect"
"runtime"
"sync"
"time"
"gopkg.in/yaml.v2"
"gopkg.in/yaml.v3"
"github.com/sot-tech/mochi/bittorrent"
"github.com/sot-tech/mochi/pkg/log"
@@ -37,7 +38,7 @@ func init() {
type driver struct{}
func (d driver) NewStorage(icfg interface{}) (storage.Storage, error) {
func (d driver) NewStorage(icfg any) (storage.Storage, error) {
// Marshal the config back into bytes.
bytes, err := yaml.Marshal(icfg)
if err != nil {
@@ -80,7 +81,7 @@ func (cfg Config) LogFields() log.Fields {
func (cfg Config) Validate() Config {
validcfg := cfg
if cfg.ShardCount <= 0 {
if cfg.ShardCount <= 0 || cfg.ShardCount > (math.MaxInt/2) {
validcfg.ShardCount = defaultShardCount
log.Warn("falling back to default configuration", log.Fields{
"name": Name + ".ShardCount",
@@ -488,7 +489,7 @@ func (ps *store) ScrapeSwarm(ih bittorrent.InfoHash, addressFamily bittorrent.Ad
return
}
func asKey(in interface{}) interface{} {
func asKey(in any) any {
if in == nil {
panic("unable to use nil map key")
}
@@ -499,12 +500,12 @@ func asKey(in interface{}) interface{} {
return fmt.Sprint(in)
}
func (ps *store) Put(ctx string, key, value interface{}) {
func (ps *store) Put(ctx string, key, value any) {
m, _ := ps.contexts.LoadOrStore(ctx, new(sync.Map))
m.(*sync.Map).Store(asKey(key), value)
}
func (ps *store) Contains(ctx string, key interface{}) bool {
func (ps *store) Contains(ctx string, key any) bool {
var exist bool
if m, found := ps.contexts.Load(ctx); found {
_, exist = m.(*sync.Map).Load(asKey(key))
@@ -522,15 +523,15 @@ func (ps *store) BulkPut(ctx string, pairs ...storage.Pair) {
}
}
func (ps *store) Load(ctx string, key interface{}) interface{} {
var v interface{}
func (ps *store) Load(ctx string, key any) any {
var v any
if m, found := ps.contexts.Load(ctx); found {
v, _ = m.(*sync.Map).Load(asKey(key))
}
return v
}
func (ps *store) Delete(ctx string, keys ...interface{}) {
func (ps *store) Delete(ctx string, keys ...any) {
if len(keys) > 0 {
if m, found := ps.contexts.Load(ctx); found {
m := m.(*sync.Map)
storage/misc.go
+1 -1
View File
@@ -8,7 +8,7 @@ import (
// Pair - some key-value pair, used for BulkPut
type Pair struct {
Left, Right interface{}
Left, Right any
}
// SerializedPeer concatenation of PeerID, net port and IP-address
storage/redis/redis.go
+22 -25
View File
@@ -1,24 +1,26 @@
package redis
import (
"context"
"errors"
"github.com/go-redsync/redsync/v4"
"github.com/go-redsync/redsync/v4/redis/redigo"
redigolib "github.com/gomodule/redigo/redis"
"net/url"
"strconv"
"strings"
"time"
"github.com/go-redsync/redsync"
"github.com/gomodule/redigo/redis"
)
// redisBackend represents a redis handler.
type redisBackend struct {
pool *redis.Pool
pool *redigolib.Pool
redsync *redsync.Redsync
}
// newRedisBackend creates a redisBackend instance.
func newRedisBackend(cfg *Config, u *redisURL, socketPath string) *redisBackend {
context.Background()
rc := &redisConnector{
URL: u,
SocketPath: socketPath,
@@ -27,18 +29,13 @@ func newRedisBackend(cfg *Config, u *redisURL, socketPath string) *redisBackend
ConnectTimeout: cfg.RedisConnectTimeout,
}
pool := rc.NewPool()
rs := redsync.New([]redsync.Pool{pool})
rs := redsync.New(redigo.NewPool(pool))
return &redisBackend{
pool: pool,
redsync: rs,
}
}
// open returns or creates instance of Redis connection.
func (rb *redisBackend) open() redis.Conn {
return rb.pool.Get()
}
type redisConnector struct {
URL *redisURL
SocketPath string
@@ -48,11 +45,11 @@ type redisConnector struct {
}
// NewPool returns a new pool of Redis connections
func (rc *redisConnector) NewPool() *redis.Pool {
return &redis.Pool{
func (rc *redisConnector) NewPool() *redigolib.Pool {
return &redigolib.Pool{
MaxIdle: 3,
IdleTimeout: 240 * time.Second,
Dial: func() (redis.Conn, error) {
Dial: func() (redigolib.Conn, error) {
c, err := rc.open()
if err != nil {
return nil, err
@@ -68,8 +65,8 @@ func (rc *redisConnector) NewPool() *redis.Pool {
return c, err
},
// PINGs connections that have been idle more than 10 seconds
TestOnBorrow: func(c redis.Conn, t time.Time) error {
if time.Since(t) < time.Duration(10*time.Second) {
TestOnBorrow: func(c redigolib.Conn, t time.Time) error {
if time.Since(t) < 10*time.Second {
return nil
}
_, err := c.Do("PING")
@@ -79,23 +76,23 @@ func (rc *redisConnector) NewPool() *redis.Pool {
}
// Open a new Redis connection
func (rc *redisConnector) open() (redis.Conn, error) {
var opts = []redis.DialOption{
redis.DialDatabase(rc.URL.DB),
redis.DialReadTimeout(rc.ReadTimeout),
redis.DialWriteTimeout(rc.WriteTimeout),
redis.DialConnectTimeout(rc.ConnectTimeout),
func (rc *redisConnector) open() (redigolib.Conn, error) {
opts := []redigolib.DialOption{
redigolib.DialDatabase(rc.URL.DB),
redigolib.DialReadTimeout(rc.ReadTimeout),
redigolib.DialWriteTimeout(rc.WriteTimeout),
redigolib.DialConnectTimeout(rc.ConnectTimeout),
}
if rc.URL.Password != "" {
opts = append(opts, redis.DialPassword(rc.URL.Password))
opts = append(opts, redigolib.DialPassword(rc.URL.Password))
}
if rc.SocketPath != "" {
return redis.Dial("unix", rc.SocketPath, opts...)
return redigolib.Dial("unix", rc.SocketPath, opts...)
}
return redis.Dial("tcp", rc.URL.Host, opts...)
return redigolib.Dial("tcp", rc.URL.Host, opts...)
}
// A redisURL represents a parsed redisURL
@@ -119,7 +116,7 @@ func parseRedisURL(target string) (*redisURL, error) {
return nil, errors.New("no redis scheme found")
}
db := 0 //default redis db
db := 0 // default redis db
parts := strings.Split(u.Path, "/")
if len(parts) != 1 {
db, err = strconv.Atoi(parts[1])
storage/redis/storage.go
+18 -17
View File
@@ -24,13 +24,14 @@
package redis
import (
"errors"
"strconv"
"strings"
"sync"
"time"
"github.com/gomodule/redigo/redis"
"gopkg.in/yaml.v2"
"gopkg.in/yaml.v3"
"github.com/sot-tech/mochi/bittorrent"
"github.com/sot-tech/mochi/pkg/log"
@@ -60,7 +61,7 @@ func init() {
type driver struct{}
func (d driver) NewStorage(icfg interface{}) (storage.Storage, error) {
func (d driver) NewStorage(icfg any) (storage.Storage, error) {
// Marshal the config back into bytes.
bytes, err := yaml.Marshal(icfg)
if err != nil {
@@ -271,7 +272,7 @@ func (ps *store) getConnection() redis.Conn {
panic("attempted to interact with stopped redis store")
default:
}
return ps.rb.open()
return ps.rb.pool.Get()
}
func closeConnection(con redis.Conn) {
@@ -291,7 +292,7 @@ func (ps *store) populateProm() {
defer closeConnection(conn)
for _, group := range ps.groups() {
if n, err := redis.Int64(conn.Do("GET", ps.infohashCountKey(group))); err != nil && err != redis.ErrNil {
if n, err := redis.Int64(conn.Do("GET", ps.infohashCountKey(group))); err != nil && !errors.Is(err, redis.ErrNil) {
log.Error("storage: GET counter failure", log.Fields{
"key": ps.infohashCountKey(group),
"error": err,
@@ -299,7 +300,7 @@ func (ps *store) populateProm() {
} else {
numInfohashes += n
}
if n, err := redis.Int64(conn.Do("GET", ps.seederCountKey(group))); err != nil && err != redis.ErrNil {
if n, err := redis.Int64(conn.Do("GET", ps.seederCountKey(group))); err != nil && !errors.Is(err, redis.ErrNil) {
log.Error("storage: GET counter failure", log.Fields{
"key": ps.seederCountKey(group),
"error": err,
@@ -307,7 +308,7 @@ func (ps *store) populateProm() {
} else {
numSeeders += n
}
if n, err := redis.Int64(conn.Do("GET", ps.leecherCountKey(group))); err != nil && err != redis.ErrNil {
if n, err := redis.Int64(conn.Do("GET", ps.leecherCountKey(group))); err != nil && !errors.Is(err, redis.ErrNil) {
log.Error("storage: GET counter failure", log.Fields{
"key": ps.leecherCountKey(group),
"error": err,
@@ -448,7 +449,7 @@ func (ps *store) DeleteLeecher(ih bittorrent.InfoHash, p bittorrent.Peer) error
}
_, err = conn.Do("DECR", ps.leecherCountKey(addressFamily))
return nil
return err
}
func (ps *store) GraduateLeecher(ih bittorrent.InfoHash, p bittorrent.Peer) error {
@@ -517,13 +518,13 @@ func (ps *store) AnnouncePeers(ih bittorrent.InfoHash, seeder bool, numWant int,
if err != nil {
return nil, err
}
conLeechers := leechers.([]interface{})
conLeechers := leechers.([]any)
seeders, err := conn.Do("HKEYS", encodedSeederInfoHash)
if err != nil {
return nil, err
}
conSeeders := seeders.([]interface{})
conSeeders := seeders.([]any)
if len(conLeechers) == 0 && len(conSeeders) == 0 {
return nil, storage.ErrResourceDoesNotExist
@@ -606,7 +607,7 @@ func (ps *store) ScrapeSwarm(ih bittorrent.InfoHash, af bittorrent.AddressFamily
return
}
func (ps *store) Put(ctx string, key, value interface{}) {
func (ps *store) Put(ctx string, key, value any) {
conn := ps.getConnection()
defer closeConnection(conn)
_, err := conn.Do("HSET", ctx, key, value)
@@ -615,7 +616,7 @@ func (ps *store) Put(ctx string, key, value interface{}) {
}
}
func (ps *store) Contains(ctx string, key interface{}) bool {
func (ps *store) Contains(ctx string, key any) bool {
conn := ps.getConnection()
defer closeConnection(conn)
exist, err := redis.Bool(conn.Do("HEXISTS", ctx, key))
@@ -636,7 +637,7 @@ func (ps *store) BulkPut(ctx string, pairs ...storage.Pair) {
default:
conn := ps.getConnection()
defer closeConnection(conn)
args := make([]interface{}, 1, l*2+1)
args := make([]any, 1, l*2+1)
args[0] = ctx
for _, p := range pairs {
args = append(args, p.Left, p.Right)
@@ -657,7 +658,7 @@ func (ps *store) BulkPut(ctx string, pairs ...storage.Pair) {
}
}
func (ps *store) Load(ctx string, key interface{}) interface{} {
func (ps *store) Load(ctx string, key any) any {
conn := ps.getConnection()
defer closeConnection(conn)
v, err := conn.Do("HGET", ctx, key)
@@ -667,7 +668,7 @@ func (ps *store) Load(ctx string, key interface{}) interface{} {
return v
}
func (ps *store) Delete(ctx string, keys ...interface{}) {
func (ps *store) Delete(ctx string, keys ...any) {
switch l := len(keys); l {
case 0:
break
@@ -681,7 +682,7 @@ func (ps *store) Delete(ctx string, keys ...interface{}) {
default:
conn := ps.getConnection()
defer closeConnection(conn)
args := make([]interface{}, 1, l+1)
args := make([]any, 1, l+1)
args[0] = ctx
args = append(args, keys...)
_, err := conn.Do("HDEL", args...)
@@ -823,7 +824,7 @@ func (ps *store) collectGarbage(cutoff time.Time) error {
_ = conn.Send("DECR", ps.infohashCountKey(group))
}
_, err = redis.Values(conn.Do("EXEC"))
if err != nil && err != redis.ErrNil {
if err != nil && !errors.Is(err, redis.ErrNil) {
log.Error("storage: Redis EXEC failure", log.Fields{
"group": group,
"infohash": ihStr,
@@ -831,7 +832,7 @@ func (ps *store) collectGarbage(cutoff time.Time) error {
})
}
} else {
if _, err = conn.Do("UNWATCH"); err != nil && err != redis.ErrNil {
if _, err = conn.Do("UNWATCH"); err != nil && !errors.Is(err, redis.ErrNil) {
log.Error("storage: Redis UNWATCH failure", log.Fields{"error": err})
}
}
storage/storage.go
+8 -9
View File
@@ -16,7 +16,7 @@ var (
// Driver is the interface used to initialize a new type of Storage.
type Driver interface {
NewStorage(cfg interface{}) (Storage, error)
NewStorage(cfg any) (Storage, error)
}
// ErrResourceDoesNotExist is the error returned by all delete methods and the
@@ -107,27 +107,26 @@ type Storage interface {
// Put used to place arbitrary k-v data with specified context
// into storage. ctx parameter used to group data
// (i.e. data only for specific middleware module)
Put(ctx string, key, value interface{})
Put(ctx string, key, value any)
// BulkPut used to place array of k-v data in specified context.
// Useful when several data entries should be added in single transaction/connection
BulkPut(ctx string, pairs ...Pair)
// Contains checks if any data in specified context exist
Contains(ctx string, key interface{}) bool
Contains(ctx string, key any) bool
// Load used to get arbitrary data in specified context by its key
Load(ctx string, key interface{}) interface{}
Load(ctx string, key any) any
// Delete used to delete arbitrary data in specified context by its keys
Delete(ctx string, keys ...interface{})
Delete(ctx string, keys ...any)
// stop.Stopper is an interface that expects a Stop method to stop the
// Storage.
// Stopper is an interface that expects a Stop method to stop the Storage.
// For more details see the documentation in the stop package.
stop.Stopper
// log.Fielder returns a loggable version of the data used to configure and
// Fielder returns a loggable version of the data used to configure and
// operate a particular Storage.
log.Fielder
}
@@ -158,7 +157,7 @@ func RegisterDriver(name string, d Driver) {
// the list of registered Drivers.
//
// If a driver does not exist, returns ErrDriverDoesNotExist.
func NewStorage(name string, cfg interface{}) (ps Storage, err error) {
func NewStorage(name string, cfg any) (ps Storage, err error) {
driversM.RLock()
defer driversM.RUnlock()
storage/test/storage_bench.go
+8 -4
View File
@@ -2,6 +2,7 @@ package test
import (
"github.com/sot-tech/mochi/bittorrent"
"github.com/sot-tech/mochi/pkg/rand_seed"
"github.com/sot-tech/mochi/storage"
"math/rand"
"net"
@@ -26,7 +27,7 @@ func generateInfohashes() (a [1000]bittorrent.InfoHash) {
}
func generatePeers() (a [1000]bittorrent.Peer) {
r := rand.New(rand.NewSource(0))
r := rand.New(rand.NewSource(rand_seed.GenSeed()))
for i := range a {
ip := make([]byte, 4)
n, err := r.Read(ip)
@@ -49,9 +50,11 @@ func generatePeers() (a [1000]bittorrent.Peer) {
return
}
type benchExecFunc func(int, storage.Storage, *benchData) error
type benchSetupFunc func(storage.Storage, *benchData) error
type benchStorageConstructor func() storage.Storage
type (
benchExecFunc func(int, storage.Storage, *benchData) error
benchSetupFunc func(storage.Storage, *benchData) error
benchStorageConstructor func() storage.Storage
)
type benchHolder struct {
st benchStorageConstructor
@@ -187,6 +190,7 @@ func (bh *benchHolder) PutDelete1kInfohash(b *testing.B) {
bh.runBenchmark(b, false, nil, func(i int, ps storage.Storage, bd *benchData) error {
err := ps.PutSeeder(bd.infohashes[i%1000], bd.peers[0])
if err != nil {
return err
}
return ps.DeleteSeeder(bd.infohashes[i%1000], bd.peers[0])
})
storage/test/storage_test_base.go
+1 -1
View File
@@ -198,7 +198,7 @@ func (th *testHolder) CustomPutContainsLoadDelete(t *testing.T) {
func (th *testHolder) CustomBulkPutContainsLoadDelete(t *testing.T) {
pairs := make([]storage.Pair, 0, len(testData))
keys := make([]interface{}, 0, len(testData))
keys := make([]any, 0, len(testData))
for _, c := range testData {
key := c.peer.String()
keys = append(keys, key)