async.c File Reference

#include "uv.h"
#include "internal.h"
#include "atomic-ops.h"
#include <errno.h>
#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sched.h>

Go to the source code of this file.

Functions
static void	uv__async_send (uv_loop_t *loop)
static int	uv__async_start (uv_loop_t *loop)
int	uv_async_init (uv_loop_t *loop, uv_async_t *handle, uv_async_cb async_cb)
int	uv_async_send (uv_async_t *handle)
static int	uv__async_spin (uv_async_t *handle)
void	uv__async_close (uv_async_t *handle)
static void	uv__async_io (uv_loop_t *loop, uv__io_t *w, unsigned int events)
int	uv__async_fork (uv_loop_t *loop)
void	uv__async_stop (uv_loop_t *loop)

Function Documentation

uv__async_close()

void uv__async_close

(

uv_async_t *

handle

)

Definition at line 115 of file async.c.

                                         {
  uv__async_spin(handle);
  QUEUE_REMOVE(&handle->queue);
  uv__handle_stop(handle);
}

References handle, QUEUE_REMOVE, uv__async_spin(), and uv__handle_stop.

Referenced by uv_close().

uv__async_fork()

int uv__async_fork

(

uv_loop_t *

loop

)

Definition at line 230 of file async.c.

                                    {
  if (loop->async_io_watcher.fd == -1) /* never started */
    return 0;
 
  uv__async_stop(loop);
 
  return uv__async_start(loop);
}

References loop, uv__async_start(), and uv__async_stop().

Referenced by uv_loop_fork().

uv__async_io()

static void uv__async_io ( uv_loop_t *  loop, uv__io_t *  w, unsigned int  events  )

static

Definition at line 122 of file async.c.

                                                                            {
  char buf[1024];
  ssize_t r;
  QUEUE queue;
  QUEUE* q;
  uv_async_t* h;
 
  assert(w == &loop->async_io_watcher);
 
  for (;;) {
    r = read(w->fd, buf, sizeof(buf));
 
    if (r == sizeof(buf))
      continue;
 
    if (r != -1)
      break;
 
    if (errno == EAGAIN || errno == EWOULDBLOCK)
      break;
 
    if (errno == EINTR)
      continue;
 
    abort();
  }
 
  QUEUE_MOVE(&loop->async_handles, &queue);
  while (!QUEUE_EMPTY(&queue)) {
    q = QUEUE_HEAD(&queue);
    h = QUEUE_DATA(q, uv_async_t, queue);
 
    QUEUE_REMOVE(q);
    QUEUE_INSERT_TAIL(&loop->async_handles, q);
 
    if (0 == uv__async_spin(h))
      continue;  /* Not pending. */
 
    if (h->async_cb == NULL)
      continue;
 
    h->async_cb(h);
  }
}

References assert(), EAGAIN, EINTR, h, loop, NULL, queue, QUEUE_DATA, QUEUE_EMPTY, QUEUE_HEAD, QUEUE_INSERT_TAIL, QUEUE_MOVE, QUEUE_REMOVE, r, read(), uv__async_spin(), and w.

Referenced by uv__async_start().

uv__async_send()

static void uv__async_send ( uv_loop_t *  loop )

static

Definition at line 168 of file async.c.

                                            {
  const void* buf;
  ssize_t len;
  int fd;
  int r;
 
  buf = "";
  len = 1;
  fd = loop->async_wfd;
 
#if defined(__linux__)
  if (fd == -1) {
    static const uint64_t val = 1;
    buf = &val;
    len = sizeof(val);
    fd = loop->async_io_watcher.fd;  /* eventfd */
  }
#endif
 
  do
    r = write(fd, buf, len);
  while (r == -1 && errno == EINTR);
 
  if (r == len)
    return;
 
  if (r == -1)
    if (errno == EAGAIN || errno == EWOULDBLOCK)
      return;
 
  abort();
}

References EAGAIN, EINTR, fd, len, loop, r, val, and write.

Referenced by uv_async_send().

