Skip to content

Commit 27bdf9b

Browse files
shankerwangmiaoshankerwangmiao
committed
git-fcgi: add buffering proxy to quickly read from fcgi
with a buffer in memory or on a local storage, to prevent cgi processes from waiting for network IO.
1 parent 85d740b commit 27bdf9b

3 files changed

Lines changed: 329 additions & 1 deletion

File tree

‎git-fcgi/Dockerfile‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,9 @@ RUN apt-get update && \
55
RUN mkdir -p /app
66
WORKDIR /app
77
COPY go-queue.go /app/
8+
COPY buffering-proxy.go /app/
89
RUN CGO_ENABLED=0 go build -o go-queue go-queue.go
10+
RUN CGO_ENABLED=0 go build -o buffering-proxy buffering-proxy.go
911

1012
FROM debian:bookworm
1113
LABEL maintainer="Justin Wong <yuzhi.wang@tuna.tsinghua.edu.cn>"
@@ -23,6 +25,7 @@ COPY queue-wrapper.sh /queue-wrapper.sh
2325
RUN mkdir -p /data
2426

2527
COPY --from=builder --chown=0:0 /app/go-queue /go-queue
28+
COPY --from=builder --chown=0:0 /app/buffering-proxy /buffering-proxy
2629

2730
EXPOSE 5000
2831
CMD [ "/cmd.sh" ]

‎git-fcgi/buffering-proxy.go‎

