Commit da537c3c7d95fcbbe52e06f9fb3ff4afbcab1911 - golang-github-cloudflare-tableflip

+7

-3

.travis.yml less more

0	0	language: go
1	1
2	2	go:
3		- "1.10.x"
4		- "1.11.x"
5		- master
	3	- "1.13.x"
	4	- "1.14.x"
	5
	6	os:
	7	- linux
	8	- osx
	9	- windows

+25

-23

README.md less more

15	15	* Crashing during initialisation is OK
16	16	* Only a single upgrade is ever run in parallel
17	17
18		`tableflip` does not work on Windows.
	18	`tableflip` works on Linux and macOS.
19	19
20		It's easy to get started:
	20	## Using the library
21	21
22	22	```Go
23		upg, err := tableflip.New(tableflip.Options{})
24		if err != nil {
25		panic(err)
26		}
	23	upg, _ := tableflip.New(tableflip.Options{})
27	24	defer upg.Stop()
28	25
29	26	go func() {
30	27	sig := make(chan os.Signal, 1)
31	28	signal.Notify(sig, syscall.SIGHUP)
32	29	for range sig {
33		err := upg.Upgrade()
34		if err != nil {
35		log.Println("Upgrade failed:", err)
36		continue
37		}
38
39		log.Println("Upgrade succeeded")
	30	upg.Upgrade()
40	31	}
41	32	}()
42	33
43		ln, err := upg.Fds.Listen("tcp", "localhost:8080")
44		if err != nil {
45		log.Fatalln("Can't listen:", err)
46		}
	34	// Listen must be called before Ready
	35	ln, _ := upg.Listen("tcp", "localhost:8080")
	36	defer ln.Close()
47	37
48		var server http.Server
49		go server.Serve(ln)
	38	go http.Serve(ln, nil)
50	39
51	40	if err := upg.Ready(); err != nil {
52	41	panic(err)
53	42	}
	43
54	44	<-upg.Exit()
	45	```
55	46
56		time.AfterFunc(30*time.Second, func() {
57		os.Exit(1)
58		})
	47	Please see the more elaborate [graceful shutdown with net/http](http_example_test.go) example.
59	48
60		_ = server.Shutdown(context.Background())
	49	## Integration with `systemd`
	50
