parent
a5699d40a8
commit
5edf4660e2
|
@ -490,17 +490,21 @@ func TestDeregisterOnFailInCopy(t *testing.T) {
|
||||||
t.Fatal("Expected file in progress")
|
t.Fatal("Expected file in progress")
|
||||||
}
|
}
|
||||||
|
|
||||||
copyChan := make(chan copyBlocksState)
|
|
||||||
pullChan := make(chan pullBlockState)
|
pullChan := make(chan pullBlockState)
|
||||||
finisherBufferChan := make(chan *sharedPullerState)
|
finisherBufferChan := make(chan *sharedPullerState)
|
||||||
finisherChan := make(chan *sharedPullerState)
|
finisherChan := make(chan *sharedPullerState)
|
||||||
dbUpdateChan := make(chan dbUpdateJob, 1)
|
dbUpdateChan := make(chan dbUpdateJob, 1)
|
||||||
|
|
||||||
go f.copierRoutine(copyChan, pullChan, finisherBufferChan)
|
copyChan, copyWg := startCopier(f, pullChan, finisherBufferChan)
|
||||||
go f.finisherRoutine(finisherChan, dbUpdateChan, make(chan string))
|
go f.finisherRoutine(finisherChan, dbUpdateChan, make(chan string))
|
||||||
defer close(copyChan)
|
|
||||||
defer close(pullChan)
|
defer func() {
|
||||||
defer close(finisherChan)
|
close(copyChan)
|
||||||
|
copyWg.Wait()
|
||||||
|
close(pullChan)
|
||||||
|
close(finisherBufferChan)
|
||||||
|
close(finisherChan)
|
||||||
|
}()
|
||||||
|
|
||||||
f.handleFile(file, copyChan, dbUpdateChan)
|
f.handleFile(file, copyChan, dbUpdateChan)
|
||||||
|
|
||||||
|
@ -508,15 +512,15 @@ func TestDeregisterOnFailInCopy(t *testing.T) {
|
||||||
// loop has been performed.
|
// loop has been performed.
|
||||||
toPull := <-pullChan
|
toPull := <-pullChan
|
||||||
|
|
||||||
// Close the file, causing errors on further access
|
|
||||||
toPull.sharedPullerState.fail(os.ErrNotExist)
|
|
||||||
|
|
||||||
// Unblock copier
|
// Unblock copier
|
||||||
go func() {
|
go func() {
|
||||||
for range pullChan {
|
for range pullChan {
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
// Close the file, causing errors on further access
|
||||||
|
toPull.sharedPullerState.fail(os.ErrNotExist)
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case state := <-finisherBufferChan:
|
case state := <-finisherBufferChan:
|
||||||
// At this point the file should still be registered with both the job
|
// At this point the file should still be registered with both the job
|
||||||
|
@ -580,26 +584,53 @@ func TestDeregisterOnFailInPull(t *testing.T) {
|
||||||
t.Fatal("Expected file in progress")
|
t.Fatal("Expected file in progress")
|
||||||
}
|
}
|
||||||
|
|
||||||
copyChan := make(chan copyBlocksState)
|
|
||||||
pullChan := make(chan pullBlockState)
|
pullChan := make(chan pullBlockState)
|
||||||
finisherBufferChan := make(chan *sharedPullerState)
|
finisherBufferChan := make(chan *sharedPullerState)
|
||||||
finisherChan := make(chan *sharedPullerState)
|
finisherChan := make(chan *sharedPullerState)
|
||||||
dbUpdateChan := make(chan dbUpdateJob, 1)
|
dbUpdateChan := make(chan dbUpdateJob, 1)
|
||||||
|
|
||||||
go f.copierRoutine(copyChan, pullChan, finisherBufferChan)
|
copyChan, copyWg := startCopier(f, pullChan, finisherBufferChan)
|
||||||
go f.pullerRoutine(pullChan, finisherBufferChan)
|
pullWg := sync.NewWaitGroup()
|
||||||
|
pullWg.Add(1)
|
||||||
|
go func() {
|
||||||
|
f.pullerRoutine(pullChan, finisherBufferChan)
|
||||||
|
pullWg.Done()
|
||||||
|
}()
|
||||||
go f.finisherRoutine(finisherChan, dbUpdateChan, make(chan string))
|
go f.finisherRoutine(finisherChan, dbUpdateChan, make(chan string))
|
||||||
defer close(copyChan)
|
defer func() {
|
||||||
defer close(pullChan)
|
// Unblock copier and puller
|
||||||
defer close(finisherChan)
|
go func() {
|
||||||
|
for range finisherBufferChan {
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
close(copyChan)
|
||||||
|
copyWg.Wait()
|
||||||
|
close(pullChan)
|
||||||
|
pullWg.Wait()
|
||||||
|
close(finisherBufferChan)
|
||||||
|
close(finisherChan)
|
||||||
|
}()
|
||||||
|
|
||||||
f.handleFile(file, copyChan, dbUpdateChan)
|
f.handleFile(file, copyChan, dbUpdateChan)
|
||||||
|
|
||||||
// Receive at finisher, we should error out as puller has nowhere to pull
|
// Receive at finisher, we should error out as puller has nowhere to pull
|
||||||
// from.
|
// from.
|
||||||
timeout = time.Second
|
timeout = time.Second
|
||||||
|
|
||||||
|
// Both the puller and copier may send to the finisherBufferChan.
|
||||||
|
var state *sharedPullerState
|
||||||
|
after := time.After(5 * time.Second)
|
||||||
|
for {
|
||||||
select {
|
select {
|
||||||
case state := <-finisherBufferChan:
|
case state = <-finisherBufferChan:
|
||||||
|
case <-after:
|
||||||
|
t.Fatal("Didn't get failed state to the finisher")
|
||||||
|
}
|
||||||
|
if state.failed() != nil {
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// At this point the file should still be registered with both the job
|
// At this point the file should still be registered with both the job
|
||||||
// queue, and the progress emitter. Verify this.
|
// queue, and the progress emitter. Verify this.
|
||||||
if f.model.progressEmitter.lenRegistry() != 1 || f.queue.lenProgress() != 1 || f.queue.lenQueued() != 0 {
|
if f.model.progressEmitter.lenRegistry() != 1 || f.queue.lenProgress() != 1 || f.queue.lenQueued() != 0 {
|
||||||
|
@ -638,9 +669,6 @@ func TestDeregisterOnFailInPull(t *testing.T) {
|
||||||
if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
|
if f.model.progressEmitter.lenRegistry() != 0 || f.queue.lenProgress() != 0 || f.queue.lenQueued() != 0 {
|
||||||
t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
|
t.Fatal("Still registered", f.model.progressEmitter.lenRegistry(), f.queue.lenProgress(), f.queue.lenQueued())
|
||||||
}
|
}
|
||||||
case <-time.After(5 * time.Second):
|
|
||||||
t.Fatal("Didn't get anything to the finisher")
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestIssue3164(t *testing.T) {
|
func TestIssue3164(t *testing.T) {
|
||||||
|
@ -830,11 +858,14 @@ func TestCopyOwner(t *testing.T) {
|
||||||
// comes the finisher is done.
|
// comes the finisher is done.
|
||||||
|
|
||||||
finisherChan := make(chan *sharedPullerState)
|
finisherChan := make(chan *sharedPullerState)
|
||||||
defer close(finisherChan)
|
copierChan, copyWg := startCopier(f, nil, finisherChan)
|
||||||
copierChan := make(chan copyBlocksState)
|
|
||||||
defer close(copierChan)
|
|
||||||
go f.copierRoutine(copierChan, nil, finisherChan)
|
|
||||||
go f.finisherRoutine(finisherChan, dbUpdateChan, nil)
|
go f.finisherRoutine(finisherChan, dbUpdateChan, nil)
|
||||||
|
defer func() {
|
||||||
|
close(copierChan)
|
||||||
|
copyWg.Wait()
|
||||||
|
close(finisherChan)
|
||||||
|
}()
|
||||||
|
|
||||||
f.handleFile(file, copierChan, nil)
|
f.handleFile(file, copierChan, nil)
|
||||||
<-dbUpdateChan
|
<-dbUpdateChan
|
||||||
|
|
||||||
|
@ -993,3 +1024,14 @@ func cleanupSharedPullerState(s *sharedPullerState) {
|
||||||
s.writer.fd.Close()
|
s.writer.fd.Close()
|
||||||
s.writer.mut.Unlock()
|
s.writer.mut.Unlock()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func startCopier(f *sendReceiveFolder, pullChan chan<- pullBlockState, finisherChan chan<- *sharedPullerState) (chan copyBlocksState, sync.WaitGroup) {
|
||||||
|
copyChan := make(chan copyBlocksState)
|
||||||
|
wg := sync.NewWaitGroup()
|
||||||
|
wg.Add(1)
|
||||||
|
go func() {
|
||||||
|
f.copierRoutine(copyChan, pullChan, finisherChan)
|
||||||
|
wg.Done()
|
||||||
|
}()
|
||||||
|
return copyChan, wg
|
||||||
|
}
|
||||||
|
|
|
@ -66,6 +66,7 @@ func TestProgressEmitter(t *testing.T) {
|
||||||
|
|
||||||
p := NewProgressEmitter(c, evLogger)
|
p := NewProgressEmitter(c, evLogger)
|
||||||
go p.Serve()
|
go p.Serve()
|
||||||
|
defer p.Stop()
|
||||||
p.interval = 0
|
p.interval = 0
|
||||||
|
|
||||||
expectTimeout(w, t)
|
expectTimeout(w, t)
|
||||||
|
|
Loading…
Reference in New Issue