Index: libcfa/src/concurrency/alarm.cfa =================================================================== --- libcfa/src/concurrency/alarm.cfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/alarm.cfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -176,4 +176,18 @@ } +//============================================================================================= +// Utilities +//============================================================================================= + +void sleep( Duration duration ) { + alarm_node_t node = { active_thread(), __kernel_get_time() + duration, 0`s }; + + register_self( &node ); + park( __cfaabi_dbg_ctx ); + + /* paranoid */ verify( !node.set ); + /* paranoid */ verify( node.next == 0p ); +} + // Local Variables: // // mode: c // Index: libcfa/src/concurrency/io.cfa =================================================================== --- libcfa/src/concurrency/io.cfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/io.cfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -45,5 +45,7 @@ memset(¶ms, 0, sizeof(params)); - int fd = syscall(__NR_io_uring_setup, entries_per_cluster(), ¶ms ); + uint32_t nentries = entries_per_cluster(); + + int fd = syscall(__NR_io_uring_setup, nentries, ¶ms ); if(fd < 0) { abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno)); @@ -52,10 +54,10 @@ // Step 2 : mmap result memset(&this.io, 0, sizeof(struct io_ring)); - struct io_uring_sq * sq = &this.io.submit_q; - struct io_uring_cq * cq = &this.io.completion_q; + struct io_uring_sq & sq = this.io.submit_q; + struct io_uring_cq & cq = this.io.completion_q; // calculate the right ring size - sq->ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned) ); - cq->ring_sz = params.cq_off.cqes + (params.cq_entries * sizeof(struct io_uring_cqe)); + sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned) ); + cq.ring_sz = params.cq_off.cqes + (params.cq_entries * sizeof(struct io_uring_cqe)); // Requires features @@ -66,6 +68,6 @@ // mmap the Submit Queue into existence - sq->ring_ptr = mmap(0, sq->ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQ_RING); - if (sq->ring_ptr == (void*)MAP_FAILED) { + sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQ_RING); + if (sq.ring_ptr == (void*)MAP_FAILED) { abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno)); } @@ -78,7 +80,7 @@ // else { // We need multiple call to MMAP - cq->ring_ptr = mmap(0, cq->ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_CQ_RING); - if (cq->ring_ptr == (void*)MAP_FAILED) { - munmap(sq->ring_ptr, sq->ring_sz); + cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_CQ_RING); + if (cq.ring_ptr == (void*)MAP_FAILED) { + munmap(sq.ring_ptr, sq.ring_sz); abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno)); } @@ -87,8 +89,8 @@ // mmap the submit queue entries size_t size = params.sq_entries * sizeof(struct io_uring_sqe); - sq->sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQES); - if (sq->sqes == (struct io_uring_sqe *)MAP_FAILED) { - munmap(sq->ring_ptr, sq->ring_sz); - if (cq->ring_ptr != sq->ring_ptr) munmap(cq->ring_ptr, cq->ring_sz); + sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQES); + if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) { + munmap(sq.ring_ptr, sq.ring_sz); + if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz); abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno)); } @@ -96,19 +98,31 @@ // Get the pointers from the kernel to fill the structure // submit queue - sq->head = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.head; - sq->tail = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.tail; - sq->mask = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.ring_mask; - sq->entries = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.ring_entries; - sq->flags = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.flags; - sq->dropped = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.dropped; - sq->array = (uint32_t *)((intptr_t)sq->ring_ptr) + params.sq_off.array; + sq.head = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head); + sq.tail = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail); + sq.mask = ( const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask); + sq.num = ( const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries); + sq.flags = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags); + sq.dropped = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped); + sq.array = ( uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array); + sq.alloc = *sq.tail; // completion queue - cq->head = (uint32_t *)((intptr_t)cq->ring_ptr) + params.cq_off.head; - cq->tail = (uint32_t *)((intptr_t)cq->ring_ptr) + params.cq_off.tail; - cq->mask = (uint32_t *)((intptr_t)cq->ring_ptr) + params.cq_off.ring_mask; - cq->entries = (struct io_uring_cqe *)((intptr_t)cq->ring_ptr) + params.cq_off.ring_entries; - cq->overflow = (uint32_t *)((intptr_t)cq->ring_ptr) + params.cq_off.overflow; - cq->cqes = (struct io_uring_cqe *)((intptr_t)cq->ring_ptr) + params.cq_off.cqes; + cq.head = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head); + cq.tail = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail); + cq.mask = ( const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask); + cq.num = ( const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries); + cq.overflow = ( uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow); + cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes); + + // some paranoid checks + /* paranoid */ verifyf( (*cq.mask) == ((*cq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*cq.num) - 1ul32, *cq.num, *cq.mask ); + /* paranoid */ verifyf( (*cq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num ); + /* paranoid */ verifyf( (*cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head ); + /* paranoid */ verifyf( (*cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail ); + + /* paranoid */ verifyf( (*sq.mask) == ((*sq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*sq.num) - 1ul32, *sq.num, *sq.mask ); + /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num ); + /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head ); + /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail ); // Update the global ring info @@ -116,4 +130,5 @@ this.io.fd = fd; this.io.done = false; + (this.io.submit){ min(*sq.num, *cq.num) }; // Create the poller thread @@ -137,5 +152,5 @@ // unmap the submit queue entries - munmap(sq.sqes, *sq.entries * sizeof(struct io_uring_sqe)); + munmap(sq.sqes, (*sq.num) * sizeof(struct io_uring_sqe)); // unmap the Submit Queue ring @@ -173,10 +188,15 @@ struct io_user_data * data = (struct io_user_data *)cqe.user_data; + __cfaabi_bits_print_safe( STDERR_FILENO, "Performed reading io cqe %p, result %d for %p\n", data, cqe.res, data->thrd ); + data->result = cqe.res; - unpark( data->thrd __cfaabi_dbg_ctx2 ); + __unpark( data->thrd __cfaabi_dbg_ctx2 ); + + // Allow new submissions to happen + V(ring.submit); // Mark to the kernel that the cqe has been seen // Ensure that the kernel only sees the new value of the head index after the CQEs have been read. - __atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELEASE ); + __atomic_fetch_add( ring.completion_q.head, 1, __ATOMIC_RELAXED ); return true; @@ -200,4 +220,5 @@ LOOP: while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) { int ret = syscall( __NR_io_uring_enter, ring.fd, 0, 1, IORING_ENTER_GETEVENTS, &mask, _NSIG / 8); + __cfaabi_bits_print_safe( STDERR_FILENO, "Performed io_wait, returned %d errmsg %s\n", ret, strerror(errno) ); if( ret < 0 ) { switch((int)errno) { @@ -221,6 +242,93 @@ //============================================================================================= -[* struct io_uring_sqe, uint32_t] io_submit_prelude( struct io_ring & ring ) { - return [0p, 0]; +// Submition steps : +// 1 - We need to make sure we don't overflow any of the buffer, P(ring.submit) to make sure +// entries are available. The semaphore make sure that there is no more operations in +// progress then the number of entries in the buffer. This probably limits concurrency +// more than necessary since submitted but not completed operations don't need any +// entries in user space. However, I don't know what happens if we overflow the buffers +// because too many requests completed at once. This is a safe approach in all cases. +// Furthermore, with hundreds of entries, this may be okay. +// +// 2 - Allocate a queue entry. The ring already has memory for all entries but only the ones +// listed in sq.array are visible by the kernel. For those not listed, the kernel does not +// offer any assurance that an entry is not being filled by multiple flags. Therefore, we +// need to write an allocator that allows allocating concurrently. +// +// 3 - Actually fill the submit entry, this is the only simple and straightforward step. +// +// 4 - Append the entry index to the array and adjust the tail accordingly. This operation +// needs to arrive to two concensus at the same time: +// A - The order in which entries are listed in the array: no two threads must pick the +// same index for their entries +// B - When can the tail be update for the kernel. EVERY entries in the array between +// head and tail must be fully filled and shouldn't ever be touched again. +// + +static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct io_ring & ring ) { + // Wait for a spot to be available + P(ring.submit); + + // Allocate the sqe + uint32_t idx = __atomic_fetch_add(&ring.submit_q.alloc, 1ul32, __ATOMIC_SEQ_CST); + + // Validate that we didn't overflow anything + // Check that nothing overflowed + /* paranoid */ verify( true ); + + // Check that it goes head -> tail -> alloc and never head -> alloc -> tail + /* paranoid */ verify( true ); + + // Return the sqe + return [&ring.submit_q.sqes[ idx & (*ring.submit_q.mask)], idx]; +} + +static inline void __submit( struct io_ring & ring, uint32_t idx ) { + // get mutual exclusion + lock(ring.submit_q.lock __cfaabi_dbg_ctx2); + + // Append to the list of ready entries + uint32_t * tail = ring.submit_q.tail; + const uint32_t mask = *ring.submit_q.mask; + + ring.submit_q.array[ (*tail) & mask ] = idx & mask; + __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST); + + // Submit however, many entries need to be submitted + int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0); + __cfaabi_bits_print_safe( STDERR_FILENO, "Performed io_submit, returned %d\n", ret ); + if( ret < 0 ) { + switch((int)errno) { + default: + abort( "KERNEL ERROR: IO_URING SUBMIT - %s\n", strerror(errno) ); + } + } + + unlock(ring.submit_q.lock); + // Make sure that idx was submitted + // Be careful to not get false positive if we cycled the entire list or that someone else submitted for us +} + +static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) { + this.opcode = opcode; + #if !defined(IOSQE_ASYNC) + this.flags = 0; + #else + this.flags = IOSQE_ASYNC; + #endif + this.ioprio = 0; + this.fd = fd; + this.off = 0; + this.addr = 0; + this.len = 0; + this.rw_flags = 0; + this.__pad2[0] = this.__pad2[1] = this.__pad2[2] = 0; +} + +static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) { + (this){ opcode, fd }; + this.off = off; + this.addr = (uint64_t)addr; + this.len = len; } @@ -228,25 +336,38 @@ // I/O Interface //============================================================================================= - -/* extern "C" { #define __USE_GNU + #define _GNU_SOURCE + #include #include - } - + #include + #include + } + + #define __submit_prelude \ + struct io_ring & ring = active_cluster()->io; \ + struct io_uring_sqe * sqe; \ + uint32_t idx; \ + [sqe, idx] = __submit_alloc( ring ); + + #define __submit_wait \ + io_user_data data = { 0, active_thread() }; \ + __cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd ); \ + sqe->user_data = (uint64_t)&data; \ + __submit( ring, idx ); \ + park( __cfaabi_dbg_ctx ); \ + return data.result; + +//----------------------------------------------------------------------------- +// Asynchronous operations ssize_t async_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 - sqe->opcode = IORING_OP_READV; - sqe->flags = 0; - sqe->ioprio = 0; - sqe->fd = fd; - sqe->off = offset; - sqe->addr = (unsigned long) iov; - sqe->len = iovcnt; - sqe->rw_flags = 0; - sqe->user_data = 0; - sqe->__pad2[0] = sqe->__pad2[1] = sqe->__pad2[2] = 0; + __submit_prelude + + (*sqe){ IORING_OP_READV, fd, iov, iovcnt, offset }; + + __submit_wait #endif } @@ -256,22 +377,345 @@ return pwritev2(fd, iov, iovcnt, offset, flags); #else - #warning not implemented - #endif - } - - bool is_async( void (*func)() ) { - switch((uintptr_t)func) { - case (uintptr_t)preadv2: - case (uintptr_t)async_preadv2: - #if defined(IORING_OP_READV) - return true; - #else - return false; - #endif - default: - return false; - } - } -*/ + __submit_prelude + + (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset }; + + __submit_wait + #endif + } + + int async_fsync(int fd) { + #if !defined(IORING_OP_FSYNC) + return fsync(fd); + #else + __submit_prelude + + (*sqe){ IORING_OP_FSYNC, fd }; + + __submit_wait + #endif + } + + int async_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 async_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 async_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 async_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 async_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 async_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 async_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 async_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 async_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 async_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 async_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 async_close(int fd) { + #if !defined(IORING_OP_CLOSE) + return close( fd ); + #else + __submit_prelude + + (*sqe){ IORING_OP_CLOSE, fd }; + + __submit_wait + #endif + } + + int async_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 ); + #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 async_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 async_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 + } + +//----------------------------------------------------------------------------- +// Check if a function is asynchronous + +// 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 ) { + + if( /*func == (fptr_t)preadv2 || */ + func == (fptr_t)async_preadv2 ) + #define _CFA_IO_FEATURE_IORING_OP_READV , + return IS_DEFINED(IORING_OP_READV); + + if( /*func == (fptr_t)pwritev2 || */ + func == (fptr_t)async_pwritev2 ) + #define _CFA_IO_FEATURE_IORING_OP_WRITEV , + return IS_DEFINED(IORING_OP_WRITEV); + + if( /*func == (fptr_t)fsync || */ + func == (fptr_t)async_fsync ) + #define _CFA_IO_FEATURE_IORING_OP_FSYNC , + return IS_DEFINED(IORING_OP_FSYNC); + + if( /*func == (fptr_t)ync_file_range || */ + func == (fptr_t)async_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)async_sendmsg ) + #define _CFA_IO_FEATURE_IORING_OP_SENDMSG , + return IS_DEFINED(IORING_OP_SENDMSG); + + if( /*func == (fptr_t)recvmsg || */ + func == (fptr_t)async_recvmsg ) + #define _CFA_IO_FEATURE_IORING_OP_RECVMSG , + return IS_DEFINED(IORING_OP_RECVMSG); + + if( /*func == (fptr_t)send || */ + func == (fptr_t)async_send ) + #define _CFA_IO_FEATURE_IORING_OP_SEND , + return IS_DEFINED(IORING_OP_SEND); + + if( /*func == (fptr_t)recv || */ + func == (fptr_t)async_recv ) + #define _CFA_IO_FEATURE_IORING_OP_RECV , + return IS_DEFINED(IORING_OP_RECV); + + if( /*func == (fptr_t)accept4 || */ + func == (fptr_t)async_accept4 ) + #define _CFA_IO_FEATURE_IORING_OP_ACCEPT , + return IS_DEFINED(IORING_OP_ACCEPT); + + if( /*func == (fptr_t)connect || */ + func == (fptr_t)async_connect ) + #define _CFA_IO_FEATURE_IORING_OP_CONNECT , + return IS_DEFINED(IORING_OP_CONNECT); + + if( /*func == (fptr_t)fallocate || */ + func == (fptr_t)async_fallocate ) + #define _CFA_IO_FEATURE_IORING_OP_FALLOCATE , + return IS_DEFINED(IORING_OP_FALLOCATE); + + if( /*func == (fptr_t)fadvise || */ + func == (fptr_t)async_fadvise ) + #define _CFA_IO_FEATURE_IORING_OP_FADVISE , + return IS_DEFINED(IORING_OP_FADVISE); + + if( /*func == (fptr_t)madvise || */ + func == (fptr_t)async_madvise ) + #define _CFA_IO_FEATURE_IORING_OP_MADVISE , + return IS_DEFINED(IORING_OP_MADVISE); + + if( /*func == (fptr_t)openat || */ + func == (fptr_t)async_openat ) + #define _CFA_IO_FEATURE_IORING_OP_OPENAT , + return IS_DEFINED(IORING_OP_OPENAT); + + if( /*func == (fptr_t)close || */ + func == (fptr_t)async_close ) + #define _CFA_IO_FEATURE_IORING_OP_CLOSE , + return IS_DEFINED(IORING_OP_CLOSE); + + if( /*func == (fptr_t)statx || */ + func == (fptr_t)async_statx ) + #define _CFA_IO_FEATURE_IORING_OP_STATX , + return IS_DEFINED(IORING_OP_STATX); + + if( /*func == (fptr_t)read || */ + func == (fptr_t)async_read ) + #define _CFA_IO_FEATURE_IORING_OP_READ , + return IS_DEFINED(IORING_OP_READ); + + if( /*func == (fptr_t)write || */ + func == (fptr_t)async_write ) + #define _CFA_IO_FEATURE_IORING_OP_WRITE , + return IS_DEFINED(IORING_OP_WRITE); + + return false; + } #endif Index: libcfa/src/concurrency/kernel.cfa =================================================================== --- libcfa/src/concurrency/kernel.cfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/kernel.cfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -615,8 +615,6 @@ } -void unpark( $thread * thrd __cfaabi_dbg_ctx_param2 ) { - if( !thrd ) return; - - disable_interrupts(); +// KERNEL ONLY unpark with out disabling interrupts +void __unpark( $thread * thrd __cfaabi_dbg_ctx_param2 ) { static_assert(sizeof(thrd->state) == sizeof(int)); @@ -647,4 +645,11 @@ abort(); } +} + +void unpark( $thread * thrd __cfaabi_dbg_ctx_param2 ) { + if( !thrd ) return; + + disable_interrupts(); + __unpark( thrd __cfaabi_dbg_ctx_fwd2 ); enable_interrupts( __cfaabi_dbg_ctx ); } Index: libcfa/src/concurrency/kernel.hfa =================================================================== --- libcfa/src/concurrency/kernel.hfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/kernel.hfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -115,30 +115,53 @@ #if defined(HAVE_LINUX_IO_URING_H) struct io_uring_sq { - uint32_t * head; - uint32_t * tail; - uint32_t * mask; - uint32_t * entries; - uint32_t * flags; - uint32_t * dropped; - uint32_t * array; - struct io_uring_sqe * sqes; - - uint32_t sqe_head; - uint32_t sqe_tail; - - size_t ring_sz; - void * ring_ptr; -}; - -struct io_uring_cq { + // Head and tail of the ring (associated with array) volatile uint32_t * head; volatile uint32_t * tail; - uint32_t * mask; - struct io_uring_cqe * entries; + + // The actual kernel ring which uses head/tail + // indexes into the sqes arrays + uint32_t * array; + + // number of entries and mask to go with it + const uint32_t * num; + const uint32_t * mask; + + // Submission flags (Not sure what for) + uint32_t * flags; + + // number of sqes not submitted (whatever that means) + uint32_t * dropped; + + // Like head/tail but not seen by the kernel + volatile uint32_t alloc; + + __spinlock_t lock; + + // A buffer of sqes (not the actual ring) + struct io_uring_sqe * sqes; + + // The location and size of the mmaped area + void * ring_ptr; + size_t ring_sz; +}; + +struct io_uring_cq { + // Head and tail of the ring + volatile uint32_t * head; + volatile uint32_t * tail; + + // number of entries and mask to go with it + const uint32_t * mask; + const uint32_t * num; + + // number of cqes not submitted (whatever that means) uint32_t * overflow; + + // the kernel ring struct io_uring_cqe * cqes; + // The location and size of the mmaped area + void * ring_ptr; size_t ring_sz; - void * ring_ptr; }; @@ -151,4 +174,5 @@ void * stack; volatile bool done; + semaphore submit; }; #endif Index: libcfa/src/concurrency/kernel_private.hfa =================================================================== --- libcfa/src/concurrency/kernel_private.hfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/kernel_private.hfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -70,4 +70,7 @@ ) +// KERNEL ONLY unpark with out disabling interrupts +void __unpark( $thread * thrd __cfaabi_dbg_ctx_param2 ); + //----------------------------------------------------------------------------- // I/O Index: libcfa/src/concurrency/preemption.cfa =================================================================== --- libcfa/src/concurrency/preemption.cfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/preemption.cfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -268,5 +268,5 @@ // reserved for future use static void timeout( $thread * this ) { - //TODO : implement waking threads + __unpark( this __cfaabi_dbg_ctx2 ); } Index: libcfa/src/concurrency/thread.hfa =================================================================== --- libcfa/src/concurrency/thread.hfa (revision 92976d95c1f6c8e40dc04d8a5960b056a5f6def8) +++ libcfa/src/concurrency/thread.hfa (revision 2d8f7b0a2facd4932b22d3d03d8138e922c42fe8) @@ -117,4 +117,8 @@ } +//---------- +// sleep: force thread to block and be rescheduled after Duration duration +void sleep( Duration duration ); + // Local Variables: // // mode: c //