aiops.c

-

{
    aio_result_t *resultp = requestp->resultp;
    int cancelled = requestp->cancelled;

    /* Free allocated structures and copy data back to user space if the */
    /* request hasn't been cancelled */
    switch (requestp->request_type) {
    case _AIO_OP_STAT:
	if (!cancelled && requestp->ret == 0)
	    xmemcpy(requestp->statp, requestp->tmpstatp, sizeof(struct stat));
	xfree(requestp->tmpstatp);
    case _AIO_OP_OPEN:
	if (cancelled && requestp->ret >= 0)
	    /* The open() was cancelled but completed */
	    close(requestp->ret);
	xfree(requestp->path);
	break;
    case _AIO_OP_CLOSE:
	if (cancelled && requestp->ret < 0)
	    /* The close() was cancelled and never got executed */
	    close(requestp->fd);
	break;
    case _AIO_OP_UNLINK:
    case _AIO_OP_OPENDIR:
	xfree(requestp->path);
	break;
    case _AIO_OP_READ:
	if (!cancelled && requestp->ret > 0)
	    xmemcpy(requestp->bufferp, requestp->tmpbufp, requestp->ret);
    case _AIO_OP_WRITE:
	xfree(requestp->tmpbufp);
	break;
    default:
	break;
    }
    if (resultp != NULL && !cancelled) {
	resultp->aio_return = requestp->ret;
	resultp->aio_errno = requestp->err;
    }
    memPoolFree(aio_request_pool, requestp);
}				/* aio_cleanup_request */


int
aio_cancel(aio_result_t * resultp)
{
    aio_thread_t *threadp;
    aio_request_t *requestp;

    for (threadp = busy_threads_head; threadp != NULL; threadp = threadp->next)
	if (threadp->processed_req->resultp == resultp) {
	    threadp->processed_req->cancelled = 1;
	    threadp->processed_req->resultp = NULL;
	    return 0;
	}
    for (requestp = request_queue_head; requestp != NULL; requestp = requestp->next)
	if (requestp->resultp == resultp) {
	    requestp->cancelled = 1;
	    requestp->resultp = NULL;
	    return 0;
	}
    for (requestp = request_done_head; requestp != NULL; requestp = requestp->next)
	if (requestp->resultp == resultp) {
	    requestp->cancelled = 1;
	    requestp->resultp = NULL;
	    return 0;
	}
    return 1;
}				/* aio_cancel */


int
aio_open(const char *path, int oflag, mode_t mode, aio_result_t * resultp)
{
    aio_request_t *requestp;
    int len;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    len = strlen(path) + 1;
    if ((requestp->path = (char *) xmalloc(len)) == NULL) {
	memPoolFree(aio_request_pool, requestp);
	errno = ENOMEM;
	return -1;
    }
    strncpy(requestp->path, path, len);
    requestp->oflag = oflag;
    requestp->mode = mode;
    requestp->resultp = resultp;
    requestp->request_type = _AIO_OP_OPEN;
    requestp->cancelled = 0;

    aio_do_request(requestp);
    return 0;
}


static void
aio_do_open(aio_request_t * requestp)
{
    requestp->ret = open(requestp->path, requestp->oflag, requestp->mode);
    requestp->err = errno;
}


int
aio_read(int fd, char *bufp, int bufs, off_t offset, int whence, aio_result_t * resultp)
{
    aio_request_t *requestp;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    requestp->fd = fd;
    requestp->bufferp = bufp;
    if ((requestp->tmpbufp = (char *) xmalloc(bufs)) == NULL) {
	memPoolFree(aio_request_pool, requestp);
	errno = ENOMEM;
	return -1;
    }
    requestp->buflen = bufs;
    requestp->offset = offset;
    requestp->whence = whence;
    requestp->resultp = resultp;
    requestp->request_type = _AIO_OP_READ;
    requestp->cancelled = 0;

    aio_do_request(requestp);
    return 0;
}


static void
aio_do_read(aio_request_t * requestp)
{
    lseek(requestp->fd, requestp->offset, requestp->whence);
    requestp->ret = read(requestp->fd, requestp->tmpbufp, requestp->buflen);
    requestp->err = errno;
}


int
aio_write(int fd, char *bufp, int bufs, off_t offset, int whence, aio_result_t * resultp)
{
    aio_request_t *requestp;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    requestp->fd = fd;
    if ((requestp->tmpbufp = (char *) xmalloc(bufs)) == NULL) {
	memPoolFree(aio_request_pool, requestp);
	errno = ENOMEM;
	return -1;
    }
    xmemcpy(requestp->tmpbufp, bufp, bufs);
    requestp->buflen = bufs;
    requestp->offset = offset;
    requestp->whence = whence;
    requestp->resultp = resultp;
    requestp->request_type = _AIO_OP_WRITE;
    requestp->cancelled = 0;

    aio_do_request(requestp);
    return 0;
}


static void
aio_do_write(aio_request_t * requestp)
{
    requestp->ret = write(requestp->fd, requestp->tmpbufp, requestp->buflen);
    requestp->err = errno;
}


int
aio_close(int fd, aio_result_t * resultp)
{
    aio_request_t *requestp;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    requestp->fd = fd;
    requestp->resultp = resultp;
    requestp->request_type = _AIO_OP_CLOSE;
    requestp->cancelled = 0;

    aio_do_request(requestp);
    return 0;
}


