#include "lib/lib.h"
#include "lib/asio.h"
+#include "lib/threads.h"
#include <string.h>
#include <unistd.h>
static struct worker_pool asio_wpool;
static void
-asio_init(void)
+asio_init_unlocked(void)
{
if (asio_num_users++)
return;
DBG("ASIO: INIT");
asio_wpool.num_threads = 1;
- asio_wpool.stack_size = 65536;
worker_pool_init(&asio_wpool);
}
static void
-asio_cleanup(void)
+asio_cleanup_unlocked(void)
{
if (--asio_num_users)
return;
void
asio_init_queue(struct asio_queue *q)
{
- asio_init();
+ ucwlib_lock();
+ asio_init_unlocked();
+ ucwlib_unlock();
DBG("ASIO: New queue %p", q);
ASSERT(q->buffer_size);
q->allocated_requests = 0;
q->running_requests = 0;
q->running_writebacks = 0;
+ q->use_count = 0;
clist_init(&q->idle_list);
clist_init(&q->done_list);
work_queue_init(&asio_wpool, &q->queue);
}
work_queue_cleanup(&q->queue);
- asio_cleanup();
+
+ ucwlib_lock();
+ asio_cleanup_unlocked();
+ ucwlib_unlock();
}
struct asio_request *
r->len = 0;
r->status = -1;
r->returned_errno = -1;
+ r->submitted = 0;
return r;
}
struct asio_request *r = (struct asio_request *) work_wait(&q->queue);
if (!r)
return 0;
+ r->submitted = 0;
q->running_requests--;
if (r->op == ASIO_WRITE_BACK)
{
struct asio_queue *q = r->queue;
DBG("ASIO: Submitting %p on queue %p", r, q);
ASSERT(r->op != ASIO_FREE);
+ ASSERT(!r->submitted);
if (r->op == ASIO_WRITE_BACK)
{
while (q->running_writebacks >= q->max_writebacks)
q->running_writebacks++;
}
q->running_requests++;
+ r->submitted = 1;
r->work.go = asio_handler;
- r->work.returned = NULL;
work_submit(&q->queue, &r->work);
}
{
struct asio_queue *q = r->queue;
DBG("ASIO: Put %p", r);
+ ASSERT(!r->submitted);
ASSERT(q->allocated_requests);
clist_add_tail(&q->idle_list, &r->work.n);
q->allocated_requests--;