libevfibers
 All Data Structures Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
fiber.c
1 /********************************************************************
2 
3  Copyright 2012 Konstantin Olkhovskiy <lupus@oxnull.net>
4 
5  This file is part of libevfibers.
6 
7  libevfibers is free software: you can redistribute it and/or modify
8  it under the terms of the GNU Lesser General Public License as
9  published by the Free Software Foundation, either version 3 of the
10  License, or any later version.
11 
12  libevfibers is distributed in the hope that it will be useful, but
13  WITHOUT ANY WARRANTY; without even the implied warranty of
14  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15  Lesser General Public License for more details.
16 
17  You should have received a copy of the GNU Lesser General
18  Public License along with libevfibers. If not, see
19  <http://www.gnu.org/licenses/>.
20 
21  ********************************************************************/
22 
23 #include <sys/mman.h>
24 #include <fcntl.h>
25 #include <assert.h>
26 #include <errno.h>
27 #include <utlist.h>
28 #include <stdio.h>
29 #include <string.h>
30 #include <strings.h>
31 #include <err.h>
32 #include <valgrind/valgrind.h>
33 
34 #include <evfibers_private/fiber.h>
35 
36 #ifndef LIST_FOREACH_SAFE
37 #define LIST_FOREACH_SAFE(var, head, field, next_var) \
38  for ((var) = ((head)->lh_first); \
39  (var) && ((next_var) = ((var)->field.le_next), 1); \
40  (var) = (next_var))
41 
42 #endif
43 
44 #ifndef TAILQ_FOREACH_SAFE
45 #define TAILQ_FOREACH_SAFE(var, head, field, next_var) \
46  for ((var) = ((head)->tqh_first); \
47  (var) ? ({ (next_var) = ((var)->field.tqe_next); 1; }) \
48  : 0; \
49  (var) = (next_var))
50 #endif
51 
52 
53 #define ENSURE_ROOT_FIBER do { \
54  assert(fctx->__p->sp->fiber == &fctx->__p->root); \
55 } while (0)
56 
57 #define CURRENT_FIBER (fctx->__p->sp->fiber)
58 #define CURRENT_FIBER_ID (fbr_id_pack(CURRENT_FIBER))
59 #define CALLED_BY_ROOT ((fctx->__p->sp - 1)->fiber == &fctx->__p->root)
60 
61 #define unpack_transfer_errno(value, ptr, id) \
62  do { \
63  if (-1 == fbr_id_unpack(fctx, ptr, id)) \
64  return (value); \
65  } while (0)
66 
67 #define return_success(value) \
68  do { \
69  fctx->f_errno = FBR_SUCCESS; \
70  return (value); \
71  } while (0)
72 
73 #define return_error(value, code) \
74  do { \
75  fctx->f_errno = (code); \
76  return (value); \
77  } while (0)
78 
79 
80 static fbr_id_t fbr_id_pack(struct fbr_fiber *fiber)
81 {
82  return ((__uint128_t)fiber->id << 64) | (uint64_t)fiber;
83 }
84 
85 static int fbr_id_unpack(FBR_P_ struct fbr_fiber **ptr, fbr_id_t id)
86 {
87  struct fbr_fiber *fiber;
88  uint64_t f_id;
89  f_id = (uint64_t)(id >> 64);
90  fiber = (struct fbr_fiber *)(uint64_t)id;
91  if (fiber->id != f_id)
92  return_error(-1, FBR_ENOFIBER);
93  if (ptr)
94  *ptr = fiber;
95  return 0;
96 }
97 
98 static void pending_async_cb(EV_P_ ev_async *w, _unused_ int revents)
99 {
100  struct fbr_context *fctx;
101  struct fbr_id_tailq_i *item;
102  fctx = (struct fbr_context *)w->data;
103  int retval;
104 
105  ENSURE_ROOT_FIBER;
106 
107  if (TAILQ_EMPTY(&fctx->__p->pending_fibers)) {
108  ev_async_stop(EV_A_ &fctx->__p->pending_async);
109  return;
110  }
111 
112  item = TAILQ_FIRST(&fctx->__p->pending_fibers);
113  assert(item->head == &fctx->__p->pending_fibers);
114  /* item shall be removed from the queue by a destructor, which shall be
115  * set by the procedure demanding delayed execution. Destructor
116  * guarantees removal upon the reclaim of fiber. */
117  ev_async_send(EV_A_ &fctx->__p->pending_async);
118 
119  retval = fbr_transfer(FBR_A_ item->id);
120  if (-1 == retval && FBR_ENOFIBER != fctx->f_errno) {
121  fbr_log_e(FBR_A_ "libevfibers: unexpected error trying to call"
122  " a fiber by id: %s",
123  fbr_strerror(FBR_A_ fctx->f_errno));
124  }
125 }
126 
127 static void *allocate_in_fiber(FBR_P_ size_t size, struct fbr_fiber *in)
128 {
129  struct mem_pool *pool_entry;
130  pool_entry = malloc(size + sizeof(struct mem_pool));
131  if (NULL == pool_entry) {
132  fbr_log_e(FBR_A_ "libevfibers: unable to allocate %zu bytes\n",
133  size + sizeof(struct mem_pool));
134  abort();
135  }
136  pool_entry->ptr = pool_entry;
137  pool_entry->destructor = NULL;
138  pool_entry->destructor_context = NULL;
139  LIST_INSERT_HEAD(&in->pool, pool_entry, entries);
140  return pool_entry + 1;
141 }
142 
143 static void stdio_logger(FBR_P_ struct fbr_logger *logger, enum fbr_log_level level,
144  const char *format, va_list ap)
145 {
146  struct fbr_fiber *fiber;
147  FILE* stream;
148  char *str_level;
149  ev_tstamp tstamp;
150 
151  if (level > logger->level)
152  return;
153 
154  fiber = CURRENT_FIBER;
155 
156  switch (level) {
157  case FBR_LOG_ERROR:
158  str_level = "ERROR";
159  stream = stderr;
160  break;
161  case FBR_LOG_WARNING:
162  str_level = "WARNING";
163  stream = stdout;
164  break;
165  case FBR_LOG_NOTICE:
166  str_level = "NOTICE";
167  stream = stdout;
168  break;
169  case FBR_LOG_INFO:
170  str_level = "INFO";
171  stream = stdout;
172  break;
173  case FBR_LOG_DEBUG:
174  str_level = "DEBUG";
175  stream = stdout;
176  break;
177  default:
178  str_level = "?????";
179  stream = stdout;
180  break;
181  }
182  tstamp = ev_now(fctx->__p->loop);
183  fprintf(stream, "%.6f %-7s %-16s ", tstamp, str_level, fiber->name);
184  vfprintf(stream, format, ap);
185  fprintf(stream, "\n");
186 }
187 
188 void fbr_init(FBR_P_ struct ev_loop *loop)
189 {
190  struct fbr_fiber *root;
191  struct fbr_logger *logger;
192 
193  fctx->__p = malloc(sizeof(struct fbr_context_private));
194  LIST_INIT(&fctx->__p->reclaimed);
195  LIST_INIT(&fctx->__p->root.children);
196  LIST_INIT(&fctx->__p->root.pool);
197  TAILQ_INIT(&fctx->__p->root.destructors);
198  TAILQ_INIT(&fctx->__p->pending_fibers);
199 
200  root = &fctx->__p->root;
201  strncpy(root->name, "root", FBR_MAX_FIBER_NAME);
202  fctx->__p->last_id = 0;
203  root->id = fctx->__p->last_id++;
204  coro_create(&root->ctx, NULL, NULL, NULL, 0);
205 
206  logger = allocate_in_fiber(FBR_A_ sizeof(struct fbr_logger), root);
207  logger->logv = stdio_logger;
208  logger->level = FBR_LOG_NOTICE;
209  fctx->logger = logger;
210 
211  fctx->__p->sp = fctx->__p->stack;
212  fctx->__p->sp->fiber = root;
213  fctx->__p->backtraces_enabled = 1;
214  fill_trace_info(FBR_A_ &fctx->__p->sp->tinfo);
215  fctx->__p->loop = loop;
216  fctx->__p->pending_async.data = fctx;
217  fctx->__p->backtraces_enabled = 0;
218  memset(&fctx->__p->key_free_mask, 0x00,
219  sizeof(fctx->__p->key_free_mask));
220  ev_async_init(&fctx->__p->pending_async, pending_async_cb);
221 }
222 
223 const char *fbr_strerror(_unused_ FBR_P_ enum fbr_error_code code)
224 {
225  switch (code) {
226  case FBR_SUCCESS:
227  return "Success";
228  case FBR_EINVAL:
229  return "Invalid argument";
230  case FBR_ENOFIBER:
231  return "No such fiber";
232  case FBR_ESYSTEM:
233  return "System error, consult system errno";
234  case FBR_EBUFFERMMAP:
235  return "Failed to mmap two adjacent regions";
236  case FBR_ENOKEY:
237  return "Fiber-local key does not exist";
238  }
239  return "Unknown error";
240 }
241 
242 void fbr_log_e(FBR_P_ const char *format, ...)
243 {
244  va_list ap;
245  va_start(ap, format);
246  (*fctx->logger->logv)(FBR_A_ fctx->logger, FBR_LOG_ERROR, format, ap);
247  va_end(ap);
248 }
249 
250 void fbr_log_w(FBR_P_ const char *format, ...)
251 {
252  va_list ap;
253  va_start(ap, format);
254  (*fctx->logger->logv)(FBR_A_ fctx->logger, FBR_LOG_WARNING, format, ap);
255  va_end(ap);
256 }
257 
258 void fbr_log_n(FBR_P_ const char *format, ...)
259 {
260  va_list ap;
261  va_start(ap, format);
262  (*fctx->logger->logv)(FBR_A_ fctx->logger, FBR_LOG_NOTICE, format, ap);
263  va_end(ap);
264 }
265 
266 void fbr_log_i(FBR_P_ const char *format, ...)
267 {
268  va_list ap;
269  va_start(ap, format);
270  (*fctx->logger->logv)(FBR_A_ fctx->logger, FBR_LOG_INFO, format, ap);
271  va_end(ap);
272 }
273 
274 void fbr_log_d(FBR_P_ const char *format, ...)
275 {
276  va_list ap;
277  va_start(ap, format);
278  (*fctx->logger->logv)(FBR_A_ fctx->logger, FBR_LOG_DEBUG, format, ap);
279  va_end(ap);
280 }
281 
282 void id_tailq_i_set(_unused_ FBR_P_
283  struct fbr_id_tailq_i *item,
284  struct fbr_fiber *fiber)
285 {
286  item->id = fbr_id_pack(fiber);
287  item->ev = NULL;
288 }
289 
290 static void reclaim_children(FBR_P_ struct fbr_fiber *fiber)
291 {
292  struct fbr_fiber *f;
293  LIST_FOREACH(f, &fiber->children, entries.children) {
294  fbr_reclaim(FBR_A_ fbr_id_pack(f));
295  }
296 }
297 
298 static void fbr_free_in_fiber(_unused_ FBR_P_ _unused_ struct fbr_fiber *fiber,
299  void *ptr, int destructor);
300 
301 void fbr_destroy(FBR_P)
302 {
303  struct fbr_fiber *fiber, *x;
304  struct mem_pool *p, *x2;
305 
306  reclaim_children(FBR_A_ &fctx->__p->root);
307 
308  LIST_FOREACH_SAFE(p, &fctx->__p->root.pool, entries, x2) {
309  fbr_free_in_fiber(FBR_A_ &fctx->__p->root, p + 1, 1);
310  }
311 
312  LIST_FOREACH_SAFE(fiber, &fctx->__p->reclaimed, entries.reclaimed, x) {
313  if (0 != munmap(fiber->stack, fiber->stack_size))
314  err(EXIT_FAILURE, "munmap");
315  free(fiber);
316  }
317 
318  free(fctx->__p);
319 }
320 
321 void fbr_enable_backtraces(FBR_P_ int enabled)
322 {
323  if (enabled)
324  fctx->__p->backtraces_enabled = 1;
325  else
326  fctx->__p->backtraces_enabled = 0;
327 
328 }
329 
330 static void cancel_ev(_unused_ FBR_P_ struct fbr_ev_base *ev)
331 {
332  fbr_destructor_remove(FBR_A_ &ev->item.dtor, 1 /* call it */);
333 }
334 
335 static void post_ev(_unused_ FBR_P_ struct fbr_fiber *fiber, struct fbr_ev_base *ev)
336 {
337  assert(NULL != fiber->ev.waiting);
338 
339  fiber->ev.arrived = 1;
340  ev->arrived = 1;
341 }
342 
343 static void ev_watcher_cb(_unused_ EV_P_ ev_watcher *w, _unused_ int event)
344 {
345  struct fbr_fiber *fiber;
346  struct fbr_ev_watcher *ev = w->data;
347  struct fbr_context *fctx = ev->ev_base.fctx;
348  int retval;
349 
350  ENSURE_ROOT_FIBER;
351 
352  retval = fbr_id_unpack(FBR_A_ &fiber, ev->ev_base.id);
353  if (-1 == retval) {
354  fbr_log_e(FBR_A_ "libevfibers: fiber is about to be called by"
355  " the watcher callback, but it's id is not valid: %s",
356  fbr_strerror(FBR_A_ fctx->f_errno));
357  abort();
358  }
359 
360  post_ev(FBR_A_ fiber, &ev->ev_base);
361 
362  retval = fbr_transfer(FBR_A_ fbr_id_pack(fiber));
363  assert(0 == retval);
364 }
365 
366 
367 static void fbr_free_in_fiber(_unused_ FBR_P_ _unused_ struct fbr_fiber *fiber,
368  void *ptr, int destructor)
369 {
370  struct mem_pool *pool_entry = NULL;
371  if (NULL == ptr)
372  return;
373  pool_entry = (struct mem_pool *)ptr - 1;
374  if (pool_entry->ptr != pool_entry) {
375  fbr_log_e(FBR_A_ "libevfibers: address %p does not look like "
376  "fiber memory pool entry", ptr);
377  if (!RUNNING_ON_VALGRIND)
378  abort();
379  }
380  LIST_REMOVE(pool_entry, entries);
381  if (destructor && pool_entry->destructor)
382  pool_entry->destructor(FBR_A_ ptr, pool_entry->destructor_context);
383  free(pool_entry);
384 }
385 
386 static void fiber_cleanup(FBR_P_ struct fbr_fiber *fiber)
387 {
388  struct mem_pool *p, *x;
389  struct fbr_destructor *dtor;
390  /* coro_destroy(&fiber->ctx); */
391  LIST_REMOVE(fiber, entries.children);
392  TAILQ_FOREACH(dtor, &fiber->destructors, entries) {
393  dtor->func(FBR_A_ dtor->arg);
394  }
395  LIST_FOREACH_SAFE(p, &fiber->pool, entries, x) {
396  fbr_free_in_fiber(FBR_A_ fiber, p + 1, 1);
397  }
398 }
399 
400 static void filter_fiber_stack(FBR_P_ struct fbr_fiber *fiber)
401 {
402  struct fbr_stack_item *sp;
403  for (sp = fctx->__p->stack; sp < fctx->__p->sp; sp++) {
404  if (sp->fiber == fiber) {
405  memmove(sp, sp + 1, (fctx->__p->sp - sp) * sizeof(*sp));
406  fctx->__p->sp--;
407  }
408  }
409 }
410 
411 int fbr_reclaim(FBR_P_ fbr_id_t id)
412 {
413  struct fbr_fiber *fiber;
414 #if 0
415  struct fbr_fiber *f;
416 #endif
417 
418  unpack_transfer_errno(-1, &fiber, id);
419 
420  fill_trace_info(FBR_A_ &fiber->reclaim_tinfo);
421  reclaim_children(FBR_A_ fiber);
422  fiber_cleanup(FBR_A_ fiber);
423  fiber->id = fctx->__p->last_id++;
424 #if 0
425  LIST_FOREACH(f, &fctx->__p->reclaimed, entries.reclaimed) {
426  assert(f != fiber);
427  }
428 #endif
429  LIST_INSERT_HEAD(&fctx->__p->reclaimed, fiber, entries.reclaimed);
430 
431  filter_fiber_stack(FBR_A_ fiber);
432 
433  if (CURRENT_FIBER == fiber)
434  fbr_yield(FBR_A);
435 
436  return_success(0);
437 }
438 
439 int fbr_is_reclaimed(_unused_ FBR_P_ fbr_id_t id)
440 {
441  if (0 == fbr_id_unpack(FBR_A_ NULL, id))
442  return 0;
443  return 1;
444 }
445 
446 fbr_id_t fbr_self(FBR_P)
447 {
448  return CURRENT_FIBER_ID;
449 }
450 
451 static void call_wrapper(FBR_P)
452 {
453  int retval;
454  struct fbr_fiber *fiber = CURRENT_FIBER;
455 
456  fiber->func(FBR_A_ fiber->func_arg);
457 
458  retval = fbr_reclaim(FBR_A_ fbr_id_pack(fiber));
459  assert(0 == retval);
460  fbr_yield(FBR_A);
461  assert(NULL);
462 }
463 
464 enum ev_action_hint {
465  EV_AH_OK = 0,
466  EV_AH_ARRIVED,
467  EV_AH_EINVAL
468 };
469 
470 static void item_dtor(_unused_ FBR_P_ void *arg)
471 {
472  struct fbr_id_tailq_i *item = arg;
473 
474  if (item->head) {
475  TAILQ_REMOVE(item->head, item, entries);
476  }
477 }
478 
479 static enum ev_action_hint prepare_ev(FBR_P_ struct fbr_ev_base *ev)
480 {
481  struct fbr_ev_watcher *e_watcher;
482  struct fbr_ev_mutex *e_mutex;
483  struct fbr_ev_cond_var *e_cond;
484  struct fbr_id_tailq_i *item = &ev->item;
485 
486  ev->arrived = 0;
487  ev->item.dtor.func = item_dtor;
488  ev->item.dtor.arg = item;
489  fbr_destructor_add(FBR_A_ &ev->item.dtor);
490 
491  switch (ev->type) {
492  case FBR_EV_WATCHER:
493  e_watcher = fbr_ev_upcast(ev, fbr_ev_watcher);
494  if (!ev_is_active(e_watcher->w)) {
495  fbr_destructor_remove(FBR_A_ &ev->item.dtor,
496  0 /* call it */);
497  return EV_AH_EINVAL;
498  }
499  e_watcher->w->data = e_watcher;
500  ev_set_cb(e_watcher->w, ev_watcher_cb);
501  break;
502  case FBR_EV_MUTEX:
503  e_mutex = fbr_ev_upcast(ev, fbr_ev_mutex);
504  if (0 == e_mutex->mutex->locked_by) {
505  e_mutex->mutex->locked_by = CURRENT_FIBER_ID;
506  return EV_AH_ARRIVED;
507  }
508  id_tailq_i_set(FBR_A_ item, CURRENT_FIBER);
509  item->ev = ev;
510  ev->data = item;
511  TAILQ_INSERT_TAIL(&e_mutex->mutex->pending, item, entries);
512  item->head = &e_mutex->mutex->pending;
513  break;
514  case FBR_EV_COND_VAR:
515  e_cond = fbr_ev_upcast(ev, fbr_ev_cond_var);
516  if (0 == e_cond->mutex->locked_by) {
517  fbr_destructor_remove(FBR_A_ &ev->item.dtor,
518  0 /* call it */);
519  return EV_AH_EINVAL;
520  }
521  id_tailq_i_set(FBR_A_ item, CURRENT_FIBER);
522  item->ev = ev;
523  ev->data = item;
524  TAILQ_INSERT_TAIL(&e_cond->cond->waiting, item, entries);
525  item->head = &e_cond->cond->waiting;
526  fbr_mutex_unlock(FBR_A_ e_cond->mutex);
527  break;
528  }
529  return EV_AH_OK;
530 }
531 
532 static void finish_ev(FBR_P_ struct fbr_ev_base *ev)
533 {
534  struct fbr_ev_cond_var *e_cond;
535  struct fbr_ev_watcher *e_watcher;
536  fbr_destructor_remove(FBR_A_ &ev->item.dtor, 1 /* call it */);
537  switch (ev->type) {
538  case FBR_EV_COND_VAR:
539  e_cond = fbr_ev_upcast(ev, fbr_ev_cond_var);
540  fbr_mutex_lock(FBR_A_ e_cond->mutex);
541  break;
542  case FBR_EV_WATCHER:
543  e_watcher = fbr_ev_upcast(ev, fbr_ev_watcher);
544  ev_set_cb(e_watcher->w, NULL);
545  break;
546  case FBR_EV_MUTEX:
547  /* NOP */
548  break;
549  }
550 }
551 
552 int fbr_ev_wait(FBR_P_ struct fbr_ev_base *events[])
553 {
554  struct fbr_fiber *fiber = CURRENT_FIBER;
555  struct fbr_ev_base **ev_pptr;
556  enum ev_action_hint hint;
557  int num = 0;
558 
559  fiber->ev.arrived = 0;
560  fiber->ev.waiting = events;
561 
562  for (ev_pptr = events; NULL != *ev_pptr; ev_pptr++) {
563  hint = prepare_ev(FBR_A_ *ev_pptr);
564  switch (hint) {
565  case EV_AH_OK:
566  break;
567  case EV_AH_ARRIVED:
568  fiber->ev.arrived = 1;
569  (*ev_pptr)->arrived = 1;
570  break;
571  case EV_AH_EINVAL:
572  return_error(-1, FBR_EINVAL);
573  }
574  }
575 
576  while (0 == fiber->ev.arrived)
577  fbr_yield(FBR_A);
578 
579  for (ev_pptr = events; NULL != *ev_pptr; ev_pptr++) {
580  if ((*ev_pptr)->arrived) {
581  num++;
582  finish_ev(FBR_A_ *ev_pptr);
583  } else
584  cancel_ev(FBR_A_ *ev_pptr);
585  }
586  return_success(num);
587 }
588 
589 int fbr_ev_wait_one(FBR_P_ struct fbr_ev_base *one)
590 {
591  int n_events;
592  struct fbr_ev_base *events[] = {one, NULL};
593  n_events = fbr_ev_wait(FBR_A_ events);
594  if (1 == n_events)
595  return 0;
596  return -1;
597 }
598 
599 int fbr_transfer(FBR_P_ fbr_id_t to)
600 {
601  struct fbr_fiber *callee;
602  struct fbr_fiber *caller = fctx->__p->sp->fiber;
603 
604  unpack_transfer_errno(-1, &callee, to);
605 
606  fctx->__p->sp++;
607 
608  fctx->__p->sp->fiber = callee;
609  fill_trace_info(FBR_A_ &fctx->__p->sp->tinfo);
610 
611  coro_transfer(&caller->ctx, &callee->ctx);
612 
613  return_success(0);
614 }
615 
616 void fbr_yield(FBR_P)
617 {
618  struct fbr_fiber *callee = fctx->__p->sp->fiber;
619  struct fbr_fiber *caller = (--fctx->__p->sp)->fiber;
620 
621  coro_transfer(&callee->ctx, &caller->ctx);
622 }
623 
624 int fbr_fd_nonblock(FBR_P_ int fd)
625 {
626  int flags, s;
627 
628  flags = fcntl(fd, F_GETFL, 0);
629  if (flags == -1)
630  return_error(-1, FBR_ESYSTEM);
631 
632  flags |= O_NONBLOCK;
633  s = fcntl(fd, F_SETFL, flags);
634  if (s == -1)
635  return_error(-1, FBR_ESYSTEM);
636 
637  return_success(0);
638 }
639 
640 static void ev_base_init(FBR_P_ struct fbr_ev_base *ev,
641  enum fbr_ev_type type)
642 {
643  memset(ev, 0x00, sizeof(*ev));
644  ev->type = type;
645  ev->id = CURRENT_FIBER_ID;
646  ev->fctx = fctx;
647 }
648 
649 void fbr_ev_watcher_init(FBR_P_ struct fbr_ev_watcher *ev, ev_watcher *w)
650 {
651  ev_base_init(FBR_A_ &ev->ev_base, FBR_EV_WATCHER);
652  ev->w = w;
653 }
654 
655 static void watcher_io_dtor(_unused_ FBR_P_ void *_arg)
656 {
657  struct ev_io *w = _arg;
658  ev_io_stop(fctx->__p->loop, w);
659 }
660 
661 ssize_t fbr_read(FBR_P_ int fd, void *buf, size_t count)
662 {
663  ssize_t r;
664  ev_io io;
665  struct fbr_ev_watcher watcher;
666  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
667 
668  ev_io_init(&io, NULL, fd, EV_READ);
669  ev_io_start(fctx->__p->loop, &io);
670  dtor.func = watcher_io_dtor;
671  dtor.arg = &io;
672  fbr_destructor_add(FBR_A_ &dtor);
673 
674  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
675  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
676 
677  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
678  do {
679  r = read(fd, buf, count);
680  } while (-1 == r && EINTR == errno);
681 
682  ev_io_stop(fctx->__p->loop, &io);
683 
684  return r;
685 }
686 
687 ssize_t fbr_read_all(FBR_P_ int fd, void *buf, size_t count)
688 {
689  ssize_t r;
690  size_t done = 0;
691  ev_io io;
692  struct fbr_ev_watcher watcher;
693  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
694 
695  ev_io_init(&io, NULL, fd, EV_READ);
696  ev_io_start(fctx->__p->loop, &io);
697  dtor.func = watcher_io_dtor;
698  dtor.arg = &io;
699  fbr_destructor_add(FBR_A_ &dtor);
700 
701  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
702 
703  while (count != done) {
704 next:
705  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
706  for (;;) {
707  r = read(fd, buf + done, count - done);
708  if (-1 == r) {
709  switch (errno) {
710  case EINTR:
711  continue;
712  case EAGAIN:
713  goto next;
714  default:
715  goto error;
716  }
717  }
718  break;
719  }
720  if (0 == r)
721  break;
722  done += r;
723  }
724  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
725  ev_io_stop(fctx->__p->loop, &io);
726  return (ssize_t)done;
727 
728 error:
729  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
730  ev_io_stop(fctx->__p->loop, &io);
731  return -1;
732 }
733 
734 ssize_t fbr_readline(FBR_P_ int fd, void *buffer, size_t n)
735 {
736  ssize_t num_read;
737  size_t total_read;
738  char *buf;
739  char ch;
740 
741  if (n <= 0 || buffer == NULL) {
742  errno = EINVAL;
743  return -1;
744  }
745 
746  buf = buffer;
747 
748  total_read = 0;
749  for (;;) {
750  num_read = fbr_read(FBR_A_ fd, &ch, 1);
751 
752  if (num_read == -1) {
753  if (errno == EINTR)
754  continue;
755  else
756  return -1;
757 
758  } else if (num_read == 0) {
759  if (total_read == 0)
760  return 0;
761  else
762  break;
763 
764  } else {
765  if (total_read < n - 1) {
766  total_read++;
767  *buf++ = ch;
768  }
769 
770  if (ch == '\n')
771  break;
772  }
773  }
774 
775  *buf = '\0';
776  return total_read;
777 }
778 
779 ssize_t fbr_write(FBR_P_ int fd, const void *buf, size_t count)
780 {
781  ssize_t r;
782  ev_io io;
783  struct fbr_ev_watcher watcher;
784  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
785 
786  ev_io_init(&io, NULL, fd, EV_WRITE);
787  ev_io_start(fctx->__p->loop, &io);
788  dtor.func = watcher_io_dtor;
789  dtor.arg = &io;
790  fbr_destructor_add(FBR_A_ &dtor);
791 
792  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
793  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
794 
795  do {
796  r = write(fd, buf, count);
797  } while (-1 == r && EINTR == errno);
798 
799  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
800  ev_io_stop(fctx->__p->loop, &io);
801  return r;
802 }
803 
804 ssize_t fbr_write_all(FBR_P_ int fd, const void *buf, size_t count)
805 {
806  ssize_t r;
807  size_t done = 0;
808  ev_io io;
809  struct fbr_ev_watcher watcher;
810  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
811 
812  ev_io_init(&io, NULL, fd, EV_WRITE);
813  ev_io_start(fctx->__p->loop, &io);
814  dtor.func = watcher_io_dtor;
815  dtor.arg = &io;
816  fbr_destructor_add(FBR_A_ &dtor);
817 
818  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
819 
820  while (count != done) {
821 next:
822  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
823  for (;;) {
824  r = write(fd, buf + done, count - done);
825  if (-1 == r) {
826  switch (errno) {
827  case EINTR:
828  continue;
829  case EAGAIN:
830  goto next;
831  default:
832  goto error;
833  }
834  }
835  break;
836  }
837  done += r;
838  }
839  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
840  ev_io_stop(fctx->__p->loop, &io);
841  return (ssize_t)done;
842 
843 error:
844  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
845  ev_io_stop(fctx->__p->loop, &io);
846  return -1;
847 }
848 
849 ssize_t fbr_recvfrom(FBR_P_ int sockfd, void *buf, size_t len, int flags,
850  struct sockaddr *src_addr, socklen_t *addrlen)
851 {
852  ev_io io;
853  struct fbr_ev_watcher watcher;
854  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
855 
856  ev_io_init(&io, NULL, sockfd, EV_READ);
857  ev_io_start(fctx->__p->loop, &io);
858  dtor.func = watcher_io_dtor;
859  dtor.arg = &io;
860  fbr_destructor_add(FBR_A_ &dtor);
861 
862  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
863  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
864 
865  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
866  ev_io_stop(fctx->__p->loop, &io);
867 
868  return recvfrom(sockfd, buf, len, flags, src_addr, addrlen);
869 }
870 
871 ssize_t fbr_sendto(FBR_P_ int sockfd, const void *buf, size_t len, int flags, const
872  struct sockaddr *dest_addr, socklen_t addrlen)
873 {
874  ev_io io;
875  struct fbr_ev_watcher watcher;
876  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
877 
878  ev_io_init(&io, NULL, sockfd, EV_WRITE);
879  ev_io_start(fctx->__p->loop, &io);
880  dtor.func = watcher_io_dtor;
881  dtor.arg = &io;
882  fbr_destructor_add(FBR_A_ &dtor);
883 
884  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
885  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
886 
887  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
888  ev_io_stop(fctx->__p->loop, &io);
889 
890  return sendto(sockfd, buf, len, flags, dest_addr, addrlen);
891 }
892 
893 int fbr_accept(FBR_P_ int sockfd, struct sockaddr *addr, socklen_t *addrlen)
894 {
895  int r;
896  ev_io io;
897  struct fbr_ev_watcher watcher;
898  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
899 
900  ev_io_init(&io, NULL, sockfd, EV_READ);
901  ev_io_start(fctx->__p->loop, &io);
902  dtor.func = watcher_io_dtor;
903  dtor.arg = &io;
904  fbr_destructor_add(FBR_A_ &dtor);
905 
906  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&io);
907  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
908 
909  do {
910  r = accept(sockfd, addr, addrlen);
911  } while (-1 == r && EINTR == errno);
912 
913  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
914  ev_io_stop(fctx->__p->loop, &io);
915 
916  return r;
917 }
918 
919 static void watcher_timer_dtor(_unused_ FBR_P_ void *_arg)
920 {
921  struct ev_timer *w = _arg;
922  ev_timer_stop(fctx->__p->loop, w);
923 }
924 
925 ev_tstamp fbr_sleep(FBR_P_ ev_tstamp seconds)
926 {
927  ev_timer timer;
928  struct fbr_ev_watcher watcher;
929  struct fbr_destructor dtor = FBR_DESTRUCTOR_INITIALIZER;
930  ev_tstamp expected = ev_now(fctx->__p->loop) + seconds;
931 
932  ev_timer_init(&timer, NULL, seconds, 0.);
933  ev_timer_start(fctx->__p->loop, &timer);
934  dtor.func = watcher_timer_dtor;
935  dtor.arg = &timer;
936  fbr_destructor_add(FBR_A_ &dtor);
937 
938  fbr_ev_watcher_init(FBR_A_ &watcher, (ev_watcher *)&timer);
939  fbr_ev_wait_one(FBR_A_ &watcher.ev_base);
940 
941  fbr_destructor_remove(FBR_A_ &dtor, 0 /* Call it? */);
942  ev_timer_stop(fctx->__p->loop, &timer);
943 
944  return max(0., expected - ev_now(fctx->__p->loop));
945 }
946 
947 static size_t round_up_to_page_size(size_t size)
948 {
949  static long sz;
950  size_t remainder;
951  if (0 == sz)
952  sz = sysconf(_SC_PAGESIZE);
953  remainder = size % sz;
954  if (remainder == 0)
955  return size;
956  return size + sz - remainder;
957 }
958 
959 fbr_id_t fbr_create(FBR_P_ const char *name, fbr_fiber_func_t func, void *arg,
960  size_t stack_size)
961 {
962  struct fbr_fiber *fiber;
963  if (!LIST_EMPTY(&fctx->__p->reclaimed)) {
964  fiber = LIST_FIRST(&fctx->__p->reclaimed);
965  LIST_REMOVE(fiber, entries.reclaimed);
966  } else {
967  fiber = malloc(sizeof(struct fbr_fiber));
968  memset(fiber, 0x00, sizeof(struct fbr_fiber));
969  if (0 == stack_size)
970  stack_size = FBR_STACK_SIZE;
971  stack_size = round_up_to_page_size(stack_size);
972  fiber->stack = mmap(NULL, stack_size, PROT_READ | PROT_WRITE,
973  MAP_PRIVATE | MAP_ANONYMOUS, -1, 0);
974  if (MAP_FAILED == fiber->stack)
975  err(EXIT_FAILURE, "mmap failed");
976  fiber->stack_size = stack_size;
977  (void)VALGRIND_STACK_REGISTER(fiber->stack, fiber->stack +
978  stack_size);
979  fiber->id = fctx->__p->last_id++;
980  }
981  coro_create(&fiber->ctx, (coro_func)call_wrapper, FBR_A, fiber->stack,
982  fiber->stack_size);
983  fiber->call_list = NULL;
984  fiber->call_list_size = 0;
985  LIST_INIT(&fiber->children);
986  LIST_INIT(&fiber->pool);
987  TAILQ_INIT(&fiber->destructors);
988  strncpy(fiber->name, name, FBR_MAX_FIBER_NAME);
989  fiber->func = func;
990  fiber->func_arg = arg;
991  LIST_INSERT_HEAD(&CURRENT_FIBER->children, fiber, entries.children);
992  fiber->parent = CURRENT_FIBER;
993  return fbr_id_pack(fiber);
994 }
995 
996 int fbr_disown(FBR_P_ fbr_id_t parent_id)
997 {
998  struct fbr_fiber *fiber, *parent;
999  if (parent_id > 0)
1000  unpack_transfer_errno(-1, &parent, parent_id);
1001  else
1002  parent = &fctx->__p->root;
1003  fiber = CURRENT_FIBER;
1004  LIST_REMOVE(fiber, entries.children);
1005  LIST_INSERT_HEAD(&parent->children, fiber, entries.children);
1006  fiber->parent = parent;
1007  return_success(0);
1008 }
1009 
1010 fbr_id_t fbr_parent(FBR_P)
1011 {
1012  struct fbr_fiber *fiber = CURRENT_FIBER;
1013  if (fiber->parent == &fctx->__p->root)
1014  return 0;
1015  return fbr_id_pack(fiber->parent);
1016 }
1017 
1018 void *fbr_calloc(FBR_P_ unsigned int nmemb, size_t size)
1019 {
1020  void *ptr;
1021  ptr = allocate_in_fiber(FBR_A_ nmemb * size, CURRENT_FIBER);
1022  memset(ptr, 0x00, nmemb * size);
1023  return ptr;
1024 }
1025 
1026 void *fbr_alloc(FBR_P_ size_t size)
1027 {
1028  return allocate_in_fiber(FBR_A_ size, CURRENT_FIBER);
1029 }
1030 
1031 void fbr_alloc_set_destructor(_unused_ FBR_P_ void *ptr,
1032  fbr_alloc_destructor_func_t func, void *context)
1033 {
1034  struct mem_pool *pool_entry;
1035  pool_entry = (struct mem_pool *)ptr - 1;
1036  pool_entry->destructor = func;
1037  pool_entry->destructor_context = context;
1038 }
1039 
1040 void fbr_free(FBR_P_ void *ptr)
1041 {
1042  fbr_free_in_fiber(FBR_A_ CURRENT_FIBER, ptr, 1);
1043 }
1044 
1045 void fbr_free_nd(FBR_P_ void *ptr)
1046 {
1047  fbr_free_in_fiber(FBR_A_ CURRENT_FIBER, ptr, 0);
1048 }
1049 
1050 void fbr_dump_stack(FBR_P_ fbr_logutil_func_t log)
1051 {
1052  struct fbr_stack_item *ptr = fctx->__p->sp;
1053  (*log)(FBR_A_ "%s", "Fiber call stack:");
1054  (*log)(FBR_A_ "%s", "-------------------------------");
1055  while (ptr >= fctx->__p->stack) {
1056  (*log)(FBR_A_ "fiber_call: %p\t%s",
1057  ptr->fiber,
1058  ptr->fiber->name);
1059  print_trace_info(FBR_A_ &ptr->tinfo, log);
1060  (*log)(FBR_A_ "%s", "-------------------------------");
1061  ptr--;
1062  }
1063 }
1064 
1065 static void transfer_later(FBR_P_ struct fbr_id_tailq_i *item)
1066 {
1067  int was_empty;
1068  was_empty = TAILQ_EMPTY(&fctx->__p->pending_fibers);
1069  TAILQ_INSERT_TAIL(&fctx->__p->pending_fibers, item, entries);
1070  item->head = &fctx->__p->pending_fibers;
1071  if (was_empty && !TAILQ_EMPTY(&fctx->__p->pending_fibers)) {
1072  ev_async_start(fctx->__p->loop, &fctx->__p->pending_async);
1073  }
1074  ev_async_send(fctx->__p->loop, &fctx->__p->pending_async);
1075 }
1076 
1077 static void transfer_later_tailq(FBR_P_ struct fbr_id_tailq *tailq)
1078 {
1079  int was_empty;
1080  struct fbr_id_tailq_i *item;
1081  TAILQ_FOREACH(item, tailq, entries) {
1082  item->head = &fctx->__p->pending_fibers;
1083  }
1084  was_empty = TAILQ_EMPTY(&fctx->__p->pending_fibers);
1085  TAILQ_CONCAT(&fctx->__p->pending_fibers, tailq, entries);
1086  if (was_empty && !TAILQ_EMPTY(&fctx->__p->pending_fibers)) {
1087  ev_async_start(fctx->__p->loop, &fctx->__p->pending_async);
1088  }
1089  ev_async_send(fctx->__p->loop, &fctx->__p->pending_async);
1090 }
1091 
1092 void fbr_ev_mutex_init(FBR_P_ struct fbr_ev_mutex *ev,
1093  struct fbr_mutex *mutex)
1094 {
1095  ev_base_init(FBR_A_ &ev->ev_base, FBR_EV_MUTEX);
1096  ev->mutex = mutex;
1097 }
1098 
1099 void fbr_mutex_init(_unused_ FBR_P_ struct fbr_mutex *mutex)
1100 {
1101  mutex->locked_by = 0;
1102  TAILQ_INIT(&mutex->pending);
1103 }
1104 
1105 void fbr_mutex_lock(FBR_P_ struct fbr_mutex *mutex)
1106 {
1107  struct fbr_ev_mutex ev;
1108 
1109  fbr_ev_mutex_init(FBR_A_ &ev, mutex);
1110  fbr_ev_wait_one(FBR_A_ &ev.ev_base);
1111  assert(mutex->locked_by == CURRENT_FIBER_ID);
1112 }
1113 
1114 int fbr_mutex_trylock(FBR_P_ struct fbr_mutex *mutex)
1115 {
1116  if (0 == mutex->locked_by) {
1117  mutex->locked_by = CURRENT_FIBER_ID;
1118  return 1;
1119  }
1120  return 0;
1121 }
1122 
1123 void fbr_mutex_unlock(FBR_P_ struct fbr_mutex *mutex)
1124 {
1125  struct fbr_id_tailq_i *item, *x;
1126  struct fbr_fiber *fiber = NULL;
1127 
1128  if (TAILQ_EMPTY(&mutex->pending)) {
1129  mutex->locked_by = 0;
1130  return;
1131  }
1132 
1133  TAILQ_FOREACH_SAFE(item, &mutex->pending, entries, x) {
1134  assert(item->head == &mutex->pending);
1135  TAILQ_REMOVE(&mutex->pending, item, entries);
1136  if (-1 == fbr_id_unpack(FBR_A_ &fiber, item->id)) {
1137  fbr_log_e(FBR_A_ "libevfibers: unexpected error trying"
1138  " to find a fiber by id: %s",
1139  fbr_strerror(FBR_A_ fctx->f_errno));
1140  continue;
1141  }
1142  break;
1143  }
1144 
1145  mutex->locked_by = item->id;
1146  post_ev(FBR_A_ fiber, item->ev);
1147 
1148  transfer_later(FBR_A_ item);
1149 }
1150 
1151 void fbr_mutex_destroy(_unused_ FBR_P_ _unused_ struct fbr_mutex *mutex)
1152 {
1153  /* Since mutex is stack allocated now, this efffeectively turns into
1154  * NOOP. But we might consider adding some cleanup in the future.
1155  */
1156 }
1157 
1158 void fbr_ev_cond_var_init(FBR_P_ struct fbr_ev_cond_var *ev,
1159  struct fbr_cond_var *cond, struct fbr_mutex *mutex)
1160 {
1161  ev_base_init(FBR_A_ &ev->ev_base, FBR_EV_COND_VAR);
1162  ev->cond = cond;
1163  ev->mutex = mutex;
1164 }
1165 
1166 void fbr_cond_init(_unused_ FBR_P_ struct fbr_cond_var *cond)
1167 {
1168  cond->mutex = NULL;
1169  TAILQ_INIT(&cond->waiting);
1170 }
1171 
1172 void fbr_cond_destroy(_unused_ FBR_P_ _unused_ struct fbr_cond_var *cond)
1173 {
1174  /* Since condvar is stack allocated now, this efffeectively turns into
1175  * NOOP. But we might consider adding some cleanup in the future.
1176  */
1177 }
1178 
1179 int fbr_cond_wait(FBR_P_ struct fbr_cond_var *cond, struct fbr_mutex *mutex)
1180 {
1181  struct fbr_ev_cond_var ev;
1182 
1183  if (0 == mutex->locked_by)
1184  return_error(-1, FBR_EINVAL);
1185 
1186  fbr_ev_cond_var_init(FBR_A_ &ev, cond, mutex);
1187  fbr_ev_wait_one(FBR_A_ &ev.ev_base);
1188  return_success(0);
1189 }
1190 
1191 void fbr_cond_broadcast(FBR_P_ struct fbr_cond_var *cond)
1192 {
1193  struct fbr_id_tailq_i *item;
1194  struct fbr_fiber *fiber;
1195  if (TAILQ_EMPTY(&cond->waiting))
1196  return;
1197  TAILQ_FOREACH(item, &cond->waiting, entries) {
1198  if(-1 == fbr_id_unpack(FBR_A_ &fiber, item->id)) {
1199  assert(FBR_ENOFIBER == fctx->f_errno);
1200  continue;
1201  }
1202  post_ev(FBR_A_ fiber, item->ev);
1203  }
1204  transfer_later_tailq(FBR_A_ &cond->waiting);
1205 }
1206 
1207 void fbr_cond_signal(FBR_P_ struct fbr_cond_var *cond)
1208 {
1209  struct fbr_id_tailq_i *item;
1210  struct fbr_fiber *fiber;
1211  if (TAILQ_EMPTY(&cond->waiting))
1212  return;
1213  item = TAILQ_FIRST(&cond->waiting);
1214  if(-1 == fbr_id_unpack(FBR_A_ &fiber, item->id)) {
1215  assert(FBR_ENOFIBER == fctx->f_errno);
1216  return;
1217  }
1218  post_ev(FBR_A_ fiber, item->ev);
1219 
1220  assert(item->head == &cond->waiting);
1221  TAILQ_REMOVE(&cond->waiting, item, entries);
1222  transfer_later(FBR_A_ item);
1223 }
1224 
1225 int fbr_buffer_init(FBR_P_ struct fbr_buffer *buffer, size_t size)
1226 {
1227  buffer->vrb = vrb_new(size, NULL);
1228  if (NULL == buffer->vrb)
1229  return_error(-1, FBR_ESYSTEM);
1230  buffer->prepared_bytes = 0;
1231  buffer->waiting_bytes = 0;
1232  fbr_cond_init(FBR_A_ &buffer->committed_cond);
1233  fbr_cond_init(FBR_A_ &buffer->bytes_freed_cond);
1234  fbr_mutex_init(FBR_A_ &buffer->write_mutex);
1235  fbr_mutex_init(FBR_A_ &buffer->read_mutex);
1236  return_success(0);
1237 }
1238 
1239 void fbr_buffer_destroy(FBR_P_ struct fbr_buffer *buffer)
1240 {
1241  vrb_destroy(buffer->vrb);
1242 
1243  fbr_mutex_destroy(FBR_A_ &buffer->read_mutex);
1244  fbr_mutex_destroy(FBR_A_ &buffer->write_mutex);
1245  fbr_cond_destroy(FBR_A_ &buffer->committed_cond);
1246  fbr_cond_destroy(FBR_A_ &buffer->bytes_freed_cond);
1247 }
1248 
1249 void *fbr_buffer_alloc_prepare(FBR_P_ struct fbr_buffer *buffer, size_t size)
1250 {
1251  if (size > vrb_capacity(buffer->vrb))
1252  return_error(NULL, FBR_EINVAL);
1253 
1254  fbr_mutex_lock(FBR_A_ &buffer->write_mutex);
1255 
1256  while (buffer->prepared_bytes > 0)
1257  fbr_cond_wait(FBR_A_ &buffer->committed_cond,
1258  &buffer->write_mutex);
1259 
1260  assert(0 == buffer->prepared_bytes);
1261 
1262  buffer->prepared_bytes = size;
1263 
1264  while ((size_t)vrb_space_len(buffer->vrb) < size)
1265  fbr_cond_wait(FBR_A_ &buffer->bytes_freed_cond,
1266  &buffer->write_mutex);
1267 
1268  return vrb_space_ptr(buffer->vrb);
1269 }
1270 
1271 void fbr_buffer_alloc_commit(FBR_P_ struct fbr_buffer *buffer)
1272 {
1273  vrb_give(buffer->vrb, buffer->prepared_bytes);
1274  buffer->prepared_bytes = 0;
1275  fbr_cond_signal(FBR_A_ &buffer->committed_cond);
1276  fbr_mutex_unlock(FBR_A_ &buffer->write_mutex);
1277 }
1278 
1279 void fbr_buffer_alloc_abort(FBR_P_ struct fbr_buffer *buffer)
1280 {
1281  buffer->prepared_bytes = 0;
1282  fbr_cond_signal(FBR_A_ &buffer->committed_cond);
1283  fbr_mutex_unlock(FBR_A_ &buffer->write_mutex);
1284 }
1285 
1286 void *fbr_buffer_read_address(FBR_P_ struct fbr_buffer *buffer, size_t size)
1287 {
1288  int retval;
1289  if (size > vrb_capacity(buffer->vrb))
1290  return_error(NULL, FBR_EINVAL);
1291 
1292  fbr_mutex_lock(FBR_A_ &buffer->read_mutex);
1293 
1294  while ((size_t)vrb_data_len(buffer->vrb) < size) {
1295  retval = fbr_cond_wait(FBR_A_ &buffer->committed_cond,
1296  &buffer->read_mutex);
1297  assert(0 == retval);
1298  }
1299 
1300  buffer->waiting_bytes = size;
1301 
1302  return_success(vrb_data_ptr(buffer->vrb));
1303 }
1304 
1305 void fbr_buffer_read_advance(FBR_P_ struct fbr_buffer *buffer)
1306 {
1307  vrb_take(buffer->vrb, buffer->waiting_bytes);
1308 
1309  fbr_cond_signal(FBR_A_ &buffer->bytes_freed_cond);
1310  fbr_mutex_unlock(FBR_A_ &buffer->read_mutex);
1311 }
1312 
1313 void fbr_buffer_read_discard(FBR_P_ struct fbr_buffer *buffer)
1314 {
1315  fbr_mutex_unlock(FBR_A_ &buffer->read_mutex);
1316 }
1317 
1318 size_t fbr_buffer_bytes(_unused_ FBR_P_ struct fbr_buffer *buffer)
1319 {
1320  return vrb_data_len(buffer->vrb);
1321 }
1322 
1323 size_t fbr_buffer_free_bytes(_unused_ FBR_P_ struct fbr_buffer *buffer)
1324 {
1325  return vrb_space_len(buffer->vrb);
1326 }
1327 
1328 void *fbr_get_user_data(FBR_P_ fbr_id_t id)
1329 {
1330  struct fbr_fiber *fiber;
1331  unpack_transfer_errno(NULL, &fiber, id);
1332  return_success(fiber->user_data);
1333 }
1334 
1335 int fbr_set_user_data(FBR_P_ fbr_id_t id, void *data)
1336 {
1337  struct fbr_fiber *fiber;
1338  unpack_transfer_errno(-1, &fiber, id);
1339  fiber->user_data = data;
1340  return_success(0);
1341 }
1342 
1343 void fbr_destructor_add(FBR_P_ struct fbr_destructor *dtor)
1344 {
1345  struct fbr_fiber *fiber = CURRENT_FIBER;
1346  TAILQ_INSERT_TAIL(&fiber->destructors, dtor, entries);
1347  dtor->active = 1;
1348 }
1349 
1350 void fbr_destructor_remove(FBR_P_ struct fbr_destructor *dtor,
1351  int call)
1352 {
1353  struct fbr_fiber *fiber = CURRENT_FIBER;
1354 
1355  if (0 == dtor->active)
1356  return;
1357 
1358  TAILQ_REMOVE(&fiber->destructors, dtor, entries);
1359  if (call)
1360  dtor->func(FBR_A_ dtor->arg);
1361  dtor->active = 0;
1362 }
1363 
1364 static inline int ffsll(uint64_t val)
1365 {
1366  /* TODO: Add some check for the existance of this builtin */
1367  return __builtin_ffsll(val);
1368 }
1369 
1370 static inline int is_key_registered(FBR_P_ fbr_key_t key)
1371 {
1372  return 0 == (fctx->__p->key_free_mask & (1 << key));
1373 }
1374 
1375 static inline void register_key(FBR_P_ fbr_key_t key)
1376 {
1377  fctx->__p->key_free_mask &= ~(1 << key);
1378 }
1379 
1380 static inline void unregister_key(FBR_P_ fbr_key_t key)
1381 {
1382  fctx->__p->key_free_mask |= (1 << key);
1383 }
1384 
1385 int fbr_key_create(FBR_P_ fbr_key_t *key_ptr)
1386 {
1387  fbr_key_t key = ffsll(fctx->__p->key_free_mask);
1388  assert(key < FBR_MAX_KEY);
1389  register_key(FBR_A_ key);
1390  *key_ptr = key;
1391  return_success(0);
1392 }
1393 
1394 int fbr_key_delete(FBR_P_ fbr_key_t key)
1395 {
1396  if (!is_key_registered(FBR_A_ key))
1397  return_error(-1, FBR_ENOKEY);
1398 
1399  unregister_key(FBR_A_ key);
1400 
1401  return_success(0);
1402 }
1403 
1404 int fbr_key_set(FBR_P_ fbr_id_t id, fbr_key_t key, void *value)
1405 {
1406  struct fbr_fiber *fiber;
1407 
1408  unpack_transfer_errno(-1, &fiber, id);
1409 
1410  if (!is_key_registered(FBR_A_ key))
1411  return_error(-1, FBR_ENOKEY);
1412 
1413  fiber->key_data[key] = value;
1414  return_success(0);
1415 }
1416 
1417 void *fbr_key_get(FBR_P_ fbr_id_t id, fbr_key_t key)
1418 {
1419  struct fbr_fiber *fiber;
1420 
1421  unpack_transfer_errno(NULL, &fiber, id);
1422 
1423  if (!is_key_registered(FBR_A_ key))
1424  return_error(NULL, FBR_ENOKEY);
1425 
1426  return fiber->key_data[key];
1427 }
1428 
1429 const char *fbr_get_name(FBR_P_ fbr_id_t id)
1430 {
1431  struct fbr_fiber *fiber;
1432  unpack_transfer_errno(NULL, &fiber, id);
1433  return_success(fiber->name);
1434 }
1435 
1436 int fbr_set_name(FBR_P_ fbr_id_t id, const char *name)
1437 {
1438  struct fbr_fiber *fiber;
1439  unpack_transfer_errno(-1, &fiber, id);
1440  strncpy(fiber->name, name, FBR_MAX_FIBER_NAME);
1441  return_success(0);
1442 }