Lines changed: 317 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,317 @@
1+
package main
2+
3+
import (
4+
"bytes"
5+
"flag"
6+
"fmt"
7+
"io"
8+
"net"
9+
"os"
10+
"strconv"
11+
"strings"
12+
"sync"
13+
"sync/atomic"
14+
"time"
15+
)
16+
17+
type config struct {
18+
listeningAddress string
19+
upstreamAddress string
20+
maxResponseDeadline time.Duration
21+
maxBuffers uint64
22+
bufferUnitSize uint64
23+
onDiskBufferPath string
24+
onDiskBufferUnitSize uint64
25+
}
26+
27+
type UnitValue uint64
28+
29+
func (v *UnitValue) String() string {
30+
return strconv.FormatUint(uint64(*v), 10)
31+
}
32+
33+
func (v *UnitValue) Set(s string) error {
34+
s = strings.TrimSpace(s)
35+
val := strings.ToUpper(s)
36+
37+
var multiplier uint64 = 1
38+
switch {
39+
case strings.HasSuffix(val, "K"):
40+
multiplier = 1024
41+
val = strings.TrimSuffix(val, "K")
42+
case strings.HasSuffix(val, "M"):
43+
multiplier = 1024 * 1024
44+
val = strings.TrimSuffix(val, "M")
45+
case strings.HasSuffix(val, "G"):
46+
multiplier = 1024 * 1024 * 1024
47+
val = strings.TrimSuffix(val, "G")
48+
case strings.HasSuffix(val, "T"):
49+
multiplier = 1024 * 1024 * 1024 * 1024
50+
val = strings.TrimSuffix(val, "T")
51+
}
52+
53+
value, err := strconv.ParseUint(val, 10, 64)
54+
if err != nil {
55+
return fmt.Errorf("invalid number format: %v", err)
56+
}
57+
*v = UnitValue(value * multiplier)
58+
return nil
59+
}
60+
61+
func parseCLIParams() config {
62+
63+
// Define command-line flags
64+
listeningAddress := flag.String("listen", ":8080", "Address to listen on (e.g., :8080)")
65+
upstreamAddress := flag.String("upstream", "localhost:80", "Upstream server address (e.g., localhost:80)")
66+
maxResponseDeadline := flag.Duration("max-deadline", 1*time.Second, "Maximum response deadline (e.g., 1s)")
67+
var maxBufferSize UnitValue = 1 * 1024 * 1024 * 1024 // Default to 1GB
68+
flag.Var(&maxBufferSize, "max-buffer-size", "Maximum buffer size in memory")
69+
var bufferUnitSize UnitValue = UnitValue(os.Getpagesize()) // Default to system page size
70+
flag.Var(&bufferUnitSize, "buffer-unit-size", "Buffer unit size in memory")
71+
var onDiskBufferUnitSize UnitValue = 4 * 1024 * 1024 // Default to 4MB
72+
flag.Var(&onDiskBufferUnitSize, "on-disk-buffer-unit-size", "Buffer unit size on disk")
73+
onDiskBufferPath := flag.String("on-disk-buffer-path", "./buffer", "Path for on-disk buffer")
74+
75+
// Parse command-line flags
76+
flag.Parse()
77+
// Validate CLI parameters that must be positive to avoid panics or busy loops.
78+
if bufferUnitSize == 0 {
79+
fmt.Fprintln(os.Stderr, "invalid value for --buffer-unit-size: must be greater than zero")
80+
os.Exit(2)
81+
}
82+
if onDiskBufferUnitSize == 0 {
83+
fmt.Fprintln(os.Stderr, "invalid value for --on-disk-buffer-unit-size: must be greater than zero")
84+
os.Exit(2)
85+
}
86+
if *maxResponseDeadline <= 0 {
87+
fmt.Fprintln(os.Stderr, "invalid value for --max-deadline: must be greater than zero")
88+
os.Exit(2)
89+
}
90+
maxBuffers := maxBufferSize / bufferUnitSize
91+
if maxBuffers == 0 {
92+
panic("max-buffer-size must be at least as large as buffer-unit-size")
93+
}
94+
95+
return config{
96+
listeningAddress: *listeningAddress,
97+
upstreamAddress: *upstreamAddress,
98+
maxResponseDeadline: *maxResponseDeadline,
99+
maxBuffers: uint64(maxBuffers),
100+
bufferUnitSize: uint64(bufferUnitSize),
101+
onDiskBufferPath: *onDiskBufferPath,
102+
onDiskBufferUnitSize: uint64(onDiskBufferUnitSize),
103+
}
104+
}
105+
106+
type Buffer struct {
107+
memBuf *[]byte
108+
diskBuf *os.File
109+
}
110+
111+
var NrInMemBuffers atomic.Uint64
112+
113+
type StreamConn interface {
114+
net.Conn
115+
CloseRead() error
116+
CloseWrite() error
117+
}
118+
119+
func handleConnection(conn *net.TCPConn, cfg config) {
120+
var _uc net.Conn
121+
var err error
122+
if strings.HasPrefix(cfg.upstreamAddress, "unix:") {
123+
socketPath := strings.TrimPrefix(cfg.upstreamAddress, "unix:")
124+
_uc, err = net.Dial("unix", socketPath)
125+
} else {
126+
_uc, err = net.Dial("tcp", cfg.upstreamAddress)
127+
}
128+
if err != nil {
129+
fmt.Printf("Failed to connect to upstream server: %v\n", err)
130+
conn.Close()
131+
return
132+
}
133+
upstreamConn, ok := _uc.(StreamConn)
134+
if !ok {
135+
fmt.Printf("Failed to cast upstream connection to TCPConn\n")
136+
conn.Close()
137+
_uc.Close()
138+
return
139+
}
140+
141+
var wg sync.WaitGroup
142+
wg.Add(3)
143+
144+
go func() { //Uploading data from client to upstream, no buffering
145+
defer wg.Done()
146+
_, err := io.Copy(upstreamConn, conn)
147+
if err != nil {
148+
fmt.Printf("Error copying from client to upstream: %v\n", err)
149+
}
150+
upstreamConn.CloseWrite()
151+
conn.CloseRead()
152+
}()
153+
154+
go func() { //Downloading data from upstream to client, with buffering
155+
channel := make(chan Buffer, int(cfg.maxBuffers))
156+
go func() { // Sender
157+
defer wg.Done()
158+
closed := false
159+
handleSendError := func() {
160+
upstreamConn.CloseRead()
161+
}
162+
for buf := range channel {
163+
if buf.memBuf != nil {
164+
if !closed {
165+
_, err := io.Copy(conn, bytes.NewReader(*buf.memBuf))
166+
if err != nil {
167+
fmt.Printf("Error writing to client: %v\n", err)
168+
closed = true
169+
conn.CloseWrite()
170+
handleSendError()
171+
}
172+
}
173+
NrInMemBuffers.Add(^uint64(0))
174+
} else {
175+
if !closed {
176+
_, err := io.Copy(conn, buf.diskBuf)
177+
if err != nil {
178+
fmt.Printf("Error writing to client from disk buffer: %v\n", err)
179+
closed = true
180+
conn.CloseWrite()
181+
handleSendError()
182+
}
183+
}
184+
buf.diskBuf.Close()
185+
}
186+
}
187+
if !closed {
188+
conn.CloseWrite()
189+
}
190+
}()
191+
go func() { // Receiver
192+
defer wg.Done()
193+
for {
194+
upstreamConn.SetReadDeadline(time.Now().Add(cfg.maxResponseDeadline))
195+
var inMem bool
196+
for {
197+
curNrBuffers := NrInMemBuffers.Load()
198+
if curNrBuffers < cfg.maxBuffers {
199+
ok := NrInMemBuffers.CompareAndSwap(curNrBuffers, curNrBuffers+1)
200+
if ok {
201+
inMem = true
202+
break
203+
}
204+
} else {
205+
inMem = false
206+
break
207+
}
208+
}
209+
var readErr error
210+
if inMem {
211+
buf := make([]byte, int(cfg.bufferUnitSize))
212+
var n int
213+
n, readErr = io.ReadFull(upstreamConn, buf)
214+
if n > 0 {
215+
buf = buf[:n]
216+
channel <- Buffer{memBuf: &buf}
217+
} else {
218+
NrInMemBuffers.Add(^uint64(0))
219+
}
220+
} else {
221+
tmpFile, err := os.CreateTemp(cfg.onDiskBufferPath, "buffer-*")
222+
if err != nil {
223+
fmt.Printf("Error creating temp file for disk buffer: %v\n", err)
224+
close(channel)
225+
upstreamConn.CloseRead()
226+
break
227+
}
228+
err = os.Remove(tmpFile.Name())
229+
if err != nil {
230+
fmt.Printf("Error removing temp file after creation: %v\n", err)
231+
tmpFile.Close()
232+
close(channel)
233+
upstreamConn.CloseRead()
234+
break
235+
}
236+
var n int64
237+
n, readErr = io.CopyN(tmpFile, upstreamConn, int64(cfg.onDiskBufferUnitSize))
238+
_, err = tmpFile.Seek(0, io.SeekStart)
239+
if err != nil {
240+
fmt.Printf("Error seeking to start of temp file: %v\n", err)
241+
tmpFile.Close()
242+
close(channel)
243+
upstreamConn.CloseRead()
244+
break
245+
}
246+
if n > 0 {
247+
channel <- Buffer{diskBuf: tmpFile}
248+
} else {
249+
tmpFile.Close()
250+
}
251+
}
252+
if readErr != nil {
253+
if ne, ok := readErr.(net.Error); ok && ne.Timeout() {
254+
continue
255+
}
256+
if readErr != io.EOF && readErr != io.ErrUnexpectedEOF {
257+
fmt.Printf("Error reading from upstream: %v\n", readErr)
258+
}
259+
close(channel)
260+
upstreamConn.CloseRead()
261+
break
262+
}
263+
}
264+
}()
265+
}()
266+
go func() {
267+
wg.Wait()
268+
upstreamConn.Close()
269+
conn.Close()
270+
}()
271+
}
272+
273+
func main() {
274+
cfg := parseCLIParams()
275+
276+
if err := os.MkdirAll(cfg.onDiskBufferPath, 0700); err != nil {
277+
fmt.Fprintf(os.Stderr, "Error creating on-disk buffer directory %q: %v\n", cfg.onDiskBufferPath, err)
278+
os.Exit(1)
279+
}
280+
281+
if cfg.maxBuffers > uint64(int(^uint(0)>>1)) {
282+
fmt.Fprintf(os.Stderr, "maxBuffers value %d exceeds maximum allowed\n", cfg.maxBuffers)
283+
os.Exit(1)
284+
}
285+
286+
if cfg.bufferUnitSize > uint64(int(^uint(0)>>1)) {
287+
fmt.Fprintf(os.Stderr, "bufferUnitSize value %d exceeds maximum allowed\n", cfg.bufferUnitSize)
288+
os.Exit(1)
289+
}
290+
291+
if cfg.onDiskBufferUnitSize > ((^uint64(0)) >> 1) {
292+
fmt.Fprintf(os.Stderr, "onDiskBufferUnitSize value %d exceeds maximum allowed\n", cfg.onDiskBufferUnitSize)
293+
os.Exit(1)
294+
}
295+
296+
connListener, err := net.Listen("tcp", cfg.listeningAddress)
297+
if err != nil {
298+
fmt.Fprintf(os.Stderr, "Error starting listener: %v\n", err)
299+
os.Exit(1)
300+
}
301+
defer connListener.Close()
302+
fmt.Printf("Listening on %s and forwarding to %s\n", cfg.listeningAddress, cfg.upstreamAddress)
303+
304+
for {
305+
conn, err := connListener.Accept()
306+
if err != nil {
307+
fmt.Printf("Failed to accept connection: %v\n", err)
308+
continue
309+
}
310+
if tcpConn, ok := conn.(*net.TCPConn); ok {
311+
go handleConnection(tcpConn, cfg)
312+
} else {
313+
fmt.Printf("Failed to cast connection to TCPConn\n")
314+
conn.Close()
315+
}
316+
}
317+
}

