mirror of
https://github.com/soypat/lneto.git
synced 2026-08-07 16:33:40 +00:00
fail on incomplete staging
This commit is contained in:
@@ -41,10 +41,10 @@ func run() error {
|
||||
|
||||
var router httphi.Router
|
||||
err = router.Configure(httphi.RouterConfig{
|
||||
FixedNumGoroutines: numGoroutines,
|
||||
RequestBufferSize: bufferSizes,
|
||||
ResponseMinBufferSize: bufferSizes,
|
||||
MaxAwaitingConns: 256,
|
||||
FixedNumGoroutines: numGoroutines,
|
||||
RequestHeaderBufferSize: bufferSizes,
|
||||
ResponseHeaderMinBufferSize: bufferSizes,
|
||||
MaxAwaitingConns: 256,
|
||||
Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) {
|
||||
return min(time.Second, time.Millisecond*time.Duration(consecutiveBackoffs))
|
||||
},
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"strconv"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/soypat/lneto"
|
||||
"github.com/soypat/lneto/http/httpraw"
|
||||
"github.com/soypat/lneto/internal"
|
||||
)
|
||||
@@ -131,6 +132,7 @@ func (exch *Exchange) StageHeader(key, value string) (enoughMemory bool) {
|
||||
// Field costs key+':'+value+CRLF, plus the CRLF [Exchange.FlushHeader]
|
||||
// appends past the last field to close the header block.
|
||||
if len(key)+len(value)+len(":\r\n")+len("\r\n") > free {
|
||||
exch.respErr = lneto.ErrBufferFull // Omit writing header back to prevent incomplete response.
|
||||
return false
|
||||
}
|
||||
n := copy(exch.rawbuf[off:], key)
|
||||
@@ -158,6 +160,7 @@ func (exch *Exchange) StageHeaderInt(key string, value int64, base int) (enoughM
|
||||
off := int(exch.respHeaderOff) + int(exch.respHeaderLen)
|
||||
free := len(exch.rawbuf) - off
|
||||
if len(key)+internal.IntLen(value, base)+len(":\r\n")+len("\r\n") > free {
|
||||
exch.respErr = lneto.ErrBufferFull // Omit writing header back to prevent incomplete response.
|
||||
return false
|
||||
}
|
||||
n := copy(exch.rawbuf[off:], key)
|
||||
|
||||
@@ -290,7 +290,7 @@ func TestExchangeReadBody(t *testing.T) {
|
||||
// SetHeader must budget every byte it writes: colon, CRLF, and the CRLF that
|
||||
// FlushHeader appends after the last field. Buffers that fit all but the last
|
||||
// byte must be refused, never overrun.
|
||||
func TestExchangeSetHeaderExactFit(t *testing.T) {
|
||||
func TestExchangeStageOKAndFail(t *testing.T) {
|
||||
const key, value = "K", "V"
|
||||
const field = len(key) + len(value) + len(":\r\n")
|
||||
for _, bufLen := range []int{field + 2, field + 1, field} {
|
||||
@@ -301,13 +301,18 @@ func TestExchangeSetHeaderExactFit(t *testing.T) {
|
||||
t.Fatal("fresh exchange failed to acquire connection")
|
||||
}
|
||||
set := exch.StageHeader(key, value)
|
||||
exch.WriteHeader(200)
|
||||
n, err := exch.FlushHeader()
|
||||
|
||||
want := "HTTP/1.1 200 OK\r\n"
|
||||
if set {
|
||||
want += key + ":" + value + "\r\n"
|
||||
want += "\r\n"
|
||||
} else {
|
||||
if err != lneto.ErrBufferFull || n != 0 {
|
||||
t.Fatal("expected buffer full and no data written:", err, n)
|
||||
}
|
||||
want = ""
|
||||
}
|
||||
want += "\r\n"
|
||||
if got := conn.ViewWritten(); got != want {
|
||||
t.Errorf("buffer %d: want %q, got %q", bufLen, want, got)
|
||||
}
|
||||
|
||||
+11
-9
@@ -67,12 +67,14 @@ type RouterConfig struct {
|
||||
// FixedNumGoroutines must be set to either -1 (freely allocate new goroutines) or to the number of goroutines
|
||||
// to spawn on [Router.Configure] being called.
|
||||
FixedNumGoroutines int
|
||||
// RequestBufferSize determines the buffer allocated
|
||||
// for processing requests.
|
||||
RequestBufferSize int
|
||||
// ResponseMinBufferSize determines buffer allocated for processing responses.
|
||||
// RequestHeaderBufferSize determines the buffer allocated
|
||||
// for processing request HTTP headers including request-target (URI), protocol and key/value pairs.
|
||||
RequestHeaderBufferSize int
|
||||
// ResponseHeaderMinBufferSize determines buffer allocated for processing response headers.
|
||||
// Response buffer will reuse unused request memory so this is not a strict limit.
|
||||
ResponseMinBufferSize int
|
||||
// "HTTP/1.1 200 OK\r\n" does not count towards this memory, only actual Headers key/value pairs use this memory.
|
||||
// After memory is fully consumed [Exchange.StageHeader] will not append more headers.
|
||||
ResponseHeaderMinBufferSize int
|
||||
|
||||
// NormalizeOutgoingKeys normalizes response header field keys as they are
|
||||
// staged, i.e: "content-type" becomes "Content-Type".
|
||||
@@ -148,8 +150,8 @@ func (r *Router) Configure(cfg RouterConfig) error {
|
||||
gen := r.gen.Load()
|
||||
numgoro := cfg.FixedNumGoroutines
|
||||
workerMode := cfg.workerMode()
|
||||
r.reqBuf = cfg.RequestBufferSize
|
||||
r.respBuf = cfg.ResponseMinBufferSize
|
||||
r.reqBuf = cfg.RequestHeaderBufferSize
|
||||
r.respBuf = cfg.ResponseHeaderMinBufferSize
|
||||
r.mux = cfg.Mux
|
||||
r.log = cfg.Logger
|
||||
r.normalizeKeys = cfg.NormalizeOutgoingKeys
|
||||
@@ -172,12 +174,12 @@ func (r *Router) Configure(cfg RouterConfig) error {
|
||||
r.freeList = nil // Freelist entries point into the buffers reused below.
|
||||
internal.SliceReuse(&r.exchs, numgoro)
|
||||
r.exchs = r.exchs[:numgoro]
|
||||
rawBuflen := cfg.RequestBufferSize + cfg.ResponseMinBufferSize
|
||||
rawBuflen := cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize
|
||||
internal.SliceReuse(&r.globbuf, numgoro*rawBuflen)
|
||||
for i := range numgoro {
|
||||
// TODO exchange buffer alloc
|
||||
goff := i * rawBuflen
|
||||
r.exchs[i].Configure(r.globbuf[goff:goff+rawBuflen], cfg.RequestBufferSize, cfg.NormalizeOutgoingKeys)
|
||||
r.exchs[i].Configure(r.globbuf[goff:goff+rawBuflen], cfg.RequestHeaderBufferSize, cfg.NormalizeOutgoingKeys)
|
||||
go r.goroWorker(gen, jobqueue, cfg.Backoff, cfg.Mux)
|
||||
}
|
||||
r.pendingConns = jobqueue
|
||||
|
||||
+16
-16
@@ -153,10 +153,10 @@ var _ Mux = (*MuxSlice)(nil)
|
||||
|
||||
func configSynchronousRouter(t *testing.T, router *Router, bufferSize int, mux Mux) {
|
||||
err := router.Configure(RouterConfig{
|
||||
FixedNumGoroutines: -1,
|
||||
Mux: mux,
|
||||
RequestBufferSize: bufferSize,
|
||||
ResponseMinBufferSize: bufferSize,
|
||||
FixedNumGoroutines: -1,
|
||||
Mux: mux,
|
||||
RequestHeaderBufferSize: bufferSize,
|
||||
ResponseHeaderMinBufferSize: bufferSize,
|
||||
Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) {
|
||||
return lneto.BackoffFlagNop
|
||||
},
|
||||
@@ -343,12 +343,12 @@ func TestRouterHandleAfterTeardown(t *testing.T) {
|
||||
)
|
||||
sm.Handle("GET /", staticPage(t, "ok"))
|
||||
err := router.Configure(RouterConfig{
|
||||
FixedNumGoroutines: 2,
|
||||
MaxAwaitingConns: 4,
|
||||
Mux: &sm,
|
||||
RequestBufferSize: 512,
|
||||
ResponseMinBufferSize: 512,
|
||||
Backoff: nopBackoff,
|
||||
FixedNumGoroutines: 2,
|
||||
MaxAwaitingConns: 4,
|
||||
Mux: &sm,
|
||||
RequestHeaderBufferSize: 512,
|
||||
ResponseHeaderMinBufferSize: 512,
|
||||
Backoff: nopBackoff,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -371,12 +371,12 @@ func TestRouterConfigureDuringWorkerHandle(t *testing.T) {
|
||||
)
|
||||
sm.Handle("GET /", staticPage(t, "ok"))
|
||||
cfg := RouterConfig{
|
||||
FixedNumGoroutines: 2,
|
||||
MaxAwaitingConns: 4,
|
||||
Mux: &sm,
|
||||
RequestBufferSize: 512,
|
||||
ResponseMinBufferSize: 512,
|
||||
Backoff: nopBackoff,
|
||||
FixedNumGoroutines: 2,
|
||||
MaxAwaitingConns: 4,
|
||||
Mux: &sm,
|
||||
RequestHeaderBufferSize: 512,
|
||||
ResponseHeaderMinBufferSize: 512,
|
||||
Backoff: nopBackoff,
|
||||
}
|
||||
if err := router.Configure(cfg); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
Reference in New Issue
Block a user