jkl1337/duplicacy
1
0
Fork 0
mirror of https://github.com/jkl1337/duplicacy.git synced 2026-01-02 11:44:45 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #236 from gilbertchen/google_rate_limit_exceeded_followup

Browse Source 
Post-review changes for GCD rate limit handling
This commit is contained in:
gilbertchen
2017-10-18 13:14:13 -04:00
committed by GitHub
parent 47137b85e3 b788b9887c
commit c08a26a0c2
1 changed files with 33 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/duplicacy_gcdstorage.go
View File

@@ -27,11 +27,11 @@ import (
type GCDStorage struct { type GCDStorage struct {
RateLimitedStorage RateLimitedStorage
service *drive.Service service *drive.Service
idCache map[string]string idCache map[string]string
idCacheLock *sync.Mutex idCacheLock *sync.Mutex
backoffs []float64 backoffs []int // desired backoff time in seconds for each thread
backoffsRetries []int attempts []int // number of failed attempts since last success for each thread
isConnected bool isConnected bool
numberOfThreads int numberOfThreads int
@@ -46,19 +46,18 @@ type GCDConfig struct {
} }
func (storage *GCDStorage) shouldRetry(threadIndex int, err error) (bool, error) { func (storage *GCDStorage) shouldRetry(threadIndex int, err error) (bool, error) {
const LIMIT_BACKOFF_TIME = 64
const MAX_NUMBER_OF_RETRIES = 15
minimumSleepRatio := 0.1
maximumSleepRatio := 0.2
minimumSleep := float64(storage.numberOfThreads) * minimumSleepRatio
maximumSleep := float64(storage.numberOfThreads) * maximumSleepRatio
rand.Seed(time.Now().UnixNano()) // unsure if this is needed
const MAX_ATTEMPTS = 15
maximumBackoff := 64
if maximumBackoff < storage.numberOfThreads {
maximumBackoff = storage.numberOfThreads
}
retry := false retry := false
message := "" message := ""
if err == nil { if err == nil {
storage.backoffs[threadIndex] = computeInitialBackoff(minimumSleep, maximumSleep) storage.backoffs[threadIndex] = 1
storage.backoffsRetries[threadIndex] = 0 storage.attempts[threadIndex] = 0
return false, nil return false, nil
} else if e, ok := err.(*googleapi.Error); ok { } else if e, ok := err.(*googleapi.Error); ok {
if 500 <= e.Code && e.Code < 600 { if 500 <= e.Code && e.Code < 600 {
@@ -71,9 +70,8 @@ func (storage *GCDStorage) shouldRetry(threadIndex int, err error) (bool, error)
retry = true retry = true
} else if e.Code == 403 { } else if e.Code == 403 {
// User Rate Limit Exceeded // User Rate Limit Exceeded
message = e.Message // "User Rate Limit Exceeded" message = e.Message
retry = true retry = true
} else if e.Code == 401 { } else if e.Code == 401 {
// Only retry on authorization error when storage has been connected before // Only retry on authorization error when storage has been connected before
if storage.isConnected { if storage.isConnected {
@@ -93,37 +91,29 @@ func (storage *GCDStorage) shouldRetry(threadIndex int, err error) (bool, error)
retry = err.Temporary() retry = err.Temporary()
} }
if !retry || storage.backoffsRetries[threadIndex] >= MAX_NUMBER_OF_RETRIES { if !retry || storage.attempts[threadIndex] >= MAX_ATTEMPTS {
LOG_INFO("GCD_RETRY", "Thread: %03d. Maximum number of retries reached. Backoff time: %.2f. Number of retries: %d", threadIndex, storage.backoffs[threadIndex], storage.backoffsRetries[threadIndex]) LOG_INFO("GCD_RETRY", "[%d] Maximum number of retries reached (backoff: %d, attempts: %d)",
storage.backoffs[threadIndex] = computeInitialBackoff(minimumSleep, maximumSleep) threadIndex, storage.backoffs[threadIndex], storage.attempts[threadIndex])
storage.backoffsRetries[threadIndex] = 0 storage.backoffs[threadIndex] = 1
storage.attempts[threadIndex] = 0
return false, err return false, err
} }
if storage.backoffs[threadIndex] < LIMIT_BACKOFF_TIME { if storage.backoffs[threadIndex] < maximumBackoff {
storage.backoffs[threadIndex] *= 2.0 storage.backoffs[threadIndex] *= 2
} else {
storage.backoffs[threadIndex] = LIMIT_BACKOFF_TIME
storage.backoffsRetries[threadIndex] += 1
} }
delay := storage.backoffs[threadIndex]*rand.Float64() + storage.backoffs[threadIndex]*rand.Float64() if storage.backoffs[threadIndex] > maximumBackoff {
LOG_DEBUG("GCD_RETRY", "Thread: %3d. Message: %s. Retrying after %6.2f seconds. Current backoff: %6.2f. Number of retries: %2d.", threadIndex, message, delay, storage.backoffs[threadIndex], storage.backoffsRetries[threadIndex]) storage.backoffs[threadIndex] = maximumBackoff
}
storage.attempts[threadIndex] += 1
delay := float64(storage.backoffs[threadIndex]) * rand.Float64() * 2
LOG_DEBUG("GCD_RETRY", "[%d] %s; retrying after %.2f seconds (backoff: %d, attempts: %d)",
threadIndex, message, delay, storage.backoffs[threadIndex], storage.attempts[threadIndex])
time.Sleep(time.Duration(delay * float64(time.Second))) time.Sleep(time.Duration(delay * float64(time.Second)))
return true, nil return true, nil
} }
/*
logic for said calculus is here: https://stackoverflow.com/questions/1527803/generating-random-whole-numbers-in-javascript-in-a-specific-range
chose 0.1*thread number as a minimum sleep time
and 0.2*thread number as a maximum sleep time
for the first sleep of the first backoff of the threads.
This would mean that both when the program is started, and when multiple threads retry, google won't be ddosed :^)
*/
func computeInitialBackoff(minimumSleep float64, maximumSleep float64) float64 {
return rand.Float64()*(maximumSleep-minimumSleep+1) + minimumSleep
}
func (storage *GCDStorage) convertFilePath(filePath string) string { func (storage *GCDStorage) convertFilePath(filePath string) string {
if strings.HasPrefix(filePath, "chunks/") && strings.HasSuffix(filePath, ".fsl") { if strings.HasPrefix(filePath, "chunks/") && strings.HasSuffix(filePath, ".fsl") {
return "fossils/" + filePath[len("chunks/"):len(filePath)-len(".fsl")] return "fossils/" + filePath[len("chunks/"):len(filePath)-len(".fsl")]
@@ -303,12 +293,13 @@ func CreateGCDStorage(tokenFile string, storagePath string, threads int) (storag
numberOfThreads: threads, numberOfThreads: threads,
idCache: make(map[string]string), idCache: make(map[string]string),
idCacheLock: &sync.Mutex{}, idCacheLock: &sync.Mutex{},
backoffs: make([]float64, threads), backoffs: make([]int, threads),
backoffsRetries: make([]int, threads), attempts: make([]int, threads),
} }
for b := range storage.backoffs { for i := range storage.backoffs {
storage.backoffs[b] = 0.1 * float64(storage.numberOfThreads) // at the first error, we should still sleep some amount storage.backoffs[i] = 1
storage.attempts[i] = 0
} }
storagePathID, err := storage.getIDFromPath(0, storagePath) storagePathID, err := storage.getIDFromPath(0, storagePath)