Changeset 4998155


Ignore:
Timestamp:
Aug 14, 2020, 12:33:26 PM (4 years ago)
Author:
Thierry Delisle <tdelisle@…>
Branches:
ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast, new-ast-unique-expr, pthread-emulation, qualifiedEnum
Children:
7fdae38
Parents:
b353a49
Message:

I/O now uses onshot semaphore instead of raw park/unpark.
I/O now uses linux/types.h types instead of stdint.h types

Location:
libcfa/src/concurrency
Files:
4 edited

Legend:

Unmodified
Added
Removed

  • libcfa/src/concurrency/io.cfa

    rb353a49 r4998155  
    9393//=============================================================================================
    9494        static unsigned __collect_submitions( struct __io_data & ring );
    95         static uint32_t __release_consumed_submission( struct __io_data & ring );
     95        static __u32 __release_consumed_submission( struct __io_data & ring );
    9696
    9797        static inline void process(struct io_uring_cqe & cqe ) {
    … …  
    100100
    101101                data->result = cqe.res;
    102                 unpark( data->thrd __cfaabi_dbg_ctx2 );
     102                post( data->sem );
    103103        }
    104104
    … …  
    136136                unsigned head = *ring.completion_q.head;
    137137                unsigned tail = *ring.completion_q.tail;
    138                 const uint32_t mask = *ring.completion_q.mask;
     138                const __u32 mask = *ring.completion_q.mask;
    139139
    140140                // Nothing was new return 0
    … …  
    143143                }
    144144
    145                 uint32_t count = tail - head;
     145                __u32 count = tail - head;
    146146                /* paranoid */ verify( count != 0 );
    147147                for(i; count) {
    … …  
    224224//
    225225
    226         [* struct io_uring_sqe, uint32_t] __submit_alloc( struct __io_data & ring, uint64_t data ) {
     226        [* struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data ) {
    227227                /* paranoid */ verify( data != 0 );
    228228
    … …  
    230230                __attribute((unused)) int len   = 0;
    231231                __attribute((unused)) int block = 0;
    232                 uint32_t cnt = *ring.submit_q.num;
    233                 uint32_t mask = *ring.submit_q.mask;
     232                __u32 cnt = *ring.submit_q.num;
     233                __u32 mask = *ring.submit_q.mask;
    234234
    235235                disable_interrupts();
    236                         uint32_t off = __tls_rand();
     236                        __u32 off = __tls_rand();
    237237                enable_interrupts( __cfaabi_dbg_ctx );
    238238
    … …  
    241241                        // Look through the list starting at some offset
    242242                        for(i; cnt) {
    243                                 uint64_t expected = 0;
    244                                 uint32_t idx = (i + off) & mask;
     243                                __u64 expected = 0;
     244                                __u32 idx = (i + off) & mask;
    245245                                struct io_uring_sqe * sqe = &ring.submit_q.sqes[idx];
    246                                 volatile uint64_t * udata = (volatile uint64_t *)&sqe->user_data;
     246                                volatile __u64 * udata = &sqe->user_data;
    247247
    248248                                if( *udata == expected &&
    … …  
    270270        }
    271271
    272         static inline uint32_t __submit_to_ready_array( struct __io_data & ring, uint32_t idx, const uint32_t mask ) {
     272        static inline __u32 __submit_to_ready_array( struct __io_data & ring, __u32 idx, const __u32 mask ) {
    273273                /* paranoid */ verify( idx <= mask   );
    274274                /* paranoid */ verify( idx != -1ul32 );
    … …  
    277277                __attribute((unused)) int len   = 0;
    278278                __attribute((unused)) int block = 0;
    279                 uint32_t ready_mask = ring.submit_q.ready_cnt - 1;
     279                __u32 ready_mask = ring.submit_q.ready_cnt - 1;
    280280
    281281                disable_interrupts();
    282                         uint32_t off = __tls_rand();
     282                        __u32 off = __tls_rand();
    283283                enable_interrupts( __cfaabi_dbg_ctx );
    284284
    285                 uint32_t picked;
     285                __u32 picked;
    286286                LOOKING: for() {
    287287                        for(i; ring.submit_q.ready_cnt) {
    288288                                picked = (i + off) & ready_mask;
    289                                 uint32_t expected = -1ul32;
     289                                __u32 expected = -1ul32;
    290290                                if( __atomic_compare_exchange_n( &ring.submit_q.ready[picked], &expected, idx, true, __ATOMIC_SEQ_CST, __ATOMIC_RELAXED ) ) {
    291291                                        break LOOKING;
    … …  
    316316        }
    317317
    318         void __submit( struct io_context * ctx, uint32_t idx ) __attribute__((nonnull (1))) {
     318        void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1))) {
    319319                __io_data & ring = *ctx->thrd.ring;
    320320                // Get now the data we definetely need
    321                 volatile uint32_t * const tail = ring.submit_q.tail;
    322                 const uint32_t mask  = *ring.submit_q.mask;
     321                volatile __u32 * const tail = ring.submit_q.tail;
     322                const __u32 mask  = *ring.submit_q.mask;
    323323
    324324                // There are 2 submission schemes, check which one we are using
    … …  
    332332                }
    333333                else if( ring.eager_submits ) {
    334                         uint32_t picked = __submit_to_ready_array( ring, idx, mask );
     334                        __u32 picked = __submit_to_ready_array( ring, idx, mask );
    335335
    336336                        for() {
    … …  
    429429
    430430                unsigned to_submit = 0;
    431                 uint32_t tail = *ring.submit_q.tail;
    432                 const uint32_t mask = *ring.submit_q.mask;
     431                __u32 tail = *ring.submit_q.tail;
     432                const __u32 mask = *ring.submit_q.mask;
    433433
    434434                // Go through the list of ready submissions
    435435                for( i; ring.submit_q.ready_cnt ) {
    436436                        // replace any submission with the sentinel, to consume it.
    437                         uint32_t idx = __atomic_exchange_n( &ring.submit_q.ready[i], -1ul32, __ATOMIC_RELAXED);
     437                        __u32 idx = __atomic_exchange_n( &ring.submit_q.ready[i], -1ul32, __ATOMIC_RELAXED);
    438438
    439439                        // If it was already the sentinel, then we are done
    … …  
    451451        }
    452452
    453         static uint32_t __release_consumed_submission( struct __io_data & ring ) {
    454                 const uint32_t smask = *ring.submit_q.mask;
     453        static __u32 __release_consumed_submission( struct __io_data & ring ) {
     454                const __u32 smask = *ring.submit_q.mask;
    455455
    456456                if( !try_lock(ring.submit_q.release_lock __cfaabi_dbg_ctx2) ) return 0;
    457                 uint32_t chead = *ring.submit_q.head;
    458                 uint32_t phead = ring.submit_q.prev_head;
     457                __u32 chead = *ring.submit_q.head;
     458                __u32 phead = ring.submit_q.prev_head;
    459459                ring.submit_q.prev_head = chead;
    460460                unlock(ring.submit_q.release_lock);
    461461
    462                 uint32_t count = chead - phead;
     462                __u32 count = chead - phead;
    463463                for( i; count ) {
    464                         uint32_t idx = ring.submit_q.array[ (phead + i) & smask ];
     464                        __u32 idx = ring.submit_q.array[ (phead + i) & smask ];
    465465                        ring.submit_q.sqes[ idx ].user_data = 0;
    466466                }

  • libcfa/src/concurrency/io/setup.cfa

    rb353a49 r4998155  
    298298                if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
    299299
    300                 uint32_t nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
     300                __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
    301301                if( !is_pow2(nentries) ) {
    302302                        abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
    … …  
    362362                // Get the pointers from the kernel to fill the structure
    363363                // submit queue
    364                 sq.head    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
    365                 sq.tail    = (volatile uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
    366                 sq.mask    = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
    367                 sq.num     = (   const uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
    368                 sq.flags   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
    369                 sq.dropped = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
    370                 sq.array   = (         uint32_t *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
     364                sq.head    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
     365                sq.tail    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
     366                sq.mask    = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
     367                sq.num     = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
     368                sq.flags   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
     369                sq.dropped = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
     370                sq.array   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
    371371                sq.prev_head = *sq.head;
    372372
    373373                {
    374                         const uint32_t num = *sq.num;
     374                        const __u32 num = *sq.num;
    375375                        for( i; num ) {
    376376                                sq.sqes[i].user_data = 0ul64;
    … …  
    395395
    396396                // completion queue
    397                 cq.head     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
    398                 cq.tail     = (volatile uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
    399                 cq.mask     = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
    400                 cq.num      = (   const uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
    401                 cq.overflow = (         uint32_t *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
    402                 cq.cqes   = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
     397                cq.head      = (volatile __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
     398                cq.tail      = (volatile __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
     399                cq.mask      = (   const __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
     400                cq.num       = (   const __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
     401                cq.overflow  = (         __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
     402                cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
    403403
    404404                // some paranoid checks
    … …  
    448448        void __ioctx_register($io_ctx_thread & ctx, struct epoll_event & ev) {
    449449                ev.events = EPOLLIN | EPOLLONESHOT;
    450                 ev.data.u64 = (uint64_t)&ctx;
     450                ev.data.u64 = (__u64)&ctx;
    451451                int ret = epoll_ctl(iopoll.epollfd, EPOLL_CTL_ADD, ctx.ring->fd, &ev);
    452452                if (ret < 0) {

  • libcfa/src/concurrency/io/types.hfa

    rb353a49 r4998155  
    1717
    1818#if defined(CFA_HAVE_LINUX_IO_URING_H)
     19        extern "C" {
     20                #include <linux/types.h>
     21        }
     22
    1923      #include "bits/locks.hfa"
    2024
    … …  
    2327      struct __submition_data {
    2428                // Head and tail of the ring (associated with array)
    25                 volatile uint32_t * head;
    26                 volatile uint32_t * tail;
    27                 volatile uint32_t prev_head;
     29                volatile __u32 * head;
     30                volatile __u32 * tail;
     31                volatile __u32 prev_head;
    2832
    2933                // The actual kernel ring which uses head/tail
    3034                // indexes into the sqes arrays
    31                 uint32_t * array;
     35                __u32 * array;
    3236
    3337                // number of entries and mask to go with it
    34                 const uint32_t * num;
    35                 const uint32_t * mask;
     38                const __u32 * num;
     39                const __u32 * mask;
    3640
    3741                // Submission flags (Not sure what for)
    38                 uint32_t * flags;
     42                __u32 * flags;
    3943
    4044                // number of sqes not submitted (whatever that means)
    41                 uint32_t * dropped;
     45                __u32 * dropped;
    4246
    4347                // Like head/tail but not seen by the kernel
    44                 volatile uint32_t * ready;
    45                 uint32_t ready_cnt;
     48                volatile __u32 * ready;
     49                __u32 ready_cnt;
    4650
    4751                __spinlock_t lock;
    … …  
    5862        struct __completion_data {
    5963                // Head and tail of the ring
    60                 volatile uint32_t * head;
    61                 volatile uint32_t * tail;
     64                volatile __u32 * head;
     65                volatile __u32 * tail;
    6266
    6367                // number of entries and mask to go with it
    64                 const uint32_t * mask;
    65                 const uint32_t * num;
     68                const __u32 * mask;
     69                const __u32 * num;
    6670
    6771                // number of cqes not submitted (whatever that means)
    68                 uint32_t * overflow;
     72                __u32 * overflow;
    6973
    7074                // the kernel ring
    … …  
    7983                struct __submition_data submit_q;
    8084                struct __completion_data completion_q;
    81                 uint32_t ring_flags;
     85                __u32 ring_flags;
    8286                int fd;
    8387                bool eager_submits:1;
    … …  
    8993        // IO user data
    9094        struct __io_user_data_t {
    91                 int32_t result;
    92                 $thread * thrd;
     95                __s32 result;
     96                oneshot sem;
    9397        };
    9498

  • libcfa/src/concurrency/iocall.cfa

    rb353a49 r4998155  
    3232        #include "io/types.hfa"
    3333
    34         extern [* struct io_uring_sqe, uint32_t] __submit_alloc( struct __io_data & ring, uint64_t data );
    35         extern void __submit( struct io_context * ctx, uint32_t idx ) __attribute__((nonnull (1)));
    36 
    37         static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd) {
     34        extern [* struct io_uring_sqe, __u32] __submit_alloc( struct __io_data & ring, __u64 data );
     35        extern void __submit( struct io_context * ctx, __u32 idx ) __attribute__((nonnull (1)));
     36
     37        static inline void ?{}(struct io_uring_sqe & this, __u8 opcode, int fd) {
    3838                this.opcode = opcode;
    3939                #if !defined(IOSQE_ASYNC)
    … …  
    5151        }
    5252
    53         static inline void ?{}(struct io_uring_sqe & this, uint8_t opcode, int fd, void * addr, uint32_t len, uint64_t off ) {
     53        static inline void ?{}(struct io_uring_sqe & this, __u8 opcode, int fd, void * addr, __u32 len, __u64 off ) {
    5454                (this){ opcode, fd };
    5555                this.off = off;
    56                 this.addr = (uint64_t)(uintptr_t)addr;
     56                this.addr = (__u64)(uintptr_t)addr;
    5757                this.len = len;
    5858        }
    … …  
    105105                (void)timeout; (void)cancellation; \
    106106                if( !context ) context = __get_io_context(); \
    107                 __io_user_data_t data = { 0, active_thread() }; \
     107                __io_user_data_t data = { 0 }; \
    108108                struct __io_data & ring = *context->thrd.ring; \
    109109                struct io_uring_sqe * sqe; \
    110                 uint32_t idx; \
    111                 uint8_t sflags = REGULAR_FLAGS & submit_flags; \
    112                 [sqe, idx] = __submit_alloc( ring, (uint64_t)(uintptr_t)&data ); \
     110                __u32 idx; \
     111                __u8 sflags = REGULAR_FLAGS & submit_flags; \
     112                [sqe, idx] = __submit_alloc( ring, (__u64)(uintptr_t)&data ); \
    113113                sqe->flags = sflags;
    114114
    115115        #define __submit_wait \
    116116                /*__cfaabi_bits_print_safe( STDERR_FILENO, "Preparing user data %p for %p\n", &data, data.thrd );*/ \
    117                 verify( sqe->user_data == (uint64_t)(uintptr_t)&data ); \
     117                verify( sqe->user_data == (__u64)(uintptr_t)&data ); \
    118118                __submit( context, idx ); \
    119                 park( __cfaabi_dbg_ctx ); \
     119                wait( data.sem ); \
    120120                if( data.result < 0 ) { \
    121121                        errno = -data.result; \
    … …  
    149149
    150150        extern int fsync(int fd);
    151         extern int sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags);
     151
     152        typedef __off64_t off_t;
     153        typedef __off64_t off64_t;
     154        extern int sync_file_range(int fd, off64_t offset, off64_t nbytes, unsigned int flags);
    152155
    153156        struct msghdr;
    … …  
    160163        extern int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
    161164
    162         extern int fallocate(int fd, int mode, uint64_t offset, uint64_t len);
    163         extern int posix_fadvise(int fd, uint64_t offset, uint64_t len, int advice);
     165        extern int fallocate(int fd, int mode, off_t offset, off_t len);
     166        extern int posix_fadvise(int fd, off_t offset, off_t len, int advice);
    164167        extern int madvise(void *addr, size_t length, int advice);
    165168
    … …  
    190193                        sqe->fd = fd;
    191194                        sqe->off = offset;
    192                         sqe->addr = (uint64_t)(uintptr_t)iov;
     195                        sqe->addr = (__u64)iov;
    193196                        sqe->len = iovcnt;
    194197                        sqe->rw_flags = 0;
    … …  
    207210                        __submit_prelude
    208211
    209                         (*sqe){ IORING_OP_WRITEV, fd, iov, iovcnt, offset };
     212                        sqe->opcode = IORING_OP_WRITEV;
     213                        sqe->ioprio = 0;
     214                        sqe->fd = fd;
     215                        sqe->off = offset;
     216                        sqe->addr = (__u64)iov;
     217                        sqe->len = iovcnt;
     218                        sqe->rw_flags = 0;
     219                        sqe->__pad2[0] = sqe->__pad2[1] = sqe->__pad2[2] = 0;
    210220
    211221                        __submit_wait
    … …  
    220230                __submit_prelude
    221231
    222                 (*sqe){ IORING_OP_FSYNC, fd };
    223 
    224                 __submit_wait
    225         #endif
    226 }
    227 
    228 int cfa_sync_file_range(int fd, int64_t offset, int64_t nbytes, unsigned int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {
     232                sqe->opcode = IORING_OP_FSYNC;
     233                sqe->ioprio = 0;
     234                sqe->fd = fd;
     235                sqe->off = 0;
     236                sqe->addr = 0;
     237                sqe->len = 0;
     238                sqe->rw_flags = 0;
     239                sqe->__pad2[0] = sqe->__pad2[1] = sqe->__pad2[2] = 0;
     240
     241                __submit_wait
     242        #endif
     243}
     244
     245int cfa_sync_file_range(int fd, off64_t offset, off64_t nbytes, unsigned int flags, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {
    229246        #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_SYNC_FILE_RANGE)
    230247                return sync_file_range(fd, offset, nbytes, flags);
    … …  
    275292
    276293                (*sqe){ IORING_OP_SEND, sockfd };
    277                 sqe->addr = (uint64_t)buf;
     294                sqe->addr = (__u64)buf;
    278295                sqe->len = len;
    279296                sqe->msg_flags = flags;
    … …  
    290307
    291308                (*sqe){ IORING_OP_RECV, sockfd };
    292                 sqe->addr = (uint64_t)buf;
     309                sqe->addr = (__u64)buf;
    293310                sqe->len = len;
    294311                sqe->msg_flags = flags;
    … …  
    305322
    306323                (*sqe){ IORING_OP_ACCEPT, sockfd };
    307                 sqe->addr = (uint64_t)(uintptr_t)addr;
    308                 sqe->addr2 = (uint64_t)(uintptr_t)addrlen;
     324                sqe->addr  = (__u64)addr;
     325                sqe->addr2 = (__u64)addrlen;
    309326                sqe->accept_flags = flags;
    310327
    … …  
    320337
    321338                (*sqe){ IORING_OP_CONNECT, sockfd };
    322                 sqe->addr = (uint64_t)(uintptr_t)addr;
    323                 sqe->off  = (uint64_t)(uintptr_t)addrlen;
    324 
    325                 __submit_wait
    326         #endif
    327 }
    328 
    329 int cfa_fallocate(int fd, int mode, uint64_t offset, uint64_t len, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {
     339                sqe->addr = (__u64)addr;
     340                sqe->off  = (__u64)addrlen;
     341
     342                __submit_wait
     343        #endif
     344}
     345
     346int cfa_fallocate(int fd, int mode, off_t offset, off_t len, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {
    330347        #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_FALLOCATE)
    331348                return fallocate( fd, mode, offset, len );
    … …  
    344361}
    345362
    346 int cfa_fadvise(int fd, uint64_t offset, uint64_t len, int advice, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {
     363int cfa_fadvise(int fd, off_t offset, off_t len, int advice, int submit_flags, Duration timeout, io_cancellation * cancellation, io_context * context) {
    347364        #if !defined(CFA_HAVE_LINUX_IO_URING_H) || !defined(CFA_HAVE_IORING_OP_FADVISE)
    348365                return posix_fadvise( fd, offset, len, advice );
    … …  
    351368
    352369                (*sqe){ IORING_OP_FADVISE, fd };
    353                 sqe->off = (uint64_t)offset;
     370                sqe->off = (__u64)offset;
    354371                sqe->len = len;
    355372                sqe->fadvise_advice = advice;
    … …  
    366383
    367384                (*sqe){ IORING_OP_MADVISE, 0 };
    368                 sqe->addr = (uint64_t)addr;
     385                sqe->addr = (__u64)addr;
    369386                sqe->len = length;
    370387                sqe->fadvise_advice = advice;
    … …  
    381398
    382399                (*sqe){ IORING_OP_OPENAT, dirfd };
    383                 sqe->addr = (uint64_t)pathname;
     400                sqe->addr = (__u64)pathname;
    384401                sqe->open_flags = flags;
    385402                sqe->len = mode;
    … …  
    414431                __submit_prelude
    415432
    416                 (*sqe){ IORING_OP_STATX, dirfd, pathname, mask, (uint64_t)statxbuf };
     433                (*sqe){ IORING_OP_STATX, dirfd, pathname, mask, (__u64)statxbuf };
    417434                sqe->statx_flags = flags;
    418435
    … …  
    456473                }
    457474                else {
    458                         sqe->off = (uint64_t)-1;
     475                        sqe->off = (__u64)-1;
    459476                }
    460477                sqe->len = len;
    … …  
    464481                }
    465482                else {
    466                         sqe->splice_off_in = (uint64_t)-1;
     483                        sqe->splice_off_in = (__u64)-1;
    467484                }
    468485                sqe->splice_flags  = flags | (SPLICE_FLAGS & submit_flags);
Note: See TracChangeset for help on using the changeset viewer.