ctdb-daemon: Stop using tevent compatibility definitions
[vlendec/samba-autobuild/.git] / ctdb / common / ctdb_io.c
1 /* 
2    ctdb database library
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
5    data.
6
7    Copyright (C) Andrew Tridgell  2006
8
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.
13    
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.
18    
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/>.
21 */
22
23 #include "includes.h"
24 #include "tdb.h"
25 #include "lib/util/dlinklist.h"
26 #include "system/network.h"
27 #include "system/filesys.h"
28 #include "../include/ctdb_private.h"
29 #include "../include/ctdb_client.h"
30 #include <stdarg.h>
31
32 #include "common/system.h"
33
34 #define QUEUE_BUFFER_SIZE       (16*1024)
35
36 /* structures for packet queueing - see common/ctdb_io.c */
37 struct ctdb_buffer {
38         uint8_t *data;
39         uint32_t length;
40         uint32_t size;
41         uint32_t extend;
42 };
43
44 struct ctdb_queue_pkt {
45         struct ctdb_queue_pkt *next, *prev;
46         uint8_t *data;
47         uint32_t length;
48         uint32_t full_length;
49         uint8_t buf[];
50 };
51
52 struct ctdb_queue {
53         struct ctdb_context *ctdb;
54         struct tevent_immediate *im;
55         struct ctdb_buffer buffer; /* input buffer */
56         struct ctdb_queue_pkt *out_queue, *out_queue_tail;
57         uint32_t out_queue_length;
58         struct tevent_fd *fde;
59         int fd;
60         size_t alignment;
61         void *private_data;
62         ctdb_queue_cb_fn_t callback;
63         bool *destroyed;
64         const char *name;
65 };
66
67
68
69 int ctdb_queue_length(struct ctdb_queue *queue)
70 {
71         return queue->out_queue_length;
72 }
73
74 static void queue_process(struct ctdb_queue *queue);
75
76 static void queue_process_event(struct tevent_context *ev, struct tevent_immediate *im,
77                                 void *private_data)
78 {
79         struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
80
81         queue_process(queue);
82 }
83
84 /*
85  * This function is used to process data in queue buffer.
86  *
87  * Queue callback function can end up freeing the queue, there should not be a
88  * loop processing packets from queue buffer.  Instead set up a timed event for
89  * immediate run to process remaining packets from buffer.
90  */
91 static void queue_process(struct ctdb_queue *queue)
92 {
93         uint32_t pkt_size;
94         uint8_t *data;
95
96         if (queue->buffer.length < sizeof(pkt_size)) {
97                 return;
98         }
99
100         pkt_size = *(uint32_t *)queue->buffer.data;
101         if (pkt_size == 0) {
102                 DEBUG(DEBUG_CRIT, ("Invalid packet of length 0\n"));
103                 goto failed;
104         }
105
106         if (queue->buffer.length < pkt_size) {
107                 if (pkt_size > QUEUE_BUFFER_SIZE) {
108                         queue->buffer.extend = pkt_size;
109                 }
110                 return;
111         }
112
113         /* Extract complete packet */
114         data = talloc_size(queue, pkt_size);
115         if (data == NULL) {
116                 DEBUG(DEBUG_ERR, ("read error alloc failed for %u\n", pkt_size));
117                 return;
118         }
119         memcpy(data, queue->buffer.data, pkt_size);
120
121         /* Shift packet out from buffer */
122         if (queue->buffer.length > pkt_size) {
123                 memmove(queue->buffer.data,
124                         queue->buffer.data + pkt_size,
125                         queue->buffer.length - pkt_size);
126         }
127         queue->buffer.length -= pkt_size;
128
129         if (queue->buffer.length > 0) {
130                 /* There is more data to be processed, schedule an event */
131                 tevent_schedule_immediate(queue->im, queue->ctdb->ev,
132                                           queue_process_event, queue);
133         } else {
134                 if (queue->buffer.size > QUEUE_BUFFER_SIZE) {
135                         TALLOC_FREE(queue->buffer.data);
136                         queue->buffer.size = 0;
137                 }
138         }
139
140         /* It is the responsibility of the callback to free 'data' */
141         queue->callback(data, pkt_size, queue->private_data);
142         return;
143
144 failed:
145         queue->callback(NULL, 0, queue->private_data);
146
147 }
148
149
150 /*
151   called when an incoming connection is readable
152   This function MUST be safe for reentry via the queue callback!
153 */
154 static void queue_io_read(struct ctdb_queue *queue)
155 {
156         int num_ready = 0;
157         ssize_t nread;
158         uint8_t *data;
159         int navail;
160
161         /* check how much data is available on the socket for immediately
162            guaranteed nonblocking access.
163            as long as we are careful never to try to read more than this
164            we know all reads will be successful and will neither block
165            nor fail with a "data not available right now" error
166         */
167         if (ioctl(queue->fd, FIONREAD, &num_ready) != 0) {
168                 return;
169         }
170         if (num_ready == 0) {
171                 /* the descriptor has been closed */
172                 goto failed;
173         }
174
175         if (queue->buffer.data == NULL) {
176                 /* starting fresh, allocate buf to read data */
177                 queue->buffer.data = talloc_size(queue, QUEUE_BUFFER_SIZE);
178                 if (queue->buffer.data == NULL) {
179                         DEBUG(DEBUG_ERR, ("read error alloc failed for %u\n", num_ready));
180                         goto failed;
181                 }
182                 queue->buffer.size = QUEUE_BUFFER_SIZE;
183         } else if (queue->buffer.extend > 0) {
184                 /* extending buffer */
185                 data = talloc_realloc_size(queue, queue->buffer.data, queue->buffer.extend);
186                 if (data == NULL) {
187                         DEBUG(DEBUG_ERR, ("read error realloc failed for %u\n", queue->buffer.extend));
188                         goto failed;
189                 }
190                 queue->buffer.data = data;
191                 queue->buffer.size = queue->buffer.extend;
192                 queue->buffer.extend = 0;
193         }
194
195         navail = queue->buffer.size - queue->buffer.length;
196         if (num_ready > navail) {
197                 num_ready = navail;
198         }
199
200         if (num_ready > 0) {
201                 nread = sys_read(queue->fd,
202                                  queue->buffer.data + queue->buffer.length,
203                                  num_ready);
204                 if (nread <= 0) {
205                         DEBUG(DEBUG_ERR, ("read error nread=%d\n", (int)nread));
206                         goto failed;
207                 }
208                 queue->buffer.length += nread;
209         }
210
211         queue_process(queue);
212         return;
213
214 failed:
215         queue->callback(NULL, 0, queue->private_data);
216 }
217
218
219 /* used when an event triggers a dead queue */
220 static void queue_dead(struct tevent_context *ev, struct tevent_immediate *im,
221                        void *private_data)
222 {
223         struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
224         queue->callback(NULL, 0, queue->private_data);
225 }
226
227
228 /*
229   called when an incoming connection is writeable
230 */
231 static void queue_io_write(struct ctdb_queue *queue)
232 {
233         while (queue->out_queue) {
234                 struct ctdb_queue_pkt *pkt = queue->out_queue;
235                 ssize_t n;
236                 if (queue->ctdb->flags & CTDB_FLAG_TORTURE) {
237                         n = write(queue->fd, pkt->data, 1);
238                 } else {
239                         n = write(queue->fd, pkt->data, pkt->length);
240                 }
241
242                 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
243                         if (pkt->length != pkt->full_length) {
244                                 /* partial packet sent - we have to drop it */
245                                 DLIST_REMOVE(queue->out_queue, pkt);
246                                 queue->out_queue_length--;
247                                 talloc_free(pkt);
248                         }
249                         talloc_free(queue->fde);
250                         queue->fde = NULL;
251                         queue->fd = -1;
252                         tevent_schedule_immediate(queue->im, queue->ctdb->ev,
253                                                   queue_dead, queue);
254                         return;
255                 }
256                 if (n <= 0) return;
257                 
258                 if (n != pkt->length) {
259                         pkt->length -= n;
260                         pkt->data += n;
261                         return;
262                 }
263
264                 DLIST_REMOVE(queue->out_queue, pkt);
265                 queue->out_queue_length--;
266                 talloc_free(pkt);
267         }
268
269         TEVENT_FD_NOT_WRITEABLE(queue->fde);
270 }
271
272 /*
273   called when an incoming connection is readable or writeable
274 */
275 static void queue_io_handler(struct tevent_context *ev, struct tevent_fd *fde,
276                              uint16_t flags, void *private_data)
277 {
278         struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
279
280         if (flags & TEVENT_FD_READ) {
281                 queue_io_read(queue);
282         } else {
283                 queue_io_write(queue);
284         }
285 }
286
287
288 /*
289   queue a packet for sending
290 */
291 int ctdb_queue_send(struct ctdb_queue *queue, uint8_t *data, uint32_t length)
292 {
293         struct ctdb_req_header *hdr = (struct ctdb_req_header *)data;
294         struct ctdb_queue_pkt *pkt;
295         uint32_t length2, full_length;
296
297         if (queue->alignment) {
298                 /* enforce the length and alignment rules from the tcp packet allocator */
299                 length2 = (length+(queue->alignment-1)) & ~(queue->alignment-1);
300                 *(uint32_t *)data = length2;
301         } else {
302                 length2 = length;
303         }
304
305         if (length2 != length) {
306                 memset(data+length, 0, length2-length);
307         }
308
309         full_length = length2;
310         
311         /* if the queue is empty then try an immediate write, avoiding
312            queue overhead. This relies on non-blocking sockets */
313         if (queue->out_queue == NULL && queue->fd != -1 &&
314             !(queue->ctdb->flags & CTDB_FLAG_TORTURE)) {
315                 ssize_t n = write(queue->fd, data, length2);
316                 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
317                         talloc_free(queue->fde);
318                         queue->fde = NULL;
319                         queue->fd = -1;
320                         tevent_schedule_immediate(queue->im, queue->ctdb->ev,
321                                                   queue_dead, queue);
322                         /* yes, we report success, as the dead node is 
323                            handled via a separate event */
324                         return 0;
325                 }
326                 if (n > 0) {
327                         data += n;
328                         length2 -= n;
329                 }
330                 if (length2 == 0) return 0;
331         }
332
333         pkt = talloc_size(
334                 queue, offsetof(struct ctdb_queue_pkt, buf) + length2);
335         CTDB_NO_MEMORY(queue->ctdb, pkt);
336         talloc_set_name_const(pkt, "struct ctdb_queue_pkt");
337
338         pkt->data = pkt->buf;
339         memcpy(pkt->data, data, length2);
340
341         pkt->length = length2;
342         pkt->full_length = full_length;
343
344         if (queue->out_queue == NULL && queue->fd != -1) {
345                 TEVENT_FD_WRITEABLE(queue->fde);
346         }
347
348         DLIST_ADD_END(queue->out_queue, pkt, NULL);
349
350         queue->out_queue_length++;
351
352         if (queue->ctdb->tunable.verbose_memory_names != 0) {
353                 switch (hdr->operation) {
354                 case CTDB_REQ_CONTROL: {
355                         struct ctdb_req_control *c = (struct ctdb_req_control *)hdr;
356                         talloc_set_name(pkt, "ctdb_queue_pkt: %s control opcode=%u srvid=%llu datalen=%u",
357                                         queue->name, (unsigned)c->opcode, (unsigned long long)c->srvid, (unsigned)c->datalen);
358                         break;
359                 }
360                 case CTDB_REQ_MESSAGE: {
361                         struct ctdb_req_message *m = (struct ctdb_req_message *)hdr;
362                         talloc_set_name(pkt, "ctdb_queue_pkt: %s message srvid=%llu datalen=%u",
363                                         queue->name, (unsigned long long)m->srvid, (unsigned)m->datalen);
364                         break;
365                 }
366                 default:
367                         talloc_set_name(pkt, "ctdb_queue_pkt: %s operation=%u length=%u src=%u dest=%u",
368                                         queue->name, (unsigned)hdr->operation, (unsigned)hdr->length,
369                                         (unsigned)hdr->srcnode, (unsigned)hdr->destnode);
370                         break;
371                 }
372         }
373
374         return 0;
375 }
376
377
378 /*
379   setup the fd used by the queue
380  */
381 int ctdb_queue_set_fd(struct ctdb_queue *queue, int fd)
382 {
383         queue->fd = fd;
384         talloc_free(queue->fde);
385         queue->fde = NULL;
386
387         if (fd != -1) {
388                 queue->fde = tevent_add_fd(queue->ctdb->ev, queue, fd,
389                                            TEVENT_FD_READ,
390                                            queue_io_handler, queue);
391                 if (queue->fde == NULL) {
392                         return -1;
393                 }
394                 tevent_fd_set_auto_close(queue->fde);
395
396                 if (queue->out_queue) {
397                         TEVENT_FD_WRITEABLE(queue->fde);
398                 }
399         }
400
401         return 0;
402 }
403
404 /* If someone sets up this pointer, they want to know if the queue is freed */
405 static int queue_destructor(struct ctdb_queue *queue)
406 {
407         TALLOC_FREE(queue->buffer.data);
408         queue->buffer.length = 0;
409         queue->buffer.size = 0;
410         if (queue->destroyed != NULL)
411                 *queue->destroyed = true;
412         return 0;
413 }
414
415 /*
416   setup a packet queue on a socket
417  */
418 struct ctdb_queue *ctdb_queue_setup(struct ctdb_context *ctdb,
419                                     TALLOC_CTX *mem_ctx, int fd, int alignment,
420                                     ctdb_queue_cb_fn_t callback,
421                                     void *private_data, const char *fmt, ...)
422 {
423         struct ctdb_queue *queue;
424         va_list ap;
425
426         queue = talloc_zero(mem_ctx, struct ctdb_queue);
427         CTDB_NO_MEMORY_NULL(ctdb, queue);
428         va_start(ap, fmt);
429         queue->name = talloc_vasprintf(mem_ctx, fmt, ap);
430         va_end(ap);
431         CTDB_NO_MEMORY_NULL(ctdb, queue->name);
432
433         queue->im= tevent_create_immediate(queue);
434         CTDB_NO_MEMORY_NULL(ctdb, queue->im);
435
436         queue->ctdb = ctdb;
437         queue->fd = fd;
438         queue->alignment = alignment;
439         queue->private_data = private_data;
440         queue->callback = callback;
441         if (fd != -1) {
442                 if (ctdb_queue_set_fd(queue, fd) != 0) {
443                         talloc_free(queue);
444                         return NULL;
445                 }
446         }
447         talloc_set_destructor(queue, queue_destructor);
448
449         return queue;
450 }