2 * UCW Library -- Asynchronous I/O
4 * (c) 2006 Martin Mares <mj@ucw.cz>
6 * This software may be freely distributed and used according to the terms
7 * of the GNU Lesser General Public License.
19 static uns asio_num_users;
20 static struct worker_pool asio_wpool;
29 asio_wpool.num_threads = 1;
30 asio_wpool.stack_size = 65536;
31 worker_pool_init(&asio_wpool);
41 worker_pool_cleanup(&asio_wpool);
45 asio_init_queue(struct asio_queue *q)
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);
60 asio_cleanup_queue(struct asio_queue *q)
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));
68 struct asio_request *r;
69 while (r = clist_head(&q->idle_list))
71 clist_remove(&r->work.n);
72 big_free(r->buffer, q->buffer_size);
76 work_queue_cleanup(&q->queue);
81 asio_get(struct asio_queue *q)
83 q->allocated_requests++;
84 struct asio_request *r = clist_head(&q->idle_list);
87 r = xmalloc_zero(sizeof(*r));
89 r->buffer = big_alloc(q->buffer_size);
90 DBG("ASIO: Got %p (new)", r);
94 clist_remove(&r->work.n);
95 DBG("ASIO: Got %p", r);
101 r->returned_errno = -1;
107 asio_raw_wait(struct asio_queue *q)
109 struct asio_request *r = (struct asio_request *) work_wait(&q->queue);
113 q->running_requests--;
114 if (r->op == ASIO_WRITE_BACK)
116 DBG("ASIO: Finished writeback %p", r);
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--;
125 clist_add_tail(&q->done_list, &r->work.n);
130 asio_handler(struct worker_thread *t UNUSED, struct work *w)
132 struct asio_request *r = (struct asio_request *) w;
134 DBG("ASIO: Servicing %p (%s on fd=%d, len=%d)", r,
135 (char*[]) { "?", "READ", "WRITE", "WRITEBACK" }[r->op], r->fd, r->len);
140 r->status = read(r->fd, r->buffer, r->len);
143 case ASIO_WRITE_BACK:
144 r->status = write(r->fd, r->buffer, r->len);
147 die("ASIO: Got unknown request type %d", r->op);
149 r->returned_errno = errno;
150 DBG("ASIO: Finished %p (status=%d, errno=%d)", r, r->status, r->returned_errno);
154 asio_submit(struct asio_request *r)
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)
162 while (q->running_writebacks >= q->max_writebacks)
164 DBG("ASIO: Waiting for free writebacks");
165 if (!asio_raw_wait(q))
168 q->running_writebacks++;
170 q->running_requests++;
172 r->work.go = asio_handler;
173 r->work.returned = NULL;
174 work_submit(&q->queue, &r->work);
177 struct asio_request *
178 asio_wait(struct asio_queue *q)
180 struct asio_request *r;
181 while (!(r = clist_head(&q->done_list)))
183 DBG("ASIO: Waiting on queue %p", q);
184 if (!asio_raw_wait(q))
187 clist_remove(&r->work.n);
188 DBG("ASIO: Done %p", r);
193 asio_put(struct asio_request *r)
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--;
204 asio_sync(struct asio_queue *q)
206 DBG("ASIO: Syncing queue %p", q);
207 while (q->running_requests)
208 if (!asio_raw_wait(q))
217 struct asio_request *r;
219 q.buffer_size = 4096;
220 q.max_writebacks = 2;
230 r->len = q.buffer_size;
239 r->op = ASIO_WRITE_BACK;
257 for (uns i=0; i<10; i++)
260 r->op = ASIO_WRITE_BACK;
263 r->buffer[0] = 'A' + i;
280 asio_cleanup_queue(&q);