Annotation of prex/sys/ipc/msg.c, Revision 1.1.1.1
1.1 nbrk 1: /*-
2: * Copyright (c) 2005-2007, Kohsuke Ohtani
3: * All rights reserved.
4: *
5: * Redistribution and use in source and binary forms, with or without
6: * modification, are permitted provided that the following conditions
7: * are met:
8: * 1. Redistributions of source code must retain the above copyright
9: * notice, this list of conditions and the following disclaimer.
10: * 2. Redistributions in binary form must reproduce the above copyright
11: * notice, this list of conditions and the following disclaimer in the
12: * documentation and/or other materials provided with the distribution.
13: * 3. Neither the name of the author nor the names of any co-contributors
14: * may be used to endorse or promote products derived from this software
15: * without specific prior written permission.
16: *
17: * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
18: * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
19: * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
20: * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
21: * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
22: * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
23: * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
24: * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25: * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
26: * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
27: * SUCH DAMAGE.
28: */
29:
30: /*
31: * msg.c - routines to transmit a message.
32: */
33:
34: /*
35: * Design:
36: *
37: * Messages are sent to the specific object by using msg_send().
38: * The transmission of a message is completely synchronous with
39: * this kernel. This means the thread which sent a message is
40: * blocked until it receives a response from another thread.
41: * msg_receive() performs reception of a message. msg_receive() is
42: * also blocked when no message is reached to the target object.
43: * The receiver thread must answer the message using msg_reply()
44: * after it finishes its message processing.
45: *
46: * The receiver thread can not receive another message until it
47: * replies to the sender. In short, a thread can receive only one
48: * message at once. Once the thread receives message, it can send
49: * another message to different object. This mechanism allows
50: * threads to redirect the sender's request to another thread.
51: *
52: * The message is copied from thread to thread directly without any
53: * kernel buffering. If sent message contains a buffer, sender's
54: * memory region is automatically mapped to the receiver's memory
55: * in kernel. Since there is no page out of memory in this system,
56: * we can copy the message data via physical memory at anytime.
57: */
58:
59: #include <kernel.h>
60: #include <queue.h>
61: #include <event.h>
62: #include <kmem.h>
63: #include <sched.h>
64: #include <thread.h>
65: #include <task.h>
66: #include <vm.h>
67: #include <ipc.h>
68:
69: #define min(a,b) (((a) < (b)) ? (a) : (b))
70:
71: /* forward declarations */
72: static thread_t msg_dequeue(queue_t);
73: static void msg_enqueue(queue_t, thread_t);
74:
75: /* event for IPC operation */
76: static struct event ipc_event;
77:
78: /*
79: * Send a message.
80: *
81: * The current thread will be blocked until any other thread
82: * receives the message and calls msg_reply() for the target
83: * object. When new message has been reached to the object, it
84: * will be received by highest priority thread waiting for
85: * that message. A thread can send a message to any object if
86: * it knows the object id.
87: */
88: int
89: msg_send(object_t obj, void *msg, size_t size)
90: {
91: struct msg_header *hdr;
92: thread_t th;
93: void *kmsg;
94: int rc;
95:
96: if (!user_area(msg))
97: return EFAULT;
98:
99: if (size < sizeof(struct msg_header))
100: return EINVAL;
101:
102: sched_lock();
103:
104: if (!object_valid(obj)) {
105: sched_unlock();
106: return EINVAL;
107: }
108: if (obj->owner != cur_task() && !task_capable(CAP_IPC)) {
109: sched_unlock();
110: return EPERM;
111: }
112: /*
113: * A thread can not send a message when the
114: * thread is already receiving from the target
115: * object. This will obviously cause a deadlock.
116: */
117: if (obj == cur_thread->recvobj) {
118: sched_unlock();
119: return EDEADLK;
120: }
121: /*
122: * Translate message address to the kernel linear
123: * address. So that a receiver thread can access
124: * the message via kernel pointer. We can catch
125: * the page fault here.
126: */
127: if ((kmsg = kmem_map(msg, size)) == NULL) {
128: /* Error - no physical address for the message */
129: sched_unlock();
130: return EFAULT;
131: }
132: /*
133: * The sender ID in the message header is filled
134: * by the kernel. So, the receiver can trust it.
135: */
136: hdr = (struct msg_header *)kmsg;
137: hdr->task = cur_task();
138:
139: /* Save information about the message block. */
140: cur_thread->msgaddr = kmsg;
141: cur_thread->msgsize = size;
142:
143: /*
144: * If receiver already exists, wake it up.
145: * Highest priority thread will get this message.
146: */
147: if (!queue_empty(&obj->recvq)) {
148: th = msg_dequeue(&obj->recvq);
149: sched_unsleep(th, 0);
150: }
151: /*
152: * Sleep until we get a reply message.
153: * Note: Do not touch any data in the object
154: * structure after we wakeup. This is because the
155: * target object may be deleted during we were
156: * sleeping.
157: */
158: cur_thread->sendobj = obj;
159: msg_enqueue(&obj->sendq, cur_thread);
160: rc = sched_sleep(&ipc_event);
161: if (rc == SLP_INTR)
162: queue_remove(&cur_thread->ipc_link);
163: cur_thread->sendobj = NULL;
164:
165: sched_unlock();
166:
167: /*
168: * Check sleep result.
169: */
170: switch (rc) {
171: case SLP_BREAK:
172: return EAGAIN; /* Receiver has been terminated */
173: case SLP_INVAL:
174: return EINVAL; /* Object has been deleted */
175: case SLP_INTR:
176: return EINTR; /* Exception */
177: default:
178: /* DO NOTHING */
179: break;
180: }
181: return 0;
182: }
183:
184: /*
185: * Receive a message.
186: *
187: * A thread can receive a message from the object which was
188: * created by any thread belongs to same task. If the message
189: * has not arrived yet, it blocks until any message comes in.
190: *
191: * The size argument specifies the "maximum" size of the message
192: * buffer to receive. If the sent message is larger than this
193: * size, the kernel will automatically clip the message to the
194: * receive buffer size.
195: *
196: * When message is received, the sender thread is removed from
197: * object's send queue. So, another thread can receive the
198: * subsequent message from that object. This is important for
199: * the multi-thread server which receives some messages
200: * simultaneously.
201: */
202: int
203: msg_receive(object_t obj, void *msg, size_t size)
204: {
205: thread_t th;
206: size_t len;
207: int rc, err = 0;
208:
209: if (!user_area(msg))
210: return EFAULT;
211:
212: sched_lock();
213:
214: if (!object_valid(obj)) {
215: err = EINVAL;
216: goto out;
217: }
218: if (obj->owner != cur_task()) {
219: err = EACCES;
220: goto out;
221: }
222: /*
223: * Check if this thread finished previous receive
224: * operation. A thread can not receive different
225: * messages at once.
226: */
227: if (cur_thread->recvobj) {
228: err = EBUSY;
229: goto out;
230: }
231: cur_thread->recvobj = obj;
232:
233: /*
234: * If no message exists, wait until message arrives.
235: */
236: while (queue_empty(&obj->sendq)) {
237: /*
238: * Block until someone sends the message.
239: */
240: msg_enqueue(&obj->recvq, cur_thread);
241: rc = sched_sleep(&ipc_event);
242: if (rc != 0) {
243: /*
244: * Receive is failed due to some reasons.
245: */
246: switch (rc) {
247: case SLP_INVAL:
248: err = EINVAL; /* Object has been deleted */
249: break;
250: case SLP_INTR:
251: queue_remove(&cur_thread->ipc_link);
252: err = EINTR; /* Got exception */
253: break;
254: default:
255: panic("msg_receive: invalid wake reason");
256: break;
257: }
258: cur_thread->recvobj = NULL;
259: goto out;
260: }
261:
262: /*
263: * Even if this thread is woken by the sender thread,
264: * the message may be received by another thread
265: * before this thread runs. This can occur when
266: * higher priority thread becomes runnable at that
267: * time. So, it is necessary to check the existence
268: * of the sender, again.
269: */
270: }
271:
272: th = msg_dequeue(&obj->sendq);
273:
274: /*
275: * Copy out the message to the user-space.
276: * The smaller buffer size is used as copy length
277: * between sender and receiver thread.
278: */
279: len = min(size, th->msgsize);
280: if (len > 0) {
281: if (umem_copyout(th->msgaddr, msg, len)) {
282: msg_enqueue(&obj->sendq, th);
283: cur_thread->recvobj = NULL;
284: err = EFAULT;
285: goto out;
286: }
287: }
288: /*
289: * Detach the message from the target object.
290: */
291: cur_thread->sender = th;
292: th->receiver = cur_thread;
293: out:
294: sched_unlock();
295: return err;
296: }
297:
298: /*
299: * Send a reply message.
300: *
301: * The target object must be an appropriate object that current
302: * thread has been received from. Otherwise, this function will
303: * be failed.
304: *
305: * Since the target object may already be deleted, we can not
306: * access the data of the object within this routine.
307: */
308: int
309: msg_reply(object_t obj, void *msg, size_t size)
310: {
311: thread_t th;
312: size_t len;
313: int err = 0;
314:
315: if (!user_area(msg))
316: return EFAULT;
317:
318: sched_lock();
319:
320: if (!object_valid(obj) || obj != cur_thread->recvobj) {
321: sched_unlock();
322: return EINVAL;
323: }
324: /*
325: * Check if sender still exists
326: */
327: if (cur_thread->sender == NULL) {
328: err = EINVAL;
329: goto out;
330: }
331: /*
332: * Copy message to the sender's buffer.
333: */
334: th = cur_thread->sender;
335: len = min(size, th->msgsize);
336: if (len > 0) {
337: if (umem_copyin(msg, th->msgaddr, len)) {
338: sched_unlock();
339: return EFAULT;
340: }
341: }
342: /*
343: * Wakeup sender with no error.
344: */
345: sched_unsleep(th, 0);
346: th->receiver = NULL;
347: out:
348: /* Clear transmit state */
349: cur_thread->sender = NULL;
350: cur_thread->recvobj = NULL;
351:
352: sched_unlock();
353: return err;
354: }
355:
356: /*
357: * Clean up pending message operation of specified thread in order
358: * to prevent deadlock. This is called when the thread is killed.
359: * It is necessary to deal with the following conditions.
360: *
361: * If killed thread is sender:
362: * 1. Killed after message is received
363: * -> The received thread will reply to the invalid thread.
364: *
365: * 2. Killed before message is received
366: * -> The thread remains in send queue of the object.
367: *
368: * When thread is receiver:
369: * 3. Killed after message is sent
370: * -> The sender thread continues waiting for reply forever.
371: *
372: * 4. Killed before message is sent
373: * -> The thread remains in receive queue of the object.
374: */
375: void
376: msg_cleanup(thread_t th)
377: {
378:
379: sched_lock();
380:
381: if (th->sendobj) {
382: if (th->receiver)
383: th->receiver->sender = NULL;
384: else
385: queue_remove(&th->ipc_link);
386: }
387: if (th->recvobj) {
388: if (th->sender) {
389: sched_unsleep(th->sender, SLP_BREAK);
390: th->sender->receiver = NULL;
391: } else
392: queue_remove(&th->ipc_link);
393: }
394: sched_unlock();
395: }
396:
397: /*
398: * Cancel all message operation relevant to the specified
399: * object.
400: *
401: * This is called when target object is deleted. All threads
402: * in message queue are woken to avoid deadlock. If the
403: * message has already been received, send/reply operation
404: * continue processing normally.
405: */
406: void
407: msg_cancel(object_t obj)
408: {
409: queue_t q;
410: thread_t th;
411:
412: sched_lock();
413:
414: /*
415: * Force wakeup all threads in the send queue.
416: */
417: while (!queue_empty(&obj->sendq)) {
418: q = dequeue(&obj->sendq);
419: th = queue_entry(q, struct thread, ipc_link);
420: sched_unsleep(th, SLP_INVAL);
421: }
422: /*
423: * Force wakeup all threads waiting for receive.
424: */
425: while (!queue_empty(&obj->recvq)) {
426: q = dequeue(&obj->sendq);
427: th = queue_entry(q, struct thread, ipc_link);
428: sched_unsleep(th, SLP_INVAL);
429: }
430: sched_unlock();
431: }
432:
433: /*
434: * Dequeue thread from specified queue.
435: * The most highest priority thread will be chosen.
436: */
437: static thread_t
438: msg_dequeue(queue_t head)
439: {
440: queue_t q;
441: thread_t th, top;
442:
443: q = queue_first(head);
444: top = queue_entry(q, struct thread, ipc_link);
445: while (!queue_end(head, q)) {
446: th = queue_entry(q, struct thread, ipc_link);
447: if (th->prio < top->prio)
448: top = th;
449: q = queue_next(q);
450: }
451: queue_remove(&top->ipc_link);
452: return top;
453: }
454:
455: static void
456: msg_enqueue(queue_t head, thread_t th)
457: {
458:
459: enqueue(head, &th->ipc_link);
460: }
461:
462: void
463: msg_init(void)
464: {
465:
466: event_init(&ipc_event, "ipc");
467: }
CVSweb
![[BACK]](/icons/back.gif){kind=icon}
Return to msg.c CVS log ![[TXT]](/icons/text.gif){kind=icon}
![[DIR]](/icons/dir.gif){kind=icon}
Up to [local] / prex / sys / ipc