Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 13, 14:15 UTC
fastcgi: Optimize FastCGI transport (#4978)
* break up code and use lazy reading and pool bufio.Writer * close underlying connection when operation failed * allocate bufWriter and streamWriter only once * refactor record writing * rebase from master * handle err * Fix type assertion Also reduce some duplication * Refactor client and clientCloser for logging Should reduce allocations * Minor cosmetic adjustments; apply Apache license * Appease the linter Co-authored-by: Matthew Holt <mholt@users.noreply.github.com>
WeidiDeng committed
Sep 3, 2022 at 06:57 UTC
83b26975bd9330c4cfc44d52b106da739240e72b
8 files changed
+399
-314
modules/caddyhttp/reverseproxy/fastcgi/client.go
+59
-286
@@ -26,9 +26,6 @@ package fastcgi
26
import (
27
"bufio"
28
"bytes"
29
- "context"
30
- "encoding/binary"
31
- "errors"
29
"io"
30
"mime/multipart"
31
"net"
@@ -40,7 +37,6 @@ import (
37
"path/filepath"
38
"strconv"
39
"strings"
43
- "sync"
40
"time"
41
42
"go.uber.org/zap"
@@ -122,285 +118,60 @@ const (
118
maxPad = 255
119
)
120
125
-type header struct {
126
- Version uint8
127
- Type uint8
128
- ID uint16
129
- ContentLength uint16
130
- PaddingLength uint8
131
- Reserved uint8
132
-}
133
-
121
// for padding so we don't have to allocate all the time
122
// not synchronized because we don't care what the contents are
123
var pad [maxPad]byte
124
138
-func (h *header) init(recType uint8, reqID uint16, contentLength int) {
139
- h.Version = 1
140
- h.Type = recType
141
- h.ID = reqID
142
- h.ContentLength = uint16(contentLength)
143
- h.PaddingLength = uint8(-contentLength & 7)
144
-}
145
-
146
-type record struct {
147
- h header
148
- rbuf []byte
149
-}
150
-
151
-func (rec *record) read(r io.Reader) (buf []byte, err error) {
152
- if err = binary.Read(r, binary.BigEndian, &rec.h); err != nil {
153
- return
154
- }
155
- if rec.h.Version != 1 {
156
- err = errors.New("fcgi: invalid header version")
157
- return
158
- }
159
- if rec.h.Type == EndRequest {
160
- err = io.EOF
161
- return
162
- }
163
- n := int(rec.h.ContentLength) + int(rec.h.PaddingLength)
164
- if len(rec.rbuf) < n {
165
- rec.rbuf = make([]byte, n)
166
- }
167
- if _, err = io.ReadFull(r, rec.rbuf[:n]); err != nil {
168
- return
169
- }
170
- buf = rec.rbuf[:int(rec.h.ContentLength)]
171
-
172
- return
173
-}
174
-
175
-// FCGIClient implements a FastCGI client, which is a standard for
125
+// client implements a FastCGI client, which is a standard for
126
// interfacing external applications with Web servers.
177
-type FCGIClient struct {
178
- mutex sync.Mutex
179
- rwc io.ReadWriteCloser
180
- h header
181
- buf bytes.Buffer
182
- stderr bytes.Buffer
183
- keepAlive bool
184
- reqID uint16
185
- logger *zap.Logger
186
-}
187
-
188
-// DialWithDialerContext connects to the fcgi responder at the specified network address, using custom net.Dialer
189
-// and a context.
190
-// See func net.Dial for a description of the network and address parameters.
191
-func DialWithDialerContext(ctx context.Context, network, address string, dialer net.Dialer) (fcgi *FCGIClient, err error) {
192
- var conn net.Conn
193
- conn, err = dialer.DialContext(ctx, network, address)
194
- if err != nil {
195
- return
196
- }
197
-
198
- fcgi = &FCGIClient{
199
- rwc: conn,
200
- keepAlive: false,
201
- reqID: 1,
202
- }
203
-
204
- return
205
-}
206
-
207
-// DialContext is like Dial but passes ctx to dialer.Dial.
208
-func DialContext(ctx context.Context, network, address string) (fcgi *FCGIClient, err error) {
209
- // TODO: why not set timeout here?
210
- return DialWithDialerContext(ctx, network, address, net.Dialer{})
211
-}
212
-
213
-// Dial connects to the fcgi responder at the specified network address, using default net.Dialer.
214
-// See func net.Dial for a description of the network and address parameters.
215
-func Dial(network, address string) (fcgi *FCGIClient, err error) {
216
- return DialContext(context.Background(), network, address)
217
-}
218
-
219
-// Close closes fcgi connection
220
-func (c *FCGIClient) Close() {
221
- c.rwc.Close()
222
-}
223
-
224
-func (c *FCGIClient) writeRecord(recType uint8, content []byte) (err error) {
225
- c.mutex.Lock()
226
- defer c.mutex.Unlock()
227
- c.buf.Reset()
228
- c.h.init(recType, c.reqID, len(content))
229
- if err := binary.Write(&c.buf, binary.BigEndian, c.h); err != nil {
230
- return err
231
- }
232
- if _, err := c.buf.Write(content); err != nil {
233
- return err
234
- }
235
- if _, err := c.buf.Write(pad[:c.h.PaddingLength]); err != nil {
236
- return err
237
- }
238
- _, err = c.rwc.Write(c.buf.Bytes())
239
- return err
240
-}
241
-
242
-func (c *FCGIClient) writeBeginRequest(role uint16, flags uint8) error {
243
- b := [8]byte{byte(role >> 8), byte(role), flags}
244
- return c.writeRecord(BeginRequest, b[:])
245
-}
246
-
247
-func (c *FCGIClient) writePairs(recType uint8, pairs map[string]string) error {
248
- w := newWriter(c, recType)
249
- b := make([]byte, 8)
250
- nn := 0
251
- for k, v := range pairs {
252
- m := 8 + len(k) + len(v)
253
- if m > maxWrite {
254
- // param data size exceed 65535 bytes"
255
- vl := maxWrite - 8 - len(k)
256
- v = v[:vl]
257
- }
258
- n := encodeSize(b, uint32(len(k)))
259
- n += encodeSize(b[n:], uint32(len(v)))
260
- m = n + len(k) + len(v)
261
- if (nn + m) > maxWrite {
262
- w.Flush()
263
- nn = 0
264
- }
265
- nn += m
266
- if _, err := w.Write(b[:n]); err != nil {
267
- return err
268
- }
269
- if _, err := w.WriteString(k); err != nil {
270
- return err
271
- }
272
- if _, err := w.WriteString(v); err != nil {
273
- return err
274
- }
275
- }
276
- w.Close()
277
- return nil
278
-}
279
-
280
-func encodeSize(b []byte, size uint32) int {
281
- if size > 127 {
282
- size |= 1 << 31
283
- binary.BigEndian.PutUint32(b, size)
284
- return 4
285
- }
286
- b[0] = byte(size)
287
- return 1
288
-}
289
-
290
-// bufWriter encapsulates bufio.Writer but also closes the underlying stream when
291
-// Closed.
292
-type bufWriter struct {
293
- closer io.Closer
294
- *bufio.Writer
295
-}
296
-
297
-func (w *bufWriter) Close() error {
298
- if err := w.Writer.Flush(); err != nil {
299
- w.closer.Close()
300
- return err
301
- }
302
- return w.closer.Close()
303
-}
304
-
305
-func newWriter(c *FCGIClient, recType uint8) *bufWriter {
306
- s := &streamWriter{c: c, recType: recType}
307
- w := bufio.NewWriterSize(s, maxWrite)
308
- return &bufWriter{s, w}
309
-}
310
-
311
-// streamWriter abstracts out the separation of a stream into discrete records.
312
-// It only writes maxWrite bytes at a time.
313
-type streamWriter struct {
314
- c *FCGIClient
315
- recType uint8
316
-}
317
-
318
-func (w *streamWriter) Write(p []byte) (int, error) {
319
- nn := 0
320
- for len(p) > 0 {
321
- n := len(p)
322
- if n > maxWrite {
323
- n = maxWrite
324
- }
325
- if err := w.c.writeRecord(w.recType, p[:n]); err != nil {
326
- return nn, err
327
- }
328
- nn += n
329
- p = p[n:]
330
- }
331
- return nn, nil
332
-}
333
-
334
-func (w *streamWriter) Close() error {
335
- // send empty record to close the stream
336
- return w.c.writeRecord(w.recType, nil)
337
-}
338
-
339
-type streamReader struct {
340
- c *FCGIClient
341
- buf []byte
342
-}
343
-
344
-func (w *streamReader) Read(p []byte) (n int, err error) {
345
- if len(p) > 0 {
346
- if len(w.buf) == 0 {
347
-
348
- // filter outputs for error log
349
- for {
350
- rec := &record{}
351
- var buf []byte
352
- buf, err = rec.read(w.c.rwc)
353
- if err != nil {
354
- return
355
- }
356
- // standard error output
357
- if rec.h.Type == Stderr {
358
- w.c.stderr.Write(buf)
359
- continue
360
- }
361
- w.buf = buf
362
- break
363
- }
364
- }
365
-
366
- n = len(p)
367
- if n > len(w.buf) {
368
- n = len(w.buf)
369
- }
370
- copy(p, w.buf[:n])
371
- w.buf = w.buf[n:]
372
- }
373
-
374
- return
127
+type client struct {
128
+ rwc net.Conn
129
+ // keepAlive bool // TODO: implement
130
+ reqID uint16
131
+ stderr bool
132
+ logger *zap.Logger
133
}
134
135
// Do made the request and returns a io.Reader that translates the data read
136
// from fcgi responder out of fcgi packet before returning it.
379
-func (c *FCGIClient) Do(p map[string]string, req io.Reader) (r io.Reader, err error) {
380
- err = c.writeBeginRequest(uint16(Responder), 0)
137
+func (c *client) Do(p map[string]string, req io.Reader) (r io.Reader, err error) {
138
+ writer := &streamWriter{c: c}
139
+ writer.buf = bufPool.Get().(*bytes.Buffer)
140
+ writer.buf.Reset()
141
+ defer bufPool.Put(writer.buf)
142
+
143
+ err = writer.writeBeginRequest(uint16(Responder), 0)
144
if err != nil {
145
return
146
}
147
385
- err = c.writePairs(Params, p)
148
+ writer.recType = Params
149
+ err = writer.writePairs(p)
150
if err != nil {
151
return
152
}
153
390
- body := newWriter(c, Stdin)
154
+ writer.recType = Stdin
155
if req != nil {
392
- _, _ = io.Copy(body, req)
156
+ _, err = io.Copy(writer, req)
157
+ if err != nil {
158
+ return nil, err
159
+ }
160
+ }
161
+ err = writer.FlushStream()
162
+ if err != nil {
163
+ return nil, err
164
}
394
- body.Close()
165
166
r = &streamReader{c: c}
167
return
168
}
169
170
// clientCloser is a io.ReadCloser. It wraps a io.Reader with a Closer
401
-// that closes FCGIClient connection.
171
+// that closes the client connection.
172
type clientCloser struct {
403
- *FCGIClient
173
+ rwc net.Conn
174
+ r *streamReader
175
io.Reader
176
177
status int
@@ -408,9 +179,9 @@ type clientCloser struct {
179
}
180
181
func (f clientCloser) Close() error {
411
- stderr := f.FCGIClient.stderr.Bytes()
182
+ stderr := f.r.stderr.Bytes()
183
if len(stderr) == 0 {
413
- return f.FCGIClient.rwc.Close()
184
+ return f.rwc.Close()
185
}
186
187
if f.status >= 400 {
@@ -418,12 +189,13 @@ func (f clientCloser) Close() error {
189
} else {
190
f.logger.Warn("stderr", zap.ByteString("body", stderr))
191
}
421
- return f.FCGIClient.rwc.Close()
192
+
193
+ return f.rwc.Close()
194
}
195
196
// Request returns a HTTP Response with Header and Body
197
// from fcgi responder
426
-func (c *FCGIClient) Request(p map[string]string, req io.Reader) (resp *http.Response, err error) {
198
+func (c *client) Request(p map[string]string, req io.Reader) (resp *http.Response, err error) {
199
r, err := c.Do(p, req)
200
if err != nil {
201
return
@@ -458,26 +230,27 @@ func (c *FCGIClient) Request(p map[string]string, req io.Reader) (resp *http.Res
230
resp.TransferEncoding = resp.Header["Transfer-Encoding"]
231
resp.ContentLength, _ = strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64)
232
233
+ // wrap the response body in our closer
234
+ closer := clientCloser{
235
+ rwc: c.rwc,
236
+ r: r.(*streamReader),
237
+ Reader: rb,
238
+ status: resp.StatusCode,
239
+ logger: noopLogger,
240
+ }
241
if chunked(resp.TransferEncoding) {
462
- resp.Body = clientCloser{
463
- FCGIClient: c,
464
- Reader: httputil.NewChunkedReader(rb),
465
- status: resp.StatusCode,
466
- logger: c.logger,
467
- }
468
- } else {
469
- resp.Body = clientCloser{
470
- FCGIClient: c,
471
- Reader: rb,
472
- status: resp.StatusCode,
473
- logger: c.logger,
474
- }
242
+ closer.Reader = httputil.NewChunkedReader(rb)
243
}
244
+ if c.stderr {
245
+ closer.logger = c.logger
246
+ }
247
+ resp.Body = closer
248
+
249
return
250
}
251
252
// Get issues a GET request to the fcgi responder.
480
-func (c *FCGIClient) Get(p map[string]string, body io.Reader, l int64) (resp *http.Response, err error) {
253
+func (c *client) Get(p map[string]string, body io.Reader, l int64) (resp *http.Response, err error) {
254
255
p["REQUEST_METHOD"] = "GET"
256
p["CONTENT_LENGTH"] = strconv.FormatInt(l, 10)
@@ -486,7 +259,7 @@ func (c *FCGIClient) Get(p map[string]string, body io.Reader, l int64) (resp *ht
259
}
260
261
// Head issues a HEAD request to the fcgi responder.
489
-func (c *FCGIClient) Head(p map[string]string) (resp *http.Response, err error) {
262
+func (c *client) Head(p map[string]string) (resp *http.Response, err error) {
263
264
p["REQUEST_METHOD"] = "HEAD"
265
p["CONTENT_LENGTH"] = "0"
@@ -495,7 +268,7 @@ func (c *FCGIClient) Head(p map[string]string) (resp *http.Response, err error)
268
}
269
270
// Options issues an OPTIONS request to the fcgi responder.
498
-func (c *FCGIClient) Options(p map[string]string) (resp *http.Response, err error) {
271
+func (c *client) Options(p map[string]string) (resp *http.Response, err error) {
272
273
p["REQUEST_METHOD"] = "OPTIONS"
274
p["CONTENT_LENGTH"] = "0"
@@ -505,7 +278,7 @@ func (c *FCGIClient) Options(p map[string]string) (resp *http.Response, err erro
278
279
// Post issues a POST request to the fcgi responder. with request body
280
// in the format that bodyType specified
508
-func (c *FCGIClient) Post(p map[string]string, method string, bodyType string, body io.Reader, l int64) (resp *http.Response, err error) {
281
+func (c *client) Post(p map[string]string, method string, bodyType string, body io.Reader, l int64) (resp *http.Response, err error) {
282
if p == nil {
283
p = make(map[string]string)
284
}
@@ -528,7 +301,7 @@ func (c *FCGIClient) Post(p map[string]string, method string, bodyType string, b
301
302
// PostForm issues a POST to the fcgi responder, with form
303
// as a string key to a list values (url.Values)
531
-func (c *FCGIClient) PostForm(p map[string]string, data url.Values) (resp *http.Response, err error) {
304
+func (c *client) PostForm(p map[string]string, data url.Values) (resp *http.Response, err error) {
305
body := bytes.NewReader([]byte(data.Encode()))
306
return c.Post(p, "POST", "application/x-www-form-urlencoded", body, int64(body.Len()))
307
}
@@ -536,7 +309,7 @@ func (c *FCGIClient) PostForm(p map[string]string, data url.Values) (resp *http.
309
// PostFile issues a POST to the fcgi responder in multipart(RFC 2046) standard,
310
// with form as a string key to a list values (url.Values),
311
// and/or with file as a string key to a list file path.
539
-func (c *FCGIClient) PostFile(p map[string]string, data url.Values, file map[string]string) (resp *http.Response, err error) {
312
+func (c *client) PostFile(p map[string]string, data url.Values, file map[string]string) (resp *http.Response, err error) {
313
buf := &bytes.Buffer{}
314
writer := multipart.NewWriter(buf)
315
bodyType := writer.FormDataContentType()
@@ -577,18 +350,18 @@ func (c *FCGIClient) PostFile(p map[string]string, data url.Values, file map[str
350
351
// SetReadTimeout sets the read timeout for future calls that read from the
352
// fcgi responder. A zero value for t means no timeout will be set.
580
-func (c *FCGIClient) SetReadTimeout(t time.Duration) error {
581
- if conn, ok := c.rwc.(net.Conn); ok && t != 0 {
582
- return conn.SetReadDeadline(time.Now().Add(t))
353
+func (c *client) SetReadTimeout(t time.Duration) error {
354
+ if t != 0 {
355
+ return c.rwc.SetReadDeadline(time.Now().Add(t))
356
}
357
return nil
358
}
359
360
// SetWriteTimeout sets the write timeout for future calls that send data to
361
// the fcgi responder. A zero value for t means no timeout will be set.
589
-func (c *FCGIClient) SetWriteTimeout(t time.Duration) error {
590
- if conn, ok := c.rwc.(net.Conn); ok && t != 0 {
591
- return conn.SetWriteDeadline(time.Now().Add(t))
362
+func (c *client) SetWriteTimeout(t time.Duration) error {
363
+ if t != 0 {
364
+ return c.rwc.SetWriteDeadline(time.Now().Add(t))
365
}
366
return nil
367
}
modules/caddyhttp/reverseproxy/fastcgi/client_test.go
+4
-2
@@ -118,12 +118,14 @@ func (s FastCGIServer) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
118
}
119
120
func sendFcgi(reqType int, fcgiParams map[string]string, data []byte, posts map[string]string, files map[string]string) (content []byte) {
121
- fcgi, err := Dial("tcp", ipPort)
121
+ conn, err := net.Dial("tcp", ipPort)
122
if err != nil {
123
log.Println("err:", err)
124
return
125
}
126
127
+ fcgi := client{rwc: conn, reqID: 1}
128
+
129
length := 0
130
131
var resp *http.Response
@@ -168,7 +170,7 @@ func sendFcgi(reqType int, fcgiParams map[string]string, data []byte, posts map[
170
content, _ = io.ReadAll(resp.Body)
171
172
log.Println("c: send data length ≈", length, string(content))
171
- fcgi.Close()
173
+ conn.Close()
174
time.Sleep(1 * time.Second)
175
176
if bytes.Contains(content, []byte("FAILED")) {
modules/caddyhttp/reverseproxy/fastcgi/fastcgi.go
+31
-26
@@ -15,7 +15,6 @@
15
package fastcgi
16
17
import (
18
- "context"
18
"crypto/tls"
19
"fmt"
20
"net"
@@ -129,13 +128,7 @@ func (t Transport) RoundTrip(r *http.Request) (*http.Response, error) {
128
return nil, fmt.Errorf("building environment: %v", err)
129
}
130
132
- // TODO: doesn't dialer have a Timeout field?
131
ctx := r.Context()
134
- if t.DialTimeout > 0 {
135
- var cancel context.CancelFunc
136
- ctx, cancel = context.WithTimeout(ctx, time.Duration(t.DialTimeout))
137
- defer cancel()
138
- }
132
133
// extract dial information from request (should have been embedded by the reverse proxy)
134
network, address := "tcp", r.URL.Host
@@ -150,32 +143,41 @@ func (t Transport) RoundTrip(r *http.Request) (*http.Response, error) {
143
ShouldLogCredentials: logCreds,
144
}
145
loggableEnv := loggableEnv{vars: env, logCredentials: logCreds}
153
- t.logger.Debug("roundtrip",
146
+
147
+ logger := t.logger.With(
148
zap.Object("request", loggableReq),
155
- zap.String("dial", address),
149
zap.Object("env", loggableEnv),
150
)
151
+ logger.Debug("roundtrip",
152
+ zap.String("dial", address),
153
+ zap.Object("env", loggableEnv),
154
+ zap.Object("request", loggableReq))
155
159
- fcgiBackend, err := DialContext(ctx, network, address)
156
+ // connect to the backend
157
+ dialer := net.Dialer{Timeout: time.Duration(t.DialTimeout)}
158
+ conn, err := dialer.DialContext(ctx, network, address)
159
if err != nil {
161
- // TODO: wrap in a special error type if the dial failed, so retries can happen if enabled
160
return nil, fmt.Errorf("dialing backend: %v", err)
161
}
164
- if t.CaptureStderr {
165
- fcgiBackend.logger = t.logger.With(
166
- zap.Object("request", loggableReq),
167
- zap.Object("env", loggableEnv),
168
- )
169
- } else {
170
- fcgiBackend.logger = noopLogger
162
+ defer func() {
163
+ // conn will be closed with the response body unless there's an error
164
+ if err != nil {
165
+ conn.Close()
166
+ }
167
+ }()
168
+
169
+ // create the client that will facilitate the protocol
170
+ client := client{
171
+ rwc: conn,
172
+ reqID: 1,
173
+ logger: logger,
174
}
172
- // fcgiBackend gets closed when response body is closed (see clientCloser)
175
176
// read/write timeouts
175
- if err := fcgiBackend.SetReadTimeout(time.Duration(t.ReadTimeout)); err != nil {
177
+ if err = client.SetReadTimeout(time.Duration(t.ReadTimeout)); err != nil {
178
return nil, fmt.Errorf("setting read timeout: %v", err)
179
}
178
- if err := fcgiBackend.SetWriteTimeout(time.Duration(t.WriteTimeout)); err != nil {
180
+ if err = client.SetWriteTimeout(time.Duration(t.WriteTimeout)); err != nil {
181
return nil, fmt.Errorf("setting write timeout: %v", err)
182
}
183
@@ -187,16 +189,19 @@ func (t Transport) RoundTrip(r *http.Request) (*http.Response, error) {
189
var resp *http.Response
190
switch r.Method {
191
case http.MethodHead:
190
- resp, err = fcgiBackend.Head(env)
192
+ resp, err = client.Head(env)
193
case http.MethodGet:
192
- resp, err = fcgiBackend.Get(env, r.Body, contentLength)
194
+ resp, err = client.Get(env, r.Body, contentLength)
195
case http.MethodOptions:
194
- resp, err = fcgiBackend.Options(env)
196
+ resp, err = client.Options(env)
197
default:
196
- resp, err = fcgiBackend.Post(env, r.Method, r.Header.Get("Content-Type"), r.Body, contentLength)
198
+ resp, err = client.Post(env, r.Method, r.Header.Get("Content-Type"), r.Body, contentLength)
199
+ }
200
+ if err != nil {
201
+ return nil, err
202
}
203
199
- return resp, err
204
+ return resp, nil
205
}
206
207
// buildEnv returns a set of CGI environment variables for the request.
modules/caddyhttp/reverseproxy/fastcgi/header.go
new
+32
@@ -0,0 +1,32 @@
1
+// Copyright 2015 Matthew Holt and The Caddy Authors
2
+//
3
+// Licensed under the Apache License, Version 2.0 (the "License");
4
+// you may not use this file except in compliance with the License.
5
+// You may obtain a copy of the License at
6
+//
7
+// http://www.apache.org/licenses/LICENSE-2.0
8
+//
9
+// Unless required by applicable law or agreed to in writing, software
10
+// distributed under the License is distributed on an "AS IS" BASIS,
11
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+// See the License for the specific language governing permissions and
13
+// limitations under the License.
14
+
15
+package fastcgi
16
+
17
+type header struct {
18
+ Version uint8
19
+ Type uint8
20
+ ID uint16
21
+ ContentLength uint16
22
+ PaddingLength uint8
23
+ Reserved uint8
24
+}
25
+
26
+func (h *header) init(recType uint8, reqID uint16, contentLength int) {
27
+ h.Version = 1
28
+ h.Type = recType
29
+ h.ID = reqID
30
+ h.ContentLength = uint16(contentLength)
31
+ h.PaddingLength = uint8(-contentLength & 7)
32
+}
modules/caddyhttp/reverseproxy/fastcgi/pool.go
new
+26
@@ -0,0 +1,26 @@
1
+// Copyright 2015 Matthew Holt and The Caddy Authors
2
+//
3
+// Licensed under the Apache License, Version 2.0 (the "License");
4
+// you may not use this file except in compliance with the License.
5
+// You may obtain a copy of the License at
6
+//
7
+// http://www.apache.org/licenses/LICENSE-2.0
8
+//
9
+// Unless required by applicable law or agreed to in writing, software
10
+// distributed under the License is distributed on an "AS IS" BASIS,
11
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+// See the License for the specific language governing permissions and
13
+// limitations under the License.
14
+
15
+package fastcgi
16
+
17
+import (
18
+ "bytes"
19
+ "sync"
20
+)
21
+
22
+var bufPool = sync.Pool{
23
+ New: func() any {
24
+ return new(bytes.Buffer)
25
+ },
26
+}
modules/caddyhttp/reverseproxy/fastcgi/reader.go
new
+44
@@ -0,0 +1,44 @@
1
+// Copyright 2015 Matthew Holt and The Caddy Authors
2
+//
3
+// Licensed under the Apache License, Version 2.0 (the "License");
4
+// you may not use this file except in compliance with the License.
5
+// You may obtain a copy of the License at
6
+//
7
+// http://www.apache.org/licenses/LICENSE-2.0
8
+//
9
+// Unless required by applicable law or agreed to in writing, software
10
+// distributed under the License is distributed on an "AS IS" BASIS,
11
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+// See the License for the specific language governing permissions and
13
+// limitations under the License.
14
+
15
+package fastcgi
16
+
17
+import (
18
+ "bytes"
19
+ "io"
20
+)
21
+
22
+type streamReader struct {
23
+ c *client
24
+ rec record
25
+ stderr bytes.Buffer
26
+}
27
+
28
+func (w *streamReader) Read(p []byte) (n int, err error) {
29
+ for !w.rec.hasMore() {
30
+ err = w.rec.fill(w.c.rwc)
31
+ if err != nil {
32
+ return 0, err
33
+ }
34
+
35
+ // standard error output
36
+ if w.rec.h.Type == Stderr {
37
+ if _, err = io.Copy(&w.stderr, &w.rec); err != nil {
38
+ return 0, err
39
+ }
40
+ }
41
+ }
42
+
43
+ return w.rec.Read(p)
44
+}
modules/caddyhttp/reverseproxy/fastcgi/record.go
new
+58
@@ -0,0 +1,58 @@
1
+// Copyright 2015 Matthew Holt and The Caddy Authors
2
+//
3
+// Licensed under the Apache License, Version 2.0 (the "License");
4
+// you may not use this file except in compliance with the License.
5
+// You may obtain a copy of the License at
6
+//
7
+// http://www.apache.org/licenses/LICENSE-2.0
8
+//
9
+// Unless required by applicable law or agreed to in writing, software
10
+// distributed under the License is distributed on an "AS IS" BASIS,
11
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+// See the License for the specific language governing permissions and
13
+// limitations under the License.
14
+
15
+package fastcgi
16
+
17
+import (
18
+ "encoding/binary"
19
+ "errors"
20
+ "io"
21
+)
22
+
23
+type record struct {
24
+ h header
25
+ lr io.LimitedReader
26
+ padding int64
27
+}
28
+
29
+func (rec *record) fill(r io.Reader) (err error) {
30
+ rec.lr.N = rec.padding
31
+ rec.lr.R = r
32
+ if _, err = io.Copy(io.Discard, rec); err != nil {
33
+ return
34
+ }
35
+
36
+ if err = binary.Read(r, binary.BigEndian, &rec.h); err != nil {
37
+ return
38
+ }
39
+ if rec.h.Version != 1 {
40
+ err = errors.New("fcgi: invalid header version")
41
+ return
42
+ }
43
+ if rec.h.Type == EndRequest {
44
+ err = io.EOF
45
+ return
46
+ }
47
+ rec.lr.N = int64(rec.h.ContentLength)
48
+ rec.padding = int64(rec.h.PaddingLength)
49
+ return
50
+}
51
+
52
+func (rec *record) Read(p []byte) (n int, err error) {
53
+ return rec.lr.Read(p)
54
+}
55
+
56
+func (rec *record) hasMore() bool {
57
+ return rec.lr.N > 0
58
+}
modules/caddyhttp/reverseproxy/fastcgi/writer.go
new
+145
@@ -0,0 +1,145 @@
1
+// Copyright 2015 Matthew Holt and The Caddy Authors
2
+//
3
+// Licensed under the Apache License, Version 2.0 (the "License");
4
+// you may not use this file except in compliance with the License.
5
+// You may obtain a copy of the License at
6
+//
7
+// http://www.apache.org/licenses/LICENSE-2.0
8
+//
9
+// Unless required by applicable law or agreed to in writing, software
10
+// distributed under the License is distributed on an "AS IS" BASIS,
11
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
+// See the License for the specific language governing permissions and
13
+// limitations under the License.
14
+
15
+package fastcgi
16
+
17
+import (
18
+ "bytes"
19
+ "encoding/binary"
20
+)
21
+
22
+// streamWriter abstracts out the separation of a stream into discrete records.
23
+// It only writes maxWrite bytes at a time.
24
+type streamWriter struct {
25
+ c *client
26
+ h header
27
+ buf *bytes.Buffer
28
+ recType uint8
29
+}
30
+
31
+func (w *streamWriter) writeRecord(recType uint8, content []byte) (err error) {
32
+ w.h.init(recType, w.c.reqID, len(content))
33
+ w.buf.Write(pad[:8])
34
+ w.writeHeader()
35
+ w.buf.Write(content)
36
+ w.buf.Write(pad[:w.h.PaddingLength])
37
+ _, err = w.buf.WriteTo(w.c.rwc)
38
+ return err
39
+}
40
+
41
+func (w *streamWriter) writeBeginRequest(role uint16, flags uint8) error {
42
+ b := [8]byte{byte(role >> 8), byte(role), flags}
43
+ return w.writeRecord(BeginRequest, b[:])
44
+}
45
+
46
+func (w *streamWriter) Write(p []byte) (int, error) {
47
+ // init header
48
+ if w.buf.Len() < 8 {
49
+ w.buf.Write(pad[:8])
50
+ }
51
+
52
+ nn := 0
53
+ for len(p) > 0 {
54
+ n := len(p)
55
+ nl := maxWrite + 8 - w.buf.Len()
56
+ if n > nl {
57
+ n = nl
58
+ w.buf.Write(p[:n])
59
+ if err := w.Flush(); err != nil {
60
+ return nn, err
61
+ }
62
+ // reset headers
63
+ w.buf.Write(pad[:8])
64
+ } else {
65
+ w.buf.Write(p[:n])
66
+ }
67
+ nn += n
68
+ p = p[n:]
69
+ }
70
+ return nn, nil
71
+}
72
+
73
+func (w *streamWriter) endStream() error {
74
+ // send empty record to close the stream
75
+ return w.writeRecord(w.recType, nil)
76
+}
77
+
78
+func (w *streamWriter) writePairs(pairs map[string]string) error {
79
+ b := make([]byte, 8)
80
+ nn := 0
81
+ // init headers
82
+ w.buf.Write(b)
83
+ for k, v := range pairs {
84
+ m := 8 + len(k) + len(v)
85
+ if m > maxWrite {
86
+ // param data size exceed 65535 bytes"
87
+ vl := maxWrite - 8 - len(k)
88
+ v = v[:vl]
89
+ }
90
+ n := encodeSize(b, uint32(len(k)))
91
+ n += encodeSize(b[n:], uint32(len(v)))
92
+ m = n + len(k) + len(v)
93
+ if (nn + m) > maxWrite {
94
+ if err := w.Flush(); err != nil {
95
+ return err
96
+ }
97
+ // reset headers
98
+ w.buf.Write(b)
99
+ nn = 0
100
+ }
101
+ nn += m
102
+ w.buf.Write(b[:n])
103
+ w.buf.WriteString(k)
104
+ w.buf.WriteString(v)
105
+ }
106
+ return w.FlushStream()
107
+}
108
+
109
+func encodeSize(b []byte, size uint32) int {
110
+ if size > 127 {
111
+ size |= 1 << 31
112
+ binary.BigEndian.PutUint32(b, size)
113
+ return 4
114
+ }
115
+ b[0] = byte(size)
116
+ return 1
117
+}
118
+
119
+// writeHeader populate header wire data in buf, it abuses buffer.Bytes() modification
120
+func (w *streamWriter) writeHeader() {
121
+ h := w.buf.Bytes()[:8]
122
+ h[0] = w.h.Version
123
+ h[1] = w.h.Type
124
+ binary.BigEndian.PutUint16(h[2:4], w.h.ID)
125
+ binary.BigEndian.PutUint16(h[4:6], w.h.ContentLength)
126
+ h[6] = w.h.PaddingLength
127
+ h[7] = w.h.Reserved
128
+}
129
+
130
+// Flush write buffer data to the underlying connection, it assumes header data is the first 8 bytes of buf
131
+func (w *streamWriter) Flush() error {
132
+ w.h.init(w.recType, w.c.reqID, w.buf.Len()-8)
133
+ w.writeHeader()
134
+ w.buf.Write(pad[:w.h.PaddingLength])
135
+ _, err := w.buf.WriteTo(w.c.rwc)
136
+ return err
137
+}
138
+
139
+// FlushStream flush data then end current stream
140
+func (w *streamWriter) FlushStream() error {
141
+ if err := w.Flush(); err != nil {
142
+ return err
143
+ }
144
+ return w.endStream()
145
+}