Index: libcfa/src/concurrency/io.cfa =================================================================== --- libcfa/src/concurrency/io.cfa (revision 2489d3171d9cea7ac76e8425d2473a9497537b10) +++ libcfa/src/concurrency/io.cfa (revision ecf6b466cf33910b43c4c1e9b980e00690da2140) @@ -1,2 +1,17 @@ +// +// Cforall Version 1.0.0 Copyright (C) 2020 University of Waterloo +// +// The contents of this file are covered under the licence agreement in the +// file "LICENCE" distributed with Cforall. +// +// io.cfa -- +// +// Author : Thierry Delisle +// Created On : Thu Apr 23 17:31:00 2020 +// Last Modified By : +// Last Modified On : +// Update Count : +// + #include "kernel.hfa" @@ -8,8 +23,4 @@ void __kernel_io_shutdown( cluster & this ) { // Nothing to do without io_uring - } - - bool is_async( void (*)() ) { - return false; } @@ -359,17 +370,19 @@ this.len = len; } +#endif //============================================================================================= // I/O Interface //============================================================================================= - extern "C" { - #define __USE_GNU - #define _GNU_SOURCE - #include - #include - #include - #include - } - +extern "C" { + #define __USE_GNU + #define _GNU_SOURCE + #include + #include + #include + #include +} + +#if defined(HAVE_LINUX_IO_URING_H) #define __submit_prelude \ struct io_ring & ring = active_cluster()->io; \ @@ -385,261 +398,262 @@ park( __cfaabi_dbg_ctx ); \ return data.result; +#endif //----------------------------------------------------------------------------- // Asynchronous operations - ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) { - #if !defined(IORING_OP_READV) - return preadv2(fd, iov, iovcnt, offset, flags); - #else - __submit_prelude - - (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset }; - - __submit_wait - #endif - } - - ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) { - #if !defined(IORING_OP_WRITEV) - return pwritev2(fd, iov, iovcnt, offset, flags); - #else - __submit_prelude - - (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset }; - - __submit_wait - #endif - } - - int cfa_fsync(int fd) { - #if !defined(IORING_OP_FSYNC) - return fsync(fd); - #else - __submit_prelude - - (*sqe){ IORING_OP_FSYNC, fd }; - - __submit_wait - #endif - } - - int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) { - #if !defined(IORING_OP_SYNC_FILE_RANGE) - return sync_file_range(fd, offset, nbytes, flags); - #else - __submit_prelude - - (*sqe){ IORING_OP_SYNC_FILE_RANGE, fd }; - sqe->off = offset; - sqe->len = nbytes; - sqe->sync_range_flags = flags; - - __submit_wait - #endif - } - - - ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags) { - #if !defined(IORING_OP_SENDMSG) - return recv(sockfd, msg, flags); - #else - __submit_prelude - - (*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 }; - sqe->msg_flags = flags; - - __submit_wait - #endif - } - - ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags) { - #if !defined(IORING_OP_RECVMSG) - return recv(sockfd, msg, flags); - #else - __submit_prelude - - (*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 }; - sqe->msg_flags = flags; - - __submit_wait - #endif - } - - ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags) { - #if !defined(IORING_OP_SEND) - return send( sockfd, buf, len, flags ); - #else - __submit_prelude - - (*sqe){ IORING_OP_SEND, sockfd }; - sqe->addr = (uint64_t)buf; - sqe->len = len; - sqe->msg_flags = flags; - - __submit_wait - #endif - } - - ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags) { - #if !defined(IORING_OP_RECV) - return recv( sockfd, buf, len, flags ); - #else - __submit_prelude - - (*sqe){ IORING_OP_RECV, sockfd }; - sqe->addr = (uint64_t)buf; - sqe->len = len; - sqe->msg_flags = flags; - - __submit_wait - #endif - } - - int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) { - #if !defined(IORING_OP_ACCEPT) - __SOCKADDR_ARG _addr; - _addr.__sockaddr__ = addr; - return accept4( sockfd, _addr, addrlen, flags ); - #else - __submit_prelude - - (*sqe){ IORING_OP_ACCEPT, sockfd }; - sqe->addr = addr; - sqe->addr2 = addrlen; - sqe->accept_flags = flags; - - __submit_wait - #endif - } - - int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) { - #if !defined(IORING_OP_CONNECT) - __CONST_SOCKADDR_ARG _addr; - _addr.__sockaddr__ = addr; - return connect( sockfd, _addr, addrlen ); - #else - __submit_prelude - - (*sqe){ IORING_OP_CONNECT, sockfd }; - sqe->addr = (uint64_t)addr; - sqe->off = addrlen; - - __submit_wait - #endif - } - - int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len) { - #if !defined(IORING_OP_FALLOCATE) - return fallocate( fd, mode, offset, len ); - #else - __submit_prelude - - (*sqe){ IORING_OP_FALLOCATE, fd }; - sqe->off = offset; - sqe->len = length; - sqe->mode = mode; - - __submit_wait - #endif - } - - int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice) { - #if !defined(IORING_OP_FADVISE) - return posix_fadvise( fd, offset, len, advice ); - #else - __submit_prelude - - (*sqe){ IORING_OP_FADVISE, fd }; - sqe->off = (uint64_t)offset; - sqe->len = length; - sqe->fadvise_advice = advice; - - __submit_wait - #endif - } - - int cfa_madvise(void *addr, size_t length, int advice) { - #if !defined(IORING_OP_MADVISE) - return madvise( addr, length, advice ); - #else - __submit_prelude - - (*sqe){ IORING_OP_MADVISE, 0 }; - sqe->addr = (uint64_t)addr; - sqe->len = length; - sqe->fadvise_advice = advice; - - __submit_wait - #endif - } - - int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode) { - #if !defined(IORING_OP_OPENAT) - return openat( dirfd, pathname, flags, mode ); - #else - __submit_prelude - - (*sqe){ IORING_OP_OPENAT, dirfd }; - sqe->addr = (uint64_t)pathname; - sqe->open_flags = flags; - sqe->mode = mode; - - __submit_wait - #endif - } - - int cfa_close(int fd) { - #if !defined(IORING_OP_CLOSE) - return close( fd ); - #else - __submit_prelude - - (*sqe){ IORING_OP_CLOSE, fd }; - - __submit_wait - #endif - } - - int cfa_statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf) { - #if !defined(IORING_OP_STATX) - //return statx( dirfd, pathname, flags, mask, statxbuf ); - return syscall( __NR_io_uring_setup, dirfd, pathname, flags, mask, statxbuf ); - #else - __submit_prelude - - (*sqe){ IORING_OP_STATX, dirfd }; - sqe->addr = (uint64_t)pathname; - sqe->statx_flags = flags; - sqe->len = mask; - sqe->off = (uint64_t)statxbuf; - - __submit_wait - #endif - } - - - ssize_t cfa_read(int fd, void *buf, size_t count) { - #if !defined(IORING_OP_READ) - return read( fd, buf, count ); - #else - __submit_prelude - - (*sqe){ IORING_OP_READ, fd, buf, count, 0 }; - - __submit_wait - #endif - } - - ssize_t cfa_write(int fd, void *buf, size_t count) { - #if !defined(IORING_OP_WRITE) - return read( fd, buf, count ); - #else - __submit_prelude - - (*sqe){ IORING_OP_WRITE, fd, buf, count, 0 }; - - __submit_wait - #endif - } +ssize_t cfa_preadv2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READV) + return preadv2(fd, iov, iovcnt, offset, flags); + #else + __submit_prelude + + (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset }; + + __submit_wait + #endif +} + +ssize_t cfa_pwritev2(int fd, const struct iovec *iov, int iovcnt, off_t offset, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITEV) + return pwritev2(fd, iov, iovcnt, offset, flags); + #else + __submit_prelude + + (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset }; + + __submit_wait + #endif +} + +int cfa_fsync(int fd) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FSYNC) + return fsync(fd); + #else + __submit_prelude + + (*sqe){ IORING_OP_FSYNC, fd }; + + __submit_wait + #endif +} + +int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SYNC_FILE_RANGE) + return sync_file_range(fd, offset, nbytes, flags); + #else + __submit_prelude + + (*sqe){ IORING_OP_SYNC_FILE_RANGE, fd }; + sqe->off = offset; + sqe->len = nbytes; + sqe->sync_range_flags = flags; + + __submit_wait + #endif +} + + +ssize_t cfa_sendmsg(int sockfd, const struct msghdr *msg, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SENDMSG) + return recv(sockfd, msg, flags); + #else + __submit_prelude + + (*sqe){ IORING_OP_SENDMSG, sockfd, msg, 1, 0 }; + sqe->msg_flags = flags; + + __submit_wait + #endif +} + +ssize_t cfa_recvmsg(int sockfd, struct msghdr *msg, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECVMSG) + return recv(sockfd, msg, flags); + #else + __submit_prelude + + (*sqe){ IORING_OP_RECVMSG, sockfd, msg, 1, 0 }; + sqe->msg_flags = flags; + + __submit_wait + #endif +} + +ssize_t cfa_send(int sockfd, const void *buf, size_t len, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_SEND) + return send( sockfd, buf, len, flags ); + #else + __submit_prelude + + (*sqe){ IORING_OP_SEND, sockfd }; + sqe->addr = (uint64_t)buf; + sqe->len = len; + sqe->msg_flags = flags; + + __submit_wait + #endif +} + +ssize_t cfa_recv(int sockfd, void *buf, size_t len, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_RECV) + return recv( sockfd, buf, len, flags ); + #else + __submit_prelude + + (*sqe){ IORING_OP_RECV, sockfd }; + sqe->addr = (uint64_t)buf; + sqe->len = len; + sqe->msg_flags = flags; + + __submit_wait + #endif +} + +int cfa_accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_ACCEPT) + __SOCKADDR_ARG _addr; + _addr.__sockaddr__ = addr; + return accept4( sockfd, _addr, addrlen, flags ); + #else + __submit_prelude + + (*sqe){ IORING_OP_ACCEPT, sockfd }; + sqe->addr = addr; + sqe->addr2 = addrlen; + sqe->accept_flags = flags; + + __submit_wait + #endif +} + +int cfa_connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CONNECT) + __CONST_SOCKADDR_ARG _addr; + _addr.__sockaddr__ = addr; + return connect( sockfd, _addr, addrlen ); + #else + __submit_prelude + + (*sqe){ IORING_OP_CONNECT, sockfd }; + sqe->addr = (uint64_t)addr; + sqe->off = addrlen; + + __submit_wait + #endif +} + +int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FALLOCATE) + return fallocate( fd, mode, offset, len ); + #else + __submit_prelude + + (*sqe){ IORING_OP_FALLOCATE, fd }; + sqe->off = offset; + sqe->len = length; + sqe->mode = mode; + + __submit_wait + #endif +} + +int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_FADVISE) + return posix_fadvise( fd, offset, len, advice ); + #else + __submit_prelude + + (*sqe){ IORING_OP_FADVISE, fd }; + sqe->off = (uint64_t)offset; + sqe->len = length; + sqe->fadvise_advice = advice; + + __submit_wait + #endif +} + +int cfa_madvise(void *addr, size_t length, int advice) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_MADVISE) + return madvise( addr, length, advice ); + #else + __submit_prelude + + (*sqe){ IORING_OP_MADVISE, 0 }; + sqe->addr = (uint64_t)addr; + sqe->len = length; + sqe->fadvise_advice = advice; + + __submit_wait + #endif +} + +int cfa_openat(int dirfd, const char *pathname, int flags, mode_t mode) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_OPENAT) + return openat( dirfd, pathname, flags, mode ); + #else + __submit_prelude + + (*sqe){ IORING_OP_OPENAT, dirfd }; + sqe->addr = (uint64_t)pathname; + sqe->open_flags = flags; + sqe->mode = mode; + + __submit_wait + #endif +} + +int cfa_close(int fd) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_CLOSE) + return close( fd ); + #else + __submit_prelude + + (*sqe){ IORING_OP_CLOSE, fd }; + + __submit_wait + #endif +} + +int cfa_statx(int dirfd, const char *pathname, int flags, unsigned int mask, struct statx *statxbuf) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_STATX) + //return statx( dirfd, pathname, flags, mask, statxbuf ); + return syscall( __NR_io_uring_setup, dirfd, pathname, flags, mask, statxbuf ); + #else + __submit_prelude + + (*sqe){ IORING_OP_STATX, dirfd }; + sqe->addr = (uint64_t)pathname; + sqe->statx_flags = flags; + sqe->len = mask; + sqe->off = (uint64_t)statxbuf; + + __submit_wait + #endif +} + + +ssize_t cfa_read(int fd, void *buf, size_t count) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_READ) + return read( fd, buf, count ); + #else + __submit_prelude + + (*sqe){ IORING_OP_READ, fd, buf, count, 0 }; + + __submit_wait + #endif +} + +ssize_t cfa_write(int fd, void *buf, size_t count) { + #if !defined(HAVE_LINUX_IO_URING_H) || !defined(IORING_OP_WRITE) + return read( fd, buf, count ); + #else + __submit_prelude + + (*sqe){ IORING_OP_WRITE, fd, buf, count, 0 }; + + __submit_wait + #endif +} //----------------------------------------------------------------------------- @@ -647,11 +661,11 @@ // Macro magic to reduce the size of the following switch case - #define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__) - #define IS_DEFINED_SECOND(first, second, ...) second - #define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion - #define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true) - - bool is_async( fptr_t func ) { - +#define IS_DEFINED_APPLY(f, ...) f(__VA_ARGS__) +#define IS_DEFINED_SECOND(first, second, ...) second +#define IS_DEFINED_TEST(expansion) _CFA_IO_FEATURE_##expansion +#define IS_DEFINED(macro) IS_DEFINED_APPLY( IS_DEFINED_SECOND,IS_DEFINED_TEST(macro) false, true) + +bool has_user_level_blocking( fptr_t func ) { + #if defined(HAVE_LINUX_IO_URING_H) if( /*func == (fptr_t)preadv2 || */ func == (fptr_t)cfa_preadv2 ) @@ -660,25 +674,25 @@ if( /*func == (fptr_t)pwritev2 || */ - func == (fptr_t)cfa_pwritev2 ) + func == (fptr_t)cfa_pwritev2 ) #define _CFA_IO_FEATURE_IORING_OP_WRITEV , return IS_DEFINED(IORING_OP_WRITEV); if( /*func == (fptr_t)fsync || */ - func == (fptr_t)cfa_fsync ) + func == (fptr_t)cfa_fsync ) #define _CFA_IO_FEATURE_IORING_OP_FSYNC , return IS_DEFINED(IORING_OP_FSYNC); if( /*func == (fptr_t)ync_file_range || */ - func == (fptr_t)cfa_sync_file_range ) + func == (fptr_t)cfa_sync_file_range ) #define _CFA_IO_FEATURE_IORING_OP_SYNC_FILE_RANGE , return IS_DEFINED(IORING_OP_SYNC_FILE_RANGE); if( /*func == (fptr_t)sendmsg || */ - func == (fptr_t)cfa_sendmsg ) + func == (fptr_t)cfa_sendmsg ) #define _CFA_IO_FEATURE_IORING_OP_SENDMSG , return IS_DEFINED(IORING_OP_SENDMSG); if( /*func == (fptr_t)recvmsg || */ - func == (fptr_t)cfa_recvmsg ) + func == (fptr_t)cfa_recvmsg ) #define _CFA_IO_FEATURE_IORING_OP_RECVMSG , return IS_DEFINED(IORING_OP_RECVMSG); @@ -735,15 +749,14 @@ if( /*func == (fptr_t)read || */ - func == (fptr_t)cfa_read ) + func == (fptr_t)cfa_read ) #define _CFA_IO_FEATURE_IORING_OP_READ , return IS_DEFINED(IORING_OP_READ); if( /*func == (fptr_t)write || */ - func == (fptr_t)cfa_write ) + func == (fptr_t)cfa_write ) #define _CFA_IO_FEATURE_IORING_OP_WRITE , return IS_DEFINED(IORING_OP_WRITE); - - return false; - } - -#endif + #endif + + return false; +}