[ViewVC] Diff of: cvs/libev/ev

Comparing libev/ev_linuxaio.c (file contents):
Revision 1.1 by root, Thu Jun 20 22:44:59 2019 UTC vs.
Revision 1.54 by sf-exg, Sun May 14 19:02:31 2023 UTC

…		…
35	* and other provisions required by the GPL. If you do not delete the	35	* and other provisions required by the GPL. If you do not delete the
36	* provisions above, a recipient may use your version of this file under	36	* provisions above, a recipient may use your version of this file under
37	* either the BSD or the GPL.	37	* either the BSD or the GPL.
38	*/	38	*/
39		39
		40	/*
		41	* general notes about linux aio:
		42	*
		43	* a) at first, the linux aio IOCB_CMD_POLL functionality introduced in
		44	* 4.18 looks too good to be true: both watchers and events can be
		45	* batched, and events can even be handled in userspace using
		46	* a ring buffer shared with the kernel. watchers can be canceled
		47	* regardless of whether the fd has been closed. no problems with fork.
		48	* ok, the ring buffer is 200% undocumented (there isn't even a
		49	* header file), but otherwise, it's pure bliss!
		50	* b) ok, watchers are one-shot, so you have to re-arm active ones
		51	* on every iteration. so much for syscall-less event handling,
		52	* but at least these re-arms can be batched, no big deal, right?
		53	* c) well, linux as usual: the documentation lies to you: io_submit
		54	* sometimes returns EINVAL because the kernel doesn't feel like
		55	* handling your poll mask - ttys can be polled for POLLOUT,
		56	* POLLOUT\|POLLIN, but polling for POLLIN fails. just great,
		57	* so we have to fall back to something else (hello, epoll),
		58	* but at least the fallback can be slow, because these are
		59	* exceptional cases, right?
		60	* d) hmm, you have to tell the kernel the maximum number of watchers
		61	* you want to queue when initialising the aio context. but of
		62	* course the real limit is magically calculated in the kernel, and
		63	* is often higher then we asked for. so we just have to destroy
		64	* the aio context and re-create it a bit larger if we hit the limit.
		65	* (starts to remind you of epoll? well, it's a bit more deterministic
		66	* and less gambling, but still ugly as hell).
		67	* e) that's when you find out you can also hit an arbitrary system-wide
		68	* limit. or the kernel simply doesn't want to handle your watchers.
		69	* what the fuck do we do then? you guessed it, in the middle
		70	* of event handling we have to switch to 100% epoll polling. and
		71	* that better is as fast as normal epoll polling, so you practically
		72	* have to use the normal epoll backend with all its quirks.
		73	* f) end result of this train wreck: it inherits all the disadvantages
		74	* from epoll, while adding a number on its own. why even bother to use
		75	* it? because if conditions are right and your fds are supported and you
		76	* don't hit a limit, this backend is actually faster, doesn't gamble with
		77	* your fds, batches watchers and events and doesn't require costly state
		78	* recreates. well, until it does.
		79	* g) all of this makes this backend use almost twice as much code as epoll.
		80	* which in turn uses twice as much code as poll. and that#s not counting
		81	* the fact that this backend also depends on the epoll backend, making
		82	* it three times as much code as poll, or kqueue.
		83	* h) bleah. why can't linux just do kqueue. sure kqueue is ugly, but by now
		84	* it's clear that whatever linux comes up with is far, far, far worse.
		85	*/
		86
40	#include <sys/time.h> /* actually linux/time.h, but we must assume they are compatible */	87	#include <sys/time.h> /* actually linux/time.h, but we must assume they are compatible */
		88	#include <poll.h>
