-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #211 from bcc-code/feat/limit-rclone
Limit and prioritize Rclone transfers
- Loading branch information
Showing
23 changed files
with
261 additions
and
30 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,85 @@ | ||
| package rclone | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/ansel1/merry/v2" | ||
| ) | ||
|
|
||
| const maxConcurrentTransfers = 5 | ||
|
|
||
| var queueLock = sync.Mutex{} | ||
| var transferQueue = map[Priority][]chan bool{} | ||
|
|
||
| func init() { | ||
| for _, priority := range Priorities.Members() { | ||
| transferQueue[priority] = []chan bool{} | ||
| } | ||
| } | ||
|
|
||
| func waitForTransferSlot(priority Priority, timeout time.Duration) error { | ||
| // Create an unbuffered channel | ||
| ch := make(chan bool) | ||
|
|
||
| queueLock.Lock() | ||
| transferQueue[priority] = append(transferQueue[priority], ch) | ||
| queueLock.Unlock() | ||
|
|
||
| select { | ||
| case <-ch: | ||
| break | ||
| case <-time.After(timeout): | ||
| return merry.Wrap(errTimeout) | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func StartFileTransferQueue() { | ||
| for { | ||
| checkFileTransferQueue() | ||
| time.Sleep(time.Second * 5) | ||
| } | ||
| } | ||
|
|
||
| func checkFileTransferQueue() { | ||
| stats, _ := GetRcloneStatus() | ||
| count := len(stats.Transferring) | ||
|
|
||
| if count >= maxConcurrentTransfers { | ||
| return | ||
| } | ||
|
|
||
| queueLock.Lock() | ||
| defer queueLock.Unlock() | ||
|
|
||
| for _, priority := range Priorities.Members() { | ||
| started := 0 | ||
| for _, ch := range transferQueue[priority] { | ||
|
|
||
| // This is a non-blocking send, so if the channel is full, we can skip it. | ||
| // It basically means that the other side is not listening and we can move on. | ||
| // this approach works because we're using an unbuffered channel | ||
| select { | ||
| case ch <- true: | ||
| count++ | ||
| started++ | ||
| default: | ||
| } | ||
|
|
||
| if count >= maxConcurrentTransfers { | ||
| // If we've reached the maximum number of concurrent transfers, then we can stop processing the queue | ||
| // and remove the items that we've already started | ||
| transferQueue[priority] = transferQueue[priority][started:] | ||
| return | ||
| } | ||
| } | ||
|
|
||
| if started > 0 { | ||
| // If we get to here, then we've exhausted the queue for this priority and can replace it with an empty slice | ||
| transferQueue[priority] = []chan bool{} | ||
| } | ||
| } | ||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| package rclone | ||
|
|
||
| import ( | ||
| "net/http" | ||
| ) | ||
|
|
||
| type CoreStats struct { | ||
| Bytes int64 `json:"bytes"` | ||
| Checks int `json:"checks"` | ||
| DeletedDirs int `json:"deletedDirs"` | ||
| Deletes int `json:"deletes"` | ||
| ElapsedTime float64 `json:"elapsedTime"` | ||
| Errors int `json:"errors"` | ||
| Eta int `json:"eta"` | ||
| FatalError bool `json:"fatalError"` | ||
| LastError string `json:"lastError"` | ||
| Renames int `json:"renames"` | ||
| RetryError bool `json:"retryError"` | ||
| ServerSideCopies int `json:"serverSideCopies"` | ||
| ServerSideCopyBytes int `json:"serverSideCopyBytes"` | ||
| ServerSideMoveBytes int `json:"serverSideMoveBytes"` | ||
| ServerSideMoves int `json:"serverSideMoves"` | ||
| Speed float64 `json:"speed"` | ||
| TotalBytes int64 `json:"totalBytes"` | ||
| TotalChecks int `json:"totalChecks"` | ||
| TotalTransfers int `json:"totalTransfers"` | ||
| TransferTime float64 `json:"transferTime"` | ||
| Transferring []Transferring `json:"transferring"` | ||
| Transfers int `json:"transfers"` | ||
| } | ||
|
|
||
| type Transferring struct { | ||
| Bytes int64 `json:"bytes"` | ||
| DstFs string `json:"dstFs"` | ||
| Eta int `json:"eta"` | ||
| Group string `json:"group"` | ||
| Name string `json:"name"` | ||
| Percentage int `json:"percentage"` | ||
| Size int64 `json:"size"` | ||
| Speed float64 `json:"speed"` | ||
| SpeedAvg float64 `json:"speedAvg"` | ||
| SrcFs string `json:"srcFs"` | ||
| } | ||
|
|
||
| func GetRcloneStatus() (*CoreStats, error) { | ||
| req, err := http.NewRequest(http.MethodPost, baseUrl+"/core/stats", nil) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| res, err := doRequest[CoreStats](req) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| if res.Transferring == nil { | ||
| res.Transferring = []Transferring{} | ||
| } | ||
| return res, nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.