src/libsystemd/sd-rtnl/sd-rtnl.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2013 Tom Gundersen <teg@jklm.no>
   7
   8   systemd is free software; you can redistribute it and/or modify it
   9   under the terms of the GNU Lesser General Public License as published by
  10   the Free Software Foundation; either version 2.1 of the License, or
  11   (at your option) any later version.
  12
  13   systemd is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   Lesser General Public License for more details.
  17
  18   You should have received a copy of the GNU Lesser General Public License
  19   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  20 ***/
  21
  22 #include <sys/socket.h>
  23 #include <poll.h>
  24
  25 #include "missing.h"
  26 #include "macro.h"
  27 #include "util.h"
  28 #include "hashmap.h"
  29
  30 #include "sd-rtnl.h"
  31 #include "rtnl-internal.h"
  32 #include "rtnl-util.h"
  33
  34 static int sd_rtnl_new(sd_rtnl **ret) {
  35         _cleanup_rtnl_unref_ sd_rtnl *rtnl = NULL;
  36
  37         assert_return(ret, -EINVAL);
  38
  39         rtnl = new0(sd_rtnl, 1);
  40         if (!rtnl)
  41                 return -ENOMEM;
  42
  43         rtnl->n_ref = REFCNT_INIT;
  44
  45         rtnl->fd = -1;
  46
  47         rtnl->sockaddr.nl.nl_family = AF_NETLINK;
  48
  49         rtnl->original_pid = getpid();
  50
  51         LIST_HEAD_INIT(rtnl->match_callbacks);
  52
  53         /* We guarantee that wqueue always has space for at least
  54          * one entry */
  55         if (!GREEDY_REALLOC(rtnl->wqueue, rtnl->wqueue_allocated, 1))
  56                 return -ENOMEM;
  57
  58         /* We guarantee that the read buffer has at least space for
  59          * a message header */
  60         if (!greedy_realloc((void**)&rtnl->rbuffer, &rtnl->rbuffer_allocated,
  61                             sizeof(struct nlmsghdr), sizeof(uint8_t)))
  62                 return -ENOMEM;
  63
  64         /* Change notification responses have sequence 0, so we must
  65          * start our request sequence numbers at 1, or we may confuse our
  66          * responses with notifications from the kernel */
  67         rtnl->serial = 1;
  68
  69         *ret = rtnl;
  70         rtnl = NULL;
  71
  72         return 0;
  73 }
  74
  75 int sd_rtnl_new_from_netlink(sd_rtnl **ret, int fd) {
  76         _cleanup_rtnl_unref_ sd_rtnl *rtnl = NULL;
  77         socklen_t addrlen;
  78         int r;
  79
  80         assert_return(ret, -EINVAL);
  81
  82         r = sd_rtnl_new(&rtnl);
  83         if (r < 0)
  84                 return r;
  85
  86         addrlen = sizeof(rtnl->sockaddr);
  87
  88         r = getsockname(fd, &rtnl->sockaddr.sa, &addrlen);
  89         if (r < 0)
  90                 return -errno;
  91
  92         rtnl->fd = fd;
  93
  94         *ret = rtnl;
  95         rtnl = NULL;
  96
  97         return 0;
  98 }
  99
 100 static bool rtnl_pid_changed(sd_rtnl *rtnl) {
 101         assert(rtnl);
 102
 103         /* We don't support people creating an rtnl connection and
 104          * keeping it around over a fork(). Let's complain. */
 105
 106         return rtnl->original_pid != getpid();
 107 }
 108
 109 static int rtnl_compute_groups_ap(uint32_t *_groups, unsigned n_groups, va_list ap) {
 110         uint32_t groups = 0;
 111         unsigned i;
 112
 113         for (i = 0; i < n_groups; i++) {
 114                 unsigned group;
 115
 116                 group = va_arg(ap, unsigned);
 117                 assert_return(group < 32, -EINVAL);
 118
 119                 groups |= group ? (1 << (group - 1)) : 0;
 120         }
 121
 122         *_groups = groups;
 123
 124         return 0;
 125 }
 126
 127 static int rtnl_open_fd_ap(sd_rtnl **ret, int fd, unsigned n_groups, va_list ap) {
 128         _cleanup_rtnl_unref_ sd_rtnl *rtnl = NULL;
 129         socklen_t addrlen;
 130         int r, one = 1;
 131
 132         assert_return(ret, -EINVAL);
 133         assert_return(fd >= 0, -EINVAL);
 134
 135         r = sd_rtnl_new(&rtnl);
 136         if (r < 0)
 137                 return r;
 138
 139         r = setsockopt(fd, SOL_SOCKET, SO_PASSCRED, &one, sizeof(one));
 140         if (r < 0)
 141                 return -errno;
 142
 143         r = setsockopt(fd, SOL_NETLINK, NETLINK_PKTINFO, &one, sizeof(one));
 144         if (r < 0)
 145                 return -errno;
 146
 147         r = rtnl_compute_groups_ap(&rtnl->sockaddr.nl.nl_groups, n_groups, ap);
 148         if (r < 0)
 149                 return r;
 150
 151         addrlen = sizeof(rtnl->sockaddr);
 152
 153         r = bind(fd, &rtnl->sockaddr.sa, addrlen);
 154         /* ignore EINVAL to allow opening an already bound socket */
 155         if (r < 0 && errno != EINVAL)
 156                 return -errno;
 157
 158         r = getsockname(fd, &rtnl->sockaddr.sa, &addrlen);
 159         if (r < 0)
 160                 return -errno;
 161
 162         rtnl->fd = fd;
 163
 164         *ret = rtnl;
 165         rtnl = NULL;
 166
 167         return 0;
 168 }
 169
 170 int sd_rtnl_open_fd(sd_rtnl **ret, int fd, unsigned n_groups, ...) {
 171         va_list ap;
 172         int r;
 173
 174         va_start(ap, n_groups);
 175         r = rtnl_open_fd_ap(ret, fd, n_groups, ap);
 176         va_end(ap);
 177
 178         return r;
 179 }
 180
 181 int sd_rtnl_open(sd_rtnl **ret, unsigned n_groups, ...) {
 182         va_list ap;
 183         int fd, r;
 184
 185         fd = socket(PF_NETLINK, SOCK_RAW|SOCK_CLOEXEC|SOCK_NONBLOCK, NETLINK_ROUTE);
 186         if (fd < 0)
 187                 return -errno;
 188
 189         va_start(ap, n_groups);
 190         r = rtnl_open_fd_ap(ret, fd, n_groups, ap);
 191         va_end(ap);
 192
 193         if (r < 0) {
 194                 safe_close(fd);
 195                 return r;
 196         }
 197
 198         return 0;
 199 }
 200
 201 int sd_rtnl_inc_rcvbuf(const sd_rtnl *const rtnl, const int size) {
 202         return fd_inc_rcvbuf(rtnl->fd, size);
 203 }
 204
 205 sd_rtnl *sd_rtnl_ref(sd_rtnl *rtnl) {
 206         assert_return(rtnl, NULL);
 207         assert_return(!rtnl_pid_changed(rtnl), NULL);
 208
 209         if (rtnl)
 210                 assert_se(REFCNT_INC(rtnl->n_ref) >= 2);
 211
 212         return rtnl;
 213 }
 214
 215 sd_rtnl *sd_rtnl_unref(sd_rtnl *rtnl) {
 216         if (!rtnl)
 217                 return NULL;
 218
 219         assert_return(!rtnl_pid_changed(rtnl), NULL);
 220
 221         if (REFCNT_DEC(rtnl->n_ref) == 0) {
 222                 struct match_callback *f;
 223                 unsigned i;
 224
 225                 for (i = 0; i < rtnl->rqueue_size; i++)
 226                         sd_rtnl_message_unref(rtnl->rqueue[i]);
 227                 free(rtnl->rqueue);
 228
 229                 for (i = 0; i < rtnl->rqueue_partial_size; i++)
 230                         sd_rtnl_message_unref(rtnl->rqueue_partial[i]);
 231                 free(rtnl->rqueue_partial);
 232
 233                 for (i = 0; i < rtnl->wqueue_size; i++)
 234                         sd_rtnl_message_unref(rtnl->wqueue[i]);
 235                 free(rtnl->wqueue);
 236
 237                 free(rtnl->rbuffer);
 238
 239                 hashmap_free_free(rtnl->reply_callbacks);
 240                 prioq_free(rtnl->reply_callbacks_prioq);
 241
 242                 sd_event_source_unref(rtnl->io_event_source);
 243                 sd_event_source_unref(rtnl->time_event_source);
 244                 sd_event_source_unref(rtnl->exit_event_source);
 245                 sd_event_unref(rtnl->event);
 246
 247                 while ((f = rtnl->match_callbacks)) {
 248                         LIST_REMOVE(match_callbacks, rtnl->match_callbacks, f);
 249                         free(f);
 250                 }
 251
 252                 safe_close(rtnl->fd);
 253                 free(rtnl);
 254         }
 255
 256         return NULL;
 257 }
 258
 259 static void rtnl_seal_message(sd_rtnl *rtnl, sd_rtnl_message *m) {
 260         assert(rtnl);
 261         assert(!rtnl_pid_changed(rtnl));
 262         assert(m);
 263         assert(m->hdr);
 264
 265         m->hdr->nlmsg_seq = rtnl->serial++;
 266
 267         rtnl_message_seal(m);
 268
 269         return;
 270 }
 271
 272 int sd_rtnl_send(sd_rtnl *nl,
 273                  sd_rtnl_message *message,
 274                  uint32_t *serial) {
 275         int r;
 276
 277         assert_return(nl, -EINVAL);
 278         assert_return(!rtnl_pid_changed(nl), -ECHILD);
 279         assert_return(message, -EINVAL);
 280         assert_return(!message->sealed, -EPERM);
 281
 282         rtnl_seal_message(nl, message);
 283
 284         if (nl->wqueue_size <= 0) {
 285                 /* send directly */
 286                 r = socket_write_message(nl, message);
 287                 if (r < 0)
 288                         return r;
 289                 else if (r == 0) {
 290                         /* nothing was sent, so let's put it on
 291                          * the queue */
 292                         nl->wqueue[0] = sd_rtnl_message_ref(message);
 293                         nl->wqueue_size = 1;
 294                 }
 295         } else {
 296                 /* append to queue */
 297                 if (nl->wqueue_size >= RTNL_WQUEUE_MAX) {
 298                         log_debug("rtnl: exhausted the write queue size (%d)", RTNL_WQUEUE_MAX);
 299                         return -ENOBUFS;
 300                 }
 301
 302                 if (!GREEDY_REALLOC(nl->wqueue, nl->wqueue_allocated, nl->wqueue_size + 1))
 303                         return -ENOMEM;
 304
 305                 nl->wqueue[nl->wqueue_size ++] = sd_rtnl_message_ref(message);
 306         }
 307
 308         if (serial)
 309                 *serial = rtnl_message_get_serial(message);
 310
 311         return 1;
 312 }
 313
 314 int rtnl_rqueue_make_room(sd_rtnl *rtnl) {
 315         assert(rtnl);
 316
 317         if (rtnl->rqueue_size >= RTNL_RQUEUE_MAX) {
 318                 log_debug("rtnl: exhausted the read queue size (%d)", RTNL_RQUEUE_MAX);
 319                 return -ENOBUFS;
 320         }
 321
 322         if (!GREEDY_REALLOC(rtnl->rqueue, rtnl->rqueue_allocated, rtnl->rqueue_size + 1))
 323                 return -ENOMEM;
 324
 325         return 0;
 326 }
 327
 328 int rtnl_rqueue_partial_make_room(sd_rtnl *rtnl) {
 329         assert(rtnl);
 330
 331         if (rtnl->rqueue_partial_size >= RTNL_RQUEUE_MAX) {
 332                 log_debug("rtnl: exhausted the partial read queue size (%d)", RTNL_RQUEUE_MAX);
 333                 return -ENOBUFS;
 334         }
 335
 336         if (!GREEDY_REALLOC(rtnl->rqueue_partial, rtnl->rqueue_partial_allocated,
 337                             rtnl->rqueue_partial_size + 1))
 338                 return -ENOMEM;
 339
 340         return 0;
 341 }
 342
 343 static int dispatch_rqueue(sd_rtnl *rtnl, sd_rtnl_message **message) {
 344         int r;
 345
 346         assert(rtnl);
 347         assert(message);
 348
 349         if (rtnl->rqueue_size <= 0) {
 350                 /* Try to read a new message */
 351                 r = socket_read_message(rtnl);
 352                 if (r <= 0)
 353                         return r;
 354         }
 355
 356         /* Dispatch a queued message */
 357         *message = rtnl->rqueue[0];
 358         rtnl->rqueue_size --;
 359         memmove(rtnl->rqueue, rtnl->rqueue + 1, sizeof(sd_rtnl_message*) * rtnl->rqueue_size);
 360
 361         return 1;
 362 }
 363
 364 static int dispatch_wqueue(sd_rtnl *rtnl) {
 365         int r, ret = 0;
 366
 367         assert(rtnl);
 368
 369         while (rtnl->wqueue_size > 0) {
 370                 r = socket_write_message(rtnl, rtnl->wqueue[0]);
 371                 if (r < 0)
 372                         return r;
 373                 else if (r == 0)
 374                         /* Didn't do anything this time */
 375                         return ret;
 376                 else {
 377                         /* see equivalent in sd-bus.c */
 378                         sd_rtnl_message_unref(rtnl->wqueue[0]);
 379                         rtnl->wqueue_size --;
 380                         memmove(rtnl->wqueue, rtnl->wqueue + 1, sizeof(sd_rtnl_message*) * rtnl->wqueue_size);
 381
 382                         ret = 1;
 383                 }
 384         }
 385
 386         return ret;
 387 }
 388
 389 static int process_timeout(sd_rtnl *rtnl) {
 390         _cleanup_rtnl_message_unref_ sd_rtnl_message *m = NULL;
 391         struct reply_callback *c;
 392         usec_t n;
 393         int r;
 394
 395         assert(rtnl);
 396
 397         c = prioq_peek(rtnl->reply_callbacks_prioq);
 398         if (!c)
 399                 return 0;
 400
 401         n = now(CLOCK_MONOTONIC);
 402         if (c->timeout > n)
 403                 return 0;
 404
 405         r = rtnl_message_new_synthetic_error(-ETIMEDOUT, c->serial, &m);
 406         if (r < 0)
 407                 return r;
 408
 409         assert_se(prioq_pop(rtnl->reply_callbacks_prioq) == c);
 410         hashmap_remove(rtnl->reply_callbacks, &c->serial);
 411
 412         r = c->callback(rtnl, m, c->userdata);
 413         if (r < 0)
 414                 log_debug_errno(r, "sd-rtnl: timedout callback failed: %m");
 415
 416         free(c);
 417
 418         return 1;
 419 }
 420
 421 static int process_reply(sd_rtnl *rtnl, sd_rtnl_message *m) {
 422         struct reply_callback *c;
 423         uint64_t serial;
 424         int r;
 425
 426         assert(rtnl);
 427         assert(m);
 428
 429         if (sd_rtnl_message_is_broadcast(m))
 430                 return 0;
 431
 432         serial = rtnl_message_get_serial(m);
 433         c = hashmap_remove(rtnl->reply_callbacks, &serial);
 434         if (!c)
 435                 return 0;
 436
 437         if (c->timeout != 0)
 438                 prioq_remove(rtnl->reply_callbacks_prioq, c, &c->prioq_idx);
 439
 440         r = c->callback(rtnl, m, c->userdata);
 441         if (r < 0)
 442                 log_debug_errno(r, "sd-rtnl: callback failed: %m");
 443
 444         free(c);
 445
 446         return 1;
 447 }
 448
 449 static int process_match(sd_rtnl *rtnl, sd_rtnl_message *m) {
 450         struct match_callback *c;
 451         uint16_t type;
 452         int r;
 453
 454         assert(rtnl);
 455         assert(m);
 456
 457         r = sd_rtnl_message_get_type(m, &type);
 458         if (r < 0)
 459                 return r;
 460
 461         LIST_FOREACH(match_callbacks, c, rtnl->match_callbacks) {
 462                 if (type == c->type) {
 463                         r = c->callback(rtnl, m, c->userdata);
 464                         if (r != 0) {
 465                                 if (r < 0)
 466                                         log_debug_errno(r, "sd-rtnl: match callback failed: %m");
 467
 468                                 break;
 469                         }
 470                 }
 471         }
 472
 473         return 1;
 474 }
 475
 476 static int process_running(sd_rtnl *rtnl, sd_rtnl_message **ret) {
 477         _cleanup_rtnl_message_unref_ sd_rtnl_message *m = NULL;
 478         int r;
 479
 480         assert(rtnl);
 481
 482         r = process_timeout(rtnl);
 483         if (r != 0)
 484                 goto null_message;
 485
 486         r = dispatch_wqueue(rtnl);
 487         if (r != 0)
 488                 goto null_message;
 489
 490         r = dispatch_rqueue(rtnl, &m);
 491         if (r < 0)
 492                 return r;
 493         if (!m)
 494                 goto null_message;
 495
 496         r = process_reply(rtnl, m);
 497         if (r != 0)
 498                 goto null_message;
 499
 500         r = process_match(rtnl, m);
 501         if (r != 0)
 502                 goto null_message;
 503
 504         if (ret) {
 505                 *ret = m;
 506                 m = NULL;
 507
 508                 return 1;
 509         }
 510
 511         return 1;
 512
 513 null_message:
 514         if (r >= 0 && ret)
 515                 *ret = NULL;
 516
 517         return r;
 518 }
 519
 520 int sd_rtnl_process(sd_rtnl *rtnl, sd_rtnl_message **ret) {
 521         RTNL_DONT_DESTROY(rtnl);
 522         int r;
 523
 524         assert_return(rtnl, -EINVAL);
 525         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
 526         assert_return(!rtnl->processing, -EBUSY);
 527
 528         rtnl->processing = true;
 529         r = process_running(rtnl, ret);
 530         rtnl->processing = false;
 531
 532         return r;
 533 }
 534
 535 static usec_t calc_elapse(uint64_t usec) {
 536         if (usec == (uint64_t) -1)
 537                 return 0;
 538
 539         if (usec == 0)
 540                 usec = RTNL_DEFAULT_TIMEOUT;
 541
 542         return now(CLOCK_MONOTONIC) + usec;
 543 }
 544
 545 static int rtnl_poll(sd_rtnl *rtnl, bool need_more, uint64_t timeout_usec) {
 546         struct pollfd p[1] = {};
 547         struct timespec ts;
 548         usec_t m = USEC_INFINITY;
 549         int r, e;
 550
 551         assert(rtnl);
 552
 553         e = sd_rtnl_get_events(rtnl);
 554         if (e < 0)
 555                 return e;
 556
 557         if (need_more)
 558                 /* Caller wants more data, and doesn't care about
 559                  * what's been read or any other timeouts. */
 560                 e |= POLLIN;
 561         else {
 562                 usec_t until;
 563                 /* Caller wants to process if there is something to
 564                  * process, but doesn't care otherwise */
 565
 566                 r = sd_rtnl_get_timeout(rtnl, &until);
 567                 if (r < 0)
 568                         return r;
 569                 if (r > 0) {
 570                         usec_t nw;
 571                         nw = now(CLOCK_MONOTONIC);
 572                         m = until > nw ? until - nw : 0;
 573                 }
 574         }
 575
 576         if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
 577                 m = timeout_usec;
 578
 579         p[0].fd = rtnl->fd;
 580         p[0].events = e;
 581
 582         r = ppoll(p, 1, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
 583         if (r < 0)
 584                 return -errno;
 585
 586         return r > 0 ? 1 : 0;
 587 }
 588
 589 int sd_rtnl_wait(sd_rtnl *nl, uint64_t timeout_usec) {
 590         assert_return(nl, -EINVAL);
 591         assert_return(!rtnl_pid_changed(nl), -ECHILD);
 592
 593         if (nl->rqueue_size > 0)
 594                 return 0;
 595
 596         return rtnl_poll(nl, false, timeout_usec);
 597 }
 598
 599 static int timeout_compare(const void *a, const void *b) {
 600         const struct reply_callback *x = a, *y = b;
 601
 602         if (x->timeout != 0 && y->timeout == 0)
 603                 return -1;
 604
 605         if (x->timeout == 0 && y->timeout != 0)
 606                 return 1;
 607
 608         if (x->timeout < y->timeout)
 609                 return -1;
 610
 611         if (x->timeout > y->timeout)
 612                 return 1;
 613
 614         return 0;
 615 }
 616
 617 int sd_rtnl_call_async(sd_rtnl *nl,
 618                        sd_rtnl_message *m,
 619                        sd_rtnl_message_handler_t callback,
 620                        void *userdata,
 621                        uint64_t usec,
 622                        uint32_t *serial) {
 623         struct reply_callback *c;
 624         uint32_t s;
 625         int r, k;
 626
 627         assert_return(nl, -EINVAL);
 628         assert_return(m, -EINVAL);
 629         assert_return(callback, -EINVAL);
 630         assert_return(!rtnl_pid_changed(nl), -ECHILD);
 631
 632         r = hashmap_ensure_allocated(&nl->reply_callbacks, &uint64_hash_ops);
 633         if (r < 0)
 634                 return r;
 635
 636         if (usec != (uint64_t) -1) {
 637                 r = prioq_ensure_allocated(&nl->reply_callbacks_prioq, timeout_compare);
 638                 if (r < 0)
 639                         return r;
 640         }
 641
 642         c = new0(struct reply_callback, 1);
 643         if (!c)
 644                 return -ENOMEM;
 645
 646         c->callback = callback;
 647         c->userdata = userdata;
 648         c->timeout = calc_elapse(usec);
 649
 650         k = sd_rtnl_send(nl, m, &s);
 651         if (k < 0) {
 652                 free(c);
 653                 return k;
 654         }
 655
 656         c->serial = s;
 657
 658         r = hashmap_put(nl->reply_callbacks, &c->serial, c);
 659         if (r < 0) {
 660                 free(c);
 661                 return r;
 662         }
 663
 664         if (c->timeout != 0) {
 665                 r = prioq_put(nl->reply_callbacks_prioq, c, &c->prioq_idx);
 666                 if (r > 0) {
 667                         c->timeout = 0;
 668                         sd_rtnl_call_async_cancel(nl, c->serial);
 669                         return r;
 670                 }
 671         }
 672
 673         if (serial)
 674                 *serial = s;
 675
 676         return k;
 677 }
 678
 679 int sd_rtnl_call_async_cancel(sd_rtnl *nl, uint32_t serial) {
 680         struct reply_callback *c;
 681         uint64_t s = serial;
 682
 683         assert_return(nl, -EINVAL);
 684         assert_return(serial != 0, -EINVAL);
 685         assert_return(!rtnl_pid_changed(nl), -ECHILD);
 686
 687         c = hashmap_remove(nl->reply_callbacks, &s);
 688         if (!c)
 689                 return 0;
 690
 691         if (c->timeout != 0)
 692                 prioq_remove(nl->reply_callbacks_prioq, c, &c->prioq_idx);
 693
 694         free(c);
 695         return 1;
 696 }
 697
 698 int sd_rtnl_call(sd_rtnl *rtnl,
 699                 sd_rtnl_message *message,
 700                 uint64_t usec,
 701                 sd_rtnl_message **ret) {
 702         usec_t timeout;
 703         uint32_t serial;
 704         unsigned i = 0;
 705         int r;
 706
 707         assert_return(rtnl, -EINVAL);
 708         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
 709         assert_return(message, -EINVAL);
 710
 711         r = sd_rtnl_send(rtnl, message, &serial);
 712         if (r < 0)
 713                 return r;
 714
 715         timeout = calc_elapse(usec);
 716
 717         for (;;) {
 718                 usec_t left;
 719
 720                 while (i < rtnl->rqueue_size) {
 721                         sd_rtnl_message *incoming;
 722                         uint32_t received_serial;
 723
 724                         incoming = rtnl->rqueue[i];
 725                         received_serial = rtnl_message_get_serial(incoming);
 726
 727                         if (received_serial == serial) {
 728                                 /* found a match, remove from rqueue and return it */
 729                                 memmove(rtnl->rqueue + i,rtnl->rqueue + i + 1,
 730                                         sizeof(sd_rtnl_message*) * (rtnl->rqueue_size - i - 1));
 731                                 rtnl->rqueue_size--;
 732
 733                                 r = sd_rtnl_message_get_errno(incoming);
 734                                 if (r < 0) {
 735                                         sd_rtnl_message_unref(incoming);
 736                                         return r;
 737                                 }
 738
 739                                 if (ret) {
 740                                         *ret = incoming;
 741                                 } else
 742                                         sd_rtnl_message_unref(incoming);
 743
 744                                 return 1;
 745                         }
 746
 747                         /* Try to read more, right away */
 748                         i ++;
 749                 }
 750
 751                 r = socket_read_message(rtnl);
 752                 if (r < 0)
 753                         return r;
 754                 if (r > 0)
 755                         /* received message, so try to process straight away */
 756                         continue;
 757
 758                 if (timeout > 0) {
 759                         usec_t n;
 760
 761                         n = now(CLOCK_MONOTONIC);
 762                         if (n >= timeout)
 763                                 return -ETIMEDOUT;
 764
 765                         left = timeout - n;
 766                 } else
 767                         left = (uint64_t) -1;
 768
 769                 r = rtnl_poll(rtnl, true, left);
 770                 if (r < 0)
 771                         return r;
 772                 else if (r == 0)
 773                         return -ETIMEDOUT;
 774
 775                 r = dispatch_wqueue(rtnl);
 776                 if (r < 0)
 777                         return r;
 778         }
 779 }
 780
 781 int sd_rtnl_flush(sd_rtnl *rtnl) {
 782         int r;
 783
 784         assert_return(rtnl, -EINVAL);
 785         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
 786
 787         if (rtnl->wqueue_size <= 0)
 788                 return 0;
 789
 790         for (;;) {
 791                 r = dispatch_wqueue(rtnl);
 792                 if (r < 0)
 793                         return r;
 794
 795                 if (rtnl->wqueue_size <= 0)
 796                         return 0;
 797
 798                 r = rtnl_poll(rtnl, false, (uint64_t) -1);
 799                 if (r < 0)
 800                         return r;
 801         }
 802 }
 803
 804 int sd_rtnl_get_events(sd_rtnl *rtnl) {
 805         int flags = 0;
 806
 807         assert_return(rtnl, -EINVAL);
 808         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
 809
 810         if (rtnl->rqueue_size <= 0)
 811                 flags |= POLLIN;
 812         if (rtnl->wqueue_size > 0)
 813                 flags |= POLLOUT;
 814
 815         return flags;
 816 }
 817
 818 int sd_rtnl_get_timeout(sd_rtnl *rtnl, uint64_t *timeout_usec) {
 819         struct reply_callback *c;
 820
 821         assert_return(rtnl, -EINVAL);
 822         assert_return(timeout_usec, -EINVAL);
 823         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
 824
 825         if (rtnl->rqueue_size > 0) {
 826                 *timeout_usec = 0;
 827                 return 1;
 828         }
 829
 830         c = prioq_peek(rtnl->reply_callbacks_prioq);
 831         if (!c) {
 832                 *timeout_usec = (uint64_t) -1;
 833                 return 0;
 834         }
 835
 836         *timeout_usec = c->timeout;
 837
 838         return 1;
 839 }
 840
 841 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
 842         sd_rtnl *rtnl = userdata;
 843         int r;
 844
 845         assert(rtnl);
 846
 847         r = sd_rtnl_process(rtnl, NULL);
 848         if (r < 0)
 849                 return r;
 850
 851         return 1;
 852 }
 853
 854 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
 855         sd_rtnl *rtnl = userdata;
 856         int r;
 857
 858         assert(rtnl);
 859
 860         r = sd_rtnl_process(rtnl, NULL);
 861         if (r < 0)
 862                 return r;
 863
 864         return 1;
 865 }
 866
 867 static int prepare_callback(sd_event_source *s, void *userdata) {
 868         sd_rtnl *rtnl = userdata;
 869         int r, e;
 870         usec_t until;
 871
 872         assert(s);
 873         assert(rtnl);
 874
 875         e = sd_rtnl_get_events(rtnl);
 876         if (e < 0)
 877                 return e;
 878
 879         r = sd_event_source_set_io_events(rtnl->io_event_source, e);
 880         if (r < 0)
 881                 return r;
 882
 883         r = sd_rtnl_get_timeout(rtnl, &until);
 884         if (r < 0)
 885                 return r;
 886         if (r > 0) {
 887                 int j;
 888
 889                 j = sd_event_source_set_time(rtnl->time_event_source, until);
 890                 if (j < 0)
 891                         return j;
 892         }
 893
 894         r = sd_event_source_set_enabled(rtnl->time_event_source, r > 0);
 895         if (r < 0)
 896                 return r;
 897
 898         return 1;
 899 }
 900
 901 static int exit_callback(sd_event_source *event, void *userdata) {
 902         sd_rtnl *rtnl = userdata;
 903
 904         assert(event);
 905
 906         sd_rtnl_flush(rtnl);
 907
 908         return 1;
 909 }
 910
 911 int sd_rtnl_attach_event(sd_rtnl *rtnl, sd_event *event, int priority) {
 912         int r;
 913
 914         assert_return(rtnl, -EINVAL);
 915         assert_return(!rtnl->event, -EBUSY);
 916
 917         assert(!rtnl->io_event_source);
 918         assert(!rtnl->time_event_source);
 919
 920         if (event)
 921                 rtnl->event = sd_event_ref(event);
 922         else {
 923                 r = sd_event_default(&rtnl->event);
 924                 if (r < 0)
 925                         return r;
 926         }
 927
 928         r = sd_event_add_io(rtnl->event, &rtnl->io_event_source, rtnl->fd, 0, io_callback, rtnl);
 929         if (r < 0)
 930                 goto fail;
 931
 932         r = sd_event_source_set_priority(rtnl->io_event_source, priority);
 933         if (r < 0)
 934                 goto fail;
 935
 936         r = sd_event_source_set_description(rtnl->io_event_source, "rtnl-receive-message");
 937         if (r < 0)
 938                 goto fail;
 939
 940         r = sd_event_source_set_prepare(rtnl->io_event_source, prepare_callback);
 941         if (r < 0)
 942                 goto fail;
 943
 944         r = sd_event_add_time(rtnl->event, &rtnl->time_event_source, CLOCK_MONOTONIC, 0, 0, time_callback, rtnl);
 945         if (r < 0)
 946                 goto fail;
 947
 948         r = sd_event_source_set_priority(rtnl->time_event_source, priority);
 949         if (r < 0)
 950                 goto fail;
 951
 952         r = sd_event_source_set_description(rtnl->time_event_source, "rtnl-timer");
 953         if (r < 0)
 954                 goto fail;
 955
 956         r = sd_event_add_exit(rtnl->event, &rtnl->exit_event_source, exit_callback, rtnl);
 957         if (r < 0)
 958                 goto fail;
 959
 960         r = sd_event_source_set_description(rtnl->exit_event_source, "rtnl-exit");
 961         if (r < 0)
 962                 goto fail;
 963
 964         return 0;
 965
 966 fail:
 967         sd_rtnl_detach_event(rtnl);
 968         return r;
 969 }
 970
 971 int sd_rtnl_detach_event(sd_rtnl *rtnl) {
 972         assert_return(rtnl, -EINVAL);
 973         assert_return(rtnl->event, -ENXIO);
 974
 975         if (rtnl->io_event_source)
 976                 rtnl->io_event_source = sd_event_source_unref(rtnl->io_event_source);
 977
 978         if (rtnl->time_event_source)
 979                 rtnl->time_event_source = sd_event_source_unref(rtnl->time_event_source);
 980
 981         if (rtnl->exit_event_source)
 982                 rtnl->exit_event_source = sd_event_source_unref(rtnl->exit_event_source);
 983
 984         if (rtnl->event)
 985                 rtnl->event = sd_event_unref(rtnl->event);
 986
 987         return 0;
 988 }
 989
 990 int sd_rtnl_add_match(sd_rtnl *rtnl,
 991                       uint16_t type,
 992                       sd_rtnl_message_handler_t callback,
 993                       void *userdata) {
 994         struct match_callback *c;
 995
 996         assert_return(rtnl, -EINVAL);
 997         assert_return(callback, -EINVAL);
 998         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
 999         assert_return(rtnl_message_type_is_link(type) ||
1000                       rtnl_message_type_is_addr(type) ||
1001                       rtnl_message_type_is_route(type), -ENOTSUP);
1002
1003         c = new0(struct match_callback, 1);
1004         if (!c)
1005                 return -ENOMEM;
1006
1007         c->callback = callback;
1008         c->type = type;
1009         c->userdata = userdata;
1010
1011         LIST_PREPEND(match_callbacks, rtnl->match_callbacks, c);
1012
1013         return 0;
1014 }
1015
1016 int sd_rtnl_remove_match(sd_rtnl *rtnl,
1017                          uint16_t type,
1018                          sd_rtnl_message_handler_t callback,
1019                          void *userdata) {
1020         struct match_callback *c;
1021
1022         assert_return(rtnl, -EINVAL);
1023         assert_return(callback, -EINVAL);
1024         assert_return(!rtnl_pid_changed(rtnl), -ECHILD);
1025
1026         LIST_FOREACH(match_callbacks, c, rtnl->match_callbacks)
1027                 if (c->callback == callback && c->type == type && c->userdata == userdata) {
1028                         LIST_REMOVE(match_callbacks, rtnl->match_callbacks, c);
1029                         free(c);
1030
1031                         return 1;
1032                 }
1033
1034         return 0;
1035 }