| /* ==================================================================== |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * ==================================================================== |
| */ |
| |
| #if defined(_DEBUG) |
| #include <assert.h> |
| #define SERF__RESOLV_assert(x) assert(x) |
| #else |
| #define SERF__RESOLV_assert(x) ((void)0) |
| #endif |
| |
| #include <apr.h> |
| #include <apr_version.h> |
| |
| #define APR_WANT_BYTEFUNC |
| #define APR_WANT_MEMFUNC |
| #include <apr_want.h> |
| |
| #include <apr_errno.h> |
| #include <apr_pools.h> |
| #include <apr_atomic.h> |
| #include <apr_strings.h> |
| #include <apr_network_io.h> |
| #include <apr_thread_mutex.h> |
| #include <apr_thread_pool.h> |
| |
| /* Third-party resolver headers. */ |
| #if SERF_HAVE_ASYNC_RESOLVER |
| #if SERF_HAVE_UNBOUND |
| #include <unbound.h> |
| #else |
| /* Really shouldn't happen, but just in case it does, fall back |
| to the apr_thread_pool-based resolver. */ |
| #undef SERF_HAVE_ASYNC_RESOLVER |
| #endif /* SERF_HAVE_UNBOUND */ |
| #endif |
| |
| #include "serf.h" |
| #include "serf_private.h" |
| |
| /* Sufficient buffer size for an IPv4 and IPv6 binary address. */ |
| #define MAX_ADDRLEN 16 |
| |
| /* Stringified address lengths. */ |
| #ifndef INET_ADDRSTRLEN |
| #define INET_ADDRSTRLEN sizeof("255.255.255.255") |
| #endif |
| #ifndef INET6_ADDRSTRLEN |
| #define INET6_ADDRSTRLEN sizeof("ffff:ffff:ffff:ffff:ffff:ffff:255.255.255.255") |
| #endif |
| |
| #define HAVE_ASYNC_RESOLVER (SERF_HAVE_ASYNC_RESOLVER || APR_HAS_THREADS) |
| |
| |
| #if HAVE_ASYNC_RESOLVER |
| |
| /* Pushes the result of a successful or failed address resolution |
| onto the context's result queue. */ |
| static void push_resolve_result(serf_context_t *ctx, |
| apr_sockaddr_t *host_address, |
| apr_status_t resolve_status, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *resolve_pool); |
| |
| /* This is the core of the asynchronous resolver implementation. */ |
| static apr_status_t resolve_address_async(serf_context_t *ctx, |
| apr_uri_t host_info, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *resolve_pool, |
| apr_pool_t *scratch_pool); |
| |
| /* Public API */ |
| apr_status_t serf_address_resolve_async(serf_context_t *ctx, |
| apr_uri_t host_info, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *pool) |
| { |
| apr_pool_t *resolve_pool; |
| |
| if (ctx->resolve_init_status != APR_SUCCESS) { |
| return ctx->resolve_init_status; |
| } |
| |
| apr_pool_create(&resolve_pool, ctx->pool); |
| return resolve_address_async(ctx, host_info, resolved, resolved_baton, |
| resolve_pool, pool); |
| } |
| |
| #else /* !HAVE_ASYNC_RESOLVER */ |
| |
| /* Public API */ |
| apr_status_t serf_address_resolve_async(serf_context_t *ctx, |
| apr_uri_t host_info, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *pool) |
| { |
| /* We have no external asynchronous resolver library, nor threads, |
| therefore no async resolver at all. */ |
| return APR_ENOTIMPL; |
| } |
| |
| #endif /* !HAVE_ASYNC_RESOLVER */ |
| |
| |
| #if SERF_HAVE_ASYNC_RESOLVER |
| |
| /* TODO: Add implementation for one or more async resolver libraries. */ |
| #if 0 |
| /* Called during context creation. Must initialize ctx->resolver_context. */ |
| static apr_status_t create_resolve_context(serf_context_t *ctx) |
| { |
| ... |
| } |
| |
| static apr_status_t resolve_address_async(serf_context_t *ctx, |
| apr_uri_t host_info, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *resolve_pool, |
| apr_pool_t *scratch_pool) |
| { |
| ... |
| } |
| |
| /* Some asynchronous resolved libraries use event loop to harvest results. |
| This function will be called from serf__process_async_resolve_results() |
| so, in effect, from serf_context_prerun(). */ |
| static apr_status_t run_async_resolver_loop(serf_context_t *ctx) |
| { |
| ... |
| } |
| #endif /* 0 */ |
| |
| #if SERF_HAVE_UNBOUND |
| |
| /*******************************************************************/ |
| /* Async resolver that uses libunbound. */ |
| |
| /* DNS classes and record types. |
| https://www.iana.org/assignments/dns-parameters/dns-parameters.xhtml */ |
| #define RR_CLASS_IN 1 /* Internet */ |
| #define RR_TYPE_A 1 /* IPv4 address */ |
| #define RR_TYPE_AAAA 28 /* IPv6 address */ |
| |
| |
| static apr_status_t err_to_status(enum ub_ctx_err err) |
| { |
| switch (err) |
| { |
| case UB_NOERROR: |
| /* no error */ |
| return APR_SUCCESS; |
| |
| case UB_SOCKET: |
| /* socket operation. Set to -1, so that if an error from _fd() is |
| passed (-1) it gives a socket error. */ |
| if (errno) |
| return APR_FROM_OS_ERROR(errno); |
| return APR_ENOTSOCK; |
| |
| case UB_NOMEM: |
| /* alloc failure */ |
| return APR_ENOMEM; |
| |
| case UB_SYNTAX: |
| /* syntax error */ |
| return APR_EINIT; |
| |
| case UB_SERVFAIL: |
| /* DNS service failed */ |
| return APR_EAGAIN; |
| |
| case UB_FORKFAIL: |
| /* fork() failed */ |
| return APR_ENOMEM; |
| |
| case UB_AFTERFINAL: |
| /* cfg change after finalize() */ |
| return APR_EINIT; |
| |
| case UB_INITFAIL: |
| /* initialization failed (bad settings) */ |
| return APR_EINIT; |
| |
| case UB_PIPE: |
| /* error in pipe communication with async bg worker */ |
| return APR_EPIPE; |
| |
| case UB_READFILE: |
| /* error reading from file (resolv.conf) */ |
| if (errno) |
| return APR_FROM_OS_ERROR(errno); |
| return APR_ENOENT; |
| |
| case UB_NOID: |
| /* error async_id does not exist or result already been delivered */ |
| return APR_EINVAL; |
| |
| default: |
| return APR_EGENERAL; |
| } |
| } |
| |
| |
| struct resolve_context |
| { |
| struct ub_ctx *ub_ctx; |
| volatile apr_uint32_t tasks; |
| }; |
| |
| static apr_status_t cleanup_resolve_context(void *baton) |
| { |
| struct resolve_context *const rctx = baton; |
| ub_ctx_delete(rctx->ub_ctx); |
| return APR_SUCCESS; |
| } |
| |
| static apr_status_t create_resolve_context(serf_context_t *ctx) |
| { |
| struct resolve_context *const rctx = apr_palloc(ctx->pool, sizeof(*rctx)); |
| int err; |
| |
| rctx->ub_ctx = ub_ctx_create(); |
| rctx->tasks = 0; |
| if (!rctx->ub_ctx) |
| return APR_ENOMEM; |
| |
| err = ub_ctx_resolvconf(rctx->ub_ctx, NULL); |
| if (!err) |
| err = ub_ctx_hosts(rctx->ub_ctx, NULL); |
| if (!err) |
| err = ub_ctx_async(rctx->ub_ctx, true); |
| |
| if (err) { |
| const apr_status_t status = err_to_status(err); |
| const char *message = apr_psprintf(ctx->pool, "unbound ctx init: %s", |
| ub_strerror(err)); |
| serf__context_error(ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| ctx->config, "%s\n", message); |
| cleanup_resolve_context(rctx); |
| return status; |
| } |
| |
| ctx->resolve_context = rctx; |
| /* pre-cleanup because the live resolve tasks contain subpools of the |
| context pool and must be canceled before their pools go away. */ |
| apr_pool_pre_cleanup_register(ctx->pool, rctx, cleanup_resolve_context); |
| return APR_SUCCESS; |
| } |
| |
| |
| /* Task data for the Unbound resolver. */ |
| typedef struct unbound_resolve_task resolve_task_t; |
| |
| struct resolve_result |
| { |
| apr_status_t status; |
| struct ub_result* ub_result; |
| resolve_task_t *task; |
| const char *qtype; |
| }; |
| |
| struct unbound_resolve_task |
| { |
| serf_context_t *ctx; |
| char *host_port_str; |
| apr_port_t host_port; |
| |
| /* There can be one or two pending results, depending on whether |
| we resolve for IPv6 as well as IPv4. */ |
| volatile apr_uint32_t pending_results; |
| struct resolve_result results[2]; |
| |
| serf_address_resolved_t resolved; |
| void *resolved_baton; |
| apr_pool_t *resolve_pool; |
| }; |
| |
| static apr_status_t resolve_convert(apr_sockaddr_t **host_address, |
| apr_int32_t family, bool free_result, |
| const struct resolve_result *result) |
| { |
| const struct unbound_resolve_task *const task = result->task; |
| const struct ub_result *const ub_result = result->ub_result; |
| apr_pool_t *const resolve_pool = task->resolve_pool; |
| apr_status_t status = result->status; |
| |
| if (status == APR_SUCCESS) |
| { |
| char *hostname = apr_pstrdup(resolve_pool, (ub_result->canonname |
| ? ub_result->canonname |
| : ub_result->qname)); |
| apr_socklen_t salen; |
| int ipaddr_len; |
| int addr_str_len; |
| int i; |
| |
| if (family == APR_INET) { |
| salen = sizeof(struct sockaddr_in); |
| ipaddr_len = sizeof(struct in_addr); |
| addr_str_len = INET_ADDRSTRLEN; |
| } |
| #if APR_HAVE_IPV6 |
| else { |
| salen = sizeof(struct sockaddr_in6); |
| ipaddr_len = sizeof(struct in6_addr); |
| addr_str_len = INET6_ADDRSTRLEN; |
| } |
| #endif |
| |
| /* Check that all result sizes are correct. */ |
| for (i = 0; ub_result->data[i]; ++i) |
| { |
| if (ub_result->len[i] != ipaddr_len) { |
| const char *message = apr_psprintf( |
| resolve_pool, |
| "unbound resolve: [%s] invalid address: " |
| " length %d, expected %d", |
| result->qtype, ub_result->len[i], ipaddr_len); |
| status = APR_EINVAL; |
| serf__context_error(task->ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| task->ctx->config, "%s\n", message); |
| goto cleanup; |
| } |
| } |
| |
| for (i = 0; ub_result->data[i]; ++i) |
| { |
| apr_sockaddr_t *addr = apr_palloc(resolve_pool, sizeof(*addr)); |
| addr->pool = resolve_pool; |
| addr->hostname = hostname; |
| addr->servname = task->host_port_str; |
| addr->port = task->host_port; |
| addr->family = family; |
| addr->salen = salen; |
| addr->ipaddr_len = ipaddr_len; |
| addr->addr_str_len = addr_str_len; |
| addr->next = *host_address; |
| |
| memset(&addr->sa, 0, sizeof(addr->sa)); |
| if (family == APR_INET) { |
| addr->ipaddr_ptr = &addr->sa.sin.sin_addr.s_addr; |
| addr->sa.sin.sin_family = APR_INET; |
| addr->sa.sin.sin_port = htons(addr->port); |
| } |
| #if APR_HAVE_IPV6 |
| else { |
| addr->ipaddr_ptr = &addr->sa.sin6.sin6_addr.s6_addr; |
| addr->sa.sin6.sin6_family = APR_INET6; |
| addr->sa.sin6.sin6_port = htons(addr->port); |
| } |
| #endif |
| memcpy(addr->ipaddr_ptr, ub_result->data[i], ipaddr_len); |
| *host_address = addr; |
| |
| if (serf__log_enabled(LOGLVL_DEBUG, LOGCOMP_CONN, |
| task->ctx->config)) { |
| char buf[INET6_ADDRSTRLEN]; |
| if (!apr_sockaddr_ip_getbuf(buf, sizeof(buf), addr)) { |
| serf__log(LOGLVL_DEBUG, LOGCOMP_CONN, |
| __FILE__, task->ctx->config, |
| "unbound resolve: [%s] %s: %s\n", |
| result->qtype, addr->hostname, buf); |
| } |
| } |
| } |
| } |
| |
| cleanup: |
| if (free_result && result->ub_result) |
| ub_resolve_free(result->ub_result); |
| |
| return status; |
| } |
| |
| static void resolve_finalize(resolve_task_t *task) |
| { |
| apr_sockaddr_t *host_address = NULL; |
| apr_status_t status, status6; |
| |
| status = resolve_convert(&host_address, APR_INET, true, &task->results[0]); |
| #if APR_HAVE_IPV6 |
| status6 = resolve_convert(&host_address, APR_INET6, true, &task->results[1]); |
| #else |
| status6 = APR_SUCCESS; |
| #endif |
| |
| if (!host_address) { |
| if (status == APR_SUCCESS) |
| status = status6; |
| if (status == APR_SUCCESS) |
| status = APR_ENOENT; |
| } |
| else { |
| /* If we got a result, ee don't care if one of the resolves failed. */ |
| status = APR_SUCCESS; |
| } |
| |
| push_resolve_result(task->ctx, host_address, status, |
| task->resolved, task->resolved_baton, |
| task->resolve_pool); |
| } |
| |
| static void resolve_callback(void* baton, int err, |
| struct ub_result* ub_result) |
| { |
| struct resolve_result *const resolve_result = baton; |
| resolve_task_t *const task = resolve_result->task; |
| apr_status_t status = err_to_status(err); |
| |
| struct resolve_context *const rctx = task->ctx->resolve_context; |
| apr_atomic_dec32(&rctx->tasks); |
| |
| if (err) |
| { |
| const char *message = apr_psprintf( |
| task->resolve_pool, |
| "unbound resolve: [%s] error %s", |
| resolve_result->qtype, ub_strerror(err)); |
| serf__context_error(task->ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| task->ctx->config, "%s\n", message); |
| } |
| else if (!ub_result->havedata) |
| { |
| const char *message; |
| if (ub_result->nxdomain) { |
| if (status == APR_SUCCESS) |
| status = APR_ENOENT; |
| message = apr_psprintf( |
| task->resolve_pool, |
| "unbound resolve: [%s] NXDOMAIN [%d]", |
| resolve_result->qtype, ub_result->rcode); |
| serf__context_error(task->ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| task->ctx->config, "%s\n", message); |
| } |
| if (ub_result->bogus) { |
| if (status == APR_SUCCESS) |
| status = APR_EINVAL; |
| message = apr_psprintf( |
| task->resolve_pool, |
| "unbound resolve: [%s] BOGUS [%d]%s%s", |
| resolve_result->qtype, ub_result->rcode, |
| ub_result->why_bogus ? " " : "", |
| ub_result->why_bogus ? ub_result->why_bogus : ""); |
| serf__context_error(task->ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| task->ctx->config, "%s\n", message); |
| } |
| if (ub_result->was_ratelimited) { |
| if (status == APR_SUCCESS) |
| status = APR_EAGAIN; |
| message = apr_psprintf( |
| task->resolve_pool, |
| "unbound resolve: [%s] SERVFAIL [%d]", |
| resolve_result->qtype, ub_result->rcode); |
| serf__context_error(task->ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| task->ctx->config, "%s\n", message); |
| } |
| |
| /* This shouldn't happen, one of the previous checks should |
| have caught an error. */ |
| if (status == APR_SUCCESS) { |
| status = APR_ENOENT; |
| message = apr_psprintf( |
| task->resolve_pool, |
| "unbound resolve: [%s] no data [%d]\n", |
| resolve_result->qtype, ub_result->rcode); |
| serf__context_error(task->ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| task->ctx->config, "%s\n", message); |
| } |
| } |
| |
| resolve_result->status = status; |
| resolve_result->ub_result = ub_result; |
| |
| /* The last pending task combines and publishes the results. */ |
| if (apr_atomic_dec32(&task->pending_results) == 0) |
| resolve_finalize(task); |
| } |
| |
| static apr_status_t resolve_address_async(serf_context_t *ctx, |
| apr_uri_t host_info, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *resolve_pool, |
| apr_pool_t *scratch_pool) |
| { |
| unsigned char addr[MAX_ADDRLEN]; |
| struct resolve_context *const rctx = ctx->resolve_context; |
| resolve_task_t *task; |
| apr_status_t status = APR_SUCCESS; |
| int err4 = 0, err6 = 0; |
| apr_int32_t family; |
| int pton, ipaddr_len, rr_type; |
| |
| /* If the hostname is an IP address, "resolve" it immediately. */ |
| pton = serf__inet_pton4(host_info.hostname, addr); |
| if (pton > 0) { |
| family = APR_INET; |
| rr_type = RR_TYPE_A; |
| ipaddr_len = sizeof(struct in_addr); |
| SERF__RESOLV_assert(ipaddr_len <= MAX_ADDRLEN); |
| } |
| else { |
| pton = serf__inet_pton6(host_info.hostname, addr); |
| if (pton > 0) { |
| #if APR_HAVE_IPV6 |
| family = APR_INET6; |
| rr_type = RR_TYPE_AAAA; |
| ipaddr_len = sizeof(struct in6_addr); |
| SERF__RESOLV_assert(ipaddr_len <= MAX_ADDRLEN); |
| #else |
| return APR_EAFNOSUPPORT; |
| #endif |
| } |
| } |
| |
| if (pton > 0) |
| { |
| static const struct ub_result ub_template = { |
| NULL, /* .qname */ |
| 0, /* .qtype */ |
| RR_CLASS_IN, /* .qclass */ |
| NULL, /* .data */ |
| NULL, /* .len */ |
| NULL, /* .canonname */ |
| 0, /* .rcode */ |
| NULL, /* .answer_packet */ |
| 0, /* .answer_len */ |
| 1, /* .havedata */ |
| 0, /* .nxdomain */ |
| 0, /* .secure */ |
| 0, /* .bogus */ |
| NULL, /* .why_bogus */ |
| 0, /* .was_ratelimited */ |
| INT_MAX /* .ttl */ |
| }; |
| |
| struct ub_result ub_result = ub_template; |
| apr_sockaddr_t *host_address = NULL; |
| char *data[2] = { NULL, NULL }; |
| int len[2] = { 0, 0 }; |
| struct resolve_result *result; |
| resolve_task_t local_task; |
| |
| memset(&local_task, 0, sizeof(local_task)); |
| data[0] = (char*) addr; |
| len[0] = ipaddr_len; |
| ub_result.qname = host_info.hostname; |
| ub_result.qtype = rr_type; |
| ub_result.data = data; |
| ub_result.len = len; |
| |
| task = &local_task; |
| task->ctx = ctx; |
| task->host_port_str = host_info.port_str; |
| task->host_port = host_info.port; |
| task->resolve_pool = resolve_pool; |
| result = &task->results[0]; |
| result->status = APR_SUCCESS; |
| result->ub_result = &ub_result; |
| result->task = task; |
| result->qtype = family == APR_INET ? "v4" : "v6"; |
| |
| status = resolve_convert(&host_address, family, false, result); |
| if (status == APR_SUCCESS) { |
| push_resolve_result(ctx, host_address, status, |
| resolved, resolved_baton, |
| resolve_pool); |
| } |
| return status; |
| } |
| |
| /* Create the async resolve tasks. */ |
| task = apr_palloc(resolve_pool, sizeof(*task)); |
| task->ctx = ctx; |
| task->host_port_str = apr_pstrdup(resolve_pool, host_info.port_str); |
| task->host_port = host_info.port; |
| |
| #if APR_HAVE_IPV6 |
| task->pending_results = 2; |
| #else |
| task->pending_results = 1; |
| #endif |
| task->results[0].status = task->results[1].status = APR_SUCCESS; |
| task->results[0].ub_result = task->results[1].ub_result = NULL; |
| task->results[0].task = task->results[1].task = task; |
| task->results[0].qtype = task->results[1].qtype = "??"; |
| |
| task->resolved = resolved; |
| task->resolved_baton = resolved_baton; |
| task->resolve_pool = resolve_pool; |
| |
| task->results[0].qtype = "v4"; |
| err4 = ub_resolve_async(rctx->ub_ctx, host_info.hostname, |
| RR_TYPE_A, RR_CLASS_IN, |
| &task->results[0], resolve_callback, NULL); |
| if (!err4) { |
| apr_atomic_inc32(&rctx->tasks); |
| } |
| |
| #if APR_HAVE_IPV6 |
| task->results[1].qtype = "v6"; |
| err6 = ub_resolve_async(rctx->ub_ctx, host_info.hostname, |
| RR_TYPE_AAAA, RR_CLASS_IN, |
| &task->results[1], resolve_callback, NULL); |
| if (!err6) { |
| apr_atomic_inc32(&rctx->tasks); |
| } |
| #endif /* APR_HAVE_IPV6 */ |
| |
| if (err4 || err6) |
| { |
| apr_uint32_t pending_results = -1; |
| const char *message; |
| |
| if (err4) { |
| pending_results = apr_atomic_dec32(&task->pending_results); |
| message = apr_psprintf(resolve_pool, |
| "unbound resolve start: [v4] %s", |
| ub_strerror(err4)); |
| status = err_to_status(err4); |
| serf__context_error(ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| ctx->config, "%s\n", message); |
| } |
| |
| #if APR_HAVE_IPV6 |
| if (err6) { |
| const apr_status_t status6 = err_to_status(err6); |
| pending_results = apr_atomic_dec32(&task->pending_results); |
| message = apr_psprintf(resolve_pool, |
| "unbound resolve start: [v6] %s", |
| ub_strerror(err6)); |
| serf__context_error(ctx, status6, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| ctx->config, "%s\n", message); |
| /* We have only one status to report. */ |
| if (!err4) |
| status = status6; |
| } |
| #endif /* APR_HAVE_IPV6 */ |
| |
| /* If one of the tasks failed and the other has already completed, |
| we have to do the result processing here. Note that the Unbound |
| callbacks can be called synchronously from ub_resolve_async(). */ |
| if (pending_results == 0) |
| resolve_finalize(task); |
| } |
| |
| return status; |
| } |
| |
| static apr_status_t run_async_resolver_loop(serf_context_t *ctx) |
| { |
| struct resolve_context *const rctx = ctx->resolve_context; |
| |
| /* No need to poll if there are no in-flight tasks. */ |
| if (apr_atomic_read32(&rctx->tasks)) |
| { |
| if (ub_poll(rctx->ub_ctx)) { |
| const int err = ub_process(rctx->ub_ctx); |
| if (err) { |
| const apr_status_t status = err_to_status(err); |
| apr_pool_t *error_pool; |
| const char *message; |
| apr_pool_create(&error_pool, ctx->pool); |
| message = apr_psprintf(error_pool, "unbound process: %s", |
| ub_strerror(err)); |
| serf__context_error(ctx, status, message); |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, |
| ctx->config, "%s\n", message); |
| apr_pool_destroy(error_pool); |
| return status; |
| } |
| } |
| } |
| |
| return APR_SUCCESS; |
| } |
| |
| #endif /* SERF_HAVE_UNBOUND */ |
| |
| #else /* !SERF_HAVE_ASYNC_RESOLVER */ |
| #if APR_HAS_THREADS |
| |
| /*******************************************************************/ |
| /* Default async resolver that uses APR thread pools. */ |
| |
| /* This could be made configurable, but given that this is a fallback |
| implementation, it really shouldn't be necessary. */ |
| #define MAX_WORK_QUEUE_THREADS 50 |
| static apr_pool_t *work_pool = NULL; |
| static apr_thread_pool_t *work_queue = NULL; |
| |
| static apr_status_t do_init_work_queue(void *baton) |
| { |
| serf_context_t *const ctx = baton; |
| apr_status_t status; |
| |
| apr_pool_create(&work_pool, NULL); |
| status = apr_thread_pool_create(&work_queue, |
| 1, MAX_WORK_QUEUE_THREADS, |
| work_pool); |
| |
| serf__log((status ? LOGLVL_ERROR : LOGLVL_DEBUG), |
| LOGCOMP_CONN, __FILE__, ctx->config, |
| "Init async resolve work queue, status %d\n", status); |
| return status; |
| } |
| |
| static apr_status_t init_work_queue(serf_context_t *ctx) |
| { |
| SERF__DECLARE_STATIC_INIT_ONCE_CONTEXT(init_ctx); |
| return serf__init_once(&init_ctx, do_init_work_queue, ctx); |
| } |
| |
| |
| static apr_status_t cleanup_resolve_tasks(void *baton) |
| { |
| /* baton is serf_context_t */ |
| return apr_thread_pool_tasks_cancel(work_queue, baton); |
| } |
| |
| static apr_status_t create_resolve_context(serf_context_t *ctx) |
| { |
| apr_status_t status; |
| |
| ctx->resolve_context = NULL; |
| status = init_work_queue(ctx); |
| if (status == APR_SUCCESS) |
| apr_pool_pre_cleanup_register(ctx->pool, ctx, cleanup_resolve_tasks); |
| |
| return status; |
| } |
| |
| |
| /* Task data for the thred pool resolver. */ |
| typedef struct threadpool_resolve_task resolve_task_t; |
| struct threadpool_resolve_task |
| { |
| serf_context_t *ctx; |
| apr_uri_t host_info; |
| serf_address_resolved_t resolved; |
| void *resolved_baton; |
| apr_pool_t *resolve_pool; |
| }; |
| |
| |
| static void *APR_THREAD_FUNC resolve(apr_thread_t *thread, void *baton) |
| { |
| resolve_task_t *task = baton; |
| apr_sockaddr_t *host_address; |
| apr_status_t status; |
| |
| status = apr_sockaddr_info_get(&host_address, |
| task->host_info.hostname, |
| APR_UNSPEC, |
| task->host_info.port, |
| 0, task->resolve_pool); |
| |
| if (status) { |
| host_address = NULL; |
| } |
| else if (serf__log_enabled(LOGLVL_DEBUG, LOGCOMP_CONN, task->ctx->config)) |
| { |
| apr_sockaddr_t *addr = host_address; |
| while (addr) |
| { |
| char buf[INET6_ADDRSTRLEN]; |
| if (!apr_sockaddr_ip_getbuf(buf, sizeof(buf), addr)) { |
| serf__log(LOGLVL_DEBUG, LOGCOMP_CONN, |
| __FILE__, task->ctx->config, |
| "thread pool resolve: %s: %s\n", addr->hostname, buf); |
| } |
| addr = addr->next; |
| } |
| } |
| |
| push_resolve_result(task->ctx, host_address, status, |
| task->resolved, task->resolved_baton, |
| task->resolve_pool); |
| return NULL; |
| } |
| |
| static apr_status_t resolve_address_async(serf_context_t *ctx, |
| apr_uri_t host_info, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *resolve_pool, |
| apr_pool_t *scratch_pool) |
| { |
| resolve_task_t *task; |
| apr_status_t status = init_work_queue(ctx); |
| if (status) |
| return status; |
| |
| task = apr_palloc(resolve_pool, sizeof(*task)); |
| task->ctx = ctx; |
| task->host_info = host_info; |
| task->resolved = resolved; |
| task->resolved_baton = resolved_baton; |
| task->resolve_pool = resolve_pool; |
| return apr_thread_pool_push(work_queue, resolve, task, |
| APR_THREAD_TASK_PRIORITY_NORMAL, |
| (void*)ctx); |
| } |
| |
| /* This is a no-op since we're using a thread pool that |
| does its own task queue management. */ |
| static apr_status_t run_async_resolver_loop(serf_context_t *ctx) |
| { |
| return APR_SUCCESS; |
| } |
| |
| #endif /* !APR_HAS_THREADS */ |
| #endif /* !SERF_HAVE_ASYNC_RESOLVER */ |
| |
| |
| /*******************************************************************/ |
| /* The result queue implementation. */ |
| #if HAVE_ASYNC_RESOLVER |
| |
| #if APR_MAJOR_VERSION < 2 |
| /* NOTE: The atomic pointer prototypes in apr-1.x are just horribly |
| wrong. They're fixed in version 2.x and these macros deal |
| with the difference. */ |
| #define apr_atomic_casptr(mem, with, cmp) \ |
| (apr_atomic_casptr)((volatile void**)(mem), (with), (cmp)) |
| #define apr_atomic_xchgptr(mem, with) \ |
| (apr_atomic_xchgptr)((volatile void**)(mem), (with)) |
| #endif /* APR_MAJOR_VERSION < 2 */ |
| |
| |
| typedef struct resolve_result_t resolve_result_t; |
| struct resolve_result_t |
| { |
| apr_sockaddr_t *host_address; |
| apr_status_t status; |
| serf_address_resolved_t resolved; |
| void *resolved_baton; |
| apr_pool_t *result_pool; |
| resolve_result_t *next; |
| }; |
| |
| |
| static void push_resolve_result(serf_context_t *ctx, |
| apr_sockaddr_t *host_address, |
| apr_status_t resolve_status, |
| serf_address_resolved_t resolved, |
| void *resolved_baton, |
| apr_pool_t *resolve_pool) |
| { |
| resolve_result_t *result; |
| void *head; |
| |
| result = apr_palloc(resolve_pool, sizeof(*result)); |
| result->host_address = host_address; |
| result->status = resolve_status; |
| result->resolved = resolved; |
| result->resolved_baton = resolved_baton; |
| result->result_pool = resolve_pool; |
| |
| /* Atomic push this result to the result stack. This might look like |
| a potential priority inversion, however, it's not likely that we'll |
| resolve several tens of thousands of results per second in hundreds |
| of separate threads. */ |
| head = apr_atomic_casptr(&ctx->resolve_head, NULL, NULL); |
| do { |
| result->next = head; |
| head = apr_atomic_casptr(&ctx->resolve_head, result, head); |
| } while(head != result->next); |
| |
| serf__context_wakeup(ctx); |
| } |
| |
| |
| /* Internal API */ |
| apr_status_t serf__create_resolve_context(serf_context_t *ctx) |
| { |
| return create_resolve_context(ctx); |
| } |
| |
| |
| /* Internal API */ |
| apr_status_t serf__process_async_resolve_results(serf_context_t *ctx) |
| { |
| resolve_result_t *result; |
| apr_status_t status; |
| unsigned counter; |
| |
| if (ctx->resolve_init_status != APR_SUCCESS) { |
| /* The async resolver initialization failed, so just return. */ |
| serf__log(LOGLVL_DEBUG, LOGCOMP_CONN, __FILE__, ctx->config, |
| "context 0x%p async resolver is not initialized\n", ctx); |
| return APR_SUCCESS; |
| } |
| |
| status = run_async_resolver_loop(ctx); |
| if (status) { |
| serf__log(LOGLVL_ERROR, LOGCOMP_CONN, __FILE__, ctx->config, |
| "context 0x%p async resolve: <%d>\n", ctx, status); |
| return status; |
| } |
| |
| /* Grab the whole stack, leaving it empty, and process the contents. */ |
| counter = 0; |
| result = apr_atomic_xchgptr(&ctx->resolve_head, NULL); |
| while (result) |
| { |
| resolve_result_t *const next = result->next; |
| result->resolved(ctx, result->resolved_baton, |
| result->host_address, result->status, |
| result->result_pool); |
| apr_pool_destroy(result->result_pool); |
| result = next; |
| ++counter; |
| } |
| |
| if (counter > 0) { |
| serf__log(LOGLVL_DEBUG, LOGCOMP_CONN, __FILE__, ctx->config, |
| "context 0x%p async resolve: %d event%s\n", |
| ctx, counter, counter == 1 ? "" : "s"); |
| } |
| return APR_SUCCESS; |
| } |
| |
| #else /* !HAVE_ASYNC_RESOLVER */ |
| |
| /* Internal API */ |
| apr_status_t serf__create_resolve_context(serf_context_t *ctx) |
| { |
| return APR_ENOTIMPL; |
| } |
| |
| /* Internal API */ |
| apr_status_t serf__process_async_resolve_results(serf_context_t *ctx) |
| { |
| /* The fallback is a no-op, the context should just continue to |
| work without an asynchronous resolver. */ |
| return APR_SUCCESS; |
| } |
| |
| #endif /* !HAVE_ASYNC_RESOLVER */ |