‎git-fcgi/cmd.sh‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,18 @@
22

33
THREADS=${THREADS:-"64"}
44
QUEUE_SIZE=${QUEUE_SIZE:-"48"}
5+
MAX_BUFFER_SIZE=${MAX_BUFFER_SIZE:-"1G"}
6+
BUFFER_PATH=${BUFFER_PATH:-"/tmp"}
7+
58
export HOME=/var/www
69
(
710
while true; do
811
/go-queue --queue-size "${QUEUE_SIZE}" --port-number 8888
912
done
1013
)&
11-
exec /usr/bin/spawn-fcgi -u www-data -g www-data -p 5000 -n -- /usr/bin/multiwatch -f ${THREADS} -- /usr/sbin/fcgiwrap
14+
(
15+
while true; do
16+
/buffering-proxy --listen ":5000" --max-buffer-size "${MAX_BUFFER_SIZE}" --on-disk-buffer-path "${BUFFER_PATH}" --upstream "unix:/tmp/fcgi.sock"
17+
done
18+
)&
19+
exec /usr/bin/spawn-fcgi -u www-data -g www-data -s /tmp/fcgi.sock -n -- /usr/bin/multiwatch -f "${THREADS}" -- /usr/sbin/fcgiwrap

0 commit comments

Comments
 (0)