uv__async_spin()

static int uv__async_spin ( uv_async_t *  handle )

static

Definition at line 84 of file async.c.

                                              {
  int i;
  int rc;
 
  for (;;) {
    /* 997 is not completely chosen at random. It's a prime number, acyclical
     * by nature, and should therefore hopefully dampen sympathetic resonance.
     */
    for (i = 0; i < 997; i++) {
      /* rc=0 -- handle is not pending.
       * rc=1 -- handle is pending, other thread is still working with it.
       * rc=2 -- handle is pending, other thread is done.
       */
      rc = cmpxchgi(&handle->pending, 2, 0);
 
      if (rc != 1)
        return rc;
 
      /* Other thread is busy with this handle, spin until it's done. */
      cpu_relax();
    }
 
    /* Yield the CPU. We may have preempted the other thread while it's
     * inside the critical section and if it's running on the same CPU
     * as us, we'll just burn CPU cycles until the end of our time slice.
     */
    sched_yield();
  }
}

References handle, and i.

Referenced by uv__async_close(), and uv__async_io().

uv__async_start()

static int uv__async_start ( uv_loop_t *  loop )

static

Definition at line 202 of file async.c.

                                            {
  int pipefd[2];
  int err;
 
  if (loop->async_io_watcher.fd != -1)
    return 0;
 
#ifdef __linux__
  err = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
  if (err < 0)
    return UV__ERR(errno);
 
  pipefd[0] = err;
  pipefd[1] = -1;
#else
  err = uv__make_pipe(pipefd, UV__F_NONBLOCK);
  if (err < 0)
    return err;
#endif
 
  uv__io_init(&loop->async_io_watcher, uv__async_io, pipefd[0]);
  uv__io_start(loop, &loop->async_io_watcher, POLLIN);
  loop->async_wfd = pipefd[1];
 
  return 0;
}

References err, loop, uv__async_io(), UV__ERR, UV__F_NONBLOCK, uv__io_init(), uv__io_start(), and uv__make_pipe().

Referenced by uv__async_fork(), and uv_async_init().

uv__async_stop()

void uv__async_stop

(

uv_loop_t *

loop

)

Definition at line 240 of file async.c.

                                     {
  if (loop->async_io_watcher.fd == -1)
    return;
 
  if (loop->async_wfd != -1) {
    if (loop->async_wfd != loop->async_io_watcher.fd)
      uv__close(loop->async_wfd);
    loop->async_wfd = -1;
  }
 
  uv__io_stop(loop, &loop->async_io_watcher, POLLIN);
  uv__close(loop->async_io_watcher.fd);
  loop->async_io_watcher.fd = -1;
}

References loop, uv__close(), and uv__io_stop().

Referenced by uv__async_fork(), and uv__loop_close().

uv_async_init()

int uv_async_init	(	uv_loop_t *	loop,
		uv_async_t *	handle,
		uv_async_cb	async_cb
	)

Definition at line 45 of file async.c.

                                                                             {
  int err;
 
  err = uv__async_start(loop);
  if (err)
    return err;
 
  uv__handle_init(loop, (uv_handle_t*)handle, UV_ASYNC);
  handle->async_cb = async_cb;
  handle->pending = 0;
 
  QUEUE_INSERT_TAIL(&loop->async_handles, &handle->queue);
  uv__handle_start(handle);
 
  return 0;
}

uv_async_send()

int uv_async_send

(

uv_async_t *

handle

)

Definition at line 63 of file async.c.

                                      {
  /* Do a cheap read first. */
  if (ACCESS_ONCE(int, handle->pending) != 0)
    return 0;
 
  /* Tell the other thread we're busy with the handle. */
  if (cmpxchgi(&handle->pending, 0, 1) != 0)
    return 0;
 
  /* Wake up the other thread's event loop. */
  uv__async_send(handle->loop);
 
  /* Tell the other thread we're done. */
  if (cmpxchgi(&handle->pending, 1, 2) != 1)
    abort();
 
  return 0;
}