3 Utility functions to read/write blobs of data from a file descriptor
4 and handle the case where we might need multiple read/writes to get all the
7 Copyright (C) Andrew Tridgell 2006
9 This program is free software; you can redistribute it and/or modify
10 it under the terms of the GNU General Public License as published by
11 the Free Software Foundation; either version 3 of the License, or
12 (at your option) any later version.
14 This program is distributed in the hope that it will be useful,
15 but WITHOUT ANY WARRANTY; without even the implied warranty of
16 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17 GNU General Public License for more details.
19 You should have received a copy of the GNU General Public License
20 along with this program; if not, see <http://www.gnu.org/licenses/>.
24 #include "lib/tdb/include/tdb.h"
25 #include "lib/tevent/tevent.h"
26 #include "lib/util/dlinklist.h"
27 #include "system/network.h"
28 #include "system/filesys.h"
29 #include "../include/ctdb_private.h"
30 #include "../include/ctdb_client.h"
33 /* structures for packet queueing - see common/ctdb_io.c */
39 struct ctdb_queue_pkt {
40 struct ctdb_queue_pkt *next, *prev;
47 struct ctdb_context *ctdb;
48 struct ctdb_partial partial; /* partial input packet */
49 struct ctdb_queue_pkt *out_queue, *out_queue_tail;
50 uint32_t out_queue_length;
55 ctdb_queue_cb_fn_t callback;
62 int ctdb_queue_length(struct ctdb_queue *queue)
64 return queue->out_queue_length;
67 static void dump_packet(unsigned char *data, size_t len)
70 char *p = talloc_array(NULL, char, len*3 + 1);
72 DEBUG(DEBUG_CRIT,("Packet talloc fail"));
76 for (i = 0; i < len; i++)
77 sprintf(p + i*3, " %02x", data[i]);
78 DEBUG(DEBUG_CRIT,("Contents: %s\n", p));
83 called when an incoming connection is readable
85 static void queue_io_read(struct ctdb_queue *queue)
88 ssize_t nread, totread, partlen;
89 uint8_t *data, *data_base;
91 if (ioctl(queue->fd, FIONREAD, &num_ready) != 0) {
95 /* the descriptor has been closed */
100 queue->partial.data = talloc_realloc_size(queue, queue->partial.data,
101 num_ready + queue->partial.length);
103 if (queue->partial.data == NULL) {
104 DEBUG(DEBUG_ERR,("%s: read error alloc failed for %u\n",
105 queue->name, num_ready + queue->partial.length));
109 nread = read(queue->fd, queue->partial.data + queue->partial.length, num_ready);
111 DEBUG(DEBUG_ERR,("%s: read error nread=%d\n",
112 queue->name, (int)nread));
116 partlen = queue->partial.length;
118 data = queue->partial.data;
119 nread += queue->partial.length;
121 queue->partial.data = NULL;
122 queue->partial.length = 0;
124 if (nread >= 4 && *(uint32_t *)data == nread) {
125 /* it is the responsibility of the incoming packet
126 function to free 'data' */
127 queue->callback(data, nread, queue->private_data);
133 while (nread >= 4 && *(uint32_t *)data <= nread) {
134 /* we have at least one packet */
137 bool destroyed = false;
139 len = *(uint32_t *)data;
141 /* bad packet! treat as EOF */
142 DEBUG(DEBUG_CRIT,("%s: Invalid packet of length 0 (nread = %zu, totread = %zu, partlen = %zu)\n",
143 queue->name, nread, totread, partlen));
144 dump_packet(data_base, totread + partlen);
147 d2 = talloc_memdup(queue, data, len);
149 DEBUG(DEBUG_ERR,("%s: read error memdup failed for %u\n",
155 queue->destroyed = &destroyed;
156 queue->callback(d2, len, queue->private_data);
157 /* If callback freed us, don't do anything else. */
161 queue->destroyed = NULL;
168 /* we have only part of a packet */
169 if (data_base == data) {
170 queue->partial.data = data;
171 queue->partial.length = nread;
173 queue->partial.data = talloc_memdup(queue, data, nread);
174 if (queue->partial.data == NULL) {
175 DEBUG(DEBUG_ERR,("%s: read error memdup partial failed for %u\n",
176 queue->name, (unsigned)nread));
179 queue->partial.length = nread;
180 talloc_free(data_base);
185 talloc_free(data_base);
189 queue->callback(NULL, 0, queue->private_data);
193 /* used when an event triggers a dead queue */
194 static void queue_dead(struct event_context *ev, struct timed_event *te,
195 struct timeval t, void *private_data)
197 struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
198 queue->callback(NULL, 0, queue->private_data);
203 called when an incoming connection is writeable
205 static void queue_io_write(struct ctdb_queue *queue)
207 while (queue->out_queue) {
208 struct ctdb_queue_pkt *pkt = queue->out_queue;
210 if (queue->ctdb->flags & CTDB_FLAG_TORTURE) {
211 n = write(queue->fd, pkt->data, 1);
213 n = write(queue->fd, pkt->data, pkt->length);
216 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
217 if (pkt->length != pkt->full_length) {
218 /* partial packet sent - we have to drop it */
219 DLIST_REMOVE(queue->out_queue, pkt);
220 queue->out_queue_length--;
223 talloc_free(queue->fde);
226 event_add_timed(queue->ctdb->ev, queue, timeval_zero(),
232 if (n != pkt->length) {
238 DLIST_REMOVE(queue->out_queue, pkt);
239 queue->out_queue_length--;
243 EVENT_FD_NOT_WRITEABLE(queue->fde);
247 called when an incoming connection is readable or writeable
249 static void queue_io_handler(struct event_context *ev, struct fd_event *fde,
250 uint16_t flags, void *private_data)
252 struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
254 if (flags & EVENT_FD_READ) {
255 queue_io_read(queue);
257 queue_io_write(queue);
263 queue a packet for sending
265 int ctdb_queue_send(struct ctdb_queue *queue, uint8_t *data, uint32_t length)
267 struct ctdb_queue_pkt *pkt;
268 uint32_t length2, full_length;
270 if (queue->alignment) {
271 /* enforce the length and alignment rules from the tcp packet allocator */
272 length2 = (length+(queue->alignment-1)) & ~(queue->alignment-1);
273 *(uint32_t *)data = length2;
278 if (length2 != length) {
279 memset(data+length, 0, length2-length);
282 full_length = length2;
284 /* if the queue is empty then try an immediate write, avoiding
285 queue overhead. This relies on non-blocking sockets */
286 if (queue->out_queue == NULL && queue->fd != -1 &&
287 !(queue->ctdb->flags & CTDB_FLAG_TORTURE)) {
288 ssize_t n = write(queue->fd, data, length2);
289 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
290 talloc_free(queue->fde);
293 event_add_timed(queue->ctdb->ev, queue, timeval_zero(),
295 /* yes, we report success, as the dead node is
296 handled via a separate event */
303 if (length2 == 0) return 0;
306 pkt = talloc(queue, struct ctdb_queue_pkt);
307 CTDB_NO_MEMORY(queue->ctdb, pkt);
309 pkt->data = talloc_memdup(pkt, data, length2);
310 CTDB_NO_MEMORY(queue->ctdb, pkt->data);
312 pkt->length = length2;
313 pkt->full_length = full_length;
315 if (queue->out_queue == NULL && queue->fd != -1) {
316 EVENT_FD_WRITEABLE(queue->fde);
319 DLIST_ADD_END(queue->out_queue, pkt, NULL);
321 queue->out_queue_length++;
323 if (queue->ctdb->tunable.verbose_memory_names != 0) {
324 struct ctdb_req_header *hdr = (struct ctdb_req_header *)pkt->data;
325 switch (hdr->operation) {
326 case CTDB_REQ_CONTROL: {
327 struct ctdb_req_control *c = (struct ctdb_req_control *)hdr;
328 talloc_set_name(pkt, "ctdb_queue_pkt: %s control opcode=%u srvid=%llu datalen=%u",
329 queue->name, (unsigned)c->opcode, (unsigned long long)c->srvid, (unsigned)c->datalen);
332 case CTDB_REQ_MESSAGE: {
333 struct ctdb_req_message *m = (struct ctdb_req_message *)hdr;
334 talloc_set_name(pkt, "ctdb_queue_pkt: %s message srvid=%llu datalen=%u",
335 queue->name, (unsigned long long)m->srvid, (unsigned)m->datalen);
339 talloc_set_name(pkt, "ctdb_queue_pkt: %s operation=%u length=%u src=%u dest=%u",
340 queue->name, (unsigned)hdr->operation, (unsigned)hdr->length,
341 (unsigned)hdr->srcnode, (unsigned)hdr->destnode);
351 setup the fd used by the queue
353 int ctdb_queue_set_fd(struct ctdb_queue *queue, int fd)
356 talloc_free(queue->fde);
360 queue->fde = event_add_fd(queue->ctdb->ev, queue, fd, EVENT_FD_READ,
361 queue_io_handler, queue);
362 if (queue->fde == NULL) {
365 tevent_fd_set_auto_close(queue->fde);
367 if (queue->out_queue) {
368 EVENT_FD_WRITEABLE(queue->fde);
375 /* If someone sets up this pointer, they want to know if the queue is freed */
376 static int queue_destructor(struct ctdb_queue *queue)
378 if (queue->destroyed != NULL)
379 *queue->destroyed = true;
384 setup a packet queue on a socket
386 struct ctdb_queue *ctdb_queue_setup(struct ctdb_context *ctdb,
387 TALLOC_CTX *mem_ctx, int fd, int alignment,
389 ctdb_queue_cb_fn_t callback,
390 void *private_data, const char *fmt, ...)
392 struct ctdb_queue *queue;
395 queue = talloc_zero(mem_ctx, struct ctdb_queue);
396 CTDB_NO_MEMORY_NULL(ctdb, queue);
398 queue->name = talloc_vasprintf(mem_ctx, fmt, ap);
400 CTDB_NO_MEMORY_NULL(ctdb, queue->name);
404 queue->alignment = alignment;
405 queue->private_data = private_data;
406 queue->callback = callback;
408 if (ctdb_queue_set_fd(queue, fd) != 0) {
413 talloc_set_destructor(queue, queue_destructor);