projects / deliverable / linux.git / blob
? search:
summary | shortlog | log | commit | commitdiff | tree
blame | history | raw | HEAD
sgi-xp: isolate xpc_vars structure to sn2 only
[deliverable/linux.git] / drivers / misc / sgi-xp / xpc_channel.c
1 /*
2 * This file is subject to the terms and conditions of the GNU General Public
3 * License. See the file "COPYING" in the main directory of this archive
4 * for more details.
5 *
6 * Copyright (c) 2004-2008 Silicon Graphics, Inc. All Rights Reserved.
7 */
8
9 /*
10 * Cross Partition Communication (XPC) channel support.
11 *
12 * This is the part of XPC that manages the channels and
13 * sends/receives messages across them to/from other partitions.
14 *
15 */
16
17 #include <linux/kernel.h>
18 #include <linux/init.h>
19 #include <linux/sched.h>
20 #include <linux/cache.h>
21 #include <linux/interrupt.h>
22 #include <linux/mutex.h>
23 #include <linux/completion.h>
24 #include <asm/sn/sn_sal.h>
25 #include "xpc.h"
26
27 /*
28 * Guarantee that the kzalloc'd memory is cacheline aligned.
29 */
30 void *
31 xpc_kzalloc_cacheline_aligned(size_t size, gfp_t flags, void **base)
32 {
33 /* see if kzalloc will give us cachline aligned memory by default */
34 *base = kzalloc(size, flags);
35 if (*base == NULL)
36 return NULL;
37
38 if ((u64)*base == L1_CACHE_ALIGN((u64)*base))
39 return *base;
40
41 kfree(*base);
42
43 /* nope, we'll have to do it ourselves */
44 *base = kzalloc(size + L1_CACHE_BYTES, flags);
45 if (*base == NULL)
46 return NULL;
47
48 return (void *)L1_CACHE_ALIGN((u64)*base);
49 }
50
51 /*
52 * Allocate the local message queue and the notify queue.
53 */
54 static enum xp_retval
55 xpc_allocate_local_msgqueue(struct xpc_channel *ch)
56 {
57 unsigned long irq_flags;
58 int nentries;
59 size_t nbytes;
60
61 for (nentries = ch->local_nentries; nentries > 0; nentries--) {
62
63 nbytes = nentries * ch->msg_size;
64 ch->local_msgqueue = xpc_kzalloc_cacheline_aligned(nbytes,
65 GFP_KERNEL,
66 &ch->local_msgqueue_base);
67 if (ch->local_msgqueue == NULL)
68 continue;
69
70 nbytes = nentries * sizeof(struct xpc_notify);
71 ch->notify_queue = kzalloc(nbytes, GFP_KERNEL);
72 if (ch->notify_queue == NULL) {
73 kfree(ch->local_msgqueue_base);
74 ch->local_msgqueue = NULL;
75 continue;
76 }
77
78 spin_lock_irqsave(&ch->lock, irq_flags);
79 if (nentries < ch->local_nentries) {
80 dev_dbg(xpc_chan, "nentries=%d local_nentries=%d, "
81 "partid=%d, channel=%d\n", nentries,
82 ch->local_nentries, ch->partid, ch->number);
83
84 ch->local_nentries = nentries;
85 }
86 spin_unlock_irqrestore(&ch->lock, irq_flags);
87 return xpSuccess;
88 }
89
90 dev_dbg(xpc_chan, "can't get memory for local message queue and notify "
91 "queue, partid=%d, channel=%d\n", ch->partid, ch->number);
92 return xpNoMemory;
93 }
94
95 /*
96 * Allocate the cached remote message queue.
97 */
98 static enum xp_retval
99 xpc_allocate_remote_msgqueue(struct xpc_channel *ch)
100 {
101 unsigned long irq_flags;
102 int nentries;
103 size_t nbytes;
104
105 DBUG_ON(ch->remote_nentries <= 0);
106
107 for (nentries = ch->remote_nentries; nentries > 0; nentries--) {
108
109 nbytes = nentries * ch->msg_size;
110 ch->remote_msgqueue = xpc_kzalloc_cacheline_aligned(nbytes,
111 GFP_KERNEL,
112 &ch->remote_msgqueue_base);
113 if (ch->remote_msgqueue == NULL)
114 continue;
115
116 spin_lock_irqsave(&ch->lock, irq_flags);
117 if (nentries < ch->remote_nentries) {
118 dev_dbg(xpc_chan, "nentries=%d remote_nentries=%d, "
119 "partid=%d, channel=%d\n", nentries,
120 ch->remote_nentries, ch->partid, ch->number);
121
122 ch->remote_nentries = nentries;
123 }
124 spin_unlock_irqrestore(&ch->lock, irq_flags);
125 return xpSuccess;
126 }
127
128 dev_dbg(xpc_chan, "can't get memory for cached remote message queue, "
129 "partid=%d, channel=%d\n", ch->partid, ch->number);
130 return xpNoMemory;
131 }
132
133 /*
134 * Allocate message queues and other stuff associated with a channel.
135 *
136 * Note: Assumes all of the channel sizes are filled in.
137 */
138 static enum xp_retval
139 xpc_allocate_msgqueues(struct xpc_channel *ch)
140 {
141 unsigned long irq_flags;
142 enum xp_retval ret;
143
144 DBUG_ON(ch->flags & XPC_C_SETUP);
145
146 ret = xpc_allocate_local_msgqueue(ch);
147 if (ret != xpSuccess)
148 return ret;
149
150 ret = xpc_allocate_remote_msgqueue(ch);
151 if (ret != xpSuccess) {
152 kfree(ch->local_msgqueue_base);
153 ch->local_msgqueue = NULL;
154 kfree(ch->notify_queue);
155 ch->notify_queue = NULL;
156 return ret;
157 }
158
159 spin_lock_irqsave(&ch->lock, irq_flags);
160 ch->flags |= XPC_C_SETUP;
161 spin_unlock_irqrestore(&ch->lock, irq_flags);
162
163 return xpSuccess;
164 }
165
166 /*
167 * Process a connect message from a remote partition.
168 *
169 * Note: xpc_process_connect() is expecting to be called with the
170 * spin_lock_irqsave held and will leave it locked upon return.
171 */
172 static void
173 xpc_process_connect(struct xpc_channel *ch, unsigned long *irq_flags)
174 {
175 enum xp_retval ret;
176
177 DBUG_ON(!spin_is_locked(&ch->lock));
178
179 if (!(ch->flags & XPC_C_OPENREQUEST) ||
180 !(ch->flags & XPC_C_ROPENREQUEST)) {
181 /* nothing more to do for now */
182 return;
183 }
184 DBUG_ON(!(ch->flags & XPC_C_CONNECTING));
185
186 if (!(ch->flags & XPC_C_SETUP)) {
187 spin_unlock_irqrestore(&ch->lock, *irq_flags);
188 ret = xpc_allocate_msgqueues(ch);
189 spin_lock_irqsave(&ch->lock, *irq_flags);
190
191 if (ret != xpSuccess)
192 XPC_DISCONNECT_CHANNEL(ch, ret, irq_flags);
193
194 if (ch->flags & (XPC_C_CONNECTED | XPC_C_DISCONNECTING))
195 return;
196
197 DBUG_ON(!(ch->flags & XPC_C_SETUP));
198 DBUG_ON(ch->local_msgqueue == NULL);
199 DBUG_ON(ch->remote_msgqueue == NULL);
200 }
201
202 if (!(ch->flags & XPC_C_OPENREPLY)) {
203 ch->flags |= XPC_C_OPENREPLY;
204 xpc_IPI_send_openreply(ch, irq_flags);
205 }
206
207 if (!(ch->flags & XPC_C_ROPENREPLY))
208 return;
209
210 DBUG_ON(ch->remote_msgqueue_pa == 0);
211
212 ch->flags = (XPC_C_CONNECTED | XPC_C_SETUP); /* clear all else */
213
214 dev_info(xpc_chan, "channel %d to partition %d connected\n",
215 ch->number, ch->partid);
216
217 spin_unlock_irqrestore(&ch->lock, *irq_flags);
218 xpc_create_kthreads(ch, 1, 0);
219 spin_lock_irqsave(&ch->lock, *irq_flags);
220 }
221
222 /*
223 * Notify those who wanted to be notified upon delivery of their message.
224 */
225 static void
226 xpc_notify_senders(struct xpc_channel *ch, enum xp_retval reason, s64 put)
227 {
228 struct xpc_notify *notify;
229 u8 notify_type;
230 s64 get = ch->w_remote_GP.get - 1;
231
232 while (++get < put && atomic_read(&ch->n_to_notify) > 0) {
233
234 notify = &ch->notify_queue[get % ch->local_nentries];
235
236 /*
237 * See if the notify entry indicates it was associated with
238 * a message who's sender wants to be notified. It is possible
239 * that it is, but someone else is doing or has done the
240 * notification.
241 */
242 notify_type = notify->type;
243 if (notify_type == 0 ||
244 cmpxchg(&notify->type, notify_type, 0) != notify_type) {
245 continue;
246 }
247
248 DBUG_ON(notify_type != XPC_N_CALL);
249
250 atomic_dec(&ch->n_to_notify);
251
252 if (notify->func != NULL) {
253 dev_dbg(xpc_chan, "notify->func() called, notify=0x%p, "
254 "msg_number=%ld, partid=%d, channel=%d\n",
255 (void *)notify, get, ch->partid, ch->number);
256
257 notify->func(reason, ch->partid, ch->number,
258 notify->key);
259
260 dev_dbg(xpc_chan, "notify->func() returned, "
261 "notify=0x%p, msg_number=%ld, partid=%d, "
262 "channel=%d\n", (void *)notify, get,
263 ch->partid, ch->number);
264 }
265 }
266 }
267
268 /*
269 * Free up message queues and other stuff that were allocated for the specified
270 * channel.
271 *
272 * Note: ch->reason and ch->reason_line are left set for debugging purposes,
273 * they're cleared when XPC_C_DISCONNECTED is cleared.
274 */
275 static void
276 xpc_free_msgqueues(struct xpc_channel *ch)
277 {
278 DBUG_ON(!spin_is_locked(&ch->lock));
279 DBUG_ON(atomic_read(&ch->n_to_notify) != 0);
280
281 ch->remote_msgqueue_pa = 0;
282 ch->func = NULL;
283 ch->key = NULL;
284 ch->msg_size = 0;
285 ch->local_nentries = 0;
286 ch->remote_nentries = 0;
287 ch->kthreads_assigned_limit = 0;
288 ch->kthreads_idle_limit = 0;
289
290 ch->local_GP->get = 0;
291 ch->local_GP->put = 0;
292 ch->remote_GP.get = 0;
293 ch->remote_GP.put = 0;
294 ch->w_local_GP.get = 0;
295 ch->w_local_GP.put = 0;
296 ch->w_remote_GP.get = 0;
297 ch->w_remote_GP.put = 0;
298 ch->next_msg_to_pull = 0;
299
300 if (ch->flags & XPC_C_SETUP) {
301 ch->flags &= ~XPC_C_SETUP;
302
303 dev_dbg(xpc_chan, "ch->flags=0x%x, partid=%d, channel=%d\n",
304 ch->flags, ch->partid, ch->number);
305
306 kfree(ch->local_msgqueue_base);
307 ch->local_msgqueue = NULL;
308 kfree(ch->remote_msgqueue_base);
309 ch->remote_msgqueue = NULL;
310 kfree(ch->notify_queue);
311 ch->notify_queue = NULL;
312 }
313 }
314
315 /*
316 * spin_lock_irqsave() is expected to be held on entry.
317 */
318 static void
319 xpc_process_disconnect(struct xpc_channel *ch, unsigned long *irq_flags)
320 {
321 struct xpc_partition *part = &xpc_partitions[ch->partid];
322 u32 channel_was_connected = (ch->flags & XPC_C_WASCONNECTED);
323
324 DBUG_ON(!spin_is_locked(&ch->lock));
325
326 if (!(ch->flags & XPC_C_DISCONNECTING))
327 return;
328
329 DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
330
331 /* make sure all activity has settled down first */
332
333 if (atomic_read(&ch->kthreads_assigned) > 0 ||
334 atomic_read(&ch->references) > 0) {
335 return;
336 }
337 DBUG_ON((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
338 !(ch->flags & XPC_C_DISCONNECTINGCALLOUT_MADE));
339
340 if (part->act_state == XPC_P_DEACTIVATING) {
341 /* can't proceed until the other side disengages from us */
342 if (xpc_partition_engaged(1UL << ch->partid))
343 return;
344
345 } else {
346
347 /* as long as the other side is up do the full protocol */
348
349 if (!(ch->flags & XPC_C_RCLOSEREQUEST))
350 return;
351
352 if (!(ch->flags & XPC_C_CLOSEREPLY)) {
353 ch->flags |= XPC_C_CLOSEREPLY;
354 xpc_IPI_send_closereply(ch, irq_flags);
355 }
356
357 if (!(ch->flags & XPC_C_RCLOSEREPLY))
358 return;
359 }
360
361 /* wake those waiting for notify completion */
362 if (atomic_read(&ch->n_to_notify) > 0) {
363 /* >>> we do callout while holding ch->lock */
364 xpc_notify_senders(ch, ch->reason, ch->w_local_GP.put);
365 }
366
367 /* both sides are disconnected now */
368
369 if (ch->flags & XPC_C_DISCONNECTINGCALLOUT_MADE) {
370 spin_unlock_irqrestore(&ch->lock, *irq_flags);
371 xpc_disconnect_callout(ch, xpDisconnected);
372 spin_lock_irqsave(&ch->lock, *irq_flags);
373 }
374
375 /* it's now safe to free the channel's message queues */
376 xpc_free_msgqueues(ch);
377
378 /* mark disconnected, clear all other flags except XPC_C_WDISCONNECT */
379 ch->flags = (XPC_C_DISCONNECTED | (ch->flags & XPC_C_WDISCONNECT));
380
381 atomic_dec(&part->nchannels_active);
382
383 if (channel_was_connected) {
384 dev_info(xpc_chan, "channel %d to partition %d disconnected, "
385 "reason=%d\n", ch->number, ch->partid, ch->reason);
386 }
387
388 if (ch->flags & XPC_C_WDISCONNECT) {
389 /* we won't lose the CPU since we're holding ch->lock */
390 complete(&ch->wdisconnect_wait);
391 } else if (ch->delayed_IPI_flags) {
392 if (part->act_state != XPC_P_DEACTIVATING) {
393 /* time to take action on any delayed IPI flags */
394 spin_lock(&part->IPI_lock);
395 XPC_SET_IPI_FLAGS(part->local_IPI_amo, ch->number,
396 ch->delayed_IPI_flags);
397 spin_unlock(&part->IPI_lock);
398 }
399 ch->delayed_IPI_flags = 0;
400 }
401 }
402
403 /*
404 * Process a change in the channel's remote connection state.
405 */
406 static void
407 xpc_process_openclose_IPI(struct xpc_partition *part, int ch_number,
408 u8 IPI_flags)
409 {
410 unsigned long irq_flags;
411 struct xpc_openclose_args *args =
412 &part->remote_openclose_args[ch_number];
413 struct xpc_channel *ch = &part->channels[ch_number];
414 enum xp_retval reason;
415
416 spin_lock_irqsave(&ch->lock, irq_flags);
417
418 again:
419
420 if ((ch->flags & XPC_C_DISCONNECTED) &&
421 (ch->flags & XPC_C_WDISCONNECT)) {
422 /*
423 * Delay processing IPI flags until thread waiting disconnect
424 * has had a chance to see that the channel is disconnected.
425 */
426 ch->delayed_IPI_flags |= IPI_flags;
427 spin_unlock_irqrestore(&ch->lock, irq_flags);
428 return;
429 }
430
431 if (IPI_flags & XPC_IPI_CLOSEREQUEST) {
432
433 dev_dbg(xpc_chan, "XPC_IPI_CLOSEREQUEST (reason=%d) received "
434 "from partid=%d, channel=%d\n", args->reason,
435 ch->partid, ch->number);
436
437 /*
438 * If RCLOSEREQUEST is set, we're probably waiting for
439 * RCLOSEREPLY. We should find it and a ROPENREQUEST packed
440 * with this RCLOSEREQUEST in the IPI_flags.
441 */
442
443 if (ch->flags & XPC_C_RCLOSEREQUEST) {
444 DBUG_ON(!(ch->flags & XPC_C_DISCONNECTING));
445 DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
446 DBUG_ON(!(ch->flags & XPC_C_CLOSEREPLY));
447 DBUG_ON(ch->flags & XPC_C_RCLOSEREPLY);
448
449 DBUG_ON(!(IPI_flags & XPC_IPI_CLOSEREPLY));
450 IPI_flags &= ~XPC_IPI_CLOSEREPLY;
451 ch->flags |= XPC_C_RCLOSEREPLY;
452
453 /* both sides have finished disconnecting */
454 xpc_process_disconnect(ch, &irq_flags);
455 DBUG_ON(!(ch->flags & XPC_C_DISCONNECTED));
456 goto again;
457 }
458
459 if (ch->flags & XPC_C_DISCONNECTED) {
460 if (!(IPI_flags & XPC_IPI_OPENREQUEST)) {
461 if ((XPC_GET_IPI_FLAGS(part->local_IPI_amo,
462 ch_number) &
463 XPC_IPI_OPENREQUEST)) {
464
465 DBUG_ON(ch->delayed_IPI_flags != 0);
466 spin_lock(&part->IPI_lock);
467 XPC_SET_IPI_FLAGS(part->local_IPI_amo,
468 ch_number,
469 XPC_IPI_CLOSEREQUEST);
470 spin_unlock(&part->IPI_lock);
471 }
472 spin_unlock_irqrestore(&ch->lock, irq_flags);
473 return;
474 }
475
476 XPC_SET_REASON(ch, 0, 0);
477 ch->flags &= ~XPC_C_DISCONNECTED;
478
479 atomic_inc(&part->nchannels_active);
480 ch->flags |= (XPC_C_CONNECTING | XPC_C_ROPENREQUEST);
481 }
482
483 IPI_flags &= ~(XPC_IPI_OPENREQUEST | XPC_IPI_OPENREPLY);
484
485 /*
486 * The meaningful CLOSEREQUEST connection state fields are:
487 * reason = reason connection is to be closed
488 */
489
490 ch->flags |= XPC_C_RCLOSEREQUEST;
491
492 if (!(ch->flags & XPC_C_DISCONNECTING)) {
493 reason = args->reason;
494 if (reason <= xpSuccess || reason > xpUnknownReason)
495 reason = xpUnknownReason;
496 else if (reason == xpUnregistering)
497 reason = xpOtherUnregistering;
498
499 XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
500
501 DBUG_ON(IPI_flags & XPC_IPI_CLOSEREPLY);
502 spin_unlock_irqrestore(&ch->lock, irq_flags);
503 return;
504 }
505
506 xpc_process_disconnect(ch, &irq_flags);
507 }
508
509 if (IPI_flags & XPC_IPI_CLOSEREPLY) {
510
511 dev_dbg(xpc_chan, "XPC_IPI_CLOSEREPLY received from partid=%d,"
512 " channel=%d\n", ch->partid, ch->number);
513
514 if (ch->flags & XPC_C_DISCONNECTED) {
515 DBUG_ON(part->act_state != XPC_P_DEACTIVATING);
516 spin_unlock_irqrestore(&ch->lock, irq_flags);
517 return;
518 }
519
520 DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
521
522 if (!(ch->flags & XPC_C_RCLOSEREQUEST)) {
523 if ((XPC_GET_IPI_FLAGS(part->local_IPI_amo, ch_number)
524 & XPC_IPI_CLOSEREQUEST)) {
525
526 DBUG_ON(ch->delayed_IPI_flags != 0);
527 spin_lock(&part->IPI_lock);
528 XPC_SET_IPI_FLAGS(part->local_IPI_amo,
529 ch_number,
530 XPC_IPI_CLOSEREPLY);
531 spin_unlock(&part->IPI_lock);
532 }
533 spin_unlock_irqrestore(&ch->lock, irq_flags);
534 return;
535 }
536
537 ch->flags |= XPC_C_RCLOSEREPLY;
538
539 if (ch->flags & XPC_C_CLOSEREPLY) {
540 /* both sides have finished disconnecting */
541 xpc_process_disconnect(ch, &irq_flags);
542 }
543 }
544
545 if (IPI_flags & XPC_IPI_OPENREQUEST) {
546
547 dev_dbg(xpc_chan, "XPC_IPI_OPENREQUEST (msg_size=%d, "
548 "local_nentries=%d) received from partid=%d, "
549 "channel=%d\n", args->msg_size, args->local_nentries,
550 ch->partid, ch->number);
551
552 if (part->act_state == XPC_P_DEACTIVATING ||
553 (ch->flags & XPC_C_ROPENREQUEST)) {
554 spin_unlock_irqrestore(&ch->lock, irq_flags);
555 return;
556 }
557
558 if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_WDISCONNECT)) {
559 ch->delayed_IPI_flags |= XPC_IPI_OPENREQUEST;
560 spin_unlock_irqrestore(&ch->lock, irq_flags);
561 return;
562 }
563 DBUG_ON(!(ch->flags & (XPC_C_DISCONNECTED |
564 XPC_C_OPENREQUEST)));
565 DBUG_ON(ch->flags & (XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
566 XPC_C_OPENREPLY | XPC_C_CONNECTED));
567
568 /*
569 * The meaningful OPENREQUEST connection state fields are:
570 * msg_size = size of channel's messages in bytes
571 * local_nentries = remote partition's local_nentries
572 */
573 if (args->msg_size == 0 || args->local_nentries == 0) {
574 /* assume OPENREQUEST was delayed by mistake */
575 spin_unlock_irqrestore(&ch->lock, irq_flags);
576 return;
577 }
578
579 ch->flags |= (XPC_C_ROPENREQUEST | XPC_C_CONNECTING);
580 ch->remote_nentries = args->local_nentries;
581
582 if (ch->flags & XPC_C_OPENREQUEST) {
583 if (args->msg_size != ch->msg_size) {
584 XPC_DISCONNECT_CHANNEL(ch, xpUnequalMsgSizes,
585 &irq_flags);
586 spin_unlock_irqrestore(&ch->lock, irq_flags);
587 return;
588 }
589 } else {
590 ch->msg_size = args->msg_size;
591
592 XPC_SET_REASON(ch, 0, 0);
593 ch->flags &= ~XPC_C_DISCONNECTED;
594
595 atomic_inc(&part->nchannels_active);
596 }
597
598 xpc_process_connect(ch, &irq_flags);
599 }
600
601 if (IPI_flags & XPC_IPI_OPENREPLY) {
602
603 dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY (local_msgqueue_pa=0x%lx, "
604 "local_nentries=%d, remote_nentries=%d) received from "
605 "partid=%d, channel=%d\n", args->local_msgqueue_pa,
606 args->local_nentries, args->remote_nentries,
607 ch->partid, ch->number);
608
609 if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
610 spin_unlock_irqrestore(&ch->lock, irq_flags);
611 return;
612 }
613 if (!(ch->flags & XPC_C_OPENREQUEST)) {
614 XPC_DISCONNECT_CHANNEL(ch, xpOpenCloseError,
615 &irq_flags);
616 spin_unlock_irqrestore(&ch->lock, irq_flags);
617 return;
618 }
619
620 DBUG_ON(!(ch->flags & XPC_C_ROPENREQUEST));
621 DBUG_ON(ch->flags & XPC_C_CONNECTED);
622
623 /*
624 * The meaningful OPENREPLY connection state fields are:
625 * local_msgqueue_pa = physical address of remote
626 * partition's local_msgqueue
627 * local_nentries = remote partition's local_nentries
628 * remote_nentries = remote partition's remote_nentries
629 */
630 DBUG_ON(args->local_msgqueue_pa == 0);
631 DBUG_ON(args->local_nentries == 0);
632 DBUG_ON(args->remote_nentries == 0);
633
634 ch->flags |= XPC_C_ROPENREPLY;
635 ch->remote_msgqueue_pa = args->local_msgqueue_pa;
636
637 if (args->local_nentries < ch->remote_nentries) {
638 dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
639 "remote_nentries=%d, old remote_nentries=%d, "
640 "partid=%d, channel=%d\n",
641 args->local_nentries, ch->remote_nentries,
642 ch->partid, ch->number);
643
644 ch->remote_nentries = args->local_nentries;
645 }
646 if (args->remote_nentries < ch->local_nentries) {
647 dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
648 "local_nentries=%d, old local_nentries=%d, "
649 "partid=%d, channel=%d\n",
650 args->remote_nentries, ch->local_nentries,
651 ch->partid, ch->number);
652
653 ch->local_nentries = args->remote_nentries;
654 }
655
656 xpc_process_connect(ch, &irq_flags);
657 }
658
659 spin_unlock_irqrestore(&ch->lock, irq_flags);
660 }
661
662 /*
663 * Attempt to establish a channel connection to a remote partition.
664 */
665 static enum xp_retval
666 xpc_connect_channel(struct xpc_channel *ch)
667 {
668 unsigned long irq_flags;
669 struct xpc_registration *registration = &xpc_registrations[ch->number];
670
671 if (mutex_trylock(&registration->mutex) == 0)
672 return xpRetry;
673
674 if (!XPC_CHANNEL_REGISTERED(ch->number)) {
675 mutex_unlock(&registration->mutex);
676 return xpUnregistered;
677 }
678
679 spin_lock_irqsave(&ch->lock, irq_flags);
680
681 DBUG_ON(ch->flags & XPC_C_CONNECTED);
682 DBUG_ON(ch->flags & XPC_C_OPENREQUEST);
683
684 if (ch->flags & XPC_C_DISCONNECTING) {
685 spin_unlock_irqrestore(&ch->lock, irq_flags);
686 mutex_unlock(&registration->mutex);
687 return ch->reason;
688 }
689
690 /* add info from the channel connect registration to the channel */
691
692 ch->kthreads_assigned_limit = registration->assigned_limit;
693 ch->kthreads_idle_limit = registration->idle_limit;
694 DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);
695 DBUG_ON(atomic_read(&ch->kthreads_idle) != 0);
696 DBUG_ON(atomic_read(&ch->kthreads_active) != 0);
697
698 ch->func = registration->func;
699 DBUG_ON(registration->func == NULL);
700 ch->key = registration->key;
701
702 ch->local_nentries = registration->nentries;
703
704 if (ch->flags & XPC_C_ROPENREQUEST) {
705 if (registration->msg_size != ch->msg_size) {
706 /* the local and remote sides aren't the same */
707
708 /*
709 * Because XPC_DISCONNECT_CHANNEL() can block we're
710 * forced to up the registration sema before we unlock
711 * the channel lock. But that's okay here because we're
712 * done with the part that required the registration
713 * sema. XPC_DISCONNECT_CHANNEL() requires that the
714 * channel lock be locked and will unlock and relock
715 * the channel lock as needed.
716 */
717 mutex_unlock(&registration->mutex);
718 XPC_DISCONNECT_CHANNEL(ch, xpUnequalMsgSizes,
719 &irq_flags);
720 spin_unlock_irqrestore(&ch->lock, irq_flags);
721 return xpUnequalMsgSizes;
722 }
723 } else {
724 ch->msg_size = registration->msg_size;
725
726 XPC_SET_REASON(ch, 0, 0);
727 ch->flags &= ~XPC_C_DISCONNECTED;
728
729 atomic_inc(&xpc_partitions[ch->partid].nchannels_active);
730 }
731
732 mutex_unlock(&registration->mutex);
733
734 /* initiate the connection */
735
736 ch->flags |= (XPC_C_OPENREQUEST | XPC_C_CONNECTING);
737 xpc_IPI_send_openrequest(ch, &irq_flags);
738
739 xpc_process_connect(ch, &irq_flags);
740
741 spin_unlock_irqrestore(&ch->lock, irq_flags);
742
743 return xpSuccess;
744 }
745
746 /*
747 * Clear some of the msg flags in the local message queue.
748 */
749 static inline void
750 xpc_clear_local_msgqueue_flags(struct xpc_channel *ch)
751 {
752 struct xpc_msg *msg;
753 s64 get;
754
755 get = ch->w_remote_GP.get;
756 do {
757 msg = (struct xpc_msg *)((u64)ch->local_msgqueue +
758 (get % ch->local_nentries) *
759 ch->msg_size);
760 msg->flags = 0;
761 } while (++get < ch->remote_GP.get);
762 }
763
764 /*
765 * Clear some of the msg flags in the remote message queue.
766 */
767 static inline void
768 xpc_clear_remote_msgqueue_flags(struct xpc_channel *ch)
769 {
770 struct xpc_msg *msg;
771 s64 put;
772
773 put = ch->w_remote_GP.put;
774 do {
775 msg = (struct xpc_msg *)((u64)ch->remote_msgqueue +
776 (put % ch->remote_nentries) *
777 ch->msg_size);
778 msg->flags = 0;
779 } while (++put < ch->remote_GP.put);
780 }
781
782 static void
783 xpc_process_msg_IPI(struct xpc_partition *part, int ch_number)
784 {
785 struct xpc_channel *ch = &part->channels[ch_number];
786 int nmsgs_sent;
787
788 ch->remote_GP = part->remote_GPs[ch_number];
789
790 /* See what, if anything, has changed for each connected channel */
791
792 xpc_msgqueue_ref(ch);
793
794 if (ch->w_remote_GP.get == ch->remote_GP.get &&
795 ch->w_remote_GP.put == ch->remote_GP.put) {
796 /* nothing changed since GPs were last pulled */
797 xpc_msgqueue_deref(ch);
798 return;
799 }
800
801 if (!(ch->flags & XPC_C_CONNECTED)) {
802 xpc_msgqueue_deref(ch);
803 return;
804 }
805
806 /*
807 * First check to see if messages recently sent by us have been
808 * received by the other side. (The remote GET value will have
809 * changed since we last looked at it.)
810 */
811
812 if (ch->w_remote_GP.get != ch->remote_GP.get) {
813
814 /*
815 * We need to notify any senders that want to be notified
816 * that their sent messages have been received by their
817 * intended recipients. We need to do this before updating
818 * w_remote_GP.get so that we don't allocate the same message
819 * queue entries prematurely (see xpc_allocate_msg()).
820 */
821 if (atomic_read(&ch->n_to_notify) > 0) {
822 /*
823 * Notify senders that messages sent have been
824 * received and delivered by the other side.
825 */
826 xpc_notify_senders(ch, xpMsgDelivered,
827 ch->remote_GP.get);
828 }
829
830 /*
831 * Clear msg->flags in previously sent messages, so that
832 * they're ready for xpc_allocate_msg().
833 */
834 xpc_clear_local_msgqueue_flags(ch);
835
836 ch->w_remote_GP.get = ch->remote_GP.get;
837
838 dev_dbg(xpc_chan, "w_remote_GP.get changed to %ld, partid=%d, "
839 "channel=%d\n", ch->w_remote_GP.get, ch->partid,
840 ch->number);
841
842 /*
843 * If anyone was waiting for message queue entries to become
844 * available, wake them up.
845 */
846 if (atomic_read(&ch->n_on_msg_allocate_wq) > 0)
847 wake_up(&ch->msg_allocate_wq);
848 }
849
850 /*
851 * Now check for newly sent messages by the other side. (The remote
852 * PUT value will have changed since we last looked at it.)
853 */
854
855 if (ch->w_remote_GP.put != ch->remote_GP.put) {
856 /*
857 * Clear msg->flags in previously received messages, so that
858 * they're ready for xpc_get_deliverable_msg().
859 */
860 xpc_clear_remote_msgqueue_flags(ch);
861
862 ch->w_remote_GP.put = ch->remote_GP.put;
863
864 dev_dbg(xpc_chan, "w_remote_GP.put changed to %ld, partid=%d, "
865 "channel=%d\n", ch->w_remote_GP.put, ch->partid,
866 ch->number);
867
868 nmsgs_sent = ch->w_remote_GP.put - ch->w_local_GP.get;
869 if (nmsgs_sent > 0) {
870 dev_dbg(xpc_chan, "msgs waiting to be copied and "
871 "delivered=%d, partid=%d, channel=%d\n",
872 nmsgs_sent, ch->partid, ch->number);
873
874 if (ch->flags & XPC_C_CONNECTEDCALLOUT_MADE)
875 xpc_activate_kthreads(ch, nmsgs_sent);
876 }
877 }
878
879 xpc_msgqueue_deref(ch);
880 }
881
882 void
883 xpc_process_channel_activity(struct xpc_partition *part)
884 {
885 unsigned long irq_flags;
886 u64 IPI_amo, IPI_flags;
887 struct xpc_channel *ch;
888 int ch_number;
889 u32 ch_flags;
890
891 IPI_amo = xpc_get_IPI_flags(part);
892
893 /*
894 * Initiate channel connections for registered channels.
895 *
896 * For each connected channel that has pending messages activate idle
897 * kthreads and/or create new kthreads as needed.
898 */
899
900 for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
901 ch = &part->channels[ch_number];
902
903 /*
904 * Process any open or close related IPI flags, and then deal
905 * with connecting or disconnecting the channel as required.
906 */
907
908 IPI_flags = XPC_GET_IPI_FLAGS(IPI_amo, ch_number);
909
910 if (XPC_ANY_OPENCLOSE_IPI_FLAGS_SET(IPI_flags))
911 xpc_process_openclose_IPI(part, ch_number, IPI_flags);
912
913 ch_flags = ch->flags; /* need an atomic snapshot of flags */
914
915 if (ch_flags & XPC_C_DISCONNECTING) {
916 spin_lock_irqsave(&ch->lock, irq_flags);
917 xpc_process_disconnect(ch, &irq_flags);
918 spin_unlock_irqrestore(&ch->lock, irq_flags);
919 continue;
920 }
921
922 if (part->act_state == XPC_P_DEACTIVATING)
923 continue;
924
925 if (!(ch_flags & XPC_C_CONNECTED)) {
926 if (!(ch_flags & XPC_C_OPENREQUEST)) {
927 DBUG_ON(ch_flags & XPC_C_SETUP);
928 (void)xpc_connect_channel(ch);
929 } else {
930 spin_lock_irqsave(&ch->lock, irq_flags);
931 xpc_process_connect(ch, &irq_flags);
932 spin_unlock_irqrestore(&ch->lock, irq_flags);
933 }
934 continue;
935 }
936
937 /*
938 * Process any message related IPI flags, this may involve the
939 * activation of kthreads to deliver any pending messages sent
940 * from the other partition.
941 */
942
943 if (XPC_ANY_MSG_IPI_FLAGS_SET(IPI_flags))
944 xpc_process_msg_IPI(part, ch_number);
945 }
946 }
947
948 /*
949 * XPC's heartbeat code calls this function to inform XPC that a partition is
950 * going down. XPC responds by tearing down the XPartition Communication
951 * infrastructure used for the just downed partition.
952 *
953 * XPC's heartbeat code will never call this function and xpc_partition_up()
954 * at the same time. Nor will it ever make multiple calls to either function
955 * at the same time.
956 */
957 void
958 xpc_partition_going_down(struct xpc_partition *part, enum xp_retval reason)
959 {
960 unsigned long irq_flags;
961 int ch_number;
962 struct xpc_channel *ch;
963
964 dev_dbg(xpc_chan, "deactivating partition %d, reason=%d\n",
965 XPC_PARTID(part), reason);
966
967 if (!xpc_part_ref(part)) {
968 /* infrastructure for this partition isn't currently set up */
969 return;
970 }
971
972 /* disconnect channels associated with the partition going down */
973
974 for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
975 ch = &part->channels[ch_number];
976
977 xpc_msgqueue_ref(ch);
978 spin_lock_irqsave(&ch->lock, irq_flags);
979
980 XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
981
982 spin_unlock_irqrestore(&ch->lock, irq_flags);
983 xpc_msgqueue_deref(ch);
984 }
985
986 xpc_wakeup_channel_mgr(part);
987
988 xpc_part_deref(part);
989 }
990
991 /*
992 * Called by XP at the time of channel connection registration to cause
993 * XPC to establish connections to all currently active partitions.
994 */
995 void
996 xpc_initiate_connect(int ch_number)
997 {
998 short partid;
999 struct xpc_partition *part;
1000 struct xpc_channel *ch;
1001
1002 DBUG_ON(ch_number < 0 || ch_number >= XPC_MAX_NCHANNELS);
1003
1004 for (partid = 0; partid < xp_max_npartitions; partid++) {
1005 part = &xpc_partitions[partid];
1006
1007 if (xpc_part_ref(part)) {
1008 ch = &part->channels[ch_number];
1009
1010 /*
1011 * Initiate the establishment of a connection on the
1012 * newly registered channel to the remote partition.
1013 */
1014 xpc_wakeup_channel_mgr(part);
1015 xpc_part_deref(part);
1016 }
1017 }
1018 }
1019
1020 void
1021 xpc_connected_callout(struct xpc_channel *ch)
1022 {
1023 /* let the registerer know that a connection has been established */
1024
1025 if (ch->func != NULL) {
1026 dev_dbg(xpc_chan, "ch->func() called, reason=xpConnected, "
1027 "partid=%d, channel=%d\n", ch->partid, ch->number);
1028
1029 ch->func(xpConnected, ch->partid, ch->number,
1030 (void *)(u64)ch->local_nentries, ch->key);
1031
1032 dev_dbg(xpc_chan, "ch->func() returned, reason=xpConnected, "
1033 "partid=%d, channel=%d\n", ch->partid, ch->number);
1034 }
1035 }
1036
1037 /*
1038 * Called by XP at the time of channel connection unregistration to cause
1039 * XPC to teardown all current connections for the specified channel.
1040 *
1041 * Before returning xpc_initiate_disconnect() will wait until all connections
1042 * on the specified channel have been closed/torndown. So the caller can be
1043 * assured that they will not be receiving any more callouts from XPC to the
1044 * function they registered via xpc_connect().
1045 *
1046 * Arguments:
1047 *
1048 * ch_number - channel # to unregister.
1049 */
1050 void
1051 xpc_initiate_disconnect(int ch_number)
1052 {
1053 unsigned long irq_flags;
1054 short partid;
1055 struct xpc_partition *part;
1056 struct xpc_channel *ch;
1057
1058 DBUG_ON(ch_number < 0 || ch_number >= XPC_MAX_NCHANNELS);
1059
1060 /* initiate the channel disconnect for every active partition */
1061 for (partid = 0; partid < xp_max_npartitions; partid++) {
1062 part = &xpc_partitions[partid];
1063
1064 if (xpc_part_ref(part)) {
1065 ch = &part->channels[ch_number];
1066 xpc_msgqueue_ref(ch);
1067
1068 spin_lock_irqsave(&ch->lock, irq_flags);
1069
1070 if (!(ch->flags & XPC_C_DISCONNECTED)) {
1071 ch->flags |= XPC_C_WDISCONNECT;
1072
1073 XPC_DISCONNECT_CHANNEL(ch, xpUnregistering,
1074 &irq_flags);
1075 }
1076
1077 spin_unlock_irqrestore(&ch->lock, irq_flags);
1078
1079 xpc_msgqueue_deref(ch);
1080 xpc_part_deref(part);
1081 }
1082 }
1083
1084 xpc_disconnect_wait(ch_number);
1085 }
1086
1087 /*
1088 * To disconnect a channel, and reflect it back to all who may be waiting.
1089 *
1090 * An OPEN is not allowed until XPC_C_DISCONNECTING is cleared by
1091 * xpc_process_disconnect(), and if set, XPC_C_WDISCONNECT is cleared by
1092 * xpc_disconnect_wait().
1093 *
1094 * THE CHANNEL IS TO BE LOCKED BY THE CALLER AND WILL REMAIN LOCKED UPON RETURN.
1095 */
1096 void
1097 xpc_disconnect_channel(const int line, struct xpc_channel *ch,
1098 enum xp_retval reason, unsigned long *irq_flags)
1099 {
1100 u32 channel_was_connected = (ch->flags & XPC_C_CONNECTED);
1101
1102 DBUG_ON(!spin_is_locked(&ch->lock));
1103
1104 if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED))
1105 return;
1106
1107 DBUG_ON(!(ch->flags & (XPC_C_CONNECTING | XPC_C_CONNECTED)));
1108
1109 dev_dbg(xpc_chan, "reason=%d, line=%d, partid=%d, channel=%d\n",
1110 reason, line, ch->partid, ch->number);
1111
1112 XPC_SET_REASON(ch, reason, line);
1113
1114 ch->flags |= (XPC_C_CLOSEREQUEST | XPC_C_DISCONNECTING);
1115 /* some of these may not have been set */
1116 ch->flags &= ~(XPC_C_OPENREQUEST | XPC_C_OPENREPLY |
1117 XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
1118 XPC_C_CONNECTING | XPC_C_CONNECTED);
1119
1120 xpc_IPI_send_closerequest(ch, irq_flags);
1121
1122 if (channel_was_connected)
1123 ch->flags |= XPC_C_WASCONNECTED;
1124
1125 spin_unlock_irqrestore(&ch->lock, *irq_flags);
1126
1127 /* wake all idle kthreads so they can exit */
1128 if (atomic_read(&ch->kthreads_idle) > 0) {
1129 wake_up_all(&ch->idle_wq);
1130
1131 } else if ((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
1132 !(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) {
1133 /* start a kthread that will do the xpDisconnecting callout */
1134 xpc_create_kthreads(ch, 1, 1);
1135 }
1136
1137 /* wake those waiting to allocate an entry from the local msg queue */
1138 if (atomic_read(&ch->n_on_msg_allocate_wq) > 0)
1139 wake_up(&ch->msg_allocate_wq);
1140
1141 spin_lock_irqsave(&ch->lock, *irq_flags);
1142 }
1143
1144 void
1145 xpc_disconnect_callout(struct xpc_channel *ch, enum xp_retval reason)
1146 {
1147 /*
1148 * Let the channel's registerer know that the channel is being
1149 * disconnected. We don't want to do this if the registerer was never
1150 * informed of a connection being made.
1151 */
1152
1153 if (ch->func != NULL) {
1154 dev_dbg(xpc_chan, "ch->func() called, reason=%d, partid=%d, "
1155 "channel=%d\n", reason, ch->partid, ch->number);
1156
1157 ch->func(reason, ch->partid, ch->number, NULL, ch->key);
1158
1159 dev_dbg(xpc_chan, "ch->func() returned, reason=%d, partid=%d, "
1160 "channel=%d\n", reason, ch->partid, ch->number);
1161 }
1162 }
1163
1164 /*
1165 * Wait for a message entry to become available for the specified channel,
1166 * but don't wait any longer than 1 jiffy.
1167 */
1168 enum xp_retval
1169 xpc_allocate_msg_wait(struct xpc_channel *ch)
1170 {
1171 enum xp_retval ret;
1172
1173 if (ch->flags & XPC_C_DISCONNECTING) {
1174 DBUG_ON(ch->reason == xpInterrupted);
1175 return ch->reason;
1176 }
1177
1178 atomic_inc(&ch->n_on_msg_allocate_wq);
1179 ret = interruptible_sleep_on_timeout(&ch->msg_allocate_wq, 1);
1180 atomic_dec(&ch->n_on_msg_allocate_wq);
1181
1182 if (ch->flags & XPC_C_DISCONNECTING) {
1183 ret = ch->reason;
1184 DBUG_ON(ch->reason == xpInterrupted);
1185 } else if (ret == 0) {
1186 ret = xpTimeout;
1187 } else {
1188 ret = xpInterrupted;
1189 }
1190
1191 return ret;
1192 }
1193
1194 /*
1195 * Allocate an entry for a message from the message queue associated with the
1196 * specified channel. NOTE that this routine can sleep waiting for a message
1197 * entry to become available. To not sleep, pass in the XPC_NOWAIT flag.
1198 *
1199 * Arguments:
1200 *
1201 * partid - ID of partition to which the channel is connected.
1202 * ch_number - channel #.
1203 * flags - see xpc.h for valid flags.
1204 * payload - address of the allocated payload area pointer (filled in on
1205 * return) in which the user-defined message is constructed.
1206 */
1207 enum xp_retval
1208 xpc_initiate_allocate(short partid, int ch_number, u32 flags, void **payload)
1209 {
1210 struct xpc_partition *part = &xpc_partitions[partid];
1211 enum xp_retval ret = xpUnknownReason;
1212 struct xpc_msg *msg = NULL;
1213
1214 DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
1215 DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
1216
1217 *payload = NULL;
1218
1219 if (xpc_part_ref(part)) {
1220 ret = xpc_allocate_msg(&part->channels[ch_number], flags, &msg);
1221 xpc_part_deref(part);
1222
1223 if (msg != NULL)
1224 *payload = &msg->payload;
1225 }
1226
1227 return ret;
1228 }
1229
1230 /*
1231 * Send a message previously allocated using xpc_initiate_allocate() on the
1232 * specified channel connected to the specified partition.
1233 *
1234 * This routine will not wait for the message to be received, nor will
1235 * notification be given when it does happen. Once this routine has returned
1236 * the message entry allocated via xpc_initiate_allocate() is no longer
1237 * accessable to the caller.
1238 *
1239 * This routine, although called by users, does not call xpc_part_ref() to
1240 * ensure that the partition infrastructure is in place. It relies on the
1241 * fact that we called xpc_msgqueue_ref() in xpc_allocate_msg().
1242 *
1243 * Arguments:
1244 *
1245 * partid - ID of partition to which the channel is connected.
1246 * ch_number - channel # to send message on.
1247 * payload - pointer to the payload area allocated via
1248 * xpc_initiate_allocate().
1249 */
1250 enum xp_retval
1251 xpc_initiate_send(short partid, int ch_number, void *payload)
1252 {
1253 struct xpc_partition *part = &xpc_partitions[partid];
1254 struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
1255 enum xp_retval ret;
1256
1257 dev_dbg(xpc_chan, "msg=0x%p, partid=%d, channel=%d\n", (void *)msg,
1258 partid, ch_number);
1259
1260 DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
1261 DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
1262 DBUG_ON(msg == NULL);
1263
1264 ret = xpc_send_msg(&part->channels[ch_number], msg, 0, NULL, NULL);
1265
1266 return ret;
1267 }
1268
1269 /*
1270 * Send a message previously allocated using xpc_initiate_allocate on the
1271 * specified channel connected to the specified partition.
1272 *
1273 * This routine will not wait for the message to be sent. Once this routine
1274 * has returned the message entry allocated via xpc_initiate_allocate() is no
1275 * longer accessable to the caller.
1276 *
1277 * Once the remote end of the channel has received the message, the function
1278 * passed as an argument to xpc_initiate_send_notify() will be called. This
1279 * allows the sender to free up or re-use any buffers referenced by the
1280 * message, but does NOT mean the message has been processed at the remote
1281 * end by a receiver.
1282 *
1283 * If this routine returns an error, the caller's function will NOT be called.
1284 *
1285 * This routine, although called by users, does not call xpc_part_ref() to
1286 * ensure that the partition infrastructure is in place. It relies on the
1287 * fact that we called xpc_msgqueue_ref() in xpc_allocate_msg().
1288 *
1289 * Arguments:
1290 *
1291 * partid - ID of partition to which the channel is connected.
1292 * ch_number - channel # to send message on.
1293 * payload - pointer to the payload area allocated via
1294 * xpc_initiate_allocate().
1295 * func - function to call with asynchronous notification of message
1296 * receipt. THIS FUNCTION MUST BE NON-BLOCKING.
1297 * key - user-defined key to be passed to the function when it's called.
1298 */
1299 enum xp_retval
1300 xpc_initiate_send_notify(short partid, int ch_number, void *payload,
1301 xpc_notify_func func, void *key)
1302 {
1303 struct xpc_partition *part = &xpc_partitions[partid];
1304 struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
1305 enum xp_retval ret;
1306
1307 dev_dbg(xpc_chan, "msg=0x%p, partid=%d, channel=%d\n", (void *)msg,
1308 partid, ch_number);
1309
1310 DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
1311 DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
1312 DBUG_ON(msg == NULL);
1313 DBUG_ON(func == NULL);
1314
1315 ret = xpc_send_msg(&part->channels[ch_number], msg, XPC_N_CALL,
1316 func, key);
1317 return ret;
1318 }
1319
1320 /*
1321 * Deliver a message to its intended recipient.
1322 */
1323 void
1324 xpc_deliver_msg(struct xpc_channel *ch)
1325 {
1326 struct xpc_msg *msg;
1327
1328 msg = xpc_get_deliverable_msg(ch);
1329 if (msg != NULL) {
1330
1331 /*
1332 * This ref is taken to protect the payload itself from being
1333 * freed before the user is finished with it, which the user
1334 * indicates by calling xpc_initiate_received().
1335 */
1336 xpc_msgqueue_ref(ch);
1337
1338 atomic_inc(&ch->kthreads_active);
1339
1340 if (ch->func != NULL) {
1341 dev_dbg(xpc_chan, "ch->func() called, msg=0x%p, "
1342 "msg_number=%ld, partid=%d, channel=%d\n",
1343 (void *)msg, msg->number, ch->partid,
1344 ch->number);
1345
1346 /* deliver the message to its intended recipient */
1347 ch->func(xpMsgReceived, ch->partid, ch->number,
1348 &msg->payload, ch->key);
1349
1350 dev_dbg(xpc_chan, "ch->func() returned, msg=0x%p, "
1351 "msg_number=%ld, partid=%d, channel=%d\n",
1352 (void *)msg, msg->number, ch->partid,
1353 ch->number);
1354 }
1355
1356 atomic_dec(&ch->kthreads_active);
1357 }
1358 }
1359
1360 /*
1361 * Acknowledge receipt of a delivered message.
1362 *
1363 * If a message has XPC_M_INTERRUPT set, send an interrupt to the partition
1364 * that sent the message.
1365 *
1366 * This function, although called by users, does not call xpc_part_ref() to
1367 * ensure that the partition infrastructure is in place. It relies on the
1368 * fact that we called xpc_msgqueue_ref() in xpc_deliver_msg().
1369 *
1370 * Arguments:
1371 *
1372 * partid - ID of partition to which the channel is connected.
1373 * ch_number - channel # message received on.
1374 * payload - pointer to the payload area allocated via
1375 * xpc_initiate_allocate().
1376 */
1377 void
1378 xpc_initiate_received(short partid, int ch_number, void *payload)
1379 {
1380 struct xpc_partition *part = &xpc_partitions[partid];
1381 struct xpc_channel *ch;
1382 struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
1383
1384 DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
1385 DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
1386
1387 ch = &part->channels[ch_number];
1388 xpc_received_msg(ch, msg);
1389
1390 /* the call to xpc_msgqueue_ref() was done by xpc_deliver_msg() */
1391 xpc_msgqueue_deref(ch);
1392 }
Linux kernel - Deliverables
This page took 0.059096 seconds and 6 git commands to generate.