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 static int rec_io_deferred_start_read(struct main_hook *ho);
  37
  38 void
  39 rec_io_add(struct main_rec_io *rio, int fd)
  40 {
  41   rio->file.fd = fd;
  42   file_add(&rio->file);
  43   rio->timer.handler = rec_io_timer_expired;
  44   rio->timer.data = rio;
  45   rio->start_read_hook.handler = rec_io_deferred_start_read;
  46   rio->start_read_hook.data = rio;
  47   clist_init(&rio->idle_write_buffers);
  48   clist_init(&rio->busy_write_buffers);
  49 }
  50
  51 void
  52 rec_io_del(struct main_rec_io *rio)
  53 {
  54   timer_del(&rio->timer);
  55   if (hook_is_active(&rio->start_read_hook))
  56     hook_del(&rio->start_read_hook);
  57   file_del(&rio->file);
  58
  59   if (rio->read_buf)
  60     {
  61       DBG("RIO: Freeing read buffer");
  62       xfree(rio->read_buf);
  63       rio->read_buf = NULL;
  64     }
  65
  66   struct rio_buffer *b;
  67   while ((b = clist_remove_head(&rio->idle_write_buffers)) || (b = clist_remove_head(&rio->busy_write_buffers)))
  68     {
  69       DBG("RIO: Freeing write buffer");
  70       xfree(b);
  71     }
  72 }
  73
  74 static int
  75 rec_io_process_read_buf(struct main_rec_io *rio)
  76 {
  77   uns got;
  78   while (rio->read_running && (got = rio->read_handler(rio)))
  79     {
  80       DBG("RIO READ: Ate %u bytes", got);
  81       if (got == ~0U)
  82         return HOOK_IDLE;
  83       rio->read_rec_start += got;
  84       rio->read_avail -= got;
  85       rio->read_prev_avail = 0;
  86       if (!rio->read_avail)
  87         {
  88           DBG("RIO READ: Resetting buffer");
  89           rio->read_rec_start = rio->read_buf;
  90           break;
  91         }
  92     }
  93   DBG("RIO READ: Want more");
  94   return (rio->read_running ? HOOK_RETRY : HOOK_IDLE);
  95 }
  96
  97 static int
  98 rec_io_read_handler(struct main_file *fi)
  99 {
 100   struct main_rec_io *rio = (struct main_rec_io *) fi;
 101
 102   if (rio->read_rec_max && rio->read_avail >= rio->read_rec_max)
 103     {
 104       rec_io_stop_read(rio);
 105       rio->notify_handler(rio, RIO_ERR_RECORD_TOO_LARGE);
 106       return HOOK_IDLE;
 107     }
 108
 109 restart: ;
 110   uns rec_start_pos = rio->read_rec_start - rio->read_buf;
 111   uns rec_end_pos = rec_start_pos + rio->read_avail;
 112   uns free_space = rio->read_buf_size - rec_end_pos;
 113   DBG("RIO READ: rec_start=%u avail=%u prev_avail=%u free=%u/%u",
 114     rec_start_pos, rio->read_avail, rio->read_prev_avail,
 115     free_space, rio->read_buf_size);
 116   if (free_space <= rio->read_buf_size/8)
 117     {
 118       if (rec_start_pos && rec_start_pos >= rio->read_buf_size/2)
 119         {
 120           // Moving the partial record to the start of the buffer
 121           DBG("RIO READ: Moving partial record to start");
 122           memmove(rio->read_buf, rio->read_rec_start, rio->read_avail);
 123           rio->read_rec_start = rio->read_buf;
 124         }
 125       else
 126         {
 127           DBG("RIO READ: Resizing buffer");
 128           rio->read_buf_size *= 2;
 129           rio->read_buf = xrealloc(rio->read_buf, rio->read_buf_size);
 130           rio->read_rec_start = rio->read_buf + rec_start_pos;
 131         }
 132       goto restart;
 133     }
 134
 135   int l = read(fi->fd, rio->read_buf + rec_end_pos, free_space);
 136   DBG("RIO READ: Read %d bytes", l);
 137   if (l < 0)
 138     {
 139       if (errno != EINTR && errno != EAGAIN)
 140         {
 141           DBG("RIO READ: Signalling error");
 142           rec_io_stop_read(rio);
 143           rio->notify_handler(rio, RIO_ERR_READ);
 144         }
 145       return HOOK_IDLE;
 146     }
 147   if (!l)
 148     {
 149       DBG("RIO READ: Signalling EOF");
 150       rec_io_stop_read(rio);
 151       rio->notify_handler(rio, RIO_EVENT_EOF);
 152       return HOOK_IDLE;
 153     }
 154   rio->read_prev_avail = rio->read_avail;
 155   rio->read_avail += l;
 156   DBG("RIO READ: Available: %u bytes", rio->read_avail);
 157
 158   return rec_io_process_read_buf(rio);
 159 }
 160
 161 static int
 162 rec_io_deferred_start_read(struct main_hook *ho)
 163 {
 164   struct main_rec_io *rio = ho->data;
 165
 166   DBG("RIO: Starting reading");
 167   if (!rio->read_buf)
 168     {
 169       if (!rio->read_buf_size)
 170         rio->read_buf_size = 256;
 171       rio->read_buf = xmalloc(rio->read_buf_size);
 172       DBG("RIO: Created read buffer (%u bytes)", rio->read_buf_size);
 173       rio->read_rec_start = rio->read_buf;
 174     }
 175
 176   rio->file.read_handler = rec_io_read_handler;
 177   file_chg(&rio->file);
 178   hook_del(ho);
 179   rio->read_running = 1;
 180
 181   rio->read_prev_avail = 0;
 182   return rec_io_process_read_buf(rio);
 183 }
 184
 185 static void
 186 rec_io_recalc_read(struct main_rec_io *rio)
 187 {
 188   uns flow = !rio->write_throttle_read || rio->write_watermark < rio->write_throttle_read;
 189   uns run = rio->read_started && flow;
 190   DBG("RIO: Recalc read (flow=%u, start=%u) -> %u", flow, rio->read_started, run);
 191   if (run != rio->read_running)
 192     {
 193       if (run)
 194         {
 195           /*
 196            * Since we need to rescan the read buffer for leftover records and we
 197            * can be deep in the call stack at this moment, we better defer most
 198            * of the work to a main_hook, which will be called in the next iteration
 199            * of the main loop.
 200            */
 201           if (!hook_is_active(&rio->start_read_hook))
 202             {
 203               DBG("RIO: Scheduling start of reading");
 204               hook_add(&rio->start_read_hook);
 205             }
 206         }
 207       else
 208         {
 209           if (hook_is_active(&rio->start_read_hook))
 210             {
 211               DBG("RIO: Descheduling start of reading");
 212               hook_del(&rio->start_read_hook);
 213             }
 214           rio->file.read_handler = NULL;
 215           file_chg(&rio->file);
 216           DBG("RIO: Reading stopped");
 217           rio->read_running = 0;
 218         }
 219     }
 220 }
 221
 222 void
 223 rec_io_start_read(struct main_rec_io *rio)
 224 {
 225   ASSERT(rec_io_is_active(rio));
 226   rio->read_started = 1;
 227   rec_io_recalc_read(rio);
 228 }
 229
 230 void
 231 rec_io_stop_read(struct main_rec_io *rio)
 232 {
 233   ASSERT(rec_io_is_active(rio));
 234   rio->read_started = 0;
 235   rec_io_recalc_read(rio);
 236 }
 237
 238 static void
 239 rec_io_stop_write(struct main_rec_io *rio)
 240 {
 241   DBG("RIO WRITE: Stopping write");
 242   ASSERT(!rio->write_watermark);
 243   rio->file.write_handler = NULL;
 244   file_chg(&rio->file);
 245 }
 246
 247 static int
 248 rec_io_write_handler(struct main_file *fi)
 249 {
 250   struct main_rec_io *rio = (struct main_rec_io *) fi;
 251   struct rio_buffer *b = clist_head(&rio->busy_write_buffers);
 252   if (!b)
 253     {
 254       rec_io_stop_write(rio);
 255       return HOOK_IDLE;
 256     }
 257
 258   int l = write(fi->fd, b->buf + b->written, b->full - b->written);
 259   DBG("RIO WRITE: Written %d bytes", l);
 260   if (l < 0)
 261     {
 262       if (errno != EINTR && errno != EAGAIN)
 263         {
 264           rec_io_stop_write(rio);
 265           rio->notify_handler(rio, RIO_ERR_WRITE);
 266         }
 267       return HOOK_IDLE;
 268     }
 269   b->written += l;
 270   if (b->written == b->full)
 271     {
 272       DBG("RIO WRITE: Written full buffer");
 273       clist_remove(&b->n);
 274       clist_add_tail(&rio->idle_write_buffers, &b->n);
 275     }
 276
 277   rio->write_watermark -= l;
 278   int ret = HOOK_RETRY;
 279   if (!rio->write_watermark)
 280     {
 281       ret = HOOK_IDLE;
 282       rec_io_stop_write(rio);
 283     }
 284   rec_io_recalc_read(rio);
 285
 286   // Call the hook, but carefully, because it can delete the RIO structure
 287   if (rio->notify_handler(rio, rio->write_watermark ? RIO_EVENT_PART_WRITTEN : RIO_EVENT_ALL_WRITTEN) == HOOK_IDLE)
 288     ret = HOOK_IDLE;
 289   return ret;
 290 }
 291
 292 static struct rio_buffer *
 293 rec_io_get_buffer(struct main_rec_io *rio)
 294 {
 295   struct rio_buffer *b = clist_remove_tail(&rio->idle_write_buffers);
 296   if (b)
 297     DBG("RIO WRITE: Recycled old buffer");
 298   else
 299     {
 300       if (!rio->write_buf_size)
 301         rio->write_buf_size = 1024;
 302       b = xmalloc(sizeof(struct rio_buffer) + rio->write_buf_size);
 303       DBG("RIO WRITE: Allocated new buffer");
 304     }
 305   b->full = b->written = 0;
 306   return b;
 307 }
 308
 309 void
 310 rec_io_write(struct main_rec_io *rio, void *data, uns len)
 311 {
 312   byte *bdata = data;
 313   ASSERT(rec_io_is_active(rio));
 314   if (!len)
 315     return;
 316
 317   while (len)
 318     {
 319       struct rio_buffer *b = clist_tail(&rio->busy_write_buffers);
 320       if (!b || b->full >= rio->write_buf_size)
 321         {
 322           b = rec_io_get_buffer(rio);
 323           clist_add_tail(&rio->busy_write_buffers, &b->n);
 324         }
 325       uns l = MIN(len, rio->write_buf_size - b->full);
 326       memcpy(b->buf + b->full, bdata, l);
 327       b->full += l;
 328       bdata += l;
 329       len -= l;
 330       rio->write_watermark += l;
 331       DBG("RIO WRITE: Buffered %u bytes of data (total %u)", l, rio->write_watermark);
 332       rec_io_recalc_read(rio);
 333     }
 334
 335   if (!rio->file.write_handler)
 336     {
 337       DBG("RIO WRITE: Starting write");
 338       rio->file.write_handler = rec_io_write_handler;
 339       file_chg(&rio->file);
 340     }
 341 }
 342
 343 void
 344 rec_io_set_timeout(struct main_rec_io *rio, timestamp_t expires_delta)
 345 {
 346   DBG("RIO: Setting timeout %u", (uns) expires_delta);
 347   if (!expires_delta)
 348     timer_del(&rio->timer);
 349   else
 350     timer_add_rel(&rio->timer, expires_delta);
 351 }
 352
 353 uns
 354 rec_io_parse_line(struct main_rec_io *rio)
 355 {
 356   for (uns i = rio->read_prev_avail; i < rio->read_avail; i++)
 357     if (rio->read_rec_start[i] == '\n')
 358       return i+1;
 359   return 0;
 360 }
 361
 362 #ifdef TEST
 363
 364 static uns rhand(struct main_rec_io *rio)
 365 {
 366   uns r = rec_io_parse_line(rio);
 367   if (r)
 368     {
 369       rio->read_rec_start[r-1] = 0;
 370       printf("Read <%s>\n", rio->read_rec_start);
 371       if (rio->read_rec_start[0] == '!')
 372         {
 373           rec_io_del(rio);
 374           main_shut_down();
 375           return ~0U;
 376         }
 377       rec_io_set_timeout(rio, 10000);
 378       rio->read_rec_start[r-1] = '\n';
 379       rec_io_write(rio, rio->read_rec_start, r);
 380     }
 381   return r;
 382 }
 383
 384 static int ehand(struct main_rec_io *rio, int cause)
 385 {
 386   if (cause < 0 || cause == RIO_EVENT_EOF)
 387     {
 388       msg(L_ERROR, "Error %d", cause);
 389       rec_io_del(rio);
 390       main_shut_down();
 391       return HOOK_IDLE;
 392     }
 393   else
 394     {
 395       msg(L_INFO, "Event %d", cause);
 396       return HOOK_RETRY;
 397     }
 398 }
 399
 400 int
 401 main(void)
 402 {
 403   log_init(NULL);
 404   main_init();
 405
 406   struct main_rec_io rio = {};
 407   rio.read_buf_size = 4;
 408   rio.read_handler = rhand;
 409   rio.notify_handler = ehand;
 410   // rio.read_rec_max = 40;
 411   rio.write_buf_size = 4;
 412   rio.write_throttle_read = 6;
 413   rec_io_add(&rio, 0);
 414   rec_io_start_read(&rio);
 415   rec_io_set_timeout(&rio, 10000);
 416
 417   main_debug();
 418
 419   main_loop();
 420   msg(L_INFO, "Finished.");
 421
 422   if (file_is_active(&rio.file))
 423     rec_io_del(&rio);
 424   main_cleanup();
 425   return 0;
 426 }
 427
 428 #endif