|
|
|
|
@ -45,6 +45,7 @@ |
|
|
|
|
#include <grpc/support/string_util.h> |
|
|
|
|
#include <grpc/support/time.h> |
|
|
|
|
#include <grpc/support/useful.h> |
|
|
|
|
#include "src/core/ext/resolver/dns/c_ares/grpc_ares_wrapper.h" |
|
|
|
|
#include "src/core/lib/iomgr/iomgr_internal.h" |
|
|
|
|
#include "src/core/lib/iomgr/sockaddr_utils.h" |
|
|
|
|
#include "src/core/lib/iomgr/unix_sockets_posix.h" |
|
|
|
|
@ -52,48 +53,82 @@ |
|
|
|
|
#include "src/core/lib/support/string.h" |
|
|
|
|
|
|
|
|
|
typedef struct fd_node { |
|
|
|
|
/** the owner of this fd node */ |
|
|
|
|
grpc_ares_ev_driver *ev_driver; |
|
|
|
|
/** refcount of the node */ |
|
|
|
|
gpr_refcount refs; |
|
|
|
|
/** the grpc_fd owned by this fd node */ |
|
|
|
|
grpc_fd *grpc_fd; |
|
|
|
|
/** a closure wrapping on_readable_cb, which should be invoked when the
|
|
|
|
|
grpc_fd in this node becomes readable. */ |
|
|
|
|
grpc_closure read_closure; |
|
|
|
|
/** a closure wrapping on_writable_cb, which should be invoked when the
|
|
|
|
|
grpc_fd in this node becomes writable. */ |
|
|
|
|
grpc_closure write_closure; |
|
|
|
|
/** next fd node in the list */ |
|
|
|
|
struct fd_node *next; |
|
|
|
|
|
|
|
|
|
/** mutex guarding the rest of the state */ |
|
|
|
|
gpr_mu mu; |
|
|
|
|
/** if the readable closure has been registered */ |
|
|
|
|
bool readable_registered; |
|
|
|
|
/** if the writable closure has been registered */ |
|
|
|
|
bool writable_registered; |
|
|
|
|
} fd_node; |
|
|
|
|
|
|
|
|
|
struct grpc_ares_ev_driver { |
|
|
|
|
/** the ares_channel owned by this event driver */ |
|
|
|
|
ares_channel channel; |
|
|
|
|
/** a closure wrapping the driver_cb, which should be invoked each time the ev
|
|
|
|
|
driver gets notified by fds. */ |
|
|
|
|
grpc_closure driver_closure; |
|
|
|
|
/** pollset set for driving the IO events of the channel */ |
|
|
|
|
grpc_pollset_set *pollset_set; |
|
|
|
|
/** has grpc_ares_ev_driver_destroy been called on this event driver? */ |
|
|
|
|
bool closing; |
|
|
|
|
/** an array of ares sockets that the ares channel owned by this event driver
|
|
|
|
|
is currently using */ |
|
|
|
|
ares_socket_t socks[ARES_GETSOCK_MAXNUM]; |
|
|
|
|
/** a bitmask that can tell if an ares socket in the socks array is readable
|
|
|
|
|
or/and writable */ |
|
|
|
|
int socks_bitmask; |
|
|
|
|
/** a list of grpc_fd that this event driver is currently using. */ |
|
|
|
|
fd_node *fds; |
|
|
|
|
|
|
|
|
|
/** mutex guarding the rest of the state */ |
|
|
|
|
gpr_mu mu; |
|
|
|
|
/** a list of grpc_fd that this event driver is currently using. */ |
|
|
|
|
fd_node *fds; |
|
|
|
|
/** is this event driver currently working? */ |
|
|
|
|
bool working; |
|
|
|
|
}; |
|
|
|
|
|
|
|
|
|
static void grpc_ares_notify_on_event(grpc_exec_ctx *exec_ctx, |
|
|
|
|
static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_ares_ev_driver *ev_driver); |
|
|
|
|
|
|
|
|
|
static fd_node *fd_node_ref(fd_node *fdn) { |
|
|
|
|
gpr_log(GPR_DEBUG, "ref %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
gpr_ref(&fdn->refs); |
|
|
|
|
return fdn; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void fd_node_unref(grpc_exec_ctx *exec_ctx, fd_node *fdn) { |
|
|
|
|
gpr_log(GPR_DEBUG, "unref %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
if (gpr_unref(&fdn->refs)) { |
|
|
|
|
gpr_log(GPR_DEBUG, "delete fd: %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
GPR_ASSERT(!fdn->readable_registered); |
|
|
|
|
GPR_ASSERT(!fdn->writable_registered); |
|
|
|
|
gpr_mu_destroy(&fdn->mu); |
|
|
|
|
grpc_pollset_set_del_fd(exec_ctx, fdn->ev_driver->pollset_set, |
|
|
|
|
fdn->grpc_fd); |
|
|
|
|
grpc_fd_shutdown(exec_ctx, fdn->grpc_fd); |
|
|
|
|
grpc_fd_orphan(exec_ctx, fdn->grpc_fd, NULL, NULL, "c-ares query finished"); |
|
|
|
|
gpr_free(fdn); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_error *grpc_ares_ev_driver_create(grpc_ares_ev_driver **ev_driver, |
|
|
|
|
grpc_pollset_set *pollset_set) { |
|
|
|
|
int status; |
|
|
|
|
grpc_error *err = grpc_ares_init(); |
|
|
|
|
if (err != GRPC_ERROR_NONE) { |
|
|
|
|
return err; |
|
|
|
|
} |
|
|
|
|
*ev_driver = gpr_malloc(sizeof(grpc_ares_ev_driver)); |
|
|
|
|
status = ares_init(&(*ev_driver)->channel); |
|
|
|
|
gpr_log(GPR_DEBUG, "grpc_ares_ev_driver_create\n"); |
|
|
|
|
if (status != ARES_SUCCESS) { |
|
|
|
|
char *err_msg; |
|
|
|
|
gpr_asprintf(&err_msg, "Failed to init ares channel. C-ares error: %s", |
|
|
|
|
ares_strerror(status)); |
|
|
|
|
grpc_error *err = GRPC_ERROR_CREATE(err_msg); |
|
|
|
|
err = GRPC_ERROR_CREATE(err_msg); |
|
|
|
|
gpr_free(err_msg); |
|
|
|
|
gpr_free(*ev_driver); |
|
|
|
|
return err; |
|
|
|
|
@ -101,17 +136,43 @@ grpc_error *grpc_ares_ev_driver_create(grpc_ares_ev_driver **ev_driver, |
|
|
|
|
gpr_mu_init(&(*ev_driver)->mu); |
|
|
|
|
(*ev_driver)->pollset_set = pollset_set; |
|
|
|
|
(*ev_driver)->fds = NULL; |
|
|
|
|
(*ev_driver)->closing = false; |
|
|
|
|
(*ev_driver)->working = false; |
|
|
|
|
return GRPC_ERROR_NONE; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void grpc_ares_ev_driver_destroy(grpc_ares_ev_driver *ev_driver) { |
|
|
|
|
ev_driver->closing = true; |
|
|
|
|
static void grpc_ares_ev_driver_cleanup(grpc_exec_ctx *exec_ctx, void *arg, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
GPR_ASSERT(error == GRPC_ERROR_NONE); |
|
|
|
|
grpc_ares_ev_driver *ev_driver = arg; |
|
|
|
|
GPR_ASSERT(ev_driver->fds == NULL); |
|
|
|
|
gpr_mu_lock(&ev_driver->mu); |
|
|
|
|
gpr_mu_unlock(&ev_driver->mu); |
|
|
|
|
gpr_mu_destroy(&ev_driver->mu); |
|
|
|
|
ares_destroy(ev_driver->channel); |
|
|
|
|
gpr_free(ev_driver); |
|
|
|
|
grpc_ares_cleanup(); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
void grpc_ares_ev_driver_destroy(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_ares_ev_driver *ev_driver) { |
|
|
|
|
// Shutdowe all the working fds, invoke their resgistered on_readable_cb and
|
|
|
|
|
// on_writable_cb.
|
|
|
|
|
gpr_mu_lock(&ev_driver->mu); |
|
|
|
|
fd_node *fdn; |
|
|
|
|
for (fdn = ev_driver->fds; fdn; fdn = fdn->next) { |
|
|
|
|
grpc_fd_shutdown(exec_ctx, fdn->grpc_fd); |
|
|
|
|
fdn = fdn->next; |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&ev_driver->mu); |
|
|
|
|
// Schedule the actual cleanup with exec_ctx, so that it happens after the
|
|
|
|
|
// fd shutdown process.
|
|
|
|
|
grpc_exec_ctx_sched( |
|
|
|
|
exec_ctx, grpc_closure_create(grpc_ares_ev_driver_cleanup, ev_driver), |
|
|
|
|
GRPC_ERROR_NONE, NULL); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Search fd in the fd_node list head. This is an O(n) search, the max possible
|
|
|
|
|
// value of n is ARES_GETSOCK_MAXNUM (16). n is typically 1 - 3 in our tests.
|
|
|
|
|
// value of n is ARES_GETSOCK_MAXNUM (16). n is typically 1 - 2 in our tests.
|
|
|
|
|
static fd_node *get_fd(fd_node **head, int fd) { |
|
|
|
|
fd_node dummy_head; |
|
|
|
|
fd_node *node; |
|
|
|
|
@ -131,93 +192,132 @@ static fd_node *get_fd(fd_node **head, int fd) { |
|
|
|
|
return NULL; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Process each file descriptor that may wake this callback up.
|
|
|
|
|
static void driver_cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { |
|
|
|
|
grpc_ares_ev_driver *d = arg; |
|
|
|
|
size_t i; |
|
|
|
|
static void on_readable_cb(grpc_exec_ctx *exec_ctx, void *arg, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
fd_node *fdn = arg; |
|
|
|
|
grpc_ares_ev_driver *ev_driver = fdn->ev_driver; |
|
|
|
|
gpr_mu_lock(&fdn->mu); |
|
|
|
|
fdn->readable_registered = false; |
|
|
|
|
gpr_mu_unlock(&fdn->mu); |
|
|
|
|
|
|
|
|
|
gpr_log(GPR_DEBUG, "readable on %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
for (i = 0; i < ARES_GETSOCK_MAXNUM; i++) { |
|
|
|
|
ares_socket_t read_fd = ARES_GETSOCK_READABLE(d->socks_bitmask, i) |
|
|
|
|
? d->socks[i] |
|
|
|
|
: ARES_SOCKET_BAD; |
|
|
|
|
ares_socket_t write_fd = ARES_GETSOCK_WRITABLE(d->socks_bitmask, i) |
|
|
|
|
? d->socks[i] |
|
|
|
|
: ARES_SOCKET_BAD; |
|
|
|
|
ares_process_fd(d->channel, read_fd, write_fd); |
|
|
|
|
ares_process_fd(ev_driver->channel, grpc_fd_wrapped_fd(fdn->grpc_fd), |
|
|
|
|
ARES_SOCKET_BAD); |
|
|
|
|
} else { |
|
|
|
|
// If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or
|
|
|
|
|
// timed out. The pending lookups made on this ev_driver will be cancelled
|
|
|
|
|
// by the following ares_canncel() and the on done callbacks will be invoked
|
|
|
|
|
// with a status of ARES_ECANCELLED. The remaining file descriptors in this
|
|
|
|
|
// ev_driver will be cleaned up in the follwing
|
|
|
|
|
// grpc_ares_notify_on_event_locked().
|
|
|
|
|
ares_cancel(ev_driver->channel); |
|
|
|
|
} |
|
|
|
|
fd_node_unref(exec_ctx, fdn); |
|
|
|
|
gpr_mu_lock(&ev_driver->mu); |
|
|
|
|
grpc_ares_notify_on_event_locked(exec_ctx, ev_driver); |
|
|
|
|
gpr_mu_unlock(&ev_driver->mu); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void on_writable_cb(grpc_exec_ctx *exec_ctx, void *arg, |
|
|
|
|
grpc_error *error) { |
|
|
|
|
fd_node *fdn = arg; |
|
|
|
|
grpc_ares_ev_driver *ev_driver = fdn->ev_driver; |
|
|
|
|
gpr_mu_lock(&fdn->mu); |
|
|
|
|
fdn->writable_registered = false; |
|
|
|
|
gpr_mu_unlock(&fdn->mu); |
|
|
|
|
|
|
|
|
|
gpr_log(GPR_DEBUG, "writable on %d", grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
if (error == GRPC_ERROR_NONE) { |
|
|
|
|
ares_process_fd(ev_driver->channel, ARES_SOCKET_BAD, |
|
|
|
|
grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
} else { |
|
|
|
|
// error != GRPC_ERROR_NONE means the waiting timed out or the fd has been
|
|
|
|
|
// shutdown. In this case, the event driver cancels all the ongoing requests
|
|
|
|
|
// that are using its channel. The fds get cleaned up in the next
|
|
|
|
|
// grpc_ares_notify_on_event.
|
|
|
|
|
ares_cancel(d->channel); |
|
|
|
|
// If error is not GRPC_ERROR_NONE, it means the fd has been shutdown or
|
|
|
|
|
// timed out. The pending lookups made on this ev_driver will be cancelled
|
|
|
|
|
// by the following ares_canncel() and the on done callbacks will be invoked
|
|
|
|
|
// with a status of ARES_ECANCELLED. The remaining file descriptors in this
|
|
|
|
|
// ev_driver will be cleaned up in the follwing
|
|
|
|
|
// grpc_ares_notify_on_event_locked().
|
|
|
|
|
ares_cancel(ev_driver->channel); |
|
|
|
|
} |
|
|
|
|
grpc_ares_notify_on_event(exec_ctx, d); |
|
|
|
|
fd_node_unref(exec_ctx, fdn); |
|
|
|
|
gpr_mu_lock(&ev_driver->mu); |
|
|
|
|
grpc_ares_notify_on_event_locked(exec_ctx, ev_driver); |
|
|
|
|
gpr_mu_unlock(&ev_driver->mu); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ares_channel *grpc_ares_ev_driver_get_channel(grpc_ares_ev_driver *ev_driver) { |
|
|
|
|
void *grpc_ares_ev_driver_get_channel(grpc_ares_ev_driver *ev_driver) { |
|
|
|
|
return &ev_driver->channel; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Get the file descriptors used by the ev_driver's ares channel, register
|
|
|
|
|
// driver_closure with these filedescriptors.
|
|
|
|
|
static void grpc_ares_notify_on_event(grpc_exec_ctx *exec_ctx, |
|
|
|
|
static void grpc_ares_notify_on_event_locked(grpc_exec_ctx *exec_ctx, |
|
|
|
|
grpc_ares_ev_driver *ev_driver) { |
|
|
|
|
size_t i; |
|
|
|
|
fd_node *new_list = NULL; |
|
|
|
|
if (!ev_driver->closing) { |
|
|
|
|
ev_driver->socks_bitmask = |
|
|
|
|
ares_getsock(ev_driver->channel, ev_driver->socks, ARES_GETSOCK_MAXNUM); |
|
|
|
|
grpc_closure_init(&ev_driver->driver_closure, driver_cb, ev_driver); |
|
|
|
|
gpr_log(GPR_DEBUG, "notify_on_event\n"); |
|
|
|
|
ares_socket_t socks[ARES_GETSOCK_MAXNUM]; |
|
|
|
|
int socks_bitmask = |
|
|
|
|
ares_getsock(ev_driver->channel, socks, ARES_GETSOCK_MAXNUM); |
|
|
|
|
size_t i; |
|
|
|
|
for (i = 0; i < ARES_GETSOCK_MAXNUM; i++) { |
|
|
|
|
if (ARES_GETSOCK_READABLE(ev_driver->socks_bitmask, i) || |
|
|
|
|
ARES_GETSOCK_WRITABLE(ev_driver->socks_bitmask, i)) { |
|
|
|
|
fd_node *fdn = get_fd(&ev_driver->fds, ev_driver->socks[i]); |
|
|
|
|
if (ARES_GETSOCK_READABLE(socks_bitmask, i) || |
|
|
|
|
ARES_GETSOCK_WRITABLE(socks_bitmask, i)) { |
|
|
|
|
fd_node *fdn = get_fd(&ev_driver->fds, socks[i]); |
|
|
|
|
// Create a new fd_node if sock[i] is not in the fd_node list.
|
|
|
|
|
if (fdn == NULL) { |
|
|
|
|
char *fd_name; |
|
|
|
|
gpr_asprintf(&fd_name, "ares_ev_driver-%" PRIuPTR, i); |
|
|
|
|
fdn = gpr_malloc(sizeof(fd_node)); |
|
|
|
|
fdn->grpc_fd = grpc_fd_create(ev_driver->socks[i], fd_name); |
|
|
|
|
grpc_pollset_set_add_fd(exec_ctx, ev_driver->pollset_set, |
|
|
|
|
fdn->grpc_fd); |
|
|
|
|
gpr_log(GPR_DEBUG, "new fd: %d", socks[i]); |
|
|
|
|
fdn->grpc_fd = grpc_fd_create(socks[i], fd_name); |
|
|
|
|
fdn->ev_driver = ev_driver; |
|
|
|
|
fdn->readable_registered = false; |
|
|
|
|
fdn->writable_registered = false; |
|
|
|
|
gpr_mu_init(&fdn->mu); |
|
|
|
|
gpr_ref_init(&fdn->refs, 1); |
|
|
|
|
grpc_closure_init(&fdn->read_closure, on_readable_cb, fdn); |
|
|
|
|
grpc_closure_init(&fdn->write_closure, on_writable_cb, fdn); |
|
|
|
|
grpc_pollset_set_add_fd(exec_ctx, ev_driver->pollset_set, fdn->grpc_fd); |
|
|
|
|
gpr_free(fd_name); |
|
|
|
|
} |
|
|
|
|
fdn->next = new_list; |
|
|
|
|
new_list = fdn; |
|
|
|
|
|
|
|
|
|
if (ARES_GETSOCK_READABLE(ev_driver->socks_bitmask, i)) { |
|
|
|
|
grpc_fd_notify_on_read(exec_ctx, fdn->grpc_fd, |
|
|
|
|
&ev_driver->driver_closure); |
|
|
|
|
gpr_mu_lock(&fdn->mu); |
|
|
|
|
// Register read_closure if the socket is readable and read_closure has
|
|
|
|
|
// not been registered with this socket.
|
|
|
|
|
if (ARES_GETSOCK_READABLE(socks_bitmask, i) && |
|
|
|
|
!fdn->readable_registered) { |
|
|
|
|
fd_node_ref(fdn); |
|
|
|
|
gpr_log(GPR_DEBUG, "notify read on: %d", |
|
|
|
|
grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
grpc_fd_notify_on_read(exec_ctx, fdn->grpc_fd, &fdn->read_closure); |
|
|
|
|
fdn->readable_registered = true; |
|
|
|
|
} |
|
|
|
|
if (ARES_GETSOCK_WRITABLE(ev_driver->socks_bitmask, i)) { |
|
|
|
|
grpc_fd_notify_on_write(exec_ctx, fdn->grpc_fd, |
|
|
|
|
&ev_driver->driver_closure); |
|
|
|
|
// Register write_closure if the socket is writable and write_closure has
|
|
|
|
|
// not been registered with this socket.
|
|
|
|
|
if (ARES_GETSOCK_WRITABLE(socks_bitmask, i) && |
|
|
|
|
!fdn->writable_registered) { |
|
|
|
|
gpr_log(GPR_DEBUG, "notify write on: %d", |
|
|
|
|
grpc_fd_wrapped_fd(fdn->grpc_fd)); |
|
|
|
|
fd_node_ref(fdn); |
|
|
|
|
grpc_fd_notify_on_write(exec_ctx, fdn->grpc_fd, &fdn->write_closure); |
|
|
|
|
fdn->writable_registered = true; |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&fdn->mu); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
while (ev_driver->fds != NULL) { |
|
|
|
|
fd_node *cur = ev_driver->fds; |
|
|
|
|
ev_driver->fds = ev_driver->fds->next; |
|
|
|
|
grpc_pollset_set_del_fd(exec_ctx, ev_driver->pollset_set, cur->grpc_fd); |
|
|
|
|
grpc_fd_shutdown(exec_ctx, cur->grpc_fd); |
|
|
|
|
grpc_fd_orphan(exec_ctx, cur->grpc_fd, NULL, NULL, "c-ares query finished"); |
|
|
|
|
gpr_free(cur); |
|
|
|
|
fd_node_unref(exec_ctx, cur); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ev_driver->fds = new_list; |
|
|
|
|
// If the ev driver has no working fd, all the tasks are done.
|
|
|
|
|
if (!new_list) { |
|
|
|
|
gpr_mu_lock(&ev_driver->mu); |
|
|
|
|
ev_driver->working = false; |
|
|
|
|
gpr_mu_unlock(&ev_driver->mu); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if (ev_driver->closing) { |
|
|
|
|
ares_destroy(ev_driver->channel); |
|
|
|
|
gpr_free(ev_driver); |
|
|
|
|
gpr_log(GPR_DEBUG, "ev driver stop working"); |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -229,8 +329,8 @@ void grpc_ares_ev_driver_start(grpc_exec_ctx *exec_ctx, |
|
|
|
|
return; |
|
|
|
|
} |
|
|
|
|
ev_driver->working = true; |
|
|
|
|
grpc_ares_notify_on_event_locked(exec_ctx, ev_driver); |
|
|
|
|
gpr_mu_unlock(&ev_driver->mu); |
|
|
|
|
grpc_ares_notify_on_event(exec_ctx, ev_driver); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#endif /* GPR_POSIX_SOCKET */ |
|
|
|
|
|
Yuchen Zeng
8 years ago