Commit 51bc0e7f authored by Nick Mathewson's avatar Nick Mathewson 🦀
Browse files

Isolate the "socketpair or a pipe" logic for alerting main thread 

This way we can use the linux eventfd extension where available.
Using EVFILT_USER on the BSDs will be a teeny bit trickier, and will
require libevent hacking.
parent c7eebe23

configure.ac

+4 −0
Original line number Diff line number Diff line
@@ -407,6 +407,7 @@ AC_CHECK_FUNCS( 
        backtrace \

        backtrace_symbols_fd \

        clock_gettime \

	eventfd \

        flock \

        ftime \

        getaddrinfo \
@@ -421,6 +422,8 @@ AC_CHECK_FUNCS( 
        localtime_r \

        lround \

        memmem \

	pipe \

	pipe2 \

        prctl \

        rint \

        sigaction \
@@ -962,6 +965,7 @@ AC_CHECK_HEADERS( 
        netinet/in6.h \

        pwd.h \

        stdint.h \

	sys/eventfd.h \

        sys/file.h \

        sys/ioctl.h \

        sys/limits.h \

src/common/compat_threads.c

+150 −0
Original line number Diff line number Diff line
@@ -3,8 +3,24 @@ 
 * Copyright (c) 2007-2015, The Tor Project, Inc. */

/* See LICENSE for licensing information */



#include "orconfig.h"

#define _GNU_SOURCE

#include <stdlib.h>

#include "compat.h"

#include "compat_threads.h"



#include "util.h"

#include "torlog.h"



#ifdef HAVE_SYS_EVENTFD_H

#include <sys/eventfd.h>

#endif

#ifdef HAVE_UNISTD_H

#include <unistd.h>

#endif

#ifdef HAVE_FCNTL_H

#include <fcntl.h>

#endif



/** Return a newly allocated, ready-for-use mutex. */

tor_mutex_t *
@@ -57,3 +73,137 @@ in_main_thread(void) 
{

  return main_thread_id == tor_get_thread_id();

}



#ifdef HAVE_EVENTFD

static int

eventfd_alert(int fd)

{

  uint64_t u = 1;

  int r = write(fd, (void*)&u, sizeof(u));

  if (r < 0 && errno != EAGAIN)

    return -1;

  return 0;

}



static int

eventfd_drain(int fd)

{

  uint64_t u = 0;

  int r = read(fd, (void*)&u, sizeof(u));

  if (r < 0 && errno != EAGAIN)

    return -1;

  return 0;

}

#endif



#ifdef HAVE_PIPE

static int

pipe_alert(int fd)

{

  ssize_t r = write(fd, "x", 1);

  if (r < 0 && errno != EAGAIN)

    return -1;

  return 0;

}



static int

pipe_drain(int fd)

{

  char buf[32];

  ssize_t r;

  while ((r = read(fd, buf, sizeof(buf))) >= 0)

    ;

  if (r == 0 || errno != EAGAIN)

    return -1;

  return 0;

}

#endif



static int

sock_alert(tor_socket_t fd)

{

  ssize_t r = send(fd, "x", 1, 0);

  if (r < 0 && !ERRNO_IS_EAGAIN(tor_socket_errno(fd)))

    return -1;

  return 0;

}



static int

sock_drain(tor_socket_t fd)

{

  char buf[32];

  ssize_t r;

  while ((r = recv(fd, buf, sizeof(buf), 0)) >= 0)

    ;

  if (r == 0 || !ERRNO_IS_EAGAIN(tor_socket_errno(fd)))

    return -1;

  return 0;

}



/** Allocate a new set of alert sockets. DOCDOC */

int

alert_sockets_create(alert_sockets_t *socks_out)

{

  tor_socket_t socks[2];



#ifdef HAVE_EVENTFD

#if defined(EFD_CLOEXEC) && defined(EFD_NONBLOCK)

  socks[0] = eventfd(0, EFD_CLOEXEC|EFD_NONBLOCK);

#else

  socks[0] = -1;

#endif

  if (socks[0] < 0) {

    socks[0] = eventfd(0,0);

    if (socks[0] >= 0) {

      if (fcntl(socks[0], F_SETFD, FD_CLOEXEC) < 0 ||

          set_socket_nonblocking(socks[0]) < 0) {

        close(socks[0]);

        return -1;

      }

    }

  }

  if (socks[0] >= 0) {

    socks_out->read_fd = socks_out->write_fd = socks[0];

    socks_out->alert_fn = eventfd_alert;

    socks_out->drain_fn = eventfd_drain;

    return 0;

  }

#endif



#ifdef HAVE_PIPE2

  if (pipe2(socks, O_NONBLOCK|O_CLOEXEC) == 0) {

    socks_out->read_fd = socks[0];

    socks_out->write_fd = socks[1];

    socks_out->alert_fn = pipe_alert;

    socks_out->drain_fn = pipe_drain;

    return 0;

  }

#endif



#ifdef HAVE_PIPE

  if (pipe(socks) == 0) {

    if (fcntl(socks[0], F_SETFD, FD_CLOEXEC) < 0 ||

        fcntl(socks[1], F_SETFD, FD_CLOEXEC) < 0 ||

        set_socket_nonblocking(socks[0]) < 0 ||

        set_socket_nonblocking(socks[1]) < 0) {

      close(socks[0]);

      close(socks[1]);

      return -1;

    }

    socks_out->read_fd = socks[0];

    socks_out->write_fd = socks[1];

    socks_out->alert_fn = pipe_alert;

    socks_out->drain_fn = pipe_drain;

    return 0;

  }

#endif



  if (tor_socketpair(AF_UNIX, SOCK_STREAM, 0, socks) == 0) {

    set_socket_nonblocking(socks[0]);

    set_socket_nonblocking(socks[1]);

    socks_out->alert_fn = sock_alert;

    socks_out->drain_fn = sock_drain;

    return 0;

  }

  return -1;

}

src/common/compat_threads.h

+11 −0
Original line number Diff line number Diff line
@@ -82,4 +82,15 @@ int tor_cond_wait(tor_cond_t *cond, tor_mutex_t *mutex, 
void tor_cond_signal_one(tor_cond_t *cond);

void tor_cond_signal_all(tor_cond_t *cond);



/** DOCDOC */

typedef struct alert_sockets_s {

  /*XXX needs a better name */

  tor_socket_t read_fd;

  tor_socket_t write_fd;

  int (*alert_fn)(tor_socket_t write_fd);

  int (*drain_fn)(tor_socket_t read_fd);

} alert_sockets_t;



int alert_sockets_create(alert_sockets_t *socks_out);



#endif

src/common/workqueue.c

+12 −56
Original line number Diff line number Diff line
@@ -9,11 +9,6 @@ 
#include "tor_queue.h"

#include "torlog.h"



#ifdef HAVE_UNISTD_H

// XXXX move wherever we move the write/send stuff

#include <unistd.h>

#endif



/*

  design:
@@ -44,9 +39,7 @@ struct replyqueue_s { 
  tor_mutex_t lock;

  TOR_TAILQ_HEAD(, workqueue_entry_s) answers;



  void (*alert_fn)(struct replyqueue_s *); // lock not held on this, next 2. 

  tor_socket_t write_sock;

  tor_socket_t read_sock;

  alert_sockets_t alert; // lock not held on this.

};



typedef struct workerthread_s {
@@ -169,20 +162,10 @@ queue_reply(replyqueue_t *queue, workqueue_entry_t *work) 
  tor_mutex_release(&queue->lock);



  if (was_empty) {

    queue->alert_fn(queue);

    if (queue->alert.alert_fn(queue->alert.write_fd) < 0) {

      /* XXXX complain! */

    }

  }





static void

alert_by_fd(replyqueue_t *queue)

{

  /* XXX extract this into new function */

#ifndef _WIN32

  (void) send(queue->write_sock, "x", 1, 0);

#else

  (void) write(queue->write_sock, "x", 1);

#endif

}



static workerthread_t *
@@ -293,59 +276,32 @@ threadpool_get_replyqueue(threadpool_t *tp) 
replyqueue_t *

replyqueue_new(void)

{

  tor_socket_t pair[2];

  replyqueue_t *rq;

  int r;



  /* XXX extract this into new function */

#ifdef _WIN32

  r = tor_socketpair(AF_UNIX, SOCK_STREAM, 0, pair);

#else

  r = pipe(pair);

#endif

  if (r < 0)

    return NULL;



  set_socket_nonblocking(pair[0]); /* the read-size should be nonblocking. */

#if defined(FD_CLOEXEC)

  fcntl(pair[0], F_SETFD, FD_CLOEXEC);

  fcntl(pair[1], F_SETFD, FD_CLOEXEC);

#endif



  rq = tor_malloc_zero(sizeof(replyqueue_t));

  if (alert_sockets_create(&rq->alert) < 0) {

    tor_free(rq);

    return NULL;

  }



  tor_mutex_init(&rq->lock);

  TOR_TAILQ_INIT(&rq->answers);



  rq->read_sock = pair[0];

  rq->write_sock = pair[1];

  rq->alert_fn = alert_by_fd;



  return rq;

}



tor_socket_t

replyqueue_get_socket(replyqueue_t *rq)

{

  return rq->read_sock;

  return rq->alert.read_fd;

}



void

replyqueue_process(replyqueue_t *queue)

{

  ssize_t r;



  /* XXX extract this into new function */

  do {

    char buf[64];

#ifdef _WIN32

    r = recv(queue->read_sock, buf, sizeof(buf), 0);

#else

    r = read(queue->read_sock, buf, sizeof(buf));

#endif

  } while (r > 0);



  /* XXXX freak out on r == 0, or r == "error, not retryable". */

  if (queue->alert.drain_fn(queue->alert.read_fd) < 0) {

    /* XXXX complain! */

  }



  tor_mutex_acquire(&queue->lock);

  while (!TOR_TAILQ_EMPTY(&queue->answers)) {