ucw/main-rec.c

   1 /*
   2  *      UCW Library -- Main Loop: Record I/O
   3  *
   4  *      (c) 2011 Martin Mares <mj@ucw.cz>
   5  *
   6  *      This software may be freely distributed and used according to the terms
   7  *      of the GNU Lesser General Public License.
   8  */
   9
  10 #undef LOCAL_DEBUG
  11
  12 #include "ucw/lib.h"
  13 #include "ucw/mainloop.h"
  14
  15 #include <stdio.h>
  16 #include <string.h>
  17 #include <stdlib.h>
  18 #include <unistd.h>
  19 #include <errno.h>
  20
  21 struct rio_buffer {
  22   cnode n;
  23   uns full;
  24   uns written;
  25   byte buf[];
  26 };
  27
  28 static void
  29 rec_io_timer_expired(struct main_timer *tm)
  30 {
  31   struct main_rec_io *rio = tm->data;
  32   timer_del(&rio->timer);
  33   rio->notify_handler(rio, RIO_ERR_TIMEOUT);
  34 }
  35
  36 void
  37 rec_io_add(struct main_rec_io *rio, int fd)
  38 {
  39   rio->file.fd = fd;
  40   file_add(&rio->file);
  41   rio->timer.handler = rec_io_timer_expired;
  42   rio->timer.data = rio;
  43   clist_init(&rio->idle_write_buffers);
  44   clist_init(&rio->busy_write_buffers);
  45 }
  46
  47 void
  48 rec_io_del(struct main_rec_io *rio)
  49 {
  50   timer_del(&rio->timer);
  51   file_del(&rio->file);
  52
  53   if (rio->read_buf)
  54     {
  55       DBG("RIO: Freeing read buffer");
  56       xfree(rio->read_buf);
  57       rio->read_buf = NULL;
  58     }
  59
  60   struct rio_buffer *b;
  61   while ((b = clist_remove_head(&rio->idle_write_buffers)) || (b = clist_remove_head(&rio->busy_write_buffers)))
  62     {
  63       DBG("RIO: Freeing write buffer");
  64       xfree(b);
  65     }
  66 }
  67
  68 static int
  69 rec_io_read_handler(struct main_file *fi)
  70 {
  71   struct main_rec_io *rio = (struct main_rec_io *) fi;
  72
  73   if (rio->read_rec_max && rio->read_avail >= rio->read_rec_max)
  74     {
  75       rec_io_stop_read(rio);
  76       rio->notify_handler(rio, RIO_ERR_READ_RECORD_TOO_LARGE);
  77       return HOOK_IDLE;
  78     }
  79
  80 restart: ;
  81   uns rec_start_pos = rio->read_rec_start - rio->read_buf;
  82   uns rec_end_pos = rec_start_pos + rio->read_avail;
  83   uns free_space = rio->read_buf_size - rec_end_pos;
  84   DBG("RIO READ: rec_start=%u avail=%u prev_avail=%u free=%u/%u",
  85     rec_start_pos, rio->read_avail, rio->read_prev_avail,
  86     free_space, rio->read_buf_size);
  87   // FIXME: Constants?
  88   if (free_space <= rio->read_buf_size/8)
  89     {
  90       if (rec_start_pos && rec_start_pos >= rio->read_buf_size/2)
  91         {
  92           // Moving the partial record to the start of the buffer
  93           DBG("RIO READ: Moving partial record to start");
  94           memmove(rio->read_buf, rio->read_rec_start, rio->read_avail);
  95           rio->read_rec_start = rio->read_buf;
  96         }
  97       else
  98         {
  99           DBG("RIO READ: Resizing buffer");
 100           rio->read_buf_size *= 2;
 101           rio->read_buf = xrealloc(rio->read_buf, rio->read_buf_size);
 102           rio->read_rec_start = rio->read_buf + rec_start_pos;
 103         }
 104       goto restart;
 105     }
 106
 107   int l = read(fi->fd, rio->read_buf + rec_end_pos, free_space);
 108   DBG("RIO READ: Read %d bytes", l);
 109   if (l < 0)
 110     {
 111       if (errno != EINTR && errno != EAGAIN)
 112         {
 113           DBG("RIO READ: Signalling error");
 114           rec_io_stop_read(rio);
 115           rio->notify_handler(rio, RIO_ERR_READ);
 116         }
 117       return HOOK_IDLE;
 118     }
 119   if (!l)
 120     {
 121       DBG("RIO READ: Signalling EOF");
 122       rec_io_stop_read(rio);
 123       rio->notify_handler(rio, RIO_ERR_READ_EOF);
 124       return HOOK_IDLE;
 125     }
 126   rio->read_prev_avail = rio->read_avail;
 127   rio->read_avail += l;
 128   DBG("RIO READ: Available: %u bytes", rio->read_avail);
 129
 130   uns got;
 131   while (rio->read_running && (got = rio->read_handler(rio)))
 132     {
 133       DBG("RIO READ: Ate %u bytes", got);
 134       rio->read_rec_start += got;
 135       rio->read_avail -= got;
 136       rio->read_prev_avail = 0;
 137       if (!rio->read_avail)
 138         {
 139           DBG("RIO READ: Resetting buffer");
 140           rio->read_rec_start = rio->read_buf;
 141           break;
 142         }
 143     }
 144   DBG("RIO READ: Want more");
 145   return (rio->read_running ? HOOK_RETRY : HOOK_IDLE);
 146 }
 147
 148 void
 149 rec_io_start_read(struct main_rec_io *rio)
 150 {
 151   ASSERT(clist_is_linked(&rio->file.n));
 152   if (rio->read_running)
 153     return;
 154   if (!rio->read_buf)
 155     {
 156       if (!rio->read_buf_size)
 157         rio->read_buf_size = 256;
 158       rio->read_buf = xmalloc(rio->read_buf_size);
 159       DBG("RIO: Created buffer (%u bytes)", rio->read_buf_size);
 160       rio->read_rec_start = rio->read_buf;
 161     }
 162   rio->file.read_handler = rec_io_read_handler;
 163   file_chg(&rio->file);
 164   rio->read_running = 1;
 165   DBG("RIO: Reading started");
 166 }
 167
 168 void
 169 rec_io_stop_read(struct main_rec_io *rio)
 170 {
 171   ASSERT(clist_is_linked(&rio->file.n));
 172   if (!rio->read_running)
 173     return;
 174   rio->file.read_handler = NULL;
 175   file_chg(&rio->file);
 176   rio->read_running = 0;
 177   DBG("RIO: Reading stopped");
 178 }
 179
 180 static void
 181 rec_io_stop_write(struct main_rec_io *rio)
 182 {
 183   DBG("RIO WRITE: Stopping write");
 184   ASSERT(!rio->write_watermark);
 185   rio->file.write_handler = NULL;
 186   file_chg(&rio->file);
 187 }
 188
 189 static int
 190 rec_io_write_handler(struct main_file *fi)
 191 {
 192   struct main_rec_io *rio = (struct main_rec_io *) fi;
 193   struct rio_buffer *b = clist_head(&rio->busy_write_buffers);
 194   if (!b)
 195     {
 196       rec_io_stop_write(rio);
 197       return HOOK_IDLE;
 198     }
 199
 200   int l = write(fi->fd, b->buf + b->written, b->full - b->written);
 201   DBG("RIO WRITE: Written %d bytes", l);
 202   if (l < 0)
 203     {
 204       if (errno != EINTR && errno != EAGAIN)
 205         {
 206           rec_io_stop_write(rio);
 207           rio->notify_handler(rio, RIO_ERR_WRITE);
 208         }
 209       return HOOK_IDLE;
 210     }
 211   b->written += l;
 212   if (b->written == b->full)
 213     {
 214       DBG("RIO WRITE: Written full buffer");
 215       clist_remove(&b->n);
 216       clist_add_tail(&rio->idle_write_buffers, &b->n);
 217     }
 218
 219   rio->write_watermark -= l;
 220   int ret = HOOK_RETRY;
 221   if (!rio->write_watermark)
 222     {
 223       ret = HOOK_IDLE;
 224       rec_io_stop_write(rio);
 225     }
 226
 227   // Call the hook, but carefully, because it can delete the RIO structure
 228   if (rio->notify_handler(rio, rio->write_watermark ? RIO_EVENT_PART_WRITTEN : RIO_EVENT_ALL_WRITTEN) == HOOK_IDLE)
 229     ret = HOOK_IDLE;
 230   return ret;
 231 }
 232
 233 static struct rio_buffer *
 234 rec_io_get_buffer(struct main_rec_io *rio)
 235 {
 236   struct rio_buffer *b = clist_remove_tail(&rio->idle_write_buffers);
 237   if (b)
 238     DBG("RIO WRITE: Recycled old buffer");
 239   else
 240     {
 241       if (!rio->write_buf_size)
 242         rio->write_buf_size = 1024;
 243       b = xmalloc(sizeof(struct rio_buffer) + rio->write_buf_size);
 244       DBG("RIO WRITE: Allocated new buffer");
 245     }
 246   b->full = b->written = 0;
 247   return b;
 248 }
 249
 250 void
 251 rec_io_write(struct main_rec_io *rio, void *data, uns len)
 252 {
 253   byte *bdata = data;
 254   ASSERT(clist_is_linked(&rio->file.n));
 255   if (!len)
 256     return;
 257
 258   while (len)
 259     {
 260       struct rio_buffer *b = clist_tail(&rio->busy_write_buffers);
 261       if (!b || b->full >= rio->write_buf_size)
 262         {
 263           b = rec_io_get_buffer(rio);
 264           clist_add_tail(&rio->busy_write_buffers, &b->n);
 265         }
 266       uns l = MIN(len, rio->write_buf_size - b->full);
 267       memcpy(b->buf + b->full, bdata, l);
 268       b->full += l;
 269       bdata += l;
 270       len -= l;
 271       rio->write_watermark += l;
 272       DBG("RIO WRITE: Buffered %u bytes of data (total %u)", l, rio->write_watermark);
 273     }
 274
 275   if (!rio->file.write_handler)
 276     {
 277       DBG("RIO WRITE: Starting write");
 278       rio->file.write_handler = rec_io_write_handler;
 279       file_chg(&rio->file);
 280     }
 281 }
 282
 283 void
 284 rec_io_set_timeout(struct main_rec_io *rio, timestamp_t expires_delta)
 285 {
 286   DBG("RIO: Setting timeout %u", (uns) expires_delta);
 287   if (!expires_delta)
 288     timer_del(&rio->timer);
 289   else
 290     timer_add_rel(&rio->timer, expires_delta);
 291 }
 292
 293 uns
 294 rec_io_parse_line(struct main_rec_io *rio)
 295 {
 296   for (uns i = rio->read_prev_avail; i < rio->read_avail; i++)
 297     if (rio->read_rec_start[i] == '\n')
 298       return i+1;
 299   return 0;
 300 }
 301
 302 #ifdef TEST
 303
 304 static uns rhand(struct main_rec_io *rio)
 305 {
 306   uns r = rec_io_parse_line(rio);
 307   if (r)
 308     {
 309       rio->read_rec_start[r-1] = 0;
 310       printf("Read <%s>\n", rio->read_rec_start);
 311       rec_io_set_timeout(rio, 10000);
 312       rio->read_rec_start[r-1] = '\n';
 313       rec_io_write(rio, rio->read_rec_start, r);
 314     }
 315   return r;
 316 }
 317
 318 static int ehand(struct main_rec_io *rio, int cause)
 319 {
 320   if (cause < 0)
 321     {
 322       msg(L_ERROR, "Error %d", cause);
 323       rec_io_del(rio);
 324       main_shut_down();
 325       return HOOK_IDLE;
 326     }
 327   else
 328     {
 329       msg(L_INFO, "Event %d", cause);
 330       return HOOK_RETRY;
 331     }
 332 }
 333
 334 int
 335 main(void)
 336 {
 337   log_init(NULL);
 338   main_init();
 339
 340   struct main_rec_io rio = {};
 341   rio.read_buf_size = 4;
 342   rio.read_handler = rhand;
 343   rio.notify_handler = ehand;
 344   // rio.read_rec_max = 40;
 345   rio.write_buf_size = 4;
 346   rec_io_add(&rio, 0);
 347   rec_io_start_read(&rio);
 348   rec_io_set_timeout(&rio, 10000);
 349
 350   main_debug();
 351
 352   main_loop();
 353   msg(L_INFO, "Finished.");
 354
 355   if (clist_is_linked(&rio.file.n))
 356     rec_io_del(&rio);
 357   main_cleanup();
 358   return 0;
 359 }
 360
 361 #endif