41	#include <linux/aio_abi.h>	89	#include <linux/aio_abi.h>
42		90
43	/* we try to fill 4kn pages exactly.	91	/*****************************************************************************/
44	* the ring buffer header is 32 bytes, every io event is 32 bytes.	92	/* syscall wrapdadoop - this section has the raw api/abi definitions */
45	* the kernel takes the io event number, doubles it, adds 2, adds the ring buffer	93
46	* so the calculation below will use "exactly" 8kB for the ring buffer	94	#include <sys/syscall.h> /* no glibc wrappers */
		95
		96	/* aio_abi.h is not versioned in any way, so we cannot test for its existence */
		97	#define IOCB_CMD_POLL 5
		98
		99	/* taken from linux/fs/aio.c. yup, that's a .c file.
		100	* not only is this totally undocumented, not even the source code
		101	* can tell you what the future semantics of compat_features and
		102	* incompat_features are, or what header_length actually is for.
47	*/	103	*/
48	#define EV_LINUXAIO_DEPTH (256 / 2 - 2 - 1) /* max. number of io events per batch */
49
50	/*****************************************************************************/
51	/* syscall wrapdadoop */
52
53	#include <sys/syscall.h> /* no glibc wrappers */
54
55	/* aio_abi.h is not verioned in any way, so we cannot test for its existance */
56	#define IOCB_CMD_POLL 5
57
58	/* taken from linux/fs/aio.c */
59	#define AIO_RING_MAGIC 0xa10a10a1	104	#define AIO_RING_MAGIC 0xa10a10a1
60	#define AIO_RING_INCOMPAT_FEATURES 0	105	#define EV_AIO_RING_INCOMPAT_FEATURES 0
61	struct aio_ring	106	struct aio_ring
62	{	107	{
63	unsigned id; /* kernel internal index number */	108	unsigned id; /* kernel internal index number */
64	unsigned nr; /* number of io_events */	109	unsigned nr; /* number of io_events */
65	unsigned head; /* Written to by userland or by kernel. */	110	unsigned head; /* Written to by userland or by kernel. */
…		…
71	unsigned header_length; /* size of aio_ring */	116	unsigned header_length; /* size of aio_ring */
72		117
73	struct io_event io_events[0];	118	struct io_event io_events[0];
74	};	119	};
75		120
76	static int	121	inline_size
		122	int
77	ev_io_setup (unsigned nr_events, aio_context_t *ctx_idp)	123	evsys_io_setup (unsigned nr_events, aio_context_t *ctx_idp)
78	{	124	{
79	return syscall (SYS_io_setup, nr_events, ctx_idp);	125	return ev_syscall2 (SYS_io_setup, nr_events, ctx_idp);
80	}	126	}
81		127
82	static int	128	inline_size
		129	int
83	ev_io_destroy (aio_context_t ctx_id)	130	evsys_io_destroy (aio_context_t ctx_id)
84	{	131	{
85	return syscall (SYS_io_destroy, ctx_id);	132	return ev_syscall1 (SYS_io_destroy, ctx_id);
86	}	133	}
87		134
88	static int	135	inline_size
		136	int
89	ev_io_submit (aio_context_t ctx_id, long nr, struct iocb *cbp[])	137	evsys_io_submit (aio_context_t ctx_id, long nr, struct iocb *cbp[])
90	{	138	{
91	return syscall (SYS_io_submit, ctx_id, nr, cbp);	139	return ev_syscall3 (SYS_io_submit, ctx_id, nr, cbp);
92	}	140	}
93		141
94	static int	142	inline_size
		143	int
95	ev_io_cancel (aio_context_t ctx_id, struct iocb cbp, struct io_event result)	144	evsys_io_cancel (aio_context_t ctx_id, struct iocb cbp, struct io_event result)
96	{	145	{
97	return syscall (SYS_io_cancel, ctx_id, cbp, result);	146	return ev_syscall3 (SYS_io_cancel, ctx_id, cbp, result);
98	}	147	}
99		148
100	static int	149	inline_size
		150	int
