Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS
Mirrored from https://github.com/Slazee/caddy.git
Synced Nov 11, 01:00 UTC
| 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 network |
| 16 | |
| 17 | import ( |
| 18 | "context" |
| 19 | "errors" |
| 20 | "strconv" |
| 21 | "sync" |
| 22 | "sync/atomic" |
| 23 | "time" |
| 24 | |
| 25 | "go.uber.org/zap" |
| 26 | "go.uber.org/zap/zapcore" |
| 27 | "golang.org/x/sync/singleflight" |
| 28 | ) |
| 29 | |
| 30 | const ( |
| 31 | // maxCacheEntries is the maximum number of entries kept in each cache. |
| 32 | maxCacheEntries = 100 |
| 33 | |
| 34 | // lookupTimeout bounds a shared lookup, which cannot use any one caller's |
| 35 | // deadline. |
| 36 | lookupTimeout = 30 * time.Second |
| 37 | |
| 38 | // maxLookups is the maximum number of distinct lookups allowed to run at |
| 39 | // once, across both caches. Callers that join a lookup already in flight |
| 40 | // do not count against it. |
| 41 | maxLookups = 64 |
| 42 | |
| 43 | // maxPendingLookups is the maximum number of lookups allowed to wait for a |
| 44 | // slot. Beyond that they are shed rather than parked, so a stream of |
| 45 | // distinct keys cannot pile up goroutines for up to lookupTimeout each. |
| 46 | maxPendingLookups = 256 |
| 47 | ) |
| 48 | |
| 49 | // errTooManyLookups is returned when the lookup gate is saturated. |
| 50 | var errTooManyLookups = errors.New("too many dynamic upstream lookups in flight") |
| 51 | |
| 52 | // lookups is shared by every cache, so total resolver work stays bounded no |
| 53 | // matter how many distinct keys come in. |
| 54 | var lookups = newLookupGate(maxLookups, maxPendingLookups) |
| 55 | |
| 56 | // cache holds the results of one kind of lookup, and deduplicates concurrent |
| 57 | // lookups for the same key. |
| 58 | type cache struct { |
| 59 | kind string // "SRV" or "A", for log messages |
| 60 | |
| 61 | mu sync.RWMutex |
| 62 | entries map[string]cacheEntry |
| 63 | epoch uint64 // bumped by every reset; guarded by mu |
| 64 | |
| 65 | group singleflight.Group |
| 66 | } |
| 67 | |
| 68 | type cacheEntry struct { |
| 69 | refresh time.Duration |
| 70 | freshness time.Time |
| 71 | targets []Target |
| 72 | |
| 73 | // gen counts targeted resets of this key. A lookup writes its result back |
| 74 | // only if gen (and the cache's epoch) are unchanged since it started, and |
| 75 | // both are part of the singleflight key, so a caller that arrives after a |
| 76 | // reset starts its own lookup instead of joining one from before it. It |
| 77 | // lives on the entry, rather than in a separate map, so it is bounded by |
| 78 | // maxCacheEntries like everything else. |
| 79 | gen uint64 |
| 80 | } |
| 81 | |
| 82 | func newCache(kind string) *cache { |
| 83 | return &cache{kind: kind, entries: make(map[string]cacheEntry)} |
| 84 | } |
| 85 | |
| 86 | func (e cacheEntry) isFresh() bool { |
| 87 | return !e.freshness.IsZero() && time.Since(e.freshness) < e.refresh |
| 88 | } |
| 89 | |
| 90 | // get returns the cached targets for key if they are fresh; otherwise it runs |
| 91 | // lookup, sharing it with concurrent callers for the same key. See [SRV] for |
| 92 | // the meaning of grace. |
| 93 | func (c *cache) get(ctx context.Context, key string, refresh, grace time.Duration, logger *zap.Logger, lookup func(context.Context) ([]Target, error)) ([]Target, error) { |
| 94 | c.mu.RLock() |
| 95 | cached := c.entries[key] |
| 96 | epoch := c.epoch |
| 97 | c.mu.RUnlock() |
| 98 | if cached.isFresh() { |
| 99 | return cached.targets, nil |
| 100 | } |
| 101 | gen := cached.gen |
| 102 | |
| 103 | flightKey := strconv.FormatUint(epoch, 10) + "\x00" + strconv.FormatUint(gen, 10) + "\x00" + |
| 104 | strconv.FormatInt(int64(grace), 10) + "\x00" + key |
| 105 | ch := c.group.DoChan(flightKey, func() (any, error) { |
| 106 | return c.refresh(ctx, key, epoch, gen, refresh, grace, logger, lookup) |
| 107 | }) |
| 108 | |
| 109 | select { |
| 110 | case res := <-ch: |
| 111 | if res.Err != nil { |
| 112 | return nil, res.Err |
| 113 | } |
| 114 | return res.Val.([]Target), nil |
| 115 | case <-ctx.Done(): |
| 116 | return nil, ctx.Err() |
| 117 | } |
| 118 | } |
| 119 | |
| 120 | // refresh runs a lookup for key on behalf of every caller sharing the flight. |
| 121 | // epoch and gen are what the caller that started it observed. |
| 122 | func (c *cache) refresh(ctx context.Context, key string, epoch, gen uint64, refresh, grace time.Duration, logger *zap.Logger, lookup func(context.Context) ([]Target, error)) ([]Target, error) { |
| 123 | // a previous flight may have refreshed the entry since the caller looked |
| 124 | c.mu.RLock() |
| 125 | cached := c.entries[key] |
| 126 | c.mu.RUnlock() |
| 127 | if cached.isFresh() { |
| 128 | return cached.targets, nil |
| 129 | } |
| 130 | |
| 131 | // detached so the caller that started the flight can be cancelled without |
| 132 | // failing it for the others, but bounded so a stuck resolver cannot |
| 133 | // outlive every waiter |
| 134 | ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), lookupTimeout) |
| 135 | defer cancel() |
| 136 | |
| 137 | if err := lookups.acquire(ctx); err != nil { |
| 138 | // slightly stale beats failing, but only if there is something to |
| 139 | // serve; caching an empty result under load would be worse |
| 140 | if grace > 0 && len(cached.targets) > 0 { |
| 141 | if ce := logger.Check(zapcore.WarnLevel, c.kind+" lookup not started; using previously cached"); ce != nil { |
| 142 | ce.Write(zap.Error(err)) |
| 143 | } |
| 144 | return c.keepStale(key, epoch, gen, cached, refresh, grace), nil |
| 145 | } |
| 146 | return nil, err |
| 147 | } |
| 148 | targets, err := lookup(ctx) |
| 149 | lookups.release() |
| 150 | if err != nil { |
| 151 | if grace > 0 { |
| 152 | if ce := logger.Check(zapcore.ErrorLevel, c.kind+" lookup failed; using previously cached"); ce != nil { |
| 153 | ce.Write(zap.Error(err)) |
| 154 | } |
| 155 | return c.keepStale(key, epoch, gen, cached, refresh, grace), nil |
| 156 | } |
| 157 | return nil, err |
| 158 | } |
| 159 | |
| 160 | c.store(key, epoch, gen, cacheEntry{refresh: refresh, freshness: time.Now(), targets: targets}) |
| 161 | return targets, nil |
| 162 | } |
| 163 | |
| 164 | // keepStale extends cached so it is served for up to grace past its refresh, |
| 165 | // and returns its targets. |
| 166 | func (c *cache) keepStale(key string, epoch, gen uint64, cached cacheEntry, refresh, grace time.Duration) []Target { |
| 167 | cached.freshness = time.Now().Add(grace - refresh) |
| 168 | c.store(key, epoch, gen, cached) |
| 169 | return cached.targets |
| 170 | } |
| 171 | |
| 172 | // store writes e for key, unless key or the whole cache was reset since a |
| 173 | // lookup observed epoch and gen. |
| 174 | func (c *cache) store(key string, epoch, gen uint64, e cacheEntry) { |
| 175 | c.mu.Lock() |
| 176 | defer c.mu.Unlock() |
| 177 | if c.epoch != epoch || c.entries[key].gen != gen { |
| 178 | return |
| 179 | } |
| 180 | e.gen = gen |
| 181 | c.makeRoom(key) |
| 182 | c.entries[key] = e |
| 183 | } |
| 184 | |
| 185 | // reset invalidates key. The entry is kept with its targets dropped and gen |
| 186 | // bumped, rather than deleted, so a lookup for it that is in flight has a |
| 187 | // generation to compare against. |
| 188 | func (c *cache) reset(key string) { |
| 189 | c.mu.Lock() |
| 190 | defer c.mu.Unlock() |
| 191 | // Keep the reset effective if cache pressure evicts its per-key marker. |
| 192 | c.epoch++ |
| 193 | c.makeRoom(key) |
| 194 | c.entries[key] = cacheEntry{gen: c.entries[key].gen + 1} |
| 195 | } |
| 196 | |
| 197 | // resetAll drops every entry. Bumping epoch covers lookups in flight, so the |
| 198 | // entries themselves can go. |
| 199 | func (c *cache) resetAll() { |
| 200 | c.mu.Lock() |
| 201 | defer c.mu.Unlock() |
| 202 | c.entries = make(map[string]cacheEntry) |
| 203 | c.epoch++ |
| 204 | } |
| 205 | |
| 206 | // makeRoom evicts a random entry if the cache is full and key is not already |
| 207 | // in it. c.mu must be held. |
| 208 | func (c *cache) makeRoom(key string) { |
| 209 | if _, ok := c.entries[key]; ok || len(c.entries) < maxCacheEntries { |
| 210 | return |
| 211 | } |
| 212 | for k := range c.entries { |
| 213 | delete(c.entries, k) |
| 214 | break |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | // lookupGate is a counting semaphore with a bounded wait queue. |
| 219 | type lookupGate struct { |
| 220 | running chan struct{} |
| 221 | pending atomic.Int64 |
| 222 | maxPending int64 |
| 223 | } |
| 224 | |
| 225 | func newLookupGate(maxRunning, maxPending int) *lookupGate { |
| 226 | return &lookupGate{ |
| 227 | running: make(chan struct{}, maxRunning), |
| 228 | maxPending: int64(maxPending), |
| 229 | } |
| 230 | } |
| 231 | |
| 232 | // acquire takes a slot, waiting until ctx is done. It fails immediately with |
| 233 | // errTooManyLookups if the wait queue is full. A free slot goes to whoever asks |
| 234 | // first, even ahead of queued callers; that keeps the common case to a single |
| 235 | // channel send. |
| 236 | func (g *lookupGate) acquire(ctx context.Context) error { |
| 237 | select { |
| 238 | case g.running <- struct{}{}: |
| 239 | return nil |
| 240 | default: |
| 241 | } |
| 242 | if g.pending.Add(1) > g.maxPending { |
| 243 | g.pending.Add(-1) |
| 244 | return errTooManyLookups |
| 245 | } |
| 246 | defer g.pending.Add(-1) |
| 247 | select { |
| 248 | case g.running <- struct{}{}: |
| 249 | return nil |
| 250 | case <-ctx.Done(): |
| 251 | return ctx.Err() |
| 252 | } |
| 253 | } |
| 254 | |
| 255 | // release gives back a slot taken by a successful acquire. |
| 256 | func (g *lookupGate) release() { |
| 257 | <-g.running |
| 258 | } |