Line data Source code
1 : /* SPDX-License-Identifier: BSD-3-Clause
2 : * Copyright (C) 2016 Intel Corporation.
3 : * All rights reserved.
4 : * Copyright (c) 2022, 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
5 : */
6 :
7 : #include "spdk/stdinc.h"
8 :
9 : #include "spdk/env.h"
10 : #include "spdk/likely.h"
11 : #include "spdk/queue.h"
12 : #include "spdk/string.h"
13 : #include "spdk/thread.h"
14 : #include "spdk/trace.h"
15 : #include "spdk/util.h"
16 : #include "spdk/fd_group.h"
17 :
18 : #include "spdk/log.h"
19 : #include "spdk_internal/thread.h"
20 : #include "spdk_internal/usdt.h"
21 : #include "thread_internal.h"
22 :
23 : #include "spdk_internal/trace_defs.h"
24 :
25 : #ifdef __linux__
26 : #include <sys/timerfd.h>
27 : #include <sys/eventfd.h>
28 : #endif
29 :
30 : #ifdef SPDK_HAVE_EXECINFO_H
31 : #include <execinfo.h>
32 : #endif
33 :
34 : #define SPDK_MSG_BATCH_SIZE 8
35 : #define SPDK_MAX_DEVICE_NAME_LEN 256
36 : #define SPDK_THREAD_EXIT_TIMEOUT_SEC 5
37 : #define SPDK_MAX_POLLER_NAME_LEN 256
38 : #define SPDK_MAX_THREAD_NAME_LEN 256
39 :
40 : static struct spdk_thread *g_app_thread;
41 :
42 : struct spdk_interrupt {
43 : int efd;
44 : struct spdk_fd_group *fgrp;
45 : struct spdk_thread *thread;
46 : spdk_interrupt_fn fn;
47 : void *arg;
48 : char name[SPDK_MAX_POLLER_NAME_LEN + 1];
49 : };
50 :
51 : enum spdk_poller_state {
52 : /* The poller is registered with a thread but not currently executing its fn. */
53 : SPDK_POLLER_STATE_WAITING,
54 :
55 : /* The poller is currently running its fn. */
56 : SPDK_POLLER_STATE_RUNNING,
57 :
58 : /* The poller was unregistered during the execution of its fn. */
59 : SPDK_POLLER_STATE_UNREGISTERED,
60 :
61 : /* The poller is in the process of being paused. It will be paused
62 : * during the next time it's supposed to be executed.
63 : */
64 : SPDK_POLLER_STATE_PAUSING,
65 :
66 : /* The poller is registered but currently paused. It's on the
67 : * paused_pollers list.
68 : */
69 : SPDK_POLLER_STATE_PAUSED,
70 : };
71 :
72 : struct spdk_poller {
73 : TAILQ_ENTRY(spdk_poller) tailq;
74 : RB_ENTRY(spdk_poller) node;
75 :
76 : /* Current state of the poller; should only be accessed from the poller's thread. */
77 : enum spdk_poller_state state;
78 :
79 : uint64_t period_ticks;
80 : uint64_t next_run_tick;
81 : uint64_t run_count;
82 : uint64_t busy_count;
83 : uint64_t id;
84 : spdk_poller_fn fn;
85 : void *arg;
86 : struct spdk_thread *thread;
87 : struct spdk_interrupt *intr;
88 : spdk_poller_set_interrupt_mode_cb set_intr_cb_fn;
89 : void *set_intr_cb_arg;
90 :
91 : char name[SPDK_MAX_POLLER_NAME_LEN + 1];
92 : };
93 :
94 : enum spdk_thread_state {
95 : /* The thread is processing poller and message by spdk_thread_poll(). */
96 : SPDK_THREAD_STATE_RUNNING,
97 :
98 : /* The thread is in the process of termination. It reaps unregistering
99 : * poller are releasing I/O channel.
100 : */
101 : SPDK_THREAD_STATE_EXITING,
102 :
103 : /* The thread is exited. It is ready to call spdk_thread_destroy(). */
104 : SPDK_THREAD_STATE_EXITED,
105 : };
106 :
107 : struct spdk_thread_post_poller_handler {
108 : spdk_post_poller_fn fn;
109 : void *fn_arg;
110 : };
111 :
112 : #define SPDK_THREAD_MAX_POST_POLLER_HANDLERS (4)
113 :
114 : struct spdk_thread {
115 : uint64_t tsc_last;
116 : struct spdk_thread_stats stats;
117 : /*
118 : * Contains pollers actively running on this thread. Pollers
119 : * are run round-robin. The thread takes one poller from the head
120 : * of the ring, executes it, then puts it back at the tail of
121 : * the ring.
122 : */
123 : TAILQ_HEAD(active_pollers_head, spdk_poller) active_pollers;
124 : /**
125 : * Contains pollers running on this thread with a periodic timer.
126 : */
127 : RB_HEAD(timed_pollers_tree, spdk_poller) timed_pollers;
128 : struct spdk_poller *first_timed_poller;
129 : /*
130 : * Contains paused pollers. Pollers on this queue are waiting until
131 : * they are resumed (in which case they're put onto the active/timer
132 : * queues) or unregistered.
133 : */
134 : TAILQ_HEAD(paused_pollers_head, spdk_poller) paused_pollers;
135 : struct spdk_thread_post_poller_handler pp_handlers[SPDK_THREAD_MAX_POST_POLLER_HANDLERS];
136 : struct spdk_ring *messages;
137 : uint8_t num_pp_handlers;
138 : int msg_fd;
139 : SLIST_HEAD(, spdk_msg) msg_cache;
140 : size_t msg_cache_count;
141 : spdk_msg_fn critical_msg;
142 : uint64_t id;
143 : uint64_t next_poller_id;
144 : enum spdk_thread_state state;
145 : int pending_unregister_count;
146 : uint32_t for_each_count;
147 :
148 : RB_HEAD(io_channel_tree, spdk_io_channel) io_channels;
149 : TAILQ_ENTRY(spdk_thread) tailq;
150 :
151 : char name[SPDK_MAX_THREAD_NAME_LEN + 1];
152 : struct spdk_cpuset cpumask;
153 : uint64_t exit_timeout_tsc;
154 :
155 : int32_t lock_count;
156 :
157 : /* spdk_thread is bound to current CPU core. */
158 : bool is_bound;
159 :
160 : /* Indicates whether this spdk_thread currently runs in interrupt. */
161 : bool in_interrupt;
162 : bool poller_unregistered;
163 : struct spdk_fd_group *fgrp;
164 :
165 : uint16_t trace_id;
166 :
167 : uint8_t reserved[6];
168 :
169 : /* User context allocated at the end */
170 : uint8_t ctx[0];
171 : };
172 :
173 : /*
174 : * Assert that spdk_thread struct is 8 byte aligned to ensure
175 : * the user ctx is also 8-byte aligned.
176 : */
177 : SPDK_STATIC_ASSERT((sizeof(struct spdk_thread)) % 8 == 0, "Incorrect size");
178 :
179 : static pthread_mutex_t g_devlist_mutex = PTHREAD_MUTEX_INITIALIZER;
180 :
181 : static spdk_new_thread_fn g_new_thread_fn = NULL;
182 : static spdk_thread_op_fn g_thread_op_fn = NULL;
183 : static spdk_thread_op_supported_fn g_thread_op_supported_fn;
184 : static size_t g_ctx_sz = 0;
185 : /* Monotonic increasing ID is set to each created thread beginning at 1. Once the
186 : * ID exceeds UINT64_MAX, further thread creation is not allowed and restarting
187 : * SPDK application is required.
188 : */
189 : static uint64_t g_thread_id = 1;
190 :
191 : enum spin_error {
192 : SPIN_ERR_NONE,
193 : /* Trying to use an SPDK lock while not on an SPDK thread */
194 : SPIN_ERR_NOT_SPDK_THREAD,
195 : /* Trying to lock a lock already held by this SPDK thread */
196 : SPIN_ERR_DEADLOCK,
197 : /* Trying to unlock a lock not held by this SPDK thread */
198 : SPIN_ERR_WRONG_THREAD,
199 : /* pthread_spin_*() returned an error */
200 : SPIN_ERR_PTHREAD,
201 : /* Trying to destroy a lock that is held */
202 : SPIN_ERR_LOCK_HELD,
203 : /* lock_count is invalid */
204 : SPIN_ERR_LOCK_COUNT,
205 : /*
206 : * An spdk_thread may migrate to another pthread. A spinlock held across migration leads to
207 : * undefined behavior. A spinlock held when an SPDK thread goes off CPU would lead to
208 : * deadlock when another SPDK thread on the same pthread tries to take that lock.
209 : */
210 : SPIN_ERR_HOLD_DURING_SWITCH,
211 : /* Trying to use a lock that was destroyed (but not re-initialized) */
212 : SPIN_ERR_DESTROYED,
213 : /* Trying to use a lock that is not initialized */
214 : SPIN_ERR_NOT_INITIALIZED,
215 :
216 : /* Must be last, not an actual error code */
217 : SPIN_ERR_LAST
218 : };
219 :
220 : static const char *spin_error_strings[] = {
221 : [SPIN_ERR_NONE] = "No error",
222 : [SPIN_ERR_NOT_SPDK_THREAD] = "Not an SPDK thread",
223 : [SPIN_ERR_DEADLOCK] = "Deadlock detected",
224 : [SPIN_ERR_WRONG_THREAD] = "Unlock on wrong SPDK thread",
225 : [SPIN_ERR_PTHREAD] = "Error from pthread_spinlock",
226 : [SPIN_ERR_LOCK_HELD] = "Destroying a held spinlock",
227 : [SPIN_ERR_LOCK_COUNT] = "Lock count is invalid",
228 : [SPIN_ERR_HOLD_DURING_SWITCH] = "Lock(s) held while SPDK thread going off CPU",
229 : [SPIN_ERR_DESTROYED] = "Lock has been destroyed",
230 : [SPIN_ERR_NOT_INITIALIZED] = "Lock has not been initialized",
231 : };
232 :
233 : #define SPIN_ERROR_STRING(err) (err < 0 || err >= SPDK_COUNTOF(spin_error_strings)) \
234 : ? "Unknown error" : spin_error_strings[err]
235 :
236 : static void
237 0 : __posix_abort(enum spin_error err)
238 : {
239 0 : abort();
240 : }
241 :
242 : typedef void (*spin_abort)(enum spin_error err);
243 : spin_abort g_spin_abort_fn = __posix_abort;
244 :
245 : #define SPIN_ASSERT_IMPL(cond, err, extra_log, ret) \
246 : do { \
247 : if (spdk_unlikely(!(cond))) { \
248 : SPDK_ERRLOG("unrecoverable spinlock error %d: %s (%s)\n", err, \
249 : SPIN_ERROR_STRING(err), #cond); \
250 : extra_log; \
251 : g_spin_abort_fn(err); \
252 : ret; \
253 : } \
254 : } while (0)
255 : #define SPIN_ASSERT_LOG_STACKS(cond, err, lock) \
256 : SPIN_ASSERT_IMPL(cond, err, sspin_stacks_print(sspin), return)
257 : #define SPIN_ASSERT_RETURN(cond, err, ret) SPIN_ASSERT_IMPL(cond, err, , return ret)
258 : #define SPIN_ASSERT(cond, err) SPIN_ASSERT_IMPL(cond, err, ,)
259 :
260 : struct io_device {
261 : void *io_device;
262 : char name[SPDK_MAX_DEVICE_NAME_LEN + 1];
263 : spdk_io_channel_create_cb create_cb;
264 : spdk_io_channel_destroy_cb destroy_cb;
265 : spdk_io_device_unregister_cb unregister_cb;
266 : struct spdk_thread *unregister_thread;
267 : uint32_t ctx_size;
268 : uint32_t for_each_count;
269 : RB_ENTRY(io_device) node;
270 :
271 : uint32_t refcnt;
272 :
273 : bool pending_unregister;
274 : bool unregistered;
275 : };
276 :
277 : static RB_HEAD(io_device_tree, io_device) g_io_devices = RB_INITIALIZER(g_io_devices);
278 :
279 : static int
280 15165 : io_device_cmp(struct io_device *dev1, struct io_device *dev2)
281 : {
282 15165 : return (dev1->io_device < dev2->io_device ? -1 : dev1->io_device > dev2->io_device);
283 : }
284 :
285 26811 : RB_GENERATE_STATIC(io_device_tree, io_device, node, io_device_cmp);
286 :
287 : static int
288 9704 : io_channel_cmp(struct spdk_io_channel *ch1, struct spdk_io_channel *ch2)
289 : {
290 9704 : return (ch1->dev < ch2->dev ? -1 : ch1->dev > ch2->dev);
291 : }
292 :
293 27523 : RB_GENERATE_STATIC(io_channel_tree, spdk_io_channel, node, io_channel_cmp);
294 :
295 : struct spdk_msg {
296 : spdk_msg_fn fn;
297 : void *arg;
298 :
299 : SLIST_ENTRY(spdk_msg) link;
300 : };
301 :
302 : static struct spdk_mempool *g_spdk_msg_mempool = NULL;
303 :
304 : static TAILQ_HEAD(, spdk_thread) g_threads = TAILQ_HEAD_INITIALIZER(g_threads);
305 : static uint32_t g_thread_count = 0;
306 :
307 : static __thread struct spdk_thread *tls_thread = NULL;
308 :
309 : static void
310 0 : thread_trace(void)
311 : {
312 0 : struct spdk_trace_tpoint_opts opts[] = {
313 : {
314 : "THREAD_IOCH_GET", TRACE_THREAD_IOCH_GET,
315 : OWNER_TYPE_NONE, OBJECT_NONE, 0,
316 : {{ "refcnt", SPDK_TRACE_ARG_TYPE_INT, 4 }}
317 : },
318 : {
319 : "THREAD_IOCH_PUT", TRACE_THREAD_IOCH_PUT,
320 : OWNER_TYPE_NONE, OBJECT_NONE, 0,
321 : {{ "refcnt", SPDK_TRACE_ARG_TYPE_INT, 4 }}
322 : }
323 : };
324 :
325 0 : spdk_trace_register_owner_type(OWNER_TYPE_THREAD, 't');
326 0 : spdk_trace_register_description_ext(opts, SPDK_COUNTOF(opts));
327 0 : }
328 49 : SPDK_TRACE_REGISTER_FN(thread_trace, "thread", TRACE_GROUP_THREAD)
329 :
330 : /*
331 : * If this compare function returns zero when two next_run_ticks are equal,
332 : * the macro RB_INSERT() returns a pointer to the element with the same
333 : * next_run_tick.
334 : *
335 : * Fortunately, the macro RB_REMOVE() takes not a key but a pointer to the element
336 : * to remove as a parameter.
337 : *
338 : * Hence we allow RB_INSERT() to insert elements with the same keys on the right
339 : * side by returning 1 when two next_run_ticks are equal.
340 : */
341 : static inline int
342 909 : timed_poller_compare(struct spdk_poller *poller1, struct spdk_poller *poller2)
343 : {
344 909 : if (poller1->next_run_tick < poller2->next_run_tick) {
345 502 : return -1;
346 : } else {
347 407 : return 1;
348 : }
349 : }
350 :
351 5013 : RB_GENERATE_STATIC(timed_pollers_tree, spdk_poller, node, timed_poller_compare);
352 :
353 : static inline struct spdk_thread *
354 935689 : _get_thread(void)
355 : {
356 935689 : return tls_thread;
357 : }
358 :
359 : static int
360 98 : _thread_lib_init(size_t ctx_sz, size_t msg_mempool_sz)
361 : {
362 : char mempool_name[SPDK_MAX_MEMZONE_NAME_LEN];
363 :
364 98 : g_ctx_sz = ctx_sz;
365 :
366 98 : snprintf(mempool_name, sizeof(mempool_name), "msgpool_%d", getpid());
367 98 : g_spdk_msg_mempool = spdk_mempool_create(mempool_name, msg_mempool_sz,
368 : sizeof(struct spdk_msg),
369 : 0, /* No cache. We do our own. */
370 : SPDK_ENV_NUMA_ID_ANY);
371 :
372 98 : SPDK_DEBUGLOG(thread, "spdk_msg_mempool was created with size: %zu\n",
373 : msg_mempool_sz);
374 :
375 98 : if (!g_spdk_msg_mempool) {
376 0 : SPDK_ERRLOG("spdk_msg_mempool creation failed\n");
377 0 : return -ENOMEM;
378 : }
379 :
380 98 : return 0;
381 : }
382 :
383 : static void thread_interrupt_destroy(struct spdk_thread *thread);
384 : static int thread_interrupt_create(struct spdk_thread *thread);
385 :
386 : static void
387 181 : _free_thread(struct spdk_thread *thread)
388 : {
389 : struct spdk_io_channel *ch;
390 : struct spdk_msg *msg;
391 : struct spdk_poller *poller, *ptmp;
392 :
393 181 : RB_FOREACH(ch, io_channel_tree, &thread->io_channels) {
394 0 : SPDK_ERRLOG("thread %s still has channel for io_device %s\n",
395 : thread->name, ch->dev->name);
396 : }
397 :
398 181 : TAILQ_FOREACH_SAFE(poller, &thread->active_pollers, tailq, ptmp) {
399 0 : if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) {
400 0 : SPDK_WARNLOG("active_poller %s still registered at thread exit\n",
401 : poller->name);
402 : }
403 0 : TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
404 0 : free(poller);
405 : }
406 :
407 209 : RB_FOREACH_SAFE(poller, timed_pollers_tree, &thread->timed_pollers, ptmp) {
408 28 : if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) {
409 0 : SPDK_WARNLOG("timed_poller %s still registered at thread exit\n",
410 : poller->name);
411 : }
412 28 : RB_REMOVE(timed_pollers_tree, &thread->timed_pollers, poller);
413 28 : free(poller);
414 : }
415 :
416 181 : TAILQ_FOREACH_SAFE(poller, &thread->paused_pollers, tailq, ptmp) {
417 0 : SPDK_WARNLOG("paused_poller %s still registered at thread exit\n", poller->name);
418 0 : TAILQ_REMOVE(&thread->paused_pollers, poller, tailq);
419 0 : free(poller);
420 : }
421 :
422 181 : pthread_mutex_lock(&g_devlist_mutex);
423 181 : assert(g_thread_count > 0);
424 181 : g_thread_count--;
425 181 : TAILQ_REMOVE(&g_threads, thread, tailq);
426 181 : pthread_mutex_unlock(&g_devlist_mutex);
427 :
428 181 : msg = SLIST_FIRST(&thread->msg_cache);
429 185377 : while (msg != NULL) {
430 185196 : SLIST_REMOVE_HEAD(&thread->msg_cache, link);
431 :
432 185196 : assert(thread->msg_cache_count > 0);
433 185196 : thread->msg_cache_count--;
434 185196 : spdk_mempool_put(g_spdk_msg_mempool, msg);
435 :
436 185196 : msg = SLIST_FIRST(&thread->msg_cache);
437 : }
438 :
439 181 : assert(thread->msg_cache_count == 0);
440 :
441 181 : if (spdk_interrupt_mode_is_enabled()) {
442 0 : thread_interrupt_destroy(thread);
443 : }
444 :
445 181 : spdk_ring_free(thread->messages);
446 181 : free(thread);
447 181 : }
448 :
449 : int
450 86 : spdk_thread_lib_init(spdk_new_thread_fn new_thread_fn, size_t ctx_sz)
451 : {
452 86 : assert(g_new_thread_fn == NULL);
453 86 : assert(g_thread_op_fn == NULL);
454 :
455 86 : if (new_thread_fn == NULL) {
456 84 : SPDK_INFOLOG(thread, "new_thread_fn was not specified at spdk_thread_lib_init\n");
457 : } else {
458 2 : g_new_thread_fn = new_thread_fn;
459 : }
460 :
461 86 : return _thread_lib_init(ctx_sz, SPDK_DEFAULT_MSG_MEMPOOL_SIZE);
462 : }
463 :
464 : int
465 12 : spdk_thread_lib_init_ext(spdk_thread_op_fn thread_op_fn,
466 : spdk_thread_op_supported_fn thread_op_supported_fn,
467 : size_t ctx_sz, size_t msg_mempool_sz)
468 : {
469 12 : assert(g_new_thread_fn == NULL);
470 12 : assert(g_thread_op_fn == NULL);
471 12 : assert(g_thread_op_supported_fn == NULL);
472 :
473 12 : if ((thread_op_fn != NULL) != (thread_op_supported_fn != NULL)) {
474 0 : SPDK_ERRLOG("Both must be defined or undefined together.\n");
475 0 : return -EINVAL;
476 : }
477 :
478 12 : if (thread_op_fn == NULL && thread_op_supported_fn == NULL) {
479 0 : SPDK_INFOLOG(thread, "thread_op_fn and thread_op_supported_fn were not specified\n");
480 : } else {
481 12 : g_thread_op_fn = thread_op_fn;
482 12 : g_thread_op_supported_fn = thread_op_supported_fn;
483 : }
484 :
485 12 : return _thread_lib_init(ctx_sz, msg_mempool_sz);
486 : }
487 :
488 : void
489 96 : spdk_thread_lib_fini(void)
490 : {
491 : struct io_device *dev;
492 :
493 97 : RB_FOREACH(dev, io_device_tree, &g_io_devices) {
494 1 : SPDK_ERRLOG("io_device %s not unregistered\n", dev->name);
495 : }
496 :
497 96 : g_new_thread_fn = NULL;
498 96 : g_thread_op_fn = NULL;
499 96 : g_thread_op_supported_fn = NULL;
500 96 : g_ctx_sz = 0;
501 96 : if (g_app_thread != NULL) {
502 92 : _free_thread(g_app_thread);
503 92 : g_app_thread = NULL;
504 : }
505 :
506 96 : if (g_spdk_msg_mempool) {
507 96 : spdk_mempool_free(g_spdk_msg_mempool);
508 96 : g_spdk_msg_mempool = NULL;
509 : }
510 96 : }
511 :
512 : struct spdk_thread *
513 218 : spdk_thread_create(const char *name, const struct spdk_cpuset *cpumask)
514 : {
515 : struct spdk_thread *thread, *null_thread;
516 218 : size_t size = SPDK_ALIGN_CEIL(sizeof(*thread) + g_ctx_sz, SPDK_CACHE_LINE_SIZE);
517 : struct spdk_msg *msgs[SPDK_MSG_MEMPOOL_CACHE_SIZE];
518 218 : int rc = 0, i;
519 :
520 : /* Since this spdk_thread object will be used by another core, ensure that it won't share a
521 : * cache line with any other object allocated on this core */
522 218 : rc = posix_memalign((void **)&thread, SPDK_CACHE_LINE_SIZE, size);
523 218 : if (rc != 0) {
524 0 : SPDK_ERRLOG("Unable to allocate memory for thread\n");
525 0 : return NULL;
526 : }
527 218 : memset(thread, 0, size);
528 :
529 218 : if (cpumask) {
530 23 : spdk_cpuset_copy(&thread->cpumask, cpumask);
531 : } else {
532 195 : spdk_cpuset_negate(&thread->cpumask);
533 : }
534 :
535 218 : RB_INIT(&thread->io_channels);
536 218 : TAILQ_INIT(&thread->active_pollers);
537 218 : RB_INIT(&thread->timed_pollers);
538 218 : TAILQ_INIT(&thread->paused_pollers);
539 218 : SLIST_INIT(&thread->msg_cache);
540 218 : thread->msg_cache_count = 0;
541 :
542 218 : thread->tsc_last = spdk_get_ticks();
543 :
544 : /* Monotonic increasing ID is set to each created poller beginning at 1. Once the
545 : * ID exceeds UINT64_MAX a warning message is logged
546 : */
547 218 : thread->next_poller_id = 1;
548 :
549 218 : thread->messages = spdk_ring_create(SPDK_RING_TYPE_MP_SC, 65536, SPDK_ENV_NUMA_ID_ANY);
550 218 : if (!thread->messages) {
551 0 : SPDK_ERRLOG("Unable to allocate memory for message ring\n");
552 0 : free(thread);
553 0 : return NULL;
554 : }
555 :
556 : /* Fill the local message pool cache. */
557 218 : rc = spdk_mempool_get_bulk(g_spdk_msg_mempool, (void **)msgs, SPDK_MSG_MEMPOOL_CACHE_SIZE);
558 218 : if (rc == 0) {
559 : /* If we can't populate the cache it's ok. The cache will get filled
560 : * up organically as messages are passed to the thread. */
561 223450 : for (i = 0; i < SPDK_MSG_MEMPOOL_CACHE_SIZE; i++) {
562 223232 : SLIST_INSERT_HEAD(&thread->msg_cache, msgs[i], link);
563 223232 : thread->msg_cache_count++;
564 : }
565 : }
566 :
567 218 : if (name) {
568 42 : snprintf(thread->name, sizeof(thread->name), "%s", name);
569 : } else {
570 176 : snprintf(thread->name, sizeof(thread->name), "%p", thread);
571 : }
572 :
573 218 : thread->trace_id = spdk_trace_register_owner(OWNER_TYPE_THREAD, thread->name);
574 :
575 218 : pthread_mutex_lock(&g_devlist_mutex);
576 218 : if (g_thread_id == 0) {
577 0 : SPDK_ERRLOG("Thread ID rolled over. Further thread creation is not allowed.\n");
578 0 : pthread_mutex_unlock(&g_devlist_mutex);
579 0 : _free_thread(thread);
580 0 : return NULL;
581 : }
582 218 : thread->id = g_thread_id++;
583 218 : TAILQ_INSERT_TAIL(&g_threads, thread, tailq);
584 218 : g_thread_count++;
585 218 : pthread_mutex_unlock(&g_devlist_mutex);
586 :
587 218 : SPDK_DEBUGLOG(thread, "Allocating new thread (%" PRIu64 ", %s)\n",
588 : thread->id, thread->name);
589 :
590 218 : if (spdk_interrupt_mode_is_enabled()) {
591 0 : thread->in_interrupt = true;
592 0 : rc = thread_interrupt_create(thread);
593 0 : if (rc != 0) {
594 0 : _free_thread(thread);
595 0 : return NULL;
596 : }
597 : }
598 :
599 218 : if (g_new_thread_fn) {
600 5 : rc = g_new_thread_fn(thread);
601 213 : } else if (g_thread_op_supported_fn && g_thread_op_supported_fn(SPDK_THREAD_OP_NEW)) {
602 16 : rc = g_thread_op_fn(thread, SPDK_THREAD_OP_NEW);
603 : }
604 :
605 218 : if (rc != 0) {
606 2 : _free_thread(thread);
607 2 : return NULL;
608 : }
609 :
610 216 : thread->state = SPDK_THREAD_STATE_RUNNING;
611 :
612 : /* If this is the first thread, save it as the app thread. Use an atomic
613 : * compare + exchange to guard against crazy users who might try to
614 : * call spdk_thread_create() simultaneously on multiple threads.
615 : */
616 216 : null_thread = NULL;
617 216 : __atomic_compare_exchange_n(&g_app_thread, &null_thread, thread, false,
618 : __ATOMIC_SEQ_CST, __ATOMIC_SEQ_CST);
619 :
620 216 : return thread;
621 : }
622 :
623 : struct spdk_thread *
624 1619 : spdk_thread_get_app_thread(void)
625 : {
626 1619 : return g_app_thread;
627 : }
628 :
629 : bool
630 156 : spdk_thread_is_app_thread(struct spdk_thread *thread)
631 : {
632 156 : if (thread == NULL) {
633 152 : thread = _get_thread();
634 : }
635 :
636 156 : return g_app_thread == thread;
637 : }
638 :
639 : void
640 1 : spdk_thread_bind(struct spdk_thread *thread, bool bind)
641 : {
642 1 : thread->is_bound = bind;
643 1 : }
644 :
645 : bool
646 10 : spdk_thread_is_bound(struct spdk_thread *thread)
647 : {
648 10 : return thread->is_bound;
649 : }
650 :
651 : void
652 466916 : spdk_set_thread(struct spdk_thread *thread)
653 : {
654 466916 : tls_thread = thread;
655 466916 : }
656 :
657 : static void
658 219 : thread_exit(struct spdk_thread *thread, uint64_t now)
659 : {
660 : struct spdk_poller *poller;
661 : struct spdk_io_channel *ch;
662 :
663 219 : if (now >= thread->exit_timeout_tsc) {
664 1 : SPDK_ERRLOG("thread %s got timeout, and move it to the exited state forcefully\n",
665 : thread->name);
666 1 : goto exited;
667 : }
668 :
669 218 : if (spdk_ring_count(thread->messages) > 0) {
670 2 : SPDK_INFOLOG(thread, "thread %s still has messages\n", thread->name);
671 2 : return;
672 : }
673 :
674 216 : if (thread->for_each_count > 0) {
675 4 : SPDK_INFOLOG(thread, "thread %s is still executing %u for_each_channels/threads\n",
676 : thread->name, thread->for_each_count);
677 4 : return;
678 : }
679 :
680 212 : TAILQ_FOREACH(poller, &thread->active_pollers, tailq) {
681 2 : if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) {
682 2 : SPDK_INFOLOG(thread,
683 : "thread %s still has active poller %s\n",
684 : thread->name, poller->name);
685 2 : return;
686 : }
687 : }
688 :
689 258 : RB_FOREACH(poller, timed_pollers_tree, &thread->timed_pollers) {
690 48 : if (poller->state != SPDK_POLLER_STATE_UNREGISTERED) {
691 0 : SPDK_INFOLOG(thread,
692 : "thread %s still has active timed poller %s\n",
693 : thread->name, poller->name);
694 0 : return;
695 : }
696 : }
697 :
698 210 : TAILQ_FOREACH(poller, &thread->paused_pollers, tailq) {
699 0 : SPDK_INFOLOG(thread,
700 : "thread %s still has paused poller %s\n",
701 : thread->name, poller->name);
702 0 : return;
703 : }
704 :
705 210 : RB_FOREACH(ch, io_channel_tree, &thread->io_channels) {
706 6 : SPDK_INFOLOG(thread,
707 : "thread %s still has channel for io_device %s\n",
708 : thread->name, ch->dev->name);
709 6 : return;
710 : }
711 :
712 204 : if (thread->pending_unregister_count > 0) {
713 2 : SPDK_INFOLOG(thread,
714 : "thread %s is still unregistering io_devices\n",
715 : thread->name);
716 2 : return;
717 : }
718 :
719 202 : exited:
720 203 : thread->state = SPDK_THREAD_STATE_EXITED;
721 203 : if (spdk_unlikely(thread->in_interrupt)) {
722 0 : g_thread_op_fn(thread, SPDK_THREAD_OP_RESCHED);
723 : }
724 : }
725 :
726 : static void _thread_exit(void *ctx);
727 :
728 : int
729 214 : spdk_thread_exit(struct spdk_thread *thread)
730 : {
731 214 : SPDK_DEBUGLOG(thread, "Exit thread %s\n", thread->name);
732 :
733 214 : assert(tls_thread == thread);
734 :
735 214 : if (thread->state >= SPDK_THREAD_STATE_EXITING) {
736 9 : SPDK_INFOLOG(thread,
737 : "thread %s is already exiting\n",
738 : thread->name);
739 9 : return 0;
740 : }
741 :
742 205 : thread->exit_timeout_tsc = spdk_get_ticks() + (spdk_get_ticks_hz() *
743 : SPDK_THREAD_EXIT_TIMEOUT_SEC);
744 205 : thread->state = SPDK_THREAD_STATE_EXITING;
745 :
746 205 : if (spdk_interrupt_mode_is_enabled()) {
747 0 : spdk_thread_send_msg(thread, _thread_exit, thread);
748 : }
749 :
750 205 : return 0;
751 : }
752 :
753 : bool
754 0 : spdk_thread_is_running(struct spdk_thread *thread)
755 : {
756 0 : return thread->state == SPDK_THREAD_STATE_RUNNING;
757 : }
758 :
759 : bool
760 407 : spdk_thread_is_exited(struct spdk_thread *thread)
761 : {
762 407 : return thread->state == SPDK_THREAD_STATE_EXITED;
763 : }
764 :
765 : void
766 184 : spdk_thread_destroy(struct spdk_thread *thread)
767 : {
768 184 : assert(thread != NULL);
769 184 : SPDK_DEBUGLOG(thread, "Destroy thread %s\n", thread->name);
770 :
771 184 : assert(thread->state == SPDK_THREAD_STATE_EXITED);
772 :
773 184 : if (tls_thread == thread) {
774 177 : tls_thread = NULL;
775 : }
776 :
777 : /* To be safe, do not free the app thread until spdk_thread_lib_fini(). */
778 184 : if (thread != g_app_thread) {
779 87 : _free_thread(thread);
780 : }
781 184 : }
782 :
783 : void *
784 49 : spdk_thread_get_ctx(struct spdk_thread *thread)
785 : {
786 49 : if (g_ctx_sz > 0) {
787 49 : return thread->ctx;
788 : }
789 :
790 0 : return NULL;
791 : }
792 :
793 : struct spdk_cpuset *
794 46 : spdk_thread_get_cpumask(struct spdk_thread *thread)
795 : {
796 46 : return &thread->cpumask;
797 : }
798 :
799 : int
800 2 : spdk_thread_set_cpumask(struct spdk_cpuset *cpumask)
801 : {
802 : struct spdk_thread *thread;
803 :
804 2 : if (!g_thread_op_supported_fn || !g_thread_op_supported_fn(SPDK_THREAD_OP_RESCHED)) {
805 0 : SPDK_ERRLOG("Framework does not support reschedule operation.\n");
806 0 : assert(false);
807 : return -ENOTSUP;
808 : }
809 :
810 2 : thread = spdk_get_thread();
811 2 : if (!thread) {
812 0 : SPDK_ERRLOG("Called from non-SPDK thread\n");
813 0 : assert(false);
814 : return -EINVAL;
815 : }
816 :
817 2 : spdk_cpuset_copy(&thread->cpumask, cpumask);
818 :
819 : /* Invoke framework's reschedule operation. If this function is called multiple times
820 : * in a single spdk_thread_poll() context, the last cpumask will be used in the
821 : * reschedule operation.
822 : */
823 2 : g_thread_op_fn(thread, SPDK_THREAD_OP_RESCHED);
824 :
825 2 : return 0;
826 : }
827 :
828 : struct spdk_thread *
829 191 : spdk_thread_get_from_ctx(void *ctx)
830 : {
831 191 : if (ctx == NULL) {
832 0 : assert(false);
833 : return NULL;
834 : }
835 :
836 191 : assert(g_ctx_sz > 0);
837 :
838 191 : return SPDK_CONTAINEROF(ctx, struct spdk_thread, ctx);
839 : }
840 :
841 : static inline uint32_t
842 414288 : msg_queue_run_batch(struct spdk_thread *thread, uint32_t max_msgs)
843 : {
844 : unsigned count, i;
845 : void *messages[SPDK_MSG_BATCH_SIZE];
846 414288 : uint64_t notify = 1;
847 : int rc;
848 :
849 : #ifdef DEBUG
850 : /*
851 : * spdk_ring_dequeue() fills messages and returns how many entries it wrote,
852 : * so we will never actually read uninitialized data from events, but just to be sure
853 : * (and to silence a static analyzer false positive), initialize the array to NULL pointers.
854 : */
855 414288 : memset(messages, 0, sizeof(messages));
856 : #endif
857 :
858 414288 : if (max_msgs > 0) {
859 365 : max_msgs = spdk_min(max_msgs, SPDK_MSG_BATCH_SIZE);
860 : } else {
861 413923 : max_msgs = SPDK_MSG_BATCH_SIZE;
862 : }
863 :
864 414288 : count = spdk_ring_dequeue(thread->messages, messages, max_msgs);
865 414288 : if (spdk_unlikely(thread->in_interrupt) &&
866 0 : spdk_ring_count(thread->messages) != 0) {
867 0 : rc = write(thread->msg_fd, ¬ify, sizeof(notify));
868 0 : if (rc < 0) {
869 0 : SPDK_ERRLOG("failed to notify msg_queue: %s.\n", spdk_strerror(errno));
870 : }
871 : }
872 414288 : if (count == 0) {
873 253701 : return 0;
874 : }
875 :
876 326532 : for (i = 0; i < count; i++) {
877 165945 : struct spdk_msg *msg = messages[i];
878 :
879 165945 : assert(msg != NULL);
880 :
881 : SPDK_DTRACE_PROBE2(msg_exec, msg->fn, msg->arg);
882 :
883 165945 : msg->fn(msg->arg);
884 :
885 165945 : SPIN_ASSERT(thread->lock_count == 0, SPIN_ERR_HOLD_DURING_SWITCH);
886 :
887 165945 : if (thread->msg_cache_count < SPDK_MSG_MEMPOOL_CACHE_SIZE) {
888 : /* Insert the messages at the head. We want to re-use the hot
889 : * ones. */
890 165806 : SLIST_INSERT_HEAD(&thread->msg_cache, msg, link);
891 165806 : thread->msg_cache_count++;
892 : } else {
893 139 : spdk_mempool_put(g_spdk_msg_mempool, msg);
894 : }
895 : }
896 :
897 160587 : return count;
898 : }
899 :
900 : static void
901 547 : poller_insert_timer(struct spdk_thread *thread, struct spdk_poller *poller, uint64_t now)
902 : {
903 : struct spdk_poller *tmp __attribute__((unused));
904 :
905 547 : poller->next_run_tick = now + poller->period_ticks;
906 :
907 : /*
908 : * Insert poller in the thread's timed_pollers tree by next scheduled run time
909 : * as its key.
910 : */
911 547 : tmp = RB_INSERT(timed_pollers_tree, &thread->timed_pollers, poller);
912 547 : assert(tmp == NULL);
913 :
914 : /* Update the cache only if it is empty or the inserted poller is earlier than it.
915 : * RB_MIN() is not necessary here because all pollers, which has exactly the same
916 : * next_run_tick as the existing poller, are inserted on the right side.
917 : */
918 547 : if (thread->first_timed_poller == NULL ||
919 458 : poller->next_run_tick < thread->first_timed_poller->next_run_tick) {
920 255 : thread->first_timed_poller = poller;
921 : }
922 547 : }
923 :
924 : static inline void
925 0 : poller_remove_timer(struct spdk_thread *thread, struct spdk_poller *poller)
926 : {
927 : struct spdk_poller *tmp __attribute__((unused));
928 :
929 0 : tmp = RB_REMOVE(timed_pollers_tree, &thread->timed_pollers, poller);
930 0 : assert(tmp != NULL);
931 :
932 : /* This function is not used in any case that is performance critical.
933 : * Update the cache simply by RB_MIN() if it needs to be changed.
934 : */
935 0 : if (thread->first_timed_poller == poller) {
936 0 : thread->first_timed_poller = RB_MIN(timed_pollers_tree, &thread->timed_pollers);
937 : }
938 0 : }
939 :
940 : static void
941 898 : thread_insert_poller(struct spdk_thread *thread, struct spdk_poller *poller)
942 : {
943 898 : if (poller->period_ticks) {
944 368 : poller_insert_timer(thread, poller, spdk_get_ticks());
945 : } else {
946 530 : TAILQ_INSERT_TAIL(&thread->active_pollers, poller, tailq);
947 : }
948 898 : }
949 :
950 : static inline void
951 414288 : thread_update_stats(struct spdk_thread *thread, uint64_t end,
952 : uint64_t start, int rc)
953 : {
954 414288 : if (rc == 0) {
955 : /* Poller status idle */
956 253153 : thread->stats.idle_tsc += end - start;
957 161135 : } else if (rc > 0) {
958 : /* Poller status busy */
959 161135 : thread->stats.busy_tsc += end - start;
960 : }
961 : /* Store end time to use it as start time of the next spdk_thread_poll(). */
962 414288 : thread->tsc_last = end;
963 414288 : }
964 :
965 : static inline int
966 4391 : thread_execute_poller(struct spdk_thread *thread, struct spdk_poller *poller)
967 : {
968 : int rc;
969 :
970 4391 : switch (poller->state) {
971 108 : case SPDK_POLLER_STATE_UNREGISTERED:
972 108 : TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
973 108 : free(poller);
974 108 : return 0;
975 5 : case SPDK_POLLER_STATE_PAUSING:
976 5 : TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
977 5 : TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq);
978 5 : poller->state = SPDK_POLLER_STATE_PAUSED;
979 5 : return 0;
980 4278 : case SPDK_POLLER_STATE_WAITING:
981 4278 : break;
982 0 : default:
983 0 : assert(false);
984 : break;
985 : }
986 :
987 4278 : poller->state = SPDK_POLLER_STATE_RUNNING;
988 4278 : rc = poller->fn(poller->arg);
989 :
990 4278 : SPIN_ASSERT(thread->lock_count == 0, SPIN_ERR_HOLD_DURING_SWITCH);
991 :
992 4278 : poller->run_count++;
993 4278 : if (rc > 0) {
994 578 : poller->busy_count++;
995 : }
996 :
997 : #ifdef DEBUG
998 4278 : if (rc == -1) {
999 13 : SPDK_DEBUGLOG(thread, "Poller %s returned -1\n", poller->name);
1000 : }
1001 : #endif
1002 :
1003 4278 : switch (poller->state) {
1004 417 : case SPDK_POLLER_STATE_UNREGISTERED:
1005 417 : TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
1006 417 : free(poller);
1007 417 : break;
1008 7 : case SPDK_POLLER_STATE_PAUSING:
1009 7 : TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
1010 7 : TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq);
1011 7 : poller->state = SPDK_POLLER_STATE_PAUSED;
1012 7 : break;
1013 0 : case SPDK_POLLER_STATE_PAUSED:
1014 : case SPDK_POLLER_STATE_WAITING:
1015 0 : break;
1016 3854 : case SPDK_POLLER_STATE_RUNNING:
1017 3854 : poller->state = SPDK_POLLER_STATE_WAITING;
1018 3854 : break;
1019 0 : default:
1020 0 : assert(false);
1021 : break;
1022 : }
1023 :
1024 4278 : return rc;
1025 : }
1026 :
1027 : static inline int
1028 499 : thread_execute_timed_poller(struct spdk_thread *thread, struct spdk_poller *poller,
1029 : uint64_t now)
1030 : {
1031 : int rc;
1032 :
1033 499 : switch (poller->state) {
1034 156 : case SPDK_POLLER_STATE_UNREGISTERED:
1035 156 : free(poller);
1036 156 : return 0;
1037 13 : case SPDK_POLLER_STATE_PAUSING:
1038 13 : TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq);
1039 13 : poller->state = SPDK_POLLER_STATE_PAUSED;
1040 13 : return 0;
1041 330 : case SPDK_POLLER_STATE_WAITING:
1042 330 : break;
1043 0 : default:
1044 0 : assert(false);
1045 : break;
1046 : }
1047 :
1048 330 : poller->state = SPDK_POLLER_STATE_RUNNING;
1049 330 : rc = poller->fn(poller->arg);
1050 :
1051 330 : SPIN_ASSERT(thread->lock_count == 0, SPIN_ERR_HOLD_DURING_SWITCH);
1052 :
1053 330 : poller->run_count++;
1054 330 : if (rc > 0) {
1055 237 : poller->busy_count++;
1056 : }
1057 :
1058 : #ifdef DEBUG
1059 330 : if (rc == -1) {
1060 5 : SPDK_DEBUGLOG(thread, "Timed poller %s returned -1\n", poller->name);
1061 : }
1062 : #endif
1063 :
1064 330 : switch (poller->state) {
1065 147 : case SPDK_POLLER_STATE_UNREGISTERED:
1066 147 : free(poller);
1067 147 : break;
1068 4 : case SPDK_POLLER_STATE_PAUSING:
1069 4 : TAILQ_INSERT_TAIL(&thread->paused_pollers, poller, tailq);
1070 4 : poller->state = SPDK_POLLER_STATE_PAUSED;
1071 4 : break;
1072 0 : case SPDK_POLLER_STATE_PAUSED:
1073 0 : break;
1074 179 : case SPDK_POLLER_STATE_RUNNING:
1075 179 : poller->state = SPDK_POLLER_STATE_WAITING;
1076 : /* fallthrough */
1077 179 : case SPDK_POLLER_STATE_WAITING:
1078 179 : poller_insert_timer(thread, poller, now);
1079 179 : break;
1080 0 : default:
1081 0 : assert(false);
1082 : break;
1083 : }
1084 :
1085 330 : return rc;
1086 : }
1087 :
1088 : static inline void
1089 0 : thread_run_pp_handlers(struct spdk_thread *thread)
1090 : {
1091 0 : uint8_t i, count = thread->num_pp_handlers;
1092 :
1093 : /* Set to max value to prevent new handlers registration within the callback */
1094 0 : thread->num_pp_handlers = SPDK_THREAD_MAX_POST_POLLER_HANDLERS;
1095 :
1096 0 : for (i = 0; i < count; i++) {
1097 0 : thread->pp_handlers[i].fn(thread->pp_handlers[i].fn_arg);
1098 0 : thread->pp_handlers[i].fn = NULL;
1099 : }
1100 :
1101 0 : thread->num_pp_handlers = 0;
1102 0 : }
1103 :
1104 : static int
1105 414288 : thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now)
1106 : {
1107 : uint32_t msg_count;
1108 : struct spdk_poller *poller, *tmp;
1109 : spdk_msg_fn critical_msg;
1110 414288 : int rc = 0;
1111 :
1112 414288 : thread->tsc_last = now;
1113 :
1114 414288 : critical_msg = thread->critical_msg;
1115 414288 : if (spdk_unlikely(critical_msg != NULL)) {
1116 0 : critical_msg(NULL);
1117 0 : thread->critical_msg = NULL;
1118 0 : rc = 1;
1119 : }
1120 :
1121 414288 : msg_count = msg_queue_run_batch(thread, max_msgs);
1122 414288 : if (msg_count) {
1123 160587 : rc = 1;
1124 : }
1125 :
1126 418679 : TAILQ_FOREACH_REVERSE_SAFE(poller, &thread->active_pollers,
1127 : active_pollers_head, tailq, tmp) {
1128 : int poller_rc;
1129 :
1130 4391 : poller_rc = thread_execute_poller(thread, poller);
1131 4391 : if (poller_rc > rc) {
1132 350 : rc = poller_rc;
1133 : }
1134 4391 : if (thread->num_pp_handlers) {
1135 0 : thread_run_pp_handlers(thread);
1136 : }
1137 : }
1138 :
1139 414288 : poller = thread->first_timed_poller;
1140 414787 : while (poller != NULL) {
1141 3433 : int timer_rc = 0;
1142 :
1143 3433 : if (now < poller->next_run_tick) {
1144 2934 : break;
1145 : }
1146 :
1147 499 : tmp = RB_NEXT(timed_pollers_tree, &thread->timed_pollers, poller);
1148 499 : RB_REMOVE(timed_pollers_tree, &thread->timed_pollers, poller);
1149 :
1150 : /* Update the cache to the next timed poller in the list
1151 : * only if the current poller is still the closest, otherwise,
1152 : * do nothing because the cache has been already updated.
1153 : */
1154 499 : if (thread->first_timed_poller == poller) {
1155 499 : thread->first_timed_poller = tmp;
1156 : }
1157 :
1158 499 : timer_rc = thread_execute_timed_poller(thread, poller, now);
1159 499 : if (timer_rc > rc) {
1160 198 : rc = timer_rc;
1161 : }
1162 :
1163 499 : poller = tmp;
1164 : }
1165 :
1166 414288 : return rc;
1167 : }
1168 :
1169 : static void
1170 0 : _thread_remove_pollers(void *ctx)
1171 : {
1172 0 : struct spdk_thread *thread = ctx;
1173 : struct spdk_poller *poller, *tmp;
1174 :
1175 0 : TAILQ_FOREACH_REVERSE_SAFE(poller, &thread->active_pollers,
1176 : active_pollers_head, tailq, tmp) {
1177 0 : if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) {
1178 0 : TAILQ_REMOVE(&thread->active_pollers, poller, tailq);
1179 0 : free(poller);
1180 : }
1181 : }
1182 :
1183 0 : RB_FOREACH_SAFE(poller, timed_pollers_tree, &thread->timed_pollers, tmp) {
1184 0 : if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) {
1185 0 : poller_remove_timer(thread, poller);
1186 0 : free(poller);
1187 : }
1188 : }
1189 :
1190 0 : thread->poller_unregistered = false;
1191 0 : }
1192 :
1193 : static void
1194 0 : _thread_exit(void *ctx)
1195 : {
1196 0 : struct spdk_thread *thread = ctx;
1197 :
1198 0 : assert(thread->state == SPDK_THREAD_STATE_EXITING);
1199 :
1200 0 : thread_exit(thread, spdk_get_ticks());
1201 :
1202 0 : if (thread->state != SPDK_THREAD_STATE_EXITED) {
1203 0 : spdk_thread_send_msg(thread, _thread_exit, thread);
1204 : }
1205 0 : }
1206 :
1207 : int
1208 414288 : spdk_thread_poll(struct spdk_thread *thread, uint32_t max_msgs, uint64_t now)
1209 : {
1210 : struct spdk_thread *orig_thread;
1211 : int rc;
1212 :
1213 414288 : orig_thread = _get_thread();
1214 414288 : tls_thread = thread;
1215 :
1216 414288 : if (now == 0) {
1217 407107 : now = spdk_get_ticks();
1218 : }
1219 :
1220 414288 : if (spdk_likely(!thread->in_interrupt)) {
1221 414288 : rc = thread_poll(thread, max_msgs, now);
1222 414288 : if (spdk_unlikely(thread->in_interrupt)) {
1223 : /* The thread transitioned to interrupt mode during the above poll.
1224 : * Poll it one more time in case that during the transition time
1225 : * there is msg received without notification.
1226 : */
1227 0 : rc = thread_poll(thread, max_msgs, now);
1228 : }
1229 :
1230 414288 : if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITING)) {
1231 219 : thread_exit(thread, now);
1232 : }
1233 : } else {
1234 : /* Non-block wait on thread's fd_group */
1235 0 : rc = spdk_fd_group_wait(thread->fgrp, 0);
1236 : }
1237 :
1238 414288 : thread_update_stats(thread, spdk_get_ticks(), now, rc);
1239 :
1240 414288 : tls_thread = orig_thread;
1241 :
1242 414288 : return rc;
1243 : }
1244 :
1245 : uint64_t
1246 0 : spdk_thread_next_poller_expiration(struct spdk_thread *thread)
1247 : {
1248 : struct spdk_poller *poller;
1249 :
1250 0 : poller = thread->first_timed_poller;
1251 0 : if (poller) {
1252 0 : return poller->next_run_tick;
1253 : }
1254 :
1255 0 : return 0;
1256 : }
1257 :
1258 : int
1259 0 : spdk_thread_has_active_pollers(struct spdk_thread *thread)
1260 : {
1261 0 : return !TAILQ_EMPTY(&thread->active_pollers);
1262 : }
1263 :
1264 : static bool
1265 25 : thread_has_unpaused_pollers(struct spdk_thread *thread)
1266 : {
1267 25 : if (TAILQ_EMPTY(&thread->active_pollers) &&
1268 25 : RB_EMPTY(&thread->timed_pollers)) {
1269 25 : return false;
1270 : }
1271 :
1272 0 : return true;
1273 : }
1274 :
1275 : bool
1276 2 : spdk_thread_has_pollers(struct spdk_thread *thread)
1277 : {
1278 2 : if (!thread_has_unpaused_pollers(thread) &&
1279 2 : TAILQ_EMPTY(&thread->paused_pollers)) {
1280 2 : return false;
1281 : }
1282 :
1283 0 : return true;
1284 : }
1285 :
1286 : bool
1287 23 : spdk_thread_is_idle(struct spdk_thread *thread)
1288 : {
1289 46 : if (spdk_ring_count(thread->messages) ||
1290 23 : thread_has_unpaused_pollers(thread) ||
1291 23 : thread->critical_msg != NULL) {
1292 0 : return false;
1293 : }
1294 :
1295 23 : return true;
1296 : }
1297 :
1298 : uint32_t
1299 17 : spdk_thread_get_count(void)
1300 : {
1301 : /*
1302 : * Return cached value of the current thread count. We could acquire the
1303 : * lock and iterate through the TAILQ of threads to count them, but that
1304 : * count could still be invalidated after we release the lock.
1305 : */
1306 17 : return g_thread_count;
1307 : }
1308 :
1309 : struct spdk_thread *
1310 348846 : spdk_get_thread(void)
1311 : {
1312 348846 : return _get_thread();
1313 : }
1314 :
1315 : const char *
1316 4 : spdk_thread_get_name(const struct spdk_thread *thread)
1317 : {
1318 4 : return thread->name;
1319 : }
1320 :
1321 : uint64_t
1322 15 : spdk_thread_get_id(const struct spdk_thread *thread)
1323 : {
1324 15 : return thread->id;
1325 : }
1326 :
1327 : struct spdk_thread *
1328 13 : spdk_thread_get_by_id(uint64_t id)
1329 : {
1330 : struct spdk_thread *thread;
1331 :
1332 13 : if (id == 0 || id >= g_thread_id) {
1333 0 : SPDK_ERRLOG("invalid thread id: %" PRIu64 ".\n", id);
1334 0 : return NULL;
1335 : }
1336 13 : pthread_mutex_lock(&g_devlist_mutex);
1337 28 : TAILQ_FOREACH(thread, &g_threads, tailq) {
1338 28 : if (thread->id == id) {
1339 13 : break;
1340 : }
1341 : }
1342 13 : pthread_mutex_unlock(&g_devlist_mutex);
1343 13 : return thread;
1344 : }
1345 :
1346 : int
1347 57 : spdk_thread_get_stats(struct spdk_thread_stats *stats)
1348 : {
1349 : struct spdk_thread *thread;
1350 :
1351 57 : thread = _get_thread();
1352 57 : if (!thread) {
1353 0 : SPDK_ERRLOG("No thread allocated\n");
1354 0 : return -EINVAL;
1355 : }
1356 :
1357 57 : if (stats == NULL) {
1358 0 : return -EINVAL;
1359 : }
1360 :
1361 57 : *stats = thread->stats;
1362 :
1363 57 : return 0;
1364 : }
1365 :
1366 : uint64_t
1367 168088 : spdk_thread_get_last_tsc(struct spdk_thread *thread)
1368 : {
1369 168088 : if (thread == NULL) {
1370 0 : thread = _get_thread();
1371 : }
1372 :
1373 168088 : return thread->tsc_last;
1374 : }
1375 :
1376 : static inline int
1377 165954 : thread_send_msg_notification(const struct spdk_thread *target_thread)
1378 : {
1379 165954 : uint64_t notify = 1;
1380 : int rc;
1381 :
1382 : /* Not necessary to do notification if interrupt facility is not enabled */
1383 165954 : if (spdk_likely(!spdk_interrupt_mode_is_enabled())) {
1384 165954 : return 0;
1385 : }
1386 :
1387 : /* When each spdk_thread can switch between poll and interrupt mode dynamically,
1388 : * after sending thread msg, it is necessary to check whether target thread runs in
1389 : * interrupt mode and then decide whether do event notification.
1390 : */
1391 0 : if (spdk_unlikely(target_thread->in_interrupt)) {
1392 0 : rc = write(target_thread->msg_fd, ¬ify, sizeof(notify));
1393 0 : if (rc < 0) {
1394 0 : SPDK_ERRLOG("failed to notify msg_queue: %s.\n", spdk_strerror(errno));
1395 0 : return -EIO;
1396 : }
1397 : }
1398 :
1399 0 : return 0;
1400 : }
1401 :
1402 : int
1403 165954 : spdk_thread_send_msg(const struct spdk_thread *thread, spdk_msg_fn fn, void *ctx)
1404 : {
1405 : struct spdk_thread *local_thread;
1406 : struct spdk_msg *msg;
1407 : int rc;
1408 :
1409 165954 : assert(thread != NULL);
1410 :
1411 165954 : if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITED)) {
1412 0 : SPDK_ERRLOG("Thread %s is marked as exited.\n", thread->name);
1413 0 : return -EIO;
1414 : }
1415 :
1416 165954 : local_thread = _get_thread();
1417 :
1418 165954 : msg = NULL;
1419 165954 : if (local_thread != NULL) {
1420 165954 : if (local_thread->msg_cache_count > 0) {
1421 165954 : msg = SLIST_FIRST(&local_thread->msg_cache);
1422 165954 : assert(msg != NULL);
1423 165954 : SLIST_REMOVE_HEAD(&local_thread->msg_cache, link);
1424 165954 : local_thread->msg_cache_count--;
1425 : }
1426 : }
1427 :
1428 165954 : if (msg == NULL) {
1429 0 : msg = spdk_mempool_get(g_spdk_msg_mempool);
1430 0 : if (!msg) {
1431 0 : SPDK_ERRLOG("msg could not be allocated\n");
1432 0 : return -ENOMEM;
1433 : }
1434 : }
1435 :
1436 165954 : msg->fn = fn;
1437 165954 : msg->arg = ctx;
1438 :
1439 165954 : rc = spdk_ring_enqueue(thread->messages, (void **)&msg, 1, NULL);
1440 165954 : if (rc != 1) {
1441 0 : SPDK_ERRLOG("msg could not be enqueued\n");
1442 0 : spdk_mempool_put(g_spdk_msg_mempool, msg);
1443 0 : return -EIO;
1444 : }
1445 :
1446 165954 : return thread_send_msg_notification(thread);
1447 : }
1448 :
1449 : int
1450 0 : spdk_thread_send_critical_msg(struct spdk_thread *thread, spdk_msg_fn fn)
1451 : {
1452 0 : spdk_msg_fn expected = NULL;
1453 :
1454 0 : if (!__atomic_compare_exchange_n(&thread->critical_msg, &expected, fn, false, __ATOMIC_SEQ_CST,
1455 : __ATOMIC_SEQ_CST)) {
1456 0 : return -EIO;
1457 : }
1458 :
1459 0 : return thread_send_msg_notification(thread);
1460 : }
1461 :
1462 : #ifdef __linux__
1463 : static int
1464 0 : interrupt_timerfd_process(void *arg)
1465 : {
1466 0 : struct spdk_poller *poller = arg;
1467 : uint64_t exp;
1468 : int rc;
1469 :
1470 : /* clear the level of interval timer */
1471 0 : rc = read(poller->intr->efd, &exp, sizeof(exp));
1472 0 : if (rc < 0) {
1473 0 : if (rc == -EAGAIN) {
1474 0 : return 0;
1475 : }
1476 :
1477 0 : return rc;
1478 : }
1479 :
1480 : SPDK_DTRACE_PROBE2(timerfd_exec, poller->fn, poller->arg);
1481 :
1482 0 : return poller->fn(poller->arg);
1483 : }
1484 :
1485 : static int
1486 0 : period_poller_interrupt_init(struct spdk_poller *poller)
1487 : {
1488 : int timerfd;
1489 :
1490 0 : SPDK_DEBUGLOG(thread, "timerfd init for periodic poller %s\n", poller->name);
1491 0 : timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC);
1492 0 : if (timerfd < 0) {
1493 0 : return -errno;
1494 : }
1495 :
1496 0 : poller->intr = spdk_interrupt_register(timerfd, interrupt_timerfd_process, poller, poller->name);
1497 0 : if (poller->intr == NULL) {
1498 0 : close(timerfd);
1499 0 : return -1;
1500 : }
1501 :
1502 0 : return 0;
1503 : }
1504 :
1505 : static void
1506 0 : period_poller_set_interrupt_mode(struct spdk_poller *poller, void *cb_arg, bool interrupt_mode)
1507 : {
1508 : int timerfd;
1509 0 : uint64_t now_tick = spdk_get_ticks();
1510 0 : uint64_t ticks = spdk_get_ticks_hz();
1511 : int ret;
1512 0 : struct itimerspec new_tv = {};
1513 0 : struct itimerspec old_tv = {};
1514 :
1515 0 : assert(poller->intr != NULL);
1516 0 : assert(poller->period_ticks != 0);
1517 :
1518 0 : timerfd = poller->intr->efd;
1519 :
1520 0 : assert(timerfd >= 0);
1521 :
1522 0 : SPDK_DEBUGLOG(thread, "timerfd set poller %s into %s mode\n", poller->name,
1523 : interrupt_mode ? "interrupt" : "poll");
1524 :
1525 0 : if (interrupt_mode) {
1526 : /* Set repeated timer expiration */
1527 0 : new_tv.it_interval.tv_sec = poller->period_ticks / ticks;
1528 0 : new_tv.it_interval.tv_nsec = poller->period_ticks % ticks * SPDK_SEC_TO_NSEC / ticks;
1529 :
1530 : /* Update next timer expiration */
1531 0 : if (poller->next_run_tick == 0) {
1532 0 : poller->next_run_tick = now_tick + poller->period_ticks;
1533 0 : } else if (poller->next_run_tick < now_tick) {
1534 0 : poller->next_run_tick = now_tick;
1535 : }
1536 :
1537 0 : new_tv.it_value.tv_sec = (poller->next_run_tick - now_tick) / ticks;
1538 0 : new_tv.it_value.tv_nsec = (poller->next_run_tick - now_tick) % ticks * SPDK_SEC_TO_NSEC / ticks;
1539 :
1540 0 : ret = timerfd_settime(timerfd, 0, &new_tv, NULL);
1541 0 : if (ret < 0) {
1542 0 : SPDK_ERRLOG("Failed to arm timerfd: error(%d)\n", errno);
1543 0 : assert(false);
1544 : }
1545 : } else {
1546 : /* Disarm the timer */
1547 0 : ret = timerfd_settime(timerfd, 0, &new_tv, &old_tv);
1548 0 : if (ret < 0) {
1549 : /* timerfd_settime's failure indicates that the timerfd is in error */
1550 0 : SPDK_ERRLOG("Failed to disarm timerfd: error(%d)\n", errno);
1551 0 : assert(false);
1552 : }
1553 :
1554 : /* In order to reuse poller_insert_timer, fix now_tick, so next_run_tick would be
1555 : * now_tick + ticks * old_tv.it_value.tv_sec + (ticks * old_tv.it_value.tv_nsec) / SPDK_SEC_TO_NSEC
1556 : */
1557 0 : now_tick = now_tick - poller->period_ticks + ticks * old_tv.it_value.tv_sec + \
1558 0 : (ticks * old_tv.it_value.tv_nsec) / SPDK_SEC_TO_NSEC;
1559 0 : poller_remove_timer(poller->thread, poller);
1560 0 : poller_insert_timer(poller->thread, poller, now_tick);
1561 : }
1562 0 : }
1563 :
1564 : static void
1565 0 : poller_interrupt_fini(struct spdk_poller *poller)
1566 : {
1567 : int fd;
1568 :
1569 0 : SPDK_DEBUGLOG(thread, "interrupt fini for poller %s\n", poller->name);
1570 0 : assert(poller->intr != NULL);
1571 0 : fd = poller->intr->efd;
1572 0 : spdk_interrupt_unregister(&poller->intr);
1573 0 : close(fd);
1574 0 : }
1575 :
1576 : static int
1577 0 : busy_poller_interrupt_init(struct spdk_poller *poller)
1578 : {
1579 : int busy_efd;
1580 :
1581 0 : SPDK_DEBUGLOG(thread, "busy_efd init for busy poller %s\n", poller->name);
1582 0 : busy_efd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
1583 0 : if (busy_efd < 0) {
1584 0 : SPDK_ERRLOG("Failed to create eventfd for Poller(%s).\n", poller->name);
1585 0 : return -errno;
1586 : }
1587 :
1588 0 : poller->intr = spdk_interrupt_register(busy_efd, poller->fn, poller->arg, poller->name);
1589 0 : if (poller->intr == NULL) {
1590 0 : close(busy_efd);
1591 0 : return -1;
1592 : }
1593 :
1594 0 : return 0;
1595 : }
1596 :
1597 : static void
1598 0 : busy_poller_set_interrupt_mode(struct spdk_poller *poller, void *cb_arg, bool interrupt_mode)
1599 : {
1600 0 : int busy_efd = poller->intr->efd;
1601 0 : uint64_t notify = 1;
1602 : int rc __attribute__((unused));
1603 :
1604 0 : assert(busy_efd >= 0);
1605 :
1606 0 : if (interrupt_mode) {
1607 : /* Write without read on eventfd will get it repeatedly triggered. */
1608 0 : if (write(busy_efd, ¬ify, sizeof(notify)) < 0) {
1609 0 : SPDK_ERRLOG("Failed to set busy wait for Poller(%s).\n", poller->name);
1610 : }
1611 : } else {
1612 : /* Read on eventfd will clear its level triggering. */
1613 0 : rc = read(busy_efd, ¬ify, sizeof(notify));
1614 : }
1615 0 : }
1616 :
1617 : #else
1618 :
1619 : static int
1620 : period_poller_interrupt_init(struct spdk_poller *poller)
1621 : {
1622 : return -ENOTSUP;
1623 : }
1624 :
1625 : static void
1626 : period_poller_set_interrupt_mode(struct spdk_poller *poller, void *cb_arg, bool interrupt_mode)
1627 : {
1628 : }
1629 :
1630 : static void
1631 : poller_interrupt_fini(struct spdk_poller *poller)
1632 : {
1633 : }
1634 :
1635 : static int
1636 : busy_poller_interrupt_init(struct spdk_poller *poller)
1637 : {
1638 : return -ENOTSUP;
1639 : }
1640 :
1641 : static void
1642 : busy_poller_set_interrupt_mode(struct spdk_poller *poller, void *cb_arg, bool interrupt_mode)
1643 : {
1644 : }
1645 :
1646 : #endif
1647 :
1648 : void
1649 7 : spdk_poller_register_interrupt(struct spdk_poller *poller,
1650 : spdk_poller_set_interrupt_mode_cb cb_fn,
1651 : void *cb_arg)
1652 : {
1653 7 : assert(poller != NULL);
1654 7 : assert(spdk_get_thread() == poller->thread);
1655 :
1656 7 : if (!spdk_interrupt_mode_is_enabled()) {
1657 7 : return;
1658 : }
1659 :
1660 : /* If this poller already had an interrupt, clean the old one up. */
1661 0 : if (poller->intr != NULL) {
1662 0 : poller_interrupt_fini(poller);
1663 : }
1664 :
1665 0 : poller->set_intr_cb_fn = cb_fn;
1666 0 : poller->set_intr_cb_arg = cb_arg;
1667 :
1668 : /* Set poller into interrupt mode if thread is in interrupt. */
1669 0 : if (poller->thread->in_interrupt && poller->set_intr_cb_fn) {
1670 0 : poller->set_intr_cb_fn(poller, poller->set_intr_cb_arg, true);
1671 : }
1672 : }
1673 :
1674 : static uint64_t
1675 878 : convert_us_to_ticks(uint64_t us)
1676 : {
1677 : uint64_t quotient, remainder, ticks;
1678 :
1679 878 : if (us) {
1680 354 : quotient = us / SPDK_SEC_TO_USEC;
1681 354 : remainder = us % SPDK_SEC_TO_USEC;
1682 354 : ticks = spdk_get_ticks_hz();
1683 :
1684 354 : return ticks * quotient + (ticks * remainder) / SPDK_SEC_TO_USEC;
1685 : } else {
1686 524 : return 0;
1687 : }
1688 : }
1689 :
1690 : static struct spdk_poller *
1691 878 : poller_register(spdk_poller_fn fn,
1692 : void *arg,
1693 : uint64_t period_microseconds,
1694 : const char *name)
1695 : {
1696 : struct spdk_thread *thread;
1697 : struct spdk_poller *poller;
1698 :
1699 878 : thread = spdk_get_thread();
1700 878 : if (!thread) {
1701 0 : assert(false);
1702 : return NULL;
1703 : }
1704 :
1705 878 : if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITED)) {
1706 0 : SPDK_ERRLOG("thread %s is marked as exited\n", thread->name);
1707 0 : return NULL;
1708 : }
1709 :
1710 878 : poller = calloc(1, sizeof(*poller));
1711 878 : if (poller == NULL) {
1712 0 : SPDK_ERRLOG("Poller memory allocation failed\n");
1713 0 : return NULL;
1714 : }
1715 :
1716 878 : if (name) {
1717 817 : snprintf(poller->name, sizeof(poller->name), "%s", name);
1718 : } else {
1719 61 : snprintf(poller->name, sizeof(poller->name), "%p", fn);
1720 : }
1721 :
1722 878 : poller->state = SPDK_POLLER_STATE_WAITING;
1723 878 : poller->fn = fn;
1724 878 : poller->arg = arg;
1725 878 : poller->thread = thread;
1726 878 : poller->intr = NULL;
1727 878 : if (thread->next_poller_id == 0) {
1728 0 : SPDK_WARNLOG("Poller ID rolled over. Poller ID is duplicated.\n");
1729 0 : thread->next_poller_id = 1;
1730 : }
1731 878 : poller->id = thread->next_poller_id++;
1732 :
1733 878 : poller->period_ticks = convert_us_to_ticks(period_microseconds);
1734 :
1735 878 : if (spdk_interrupt_mode_is_enabled()) {
1736 : int rc;
1737 :
1738 0 : if (period_microseconds) {
1739 0 : rc = period_poller_interrupt_init(poller);
1740 0 : if (rc < 0) {
1741 0 : SPDK_ERRLOG("Failed to register interruptfd for periodic poller: %s\n", spdk_strerror(-rc));
1742 0 : free(poller);
1743 0 : return NULL;
1744 : }
1745 :
1746 0 : poller->set_intr_cb_fn = period_poller_set_interrupt_mode;
1747 0 : poller->set_intr_cb_arg = NULL;
1748 :
1749 : } else {
1750 : /* If the poller doesn't have a period, create interruptfd that's always
1751 : * busy automatically when running in interrupt mode.
1752 : */
1753 0 : rc = busy_poller_interrupt_init(poller);
1754 0 : if (rc > 0) {
1755 0 : SPDK_ERRLOG("Failed to register interruptfd for busy poller: %s\n", spdk_strerror(-rc));
1756 0 : free(poller);
1757 0 : return NULL;
1758 : }
1759 :
1760 0 : poller->set_intr_cb_fn = busy_poller_set_interrupt_mode;
1761 0 : poller->set_intr_cb_arg = NULL;
1762 : }
1763 :
1764 : /* Set poller into interrupt mode if thread is in interrupt. */
1765 0 : if (poller->thread->in_interrupt) {
1766 0 : poller->set_intr_cb_fn(poller, poller->set_intr_cb_arg, true);
1767 : }
1768 : }
1769 :
1770 878 : thread_insert_poller(thread, poller);
1771 :
1772 878 : return poller;
1773 : }
1774 :
1775 : struct spdk_poller *
1776 61 : spdk_poller_register(spdk_poller_fn fn,
1777 : void *arg,
1778 : uint64_t period_microseconds)
1779 : {
1780 61 : return poller_register(fn, arg, period_microseconds, NULL);
1781 : }
1782 :
1783 : struct spdk_poller *
1784 817 : spdk_poller_register_named(spdk_poller_fn fn,
1785 : void *arg,
1786 : uint64_t period_microseconds,
1787 : const char *name)
1788 : {
1789 817 : return poller_register(fn, arg, period_microseconds, name);
1790 : }
1791 :
1792 : static void
1793 0 : wrong_thread(const char *func, const char *name, struct spdk_thread *thread,
1794 : struct spdk_thread *curthread)
1795 : {
1796 0 : if (thread == NULL) {
1797 0 : SPDK_ERRLOG("%s(%s) called with NULL thread\n", func, name);
1798 0 : abort();
1799 : }
1800 0 : SPDK_ERRLOG("%s(%s) called from wrong thread %s:%" PRIu64 " (should be "
1801 : "%s:%" PRIu64 ")\n", func, name, curthread->name, curthread->id,
1802 : thread->name, thread->id);
1803 0 : assert(false);
1804 : }
1805 :
1806 : void
1807 1326 : spdk_poller_unregister(struct spdk_poller **ppoller)
1808 : {
1809 : struct spdk_thread *thread;
1810 : struct spdk_poller *poller;
1811 :
1812 1326 : poller = *ppoller;
1813 1326 : if (poller == NULL) {
1814 448 : return;
1815 : }
1816 :
1817 878 : *ppoller = NULL;
1818 :
1819 878 : thread = spdk_get_thread();
1820 878 : if (!thread) {
1821 0 : assert(false);
1822 : return;
1823 : }
1824 :
1825 878 : if (poller->thread != thread) {
1826 0 : wrong_thread(__func__, poller->name, poller->thread, thread);
1827 0 : return;
1828 : }
1829 :
1830 878 : if (spdk_interrupt_mode_is_enabled()) {
1831 : /* Release the interrupt resource for period or busy poller */
1832 0 : if (poller->intr != NULL) {
1833 0 : poller_interrupt_fini(poller);
1834 : }
1835 :
1836 : /* If there is not already a pending poller removal, generate
1837 : * a message to go process removals. */
1838 0 : if (!thread->poller_unregistered) {
1839 0 : thread->poller_unregistered = true;
1840 0 : spdk_thread_send_msg(thread, _thread_remove_pollers, thread);
1841 : }
1842 : }
1843 :
1844 : /* If the poller was paused, put it on the active_pollers list so that
1845 : * its unregistration can be processed by spdk_thread_poll().
1846 : */
1847 878 : if (poller->state == SPDK_POLLER_STATE_PAUSED) {
1848 9 : TAILQ_REMOVE(&thread->paused_pollers, poller, tailq);
1849 9 : TAILQ_INSERT_TAIL(&thread->active_pollers, poller, tailq);
1850 9 : poller->period_ticks = 0;
1851 : }
1852 :
1853 : /* Simply set the state to unregistered. The poller will get cleaned up
1854 : * in a subsequent call to spdk_thread_poll().
1855 : */
1856 878 : poller->state = SPDK_POLLER_STATE_UNREGISTERED;
1857 : }
1858 :
1859 : void
1860 41 : spdk_poller_pause(struct spdk_poller *poller)
1861 : {
1862 : struct spdk_thread *thread;
1863 :
1864 41 : thread = spdk_get_thread();
1865 41 : if (!thread) {
1866 0 : assert(false);
1867 : return;
1868 : }
1869 :
1870 41 : if (poller->thread != thread) {
1871 0 : wrong_thread(__func__, poller->name, poller->thread, thread);
1872 0 : return;
1873 : }
1874 :
1875 : /* We just set its state to SPDK_POLLER_STATE_PAUSING and let
1876 : * spdk_thread_poll() move it. It allows a poller to be paused from
1877 : * another one's context without breaking the TAILQ_FOREACH_REVERSE_SAFE
1878 : * iteration, or from within itself without breaking the logic to always
1879 : * remove the closest timed poller in the TAILQ_FOREACH_SAFE iteration.
1880 : */
1881 41 : switch (poller->state) {
1882 2 : case SPDK_POLLER_STATE_PAUSED:
1883 : case SPDK_POLLER_STATE_PAUSING:
1884 2 : break;
1885 39 : case SPDK_POLLER_STATE_RUNNING:
1886 : case SPDK_POLLER_STATE_WAITING:
1887 39 : poller->state = SPDK_POLLER_STATE_PAUSING;
1888 39 : break;
1889 0 : default:
1890 0 : assert(false);
1891 : break;
1892 : }
1893 : }
1894 :
1895 : void
1896 40 : spdk_poller_resume(struct spdk_poller *poller)
1897 : {
1898 : struct spdk_thread *thread;
1899 :
1900 40 : thread = spdk_get_thread();
1901 40 : if (!thread) {
1902 0 : assert(false);
1903 : return;
1904 : }
1905 :
1906 40 : if (poller->thread != thread) {
1907 0 : wrong_thread(__func__, poller->name, poller->thread, thread);
1908 0 : return;
1909 : }
1910 :
1911 : /* If a poller is paused it has to be removed from the paused pollers
1912 : * list and put on the active list or timer tree depending on its
1913 : * period_ticks. If a poller is still in the process of being paused,
1914 : * we just need to flip its state back to waiting, as it's already on
1915 : * the appropriate list or tree.
1916 : */
1917 40 : switch (poller->state) {
1918 20 : case SPDK_POLLER_STATE_PAUSED:
1919 20 : TAILQ_REMOVE(&thread->paused_pollers, poller, tailq);
1920 20 : thread_insert_poller(thread, poller);
1921 : /* fallthrough */
1922 29 : case SPDK_POLLER_STATE_PAUSING:
1923 29 : poller->state = SPDK_POLLER_STATE_WAITING;
1924 29 : break;
1925 11 : case SPDK_POLLER_STATE_RUNNING:
1926 : case SPDK_POLLER_STATE_WAITING:
1927 11 : break;
1928 0 : default:
1929 0 : assert(false);
1930 : break;
1931 : }
1932 : }
1933 :
1934 : const char *
1935 2 : spdk_poller_get_name(struct spdk_poller *poller)
1936 : {
1937 2 : return poller->name;
1938 : }
1939 :
1940 : uint64_t
1941 3 : spdk_poller_get_id(struct spdk_poller *poller)
1942 : {
1943 3 : return poller->id;
1944 : }
1945 :
1946 : const char *
1947 9 : spdk_poller_get_state_str(struct spdk_poller *poller)
1948 : {
1949 9 : switch (poller->state) {
1950 4 : case SPDK_POLLER_STATE_WAITING:
1951 4 : return "waiting";
1952 2 : case SPDK_POLLER_STATE_RUNNING:
1953 2 : return "running";
1954 0 : case SPDK_POLLER_STATE_UNREGISTERED:
1955 0 : return "unregistered";
1956 1 : case SPDK_POLLER_STATE_PAUSING:
1957 1 : return "pausing";
1958 2 : case SPDK_POLLER_STATE_PAUSED:
1959 2 : return "paused";
1960 0 : default:
1961 0 : return NULL;
1962 : }
1963 : }
1964 :
1965 : uint64_t
1966 2 : spdk_poller_get_period_ticks(struct spdk_poller *poller)
1967 : {
1968 2 : return poller->period_ticks;
1969 : }
1970 :
1971 : void
1972 2 : spdk_poller_get_stats(struct spdk_poller *poller, struct spdk_poller_stats *stats)
1973 : {
1974 2 : stats->run_count = poller->run_count;
1975 2 : stats->busy_count = poller->busy_count;
1976 2 : }
1977 :
1978 : struct spdk_poller *
1979 0 : spdk_thread_get_first_active_poller(struct spdk_thread *thread)
1980 : {
1981 0 : return TAILQ_FIRST(&thread->active_pollers);
1982 : }
1983 :
1984 : struct spdk_poller *
1985 0 : spdk_thread_get_next_active_poller(struct spdk_poller *prev)
1986 : {
1987 0 : return TAILQ_NEXT(prev, tailq);
1988 : }
1989 :
1990 : struct spdk_poller *
1991 0 : spdk_thread_get_first_timed_poller(struct spdk_thread *thread)
1992 : {
1993 0 : return RB_MIN(timed_pollers_tree, &thread->timed_pollers);
1994 : }
1995 :
1996 : struct spdk_poller *
1997 0 : spdk_thread_get_next_timed_poller(struct spdk_poller *prev)
1998 : {
1999 0 : return RB_NEXT(timed_pollers_tree, &thread->timed_pollers, prev);
2000 : }
2001 :
2002 : struct spdk_poller *
2003 0 : spdk_thread_get_first_paused_poller(struct spdk_thread *thread)
2004 : {
2005 0 : return TAILQ_FIRST(&thread->paused_pollers);
2006 : }
2007 :
2008 : struct spdk_poller *
2009 0 : spdk_thread_get_next_paused_poller(struct spdk_poller *prev)
2010 : {
2011 0 : return TAILQ_NEXT(prev, tailq);
2012 : }
2013 :
2014 : struct spdk_io_channel *
2015 0 : spdk_thread_get_first_io_channel(struct spdk_thread *thread)
2016 : {
2017 0 : return RB_MIN(io_channel_tree, &thread->io_channels);
2018 : }
2019 :
2020 : struct spdk_io_channel *
2021 0 : spdk_thread_get_next_io_channel(struct spdk_io_channel *prev)
2022 : {
2023 0 : return RB_NEXT(io_channel_tree, &thread->io_channels, prev);
2024 : }
2025 :
2026 : uint16_t
2027 0 : spdk_thread_get_trace_id(struct spdk_thread *thread)
2028 : {
2029 0 : return thread->trace_id;
2030 : }
2031 :
2032 : struct call_thread {
2033 : struct spdk_thread *cur_thread;
2034 : spdk_msg_fn fn;
2035 : void *ctx;
2036 :
2037 : struct spdk_thread *orig_thread;
2038 : spdk_msg_fn cpl;
2039 : };
2040 :
2041 : static void
2042 2 : _back_to_orig_thread(void *ctx)
2043 : {
2044 2 : struct call_thread *ct = ctx;
2045 :
2046 2 : assert(ct->orig_thread->for_each_count > 0);
2047 2 : ct->orig_thread->for_each_count--;
2048 :
2049 2 : ct->cpl(ct->ctx);
2050 2 : free(ctx);
2051 2 : }
2052 :
2053 : static void
2054 6 : _on_thread(void *ctx)
2055 : {
2056 6 : struct call_thread *ct = ctx;
2057 : int rc __attribute__((unused));
2058 :
2059 6 : ct->fn(ct->ctx);
2060 :
2061 6 : pthread_mutex_lock(&g_devlist_mutex);
2062 6 : ct->cur_thread = TAILQ_NEXT(ct->cur_thread, tailq);
2063 6 : while (ct->cur_thread && ct->cur_thread->state != SPDK_THREAD_STATE_RUNNING) {
2064 0 : SPDK_DEBUGLOG(thread, "thread %s is not running but still not destroyed.\n",
2065 : ct->cur_thread->name);
2066 0 : ct->cur_thread = TAILQ_NEXT(ct->cur_thread, tailq);
2067 : }
2068 6 : pthread_mutex_unlock(&g_devlist_mutex);
2069 :
2070 6 : if (!ct->cur_thread) {
2071 2 : SPDK_DEBUGLOG(thread, "Completed thread iteration\n");
2072 :
2073 2 : rc = spdk_thread_send_msg(ct->orig_thread, _back_to_orig_thread, ctx);
2074 : } else {
2075 4 : SPDK_DEBUGLOG(thread, "Continuing thread iteration to %s\n",
2076 : ct->cur_thread->name);
2077 :
2078 4 : rc = spdk_thread_send_msg(ct->cur_thread, _on_thread, ctx);
2079 : }
2080 6 : assert(rc == 0);
2081 6 : }
2082 :
2083 : void
2084 2 : spdk_for_each_thread(spdk_msg_fn fn, void *ctx, spdk_msg_fn cpl)
2085 : {
2086 : struct call_thread *ct;
2087 : struct spdk_thread *thread;
2088 : int rc __attribute__((unused));
2089 :
2090 2 : ct = calloc(1, sizeof(*ct));
2091 2 : if (!ct) {
2092 0 : SPDK_ERRLOG("Unable to perform thread iteration\n");
2093 0 : cpl(ctx);
2094 0 : return;
2095 : }
2096 :
2097 2 : ct->fn = fn;
2098 2 : ct->ctx = ctx;
2099 2 : ct->cpl = cpl;
2100 :
2101 2 : thread = _get_thread();
2102 2 : if (!thread) {
2103 0 : SPDK_ERRLOG("No thread allocated\n");
2104 0 : free(ct);
2105 0 : cpl(ctx);
2106 0 : return;
2107 : }
2108 2 : ct->orig_thread = thread;
2109 :
2110 2 : ct->orig_thread->for_each_count++;
2111 :
2112 2 : pthread_mutex_lock(&g_devlist_mutex);
2113 2 : ct->cur_thread = TAILQ_FIRST(&g_threads);
2114 2 : pthread_mutex_unlock(&g_devlist_mutex);
2115 :
2116 2 : SPDK_DEBUGLOG(thread, "Starting thread iteration from %s\n",
2117 : ct->orig_thread->name);
2118 :
2119 2 : rc = spdk_thread_send_msg(ct->cur_thread, _on_thread, ct);
2120 2 : assert(rc == 0);
2121 : }
2122 :
2123 : static inline void
2124 0 : poller_set_interrupt_mode(struct spdk_poller *poller, bool interrupt_mode)
2125 : {
2126 0 : if (poller->state == SPDK_POLLER_STATE_UNREGISTERED) {
2127 0 : return;
2128 : }
2129 :
2130 0 : if (poller->set_intr_cb_fn) {
2131 0 : poller->set_intr_cb_fn(poller, poller->set_intr_cb_arg, interrupt_mode);
2132 : }
2133 : }
2134 :
2135 : void
2136 0 : spdk_thread_set_interrupt_mode(bool enable_interrupt)
2137 : {
2138 0 : struct spdk_thread *thread = _get_thread();
2139 : struct spdk_poller *poller, *tmp;
2140 :
2141 0 : assert(thread);
2142 0 : assert(spdk_interrupt_mode_is_enabled());
2143 :
2144 0 : SPDK_NOTICELOG("Set spdk_thread (%s) to %s mode from %s mode.\n",
2145 : thread->name, enable_interrupt ? "intr" : "poll",
2146 : thread->in_interrupt ? "intr" : "poll");
2147 :
2148 0 : if (thread->in_interrupt == enable_interrupt) {
2149 0 : return;
2150 : }
2151 :
2152 : /* Set pollers to expected mode */
2153 0 : RB_FOREACH_SAFE(poller, timed_pollers_tree, &thread->timed_pollers, tmp) {
2154 0 : poller_set_interrupt_mode(poller, enable_interrupt);
2155 : }
2156 0 : TAILQ_FOREACH_SAFE(poller, &thread->active_pollers, tailq, tmp) {
2157 0 : poller_set_interrupt_mode(poller, enable_interrupt);
2158 : }
2159 : /* All paused pollers will go to work in interrupt mode */
2160 0 : TAILQ_FOREACH_SAFE(poller, &thread->paused_pollers, tailq, tmp) {
2161 0 : poller_set_interrupt_mode(poller, enable_interrupt);
2162 : }
2163 :
2164 0 : thread->in_interrupt = enable_interrupt;
2165 0 : return;
2166 : }
2167 :
2168 : static struct io_device *
2169 9011 : io_device_get(void *io_device)
2170 : {
2171 9011 : struct io_device find = {};
2172 :
2173 9011 : find.io_device = io_device;
2174 9011 : return RB_FIND(io_device_tree, &g_io_devices, &find);
2175 : }
2176 :
2177 : void
2178 2618 : spdk_io_device_register(void *io_device, spdk_io_channel_create_cb create_cb,
2179 : spdk_io_channel_destroy_cb destroy_cb, uint32_t ctx_size,
2180 : const char *name)
2181 : {
2182 : struct io_device *dev, *tmp;
2183 : struct spdk_thread *thread;
2184 :
2185 2618 : assert(io_device != NULL);
2186 2618 : assert(create_cb != NULL);
2187 2618 : assert(destroy_cb != NULL);
2188 :
2189 2618 : thread = spdk_get_thread();
2190 2618 : if (!thread) {
2191 0 : SPDK_ERRLOG("called from non-SPDK thread\n");
2192 0 : assert(false);
2193 : return;
2194 : }
2195 :
2196 2618 : dev = calloc(1, sizeof(struct io_device));
2197 2618 : if (dev == NULL) {
2198 0 : SPDK_ERRLOG("could not allocate io_device\n");
2199 0 : return;
2200 : }
2201 :
2202 2618 : dev->io_device = io_device;
2203 2618 : if (name) {
2204 1704 : snprintf(dev->name, sizeof(dev->name), "%s", name);
2205 : } else {
2206 914 : snprintf(dev->name, sizeof(dev->name), "%p", dev);
2207 : }
2208 2618 : dev->create_cb = create_cb;
2209 2618 : dev->destroy_cb = destroy_cb;
2210 2618 : dev->unregister_cb = NULL;
2211 2618 : dev->ctx_size = ctx_size;
2212 2618 : dev->for_each_count = 0;
2213 2618 : dev->unregistered = false;
2214 2618 : dev->refcnt = 0;
2215 :
2216 2618 : SPDK_DEBUGLOG(thread, "Registering io_device %s (%p) on thread %s\n",
2217 : dev->name, dev->io_device, thread->name);
2218 :
2219 2618 : pthread_mutex_lock(&g_devlist_mutex);
2220 2618 : tmp = RB_INSERT(io_device_tree, &g_io_devices, dev);
2221 2618 : if (tmp != NULL) {
2222 2 : SPDK_ERRLOG("io_device %p already registered (old:%s new:%s)\n",
2223 : io_device, tmp->name, dev->name);
2224 2 : free(dev);
2225 : }
2226 :
2227 2618 : pthread_mutex_unlock(&g_devlist_mutex);
2228 : }
2229 :
2230 : static void
2231 2420 : _finish_unregister(void *arg)
2232 : {
2233 2420 : struct io_device *dev = arg;
2234 : struct spdk_thread *thread;
2235 :
2236 2420 : thread = spdk_get_thread();
2237 2420 : assert(thread == dev->unregister_thread);
2238 :
2239 2420 : SPDK_DEBUGLOG(thread, "Finishing unregistration of io_device %s (%p) on thread %s\n",
2240 : dev->name, dev->io_device, thread->name);
2241 :
2242 2420 : assert(thread->pending_unregister_count > 0);
2243 2420 : thread->pending_unregister_count--;
2244 :
2245 2420 : dev->unregister_cb(dev->io_device);
2246 2420 : free(dev);
2247 2420 : }
2248 :
2249 : static void
2250 2615 : io_device_free(struct io_device *dev)
2251 : {
2252 : int rc __attribute__((unused));
2253 :
2254 2615 : if (dev->unregister_cb == NULL) {
2255 195 : free(dev);
2256 : } else {
2257 2420 : assert(dev->unregister_thread != NULL);
2258 2420 : SPDK_DEBUGLOG(thread, "io_device %s (%p) needs to unregister from thread %s\n",
2259 : dev->name, dev->io_device, dev->unregister_thread->name);
2260 2420 : rc = spdk_thread_send_msg(dev->unregister_thread, _finish_unregister, dev);
2261 2420 : assert(rc == 0);
2262 : }
2263 2615 : }
2264 :
2265 : void
2266 2617 : spdk_io_device_unregister(void *io_device, spdk_io_device_unregister_cb unregister_cb)
2267 : {
2268 : struct io_device *dev;
2269 : uint32_t refcnt;
2270 : struct spdk_thread *thread;
2271 :
2272 2617 : thread = spdk_get_thread();
2273 2617 : if (!thread) {
2274 0 : SPDK_ERRLOG("called from non-SPDK thread\n");
2275 0 : assert(false);
2276 : return;
2277 : }
2278 :
2279 2617 : pthread_mutex_lock(&g_devlist_mutex);
2280 2617 : dev = io_device_get(io_device);
2281 2617 : if (!dev) {
2282 0 : SPDK_ERRLOG("io_device %p not found\n", io_device);
2283 0 : assert(false);
2284 : pthread_mutex_unlock(&g_devlist_mutex);
2285 : return;
2286 : }
2287 :
2288 : /* The for_each_count check differentiates the user attempting to unregister the
2289 : * device a second time, from the internal call to this function that occurs
2290 : * after the for_each_count reaches 0.
2291 : */
2292 2617 : if (dev->pending_unregister && dev->for_each_count > 0) {
2293 0 : SPDK_ERRLOG("io_device %p already has a pending unregister\n", io_device);
2294 0 : assert(false);
2295 : pthread_mutex_unlock(&g_devlist_mutex);
2296 : return;
2297 : }
2298 :
2299 2617 : dev->unregister_cb = unregister_cb;
2300 2617 : dev->unregister_thread = thread;
2301 :
2302 2617 : if (dev->for_each_count > 0) {
2303 1 : SPDK_WARNLOG("io_device %s (%p) has %u for_each calls outstanding\n",
2304 : dev->name, io_device, dev->for_each_count);
2305 1 : dev->pending_unregister = true;
2306 1 : pthread_mutex_unlock(&g_devlist_mutex);
2307 1 : return;
2308 : }
2309 :
2310 2616 : dev->unregistered = true;
2311 2616 : RB_REMOVE(io_device_tree, &g_io_devices, dev);
2312 2616 : refcnt = dev->refcnt;
2313 2616 : pthread_mutex_unlock(&g_devlist_mutex);
2314 :
2315 2616 : SPDK_DEBUGLOG(thread, "Unregistering io_device %s (%p) from thread %s\n",
2316 : dev->name, dev->io_device, thread->name);
2317 :
2318 2616 : if (unregister_cb) {
2319 2420 : thread->pending_unregister_count++;
2320 : }
2321 :
2322 2616 : if (refcnt > 0) {
2323 : /* defer deletion */
2324 1009 : return;
2325 : }
2326 :
2327 1607 : io_device_free(dev);
2328 : }
2329 :
2330 : const char *
2331 0 : spdk_io_device_get_name(struct io_device *dev)
2332 : {
2333 0 : return dev->name;
2334 : }
2335 :
2336 : static struct spdk_io_channel *
2337 10442 : thread_get_io_channel(struct spdk_thread *thread, struct io_device *dev)
2338 : {
2339 10442 : struct spdk_io_channel find = {};
2340 :
2341 10442 : find.dev = dev;
2342 10442 : return RB_FIND(io_channel_tree, &thread->io_channels, &find);
2343 : }
2344 :
2345 : struct spdk_io_channel *
2346 4535 : spdk_get_io_channel(void *io_device)
2347 : {
2348 : struct spdk_io_channel *ch;
2349 : struct spdk_thread *thread;
2350 : struct io_device *dev;
2351 : int rc;
2352 :
2353 4535 : pthread_mutex_lock(&g_devlist_mutex);
2354 4535 : dev = io_device_get(io_device);
2355 4535 : if (dev == NULL) {
2356 1 : SPDK_ERRLOG("could not find io_device %p\n", io_device);
2357 1 : pthread_mutex_unlock(&g_devlist_mutex);
2358 1 : return NULL;
2359 : }
2360 :
2361 4534 : thread = _get_thread();
2362 4534 : if (!thread) {
2363 0 : SPDK_ERRLOG("No thread allocated\n");
2364 0 : pthread_mutex_unlock(&g_devlist_mutex);
2365 0 : return NULL;
2366 : }
2367 :
2368 4534 : if (spdk_unlikely(thread->state == SPDK_THREAD_STATE_EXITED)) {
2369 0 : SPDK_ERRLOG("Thread %s is marked as exited\n", thread->name);
2370 0 : pthread_mutex_unlock(&g_devlist_mutex);
2371 0 : return NULL;
2372 : }
2373 :
2374 4534 : ch = thread_get_io_channel(thread, dev);
2375 4534 : if (ch != NULL) {
2376 1307 : ch->ref++;
2377 :
2378 1307 : SPDK_DEBUGLOG(thread, "Get io_channel %p for io_device %s (%p) on thread %s refcnt %u\n",
2379 : ch, dev->name, dev->io_device, thread->name, ch->ref);
2380 :
2381 : /*
2382 : * An I/O channel already exists for this device on this
2383 : * thread, so return it.
2384 : */
2385 1307 : pthread_mutex_unlock(&g_devlist_mutex);
2386 1307 : spdk_trace_record(TRACE_THREAD_IOCH_GET, 0, 0,
2387 : (uint64_t)spdk_io_channel_get_ctx(ch), ch->ref);
2388 1307 : return ch;
2389 : }
2390 :
2391 3227 : ch = calloc(1, sizeof(*ch) + dev->ctx_size);
2392 3227 : if (ch == NULL) {
2393 0 : SPDK_ERRLOG("could not calloc spdk_io_channel\n");
2394 0 : pthread_mutex_unlock(&g_devlist_mutex);
2395 0 : return NULL;
2396 : }
2397 :
2398 3227 : ch->dev = dev;
2399 3227 : ch->destroy_cb = dev->destroy_cb;
2400 3227 : ch->thread = thread;
2401 3227 : ch->ref = 1;
2402 3227 : ch->destroy_ref = 0;
2403 3227 : RB_INSERT(io_channel_tree, &thread->io_channels, ch);
2404 :
2405 3227 : SPDK_DEBUGLOG(thread, "Get io_channel %p for io_device %s (%p) on thread %s refcnt %u\n",
2406 : ch, dev->name, dev->io_device, thread->name, ch->ref);
2407 :
2408 3227 : dev->refcnt++;
2409 :
2410 3227 : pthread_mutex_unlock(&g_devlist_mutex);
2411 :
2412 3227 : rc = dev->create_cb(io_device, (uint8_t *)ch + sizeof(*ch));
2413 3227 : if (rc != 0) {
2414 3 : pthread_mutex_lock(&g_devlist_mutex);
2415 3 : RB_REMOVE(io_channel_tree, &ch->thread->io_channels, ch);
2416 3 : dev->refcnt--;
2417 3 : free(ch);
2418 3 : SPDK_ERRLOG("could not create io_channel for io_device %s (%p): %s (rc=%d)\n",
2419 : dev->name, io_device, spdk_strerror(-rc), rc);
2420 3 : pthread_mutex_unlock(&g_devlist_mutex);
2421 3 : return NULL;
2422 : }
2423 :
2424 3224 : spdk_trace_record(TRACE_THREAD_IOCH_GET, 0, 0, (uint64_t)spdk_io_channel_get_ctx(ch), 1);
2425 3224 : return ch;
2426 : }
2427 :
2428 : static void
2429 3226 : put_io_channel(void *arg)
2430 : {
2431 3226 : struct spdk_io_channel *ch = arg;
2432 3226 : bool do_remove_dev = true;
2433 : struct spdk_thread *thread;
2434 :
2435 3226 : thread = spdk_get_thread();
2436 3226 : if (!thread) {
2437 0 : SPDK_ERRLOG("called from non-SPDK thread\n");
2438 0 : assert(false);
2439 : return;
2440 : }
2441 :
2442 3226 : SPDK_DEBUGLOG(thread,
2443 : "Releasing io_channel %p for io_device %s (%p) on thread %s\n",
2444 : ch, ch->dev->name, ch->dev->io_device, thread->name);
2445 :
2446 3226 : assert(ch->thread == thread);
2447 :
2448 3226 : ch->destroy_ref--;
2449 :
2450 3226 : if (ch->ref > 0 || ch->destroy_ref > 0) {
2451 : /*
2452 : * Another reference to the associated io_device was requested
2453 : * after this message was sent but before it had a chance to
2454 : * execute.
2455 : */
2456 4 : return;
2457 : }
2458 :
2459 3222 : pthread_mutex_lock(&g_devlist_mutex);
2460 3222 : RB_REMOVE(io_channel_tree, &ch->thread->io_channels, ch);
2461 3222 : pthread_mutex_unlock(&g_devlist_mutex);
2462 :
2463 : /* Don't hold the devlist mutex while the destroy_cb is called. */
2464 3222 : ch->destroy_cb(ch->dev->io_device, spdk_io_channel_get_ctx(ch));
2465 :
2466 3222 : pthread_mutex_lock(&g_devlist_mutex);
2467 3222 : ch->dev->refcnt--;
2468 :
2469 3222 : if (!ch->dev->unregistered) {
2470 2207 : do_remove_dev = false;
2471 : }
2472 :
2473 3222 : if (ch->dev->refcnt > 0) {
2474 146 : do_remove_dev = false;
2475 : }
2476 :
2477 3222 : pthread_mutex_unlock(&g_devlist_mutex);
2478 :
2479 3222 : if (do_remove_dev) {
2480 1008 : io_device_free(ch->dev);
2481 : }
2482 3222 : free(ch);
2483 : }
2484 :
2485 : void
2486 4530 : spdk_put_io_channel(struct spdk_io_channel *ch)
2487 : {
2488 : struct spdk_thread *thread;
2489 : int rc __attribute__((unused));
2490 :
2491 4530 : spdk_trace_record(TRACE_THREAD_IOCH_PUT, 0, 0,
2492 : (uint64_t)spdk_io_channel_get_ctx(ch), ch->ref);
2493 :
2494 4530 : thread = spdk_get_thread();
2495 4530 : if (!thread) {
2496 0 : SPDK_ERRLOG("called from non-SPDK thread\n");
2497 0 : assert(false);
2498 : return;
2499 : }
2500 :
2501 4530 : if (ch->thread != thread) {
2502 0 : wrong_thread(__func__, "ch", ch->thread, thread);
2503 0 : return;
2504 : }
2505 :
2506 4530 : SPDK_DEBUGLOG(thread,
2507 : "Putting io_channel %p for io_device %s (%p) on thread %s refcnt %u\n",
2508 : ch, ch->dev->name, ch->dev->io_device, thread->name, ch->ref);
2509 :
2510 4530 : ch->ref--;
2511 :
2512 4530 : if (ch->ref == 0) {
2513 3226 : ch->destroy_ref++;
2514 3226 : rc = spdk_thread_send_msg(thread, put_io_channel, ch);
2515 3226 : assert(rc == 0);
2516 : }
2517 : }
2518 :
2519 : struct spdk_io_channel *
2520 656910 : spdk_io_channel_from_ctx(void *ctx)
2521 : {
2522 656910 : return (struct spdk_io_channel *)((uint8_t *)ctx - sizeof(struct spdk_io_channel));
2523 : }
2524 :
2525 : struct spdk_thread *
2526 2429 : spdk_io_channel_get_thread(struct spdk_io_channel *ch)
2527 : {
2528 2429 : return ch->thread;
2529 : }
2530 :
2531 : void *
2532 654010 : spdk_io_channel_get_io_device(struct spdk_io_channel *ch)
2533 : {
2534 654010 : return ch->dev->io_device;
2535 : }
2536 :
2537 : const char *
2538 0 : spdk_io_channel_get_io_device_name(struct spdk_io_channel *ch)
2539 : {
2540 0 : return spdk_io_device_get_name(ch->dev);
2541 : }
2542 :
2543 : int
2544 0 : spdk_io_channel_get_ref_count(struct spdk_io_channel *ch)
2545 : {
2546 0 : return ch->ref;
2547 : }
2548 :
2549 : struct spdk_io_channel_iter {
2550 : void *io_device;
2551 : struct io_device *dev;
2552 : spdk_channel_msg fn;
2553 : int status;
2554 : void *ctx;
2555 : struct spdk_io_channel *ch;
2556 :
2557 : struct spdk_thread *cur_thread;
2558 :
2559 : struct spdk_thread *orig_thread;
2560 : spdk_channel_for_each_cpl cpl;
2561 : };
2562 :
2563 : void *
2564 768 : spdk_io_channel_iter_get_io_device(struct spdk_io_channel_iter *i)
2565 : {
2566 768 : return i->io_device;
2567 : }
2568 :
2569 : struct spdk_io_channel *
2570 1221 : spdk_io_channel_iter_get_channel(struct spdk_io_channel_iter *i)
2571 : {
2572 1221 : return i->ch;
2573 : }
2574 :
2575 : void *
2576 3068 : spdk_io_channel_iter_get_ctx(struct spdk_io_channel_iter *i)
2577 : {
2578 3068 : return i->ctx;
2579 : }
2580 :
2581 : static void
2582 1856 : _call_completion(void *ctx)
2583 : {
2584 1856 : struct spdk_io_channel_iter *i = ctx;
2585 :
2586 1856 : assert(i->orig_thread->for_each_count > 0);
2587 1856 : i->orig_thread->for_each_count--;
2588 :
2589 1856 : if (i->cpl != NULL) {
2590 1856 : i->cpl(i, i->status);
2591 : }
2592 1856 : free(i);
2593 1856 : }
2594 :
2595 : static void
2596 1917 : _call_channel(void *ctx)
2597 : {
2598 1917 : struct spdk_io_channel_iter *i = ctx;
2599 : struct spdk_io_channel *ch;
2600 :
2601 : /*
2602 : * It is possible that the channel was deleted before this
2603 : * message had a chance to execute. If so, skip calling
2604 : * the fn() on this thread.
2605 : */
2606 1917 : pthread_mutex_lock(&g_devlist_mutex);
2607 1917 : ch = thread_get_io_channel(i->cur_thread, i->dev);
2608 1917 : pthread_mutex_unlock(&g_devlist_mutex);
2609 :
2610 1917 : if (ch) {
2611 1875 : i->fn(i);
2612 : } else {
2613 42 : spdk_for_each_channel_continue(i, 0);
2614 : }
2615 1917 : }
2616 :
2617 : void
2618 1856 : spdk_for_each_channel(void *io_device, spdk_channel_msg fn, void *ctx,
2619 : spdk_channel_for_each_cpl cpl)
2620 : {
2621 : struct spdk_thread *thread;
2622 : struct spdk_io_channel *ch;
2623 : struct spdk_io_channel_iter *i;
2624 : int rc __attribute__((unused));
2625 :
2626 1856 : i = calloc(1, sizeof(*i));
2627 1856 : if (!i) {
2628 0 : SPDK_ERRLOG("Unable to allocate iterator\n");
2629 0 : assert(false);
2630 : return;
2631 : }
2632 :
2633 1856 : i->io_device = io_device;
2634 1856 : i->fn = fn;
2635 1856 : i->ctx = ctx;
2636 1856 : i->cpl = cpl;
2637 1856 : i->orig_thread = _get_thread();
2638 :
2639 1856 : i->orig_thread->for_each_count++;
2640 :
2641 1856 : pthread_mutex_lock(&g_devlist_mutex);
2642 1856 : i->dev = io_device_get(io_device);
2643 1856 : if (i->dev == NULL) {
2644 0 : SPDK_ERRLOG("could not find io_device %p\n", io_device);
2645 0 : assert(false);
2646 : i->status = -ENODEV;
2647 : goto end;
2648 : }
2649 :
2650 : /* Do not allow new for_each operations if we are already waiting to unregister
2651 : * the device for other for_each operations to complete.
2652 : */
2653 1856 : if (i->dev->pending_unregister) {
2654 0 : SPDK_ERRLOG("io_device %p has a pending unregister\n", io_device);
2655 0 : i->status = -ENODEV;
2656 0 : goto end;
2657 : }
2658 :
2659 2156 : TAILQ_FOREACH(thread, &g_threads, tailq) {
2660 2019 : ch = thread_get_io_channel(thread, i->dev);
2661 2019 : if (ch != NULL) {
2662 1719 : ch->dev->for_each_count++;
2663 1719 : i->cur_thread = thread;
2664 1719 : i->ch = ch;
2665 1719 : pthread_mutex_unlock(&g_devlist_mutex);
2666 1719 : rc = spdk_thread_send_msg(thread, _call_channel, i);
2667 1719 : assert(rc == 0);
2668 1719 : return;
2669 : }
2670 : }
2671 :
2672 137 : end:
2673 137 : pthread_mutex_unlock(&g_devlist_mutex);
2674 :
2675 137 : rc = spdk_thread_send_msg(i->orig_thread, _call_completion, i);
2676 137 : assert(rc == 0);
2677 : }
2678 :
2679 : static void
2680 1 : __pending_unregister(void *arg)
2681 : {
2682 1 : struct io_device *dev = arg;
2683 :
2684 1 : assert(dev->pending_unregister);
2685 1 : assert(dev->for_each_count == 0);
2686 1 : spdk_io_device_unregister(dev->io_device, dev->unregister_cb);
2687 1 : }
2688 :
2689 : void
2690 1917 : spdk_for_each_channel_continue(struct spdk_io_channel_iter *i, int status)
2691 : {
2692 : struct spdk_thread *thread;
2693 : struct spdk_io_channel *ch;
2694 : struct io_device *dev;
2695 : int rc __attribute__((unused));
2696 :
2697 1917 : assert(i->cur_thread == spdk_get_thread());
2698 :
2699 1917 : i->status = status;
2700 :
2701 1917 : pthread_mutex_lock(&g_devlist_mutex);
2702 1917 : dev = i->dev;
2703 1917 : if (status) {
2704 11 : goto end;
2705 : }
2706 :
2707 1906 : thread = TAILQ_NEXT(i->cur_thread, tailq);
2708 3680 : while (thread) {
2709 1972 : ch = thread_get_io_channel(thread, dev);
2710 1972 : if (ch != NULL) {
2711 198 : i->cur_thread = thread;
2712 198 : i->ch = ch;
2713 198 : pthread_mutex_unlock(&g_devlist_mutex);
2714 198 : rc = spdk_thread_send_msg(thread, _call_channel, i);
2715 198 : assert(rc == 0);
2716 198 : return;
2717 : }
2718 1774 : thread = TAILQ_NEXT(thread, tailq);
2719 : }
2720 :
2721 1708 : end:
2722 1719 : dev->for_each_count--;
2723 1719 : i->ch = NULL;
2724 1719 : pthread_mutex_unlock(&g_devlist_mutex);
2725 :
2726 1719 : rc = spdk_thread_send_msg(i->orig_thread, _call_completion, i);
2727 1719 : assert(rc == 0);
2728 :
2729 1719 : pthread_mutex_lock(&g_devlist_mutex);
2730 1719 : if (dev->pending_unregister && dev->for_each_count == 0) {
2731 1 : rc = spdk_thread_send_msg(dev->unregister_thread, __pending_unregister, dev);
2732 1 : assert(rc == 0);
2733 : }
2734 1719 : pthread_mutex_unlock(&g_devlist_mutex);
2735 : }
2736 :
2737 : static void
2738 0 : thread_interrupt_destroy(struct spdk_thread *thread)
2739 : {
2740 0 : struct spdk_fd_group *fgrp = thread->fgrp;
2741 :
2742 0 : SPDK_INFOLOG(thread, "destroy fgrp for thread (%s)\n", thread->name);
2743 :
2744 0 : if (thread->msg_fd < 0) {
2745 0 : return;
2746 : }
2747 :
2748 0 : spdk_fd_group_remove(fgrp, thread->msg_fd);
2749 0 : close(thread->msg_fd);
2750 0 : thread->msg_fd = -1;
2751 :
2752 0 : spdk_fd_group_destroy(fgrp);
2753 0 : thread->fgrp = NULL;
2754 : }
2755 :
2756 : #ifdef __linux__
2757 : static int
2758 0 : thread_interrupt_msg_process(void *arg)
2759 : {
2760 0 : struct spdk_thread *thread = arg;
2761 : struct spdk_thread *orig_thread;
2762 : uint32_t msg_count;
2763 : spdk_msg_fn critical_msg;
2764 0 : int rc = 0;
2765 0 : uint64_t notify = 1;
2766 :
2767 0 : assert(spdk_interrupt_mode_is_enabled());
2768 :
2769 0 : orig_thread = spdk_get_thread();
2770 0 : spdk_set_thread(thread);
2771 :
2772 0 : critical_msg = thread->critical_msg;
2773 0 : if (spdk_unlikely(critical_msg != NULL)) {
2774 0 : critical_msg(NULL);
2775 0 : thread->critical_msg = NULL;
2776 0 : rc = 1;
2777 : }
2778 :
2779 0 : msg_count = msg_queue_run_batch(thread, 0);
2780 0 : if (msg_count) {
2781 0 : rc = 1;
2782 : }
2783 :
2784 0 : SPIN_ASSERT(thread->lock_count == 0, SPIN_ERR_HOLD_DURING_SWITCH);
2785 0 : if (spdk_unlikely(!thread->in_interrupt)) {
2786 : /* The thread transitioned to poll mode in a msg during the above processing.
2787 : * Clear msg_fd since thread messages will be polled directly in poll mode.
2788 : */
2789 0 : rc = read(thread->msg_fd, ¬ify, sizeof(notify));
2790 0 : if (rc < 0 && errno != EAGAIN) {
2791 0 : SPDK_ERRLOG("failed to acknowledge msg queue: %s.\n", spdk_strerror(errno));
2792 : }
2793 : }
2794 :
2795 0 : spdk_set_thread(orig_thread);
2796 0 : return rc;
2797 : }
2798 :
2799 : static int
2800 0 : thread_interrupt_create(struct spdk_thread *thread)
2801 : {
2802 0 : struct spdk_event_handler_opts opts = {};
2803 : int rc;
2804 :
2805 0 : SPDK_INFOLOG(thread, "Create fgrp for thread (%s)\n", thread->name);
2806 :
2807 0 : rc = spdk_fd_group_create(&thread->fgrp);
2808 0 : if (rc) {
2809 0 : thread->msg_fd = -1;
2810 0 : return rc;
2811 : }
2812 :
2813 0 : thread->msg_fd = eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
2814 0 : if (thread->msg_fd < 0) {
2815 0 : rc = -errno;
2816 0 : spdk_fd_group_destroy(thread->fgrp);
2817 0 : thread->fgrp = NULL;
2818 :
2819 0 : return rc;
2820 : }
2821 :
2822 0 : spdk_fd_group_get_default_event_handler_opts(&opts, sizeof(opts));
2823 0 : opts.fd_type = SPDK_FD_TYPE_EVENTFD;
2824 :
2825 0 : return SPDK_FD_GROUP_ADD_EXT(thread->fgrp, thread->msg_fd,
2826 : thread_interrupt_msg_process, thread, &opts);
2827 : }
2828 : #else
2829 : static int
2830 : thread_interrupt_create(struct spdk_thread *thread)
2831 : {
2832 : return -ENOTSUP;
2833 : }
2834 : #endif
2835 :
2836 : static int
2837 0 : _interrupt_wrapper(void *ctx)
2838 : {
2839 0 : struct spdk_interrupt *intr = ctx;
2840 : struct spdk_thread *orig_thread, *thread;
2841 : int rc;
2842 :
2843 0 : orig_thread = spdk_get_thread();
2844 0 : thread = intr->thread;
2845 :
2846 0 : spdk_set_thread(thread);
2847 :
2848 : SPDK_DTRACE_PROBE4(interrupt_fd_process, intr->name, intr->efd,
2849 : intr->fn, intr->arg);
2850 :
2851 0 : rc = intr->fn(intr->arg);
2852 :
2853 0 : SPIN_ASSERT(thread->lock_count == 0, SPIN_ERR_HOLD_DURING_SWITCH);
2854 :
2855 0 : spdk_set_thread(orig_thread);
2856 :
2857 0 : return rc;
2858 : }
2859 :
2860 : struct spdk_interrupt *
2861 0 : spdk_interrupt_register(int efd, spdk_interrupt_fn fn,
2862 : void *arg, const char *name)
2863 : {
2864 0 : return spdk_interrupt_register_for_events(efd, SPDK_INTERRUPT_EVENT_IN, fn, arg, name);
2865 : }
2866 :
2867 : struct spdk_interrupt *
2868 0 : spdk_interrupt_register_for_events(int efd, uint32_t events, spdk_interrupt_fn fn, void *arg,
2869 : const char *name)
2870 : {
2871 0 : struct spdk_event_handler_opts opts = {};
2872 :
2873 0 : spdk_fd_group_get_default_event_handler_opts(&opts, sizeof(opts));
2874 0 : opts.events = events;
2875 0 : opts.fd_type = SPDK_FD_TYPE_DEFAULT;
2876 :
2877 0 : return spdk_interrupt_register_ext(efd, fn, arg, name, &opts);
2878 : }
2879 :
2880 : static struct spdk_interrupt *
2881 0 : alloc_interrupt(int efd, struct spdk_fd_group *fgrp, spdk_interrupt_fn fn, void *arg,
2882 : const char *name)
2883 : {
2884 : struct spdk_thread *thread;
2885 : struct spdk_interrupt *intr;
2886 :
2887 0 : thread = spdk_get_thread();
2888 0 : if (!thread) {
2889 0 : assert(false);
2890 : return NULL;
2891 : }
2892 :
2893 0 : if (spdk_unlikely(thread->state != SPDK_THREAD_STATE_RUNNING)) {
2894 0 : SPDK_ERRLOG("thread %s is marked as exited\n", thread->name);
2895 0 : return NULL;
2896 : }
2897 :
2898 0 : intr = calloc(1, sizeof(*intr));
2899 0 : if (intr == NULL) {
2900 0 : SPDK_ERRLOG("Interrupt handler allocation failed\n");
2901 0 : return NULL;
2902 : }
2903 :
2904 0 : if (name) {
2905 0 : snprintf(intr->name, sizeof(intr->name), "%s", name);
2906 : } else {
2907 0 : snprintf(intr->name, sizeof(intr->name), "%p", fn);
2908 : }
2909 :
2910 0 : assert(efd < 0 || fgrp == NULL);
2911 0 : intr->efd = efd;
2912 0 : intr->fgrp = fgrp;
2913 0 : intr->thread = thread;
2914 0 : intr->fn = fn;
2915 0 : intr->arg = arg;
2916 :
2917 0 : return intr;
2918 : }
2919 :
2920 : struct spdk_interrupt *
2921 0 : spdk_interrupt_register_ext(int efd, spdk_interrupt_fn fn, void *arg, const char *name,
2922 : struct spdk_event_handler_opts *opts)
2923 : {
2924 : struct spdk_interrupt *intr;
2925 : int ret;
2926 :
2927 0 : intr = alloc_interrupt(efd, NULL, fn, arg, name);
2928 0 : if (intr == NULL) {
2929 0 : return NULL;
2930 : }
2931 :
2932 0 : ret = spdk_fd_group_add_ext(intr->thread->fgrp, efd,
2933 0 : _interrupt_wrapper, intr, intr->name, opts);
2934 0 : if (ret != 0) {
2935 0 : SPDK_ERRLOG("thread %s: failed to add fd %d: %s\n",
2936 : intr->thread->name, efd, spdk_strerror(-ret));
2937 0 : free(intr);
2938 0 : return NULL;
2939 : }
2940 :
2941 0 : return intr;
2942 : }
2943 :
2944 : static int
2945 0 : interrupt_fd_group_wrapper(void *wrap_ctx, spdk_fd_fn cb_fn, void *cb_ctx)
2946 : {
2947 0 : struct spdk_interrupt *intr = wrap_ctx;
2948 : struct spdk_thread *orig_thread, *thread;
2949 : int rc;
2950 :
2951 0 : orig_thread = spdk_get_thread();
2952 0 : thread = intr->thread;
2953 :
2954 0 : spdk_set_thread(thread);
2955 0 : rc = cb_fn(cb_ctx);
2956 0 : SPIN_ASSERT(thread->lock_count == 0, SPIN_ERR_HOLD_DURING_SWITCH);
2957 0 : spdk_set_thread(orig_thread);
2958 :
2959 0 : return rc;
2960 : }
2961 :
2962 : struct spdk_interrupt *
2963 0 : spdk_interrupt_register_fd_group(struct spdk_fd_group *fgrp, const char *name)
2964 : {
2965 : struct spdk_interrupt *intr;
2966 : int rc;
2967 :
2968 0 : intr = alloc_interrupt(-1, fgrp, NULL, NULL, name);
2969 0 : if (intr == NULL) {
2970 0 : return NULL;
2971 : }
2972 :
2973 0 : rc = spdk_fd_group_set_wrapper(fgrp, interrupt_fd_group_wrapper, intr);
2974 0 : if (rc != 0) {
2975 0 : SPDK_ERRLOG("thread %s: failed to set wrapper for fd_group %d: %s\n",
2976 : intr->thread->name, spdk_fd_group_get_fd(fgrp), spdk_strerror(-rc));
2977 0 : free(intr);
2978 0 : return NULL;
2979 : }
2980 :
2981 0 : rc = spdk_fd_group_nest(intr->thread->fgrp, fgrp);
2982 0 : if (rc != 0) {
2983 0 : SPDK_ERRLOG("thread %s: failed to nest fd_group %d: %s\n",
2984 : intr->thread->name, spdk_fd_group_get_fd(fgrp), spdk_strerror(-rc));
2985 0 : spdk_fd_group_set_wrapper(fgrp, NULL, NULL);
2986 0 : free(intr);
2987 0 : return NULL;
2988 : }
2989 :
2990 0 : return intr;
2991 : }
2992 :
2993 : void
2994 1 : spdk_interrupt_unregister(struct spdk_interrupt **pintr)
2995 : {
2996 : struct spdk_thread *thread;
2997 : struct spdk_interrupt *intr;
2998 :
2999 1 : intr = *pintr;
3000 1 : if (intr == NULL) {
3001 1 : return;
3002 : }
3003 :
3004 0 : *pintr = NULL;
3005 :
3006 0 : thread = spdk_get_thread();
3007 0 : if (!thread) {
3008 0 : assert(false);
3009 : return;
3010 : }
3011 :
3012 0 : if (intr->thread != thread) {
3013 0 : wrong_thread(__func__, intr->name, intr->thread, thread);
3014 0 : return;
3015 : }
3016 :
3017 0 : if (intr->fgrp != NULL) {
3018 0 : assert(intr->efd < 0);
3019 0 : spdk_fd_group_unnest(thread->fgrp, intr->fgrp);
3020 0 : spdk_fd_group_set_wrapper(thread->fgrp, NULL, NULL);
3021 : } else {
3022 0 : spdk_fd_group_remove(thread->fgrp, intr->efd);
3023 : }
3024 :
3025 0 : free(intr);
3026 : }
3027 :
3028 : int
3029 0 : spdk_interrupt_set_event_types(struct spdk_interrupt *intr,
3030 : enum spdk_interrupt_event_types event_types)
3031 : {
3032 : struct spdk_thread *thread;
3033 :
3034 0 : thread = spdk_get_thread();
3035 0 : if (!thread) {
3036 0 : assert(false);
3037 : return -EINVAL;
3038 : }
3039 :
3040 0 : if (intr->thread != thread) {
3041 0 : wrong_thread(__func__, intr->name, intr->thread, thread);
3042 0 : return -EINVAL;
3043 : }
3044 :
3045 0 : if (intr->efd < 0) {
3046 0 : assert(false);
3047 : return -EINVAL;
3048 : }
3049 :
3050 0 : return spdk_fd_group_event_modify(thread->fgrp, intr->efd, event_types);
3051 : }
3052 :
3053 : int
3054 0 : spdk_thread_get_interrupt_fd(struct spdk_thread *thread)
3055 : {
3056 0 : return spdk_fd_group_get_fd(thread->fgrp);
3057 : }
3058 :
3059 : struct spdk_fd_group *
3060 0 : spdk_thread_get_interrupt_fd_group(struct spdk_thread *thread)
3061 : {
3062 0 : return thread->fgrp;
3063 : }
3064 :
3065 : static bool g_interrupt_mode = false;
3066 :
3067 : int
3068 0 : spdk_interrupt_mode_enable(void)
3069 : {
3070 : /* It must be called once prior to initializing the threading library.
3071 : * g_spdk_msg_mempool will be valid if thread library is initialized.
3072 : */
3073 0 : if (g_spdk_msg_mempool) {
3074 0 : SPDK_ERRLOG("Failed due to threading library is already initialized.\n");
3075 0 : return -1;
3076 : }
3077 :
3078 : #ifdef __linux__
3079 0 : SPDK_NOTICELOG("Set SPDK running in interrupt mode.\n");
3080 0 : g_interrupt_mode = true;
3081 0 : return 0;
3082 : #else
3083 : SPDK_ERRLOG("SPDK interrupt mode supports only Linux platform now.\n");
3084 : g_interrupt_mode = false;
3085 : return -ENOTSUP;
3086 : #endif
3087 : }
3088 :
3089 : bool
3090 169145 : spdk_interrupt_mode_is_enabled(void)
3091 : {
3092 169145 : return g_interrupt_mode;
3093 : }
3094 :
3095 : #define SSPIN_DEBUG_STACK_FRAMES 16
3096 :
3097 : struct sspin_stack {
3098 : void *addrs[SSPIN_DEBUG_STACK_FRAMES];
3099 : uint32_t depth;
3100 : };
3101 :
3102 : struct spdk_spinlock_internal {
3103 : struct sspin_stack init_stack;
3104 : struct sspin_stack lock_stack;
3105 : struct sspin_stack unlock_stack;
3106 : };
3107 :
3108 : static void
3109 1504 : sspin_init_internal(struct spdk_spinlock *sspin)
3110 : {
3111 : #ifdef DEBUG
3112 1504 : sspin->internal = calloc(1, sizeof(*sspin->internal));
3113 : #endif
3114 1504 : }
3115 :
3116 : static void
3117 1493 : sspin_fini_internal(struct spdk_spinlock *sspin)
3118 : {
3119 : #ifdef DEBUG
3120 1493 : free(sspin->internal);
3121 1493 : sspin->internal = NULL;
3122 : #endif
3123 1493 : }
3124 :
3125 : #if defined(DEBUG) && defined(SPDK_HAVE_EXECINFO_H)
3126 : #define SSPIN_GET_STACK(sspin, which) \
3127 : do { \
3128 : if (sspin->internal != NULL) { \
3129 : struct sspin_stack *stack = &sspin->internal->which ## _stack; \
3130 : stack->depth = backtrace(stack->addrs, SPDK_COUNTOF(stack->addrs)); \
3131 : } \
3132 : } while (0)
3133 : #else
3134 : #define SSPIN_GET_STACK(sspin, which) do { } while (0)
3135 : #endif
3136 :
3137 : static void
3138 15 : sspin_stack_print(const char *title, const struct sspin_stack *sspin_stack)
3139 : {
3140 : #ifdef SPDK_HAVE_EXECINFO_H
3141 : char **stack;
3142 : size_t i;
3143 :
3144 : stack = backtrace_symbols(sspin_stack->addrs, sspin_stack->depth);
3145 : if (stack == NULL) {
3146 : SPDK_ERRLOG("Out of memory while allocate stack for %s\n", title);
3147 : return;
3148 : }
3149 : SPDK_ERRLOG(" %s:\n", title);
3150 : for (i = 0; i < sspin_stack->depth; i++) {
3151 : /*
3152 : * This does not print line numbers. In gdb, use something like "list *0x444b6b" or
3153 : * "list *sspin_stack->addrs[0]". Or more conveniently, load the spdk gdb macros
3154 : * and use use "print *sspin" or "print sspin->internal.lock_stack". See
3155 : * gdb_macros.md in the docs directory for details.
3156 : */
3157 : SPDK_ERRLOG(" #%" PRIu64 ": %s\n", i, stack[i]);
3158 : }
3159 : free(stack);
3160 : #endif /* SPDK_HAVE_EXECINFO_H */
3161 15 : }
3162 :
3163 : static void
3164 5 : sspin_stacks_print(const struct spdk_spinlock *sspin)
3165 : {
3166 5 : if (sspin->internal == NULL) {
3167 0 : return;
3168 : }
3169 5 : SPDK_ERRLOG("spinlock %p\n", sspin);
3170 5 : sspin_stack_print("Lock initialized at", &sspin->internal->init_stack);
3171 5 : sspin_stack_print("Last locked at", &sspin->internal->lock_stack);
3172 5 : sspin_stack_print("Last unlocked at", &sspin->internal->unlock_stack);
3173 : }
3174 :
3175 : void
3176 1504 : spdk_spin_init(struct spdk_spinlock *sspin)
3177 : {
3178 : int rc;
3179 :
3180 1504 : memset(sspin, 0, sizeof(*sspin));
3181 1504 : rc = pthread_spin_init(&sspin->spinlock, PTHREAD_PROCESS_PRIVATE);
3182 1504 : SPIN_ASSERT_LOG_STACKS(rc == 0, SPIN_ERR_PTHREAD, sspin);
3183 1504 : sspin_init_internal(sspin);
3184 : SSPIN_GET_STACK(sspin, init);
3185 1504 : sspin->initialized = true;
3186 : }
3187 :
3188 : void
3189 1494 : spdk_spin_destroy(struct spdk_spinlock *sspin)
3190 : {
3191 : int rc;
3192 :
3193 1494 : SPIN_ASSERT_LOG_STACKS(!sspin->destroyed, SPIN_ERR_DESTROYED, sspin);
3194 1494 : SPIN_ASSERT_LOG_STACKS(sspin->initialized, SPIN_ERR_NOT_INITIALIZED, sspin);
3195 1494 : SPIN_ASSERT_LOG_STACKS(sspin->thread == NULL, SPIN_ERR_LOCK_HELD, sspin);
3196 :
3197 1493 : rc = pthread_spin_destroy(&sspin->spinlock);
3198 1493 : SPIN_ASSERT_LOG_STACKS(rc == 0, SPIN_ERR_PTHREAD, sspin);
3199 :
3200 1493 : sspin_fini_internal(sspin);
3201 1493 : sspin->initialized = false;
3202 1493 : sspin->destroyed = true;
3203 : }
3204 :
3205 : void
3206 33128 : spdk_spin_lock(struct spdk_spinlock *sspin)
3207 : {
3208 33128 : struct spdk_thread *thread = spdk_get_thread();
3209 : int rc;
3210 :
3211 33128 : SPIN_ASSERT_LOG_STACKS(!sspin->destroyed, SPIN_ERR_DESTROYED, sspin);
3212 33128 : SPIN_ASSERT_LOG_STACKS(sspin->initialized, SPIN_ERR_NOT_INITIALIZED, sspin);
3213 33128 : SPIN_ASSERT_LOG_STACKS(thread != NULL, SPIN_ERR_NOT_SPDK_THREAD, sspin);
3214 33127 : SPIN_ASSERT_LOG_STACKS(thread != sspin->thread, SPIN_ERR_DEADLOCK, sspin);
3215 :
3216 33126 : rc = pthread_spin_lock(&sspin->spinlock);
3217 33126 : SPIN_ASSERT_LOG_STACKS(rc == 0, SPIN_ERR_PTHREAD, sspin);
3218 :
3219 33126 : sspin->thread = thread;
3220 33126 : sspin->thread->lock_count++;
3221 :
3222 : SSPIN_GET_STACK(sspin, lock);
3223 : }
3224 :
3225 : void
3226 33128 : spdk_spin_unlock(struct spdk_spinlock *sspin)
3227 : {
3228 33128 : struct spdk_thread *thread = spdk_get_thread();
3229 : int rc;
3230 :
3231 33128 : SPIN_ASSERT_LOG_STACKS(!sspin->destroyed, SPIN_ERR_DESTROYED, sspin);
3232 33128 : SPIN_ASSERT_LOG_STACKS(sspin->initialized, SPIN_ERR_NOT_INITIALIZED, sspin);
3233 33128 : SPIN_ASSERT_LOG_STACKS(thread != NULL, SPIN_ERR_NOT_SPDK_THREAD, sspin);
3234 33128 : SPIN_ASSERT_LOG_STACKS(thread == sspin->thread, SPIN_ERR_WRONG_THREAD, sspin);
3235 :
3236 33126 : SPIN_ASSERT_LOG_STACKS(thread->lock_count > 0, SPIN_ERR_LOCK_COUNT, sspin);
3237 33126 : thread->lock_count--;
3238 33126 : sspin->thread = NULL;
3239 :
3240 : SSPIN_GET_STACK(sspin, unlock);
3241 :
3242 33126 : rc = pthread_spin_unlock(&sspin->spinlock);
3243 33126 : SPIN_ASSERT_LOG_STACKS(rc == 0, SPIN_ERR_PTHREAD, sspin);
3244 : }
3245 :
3246 : bool
3247 38579 : spdk_spin_held(struct spdk_spinlock *sspin)
3248 : {
3249 38579 : struct spdk_thread *thread = spdk_get_thread();
3250 :
3251 38579 : SPIN_ASSERT_RETURN(thread != NULL, SPIN_ERR_NOT_SPDK_THREAD, false);
3252 :
3253 38578 : return sspin->thread == thread;
3254 : }
3255 :
3256 : void
3257 0 : spdk_thread_register_post_poller_handler(spdk_post_poller_fn fn, void *fn_arg)
3258 : {
3259 : struct spdk_thread *thr;
3260 :
3261 0 : thr = _get_thread();
3262 0 : assert(thr);
3263 0 : if (spdk_unlikely(thr->num_pp_handlers == SPDK_THREAD_MAX_POST_POLLER_HANDLERS)) {
3264 0 : SPDK_ERRLOG("Too many handlers registered");
3265 0 : return;
3266 : }
3267 :
3268 0 : thr->pp_handlers[thr->num_pp_handlers].fn = fn;
3269 0 : thr->pp_handlers[thr->num_pp_handlers].fn_arg = fn_arg;
3270 0 : thr->num_pp_handlers++;
3271 : }
3272 :
3273 49 : SPDK_LOG_REGISTER_COMPONENT(thread)
|
