summaryrefslogtreecommitdiff
path: root/ev_iouring.c
blob: 33fbcdde7c081d33510733d4b9f56ec05c2e1049 (plain)
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
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
/*
 * libev linux io_uring fd activity backend
 *
 * Copyright (c) 2019 Marc Alexander Lehmann <libev@schmorp.de>
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without modifica-
 * tion, are permitted provided that the following conditions are met:
 *
 *   1.  Redistributions of source code must retain the above copyright notice,
 *       this list of conditions and the following disclaimer.
 *
 *   2.  Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *
 * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MER-
 * CHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO
 * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPE-
 * CIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
 * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTH-
 * ERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
 * OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Alternatively, the contents of this file may be used under the terms of
 * the GNU General Public License ("GPL") version 2 or any later version,
 * in which case the provisions of the GPL are applicable instead of
 * the above. If you wish to allow the use of your version of this file
 * only under the terms of the GPL and not to allow others to use your
 * version of this file under the BSD license, indicate your decision
 * by deleting the provisions above and replace them with the notice
 * and other provisions required by the GPL. If you do not delete the
 * provisions above, a recipient may use your version of this file under
 * either the BSD or the GPL.
 */

/*
 * general notes about linux io_uring:
 *
 * a) it's the best interface I have seen so far. on linux.
 * b) best is not necessarily very good.
 * c) it's better than the aio mess, doesn't suffer from the fork problems
 *    of linux aio or epoll and so on and so on. and you could do event stuff
 *    without any syscalls. what's not to like?
 * d) ok, it's vastly more complex, but that's ok, really.
 * e) why 3 mmaps instead of one? one would be more space-efficient,
 *    and I can't see what benefit three would have (other than being
 *    somehow resizable/relocatable, but that's apparently not possible).
 * f) hmm, it's practiclaly undebuggable (gdb can't access the memory, and
      the bizarre way structure offsets are commuinicated makes it hard to
 *    just print the ring buffer heads, even *iff* the memory were visible
 *    in gdb. but then, that's also ok, really.
 * g) well, you cannot specify a timeout when waiting for events. no,
 *    seriously, the interface doesn't support a timeout. never seen _that_
 *    before. sure, you can use a timerfd, but that's another syscall
 *    you could have avoided. overall, this bizarre omission smells
 *    like a ยต-optimisation by the io_uring author for his personal
 *    applications, to the detriment of everybody else who just wants
 *    an event loop. but, umm, ok, if that's all, it could be worse.
 * h) there is a hardcoded limit of 4096 outstanding events. okay,
 *    at least there is no arbitrary low system-wide limit...
 * i) unlike linux aio, you *can* register more then the limit
 *    of fd events, and the kernel will "gracefully" signal an
 *    overflow, after which you could destroy and recreate the kernel
 *    state, a bit bigger, or fall back to e.g. poll. thats not
 *    totally insane, but kind of questions the point a high
 *    performance I/O framework when it doesn't really work
 *    under stress.
 * j) but, oh my! is has exactly the same bugs as the linux aio backend,
 *    where some undocumented poll combinations just fail.
 *    so we need epoll AGAIN as a fallback. AGAIN! epoll!! and of course,
 *    this is completely undocumented, have I mantioned this already?
 * k) overall, the *API* itself is, I dare to say, not a total trainwreck.
 *    the big isuess with it are the bugs requiring epoll, which might
 *    or might not get fixed (do I hold my breath?).
 */

#include <sys/timerfd.h>
#include <sys/mman.h>
#include <poll.h>

#define IOURING_INIT_ENTRIES 32

/*****************************************************************************/
/* syscall wrapdadoop - this section has the raw api/abi definitions */

#include <linux/fs.h>
#include <linux/types.h>

/* mostly directly taken from the kernel or documentation */

struct io_uring_sqe
{
  __u8 opcode;
  __u8 flags;
  __u16 ioprio;
  __s32 fd;
  __u64 off;
  __u64 addr;
  __u32 len;
  union {
    __kernel_rwf_t rw_flags;
    __u32 fsync_flags;
    __u16 poll_events;
    __u32 sync_range_flags;
    __u32 msg_flags;
  };
  __u64 user_data;
  union {
    __u16 buf_index;
    __u64 __pad2[3];
  };
};

struct io_uring_cqe
{
  __u64 user_data;
  __s32 res;
  __u32 flags;
};