101	ev_io_getevents (aio_context_t ctx_id, long min_nr, long nr, struct io_event events, struct timespec timeout)	151	evsys_io_getevents (aio_context_t ctx_id, long min_nr, long nr, struct io_event events, struct timespec timeout)
102	{	152	{
103	return syscall (SYS_io_getevents, ctx_id, min_nr, nr, events, timeout);	153	return ev_syscall5 (SYS_io_getevents, ctx_id, min_nr, nr, events, timeout);
104	}	154	}
105
106	typedef void (ev_io_cb) (long nr, struct io_event events);
107		155
108	/*****************************************************************************/	156	/*****************************************************************************/
109	/* actual backed implementation */	157	/* actual backed implementation */
110		158
111	/* two iocbs for every fd, one for read, one for write */	159	ecb_cold
		160	static int
		161	linuxaio_nr_events (EV_P)
		162	{
		163	/* we start with 16 iocbs and incraese from there
		164	* that's tiny, but the kernel has a rather low system-wide
		165	* limit that can be reached quickly, so let's be parsimonious
		166	* with this resource.
		167	* Rest assured, the kernel generously rounds up small and big numbers
		168	* in different ways (but doesn't seem to charge you for it).
		169	* The 15 here is because the kernel usually has a power of two as aio-max-nr,
		170	* and this helps to take advantage of that limit.
		171	*/
		172
		173	/* we try to fill 4kB pages exactly.
		174	* the ring buffer header is 32 bytes, every io event is 32 bytes.
		175	* the kernel takes the io requests number, doubles it, adds 2
		176	* and adds the ring buffer.
		177	* the way we use this is by starting low, and then roughly doubling the
		178	* size each time we hit a limit.
		179	*/
		180
		181	int requests = 15 << linuxaio_iteration;
		182	int one_page = (4096
		183	/ sizeof (struct io_event) ) / 2; /* how many fit into one page */
		184	int first_page = ((4096 - sizeof (struct aio_ring))
		185	/ sizeof (struct io_event) - 2) / 2; /* how many fit into the first page */
		186
		187	/* if everything fits into one page, use count exactly */
		188	if (requests > first_page)
		189	/* otherwise, round down to full pages and add the first page */
		190	requests = requests / one_page * one_page + first_page;
		191
		192	return requests;
		193	}
		194
		195	/* we use out own wrapper structure in case we ever want to do something "clever" */
112	typedef struct aniocb	196	typedef struct aniocb
113	{	197	{
114	struct iocb io;	198	struct iocb io;
115	/int inuse;/	199	/int inuse;/
116	} *ANIOCBP;	200	} *ANIOCBP;
117		201
118	inline_size	202	inline_size
119	void	203	void
120	linuxaio_array_needsize_iocbp (ANIOCBP *base, int count)	204	linuxaio_array_needsize_iocbp (ANIOCBP *base, int offset, int count)
121	{	205	{
122	/* TODO: quite the overhead to allocate every iocb separately */
123	while (count--)	206	while (count--)
124	{	207	{
		208	/* TODO: quite the overhead to allocate every iocb separately, maybe use our own allocator? */
125	base = (ANIOCBP)ev_malloc (sizeof (*base));	209	ANIOCBP iocb = (ANIOCBP)ev_malloc (sizeof (*iocb));
		210
		211	/* full zero initialise is probably not required at the moment, but
		212	* this is not well documented, so we better do it.
		213	*/
126	memset (base, 0, sizeof (*base));	214	memset (iocb, 0, sizeof (*iocb));
127	/* would be nice to initialize fd/data as well */	215
128	(*base)->io.aio_lio_opcode = IOCB_CMD_POLL;	216	iocb->io.aio_lio_opcode = IOCB_CMD_POLL;
129	++base;	217	iocb->io.aio_fildes = offset;
130	}
131	}
132		218
		219	base [offset++] = iocb;
		220	}
		221	}
		222
		223	ecb_cold
