full aio implementation (without signals) (0d3b8e50) · Commits · hpc / gekkofs

src/client/gkfs_libc.cpp

+85 −18

Original line number	Diff line number	Diff line
		@@ -59,7 +59,7 @@ std::atomic<bool> initializing{false};
		*/
		// Define a debug macro, can be easily disabled

		#define GKFS_TRACE
		// #define GKFS_TRACE
		// #define PATH_SHORTCUT

		#ifdef GKFS_DEBUG_BUILD
		@@ -2264,19 +2264,43 @@ pwritev2(int fd, const struct iovec* iov, int iovcnt, off_t offset, int flags) {
		// tails.

		// */
		/* Add result to tracking list */
		typedef struct ResultEntry {
		struct aiocb* aiocbp;
		ssize_t result;
		struct ResultEntry* next;
		} ResultEntry;

		static pthread_mutex_t result_mutex = PTHREAD_MUTEX_INITIALIZER;
		static ResultEntry* results = NULL;

		static void
		add_result(struct aiocb* aiocbp, ssize_t res) {
		ResultEntry* entry = (ResultEntry*) malloc(sizeof(ResultEntry));
		entry->aiocbp = aiocbp;
		entry->result = res;
		entry->next = NULL;

		pthread_mutex_lock(&result_mutex);
		entry->next = results;
		results = entry;
		pthread_mutex_unlock(&result_mutex);
		}


		DLSYM_WRAPPER(int, aio_write, (struct aiocb * aiocbp), (aiocbp), "aio_write")
		int
		aio_write(struct aiocb* aiocbp) {
		initializeGekko();
		if(CTX->interception_enabled() && is_gkfs_fd(aiocbp->aio_fildes)) {
		gkfs::syscall::gkfs_write(aiocbp->aio_fildes,
		auto res = gkfs::syscall::gkfs_write(aiocbp->aio_fildes,
		(const void*) aiocbp->aio_buf,
		aiocbp->aio_nbytes);


		// TODO : SIGNALING IS WRONG*/
		// return 0;
		add_result(aiocbp, res);
		return 0;
		}

		GKFS_FALLBACK(aio_write, aiocbp)
		@@ -2287,23 +2311,66 @@ int
		aio_read(struct aiocb* aiocbp) {
		initializeGekko();
		if(CTX->interception_enabled() && is_gkfs_fd(aiocbp->aio_fildes)) {
		gkfs::syscall::gkfs_read(aiocbp->aio_fildes, (void*) aiocbp->aio_buf,
		auto res = gkfs::syscall::gkfs_read(aiocbp->aio_fildes,
		(void*) aiocbp->aio_buf,
		aiocbp->aio_nbytes);


		if(aiocbp->aio_sigevent.sigev_notify == SIGEV_SIGNAL) {
		// send signal of completation
		kill(getpid(), aiocbp->aio_sigevent.sigev_signo);
		// TODO : SIGNALING IS WRONG
		add_result(aiocbp, res);
		return 0;
		}
		// aiocbp->aio_nbytes = 0; // TODO : NOt overwrite the buffer
		GKFS_FALLBACK(aio_read, aiocbp)
		}

		/* Find and remove result from tracking list */
		static ssize_t
		get_result(struct aiocb* aiocbp) {
		pthread_mutex_lock(&result_mutex);
		ResultEntry** prev = &results;
		ResultEntry* current = results;
		ssize_t ret = -1;

		} else if(aiocbp->aio_sigevent.sigev_notify == SIGEV_THREAD) {
		aiocbp->aio_sigevent.sigev_notify_function(
		aiocbp->aio_sigevent.sigev_value);
		} else if(aiocbp->aio_sigevent.sigev_notify == SIGEV_NONE) {
		while(current) {
		if(current->aiocbp == aiocbp) {
		*prev = current->next;
		ret = current->result;
		free(current);
		break;
		}
		prev = &current->next;
		current = current->next;
		}

		// TODO : SIGNALING IS WRONG
		// return 0;
		pthread_mutex_unlock(&result_mutex);
		return ret;
		}
		aiocbp->aio_nbytes = 0; // TODO : NOt overwrite the buffer
		GKFS_FALLBACK(aio_read, aiocbp)

		DLSYM_WRAPPER(ssize_t, aio_return, (struct aiocb * aiocbp), (aiocbp),
		"aio_return")
		ssize_t
		aio_return(struct aiocb* aiocbp) {
		initializeGekko();
		if(CTX->interception_enabled() && is_gkfs_fd(aiocbp->aio_fildes)) {
		ssize_t ret = get_result(aiocbp);
		DEBUG_INFO("AIO_RETURN {} -> {}", aiocbp->aio_fildes, ret);

		if(ret == -1)
		errno = EINVAL;
		return ret;
		}
		GKFS_FALLBACK(aio_return, aiocbp)
		}
		DLSYM_WRAPPER(int, aio_error, (const struct aiocb* aiocbp), (aiocbp),
		"aio_error")
		int
		aio_error(const struct aiocb* aiocbp) {

		initializeGekko();
		if(CTX->interception_enabled() && is_gkfs_fd(aiocbp->aio_fildes)) {
		DEBUG_INFO("AIO_ERROR {}", aiocbp->aio_fildes);
		return 0;
		}
		GKFS_FALLBACK(aio_error, aiocbp)
		}