/* * Ring buffer management * This is a fork of ring.c for DNS usage. * * Copyright (C) 2000-2019 Willy Tarreau - w@1wt.eu * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public * License as published by the Free Software Foundation, version 2.1 * exclusively. * * This library is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Lesser General Public License for more details. * * You should have received a copy of the GNU Lesser General Public * License along with this library; if not, write to the Free Software * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA */ #include #include #include #include #include #include #include #include #include /* Initialize a pre-allocated ring with the buffer area * of size */ void dns_ring_init(struct dns_ring *ring, void *area, size_t size) { HA_RWLOCK_INIT(&ring->lock); MT_LIST_INIT(&ring->waiters); ring->readers_count = 0; ring->buf = b_make(area, size, 0, 0); /* write the initial RC byte */ b_putchr(&ring->buf, 0); } /* Creates and returns a ring buffer of size bytes. Returns NULL on * allocation failure. */ struct dns_ring *dns_ring_new(size_t size) { struct dns_ring *ring = NULL; void *area = NULL; if (size < 2) goto fail; ring = malloc(sizeof(*ring)); if (!ring) goto fail; area = malloc(size); if (!area) goto fail; dns_ring_init(ring, area, size); return ring; fail: free(area); free(ring); return NULL; } /* destroys and frees ring */ void dns_ring_free(struct dns_ring *ring) { if (!ring) return; free(ring->buf.area); free(ring); } /* Tries to send parts from followed by parts from * to ring . The message is sent atomically. It may be truncated to * bytes if is non-null. There is no distinction between the * two lists, it's just a convenience to help the caller prepend some prefixes * when necessary. It takes the ring's write lock to make sure no other thread * will touch the buffer during the update. Returns the number of bytes sent, * or <=0 on failure. */ ssize_t dns_ring_write(struct dns_ring *ring, size_t maxlen, const struct ist pfx[], size_t npfx, const struct ist msg[], size_t nmsg) { struct buffer *buf = &ring->buf; struct appctx *appctx; size_t totlen = 0; size_t lenlen; uint64_t dellen; int dellenlen; struct mt_list *elt1, elt2; ssize_t sent = 0; int i; /* we have to find some room to add our message (the buffer is * never empty and at least contains the previous counter) and * to update both the buffer contents and heads at the same * time (it's doable using atomic ops but not worth the * trouble, let's just lock). For this we first need to know * the total message's length. We cannot measure it while * copying due to the varint encoding of the length. */ for (i = 0; i < npfx; i++) totlen += pfx[i].len; for (i = 0; i < nmsg; i++) totlen += msg[i].len; if (totlen > maxlen) totlen = maxlen; lenlen = varint_bytes(totlen); HA_RWLOCK_WRLOCK(RING_LOCK, &ring->lock); if (lenlen + totlen + 1 + 1 > b_size(buf)) goto done_buf; while (b_room(buf) < lenlen + totlen + 1) { /* we need to delete the oldest message (from the end), * and we have to stop if there's a reader stuck there. * Unless there's corruption in the buffer it's guaranteed * that we have enough data to find 1 counter byte, a * varint-encoded length (1 byte min) and the message * payload (0 bytes min). */ if (*b_head(buf)) goto done_buf; dellenlen = b_peek_varint(buf, 1, &dellen); if (!dellenlen) goto done_buf; BUG_ON(b_data(buf) < 1 + dellenlen + dellen); b_del(buf, 1 + dellenlen + dellen); } /* OK now we do have room */ __b_put_varint(buf, totlen); totlen = 0; for (i = 0; i < npfx; i++) { size_t len = pfx[i].len; if (len + totlen > maxlen) len = maxlen - totlen; if (len) __b_putblk(buf, pfx[i].ptr, len); totlen += len; } for (i = 0; i < nmsg; i++) { size_t len = msg[i].len; if (len + totlen > maxlen) len = maxlen - totlen; if (len) __b_putblk(buf, msg[i].ptr, len); totlen += len; } *b_tail(buf) = 0; buf->data++; // new read counter sent = lenlen + totlen + 1; /* notify potential readers */ mt_list_for_each_entry_safe(appctx, &ring->waiters, wait_entry, elt1, elt2) appctx_wakeup(appctx); done_buf: HA_RWLOCK_WRUNLOCK(RING_LOCK, &ring->lock); return sent; } /* Tries to attach appctx as a new reader on ring . This is * meant to be used by low level appctx code such as CLI or ring forwarding. * For higher level functions, please see the relevant parts in appctx or CLI. * It returns non-zero on success or zero on failure if too many users are * already attached. On success, the caller MUST call dns_ring_detach_appctx() * to detach itself, even if it was never woken up. */ int dns_ring_attach(struct dns_ring *ring) { int users = ring->readers_count; do { if (users >= 255) return 0; } while (!_HA_ATOMIC_CAS(&ring->readers_count, &users, users + 1)); return 1; } /* detach an appctx from a ring. The appctx is expected to be waiting at offset * relative to the beginning of the storage, or ~0 if not waiting yet. * Nothing is done if is NULL. */ void dns_ring_detach_appctx(struct dns_ring *ring, struct appctx *appctx, size_t ofs) { if (!ring) return; HA_RWLOCK_WRLOCK(RING_LOCK, &ring->lock); if (ofs != ~0) { /* reader was still attached */ if (ofs < b_head_ofs(&ring->buf)) ofs += b_size(&ring->buf) - b_head_ofs(&ring->buf); else ofs -= b_head_ofs(&ring->buf); BUG_ON(ofs >= b_size(&ring->buf)); MT_LIST_DELETE(&appctx->wait_entry); HA_ATOMIC_DEC(b_peek(&ring->buf, ofs)); } HA_ATOMIC_DEC(&ring->readers_count); HA_RWLOCK_WRUNLOCK(RING_LOCK, &ring->lock); } /* * Local variables: * c-indent-level: 8 * c-basic-offset: 8 * End: */