133728e barerepo 1mo
1
package gitx
133728e barerepo 1mo
3
import (
133728e barerepo 1mo
4
"bufio"
133728e barerepo 1mo
5
"container/list"
133728e barerepo 1mo
6
"context"
133728e barerepo 1mo
7
"fmt"
133728e barerepo 1mo
8
"io"
133728e barerepo 1mo
9
"os/exec"
133728e barerepo 1mo
10
"strconv"
133728e barerepo 1mo
11
"strings"
133728e barerepo 1mo
12
"sync"
133728e barerepo 1mo
13
"time"
133728e barerepo 1mo
16
// idlePerDir bounds how many readers one repository keeps, so a hot repo does not serialise.
133728e barerepo 1mo
17
const idlePerDir = 4
133728e barerepo 1mo
19
// idleTotal bounds the whole server, because a barerepo holds more repositories than processes.
133728e barerepo 1mo
20
const idleTotal = 64
133728e barerepo 1mo
22
// IdleLife is how long an unused reader is kept, and Sweep closes what is older.
133728e barerepo 1mo
23
const IdleLife = 10 * time.Minute
133728e barerepo 1mo
25
// reader is one cat-file --batch kept open, because forking git costs more than the read does.
133728e barerepo 1mo
26
type reader struct {
133728e barerepo 1mo
27
dir string
133728e barerepo 1mo
28
cmd *exec.Cmd
133728e barerepo 1mo
29
in io.WriteCloser
133728e barerepo 1mo
30
out *bufio.Reader
133728e barerepo 1mo
31
idle time.Time
133728e barerepo 1mo
32
spot *list.Element
133728e barerepo 1mo
35
var (
133728e barerepo 1mo
36
poolMu sync.Mutex
133728e barerepo 1mo
37
// free holds readers by directory, and order holds the same readers oldest first.
133728e barerepo 1mo
38
free = map[string][]*reader{}
133728e barerepo 1mo
39
order = list.New()
133728e barerepo 1mo
42
// take returns an idle reader for dir, or nothing if none is waiting.
133728e barerepo 1mo
43
func take(dir string) *reader {
133728e barerepo 1mo
44
poolMu.Lock()
133728e barerepo 1mo
45
defer poolMu.Unlock()
133728e barerepo 1mo
46
have := free[dir]
133728e barerepo 1mo
47
if len(have) == 0 {
133728e barerepo 1mo
48
return nil
133728e barerepo 1mo
50
r := have[len(have)-1]
133728e barerepo 1mo
51
free[dir] = have[:len(have)-1]
133728e barerepo 1mo
52
if len(free[dir]) == 0 {
133728e barerepo 1mo
53
delete(free, dir)
133728e barerepo 1mo
55
order.Remove(r.spot)
133728e barerepo 1mo
56
r.spot = nil
133728e barerepo 1mo
57
return r
133728e barerepo 1mo
60
// put keeps a reader for the next request, or closes it when the pool is full.
133728e barerepo 1mo
61
func put(r *reader) {
133728e barerepo 1mo
62
poolMu.Lock()
133728e barerepo 1mo
63
if len(free[r.dir]) >= idlePerDir {
133728e barerepo 1mo
64
poolMu.Unlock()
133728e barerepo 1mo
65
r.close()
133728e barerepo 1mo
66
return
133728e barerepo 1mo
68
r.idle = time.Now()
133728e barerepo 1mo
69
free[r.dir] = append(free[r.dir], r)
133728e barerepo 1mo
70
r.spot = order.PushBack(r)
133728e barerepo 1mo
71
var evict *reader
133728e barerepo 1mo
72
if order.Len() > idleTotal {
133728e barerepo 1mo
73
evict = drop(order.Front())
133728e barerepo 1mo
75
poolMu.Unlock()
133728e barerepo 1mo
76
if evict != nil {
133728e barerepo 1mo
77
evict.close()
133728e barerepo 1mo
81
// drop removes one reader from both structures, and the caller closes it outside the lock.
133728e barerepo 1mo
82
func drop(e *list.Element) *reader {
133728e barerepo 1mo
83
if e == nil {
133728e barerepo 1mo
84
return nil
133728e barerepo 1mo
86
r := e.Value.(*reader)
133728e barerepo 1mo
87
order.Remove(e)
133728e barerepo 1mo
88
r.spot = nil
133728e barerepo 1mo
89
have := free[r.dir]
133728e barerepo 1mo
90
for i, other := range have {
133728e barerepo 1mo
91
if other == r {
133728e barerepo 1mo
92
free[r.dir] = append(have[:i], have[i+1:]...)
133728e barerepo 1mo
93
break
133728e barerepo 1mo
96
if len(free[r.dir]) == 0 {
133728e barerepo 1mo
97
delete(free, r.dir)
133728e barerepo 1mo
99
return r
133728e barerepo 1mo
102
// CloseIdleReaders closes every reader unused for longer than age, and returns how many.
133728e barerepo 1mo
103
func CloseIdleReaders(age time.Duration) int {
133728e barerepo 1mo
104
cutoff := time.Now().Add(-age)
133728e barerepo 1mo
105
var stale []*reader
133728e barerepo 1mo
106
poolMu.Lock()
133728e barerepo 1mo
107
for e := order.Front(); e != nil; {
133728e barerepo 1mo
108
next := e.Next()
133728e barerepo 1mo
109
if e.Value.(*reader).idle.After(cutoff) {
133728e barerepo 1mo
110
break
133728e barerepo 1mo
111
}
133728e barerepo 1mo
112
stale = append(stale, drop(e))
133728e barerepo 1mo
113
e = next
133728e barerepo 1mo
114
}
133728e barerepo 1mo
115
poolMu.Unlock()
133728e barerepo 1mo
116
for _, r := range stale {
133728e barerepo 1mo
117
r.close()
133728e barerepo 1mo
118
}
133728e barerepo 1mo
119
return len(stale)
133728e barerepo 1mo
122
// start opens a new reader, which is the only place a cat-file process is created.
133728e barerepo 1mo
123
func start(dir string) (*reader, error) {
133728e barerepo 1mo
124
cmd := exec.Command(Bin, "cat-file", "--batch")
133728e barerepo 1mo
125
cmd.Dir = dir
133728e barerepo 1mo
126
cmd.Env = env()
133728e barerepo 1mo
127
in, err := cmd.StdinPipe()
133728e barerepo 1mo
128
if err != nil {
133728e barerepo 1mo
129
return nil, err
133728e barerepo 1mo
130
}
133728e barerepo 1mo
131
out, err := cmd.StdoutPipe()
133728e barerepo 1mo
132
if err != nil {
133728e barerepo 1mo
133
in.Close()
133728e barerepo 1mo
134
return nil, err
133728e barerepo 1mo
135
}
133728e barerepo 1mo
136
if err := cmd.Start(); err != nil {
133728e barerepo 1mo
137
in.Close()
133728e barerepo 1mo
138
return nil, err
133728e barerepo 1mo
139
}
133728e barerepo 1mo
140
return &reader{dir: dir, cmd: cmd, in: in, out: bufio.NewReaderSize(out, 64<<10)}, nil
133728e barerepo 1mo
143
func (r *reader) close() {
133728e barerepo 1mo
144
r.in.Close()
133728e barerepo 1mo
145
if r.cmd.Process != nil {
133728e barerepo 1mo
146
r.cmd.Process.Kill()
133728e barerepo 1mo
147
}
133728e barerepo 1mo
148
r.cmd.Wait()
133728e barerepo 1mo
151
// ask writes every spec and reads every answer, and any surprise means the stream is out of step.
133728e barerepo 1mo
152
func (r *reader) ask(specs []string) (map[string]*Object, error) {
133728e barerepo 1mo
153
// Written from another goroutine, because a big batch fills the pipe before git has answered.
133728e barerepo 1mo
154
sent := make(chan error, 1)
133728e barerepo 1mo
155
go func() {
133728e barerepo 1mo
156
_, err := io.WriteString(r.in, strings.Join(specs, "\n")+"\n")
133728e barerepo 1mo
157
sent <- err
133728e barerepo 1mo
158
}()
133728e barerepo 1mo
159
out := make(map[string]*Object, len(specs))
133728e barerepo 1mo
160
for _, spec := range specs {
133728e barerepo 1mo
161
header, err := r.out.ReadString('\n')
133728e barerepo 1mo
162
if err != nil {
133728e barerepo 1mo
163
<-sent
133728e barerepo 1mo
164
return nil, err
133728e barerepo 1mo
165
}
133728e barerepo 1mo
166
fields := strings.Fields(header)
133728e barerepo 1mo
167
// "<spec> missing" is git's answer, and it rereads the pack directory before saying it.
133728e barerepo 1mo
168
if len(fields) < 3 {
133728e barerepo 1mo
169
continue
133728e barerepo 1mo
170
}
133728e barerepo 1mo
171
size, err := strconv.ParseInt(fields[2], 10, 64)
133728e barerepo 1mo
172
if err != nil || size < 0 {
133728e barerepo 1mo
173
<-sent
133728e barerepo 1mo
174
return nil, fmt.Errorf("cat-file said %q", strings.TrimSpace(header))
133728e barerepo 1mo
175
}
133728e barerepo 1mo
176
body := make([]byte, size+1)
133728e barerepo 1mo
177
if _, err := io.ReadFull(r.out, body); err != nil {
133728e barerepo 1mo
178
<-sent
133728e barerepo 1mo
179
return nil, err
133728e barerepo 1mo
180
}
133728e barerepo 1mo
181
out[spec] = &Object{SHA: fields[0], Type: fields[1], Size: size, Body: string(body[:size])}
133728e barerepo 1mo
182
}
133728e barerepo 1mo
183
if err := <-sent; err != nil {
133728e barerepo 1mo
184
return nil, err
133728e barerepo 1mo
185
}
133728e barerepo 1mo
186
return out, nil
133728e barerepo 1mo
189
// Batch reads many objects without starting a process, which is what chapter 25's budget needs.
133728e barerepo 1mo
190
func Batch(ctx context.Context, dir string, specs []string) (map[string]*Object, error) {
133728e barerepo 1mo
191
if len(specs) == 0 {
133728e barerepo 1mo
192
return map[string]*Object{}, nil
133728e barerepo 1mo
193
}
133728e barerepo 1mo
194
if err := ctx.Err(); err != nil {
133728e barerepo 1mo
195
return nil, err
133728e barerepo 1mo
196
}
133728e barerepo 1mo
197
if r := take(dir); r != nil {
133728e barerepo 1mo
198
out, err := r.ask(specs)
133728e barerepo 1mo
199
if err == nil {
133728e barerepo 1mo
200
put(r)
133728e barerepo 1mo
201
return out, nil
133728e barerepo 1mo
202
}
133728e barerepo 1mo
203
r.close()
133728e barerepo 1mo
204
}
133728e barerepo 1mo
205
r, err := start(dir)
133728e barerepo 1mo
206
if err != nil {
133728e barerepo 1mo
207
return nil, err
133728e barerepo 1mo
208
}
133728e barerepo 1mo
209
out, err := r.ask(specs)
133728e barerepo 1mo
210
if err != nil {
133728e barerepo 1mo
211
r.close()
133728e barerepo 1mo
212
return nil, err
133728e barerepo 1mo
213
}
133728e barerepo 1mo
214
put(r)
133728e barerepo 1mo
215
return out, nil