133	static void	224	static void
134	linuxaio_free_iocbp (EV_P)	225	linuxaio_free_iocbp (EV_P)
135	{	226	{
136	while (linuxaio_iocbpmax--)	227	while (linuxaio_iocbpmax--)
137	ev_free (linuxaio_iocbps [linuxaio_iocbpmax]);	228	ev_free (linuxaio_iocbps [linuxaio_iocbpmax]);
138		229
139	linuxaio_iocbpmax = 0;	230	linuxaio_iocbpmax = 0; /* next resize will completely reallocate the array, at some overhead */
140	}	231	}
141		232
142	static void	233	static void
143	linuxaio_modify (EV_P_ int fd, int oev, int nev)	234	linuxaio_modify (EV_P_ int fd, int oev, int nev)
144	{	235	{
145	/* TODO: full zero initialize required? */
146	array_needsize (ANIOCBP, linuxaio_iocbps, linuxaio_iocbpmax, fd + 1, linuxaio_array_needsize_iocbp);	236	array_needsize (ANIOCBP, linuxaio_iocbps, linuxaio_iocbpmax, fd + 1, linuxaio_array_needsize_iocbp);
147	struct aniocb *iocb = linuxaio_iocbps [fd];	237	ANIOCBP iocb = linuxaio_iocbps [fd];
		238	ANFD *anfd = &anfds [fd];
148		239
149	if (iocb->io.aio_buf)	240	if (ecb_expect_false (iocb->io.aio_reqprio < 0))
150	ev_io_cancel (linuxaio_ctx, &iocb->io, (void *)0);	241	{
		242	/* we handed this fd over to epoll, so undo this first */
		243	/* we do it manually because the optimisations on epoll_modify won't do us any good */
		244	epoll_ctl (backend_fd, EPOLL_CTL_DEL, fd, 0);
		245	anfd->emask = 0;
		246	iocb->io.aio_reqprio = 0;
		247	}
		248	else if (ecb_expect_false (iocb->io.aio_buf))
		249	{
		250	/* iocb active, so cancel it first before resubmit */
		251	/* this assumes we only ever get one call per fd per loop iteration */
		252	for (;;)
		253	{
		254	/* on all relevant kernels, io_cancel fails with EINPROGRESS on "success" */
		255	if (ecb_expect_false (evsys_io_cancel (linuxaio_ctx, &iocb->io, (struct io_event *)0) == 0))
		256	break;
		257
		258	if (ecb_expect_true (errno == EINPROGRESS))
		259	break;
		260
		261	/* the EINPROGRESS test is for nicer error message. clumsy. */
		262	if (errno != EINTR)
		263	{
		264	assert (("libev: linuxaio unexpected io_cancel failed", errno != EINTR && errno != EINPROGRESS));
		265	break;
		266	}
		267	}
		268
		269	/* increment generation counter to avoid handling old events */
		270	++anfd->egen;
		271	}
		272
		273	iocb->io.aio_buf = (nev & EV_READ ? POLLIN : 0)
		274	\| (nev & EV_WRITE ? POLLOUT : 0);
151		275
152	if (nev)	276	if (nev)
153	{	277	{
154	iocb->io.aio_data = fd;	278	iocb->io.aio_data = (uint32_t)fd \| ((__u64)(uint32_t)anfd->egen << 32);
155	iocb->io.aio_fildes = fd;
156	iocb->io.aio_buf =
157	(nev & EV_READ ? POLLIN : 0)
158	\| (nev & EV_WRITE ? POLLOUT : 0);
159		279
160	/* queue iocb up for io_submit */	280	/* queue iocb up for io_submit */
161	/* this assumes we only ever get one call per fd per loop iteration */	281	/* this assumes we only ever get one call per fd per loop iteration */
162	++linuxaio_submitcnt;	282	++linuxaio_submitcnt;
163	array_needsize (struct iocb *, linuxaio_submits, linuxaio_submitmax, linuxaio_submitcnt, array_needsize_noinit);	283	array_needsize (struct iocb *, linuxaio_submits, linuxaio_submitmax, linuxaio_submitcnt, array_needsize_noinit);
164	linuxaio_submits [linuxaio_submitcnt - 1] = &iocb->io;	284	linuxaio_submits [linuxaio_submitcnt - 1] = &iocb->io;
165	}	285	}
166	}	286	}
167		287
168	static void	288	static void
		289	linuxaio_epoll_cb (EV_P_ struct ev_io *w, int revents)
		290	{
		291	epoll_poll (EV_A_ 0);
		292	}
		293
		294	inline_speed
		295	void
		296	linuxaio_fd_rearm (EV_P_ int fd)
		297	{
		298	anfds [fd].events = 0;
		299	linuxaio_iocbps [fd]->io.aio_buf = 0;
		300	fd_change (EV_A_ fd, EV_ANFD_REIFY);
		301	}
		302
		303	static void
