Ignore:
Timestamp:
Feb 19, 2021, 1:47:09 PM (8 months ago)
Author:
Thierry Delisle <tdelisle@…>
Branches:
arm-eh, jacob/cs343-translation, master, new-ast-unique-expr
Children:
4f762d3
Parents:
b44959f
Message:

New implementation of io based on instance burrowing.
Trying to avoid the unbounded growth of the previous flat combining approach.

File:
1 edited

Legend:

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

    rb44959f r78da4ab  
    3636        void ?{}(io_context_params & this) {}
    3737
    38         void ?{}(io_context & this, struct cluster & cl) {}
    39         void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {}
    40 
    41         void ^?{}(io_context & this) {}
    42         void ^?{}(io_context & this, bool cluster_context) {}
    43 
    44         void register_fixed_files( io_context &, int *, unsigned ) {}
    45         void register_fixed_files( cluster    &, int *, unsigned ) {}
     38        void  ?{}($io_context & this, struct cluster & cl) {}
     39        void ^?{}($io_context & this) {}
     40
     41        $io_arbiter * create(void) { return 0p; }
     42        void destroy($io_arbiter *) {}
    4643
    4744#else
     
    6865        void ?{}(io_context_params & this) {
    6966                this.num_entries = 256;
    70                 this.num_ready = 256;
    71                 this.submit_aff = -1;
    72                 this.eager_submits = false;
    73                 this.poller_submits = false;
    74                 this.poll_submit = false;
    75                 this.poll_complete = false;
    7667        }
    7768
     
    194185
    195186                        for(i; nfds) {
    196                                 $io_ctx_thread * io_ctx = ($io_ctx_thread *)(uintptr_t)events[i].data.u64;
     187                                $io_context * io_ctx = ($io_context *)(uintptr_t)events[i].data.u64;
    197188                                /* paranoid */ verify( io_ctx );
    198                                 __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Unparking io poller %d (%p)\n", io_ctx->ring->fd, io_ctx);
     189                                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Unparking io poller %d (%p)\n", io_ctx->fd, io_ctx);
    199190                                #if !defined( __CFA_NO_STATISTICS__ )
    200191                                        __cfaabi_tls.this_stats = io_ctx->self.curr_cluster->stats;
     
    202193
    203194                                eventfd_t v;
    204                                 eventfd_read(io_ctx->ring->efd, &v);
     195                                eventfd_read(io_ctx->efd, &v);
    205196
    206197                                post( io_ctx->sem );
     
    219210//=============================================================================================
    220211
    221         void ?{}($io_ctx_thread & this, struct cluster & cl) { (this.self){ "IO Poller", cl }; }
    222         void main( $io_ctx_thread & this );
    223         static inline $thread * get_thread( $io_ctx_thread & this ) { return &this.self; }
    224         void ^?{}( $io_ctx_thread & mutex this ) {}
    225 
    226         static void __io_create ( __io_data & this, const io_context_params & params_in );
    227         static void __io_destroy( __io_data & this );
    228 
    229         void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {
    230                 (this.thrd){ cl };
    231                 this.thrd.ring = malloc();
    232                 __cfadbg_print_safe(io_core, "Kernel I/O : Creating ring for io_context %p\n", &this);
    233                 __io_create( *this.thrd.ring, params );
    234 
    235                 __cfadbg_print_safe(io_core, "Kernel I/O : Starting poller thread for io_context %p\n", &this);
    236                 this.thrd.done = false;
    237                 __thrd_start( this.thrd, main );
    238 
    239                 __cfadbg_print_safe(io_core, "Kernel I/O : io_context %p ready\n", &this);
     212        static void __io_uring_setup ( $io_context & this, const io_context_params & params_in );
     213        static void __io_uring_teardown( $io_context & this );
     214        static void __epoll_register($io_context & ctx);
     215        static void __epoll_unregister($io_context & ctx);
     216        void __ioarbiter_register( $io_arbiter & mutex, $io_context & ctx );
     217        void __ioarbiter_unregister( $io_arbiter & mutex, $io_context & ctx );
     218
     219        void ?{}($io_context & this, struct cluster & cl) {
     220                (this.self){ "IO Poller", cl };
     221                this.ext_sq.empty = true;
     222                __io_uring_setup( this, cl.io.params );
     223                __cfadbg_print_safe(io_core, "Kernel I/O : Created ring for io_context %u (%p)\n", this.fd, &this);
     224
     225                __epoll_register(this);
     226
     227                __ioarbiter_register(*cl.io.arbiter, this);
     228
     229                __thrd_start( this, main );
     230                __cfadbg_print_safe(io_core, "Kernel I/O : Started poller thread for io_context %u\n", this.fd);
     231        }
     232
     233        void ^?{}($io_context & mutex this) {
     234                __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %u\n", this.fd);
     235
     236                ^(this.self){};
     237                __cfadbg_print_safe(io_core, "Kernel I/O : Stopped poller thread for io_context %u\n", this.fd);
     238
     239                __ioarbiter_unregister(*this.arbiter, this);
     240
     241                __epoll_unregister(this);
     242
     243                __io_uring_teardown( this );
     244                __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %u\n", this.fd);
    240245        }
    241246
    242247        void ?{}(io_context & this, struct cluster & cl) {
    243                 io_context_params params;
    244                 (this){ cl, params };
    245         }
    246 
    247         void ^?{}(io_context & this, bool cluster_context) {
    248                 __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %p\n", &this);
    249 
    250                 // Notify the thread of the shutdown
    251                 __atomic_store_n(&this.thrd.done, true, __ATOMIC_SEQ_CST);
    252 
    253                 // If this is an io_context within a cluster, things get trickier
    254                 $thread & thrd = this.thrd.self;
    255                 if( cluster_context ) {
    256                         // We are about to do weird things with the threads
    257                         // we don't need interrupts to complicate everything
    258                         disable_interrupts();
    259 
    260                         // Get cluster info
    261                         cluster & cltr = *thrd.curr_cluster;
    262                         /* paranoid */ verify( cltr.idles.total == 0 || &cltr == mainCluster );
    263                         /* paranoid */ verify( !ready_mutate_islocked() );
    264 
    265                         // We need to adjust the clean-up based on where the thread is
    266                         if( thrd.state == Ready || thrd.preempted != __NO_PREEMPTION ) {
    267                                 // This is the tricky case
    268                                 // The thread was preempted or ready to run and now it is on the ready queue
    269                                 // but the cluster is shutting down, so there aren't any processors to run the ready queue
    270                                 // the solution is to steal the thread from the ready-queue and pretend it was blocked all along
    271 
    272                                 ready_schedule_lock();
    273                                         // The thread should on the list
    274                                         /* paranoid */ verify( thrd.link.next != 0p );
    275 
    276                                         // Remove the thread from the ready queue of this cluster
    277                                         // The thread should be the last on the list
    278                                         __attribute__((unused)) bool removed = remove_head( &cltr, &thrd );
    279                                         /* paranoid */ verify( removed );
    280                                         thrd.link.next = 0p;
    281                                         thrd.link.prev = 0p;
    282 
    283                                         // Fixup the thread state
    284                                         thrd.state = Blocked;
    285                                         thrd.ticket = TICKET_BLOCKED;
    286                                         thrd.preempted = __NO_PREEMPTION;
    287 
    288                                 ready_schedule_unlock();
    289 
    290                                 // Pretend like the thread was blocked all along
    291                         }
    292                         // !!! This is not an else if !!!
    293                         // Ok, now the thread is blocked (whether we cheated to get here or not)
    294                         if( thrd.state == Blocked ) {
    295                                 // This is the "easy case"
    296                                 // The thread is parked and can easily be moved to active cluster
    297                                 verify( thrd.curr_cluster != active_cluster() || thrd.curr_cluster == mainCluster );
    298                                 thrd.curr_cluster = active_cluster();
    299 
    300                                 // unpark the fast io_poller
    301                                 unpark( &thrd );
    302                         }
    303                         else {
    304                                 // The thread is in a weird state
    305                                 // I don't know what to do here
    306                                 abort("io_context poller thread is in unexpected state, cannot clean-up correctly\n");
    307                         }
    308 
    309                         // The weird thread kidnapping stuff is over, restore interrupts.
    310                         enable_interrupts( __cfaabi_dbg_ctx );
    311                 } else {
    312                         post( this.thrd.sem );
    313                 }
    314 
    315                 ^(this.thrd){};
    316                 __cfadbg_print_safe(io_core, "Kernel I/O : Stopped poller thread for io_context %p\n", &this);
    317 
    318                 __io_destroy( *this.thrd.ring );
    319                 __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %p\n", &this);
    320 
    321                 free(this.thrd.ring);
     248                // this.ctx = new(cl);
     249                this.ctx = alloc();
     250                (*this.ctx){ cl };
     251
     252                __cfadbg_print_safe(io_core, "Kernel I/O : io_context %u ready\n", this.ctx->fd);
    322253        }
    323254
    324255        void ^?{}(io_context & this) {
    325                 ^(this){ false };
     256                post( this.ctx->sem );
     257
     258                delete(this.ctx);
    326259        }
    327260
     
    329262        extern void __enable_interrupts_hard();
    330263
    331         static void __io_create( __io_data & this, const io_context_params & params_in ) {
     264        static void __io_uring_setup( $io_context & this, const io_context_params & params_in ) {
    332265                // Step 1 : call to setup
    333266                struct io_uring_params params;
    334267                memset(&params, 0, sizeof(params));
    335                 if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
    336                 if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
     268                // if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
     269                // if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
    337270
    338271                __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
     
    340273                        abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
    341274                }
    342                 if( params_in.poller_submits && params_in.eager_submits ) {
    343                         abort("ERROR: I/O setup 'poller_submits' and 'eager_submits' cannot be used together\n");
    344                 }
    345275
    346276                int fd = syscall(__NR_io_uring_setup, nentries, &params );
     
    350280
    351281                // Step 2 : mmap result
    352                 memset( &this, 0, sizeof(struct __io_data) );
    353                 struct __submition_data  & sq = this.submit_q;
    354                 struct __completion_data & cq = this.completion_q;
     282                struct __sub_ring_t & sq = this.sq;
     283                struct __cmp_ring_t & cq = this.cq;
    355284
    356285                // calculate the right ring size
     
    401330                // Get the pointers from the kernel to fill the structure
    402331                // submit queue
    403                 sq.head    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
    404                 sq.tail    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
    405                 sq.mask    = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
    406                 sq.num     = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
    407                 sq.flags   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
    408                 sq.dropped = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
    409                 sq.array   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
    410                 sq.prev_head = *sq.head;
    411 
    412                 {
    413                         const __u32 num = *sq.num;
    414                         for( i; num ) {
    415                                 __sqe_clean( &sq.sqes[i] );
    416                         }
    417                 }
    418 
    419                 (sq.submit_lock){};
    420                 (sq.release_lock){};
    421 
    422                 if( params_in.poller_submits || params_in.eager_submits ) {
    423                         /* paranoid */ verify( is_pow2( params_in.num_ready ) || (params_in.num_ready < 8) );
    424                         sq.ready_cnt = max( params_in.num_ready, 8 );
    425                         sq.ready = alloc( sq.ready_cnt, 64`align );
    426                         for(i; sq.ready_cnt) {
    427                                 sq.ready[i] = -1ul32;
    428                         }
    429                         sq.prev_ready = 0;
    430                 }
    431                 else {
    432                         sq.ready_cnt = 0;
    433                         sq.ready = 0p;
    434                         sq.prev_ready = 0;
    435                 }
     332                sq.kring.head  = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
     333                sq.kring.tail  = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
     334                sq.kring.array = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
     335                sq.mask        = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
     336                sq.num         = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
     337                sq.flags       = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
     338                sq.dropped     = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
     339
     340                sq.kring.ready = 0;
     341                sq.kring.released = 0;
     342
     343                sq.free_ring.head = 0;
     344                sq.free_ring.tail = *sq.num;
     345                sq.free_ring.array = alloc( *sq.num, 128`align );
     346                for(i; (__u32)*sq.num) {
     347                        sq.free_ring.array[i] = i;
     348                }
     349
     350                sq.to_submit = 0;
    436351
    437352                // completion queue
     
    468383                /* 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 );
    469384                /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
    470                 /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
    471                 /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
     385                /* paranoid */ verifyf( (*sq.kring.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.kring.head );
     386                /* paranoid */ verifyf( (*sq.kring.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.kring.tail );
    472387
    473388                // Update the global ring info
    474                 this.ring_flags = params.flags;
     389                this.ring_flags = 0;
    475390                this.fd         = fd;
    476391                this.efd        = efd;
    477                 this.eager_submits  = params_in.eager_submits;
    478                 this.poller_submits = params_in.poller_submits;
    479         }
    480 
    481         static void __io_destroy( __io_data & this ) {
     392        }
     393
     394        static void __io_uring_teardown( $io_context & this ) {
    482395                // Shutdown the io rings
    483                 struct __submition_data  & sq = this.submit_q;
    484                 struct __completion_data & cq = this.completion_q;
     396                struct __sub_ring_t & sq = this.sq;
     397                struct __cmp_ring_t & cq = this.cq;
    485398
    486399                // unmap the submit queue entries
     
    499412                close(this.efd);
    500413
    501                 free( this.submit_q.ready ); // Maybe null, doesn't matter
     414                free( this.sq.free_ring.array ); // Maybe null, doesn't matter
    502415        }
    503416
     
    505418// I/O Context Sleep
    506419//=============================================================================================
    507         static inline void __ioctx_epoll_ctl($io_ctx_thread & ctx, int op, const char * error) {
     420        static inline void __epoll_ctl($io_context & ctx, int op, const char * error) {
    508421                struct epoll_event ev;
    509422                ev.events = EPOLLIN | EPOLLONESHOT;
    510423                ev.data.u64 = (__u64)&ctx;
    511                 int ret = epoll_ctl(iopoll.epollfd, op, ctx.ring->efd, &ev);
     424                int ret = epoll_ctl(iopoll.epollfd, op, ctx.efd, &ev);
    512425                if (ret < 0) {
    513426                        abort( "KERNEL ERROR: EPOLL %s - (%d) %s\n", error, (int)errno, strerror(errno) );
     
    515428        }
    516429
    517         void __ioctx_register($io_ctx_thread & ctx) {
    518                 __ioctx_epoll_ctl(ctx, EPOLL_CTL_ADD, "ADD");
    519         }
    520 
    521         void __ioctx_prepare_block($io_ctx_thread & ctx) {
    522                 __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %d (%p)\n", ctx.ring->fd, &ctx);
    523                 __ioctx_epoll_ctl(ctx, EPOLL_CTL_MOD, "REARM");
    524         }
    525 
    526         void __ioctx_unregister($io_ctx_thread & ctx) {
     430        static void __epoll_register($io_context & ctx) {
     431                __epoll_ctl(ctx, EPOLL_CTL_ADD, "ADD");
     432        }
     433
     434        static void __epoll_unregister($io_context & ctx) {
    527435                // Read the current epoch so we know when to stop
    528436                size_t curr = __atomic_load_n(&iopoll.epoch, __ATOMIC_SEQ_CST);
    529437
    530438                // Remove the fd from the iopoller
    531                 __ioctx_epoll_ctl(ctx, EPOLL_CTL_DEL, "REMOVE");
     439                __epoll_ctl(ctx, EPOLL_CTL_DEL, "REMOVE");
    532440
    533441                // Notify the io poller thread of the shutdown
     
    543451        }
    544452
     453        void __ioctx_prepare_block($io_context & ctx) {
     454                __cfadbg_print_safe(io_core, "Kernel I/O - epoll : Re-arming io poller %d (%p)\n", ctx.fd, &ctx);
     455                __epoll_ctl(ctx, EPOLL_CTL_MOD, "REARM");
     456        }
     457
     458
    545459//=============================================================================================
    546460// I/O Context Misc Setup
    547461//=============================================================================================
    548         void register_fixed_files( io_context & ctx, int * files, unsigned count ) {
    549                 int ret = syscall( __NR_io_uring_register, ctx.thrd.ring->fd, IORING_REGISTER_FILES, files, count );
    550                 if( ret < 0 ) {
    551                         abort( "KERNEL ERROR: IO_URING REGISTER - (%d) %s\n", (int)errno, strerror(errno) );
    552                 }
    553 
    554                 __cfadbg_print_safe( io_core, "Kernel I/O : Performed io_register for %p, returned %d\n", active_thread(), ret );
    555         }
    556 
    557         void register_fixed_files( cluster & cltr, int * files, unsigned count ) {
    558                 for(i; cltr.io.cnt) {
    559                         register_fixed_files( cltr.io.ctxs[i], files, count );
    560                 }
    561         }
     462        void ?{}( $io_arbiter & this ) {
     463                this.pending.flag = false;
     464        }
     465
     466        void ^?{}( $io_arbiter & mutex this ) {
     467                /* paranoid */ verify( empty(this.assigned) );
     468                /* paranoid */ verify( empty(this.available) );
     469                /* paranoid */ verify( is_empty(this.pending.blocked) );
     470        }
     471
     472        $io_arbiter * create(void) {
     473                return new();
     474        }
     475        void destroy($io_arbiter * arbiter) {
     476                delete(arbiter);
     477        }
     478
     479//=============================================================================================
     480// I/O Context Misc Setup
     481//=============================================================================================
     482
    562483#endif
Note: See TracChangeset for help on using the changeset viewer.