File: freeport.go

package info (click to toggle)
consul 1.8.7%2Bdfsg1-2
  • links: PTS, VCS
  • area: main
  • in suites: bullseye, bullseye-backports
  • size: 57,848 kB
  • sloc: javascript: 25,918; sh: 3,807; makefile: 135; cpp: 102
file content (391 lines) | stat: -rw-r--r-- 9,994 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
// Package freeport provides a helper for allocating free ports across multiple
// processes on the same machine.
package freeport

import (
	"container/list"
	"fmt"
	"math/rand"
	"net"
	"os"
	"runtime"
	"sync"
	"time"

	"github.com/mitchellh/go-testing-interface"
)

const (
	// maxBlocks is the number of available port blocks before exclusions.
	maxBlocks = 30

	// lowPort is the lowest port number that should be used.
	lowPort = 10000

	// attempts is how often we try to allocate a port block
	// before giving up.
	attempts = 10
)

var (
	// blockSize is the size of the allocated port block. ports are given out
	// consecutively from that block and after that point in a LRU fashion.
	blockSize int

	// effectiveMaxBlocks is the number of available port blocks.
	// lowPort + effectiveMaxBlocks * blockSize must be less than 65535.
	effectiveMaxBlocks int

	// firstPort is the first port of the allocated block.
	firstPort int

	// lockLn is the system-wide mutex for the port block.
	lockLn net.Listener

	// mu guards:
	// - pendingPorts
	// - freePorts
	// - total
	mu sync.Mutex

	// once is used to do the initialization on the first call to retrieve free
	// ports
	once sync.Once

	// condNotEmpty is a condition variable to wait for freePorts to be not
	// empty. Linked to 'mu'
	condNotEmpty *sync.Cond

	// freePorts is a FIFO of all currently free ports. Take from the front,
	// and return to the back.
	freePorts *list.List

	// pendingPorts is a FIFO of recently freed ports that have not yet passed
	// the not-in-use check.
	pendingPorts *list.List

	// total is the total number of available ports in the block for use.
	total int

	// stopCh is used to signal to background goroutines to terminate. Only
	// really exists for the safety of reset() during unit tests.
	stopCh chan struct{}

	// stopWg is used to keep track of background goroutines that are still
	// alive. Only really exists for the safety of reset() during unit tests.
	stopWg sync.WaitGroup
)

// initialize is used to initialize freeport.
func initialize() {
	var err error

	blockSize = 1500
	limit, err := systemLimit()
	if err != nil {
		panic("freeport: error getting system limit: " + err.Error())
	}
	if limit > 0 && limit < blockSize {
		logf("INFO", "blockSize %d too big for system limit %d. Adjusting...", blockSize, limit)
		blockSize = limit - 3
	}

	effectiveMaxBlocks, err = adjustMaxBlocks()
	if err != nil {
		panic("freeport: ephemeral port range detection failed: " + err.Error())
	}
	if effectiveMaxBlocks < 0 {
		panic("freeport: no blocks of ports available outside of ephemeral range")
	}
	if lowPort+effectiveMaxBlocks*blockSize > 65535 {
		panic("freeport: block size too big or too many blocks requested")
	}

	rand.Seed(time.Now().UnixNano())
	firstPort, lockLn = alloc()

	condNotEmpty = sync.NewCond(&mu)
	freePorts = list.New()
	pendingPorts = list.New()

	// fill with all available free ports
	for port := firstPort + 1; port < firstPort+blockSize; port++ {
		if used := isPortInUse(port); !used {
			freePorts.PushBack(port)
		}
	}
	total = freePorts.Len()

	stopWg.Add(1)
	stopCh = make(chan struct{})
	// Note: we pass this param explicitly to the goroutine so that we can
	// freely recreate the underlying stop channel during reset() after closing
	// the original.
	go checkFreedPorts(stopCh)
}

func shutdownGoroutine() {
	mu.Lock()
	if stopCh == nil {
		mu.Unlock()
		return
	}

	close(stopCh)
	stopCh = nil
	mu.Unlock()

	stopWg.Wait()
}

// reset will reverse the setup from initialize() and then redo it (for tests)
func reset() {
	logf("INFO", "resetting the freeport package state")
	shutdownGoroutine()

	mu.Lock()
	defer mu.Unlock()

	effectiveMaxBlocks = 0
	firstPort = 0
	if lockLn != nil {
		lockLn.Close()
		lockLn = nil
	}

	once = sync.Once{}

	freePorts = nil
	pendingPorts = nil
	total = 0
}

func checkFreedPorts(stopCh <-chan struct{}) {
	defer stopWg.Done()

	ticker := time.NewTicker(250 * time.Millisecond)
	for {
		select {
		case <-stopCh:
			logf("INFO", "Closing checkFreedPorts()")
			return
		case <-ticker.C:
			checkFreedPortsOnce()
		}
	}
}

func checkFreedPortsOnce() {
	mu.Lock()
	defer mu.Unlock()

	pending := pendingPorts.Len()
	remove := make([]*list.Element, 0, pending)
	for elem := pendingPorts.Front(); elem != nil; elem = elem.Next() {
		port := elem.Value.(int)
		if used := isPortInUse(port); !used {
			freePorts.PushBack(port)
			remove = append(remove, elem)
		}
	}

	retained := pending - len(remove)

	if retained > 0 {
		logf("WARN", "%d out of %d pending ports are still in use; something probably didn't wait around for the port to be closed!", retained, pending)
	}

	if len(remove) == 0 {
		return
	}

	for _, elem := range remove {
		pendingPorts.Remove(elem)
	}

	condNotEmpty.Broadcast()
}