struct io_sqring_offsets
{
  __u32 head;
  __u32 tail;
  __u32 ring_mask;
  __u32 ring_entries;
  __u32 flags;
  __u32 dropped;
  __u32 array;
  __u32 resv1;
  __u64 resv2;
};

struct io_cqring_offsets
{
  __u32 head;
  __u32 tail;
  __u32 ring_mask;
  __u32 ring_entries;
  __u32 overflow;
  __u32 cqes;
  __u64 resv[2];
};

struct io_uring_params
{
  __u32 sq_entries;
  __u32 cq_entries;
  __u32 flags;
  __u32 sq_thread_cpu;
  __u32 sq_thread_idle;
  __u32 resv[5];
  struct io_sqring_offsets sq_off;
  struct io_cqring_offsets cq_off;
};

#define IORING_OP_POLL_ADD    6
#define IORING_OP_POLL_REMOVE 7

#define IORING_ENTER_GETEVENTS 0x01

#define IORING_OFF_SQ_RING 0x00000000ULL
#define IORING_OFF_CQ_RING 0x08000000ULL
#define IORING_OFF_SQES	   0x10000000ULL

inline_size
int
evsys_io_uring_setup (unsigned entries, struct io_uring_params *params)
{
  return ev_syscall2 (SYS_io_uring_setup, entries, params);
}

inline_size
int
evsys_io_uring_enter (int fd, unsigned to_submit, unsigned min_complete, unsigned flags, const sigset_t *sig, size_t sigsz)
{
  return ev_syscall6 (SYS_io_uring_enter, fd, to_submit, min_complete, flags, sig, sigsz);
}

/*****************************************************************************/
/* actual backed implementation */

/* we hope that volatile will make the compiler access this variables only once */
#define EV_SQ_VAR(name) *(volatile unsigned *)((char *)iouring_sq_ring + iouring_sq_ ## name)
#define EV_CQ_VAR(name) *(volatile unsigned *)((char *)iouring_cq_ring + iouring_cq_ ## name)

/* the index array */
#define EV_SQ_ARRAY     ((unsigned *)((char *)iouring_sq_ring + iouring_sq_array))

/* the submit/completion queue entries */
#define EV_SQES         ((struct io_uring_sqe *)         iouring_sqes)
#define EV_CQES         ((struct io_uring_cqe *)((char *)iouring_cq_ring + iouring_cq_cqes))

static
struct io_uring_sqe *
iouring_sqe_get (EV_P)
{
  unsigned tail = EV_SQ_VAR (tail);

  if (tail + 1 - EV_SQ_VAR (head) > EV_SQ_VAR (ring_entries))
    {
      /* queue full, flush */
      evsys_io_uring_enter (iouring_fd, iouring_to_submit, 0, 0, 0, 0);
      iouring_to_submit = 0;
    }

  assert (("libev: io_uring queue full after flush", tail + 1 - EV_SQ_VAR (head) <= EV_SQ_VAR (ring_entries)));

  return EV_SQES + (tail & EV_SQ_VAR (ring_mask));
}

inline_size
struct io_uring_sqe *
iouring_sqe_submit (EV_P_ struct io_uring_sqe *sqe)
{
  unsigned idx = sqe - EV_SQES;

  EV_SQ_ARRAY [idx] = idx;
  ECB_MEMORY_FENCE_RELEASE;
  ++EV_SQ_VAR (tail);
  /*ECB_MEMORY_FENCE_RELEASE; /* for the time being we assume this is not needed */
  ++iouring_to_submit;
}

/*****************************************************************************/

/* when the timerfd expires we simply note the fact,
 * as the purpose of the timerfd is to wake us up, nothing else.
 * the next iteration should re-set it.
 */
static void
iouring_tfd_cb (EV_P_ struct ev_io *w, int revents)
{
  iouring_tfd_to = EV_TSTAMP_HUGE;
}

static void
iouring_epoll_cb (EV_P_ struct ev_io *w, int revents)
{
  epoll_poll (EV_A_ 0);
}

/* called for full and partial cleanup */
ecb_cold
static int
iouring_internal_destroy (EV_P)
{
  close (iouring_tfd);
  close (iouring_fd);

  if (iouring_sq_ring != MAP_FAILED) munmap (iouring_sq_ring, iouring_sq_ring_size);
  if (iouring_cq_ring != MAP_FAILED) munmap (iouring_cq_ring, iouring_cq_ring_size);
  if (iouring_sqes    != MAP_FAILED) munmap (iouring_sqes   , iouring_sqes_size   );

  if (ev_is_active (&iouring_epoll_w)) ev_ref (EV_A); ev_io_stop (EV_A_ &iouring_epoll_w);
  if (ev_is_active (&iouring_tfd_w  )) ev_ref (EV_A); ev_io_stop (EV_A_ &iouring_tfd_w  );
}

