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   return r;
 103 }
 104
 105 static int
 106 asio_raw_wait(struct asio_queue *q)
 107 {
 108   struct asio_request *r = (struct asio_request *) work_wait(&q->queue);
 109   if (!r)
 110     return 0;
 111   q->running_requests--;
 112   if (r->op == ASIO_WRITE_BACK)
 113     {
 114       DBG("ASIO: Finished writeback %p", r);
 115       if (r->status < 0)
 116         die("Asynchronous write to fd %d failed: %s", r->fd, strerror(r->returned_errno));
 117       if (r->status != (int)r->len)
 118         die("Asynchronous write to fd %d wrote only %d bytes out of %d", r->fd, r->status, r->len);
 119       q->running_writebacks--;
 120       asio_put(r);
 121     }
 122   else
 123     clist_add_tail(&q->done_list, &r->work.n);
 124   return 1;
 125 }
 126
 127 static void
 128 asio_handler(struct worker_thread *t UNUSED, struct work *w)
 129 {
 130   struct asio_request *r = (struct asio_request *) w;
 131
 132   DBG("ASIO: Servicing %p (%s on fd=%d, len=%d)", r,
 133       (char*[]) { "?", "READ", "WRITE", "WRITEBACK" }[r->op], r->fd, r->len);
 134   errno = 0;
 135   switch (r->op)
 136     {
 137     case ASIO_READ:
 138       r->status = read(r->fd, r->buffer, r->len);
 139       break;
 140     case ASIO_WRITE:
 141     case ASIO_WRITE_BACK:
 142       r->status = write(r->fd, r->buffer, r->len);
 143       break;
 144     default:
 145       die("ASIO: Got unknown request type %d", r->op);
 146     }
 147   r->returned_errno = errno;
 148   DBG("ASIO: Finished %p (status=%d, errno=%d)", r, r->status, r->returned_errno);
 149 }
 150
 151 void
 152 asio_submit(struct asio_request *r)
 153 {
 154   struct asio_queue *q = r->queue;
 155   DBG("ASIO: Submitting %p on queue %p", r, q);
 156   ASSERT(r->op != ASIO_FREE);
 157   if (r->op == ASIO_WRITE_BACK)
 158     {
 159       while (q->running_writebacks >= q->max_writebacks)
 160         {
 161           DBG("ASIO: Waiting for free writebacks");
 162           if (!asio_raw_wait(q))
 163             ASSERT(0);
 164         }
 165       q->running_writebacks++;
 166     }
 167   q->running_requests++;
 168   r->work.go = asio_handler;
 169   r->work.returned = NULL;
 170   work_submit(&q->queue, &r->work);
 171 }
 172
 173 struct asio_request *
 174 asio_wait(struct asio_queue *q)
 175 {
 176   struct asio_request *r;
 177   while (!(r = clist_head(&q->done_list)))
 178     {
 179       DBG("ASIO: Waiting on queue %p", q);
 180       if (!asio_raw_wait(q))
 181         return NULL;
 182     }
 183   clist_remove(&r->work.n);
 184   DBG("ASIO: Done %p", r);
 185   return r;
 186 }
 187
 188 void
 189 asio_put(struct asio_request *r)
 190 {
 191   struct asio_queue *q = r->queue;
 192   DBG("ASIO: Put %p", r);
 193   ASSERT(q->allocated_requests);
 194   clist_add_tail(&q->idle_list, &r->work.n);
 195   q->allocated_requests--;
 196 }
 197
 198 void
 199 asio_sync(struct asio_queue *q)
 200 {
 201   DBG("ASIO: Syncing queue %p", q);
 202   while (q->running_requests)
 203     if (!asio_raw_wait(q))
 204       ASSERT(0);
 205 }
 206
 207 #ifdef TEST
 208
 209 int main(void)
 210 {
 211   struct asio_queue q;
 212   struct asio_request *r;
 213
 214   q.buffer_size = 4096;
 215   q.max_writebacks = 2;
 216   asio_init_queue(&q);
 217
 218 #if 0
 219
 220   for (;;)
 221     {
 222       r = asio_get(&q);
 223       r->op = ASIO_READ;
 224       r->fd = 0;
 225       r->len = q.buffer_size;
 226       asio_submit(r);
 227       r = asio_wait(&q);
 228       ASSERT(r);
 229       if (r->status <= 0)
 230         {
 231           asio_put(r);
 232           break;
 233         }
 234       r->op = ASIO_WRITE_BACK;
 235       r->fd = 1;
 236       r->len = r->status;
 237       asio_submit(r);
 238     }
 239   asio_sync(&q);
 240
 241 #else
 242
 243   r = asio_get(&q);
 244   r->op = ASIO_READ;
 245   r->fd = 0;
 246   r->len = 1;
 247   asio_submit(r);
 248   r = asio_wait(&q);
 249   ASSERT(r);
 250   asio_put(r);
 251
 252   for (uns i=0; i<10; i++)
 253     {
 254       r = asio_get(&q);
 255       r->op = ASIO_WRITE_BACK;
 256       r->fd = 1;
 257       r->len = 1;
 258       r->buffer[0] = 'A' + i;
 259       asio_submit(r);
 260     }
 261   asio_sync(&q);
 262
 263   r = asio_get(&q);
 264   r->op = ASIO_WRITE;
 265   r->fd = 1;
 266   r->len = 1;
 267   r->buffer[0] = '\n';
 268   asio_submit(r);
 269   r = asio_wait(&q);
 270   ASSERT(r);
 271   asio_put(r);
 272
 273 #endif
 274
 275   asio_cleanup_queue(&q);
 276   return 0;
 277 }
 278
 279 #endif