// adjustMaxBlocks avoids having the allocation ranges overlap the ephemeral
// port range.
func adjustMaxBlocks() (int, error) {
	ephemeralPortMin, ephemeralPortMax, err := getEphemeralPortRange()
	if err != nil {
		return 0, err
	}

	if ephemeralPortMin <= 0 || ephemeralPortMax <= 0 {
		logf("INFO", "ephemeral port range detection not configured for GOOS=%q", runtime.GOOS)
		return maxBlocks, nil
	}

	logf("INFO", "detected ephemeral port range of [%d, %d]", ephemeralPortMin, ephemeralPortMax)
	for block := 0; block < maxBlocks; block++ {
		min := lowPort + block*blockSize
		max := min + blockSize
		overlap := intervalOverlap(min, max-1, ephemeralPortMin, ephemeralPortMax)
		if overlap {
			logf("INFO", "reducing max blocks from %d to %d to avoid the ephemeral port range", maxBlocks, block)
			return block, nil
		}
	}
	return maxBlocks, nil
}

// alloc reserves a port block for exclusive use for the lifetime of the
// application. lockLn serves as a system-wide mutex for the port block and is
// implemented as a TCP listener which is bound to the firstPort and which will
// be automatically released when the application terminates.
func alloc() (int, net.Listener) {
	for i := 0; i < attempts; i++ {
		block := int(rand.Int31n(int32(effectiveMaxBlocks)))
		firstPort := lowPort + block*blockSize
		ln, err := net.ListenTCP("tcp", tcpAddr("127.0.0.1", firstPort))
		if err != nil {
			continue
		}
		// logf("DEBUG", "allocated port block %d (%d-%d)", block, firstPort, firstPort+blockSize-1)
		return firstPort, ln
	}
	panic("freeport: cannot allocate port block")
}

// MustTake is the same as Take except it panics on error.
func MustTake(n int) (ports []int) {
	ports, err := Take(n)
	if err != nil {
		panic(err)
	}
	return ports
}

// Take returns a list of free ports from the allocated port block. It is safe
// to call this method concurrently. Ports have been tested to be available on
// 127.0.0.1 TCP but there is no guarantee that they will remain free in the
// future.
func Take(n int) (ports []int, err error) {
	if n <= 0 {
		return nil, fmt.Errorf("freeport: cannot take %d ports", n)
	}

	mu.Lock()
	defer mu.Unlock()

	// Reserve a port block
	once.Do(initialize)

	if n > total {
		return nil, fmt.Errorf("freeport: block size too small")
	}

	for len(ports) < n {
		for freePorts.Len() == 0 {
			if total == 0 {
				return nil, fmt.Errorf("freeport: impossible to satisfy request; there are no actual free ports in the block anymore")
			}
			condNotEmpty.Wait()
		}

		elem := freePorts.Front()
		freePorts.Remove(elem)
		port := elem.Value.(int)

		if used := isPortInUse(port); used {
			// Something outside of the test suite has stolen this port, possibly
			// due to assignment to an ephemeral port, remove it completely.
			logf("WARN", "leaked port %d due to theft; removing from circulation", port)
			total--
			continue
		}

		ports = append(ports, port)
	}

	// logf("DEBUG", "free ports: %v", ports)
	return ports, nil
}

// peekFree returns the next port that will be returned by Take to aid in testing.
func peekFree() int {
	mu.Lock()
	defer mu.Unlock()
	return freePorts.Front().Value.(int)
}

// peekAllFree returns all free ports that could be returned by Take to aid in testing.
func peekAllFree() []int {
	mu.Lock()
	defer mu.Unlock()

	var out []int
	for elem := freePorts.Front(); elem != nil; elem = elem.Next() {
		port := elem.Value.(int)
		out = append(out, port)
	}

	return out
}

// stats returns diagnostic data to aid in testing
func stats() (numTotal, numPending, numFree int) {
	mu.Lock()
	defer mu.Unlock()
	return total, pendingPorts.Len(), freePorts.Len()
}

// Return returns a block of ports back to the general pool. These ports should
// have been returned from a call to Take().
func Return(ports []int) {
	if len(ports) == 0 {
		return // convenience short circuit for test ergonomics
	}

	mu.Lock()
	defer mu.Unlock()

	for _, port := range ports {
		if port > firstPort && port < firstPort+blockSize {
			pendingPorts.PushBack(port)
		}
	}
}

func isPortInUse(port int) bool {
	ln, err := net.ListenTCP("tcp", tcpAddr("127.0.0.1", port))
	if err != nil {
		return true
	}
	ln.Close()
	return false
}

func tcpAddr(ip string, port int) *net.TCPAddr {
	return &net.TCPAddr{IP: net.ParseIP(ip), Port: port}
}

// intervalOverlap returns true if the doubly-inclusive integer intervals
// represented by [min1, max1] and [min2, max2] overlap.
func intervalOverlap(min1, max1, min2, max2 int) bool {
	if min1 > max1 {
		logf("WARN", "interval1 is not ordered [%d, %d]", min1, max1)
		return false
	}
	if min2 > max2 {
		logf("WARN", "interval2 is not ordered [%d, %d]", min2, max2)
		return false
	}
	return min1 <= max2 && min2 <= max1
}

func logf(severity string, format string, a ...interface{}) {
	fmt.Fprintf(os.Stderr, "["+severity+"] freeport: "+format+"\n", a...)
}

// Deprecated: Please use Take/Return calls instead.
func Get(n int) (ports []int) { return MustTake(n) }

// Deprecated: Please use Take/Return calls instead.
func GetT(t testing.T, n int) (ports []int) { return MustTake(n) }

// Deprecated: Please use Take/Return calls instead.
func Free(n int) (ports []int, err error) { return MustTake(n), nil }