xref: /spdk/lib/thread/thread.c (revision c10f8e160e42a2a642e8a593b60c2f84561d5eba)
1 /*-
2  *   BSD LICENSE
3  *
4  *   Copyright (c) Intel Corporation.
5  *   All rights reserved.
6  *
7  *   Redistribution and use in source and binary forms, with or without
8  *   modification, are permitted provided that the following conditions
9  *   are met:
10  *
11  *     * Redistributions of source code must retain the above copyright
12  *       notice, this list of conditions and the following disclaimer.
13  *     * Redistributions in binary form must reproduce the above copyright
14  *       notice, this list of conditions and the following disclaimer in
15  *       the documentation and/or other materials provided with the
16  *       distribution.
17  *     * Neither the name of Intel Corporation nor the names of its
18  *       contributors may be used to endorse or promote products derived
19  *       from this software without specific prior written permission.
20  *
21  *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
24  *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
25  *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
26  *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
27  *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
28  *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
29  *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
30  *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31  *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32  */
33 
34 #include "spdk/stdinc.h"
35 
36 #include "spdk/string.h"
37 #include "spdk/thread.h"
38 
39 #include "spdk_internal/log.h"
40 
41 #ifdef __linux__
42 #include <sys/prctl.h>
43 #endif
44 
45 #ifdef __FreeBSD__
46 #include <pthread_np.h>
47 #endif
48 
49 static pthread_mutex_t g_devlist_mutex = PTHREAD_MUTEX_INITIALIZER;
50 
51 struct io_device {
52 	void				*io_device;
53 	char				*name;
54 	spdk_io_channel_create_cb	create_cb;
55 	spdk_io_channel_destroy_cb	destroy_cb;
56 	spdk_io_device_unregister_cb	unregister_cb;
57 	struct spdk_thread		*unregister_thread;
58 	uint32_t			ctx_size;
59 	uint32_t			for_each_count;
60 	TAILQ_ENTRY(io_device)		tailq;
61 
62 	uint32_t			refcnt;
63 
64 	bool				unregistered;
65 };
66 
67 static TAILQ_HEAD(, io_device) g_io_devices = TAILQ_HEAD_INITIALIZER(g_io_devices);
68 
69 struct spdk_thread {
70 	pthread_t			thread_id;
71 	spdk_thread_pass_msg		msg_fn;
72 	spdk_start_poller		start_poller_fn;
73 	spdk_stop_poller		stop_poller_fn;
74 	void				*thread_ctx;
75 	TAILQ_HEAD(, spdk_io_channel)	io_channels;
76 	TAILQ_ENTRY(spdk_thread)	tailq;
77 	char				*name;
78 };
79 
80 static TAILQ_HEAD(, spdk_thread) g_threads = TAILQ_HEAD_INITIALIZER(g_threads);
81 static uint32_t g_thread_count = 0;
82 
83 static struct spdk_thread *
84 _get_thread(void)
85 {
86 	pthread_t thread_id;
87 	struct spdk_thread *thread;
88 
89 	thread_id = pthread_self();
90 
91 	thread = NULL;
92 	TAILQ_FOREACH(thread, &g_threads, tailq) {
93 		if (thread->thread_id == thread_id) {
94 			return thread;
95 		}
96 	}
97 
98 	return NULL;
99 }
100 
101 static void
102 _set_thread_name(const char *thread_name)
103 {
104 #if defined(__linux__)
105 	prctl(PR_SET_NAME, thread_name, 0, 0, 0);
106 #elif defined(__FreeBSD__)
107 	pthread_set_name_np(pthread_self(), thread_name);
108 #else
109 #error missing platform support for thread name
110 #endif
111 }
112 
113 int
114 spdk_thread_lib_init(void)
115 {
116 	return 0;
117 }
118 
119 void
120 spdk_thread_lib_fini(void)
121 {
122 }
123 
124 struct spdk_thread *
125 spdk_allocate_thread(spdk_thread_pass_msg msg_fn,
126 		     spdk_start_poller start_poller_fn,
127 		     spdk_stop_poller stop_poller_fn,
128 		     void *thread_ctx, const char *name)
129 {
130 	struct spdk_thread *thread;
131 
132 	pthread_mutex_lock(&g_devlist_mutex);
133 
134 	thread = _get_thread();
135 	if (thread) {
136 		SPDK_ERRLOG("Double allocated SPDK thread\n");
137 		pthread_mutex_unlock(&g_devlist_mutex);
138 		return NULL;
139 	}
140 
141 	thread = calloc(1, sizeof(*thread));
142 	if (!thread) {
143 		SPDK_ERRLOG("Unable to allocate memory for thread\n");
144 		pthread_mutex_unlock(&g_devlist_mutex);
145 		return NULL;
146 	}
147 
148 	thread->thread_id = pthread_self();
149 	thread->msg_fn = msg_fn;
150 	thread->start_poller_fn = start_poller_fn;
151 	thread->stop_poller_fn = stop_poller_fn;
152 	thread->thread_ctx = thread_ctx;
153 	TAILQ_INIT(&thread->io_channels);
154 	TAILQ_INSERT_TAIL(&g_threads, thread, tailq);
155 	g_thread_count++;
156 	if (name) {
157 		_set_thread_name(name);
158 		thread->name = strdup(name);
159 	} else {
160 		thread->name = spdk_sprintf_alloc("%p", thread);
161 	}
162 
163 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Allocating new thread %s\n", thread->name);
164 
165 	pthread_mutex_unlock(&g_devlist_mutex);
166 
167 	return thread;
168 }
169 
170 void
171 spdk_free_thread(void)
172 {
173 	struct spdk_thread *thread;
174 
175 	pthread_mutex_lock(&g_devlist_mutex);
176 
177 	thread = _get_thread();
178 	if (!thread) {
179 		SPDK_ERRLOG("No thread allocated\n");
180 		pthread_mutex_unlock(&g_devlist_mutex);
181 		return;
182 	}
183 
184 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Freeing thread %s\n", thread->name);
185 
186 	assert(g_thread_count > 0);
187 	g_thread_count--;
188 	TAILQ_REMOVE(&g_threads, thread, tailq);
189 	free(thread->name);
190 	free(thread);
191 
192 	pthread_mutex_unlock(&g_devlist_mutex);
193 }
194 
195 uint32_t
196 spdk_thread_get_count(void)
197 {
198 	/*
199 	 * Return cached value of the current thread count.  We could acquire the
200 	 *  lock and iterate through the TAILQ of threads to count them, but that
201 	 *  count could still be invalidated after we release the lock.
202 	 */
203 	return g_thread_count;
204 }
205 
206 struct spdk_thread *
207 spdk_get_thread(void)
208 {
209 	struct spdk_thread *thread;
210 
211 	pthread_mutex_lock(&g_devlist_mutex);
212 
213 	thread = _get_thread();
214 	if (!thread) {
215 		SPDK_ERRLOG("No thread allocated\n");
216 	}
217 
218 	pthread_mutex_unlock(&g_devlist_mutex);
219 
220 	return thread;
221 }
222 
223 const char *
224 spdk_thread_get_name(const struct spdk_thread *thread)
225 {
226 	return thread->name;
227 }
228 
229 void
230 spdk_thread_send_msg(const struct spdk_thread *thread, spdk_thread_fn fn, void *ctx)
231 {
232 	thread->msg_fn(fn, ctx, thread->thread_ctx);
233 }
234 
235 
236 struct spdk_poller *
237 spdk_poller_register(spdk_poller_fn fn,
238 		     void *arg,
239 		     uint64_t period_microseconds)
240 {
241 	struct spdk_thread *thread;
242 	struct spdk_poller *poller;
243 
244 	thread = spdk_get_thread();
245 	if (!thread) {
246 		assert(false);
247 		return NULL;
248 	}
249 
250 	if (!thread->start_poller_fn || !thread->stop_poller_fn) {
251 		SPDK_ERRLOG("No related functions to start requested poller\n");
252 		assert(false);
253 		return NULL;
254 	}
255 
256 	poller = thread->start_poller_fn(thread->thread_ctx, fn, arg, period_microseconds);
257 	if (!poller) {
258 		SPDK_ERRLOG("Unable to start requested poller\n");
259 		assert(false);
260 		return NULL;
261 	}
262 
263 	return poller;
264 }
265 
266 void
267 spdk_poller_unregister(struct spdk_poller **ppoller)
268 {
269 	struct spdk_thread *thread;
270 	struct spdk_poller *poller;
271 
272 	poller = *ppoller;
273 	if (poller == NULL) {
274 		return;
275 	}
276 
277 	*ppoller = NULL;
278 
279 	thread = spdk_get_thread();
280 
281 	if (thread) {
282 		thread->stop_poller_fn(poller, thread->thread_ctx);
283 	}
284 }
285 
286 struct call_thread {
287 	struct spdk_thread *cur_thread;
288 	spdk_thread_fn fn;
289 	void *ctx;
290 
291 	struct spdk_thread *orig_thread;
292 	spdk_thread_fn cpl;
293 };
294 
295 static void
296 spdk_on_thread(void *ctx)
297 {
298 	struct call_thread *ct = ctx;
299 
300 	ct->fn(ct->ctx);
301 
302 	pthread_mutex_lock(&g_devlist_mutex);
303 	ct->cur_thread = TAILQ_NEXT(ct->cur_thread, tailq);
304 	pthread_mutex_unlock(&g_devlist_mutex);
305 
306 	if (!ct->cur_thread) {
307 		SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Completed thread iteration");
308 
309 		spdk_thread_send_msg(ct->orig_thread, ct->cpl, ct->ctx);
310 		free(ctx);
311 	} else {
312 		SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Continuing thread iteration to %s\n",
313 			      ct->cur_thread->name);
314 
315 		spdk_thread_send_msg(ct->cur_thread, spdk_on_thread, ctx);
316 	}
317 }
318 
319 void
320 spdk_for_each_thread(spdk_thread_fn fn, void *ctx, spdk_thread_fn cpl)
321 {
322 	struct call_thread *ct;
323 
324 	ct = calloc(1, sizeof(*ct));
325 	if (!ct) {
326 		SPDK_ERRLOG("Unable to perform thread iteration\n");
327 		cpl(ctx);
328 		return;
329 	}
330 
331 	ct->fn = fn;
332 	ct->ctx = ctx;
333 	ct->cpl = cpl;
334 
335 	pthread_mutex_lock(&g_devlist_mutex);
336 	ct->orig_thread = _get_thread();
337 	ct->cur_thread = TAILQ_FIRST(&g_threads);
338 	pthread_mutex_unlock(&g_devlist_mutex);
339 
340 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Starting thread iteration from %s\n",
341 		      ct->orig_thread->name);
342 
343 	spdk_thread_send_msg(ct->cur_thread, spdk_on_thread, ct);
344 }
345 
346 void
347 spdk_io_device_register(void *io_device, spdk_io_channel_create_cb create_cb,
348 			spdk_io_channel_destroy_cb destroy_cb, uint32_t ctx_size,
349 			const char *name)
350 {
351 	struct io_device *dev, *tmp;
352 
353 	assert(io_device != NULL);
354 	assert(create_cb != NULL);
355 	assert(destroy_cb != NULL);
356 
357 	dev = calloc(1, sizeof(struct io_device));
358 	if (dev == NULL) {
359 		SPDK_ERRLOG("could not allocate io_device\n");
360 		return;
361 	}
362 
363 	dev->io_device = io_device;
364 	if (name) {
365 		dev->name = strdup(name);
366 	} else {
367 		dev->name = spdk_sprintf_alloc("%p", dev);
368 	}
369 	dev->create_cb = create_cb;
370 	dev->destroy_cb = destroy_cb;
371 	dev->unregister_cb = NULL;
372 	dev->ctx_size = ctx_size;
373 	dev->for_each_count = 0;
374 	dev->unregistered = false;
375 	dev->refcnt = 0;
376 
377 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Registering io_device %s (%p) on thread %s\n",
378 		      dev->name, dev->io_device, spdk_get_thread()->name);
379 
380 	pthread_mutex_lock(&g_devlist_mutex);
381 	TAILQ_FOREACH(tmp, &g_io_devices, tailq) {
382 		if (tmp->io_device == io_device) {
383 			SPDK_ERRLOG("io_device %p already registered\n", io_device);
384 			free(dev->name);
385 			free(dev);
386 			pthread_mutex_unlock(&g_devlist_mutex);
387 			return;
388 		}
389 	}
390 	TAILQ_INSERT_TAIL(&g_io_devices, dev, tailq);
391 	pthread_mutex_unlock(&g_devlist_mutex);
392 }
393 
394 static void
395 _finish_unregister(void *arg)
396 {
397 	struct io_device *dev = arg;
398 
399 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Finishing unregistration of io_device %s (%p) on thread %s\n",
400 		      dev->name, dev->io_device, dev->unregister_thread->name);
401 
402 	dev->unregister_cb(dev->io_device);
403 	free(dev->name);
404 	free(dev);
405 }
406 
407 static void
408 _spdk_io_device_free(struct io_device *dev)
409 {
410 	if (dev->unregister_cb == NULL) {
411 		free(dev->name);
412 		free(dev);
413 	} else {
414 		assert(dev->unregister_thread != NULL);
415 		SPDK_DEBUGLOG(SPDK_LOG_THREAD, "io_device %s (%p) needs to unregister from thread %s\n",
416 			      dev->name, dev->io_device, dev->unregister_thread->name);
417 		spdk_thread_send_msg(dev->unregister_thread, _finish_unregister, dev);
418 	}
419 }
420 
421 void
422 spdk_io_device_unregister(void *io_device, spdk_io_device_unregister_cb unregister_cb)
423 {
424 	struct io_device *dev;
425 	uint32_t refcnt;
426 	struct spdk_thread *thread;
427 
428 	thread = spdk_get_thread();
429 
430 	pthread_mutex_lock(&g_devlist_mutex);
431 	TAILQ_FOREACH(dev, &g_io_devices, tailq) {
432 		if (dev->io_device == io_device) {
433 			break;
434 		}
435 	}
436 
437 	if (!dev) {
438 		SPDK_ERRLOG("io_device %p not found\n", io_device);
439 		assert(false);
440 		pthread_mutex_unlock(&g_devlist_mutex);
441 		return;
442 	}
443 
444 	if (dev->for_each_count > 0) {
445 		SPDK_ERRLOG("io_device %p has %u for_each calls outstanding\n", io_device, dev->for_each_count);
446 		pthread_mutex_unlock(&g_devlist_mutex);
447 		return;
448 	}
449 
450 	dev->unregister_cb = unregister_cb;
451 	dev->unregistered = true;
452 	TAILQ_REMOVE(&g_io_devices, dev, tailq);
453 	refcnt = dev->refcnt;
454 	dev->unregister_thread = thread;
455 	pthread_mutex_unlock(&g_devlist_mutex);
456 
457 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Unregistering io_device %s (%p) from thread %s\n",
458 		      dev->name, dev->io_device, thread->name);
459 
460 	if (refcnt > 0) {
461 		/* defer deletion */
462 		return;
463 	}
464 
465 	_spdk_io_device_free(dev);
466 }
467 
468 struct spdk_io_channel *
469 spdk_get_io_channel(void *io_device)
470 {
471 	struct spdk_io_channel *ch;
472 	struct spdk_thread *thread;
473 	struct io_device *dev;
474 	int rc;
475 
476 	pthread_mutex_lock(&g_devlist_mutex);
477 	TAILQ_FOREACH(dev, &g_io_devices, tailq) {
478 		if (dev->io_device == io_device) {
479 			break;
480 		}
481 	}
482 	if (dev == NULL) {
483 		SPDK_ERRLOG("could not find io_device %p\n", io_device);
484 		pthread_mutex_unlock(&g_devlist_mutex);
485 		return NULL;
486 	}
487 
488 	thread = _get_thread();
489 	if (!thread) {
490 		SPDK_ERRLOG("No thread allocated\n");
491 		pthread_mutex_unlock(&g_devlist_mutex);
492 		return NULL;
493 	}
494 
495 	TAILQ_FOREACH(ch, &thread->io_channels, tailq) {
496 		if (ch->dev == dev) {
497 			ch->ref++;
498 
499 			SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Get io_channel %p for io_device %s (%p) on thread %s refcnt %u\n",
500 				      ch, dev->name, dev->io_device, thread->name, ch->ref);
501 
502 			/*
503 			 * An I/O channel already exists for this device on this
504 			 *  thread, so return it.
505 			 */
506 			pthread_mutex_unlock(&g_devlist_mutex);
507 			return ch;
508 		}
509 	}
510 
511 	ch = calloc(1, sizeof(*ch) + dev->ctx_size);
512 	if (ch == NULL) {
513 		SPDK_ERRLOG("could not calloc spdk_io_channel\n");
514 		pthread_mutex_unlock(&g_devlist_mutex);
515 		return NULL;
516 	}
517 
518 	ch->dev = dev;
519 	ch->destroy_cb = dev->destroy_cb;
520 	ch->thread = thread;
521 	ch->ref = 1;
522 	ch->destroy_ref = 0;
523 	TAILQ_INSERT_TAIL(&thread->io_channels, ch, tailq);
524 
525 	SPDK_DEBUGLOG(SPDK_LOG_THREAD, "Get io_channel %p for io_device %s (%p) on thread %s refcnt %u\n",
526 		      ch, dev->name, dev->io_device, thread->name, ch->ref);
527 
528 	dev->refcnt++;
529 
530 	pthread_mutex_unlock(&g_devlist_mutex);
531 
532 	rc = dev->create_cb(io_device, (uint8_t *)ch + sizeof(*ch));
533 	if (rc == -1) {
534 		pthread_mutex_lock(&g_devlist_mutex);
535 		TAILQ_REMOVE(&ch->thread->io_channels, ch, tailq);
536 		dev->refcnt--;
537 		free(ch);
538 		pthread_mutex_unlock(&g_devlist_mutex);
539 		return NULL;
540 	}
541 
542 	return ch;
543 }
544 
545 static void
546 _spdk_put_io_channel(void *arg)
547 {
548 	struct spdk_io_channel *ch = arg;
549 	bool do_remove_dev = true;
550 
551 	SPDK_DEBUGLOG(SPDK_LOG_THREAD,
552 		      "Releasing io_channel %p for io_device %s (%p). Channel thread %p. Current thread %s\n",
553 		      ch, ch->dev->name, ch->dev->io_device, ch->thread, spdk_get_thread()->name);
554 
555 	assert(ch->thread == spdk_get_thread());
556 
557 	ch->destroy_ref--;
558 
559 	if (ch->ref > 0 || ch->destroy_ref > 0) {
560 		/*
561 		 * Another reference to the associated io_device was requested
562 		 *  after this message was sent but before it had a chance to
563 		 *  execute.
564 		 */
565 		return;
566 	}
567 
568 	pthread_mutex_lock(&g_devlist_mutex);
569 	TAILQ_REMOVE(&ch->thread->io_channels, ch, tailq);
570 	pthread_mutex_unlock(&g_devlist_mutex);
571 
572 	/* Don't hold the devlist mutex while the destroy_cb is called. */
573 	ch->destroy_cb(ch->dev->io_device, spdk_io_channel_get_ctx(ch));
574 
575 	pthread_mutex_lock(&g_devlist_mutex);
576 	ch->dev->refcnt--;
577 
578 	if (!ch->dev->unregistered) {
579 		do_remove_dev = false;
580 	}
581 
582 	if (ch->dev->refcnt > 0) {
583 		do_remove_dev = false;
584 	}
585 
586 	pthread_mutex_unlock(&g_devlist_mutex);
587 
588 	if (do_remove_dev) {
589 		_spdk_io_device_free(ch->dev);
590 	}
591 	free(ch);
592 }
593 
594 void
595 spdk_put_io_channel(struct spdk_io_channel *ch)
596 {
597 	SPDK_DEBUGLOG(SPDK_LOG_THREAD,
598 		      "Putting io_channel %p for io_device %s (%p) on thread %s refcnt %u\n",
599 		      ch, ch->dev->name, ch->dev->io_device, ch->thread->name, ch->ref);
600 
601 	ch->ref--;
602 
603 	if (ch->ref == 0) {
604 		ch->destroy_ref++;
605 		spdk_thread_send_msg(ch->thread, _spdk_put_io_channel, ch);
606 	}
607 }
608 
609 struct spdk_io_channel *
610 spdk_io_channel_from_ctx(void *ctx)
611 {
612 	return (struct spdk_io_channel *)((uint8_t *)ctx - sizeof(struct spdk_io_channel));
613 }
614 
615 struct spdk_thread *
616 spdk_io_channel_get_thread(struct spdk_io_channel *ch)
617 {
618 	return ch->thread;
619 }
620 
621 struct spdk_io_channel_iter {
622 	void *io_device;
623 	struct io_device *dev;
624 	spdk_channel_msg fn;
625 	int status;
626 	void *ctx;
627 	struct spdk_io_channel *ch;
628 
629 	struct spdk_thread *cur_thread;
630 
631 	struct spdk_thread *orig_thread;
632 	spdk_channel_for_each_cpl cpl;
633 };
634 
635 void *
636 spdk_io_channel_iter_get_io_device(struct spdk_io_channel_iter *i)
637 {
638 	return i->io_device;
639 }
640 
641 struct spdk_io_channel *
642 spdk_io_channel_iter_get_channel(struct spdk_io_channel_iter *i)
643 {
644 	return i->ch;
645 }
646 
647 void *
648 spdk_io_channel_iter_get_ctx(struct spdk_io_channel_iter *i)
649 {
650 	return i->ctx;
651 }
652 
653 static void
654 _call_completion(void *ctx)
655 {
656 	struct spdk_io_channel_iter *i = ctx;
657 
658 	if (i->cpl != NULL) {
659 		i->cpl(i, i->status);
660 	}
661 	free(i);
662 }
663 
664 static void
665 _call_channel(void *ctx)
666 {
667 	struct spdk_io_channel_iter *i = ctx;
668 	struct spdk_io_channel *ch;
669 
670 	/*
671 	 * It is possible that the channel was deleted before this
672 	 *  message had a chance to execute.  If so, skip calling
673 	 *  the fn() on this thread.
674 	 */
675 	pthread_mutex_lock(&g_devlist_mutex);
676 	TAILQ_FOREACH(ch, &i->cur_thread->io_channels, tailq) {
677 		if (ch->dev->io_device == i->io_device) {
678 			break;
679 		}
680 	}
681 	pthread_mutex_unlock(&g_devlist_mutex);
682 
683 	if (ch) {
684 		i->fn(i);
685 	} else {
686 		spdk_for_each_channel_continue(i, 0);
687 	}
688 }
689 
690 void
691 spdk_for_each_channel(void *io_device, spdk_channel_msg fn, void *ctx,
692 		      spdk_channel_for_each_cpl cpl)
693 {
694 	struct spdk_thread *thread;
695 	struct spdk_io_channel *ch;
696 	struct spdk_io_channel_iter *i;
697 
698 	i = calloc(1, sizeof(*i));
699 	if (!i) {
700 		SPDK_ERRLOG("Unable to allocate iterator\n");
701 		return;
702 	}
703 
704 	i->io_device = io_device;
705 	i->fn = fn;
706 	i->ctx = ctx;
707 	i->cpl = cpl;
708 
709 	pthread_mutex_lock(&g_devlist_mutex);
710 	i->orig_thread = _get_thread();
711 
712 	TAILQ_FOREACH(thread, &g_threads, tailq) {
713 		TAILQ_FOREACH(ch, &thread->io_channels, tailq) {
714 			if (ch->dev->io_device == io_device) {
715 				ch->dev->for_each_count++;
716 				i->dev = ch->dev;
717 				i->cur_thread = thread;
718 				i->ch = ch;
719 				pthread_mutex_unlock(&g_devlist_mutex);
720 				spdk_thread_send_msg(thread, _call_channel, i);
721 				return;
722 			}
723 		}
724 	}
725 
726 	pthread_mutex_unlock(&g_devlist_mutex);
727 
728 	spdk_thread_send_msg(i->orig_thread, _call_completion, i);
729 }
730 
731 void
732 spdk_for_each_channel_continue(struct spdk_io_channel_iter *i, int status)
733 {
734 	struct spdk_thread *thread;
735 	struct spdk_io_channel *ch;
736 
737 	assert(i->cur_thread == spdk_get_thread());
738 
739 	i->status = status;
740 
741 	pthread_mutex_lock(&g_devlist_mutex);
742 	if (status) {
743 		goto end;
744 	}
745 	thread = TAILQ_NEXT(i->cur_thread, tailq);
746 	while (thread) {
747 		TAILQ_FOREACH(ch, &thread->io_channels, tailq) {
748 			if (ch->dev->io_device == i->io_device) {
749 				i->cur_thread = thread;
750 				i->ch = ch;
751 				pthread_mutex_unlock(&g_devlist_mutex);
752 				spdk_thread_send_msg(thread, _call_channel, i);
753 				return;
754 			}
755 		}
756 		thread = TAILQ_NEXT(thread, tailq);
757 	}
758 
759 end:
760 	i->dev->for_each_count--;
761 	i->ch = NULL;
762 	pthread_mutex_unlock(&g_devlist_mutex);
763 
764 	spdk_thread_send_msg(i->orig_thread, _call_completion, i);
765 }
766 
767 
768 SPDK_LOG_REGISTER_COMPONENT("thread", SPDK_LOG_THREAD)
769