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;
106 asio_raw_wait(struct asio_queue *q)
108 struct asio_request *r = (struct asio_request *) work_wait(&q->queue);
111 q->running_requests--;
112 if (r->op == ASIO_WRITE_BACK)
114 DBG("ASIO: Finished writeback %p", r);
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--;
123 clist_add_tail(&q->done_list, &r->work.n);
128 asio_handler(struct worker_thread *t UNUSED, struct work *w)
130 struct asio_request *r = (struct asio_request *) w;
132 DBG("ASIO: Servicing %p (%s on fd=%d, len=%d)", r,
133 (char*[]) { "?", "READ", "WRITE", "WRITEBACK" }[r->op], r->fd, r->len);
138 r->status = read(r->fd, r->buffer, r->len);
141 case ASIO_WRITE_BACK:
142 r->status = write(r->fd, r->buffer, r->len);
145 die("ASIO: Got unknown request type %d", r->op);
147 r->returned_errno = errno;
148 DBG("ASIO: Finished %p (status=%d, errno=%d)", r, r->status, r->returned_errno);
152 asio_submit(struct asio_request *r)
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)
159 while (q->running_writebacks >= q->max_writebacks)
161 DBG("ASIO: Waiting for free writebacks");
162 if (!asio_raw_wait(q))
165 q->running_writebacks++;
167 q->running_requests++;
168 r->work.go = asio_handler;
169 r->work.returned = NULL;
170 work_submit(&q->queue, &r->work);
173 struct asio_request *
174 asio_wait(struct asio_queue *q)
176 struct asio_request *r;
177 while (!(r = clist_head(&q->done_list)))
179 DBG("ASIO: Waiting on queue %p", q);
180 if (!asio_raw_wait(q))
183 clist_remove(&r->work.n);
184 DBG("ASIO: Done %p", r);
189 asio_put(struct asio_request *r)
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--;
199 asio_sync(struct asio_queue *q)
201 DBG("ASIO: Syncing queue %p", q);
202 while (q->running_requests)
203 if (!asio_raw_wait(q))
212 struct asio_request *r;
214 q.buffer_size = 4096;
215 q.max_writebacks = 2;
225 r->len = q.buffer_size;
234 r->op = ASIO_WRITE_BACK;
252 for (uns i=0; i<10; i++)
255 r->op = ASIO_WRITE_BACK;
258 r->buffer[0] = 'A' + i;
275 asio_cleanup_queue(&q);