ctdb/libctdb/io_elem.c

   1 /*
   2    Simple queuing of input and output records for libctdb
   3
   4    Copyright (C) Rusty Russell 2010
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19 #include <sys/types.h>
  20 #include <sys/socket.h>
  21 #include <string.h>
  22 #include <stdint.h>
  23 #include <stdbool.h>
  24 #include <unistd.h>
  25 #include <errno.h>
  26 #include <stdlib.h>
  27 #include "libctdb_private.h"
  28 #include "io_elem.h"
  29 #include <tdb.h>
  30 #include <netinet/in.h>
  31 #include <dlinklist.h>
  32 #include <ctdb_protocol.h> // For CTDB_DS_ALIGNMENT and ctdb_req_header
  33
  34 struct io_elem {
  35         struct io_elem *next, *prev;
  36         size_t len, off;
  37         char *data;
  38 };
  39
  40 struct io_elem *new_io_elem(size_t len)
  41 {
  42         struct io_elem *elem;
  43         size_t ask = len;
  44
  45         len = (len + (CTDB_DS_ALIGNMENT-1)) & ~(CTDB_DS_ALIGNMENT-1);
  46
  47         elem = malloc(sizeof(*elem));
  48         if (!elem)
  49                 return NULL;
  50         elem->data = malloc(len);
  51         if (!elem->data) {
  52                 free(elem);
  53                 return NULL;
  54         }
  55
  56         /* stamp out any padding to keep valgrind happy */
  57         if (ask != len) {
  58                 memset(elem->data + ask, 0, len-ask);
  59         }
  60         elem->len = len;
  61         elem->off = 0;
  62         elem->next = NULL;
  63         elem->prev = NULL;
  64         return elem;
  65 }
  66
  67 void free_io_elem(struct io_elem *io)
  68 {
  69         free(io->data);
  70         free(io);
  71 }
  72
  73 bool io_elem_finished(const struct io_elem *io)
  74 {
  75         return io->off == io->len;
  76 }
  77
  78 void io_elem_init_req_header(struct io_elem *io,
  79                              uint32_t operation,
  80                              uint32_t destnode,
  81                              uint32_t reqid)
  82 {
  83         struct ctdb_req_header *hdr = io_elem_data(io, NULL);
  84
  85         hdr->length = io->len;
  86         hdr->ctdb_magic = CTDB_MAGIC;
  87         hdr->ctdb_version = CTDB_VERSION;
  88         /* Generation and srcnode only used for inter-ctdbd communication. */
  89         hdr->generation = 0;
  90         hdr->destnode = destnode;
  91         hdr->srcnode = 0;
  92         hdr->operation = operation;
  93         hdr->reqid = reqid;
  94 }
  95
  96 /* Access to raw data: if len is non-NULL it is filled in. */
  97 void *io_elem_data(const struct io_elem *io, size_t *len)
  98 {
  99         if (len)
 100                 *len = io->len;
 101         return io->data;
 102 }
 103
 104 /* Returns -1 if we hit an error.  Errno will be set. */
 105 int read_io_elem(int fd, struct io_elem *io)
 106 {
 107         ssize_t ret;
 108
 109         ret = read(fd, io->data + io->off, io->len - io->off);
 110         if (ret < 0)
 111                 return ret;
 112
 113         io->off += ret;
 114         if (io_elem_finished(io)) {
 115                 struct ctdb_req_header *hdr = (void *)io->data;
 116
 117                 /* Finished.  But maybe this was just header? */
 118                 if (io->len == sizeof(*hdr) && hdr->length > io->len) {
 119                         int reret;
 120                         void *newdata;
 121                         /* Enlarge and re-read. */
 122                         io->len = hdr->length;
 123                         newdata = realloc(io->data, io->len);
 124                         if (!newdata)
 125                                 return -1;
 126                         io->data = newdata;
 127                         /* Try reading again immediately. */
 128                         reret = read_io_elem(fd, io);
 129                         if (reret >= 0)
 130                                 reret += ret;
 131                         return reret;
 132                 }
 133         }
 134         return ret;
 135 }
 136
 137 /* Returns -1 if we hit an error.  Errno will be set. */
 138 int write_io_elem(int fd, struct io_elem *io)
 139 {
 140         ssize_t ret;
 141
 142         ret = write(fd, io->data + io->off, io->len - io->off);
 143         if (ret < 0)
 144                 return ret;
 145
 146         io->off += ret;
 147         return ret;
 148 }
 149
 150 void io_elem_reset(struct io_elem *io)
 151 {
 152         io->off = 0;
 153 }
 154
 155 void io_elem_queue(struct ctdb_connection *ctdb, struct io_elem *io)
 156 {
 157         DLIST_ADD_END(ctdb->inqueue, io, struct io_elem);
 158 }
 159
 160 void io_elem_dequeue(struct ctdb_connection *ctdb, struct io_elem *io)
 161 {
 162         DLIST_REMOVE(ctdb->inqueue, io);
 163 }
 164