octoleo/restic
2
2
Fork 0
mirror of https://github.com/octoleo/restic.git synced 2024-12-23 11:28:54 +00:00
Code Issues Releases Wiki Activity

Merge pull request #999 from restic/backend-use-semaphore

Browse Source 
backends: Use new semaphore
This commit is contained in:
Alexander Neumann 2017-06-07 19:48:32 +02:00
commit 550e1feaec
15 changed files with 249 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/restic/backend/b2/b2.go
View File

@ -53,6 +53,11 @@ func Open(cfg Config) (restic.Backend, error) {
return nil, errors.Wrap(err, "Bucket")
}
sem, err := backend.NewSemaphore(cfg.Connections)
if err != nil {
return nil, err
}
be := &b2Backend{
client: client,
bucket: bucket,
@ -61,7 +66,7 @@ func Open(cfg Config) (restic.Backend, error) {
Join: path.Join,
Path: cfg.Prefix,
},
sem: backend.NewSemaphore(cfg.Connections),
sem: sem,
}
return be, nil
@ -88,6 +93,11 @@ func Create(cfg Config) (restic.Backend, error) {
return nil, errors.Wrap(err, "NewBucket")
}
sem, err := backend.NewSemaphore(cfg.Connections)
if err != nil {
return nil, err
}
be := &b2Backend{
client: client,
bucket: bucket,
@ -96,7 +106,7 @@ func Create(cfg Config) (restic.Backend, error) {
Join: path.Join,
Path: cfg.Prefix,
},
sem: backend.NewSemaphore(cfg.Connections),
sem: sem,
}
present, err := be.Test(context.TODO(), restic.Handle{Type: restic.ConfigFile})

2
src/restic/backend/b2/config.go
View File

@ -17,7 +17,7 @@ type Config struct {
Bucket string
Prefix string
Connections int `option:"connections" help:"set a limit for the number of concurrent connections (default: 5)"`
Connections uint `option:"connections" help:"set a limit for the number of concurrent connections (default: 5)"`
}
// NewConfig returns a new config with default options applied.

71
src/restic/backend/location/location_test.go
View File

@ -120,9 +120,10 @@ var parseTests = []struct {
"s3://eu-central-1/bucketname",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Connections: 20,
},
},
},
@ -130,9 +131,10 @@ var parseTests = []struct {
"s3://hostname.foo/bucketname",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "hostname.foo",
Bucket: "bucketname",
Prefix: "restic",
Endpoint: "hostname.foo",
Bucket: "bucketname",
Prefix: "restic",
Connections: 20,
},
},
},
@ -140,9 +142,10 @@ var parseTests = []struct {
"s3://hostname.foo/bucketname/prefix/directory",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "hostname.foo",
Bucket: "bucketname",
Prefix: "prefix/directory",
Endpoint: "hostname.foo",
Bucket: "bucketname",
Prefix: "prefix/directory",
Connections: 20,
},
},
},
@ -150,9 +153,10 @@ var parseTests = []struct {
"s3:eu-central-1/repo",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "eu-central-1",
Bucket: "repo",
Prefix: "restic",
Endpoint: "eu-central-1",
Bucket: "repo",
Prefix: "restic",
Connections: 20,
},
},
},
@ -160,9 +164,10 @@ var parseTests = []struct {
"s3:eu-central-1/repo/prefix/directory",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "eu-central-1",
Bucket: "repo",
Prefix: "prefix/directory",
Endpoint: "eu-central-1",
Bucket: "repo",
Prefix: "prefix/directory",
Connections: 20,
},
},
},
@ -170,9 +175,10 @@ var parseTests = []struct {
"s3:https://hostname.foo/repo",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "hostname.foo",
Bucket: "repo",
Prefix: "restic",
Endpoint: "hostname.foo",
Bucket: "repo",
Prefix: "restic",
Connections: 20,
},
},
},
@ -180,9 +186,10 @@ var parseTests = []struct {
"s3:https://hostname.foo/repo/prefix/directory",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "hostname.foo",
Bucket: "repo",
Prefix: "prefix/directory",
Endpoint: "hostname.foo",
Bucket: "repo",
Prefix: "prefix/directory",
Connections: 20,
},
},
},
@ -190,10 +197,11 @@ var parseTests = []struct {
"s3:http://hostname.foo/repo",
Location{Scheme: "s3",
Config: s3.Config{
Endpoint: "hostname.foo",
Bucket: "repo",
Prefix: "restic",
UseHTTP: true,
Endpoint: "hostname.foo",
Bucket: "repo",
Prefix: "restic",
UseHTTP: true,
Connections: 20,
},
},
},
@ -201,8 +209,9 @@ var parseTests = []struct {
"swift:container17:/",
Location{Scheme: "swift",
Config: swift.Config{
Container: "container17",
Prefix: "",
Container: "container17",
Prefix: "",
Connections: 20,
},
},
},
@ -210,8 +219,9 @@ var parseTests = []struct {
"swift:container17:/prefix97",
Location{Scheme: "swift",
Config: swift.Config{
Container: "container17",
Prefix: "prefix97",
Container: "container17",
Prefix: "prefix97",
Connections: 20,
},
},
},
@ -219,7 +229,8 @@ var parseTests = []struct {
"rest:http://hostname.foo:1234/",
Location{Scheme: "rest",
Config: rest.Config{
URL: parseURL("http://hostname.foo:1234/"),
URL: parseURL("http://hostname.foo:1234/"),
Connections: 20,
},
},
},