61	51	```
	52	[Unit]
	53	Description=Service using tableflip
	54
	55	[Service]
	56	ExecStart=/path/to/binary -some-flag /path/to/pid-file
	57	ExecReload=/bin/kill -HUP $MAINPID
	58	PIDFile=/path/to/pid-file
	59	```
	60
	61	See the [documentation](https://godoc.org/github.com/cloudflare/tableflip) as well.
	62
	63	The logs of a process using `tableflip` may go missing due to a [bug in journald](https://github.com/systemd/systemd/issues/13708). You can work around this by logging directly to journald, for example by using [go-systemd/journal](https://godoc.org/github.com/coreos/go-systemd/journal) and looking for the [$JOURNAL_STREAM](https://www.freedesktop.org/software/systemd/man/systemd.exec.html#$JOURNAL_STREAM) environment variable.

+3

-5

child.go less more

3	3	"encoding/gob"
4	4	"fmt"
5	5	"os"
6
7		"github.com/pkg/errors"
8	6	)
9	7
10	8	type child struct {

21	19	// readyW is passed to the child, readyR stays with the parent
22	20	readyR, readyW, err := os.Pipe()
23	21	if err != nil {
24		return nil, errors.Wrap(err, "pipe failed")
	22	return nil, fmt.Errorf("pipe failed: %s", err)
25	23	}
26	24
27	25	namesR, namesW, err := os.Pipe()
28	26	if err != nil {
29	27	readyR.Close()
30	28	readyW.Close()
31		return nil, errors.Wrap(err, "pipe failed")
	29	return nil, fmt.Errorf("pipe failed: %s", err)
32	30	}
33	31
34	32	// Copy passed fds and append the notification pipe

52	50	readyW.Close()
53	51	namesR.Close()
54	52	namesW.Close()
55		return nil, errors.Wrapf(err, "can't start process %s", os.Args[0])
	53	return nil, fmt.Errorf("can't start process %s: %s", os.Args[0], err)
56	54	}
57	55
58	56	exited := make(chan struct{})

+7

-0

doc.go less more

37	37	// you're probably using "go run main.go", for graceful reloads to work,
38	38	// you'll need use "go build main.go".
39	39	//
	40	// Tableflip does not work on Windows, because Windows does not have
	41	// the mechanisms required to support this method of graceful restarting.
	42	// It is still possible to include this package in code that runs on Windows,
	43	// which may be necessary in certain development circumstances, but it will not
	44	// provide zero downtime upgrades when running on Windows. See the `testing`
	45	// package for an example of how to use it.
	46	//
40	47	package tableflip

+17

-0

dup_fd.go less more

	0	// +build !windows
	1
	2	package tableflip
	3
	4	import (
	5	"fmt"
	6	"syscall"
	7	)
	8
	9	func dupFd(fd uintptr, name fileName) (*file, error) {
	10	dupfd, _, errno := syscall.Syscall(syscall.SYS_FCNTL, fd, syscall.F_DUPFD_CLOEXEC, 0)
	11	if errno != 0 {
	12	return nil, fmt.Errorf("can't dup fd using fcntl: %s", errno)
	13	}
	14
	15	return newFile(dupfd, name), nil
	16	}

+7

-0

dup_fd_windows.go less more

	0	package tableflip
	1
	2	import "errors"
	3
	4	func dupFd(fd uintptr, name fileName) (*file, error) {
	5	return nil, errors.New("tableflip: duplicating file descriptors is not supported on this platform")
	6	}

+0

-9

env.go less more

1	1
2	2	import (
3	3	"os"
4		"syscall"
5	4	)
6
7		var stdEnv = &env{
8		newProc: newOSProcess,
9		newFile: os.NewFile,
10		environ: os.Environ,
11		getenv: os.Getenv,
12		closeOnExec: syscall.CloseOnExec,
13		}
14	5
15	6	type env struct {
16	7	newProc func(string, []string, []*os.File, []string) (process, error)

+16

-0

env_syscalls.go less more

	0	// +build !windows
	1
	2	package tableflip
	3
	4	import (
	5	"os"
	6	"syscall"
	7	)
	8
	9	var stdEnv = &env{
	10	newProc: newOSProcess,
	11	newFile: os.NewFile,
	12	environ: os.Environ,
	13	getenv: os.Getenv,
	14	closeOnExec: syscall.CloseOnExec,
	15	}

+6

-0

env_windows.go less more

	0	package tableflip
	1
	2	// replace Unix-specific syscall with a no-op so it will build
	3	// without errors.
	4
	5	var stdEnv *env = nil

+122

-23

fds.go less more

0	0	package tableflip
1	1
2	2	import (
	3	"fmt"
3	4	"net"
4	5	"os"
	6	"runtime"
5	7	"strings"
6	8	"sync"
7	9	"syscall"
8
9		"github.com/pkg/errors"
10	10	)
11	11
12	12	// Listener can be shared between processes.

15	15	syscall.Conn
16	16	}
17	17
	18	// PacketConn can be shared between processes.
	19	type PacketConn interface {
	20	net.PacketConn
	21	syscall.Conn
	22	}
	23
18	24	// Conn can be shared between processes.
19	25	type Conn interface {
20	26	net.Conn

23	29
24	30	const (
25	31	listenKind = "listener"
	32	packetKind = "packet"
26	33	connKind = "conn"
27	34	fdKind = "fd"
28	35	)

31	38
32	39	func (name fileName) String() string {
33	40	return strings.Join(name[:], ":")
	41	}
	42
	43	func (name fileName) isUnix() bool {
	44	if name[0] == listenKind && (name[1] == "unix" \|\| name[1] == "unixpacket") {
	45	return true
	46	}
	47	if name[0] == packetKind && (name[1] == "unixgram") {
	48	return true
	49	}
	50	return false
34	51	}
35	52
36	53	// file works around the fact that it's not possible

88	105
89	106	ln, err = net.Listen(network, addr)
90	107	if err != nil {
91		return nil, errors.Wrap(err, "can't create new listener")
	108	return nil, fmt.Errorf("can't create new listener: %s", err)
92	109	}
93	110
94	111	if _, ok := ln.(Listener); !ok {
95	112	ln.Close()
96		return nil, errors.Errorf("%T doesn't implement tableflip.Listener", ln)
	113	return nil, fmt.Errorf("%T doesn't implement tableflip.Listener", ln)
97	114	}
98	115
99	116	err = f.addListenerLocked(network, addr, ln.(Listener))

124	141
125	142	ln, err := net.FileListener(file.File)
126	143	if err != nil {
127		return nil, errors.Wrapf(err, "can't inherit listener %s %s", network, addr)
	144	return nil, fmt.Errorf("can't inherit listener %s %s: %s", network, addr, err)
128	145	}
129	146
130	147	delete(f.inherited, key)

152	169	ifc.SetUnlinkOnClose(false)
153	170	}
154	171
155		return f.addConnLocked(listenKind, network, addr, ln)
	172	return f.addSyscallConnLocked(listenKind, network, addr, ln)
	173	}
	174
	175	// ListenPacket returns a packet conn inherited from the parent process, or creates a new one.
	176	func (f *Fds) ListenPacket(network, addr string) (net.PacketConn, error) {
	177	f.mu.Lock()
	178	defer f.mu.Unlock()
	179
	180	conn, err := f.packetConnLocked(network, addr)
	181	if err != nil {
	182	return nil, err
	183	}
	184
	185	if conn != nil {
	186	return conn, nil
	187	}
	188
	189	conn, err = net.ListenPacket(network, addr)
	190	if err != nil {
	191	return nil, fmt.Errorf("can't create new listener: %s", err)
	192	}
	193
	194	if _, ok := conn.(PacketConn); !ok {
	195	return nil, fmt.Errorf("%T doesn't implement tableflip.PacketConn", conn)
	196	}
	197
	198	err = f.addSyscallConnLocked(packetKind, network, addr, conn.(PacketConn))
	199	if err != nil {
	200	conn.Close()
	201	return nil, err
	202	}
	203
	204	return conn, nil
	205	}
	206
	207	// PacketConn returns an inherited packet connection or nil.
	208	//
	209	// It is safe to close the returned packet connection.
	210	func (f *Fds) PacketConn(network, addr string) (net.PacketConn, error) {
	211	f.mu.Lock()
	212	defer f.mu.Unlock()
	213
	214	return f.packetConnLocked(network, addr)
	215	}
	216
	217	// AddPacketConn adds a PacketConn.
	218	//
	219	// It is safe to close conn after calling the method.
	220	// Any existing packet connection with the same address is overwitten.
	221	func (f *Fds) AddPacketConn(network, addr string, conn PacketConn) error {
	222	f.mu.Lock()
	223	defer f.mu.Unlock()
	224
	225	return f.addSyscallConnLocked(packetKind, network, addr, conn)
	226	}
	227
	228	func (f *Fds) packetConnLocked(network, addr string) (net.PacketConn, error) {
	229	key := fileName{packetKind, network, addr}
	230	file := f.inherited[key]
	231	if file == nil {
	232	return nil, nil
	233	}
	234
	235	conn, err := net.FilePacketConn(file.File)
	236	if err != nil {
	237	return nil, fmt.Errorf("can't inherit packet conn %s %s: %s", network, addr, err)
	238	}
	239
	240	delete(f.inherited, key)
	241	f.used[key] = file
	242	return conn, nil
156	243	}
157	244
158	245	// Conn returns an inherited connection or nil.

