src/shared/barrier.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2014 David Herrmann <dh.herrmann@gmail.com>
   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 <errno.h>
  23 #include <fcntl.h>
  24 #include <limits.h>
  25 #include <poll.h>
  26 #include <stdbool.h>
  27 #include <stdint.h>
  28 #include <stdio.h>
  29 #include <stdlib.h>
  30 #include <string.h>
  31 #include <sys/eventfd.h>
  32 #include <sys/types.h>
  33 #include <unistd.h>
  34
  35 #include "barrier.h"
  36 #include "macro.h"
  37 #include "util.h"
  38
  39 /**
  40  * Barriers
  41  * This barrier implementation provides a simple synchronization method based
  42  * on file-descriptors that can safely be used between threads and processes. A
  43  * barrier object contains 2 shared counters based on eventfd. Both processes
  44  * can now place barriers and wait for the other end to reach a random or
  45  * specific barrier.
  46  * Barriers are numbered, so you can either wait for the other end to reach any
  47  * barrier or the last barrier that you placed. This way, you can use barriers
  48  * for one-way *and* full synchronization. Note that even-though barriers are
  49  * numbered, these numbers are internal and recycled once both sides reached the
  50  * same barrier (implemented as a simple signed counter). It is thus not
  51  * possible to address barriers by their ID.
  52  *
  53  * Barrier-API: Both ends can place as many barriers via barrier_place() as
  54  * they want and each pair of barriers on both sides will be implicitly linked.
  55  * Each side can use the barrier_wait/sync_*() family of calls to wait for the
  56  * other side to place a specific barrier. barrier_wait_next() waits until the
  57  * other side calls barrier_place(). No links between the barriers are
  58  * considered and this simply serves as most basic asynchronous barrier.
  59  * barrier_sync_next() is like barrier_wait_next() and waits for the other side
  60  * to place their next barrier via barrier_place(). However, it only waits for
  61  * barriers that are linked to a barrier we already placed. If the other side
  62  * already placed more barriers than we did, barrier_sync_next() returns
  63  * immediately.
  64  * barrier_sync() extends barrier_sync_next() and waits until the other end
  65  * placed as many barriers via barrier_place() as we did. If they already placed
  66  * as many as we did (or more), it returns immediately.
  67  *
  68  * Additionally to basic barriers, an abortion event is available.
  69  * barrier_abort() places an abortion event that cannot be undone. An abortion
  70  * immediately cancels all placed barriers and replaces them. Any running and
  71  * following wait/sync call besides barrier_wait_abortion() will immediately
  72  * return false on both sides (otherwise, they always return true).
  73  * barrier_abort() can be called multiple times on both ends and will be a
  74  * no-op if already called on this side.
  75  * barrier_wait_abortion() can be used to wait for the other side to call
  76  * barrier_abort() and is the only wait/sync call that does not return
  77  * immediately if we aborted outself. It only returns once the other side
  78  * called barrier_abort().
  79  *
  80  * Barriers can be used for in-process and inter-process synchronization.
  81  * However, for in-process synchronization you could just use mutexes.
  82  * Therefore, main target is IPC and we require both sides to *not* share the FD
  83  * table. If that's given, barriers provide target tracking: If the remote side
  84  * exit()s, an abortion event is implicitly queued on the other side. This way,
  85  * a sync/wait call will be woken up if the remote side crashed or exited
  86  * unexpectedly. However, note that these abortion events are only queued if the
  87  * barrier-queue has been drained. Therefore, it is safe to place a barrier and
  88  * exit. The other side can safely wait on the barrier even though the exit
  89  * queued an abortion event. Usually, the abortion event would overwrite the
  90  * barrier, however, that's not true for exit-abortion events. Those are only
  91  * queued if the barrier-queue is drained (thus, the receiving side has placed
  92  * more barriers than the remote side).
  93  */
  94
  95 /**
  96  * barrier_create() - Initialize a barrier object
  97  * @obj: barrier to initialize
  98  *
  99  * This initializes a barrier object. The caller is responsible of allocating
 100  * the memory and keeping it valid. The memory does not have to be zeroed
 101  * beforehand.
 102  * Two eventfd objects are allocated for each barrier. If allocation fails, an
 103  * error is returned.
 104  *
 105  * If this function fails, the barrier is reset to an invalid state so it is
 106  * safe to call barrier_destroy() on the object regardless whether the
 107  * initialization succeeded or not.
 108  *
 109  * The caller is responsible to destroy the object via barrier_destroy() before
 110  * releasing the underlying memory.
 111  *
 112  * Returns: 0 on success, negative error code on failure.
 113  */
 114 int barrier_create(Barrier *b) {
 115         assert(b);
 116
 117         if ((b->me = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK)) < 0 ||
 118             (b->them = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK)) < 0 ||
 119             pipe2(b->pipe, O_CLOEXEC | O_NONBLOCK) < 0) {
 120                 barrier_destroy(b);
 121                 return -errno;
 122         }
 123
 124         return 0;
 125 }
 126
 127 /**
 128  * barrier_destroy() - Destroy a barrier object
 129  * @b: barrier to destroy or NULL
 130  *
 131  * This destroys a barrier object that has previously been passed to
 132  * barrier_create(). The object is released and reset to invalid
 133  * state. Therefore, it is safe to call barrier_destroy() multiple
 134  * times or even if barrier_create() failed. However, barrier must be
 135  * always initalized with BARRIER_NULL.
 136  *
 137  * If @b is NULL, this is a no-op.
 138  */
 139 void barrier_destroy(Barrier *b) {
 140         if (!b)
 141                 return;
 142
 143         b->me = safe_close(b->me);
 144         b->them = safe_close(b->them);
 145         safe_close_pair(b->pipe);
 146         b->barriers = 0;
 147 }
 148
 149 /**
 150  * barrier_set_role() - Set the local role of the barrier
 151  * @b: barrier to operate on
 152  * @role: role to set on the barrier
 153  *
 154  * This sets the roles on a barrier object. This is needed to know
 155  * which side of the barrier you're on. Usually, the parent creates
 156  * the barrier via barrier_create() and then calls fork() or clone().
 157  * Therefore, the FDs are duplicated and the child retains the same
 158  * barrier object.
 159  *
 160  * Both sides need to call barrier_set_role() after fork() or clone()
 161  * are done. If this is not done, barriers will not work correctly.
 162  *
 163  * Note that barriers could be supported without fork() or clone(). However,
 164  * this is currently not needed so it hasn't been implemented.
 165  */
 166 void barrier_set_role(Barrier *b, unsigned int role) {
 167         int fd;
 168
 169         assert(b);
 170         assert(role == BARRIER_PARENT || role == BARRIER_CHILD);
 171         /* make sure this is only called once */
 172         assert(b->pipe[1] >= 0 && b->pipe[1] >= 0);
 173
 174         if (role == BARRIER_PARENT)
 175                 b->pipe[1] = safe_close(b->pipe[1]);
 176         else {
 177                 b->pipe[0] = safe_close(b->pipe[0]);
 178
 179                 /* swap me/them for children */
 180                 fd = b->me;
 181                 b->me = b->them;
 182                 b->them = fd;
 183         }
 184 }
 185
 186 /* places barrier; returns false if we aborted, otherwise true */
 187 static bool barrier_write(Barrier *b, uint64_t buf) {
 188         ssize_t len;
 189
 190         /* prevent new sync-points if we already aborted */
 191         if (barrier_i_aborted(b))
 192                 return false;
 193
 194         do {
 195                 len = write(b->me, &buf, sizeof(buf));
 196         } while (len < 0 && IN_SET(errno, EAGAIN, EINTR));
 197
 198         if (len != sizeof(buf))
 199                 goto error;
 200
 201         /* lock if we aborted */
 202         if (buf >= (uint64_t)BARRIER_ABORTION) {
 203                 if (barrier_they_aborted(b))
 204                         b->barriers = BARRIER_WE_ABORTED;
 205                 else
 206                         b->barriers = BARRIER_I_ABORTED;
 207         } else if (!barrier_is_aborted(b))
 208                 b->barriers += buf;
 209
 210         return !barrier_i_aborted(b);
 211
 212 error:
 213         /* If there is an unexpected error, we have to make this fatal. There
 214          * is no way we can recover from sync-errors. Therefore, we close the
 215          * pipe-ends and treat this as abortion. The other end will notice the
 216          * pipe-close and treat it as abortion, too. */
 217
 218         safe_close_pair(b->pipe);
 219         b->barriers = BARRIER_WE_ABORTED;
 220         return false;
 221 }
 222
 223 /* waits for barriers; returns false if they aborted, otherwise true */
 224 static bool barrier_read(Barrier *b, int64_t comp) {
 225         if (barrier_they_aborted(b))
 226                 return false;
 227
 228         while (b->barriers > comp) {
 229                 struct pollfd pfd[2] = {
 230                         { .fd = b->pipe[0] >= 0 ? b->pipe[0] : b->pipe[1],
 231                           .events = POLLHUP },
 232                         { .fd = b->them,
 233                           .events = POLLIN }};
 234                 uint64_t buf;
 235                 int r;
 236
 237                 r = poll(pfd, 2, -1);
 238                 if (r < 0 && IN_SET(errno, EAGAIN, EINTR))
 239                         continue;
 240                 else if (r < 0)
 241                         goto error;
 242
 243                 if (pfd[1].revents) {
 244                         ssize_t len;
 245
 246                         /* events on @them signal new data for us */
 247                         len = read(b->them, &buf, sizeof(buf));
 248                         if (len < 0 && IN_SET(errno, EAGAIN, EINTR))
 249                                 continue;
 250
 251                         if (len != sizeof(buf))
 252                                 goto error;
 253                 } else if (pfd[0].revents & (POLLHUP | POLLERR | POLLNVAL))
 254                         /* POLLHUP on the pipe tells us the other side exited.
 255                          * We treat this as implicit abortion. But we only
 256                          * handle it if there's no event on the eventfd. This
 257                          * guarantees that exit-abortions do not overwrite real
 258                          * barriers. */
 259                         buf = BARRIER_ABORTION;
 260                 else
 261                         continue;
 262
 263                 /* lock if they aborted */
 264                 if (buf >= (uint64_t)BARRIER_ABORTION) {
 265                         if (barrier_i_aborted(b))
 266                                 b->barriers = BARRIER_WE_ABORTED;
 267                         else
 268                                 b->barriers = BARRIER_THEY_ABORTED;
 269                 } else if (!barrier_is_aborted(b))
 270                         b->barriers -= buf;
 271         }
 272
 273         return !barrier_they_aborted(b);
 274
 275 error:
 276         /* If there is an unexpected error, we have to make this fatal. There
 277          * is no way we can recover from sync-errors. Therefore, we close the
 278          * pipe-ends and treat this as abortion. The other end will notice the
 279          * pipe-close and treat it as abortion, too. */
 280
 281         safe_close_pair(b->pipe);
 282         b->barriers = BARRIER_WE_ABORTED;
 283         return false;
 284 }
 285
 286 /**
 287  * barrier_place() - Place a new barrier
 288  * @b: barrier object
 289  *
 290  * This places a new barrier on the barrier object. If either side already
 291  * aborted, this is a no-op and returns "false". Otherwise, the barrier is
 292  * placed and this returns "true".
 293  *
 294  * Returns: true if barrier was placed, false if either side aborted.
 295  */
 296 bool barrier_place(Barrier *b) {
 297         assert(b);
 298
 299         if (barrier_is_aborted(b))
 300                 return false;
 301
 302         barrier_write(b, BARRIER_SINGLE);
 303         return true;
 304 }
 305
 306 /**
 307  * barrier_abort() - Abort the synchronization
 308  * @b: barrier object to abort
 309  *
 310  * This aborts the barrier-synchronization. If barrier_abort() was already
 311  * called on this side, this is a no-op. Otherwise, the barrier is put into the
 312  * ABORT-state and will stay there. The other side is notified about the
 313  * abortion. Any following attempt to place normal barriers or to wait on normal
 314  * barriers will return immediately as "false".
 315  *
 316  * You can wait for the other side to call barrier_abort(), too. Use
 317  * barrier_wait_abortion() for that.
 318  *
 319  * Returns: false if the other side already aborted, true otherwise.
 320  */
 321 bool barrier_abort(Barrier *b) {
 322         assert(b);
 323
 324         barrier_write(b, BARRIER_ABORTION);
 325         return !barrier_they_aborted(b);
 326 }
 327
 328 /**
 329  * barrier_wait_next() - Wait for the next barrier of the other side
 330  * @b: barrier to operate on
 331  *
 332  * This waits until the other side places its next barrier. This is independent
 333  * of any barrier-links and just waits for any next barrier of the other side.
 334  *
 335  * If either side aborted, this returns false.
 336  *
 337  * Returns: false if either side aborted, true otherwise.
 338  */
 339 bool barrier_wait_next(Barrier *b) {
 340         assert(b);
 341
 342         if (barrier_is_aborted(b))
 343                 return false;
 344
 345         barrier_read(b, b->barriers - 1);
 346         return !barrier_is_aborted(b);
 347 }
 348
 349 /**
 350  * barrier_wait_abortion() - Wait for the other side to abort
 351  * @b: barrier to operate on
 352  *
 353  * This waits until the other side called barrier_abort(). This can be called
 354  * regardless whether the local side already called barrier_abort() or not.
 355  *
 356  * If the other side has already aborted, this returns immediately.
 357  *
 358  * Returns: false if the local side aborted, true otherwise.
 359  */
 360 bool barrier_wait_abortion(Barrier *b) {
 361         assert(b);
 362
 363         barrier_read(b, BARRIER_THEY_ABORTED);
 364         return !barrier_i_aborted(b);
 365 }
 366
 367 /**
 368  * barrier_sync_next() - Wait for the other side to place a next linked barrier
 369  * @b: barrier to operate on
 370  *
 371  * This is like barrier_wait_next() and waits for the other side to call
 372  * barrier_place(). However, this only waits for linked barriers. That means, if
 373  * the other side already placed more barriers than (or as much as) we did, this
 374  * returns immediately instead of waiting.
 375  *
 376  * If either side aborted, this returns false.
 377  *
 378  * Returns: false if either side aborted, true otherwise.
 379  */
 380 bool barrier_sync_next(Barrier *b) {
 381         assert(b);
 382
 383         if (barrier_is_aborted(b))
 384                 return false;
 385
 386         barrier_read(b, MAX((int64_t)0, b->barriers - 1));
 387         return !barrier_is_aborted(b);
 388 }
 389
 390 /**
 391  * barrier_sync() - Wait for the other side to place as many barriers as we did
 392  * @b: barrier to operate on
 393  *
 394  * This is like barrier_sync_next() but waits for the other side to call
 395  * barrier_place() as often as we did (in total). If they already placed as much
 396  * as we did (or more), this returns immediately instead of waiting.
 397  *
 398  * If either side aborted, this returns false.
 399  *
 400  * Returns: false if either side aborted, true otherwise.
 401  */
 402 bool barrier_sync(Barrier *b) {
 403         assert(b);
 404
 405         if (barrier_is_aborted(b))
 406                 return false;
 407
 408         barrier_read(b, 0);
 409         return !barrier_is_aborted(b);
 410 }