source: libcfa/src/concurrency/io/setup.cfa @ 58fe85a

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 58fe85a was 80444bb, checked in by Thierry Delisle <tdelisle@…>, 4 years ago

Fixed prints in io setup.

  • Property mode set to 100644
File size: 16.2 KB
Line 
1//
2// Cforall Version 1.0.0 Copyright (C) 2020 University of Waterloo
3//
4// The contents of this file are covered under the licence agreement in the
5// file "LICENCE" distributed with Cforall.
6//
7// io/setup.cfa --
8//
9// Author           : Thierry Delisle
10// Created On       : Fri Jul 31 16:25:51 2020
11// Last Modified By :
12// Last Modified On :
13// Update Count     :
14//
15
16#define __cforall_thread__
17#define _GNU_SOURCE         /* See feature_test_macros(7) */
18
19#if defined(__CFA_DEBUG__)
20        // #define __CFA_DEBUG_PRINT_IO__
21        // #define __CFA_DEBUG_PRINT_IO_CORE__
22#endif
23
24#include "io/types.hfa"
25#include "kernel.hfa"
26
27#if !defined(CFA_HAVE_LINUX_IO_URING_H)
28        void __kernel_io_startup() {
29                // Nothing to do without io_uring
30        }
31
32        void __kernel_io_shutdown() {
33                // Nothing to do without io_uring
34        }
35
36        void ?{}(io_context_params & this) {}
37
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#else
45        #include <errno.h>
46        #include <stdint.h>
47        #include <string.h>
48        #include <signal.h>
49        #include <unistd.h>
50
51        extern "C" {
52                #include <pthread.h>
53                #include <sys/epoll.h>
54                #include <sys/mman.h>
55                #include <sys/syscall.h>
56
57                #include <linux/io_uring.h>
58        }
59
60        #include "bitmanip.hfa"
61        #include "kernel_private.hfa"
62        #include "thread.hfa"
63
64        void ?{}(io_context_params & this) {
65                this.num_entries = 256;
66                this.num_ready = 256;
67                this.submit_aff = -1;
68                this.eager_submits = false;
69                this.poller_submits = false;
70                this.poll_submit = false;
71                this.poll_complete = false;
72        }
73
74        static void * __io_poller_slow( void * arg );
75
76        // Weirdly, some systems that do support io_uring don't actually define these
77        #ifdef __alpha__
78                /*
79                * alpha is the only exception, all other architectures
80                * have common numbers for new system calls.
81                */
82                #ifndef __NR_io_uring_setup
83                        #define __NR_io_uring_setup           535
84                #endif
85                #ifndef __NR_io_uring_enter
86                        #define __NR_io_uring_enter           536
87                #endif
88                #ifndef __NR_io_uring_register
89                        #define __NR_io_uring_register        537
90                #endif
91        #else /* !__alpha__ */
92                #ifndef __NR_io_uring_setup
93                        #define __NR_io_uring_setup           425
94                #endif
95                #ifndef __NR_io_uring_enter
96                        #define __NR_io_uring_enter           426
97                #endif
98                #ifndef __NR_io_uring_register
99                        #define __NR_io_uring_register        427
100                #endif
101        #endif
102
103//=============================================================================================
104// I/O Startup / Shutdown logic + Master Poller
105//=============================================================================================
106
107        // IO Master poller loop forward
108        static void * iopoll_loop( __attribute__((unused)) void * args );
109
110        static struct {
111                pthread_t     thrd;    // pthread handle to io poller thread
112                void *        stack;   // pthread stack for io poller thread
113                int           epollfd; // file descriptor to the epoll instance
114                volatile bool run;     // Whether or not to continue
115        } iopoll;
116
117        void __kernel_io_startup(void) {
118                __cfadbg_print_safe(io_core, "Kernel : Creating EPOLL instance\n" );
119
120                iopoll.epollfd = epoll_create1(0);
121                if (iopoll.epollfd == -1) {
122                        abort( "internal error, epoll_create1\n");
123                }
124
125                __cfadbg_print_safe(io_core, "Kernel : Starting io poller thread\n" );
126
127                iopoll.run = true;
128                iopoll.stack = __create_pthread( &iopoll.thrd, iopoll_loop, 0p );
129        }
130
131        void __kernel_io_shutdown(void) {
132                // Notify the io poller thread of the shutdown
133                iopoll.run = false;
134                sigval val = { 1 };
135                pthread_sigqueue( iopoll.thrd, SIGUSR1, val );
136
137                // Wait for the io poller thread to finish
138
139                __destroy_pthread( iopoll.thrd, iopoll.stack, 0p );
140
141                int ret = close(iopoll.epollfd);
142                if (ret == -1) {
143                        abort( "internal error, close epoll\n");
144                }
145
146                // Io polling is now fully stopped
147
148                __cfadbg_print_safe(io_core, "Kernel : IO poller stopped\n" );
149        }
150
151        static void * iopoll_loop( __attribute__((unused)) void * args ) {
152                __processor_id_t id;
153                id.full_proc = false;
154                id.id = doregister(&id);
155                __cfaabi_tls.this_proc_id = &id;
156                __cfadbg_print_safe(io_core, "Kernel : IO poller thread starting\n" );
157
158                // Block signals to control when they arrive
159                sigset_t mask;
160                sigfillset(&mask);
161                if ( pthread_sigmask( SIG_BLOCK, &mask, 0p ) == -1 ) {
162                abort( "internal error, pthread_sigmask" );
163                }
164
165                sigdelset( &mask, SIGUSR1 );
166
167                // Create sufficient events
168                struct epoll_event events[10];
169                // Main loop
170                while( iopoll.run ) {
171                        // Wait for events
172                        int nfds = epoll_pwait( iopoll.epollfd, events, 10, -1, &mask );
173
174                        // Check if an error occured
175                        if (nfds == -1) {
176                                if( errno == EINTR ) continue;
177                                abort( "internal error, pthread_sigmask" );
178                        }
179
180                        for(i; nfds) {
181                                $io_ctx_thread * io_ctx = ($io_ctx_thread *)(uintptr_t)events[i].data.u64;
182                                /* paranoid */ verify( io_ctx );
183                                __cfadbg_print_safe(io_core, "Kernel I/O : Unparking io poller %p\n", io_ctx);
184                                #if !defined( __CFA_NO_STATISTICS__ )
185                                        __cfaabi_tls.this_stats = io_ctx->self.curr_cluster->stats;
186                                #endif
187                                post( io_ctx->sem );
188                        }
189                }
190
191                __cfadbg_print_safe(io_core, "Kernel : IO poller thread stopping\n" );
192                unregister(&id);
193                return 0p;
194        }
195
196//=============================================================================================
197// I/O Context Constrution/Destruction
198//=============================================================================================
199
200        void ?{}($io_ctx_thread & this, struct cluster & cl) { (this.self){ "IO Poller", cl }; }
201        void main( $io_ctx_thread & this );
202        static inline $thread * get_thread( $io_ctx_thread & this ) { return &this.self; }
203        void ^?{}( $io_ctx_thread & mutex this ) {}
204
205        static void __io_create ( __io_data & this, const io_context_params & params_in );
206        static void __io_destroy( __io_data & this );
207
208        void ?{}(io_context & this, struct cluster & cl, const io_context_params & params) {
209                (this.thrd){ cl };
210                this.thrd.ring = malloc();
211                __cfadbg_print_safe(io_core, "Kernel I/O : Creating ring for io_context %p\n", &this);
212                __io_create( *this.thrd.ring, params );
213
214                __cfadbg_print_safe(io_core, "Kernel I/O : Starting poller thread for io_context %p\n", &this);
215                this.thrd.done = false;
216                __thrd_start( this.thrd, main );
217
218                __cfadbg_print_safe(io_core, "Kernel I/O : io_context %p ready\n", &this);
219        }
220
221        void ?{}(io_context & this, struct cluster & cl) {
222                io_context_params params;
223                (this){ cl, params };
224        }
225
226        void ^?{}(io_context & this, bool cluster_context) {
227                __cfadbg_print_safe(io_core, "Kernel I/O : tearing down io_context %p\n", &this);
228
229                // Notify the thread of the shutdown
230                __atomic_store_n(&this.thrd.done, true, __ATOMIC_SEQ_CST);
231
232                // If this is an io_context within a cluster, things get trickier
233                $thread & thrd = this.thrd.self;
234                if( cluster_context ) {
235                        cluster & cltr = *thrd.curr_cluster;
236                        /* paranoid */ verify( cltr.idles.total == 0 || &cltr == mainCluster );
237                        /* paranoid */ verify( !ready_mutate_islocked() );
238
239                        // We need to adjust the clean-up based on where the thread is
240                        if( thrd.state == Ready || thrd.preempted != __NO_PREEMPTION ) {
241
242                                ready_schedule_lock();
243
244                                        // This is the tricky case
245                                        // The thread was preempted and now it is on the ready queue
246                                        // The thread should be the last on the list
247                                        /* paranoid */ verify( thrd.link.next != 0p );
248
249                                        // Remove the thread from the ready queue of this cluster
250                                        __attribute__((unused)) bool removed = remove_head( &cltr, &thrd );
251                                        /* paranoid */ verify( removed );
252                                        thrd.link.next = 0p;
253                                        thrd.link.prev = 0p;
254
255                                        // Fixup the thread state
256                                        thrd.state = Blocked;
257                                        thrd.ticket = TICKET_BLOCKED;
258                                        thrd.preempted = __NO_PREEMPTION;
259
260                                ready_schedule_unlock();
261
262                                // Pretend like the thread was blocked all along
263                        }
264                        // !!! This is not an else if !!!
265                        if( thrd.state == Blocked ) {
266
267                                // This is the "easy case"
268                                // The thread is parked and can easily be moved to active cluster
269                                verify( thrd.curr_cluster != active_cluster() || thrd.curr_cluster == mainCluster );
270                                thrd.curr_cluster = active_cluster();
271
272                                // unpark the fast io_poller
273                                unpark( &thrd );
274                        }
275                        else {
276
277                                // The thread is in a weird state
278                                // I don't know what to do here
279                                abort("io_context poller thread is in unexpected state, cannot clean-up correctly\n");
280                        }
281                } else {
282                        post( this.thrd.sem );
283                }
284
285                ^(this.thrd){};
286                __cfadbg_print_safe(io_core, "Kernel I/O : Stopped poller thread for io_context %p\n", &this);
287
288                __io_destroy( *this.thrd.ring );
289                __cfadbg_print_safe(io_core, "Kernel I/O : Destroyed ring for io_context %p\n", &this);
290
291                free(this.thrd.ring);
292        }
293
294        void ^?{}(io_context & this) {
295                ^(this){ false };
296        }
297
298        static void __io_create( __io_data & this, const io_context_params & params_in ) {
299                // Step 1 : call to setup
300                struct io_uring_params params;
301                memset(&params, 0, sizeof(params));
302                if( params_in.poll_submit   ) params.flags |= IORING_SETUP_SQPOLL;
303                if( params_in.poll_complete ) params.flags |= IORING_SETUP_IOPOLL;
304
305                __u32 nentries = params_in.num_entries != 0 ? params_in.num_entries : 256;
306                if( !is_pow2(nentries) ) {
307                        abort("ERROR: I/O setup 'num_entries' must be a power of 2\n");
308                }
309                if( params_in.poller_submits && params_in.eager_submits ) {
310                        abort("ERROR: I/O setup 'poller_submits' and 'eager_submits' cannot be used together\n");
311                }
312
313                int fd = syscall(__NR_io_uring_setup, nentries, &params );
314                if(fd < 0) {
315                        abort("KERNEL ERROR: IO_URING SETUP - %s\n", strerror(errno));
316                }
317
318                // Step 2 : mmap result
319                memset( &this, 0, sizeof(struct __io_data) );
320                struct __submition_data  & sq = this.submit_q;
321                struct __completion_data & cq = this.completion_q;
322
323                // calculate the right ring size
324                sq.ring_sz = params.sq_off.array + (params.sq_entries * sizeof(unsigned)           );
325                cq.ring_sz = params.cq_off.cqes  + (params.cq_entries * sizeof(struct io_uring_cqe));
326
327                // Requires features
328                #if defined(IORING_FEAT_SINGLE_MMAP)
329                        // adjust the size according to the parameters
330                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
331                                cq.ring_sz = sq.ring_sz = max(cq.ring_sz, sq.ring_sz);
332                        }
333                #endif
334
335                // mmap the Submit Queue into existence
336                sq.ring_ptr = mmap(0, sq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQ_RING);
337                if (sq.ring_ptr == (void*)MAP_FAILED) {
338                        abort("KERNEL ERROR: IO_URING MMAP1 - %s\n", strerror(errno));
339                }
340
341                // Requires features
342                #if defined(IORING_FEAT_SINGLE_MMAP)
343                        // mmap the Completion Queue into existence (may or may not be needed)
344                        if ((params.features & IORING_FEAT_SINGLE_MMAP) != 0) {
345                                cq.ring_ptr = sq.ring_ptr;
346                        }
347                        else
348                #endif
349                {
350                        // We need multiple call to MMAP
351                        cq.ring_ptr = mmap(0, cq.ring_sz, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_CQ_RING);
352                        if (cq.ring_ptr == (void*)MAP_FAILED) {
353                                munmap(sq.ring_ptr, sq.ring_sz);
354                                abort("KERNEL ERROR: IO_URING MMAP2 - %s\n", strerror(errno));
355                        }
356                }
357
358                // mmap the submit queue entries
359                size_t size = params.sq_entries * sizeof(struct io_uring_sqe);
360                sq.sqes = (struct io_uring_sqe *)mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_POPULATE, fd, IORING_OFF_SQES);
361                if (sq.sqes == (struct io_uring_sqe *)MAP_FAILED) {
362                        munmap(sq.ring_ptr, sq.ring_sz);
363                        if (cq.ring_ptr != sq.ring_ptr) munmap(cq.ring_ptr, cq.ring_sz);
364                        abort("KERNEL ERROR: IO_URING MMAP3 - %s\n", strerror(errno));
365                }
366
367                // Get the pointers from the kernel to fill the structure
368                // submit queue
369                sq.head    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.head);
370                sq.tail    = (volatile __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.tail);
371                sq.mask    = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_mask);
372                sq.num     = (   const __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.ring_entries);
373                sq.flags   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.flags);
374                sq.dropped = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.dropped);
375                sq.array   = (         __u32 *)(((intptr_t)sq.ring_ptr) + params.sq_off.array);
376                sq.prev_head = *sq.head;
377
378                {
379                        const __u32 num = *sq.num;
380                        for( i; num ) {
381                                sq.sqes[i].user_data = 0ul64;
382                        }
383                }
384
385                (sq.submit_lock){};
386                (sq.release_lock){};
387
388                if( params_in.poller_submits || params_in.eager_submits ) {
389                        /* paranoid */ verify( is_pow2( params_in.num_ready ) || (params_in.num_ready < 8) );
390                        sq.ready_cnt = max( params_in.num_ready, 8 );
391                        sq.ready = alloc( sq.ready_cnt, 64`align );
392                        for(i; sq.ready_cnt) {
393                                sq.ready[i] = -1ul32;
394                        }
395                        sq.prev_ready = 0;
396                }
397                else {
398                        sq.ready_cnt = 0;
399                        sq.ready = 0p;
400                        sq.prev_ready = 0;
401                }
402
403                // completion queue
404                cq.head      = (volatile __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.head);
405                cq.tail      = (volatile __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.tail);
406                cq.mask      = (   const __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_mask);
407                cq.num       = (   const __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.ring_entries);
408                cq.overflow  = (         __u32 *)(((intptr_t)cq.ring_ptr) + params.cq_off.overflow);
409                cq.cqes = (struct io_uring_cqe *)(((intptr_t)cq.ring_ptr) + params.cq_off.cqes);
410
411                // some paranoid checks
412                /* paranoid */ verifyf( (*cq.mask) == ((*cq.num) - 1ul32), "IO_URING Expected mask to be %u (%u entries), was %u", (*cq.num) - 1ul32, *cq.num, *cq.mask  );
413                /* paranoid */ verifyf( (*cq.num)  >= nentries, "IO_URING Expected %u entries, got %u", nentries, *cq.num );
414                /* paranoid */ verifyf( (*cq.head) == 0, "IO_URING Expected head to be 0, got %u", *cq.head );
415                /* paranoid */ verifyf( (*cq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *cq.tail );
416
417                /* 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 );
418                /* paranoid */ verifyf( (*sq.num) >= nentries, "IO_URING Expected %u entries, got %u", nentries, *sq.num );
419                /* paranoid */ verifyf( (*sq.head) == 0, "IO_URING Expected head to be 0, got %u", *sq.head );
420                /* paranoid */ verifyf( (*sq.tail) == 0, "IO_URING Expected tail to be 0, got %u", *sq.tail );
421
422                // Update the global ring info
423                this.ring_flags = params.flags;
424                this.fd         = fd;
425                this.eager_submits  = params_in.eager_submits;
426                this.poller_submits = params_in.poller_submits;
427        }
428
429        static void __io_destroy( __io_data & this ) {
430                // Shutdown the io rings
431                struct __submition_data  & sq = this.submit_q;
432                struct __completion_data & cq = this.completion_q;
433
434                // unmap the submit queue entries
435                munmap(sq.sqes, (*sq.num) * sizeof(struct io_uring_sqe));
436
437                // unmap the Submit Queue ring
438                munmap(sq.ring_ptr, sq.ring_sz);
439
440                // unmap the Completion Queue ring, if it is different
441                if (cq.ring_ptr != sq.ring_ptr) {
442                        munmap(cq.ring_ptr, cq.ring_sz);
443                }
444
445                // close the file descriptor
446                close(this.fd);
447
448                free( this.submit_q.ready ); // Maybe null, doesn't matter
449        }
450
451//=============================================================================================
452// I/O Context Sleep
453//=============================================================================================
454
455        void __ioctx_register($io_ctx_thread & ctx, struct epoll_event & ev) {
456                ev.events = EPOLLIN | EPOLLONESHOT;
457                ev.data.u64 = (__u64)&ctx;
458                int ret = epoll_ctl(iopoll.epollfd, EPOLL_CTL_ADD, ctx.ring->fd, &ev);
459                if (ret < 0) {
460                        abort( "KERNEL ERROR: EPOLL ADD - (%d) %s\n", (int)errno, strerror(errno) );
461                }
462        }
463
464        void __ioctx_prepare_block($io_ctx_thread & ctx, struct epoll_event & ev) {
465                int ret = epoll_ctl(iopoll.epollfd, EPOLL_CTL_MOD, ctx.ring->fd, &ev);
466                if (ret < 0) {
467                        abort( "KERNEL ERROR: EPOLL REARM - (%d) %s\n", (int)errno, strerror(errno) );
468                }
469        }
470
471//=============================================================================================
472// I/O Context Misc Setup
473//=============================================================================================
474        void register_fixed_files( io_context & ctx, int * files, unsigned count ) {
475                int ret = syscall( __NR_io_uring_register, ctx.thrd.ring->fd, IORING_REGISTER_FILES, files, count );
476                if( ret < 0 ) {
477                        abort( "KERNEL ERROR: IO_URING SYSCALL - (%d) %s\n", (int)errno, strerror(errno) );
478                }
479
480                __cfadbg_print_safe( io_core, "Kernel I/O : Performed io_register for %p, returned %d\n", active_thread(), ret );
481        }
482
483        void register_fixed_files( cluster & cltr, int * files, unsigned count ) {
484                for(i; cltr.io.cnt) {
485                        register_fixed_files( cltr.io.ctxs[i], files, count );
486                }
487        }
488#endif
Note: See TracBrowser for help on using the repository browser.