170	257
171	258	conn, err := net.FileConn(file.File)
172	259	if err != nil {
173		return nil, errors.Wrapf(err, "can't inherit connection %s %s", network, addr)
	260	return nil, fmt.Errorf("can't inherit connection %s %s: %s", network, addr, err)
174	261	}
175	262
176	263	delete(f.inherited, key)

185	272	f.mu.Lock()
186	273	defer f.mu.Unlock()
187	274
188		return f.addConnLocked(connKind, network, addr, conn)
189		}
190
191		func (f *Fds) addConnLocked(kind, network, addr string, conn syscall.Conn) error {
	275	return f.addSyscallConnLocked(connKind, network, addr, conn)
	276	}
	277
	278	func (f *Fds) addSyscallConnLocked(kind, network, addr string, conn syscall.Conn) error {
192	279	key := fileName{kind, network, addr}
193	280	file, err := dupConn(conn, key)
194	281	if err != nil {
195		return errors.Wrapf(err, "can't dup listener %s %s", network, addr)
	282	return fmt.Errorf("can't dup %s (%s %s): %s", kind, network, addr, err)
196	283	}
197	284
198	285	delete(f.inherited, key)

262	349	defer f.mu.Unlock()
263	350
264	351	for key, file := range f.inherited {
265		if key[0] == listenKind && (key[1] == "unix" \|\| key[1] == "unixpacket") {
	352	if key.isUnix() {
266	353	// Remove inherited but unused Unix sockets from the file system.
267	354	// This undoes the effect of SetUnlinkOnClose(false).
268	355	_ = unlinkUnixSocket(key[2])

273	360	}
274	361
275	362	func unlinkUnixSocket(path string) error {
	363	if runtime.GOOS == "linux" && strings.HasPrefix(path, "@") {
	364	// Don't unlink sockets using the abstract namespace.
	365	return nil
	366	}
	367
276	368	info, err := os.Stat(path)
277	369	if err != nil {
278	370	return err

290	382	defer f.mu.Unlock()
291	383
292	384	for _, file := range f.used {
	385	_ = file.Close()
	386	}
	387	f.used = make(map[fileName]*file)
	388	}
	389
	390	func (f *Fds) closeAndRemoveUsed() {
	391	f.mu.Lock()
	392	defer f.mu.Unlock()
	393
	394	for key, file := range f.used {
	395	if key.isUnix() {
	396	// Remove used Unix Domain Sockets if we are shutting
	397	// down without having done an upgrade.
	398	// This undoes the effect of SetUnlinkOnClose(false).
	399	_ = unlinkUnixSocket(key[2])
	400	}
293	401	_ = file.Close()
294	402	}
295	403	f.used = make(map[fileName]*file)

309	417	dup, duperr = dupFd(fd, name)
310	418	})
311	419	if err != nil {
312		return nil, errors.Wrap(err, "can't access fd")
	420	return nil, fmt.Errorf("can't access fd: %s", err)
313	421	}
314	422	return dup, duperr
315	423	}
316
317		func dupFd(fd uintptr, name fileName) (*file, error) {
318		dupfd, _, errno := syscall.Syscall(syscall.SYS_FCNTL, fd, syscall.F_DUPFD_CLOEXEC, 0)
319		if errno != 0 {
320		return nil, errors.Wrap(errno, "can't dup fd using fcntl")
321		}
322
323		return newFile(dupfd, name), nil
324		}

+178

-66

fds_test.go less more

0	0	package tableflip
1	1
2	2	import (
	3	"io"
3	4	"io/ioutil"
4	5	"net"
5	6	"os"
6	7	"path/filepath"
	8	"runtime"
7	9	"testing"
8	10	)
9	11
	12	func TestFdsAddListener(t *testing.T) {
	13	socketPath, cleanup := tempSocket(t)
	14	defer cleanup()
	15
	16	addrs := [][2]string{
	17	{"unix", socketPath},
	18	{"tcp", "localhost:0"},
	19	}
	20
	21	fds := newFds(nil)
	22	for _, addr := range addrs {
	23	ln, err := net.Listen(addr[0], addr[1])
	24	if err != nil {
	25	t.Fatal(err)
	26	}
	27	if err := fds.AddListener(addr[0], addr[1], ln.(Listener)); err != nil {
	28	t.Fatalf("Can't add %s listener: %s", addr[0], err)
	29	}
	30	ln.Close()
	31	}
	32	}
	33
	34	func TestFdsAddPacketConn(t *testing.T) {
	35	socketPath, cleanup := tempSocket(t)
	36	defer cleanup()
	37
	38	addrs := [][2]string{
	39	{"unix", socketPath},
	40	{"udp", "localhost:0"},
	41	}
	42
	43	fds := newFds(nil)
	44	for _, addr := range addrs {
	45	conn, err := net.ListenPacket(addr[0], addr[1])
	46	if err != nil {
	47	t.Fatal(err)
	48	}
	49	if err := fds.AddPacketConn(addr[0], addr[1], conn.(PacketConn)); err != nil {
	50	t.Fatalf("Can't add %s listener: %s", addr[0], err)
	51	}
	52	conn.Close()
	53	}
	54	}
	55
	56	func tempSocket(t *testing.T) (string, func()) {
	57	t.Helper()
	58
	59	temp, err := ioutil.TempDir("", "tableflip")
	60	if err != nil {
	61	t.Fatal(err)
	62	}
	63
	64	return filepath.Join(temp, "socket"), func() { os.RemoveAll(temp) }
	65	}
	66
10	67	func TestFdsListen(t *testing.T) {
11		addrs := [][2]string{
12		{"unix", ""},
	68	socketPath, cleanup := tempSocket(t)
	69	defer cleanup()
	70
	71	addrs := [][2]string{
13	72	{"tcp", "localhost:0"},
14		}
15
16		fds := newFds(nil)
17
18		for _, addr := range addrs {
19		ln, err := fds.Listen(addr[0], addr[1])
20		if err != nil {
21		t.Fatal(err)
	73	{"udp", "localhost:0"},
	74	{"unix", socketPath},
	75	{"unixgram", socketPath + "Unixgram"},
	76	}
	77
	78	// Linux supports the abstract namespace for domain sockets.
	79	if runtime.GOOS == "linux" {
	80	addrs = append(addrs,
	81	[2]string{"unixpacket", socketPath + "Unixpacket"},
	82	[2]string{"unix", ""},
	83	[2]string{"unixpacket", ""},
	84	[2]string{"unixgram", ""},
	85	)
	86	}
	87
	88	var (
	89	ln io.Closer
	90	err error
	91	)
	92
	93	parent := newFds(nil)
	94	for _, addr := range addrs {
	95	switch addr[0] {
	96	case "udp", "unixgram":
	97	ln, err = parent.ListenPacket(addr[0], addr[1])
	98	default:
	99	ln, err = parent.Listen(addr[0], addr[1])
	100	}
	101	if err != nil {
	102	t.Fatalf("Can't create %s listener: %s", addr[0], err)
22	103	}
23	104	if ln == nil {
24		t.Fatal("Missing listener", addr)
	105	t.Fatalf("Got a nil %s listener", addr[0])
25	106	}
26	107	ln.Close()
27	108	}
28		}
29
30		func TestFdsListener(t *testing.T) {
31		addr := &net.TCPAddr{
32		IP: net.ParseIP("127.0.0.1"),
33		Port: 0,
34		}
35
36		tcp, err := net.ListenTCP("tcp", addr)
37		if err != nil {
38		t.Fatal(err)
39		}
40		defer tcp.Close()
41
42		temp, err := ioutil.TempDir("", "tableflip")
43		if err != nil {
44		t.Fatal(err)
45		}
46		defer os.RemoveAll(temp)
47
48		socketPath := filepath.Join(temp, "socket")
49		unix, err := net.Listen("unix", socketPath)
50		if err != nil {
51		t.Fatal(err)
52		}
53		defer unix.Close()
54
55		parent := newFds(nil)
56		if err := parent.AddListener(addr.Network(), addr.String(), tcp); err != nil {
57		t.Fatal("Can't add listener:", err)
58		}
59		tcp.Close()
60
61		if err := parent.AddListener("unix", socketPath, unix.(Listener)); err != nil {
62		t.Fatal("Can't add listener:", err)
63		}
64		unix.Close()
65
66		if _, err := os.Stat(socketPath); err != nil {
67		t.Error("Unix.Close() unlinked socketPath:", err)
68		}
69	109
70	110	child := newFds(parent.copy())
71		ln, err := child.Listener(addr.Network(), addr.String())
72		if err != nil {
73		t.Fatal("Can't get listener:", err)
74		}
75		if ln == nil {
76		t.Fatal("Missing listener")
77		}
78		ln.Close()
79
80		child.closeInherited()
81		if _, err := os.Stat(socketPath); err == nil {
82		t.Error("closeInherited() did not unlink socketPath")
83		}
	111	for _, addr := range addrs {
	112	switch addr[0] {
	113	case "udp", "unixgram":
	114	ln, err = child.PacketConn(addr[0], addr[1])
	115	default:
	116	ln, err = child.Listener(addr[0], addr[1])
	117	}
	118	if err != nil {
	119	t.Fatalf("Can't get retrieve %s from child: %s", addr[0], err)
	120	}
	121	if ln == nil {
	122	t.Fatalf("Missing %s listener", addr[0])
	123	}
	124	ln.Close()
	125	}
	126	}
	127
	128	func TestFdsRemoveUnix(t *testing.T) {
	129	socketPath, cleanup := tempSocket(t)
	130	defer cleanup()
	131
	132	addrs := [][2]string{
	133	{"unix", socketPath},
	134	{"unixgram", socketPath + "Unixgram"},
	135	}
	136
	137	if runtime.GOOS == "linux" {
	138	addrs = append(addrs,
	139	[2]string{"unixpacket", socketPath + "Unixpacket"},
	140	)
	141	}
	142
	143	makeFds := func(t testing.T) Fds {
	144	fds := newFds(nil)
	145	for _, addr := range addrs {
	146	var c io.Closer
	147	var err error
	148	if addr[0] == "unixgram" {
	149	c, err = fds.ListenPacket(addr[0], addr[1])
	150	} else {
	151	c, err = fds.Listen(addr[0], addr[1])
	152	}
	153	if err != nil {
	154	t.Fatalf("Can't listen on socket %v: %v", addr, err)
	155	}
	156	c.Close()
	157	if _, err := os.Stat(addr[1]); err != nil {
	158	t.Errorf("%s Close() unlinked socket: %s", addr[0], err)
	159	}
	160	}
	161	return fds
	162	}
	163
	164	t.Run("closeAndRemoveUsed", func(t *testing.T) {
	165	parent := makeFds(t)
	166	parent.closeAndRemoveUsed()
	167	for _, addr := range addrs {
	168	if _, err := os.Stat(addr[1]); err == nil {
	169	t.Errorf("Used %s listeners are not removed", addr[0])
	170	}
	171	}
	172	})
	173
	174	t.Run("closeInherited", func(t *testing.T) {
	175	parent := makeFds(t)
	176	child := newFds(parent.copy())
	177	child.closeInherited()
	178	for _, addr := range addrs {
	179	if _, err := os.Stat(addr[1]); err == nil {
	180	t.Errorf("Inherited but unused %s listeners are not removed", addr[0])
	181	}
	182	}
	183	})
	184
	185	t.Run("closeUsed", func(t *testing.T) {
	186	parent := makeFds(t)
	187	parent.closeUsed()
	188	for _, addr := range addrs {
	189	if _, err := os.Stat(addr[1]); err != nil {
	190	t.Errorf("Used %s listeners are removed", addr[0])
	191	}
	192	}
	193	})
84	194	}
85	195
86	196	func TestFdsConn(t *testing.T) {
	197	socketPath, cleanup := tempSocket(t)
	198	defer cleanup()
87	199	unix, err := net.ListenUnixgram("unixgram", &net.UnixAddr{
88	200	Net: "unixgram",
89		Name: "",
	201	Name: socketPath,
90	202	})
91	203	if err != nil {
92	204	t.Fatal(err)

+3

-0

go.mod less more

	0	module github.com/cloudflare/tableflip
	1
	2	go 1.13

+0

-0

go.sum less more

(New empty file)

+2

-1

http_example_test.go less more

44	44	}
45	45	}()
46	46
47		ln, err := upg.Fds.Listen("tcp", *listenAddr)
	47	// Listen must be called before Ready
	48	ln, err := upg.Listen("tcp", *listenAddr)
48	49	if err != nil {
49	50	log.Fatalln("Can't listen:", err)
50	51	}

+5

-5

parent.go less more

1	1
2	2	import (
3	3	"encoding/gob"
	4	"errors"
	5	"fmt"
4	6	"io"
5	7	"io/ioutil"
6	8	"os"
7
8		"github.com/pkg/errors"
9	9	)
10	10
11	11	const (

30	30	var names [][]string
31	31	dec := gob.NewDecoder(rd)
32	32	if err := dec.Decode(&names); err != nil {
33		return nil, nil, errors.Wrap(err, "can't decode names from parent process")
	33	return nil, nil, fmt.Errorf("can't decode names from parent process: %s", err)
34	34	}
35	35
36	36	files := make(map[fileName]*file)

57	57	if n != 0 {
58	58	err = errors.New("unexpected data from parent process")
59	59	} else if err != nil {
60		err = errors.Wrap(err, "unexpected error while waiting for parent to exit")
	60	err = fmt.Errorf("unexpected error while waiting for parent to exit: %s", err)
61	61	}
62	62	result <- err
63	63	close(exited)

73	73	func (ps *parent) sendReady() error {
74	74	defer ps.wr.Close()
75	75	if _, err := ps.wr.Write([]byte{notifyReady}); err != nil {
76		return errors.Wrap(err, "can't notify parent process")
	76	return fmt.Errorf("can't notify parent process: %s", err)
77	77	}
78	78	return nil
79	79	}

+52

-0

testing/fds.go less more

	0	package testing
	1
	2	import (
	3	"net"
	4	"os"
	5	)
	6
	7	type Fds struct{}
	8
	9	// Listen returns a listener by calling net.Listen directly
	10	//
	11	// Note: In the stub implementation, this is the only function that
	12	// actually does anything
	13	func (f *Fds) Listen(network, addr string) (net.Listener, error) {
	14	return net.Listen(network, addr)
	15	}
	16
	17	// Listener always returns nil, since it is impossible to inherit with
	18	// the stub implementation
	19	func (f *Fds) Listener(network, addr string) (net.Listener, error) {
	20	return nil, nil
	21	}
	22
	23	// AddListener does nothing, since there is no reason to track connections
	24	// in the stub implementation
	25	func (f *Fds) AddListener(network, addr string, ln net.Listener) error {
	26	return nil
	27	}
	28
	29	// Conn always returns nil, since it is impossible to inherit with
	30	// the stub implementation
	31	func (f *Fds) Conn(network, addr string) (net.Conn, error) {
	32	return nil, nil
	33	}
	34
	35	// AddConn does nothing, since there is no reason to track connections
	36	// in the stub implementation
	37	func (f *Fds) AddConn(network, addr string, conn net.Conn) error {
	38	return nil
	39	}
	40
	41	// File always returns nil, since it is impossible to inherit with
	42	// the stub implementation
	43	func (f Fds) File(name string) (os.File, error) {
	44	return nil, nil
	45	}
	46
	47	// AddFile does nothing, since there is no reason to track connections
	48	// in the stub implementation
	49	func (f Fds) AddFile(name string, file os.File) error {
	50	return nil
	51	}

+24

-0

testing/fds_test.go less more

	0	package testing
	1
	2	import (
	3	"testing"
	4	)
	5
	6	func TestFdsListen(t *testing.T) {
	7	addrs := [][2]string{
	8	{"tcp", "localhost:0"},
	9	}
	10
	11	fds := &Fds{}
	12
	13	for _, addr := range addrs {
	14	ln, err := fds.Listen(addr[0], addr[1])
	15	if err != nil {
	16	t.Fatal(err)
	17	}
	18	if ln == nil {
	19	t.Fatal("Missing listener", addr)
	20	}
	21	ln.Close()
	22	}
	23	}

+98

-0

testing/http_example_test.go less more

	0	package testing_test
	1
	2	import (
	3	"context"
	4	"errors"
	5	"flag"
	6	"fmt"
	7	"log"
	8	"net"
	9	"net/http"
	10	"os"
	11	"os/signal"
	12	"syscall"
	13	"time"
	14
	15	"github.com/cloudflare/tableflip"
	16	"github.com/cloudflare/tableflip/testing"
	17	)
	18
	19	type upgrader interface {
	20	Listen(network, addr string) (net.Listener, error)
	21	Stop()
	22	Upgrade() error
	23	Ready() error
	24	Exit() <-chan struct{}
	25	}
	26
	27	// This shows how to use the upgrader
	28	// with the graceful shutdown facilities of net/http
	29	// and using the stub implementation if on an unsupported platform.
	30	func Example_httpShutdown() {
	31	var (
	32	listenAddr = flag.String("listen", "localhost:8080", "`Address` to listen on")
	33	pidFile = flag.String("pid-file", "", "`Path` to pid file")
	34	)
	35
	36	flag.Parse()
	37	log.SetPrefix(fmt.Sprintf("%d ", os.Getpid()))
	38
	39	var upg upgrader
	40	upg, err := tableflip.New(tableflip.Options{
	41	PIDFile: *pidFile,
	42	})
	43	if errors.Is(err, tableflip.ErrNotSupported) {
	44	upg, _ = testing.New()
	45	} else if err != nil {
	46	panic(err)
	47	}
	48	defer upg.Stop()
	49
	50	// Do an upgrade on SIGHUP
	51	// NOTE: With `testing.Upgrader` this goroutine is useless
	52	// You may choose to enclose it inside an `if` statement block.
	53	go func() {
	54	sig := make(chan os.Signal, 1)
	55	signal.Notify(sig, syscall.SIGHUP)
	56	for range sig {
	57	err := upg.Upgrade()
	58	if err != nil {
	59	log.Println("Upgrade failed:", err)
	60	}
	61	}
	62	}()
	63
	64	// Listen must be called before Ready
	65	ln, err := upg.Listen("tcp", *listenAddr)
	66	if err != nil {
	67	log.Fatalln("Can't listen:", err)
	68	}
	69
	70	server := http.Server{
	71	// Set timeouts, etc.
	72	}
	73
	74	go func() {
	75	err := server.Serve(ln)
	76	if err != http.ErrServerClosed {
	77	log.Println("HTTP server:", err)
	78	}
	79	}()
	80
	81	log.Printf("ready")
	82	if err := upg.Ready(); err != nil {
	83	panic(err)
	84	}
	85	<-upg.Exit()
	86
	87	// Make sure to set a deadline on exiting the process
	88	// after upg.Exit() is closed. No new upgrades can be
	89	// performed if the parent doesn't exit.
	90	time.AfterFunc(30*time.Second, func() {
	91	log.Println("Graceful shutdown timed out")
	92	os.Exit(1)
	93	})
	94
	95	// Wait for connections to drain.
	96	server.Shutdown(context.Background())
	97	}

+70

-0

testing/upgrader.go less more

	0	// Package testing provides a stub implementation that can be used for
	1	// simplified testing of applications that normally use tableflip.
	2	// It is also helpful for allowing projects that use tableflip
	3	// able to run on Windows, which does not support tableflip.
	4	package testing
	5
	6	import (
	7	"context"
	8
	9	"github.com/cloudflare/tableflip"
	10	)
	11
	12	// Upgrader has all the methods of tableflip.Upgrader, but they don't
	13	// actually do anything special.
	14	type Upgrader struct {
	15	*Fds
	16	}
	17
	18	// New creates a new stub Upgrader.
	19	//
	20	// Unlike the real version, this can be called many times.
	21	func New() (*Upgrader, error) {
	22	upg := newStubUpgrader()
	23
	24	return upg, nil
	25	}
	26
	27	func newStubUpgrader() *Upgrader {
	28	return &Upgrader{
	29	&Fds{},
	30	}
	31	}
	32
	33	// Ready does nothing, since it is impossible to inherit with
	34	// the stub implementation.
	35	// However, the function still needs to be callable without errors
	36	// in order to be useful.
	37	func (u *Upgrader) Ready() error {
	38	return nil
	39	}
	40
	41	// Exit returns a channel which is closed when the process should
	42	// exit.
	43	// We can return nil here because reading from a nil channel blocks
	44	func (u *Upgrader) Exit() <-chan struct{} {
	45	return nil
	46	}
	47
	48	// Stop does nothing, since there will never be anything to stop
	49	// in the stub implementation
	50	func (u *Upgrader) Stop() {
	51	}
	52
	53	// WaitForParent returns immediately, since the stub implementation
	54	// can never be a parent
	55	func (u *Upgrader) WaitForParent(ctx context.Context) error {
	56	return nil
	57	}
	58
	59	// HasParent is always false, since the stub implementation can never
	60	// have a parent
	61	func (u *Upgrader) HasParent() bool {
	62	return false
	63	}
	64
	65	// Upgrade always returns an error in the stub implementation,
	66	// since nothing can be done.
	67	func (u *Upgrader) Upgrade() error {
	68	return tableflip.ErrNotSupported
	69	}

+46

-11

upgrader.go less more

1	1
2	2	import (
3	3	"context"
	4	"errors"
	5	"fmt"
4	6	"io/ioutil"
5	7	"os"
6	8	"path/filepath"
	9	"runtime"
7	10	"strconv"
8	11	"sync"
9	12	"time"
10
11		"github.com/pkg/errors"
12	13	)
13	14
14	15	// DefaultUpgradeTimeout is the duration before the Upgrader kills the new process if no

26	27
27	28	// Upgrader handles zero downtime upgrades and passing files between processes.
28	29	type Upgrader struct {
	30	*Fds
	31
29	32	*env
30	33	opts Options
31	34	parent *parent

37	40	upgradeC chan chan<- error
38	41	exitC chan struct{}
39	42	exitFd chan neverCloseThisFile
40
41		Fds *Fds
42	43	}
43	44
44	45	var (

46	47	stdEnvUpgrader *Upgrader
47	48	)
48	49
	50	var ErrNotSupported = errors.New("tableflip: platform does not support graceful restart")
	51
49	52	// New creates a new Upgrader. Files are passed from the parent and may be empty.
50	53	//
51		// Only the first call to this function will succeed.
	54	// Only the first call to this function will succeed. May return ErrNotSupported.
52	55	func New(opts Options) (upg *Upgrader, err error) {
53	56	stdEnvMu.Lock()
54	57	defer stdEnvMu.Unlock()
	58
	59	if !isSupportedOS() {
	60	return nil, fmt.Errorf("%w", ErrNotSupported)
	61	}
55	62
56	63	if stdEnvUpgrader != nil {
57	64	return nil, errors.New("tableflip: only a single Upgrader allowed")

65	72	}
66	73
67	74	func newUpgrader(env env, opts Options) (Upgrader, error) {
	75	if initialWD == "" {
	76	return nil, errors.New("couldn't determine initial working directory")
	77	}
	78
68	79	parent, files, err := newParent(env)
69	80	if err != nil {
70	81	return nil, err

104	115
105	116	if u.opts.PIDFile != "" {
106	117	if err := writePIDFile(u.opts.PIDFile); err != nil {
107		return errors.Wrap(err, "tableflip: can't write PID file")
	118	return fmt.Errorf("tableflip: can't write PID file: %s", err)
108	119	}
109	120	}
110	121

122	133
123	134	// Stop prevents any more upgrades from happening, and closes
124	135	// the exit channel.
	136	//
	137	// If this function is called before a call to Upgrade() has
	138	// succeeded, it is assumed that the process is being shut down
	139	// completely. All Unix sockets known to Upgrader.Fds are then
	140	// unlinked from the filesystem.
125	141	func (u *Upgrader) Stop() {
126	142	u.stopOnce.Do(func() {
127	143	// Interrupt any running Upgrade(), and

176	192
177	193	func (u *Upgrader) run() {
178	194	defer close(u.exitC)
179		defer u.Fds.closeUsed()
180	195
181	196	var (
182	197	parentExited <-chan struct{}

196	211	processReady = nil
197	212
198	213	case <-u.stopC:
	214	u.Fds.closeAndRemoveUsed()
199	215	return
200	216
201	217	case request := <-u.upgradeC:

217	233	// exits. This signals to the new process that the old process
218	234	// has exited.
219	235	u.exitFd <- neverCloseThisFile{file}
	236	u.Fds.closeUsed()
220	237	return
221	238	}
222	239	}

226	243	func (u Upgrader) doUpgrade() (os.File, error) {
227	244	child, err := startChild(u.env, u.Fds.copy())
228	245	if err != nil {
229		return nil, errors.Wrap(err, "can't start child")
	246	return nil, fmt.Errorf("can't start child: %s", err)
230	247	}
231	248
232	249	readyTimeout := time.After(u.opts.UpgradeTimeout)

237	254
238	255	case err := <-child.result:
239	256	if err == nil {
240		return nil, errors.Errorf("child %s exited", child)
	257	return nil, fmt.Errorf("child %s exited", child)
241	258	}
242		return nil, errors.Wrapf(err, "child %s exited", child)
	259	return nil, fmt.Errorf("child %s exited: %s", child, err)
243	260
244	261	case <-u.stopC:
245	262	child.Kill()

247	264
248	265	case <-readyTimeout:
249	266	child.Kill()
250		return nil, errors.Errorf("new child %s timed out", child)
	267	return nil, fmt.Errorf("new child %s timed out", child)
251	268
252	269	case file := <-child.ready:
253	270	return file, nil

266	283
267	284	func writePIDFile(path string) error {
268	285	dir, file := filepath.Split(path)
	286
	287	// if dir is empty, the user probably specified just the name
	288	// of the pid file expecting it to be created in the current work directory
	289	if dir == "" {
	290	dir = initialWD
	291	}
	292
	293	if dir == "" {
	294	return errors.New("empty initial working directory")
	295	}
	296
269	297	fh, err := ioutil.TempFile(dir, file)
270	298	if err != nil {
271	299	return err

281	309
282	310	return os.Rename(fh.Name(), path)
283	311	}
	312
	313	// Check if this is a supported OS.
	314	// That is currently all Unix-like OS's.
	315	// At the moment, we assume that is everything except Windows.
	316	func isSupportedOS() bool {
	317	return runtime.GOOS != "windows"
	318	}

+124

-74

upgrader_test.go less more

2	2	import (
3	3	"bytes"
4	4	"context"
5		"encoding/binary"
	5	"encoding/gob"
6	6	"errors"
7	7	"fmt"
8	8	"io"

63	63
64	64	func TestMain(m *testing.M) {
65	65	upg, err := New(Options{})
	66	if errors.Is(err, ErrNotSupported) {
	67	fmt.Fprintln(os.Stderr, "Skipping tests, OS is not supported")
	68	os.Exit(0)
	69	}
66	70	if err != nil {
67	71	panic(err)
68	72	}

72	76	os.Exit(m.Run())
73	77	}
74	78
75		pid, err := upg.Fds.File("pid")
76		if err != nil {
77		panic(err)
78		}
79
80		if pid != nil {
81		buf := make([]byte, 8)
82		binary.LittleEndian.PutUint64(buf, uint64(os.Getpid()))
83		pid.Write(buf)
84		pid.Close()
85		}
86
87		parent, err := upg.Fds.File("hasParent")
88		if err != nil {
89		panic(err)
90		}
91
92		if parent != nil {
93		if _, err := io.WriteString(parent, fmt.Sprint(upg.HasParent())); err != nil {
94		panic(err)
95		}
96		parent.Close()
	79	if err := childProcess(upg); err != nil {
	80	fmt.Fprintf(os.Stderr, "Error: %s\n", err)
	81	os.Exit(1)
	82	}
	83	}
	84
	85	type childState struct {
	86	PID int
	87	}
	88
	89	// Used by Benchmark and TestUpgraderOnOS
	90	func childProcess(upg *Upgrader) error {
	91	if !upg.HasParent() {
	92	return errors.New("Upgrader doesn't recognize parent")
	93	}
	94
	95	wState, err := upg.Fds.File("wState")
	96	if err != nil {
	97	return err
	98	}
	99	if wState != nil {
	100	state := &childState{
	101	PID: os.Getpid(),
	102	}
	103	if err := gob.NewEncoder(wState).Encode(state); err != nil {
	104	return err
	105	}
	106	wState.Close()
97	107	}
98	108
99	109	for _, name := range names {
100	110	file, err := upg.Fds.File(name)
101	111	if err != nil {
102		panic(err)
	112	return fmt.Errorf("can't get file %s: %s", name, err)
103	113	}
104	114	if file == nil {
105	115	continue
106	116	}
107	117	if _, err := io.WriteString(file, name); err != nil {
108		panic(err)
109		}
110		}
111
	118	return fmt.Errorf("can't write to %s: %s", name, err)
	119	}
	120	file.Close()
	121	}
	122
	123	rExit, err := upg.Fds.File("rExit")
	124	if err != nil {
	125	return err
	126	}
	127
	128	// Ready closes all inherited but unused files.
112	129	if err := upg.Ready(); err != nil {
113		panic(err)
114		}
	130	return fmt.Errorf("can't signal ready: %s", err)
	131	}
	132
	133	// Block until the parent is done with us. Returning an
	134	// error here won't make the parent fail, so don't bother.
	135	if rExit != nil {
	136	var b [1]byte
	137	rExit.Read(b[:])
	138	}
	139
	140	return nil
115	141	}
116	142
117	143	func TestUpgraderOnOS(t *testing.T) {

121	147	}
122	148	defer u.Stop()
123	149
124		rPid, wPid, err := os.Pipe()
125		if err != nil {
126		t.Fatal(err)
127		}
128		defer rPid.Close()
129
130		if err := u.Fds.AddFile("pid", wPid); err != nil {
131		t.Fatal(err)
132		}
133		wPid.Close()
134
135		rHasParent, wHasParent, err := os.Pipe()
136		if err != nil {
137		t.Fatal(err)
138		}
139		defer rHasParent.Close()
140
141		if err := u.Fds.AddFile("hasParent", wHasParent); err != nil {
142		t.Fatal(err)
143		}
144		wHasParent.Close()
	150	pipe := func() (r, w *os.File) {
	151	t.Helper()
	152
	153	r, w, err := os.Pipe()
	154	if err != nil {
	155	t.Fatal(err)
	156	}
	157	return r, w
	158	}
	159
	160	addPipe := func(name string, file *os.File) {
	161	t.Helper()
	162
	163	if err := u.Fds.AddFile(name, file); err != nil {
	164	t.Fatal(err)
	165	}
	166	file.Close()
	167	}
	168
	169	rState, wState := pipe()
	170	defer rState.Close()
	171
	172	addPipe("wState", wState)
	173
	174	rExit, wExit := pipe()
	175	defer wExit.Close()
	176
	177	addPipe("rExit", rExit)
145	178
146	179	var readers []*os.File
147	180	defer func() {

151	184	}()
152	185
153	186	for _, name := range names {
154		r, w, err := os.Pipe()
155		if err != nil {
156		t.Fatal(err)
157		}
	187	r, w := pipe()
	188	addPipe(name, w)
158	189	readers = append(readers, r)
159
160		if err := u.Fds.AddFile(name, w); err != nil {
161		t.Fatal(err)
162		}
163		w.Close()
164	190	}
165	191
166	192	if err := u.Upgrade(); err == nil {

179	205	}
180	206	}
181	207
	208	// Tell child it's OK to exit now.
	209	wExit.Close()
	210
182	211	// Close copies of write pipes, so that
183	212	// reads below return EOF.
184	213	u.Stop()
185	214
186		buf := make([]byte, 8)
187		if _, err := rPid.Read(buf); err != nil {
188		t.Fatal(err)
189		}
190
191		if int(binary.LittleEndian.Uint64(buf)) == os.Getpid() {
	215	var state childState
	216	if err := gob.NewDecoder(rState).Decode(&state); err != nil {
	217	t.Fatal("Can't decode state from child:", err)
	218	}
	219
	220	if state.PID == os.Getpid() {
192	221	t.Error("Child did not execute in new process")
193		}
194
195		hasParentBytes, err := ioutil.ReadAll(rHasParent)
196		if err != nil {
197		t.Fatal(err)
198		}
199		if !bytes.Equal(hasParentBytes, []byte("true")) {
200		t.Fatal("Child did not recognize parent")
201	222	}
202	223
203	224	for i, name := range names {

409	430	}
410	431	defer fh.Close()
411	432
	433	var pid int
	434	if _, err := fmt.Fscan(fh, &pid); err != nil {
	435	t.Fatal("Can't read PID:", err)
	436	}
	437
	438	if pid != os.Getpid() {
	439	t.Error("PID doesn't match")
	440	}
	441	}
	442
	443	func TestWritePidFileWithoutPath(t *testing.T) {
	444	pidFile := "tableflip-test.pid"
	445
	446	err := writePIDFile(pidFile)
	447	if err != nil {
	448	t.Fatal("Could not write pidfile:", err)
	449	}
	450	defer os.Remove(pidFile)
	451
	452	// lets see if we are able to read the file back
	453	fh, err := os.Open(pidFile)
	454	if err != nil {
	455	t.Fatal("PID file doesn't exist:", err)
	456	}
	457	defer fh.Close()
	458
	459	// just to be sure: check the pid for correctness
	460	// if something failed at a previous run we could be reading
	461	// a bogus pidfile
412	462	var pid int
413	463	if _, err := fmt.Fscan(fh, &pid); err != nil {
414	464	t.Fatal("Can't read PID:", err)