// +build !plan9 package cache import (	"context"	"fmt"	"io"	"math"	"os"	"os/signal"	"path"	"path/filepath"	"sort"	"strconv"	"strings"	"sync"	"syscall"	"time" 	"github.com/pkg/errors"	"github.com/rclone/rclone/backend/crypt"	"github.com/rclone/rclone/fs"	"github.com/rclone/rclone/fs/cache"	"github.com/rclone/rclone/fs/config"	"github.com/rclone/rclone/fs/config/configmap"	"github.com/rclone/rclone/fs/config/configstruct"	"github.com/rclone/rclone/fs/config/obscure"	"github.com/rclone/rclone/fs/fspath"	"github.com/rclone/rclone/fs/hash"	"github.com/rclone/rclone/fs/rc"	"github.com/rclone/rclone/fs/walk"	"github.com/rclone/rclone/lib/atexit"	"golang.org/x/time/rate") const (	// DefCacheChunkSize is the default value for chunk size	DefCacheChunkSize = fs.SizeSuffix(5 * 1024 * 1024)	// DefCacheTotalChunkSize is the default value for the maximum size of stored chunks	DefCacheTotalChunkSize = fs.SizeSuffix(10 * 1024 * 1024 * 1024)	// DefCacheChunkCleanInterval is the interval at which chunks are cleaned	DefCacheChunkCleanInterval = fs.Duration(time.Minute)	// DefCacheInfoAge is the default value for object info age	DefCacheInfoAge = fs.Duration(6 * time.Hour)	// DefCacheReadRetries is the default value for read retries	DefCacheReadRetries = 10	// DefCacheTotalWorkers is how many workers run in parallel to download chunks	DefCacheTotalWorkers = 4	// DefCacheChunkNoMemory will enable or disable in-memory storage for chunks	DefCacheChunkNoMemory = false	// DefCacheRps limits the number of requests per second to the source FS	DefCacheRps = -1	// DefCacheWrites will cache file data on writes through the cache	DefCacheWrites = false	// DefCacheTmpWaitTime says how long should files be stored in local cache before being uploaded	DefCacheTmpWaitTime = fs.Duration(15 * time.Second)	// DefCacheDbWaitTime defines how long the cache backend should wait for the DB to be available	DefCacheDbWaitTime = fs.Duration(1 * time.Second)) // Register with Fsfunc init() {	fs.Register(&fs.RegInfo{		Name:        "cache",		Description: "Cache a remote",		NewFs:       NewFs,		Options: []fs.Option{{			Name:     "remote",			Help:     "Remote to cache.\nNormally should contain a ':' and a path, eg \"myremote:path/to/dir\",\n\"myremote:bucket\" or maybe \"myremote:\" (not recommended).",			Required: true,		}, {			Name: "plex_url",			Help: "The URL of the Plex server",		}, {			Name: "plex_username",			Help: "The username of the Plex user",		}, {			Name:       "plex_password",			Help:       "The password of the Plex user",			IsPassword: true,		}, {			Name:     "plex_token",			Help:     "The plex token for authentication - auto set normally",			Hide:     fs.OptionHideBoth,			Advanced: true,		}, {			Name:     "plex_insecure",			Help:     "Skip all certificate verifications when connecting to the Plex server",			Advanced: true,		}, {			Name: "chunk_size",			Help: `The size of a chunk (partial file data). Use lower numbers for slower connections. If the chunk size ischanged, any downloaded chunks will be invalid and cache-chunk-pathwill need to be cleared or unexpected EOF errors will occur.`,			Default: DefCacheChunkSize,			Examples: []fs.OptionExample{{				Value: "1m",				Help:  "1MB",			}, {				Value: "5M",				Help:  "5 MB",			}, {				Value: "10M",				Help:  "10 MB",			}},		}, {			Name: "info_age",			Help: `How long to cache file structure information (directory listings, file size, times etc). If all write operations are done through the cache then you can safely makethis value very large as the cache store will also be updated in real time.`,			Default: DefCacheInfoAge,			Examples: []fs.OptionExample{{				Value: "1h",				Help:  "1 hour",			}, {				Value: "24h",				Help:  "24 hours",			}, {				Value: "48h",				Help:  "48 hours",			}},		}, {			Name: "chunk_total_size",			Help: `The total size that the chunks can take up on the local disk. If the cache exceeds this value then it will start to delete theoldest chunks until it goes under this value.`,			Default: DefCacheTotalChunkSize,			Examples: []fs.OptionExample{{				Value: "500M",				Help:  "500 MB",			}, {				Value: "1G",				Help:  "1 GB",			}, {				Value: "10G",				Help:  "10 GB",			}},		}, {			Name:     "db_path",			Default:  filepath.Join(config.CacheDir, "cache-backend"),			Help:     "Directory to store file structure metadata DB.\nThe remote name is used as the DB file name.",			Advanced: true,		}, {			Name:    "chunk_path",			Default: filepath.Join(config.CacheDir, "cache-backend"),			Help: `Directory to cache chunk files. Path to where partial file data (chunks) are stored locally. The remotename is appended to the final path. This config follows the "--cache-db-path". If you specify a customlocation for "--cache-db-path" and don't specify one for "--cache-chunk-path"then "--cache-chunk-path" will use the same path as "--cache-db-path".`,			Advanced: true,		}, {			Name:     "db_purge",			Default:  false,			Help:     "Clear all the cached data for this remote on start.",			Hide:     fs.OptionHideConfigurator,			Advanced: true,		}, {			Name:    "chunk_clean_interval",			Default: DefCacheChunkCleanInterval,			Help: `How often should the cache perform cleanups of the chunk storage.The default value should be ok for most people. If you find that thecache goes over "cache-chunk-total-size" too often then try to lowerthis value to force it to perform cleanups more often.`,			Advanced: true,		}, {			Name:    "read_retries",			Default: DefCacheReadRetries,			Help: `How many times to retry a read from a cache storage. Since reading from a cache stream is independent from downloading filedata, readers can get to a point where there's no more data in thecache.  Most of the times this can indicate a connectivity issue ifcache isn't able to provide file data anymore. For really slow connections, increase this to a point where the stream isable to provide data but your experience will be very stuttering.`,			Advanced: true,		}, {			Name:    "workers",			Default: DefCacheTotalWorkers,			Help: `How many workers should run in parallel to download chunks. Higher values will mean more parallel processing (better CPU needed)and more concurrent requests on the cloud provider.  This impactsseveral aspects like the cloud provider API limits, more stress on thehardware that rclone runs on but it also means that streams will bemore fluid and data will be available much more faster to readers. **Note**: If the optional Plex integration is enabled then thissetting will adapt to the type of reading performed and the valuespecified here will be used as a maximum number of workers to use.`,			Advanced: true,		}, {			Name:    "chunk_no_memory",			Default: DefCacheChunkNoMemory,			Help: `Disable the in-memory cache for storing chunks during streaming. By default, cache will keep file data during streaming in RAM as wellto provide it to readers as fast as possible. This transient data is evicted as soon as it is read and the number ofchunks stored doesn't exceed the number of workers. However, dependingon other settings like "cache-chunk-size" and "cache-workers" this footprintcan increase if there are parallel streams too (multiple files being readat the same time). If the hardware permits it, use this feature to provide an overall betterperformance during streaming but it can also be disabled if RAM is notavailable on the local machine.`,			Advanced: true,		}, {			Name:    "rps",			Default: int(DefCacheRps),			Help: `Limits the number of requests per second to the source FS (-1 to disable) This setting places a hard limit on the number of requests per secondthat cache will be doing to the cloud provider remote and try torespect that value by setting waits between reads. If you find that you're getting banned or limited on the cloudprovider through cache and know that a smaller number of requests persecond will allow you to work with it then you can use this settingfor that. A good balance of all the other settings should make this settinguseless but it is available to set for more special cases. **NOTE**: This will limit the number of requests during streams butother API calls to the cloud provider like directory listings willstill pass.`,			Advanced: true,		}, {			Name:    "writes",			Default: DefCacheWrites,			Help: `Cache file data on writes through the FS If you need to read files immediately after you upload them throughcache you can enable this flag to have their data stored in thecache store at the same time during upload.`,			Advanced: true,		}, {			Name:    "tmp_upload_path",			Default: "",			Help: `Directory to keep temporary files until they are uploaded. This is the path where cache will use as a temporary storage for newfiles that need to be uploaded to the cloud provider. Specifying a value will enable this feature. Without it, it iscompletely disabled and files will be uploaded directly to the cloudprovider`,			Advanced: true,		}, {			Name:    "tmp_wait_time",			Default: DefCacheTmpWaitTime,			Help: `How long should files be stored in local cache before being uploaded This is the duration that a file must wait in the temporary location_cache-tmp-upload-path_ before it is selected for upload. Note that only one file is uploaded at a time and it can take longerto start the upload if a queue formed for this purpose.`,			Advanced: true,		}, {			Name:    "db_wait_time",			Default: DefCacheDbWaitTime,			Help: `How long to wait for the DB to be available - 0 is unlimited Only one process can have the DB open at any one time, so rclone waitsfor this duration for the DB to become available before it gives anerror. If you set it to 0 then it will wait forever.`,			Advanced: true,		}},	})} // Options defines the configuration for this backendtype Options struct {	Remote             string        `config:"remote"`	PlexURL            string        `config:"plex_url"`	PlexUsername       string        `config:"plex_username"`	PlexPassword       string        `config:"plex_password"`	PlexToken          string        `config:"plex_token"`	PlexInsecure       bool          `config:"plex_insecure"`	ChunkSize          fs.SizeSuffix `config:"chunk_size"`	InfoAge            fs.Duration   `config:"info_age"`	ChunkTotalSize     fs.SizeSuffix `config:"chunk_total_size"`	DbPath             string        `config:"db_path"`	ChunkPath          string        `config:"chunk_path"`	DbPurge            bool          `config:"db_purge"`	ChunkCleanInterval fs.Duration   `config:"chunk_clean_interval"`	ReadRetries        int           `config:"read_retries"`	TotalWorkers       int           `config:"workers"`	ChunkNoMemory      bool          `config:"chunk_no_memory"`	Rps                int           `config:"rps"`	StoreWrites        bool          `config:"writes"`	TempWritePath      string        `config:"tmp_upload_path"`	TempWaitTime       fs.Duration   `config:"tmp_wait_time"`	DbWaitTime         fs.Duration   `config:"db_wait_time"`} // Fs represents a wrapped fs.Fstype Fs struct {	fs.Fs	wrapper fs.Fs 	name     string	root     string	opt      Options      // parsed options	features *fs.Features // optional features	cache    *Persistent	tempFs   fs.Fs 	lastChunkCleanup time.Time	cleanupMu        sync.Mutex	rateLimiter      *rate.Limiter	plexConnector    *plexConnector	backgroundRunner *backgroundWriter	cleanupChan      chan bool	parentsForgetFn  []func(string, fs.EntryType)	notifiedRemotes  map[string]bool	notifiedMu       sync.Mutex	parentsForgetMu  sync.Mutex} // parseRootPath returns a cleaned root path and a nil error or "" and an error when the path is invalidfunc parseRootPath(path string) (string, error) {	return strings.Trim(path, "/"), nil} // NewFs constructs a Fs from the path, container:pathfunc NewFs(name, rootPath string, m configmap.Mapper) (fs.Fs, error) {	// Parse config into Options struct	opt := new(Options)	err := configstruct.Set(m, opt)	if err != nil {		return nil, err	}	if opt.ChunkTotalSize < opt.ChunkSize*fs.SizeSuffix(opt.TotalWorkers) {		return nil, errors.Errorf("don't set cache-chunk-total-size(%v) less than cache-chunk-size(%v) * cache-workers(%v)",			opt.ChunkTotalSize, opt.ChunkSize, opt.TotalWorkers)	} 	if strings.HasPrefix(opt.Remote, name+":") {		return nil, errors.New("can't point cache remote at itself - check the value of the remote setting")	} 	rpath, err := parseRootPath(rootPath)	if err != nil {		return nil, errors.Wrapf(err, "failed to clean root path %q", rootPath)	} 	wInfo, wName, wPath, wConfig, err := fs.ConfigFs(opt.Remote)	if err != nil {		return nil, errors.Wrapf(err, "failed to parse remote %q to wrap", opt.Remote)	} 	remotePath := fspath.JoinRootPath(wPath, rootPath)	wrappedFs, wrapErr := wInfo.NewFs(wName, remotePath, wConfig)	if wrapErr != nil && wrapErr != fs.ErrorIsFile {		return nil, errors.Wrapf(wrapErr, "failed to make remote %s:%s to wrap", wName, remotePath)	}	var fsErr error	fs.Debugf(name, "wrapped %v:%v at root %v", wrappedFs.Name(), wrappedFs.Root(), rpath)	if wrapErr == fs.ErrorIsFile {		fsErr = fs.ErrorIsFile		rpath = cleanPath(path.Dir(rpath))	}	// configure cache backend	if opt.DbPurge {		fs.Debugf(name, "Purging the DB")	}	f := &Fs{		Fs:               wrappedFs,		name:             name,		root:             rpath,		opt:              *opt,		lastChunkCleanup: time.Now().Truncate(time.Hour * 24 * 30),		cleanupChan:      make(chan bool, 1),		notifiedRemotes:  make(map[string]bool),	}	f.rateLimiter = rate.NewLimiter(rate.Limit(float64(opt.Rps)), opt.TotalWorkers) 	f.plexConnector = &plexConnector{}	if opt.PlexURL != "" {		if opt.PlexToken != "" {			f.plexConnector, err = newPlexConnectorWithToken(f, opt.PlexURL, opt.PlexToken, opt.PlexInsecure)			if err != nil {				return nil, errors.Wrapf(err, "failed to connect to the Plex API %v", opt.PlexURL)			}		} else {			if opt.PlexPassword != "" && opt.PlexUsername != "" {				decPass, err := obscure.Reveal(opt.PlexPassword)				if err != nil {					decPass = opt.PlexPassword				}				f.plexConnector, err = newPlexConnector(f, opt.PlexURL, opt.PlexUsername, decPass, opt.PlexInsecure, func(token string) {					m.Set("plex_token", token)				})				if err != nil {					return nil, errors.Wrapf(err, "failed to connect to the Plex API %v", opt.PlexURL)				}			}		}	} 	dbPath := f.opt.DbPath	chunkPath := f.opt.ChunkPath	// if the dbPath is non default but the chunk path is default, we overwrite the last to follow the same one as dbPath	if dbPath != filepath.Join(config.CacheDir, "cache-backend") &&		chunkPath == filepath.Join(config.CacheDir, "cache-backend") {		chunkPath = dbPath	}	if filepath.Ext(dbPath) != "" {		dbPath = filepath.Dir(dbPath)	}	if filepath.Ext(chunkPath) != "" {		chunkPath = filepath.Dir(chunkPath)	}	err = os.MkdirAll(dbPath, os.ModePerm)	if err != nil {		return nil, errors.Wrapf(err, "failed to create cache directory %v", dbPath)	}	err = os.MkdirAll(chunkPath, os.ModePerm)	if err != nil {		return nil, errors.Wrapf(err, "failed to create cache directory %v", chunkPath)	} 	dbPath = filepath.Join(dbPath, name+".db")	chunkPath = filepath.Join(chunkPath, name)	fs.Infof(name, "Cache DB path: %v", dbPath)	fs.Infof(name, "Cache chunk path: %v", chunkPath)	f.cache, err = GetPersistent(dbPath, chunkPath, &Features{		PurgeDb:    opt.DbPurge,		DbWaitTime: time.Duration(opt.DbWaitTime),	})	if err != nil {		return nil, errors.Wrapf(err, "failed to start cache db")	}	// Trap SIGINT and SIGTERM to close the DB handle gracefully	c := make(chan os.Signal, 1)	signal.Notify(c, syscall.SIGHUP)	atexit.Register(func() {		if opt.PlexURL != "" {			f.plexConnector.closeWebsocket()		}		f.StopBackgroundRunners()	})	go func() {		for {			s := <-c			if s == syscall.SIGHUP {				fs.Infof(f, "Clearing cache from signal")				f.DirCacheFlush()			}		}	}() 	fs.Infof(name, "Chunk Memory: %v", !f.opt.ChunkNoMemory)	fs.Infof(name, "Chunk Size: %v", f.opt.ChunkSize)	fs.Infof(name, "Chunk Total Size: %v", f.opt.ChunkTotalSize)	fs.Infof(name, "Chunk Clean Interval: %v", f.opt.ChunkCleanInterval)	fs.Infof(name, "Workers: %v", f.opt.TotalWorkers)	fs.Infof(name, "File Age: %v", f.opt.InfoAge)	if f.opt.StoreWrites {		fs.Infof(name, "Cache Writes: enabled")	} 	if f.opt.TempWritePath != "" {		err = os.MkdirAll(f.opt.TempWritePath, os.ModePerm)		if err != nil {			return nil, errors.Wrapf(err, "failed to create cache directory %v", f.opt.TempWritePath)		}		f.opt.TempWritePath = filepath.ToSlash(f.opt.TempWritePath)		f.tempFs, err = cache.Get(f.opt.TempWritePath)		if err != nil {			return nil, errors.Wrapf(err, "failed to create temp fs: %v", err)		}		fs.Infof(name, "Upload Temp Rest Time: %v", f.opt.TempWaitTime)		fs.Infof(name, "Upload Temp FS: %v", f.opt.TempWritePath)		f.backgroundRunner, _ = initBackgroundUploader(f)		go f.backgroundRunner.run()	} 	go func() {		for {			time.Sleep(time.Duration(f.opt.ChunkCleanInterval))			select {			case <-f.cleanupChan:				fs.Infof(f, "stopping cleanup")

Lines 1–500 of 1902. Load more