src/coordinator/UnixSocketIo.bend

mylsm-lsm-store@0.6.0.1 · 250 lines · 32 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/coordinator/UnixSocketPolicy.bend as Policy
3import mylsm-lsm-store@0.6.0.1/src/coordinator/UnixSocketTypes.bend as SocketTypes
4
5# Represents an unavailable socket primitive or another raw host failure.
6type Error is Data:
7  UnsupportedPlatform{}
8  Transient{code: U32}
9  Host{code: U32, message: String}
10
11# Carries the state for PollStatus.
12type PollStatus is Data:
13  Readable{}
14  ReadableFailed{}
15  Idle{}
16  Failed{}
17
18# Classifies the platform-specific unsupported syscall codes.
19def unsupported_code(+code: U32) -> Bool:
20  U32.is_eq(code, 95) || U32.is_eq(code, 45) ||
21    U32.is_eq(code, 38) || U32.is_eq(code, 78)
22
23# Converts one host code and message to the socket error domain.
24def classify.mapping(
25  classification: Policy.Classification,
26  message: String
27) -> Error:
28  match classification:
29    case Policy.UnsupportedPlatform{}: UnsupportedPlatform{}
30    case Policy.Transient{code}: Transient{code}
31    case Policy.Host{code}: Host{code, message}
32
33# Converts a pure platform classification into the socket error surface.
34def classify(pair: U32 & String) -> Error:
35  match pair:
36    case (code, message):
37      classify.mapping(Policy.classify_code(code), message)
38
39# Preserves ordinary host errors and types unsupported platforms.
40def classify_result(
41  result: Result<&1, &1, U32 & String, Unit>
42) -> Result<&1, &1, Error, Unit>:
43  match result:
44    case Done{Unit{}}: Done{Unit{}}
45    case Fail{error}: Fail{classify(error)}
46
47# Rewrites the error domain without consuming the socket descriptor.
48def classify_file_unit(
49  pair: File & Result<&1, &1, U32 & String, Unit>
50) -> File & Result<&1, &1, Error, Unit>:
51  match pair:
52    case (file, result): (file, classify_result(result))
53
54# Rewrites an accepted descriptor's error domain.
55def classify_file_result(
56  pair: File & Result<&1, &1, U32 & String, File>
57) -> File & Result<&1, &1, Error, File>:
58  match pair:
59    case (server, Done{client}): (server, Done{client})
60    case (server, Fail{error}): (server, Fail{classify(error)})
61
62# Handles classify poll result.
63def classify_poll_result(
64  result: Result<&1, &1, U32 & String, U32>
65) -> Result<&1, &1, Error, U32>:
66  match result:
67    case Done{ready}: Done{ready}
68    case Fail{error}: Fail{classify(error)}
69
70def poll.status.classified(status: Policy.PollStatus) -> PollStatus:
71  match status:
72    case Policy.Readable{}: Readable{}
73    case Policy.ReadableFailed{}: ReadableFailed{}
74    case Policy.Idle{}: Idle{}
75    case Policy.Failed{}: Failed{}
76
77def poll.status(flags: U32) -> PollStatus:
78  poll.status.classified(Policy.poll_status(flags))
79
80def poll.failure.retryable(action: Policy.PollFailureAction) -> Bool:
81  match action:
82    case Policy.Retry{}: True{}
83    case Policy.Stop{}: False{}
84
85def poll.error.retryable(error: Error) -> Bool:
86  match error:
87    case Transient{code}: poll.failure.retryable(Policy.poll.failure(code))
88    case _: False{}
89
90# Rewrites a socket-creation error into the typed platform domain.
91def classify_created(
92  result: Result<&1, &1, U32 & String, File>
93) -> Result<&1, &1, Error, File>:
94  match result:
95    case Done{file}: Done{file}
96    case Fail{error}: Fail{classify(error)}
97
98# Creates one stream socket through the paired host effects.
99def unix_socket_create() -> IO(Result<&1, &1, U32 & String, File>):
100  import "../effs/unix_socket_create.c"
101  import "../effs/unix_socket_create.js"
102
103# Creates one stream socket and types unsupported-platform failures.
104def create() -> IO(Result<&1, &1, Error, File>):
105  do IO<Result<&1, &1, Error, File>>:
106    result : Result<&1, &1, U32 & String, File> <- unix_socket_create()
107    return classify_created(result)
108
109# Binds one socket descriptor to an already validated local endpoint path.
110def unix_socket_bind(
111  server: File,
112  path: String
113) -> IO(File & Result<&1, &1, U32 & String, Unit>):
114  import "../effs/unix_socket_bind.c"
115  import "../effs/unix_socket_bind.js"
116
117# Binds one socket descriptor to an already validated local endpoint path.
118def bind(
119  server: File,
120  path: String
121) -> IO(File & Result<&1, &1, Error, Unit>):
122  do IO<File & Result<&1, &1, Error, Unit>>:
123    pair : File & Result<&1, &1, U32 & String, Unit> <-
124      unix_socket_bind(server, path)
125    return classify_file_unit(pair)
126
127# Starts listening on one bound socket descriptor.
128def unix_socket_listen(
129  server: File,
130  backlog: U32
131) -> IO(File & Result<&1, &1, U32 & String, Unit>):
132  import "../effs/unix_socket_listen.c"
133  import "../effs/unix_socket_listen.js"
134
135# Starts listening on one bound socket descriptor.
136def listen(
137  server: File,
138  backlog: U32
139) -> IO(File & Result<&1, &1, Error, Unit>):
140  do IO<File & Result<&1, &1, Error, Unit>>:
141    pair : File & Result<&1, &1, U32 & String, Unit> <-
142      unix_socket_listen(server, backlog)
143    return classify_file_unit(pair)
144
145# Accepts one queued connection and preserves the listening descriptor.
146def unix_socket_accept(
147  server: File
148) -> IO(File & Result<&1, &1, U32 & String, File>):
149  import "../effs/unix_socket_accept.c"
150  import "../effs/unix_socket_accept.js"
151
152# Accepts one queued connection and preserves the listening descriptor.
153def accept(
154  server: File
155) -> IO(File & Result<&1, &1, Error, File>):
156  do IO<File & Result<&1, &1, Error, File>>:
157    pair : File & Result<&1, &1, U32 & String, File> <-
158      unix_socket_accept(server)
159    return classify_file_result(pair)
160
161# Handles unix socket nonblocking.
162def unix_socket_nonblocking(
163  socket: File
164) -> IO(File & Result<&1, &1, U32 & String, Unit>):
165  import "../effs/unix_socket_nonblocking.c"
166  import "../effs/unix_socket_nonblocking.js"
167
168# Handles nonblocking.
169def nonblocking(
170  socket: File
171) -> IO(File & Result<&1, &1, Error, Unit>):
172  do IO<File & Result<&1, &1, Error, Unit>>:
173    result : File & Result<&1, &1, U32 & String, Unit> <-
174      unix_socket_nonblocking(socket)
175    return classify_file_unit(result)
176
177# Handles unix socket poll.
178def unix_socket_poll(
179  socket: File,
180  events: U32,
181  timeout_ms: U32
182) -> IO(File & Result<&1, &1, U32 & String, U32>):
183  import "../effs/unix_socket_poll.c"
184  import "../effs/unix_socket_poll.js"
185
186# Handles classify poll pair.
187def classify_poll_pair(
188  pair: File & Result<&1, &1, U32 & String, U32>
189) -> File & Result<&1, &1, Error, U32>:
190  match pair:
191    case (file, result): (file, classify_poll_result(result))
192
193# Handles poll.
194def poll(
195  socket: File,
196  events: U32,
197  timeout_ms: U32
198) -> IO(File & Result<&1, &1, Error, U32>):
199  do IO<File & Result<&1, &1, Error, U32>>:
200    pair : File & Result<&1, &1, U32 & String, U32> <-
201      unix_socket_poll(socket, events, timeout_ms)
202    return classify_poll_pair(pair)
203
204# Handles unix socket poll many.
205def unix_socket_poll_many(
206  interests: List<SocketTypes.PollInterest>,
207  timeout_ms: U32
208) -> IO(List<SocketTypes.PollInterest> &
209  Result<&1, &1, U32 & String, List<&2, SocketTypes.ReadyEvent>>):
210  import "../effs/unix_socket_poll_many.c"
211  import "../effs/unix_socket_poll_many.js"
212
213def poll_many.result(
214  pair: List<SocketTypes.PollInterest> & Result<&1, &1, U32 & String, List<&2, SocketTypes.ReadyEvent>>
215) -> List<SocketTypes.PollInterest> &
216  Result<&1, &1, Error, List<&2, SocketTypes.ReadyEvent>>:
217  match pair:
218    case (interests, Done{flags}): (interests, Done{flags})
219    case (interests, Fail{error}): (interests, Fail{classify(error)})
220
221# Handles poll many.
222def poll_many(
223  interests: List<SocketTypes.PollInterest>,
224  timeout_ms: U32
225) -> IO(List<SocketTypes.PollInterest> &
226  Result<&1, &1, Error, List<&2, SocketTypes.ReadyEvent>>):
227  do IO<List<SocketTypes.PollInterest> &
228    Result<&1, &1, Error, List<&2, SocketTypes.ReadyEvent>>>:
229    pair : List<SocketTypes.PollInterest> &
230      Result<&1, &1, U32 & String, List<&2, SocketTypes.ReadyEvent>> <-
231        unix_socket_poll_many(interests, timeout_ms)
232    return poll_many.result(pair)
233
234# Connects one client socket descriptor to a validated local endpoint path.
235def unix_socket_connect(
236  client: File,
237  path: String
238) -> IO(File & Result<&1, &1, U32 & String, Unit>):
239  import "../effs/unix_socket_connect.c"
240  import "../effs/unix_socket_connect.js"
241
242# Connects one client socket descriptor to a validated local endpoint path.
243def connect(
244  client: File,
245  path: String
246) -> IO(File & Result<&1, &1, Error, Unit>):
247  do IO<File & Result<&1, &1, Error, Unit>>:
248    pair : File & Result<&1, &1, U32 & String, Unit> <-
249      unix_socket_connect(client, path)
250    return classify_file_unit(pair)