ecb_cold
static int
iouring_internal_init (EV_P)
{
  struct io_uring_params params = { 0 };

  iouring_to_submit = 0;

  iouring_tfd     = -1;
  iouring_sq_ring = MAP_FAILED;
  iouring_cq_ring = MAP_FAILED;
  iouring_sqes    = MAP_FAILED;

  for (;;)
    {
      iouring_fd = evsys_io_uring_setup (iouring_entries, &params);

      if (iouring_fd >= 0)
        break; /* yippie */

      if (errno != EINVAL)
        return -1; /* we failed */

      /* EINVAL: lots of possible reasons, but maybe
       * it is because we hit the unqueryable hardcoded size limit
       */

      /* we hit the limit already, give up */
      if (iouring_max_entries)
        return -1;

      /* first time we hit EINVAL? assume we hit the limit, so go back and retry */
      iouring_entries >>= 1;
      iouring_max_entries = iouring_entries;
    }

  iouring_sq_ring_size = params.sq_off.array + params.sq_entries * sizeof (unsigned);
  iouring_cq_ring_size = params.cq_off.cqes  + params.cq_entries * sizeof (struct io_uring_cqe);
  iouring_sqes_size    =                       params.sq_entries * sizeof (struct io_uring_sqe);

  iouring_sq_ring = mmap (0, iouring_sq_ring_size, PROT_READ | PROT_WRITE,
                          MAP_SHARED | MAP_POPULATE, iouring_fd, IORING_OFF_SQ_RING);
  iouring_cq_ring = mmap (0, iouring_cq_ring_size, PROT_READ | PROT_WRITE,
                          MAP_SHARED | MAP_POPULATE, iouring_fd, IORING_OFF_CQ_RING);
  iouring_sqes    = mmap (0, iouring_sqes_size, PROT_READ | PROT_WRITE,
                          MAP_SHARED | MAP_POPULATE, iouring_fd, IORING_OFF_SQES);

  if (iouring_sq_ring == MAP_FAILED || iouring_cq_ring == MAP_FAILED || iouring_sqes == MAP_FAILED)
    return -1;

  iouring_sq_head         = params.sq_off.head;
  iouring_sq_tail         = params.sq_off.tail;
  iouring_sq_ring_mask    = params.sq_off.ring_mask;
  iouring_sq_ring_entries = params.sq_off.ring_entries;
  iouring_sq_flags        = params.sq_off.flags;
  iouring_sq_dropped      = params.sq_off.dropped;
  iouring_sq_array        = params.sq_off.array;

  iouring_cq_head         = params.cq_off.head;
  iouring_cq_tail         = params.cq_off.tail;
  iouring_cq_ring_mask    = params.cq_off.ring_mask;
  iouring_cq_ring_entries = params.cq_off.ring_entries;
  iouring_cq_overflow     = params.cq_off.overflow;
  iouring_cq_cqes         = params.cq_off.cqes;

  iouring_tfd = timerfd_create (CLOCK_MONOTONIC, TFD_CLOEXEC);

  if (iouring_tfd < 0)
    return iouring_tfd;

  iouring_tfd_to = EV_TSTAMP_HUGE;

  return 0;
}

ecb_cold
static void
iouring_fork (EV_P)
{
  iouring_internal_destroy (EV_A);

  while (iouring_internal_init (EV_A) < 0)
    ev_syserr ("(libev) io_uring_setup");

  /* forking epoll should also effectively unregister all fds from the backend */
  epoll_fork (EV_A);
  /* epoll_fork already did this. hopefully */
  /*fd_rearm_all (EV_A);*/

  ev_io_stop  (EV_A_ &iouring_epoll_w);
  ev_io_set   (EV_A_ &iouring_epoll_w, backend_fd, EV_READ);
  ev_io_start (EV_A_ &iouring_epoll_w);

  ev_io_stop  (EV_A_ &iouring_tfd_w);
  ev_io_set   (EV_A_ &iouring_tfd_w, iouring_tfd, EV_READ);
  ev_io_start (EV_A_ &iouring_tfd_w);
}

