syncthing/model/filequeue.go

240 lines
4.4 KiB
Go
Raw Normal View History

package model
import (
"log"
"sort"
"sync"
"time"
)
type Monitor interface {
FileBegins(<-chan content) error
FileDone() error
}
type FileQueue struct {
2014-01-13 17:29:23 +00:00
files queuedFileList
sorted bool
2014-01-18 03:06:44 +00:00
fmut sync.Mutex // protects files and sorted
2014-01-13 17:29:23 +00:00
availability map[string][]string
2014-01-18 03:06:44 +00:00
amut sync.Mutex // protects availability
2014-01-23 15:39:12 +00:00
queued map[string]bool
}
type queuedFile struct {
name string
blocks []Block
activeBlocks []bool
given int
remaining int
channel chan content
nodes []string
nodesChecked time.Time
monitor Monitor
}
type content struct {
offset int64
data []byte
}
type queuedFileList []queuedFile
func (l queuedFileList) Len() int { return len(l) }
func (l queuedFileList) Swap(a, b int) { l[a], l[b] = l[b], l[a] }
func (l queuedFileList) Less(a, b int) bool {
// Sort by most blocks already given out, then alphabetically
if l[a].given != l[b].given {
return l[a].given > l[b].given
}
return l[a].name < l[b].name
}
type queuedBlock struct {
name string
block Block
index int
}
2014-01-20 21:22:27 +00:00
func NewFileQueue() *FileQueue {
return &FileQueue{
availability: make(map[string][]string),
2014-01-23 15:39:12 +00:00
queued: make(map[string]bool),
2014-01-20 21:22:27 +00:00
}
}
func (q *FileQueue) Add(name string, blocks []Block, monitor Monitor) {
2014-01-18 03:06:44 +00:00
q.fmut.Lock()
defer q.fmut.Unlock()
2014-01-23 15:39:12 +00:00
if q.queued[name] {
return
2014-01-18 03:06:44 +00:00
}
q.files = append(q.files, queuedFile{
name: name,
blocks: blocks,
activeBlocks: make([]bool, len(blocks)),
remaining: len(blocks),
channel: make(chan content),
monitor: monitor,
})
2014-01-23 15:39:12 +00:00
q.queued[name] = true
q.sorted = false
}
func (q *FileQueue) Len() int {
2014-01-18 03:06:44 +00:00
q.fmut.Lock()
defer q.fmut.Unlock()
return len(q.files)
}
func (q *FileQueue) Get(nodeID string) (queuedBlock, bool) {
2014-01-18 03:06:44 +00:00
q.fmut.Lock()
defer q.fmut.Unlock()
if !q.sorted {
sort.Sort(q.files)
q.sorted = true
}
for i := range q.files {
2014-01-13 17:29:23 +00:00
qf := &q.files[i]
2014-01-18 03:06:44 +00:00
q.amut.Lock()
av := q.availability[qf.name]
q.amut.Unlock()
if len(av) == 0 {
// Noone has the file we want; abort.
2014-01-13 17:29:23 +00:00
if qf.remaining != len(qf.blocks) {
// We have already started on this file; close it down
2014-01-13 17:29:23 +00:00
close(qf.channel)
if mon := qf.monitor; mon != nil {
mon.FileDone()
}
}
2014-01-23 15:39:12 +00:00
delete(q.queued, qf.name)
2014-01-13 17:29:23 +00:00
q.deleteAt(i)
return queuedBlock{}, false
}
2014-01-18 03:06:44 +00:00
for _, ni := range av {
// Find and return the next block in the queue
if ni == nodeID {
for j, b := range qf.blocks {
if !qf.activeBlocks[j] {
2014-01-13 17:29:23 +00:00
qf.activeBlocks[j] = true
qf.given++
return queuedBlock{
name: qf.name,
block: b,
index: j,
}, true
}
}
break
}
}
}
// We found nothing to do
return queuedBlock{}, false
}
func (q *FileQueue) Done(file string, offset int64, data []byte) {
2014-01-18 03:06:44 +00:00
q.fmut.Lock()
defer q.fmut.Unlock()
c := content{
offset: offset,
data: data,
}
2014-01-13 17:29:23 +00:00
for i := range q.files {
qf := &q.files[i]
if qf.name == file {
if qf.monitor != nil && qf.remaining == len(qf.blocks) {
err := qf.monitor.FileBegins(qf.channel)
if err != nil {
log.Printf("WARNING: %s: %v (not synced)", qf.name, err)
2014-01-23 15:39:12 +00:00
delete(q.queued, qf.name)
2014-01-13 17:29:23 +00:00
q.deleteAt(i)
return
}
}
qf.channel <- c
2014-01-13 17:29:23 +00:00
qf.remaining--
2014-01-13 17:29:23 +00:00
if qf.remaining == 0 {
close(qf.channel)
if qf.monitor != nil {
err := qf.monitor.FileDone()
if err != nil {
log.Printf("WARNING: %s: %v", qf.name, err)
}
}
2014-01-23 15:39:12 +00:00
delete(q.queued, qf.name)
2014-01-13 17:29:23 +00:00
q.deleteAt(i)
}
return
}
}
// We found nothing, might have errored out already
}
func (q *FileQueue) QueuedFiles() (files []string) {
2014-01-18 03:06:44 +00:00
q.fmut.Lock()
defer q.fmut.Unlock()
for _, qf := range q.files {
files = append(files, qf.name)
}
return
}
2014-01-13 17:29:23 +00:00
func (q *FileQueue) deleteAt(i int) {
q.files = append(q.files[:i], q.files[i+1:]...)
}
2014-01-13 17:29:23 +00:00
func (q *FileQueue) deleteFile(n string) {
for i, file := range q.files {
if n == file.name {
q.deleteAt(i)
2014-01-23 15:39:12 +00:00
delete(q.queued, file.name)
return
}
}
}
2014-01-20 21:22:27 +00:00
func (q *FileQueue) SetAvailable(file string, nodes []string) {
2014-01-18 03:06:44 +00:00
q.amut.Lock()
defer q.amut.Unlock()
2014-01-20 21:22:27 +00:00
q.availability[file] = nodes
2014-01-13 17:29:23 +00:00
}
func (q *FileQueue) RemoveAvailable(toRemove string) {
2014-01-18 03:06:44 +00:00
q.amut.Lock()
defer q.amut.Unlock()
for file, nodes := range q.availability {
for i, node := range nodes {
if node == toRemove {
q.availability[file] = nodes[:i+copy(nodes[i:], nodes[i+1:])]
if len(q.availability[file]) == 0 {
q.fmut.Lock()
q.deleteFile(file)
q.fmut.Unlock()
}
}
break
}
}
}