forked from keybase/client
-
Notifications
You must be signed in to change notification settings - Fork 0
/
block_retrieval_worker.go
127 lines (113 loc) · 3.06 KB
/
block_retrieval_worker.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
// Copyright 2016 Keybase Inc. All rights reserved.
// Use of this source code is governed by a BSD
// license that can be found in the LICENSE file.
package libkbfs
import (
"io"
"github.com/eapache/channels"
"github.com/keybase/client/go/kbfs/data"
)
// blockRetrievalWorker processes blockRetrievalQueue requests
type blockRetrievalWorker struct {
blockGetter
stopCh chan struct{}
doneCh chan struct{}
queue *blockRetrievalQueue
workCh channels.Channel
}
// run runs the worker loop until Shutdown is called
func (brw *blockRetrievalWorker) run() {
defer close(brw.doneCh)
for {
err := brw.HandleRequest()
// Only io.EOF is relevant to the loop; other errors are handled in
// FinalizeRequest
if err == io.EOF {
return
}
}
}
// newBlockRetrievalWorker returns a blockRetrievalWorker for a given
// blockRetrievalQueue, using the passed in blockGetter to obtain blocks for
// requests.
func newBlockRetrievalWorker(bg blockGetter, q *blockRetrievalQueue,
workCh channels.Channel) *blockRetrievalWorker {
brw := &blockRetrievalWorker{
blockGetter: bg,
stopCh: make(chan struct{}),
doneCh: make(chan struct{}),
queue: q,
workCh: workCh,
}
go brw.run()
return brw
}
// HandleRequest is the main work method for the worker. It obtains a
// blockRetrieval from the queue, retrieves the block using
// blockGetter.getBlock, and responds to the subscribed requestors with the
// results.
func (brw *blockRetrievalWorker) HandleRequest() (err error) {
var retrieval *blockRetrieval
select {
case <-brw.workCh.Out():
retrieval = brw.queue.popIfNotEmpty()
if retrieval == nil {
return nil
}
case <-brw.stopCh:
return io.EOF
}
var block data.Block
var cacheType DiskBlockCacheType
defer func() {
brw.queue.FinalizeRequest(retrieval, block, cacheType, err)
}()
// Handle canceled contexts.
select {
case <-retrieval.ctx.Done():
return retrieval.ctx.Err()
default:
}
var action BlockRequestAction
func() {
retrieval.reqMtx.RLock()
defer retrieval.reqMtx.RUnlock()
block = retrieval.requests[0].block.NewEmpty()
action = retrieval.action
}()
// If we running with a "stop-if-full" action, before we fetch the
// block, make sure the disk cache has room for it.
if action.StopIfFull() {
dbc := brw.queue.config.DiskBlockCache()
if dbc != nil {
hasRoom, _, err := dbc.DoesCacheHaveSpace(
retrieval.ctx, action.CacheType())
if err != nil {
return err
}
if !hasRoom {
return DiskCacheTooFullForBlockError{retrieval.blockPtr, action}
}
}
}
cacheType = action.CacheType()
if action.DelayCacheCheck() {
_, err := brw.queue.checkCaches(
retrieval.ctx, retrieval.kmd, retrieval.blockPtr, block,
action.WithoutDelayedCacheCheckAction())
if err == nil {
return nil
}
}
return brw.getBlock(
retrieval.ctx, retrieval.kmd, retrieval.blockPtr, block, cacheType)
}
// Shutdown shuts down the blockRetrievalWorker once its current work is done.
func (brw *blockRetrievalWorker) Shutdown() <-chan struct{} {
select {
case <-brw.stopCh:
default:
close(brw.stopCh)
}
return brw.doneCh
}