18
src/restic/backend/rest/config.go
View File

@ -5,11 +5,24 @@ import (
"strings"
"restic/errors"
"restic/options"
)
// Config contains all configuration necessary to connect to a REST server.
type Config struct {
URL *url.URL
URL *url.URL
Connections uint `option:"connections" help:"set a limit for the number of concurrent connections (default: 20)"`
}
func init() {
options.Register("rest", Config{})
}
// NewConfig returns a new Config with the default values filled in.
func NewConfig() Config {
return Config{
Connections: 20,
}
}
// ParseConfig parses the string s and extracts the REST server URL.
@ -25,6 +38,7 @@ func ParseConfig(s string) (interface{}, error) {
return nil, errors.Wrap(err, "url.Parse")
}
cfg := Config{URL: u}
cfg := NewConfig()
cfg.URL = u
return cfg, nil
}

3
src/restic/backend/rest/config_test.go
View File

@ -20,7 +20,8 @@ var configTests = []struct {
cfg Config
}{
{"rest:http://localhost:1234", Config{
URL: parseURL("http://localhost:1234"),
URL: parseURL("http://localhost:1234"),
Connections: 20,
}},
}

44
src/restic/backend/rest/rest.go
View File

@ -26,21 +26,21 @@ const connLimit = 40
var _ restic.Backend = &restBackend{}
type restBackend struct {
url *url.URL
connChan chan struct{}
client *http.Client
url *url.URL
sem *backend.Semaphore
client *http.Client
backend.Layout
}
// Open opens the REST backend with the given config.
func Open(cfg Config) (restic.Backend, error) {
connChan := make(chan struct{}, connLimit)
for i := 0; i < connLimit; i++ {
connChan <- struct{}{}
}
client := &http.Client{Transport: backend.Transport()}
sem, err := backend.NewSemaphore(cfg.Connections)
if err != nil {
return nil, err
}
// use url without trailing slash for layout
url := cfg.URL.String()
if url[len(url)-1] == '/' {
@ -48,10 +48,10 @@ func Open(cfg Config) (restic.Backend, error) {
}
be := &restBackend{
url: cfg.URL,
connChan: connChan,
client: client,
Layout: &backend.RESTLayout{URL: url, Join: path.Join},
url: cfg.URL,
client: client,
Layout: &backend.RESTLayout{URL: url, Join: path.Join},
sem: sem,
}
return be, nil
@ -114,9 +114,9 @@ func (b *restBackend) Save(ctx context.Context, h restic.Handle, rd io.Reader) (
// backend.Closer, which has a noop method.
rd = backend.Closer{Reader: rd}
<-b.connChan
b.sem.GetToken()
resp, err := ctxhttp.Post(ctx, b.client, b.Filename(h), "binary/octet-stream", rd)
b.connChan <- struct{}{}
b.sem.ReleaseToken()
if resp != nil {
defer func() {
@ -169,9 +169,9 @@ func (b *restBackend) Load(ctx context.Context, h restic.Handle, length int, off
req.Header.Add("Range", byteRange)
debug.Log("Load(%v) send range %v", h, byteRange)
<-b.connChan
b.sem.GetToken()
resp, err := ctxhttp.Do(ctx, b.client, req)
b.connChan <- struct{}{}
b.sem.ReleaseToken()
if err != nil {
if resp != nil {
@ -195,9 +195,9 @@ func (b *restBackend) Stat(ctx context.Context, h restic.Handle) (restic.FileInf
return restic.FileInfo{}, err
}
<-b.connChan
b.sem.GetToken()
resp, err := ctxhttp.Head(ctx, b.client, b.Filename(h))
b.connChan <- struct{}{}
b.sem.ReleaseToken()
if err != nil {
return restic.FileInfo{}, errors.Wrap(err, "client.Head")
}
@ -242,9 +242,9 @@ func (b *restBackend) Remove(ctx context.Context, h restic.Handle) error {
if err != nil {
return errors.Wrap(err, "http.NewRequest")
}
<-b.connChan
b.sem.GetToken()
resp, err := ctxhttp.Do(ctx, b.client, req)
b.connChan <- struct{}{}
b.sem.ReleaseToken()
if err != nil {
return errors.Wrap(err, "client.Do")
@ -273,9 +273,9 @@ func (b *restBackend) List(ctx context.Context, t restic.FileType) <-chan string
url += "/"
}
<-b.connChan
b.sem.GetToken()
resp, err := ctxhttp.Get(ctx, b.client, url)
b.connChan <- struct{}{}
b.sem.ReleaseToken()
if resp != nil {
defer func() {

5
src/restic/backend/rest/rest_test.go
View File

@ -76,9 +76,8 @@ func newTestSuite(ctx context.Context, t testing.TB) *test.Suite {
t.Fatal(err)
}
cfg := rest.Config{
URL: url,
}
cfg := rest.NewConfig()
cfg.URL = url
return cfg, nil
},

21
src/restic/backend/s3/config.go
View File

@ -18,6 +18,15 @@ type Config struct {
Bucket string
Prefix string
Layout string `option:"layout" help:"use this backend layout (default: auto-detect)"`
Connections uint `option:"connections" help:"set a limit for the number of concurrent connections (default: 20)"`
}
// NewConfig returns a new Config with the default values filled in.
func NewConfig() Config {
return Config{
Connections: 20,
}
}
func init() {
@ -70,10 +79,10 @@ func createConfig(endpoint string, p []string, useHTTP bool) (interface{}, error
default:
prefix = path.Clean(p[1])
}
return Config{
Endpoint: endpoint,
UseHTTP: useHTTP,
Bucket: p[0],
Prefix: prefix,
}, nil
cfg := NewConfig()
cfg.Endpoint = endpoint
cfg.UseHTTP = useHTTP
cfg.Bucket = p[0]
cfg.Prefix = prefix
return cfg, nil
}

106
src/restic/backend/s3/config_test.go
View File

@ -7,78 +7,92 @@ var configTests = []struct {
cfg Config
}{
{"s3://eu-central-1/bucketname", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Connections: 20,
}},
{"s3://eu-central-1/bucketname/", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "restic",
Connections: 20,
}},
{"s3://eu-central-1/bucketname/prefix/directory", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "prefix/directory",
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "prefix/directory",
Connections: 20,
}},
{"s3://eu-central-1/bucketname/prefix/directory/", Config{
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "prefix/directory",
Endpoint: "eu-central-1",
Bucket: "bucketname",
Prefix: "prefix/directory",
Connections: 20,
}},
{"s3:eu-central-1/foobar", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "restic",
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
}},
{"s3:eu-central-1/foobar/", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "restic",
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
}},
{"s3:eu-central-1/foobar/prefix/directory", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "prefix/directory",
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "prefix/directory",
Connections: 20,
}},
{"s3:eu-central-1/foobar/prefix/directory/", Config{
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "prefix/directory",
Endpoint: "eu-central-1",
Bucket: "foobar",
Prefix: "prefix/directory",
Connections: 20,
}},
{"s3:https://hostname:9999/foobar", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
}},
{"s3:https://hostname:9999/foobar/", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
Connections: 20,
}},
{"s3:http://hostname:9999/foobar", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
UseHTTP: true,
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
UseHTTP: true,
Connections: 20,
}},
{"s3:http://hostname:9999/foobar/", Config{
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
UseHTTP: true,
Endpoint: "hostname:9999",
Bucket: "foobar",
Prefix: "restic",
UseHTTP: true,
Connections: 20,
}},
{"s3:http://hostname:9999/bucket/prefix/directory", Config{
Endpoint: "hostname:9999",
Bucket: "bucket",
Prefix: "prefix/directory",
UseHTTP: true,
Endpoint: "hostname:9999",
Bucket: "bucket",
Prefix: "prefix/directory",
UseHTTP: true,
Connections: 20,
}},
{"s3:http://hostname:9999/bucket/prefix/directory/", Config{
Endpoint: "hostname:9999",
Bucket: "bucket",
Prefix: "prefix/directory",
UseHTTP: true,
Endpoint: "hostname:9999",
Bucket: "bucket",
Prefix: "prefix/directory",
UseHTTP: true,
Connections: 20,
}},
}

35
src/restic/backend/s3/s3.go
View File

@ -24,7 +24,7 @@ const connLimit = 10
// s3 is a backend which stores the data on an S3 endpoint.
type s3 struct {
client *minio.Client
connChan chan struct{}
sem *backend.Semaphore
bucketname string
prefix string
cacheMutex sync.RWMutex
@ -47,8 +47,14 @@ func Open(cfg Config) (restic.Backend, error) {
return nil, errors.Wrap(err, "minio.New")
}
sem, err := backend.NewSemaphore(cfg.Connections)
if err != nil {
return nil, err
}
be := &s3{
client: client,
sem: sem,
bucketname: cfg.Bucket,
prefix: cfg.Prefix,
cacheObjSize: make(map[string]int64),
@ -63,8 +69,6 @@ func Open(cfg Config) (restic.Backend, error) {
be.Layout = l
be.createConnections()
found, err := client.BucketExists(cfg.Bucket)
if err != nil {
debug.Log("BucketExists(%v) returned err %v", cfg.Bucket, err)
@ -82,13 +86,6 @@ func Open(cfg Config) (restic.Backend, error) {
return be, nil
}
func (be *s3) createConnections() {
be.connChan = make(chan struct{}, connLimit)
for i := 0; i < connLimit; i++ {
be.connChan <- struct{}{}
}
}
// IsNotExist returns true if the error is caused by a not existing file.
func (be *s3) IsNotExist(err error) bool {
debug.Log("IsNotExist(%T, %#v)", err, err)
@ -226,7 +223,7 @@ func (be *s3) Save(ctx context.Context, h restic.Handle, rd io.Reader) (err erro
return errors.New("key already exists")
}
<-be.connChan
be.sem.GetToken()
// wrap the reader so that net/http client cannot close the reader, return
// the token instead.
@ -238,11 +235,10 @@ func (be *s3) Save(ctx context.Context, h restic.Handle, rd io.Reader) (err erro
}
debug.Log("PutObject(%v, %v)", be.bucketname, objName)
coreClient := minio.Core{be.client}
coreClient := minio.Core{Client: be.client}
info, err := coreClient.PutObject(be.bucketname, objName, size, rd, nil, nil, nil)
// return token
be.connChan <- struct{}{}
be.sem.ReleaseToken()
debug.Log("%v -> %v bytes, err %#v", objName, info.Size, err)
return errors.Wrap(err, "client.PutObject")
@ -279,8 +275,7 @@ func (be *s3) Load(ctx context.Context, h restic.Handle, length int, offset int6
objName := be.Filename(h)
// get token for connection
<-be.connChan
be.sem.GetToken()
byteRange := fmt.Sprintf("bytes=%d-", offset)
if length > 0 {
@ -290,11 +285,10 @@ func (be *s3) Load(ctx context.Context, h restic.Handle, length int, offset int6
headers.Add("Range", byteRange)
debug.Log("Load(%v) send range %v", h, byteRange)
coreClient := minio.Core{be.client}
coreClient := minio.Core{Client: be.client}
rd, _, err := coreClient.GetObject(be.bucketname, objName, headers)
if err != nil {
// return token
be.connChan <- struct{}{}
be.sem.ReleaseToken()
return nil, err
}
@ -302,8 +296,7 @@ func (be *s3) Load(ctx context.Context, h restic.Handle, length int, offset int6
ReadCloser: rd,
f: func() {
debug.Log("Close()")
// return token
be.connChan <- struct{}{}
be.sem.ReleaseToken()
},
}

15
src/restic/backend/s3/s3_test.go
View File

@ -114,14 +114,13 @@ func newMinioTestSuite(ctx context.Context, t testing.TB) *test.Suite {
key, secret := newRandomCredentials(t)
cfg.stopServer = runMinio(ctx, t, cfg.tempdir, key, secret)
cfg.Config = s3.Config{
Endpoint: "localhost:9000",
Bucket: "restictestbucket",
Prefix: fmt.Sprintf("test-%d", time.Now().UnixNano()),
UseHTTP: true,
KeyID: key,
Secret: secret,
}
cfg.Config = s3.NewConfig()
cfg.Config.Endpoint = "localhost:9000"
cfg.Config.Bucket = "restictestbucket"
cfg.Config.Prefix = fmt.Sprintf("test-%d", time.Now().UnixNano())
cfg.Config.UseHTTP = true
cfg.Config.KeyID = key
cfg.Config.Secret = secret
return cfg, nil
},

9
src/restic/backend/semaphore.go
View File

@ -1,15 +1,20 @@
package backend
import "restic/errors"
// Semaphore limits access to a restricted resource.
type Semaphore struct {
ch chan struct{}
}
// NewSemaphore returns a new semaphore with capacity n.
func NewSemaphore(n int) *Semaphore {
func NewSemaphore(n uint) (*Semaphore, error) {
if n <= 0 {
return nil, errors.New("must be a positive number")
}
return &Semaphore{
ch: make(chan struct{}, n),
}
}, nil
}
// GetToken blocks until a Token is available.

21
src/restic/backend/swift/config.go
View File

@ -3,6 +3,7 @@ package swift
import (
"os"
"restic/errors"
"restic/options"
"strings"
)
@ -24,6 +25,19 @@ type Config struct {
Container string
Prefix string
DefaultContainerPolicy string
Connections uint `option:"connections" help:"set a limit for the number of concurrent connections (default: 20)"`
}
func init() {
options.Register("swift", Config{})
}
// NewConfig returns a new config with the default values filled in.
func NewConfig() Config {
return Config{
Connections: 20,
}
}
// ParseConfig parses the string s and extract swift's container name and prefix.
@ -47,10 +61,9 @@ func ParseConfig(s string) (interface{}, error) {
}
prefix = prefix[1:]
cfg := Config{
Container: container,
Prefix: prefix,
}
cfg := NewConfig()
cfg.Container = container
cfg.Prefix = prefix
return cfg, nil
}

25
src/restic/backend/swift/config_test.go
View File

@ -6,9 +6,28 @@ var configTests = []struct {
s string
cfg Config
}{
{"swift:cnt1:/", Config{Container: "cnt1", Prefix: ""}},
{"swift:cnt2:/prefix", Config{Container: "cnt2", Prefix: "prefix"}},
{"swift:cnt3:/prefix/longer", Config{Container: "cnt3", Prefix: "prefix/longer"}},
{
"swift:cnt1:/",
Config{
Container: "cnt1",
Prefix: "",
Connections: 20,
},
},
{
"swift:cnt2:/prefix",
Config{Container: "cnt2",
Prefix: "prefix",
Connections: 20,
},
},
{
"swift:cnt3:/prefix/longer",
Config{Container: "cnt3",
Prefix: "prefix/longer",
Connections: 20,
},
},
}
func TestParseConfig(t *testing.T) {

24
src/restic/backend/swift/swift.go
View File

@ -22,7 +22,7 @@ const connLimit = 10
// beSwift is a backend which stores the data on a swift endpoint.
type beSwift struct {
conn *swift.Connection
connChan chan struct{}
sem *backend.Semaphore
container string // Container name
prefix string // Prefix of object names in the container
backend.Layout
@ -36,6 +36,11 @@ var _ restic.Backend = &beSwift{}
func Open(cfg Config) (restic.Backend, error) {
debug.Log("config %#v", cfg)
sem, err := backend.NewSemaphore(cfg.Connections)
if err != nil {
return nil, err
}
be := &beSwift{
conn: &swift.Connection{
UserName: cfg.UserName,
@ -54,6 +59,7 @@ func Open(cfg Config) (restic.Backend, error) {
Transport: backend.Transport(),
},
sem: sem,
container: cfg.Container,
prefix: cfg.Prefix,
Layout: &backend.DefaultLayout{
@ -61,7 +67,6 @@ func Open(cfg Config) (restic.Backend, error) {
Join: path.Join,
},
}
be.createConnections()
// Authenticate if needed
if !be.conn.Authenticated() {
@ -98,13 +103,6 @@ func Open(cfg Config) (restic.Backend, error) {
return be, nil
}
func (be *beSwift) createConnections() {
be.connChan = make(chan struct{}, connLimit)
for i := 0; i < connLimit; i++ {
be.connChan <- struct{}{}
}
}
func (be *beSwift) createContainer(policy string) error {
var h swift.Headers
if policy != "" {
@ -140,9 +138,9 @@ func (be *beSwift) Load(ctx context.Context, h restic.Handle, length int, offset
objName := be.Filename(h)
<-be.connChan
be.sem.GetToken()
defer func() {
be.connChan <- struct{}{}
be.sem.ReleaseToken()
}()
headers := swift.Headers{}
@ -190,9 +188,9 @@ func (be *beSwift) Save(ctx context.Context, h restic.Handle, rd io.Reader) (err
return errors.Wrap(err, "conn.Object")
}
<-be.connChan
be.sem.GetToken()
defer func() {
be.connChan <- struct{}{}
be.sem.ReleaseToken()
}()
encoding := "binary/octet-stream"