For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.
tunnel/gated.go
| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "sync" |
| 5 | "time" |
| 6 | ) |
| 7 | |
| 8 | // The gated thin pipe (the design plan, sections 3 and 5). A device whose trial ended, or whose license lapsed within the policy's cut-off, is |
| 9 | // authorized as gated: it is served, but only a little (policy gated.bytes_per_day and gated.streams_per_day), enough for its paywall card to reach |
| 10 | // ChatGPT. The device's own gate is not a boundary against its owner (he can edit his files), so THIS is: the node counts, and it counts on every |
| 11 | // status the control plane gives, a RENEW included, so a device that turns gated in the middle of a connection is held to the budget from that moment |
| 12 | // on, its streams already open included (they are cut when the bytes run out). |
| 13 | // |
| 14 | // The count is per UTC day and per device host, in memory. What a device used while gated is not forgotten by reconnecting; it is forgotten at midnight. |
| 15 | |
| 16 | // dayUsage is what one device host used of its gated budget today. |
| 17 | type dayUsage struct { |
| 18 | mu sync.Mutex |
| 19 | day int64 |
| 20 | bytes int64 |
| 21 | streams int |
| 22 | users int // streams holding this record: it is never swept while there are any |
| 23 | } |
| 24 | |
| 25 | // rollLocked starts a new day's count when the day changed. |
| 26 | func (u *dayUsage) rollLocked(now time.Time) { |
| 27 | if day := now.Unix() / 86400; day != u.day { |
| 28 | u.day, u.bytes, u.streams = day, 0, 0 |
| 29 | } |
| 30 | } |
| 31 | |
| 32 | // admit counts a new stream of a gated device, or refuses it (reason says which budget is spent). A byte budget that is already spent also refuses: |
| 33 | // the stream would get nothing. |
| 34 | func (u *dayUsage) admit(now time.Time, bytesMax int64, streamsMax int) (ok bool, reason string) { |
| 35 | u.mu.Lock() |
| 36 | defer u.mu.Unlock() |
| 37 | u.rollLocked(now) |
| 38 | switch { |
| 39 | case u.streams >= streamsMax: |
| 40 | return false, "gated_streams" |
| 41 | case u.bytes >= bytesMax: |
| 42 | return false, "gated_bytes" |
| 43 | } |
| 44 | u.streams++ |
| 45 | return true, "" |
| 46 | } |
| 47 | |
| 48 | // charge takes n bytes of today's budget and returns how many of them may be forwarded: all of them while there is budget, fewer at the edge, none |
| 49 | // after it. |
| 50 | func (u *dayUsage) charge(now time.Time, n, bytesMax int64) int64 { |
| 51 | u.mu.Lock() |
| 52 | defer u.mu.Unlock() |
| 53 | u.rollLocked(now) |
| 54 | take := min(n, max(bytesMax-u.bytes, 0)) |
| 55 | u.bytes += take |
| 56 | return take |
| 57 | } |
| 58 | |
| 59 | // gateTable keeps the usage of every device host that was gated today. |
| 60 | type gateTable struct { |
| 61 | clk clock |
| 62 | |
| 63 | mu sync.Mutex |
| 64 | byHost map[string]*dayUsage |
| 65 | } |
| 66 | |
| 67 | func newGateTable(clk clock) *gateTable { return &gateTable{clk: clk, byHost: map[string]*dayUsage{}} } |
| 68 | |
| 69 | // acquire gives the usage record of host (made when there is none) and counts the caller as a user; release gives it back. |
| 70 | func (g *gateTable) acquire(host string) *dayUsage { |
| 71 | g.mu.Lock() |
| 72 | defer g.mu.Unlock() |
| 73 | u := g.byHost[host] |
| 74 | if u == nil { |
| 75 | u = &dayUsage{day: g.clk.Now().Unix() / 86400} |
| 76 | g.byHost[host] = u |
| 77 | } |
| 78 | u.mu.Lock() |
| 79 | u.users++ |
| 80 | u.mu.Unlock() |
| 81 | return u |
| 82 | } |
| 83 | |
| 84 | func (g *gateTable) release(u *dayUsage) { |
| 85 | u.mu.Lock() |
| 86 | u.users-- |
| 87 | u.mu.Unlock() |
| 88 | } |
| 89 | |
| 90 | // Sweep forgets the records nobody holds that have nothing to remember: a previous day's, or an empty one (a background task calls it). |
| 91 | func (g *gateTable) Sweep() { |
| 92 | now := g.clk.Now() |
| 93 | g.mu.Lock() |
| 94 | defer g.mu.Unlock() |
| 95 | for host, u := range g.byHost { |
| 96 | u.mu.Lock() |
| 97 | if u.users == 0 { |
| 98 | u.rollLocked(now) |
| 99 | if u.bytes == 0 && u.streams == 0 { |
| 100 | delete(g.byHost, host) |
| 101 | } |
| 102 | } |
| 103 | u.mu.Unlock() |
| 104 | } |
| 105 | } |
| 106 | |
| 107 | // size is the number of device hosts with a record, for the metrics. |
| 108 | func (g *gateTable) size() int { |
| 109 | g.mu.Lock() |
| 110 | defer g.mu.Unlock() |
| 111 | return len(g.byHost) |
| 112 | } |