Ignore:
File:
1 edited

Legend:

Unmodified
Added
Removed
  • libcfa/src/concurrency/io.cfa

    rfb98462 r20ab637  
    1414//
    1515
    16 // #define __CFA_DEBUG_PRINT_IO__
    17 // #define __CFA_DEBUG_PRINT_IO_CORE__
     16#if defined(__CFA_DEBUG__)
     17        // #define __CFA_DEBUG_PRINT_IO__
     18        #define __CFA_DEBUG_PRINT_IO_CORE__
     19#endif
    1820
    1921#include "kernel.hfa"
     
    216218                        // adjust the size according to the parameters
    217219                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
    218                                 cq.ring_sz = sq.ring_sz = max(cq.ring_sz, sq.ring_sz);
     220                                cq->ring_sz = sq->ring_sz = max(cq->ring_sz, sq->ring_sz);
    219221                        }
    220222                #endif
     
    230232                        // mmap the Completion Queue into existence (may or may not be needed)
    231233                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
    232                                 cq.ring_ptr = sq.ring_ptr;
     234                                cq->ring_ptr = sq->ring_ptr;
    233235                        }
    234236                        else
     
    325327
    326328                // Create the poller thread
    327                 __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this);
     329                __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluster %p\n", &this);
    328330                this.io->poller.slow.blocked = false;
    329331                this.io->poller.slow.stack = __create_pthread( &this.io->poller.slow.kthrd, __io_poller_slow, &this );
     
    430432        }
    431433
     434        int __io_uring_enter( struct __io_data & ring, unsigned to_submit, bool get, sigset_t * mask ) {
     435                bool need_sys_to_submit = false;
     436                bool need_sys_to_complete = false;
     437                unsigned min_complete = 0;
     438                unsigned flags = 0;
     439
     440
     441                TO_SUBMIT:
     442                if( to_submit > 0 ) {
     443                        if( !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
     444                                need_sys_to_submit = true;
     445                                break TO_SUBMIT;
     446                        }
     447                        if( (*ring.submit_q.flags) & IORING_SQ_NEED_WAKEUP ) {
     448                                need_sys_to_submit = true;
     449                                flags |= IORING_ENTER_SQ_WAKEUP;
     450                        }
     451                }
     452
     453                TO_COMPLETE:
     454                if( get && !(ring.ring_flags & IORING_SETUP_SQPOLL) ) {
     455                        flags |= IORING_ENTER_GETEVENTS;
     456                        if( mask ) {
     457                                need_sys_to_complete = true;
     458                                min_complete = 1;
     459                                break TO_COMPLETE;
     460                        }
     461                        if( (ring.ring_flags & IORING_SETUP_IOPOLL) ) {
     462                                need_sys_to_complete = true;
     463                        }
     464                }
     465
     466                int ret = 0;
     467                if( need_sys_to_submit || need_sys_to_complete ) {
     468                        ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, min_complete, flags, mask, _NSIG / 8);
     469                        if( ret < 0 ) {
     470                                switch((int)errno) {
     471                                case EAGAIN:
     472                                case EINTR:
     473                                        ret = -1;
     474                                        break;
     475                                default:
     476                                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
     477                                }
     478                        }
     479                }
     480
     481                // Memory barrier
     482                __atomic_thread_fence( __ATOMIC_SEQ_CST );
     483                return ret;
     484        }
     485
    432486//=============================================================================================
    433487// I/O Polling
     
    438492        // Process a single completion message from the io_uring
    439493        // This is NOT thread-safe
    440         static [int, bool] __drain_io( & struct __io_data ring, * sigset_t mask, int waitcnt, bool in_kernel ) {
     494        static [int, bool] __drain_io( & struct __io_data ring, * sigset_t mask ) {
    441495                /* paranoid */ verify( !kernelTLS.preemption_state.enabled );
     496                const uint32_t smask = *ring.submit_q.mask;
    442497
    443498                unsigned to_submit = 0;
     
    447502                }
    448503
    449                 if (to_submit > 0 || waitcnt > 0) {
    450                         int ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, waitcnt, IORING_ENTER_GETEVENTS, mask, _NSIG / 8);
    451                         if( ret < 0 ) {
    452                                 switch((int)errno) {
    453                                 case EAGAIN:
    454                                 case EINTR:
    455                                         return [0, true];
    456                                 default:
    457                                         abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
    458                                 }
    459                         }
    460 
    461                         // Release the consumed SQEs
    462                         __release_consumed_submission( ring );
    463 
    464                         // update statistics
     504                int ret = __io_uring_enter(ring, to_submit, true, mask);
     505                if( ret < 0 ) {
     506                        return [0, true];
     507                }
     508
     509                // update statistics
     510                if (to_submit > 0) {
    465511                        __STATS__( true,
    466512                                if( to_submit > 0 ) {
     
    472518                }
    473519
    474                 // Memory barrier
    475                 __atomic_thread_fence( __ATOMIC_SEQ_CST );
     520                // Release the consumed SQEs
     521                __release_consumed_submission( ring );
    476522
    477523                // Drain the queue
     
    497543
    498544                        data->result = cqe.res;
    499                         if(!in_kernel) { unpark( data->thrd __cfaabi_dbg_ctx2 ); }
    500                         else         { __unpark( &ring.poller.slow.id, data->thrd __cfaabi_dbg_ctx2 ); }
     545                        if(!mask) { unpark( data->thrd __cfaabi_dbg_ctx2 ); }
     546                        else      { __unpark( &ring.poller.slow.id, data->thrd __cfaabi_dbg_ctx2 ); }
    501547                }
    502548
     
    546592                                int count;
    547593                                bool again;
    548                                 [count, again] = __drain_io( ring, &mask, 1, true );
     594                                [count, again] = __drain_io( ring, &mask );
    549595
    550596                                __atomic_store_n( &ring.poller.slow.blocked, false, __ATOMIC_SEQ_CST );
     
    568614                                int count;
    569615                                bool again;
    570                                 [count, again] = __drain_io( ring, &mask, 1, true );
     616                                [count, again] = __drain_io( ring, &mask );
    571617
    572618                                // Update statistics
     
    606652                        bool again;
    607653                        disable_interrupts();
    608                                 [count, again] = __drain_io( *this.ring, 0p, 0, false );
     654                                [count, again] = __drain_io( *this.ring, 0p );
    609655
    610656                                if(!again) reset++;
     
    800846                        // We got the lock
    801847                        unsigned to_submit = __collect_submitions( ring );
    802                         int ret = syscall( __NR_io_uring_enter, ring.fd, to_submit, 0, 0, 0p, _NSIG / 8);
     848                        int ret = __io_uring_enter( ring, to_submit, false, 0p );
    803849                        if( ret < 0 ) {
    804                                 switch((int)errno) {
    805                                 case EAGAIN:
    806                                 case EINTR:
    807                                         unlock(ring.submit_q.lock);
    808                                         return;
    809                                 default:
    810                                         abort( "KERNEL ERROR: IO_URING WAIT - %s\n", strerror(errno) );
    811                                 }
    812                         }
    813 
    814                         /* paranoid */ verify( ret > 0 );
     850                                unlock(ring.submit_q.lock);
     851                                return;
     852                        }
     853
     854                        /* paranoid */ verify( ret > 0 || (ring.ring_flags & IORING_SETUP_SQPOLL) );
    815855
    816856                        // Release the consumed SQEs
     
    830870                        lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
    831871
     872                        /* paranoid */ verifyf( ring.submit_q.sqes[ idx ].user_data != 0,
     873                        /* paranoid */  "index %u already reclaimed\n"
     874                        /* paranoid */  "head %u, prev %u, tail %u\n"
     875                        /* paranoid */  "[-0: %u,-1: %u,-2: %u,-3: %u]\n",
     876                        /* paranoid */  idx,
     877                        /* paranoid */  *ring.submit_q.head, ring.submit_q.prev_head, *tail
     878                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 0) & (*ring.submit_q.mask) ]
     879                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 1) & (*ring.submit_q.mask) ]
     880                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 2) & (*ring.submit_q.mask) ]
     881                        /* paranoid */  ,ring.submit_q.array[ ((*ring.submit_q.head) - 3) & (*ring.submit_q.mask) ]
     882                        /* paranoid */ );
     883
    832884                        // Append to the list of ready entries
    833885
    834886                        /* paranoid */ verify( idx <= mask );
    835 
    836                         ring.submit_q.array[ (*tail) & mask ] = idx & mask;
     887                        ring.submit_q.array[ (*tail) & mask ] = idx;
    837888                        __atomic_fetch_add(tail, 1ul32, __ATOMIC_SEQ_CST);
    838889
    839                         /* paranoid */ verify( ring.submit_q.sqes[ idx ].user_data != 0 );
    840 
    841890                        // Submit however, many entries need to be submitted
    842                         int ret = syscall( __NR_io_uring_enter, ring.fd, 1, 0, 0, 0p, 0);
     891                        int ret = __io_uring_enter( ring, 1, false, 0p );
    843892                        if( ret < 0 ) {
    844893                                switch((int)errno) {
     
    906955                return count;
    907956        }
     957
     958//=============================================================================================
     959// I/O Submissions
     960//=============================================================================================
     961
     962        void register_fixed_files( cluster & cl, int * files, unsigned count ) {
     963                int ret = syscall( __NR_io_uring_register, cl.io->fd, IORING_REGISTER_FILES, files, count );
     964                if( ret < 0 ) {
     965                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
     966                }
     967
     968                __cfadbg_print_safe( io_core, "Kernel I/O : Performed io_register for %p, returned %d\n", active_thread(), ret );
     969        }
    908970#endif
Note: See TracChangeset for help on using the changeset viewer.