Changes in libcfa/src/concurrency/io/call.cfa.in [44f09ea:ec19b21]

File:

• libcfa/src/concurrency/io/call.cfa.in (modified) (10 diffs)

libcfa/src/concurrency/io/call.cfa.in

@@ -54 +54 @@
                         | IOSQE_IO_DRAIN
                 #endif
+                #if defined(CFA_HAVE_IOSQE_ASYNC)
+                        | IOSQE_ASYNC
+                #endif
+        ;
+
+        static const __u32 LINK_FLAGS = 0
                 #if defined(CFA_HAVE_IOSQE_IO_LINK)
                         | IOSQE_IO_LINK
@@ -60 +66 @@
                         | IOSQE_IO_HARDLINK
                 #endif
-                #if defined(CFA_HAVE_IOSQE_ASYNC)
-                        | IOSQE_ASYNC
-                #endif
-                #if defined(CFA_HAVE_IOSQE_BUFFER_SELECTED)
-                        | IOSQE_BUFFER_SELECTED
-                #endif
         ;
 
@@ -74 +74 @@
         ;
 
-        extern struct $io_context * cfa_io_allocate(struct io_uring_sqe * out_sqes[], __u32 out_idxs[], __u32 want)  __attribute__((nonnull (1,2)));
-        extern void cfa_io_submit( struct $io_context * in_ctx, __u32 in_idxs[], __u32 have, bool lazy ) __attribute__((nonnull (1,2)));
+        extern [* volatile struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data );
+        extern void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1)));
+
+        static inline io_context * __get_io_context( void ) {
+                cluster * cltr = active_cluster();
+
+                /* paranoid */ verifyf( cltr, "No active cluster for io operation\\n");
+                assertf( cltr->io.cnt > 0, "Cluster %p has no default io contexts and no context was specified\\n", cltr );
+
+                /* paranoid */ verifyf( cltr->io.ctxs, "default io contexts for cluster %p are missing\\n", cltr);
+                return &cltr->io.ctxs[ thread_rand() % cltr->io.cnt ];
+        }
 #endif
 
@@ -88 +98 @@
 
 extern "C" {
-        #include <asm/types.h>
+        #include <sys/types.h>
         #include <sys/socket.h>
         #include <sys/syscall.h>
@@ -185 +195 @@
                 return ', '.join(args_a)
 
-AsyncTemplate = """inline void async_{name}(io_future_t & future, {params}, __u64 submit_flags) {{
+AsyncTemplate = """inline void async_{name}(io_future_t & future, {params}, int submit_flags, io_cancellation * cancellation, io_context * context) {{
         #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_{op})
                 ssize_t res = {name}({args});
@@ -195 +205 @@
                 }}
         #else
+                // we don't support LINK yet
+                if( 0 != (submit_flags & LINK_FLAGS) ) {{
+                        errno = ENOTSUP; return -1;
+                }}
+
+                if( !context ) {{
+                        context = __get_io_context();
+                }}
+                if(cancellation) {{
+                        cancellation->target = (__u64)(uintptr_t)&future;
+                }}
+
                 __u8 sflags = REGULAR_FLAGS & submit_flags;
+                struct __io_data & ring = *context->thrd.ring;
+
                 __u32 idx;
                 struct io_uring_sqe * sqe;
-                struct $io_context * ctx = cfa_io_allocate( &sqe, &idx, 1 );
+                [(volatile struct io_uring_sqe *) sqe, idx] = __submit_alloc( ring, (__u64)(uintptr_t)&future );
 
                 sqe->opcode = IORING_OP_{op};
-                sqe->user_data = (__u64)(uintptr_t)&future;
                 sqe->flags = sflags;
                 sqe->ioprio = 0;
@@ -216 +239 @@
 
                 verify( sqe->user_data == (__u64)(uintptr_t)&future );
-                cfa_io_submit( ctx, &idx, 1, 0 != (submit_flags & CFA_IO_LAZY) );
+                __submit( context, idx );
         #endif
 }}"""
 
-SyncTemplate = """{ret} cfa_{name}({params}, __u64 submit_flags) {{
+SyncTemplate = """{ret} cfa_{name}({params}, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {{
+        if( timeout >= 0 ) {{
+                errno = ENOTSUP;
+                return -1;
+        }}
         io_future_t future;
 
-        async_{name}( future, {args}, submit_flags );
+        async_{name}( future, {args}, submit_flags, cancellation, context );
 
         wait( future );
@@ -388 +415 @@
         if c.define:
                 print("""#if defined({define})
-        {ret} cfa_{name}({params}, __u64 submit_flags);
+        {ret} cfa_{name}({params}, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);
 #endif""".format(define=c.define,ret=c.ret, name=c.name, params=c.params))
         else:
-                print("{ret} cfa_{name}({params}, __u64 submit_flags);"
+                print("{ret} cfa_{name}({params}, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context);"
                 .format(ret=c.ret, name=c.name, params=c.params))
 
@@ -399 +426 @@
         if c.define:
                 print("""#if defined({define})
-        void async_{name}(io_future_t & future, {params}, __u64 submit_flags);
+        void async_{name}(io_future_t & future, {params}, int submit_flags, io_cancellation * cancellation, io_context * context);
 #endif""".format(define=c.define,name=c.name, params=c.params))
         else:
-                print("void async_{name}(io_future_t & future, {params}, __u64 submit_flags);"
+                print("void async_{name}(io_future_t & future, {params}, int submit_flags, io_cancellation * cancellation, io_context * context);"
                 .format(name=c.name, params=c.params))
 print("\n")
@@ -447 +474 @@
 
 print("""
+//-----------------------------------------------------------------------------
+bool cancel(io_cancellation & this) {
+        #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_ASYNC_CANCEL)
+                return false;
+        #else
+                io_future_t future;
+
+                io_context * context = __get_io_context();
+
+                __u8 sflags = 0;
+                struct __io_data & ring = *context->thrd.ring;
+
+                __u32 idx;
+                volatile struct io_uring_sqe * sqe;
+                [sqe, idx] = __submit_alloc( ring, (__u64)(uintptr_t)&future );
+
+                sqe->__pad2[0] = sqe->__pad2[1] = sqe->__pad2[2] = 0;
+                sqe->opcode = IORING_OP_ASYNC_CANCEL;
+                sqe->flags = sflags;
+                sqe->addr = this.target;
+
+                verify( sqe->user_data == (__u64)(uintptr_t)&future );
+                __submit( context, idx );
+
+                wait(future);
+
+                if( future.result == 0 ) return true; // Entry found
+                if( future.result == -EALREADY) return true; // Entry found but in progress
+                if( future.result == -ENOENT ) return false; // Entry not found
+                return false;
+        #endif
+}
+
 //-----------------------------------------------------------------------------
 // Check if a function is has asynchronous