static void
aio_do_close(aio_request_t * requestp)
{
    requestp->ret = close(requestp->fd);
    requestp->err = errno;
}


int
aio_stat(const char *path, struct stat *sb, aio_result_t * resultp)
{
    aio_request_t *requestp;
    int len;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    len = strlen(path) + 1;
    if ((requestp->path = (char *) xmalloc(len)) == NULL) {
	memPoolFree(aio_request_pool, requestp);
	errno = ENOMEM;
	return -1;
    }
    strncpy(requestp->path, path, len);
    requestp->statp = sb;
    if ((requestp->tmpstatp = (struct stat *) xmalloc(sizeof(struct stat))) == NULL) {
	xfree(requestp->path);
	memPoolFree(aio_request_pool, requestp);
	errno = ENOMEM;
	return -1;
    }
    requestp->resultp = resultp;
    requestp->request_type = _AIO_OP_STAT;
    requestp->cancelled = 0;

    aio_do_request(requestp);
    return 0;
}


static void
aio_do_stat(aio_request_t * requestp)
{
    requestp->ret = stat(requestp->path, requestp->tmpstatp);
    requestp->err = errno;
}


int
aio_unlink(const char *path, aio_result_t * resultp)
{
    aio_request_t *requestp;
    int len;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    len = strlen(path) + 1;
    if ((requestp->path = (char *) xmalloc(len)) == NULL) {
	memPoolFree(aio_request_pool, requestp);
	errno = ENOMEM;
	return -1;
    }
    strncpy(requestp->path, path, len);
    requestp->resultp = resultp;
    requestp->request_type = _AIO_OP_UNLINK;
    requestp->cancelled = 0;

    aio_do_request(requestp);
    return 0;
}


static void
aio_do_unlink(aio_request_t * requestp)
{
    requestp->ret = unlink(requestp->path);
    requestp->err = errno;
}


#if AIO_OPENDIR
/* XXX aio_opendir NOT implemented? */

int
aio_opendir(const char *path, aio_result_t * resultp)
{
    aio_request_t *requestp;
    int len;

    if (!aio_initialised)
	aio_init();
    if ((requestp = memPoolAlloc(aio_request_pool)) == NULL) {
	errno = ENOMEM;
	return -1;
    }
    return -1;
}

static void
aio_do_opendir(aio_request_t * requestp)
{
    /* NOT IMPLEMENTED */
}

#endif


void
aio_poll_threads(void)
{
    aio_thread_t *prev;
    aio_thread_t *threadp;
    aio_request_t *requestp;

    do {			/* while found completed thread */
	prev = NULL;
	threadp = busy_threads_head;
	while (threadp) {
	    debug(43, 5) ("%d: %d -> %d\n",
		threadp->thread,
		threadp->processed_req->request_type,
		threadp->status);
#if AIO_PROPER_MUTEX
	    if (threadp->req == NULL)
		if (pthread_mutex_trylock(&threadp->mutex) == 0)
		    break;
#else
	    if (threadp->req == NULL)
		break;
#endif
	    prev = threadp;
	    threadp = threadp->next;
	}
	if (threadp == NULL)
	    break;

	if (prev == NULL)
	    busy_threads_head = busy_threads_head->next;
	else
	    prev->next = threadp->next;

	if (busy_threads_tail == threadp)
	    busy_threads_tail = prev;

	requestp = threadp->processed_req;
	threadp->processed_req = NULL;

	threadp->next = wait_threads;
	wait_threads = threadp;

	if (request_done_tail != NULL)
	    request_done_tail->next = requestp;
	else
	    request_done_head = requestp;
	request_done_tail = requestp;
    } while (threadp);

    aio_process_request_queue();
}				/* aio_poll_threads */

aio_result_t *
aio_poll_done()
{
    aio_request_t *requestp;
    aio_result_t *resultp;
    int cancelled;

  AIO_REPOLL:
    aio_poll_threads();
    if (request_done_head == NULL) {
	return NULL;
    }
    requestp = request_done_head;
    request_done_head = requestp->next;
    if (!request_done_head)
	request_done_tail = NULL;

    resultp = requestp->resultp;
    cancelled = requestp->cancelled;
    aio_debug(requestp);
    debug(43, 5) ("DONE: %d -> %d\n", requestp->ret, requestp->err);
    aio_cleanup_request(requestp);
    if (cancelled)
	goto AIO_REPOLL;
    return resultp;
}				/* aio_poll_done */

int
aio_operations_pending(void)
{
    if (request_done_head)
	return 1;
    if (busy_threads_head)
	return 1;
    return 0;
}

static void
aio_debug(aio_request_t * requestp)
{
    switch (requestp->request_type) {
    case _AIO_OP_OPEN:
	debug(43, 5) ("OPEN of %s to FD %d\n", requestp->path, requestp->ret);
	break;
    case _AIO_OP_READ:
	debug(43, 5) ("READ on fd: %d\n", requestp->fd);
	break;
    case _AIO_OP_WRITE:
	debug(43, 5) ("WRITE on fd: %d\n", requestp->fd);
	break;
    case _AIO_OP_CLOSE:
	debug(43, 5) ("CLOSE of fd: %d\n", requestp->fd);
	break;
    case _AIO_OP_UNLINK:
	debug(43, 5) ("UNLINK of %s\n", requestp->path);
	break;
    default:
	break;
    }
}