Rename serialize_sleep() to zsleep()
[dragonfly.git] / sys / kern / lwkt_msgport.c
CommitLineData
ece04fd0 1/*
8c10bfcf
MD
2 * Copyright (c) 2003,2004 The DragonFly Project. All rights reserved.
3 *
4 * This code is derived from software contributed to The DragonFly Project
5 * by Matthew Dillon <dillon@backplane.com>
6 *
ece04fd0
MD
7 * Redistribution and use in source and binary forms, with or without
8 * modification, are permitted provided that the following conditions
9 * are met:
8c10bfcf 10 *
ece04fd0
MD
11 * 1. Redistributions of source code must retain the above copyright
12 * notice, this list of conditions and the following disclaimer.
13 * 2. Redistributions in binary form must reproduce the above copyright
8c10bfcf
MD
14 * notice, this list of conditions and the following disclaimer in
15 * the documentation and/or other materials provided with the
16 * distribution.
17 * 3. Neither the name of The DragonFly Project 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
24 * FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
25 * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
26 * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
27 * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
28 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29 * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
31 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
ece04fd0 32 * SUCH DAMAGE.
8c10bfcf 33 *
df2244e3
MD
34 * NOTE! This file may be compiled for userland libraries as well as for
35 * the kernel.
36 *
f2c8b5a3 37 * $DragonFly: src/sys/kern/lwkt_msgport.c,v 1.54 2008/11/26 15:05:42 sephe Exp $
ece04fd0
MD
38 */
39
40#include <sys/param.h>
41#include <sys/systm.h>
42#include <sys/kernel.h>
43#include <sys/proc.h>
44#include <sys/rtprio.h>
45#include <sys/queue.h>
46#include <sys/sysctl.h>
47#include <sys/kthread.h>
e7906ee5 48#include <sys/signalvar.h>
b1b4e5a6 49#include <sys/signal2.h>
ece04fd0
MD
50#include <machine/cpu.h>
51#include <sys/lock.h>
52
53#include <vm/vm.h>
54#include <vm/vm_param.h>
55#include <vm/vm_kern.h>
56#include <vm/vm_object.h>
57#include <vm/vm_page.h>
58#include <vm/vm_map.h>
59#include <vm/vm_pager.h>
60#include <vm/vm_extern.h>
61#include <vm/vm_zone.h>
62
63#include <sys/thread2.h>
64#include <sys/msgport2.h>
fb0f29c4 65#include <sys/spinlock2.h>
74208df3 66#include <sys/serialize.h>
ece04fd0
MD
67
68#include <machine/stdarg.h>
e7906ee5 69#include <machine/cpufunc.h>
ece04fd0
MD
70#ifdef SMP
71#include <machine/smp.h>
72#endif
73
9eeaa8a9
JH
74#include <sys/malloc.h>
75MALLOC_DEFINE(M_LWKTMSG, "lwkt message", "lwkt message");
76
ece04fd0
MD
77/************************************************************************
78 * MESSAGE FUNCTIONS *
79 ************************************************************************/
80
ece04fd0
MD
81/*
82 * lwkt_sendmsg()
83 *
4599cf19
MD
84 * Request asynchronous completion and call lwkt_beginmsg(). The
85 * target port can opt to execute the message synchronously or
86 * asynchronously and this function will automatically queue the
87 * response if the target executes the message synchronously.
ece04fd0 88 *
4599cf19
MD
89 * NOTE: The message is in an indeterminant state until this call
90 * returns. The caller should not mess with it (e.g. try to abort it)
91 * until then.
ece04fd0
MD
92 */
93void
94lwkt_sendmsg(lwkt_port_t port, lwkt_msg_t msg)
95{
96 int error;
97
4599cf19
MD
98 KKASSERT(msg->ms_reply_port != NULL &&
99 (msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == MSGF_DONE);
100 msg->ms_flags &= ~(MSGF_REPLY | MSGF_SYNC | MSGF_DONE);
4fd10eb6 101 if ((error = lwkt_beginmsg(port, msg)) != EASYNC) {
ece04fd0
MD
102 lwkt_replymsg(msg, error);
103 }
104}
105
106/*
107 * lwkt_domsg()
108 *
4599cf19
MD
109 * Request asynchronous completion and call lwkt_beginmsg(). The
110 * target port can opt to execute the message synchronously or
111 * asynchronously and this function will automatically queue the
112 * response if the target executes the message synchronously.
ece04fd0
MD
113 */
114int
a22c590e 115lwkt_domsg(lwkt_port_t port, lwkt_msg_t msg, int flags)
ece04fd0
MD
116{
117 int error;
118
4599cf19
MD
119 KKASSERT(msg->ms_reply_port != NULL &&
120 (msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == MSGF_DONE);
121 msg->ms_flags &= ~(MSGF_REPLY | MSGF_DONE);
122 msg->ms_flags |= MSGF_SYNC;
4fd10eb6 123 if ((error = lwkt_beginmsg(port, msg)) == EASYNC) {
a22c590e 124 error = lwkt_waitmsg(msg, flags);
4599cf19
MD
125 } else {
126 msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
ece04fd0
MD
127 }
128 return(error);
129}
130
fb0f29c4
MD
131/*
132 * lwkt_forwardmsg()
133 *
134 * Forward a message received on one port to another port.
135 */
136int
137lwkt_forwardmsg(lwkt_port_t port, lwkt_msg_t msg)
138{
139 int error;
140
141 crit_enter();
142 KKASSERT((msg->ms_flags & (MSGF_QUEUED|MSGF_DONE|MSGF_REPLY)) == 0);
143 if ((error = port->mp_putport(port, msg)) != EASYNC)
144 lwkt_replymsg(msg, error);
145 crit_exit();
146 return(error);
147}
148
149/*
150 * lwkt_abortmsg()
151 *
152 * Attempt to abort a message. This only works if MSGF_ABORTABLE is set.
153 * The caller must ensure that the message will not be both replied AND
154 * destroyed while the abort is in progress.
155 *
156 * This function issues a callback which might block!
157 */
158void
159lwkt_abortmsg(lwkt_msg_t msg)
160{
161 /*
162 * A critical section protects us from reply IPIs on this cpu.
163 */
164 crit_enter();
165
166 /*
167 * Shortcut the operation if the message has already been returned.
168 * The callback typically constructs a lwkt_msg with the abort request,
169 * issues it synchronously, and waits for completion. The callback
170 * is not required to actually abort the message and the target port,
171 * upon receiving an abort request message generated by the callback
172 * should check whether the original message has already completed or
173 * not.
174 */
175 if (msg->ms_flags & MSGF_ABORTABLE) {
176 if ((msg->ms_flags & (MSGF_DONE|MSGF_REPLY)) == 0)
177 msg->ms_abortfn(msg);
178 }
179 crit_exit();
180}
181
ece04fd0 182/************************************************************************
fb0f29c4 183 * PORT INITIALIZATION API *
ece04fd0
MD
184 ************************************************************************/
185
fb0f29c4
MD
186static void *lwkt_thread_getport(lwkt_port_t port);
187static int lwkt_thread_putport(lwkt_port_t port, lwkt_msg_t msg);
a22c590e
MD
188static int lwkt_thread_waitmsg(lwkt_msg_t msg, int flags);
189static void *lwkt_thread_waitport(lwkt_port_t port, int flags);
fb0f29c4 190static void lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg);
e2ff0223 191static void lwkt_thread_dropmsg(lwkt_port_t port, lwkt_msg_t msg);
fb0f29c4
MD
192
193static void *lwkt_spin_getport(lwkt_port_t port);
194static int lwkt_spin_putport(lwkt_port_t port, lwkt_msg_t msg);
a22c590e
MD
195static int lwkt_spin_waitmsg(lwkt_msg_t msg, int flags);
196static void *lwkt_spin_waitport(lwkt_port_t port, int flags);
fb0f29c4
MD
197static void lwkt_spin_replyport(lwkt_port_t port, lwkt_msg_t msg);
198
74208df3
SZ
199static void *lwkt_serialize_getport(lwkt_port_t port);
200static int lwkt_serialize_putport(lwkt_port_t port, lwkt_msg_t msg);
201static int lwkt_serialize_waitmsg(lwkt_msg_t msg, int flags);
202static void *lwkt_serialize_waitport(lwkt_port_t port, int flags);
203static void lwkt_serialize_replyport(lwkt_port_t port, lwkt_msg_t msg);
204
fb0f29c4
MD
205static void lwkt_null_replyport(lwkt_port_t port, lwkt_msg_t msg);
206static void *lwkt_panic_getport(lwkt_port_t port);
207static int lwkt_panic_putport(lwkt_port_t port, lwkt_msg_t msg);
a22c590e
MD
208static int lwkt_panic_waitmsg(lwkt_msg_t msg, int flags);
209static void *lwkt_panic_waitport(lwkt_port_t port, int flags);
fb0f29c4 210static void lwkt_panic_replyport(lwkt_port_t port, lwkt_msg_t msg);
e2ff0223 211static void lwkt_panic_dropmsg(lwkt_port_t port, lwkt_msg_t msg);
fb0f29c4 212
ece04fd0 213/*
fb0f29c4 214 * Core port initialization (internal)
ece04fd0 215 */
fb0f29c4 216static __inline
ece04fd0 217void
fb0f29c4
MD
218_lwkt_initport(lwkt_port_t port,
219 void *(*gportfn)(lwkt_port_t),
220 int (*pportfn)(lwkt_port_t, lwkt_msg_t),
a22c590e
MD
221 int (*wmsgfn)(lwkt_msg_t, int),
222 void *(*wportfn)(lwkt_port_t, int),
e2ff0223
SZ
223 void (*rportfn)(lwkt_port_t, lwkt_msg_t),
224 void (*dmsgfn)(lwkt_port_t, lwkt_msg_t))
ece04fd0
MD
225{
226 bzero(port, sizeof(*port));
227 TAILQ_INIT(&port->mp_msgq);
f2c8b5a3 228 TAILQ_INIT(&port->mp_msgq_prio);
fb0f29c4
MD
229 port->mp_getport = gportfn;
230 port->mp_putport = pportfn;
a22c590e 231 port->mp_waitmsg = wmsgfn;
fb0f29c4
MD
232 port->mp_waitport = wportfn;
233 port->mp_replyport = rportfn;
e2ff0223 234 port->mp_dropmsg = dmsgfn;
fb0f29c4
MD
235}
236
237/*
361d01dd
MD
238 * Schedule the target thread. If the message flags contains MSGF_NORESCHED
239 * we tell the scheduler not to reschedule if td is at a higher priority.
240 *
241 * This routine is called even if the thread is already scheduled so messages
242 * without NORESCHED will cause the target thread to be rescheduled even if
243 * prior messages did not.
244 */
245static __inline
246void
247_lwkt_schedule_msg(thread_t td, int flags)
248{
60ad1997
SZ
249 if (flags & MSGF_NORESCHED)
250 lwkt_schedule_noresched(td);
251 else
252 lwkt_schedule(td);
361d01dd
MD
253}
254
255/*
fb0f29c4
MD
256 * lwkt_initport_thread()
257 *
258 * Initialize a port for use by a particular thread. The port may
259 * only be used by <td>.
260 */
261void
262lwkt_initport_thread(lwkt_port_t port, thread_t td)
263{
264 _lwkt_initport(port,
265 lwkt_thread_getport,
266 lwkt_thread_putport,
a22c590e 267 lwkt_thread_waitmsg,
fb0f29c4 268 lwkt_thread_waitport,
e2ff0223
SZ
269 lwkt_thread_replyport,
270 lwkt_thread_dropmsg);
fb0f29c4
MD
271 port->mpu_td = td;
272}
273
274/*
275 * lwkt_initport_spin()
276 *
277 * Initialize a port for use with descriptors that might be accessed
278 * via multiple LWPs, processes, or threads. Has somewhat more
279 * overhead then thread ports.
280 */
281void
282lwkt_initport_spin(lwkt_port_t port)
283{
284 _lwkt_initport(port,
285 lwkt_spin_getport,
286 lwkt_spin_putport,
a22c590e 287 lwkt_spin_waitmsg,
fb0f29c4 288 lwkt_spin_waitport,
e2ff0223
SZ
289 lwkt_spin_replyport,
290 lwkt_panic_dropmsg);
fb0f29c4 291 spin_init(&port->mpu_spin);
ece04fd0
MD
292}
293
e7981050
SZ
294/*
295 * lwkt_initport_serialize()
296 *
297 * Initialize a port for use with descriptors that might be accessed
298 * via multiple LWPs, processes, or threads. Callers are assumed to
299 * have held the serializer (slz).
300 */
74208df3
SZ
301void
302lwkt_initport_serialize(lwkt_port_t port, struct lwkt_serialize *slz)
303{
304 _lwkt_initport(port,
305 lwkt_serialize_getport,
306 lwkt_serialize_putport,
307 lwkt_serialize_waitmsg,
308 lwkt_serialize_waitport,
e2ff0223
SZ
309 lwkt_serialize_replyport,
310 lwkt_panic_dropmsg);
74208df3
SZ
311 port->mpu_serialize = slz;
312}
313
a36d140d
MD
314/*
315 * Similar to the standard initport, this function simply marks the message
316 * as being done and does not attempt to return it to an originating port.
317 */
be86a874 318void
fb0f29c4 319lwkt_initport_replyonly_null(lwkt_port_t port)
be86a874 320{
fb0f29c4
MD
321 _lwkt_initport(port,
322 lwkt_panic_getport,
323 lwkt_panic_putport,
a22c590e 324 lwkt_panic_waitmsg,
fb0f29c4 325 lwkt_panic_waitport,
e2ff0223
SZ
326 lwkt_null_replyport,
327 lwkt_panic_dropmsg);
fb0f29c4
MD
328}
329
330/*
331 * Initialize a reply-only port, typically used as a message sink. Such
332 * ports can only be used as a reply port.
333 */
334void
335lwkt_initport_replyonly(lwkt_port_t port,
336 void (*rportfn)(lwkt_port_t, lwkt_msg_t))
337{
338 _lwkt_initport(port, lwkt_panic_getport, lwkt_panic_putport,
a22c590e 339 lwkt_panic_waitmsg, lwkt_panic_waitport,
e2ff0223 340 rportfn, lwkt_panic_dropmsg);
fb0f29c4
MD
341}
342
343void
344lwkt_initport_putonly(lwkt_port_t port,
345 int (*pportfn)(lwkt_port_t, lwkt_msg_t))
346{
347 _lwkt_initport(port, lwkt_panic_getport, pportfn,
a22c590e 348 lwkt_panic_waitmsg, lwkt_panic_waitport,
e2ff0223 349 lwkt_panic_replyport, lwkt_panic_dropmsg);
fb0f29c4
MD
350}
351
352void
353lwkt_initport_panic(lwkt_port_t port)
354{
355 _lwkt_initport(port,
a22c590e
MD
356 lwkt_panic_getport, lwkt_panic_putport,
357 lwkt_panic_waitmsg, lwkt_panic_waitport,
e2ff0223 358 lwkt_panic_replyport, lwkt_panic_dropmsg);
be86a874
MD
359}
360
b44419cb
MD
361static __inline
362void
363_lwkt_pullmsg(lwkt_port_t port, lwkt_msg_t msg)
364{
f2c8b5a3
SZ
365 lwkt_msg_queue *queue;
366
4599cf19
MD
367 /*
368 * normal case, remove and return the message.
369 */
f2c8b5a3
SZ
370 if (__predict_false(msg->ms_flags & MSGF_PRIORITY))
371 queue = &port->mp_msgq_prio;
372 else
373 queue = &port->mp_msgq;
374 TAILQ_REMOVE(queue, msg, ms_node);
4599cf19 375 msg->ms_flags &= ~MSGF_QUEUED;
b44419cb
MD
376}
377
1f2810a4
SZ
378static __inline
379void
380_lwkt_pushmsg(lwkt_port_t port, lwkt_msg_t msg)
381{
f2c8b5a3
SZ
382 lwkt_msg_queue *queue;
383
1f2810a4 384 msg->ms_flags |= MSGF_QUEUED;
f2c8b5a3
SZ
385 if (__predict_false(msg->ms_flags & MSGF_PRIORITY))
386 queue = &port->mp_msgq_prio;
387 else
388 queue = &port->mp_msgq;
389 TAILQ_INSERT_TAIL(queue, msg, ms_node);
e2ff0223
SZ
390}
391
392static __inline
393lwkt_msg_t
394_lwkt_pollmsg(lwkt_port_t port)
395{
f2c8b5a3
SZ
396 lwkt_msg_t msg;
397
398 msg = TAILQ_FIRST(&port->mp_msgq_prio);
399 if (__predict_false(msg != NULL))
400 return msg;
401
402 /*
403 * Priority queue has no message, fallback to non-priority queue.
404 */
e2ff0223 405 return TAILQ_FIRST(&port->mp_msgq);
1f2810a4
SZ
406}
407
408static __inline
409void
410_lwkt_enqueue_reply(lwkt_port_t port, lwkt_msg_t msg)
411{
e2ff0223
SZ
412 _lwkt_pushmsg(port, msg);
413 msg->ms_flags |= MSGF_REPLY | MSGF_DONE;
1f2810a4
SZ
414}
415
fb0f29c4
MD
416/************************************************************************
417 * THREAD PORT BACKEND *
418 ************************************************************************
419 *
420 * This backend is used when the port a message is retrieved from is owned
421 * by a single thread (the calling thread). Messages are IPId to the
422 * correct cpu before being enqueued to a port. Note that this is fairly
423 * optimal since scheduling would have had to do an IPI anyway if the
424 * message were headed to a different cpu.
425 */
df2244e3 426
b8a98473
MD
427#ifdef SMP
428
e7906ee5
MD
429/*
430 * This function completes reply processing for the default case in the
431 * context of the originating cpu.
432 */
ece04fd0
MD
433static
434void
fb0f29c4 435lwkt_thread_replyport_remote(lwkt_msg_t msg)
ece04fd0 436{
4599cf19 437 lwkt_port_t port = msg->ms_reply_port;
361d01dd 438 int flags;
4599cf19 439
fb0f29c4
MD
440 /*
441 * Chase any thread migration that occurs
442 */
443 if (port->mpu_td->td_gd != mycpu) {
444 lwkt_send_ipiq(port->mpu_td->td_gd,
445 (ipifunc1_t)lwkt_thread_replyport_remote, msg);
446 return;
447 }
448
449 /*
450 * Cleanup
451 */
4599cf19
MD
452#ifdef INVARIANTS
453 KKASSERT(msg->ms_flags & MSGF_INTRANSIT);
454 msg->ms_flags &= ~MSGF_INTRANSIT;
455#endif
361d01dd 456 flags = msg->ms_flags;
fb0f29c4 457 if (msg->ms_flags & MSGF_SYNC) {
361d01dd
MD
458 cpu_sfence();
459 msg->ms_flags |= MSGF_REPLY | MSGF_DONE;
fb0f29c4 460 } else {
1f2810a4 461 _lwkt_enqueue_reply(port, msg);
fb0f29c4 462 }
4599cf19 463 if (port->mp_flags & MSGPORTF_WAITING)
361d01dd 464 _lwkt_schedule_msg(port->mpu_td, flags);
ece04fd0
MD
465}
466
b8a98473
MD
467#endif
468
e7906ee5 469/*
fb0f29c4 470 * lwkt_thread_replyport() - Backend to lwkt_replymsg()
4599cf19
MD
471 *
472 * Called with the reply port as an argument but in the context of the
fb0f29c4
MD
473 * original target port. Completion must occur on the target port's
474 * cpu.
4599cf19 475 *
19ab3d9d 476 * The critical section protects us from IPIs on the this CPU.
e7906ee5 477 */
83cbde16 478static
ece04fd0 479void
fb0f29c4 480lwkt_thread_replyport(lwkt_port_t port, lwkt_msg_t msg)
ece04fd0 481{
361d01dd
MD
482 int flags;
483
1a69df94 484 KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED|MSGF_INTRANSIT)) == 0);
6d11f77a 485
4599cf19
MD
486 if (msg->ms_flags & MSGF_SYNC) {
487 /*
488 * If a synchronous completion has been requested, just wakeup
489 * the message without bothering to queue it to the target port.
fb0f29c4
MD
490 *
491 * Assume the target thread is non-preemptive, so no critical
492 * section is required.
4599cf19 493 */
fb0f29c4
MD
494#ifdef SMP
495 if (port->mpu_td->td_gd == mycpu) {
496#endif
361d01dd
MD
497 flags = msg->ms_flags;
498 cpu_sfence();
fb0f29c4
MD
499 msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
500 if (port->mp_flags & MSGPORTF_WAITING)
361d01dd 501 _lwkt_schedule_msg(port->mpu_td, flags);
fb0f29c4
MD
502#ifdef SMP
503 } else {
504#ifdef INVARIANTS
505 msg->ms_flags |= MSGF_INTRANSIT;
506#endif
507 msg->ms_flags |= MSGF_REPLY;
508 lwkt_send_ipiq(port->mpu_td->td_gd,
509 (ipifunc1_t)lwkt_thread_replyport_remote, msg);
510 }
511#endif
4599cf19
MD
512 } else {
513 /*
514 * If an asynchronous completion has been requested the message
9740d7df 515 * must be queued to the reply port.
fb0f29c4
MD
516 *
517 * A critical section is required to interlock the port queue.
4599cf19
MD
518 */
519#ifdef SMP
fb0f29c4 520 if (port->mpu_td->td_gd == mycpu) {
4599cf19 521#endif
fb0f29c4 522 crit_enter();
1f2810a4 523 _lwkt_enqueue_reply(port, msg);
4599cf19 524 if (port->mp_flags & MSGPORTF_WAITING)
361d01dd 525 _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
fb0f29c4 526 crit_exit();
4599cf19
MD
527#ifdef SMP
528 } else {
529#ifdef INVARIANTS
530 msg->ms_flags |= MSGF_INTRANSIT;
531#endif
532 msg->ms_flags |= MSGF_REPLY;
fb0f29c4
MD
533 lwkt_send_ipiq(port->mpu_td->td_gd,
534 (ipifunc1_t)lwkt_thread_replyport_remote, msg);
4599cf19
MD
535 }
536#endif
ece04fd0 537 }
be86a874
MD
538}
539
540/*
e2ff0223
SZ
541 * lwkt_thread_dropmsg() - Backend to lwkt_dropmsg()
542 *
543 * This function could _only_ be used when caller is in the same thread
544 * as the message's target port owner thread.
545 */
546static void
547lwkt_thread_dropmsg(lwkt_port_t port, lwkt_msg_t msg)
548{
549 KASSERT(port->mpu_td == curthread,
550 ("message could only be dropped in the same thread "
551 "as the message target port thread\n"));
552 crit_enter_quick(port->mpu_td);
553 _lwkt_pullmsg(port, msg);
554 msg->ms_flags |= MSGF_DONE;
555 crit_exit_quick(port->mpu_td);
556}
557
558/*
fb0f29c4 559 * lwkt_thread_putport() - Backend to lwkt_beginmsg()
2179a730 560 *
4599cf19
MD
561 * Called with the target port as an argument but in the context of the
562 * reply port. This function always implements an asynchronous put to
563 * the target message port, and thus returns EASYNC.
1fa2b4b4 564 *
4599cf19 565 * The message must already have cleared MSGF_DONE and MSGF_REPLY
ece04fd0 566 */
ece04fd0 567
b8a98473
MD
568#ifdef SMP
569
ece04fd0
MD
570static
571void
fb0f29c4 572lwkt_thread_putport_remote(lwkt_msg_t msg)
ece04fd0 573{
4599cf19
MD
574 lwkt_port_t port = msg->ms_target_port;
575
fb0f29c4
MD
576 /*
577 * Chase any thread migration that occurs
578 */
579 if (port->mpu_td->td_gd != mycpu) {
580 lwkt_send_ipiq(port->mpu_td->td_gd,
581 (ipifunc1_t)lwkt_thread_putport_remote, msg);
582 return;
583 }
584
585 /*
586 * Cleanup
587 */
61d46876 588#ifdef INVARIANTS
4599cf19
MD
589 KKASSERT(msg->ms_flags & MSGF_INTRANSIT);
590 msg->ms_flags &= ~MSGF_INTRANSIT;
61d46876 591#endif
1f2810a4 592 _lwkt_pushmsg(port, msg);
4599cf19 593 if (port->mp_flags & MSGPORTF_WAITING)
361d01dd 594 _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
ece04fd0
MD
595}
596
b8a98473
MD
597#endif
598
fb0f29c4 599static
ece04fd0 600int
fb0f29c4 601lwkt_thread_putport(lwkt_port_t port, lwkt_msg_t msg)
ece04fd0 602{
4599cf19
MD
603 KKASSERT((msg->ms_flags & (MSGF_DONE | MSGF_REPLY)) == 0);
604
e7906ee5 605 msg->ms_target_port = port;
4599cf19 606#ifdef SMP
fb0f29c4 607 if (port->mpu_td->td_gd == mycpu) {
4599cf19 608#endif
fb0f29c4 609 crit_enter();
1f2810a4 610 _lwkt_pushmsg(port, msg);
4599cf19 611 if (port->mp_flags & MSGPORTF_WAITING)
361d01dd 612 _lwkt_schedule_msg(port->mpu_td, msg->ms_flags);
fb0f29c4 613 crit_exit();
4599cf19
MD
614#ifdef SMP
615 } else {
616#ifdef INVARIANTS
617 msg->ms_flags |= MSGF_INTRANSIT;
618#endif
fb0f29c4
MD
619 lwkt_send_ipiq(port->mpu_td->td_gd,
620 (ipifunc1_t)lwkt_thread_putport_remote, msg);
4599cf19
MD
621 }
622#endif
4599cf19 623 return (EASYNC);
ece04fd0
MD
624}
625
626/*
fb0f29c4 627 * lwkt_thread_getport()
ece04fd0 628 *
fb0f29c4
MD
629 * Retrieve the next message from the port or NULL if no messages
630 * are ready.
e7906ee5 631 */
83cbde16 632static
fb0f29c4
MD
633void *
634lwkt_thread_getport(lwkt_port_t port)
e7906ee5 635{
fb0f29c4 636 lwkt_msg_t msg;
e7906ee5 637
fb0f29c4
MD
638 KKASSERT(port->mpu_td == curthread);
639
640 crit_enter_quick(port->mpu_td);
e2ff0223 641 if ((msg = _lwkt_pollmsg(port)) != NULL)
fb0f29c4
MD
642 _lwkt_pullmsg(port, msg);
643 crit_exit_quick(port->mpu_td);
644 return(msg);
ece04fd0
MD
645}
646
647/*
a22c590e 648 * lwkt_thread_waitmsg()
2179a730 649 *
a22c590e
MD
650 * Wait for a particular message to be replied. We must be the only
651 * thread waiting on the message. The port must be owned by the
652 * caller.
ece04fd0 653 */
83cbde16 654static
a22c590e
MD
655int
656lwkt_thread_waitmsg(lwkt_msg_t msg, int flags)
ece04fd0 657{
e2ff0223
SZ
658 KASSERT((msg->ms_flags & MSGF_DROPABLE) == 0,
659 ("can't wait dropable message\n"));
660
a22c590e 661 if ((msg->ms_flags & MSGF_DONE) == 0) {
4599cf19 662 /*
a22c590e 663 * If the done bit was not set we have to block until it is.
4599cf19 664 */
a22c590e
MD
665 lwkt_port_t port = msg->ms_reply_port;
666 thread_t td = curthread;
667 int sentabort;
668
669 KKASSERT(port->mpu_td == td);
a22c590e
MD
670 crit_enter_quick(td);
671 sentabort = 0;
672
673 while ((msg->ms_flags & MSGF_DONE) == 0) {
df2244e3 674 port->mp_flags |= MSGPORTF_WAITING;
a22c590e
MD
675 if (sentabort == 0) {
676 if ((sentabort = lwkt_sleep("waitmsg", flags)) != 0) {
677 lwkt_abortmsg(msg);
678 }
679 } else {
680 lwkt_sleep("waitabt", 0);
681 }
df2244e3
MD
682 port->mp_flags &= ~MSGPORTF_WAITING;
683 }
a22c590e
MD
684 if (msg->ms_flags & MSGF_QUEUED)
685 _lwkt_pullmsg(port, msg);
686 crit_exit_quick(td);
df2244e3
MD
687 } else {
688 /*
a22c590e
MD
689 * If the done bit was set we only have to mess around with the
690 * message if it is queued on the reply port.
df2244e3 691 */
a22c590e
MD
692 if (msg->ms_flags & MSGF_QUEUED) {
693 lwkt_port_t port = msg->ms_reply_port;
694 thread_t td = curthread;
4599cf19 695
a22c590e 696 KKASSERT(port->mpu_td == td);
a22c590e
MD
697 crit_enter_quick(td);
698 _lwkt_pullmsg(port, msg);
699 crit_exit_quick(td);
df2244e3 700 }
a22c590e
MD
701 }
702 return(msg->ms_error);
703}
4599cf19 704
83cbde16 705static
a22c590e
MD
706void *
707lwkt_thread_waitport(lwkt_port_t port, int flags)
708{
709 thread_t td = curthread;
710 lwkt_msg_t msg;
711 int error;
712
713 KKASSERT(port->mpu_td == td);
714 crit_enter_quick(td);
e2ff0223 715 while ((msg = _lwkt_pollmsg(port)) == NULL) {
a22c590e
MD
716 port->mp_flags |= MSGPORTF_WAITING;
717 error = lwkt_sleep("waitport", flags);
718 port->mp_flags &= ~MSGPORTF_WAITING;
719 if (error)
60ad1997 720 goto done;
ece04fd0 721 }
a22c590e
MD
722 _lwkt_pullmsg(port, msg);
723done:
37af14fe 724 crit_exit_quick(td);
ece04fd0
MD
725 return(msg);
726}
727
fb0f29c4
MD
728/************************************************************************
729 * SPIN PORT BACKEND *
730 ************************************************************************
731 *
732 * This backend uses spinlocks instead of making assumptions about which
733 * thread is accessing the port. It must be used when a port is not owned
734 * by a particular thread. This is less optimal then thread ports but
735 * you don't have a choice if there are multiple threads accessing the port.
a22c590e
MD
736 *
737 * Note on MSGPORTF_WAITING - because there may be multiple threads blocked
738 * on the message port, it is the responsibility of the code doing the
739 * wakeup to clear this flag rather then the blocked threads. Some
740 * superfluous wakeups may occur, which is ok.
741 *
742 * XXX synchronous message wakeups are not current optimized.
fb0f29c4
MD
743 */
744
745static
746void *
747lwkt_spin_getport(lwkt_port_t port)
748{
749 lwkt_msg_t msg;
750
751 spin_lock_wr(&port->mpu_spin);
e2ff0223 752 if ((msg = _lwkt_pollmsg(port)) != NULL)
fb0f29c4
MD
753 _lwkt_pullmsg(port, msg);
754 spin_unlock_wr(&port->mpu_spin);
755 return(msg);
756}
757
758static
759int
760lwkt_spin_putport(lwkt_port_t port, lwkt_msg_t msg)
761{
762 int dowakeup;
763
764 KKASSERT((msg->ms_flags & (MSGF_DONE | MSGF_REPLY)) == 0);
765
766 msg->ms_target_port = port;
767 spin_lock_wr(&port->mpu_spin);
1f2810a4 768 _lwkt_pushmsg(port, msg);
fb0f29c4
MD
769 dowakeup = 0;
770 if (port->mp_flags & MSGPORTF_WAITING) {
771 port->mp_flags &= ~MSGPORTF_WAITING;
772 dowakeup = 1;
773 }
774 spin_unlock_wr(&port->mpu_spin);
775 if (dowakeup)
776 wakeup(port);
777 return (EASYNC);
778}
779
780static
a22c590e
MD
781int
782lwkt_spin_waitmsg(lwkt_msg_t msg, int flags)
fb0f29c4 783{
a22c590e 784 lwkt_port_t port;
fb0f29c4
MD
785 int sentabort;
786 int error;
787
e2ff0223
SZ
788 KASSERT((msg->ms_flags & MSGF_DROPABLE) == 0,
789 ("can't wait dropable message\n"));
790
a22c590e
MD
791 if ((msg->ms_flags & MSGF_DONE) == 0) {
792 port = msg->ms_reply_port;
793 sentabort = 0;
794 spin_lock_wr(&port->mpu_spin);
795 while ((msg->ms_flags & MSGF_DONE) == 0) {
796 void *won;
fb0f29c4 797
fb0f29c4 798 /*
a22c590e
MD
799 * If message was sent synchronously from the beginning
800 * the wakeup will be on the message structure, else it
801 * will be on the port structure.
fb0f29c4 802 */
a22c590e
MD
803 if (msg->ms_flags & MSGF_SYNC) {
804 won = msg;
805 } else {
806 won = port;
807 port->mp_flags |= MSGPORTF_WAITING;
808 }
fb0f29c4
MD
809
810 /*
a22c590e
MD
811 * Only messages which support abort can be interrupted.
812 * We must still wait for message completion regardless.
fb0f29c4 813 */
a22c590e
MD
814 if ((flags & PCATCH) && sentabort == 0) {
815 error = msleep(won, &port->mpu_spin, PCATCH, "waitmsg", 0);
816 if (error) {
817 sentabort = error;
818 spin_unlock_wr(&port->mpu_spin);
819 lwkt_abortmsg(msg);
820 spin_lock_wr(&port->mpu_spin);
821 }
822 } else {
823 error = msleep(won, &port->mpu_spin, 0, "waitmsg", 0);
824 }
825 /* see note at the top on the MSGPORTF_WAITING flag */
826 }
827 /*
828 * Turn EINTR into ERESTART if the signal indicates.
829 */
830 if (sentabort && msg->ms_error == EINTR)
831 msg->ms_error = sentabort;
832 if (msg->ms_flags & MSGF_QUEUED)
60ad1997 833 _lwkt_pullmsg(port, msg);
a22c590e
MD
834 spin_unlock_wr(&port->mpu_spin);
835 } else {
836 if (msg->ms_flags & MSGF_QUEUED) {
837 port = msg->ms_reply_port;
838 spin_lock_wr(&port->mpu_spin);
839 _lwkt_pullmsg(port, msg);
840 spin_unlock_wr(&port->mpu_spin);
841 }
842 }
843 return(msg->ms_error);
844}
845
846static
847void *
848lwkt_spin_waitport(lwkt_port_t port, int flags)
849{
850 lwkt_msg_t msg;
851 int error;
852
853 spin_lock_wr(&port->mpu_spin);
e2ff0223 854 while ((msg = _lwkt_pollmsg(port)) == NULL) {
a22c590e
MD
855 port->mp_flags |= MSGPORTF_WAITING;
856 error = msleep(port, &port->mpu_spin, flags, "waitport", 0);
857 /* see note at the top on the MSGPORTF_WAITING flag */
858 if (error) {
859 spin_unlock_wr(&port->mpu_spin);
860 return(NULL);
fb0f29c4
MD
861 }
862 }
a22c590e 863 _lwkt_pullmsg(port, msg);
fb0f29c4
MD
864 spin_unlock_wr(&port->mpu_spin);
865 return(msg);
866}
867
868static
869void
870lwkt_spin_replyport(lwkt_port_t port, lwkt_msg_t msg)
871{
872 int dowakeup;
873
874 KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == 0);
875
876 if (msg->ms_flags & MSGF_SYNC) {
877 /*
878 * If a synchronous completion has been requested, just wakeup
879 * the message without bothering to queue it to the target port.
880 */
881 msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
882 wakeup(msg);
883 } else {
884 /*
885 * If an asynchronous completion has been requested the message
9740d7df 886 * must be queued to the reply port.
fb0f29c4
MD
887 */
888 spin_lock_wr(&port->mpu_spin);
1f2810a4 889 _lwkt_enqueue_reply(port, msg);
fb0f29c4
MD
890 dowakeup = 0;
891 if (port->mp_flags & MSGPORTF_WAITING) {
892 port->mp_flags &= ~MSGPORTF_WAITING;
893 dowakeup = 1;
894 }
895 spin_unlock_wr(&port->mpu_spin);
896 if (dowakeup)
897 wakeup(port);
898 }
899}
900
901/************************************************************************
74208df3
SZ
902 * SERIALIZER PORT BACKEND *
903 ************************************************************************
904 *
905 * This backend uses serializer to protect port accessing. Callers are
906 * assumed to have serializer held. This kind of port is usually created
907 * by network device driver along with _one_ lwkt thread to pipeline
908 * operations which may temporarily release serializer.
909 *
910 * Implementation is based on SPIN PORT BACKEND.
911 */
912
913static
914void *
915lwkt_serialize_getport(lwkt_port_t port)
916{
917 lwkt_msg_t msg;
918
919 ASSERT_SERIALIZED(port->mpu_serialize);
920
e2ff0223 921 if ((msg = _lwkt_pollmsg(port)) != NULL)
74208df3
SZ
922 _lwkt_pullmsg(port, msg);
923 return(msg);
924}
925
926static
927int
928lwkt_serialize_putport(lwkt_port_t port, lwkt_msg_t msg)
929{
930 KKASSERT((msg->ms_flags & (MSGF_DONE | MSGF_REPLY)) == 0);
931 ASSERT_SERIALIZED(port->mpu_serialize);
932
933 msg->ms_target_port = port;
1f2810a4 934 _lwkt_pushmsg(port, msg);
74208df3
SZ
935 if (port->mp_flags & MSGPORTF_WAITING) {
936 port->mp_flags &= ~MSGPORTF_WAITING;
937 wakeup(port);
938 }
939 return (EASYNC);
940}
941
942static
943int
944lwkt_serialize_waitmsg(lwkt_msg_t msg, int flags)
945{
946 lwkt_port_t port;
947 int sentabort;
948 int error;
949
e2ff0223
SZ
950 KASSERT((msg->ms_flags & MSGF_DROPABLE) == 0,
951 ("can't wait dropable message\n"));
952
74208df3
SZ
953 if ((msg->ms_flags & MSGF_DONE) == 0) {
954 port = msg->ms_reply_port;
955
956 ASSERT_SERIALIZED(port->mpu_serialize);
957
958 sentabort = 0;
959 while ((msg->ms_flags & MSGF_DONE) == 0) {
960 void *won;
961
962 /*
963 * If message was sent synchronously from the beginning
964 * the wakeup will be on the message structure, else it
965 * will be on the port structure.
966 */
967 if (msg->ms_flags & MSGF_SYNC) {
968 won = msg;
969 } else {
970 won = port;
971 port->mp_flags |= MSGPORTF_WAITING;
972 }
973
974 /*
975 * Only messages which support abort can be interrupted.
976 * We must still wait for message completion regardless.
977 */
978 if ((flags & PCATCH) && sentabort == 0) {
ed3f6624 979 error = zsleep(won, port->mpu_serialize, PCATCH, "waitmsg", 0);
74208df3
SZ
980 if (error) {
981 sentabort = error;
982 lwkt_serialize_exit(port->mpu_serialize);
983 lwkt_abortmsg(msg);
984 lwkt_serialize_enter(port->mpu_serialize);
985 }
986 } else {
ed3f6624 987 error = zsleep(won, port->mpu_serialize, 0, "waitmsg", 0);
74208df3
SZ
988 }
989 /* see note at the top on the MSGPORTF_WAITING flag */
990 }
991 /*
992 * Turn EINTR into ERESTART if the signal indicates.
993 */
994 if (sentabort && msg->ms_error == EINTR)
995 msg->ms_error = sentabort;
996 if (msg->ms_flags & MSGF_QUEUED)
60ad1997 997 _lwkt_pullmsg(port, msg);
74208df3
SZ
998 } else {
999 if (msg->ms_flags & MSGF_QUEUED) {
1000 port = msg->ms_reply_port;
1001
1002 ASSERT_SERIALIZED(port->mpu_serialize);
1003 _lwkt_pullmsg(port, msg);
1004 }
1005 }
1006 return(msg->ms_error);
1007}
1008
1009static
1010void *
1011lwkt_serialize_waitport(lwkt_port_t port, int flags)
1012{
1013 lwkt_msg_t msg;
1014 int error;
1015
1016 ASSERT_SERIALIZED(port->mpu_serialize);
1017
e2ff0223 1018 while ((msg = _lwkt_pollmsg(port)) == NULL) {
74208df3 1019 port->mp_flags |= MSGPORTF_WAITING;
ed3f6624 1020 error = zsleep(port, port->mpu_serialize, flags, "waitport", 0);
74208df3
SZ
1021 /* see note at the top on the MSGPORTF_WAITING flag */
1022 if (error)
1023 return(NULL);
1024 }
1025 _lwkt_pullmsg(port, msg);
1026 return(msg);
1027}
1028
1029static
1030void
1031lwkt_serialize_replyport(lwkt_port_t port, lwkt_msg_t msg)
1032{
1033 KKASSERT((msg->ms_flags & (MSGF_DONE|MSGF_QUEUED)) == 0);
1034 ASSERT_SERIALIZED(port->mpu_serialize);
1035
1036 if (msg->ms_flags & MSGF_SYNC) {
1037 /*
1038 * If a synchronous completion has been requested, just wakeup
1039 * the message without bothering to queue it to the target port.
1040 */
1041 msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
1042 wakeup(msg);
1043 } else {
1044 /*
1045 * If an asynchronous completion has been requested the message
1046 * must be queued to the reply port.
1047 */
1f2810a4 1048 _lwkt_enqueue_reply(port, msg);
74208df3
SZ
1049 if (port->mp_flags & MSGPORTF_WAITING) {
1050 port->mp_flags &= ~MSGPORTF_WAITING;
1051 wakeup(port);
1052 }
1053 }
1054}
1055
1056/************************************************************************
fb0f29c4
MD
1057 * PANIC AND SPECIAL PORT FUNCTIONS *
1058 ************************************************************************/
1059
1060/*
1061 * You can point a port's reply vector at this function if you just want
1062 * the message marked done, without any queueing or signaling. This is
1063 * often used for structure-embedded messages.
1064 */
1065static
1066void
1067lwkt_null_replyport(lwkt_port_t port, lwkt_msg_t msg)
1068{
1069 msg->ms_flags |= MSGF_DONE | MSGF_REPLY;
1070}
1071
1072static
1073void *
1074lwkt_panic_getport(lwkt_port_t port)
1075{
1076 panic("lwkt_getport() illegal on port %p", port);
1077}
1078
1079static
1080int
1081lwkt_panic_putport(lwkt_port_t port, lwkt_msg_t msg)
1082{
1083 panic("lwkt_begin/do/sendmsg() illegal on port %p msg %p", port, msg);
1084}
1085
1086static
a22c590e
MD
1087int
1088lwkt_panic_waitmsg(lwkt_msg_t msg, int flags)
1089{
1090 panic("port %p msg %p cannot be waited on", msg->ms_reply_port, msg);
1091}
1092
1093static
fb0f29c4 1094void *
a22c590e 1095lwkt_panic_waitport(lwkt_port_t port, int flags)
fb0f29c4 1096{
a22c590e 1097 panic("port %p cannot be waited on", port);
fb0f29c4
MD
1098}
1099
1100static
1101void
1102lwkt_panic_replyport(lwkt_port_t port, lwkt_msg_t msg)
1103{
1104 panic("lwkt_replymsg() is illegal on port %p msg %p", port, msg);
1105}
1106
e2ff0223
SZ
1107static
1108void
1109lwkt_panic_dropmsg(lwkt_port_t port, lwkt_msg_t msg)
1110{
1111 panic("lwkt_dropmsg() is illegal on port %p msg %p", port, msg);
1112}