Merge branch 'for-2.6.30' of git://linux-nfs.org/~bfields/linux
[linux-2.6] / 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/device.h>
18 #include "xpc.h"
19
20 /*
21  * Process a connect message from a remote partition.
22  *
23  * Note: xpc_process_connect() is expecting to be called with the
24  * spin_lock_irqsave held and will leave it locked upon return.
25  */
26 static void
27 xpc_process_connect(struct xpc_channel *ch, unsigned long *irq_flags)
28 {
29         enum xp_retval ret;
30
31         DBUG_ON(!spin_is_locked(&ch->lock));
32
33         if (!(ch->flags & XPC_C_OPENREQUEST) ||
34             !(ch->flags & XPC_C_ROPENREQUEST)) {
35                 /* nothing more to do for now */
36                 return;
37         }
38         DBUG_ON(!(ch->flags & XPC_C_CONNECTING));
39
40         if (!(ch->flags & XPC_C_SETUP)) {
41                 spin_unlock_irqrestore(&ch->lock, *irq_flags);
42                 ret = xpc_setup_msg_structures(ch);
43                 spin_lock_irqsave(&ch->lock, *irq_flags);
44
45                 if (ret != xpSuccess)
46                         XPC_DISCONNECT_CHANNEL(ch, ret, irq_flags);
47
48                 ch->flags |= XPC_C_SETUP;
49
50                 if (ch->flags & (XPC_C_CONNECTED | XPC_C_DISCONNECTING))
51                         return;
52         }
53
54         if (!(ch->flags & XPC_C_OPENREPLY)) {
55                 ch->flags |= XPC_C_OPENREPLY;
56                 xpc_send_chctl_openreply(ch, irq_flags);
57         }
58
59         if (!(ch->flags & XPC_C_ROPENREPLY))
60                 return;
61
62         ch->flags = (XPC_C_CONNECTED | XPC_C_SETUP);    /* clear all else */
63
64         dev_info(xpc_chan, "channel %d to partition %d connected\n",
65                  ch->number, ch->partid);
66
67         spin_unlock_irqrestore(&ch->lock, *irq_flags);
68         xpc_create_kthreads(ch, 1, 0);
69         spin_lock_irqsave(&ch->lock, *irq_flags);
70 }
71
72 /*
73  * spin_lock_irqsave() is expected to be held on entry.
74  */
75 static void
76 xpc_process_disconnect(struct xpc_channel *ch, unsigned long *irq_flags)
77 {
78         struct xpc_partition *part = &xpc_partitions[ch->partid];
79         u32 channel_was_connected = (ch->flags & XPC_C_WASCONNECTED);
80
81         DBUG_ON(!spin_is_locked(&ch->lock));
82
83         if (!(ch->flags & XPC_C_DISCONNECTING))
84                 return;
85
86         DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
87
88         /* make sure all activity has settled down first */
89
90         if (atomic_read(&ch->kthreads_assigned) > 0 ||
91             atomic_read(&ch->references) > 0) {
92                 return;
93         }
94         DBUG_ON((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
95                 !(ch->flags & XPC_C_DISCONNECTINGCALLOUT_MADE));
96
97         if (part->act_state == XPC_P_AS_DEACTIVATING) {
98                 /* can't proceed until the other side disengages from us */
99                 if (xpc_partition_engaged(ch->partid))
100                         return;
101
102         } else {
103
104                 /* as long as the other side is up do the full protocol */
105
106                 if (!(ch->flags & XPC_C_RCLOSEREQUEST))
107                         return;
108
109                 if (!(ch->flags & XPC_C_CLOSEREPLY)) {
110                         ch->flags |= XPC_C_CLOSEREPLY;
111                         xpc_send_chctl_closereply(ch, irq_flags);
112                 }
113
114                 if (!(ch->flags & XPC_C_RCLOSEREPLY))
115                         return;
116         }
117
118         /* wake those waiting for notify completion */
119         if (atomic_read(&ch->n_to_notify) > 0) {
120                 /* we do callout while holding ch->lock, callout can't block */
121                 xpc_notify_senders_of_disconnect(ch);
122         }
123
124         /* both sides are disconnected now */
125
126         if (ch->flags & XPC_C_DISCONNECTINGCALLOUT_MADE) {
127                 spin_unlock_irqrestore(&ch->lock, *irq_flags);
128                 xpc_disconnect_callout(ch, xpDisconnected);
129                 spin_lock_irqsave(&ch->lock, *irq_flags);
130         }
131
132         DBUG_ON(atomic_read(&ch->n_to_notify) != 0);
133
134         /* it's now safe to free the channel's message queues */
135         xpc_teardown_msg_structures(ch);
136
137         ch->func = NULL;
138         ch->key = NULL;
139         ch->entry_size = 0;
140         ch->local_nentries = 0;
141         ch->remote_nentries = 0;
142         ch->kthreads_assigned_limit = 0;
143         ch->kthreads_idle_limit = 0;
144
145         /*
146          * Mark the channel disconnected and clear all other flags, including
147          * XPC_C_SETUP (because of call to xpc_teardown_msg_structures()) but
148          * not including XPC_C_WDISCONNECT (if it was set).
149          */
150         ch->flags = (XPC_C_DISCONNECTED | (ch->flags & XPC_C_WDISCONNECT));
151
152         atomic_dec(&part->nchannels_active);
153
154         if (channel_was_connected) {
155                 dev_info(xpc_chan, "channel %d to partition %d disconnected, "
156                          "reason=%d\n", ch->number, ch->partid, ch->reason);
157         }
158
159         if (ch->flags & XPC_C_WDISCONNECT) {
160                 /* we won't lose the CPU since we're holding ch->lock */
161                 complete(&ch->wdisconnect_wait);
162         } else if (ch->delayed_chctl_flags) {
163                 if (part->act_state != XPC_P_AS_DEACTIVATING) {
164                         /* time to take action on any delayed chctl flags */
165                         spin_lock(&part->chctl_lock);
166                         part->chctl.flags[ch->number] |=
167                             ch->delayed_chctl_flags;
168                         spin_unlock(&part->chctl_lock);
169                 }
170                 ch->delayed_chctl_flags = 0;
171         }
172 }
173
174 /*
175  * Process a change in the channel's remote connection state.
176  */
177 static void
178 xpc_process_openclose_chctl_flags(struct xpc_partition *part, int ch_number,
179                                   u8 chctl_flags)
180 {
181         unsigned long irq_flags;
182         struct xpc_openclose_args *args =
183             &part->remote_openclose_args[ch_number];
184         struct xpc_channel *ch = &part->channels[ch_number];
185         enum xp_retval reason;
186         enum xp_retval ret;
187
188         spin_lock_irqsave(&ch->lock, irq_flags);
189
190 again:
191
192         if ((ch->flags & XPC_C_DISCONNECTED) &&
193             (ch->flags & XPC_C_WDISCONNECT)) {
194                 /*
195                  * Delay processing chctl flags until thread waiting disconnect
196                  * has had a chance to see that the channel is disconnected.
197                  */
198                 ch->delayed_chctl_flags |= chctl_flags;
199                 spin_unlock_irqrestore(&ch->lock, irq_flags);
200                 return;
201         }
202
203         if (chctl_flags & XPC_CHCTL_CLOSEREQUEST) {
204
205                 dev_dbg(xpc_chan, "XPC_CHCTL_CLOSEREQUEST (reason=%d) received "
206                         "from partid=%d, channel=%d\n", args->reason,
207                         ch->partid, ch->number);
208
209                 /*
210                  * If RCLOSEREQUEST is set, we're probably waiting for
211                  * RCLOSEREPLY. We should find it and a ROPENREQUEST packed
212                  * with this RCLOSEREQUEST in the chctl_flags.
213                  */
214
215                 if (ch->flags & XPC_C_RCLOSEREQUEST) {
216                         DBUG_ON(!(ch->flags & XPC_C_DISCONNECTING));
217                         DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
218                         DBUG_ON(!(ch->flags & XPC_C_CLOSEREPLY));
219                         DBUG_ON(ch->flags & XPC_C_RCLOSEREPLY);
220
221                         DBUG_ON(!(chctl_flags & XPC_CHCTL_CLOSEREPLY));
222                         chctl_flags &= ~XPC_CHCTL_CLOSEREPLY;
223                         ch->flags |= XPC_C_RCLOSEREPLY;
224
225                         /* both sides have finished disconnecting */
226                         xpc_process_disconnect(ch, &irq_flags);
227                         DBUG_ON(!(ch->flags & XPC_C_DISCONNECTED));
228                         goto again;
229                 }
230
231                 if (ch->flags & XPC_C_DISCONNECTED) {
232                         if (!(chctl_flags & XPC_CHCTL_OPENREQUEST)) {
233                                 if (part->chctl.flags[ch_number] &
234                                     XPC_CHCTL_OPENREQUEST) {
235
236                                         DBUG_ON(ch->delayed_chctl_flags != 0);
237                                         spin_lock(&part->chctl_lock);
238                                         part->chctl.flags[ch_number] |=
239                                             XPC_CHCTL_CLOSEREQUEST;
240                                         spin_unlock(&part->chctl_lock);
241                                 }
242                                 spin_unlock_irqrestore(&ch->lock, irq_flags);
243                                 return;
244                         }
245
246                         XPC_SET_REASON(ch, 0, 0);
247                         ch->flags &= ~XPC_C_DISCONNECTED;
248
249                         atomic_inc(&part->nchannels_active);
250                         ch->flags |= (XPC_C_CONNECTING | XPC_C_ROPENREQUEST);
251                 }
252
253                 chctl_flags &= ~(XPC_CHCTL_OPENREQUEST | XPC_CHCTL_OPENREPLY);
254
255                 /*
256                  * The meaningful CLOSEREQUEST connection state fields are:
257                  *      reason = reason connection is to be closed
258                  */
259
260                 ch->flags |= XPC_C_RCLOSEREQUEST;
261
262                 if (!(ch->flags & XPC_C_DISCONNECTING)) {
263                         reason = args->reason;
264                         if (reason <= xpSuccess || reason > xpUnknownReason)
265                                 reason = xpUnknownReason;
266                         else if (reason == xpUnregistering)
267                                 reason = xpOtherUnregistering;
268
269                         XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
270
271                         DBUG_ON(chctl_flags & XPC_CHCTL_CLOSEREPLY);
272                         spin_unlock_irqrestore(&ch->lock, irq_flags);
273                         return;
274                 }
275
276                 xpc_process_disconnect(ch, &irq_flags);
277         }
278
279         if (chctl_flags & XPC_CHCTL_CLOSEREPLY) {
280
281                 dev_dbg(xpc_chan, "XPC_CHCTL_CLOSEREPLY received from partid="
282                         "%d, channel=%d\n", ch->partid, ch->number);
283
284                 if (ch->flags & XPC_C_DISCONNECTED) {
285                         DBUG_ON(part->act_state != XPC_P_AS_DEACTIVATING);
286                         spin_unlock_irqrestore(&ch->lock, irq_flags);
287                         return;
288                 }
289
290                 DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
291
292                 if (!(ch->flags & XPC_C_RCLOSEREQUEST)) {
293                         if (part->chctl.flags[ch_number] &
294                             XPC_CHCTL_CLOSEREQUEST) {
295
296                                 DBUG_ON(ch->delayed_chctl_flags != 0);
297                                 spin_lock(&part->chctl_lock);
298                                 part->chctl.flags[ch_number] |=
299                                     XPC_CHCTL_CLOSEREPLY;
300                                 spin_unlock(&part->chctl_lock);
301                         }
302                         spin_unlock_irqrestore(&ch->lock, irq_flags);
303                         return;
304                 }
305
306                 ch->flags |= XPC_C_RCLOSEREPLY;
307
308                 if (ch->flags & XPC_C_CLOSEREPLY) {
309                         /* both sides have finished disconnecting */
310                         xpc_process_disconnect(ch, &irq_flags);
311                 }
312         }
313
314         if (chctl_flags & XPC_CHCTL_OPENREQUEST) {
315
316                 dev_dbg(xpc_chan, "XPC_CHCTL_OPENREQUEST (entry_size=%d, "
317                         "local_nentries=%d) received from partid=%d, "
318                         "channel=%d\n", args->entry_size, args->local_nentries,
319                         ch->partid, ch->number);
320
321                 if (part->act_state == XPC_P_AS_DEACTIVATING ||
322                     (ch->flags & XPC_C_ROPENREQUEST)) {
323                         spin_unlock_irqrestore(&ch->lock, irq_flags);
324                         return;
325                 }
326
327                 if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_WDISCONNECT)) {
328                         ch->delayed_chctl_flags |= XPC_CHCTL_OPENREQUEST;
329                         spin_unlock_irqrestore(&ch->lock, irq_flags);
330                         return;
331                 }
332                 DBUG_ON(!(ch->flags & (XPC_C_DISCONNECTED |
333                                        XPC_C_OPENREQUEST)));
334                 DBUG_ON(ch->flags & (XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
335                                      XPC_C_OPENREPLY | XPC_C_CONNECTED));
336
337                 /*
338                  * The meaningful OPENREQUEST connection state fields are:
339                  *      entry_size = size of channel's messages in bytes
340                  *      local_nentries = remote partition's local_nentries
341                  */
342                 if (args->entry_size == 0 || args->local_nentries == 0) {
343                         /* assume OPENREQUEST was delayed by mistake */
344                         spin_unlock_irqrestore(&ch->lock, irq_flags);
345                         return;
346                 }
347
348                 ch->flags |= (XPC_C_ROPENREQUEST | XPC_C_CONNECTING);
349                 ch->remote_nentries = args->local_nentries;
350
351                 if (ch->flags & XPC_C_OPENREQUEST) {
352                         if (args->entry_size != ch->entry_size) {
353                                 XPC_DISCONNECT_CHANNEL(ch, xpUnequalMsgSizes,
354                                                        &irq_flags);
355                                 spin_unlock_irqrestore(&ch->lock, irq_flags);
356                                 return;
357                         }
358                 } else {
359                         ch->entry_size = args->entry_size;
360
361                         XPC_SET_REASON(ch, 0, 0);
362                         ch->flags &= ~XPC_C_DISCONNECTED;
363
364                         atomic_inc(&part->nchannels_active);
365                 }
366
367                 xpc_process_connect(ch, &irq_flags);
368         }
369
370         if (chctl_flags & XPC_CHCTL_OPENREPLY) {
371
372                 dev_dbg(xpc_chan, "XPC_CHCTL_OPENREPLY (local_msgqueue_pa="
373                         "0x%lx, local_nentries=%d, remote_nentries=%d) "
374                         "received from partid=%d, channel=%d\n",
375                         args->local_msgqueue_pa, args->local_nentries,
376                         args->remote_nentries, ch->partid, ch->number);
377
378                 if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
379                         spin_unlock_irqrestore(&ch->lock, irq_flags);
380                         return;
381                 }
382                 if (!(ch->flags & XPC_C_OPENREQUEST)) {
383                         XPC_DISCONNECT_CHANNEL(ch, xpOpenCloseError,
384                                                &irq_flags);
385                         spin_unlock_irqrestore(&ch->lock, irq_flags);
386                         return;
387                 }
388
389                 DBUG_ON(!(ch->flags & XPC_C_ROPENREQUEST));
390                 DBUG_ON(ch->flags & XPC_C_CONNECTED);
391
392                 /*
393                  * The meaningful OPENREPLY connection state fields are:
394                  *      local_msgqueue_pa = physical address of remote
395                  *                          partition's local_msgqueue
396                  *      local_nentries = remote partition's local_nentries
397                  *      remote_nentries = remote partition's remote_nentries
398                  */
399                 DBUG_ON(args->local_msgqueue_pa == 0);
400                 DBUG_ON(args->local_nentries == 0);
401                 DBUG_ON(args->remote_nentries == 0);
402
403                 ret = xpc_save_remote_msgqueue_pa(ch, args->local_msgqueue_pa);
404                 if (ret != xpSuccess) {
405                         XPC_DISCONNECT_CHANNEL(ch, ret, &irq_flags);
406                         spin_unlock_irqrestore(&ch->lock, irq_flags);
407                         return;
408                 }
409                 ch->flags |= XPC_C_ROPENREPLY;
410
411                 if (args->local_nentries < ch->remote_nentries) {
412                         dev_dbg(xpc_chan, "XPC_CHCTL_OPENREPLY: new "
413                                 "remote_nentries=%d, old remote_nentries=%d, "
414                                 "partid=%d, channel=%d\n",
415                                 args->local_nentries, ch->remote_nentries,
416                                 ch->partid, ch->number);
417
418                         ch->remote_nentries = args->local_nentries;
419                 }
420                 if (args->remote_nentries < ch->local_nentries) {
421                         dev_dbg(xpc_chan, "XPC_CHCTL_OPENREPLY: new "
422                                 "local_nentries=%d, old local_nentries=%d, "
423                                 "partid=%d, channel=%d\n",
424                                 args->remote_nentries, ch->local_nentries,
425                                 ch->partid, ch->number);
426
427                         ch->local_nentries = args->remote_nentries;
428                 }
429
430                 xpc_process_connect(ch, &irq_flags);
431         }
432
433         spin_unlock_irqrestore(&ch->lock, irq_flags);
434 }
435
436 /*
437  * Attempt to establish a channel connection to a remote partition.
438  */
439 static enum xp_retval
440 xpc_connect_channel(struct xpc_channel *ch)
441 {
442         unsigned long irq_flags;
443         struct xpc_registration *registration = &xpc_registrations[ch->number];
444
445         if (mutex_trylock(&registration->mutex) == 0)
446                 return xpRetry;
447
448         if (!XPC_CHANNEL_REGISTERED(ch->number)) {
449                 mutex_unlock(&registration->mutex);
450                 return xpUnregistered;
451         }
452
453         spin_lock_irqsave(&ch->lock, irq_flags);
454
455         DBUG_ON(ch->flags & XPC_C_CONNECTED);
456         DBUG_ON(ch->flags & XPC_C_OPENREQUEST);
457
458         if (ch->flags & XPC_C_DISCONNECTING) {
459                 spin_unlock_irqrestore(&ch->lock, irq_flags);
460                 mutex_unlock(&registration->mutex);
461                 return ch->reason;
462         }
463
464         /* add info from the channel connect registration to the channel */
465
466         ch->kthreads_assigned_limit = registration->assigned_limit;
467         ch->kthreads_idle_limit = registration->idle_limit;
468         DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);
469         DBUG_ON(atomic_read(&ch->kthreads_idle) != 0);
470         DBUG_ON(atomic_read(&ch->kthreads_active) != 0);
471
472         ch->func = registration->func;
473         DBUG_ON(registration->func == NULL);
474         ch->key = registration->key;
475
476         ch->local_nentries = registration->nentries;
477
478         if (ch->flags & XPC_C_ROPENREQUEST) {
479                 if (registration->entry_size != ch->entry_size) {
480                         /* the local and remote sides aren't the same */
481
482                         /*
483                          * Because XPC_DISCONNECT_CHANNEL() can block we're
484                          * forced to up the registration sema before we unlock
485                          * the channel lock. But that's okay here because we're
486                          * done with the part that required the registration
487                          * sema. XPC_DISCONNECT_CHANNEL() requires that the
488                          * channel lock be locked and will unlock and relock
489                          * the channel lock as needed.
490                          */
491                         mutex_unlock(&registration->mutex);
492                         XPC_DISCONNECT_CHANNEL(ch, xpUnequalMsgSizes,
493                                                &irq_flags);
494                         spin_unlock_irqrestore(&ch->lock, irq_flags);
495                         return xpUnequalMsgSizes;
496                 }
497         } else {
498                 ch->entry_size = registration->entry_size;
499
500                 XPC_SET_REASON(ch, 0, 0);
501                 ch->flags &= ~XPC_C_DISCONNECTED;
502
503                 atomic_inc(&xpc_partitions[ch->partid].nchannels_active);
504         }
505
506         mutex_unlock(&registration->mutex);
507
508         /* initiate the connection */
509
510         ch->flags |= (XPC_C_OPENREQUEST | XPC_C_CONNECTING);
511         xpc_send_chctl_openrequest(ch, &irq_flags);
512
513         xpc_process_connect(ch, &irq_flags);
514
515         spin_unlock_irqrestore(&ch->lock, irq_flags);
516
517         return xpSuccess;
518 }
519
520 void
521 xpc_process_sent_chctl_flags(struct xpc_partition *part)
522 {
523         unsigned long irq_flags;
524         union xpc_channel_ctl_flags chctl;
525         struct xpc_channel *ch;
526         int ch_number;
527         u32 ch_flags;
528
529         chctl.all_flags = xpc_get_chctl_all_flags(part);
530
531         /*
532          * Initiate channel connections for registered channels.
533          *
534          * For each connected channel that has pending messages activate idle
535          * kthreads and/or create new kthreads as needed.
536          */
537
538         for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
539                 ch = &part->channels[ch_number];
540
541                 /*
542                  * Process any open or close related chctl flags, and then deal
543                  * with connecting or disconnecting the channel as required.
544                  */
545
546                 if (chctl.flags[ch_number] & XPC_OPENCLOSE_CHCTL_FLAGS) {
547                         xpc_process_openclose_chctl_flags(part, ch_number,
548                                                         chctl.flags[ch_number]);
549                 }
550
551                 ch_flags = ch->flags;   /* need an atomic snapshot of flags */
552
553                 if (ch_flags & XPC_C_DISCONNECTING) {
554                         spin_lock_irqsave(&ch->lock, irq_flags);
555                         xpc_process_disconnect(ch, &irq_flags);
556                         spin_unlock_irqrestore(&ch->lock, irq_flags);
557                         continue;
558                 }
559
560                 if (part->act_state == XPC_P_AS_DEACTIVATING)
561                         continue;
562
563                 if (!(ch_flags & XPC_C_CONNECTED)) {
564                         if (!(ch_flags & XPC_C_OPENREQUEST)) {
565                                 DBUG_ON(ch_flags & XPC_C_SETUP);
566                                 (void)xpc_connect_channel(ch);
567                         } else {
568                                 spin_lock_irqsave(&ch->lock, irq_flags);
569                                 xpc_process_connect(ch, &irq_flags);
570                                 spin_unlock_irqrestore(&ch->lock, irq_flags);
571                         }
572                         continue;
573                 }
574
575                 /*
576                  * Process any message related chctl flags, this may involve
577                  * the activation of kthreads to deliver any pending messages
578                  * sent from the other partition.
579                  */
580
581                 if (chctl.flags[ch_number] & XPC_MSG_CHCTL_FLAGS)
582                         xpc_process_msg_chctl_flags(part, ch_number);
583         }
584 }
585
586 /*
587  * XPC's heartbeat code calls this function to inform XPC that a partition is
588  * going down.  XPC responds by tearing down the XPartition Communication
589  * infrastructure used for the just downed partition.
590  *
591  * XPC's heartbeat code will never call this function and xpc_partition_up()
592  * at the same time. Nor will it ever make multiple calls to either function
593  * at the same time.
594  */
595 void
596 xpc_partition_going_down(struct xpc_partition *part, enum xp_retval reason)
597 {
598         unsigned long irq_flags;
599         int ch_number;
600         struct xpc_channel *ch;
601
602         dev_dbg(xpc_chan, "deactivating partition %d, reason=%d\n",
603                 XPC_PARTID(part), reason);
604
605         if (!xpc_part_ref(part)) {
606                 /* infrastructure for this partition isn't currently set up */
607                 return;
608         }
609
610         /* disconnect channels associated with the partition going down */
611
612         for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
613                 ch = &part->channels[ch_number];
614
615                 xpc_msgqueue_ref(ch);
616                 spin_lock_irqsave(&ch->lock, irq_flags);
617
618                 XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
619
620                 spin_unlock_irqrestore(&ch->lock, irq_flags);
621                 xpc_msgqueue_deref(ch);
622         }
623
624         xpc_wakeup_channel_mgr(part);
625
626         xpc_part_deref(part);
627 }
628
629 /*
630  * Called by XP at the time of channel connection registration to cause
631  * XPC to establish connections to all currently active partitions.
632  */
633 void
634 xpc_initiate_connect(int ch_number)
635 {
636         short partid;
637         struct xpc_partition *part;
638         struct xpc_channel *ch;
639
640         DBUG_ON(ch_number < 0 || ch_number >= XPC_MAX_NCHANNELS);
641
642         for (partid = 0; partid < xp_max_npartitions; partid++) {
643                 part = &xpc_partitions[partid];
644
645                 if (xpc_part_ref(part)) {
646                         ch = &part->channels[ch_number];
647
648                         /*
649                          * Initiate the establishment of a connection on the
650                          * newly registered channel to the remote partition.
651                          */
652                         xpc_wakeup_channel_mgr(part);
653                         xpc_part_deref(part);
654                 }
655         }
656 }
657
658 void
659 xpc_connected_callout(struct xpc_channel *ch)
660 {
661         /* let the registerer know that a connection has been established */
662
663         if (ch->func != NULL) {
664                 dev_dbg(xpc_chan, "ch->func() called, reason=xpConnected, "
665                         "partid=%d, channel=%d\n", ch->partid, ch->number);
666
667                 ch->func(xpConnected, ch->partid, ch->number,
668                          (void *)(u64)ch->local_nentries, ch->key);
669
670                 dev_dbg(xpc_chan, "ch->func() returned, reason=xpConnected, "
671                         "partid=%d, channel=%d\n", ch->partid, ch->number);
672         }
673 }
674
675 /*
676  * Called by XP at the time of channel connection unregistration to cause
677  * XPC to teardown all current connections for the specified channel.
678  *
679  * Before returning xpc_initiate_disconnect() will wait until all connections
680  * on the specified channel have been closed/torndown. So the caller can be
681  * assured that they will not be receiving any more callouts from XPC to the
682  * function they registered via xpc_connect().
683  *
684  * Arguments:
685  *
686  *      ch_number - channel # to unregister.
687  */
688 void
689 xpc_initiate_disconnect(int ch_number)
690 {
691         unsigned long irq_flags;
692         short partid;
693         struct xpc_partition *part;
694         struct xpc_channel *ch;
695
696         DBUG_ON(ch_number < 0 || ch_number >= XPC_MAX_NCHANNELS);
697
698         /* initiate the channel disconnect for every active partition */
699         for (partid = 0; partid < xp_max_npartitions; partid++) {
700                 part = &xpc_partitions[partid];
701
702                 if (xpc_part_ref(part)) {
703                         ch = &part->channels[ch_number];
704                         xpc_msgqueue_ref(ch);
705
706                         spin_lock_irqsave(&ch->lock, irq_flags);
707
708                         if (!(ch->flags & XPC_C_DISCONNECTED)) {
709                                 ch->flags |= XPC_C_WDISCONNECT;
710
711                                 XPC_DISCONNECT_CHANNEL(ch, xpUnregistering,
712                                                        &irq_flags);
713                         }
714
715                         spin_unlock_irqrestore(&ch->lock, irq_flags);
716
717                         xpc_msgqueue_deref(ch);
718                         xpc_part_deref(part);
719                 }
720         }
721
722         xpc_disconnect_wait(ch_number);
723 }
724
725 /*
726  * To disconnect a channel, and reflect it back to all who may be waiting.
727  *
728  * An OPEN is not allowed until XPC_C_DISCONNECTING is cleared by
729  * xpc_process_disconnect(), and if set, XPC_C_WDISCONNECT is cleared by
730  * xpc_disconnect_wait().
731  *
732  * THE CHANNEL IS TO BE LOCKED BY THE CALLER AND WILL REMAIN LOCKED UPON RETURN.
733  */
734 void
735 xpc_disconnect_channel(const int line, struct xpc_channel *ch,
736                        enum xp_retval reason, unsigned long *irq_flags)
737 {
738         u32 channel_was_connected = (ch->flags & XPC_C_CONNECTED);
739
740         DBUG_ON(!spin_is_locked(&ch->lock));
741
742         if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED))
743                 return;
744
745         DBUG_ON(!(ch->flags & (XPC_C_CONNECTING | XPC_C_CONNECTED)));
746
747         dev_dbg(xpc_chan, "reason=%d, line=%d, partid=%d, channel=%d\n",
748                 reason, line, ch->partid, ch->number);
749
750         XPC_SET_REASON(ch, reason, line);
751
752         ch->flags |= (XPC_C_CLOSEREQUEST | XPC_C_DISCONNECTING);
753         /* some of these may not have been set */
754         ch->flags &= ~(XPC_C_OPENREQUEST | XPC_C_OPENREPLY |
755                        XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
756                        XPC_C_CONNECTING | XPC_C_CONNECTED);
757
758         xpc_send_chctl_closerequest(ch, irq_flags);
759
760         if (channel_was_connected)
761                 ch->flags |= XPC_C_WASCONNECTED;
762
763         spin_unlock_irqrestore(&ch->lock, *irq_flags);
764
765         /* wake all idle kthreads so they can exit */
766         if (atomic_read(&ch->kthreads_idle) > 0) {
767                 wake_up_all(&ch->idle_wq);
768
769         } else if ((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
770                    !(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) {
771                 /* start a kthread that will do the xpDisconnecting callout */
772                 xpc_create_kthreads(ch, 1, 1);
773         }
774
775         /* wake those waiting to allocate an entry from the local msg queue */
776         if (atomic_read(&ch->n_on_msg_allocate_wq) > 0)
777                 wake_up(&ch->msg_allocate_wq);
778
779         spin_lock_irqsave(&ch->lock, *irq_flags);
780 }
781
782 void
783 xpc_disconnect_callout(struct xpc_channel *ch, enum xp_retval reason)
784 {
785         /*
786          * Let the channel's registerer know that the channel is being
787          * disconnected. We don't want to do this if the registerer was never
788          * informed of a connection being made.
789          */
790
791         if (ch->func != NULL) {
792                 dev_dbg(xpc_chan, "ch->func() called, reason=%d, partid=%d, "
793                         "channel=%d\n", reason, ch->partid, ch->number);
794
795                 ch->func(reason, ch->partid, ch->number, NULL, ch->key);
796
797                 dev_dbg(xpc_chan, "ch->func() returned, reason=%d, partid=%d, "
798                         "channel=%d\n", reason, ch->partid, ch->number);
799         }
800 }
801
802 /*
803  * Wait for a message entry to become available for the specified channel,
804  * but don't wait any longer than 1 jiffy.
805  */
806 enum xp_retval
807 xpc_allocate_msg_wait(struct xpc_channel *ch)
808 {
809         enum xp_retval ret;
810
811         if (ch->flags & XPC_C_DISCONNECTING) {
812                 DBUG_ON(ch->reason == xpInterrupted);
813                 return ch->reason;
814         }
815
816         atomic_inc(&ch->n_on_msg_allocate_wq);
817         ret = interruptible_sleep_on_timeout(&ch->msg_allocate_wq, 1);
818         atomic_dec(&ch->n_on_msg_allocate_wq);
819
820         if (ch->flags & XPC_C_DISCONNECTING) {
821                 ret = ch->reason;
822                 DBUG_ON(ch->reason == xpInterrupted);
823         } else if (ret == 0) {
824                 ret = xpTimeout;
825         } else {
826                 ret = xpInterrupted;
827         }
828
829         return ret;
830 }
831
832 /*
833  * Send a message that contains the user's payload on the specified channel
834  * connected to the specified partition.
835  *
836  * NOTE that this routine can sleep waiting for a message entry to become
837  * available. To not sleep, pass in the XPC_NOWAIT flag.
838  *
839  * Once sent, this routine will not wait for the message to be received, nor
840  * will notification be given when it does happen.
841  *
842  * Arguments:
843  *
844  *      partid - ID of partition to which the channel is connected.
845  *      ch_number - channel # to send message on.
846  *      flags - see xp.h for valid flags.
847  *      payload - pointer to the payload which is to be sent.
848  *      payload_size - size of the payload in bytes.
849  */
850 enum xp_retval
851 xpc_initiate_send(short partid, int ch_number, u32 flags, void *payload,
852                   u16 payload_size)
853 {
854         struct xpc_partition *part = &xpc_partitions[partid];
855         enum xp_retval ret = xpUnknownReason;
856
857         dev_dbg(xpc_chan, "payload=0x%p, partid=%d, channel=%d\n", payload,
858                 partid, ch_number);
859
860         DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
861         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
862         DBUG_ON(payload == NULL);
863
864         if (xpc_part_ref(part)) {
865                 ret = xpc_send_payload(&part->channels[ch_number], flags,
866                                        payload, payload_size, 0, NULL, NULL);
867                 xpc_part_deref(part);
868         }
869
870         return ret;
871 }
872
873 /*
874  * Send a message that contains the user's payload on the specified channel
875  * connected to the specified partition.
876  *
877  * NOTE that this routine can sleep waiting for a message entry to become
878  * available. To not sleep, pass in the XPC_NOWAIT flag.
879  *
880  * This routine will not wait for the message to be sent or received.
881  *
882  * Once the remote end of the channel has received the message, the function
883  * passed as an argument to xpc_initiate_send_notify() will be called. This
884  * allows the sender to free up or re-use any buffers referenced by the
885  * message, but does NOT mean the message has been processed at the remote
886  * end by a receiver.
887  *
888  * If this routine returns an error, the caller's function will NOT be called.
889  *
890  * Arguments:
891  *
892  *      partid - ID of partition to which the channel is connected.
893  *      ch_number - channel # to send message on.
894  *      flags - see xp.h for valid flags.
895  *      payload - pointer to the payload which is to be sent.
896  *      payload_size - size of the payload in bytes.
897  *      func - function to call with asynchronous notification of message
898  *                receipt. THIS FUNCTION MUST BE NON-BLOCKING.
899  *      key - user-defined key to be passed to the function when it's called.
900  */
901 enum xp_retval
902 xpc_initiate_send_notify(short partid, int ch_number, u32 flags, void *payload,
903                          u16 payload_size, xpc_notify_func func, void *key)
904 {
905         struct xpc_partition *part = &xpc_partitions[partid];
906         enum xp_retval ret = xpUnknownReason;
907
908         dev_dbg(xpc_chan, "payload=0x%p, partid=%d, channel=%d\n", payload,
909                 partid, ch_number);
910
911         DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
912         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
913         DBUG_ON(payload == NULL);
914         DBUG_ON(func == NULL);
915
916         if (xpc_part_ref(part)) {
917                 ret = xpc_send_payload(&part->channels[ch_number], flags,
918                                        payload, payload_size, XPC_N_CALL, func,
919                                        key);
920                 xpc_part_deref(part);
921         }
922         return ret;
923 }
924
925 /*
926  * Deliver a message's payload to its intended recipient.
927  */
928 void
929 xpc_deliver_payload(struct xpc_channel *ch)
930 {
931         void *payload;
932
933         payload = xpc_get_deliverable_payload(ch);
934         if (payload != NULL) {
935
936                 /*
937                  * This ref is taken to protect the payload itself from being
938                  * freed before the user is finished with it, which the user
939                  * indicates by calling xpc_initiate_received().
940                  */
941                 xpc_msgqueue_ref(ch);
942
943                 atomic_inc(&ch->kthreads_active);
944
945                 if (ch->func != NULL) {
946                         dev_dbg(xpc_chan, "ch->func() called, payload=0x%p "
947                                 "partid=%d channel=%d\n", payload, ch->partid,
948                                 ch->number);
949
950                         /* deliver the message to its intended recipient */
951                         ch->func(xpMsgReceived, ch->partid, ch->number, payload,
952                                  ch->key);
953
954                         dev_dbg(xpc_chan, "ch->func() returned, payload=0x%p "
955                                 "partid=%d channel=%d\n", payload, ch->partid,
956                                 ch->number);
957                 }
958
959                 atomic_dec(&ch->kthreads_active);
960         }
961 }
962
963 /*
964  * Acknowledge receipt of a delivered message's payload.
965  *
966  * This function, although called by users, does not call xpc_part_ref() to
967  * ensure that the partition infrastructure is in place. It relies on the
968  * fact that we called xpc_msgqueue_ref() in xpc_deliver_payload().
969  *
970  * Arguments:
971  *
972  *      partid - ID of partition to which the channel is connected.
973  *      ch_number - channel # message received on.
974  *      payload - pointer to the payload area allocated via
975  *                      xpc_initiate_send() or xpc_initiate_send_notify().
976  */
977 void
978 xpc_initiate_received(short partid, int ch_number, void *payload)
979 {
980         struct xpc_partition *part = &xpc_partitions[partid];
981         struct xpc_channel *ch;
982
983         DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
984         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
985
986         ch = &part->channels[ch_number];
987         xpc_received_payload(ch, payload);
988
989         /* the call to xpc_msgqueue_ref() was done by xpc_deliver_payload()  */
990         xpc_msgqueue_deref(ch);
991 }