/*****************************************************************************/

static void
iouring_modify (EV_P_ int fd, int oev, int nev)
{
  if (ecb_expect_false (anfds [fd].eflags))
    {
      /* we handed this fd over to epoll, so undo this first */
      /* we do it manually because the optimisations on epoll_modify won't do us any good */
      epoll_ctl (iouring_fd, EPOLL_CTL_DEL, fd, 0);
      anfds [fd].eflags = 0;
      oev = 0;
    }

  if (oev)
    {
      /* we assume the sqe's are all "properly" initialised */
      struct io_uring_sqe *sqe = iouring_sqe_get (EV_A);
      sqe->opcode    = IORING_OP_POLL_REMOVE;
      sqe->fd        = fd;
      sqe->user_data = -1;
      iouring_sqe_submit (EV_A_ sqe);

      /* increment generation counter to avoid handling old events */
      ++anfds [fd].egen;
    }

  if (nev)
    {
      struct io_uring_sqe *sqe = iouring_sqe_get (EV_A);
      sqe->opcode      = IORING_OP_POLL_ADD;
      sqe->fd          = fd;
      sqe->user_data   = (uint32_t)fd | ((__u64)(uint32_t)anfds [fd].egen << 32);
      sqe->poll_events =
        (nev & EV_READ ? POLLIN : 0)
        | (nev & EV_WRITE ? POLLOUT : 0);
      iouring_sqe_submit (EV_A_ sqe);
    }
}

inline_size
void
iouring_tfd_update (EV_P_ ev_tstamp timeout)
{
  ev_tstamp tfd_to = mn_now + timeout;

  /* we assume there will be many iterations per timer change, so
   * we only re-set the timerfd when we have to because its expiry
   * is too late.
   */
  if (ecb_expect_false (tfd_to < iouring_tfd_to))
    {
       struct itimerspec its;

       iouring_tfd_to = tfd_to;
       EV_TS_SET (its.it_interval, 0.);
       EV_TS_SET (its.it_value, tfd_to);

       if (timerfd_settime (iouring_tfd, TFD_TIMER_ABSTIME, &its, 0) < 0)
         assert (("libev: iouring timerfd_settime failed", 0));
    }
}

inline_size
void
iouring_process_cqe (EV_P_ struct io_uring_cqe *cqe)
{
  int      fd  = cqe->user_data & 0xffffffffU;
  uint32_t gen = cqe->user_data >> 32;
  int      res = cqe->res;

  /* ignore fd removal events, if there are any. TODO: verify */
  if (cqe->user_data == (__u64)-1)
    abort ();//D

  assert (("libev: io_uring fd must be in-bounds", fd >= 0 && fd < anfdmax));

  /* documentation lies, of course. the result value is NOT like
   * normal syscalls, but like linux raw syscalls, i.e. negative
   * error numbers. fortunate, as otherwise there would be no way
   * to get error codes at all. still, why not document this?
   */

  /* ignore event if generation doesn't match */
  /* this should actually be very rare */
  if (ecb_expect_false (gen != (uint32_t)anfds [fd].egen))
    return;

  if (ecb_expect_false (res < 0))
    {
      if (res == -EINVAL)
        {
          /* we assume this error code means the fd/poll combination is buggy
           * and fall back to epoll.
           * this error code might also indicate a bug, but the kernel doesn't
           * distinguish between those two conditions, so... sigh...
           */

          epoll_modify (EV_A_ fd, 0, anfds [fd].events);
        }
      else if (res == -EBADF)
        {
          assert (("libev: event loop rejected bad fd", res != -EBADF));
          fd_kill (EV_A_ fd);
        }
      else
        {
          errno = -res;
          ev_syserr ("(libev) IORING_OP_POLL_ADD");
        }

      return;
    }

  /* feed events, we do not expect or handle POLLNVAL */
  fd_event (
    EV_A_
    fd,
    (res & (POLLOUT | POLLERR | POLLHUP) ? EV_WRITE : 0)
    | (res & (POLLIN | POLLERR | POLLHUP) ? EV_READ : 0)
  );

  /* io_uring is oneshot, so we need to re-arm the fd next iteration */
  /* this also means we usually have to do at least one syscall per iteration */
  anfds [fd].events = 0;
  fd_change (EV_A_ fd, EV_ANFD_REIFY);
}

