diff options
Diffstat (limited to '')
| -rw-r--r-- | src/dns.c | 1350 |
1 files changed, 1350 insertions, 0 deletions
diff --git a/src/dns.c b/src/dns.c new file mode 100644 index 0000000..8bf8dcd --- /dev/null +++ b/src/dns.c @@ -0,0 +1,1350 @@ +/* + * Name server resolution + * + * Copyright 2020 HAProxy Technologies + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version + * 2 of the License, or (at your option) any later version. + * + */ + +#include <errno.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> + +#include <sys/types.h> + +#include <haproxy/action.h> +#include <haproxy/api.h> +#include <haproxy/applet.h> +#include <haproxy/cfgparse.h> +#include <haproxy/channel.h> +#include <haproxy/check.h> +#include <haproxy/cli.h> +#include <haproxy/dgram.h> +#include <haproxy/dns.h> +#include <haproxy/errors.h> +#include <haproxy/fd.h> +#include <haproxy/log.h> +#include <haproxy/ring.h> +#include <haproxy/sc_strm.h> +#include <haproxy/stconn.h> +#include <haproxy/stream.h> +#include <haproxy/tools.h> + +static THREAD_LOCAL char *dns_msg_trash; + +DECLARE_STATIC_POOL(dns_session_pool, "dns_session", sizeof(struct dns_session)); +DECLARE_STATIC_POOL(dns_query_pool, "dns_query", sizeof(struct dns_query)); +DECLARE_STATIC_POOL(dns_msg_buf, "dns_msg_buf", DNS_TCP_MSG_RING_MAX_SIZE); + +/* Opens an UDP socket on the namesaver's IP/Port, if required. Returns 0 on + * success, -1 otherwise. ns->dgram must be defined. + */ +static int dns_connect_nameserver(struct dns_nameserver *ns) +{ + struct dgram_conn *dgram = &ns->dgram->conn; + int fd; + + /* Already connected */ + if (dgram->t.sock.fd != -1) + return 0; + + /* Create an UDP socket and connect it on the nameserver's IP/Port */ + if ((fd = socket(dgram->addr.to.ss_family, SOCK_DGRAM, IPPROTO_UDP)) == -1) { + send_log(NULL, LOG_WARNING, + "DNS : section '%s': can't create socket for nameserver '%s'.\n", + ns->counters->pid, ns->id); + return -1; + } + if (connect(fd, (struct sockaddr*)&dgram->addr.to, get_addr_len(&dgram->addr.to)) == -1) { + send_log(NULL, LOG_WARNING, + "DNS : section '%s': can't connect socket for nameserver '%s'.\n", + ns->counters->id, ns->id); + close(fd); + return -1; + } + + /* Make the socket non blocking */ + fd_set_nonblock(fd); + + /* Add the fd in the fd list and update its parameters */ + dgram->t.sock.fd = fd; + fd_insert(fd, dgram, dgram_fd_handler, MAX_THREADS_MASK); + fd_want_recv(fd); + return 0; +} + +/* Sends a message to a name server + * It returns message length on success + * or -1 in error case + * 0 is returned in case of output ring buffer is full + */ +int dns_send_nameserver(struct dns_nameserver *ns, void *buf, size_t len) +{ + int ret = -1; + + if (ns->dgram) { + struct dgram_conn *dgram = &ns->dgram->conn; + int fd; + + HA_SPIN_LOCK(DNS_LOCK, &dgram->lock); + fd = dgram->t.sock.fd; + if (fd == -1) { + if (dns_connect_nameserver(ns) == -1) { + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return -1; + } + fd = dgram->t.sock.fd; + } + + ret = send(fd, buf, len, 0); + if (ret < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + struct ist myist; + + myist = ist2(buf, len); + ret = ring_write(ns->dgram->ring_req, DNS_TCP_MSG_MAX_SIZE, NULL, 0, &myist, 1); + if (!ret) { + ns->counters->snd_error++; + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return -1; + } + fd_cant_send(fd); + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return ret; + } + ns->counters->snd_error++; + fd_delete(fd); + dgram->t.sock.fd = -1; + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return -1; + } + ns->counters->sent++; + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + } + else if (ns->stream) { + struct ist myist; + + myist = ist2(buf, len); + ret = ring_write(ns->stream->ring_req, DNS_TCP_MSG_MAX_SIZE, NULL, 0, &myist, 1); + if (!ret) { + ns->counters->snd_error++; + return -1; + } + task_wakeup(ns->stream->task_req, TASK_WOKEN_MSG); + return ret; + } + + return ret; +} + +void dns_session_free(struct dns_session *); + +/* Receives a dns message + * Returns message length + * 0 is returned if no more message available + * -1 in error case + */ +ssize_t dns_recv_nameserver(struct dns_nameserver *ns, void *data, size_t size) +{ + ssize_t ret = -1; + + if (ns->dgram) { + struct dgram_conn *dgram = &ns->dgram->conn; + int fd; + + HA_SPIN_LOCK(DNS_LOCK, &dgram->lock); + fd = dgram->t.sock.fd; + if (fd == -1) { + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return -1; + } + + if ((ret = recv(fd, data, size, 0)) < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + fd_cant_recv(fd); + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return 0; + } + fd_delete(fd); + dgram->t.sock.fd = -1; + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return -1; + } + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + } + else if (ns->stream) { + struct dns_stream_server *dss = ns->stream; + struct dns_session *ds; + + HA_SPIN_LOCK(DNS_LOCK, &dss->lock); + + if (!LIST_ISEMPTY(&dss->wait_sess)) { + ds = LIST_NEXT(&dss->wait_sess, struct dns_session *, waiter); + ret = ds->rx_msg.len < size ? ds->rx_msg.len : size; + memcpy(data, ds->rx_msg.area, ret); + + ds->rx_msg.len = 0; + + /* This barrier is here to ensure that all data is + * stored if the appctx detect the elem is out of the + * list. + */ + __ha_barrier_store(); + + LIST_DEL_INIT(&ds->waiter); + + if (ds->appctx) { + /* This second barrier is here to ensure that + * the waked up appctx won't miss that the elem + * is removed from the list. + */ + __ha_barrier_store(); + + /* awake appctx because it may have other + * message to receive + */ + appctx_wakeup(ds->appctx); + + /* dns_session could already be into free_sess list + * so we firstly remove it */ + LIST_DEL_INIT(&ds->list); + + /* decrease nb_queries to free a slot for a new query on that sess */ + ds->nb_queries--; + if (ds->nb_queries) { + /* it remains pipelined unanswered request + * into this session but we just decrease + * the counter so the session + * can not be full of pipelined requests + * so we can add if to free_sess list + * to receive a new request + */ + LIST_INSERT(&ds->dss->free_sess, &ds->list); + } + else { + /* there is no more pipelined requests + * into this session, so we move it + * to idle_sess list */ + LIST_INSERT(&ds->dss->idle_sess, &ds->list); + + /* update the counter of idle sessions */ + ds->dss->idle_conns++; + + /* Note: this is useless there to update + * the max_active_conns since we increase + * the idle count */ + } + } + else { + /* there is no more appctx for this session + * it means it is ready to die + */ + dns_session_free(ds); + } + + + } + + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + } + + return ret; +} + +static void dns_resolve_recv(struct dgram_conn *dgram) +{ + struct dns_nameserver *ns; + int fd; + + HA_SPIN_LOCK(DNS_LOCK, &dgram->lock); + + fd = dgram->t.sock.fd; + + /* check if ready for reading */ + if ((fd == -1) || !fd_recv_ready(fd)) { + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return; + } + + /* no need to go further if we can't retrieve the nameserver */ + if ((ns = dgram->owner) == NULL) { + _HA_ATOMIC_AND(&fdtab[fd].state, ~(FD_POLL_HUP|FD_POLL_ERR)); + fd_stop_recv(fd); + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return; + } + + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + + ns->process_responses(ns); +} + +/* Called when a dns network socket is ready to send data */ +static void dns_resolve_send(struct dgram_conn *dgram) +{ + int fd; + struct dns_nameserver *ns; + struct ring *ring; + struct buffer *buf; + uint64_t msg_len; + size_t len, cnt, ofs; + + HA_SPIN_LOCK(DNS_LOCK, &dgram->lock); + + fd = dgram->t.sock.fd; + + /* check if ready for sending */ + if ((fd == -1) || !fd_send_ready(fd)) { + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return; + } + + /* no need to go further if we can't retrieve the nameserver */ + if ((ns = dgram->owner) == NULL) { + _HA_ATOMIC_AND(&fdtab[fd].state, ~(FD_POLL_HUP|FD_POLL_ERR)); + fd_stop_send(fd); + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + return; + } + + ring = ns->dgram->ring_req; + buf = &ring->buf; + + HA_RWLOCK_RDLOCK(DNS_LOCK, &ring->lock); + ofs = ns->dgram->ofs_req; + + /* explanation for the initialization below: it would be better to do + * this in the parsing function but this would occasionally result in + * dropped events because we'd take a reference on the oldest message + * and keep it while being scheduled. Thus instead let's take it the + * first time we enter here so that we have a chance to pass many + * existing messages before grabbing a reference to a location. This + * value cannot be produced after initialization. + */ + if (unlikely(ofs == ~0)) { + ofs = 0; + HA_ATOMIC_INC(b_peek(buf, ofs)); + ofs += ring->ofs; + } + + /* we were already there, adjust the offset to be relative to + * the buffer's head and remove us from the counter. + */ + ofs -= ring->ofs; + BUG_ON(ofs >= buf->size); + HA_ATOMIC_DEC(b_peek(buf, ofs)); + + while (ofs + 1 < b_data(buf)) { + int ret; + + cnt = 1; + len = b_peek_varint(buf, ofs + cnt, &msg_len); + if (!len) + break; + cnt += len; + BUG_ON(msg_len + ofs + cnt + 1 > b_data(buf)); + if (unlikely(msg_len > DNS_TCP_MSG_MAX_SIZE)) { + /* too large a message to ever fit, let's skip it */ + ofs += cnt + msg_len; + continue; + } + + len = b_getblk(buf, dns_msg_trash, msg_len, ofs + cnt); + + ret = send(fd, dns_msg_trash, len, 0); + if (ret < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + fd_cant_send(fd); + goto out; + } + ns->counters->snd_error++; + fd_delete(fd); + fd = dgram->t.sock.fd = -1; + goto out; + } + ns->counters->sent++; + + ofs += cnt + len; + } + + /* we don't want/need to be waked up any more for sending + * because all ring content is sent */ + fd_stop_send(fd); + +out: + + HA_ATOMIC_INC(b_peek(buf, ofs)); + ofs += ring->ofs; + ns->dgram->ofs_req = ofs; + HA_RWLOCK_RDUNLOCK(DNS_LOCK, &ring->lock); + HA_SPIN_UNLOCK(DNS_LOCK, &dgram->lock); + +} + +/* proto_udp callback functions for a DNS resolution */ +struct dgram_data_cb dns_dgram_cb = { + .recv = dns_resolve_recv, + .send = dns_resolve_send, +}; + +int dns_dgram_init(struct dns_nameserver *ns, struct sockaddr_storage *sk) +{ + struct dns_dgram_server *dgram; + + if ((dgram = calloc(1, sizeof(*dgram))) == NULL) + return -1; + + /* Leave dgram partially initialized, no FD attached for + * now. */ + dgram->conn.owner = ns; + dgram->conn.data = &dns_dgram_cb; + dgram->conn.t.sock.fd = -1; + dgram->conn.addr.to = *sk; + HA_SPIN_INIT(&dgram->conn.lock); + ns->dgram = dgram; + + dgram->ofs_req = ~0; /* init ring offset */ + dgram->ring_req = ring_new(2*DNS_TCP_MSG_RING_MAX_SIZE); + if (!dgram->ring_req) { + ha_alert("memory allocation error initializing the ring for nameserver.\n"); + goto out; + } + + /* attach the task as reader */ + if (!ring_attach(dgram->ring_req)) { + /* mark server attached to the ring */ + ha_alert("nameserver sets too many watchers > 255 on ring. This is a bug and should not happen.\n"); + goto out; + } + return 0; +out: + if (dgram->ring_req) + ring_free(dgram->ring_req); + + free(dgram); + + return -1; +} + +/* + * IO Handler to handle message push to dns tcp server + * It takes its context from appctx->svcctx. + */ +static void dns_session_io_handler(struct appctx *appctx) +{ + struct stconn *sc = appctx_sc(appctx); + struct dns_session *ds = appctx->svcctx; + struct ring *ring = &ds->ring; + struct buffer *buf = &ring->buf; + uint64_t msg_len; + int available_room; + size_t len, cnt, ofs; + int ret = 0; + + /* if stopping was requested, close immediately */ + if (unlikely(stopping)) + goto close; + + /* we want to be sure to not miss that we have been awaked for a shutdown */ + __ha_barrier_load(); + + /* that means the connection was requested to shutdown + * for instance idle expire */ + if (ds->shutdown) + goto close; + + /* an error was detected */ + if (unlikely(sc_ic(sc)->flags & (CF_WRITE_ERROR|CF_SHUTW))) + goto close; + + /* con closed by server side, we will skip data write and drain data from channel */ + if ((sc_oc(sc)->flags & CF_SHUTW)) { + goto read; + } + + /* if the connection is not established, inform the stream that we want + * to be notified whenever the connection completes. + */ + if (sc_opposite(sc)->state < SC_ST_EST) { + applet_need_more_data(appctx); + se_need_remote_conn(appctx->sedesc); + applet_have_more_data(appctx); + return; + } + + + ofs = ds->ofs; + + HA_RWLOCK_WRLOCK(DNS_LOCK, &ring->lock); + LIST_DEL_INIT(&appctx->wait_entry); + HA_RWLOCK_WRUNLOCK(DNS_LOCK, &ring->lock); + + HA_RWLOCK_RDLOCK(DNS_LOCK, &ring->lock); + + /* explanation for the initialization below: it would be better to do + * this in the parsing function but this would occasionally result in + * dropped events because we'd take a reference on the oldest message + * and keep it while being scheduled. Thus instead let's take it the + * first time we enter here so that we have a chance to pass many + * existing messages before grabbing a reference to a location. This + * value cannot be produced after initialization. + */ + if (unlikely(ofs == ~0)) { + ofs = 0; + + HA_ATOMIC_INC(b_peek(buf, ofs)); + ofs += ring->ofs; + } + + /* in this loop, ofs always points to the counter byte that precedes + * the message so that we can take our reference there if we have to + * stop before the end (ret=0). + */ + if (sc_opposite(sc)->state == SC_ST_EST) { + /* we were already there, adjust the offset to be relative to + * the buffer's head and remove us from the counter. + */ + ofs -= ring->ofs; + BUG_ON(ofs >= buf->size); + HA_ATOMIC_DEC(b_peek(buf, ofs)); + + ret = 1; + while (ofs + 1 < b_data(buf)) { + struct dns_query *query; + uint16_t original_qid; + uint16_t new_qid; + + cnt = 1; + len = b_peek_varint(buf, ofs + cnt, &msg_len); + if (!len) + break; + cnt += len; + BUG_ON(msg_len + ofs + cnt + 1 > b_data(buf)); + + /* retrieve available room on output channel */ + available_room = channel_recv_max(sc_ic(sc)); + + /* tx_msg_offset null means we are at the start of a new message */ + if (!ds->tx_msg_offset) { + uint16_t slen; + + /* check if there is enough room to put message len and query id */ + if (available_room < sizeof(slen) + sizeof(new_qid)) { + sc_need_room(sc); + ret = 0; + break; + } + + /* put msg len into then channel */ + slen = (uint16_t)msg_len; + slen = htons(slen); + applet_putblk(appctx, (char *)&slen, sizeof(slen)); + available_room -= sizeof(slen); + + /* backup original query id */ + len = b_getblk(buf, (char *)&original_qid, sizeof(original_qid), ofs + cnt); + if (!len) { + /* should never happen since messages are atomically + * written into ring + */ + ret = 0; + break; + } + + /* generates new query id */ + new_qid = ++ds->query_counter; + new_qid = htons(new_qid); + + /* put new query id into the channel */ + applet_putblk(appctx, (char *)&new_qid, sizeof(new_qid)); + available_room -= sizeof(new_qid); + + /* keep query id mapping */ + + query = pool_alloc(dns_query_pool); + if (query) { + query->qid.key = new_qid; + query->original_qid = original_qid; + query->expire = tick_add(now_ms, 5000); + LIST_INIT(&query->list); + if (LIST_ISEMPTY(&ds->queries)) { + /* enable task to handle expire */ + ds->task_exp->expire = query->expire; + /* ensure this will be executed by the same + * thread than ds_session_release + * to ensure session_release is free + * to destroy the task */ + task_queue(ds->task_exp); + } + LIST_APPEND(&ds->queries, &query->list); + eb32_insert(&ds->query_ids, &query->qid); + ds->onfly_queries++; + } + + /* update the tx_offset to handle output in 16k streams */ + ds->tx_msg_offset = sizeof(original_qid); + + } + + /* check if it remains available room on output chan */ + if (unlikely(!available_room)) { + sc_need_room(sc); + ret = 0; + break; + } + + chunk_reset(&trash); + if ((msg_len - ds->tx_msg_offset) > available_room) { + /* remaining msg data is too large to be written in output channel at one time */ + + len = b_getblk(buf, trash.area, available_room, ofs + cnt + ds->tx_msg_offset); + + /* update offset to complete mesg forwarding later */ + ds->tx_msg_offset += len; + } + else { + /* remaining msg data can be written in output channel at one time */ + len = b_getblk(buf, trash.area, msg_len - ds->tx_msg_offset, ofs + cnt + ds->tx_msg_offset); + + /* reset tx_msg_offset to mark forward fully processed */ + ds->tx_msg_offset = 0; + } + trash.data += len; + + if (applet_putchk(appctx, &trash) == -1) { + /* should never happen since we + * check available_room is large + * enough here. + */ + ret = 0; + break; + } + + if (ds->tx_msg_offset) { + /* msg was not fully processed, we must be awake to drain pending data */ + + sc_need_room(sc); + ret = 0; + break; + } + /* switch to next message */ + ofs += cnt + msg_len; + } + + HA_ATOMIC_INC(b_peek(buf, ofs)); + ofs += ring->ofs; + ds->ofs = ofs; + } + HA_RWLOCK_RDUNLOCK(DNS_LOCK, &ring->lock); + + if (ret) { + /* let's be woken up once new request to write arrived */ + HA_RWLOCK_WRLOCK(DNS_LOCK, &ring->lock); + BUG_ON(LIST_INLIST(&appctx->wait_entry)); + LIST_APPEND(&ring->waiters, &appctx->wait_entry); + HA_RWLOCK_WRUNLOCK(DNS_LOCK, &ring->lock); + applet_have_no_more_data(appctx); + } + +read: + + /* if session is not a waiter it means there is no committed + * message into rx_buf and we are free to use it + * Note: we need a load barrier here to not miss the + * delete from the list + */ + + __ha_barrier_load(); + if (!LIST_INLIST_ATOMIC(&ds->waiter)) { + while (1) { + uint16_t query_id; + struct eb32_node *eb; + struct dns_query *query; + + if (!ds->rx_msg.len) { + /* next message len is not fully available into the channel */ + if (co_data(sc_oc(sc)) < 2) + break; + + /* retrieve message len */ + co_getblk(sc_oc(sc), (char *)&msg_len, 2, 0); + + /* mark as consumed */ + co_skip(sc_oc(sc), 2); + + /* store message len */ + ds->rx_msg.len = ntohs(msg_len); + } + + if (!co_data(sc_oc(sc))) { + /* we need more data but nothing is available */ + break; + } + + if (co_data(sc_oc(sc)) + ds->rx_msg.offset < ds->rx_msg.len) { + /* message only partially available */ + + /* read available data */ + co_getblk(sc_oc(sc), ds->rx_msg.area + ds->rx_msg.offset, co_data(sc_oc(sc)), 0); + + /* update message offset */ + ds->rx_msg.offset += co_data(sc_oc(sc)); + + /* consume all pending data from the channel */ + co_skip(sc_oc(sc), co_data(sc_oc(sc))); + + /* we need to wait for more data */ + break; + } + + /* enough data is available into the channel to read the message until the end */ + + /* read from the channel until the end of the message */ + co_getblk(sc_oc(sc), ds->rx_msg.area + ds->rx_msg.offset, ds->rx_msg.len - ds->rx_msg.offset, 0); + + /* consume all data until the end of the message from the channel */ + co_skip(sc_oc(sc), ds->rx_msg.len - ds->rx_msg.offset); + + /* reset reader offset to 0 for next message reand */ + ds->rx_msg.offset = 0; + + /* try remap query id to original */ + memcpy(&query_id, ds->rx_msg.area, sizeof(query_id)); + eb = eb32_lookup(&ds->query_ids, query_id); + if (!eb) { + /* query id not found means we have an unknown corresponding + * request, perhaps server's bug or or the query reached + * timeout + */ + ds->rx_msg.len = 0; + continue; + } + + /* re-map the original query id set by the requester */ + query = eb32_entry(eb, struct dns_query, qid); + memcpy(ds->rx_msg.area, &query->original_qid, sizeof(query->original_qid)); + + /* remove query ids mapping from pending queries list/tree */ + eb32_delete(&query->qid); + LIST_DELETE(&query->list); + pool_free(dns_query_pool, query); + ds->onfly_queries--; + + /* the dns_session is also added in queue of the + * wait_sess list where the task processing + * response will pop available responses + */ + HA_SPIN_LOCK(DNS_LOCK, &ds->dss->lock); + + BUG_ON(LIST_INLIST(&ds->waiter)); + LIST_APPEND(&ds->dss->wait_sess, &ds->waiter); + + HA_SPIN_UNLOCK(DNS_LOCK, &ds->dss->lock); + + /* awake the task processing the responses */ + task_wakeup(ds->dss->task_rsp, TASK_WOKEN_INIT); + + break; + } + + if (!LIST_INLIST(&ds->waiter)) { + /* there is no more pending data to read and the con was closed by the server side */ + if (!co_data(sc_oc(sc)) && (sc_oc(sc)->flags & CF_SHUTW)) { + goto close; + } + } + + } + + return; +close: + sc_shutw(sc); + sc_shutr(sc); + sc_ic(sc)->flags |= CF_READ_NULL; +} + +void dns_queries_flush(struct dns_session *ds) +{ + struct dns_query *query, *queryb; + + list_for_each_entry_safe(query, queryb, &ds->queries, list) { + eb32_delete(&query->qid); + LIST_DELETE(&query->list); + pool_free(dns_query_pool, query); + } +} + +void dns_session_free(struct dns_session *ds) +{ + if (ds->rx_msg.area) + pool_free(dns_msg_buf, ds->rx_msg.area); + if (ds->tx_ring_area) + pool_free(dns_msg_buf, ds->tx_ring_area); + if (ds->task_exp) + task_destroy(ds->task_exp); + + dns_queries_flush(ds); + + /* Ensure to remove this session from external lists + * Note: we are under the lock of dns_stream_server + * which own the heads of those lists. + */ + LIST_DEL_INIT(&ds->waiter); + LIST_DEL_INIT(&ds->list); + + ds->dss->cur_conns--; + /* Note: this is useless to update + * max_active_conns here because + * we decrease the value + */ + + BUG_ON(!LIST_ISEMPTY(&ds->list)); + BUG_ON(!LIST_ISEMPTY(&ds->waiter)); + BUG_ON(!LIST_ISEMPTY(&ds->queries)); + BUG_ON(!LIST_ISEMPTY(&ds->ring.waiters)); + BUG_ON(!eb_is_empty(&ds->query_ids)); + pool_free(dns_session_pool, ds); +} + +static struct appctx *dns_session_create(struct dns_session *ds); + +static int dns_session_init(struct appctx *appctx) +{ + struct dns_session *ds = appctx->svcctx; + struct stream *s; + struct sockaddr_storage *addr = NULL; + + if (!sockaddr_alloc(&addr, &ds->dss->srv->addr, sizeof(ds->dss->srv->addr))) + goto error; + + if (appctx_finalize_startup(appctx, ds->dss->srv->proxy, &BUF_NULL) == -1) + goto error; + + s = appctx_strm(appctx); + s->scb->dst = addr; + s->scb->flags |= SC_FL_NOLINGER; + s->target = &ds->dss->srv->obj_type; + s->flags = SF_ASSIGNED; + + s->do_log = NULL; + s->uniq_id = 0; + + s->res.flags |= CF_READ_DONTWAIT; + /* for rto and rex to eternity to not expire on idle recv: + * We are using a syslog server. + */ + s->res.rto = TICK_ETERNITY; + s->res.rex = TICK_ETERNITY; + + ds->appctx = appctx; + return 0; + + error: + return -1; +} + +/* + * Function to release a DNS tcp session + */ +static void dns_session_release(struct appctx *appctx) +{ + struct dns_session *ds = appctx->svcctx; + struct dns_stream_server *dss __maybe_unused; + + if (!ds) + return; + + /* We do not call ring_appctx_detach here + * because we want to keep readers counters + * to retry a conn with a different appctx. + */ + HA_RWLOCK_WRLOCK(DNS_LOCK, &ds->ring.lock); + LIST_DEL_INIT(&appctx->wait_entry); + HA_RWLOCK_WRUNLOCK(DNS_LOCK, &ds->ring.lock); + + dss = ds->dss; + + HA_SPIN_LOCK(DNS_LOCK, &dss->lock); + LIST_DEL_INIT(&ds->list); + + if (stopping) { + dns_session_free(ds); + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + return; + } + + if (!ds->nb_queries) { + /* this is an idle session */ + /* Note: this is useless to update max_active_sess + * here because we decrease idle_conns but + * dns_session_free decrease curconns + */ + + ds->dss->idle_conns--; + dns_session_free(ds); + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + return; + } + + if (ds->onfly_queries == ds->nb_queries) { + /* the session can be released because + * it means that all queries AND + * responses are in fly */ + dns_session_free(ds); + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + return; + } + + /* if there is no pending complete response + * message, ensure to reset + * message offsets if the session + * was closed with an incomplete pending response + */ + if (!LIST_INLIST(&ds->waiter)) + ds->rx_msg.len = ds->rx_msg.offset = 0; + + /* we flush pending sent queries because we never + * have responses + */ + ds->nb_queries -= ds->onfly_queries; + dns_queries_flush(ds); + + /* reset offset to be sure to start from message start */ + ds->tx_msg_offset = 0; + + /* here the ofs and the attached counter + * are kept unchanged + */ + + /* Create a new appctx, We hope we can + * create from the release callback! */ + ds->appctx = dns_session_create(ds); + if (!ds->appctx) { + dns_session_free(ds); + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + return; + } + + if (ds->nb_queries < DNS_STREAM_MAX_PIPELINED_REQ) + LIST_INSERT(&ds->dss->free_sess, &ds->list); + + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); +} + +/* DNS tcp session applet */ +static struct applet dns_session_applet = { + .obj_type = OBJ_TYPE_APPLET, + .name = "<STRMDNS>", /* used for logging */ + .fct = dns_session_io_handler, + .init = dns_session_init, + .release = dns_session_release, +}; + +/* + * Function used to create an appctx for a DNS session + * It sets its context into appctx->svcctx. + */ +static struct appctx *dns_session_create(struct dns_session *ds) +{ + struct appctx *appctx; + + appctx = appctx_new_here(&dns_session_applet, NULL); + if (!appctx) + goto out_close; + appctx->svcctx = (void *)ds; + + if (appctx_init(appctx) == -1) { + ha_alert("out of memory in dns_session_create().\n"); + goto out_free_appctx; + } + + return appctx; + + /* Error unrolling */ + out_free_appctx: + appctx_free_on_early_error(appctx); + out_close: + return NULL; +} + +/* Task processing expiration of unresponded queries, this one is supposed + * to be stuck on the same thread than the appctx handler + */ +static struct task *dns_process_query_exp(struct task *t, void *context, unsigned int state) +{ + struct dns_session *ds = (struct dns_session *)context; + struct dns_query *query, *queryb; + + t->expire = TICK_ETERNITY; + + list_for_each_entry_safe(query, queryb, &ds->queries, list) { + if (tick_is_expired(query->expire, now_ms)) { + eb32_delete(&query->qid); + LIST_DELETE(&query->list); + pool_free(dns_query_pool, query); + ds->onfly_queries--; + } + else { + t->expire = query->expire; + break; + } + } + + return t; +} + +/* Task processing expiration of idle sessions */ +static struct task *dns_process_idle_exp(struct task *t, void *context, unsigned int state) +{ + struct dns_stream_server *dss = (struct dns_stream_server *)context; + struct dns_session *ds, *dsb; + int target = 0; + int cur_active_conns; + + HA_SPIN_LOCK(DNS_LOCK, &dss->lock); + + + cur_active_conns = dss->cur_conns - dss->idle_conns; + if (cur_active_conns > dss->max_active_conns) + dss->max_active_conns = cur_active_conns; + + target = (dss->max_active_conns - cur_active_conns) / 2; + list_for_each_entry_safe(ds, dsb, &dss->idle_sess, list) { + if (!target) + break; + + /* remove conn to pending list to ensure it won't be reused */ + LIST_DEL_INIT(&ds->list); + + /* force session shutdown */ + ds->shutdown = 1; + + /* to be sure that the appctx won't miss shutdown */ + __ha_barrier_store(); + + /* wake appctx to perform the shutdown */ + appctx_wakeup(ds->appctx); + } + + /* reset max to current active conns */ + dss->max_active_conns = cur_active_conns; + + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + + t->expire = tick_add(now_ms, 5000); + + return t; +} + +struct dns_session *dns_session_new(struct dns_stream_server *dss) +{ + struct dns_session *ds; + + if (dss->maxconn && (dss->maxconn <= dss->cur_conns)) + return NULL; + + ds = pool_zalloc(dns_session_pool); + if (!ds) + return NULL; + + ds->ofs = ~0; + ds->dss = dss; + LIST_INIT(&ds->list); + LIST_INIT(&ds->queries); + LIST_INIT(&ds->waiter); + ds->rx_msg.offset = ds->rx_msg.len = 0; + ds->rx_msg.area = NULL; + ds->tx_ring_area = NULL; + ds->task_exp = NULL; + ds->appctx = NULL; + ds->shutdown = 0; + ds->nb_queries = 0; + ds->query_ids = EB_ROOT_UNIQUE; + ds->rx_msg.area = pool_alloc(dns_msg_buf); + if (!ds->rx_msg.area) + goto error; + + ds->tx_ring_area = pool_alloc(dns_msg_buf); + if (!ds->tx_ring_area) + goto error; + + ring_init(&ds->ring, ds->tx_ring_area, DNS_TCP_MSG_RING_MAX_SIZE); + /* never fail because it is the first watcher attached to the ring */ + DISGUISE(ring_attach(&ds->ring)); + + if ((ds->task_exp = task_new_here()) == NULL) + goto error; + + ds->task_exp->process = dns_process_query_exp; + ds->task_exp->context = ds; + + ds->appctx = dns_session_create(ds); + if (!ds->appctx) + goto error; + + dss->cur_conns++; + + return ds; + +error: + if (ds->task_exp) + task_destroy(ds->task_exp); + if (ds->rx_msg.area) + pool_free(dns_msg_buf, ds->rx_msg.area); + if (ds->tx_ring_area) + pool_free(dns_msg_buf, ds->tx_ring_area); + + pool_free(dns_session_pool, ds); + + return NULL; +} + +/* + * Task used to consume pending messages from nameserver ring + * and forward them to dns_session ring. + * Note: If no slot found a new dns_session is allocated + */ +static struct task *dns_process_req(struct task *t, void *context, unsigned int state) +{ + struct dns_nameserver *ns = (struct dns_nameserver *)context; + struct dns_stream_server *dss = ns->stream; + struct ring *ring = dss->ring_req; + struct buffer *buf = &ring->buf; + uint64_t msg_len; + size_t len, cnt, ofs; + struct dns_session *ds, *ads; + HA_SPIN_LOCK(DNS_LOCK, &dss->lock); + + ofs = dss->ofs_req; + + HA_RWLOCK_RDLOCK(DNS_LOCK, &ring->lock); + + /* explanation for the initialization below: it would be better to do + * this in the parsing function but this would occasionally result in + * dropped events because we'd take a reference on the oldest message + * and keep it while being scheduled. Thus instead let's take it the + * first time we enter here so that we have a chance to pass many + * existing messages before grabbing a reference to a location. This + * value cannot be produced after initialization. + */ + if (unlikely(ofs == ~0)) { + ofs = 0; + HA_ATOMIC_INC(b_peek(buf, ofs)); + ofs += ring->ofs; + } + + /* we were already there, adjust the offset to be relative to + * the buffer's head and remove us from the counter. + */ + ofs -= ring->ofs; + BUG_ON(ofs >= buf->size); + HA_ATOMIC_DEC(b_peek(buf, ofs)); + + while (ofs + 1 < b_data(buf)) { + struct ist myist; + + cnt = 1; + len = b_peek_varint(buf, ofs + cnt, &msg_len); + if (!len) + break; + cnt += len; + BUG_ON(msg_len + ofs + cnt + 1 > b_data(buf)); + if (unlikely(msg_len > DNS_TCP_MSG_MAX_SIZE)) { + /* too large a message to ever fit, let's skip it */ + ofs += cnt + msg_len; + continue; + } + + len = b_getblk(buf, dns_msg_trash, msg_len, ofs + cnt); + + myist = ist2(dns_msg_trash, len); + + ads = NULL; + /* try to push request into active sess with free slot */ + if (!LIST_ISEMPTY(&dss->free_sess)) { + ds = LIST_NEXT(&dss->free_sess, struct dns_session *, list); + + if (ring_write(&ds->ring, DNS_TCP_MSG_MAX_SIZE, NULL, 0, &myist, 1) > 0) { + ds->nb_queries++; + if (ds->nb_queries >= DNS_STREAM_MAX_PIPELINED_REQ) + LIST_DEL_INIT(&ds->list); + ads = ds; + } + else { + /* it means we were unable to put a request in this slot, + * it may be close to be full so we put it at the end + * of free conn list */ + LIST_DEL_INIT(&ds->list); + LIST_APPEND(&dss->free_sess, &ds->list); + } + } + + if (!ads) { + /* try to push request into idle, this one should have enough free space */ + if (!LIST_ISEMPTY(&dss->idle_sess)) { + ds = LIST_NEXT(&dss->idle_sess, struct dns_session *, list); + + /* ring is empty so this ring_write should never fail */ + ring_write(&ds->ring, DNS_TCP_MSG_MAX_SIZE, NULL, 0, &myist, 1); + ds->nb_queries++; + LIST_DEL_INIT(&ds->list); + + ds->dss->idle_conns--; + + /* we may have to update the max_active_conns */ + if (ds->dss->max_active_conns < ds->dss->cur_conns - ds->dss->idle_conns) + ds->dss->max_active_conns = ds->dss->cur_conns - ds->dss->idle_conns; + + /* since we may unable to find a free list to handle + * this request, this request may be large and fill + * the ring buffer so we prefer to put at the end of free + * list. */ + LIST_APPEND(&dss->free_sess, &ds->list); + ads = ds; + } + } + + /* we didn't find a session available with large enough room */ + if (!ads) { + /* allocate a new session */ + ads = dns_session_new(dss); + if (ads) { + /* ring is empty so this ring_write should never fail */ + ring_write(&ads->ring, DNS_TCP_MSG_MAX_SIZE, NULL, 0, &myist, 1); + ads->nb_queries++; + LIST_INSERT(&dss->free_sess, &ads->list); + } + else + ns->counters->snd_error++; + } + + if (ads) + ns->counters->sent++; + + ofs += cnt + len; + } + + HA_ATOMIC_INC(b_peek(buf, ofs)); + ofs += ring->ofs; + dss->ofs_req = ofs; + HA_RWLOCK_RDUNLOCK(DNS_LOCK, &ring->lock); + + + HA_SPIN_UNLOCK(DNS_LOCK, &dss->lock); + return t; +} + +/* + * Task used to consume response + * Note: upper layer callback is called + */ +static struct task *dns_process_rsp(struct task *t, void *context, unsigned int state) +{ + struct dns_nameserver *ns = (struct dns_nameserver *)context; + + ns->process_responses(ns); + + return t; +} + +/* Function used to initialize an TCP nameserver */ +int dns_stream_init(struct dns_nameserver *ns, struct server *srv) +{ + struct dns_stream_server *dss = NULL; + + dss = calloc(1, sizeof(*dss)); + if (!dss) { + ha_alert("memory allocation error initializing dns tcp server '%s'.\n", srv->id); + goto out; + } + + dss->srv = srv; + dss->maxconn = srv->maxconn; + + dss->ofs_req = ~0; /* init ring offset */ + dss->ring_req = ring_new(2*DNS_TCP_MSG_RING_MAX_SIZE); + if (!dss->ring_req) { + ha_alert("memory allocation error initializing the ring for dns tcp server '%s'.\n", srv->id); + goto out; + } + /* Create the task associated to the resolver target handling conns */ + if ((dss->task_req = task_new_anywhere()) == NULL) { + ha_alert("memory allocation error initializing the ring for dns tcp server '%s'.\n", srv->id); + goto out; + } + + /* Update task's parameters */ + dss->task_req->process = dns_process_req; + dss->task_req->context = ns; + + /* attach the task as reader */ + if (!ring_attach(dss->ring_req)) { + /* mark server attached to the ring */ + ha_alert("server '%s': too many watchers for ring. this should never happen.\n", srv->id); + goto out; + } + + /* Create the task associated to the resolver target handling conns */ + if ((dss->task_rsp = task_new_anywhere()) == NULL) { + ha_alert("memory allocation error initializing the ring for dns tcp server '%s'.\n", srv->id); + goto out; + } + + /* Update task's parameters */ + dss->task_rsp->process = dns_process_rsp; + dss->task_rsp->context = ns; + + /* Create the task associated to the resolver target handling conns */ + if ((dss->task_idle = task_new_anywhere()) == NULL) { + ha_alert("memory allocation error initializing the ring for dns tcp server '%s'.\n", srv->id); + goto out; + } + + /* Update task's parameters */ + dss->task_idle->process = dns_process_idle_exp; + dss->task_idle->context = dss; + dss->task_idle->expire = tick_add(now_ms, 5000); + + /* let start the task to free idle conns immediately */ + task_queue(dss->task_idle); + + LIST_INIT(&dss->free_sess); + LIST_INIT(&dss->idle_sess); + LIST_INIT(&dss->wait_sess); + HA_SPIN_INIT(&dss->lock); + ns->stream = dss; + return 0; +out: + if (dss && dss->task_rsp) + task_destroy(dss->task_rsp); + if (dss && dss->task_req) + task_destroy(dss->task_req); + if (dss && dss->ring_req) + ring_free(dss->ring_req); + + free(dss); + return -1; +} + +int init_dns_buffers() +{ + dns_msg_trash = malloc(DNS_TCP_MSG_MAX_SIZE); + if (!dns_msg_trash) + return 0; + + return 1; +} + +void deinit_dns_buffers() +{ + ha_free(&dns_msg_trash); +} + +REGISTER_PER_THREAD_ALLOC(init_dns_buffers); +REGISTER_PER_THREAD_FREE(deinit_dns_buffers); |