carlos / platform Public

Download this file

1	//go:build unix
2
3	package hostexec
4
5	import (
6	"context"
7	"crypto/rand"
8	"encoding/hex"
9	"encoding/json"
10	"errors"
11	"io"
12	"os"
13	"path/filepath"
14	"time"
15
16	"golang.org/x/sys/unix"
17	)
18
19	type localDir struct {
20	fd int
21	sync func(int) error
22	}
23
24	func openDir(path string) (int, error) {
25	fd, err := unix.Open(path, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0)
26	if err != nil {
27	return -1, err
28	}
29	var st unix.Stat_t
30	if err = unix.Fstat(fd, &st); err != nil {
31	unix.Close(fd)
32	return -1, err
33	}
34	if st.Mode&unix.S_IFMT != unix.S_IFDIR {
35	unix.Close(fd)
36	return -1, ErrState
37	}
38	return fd, nil
39	}
40	func openLocal(dataDir string) (*localDir, error) {
41	fd, err := openDir(filepath.Join(dataDir, ".hostexec"))
42	if err != nil {
43	return nil, err
44	}
45	return &localDir{fd: fd, sync: unix.Fsync}, nil
46	}
47	func (d *localDir) close() { _ = unix.Close(d.fd) }
48	func (d *localDir) read(host string) (State, error) {
49	var s State
50	if _, err := Key(host, "check"); err != nil {
51	return s, err
52	}
53	fd, err := unix.Openat(d.fd, host+".json", unix.O_RDONLY|unix.O_NOFOLLOW|unix.O_CLOEXEC|unix.O_NONBLOCK, 0)
54	if err != nil {
55	return s, err
56	}
57	f := os.NewFile(uintptr(fd), host)
58	defer f.Close()
59	var st unix.Stat_t
60	if err = unix.Fstat(fd, &st); err != nil {
61	return s, err
62	}
63	if st.Mode&unix.S_IFMT != unix.S_IFREG {
64	return s, ErrState
65	}
66	b, err := io.ReadAll(io.LimitReader(f, 65537))
67	if err != nil {
68	return s, err
69	}
70	if len(b) > 65536 {
71	return s, ErrState
72	}
73	if err = decode(b, &s); err != nil {
74	return s, err
75	}
76	if err = s.validate(); err != nil {
77	return s, err
78	}
79	if s.Host != host {
80	return s, ErrState
81	}
82	return s, nil
83	}
84	func (d *localDir) write(s State) error {
85	if err := s.validate(); err != nil {
86	return err
87	}
88	if _, err := d.read(s.Host); err != nil && !errors.Is(err, os.ErrNotExist) {
89	return err
90	}
91	var nonce [16]byte
92	if _, err := rand.Read(nonce[:]); err != nil {
93	return err
94	}
95	tmp := ".tmp-" + hex.EncodeToString(nonce[:])
96	fd, err := unix.Openat(d.fd, tmp, unix.O_WRONLY|unix.O_CREAT|unix.O_EXCL|unix.O_NOFOLLOW|unix.O_CLOEXEC, 0660)
97	if err != nil {
98	return err
99	}
100	defer unix.Unlinkat(d.fd, tmp, 0)
101	f := os.NewFile(uintptr(fd), tmp)
102	defer f.Close()
103	if err = unix.Fchmod(fd, 0660); err != nil {
104	return err
105	}
106	b, err := json.Marshal(s)
107	if err != nil {
108	return err
109	}
110	if _, err = f.Write(append(b, '\n')); err != nil {
111	return err
112	}
113	if err = d.sync(fd); err != nil {
114	return err
115	}
116	if err = f.Close(); err != nil {
117	return err
118	}
119	if err = unix.Renameat(d.fd, tmp, d.fd, s.Host+".json"); err != nil {
120	return err
121	}
122	return d.sync(d.fd)
123	}
124	func (d *localDir) remove(host string) error {
125	if _, err := d.read(host); err != nil {
126	return err
127	}
128	if err := unix.Unlinkat(d.fd, host+".json", 0); err != nil {
129	return err
130	}
131	return d.sync(d.fd)
132	}
133	func tryAcquire(dir, host string) (func(), error) {
134	return tryAcquireNamed(dir, host, ".lock")
135	}
136
137	func tryAcquireLive(dir, host string) (func(), error) {
138	return tryAcquireNamed(dir, host, ".live")
139	}
140
141	func tryAcquireNamed(dir, host, suffix string) (func(), error) {
142	if _, err := Key(host, "check"); err != nil {
143	return nil, err
144	}
145	fd, err := openDir(dir)
146	if err != nil {
147	return nil, err
148	}
149	defer unix.Close(fd)
150	flags := unix.O_RDWR | unix.O_CREAT | unix.O_NOFOLLOW | unix.O_CLOEXEC | unix.O_NONBLOCK
151	lock, err := unix.Openat(fd, host+suffix, flags|unix.O_EXCL, 0660)
152	created := err == nil
153	if errors.Is(err, unix.EEXIST) {
154	lock, err = unix.Openat(fd, host+suffix, flags, 0660)
155	}
156	if err != nil {
157	return nil, err
158	}
159	fail := func(e error) (func(), error) { unix.Close(lock); return nil, e }
160	var st unix.Stat_t
161	if err = unix.Fstat(lock, &st); err != nil {
162	return fail(err)
163	}
164	if st.Mode&unix.S_IFMT != unix.S_IFREG {
165	return fail(ErrState)
166	}
167	if created || st.Mode&07777 != 0660 {
168	if err = unix.Fchmod(lock, 0660); err != nil {
169	return fail(err)
170	}
171	}
172	err = unix.Flock(lock, unix.LOCK_EX|unix.LOCK_NB)
173	if err == nil {
174	return func() { _ = unix.Flock(lock, unix.LOCK_UN); _ = unix.Close(lock) }, nil
175	}
176	if errors.Is(err, unix.EWOULDBLOCK) || errors.Is(err, unix.EAGAIN) {
177	return fail(ErrControllerBusy)
178	}
179	return fail(err)
180	}
181
182	func acquire(ctx context.Context, dir, host string) (func(), error) {
183	for {
184	if err := ctx.Err(); err != nil {
185	return nil, err
186	}
187	unlock, err := tryAcquire(dir, host)
188	if !errors.Is(err, ErrControllerBusy) {
189	return unlock, err
190	}
191	select {
192	case <-ctx.Done():
193	return nil, ctx.Err()
194	case <-time.After(10 * time.Millisecond):
195	}
196	}
197	}
198