/* called when the event queue overflows */
ecb_cold
static void
iouring_overflow (EV_P)
{
  /* we have two options, resize the queue (by tearing down
   * everything and recreating it, or living with it
   * and polling.
   * we implement this by resizing tghe queue, and, if that fails,
   * we just recreate the state on every failure, which
   * kind of is a very inefficient poll.
   * one danger is, due to the bios toward lower fds,
   * we will only really get events for those, so
   * maybe we need a poll() fallback, after all.
   */
  /*EV_CQ_VAR (overflow) = 0;*/ /* need to do this if we keep the state and poll manually */

  fd_rearm_all (EV_A);

  /* we double the size until we hit the hard-to-probe maximum */
  if (!iouring_max_entries)
    {
      iouring_entries <<= 1;
      iouring_fork (EV_A);
    }
  else
    {
      /* we hit the kernel limit, we should fall back to something else.
       * we can either poll() a few times and hope for the best,
       * poll always, or switch to epoll.
       * since we use epoll anyways, go epoll.
       */

      iouring_internal_destroy (EV_A);

      /* this should make it so that on return, we don'T call any uring functions */
      iouring_to_submit = 0;

      for (;;)
        {
          backend = epoll_init (EV_A_ 0);

          if (backend)
            break;

          ev_syserr ("(libev) iouring switch to epoll");
        }
    }
}

/* handle any events in the completion queue, return true if there were any */
static int
iouring_handle_cq (EV_P)
{
  unsigned head, tail, mask;
  
  head = EV_CQ_VAR (head);
  ECB_MEMORY_FENCE_ACQUIRE;
  tail = EV_CQ_VAR (tail);

  if (head == tail)
    return 0;

  /* it can only overflow if we have events, yes, yes? */
  if (ecb_expect_false (EV_CQ_VAR (overflow)))
    {
      iouring_overflow (EV_A);
      return 1;
    }

  mask = EV_CQ_VAR (ring_mask);

  do
    iouring_process_cqe (EV_A_ &EV_CQES [head++ & mask]);
  while (head != tail);

  EV_CQ_VAR (head) = head;
  ECB_MEMORY_FENCE_RELEASE;

  return 1;
}

static void
iouring_poll (EV_P_ ev_tstamp timeout)
{
  /* if we have events, no need for extra syscalls, but we might have to queue events */
  if (iouring_handle_cq (EV_A))
    timeout = EV_TS_CONST (0.);
  else
    /* no events, so maybe wait for some */
    iouring_tfd_update (EV_A_ timeout);

  /* only enter the kernel if we have somethign to submit, or we need to wait */
  if (timeout || iouring_to_submit)
    {
      int res;

      EV_RELEASE_CB;

      res = evsys_io_uring_enter (iouring_fd, iouring_to_submit, 1,
                                  timeout > EV_TS_CONST (0.) ? IORING_ENTER_GETEVENTS : 0, 0, 0);
      iouring_to_submit = 0;

      EV_ACQUIRE_CB;

      if (ecb_expect_false (res < 0))
        if (errno == EINTR)
          /* ignore */;
        else
          ev_syserr ("(libev) iouring setup");
      else
        iouring_handle_cq (EV_A);
    }
}

inline_size
int
iouring_init (EV_P_ int flags)
{
  if (!epoll_init (EV_A_ 0))
    return 0;

  ev_io_init  (&iouring_epoll_w, iouring_epoll_cb, backend_fd, EV_READ);
  ev_set_priority (&iouring_epoll_w, EV_MAXPRI);

  ev_io_init  (&iouring_tfd_w, iouring_tfd_cb, iouring_tfd, EV_READ);
  ev_set_priority (&iouring_tfd_w, EV_MAXPRI);

  iouring_entries     = IOURING_INIT_ENTRIES;
  iouring_max_entries = 0;

  if (iouring_internal_init (EV_A) < 0)
    {
      iouring_internal_destroy (EV_A);
      return 0;
    }

  ev_io_start (EV_A_ &iouring_epoll_w);
  ev_unref (EV_A); /* watcher should not keep loop alive */

  ev_io_start (EV_A_ &iouring_tfd_w);
  ev_unref (EV_A); /* watcher should not keep loop alive */

  backend_modify = iouring_modify;
  backend_poll   = iouring_poll;

  return EVBACKEND_IOURING;
}

inline_size
void
iouring_destroy (EV_P)
{
  iouring_internal_destroy (EV_A);
  epoll_destroy (EV_A);
}