lib/asio.c

   1 /*
   2  *      UCW Library -- Asynchronous I/O
   3  *
   4  *      (c) 2006 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 "lib/lib.h"
  13 #include "lib/asio.h"
  14
  15 #include <string.h>
  16 #include <unistd.h>
  17 #include <errno.h>
  18
  19 static uns asio_num_users;
  20 static struct worker_pool asio_wpool;
  21
  22 static void
  23 asio_init(void)
  24 {
  25   if (asio_num_users++)
  26     return;
  27
  28   DBG("ASIO: INIT");
  29   asio_wpool.num_threads = 1;
  30   asio_wpool.stack_size = 65536;
  31   worker_pool_init(&asio_wpool);
  32 }
  33
  34 static void
  35 asio_cleanup(void)
  36 {
  37   if (--asio_num_users)
  38     return;
  39
  40   DBG("ASIO: CLEANUP");
  41   worker_pool_cleanup(&asio_wpool);
  42 }
  43
  44 void
  45 asio_init_queue(struct asio_queue *q)
  46 {
  47   asio_init();
  48
  49   DBG("ASIO: New queue %p", q);
  50   ASSERT(q->buffer_size);
  51   q->allocated_requests = 0;
  52   q->running_requests = 0;
  53   q->running_writebacks = 0;
  54   clist_init(&q->idle_list);
  55   clist_init(&q->done_list);
  56   work_queue_init(&asio_wpool, &q->queue);
  57 }
  58
  59 void
  60 asio_cleanup_queue(struct asio_queue *q)
  61 {
  62   DBG("ASIO: Removing queue %p", q);
  63   ASSERT(!q->running_requests);
  64   ASSERT(!q->running_writebacks);
  65   ASSERT(!q->allocated_requests);
  66   ASSERT(clist_empty(&q->done_list));
  67
  68   struct asio_request *r;
  69   while (r = clist_head(&q->idle_list))
  70     {
  71       clist_remove(&r->work.n);
  72       big_free(r->buffer, q->buffer_size);
  73       xfree(r);
  74     }
  75
  76   work_queue_cleanup(&q->queue);
  77   asio_cleanup();
  78 }
  79
  80 struct asio_request *
  81 asio_get(struct asio_queue *q)
  82 {
  83   q->allocated_requests++;
  84   struct asio_request *r = clist_head(&q->idle_list);
  85   if (!r)
  86     {
  87       r = xmalloc_zero(sizeof(*r));
  88       r->queue = q;
  89       r->buffer = big_alloc(q->buffer_size);
  90       DBG("ASIO: Got %p (new)", r);
  91     }
  92   else
  93     {
  94       clist_remove(&r->work.n);
  95       DBG("ASIO: Got %p", r);
  96     }
  97   r->op = ASIO_FREE;
  98   r->fd = -1;
  99   r->len = 0;
 100   r->status = -1;
 101   r->returned_errno = -1;
 102   r->submitted = 0;
 103   return r;
 104 }
 105
 106 static int
 107 asio_raw_wait(struct asio_queue *q)
 108 {
 109   struct asio_request *r = (struct asio_request *) work_wait(&q->queue);
 110   if (!r)
 111     return 0;
 112   r->submitted = 0;
 113   q->running_requests--;
 114   if (r->op == ASIO_WRITE_BACK)
 115     {
 116       DBG("ASIO: Finished writeback %p", r);
 117       if (r->status < 0)
 118         die("Asynchronous write to fd %d failed: %s", r->fd, strerror(r->returned_errno));
 119       if (r->status != (int)r->len)
 120         die("Asynchronous write to fd %d wrote only %d bytes out of %d", r->fd, r->status, r->len);
 121       q->running_writebacks--;
 122       asio_put(r);
 123     }
 124   else
 125     clist_add_tail(&q->done_list, &r->work.n);
 126   return 1;
 127 }
 128
 129 static void
 130 asio_handler(struct worker_thread *t UNUSED, struct work *w)
 131 {
 132   struct asio_request *r = (struct asio_request *) w;
 133
 134   DBG("ASIO: Servicing %p (%s on fd=%d, len=%d)", r,
 135       (char*[]) { "?", "READ", "WRITE", "WRITEBACK" }[r->op], r->fd, r->len);
 136   errno = 0;
 137   switch (r->op)
 138     {
 139     case ASIO_READ:
 140       r->status = read(r->fd, r->buffer, r->len);
 141       break;
 142     case ASIO_WRITE:
 143     case ASIO_WRITE_BACK:
 144       r->status = write(r->fd, r->buffer, r->len);
 145       break;
 146     default:
 147       die("ASIO: Got unknown request type %d", r->op);
 148     }
 149   r->returned_errno = errno;
 150   DBG("ASIO: Finished %p (status=%d, errno=%d)", r, r->status, r->returned_errno);
 151 }
 152
 153 void
 154 asio_submit(struct asio_request *r)
 155 {
 156   struct asio_queue *q = r->queue;
 157   DBG("ASIO: Submitting %p on queue %p", r, q);
 158   ASSERT(r->op != ASIO_FREE);
 159   ASSERT(!r->submitted);
 160   if (r->op == ASIO_WRITE_BACK)
 161     {
 162       while (q->running_writebacks >= q->max_writebacks)
 163         {
 164           DBG("ASIO: Waiting for free writebacks");
 165           if (!asio_raw_wait(q))
 166             ASSERT(0);
 167         }
 168       q->running_writebacks++;
 169     }
 170   q->running_requests++;
 171   r->submitted = 1;
 172   r->work.go = asio_handler;
 173   r->work.returned = NULL;
 174   work_submit(&q->queue, &r->work);
 175 }
 176
 177 struct asio_request *
 178 asio_wait(struct asio_queue *q)
 179 {
 180   struct asio_request *r;
 181   while (!(r = clist_head(&q->done_list)))
 182     {
 183       DBG("ASIO: Waiting on queue %p", q);
 184       if (!asio_raw_wait(q))
 185         return NULL;
 186     }
 187   clist_remove(&r->work.n);
 188   DBG("ASIO: Done %p", r);
 189   return r;
 190 }
 191
 192 void
 193 asio_put(struct asio_request *r)
 194 {
 195   struct asio_queue *q = r->queue;
 196   DBG("ASIO: Put %p", r);
 197   ASSERT(!r->submitted);
 198   ASSERT(q->allocated_requests);
 199   clist_add_tail(&q->idle_list, &r->work.n);
 200   q->allocated_requests--;
 201 }
 202
 203 void
 204 asio_sync(struct asio_queue *q)
 205 {
 206   DBG("ASIO: Syncing queue %p", q);
 207   while (q->running_requests)
 208     if (!asio_raw_wait(q))
 209       ASSERT(0);
 210 }
 211
 212 #ifdef TEST
 213
 214 int main(void)
 215 {
 216   struct asio_queue q;
 217   struct asio_request *r;
 218
 219   q.buffer_size = 4096;
 220   q.max_writebacks = 2;
 221   asio_init_queue(&q);
 222
 223 #if 0
 224
 225   for (;;)
 226     {
 227       r = asio_get(&q);
 228       r->op = ASIO_READ;
 229       r->fd = 0;
 230       r->len = q.buffer_size;
 231       asio_submit(r);
 232       r = asio_wait(&q);
 233       ASSERT(r);
 234       if (r->status <= 0)
 235         {
 236           asio_put(r);
 237           break;
 238         }
 239       r->op = ASIO_WRITE_BACK;
 240       r->fd = 1;
 241       r->len = r->status;
 242       asio_submit(r);
 243     }
 244   asio_sync(&q);
 245
 246 #else
 247
 248   r = asio_get(&q);
 249   r->op = ASIO_READ;
 250   r->fd = 0;
 251   r->len = 1;
 252   asio_submit(r);
 253   r = asio_wait(&q);
 254   ASSERT(r);
 255   asio_put(r);
 256
 257   for (uns i=0; i<10; i++)
 258     {
 259       r = asio_get(&q);
 260       r->op = ASIO_WRITE_BACK;
 261       r->fd = 1;
 262       r->len = 1;
 263       r->buffer[0] = 'A' + i;
 264       asio_submit(r);
 265     }
 266   asio_sync(&q);
 267
 268   r = asio_get(&q);
 269   r->op = ASIO_WRITE;
 270   r->fd = 1;
 271   r->len = 1;
 272   r->buffer[0] = '\n';
 273   asio_submit(r);
 274   r = asio_wait(&q);
 275   ASSERT(r);
 276   asio_put(r);
 277
 278 #endif
 279
 280   asio_cleanup_queue(&q);
 281   return 0;
 282 }
 283
 284 #endif