Changes in / [f90d10f:d3ab183]


Ignore:
Files:
5 edited

Legend:

Unmodified
Added
Removed

  • benchmark/io/readv.cfa

    rf90d10f rd3ab183  
    1616#include <thread.hfa>
    1717#include <time.hfa>
     18
     19#if !defined(HAVE_LINUX_IO_URING_H)
     20#warning no io uring
     21#endif
    1822
    1923extern bool traceHeapOn();

  • libcfa/prelude/defines.hfa.in

    rf90d10f rd3ab183  
    1616#undef HAVE_LINUX_IO_URING_H
    1717
    18 #undef __CFA_NO_STATISTICS__
     18// #define __CFA_IO_POLLING_USER__
     19// #define __CFA_IO_POLLING_KERNEL__

  • libcfa/src/concurrency/io.cfa

    rf90d10f rd3ab183  
    8686        #endif
    8787
    88         // Fast poller user-thread
    89         // Not using the "thread" keyword because we want to control
    90         // more carefully when to start/stop it
    91         struct __io_poller_fast {
    92                 struct __io_data * ring;
    93                 bool waiting;
    94                 $thread thrd;
    95         };
    96 
    97         void ?{}( __io_poller_fast & this, struct cluster & cltr ) {
    98                 this.ring = cltr.io;
    99                 this.waiting = true;
    100                 (this.thrd){ "Fast I/O Poller", cltr };
    101         }
    102         void ^?{}( __io_poller_fast & mutex this );
    103         void main( __io_poller_fast & this );
    104         static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; }
    105         void ^?{}( __io_poller_fast & mutex this ) {}
    106 
    107         struct __submition_data {
    108                 // Head and tail of the ring (associated with array)
    109                 volatile uint32_t * head;
    110                 volatile uint32_t * tail;
    111 
    112                 // The actual kernel ring which uses head/tail
    113                 // indexes into the sqes arrays
    114                 uint32_t * array;
    115 
    116                 // number of entries and mask to go with it
    117                 const uint32_t * num;
    118                 const uint32_t * mask;
    119 
    120                 // Submission flags (Not sure what for)
    121                 uint32_t * flags;
    122 
    123                 // number of sqes not submitted (whatever that means)
    124                 uint32_t * dropped;
    125 
    126                 // Like head/tail but not seen by the kernel
    127                 volatile uint32_t alloc;
    128                 volatile uint32_t ready;
    129 
    130                 __spinlock_t lock;
    131 
    132                 // A buffer of sqes (not the actual ring)
    133                 struct io_uring_sqe * sqes;
    134 
    135                 // The location and size of the mmaped area
    136                 void * ring_ptr;
    137                 size_t ring_sz;
    138 
    139                 // Statistics
    140                 #if !defined(__CFA_NO_STATISTICS__)
    141                         struct {
    142                                 struct {
    143                                         unsigned long long int val;
    144                                         unsigned long long int cnt;
    145                                 } submit_avg;
    146                         } stats;
    147                 #endif
    148         };
    149 
    150         struct __completion_data {
    151                 // Head and tail of the ring
    152                 volatile uint32_t * head;
    153                 volatile uint32_t * tail;
    154 
    155                 // number of entries and mask to go with it
    156                 const uint32_t * mask;
    157                 const uint32_t * num;
    158 
    159                 // number of cqes not submitted (whatever that means)
    160                 uint32_t * overflow;
    161 
    162                 // the kernel ring
    163                 struct io_uring_cqe * cqes;
    164 
    165                 // The location and size of the mmaped area
    166                 void * ring_ptr;
    167                 size_t ring_sz;
    168 
    169                 // Statistics
    170                 #if !defined(__CFA_NO_STATISTICS__)
    171                         struct {
    172                                 struct {
    173                                         unsigned long long int val;
    174                                         unsigned long long int slow_cnt;
    175                                         unsigned long long int fast_cnt;
    176                                 } completed_avg;
    177                         } stats;
    178                 #endif
    179         };
    180 
    181         struct __io_data {
    182                 struct __submition_data submit_q;
    183                 struct __completion_data completion_q;
    184                 uint32_t flags;
    185                 int fd;
    186                 semaphore submit;
    187                 volatile bool done;
    188                 struct {
    189                         struct {
    190                                 void * stack;
    191                                 pthread_t kthrd;
    192                         } slow;
    193                         __io_poller_fast fast;
    194                         __bin_sem_t sem;
    195                 } poller;
    196         };
     88        #if defined(__CFA_IO_POLLING_USER__)
     89                void ?{}( __io_poller_fast & this, struct cluster & cltr ) {
     90                        this.ring = &cltr.io;
     91                        (this.thrd){ "Fast I/O Poller", cltr };
     92                }
     93                void ^?{}( __io_poller_fast & mutex this );
     94        void main( __io_poller_fast & this );
     95        static inline $thread * get_thread( __io_poller_fast & this ) { return &this.thrd; }
     96                void ^?{}( __io_poller_fast & mutex this ) {}
     97        #endif
    19798
    19899//=============================================================================================
    … …  
    200101//=============================================================================================
    201102        void __kernel_io_startup( cluster & this, bool main_cluster ) {
    202                 this.io = malloc();
    203 
    204103                // Step 1 : call to setup
    205104                struct io_uring_params params;
    … …  
    214113
    215114                // Step 2 : mmap result
    216                 memset( this.io, 0, sizeof(struct __io_data) );
    217                 struct __submition_data  & sq = this.io->submit_q;
    218                 struct __completion_data & cq = this.io->completion_q;
     115                memset(&this.io, 0, sizeof(struct io_ring));
     116                struct io_uring_sq & sq = this.io.submit_q;
     117                struct io_uring_cq & cq = this.io.completion_q;
    219118
    220119                // calculate the right ring size
    … …  
    294193
    295194                // Update the global ring info
    296                 this.io->flags = params.flags;
    297                 this.io->fd    = fd;
    298                 this.io->done  = false;
    299                 (this.io->submit){ min(*sq.num, *cq.num) };
     195                this.io.flags = params.flags;
     196                this.io.fd    = fd;
     197                this.io.done  = false;
     198                (this.io.submit){ min(*sq.num, *cq.num) };
    300199
    301200                // Initialize statistics
    302201                #if !defined(__CFA_NO_STATISTICS__)
    303                         this.io->submit_q.stats.submit_avg.val = 0;
    304                         this.io->submit_q.stats.submit_avg.cnt = 0;
    305                         this.io->completion_q.stats.completed_avg.val = 0;
    306                         this.io->completion_q.stats.completed_avg.slow_cnt = 0;
    307                         this.io->completion_q.stats.completed_avg.fast_cnt = 0;
     202                        this.io.submit_q.stats.submit_avg.val = 0;
     203                        this.io.submit_q.stats.submit_avg.cnt = 0;
     204                        this.io.completion_q.stats.completed_avg.val = 0;
     205                        this.io.completion_q.stats.completed_avg.cnt = 0;
    308206                #endif
    309207
    … …  
    314212
    315213        void __kernel_io_finish_start( cluster & this ) {
    316                 __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
    317                 (this.io->poller.fast){ this };
    318                 __thrd_start( this.io->poller.fast, main );
     214                #if defined(__CFA_IO_POLLING_USER__)
     215                        __cfadbg_print_safe(io_core, "Kernel I/O : Creating fast poller for cluter %p\n", &this);
     216                        (this.io.poller.fast){ this };
     217                        __thrd_start( this.io.poller.fast, main );
     218                #endif
    319219
    320220                // Create the poller thread
    321221                __cfadbg_print_safe(io_core, "Kernel I/O : Creating slow poller for cluter %p\n", &this);
    322                 this.io->poller.slow.stack = __create_pthread( &this.io->poller.slow.kthrd, __io_poller_slow, &this );
     222                this.io.poller.slow.stack = __create_pthread( &this.io.poller.slow.kthrd, __io_poller_slow, &this );
    323223        }
    324224
    … …  
    326226                __cfadbg_print_safe(io_core, "Kernel I/O : Stopping pollers for cluster\n", &this);
    327227                // Notify the poller thread of the shutdown
    328                 __atomic_store_n(&this.io->done, true, __ATOMIC_SEQ_CST);
     228                __atomic_store_n(&this.io.done, true, __ATOMIC_SEQ_CST);
    329229
    330230                // Stop the IO Poller
    331231                sigval val = { 1 };
    332                 pthread_sigqueue( this.io->poller.slow.kthrd, SIGUSR1, val );
    333                 post( this.io->poller.sem );
     232                pthread_sigqueue( this.io.poller.slow.kthrd, SIGUSR1, val );
     233                #if defined(__CFA_IO_POLLING_USER__)
     234                        post( this.io.poller.sem );
     235                #endif
    334236
    335237                // Wait for the poller thread to finish
    336                 pthread_join( this.io->poller.slow.kthrd, 0p );
    337                 free( this.io->poller.slow.stack );
     238                pthread_join( this.io.poller.slow.kthrd, 0p );
     239                free( this.io.poller.slow.stack );
    338240
    339241                __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller stopped for cluster\n", &this);
    340242
    341243                #if defined(__CFA_IO_POLLING_USER__)
    342                         verify( this.io->poller.fast.waiting );
    343                         verify( this.io->poller.fast.thrd.state == Blocked );
    344 
    345                         this.io->poller.fast.thrd.curr_cluster = mainCluster;
    346 
    347244                        // unpark the fast io_poller
    348                         unpark( &this.io->poller.fast.thrd __cfaabi_dbg_ctx2 );
    349 
    350                         ^(this.io->poller.fast){};
     245                        unpark( &this.io.poller.fast.thrd __cfaabi_dbg_ctx2 );
     246
     247                        ^(this.io.poller.fast){};
    351248
    352249                        __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller stopped for cluster\n", &this);
    … …  
    362259                #if !defined(__CFA_NO_STATISTICS__)
    363260                        if(this.print_stats) {
    364                                 with(this.io->submit_q.stats, this.io->completion_q.stats) {
    365                                         __cfaabi_bits_print_safe( STDERR_FILENO,
    366                                                 "----- I/O uRing Stats -----\n"
    367                                                 "- total submit calls  : %llu\n"
    368                                                 "- avg submit          : %lf\n"
    369                                                 "- total wait calls    : %llu (%llu slow, %llu fast)\n"
    370                                                 "- avg completion/wait : %lf\n",
    371                                                 submit_avg.cnt,
    372                                                 ((double)submit_avg.val) / submit_avg.cnt,
    373                                                 completed_avg.slow_cnt + completed_avg.fast_cnt,
    374                                                 completed_avg.slow_cnt,  completed_avg.fast_cnt,
    375                                                 ((double)completed_avg.val) / (completed_avg.slow_cnt + completed_avg.fast_cnt)
    376                                         );
    377                                 }
     261                                __cfaabi_bits_print_safe( STDERR_FILENO,
     262                                        "----- I/O uRing Stats -----\n"
     263                                        "- total submit calls  : %llu\n"
     264                                        "- avg submit          : %lf\n"
     265                                        "- total wait calls    : %llu\n"
     266                                        "- avg completion/wait : %lf\n",
     267                                        this.io.submit_q.stats.submit_avg.cnt,
     268                                        ((double)this.io.submit_q.stats.submit_avg.val) / this.io.submit_q.stats.submit_avg.cnt,
     269                                        this.io.completion_q.stats.completed_avg.cnt,
     270                                        ((double)this.io.completion_q.stats.completed_avg.val) / this.io.completion_q.stats.completed_avg.cnt
     271                                );
    378272                        }
    379273                #endif
    380274
    381275                // Shutdown the io rings
    382                 struct __submition_data  & sq = this.io->submit_q;
    383                 struct __completion_data & cq = this.io->completion_q;
     276                struct io_uring_sq & sq = this.io.submit_q;
     277                struct io_uring_cq & cq = this.io.completion_q;
    384278
    385279                // unmap the submit queue entries
    … …  
    395289
    396290                // close the file descriptor
    397                 close(this.io->fd);
    398 
    399                 free( this.io );
     291                close(this.io.fd);
    400292        }
    401293
    … …  
    410302        // Process a single completion message from the io_uring
    411303        // This is NOT thread-safe
    412         static int __drain_io( struct __io_data & ring, sigset_t * mask, int waitcnt, bool in_kernel ) {
     304        static int __drain_io( struct io_ring & ring, sigset_t * mask, int waitcnt, bool in_kernel ) {
    413305                int ret = syscall( __NR_io_uring_enter, ring.fd, 0, waitcnt, IORING_ENTER_GETEVENTS, mask, _NSIG / 8);
    414306                if( ret < 0 ) {
    … …  
    428320                // Nothing was new return 0
    429321                if (head == tail) {
     322                        #if !defined(__CFA_NO_STATISTICS__)
     323                                ring.completion_q.stats.completed_avg.cnt += 1;
     324                        #endif
    430325                        return 0;
    431326                }
    … …  
    453348                __atomic_fetch_add( ring.completion_q.head, count, __ATOMIC_RELAXED );
    454349
     350                // Update statistics
     351                #if !defined(__CFA_NO_STATISTICS__)
     352                        ring.completion_q.stats.completed_avg.val += count;
     353                        ring.completion_q.stats.completed_avg.cnt += 1;
     354                #endif
     355
    455356                return count;
    456357        }
    … …  
    458359        static void * __io_poller_slow( void * arg ) {
    459360                cluster * cltr = (cluster *)arg;
    460                 struct __io_data & ring = *cltr->io;
     361                struct io_ring & ring = cltr->io;
    461362
    462363                sigset_t mask;
    … …  
    471372                verify( (*ring.completion_q.head) == (*ring.completion_q.tail) );
    472373
    473                 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p ready\n", &ring);
    474 
    475374                while(!__atomic_load_n(&ring.done, __ATOMIC_SEQ_CST)) {
    476375                        #if defined(__CFA_IO_POLLING_USER__)
    … …  
    479378                                // batton pass to the user-thread
    480379                                int count = __drain_io( ring, &mask, 1, true );
    481 
    482                                 // Update statistics
    483                                 #if !defined(__CFA_NO_STATISTICS__)
    484                                         ring.completion_q.stats.completed_avg.val += count;
    485                                         ring.completion_q.stats.completed_avg.slow_cnt += 1;
    486                                 #endif
    487 
    488380                                if(count > 0) {
    489381                                        __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to fast poller\n", &ring);
    … …  
    495387
    496388                                //In the naive approach, just poll the io completion queue directly
    497                                 int count = __drain_io( ring, &mask, 1, true );
    498 
    499                                 // Update statistics
    500                                 #if !defined(__CFA_NO_STATISTICS__)
    501                                         ring.completion_q.stats.completed_avg.val += count;
    502                                         ring.completion_q.stats.completed_avg.slow_cnt += 1;
    503                                 #endif
     389                                __drain_io( ring, &mask, 1, true );
    504390
    505391                        #endif
    506392                }
    507393
    508                 __cfadbg_print_safe(io_core, "Kernel I/O : Slow poller for ring %p stopping\n", &ring);
    509 
    510394                return 0p;
    511395        }
    512396
    513         void main( __io_poller_fast & this ) {
    514                 // Start parked
    515                 park( __cfaabi_dbg_ctx );
    516 
    517                 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p ready\n", &this.ring);
    518 
    519                 // Then loop until we need to start
    520                 while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) {
    521                         // Drain the io
    522                         this.waiting = false;
    523                         int count = __drain_io( *this.ring, 0p, 0, false );
    524 
    525                         // Update statistics
    526                         #if !defined(__CFA_NO_STATISTICS__)
    527                                 this.ring->completion_q.stats.completed_avg.val += count;
    528                                 this.ring->completion_q.stats.completed_avg.fast_cnt += 1;
    529                         #endif
    530 
    531                         this.waiting = true;
    532                         if(0 > count) {
    533                                 // If we got something, just yield and check again
    534                                 yield();
     397        #if defined(__CFA_IO_POLLING_USER__)
     398                void main( __io_poller_fast & this ) {
     399                        // Start parked
     400                        park( __cfaabi_dbg_ctx );
     401
     402                        // Then loop until we need to start
     403                        while(!__atomic_load_n(&this.ring->done, __ATOMIC_SEQ_CST)) {
     404                                // Drain the io
     405                                if(0 > __drain_io( *this.ring, 0p, 0, false )) {
     406                                        // If we got something, just yield and check again
     407                                        yield();
     408                                }
     409                                else {
     410                                        // We didn't get anything baton pass to the slow poller
     411                                        __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring);
     412                                        post( this.ring->poller.sem );
     413                                        park( __cfaabi_dbg_ctx );
     414                                }
    535415                        }
    536                         else {
    537                                 // We didn't get anything baton pass to the slow poller
    538                                 __cfadbg_print_safe(io_core, "Kernel I/O : Moving to ring %p to slow poller\n", &this.ring);
    539                                 post( this.ring->poller.sem );
    540                                 park( __cfaabi_dbg_ctx );
    541                         }
    542                 }
    543 
    544                 __cfadbg_print_safe(io_core, "Kernel I/O : Fast poller for ring %p stopping\n", &this.ring);
    545         }
     416                }
     417        #endif
    546418
    547419//=============================================================================================
    … …  
    573445//
    574446
    575         static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct __io_data & ring ) {
     447        static inline [* struct io_uring_sqe, uint32_t] __submit_alloc( struct io_ring & ring ) {
    576448                // Wait for a spot to be available
    577449                P(ring.submit);
    … …  
    591463        }
    592464
    593         static inline void __submit( struct __io_data & ring, uint32_t idx ) {
     465        static inline void __submit( struct io_ring & ring, uint32_t idx ) {
    594466                // get mutual exclusion
    595467                lock(ring.submit_q.lock __cfaabi_dbg_ctx2);
    … …  
    652524
    653525        #define __submit_prelude \
    654                 struct __io_data & ring = *active_cluster()->io; \
     526                struct io_ring & ring = active_cluster()->io; \
    655527                struct io_uring_sqe * sqe; \
    656528                uint32_t idx; \

  • libcfa/src/concurrency/kernel.hfa

    rf90d10f rd3ab183  
    114114//-----------------------------------------------------------------------------
    115115// I/O
    116 struct __io_data;
     116#if defined(HAVE_LINUX_IO_URING_H)
     117struct io_uring_sq {
     118        // Head and tail of the ring (associated with array)
     119        volatile uint32_t * head;
     120        volatile uint32_t * tail;
     121
     122        // The actual kernel ring which uses head/tail
     123        // indexes into the sqes arrays
     124        uint32_t * array;
     125
     126        // number of entries and mask to go with it
     127        const uint32_t * num;
     128        const uint32_t * mask;
     129
     130        // Submission flags (Not sure what for)
     131        uint32_t * flags;
     132
     133        // number of sqes not submitted (whatever that means)
     134        uint32_t * dropped;
     135
     136        // Like head/tail but not seen by the kernel
     137        volatile uint32_t alloc;
     138        volatile uint32_t ready;
     139
     140        __spinlock_t lock;
     141
     142        // A buffer of sqes (not the actual ring)
     143        struct io_uring_sqe * sqes;
     144
     145        // The location and size of the mmaped area
     146        void * ring_ptr;
     147        size_t ring_sz;
     148
     149        // Statistics
     150        #if !defined(__CFA_NO_STATISTICS__)
     151                struct {
     152                        struct {
     153                                unsigned long long int val;
     154                                unsigned long long int cnt;
     155                        } submit_avg;
     156                } stats;
     157        #endif
     158};
     159
     160struct io_uring_cq {
     161        // Head and tail of the ring
     162        volatile uint32_t * head;
     163        volatile uint32_t * tail;
     164
     165        // number of entries and mask to go with it
     166        const uint32_t * mask;
     167        const uint32_t * num;
     168
     169        // number of cqes not submitted (whatever that means)
     170        uint32_t * overflow;
     171
     172        // the kernel ring
     173        struct io_uring_cqe * cqes;
     174
     175        // The location and size of the mmaped area
     176        void * ring_ptr;
     177        size_t ring_sz;
     178
     179        // Statistics
     180        #if !defined(__CFA_NO_STATISTICS__)
     181                struct {
     182                        struct {
     183                                unsigned long long int val;
     184                                unsigned long long int cnt;
     185                        } completed_avg;
     186                } stats;
     187        #endif
     188};
     189
     190#if defined(__CFA_IO_POLLING_USER__)
     191        struct __io_poller_fast {
     192                struct io_ring * ring;
     193                $thread thrd;
     194        };
     195#endif
     196
     197struct io_ring {
     198        struct io_uring_sq submit_q;
     199        struct io_uring_cq completion_q;
     200        uint32_t flags;
     201        int fd;
     202        semaphore submit;
     203        volatile bool done;
     204        struct {
     205                struct {
     206                        void * stack;
     207                        pthread_t kthrd;
     208                } slow;
     209                #if defined(__CFA_IO_POLLING_USER__)
     210                        __io_poller_fast fast;
     211                        __bin_sem_t sem;
     212                #endif
     213        } poller;
     214};
     215#endif
    117216
    118217//-----------------------------------------------------------------------------
    … …  
    148247        } node;
    149248
    150         struct __io_data * io;
     249        #if defined(HAVE_LINUX_IO_URING_H)
     250                struct io_ring io;
     251        #endif
    151252
    152253        #if !defined(__CFA_NO_STATISTICS__)

  • libcfa/src/concurrency/kernel_private.hfa

    rf90d10f rd3ab183  
    5959extern volatile thread_local __cfa_kernel_preemption_state_t preemption_state __attribute__ ((tls_model ( "initial-exec" )));
    6060
    61 extern cluster * mainCluster;
    62 
    6361//-----------------------------------------------------------------------------
    6462// Threads
Note: See TracChangeset for help on using the changeset viewer.