lib/async-append-aio.c

   1 /* Copyright (c) 2013 Nicira, Inc.
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <config.h>
  17
  18 /* This implementation of the async-append.h interface uses the POSIX
  19  * asynchronous I/O interface.  */
  20
  21 #include "async-append.h"
  22
  23 #include <aio.h>
  24 #include <errno.h>
  25 #include <stdlib.h>
  26 #include <unistd.h>
  27
  28 #include "byteq.h"
  29 #include "ovs-thread.h"
  30 #include "util.h"
  31
  32 /* Maximum number of bytes of buffered data. */
  33 enum { BUFFER_SIZE = 65536 };
  34
  35 /* Maximum number of aiocbs to use.
  36  *
  37  * aiocbs are big (144 bytes with glibc 2.11 on i386) so we try to allow for a
  38  * reasonable number by basing the number we allocate on the amount of buffer
  39  * space. */
  40 enum { MAX_CBS = ROUND_DOWN_POW2(BUFFER_SIZE / sizeof(struct aiocb)) };
  41 BUILD_ASSERT_DECL(IS_POW2(MAX_CBS));
  42
  43 struct async_append {
  44     int fd;
  45
  46     struct aiocb *aiocbs;
  47     unsigned int aiocb_head, aiocb_tail;
  48
  49     uint8_t *buffer;
  50     struct byteq byteq;
  51 };
  52
  53 struct async_append *
  54 async_append_create(int fd)
  55 {
  56     struct async_append *ap;
  57
  58     ap = xmalloc(sizeof *ap);
  59     ap->fd = fd;
  60     ap->aiocbs = xmalloc(MAX_CBS * sizeof *ap->aiocbs);
  61     ap->aiocb_head = ap->aiocb_tail = 0;
  62     ap->buffer = xmalloc(BUFFER_SIZE);
  63     byteq_init(&ap->byteq, ap->buffer, BUFFER_SIZE);
  64
  65     return ap;
  66 }
  67
  68 void
  69 async_append_destroy(struct async_append *ap)
  70 {
  71     if (ap) {
  72         async_append_flush(ap);
  73         free(ap->aiocbs);
  74         free(ap->buffer);
  75         free(ap);
  76     }
  77 }
  78
  79 static bool
  80 async_append_is_full(const struct async_append *ap)
  81 {
  82     return (ap->aiocb_head - ap->aiocb_tail >= MAX_CBS
  83             || byteq_is_full(&ap->byteq));
  84 }
  85
  86 static bool
  87 async_append_is_empty(const struct async_append *ap)
  88 {
  89     return byteq_is_empty(&ap->byteq);
  90 }
  91
  92 static void
  93 async_append_wait(struct async_append *ap)
  94 {
  95     int n = 0;
  96
  97     while (!async_append_is_empty(ap)) {
  98         struct aiocb *aiocb = &ap->aiocbs[ap->aiocb_tail & (MAX_CBS - 1)];
  99         int error = aio_error(aiocb);
 100
 101         if (error == EINPROGRESS) {
 102             const struct aiocb *p = aiocb;
 103             if (n > 0) {
 104                 return;
 105             }
 106             aio_suspend(&p, 1, NULL);
 107         } else {
 108             ignore(aio_return(aiocb));
 109             ap->aiocb_tail++;
 110             byteq_advance_tail(&ap->byteq, aiocb->aio_nbytes);
 111             n++;
 112         }
 113     }
 114 }
 115
 116 void
 117 async_append_write(struct async_append *ap, const void *data_, size_t size)
 118 {
 119     const uint8_t *data = data_;
 120
 121     while (size > 0) {
 122         struct aiocb *aiocb;
 123         size_t chunk_size;
 124         void *chunk;
 125
 126         while (async_append_is_full(ap)) {
 127             async_append_wait(ap);
 128         }
 129
 130         chunk = byteq_head(&ap->byteq);
 131         chunk_size = byteq_headroom(&ap->byteq);
 132         if (chunk_size > size) {
 133             chunk_size = size;
 134         }
 135         memcpy(chunk, data, chunk_size);
 136
 137         aiocb = &ap->aiocbs[ap->aiocb_head & (MAX_CBS - 1)];
 138         memset(aiocb, 0, sizeof *aiocb);
 139         aiocb->aio_fildes = ap->fd;
 140         aiocb->aio_offset = 0;
 141         aiocb->aio_buf = chunk;
 142         aiocb->aio_nbytes = chunk_size;
 143         aiocb->aio_sigevent.sigev_notify = SIGEV_NONE;
 144         if (aio_write(aiocb) == -1) {
 145             async_append_flush(ap);
 146             ignore(write(ap->fd, data, size));
 147             return;
 148         }
 149
 150         data += chunk_size;
 151         size -= chunk_size;
 152         byteq_advance_head(&ap->byteq, chunk_size);
 153         ap->aiocb_head++;
 154     }
 155 }
 156
 157 void
 158 async_append_flush(struct async_append *ap)
 159 {
 160     while (!async_append_is_empty(ap)) {
 161         async_append_wait(ap);
 162     }
 163 }