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 library is free software; you can redistribute it and/or
  10    modify it under the terms of the GNU Lesser General Public
  11    License as published by the Free Software Foundation; either
  12    version 2 of the License, or (at your option) any later version.
  13
  14    This library 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 GNU
  17    Lesser General Public License for more details.
  18
  19    You should have received a copy of the GNU Lesser General Public
  20    License along with this library; if not, write to the Free Software
  21    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  22 */
  23
  24 #include "includes.h"
  25 #include "lib/tdb/include/tdb.h"
  26 #include "lib/events/events.h"
  27 #include "lib/util/dlinklist.h"
  28 #include "system/network.h"
  29 #include "system/filesys.h"
  30 #include "../include/ctdb_private.h"
  31 #include "../include/ctdb.h"
  32
  33 /* structures for packet queueing - see common/ctdb_io.c */
  34 struct ctdb_partial {
  35         uint8_t *data;
  36         uint32_t length;
  37 };
  38
  39 struct ctdb_queue_pkt {
  40         struct ctdb_queue_pkt *next, *prev;
  41         uint8_t *data;
  42         uint32_t length;
  43 };
  44
  45 struct ctdb_queue {
  46         struct ctdb_context *ctdb;
  47         struct ctdb_partial partial; /* partial input packet */
  48         struct ctdb_queue_pkt *out_queue;
  49         struct fd_event *fde;
  50         int fd;
  51         size_t alignment;
  52         void *private_data;
  53         ctdb_queue_cb_fn_t callback;
  54 };
  55
  56
  57
  58 /*
  59   called when an incoming connection is readable
  60 */
  61 static void queue_io_read(struct ctdb_queue *queue)
  62 {
  63         int num_ready = 0;
  64         ssize_t nread;
  65         uint8_t *data, *data_base;
  66
  67         if (ioctl(queue->fd, FIONREAD, &num_ready) != 0 ||
  68             num_ready == 0) {
  69                 /* the descriptor has been closed */
  70                 goto failed;
  71         }
  72
  73
  74         queue->partial.data = talloc_realloc_size(queue, queue->partial.data,
  75                                                   num_ready + queue->partial.length);
  76
  77         if (queue->partial.data == NULL) {
  78                 goto failed;
  79         }
  80
  81         nread = read(queue->fd, queue->partial.data + queue->partial.length, num_ready);
  82         if (nread <= 0) {
  83                 goto failed;
  84         }
  85
  86
  87         data = queue->partial.data;
  88         nread += queue->partial.length;
  89
  90         queue->partial.data = NULL;
  91         queue->partial.length = 0;
  92
  93         if (nread >= 4 && *(uint32_t *)data == nread) {
  94                 /* it is the responsibility of the incoming packet
  95                  function to free 'data' */
  96                 queue->callback(data, nread, queue->private_data);
  97                 return;
  98         }
  99
 100         data_base = data;
 101
 102         while (nread >= 4 && *(uint32_t *)data <= nread) {
 103                 /* we have at least one packet */
 104                 uint8_t *d2;
 105                 uint32_t len;
 106                 len = *(uint32_t *)data;
 107                 d2 = talloc_memdup(queue, data, len);
 108                 if (d2 == NULL) {
 109                         /* sigh */
 110                         goto failed;
 111                 }
 112                 queue->callback(d2, len, queue->private_data);
 113                 data += len;
 114                 nread -= len;
 115         }
 116
 117         if (nread > 0) {
 118                 /* we have only part of a packet */
 119                 if (data_base == data) {
 120                         queue->partial.data = data;
 121                         queue->partial.length = nread;
 122                 } else {
 123                         queue->partial.data = talloc_memdup(queue, data, nread);
 124                         if (queue->partial.data == NULL) {
 125                                 goto failed;
 126                         }
 127                         queue->partial.length = nread;
 128                         talloc_free(data_base);
 129                 }
 130                 return;
 131         }
 132
 133         talloc_free(data_base);
 134         return;
 135
 136 failed:
 137         queue->callback(NULL, 0, queue->private_data);
 138 }
 139
 140
 141 /* used when an event triggers a dead queue */
 142 static void queue_dead(struct event_context *ev, struct timed_event *te,
 143                        struct timeval t, void *private_data)
 144 {
 145         struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
 146         queue->callback(NULL, 0, queue->private_data);
 147 }
 148
 149
 150 /*
 151   called when an incoming connection is writeable
 152 */
 153 static void queue_io_write(struct ctdb_queue *queue)
 154 {
 155         while (queue->out_queue) {
 156                 struct ctdb_queue_pkt *pkt = queue->out_queue;
 157                 ssize_t n;
 158
 159                 n = write(queue->fd, pkt->data, pkt->length);
 160
 161                 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
 162                         event_add_timed(queue->ctdb->ev, queue, timeval_zero(),
 163                                         queue_dead, queue);
 164                         EVENT_FD_NOT_WRITEABLE(queue->fde);
 165                         return;
 166                 }
 167                 if (n <= 0) return;
 168
 169                 if (n != pkt->length) {
 170                         pkt->length -= n;
 171                         pkt->data += n;
 172                         return;
 173                 }
 174
 175                 DLIST_REMOVE(queue->out_queue, pkt);
 176                 talloc_free(pkt);
 177         }
 178
 179         EVENT_FD_NOT_WRITEABLE(queue->fde);
 180 }
 181
 182 /*
 183   called when an incoming connection is readable or writeable
 184 */
 185 static void queue_io_handler(struct event_context *ev, struct fd_event *fde,
 186                              uint16_t flags, void *private_data)
 187 {
 188         struct ctdb_queue *queue = talloc_get_type(private_data, struct ctdb_queue);
 189
 190         if (flags & EVENT_FD_READ) {
 191                 queue_io_read(queue);
 192         } else {
 193                 queue_io_write(queue);
 194         }
 195 }
 196
 197
 198 /*
 199   queue a packet for sending
 200 */
 201 int ctdb_queue_send(struct ctdb_queue *queue, uint8_t *data, uint32_t length)
 202 {
 203         struct ctdb_queue_pkt *pkt;
 204         uint32_t length2;
 205
 206         /* enforce the length and alignment rules from the tcp packet allocator */
 207         length2 = (length+(queue->alignment-1)) & ~(queue->alignment-1);
 208         *(uint32_t *)data = length2;
 209
 210         if (length2 != length) {
 211                 memset(data+length, 0, length2-length);
 212         }
 213
 214         /* if the queue is empty then try an immediate write, avoiding
 215            queue overhead. This relies on non-blocking sockets */
 216         if (queue->out_queue == NULL && queue->fd != -1) {
 217                 ssize_t n = write(queue->fd, data, length2);
 218                 if (n == -1 && errno != EAGAIN && errno != EWOULDBLOCK) {
 219                         event_add_timed(queue->ctdb->ev, queue, timeval_zero(),
 220                                         queue_dead, queue);
 221                         /* yes, we report success, as the dead node is
 222                            handled via a separate event */
 223                         return 0;
 224                 }
 225                 if (n > 0) {
 226                         data += n;
 227                         length2 -= n;
 228                 }
 229                 if (length2 == 0) return 0;
 230         }
 231
 232         pkt = talloc(queue, struct ctdb_queue_pkt);
 233         CTDB_NO_MEMORY(queue->ctdb, pkt);
 234
 235         pkt->data = talloc_memdup(pkt, data, length2);
 236         CTDB_NO_MEMORY(queue->ctdb, pkt->data);
 237
 238         pkt->length = length2;
 239
 240         if (queue->out_queue == NULL && queue->fd != -1) {
 241                 EVENT_FD_WRITEABLE(queue->fde);
 242         }
 243
 244         DLIST_ADD_END(queue->out_queue, pkt, struct ctdb_queue_pkt *);
 245
 246         return 0;
 247 }
 248
 249
 250 /*
 251   setup the fd used by the queue
 252  */
 253 int ctdb_queue_set_fd(struct ctdb_queue *queue, int fd)
 254 {
 255         queue->fd = fd;
 256         talloc_free(queue->fde);
 257         queue->fde = NULL;
 258
 259         if (fd != -1) {
 260                 queue->fde = event_add_fd(queue->ctdb->ev, queue, fd, EVENT_FD_READ,
 261                                           queue_io_handler, queue);
 262                 if (queue->fde == NULL) {
 263                         return -1;
 264                 }
 265
 266                 if (queue->out_queue) {
 267                         EVENT_FD_WRITEABLE(queue->fde);
 268                 }
 269         }
 270
 271         return 0;
 272 }
 273
 274
 275
 276 /*
 277   setup a packet queue on a socket
 278  */
 279 struct ctdb_queue *ctdb_queue_setup(struct ctdb_context *ctdb,
 280                                     TALLOC_CTX *mem_ctx, int fd, int alignment,
 281
 282                                     ctdb_queue_cb_fn_t callback,
 283                                     void *private_data)
 284 {
 285         struct ctdb_queue *queue;
 286
 287         queue = talloc_zero(mem_ctx, struct ctdb_queue);
 288         CTDB_NO_MEMORY_NULL(ctdb, queue);
 289
 290         queue->ctdb = ctdb;
 291         queue->fd = fd;
 292         queue->alignment = alignment;
 293         queue->private_data = private_data;
 294         queue->callback = callback;
 295         if (fd != -1) {
 296                 if (ctdb_queue_set_fd(queue, fd) != 0) {
 297                         talloc_free(queue);
 298                         return NULL;
 299                 }
 300         }
 301
 302         return queue;
 303 }