169	linuxaio_parse_events (EV_P_ struct io_event *ev, int nr)	304	linuxaio_parse_events (EV_P_ struct io_event *ev, int nr)
170	{	305	{
171	while (nr)	306	while (nr)
172	{	307	{
173	int fd = ev->data;	308	int fd = ev->data & 0xffffffff;
		309	uint32_t gen = ev->data >> 32;
174	int res = ev->res;	310	int res = ev->res;
175		311
176	assert (("libev: iocb fd must be in-bounds", fd >= 0 && fd < anfdxmax));	312	assert (("libev: iocb fd must be in-bounds", fd >= 0 && fd < anfdmax));
177		313
178	/* linux aio is oneshot: rearm fd */	314	/* only accept events if generation counter matches */
179	linuxaio_iocbps [fd]->io.aio_buf = 0;	315	if (ecb_expect_true (gen == (uint32_t)anfds [fd].egen))
180	anfds [fd].events = 0;	316	{
181	fd_change (EV_A_ fd, 0);
182
183	/* feed events, we do not expect or handle POLLNVAL */	317	/* feed events, we do not expect or handle POLLNVAL */
184	if (ecb_expect_false (res & POLLNVAL))
185	fd_kill (EV_A_ fd);
186	else
187	fd_event (	318	fd_event (
188	EV_A_	319	EV_A_
189	fd,	320	fd,
190	(res & (POLLOUT \| POLLERR \| POLLHUP) ? EV_WRITE : 0)	321	(res & (POLLOUT \| POLLERR \| POLLHUP) ? EV_WRITE : 0)
191	\| (res & (POLLIN \| POLLERR \| POLLHUP) ? EV_READ : 0)	322	\| (res & (POLLIN \| POLLERR \| POLLHUP) ? EV_READ : 0)
192	);	323	);
		324
		325	/* linux aio is oneshot: rearm fd. TODO: this does more work than strictly needed */
		326	linuxaio_fd_rearm (EV_A_ fd);
		327	}
193		328
194	--nr;	329	--nr;
195	++ev;	330	++ev;
196	}	331	}
197	}	332	}
198		333
199	/* get any events from ringbuffer, return true if any were handled */	334	/* get any events from ring buffer, return true if any were handled */
200	static int	335	static int
201	linuxaio_get_events_from_ring (EV_P)	336	linuxaio_get_events_from_ring (EV_P)
202	{	337	{
203	struct aio_ring ring = (struct aio_ring )linuxaio_ctx;	338	struct aio_ring ring = (struct aio_ring )linuxaio_ctx;
		339	unsigned head, tail;
204		340
		341	/* the kernel reads and writes both of these variables, */
		342	/* as a C extension, we assume that volatile use here */
		343	/* both makes reads atomic and once-only */
		344	head = (volatile unsigned )&ring->head;
205	ECB_MEMORY_FENCE_ACQUIRE;	345	ECB_MEMORY_FENCE_ACQUIRE;
206
207	unsigned head = ring->head;
208	unsigned tail = (volatile unsigned )&ring->tail;	346	tail = (volatile unsigned )&ring->tail;
209		347
210	if (ring->magic != AIO_RING_MAGIC	348	if (head == tail)
211	\|\| ring->incompat_features != AIO_RING_INCOMPAT_FEATURES
212	\|\| ring->header_length != sizeof (struct aio_ring) /* TODO: or use it to find io_event[0]? */
213	\|\| head == tail)
214	return 0;	349	return 0;
215		350
216	/* parse all available events, but only once, to avoid starvation */	351	/* parse all available events, but only once, to avoid starvation */
217	if (tail > head) /* normal case around */	352	if (ecb_expect_true (tail > head)) /* normal case around */
218	linuxaio_parse_events (EV_A_ ring->io_events + head, tail - head);	353	linuxaio_parse_events (EV_A_ ring->io_events + head, tail - head);
219	else
220	{
221	/* wrapped around */	354	else /* wrapped around */
		355	{
222	linuxaio_parse_events (EV_A_ ring->io_events + head, ring->nr - head);	356	linuxaio_parse_events (EV_A_ ring->io_events + head, ring->nr - head);
223	linuxaio_parse_events (EV_A_ ring->io_events, tail);	357	linuxaio_parse_events (EV_A_ ring->io_events, tail);
224	}	358	}
225		359
226	ring->head = tail;	360	ECB_MEMORY_FENCE_RELEASE;
		361	/* as an extension to C, we hope that the volatile will make this atomic and once-only */
		362	(volatile unsigned )&ring->head = tail;
227		363
228	return 1;	364	return 1;
		365	}
		366
		367	inline_size
		368	int
		369	linuxaio_ringbuf_valid (EV_P)
		370	{
		371	struct aio_ring ring = (struct aio_ring )linuxaio_ctx;
		372
		373	return ecb_expect_true (ring->magic == AIO_RING_MAGIC)
		374	&& ring->incompat_features == EV_AIO_RING_INCOMPAT_FEATURES
		375	&& ring->header_length == sizeof (struct aio_ring); /* TODO: or use it to find io_event[0]? */
229	}	376	}
230		377
231	/* read at least one event from kernel, or timeout */	378	/* read at least one event from kernel, or timeout */
232	inline_size	379	inline_size
233	void	380	void
234	linuxaio_get_events (EV_P_ ev_tstamp timeout)	381	linuxaio_get_events (EV_P_ ev_tstamp timeout)
235	{	382	{
236	struct timespec ts;	383	struct timespec ts;
237	struct io_event ioev;	384	struct io_event ioev[8]; /* 256 octet stack space */
238	int res;	385	int want = 1; /* how many events to request */
		386	int ringbuf_valid = linuxaio_ringbuf_valid (EV_A);
239		387
		388	if (ecb_expect_true (ringbuf_valid))
		389	{
		390	/* if the ring buffer has any events, we don't wait or call the kernel at all */
240	if (linuxaio_get_events_from_ring (EV_A))	391	if (linuxaio_get_events_from_ring (EV_A))
241	return;	392	return;
242		393
243	/* no events, so wait for at least one, then poll ring buffer again */	394	/* if the ring buffer is empty, and we don't have a timeout, then don't call the kernel */
244	/* this degraded to one event per loop iteration */	395	if (!timeout)
245	/* if the ring buffer changes layout, but so be it */	396	return;
		397	}
		398	else
		399	/* no ringbuffer, request slightly larger batch */
		400	want = sizeof (ioev) / sizeof (ioev [0]);
246		401
247	ts.tv_sec = (long)timeout;	402	/* no events, so wait for some
248	ts.tv_nsec = (long)((timeout - ts.tv_sec) * 1e9);	403	* for fairness reasons, we do this in a loop, to fetch all events
		404	*/
		405	for (;;)
		406	{
		407	int res;
249		408
		409	EV_RELEASE_CB;
		410
		411	EV_TS_SET (ts, timeout);
250	res = ev_io_getevents (linuxaio_ctx, 1, 1, &ioev, &ts);	412	res = evsys_io_getevents (linuxaio_ctx, 1, want, ioev, &ts);
251		413
		414	EV_ACQUIRE_CB;
		415
252	if (res < 0)	416	if (res < 0)
		417	if (errno == EINTR)
		418	/* ignored, retry */;
		419	else
253	ev_syserr ("(libev) io_getevents");	420	ev_syserr ("(libev) linuxaio io_getevents");
254	else if (res)	421	else if (res)
255	{	422	{
256	/* at least one event received, handle it and any remaining ones in the ring buffer */	423	/* at least one event available, handle them */
257	linuxaio_parse_events (EV_A_ &ioev, 1);	424	linuxaio_parse_events (EV_A_ ioev, res);
		425
		426	if (ecb_expect_true (ringbuf_valid))
		427	{
		428	/* if we have a ring buffer, handle any remaining events in it */
258	linuxaio_get_events_from_ring (EV_A);	429	linuxaio_get_events_from_ring (EV_A);
		430
		431	/* at this point, we should have handled all outstanding events */
		432	break;
		433	}
		434	else if (res < want)
		435	/* otherwise, if there were fewere events than we wanted, we assume there are no more */
		436	break;
		437	}
		438	else
		439	break; /* no events from the kernel, we are done */
		440
		441	timeout = EV_TS_CONST (0.); /* only wait in the first iteration */
259	}	442	}
		443	}
		444
		445	inline_size
		446	int
		447	linuxaio_io_setup (EV_P)
		448	{
		449	linuxaio_ctx = 0;
		450	return evsys_io_setup (linuxaio_nr_events (EV_A), &linuxaio_ctx);
260	}	451	}
261		452
262	static void	453	static void
263	linuxaio_poll (EV_P_ ev_tstamp timeout)	454	linuxaio_poll (EV_P_ ev_tstamp timeout)
264	{	455	{
…		…
266		457
267	/* first phase: submit new iocbs */	458	/* first phase: submit new iocbs */
268		459
269	/* io_submit might return less than the requested number of iocbs */	460	/* io_submit might return less than the requested number of iocbs */
270	/* this is, afaics, only because of errors, but we go by the book and use a loop, */	461	/* this is, afaics, only because of errors, but we go by the book and use a loop, */
271	/* which allows us to pinpoint the errornous iocb */	462	/* which allows us to pinpoint the erroneous iocb */
272	for (submitted = 0; submitted < linuxaio_submitcnt; )	463	for (submitted = 0; submitted < linuxaio_submitcnt; )
273	{	464	{
274	int res = ev_io_submit (linuxaio_ctx, linuxaio_submitcnt - submitted, linuxaio_submits + submitted);	465	int res = evsys_io_submit (linuxaio_ctx, linuxaio_submitcnt - submitted, linuxaio_submits + submitted);
275		466
276	if (res < 0)	467	if (ecb_expect_false (res < 0))
277	if (errno == EAGAIN)	468	if (errno == EINVAL)
278	{	469	{
279	/* This happens when the ring buffer is full, at least. I assume this means	470	/* This happens for unsupported fds, officially, but in my testing,
280	* that the event was queued synchronously during io_submit, and thus	471	* also randomly happens for supported fds. We fall back to good old
281	* the buffer overflowd.	472	* poll() here, under the assumption that this is a very rare case.
282	* In this case, we just try next loop iteration.	473	* See https://lore.kernel.org/patchwork/patch/1047453/ to see
		474	* discussion about such a case (ttys) where polling for POLLIN
		475	* fails but POLLIN\|POLLOUT works.
283	*/	476	*/
284	memcpy (linuxaio_submits, linuxaio_submits + submitted, (linuxaio_submitcnt - submitted) * sizeof (*linuxaio_submits));	477	struct iocb *iocb = linuxaio_submits [submitted];
285	linuxaio_submitcnt -= submitted;	478	epoll_modify (EV_A_ iocb->aio_fildes, 0, anfds [iocb->aio_fildes].events);
286	timeout = 0;	479	iocb->aio_reqprio = -1; /* mark iocb as epoll */
287	break;	480
		481	res = 1; /* skip this iocb - another iocb, another chance */
288	}	482	}
		483	else if (errno == EAGAIN)
		484	{
		485	/* This happens when the ring buffer is full, or some other shit we
		486	* don't know and isn't documented. Most likely because we have too
		487	* many requests and linux aio can't be assed to handle them.
		488	* In this case, we try to allocate a larger ring buffer, freeing
		489	* ours first. This might fail, in which case we have to fall back to 100%
		490	* epoll.
		491	* God, how I hate linux not getting its act together. Ever.
		492	*/
		493	evsys_io_destroy (linuxaio_ctx);
		494	linuxaio_submitcnt = 0;
		495
		496	/* rearm all fds with active iocbs */
		497	{
		498	int fd;
		499	for (fd = 0; fd < linuxaio_iocbpmax; ++fd)
		500	if (linuxaio_iocbps [fd]->io.aio_buf)
		501	linuxaio_fd_rearm (EV_A_ fd);
		502	}
		503
		504	++linuxaio_iteration;
		505	if (linuxaio_io_setup (EV_A) < 0)
		506	{
		507	/* TODO: rearm all and recreate epoll backend from scratch */
		508	/* TODO: might be more prudent? */
		509
		510	/* to bad, we can't get a new aio context, go 100% epoll */
		511	linuxaio_free_iocbp (EV_A);
		512	ev_io_stop (EV_A_ &linuxaio_epoll_w);
		513	ev_ref (EV_A);
		514	linuxaio_ctx = 0;
		515
		516	backend = EVBACKEND_EPOLL;
		517	backend_modify = epoll_modify;
		518	backend_poll = epoll_poll;
		519	}
		520
		521	timeout = EV_TS_CONST (0.);
		522	/* it's easiest to handle this mess in another iteration */
		523	return;
		524	}
		525	else if (errno == EBADF)
		526	{
		527	assert (("libev: event loop rejected bad fd", errno != EBADF));
		528	fd_kill (EV_A_ linuxaio_submits [submitted]->aio_fildes);
		529
		530	res = 1; /* skip this iocb */
		531	}
		532	else if (errno == EINTR) /* not seen in reality, not documented */
		533	res = 0; /* silently ignore and retry */
289	else	534	else
290	/* TODO: we get EAGAIN when the ring buffer is full for some reason */	535	{
291	/* TODO: should we always just try next time? */
292	ev_syserr ("(libev) io_submit");	536	ev_syserr ("(libev) linuxaio io_submit");
		537	res = 0;
		538	}
293		539
294	submitted += res;	540	submitted += res;
295	}	541	}
296		542
297	linuxaio_submitcnt = 0;	543	linuxaio_submitcnt = 0;
…		…
304	inline_size	550	inline_size
305	int	551	int
306	linuxaio_init (EV_P_ int flags)	552	linuxaio_init (EV_P_ int flags)
307	{	553	{
308	/* would be great to have a nice test for IOCB_CMD_POLL instead */	554	/* would be great to have a nice test for IOCB_CMD_POLL instead */
309	if (ev_linux_version () < 0x041200) /* 4.18 introduced IOCB_CMD_POLL */	555	/* also: test some semi-common fd types, such as files and ttys in recommended_backends */
		556	/* 4.18 introduced IOCB_CMD_POLL, 4.19 made epoll work, and we need that */
		557	if (ev_linux_version () < 0x041300)
310	return 0;	558	return 0;
311		559
312	if (ev_io_setup (EV_LINUXAIO_DEPTH, &linuxaio_ctx) < 0)	560	if (!epoll_init (EV_A_ 0))
313	return 0;	561	return 0;
314		562
		563	linuxaio_iteration = 0;
		564
		565	if (linuxaio_io_setup (EV_A) < 0)
		566	{
		567	epoll_destroy (EV_A);
		568	return 0;
		569	}
		570
		571	ev_io_init (&linuxaio_epoll_w, linuxaio_epoll_cb, backend_fd, EV_READ);
		572	ev_set_priority (&linuxaio_epoll_w, EV_MAXPRI);
		573	ev_io_start (EV_A_ &linuxaio_epoll_w);
		574	ev_unref (EV_A); /* watcher should not keep loop alive */
		575
315	backend_modify = linuxaio_modify;	576	backend_modify = linuxaio_modify;
316	backend_poll = linuxaio_poll;	577	backend_poll = linuxaio_poll;
317		578
318	linuxaio_iocbpmax = 0;	579	linuxaio_iocbpmax = 0;
319	linuxaio_iocbps = 0;	580	linuxaio_iocbps = 0;
320		581
321	linuxaio_submits = 0;	582	linuxaio_submits = 0;
…		…
327		588
328	inline_size	589	inline_size
329	void	590	void
330	linuxaio_destroy (EV_P)	591	linuxaio_destroy (EV_P)
331	{	592	{
		593	epoll_destroy (EV_A);
332	linuxaio_free_iocbp (EV_A);	594	linuxaio_free_iocbp (EV_A);
333	ev_io_destroy (linuxaio_ctx);	595	evsys_io_destroy (linuxaio_ctx); /* fails in child, aio context is destroyed */
334	}	596	}
335		597
336	inline_size	598	ecb_cold
337	void	599	static void
338	linuxaio_fork (EV_P)	600	linuxaio_fork (EV_P)
339	{	601	{
340	abort ();//D	602	linuxaio_submitcnt = 0; /* all pointers were invalidated */
341	}	603	linuxaio_free_iocbp (EV_A); /* this frees all iocbs, which is very heavy-handed */
		604	evsys_io_destroy (linuxaio_ctx); /* fails in child, aio context is destroyed */
342		605
		606	linuxaio_iteration = 0; /* we start over in the child */
		607
		608	while (linuxaio_io_setup (EV_A) < 0)
		609	ev_syserr ("(libev) linuxaio io_setup");
		610
		611	/* forking epoll should also effectively unregister all fds from the backend */
		612	epoll_fork (EV_A);
		613	/* epoll_fork already did this. hopefully */
		614	/fd_rearm_all (EV_A);/
		615
		616	ev_io_stop (EV_A_ &linuxaio_epoll_w);
		617	ev_io_set (EV_A_ &linuxaio_epoll_w, backend_fd, EV_READ);
		618	ev_io_start (EV_A_ &linuxaio_epoll_w);
		619	}
		620

Diff Legend

-–
+Removed lines
-+
+Added lines
-<
+Changed lines
->
+Changed lines

Comparing libev/ev_linuxaio.c (file contents): Revision 1.1 by root, Thu Jun 20 22:44:59 2019 UTC vs. Revision 1.54 by sf-exg, Sun May 14 19:02:31 2023 UTC

Diff Legend

Comparing libev/ev_linuxaio.c (file contents):
Revision 1.1 by root, Thu Jun 20 22:44:59 2019 UTC vs.
Revision 1.54 by sf-exg, Sun May 14 19:02:31 2023 UTC