[PATCH] inotify: misc cleanup
[linux-3.10.git] / arch / ia64 / sn / kernel / 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-2005 Silicon Graphics, Inc.  All Rights Reserved.
7  */
8
9
10 /*
11  * Cross Partition Communication (XPC) channel support.
12  *
13  *      This is the part of XPC that manages the channels and
14  *      sends/receives messages across them to/from other partitions.
15  *
16  */
17
18
19 #include <linux/kernel.h>
20 #include <linux/init.h>
21 #include <linux/sched.h>
22 #include <linux/cache.h>
23 #include <linux/interrupt.h>
24 #include <linux/slab.h>
25 #include <asm/sn/bte.h>
26 #include <asm/sn/sn_sal.h>
27 #include "xpc.h"
28
29
30 /*
31  * Set up the initial values for the XPartition Communication channels.
32  */
33 static void
34 xpc_initialize_channels(struct xpc_partition *part, partid_t partid)
35 {
36         int ch_number;
37         struct xpc_channel *ch;
38
39
40         for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
41                 ch = &part->channels[ch_number];
42
43                 ch->partid = partid;
44                 ch->number = ch_number;
45                 ch->flags = XPC_C_DISCONNECTED;
46
47                 ch->local_GP = &part->local_GPs[ch_number];
48                 ch->local_openclose_args =
49                                         &part->local_openclose_args[ch_number];
50
51                 atomic_set(&ch->kthreads_assigned, 0);
52                 atomic_set(&ch->kthreads_idle, 0);
53                 atomic_set(&ch->kthreads_active, 0);
54
55                 atomic_set(&ch->references, 0);
56                 atomic_set(&ch->n_to_notify, 0);
57
58                 spin_lock_init(&ch->lock);
59                 sema_init(&ch->msg_to_pull_sema, 1);    /* mutex */
60
61                 atomic_set(&ch->n_on_msg_allocate_wq, 0);
62                 init_waitqueue_head(&ch->msg_allocate_wq);
63                 init_waitqueue_head(&ch->idle_wq);
64         }
65 }
66
67
68 /*
69  * Setup the infrastructure necessary to support XPartition Communication
70  * between the specified remote partition and the local one.
71  */
72 enum xpc_retval
73 xpc_setup_infrastructure(struct xpc_partition *part)
74 {
75         int ret;
76         struct timer_list *timer;
77         partid_t partid = XPC_PARTID(part);
78
79
80         /*
81          * Zero out MOST of the entry for this partition. Only the fields
82          * starting with `nchannels' will be zeroed. The preceding fields must
83          * remain `viable' across partition ups and downs, since they may be
84          * referenced during this memset() operation.
85          */
86         memset(&part->nchannels, 0, sizeof(struct xpc_partition) -
87                                 offsetof(struct xpc_partition, nchannels));
88
89         /*
90          * Allocate all of the channel structures as a contiguous chunk of
91          * memory.
92          */
93         part->channels = kmalloc(sizeof(struct xpc_channel) * XPC_NCHANNELS,
94                                                                 GFP_KERNEL);
95         if (part->channels == NULL) {
96                 dev_err(xpc_chan, "can't get memory for channels\n");
97                 return xpcNoMemory;
98         }
99         memset(part->channels, 0, sizeof(struct xpc_channel) * XPC_NCHANNELS);
100
101         part->nchannels = XPC_NCHANNELS;
102
103
104         /* allocate all the required GET/PUT values */
105
106         part->local_GPs = xpc_kmalloc_cacheline_aligned(XPC_GP_SIZE,
107                                         GFP_KERNEL, &part->local_GPs_base);
108         if (part->local_GPs == NULL) {
109                 kfree(part->channels);
110                 part->channels = NULL;
111                 dev_err(xpc_chan, "can't get memory for local get/put "
112                         "values\n");
113                 return xpcNoMemory;
114         }
115         memset(part->local_GPs, 0, XPC_GP_SIZE);
116
117         part->remote_GPs = xpc_kmalloc_cacheline_aligned(XPC_GP_SIZE,
118                                         GFP_KERNEL, &part->remote_GPs_base);
119         if (part->remote_GPs == NULL) {
120                 kfree(part->channels);
121                 part->channels = NULL;
122                 kfree(part->local_GPs_base);
123                 part->local_GPs = NULL;
124                 dev_err(xpc_chan, "can't get memory for remote get/put "
125                         "values\n");
126                 return xpcNoMemory;
127         }
128         memset(part->remote_GPs, 0, XPC_GP_SIZE);
129
130
131         /* allocate all the required open and close args */
132
133         part->local_openclose_args = xpc_kmalloc_cacheline_aligned(
134                                         XPC_OPENCLOSE_ARGS_SIZE, GFP_KERNEL,
135                                         &part->local_openclose_args_base);
136         if (part->local_openclose_args == NULL) {
137                 kfree(part->channels);
138                 part->channels = NULL;
139                 kfree(part->local_GPs_base);
140                 part->local_GPs = NULL;
141                 kfree(part->remote_GPs_base);
142                 part->remote_GPs = NULL;
143                 dev_err(xpc_chan, "can't get memory for local connect args\n");
144                 return xpcNoMemory;
145         }
146         memset(part->local_openclose_args, 0, XPC_OPENCLOSE_ARGS_SIZE);
147
148         part->remote_openclose_args = xpc_kmalloc_cacheline_aligned(
149                                         XPC_OPENCLOSE_ARGS_SIZE, GFP_KERNEL,
150                                         &part->remote_openclose_args_base);
151         if (part->remote_openclose_args == NULL) {
152                 kfree(part->channels);
153                 part->channels = NULL;
154                 kfree(part->local_GPs_base);
155                 part->local_GPs = NULL;
156                 kfree(part->remote_GPs_base);
157                 part->remote_GPs = NULL;
158                 kfree(part->local_openclose_args_base);
159                 part->local_openclose_args = NULL;
160                 dev_err(xpc_chan, "can't get memory for remote connect args\n");
161                 return xpcNoMemory;
162         }
163         memset(part->remote_openclose_args, 0, XPC_OPENCLOSE_ARGS_SIZE);
164
165
166         xpc_initialize_channels(part, partid);
167
168         atomic_set(&part->nchannels_active, 0);
169
170
171         /* local_IPI_amo were set to 0 by an earlier memset() */
172
173         /* Initialize this partitions AMO_t structure */
174         part->local_IPI_amo_va = xpc_IPI_init(partid);
175
176         spin_lock_init(&part->IPI_lock);
177
178         atomic_set(&part->channel_mgr_requests, 1);
179         init_waitqueue_head(&part->channel_mgr_wq);
180
181         sprintf(part->IPI_owner, "xpc%02d", partid);
182         ret = request_irq(SGI_XPC_NOTIFY, xpc_notify_IRQ_handler, SA_SHIRQ,
183                                 part->IPI_owner, (void *) (u64) partid);
184         if (ret != 0) {
185                 kfree(part->channels);
186                 part->channels = NULL;
187                 kfree(part->local_GPs_base);
188                 part->local_GPs = NULL;
189                 kfree(part->remote_GPs_base);
190                 part->remote_GPs = NULL;
191                 kfree(part->local_openclose_args_base);
192                 part->local_openclose_args = NULL;
193                 kfree(part->remote_openclose_args_base);
194                 part->remote_openclose_args = NULL;
195                 dev_err(xpc_chan, "can't register NOTIFY IRQ handler, "
196                         "errno=%d\n", -ret);
197                 return xpcLackOfResources;
198         }
199
200         /* Setup a timer to check for dropped IPIs */
201         timer = &part->dropped_IPI_timer;
202         init_timer(timer);
203         timer->function = (void (*)(unsigned long)) xpc_dropped_IPI_check;
204         timer->data = (unsigned long) part;
205         timer->expires = jiffies + XPC_P_DROPPED_IPI_WAIT;
206         add_timer(timer);
207
208         /*
209          * With the setting of the partition setup_state to XPC_P_SETUP, we're
210          * declaring that this partition is ready to go.
211          */
212         part->setup_state = XPC_P_SETUP;
213
214
215         /*
216          * Setup the per partition specific variables required by the
217          * remote partition to establish channel connections with us.
218          *
219          * The setting of the magic # indicates that these per partition
220          * specific variables are ready to be used.
221          */
222         xpc_vars_part[partid].GPs_pa = __pa(part->local_GPs);
223         xpc_vars_part[partid].openclose_args_pa =
224                                         __pa(part->local_openclose_args);
225         xpc_vars_part[partid].IPI_amo_pa = __pa(part->local_IPI_amo_va);
226         xpc_vars_part[partid].IPI_nasid = cpuid_to_nasid(smp_processor_id());
227         xpc_vars_part[partid].IPI_phys_cpuid =
228                                         cpu_physical_id(smp_processor_id());
229         xpc_vars_part[partid].nchannels = part->nchannels;
230         xpc_vars_part[partid].magic = XPC_VP_MAGIC1;
231
232         return xpcSuccess;
233 }
234
235
236 /*
237  * Create a wrapper that hides the underlying mechanism for pulling a cacheline
238  * (or multiple cachelines) from a remote partition.
239  *
240  * src must be a cacheline aligned physical address on the remote partition.
241  * dst must be a cacheline aligned virtual address on this partition.
242  * cnt must be an cacheline sized
243  */
244 static enum xpc_retval
245 xpc_pull_remote_cachelines(struct xpc_partition *part, void *dst,
246                                 const void *src, size_t cnt)
247 {
248         bte_result_t bte_ret;
249
250
251         DBUG_ON((u64) src != L1_CACHE_ALIGN((u64) src));
252         DBUG_ON((u64) dst != L1_CACHE_ALIGN((u64) dst));
253         DBUG_ON(cnt != L1_CACHE_ALIGN(cnt));
254
255         if (part->act_state == XPC_P_DEACTIVATING) {
256                 return part->reason;
257         }
258
259         bte_ret = xp_bte_copy((u64) src, (u64) ia64_tpa((u64) dst),
260                                 (u64) cnt, (BTE_NORMAL | BTE_WACQUIRE), NULL);
261         if (bte_ret == BTE_SUCCESS) {
262                 return xpcSuccess;
263         }
264
265         dev_dbg(xpc_chan, "xp_bte_copy() from partition %d failed, ret=%d\n",
266                 XPC_PARTID(part), bte_ret);
267
268         return xpc_map_bte_errors(bte_ret);
269 }
270
271
272 /*
273  * Pull the remote per partititon specific variables from the specified
274  * partition.
275  */
276 enum xpc_retval
277 xpc_pull_remote_vars_part(struct xpc_partition *part)
278 {
279         u8 buffer[L1_CACHE_BYTES * 2];
280         struct xpc_vars_part *pulled_entry_cacheline =
281                         (struct xpc_vars_part *) L1_CACHE_ALIGN((u64) buffer);
282         struct xpc_vars_part *pulled_entry;
283         u64 remote_entry_cacheline_pa, remote_entry_pa;
284         partid_t partid = XPC_PARTID(part);
285         enum xpc_retval ret;
286
287
288         /* pull the cacheline that contains the variables we're interested in */
289
290         DBUG_ON(part->remote_vars_part_pa !=
291                                 L1_CACHE_ALIGN(part->remote_vars_part_pa));
292         DBUG_ON(sizeof(struct xpc_vars_part) != L1_CACHE_BYTES / 2);
293
294         remote_entry_pa = part->remote_vars_part_pa +
295                         sn_partition_id * sizeof(struct xpc_vars_part);
296
297         remote_entry_cacheline_pa = (remote_entry_pa & ~(L1_CACHE_BYTES - 1));
298
299         pulled_entry = (struct xpc_vars_part *) ((u64) pulled_entry_cacheline +
300                                 (remote_entry_pa & (L1_CACHE_BYTES - 1)));
301
302         ret = xpc_pull_remote_cachelines(part, pulled_entry_cacheline,
303                                         (void *) remote_entry_cacheline_pa,
304                                         L1_CACHE_BYTES);
305         if (ret != xpcSuccess) {
306                 dev_dbg(xpc_chan, "failed to pull XPC vars_part from "
307                         "partition %d, ret=%d\n", partid, ret);
308                 return ret;
309         }
310
311
312         /* see if they've been set up yet */
313
314         if (pulled_entry->magic != XPC_VP_MAGIC1 &&
315                                 pulled_entry->magic != XPC_VP_MAGIC2) {
316
317                 if (pulled_entry->magic != 0) {
318                         dev_dbg(xpc_chan, "partition %d's XPC vars_part for "
319                                 "partition %d has bad magic value (=0x%lx)\n",
320                                 partid, sn_partition_id, pulled_entry->magic);
321                         return xpcBadMagic;
322                 }
323
324                 /* they've not been initialized yet */
325                 return xpcRetry;
326         }
327
328         if (xpc_vars_part[partid].magic == XPC_VP_MAGIC1) {
329
330                 /* validate the variables */
331
332                 if (pulled_entry->GPs_pa == 0 ||
333                                 pulled_entry->openclose_args_pa == 0 ||
334                                         pulled_entry->IPI_amo_pa == 0) {
335
336                         dev_err(xpc_chan, "partition %d's XPC vars_part for "
337                                 "partition %d are not valid\n", partid,
338                                 sn_partition_id);
339                         return xpcInvalidAddress;
340                 }
341
342                 /* the variables we imported look to be valid */
343
344                 part->remote_GPs_pa = pulled_entry->GPs_pa;
345                 part->remote_openclose_args_pa =
346                                         pulled_entry->openclose_args_pa;
347                 part->remote_IPI_amo_va =
348                                       (AMO_t *) __va(pulled_entry->IPI_amo_pa);
349                 part->remote_IPI_nasid = pulled_entry->IPI_nasid;
350                 part->remote_IPI_phys_cpuid = pulled_entry->IPI_phys_cpuid;
351
352                 if (part->nchannels > pulled_entry->nchannels) {
353                         part->nchannels = pulled_entry->nchannels;
354                 }
355
356                 /* let the other side know that we've pulled their variables */
357
358                 xpc_vars_part[partid].magic = XPC_VP_MAGIC2;
359         }
360
361         if (pulled_entry->magic == XPC_VP_MAGIC1) {
362                 return xpcRetry;
363         }
364
365         return xpcSuccess;
366 }
367
368
369 /*
370  * Get the IPI flags and pull the openclose args and/or remote GPs as needed.
371  */
372 static u64
373 xpc_get_IPI_flags(struct xpc_partition *part)
374 {
375         unsigned long irq_flags;
376         u64 IPI_amo;
377         enum xpc_retval ret;
378
379
380         /*
381          * See if there are any IPI flags to be handled.
382          */
383
384         spin_lock_irqsave(&part->IPI_lock, irq_flags);
385         if ((IPI_amo = part->local_IPI_amo) != 0) {
386                 part->local_IPI_amo = 0;
387         }
388         spin_unlock_irqrestore(&part->IPI_lock, irq_flags);
389
390
391         if (XPC_ANY_OPENCLOSE_IPI_FLAGS_SET(IPI_amo)) {
392                 ret = xpc_pull_remote_cachelines(part,
393                                         part->remote_openclose_args,
394                                         (void *) part->remote_openclose_args_pa,
395                                         XPC_OPENCLOSE_ARGS_SIZE);
396                 if (ret != xpcSuccess) {
397                         XPC_DEACTIVATE_PARTITION(part, ret);
398
399                         dev_dbg(xpc_chan, "failed to pull openclose args from "
400                                 "partition %d, ret=%d\n", XPC_PARTID(part),
401                                 ret);
402
403                         /* don't bother processing IPIs anymore */
404                         IPI_amo = 0;
405                 }
406         }
407
408         if (XPC_ANY_MSG_IPI_FLAGS_SET(IPI_amo)) {
409                 ret = xpc_pull_remote_cachelines(part, part->remote_GPs,
410                                                 (void *) part->remote_GPs_pa,
411                                                 XPC_GP_SIZE);
412                 if (ret != xpcSuccess) {
413                         XPC_DEACTIVATE_PARTITION(part, ret);
414
415                         dev_dbg(xpc_chan, "failed to pull GPs from partition "
416                                 "%d, ret=%d\n", XPC_PARTID(part), ret);
417
418                         /* don't bother processing IPIs anymore */
419                         IPI_amo = 0;
420                 }
421         }
422
423         return IPI_amo;
424 }
425
426
427 /*
428  * Allocate the local message queue and the notify queue.
429  */
430 static enum xpc_retval
431 xpc_allocate_local_msgqueue(struct xpc_channel *ch)
432 {
433         unsigned long irq_flags;
434         int nentries;
435         size_t nbytes;
436
437
438         // >>> may want to check for ch->flags & XPC_C_DISCONNECTING between
439         // >>> iterations of the for-loop, bail if set?
440
441         // >>> should we impose a minumum #of entries? like 4 or 8?
442         for (nentries = ch->local_nentries; nentries > 0; nentries--) {
443
444                 nbytes = nentries * ch->msg_size;
445                 ch->local_msgqueue = xpc_kmalloc_cacheline_aligned(nbytes,
446                                                 (GFP_KERNEL | GFP_DMA),
447                                                 &ch->local_msgqueue_base);
448                 if (ch->local_msgqueue == NULL) {
449                         continue;
450                 }
451                 memset(ch->local_msgqueue, 0, nbytes);
452
453                 nbytes = nentries * sizeof(struct xpc_notify);
454                 ch->notify_queue = kmalloc(nbytes, (GFP_KERNEL | GFP_DMA));
455                 if (ch->notify_queue == NULL) {
456                         kfree(ch->local_msgqueue_base);
457                         ch->local_msgqueue = NULL;
458                         continue;
459                 }
460                 memset(ch->notify_queue, 0, nbytes);
461
462                 spin_lock_irqsave(&ch->lock, irq_flags);
463                 if (nentries < ch->local_nentries) {
464                         dev_dbg(xpc_chan, "nentries=%d local_nentries=%d, "
465                                 "partid=%d, channel=%d\n", nentries,
466                                 ch->local_nentries, ch->partid, ch->number);
467
468                         ch->local_nentries = nentries;
469                 }
470                 spin_unlock_irqrestore(&ch->lock, irq_flags);
471                 return xpcSuccess;
472         }
473
474         dev_dbg(xpc_chan, "can't get memory for local message queue and notify "
475                 "queue, partid=%d, channel=%d\n", ch->partid, ch->number);
476         return xpcNoMemory;
477 }
478
479
480 /*
481  * Allocate the cached remote message queue.
482  */
483 static enum xpc_retval
484 xpc_allocate_remote_msgqueue(struct xpc_channel *ch)
485 {
486         unsigned long irq_flags;
487         int nentries;
488         size_t nbytes;
489
490
491         DBUG_ON(ch->remote_nentries <= 0);
492
493         // >>> may want to check for ch->flags & XPC_C_DISCONNECTING between
494         // >>> iterations of the for-loop, bail if set?
495
496         // >>> should we impose a minumum #of entries? like 4 or 8?
497         for (nentries = ch->remote_nentries; nentries > 0; nentries--) {
498
499                 nbytes = nentries * ch->msg_size;
500                 ch->remote_msgqueue = xpc_kmalloc_cacheline_aligned(nbytes,
501                                                 (GFP_KERNEL | GFP_DMA),
502                                                 &ch->remote_msgqueue_base);
503                 if (ch->remote_msgqueue == NULL) {
504                         continue;
505                 }
506                 memset(ch->remote_msgqueue, 0, nbytes);
507
508                 spin_lock_irqsave(&ch->lock, irq_flags);
509                 if (nentries < ch->remote_nentries) {
510                         dev_dbg(xpc_chan, "nentries=%d remote_nentries=%d, "
511                                 "partid=%d, channel=%d\n", nentries,
512                                 ch->remote_nentries, ch->partid, ch->number);
513
514                         ch->remote_nentries = nentries;
515                 }
516                 spin_unlock_irqrestore(&ch->lock, irq_flags);
517                 return xpcSuccess;
518         }
519
520         dev_dbg(xpc_chan, "can't get memory for cached remote message queue, "
521                 "partid=%d, channel=%d\n", ch->partid, ch->number);
522         return xpcNoMemory;
523 }
524
525
526 /*
527  * Allocate message queues and other stuff associated with a channel.
528  *
529  * Note: Assumes all of the channel sizes are filled in.
530  */
531 static enum xpc_retval
532 xpc_allocate_msgqueues(struct xpc_channel *ch)
533 {
534         unsigned long irq_flags;
535         int i;
536         enum xpc_retval ret;
537
538
539         DBUG_ON(ch->flags & XPC_C_SETUP);
540
541         if ((ret = xpc_allocate_local_msgqueue(ch)) != xpcSuccess) {
542                 return ret;
543         }
544
545         if ((ret = xpc_allocate_remote_msgqueue(ch)) != xpcSuccess) {
546                 kfree(ch->local_msgqueue_base);
547                 ch->local_msgqueue = NULL;
548                 kfree(ch->notify_queue);
549                 ch->notify_queue = NULL;
550                 return ret;
551         }
552
553         for (i = 0; i < ch->local_nentries; i++) {
554                 /* use a semaphore as an event wait queue */
555                 sema_init(&ch->notify_queue[i].sema, 0);
556         }
557
558         sema_init(&ch->teardown_sema, 0);       /* event wait */
559
560         spin_lock_irqsave(&ch->lock, irq_flags);
561         ch->flags |= XPC_C_SETUP;
562         spin_unlock_irqrestore(&ch->lock, irq_flags);
563
564         return xpcSuccess;
565 }
566
567
568 /*
569  * Process a connect message from a remote partition.
570  *
571  * Note: xpc_process_connect() is expecting to be called with the
572  * spin_lock_irqsave held and will leave it locked upon return.
573  */
574 static void
575 xpc_process_connect(struct xpc_channel *ch, unsigned long *irq_flags)
576 {
577         enum xpc_retval ret;
578
579
580         DBUG_ON(!spin_is_locked(&ch->lock));
581
582         if (!(ch->flags & XPC_C_OPENREQUEST) ||
583                                 !(ch->flags & XPC_C_ROPENREQUEST)) {
584                 /* nothing more to do for now */
585                 return;
586         }
587         DBUG_ON(!(ch->flags & XPC_C_CONNECTING));
588
589         if (!(ch->flags & XPC_C_SETUP)) {
590                 spin_unlock_irqrestore(&ch->lock, *irq_flags);
591                 ret = xpc_allocate_msgqueues(ch);
592                 spin_lock_irqsave(&ch->lock, *irq_flags);
593
594                 if (ret != xpcSuccess) {
595                         XPC_DISCONNECT_CHANNEL(ch, ret, irq_flags);
596                 }
597                 if (ch->flags & (XPC_C_CONNECTED | XPC_C_DISCONNECTING)) {
598                         return;
599                 }
600
601                 DBUG_ON(!(ch->flags & XPC_C_SETUP));
602                 DBUG_ON(ch->local_msgqueue == NULL);
603                 DBUG_ON(ch->remote_msgqueue == NULL);
604         }
605
606         if (!(ch->flags & XPC_C_OPENREPLY)) {
607                 ch->flags |= XPC_C_OPENREPLY;
608                 xpc_IPI_send_openreply(ch, irq_flags);
609         }
610
611         if (!(ch->flags & XPC_C_ROPENREPLY)) {
612                 return;
613         }
614
615         DBUG_ON(ch->remote_msgqueue_pa == 0);
616
617         ch->flags = (XPC_C_CONNECTED | XPC_C_SETUP);    /* clear all else */
618
619         dev_info(xpc_chan, "channel %d to partition %d connected\n",
620                 ch->number, ch->partid);
621
622         spin_unlock_irqrestore(&ch->lock, *irq_flags);
623         xpc_create_kthreads(ch, 1);
624         spin_lock_irqsave(&ch->lock, *irq_flags);
625 }
626
627
628 /*
629  * Free up message queues and other stuff that were allocated for the specified
630  * channel.
631  *
632  * Note: ch->reason and ch->reason_line are left set for debugging purposes,
633  * they're cleared when XPC_C_DISCONNECTED is cleared.
634  */
635 static void
636 xpc_free_msgqueues(struct xpc_channel *ch)
637 {
638         DBUG_ON(!spin_is_locked(&ch->lock));
639         DBUG_ON(atomic_read(&ch->n_to_notify) != 0);
640
641         ch->remote_msgqueue_pa = 0;
642         ch->func = NULL;
643         ch->key = NULL;
644         ch->msg_size = 0;
645         ch->local_nentries = 0;
646         ch->remote_nentries = 0;
647         ch->kthreads_assigned_limit = 0;
648         ch->kthreads_idle_limit = 0;
649
650         ch->local_GP->get = 0;
651         ch->local_GP->put = 0;
652         ch->remote_GP.get = 0;
653         ch->remote_GP.put = 0;
654         ch->w_local_GP.get = 0;
655         ch->w_local_GP.put = 0;
656         ch->w_remote_GP.get = 0;
657         ch->w_remote_GP.put = 0;
658         ch->next_msg_to_pull = 0;
659
660         if (ch->flags & XPC_C_SETUP) {
661                 ch->flags &= ~XPC_C_SETUP;
662
663                 dev_dbg(xpc_chan, "ch->flags=0x%x, partid=%d, channel=%d\n",
664                         ch->flags, ch->partid, ch->number);
665
666                 kfree(ch->local_msgqueue_base);
667                 ch->local_msgqueue = NULL;
668                 kfree(ch->remote_msgqueue_base);
669                 ch->remote_msgqueue = NULL;
670                 kfree(ch->notify_queue);
671                 ch->notify_queue = NULL;
672
673                 /* in case someone is waiting for the teardown to complete */
674                 up(&ch->teardown_sema);
675         }
676 }
677
678
679 /*
680  * spin_lock_irqsave() is expected to be held on entry.
681  */
682 static void
683 xpc_process_disconnect(struct xpc_channel *ch, unsigned long *irq_flags)
684 {
685         struct xpc_partition *part = &xpc_partitions[ch->partid];
686         u32 ch_flags = ch->flags;
687
688
689         DBUG_ON(!spin_is_locked(&ch->lock));
690
691         if (!(ch->flags & XPC_C_DISCONNECTING)) {
692                 return;
693         }
694
695         DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
696
697         /* make sure all activity has settled down first */
698
699         if (atomic_read(&ch->references) > 0) {
700                 return;
701         }
702         DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);
703
704         /* it's now safe to free the channel's message queues */
705
706         xpc_free_msgqueues(ch);
707         DBUG_ON(ch->flags & XPC_C_SETUP);
708
709         if (part->act_state != XPC_P_DEACTIVATING) {
710
711                 /* as long as the other side is up do the full protocol */
712
713                 if (!(ch->flags & XPC_C_RCLOSEREQUEST)) {
714                         return;
715                 }
716
717                 if (!(ch->flags & XPC_C_CLOSEREPLY)) {
718                         ch->flags |= XPC_C_CLOSEREPLY;
719                         xpc_IPI_send_closereply(ch, irq_flags);
720                 }
721
722                 if (!(ch->flags & XPC_C_RCLOSEREPLY)) {
723                         return;
724                 }
725         }
726
727         /* both sides are disconnected now */
728
729         ch->flags = XPC_C_DISCONNECTED; /* clear all flags, but this one */
730
731         atomic_dec(&part->nchannels_active);
732
733         if (ch_flags & XPC_C_WASCONNECTED) {
734                 dev_info(xpc_chan, "channel %d to partition %d disconnected, "
735                         "reason=%d\n", ch->number, ch->partid, ch->reason);
736         }
737 }
738
739
740 /*
741  * Process a change in the channel's remote connection state.
742  */
743 static void
744 xpc_process_openclose_IPI(struct xpc_partition *part, int ch_number,
745                                 u8 IPI_flags)
746 {
747         unsigned long irq_flags;
748         struct xpc_openclose_args *args =
749                                 &part->remote_openclose_args[ch_number];
750         struct xpc_channel *ch = &part->channels[ch_number];
751         enum xpc_retval reason;
752
753
754
755         spin_lock_irqsave(&ch->lock, irq_flags);
756
757
758         if (IPI_flags & XPC_IPI_CLOSEREQUEST) {
759
760                 dev_dbg(xpc_chan, "XPC_IPI_CLOSEREQUEST (reason=%d) received "
761                         "from partid=%d, channel=%d\n", args->reason,
762                         ch->partid, ch->number);
763
764                 /*
765                  * If RCLOSEREQUEST is set, we're probably waiting for
766                  * RCLOSEREPLY. We should find it and a ROPENREQUEST packed
767                  * with this RCLOSEQREUQEST in the IPI_flags.
768                  */
769
770                 if (ch->flags & XPC_C_RCLOSEREQUEST) {
771                         DBUG_ON(!(ch->flags & XPC_C_DISCONNECTING));
772                         DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
773                         DBUG_ON(!(ch->flags & XPC_C_CLOSEREPLY));
774                         DBUG_ON(ch->flags & XPC_C_RCLOSEREPLY);
775
776                         DBUG_ON(!(IPI_flags & XPC_IPI_CLOSEREPLY));
777                         IPI_flags &= ~XPC_IPI_CLOSEREPLY;
778                         ch->flags |= XPC_C_RCLOSEREPLY;
779
780                         /* both sides have finished disconnecting */
781                         xpc_process_disconnect(ch, &irq_flags);
782                 }
783
784                 if (ch->flags & XPC_C_DISCONNECTED) {
785                         // >>> explain this section
786
787                         if (!(IPI_flags & XPC_IPI_OPENREQUEST)) {
788                                 DBUG_ON(part->act_state !=
789                                                         XPC_P_DEACTIVATING);
790                                 spin_unlock_irqrestore(&ch->lock, irq_flags);
791                                 return;
792                         }
793
794                         XPC_SET_REASON(ch, 0, 0);
795                         ch->flags &= ~XPC_C_DISCONNECTED;
796
797                         atomic_inc(&part->nchannels_active);
798                         ch->flags |= (XPC_C_CONNECTING | XPC_C_ROPENREQUEST);
799                 }
800
801                 IPI_flags &= ~(XPC_IPI_OPENREQUEST | XPC_IPI_OPENREPLY);
802
803                 /*
804                  * The meaningful CLOSEREQUEST connection state fields are:
805                  *      reason = reason connection is to be closed
806                  */
807
808                 ch->flags |= XPC_C_RCLOSEREQUEST;
809
810                 if (!(ch->flags & XPC_C_DISCONNECTING)) {
811                         reason = args->reason;
812                         if (reason <= xpcSuccess || reason > xpcUnknownReason) {
813                                 reason = xpcUnknownReason;
814                         } else if (reason == xpcUnregistering) {
815                                 reason = xpcOtherUnregistering;
816                         }
817
818                         XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
819                 } else {
820                         xpc_process_disconnect(ch, &irq_flags);
821                 }
822         }
823
824
825         if (IPI_flags & XPC_IPI_CLOSEREPLY) {
826
827                 dev_dbg(xpc_chan, "XPC_IPI_CLOSEREPLY received from partid=%d,"
828                         " channel=%d\n", ch->partid, ch->number);
829
830                 if (ch->flags & XPC_C_DISCONNECTED) {
831                         DBUG_ON(part->act_state != XPC_P_DEACTIVATING);
832                         spin_unlock_irqrestore(&ch->lock, irq_flags);
833                         return;
834                 }
835
836                 DBUG_ON(!(ch->flags & XPC_C_CLOSEREQUEST));
837                 DBUG_ON(!(ch->flags & XPC_C_RCLOSEREQUEST));
838
839                 ch->flags |= XPC_C_RCLOSEREPLY;
840
841                 if (ch->flags & XPC_C_CLOSEREPLY) {
842                         /* both sides have finished disconnecting */
843                         xpc_process_disconnect(ch, &irq_flags);
844                 }
845         }
846
847
848         if (IPI_flags & XPC_IPI_OPENREQUEST) {
849
850                 dev_dbg(xpc_chan, "XPC_IPI_OPENREQUEST (msg_size=%d, "
851                         "local_nentries=%d) received from partid=%d, "
852                         "channel=%d\n", args->msg_size, args->local_nentries,
853                         ch->partid, ch->number);
854
855                 if ((ch->flags & XPC_C_DISCONNECTING) ||
856                                         part->act_state == XPC_P_DEACTIVATING) {
857                         spin_unlock_irqrestore(&ch->lock, irq_flags);
858                         return;
859                 }
860                 DBUG_ON(!(ch->flags & (XPC_C_DISCONNECTED |
861                                                         XPC_C_OPENREQUEST)));
862                 DBUG_ON(ch->flags & (XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
863                                         XPC_C_OPENREPLY | XPC_C_CONNECTED));
864
865                 /*
866                  * The meaningful OPENREQUEST connection state fields are:
867                  *      msg_size = size of channel's messages in bytes
868                  *      local_nentries = remote partition's local_nentries
869                  */
870                 DBUG_ON(args->msg_size == 0);
871                 DBUG_ON(args->local_nentries == 0);
872
873                 ch->flags |= (XPC_C_ROPENREQUEST | XPC_C_CONNECTING);
874                 ch->remote_nentries = args->local_nentries;
875
876
877                 if (ch->flags & XPC_C_OPENREQUEST) {
878                         if (args->msg_size != ch->msg_size) {
879                                 XPC_DISCONNECT_CHANNEL(ch, xpcUnequalMsgSizes,
880                                                                 &irq_flags);
881                                 spin_unlock_irqrestore(&ch->lock, irq_flags);
882                                 return;
883                         }
884                 } else {
885                         ch->msg_size = args->msg_size;
886
887                         XPC_SET_REASON(ch, 0, 0);
888                         ch->flags &= ~XPC_C_DISCONNECTED;
889
890                         atomic_inc(&part->nchannels_active);
891                 }
892
893                 xpc_process_connect(ch, &irq_flags);
894         }
895
896
897         if (IPI_flags & XPC_IPI_OPENREPLY) {
898
899                 dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY (local_msgqueue_pa=0x%lx, "
900                         "local_nentries=%d, remote_nentries=%d) received from "
901                         "partid=%d, channel=%d\n", args->local_msgqueue_pa,
902                         args->local_nentries, args->remote_nentries,
903                         ch->partid, ch->number);
904
905                 if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
906                         spin_unlock_irqrestore(&ch->lock, irq_flags);
907                         return;
908                 }
909                 DBUG_ON(!(ch->flags & XPC_C_OPENREQUEST));
910                 DBUG_ON(!(ch->flags & XPC_C_ROPENREQUEST));
911                 DBUG_ON(ch->flags & XPC_C_CONNECTED);
912
913                 /*
914                  * The meaningful OPENREPLY connection state fields are:
915                  *      local_msgqueue_pa = physical address of remote
916                  *                          partition's local_msgqueue
917                  *      local_nentries = remote partition's local_nentries
918                  *      remote_nentries = remote partition's remote_nentries
919                  */
920                 DBUG_ON(args->local_msgqueue_pa == 0);
921                 DBUG_ON(args->local_nentries == 0);
922                 DBUG_ON(args->remote_nentries == 0);
923
924                 ch->flags |= XPC_C_ROPENREPLY;
925                 ch->remote_msgqueue_pa = args->local_msgqueue_pa;
926
927                 if (args->local_nentries < ch->remote_nentries) {
928                         dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
929                                 "remote_nentries=%d, old remote_nentries=%d, "
930                                 "partid=%d, channel=%d\n",
931                                 args->local_nentries, ch->remote_nentries,
932                                 ch->partid, ch->number);
933
934                         ch->remote_nentries = args->local_nentries;
935                 }
936                 if (args->remote_nentries < ch->local_nentries) {
937                         dev_dbg(xpc_chan, "XPC_IPI_OPENREPLY: new "
938                                 "local_nentries=%d, old local_nentries=%d, "
939                                 "partid=%d, channel=%d\n",
940                                 args->remote_nentries, ch->local_nentries,
941                                 ch->partid, ch->number);
942
943                         ch->local_nentries = args->remote_nentries;
944                 }
945
946                 xpc_process_connect(ch, &irq_flags);
947         }
948
949         spin_unlock_irqrestore(&ch->lock, irq_flags);
950 }
951
952
953 /*
954  * Attempt to establish a channel connection to a remote partition.
955  */
956 static enum xpc_retval
957 xpc_connect_channel(struct xpc_channel *ch)
958 {
959         unsigned long irq_flags;
960         struct xpc_registration *registration = &xpc_registrations[ch->number];
961
962
963         if (down_interruptible(&registration->sema) != 0) {
964                 return xpcInterrupted;
965         }
966
967         if (!XPC_CHANNEL_REGISTERED(ch->number)) {
968                 up(&registration->sema);
969                 return xpcUnregistered;
970         }
971
972         spin_lock_irqsave(&ch->lock, irq_flags);
973
974         DBUG_ON(ch->flags & XPC_C_CONNECTED);
975         DBUG_ON(ch->flags & XPC_C_OPENREQUEST);
976
977         if (ch->flags & XPC_C_DISCONNECTING) {
978                 spin_unlock_irqrestore(&ch->lock, irq_flags);
979                 up(&registration->sema);
980                 return ch->reason;
981         }
982
983
984         /* add info from the channel connect registration to the channel */
985
986         ch->kthreads_assigned_limit = registration->assigned_limit;
987         ch->kthreads_idle_limit = registration->idle_limit;
988         DBUG_ON(atomic_read(&ch->kthreads_assigned) != 0);
989         DBUG_ON(atomic_read(&ch->kthreads_idle) != 0);
990         DBUG_ON(atomic_read(&ch->kthreads_active) != 0);
991
992         ch->func = registration->func;
993         DBUG_ON(registration->func == NULL);
994         ch->key = registration->key;
995
996         ch->local_nentries = registration->nentries;
997
998         if (ch->flags & XPC_C_ROPENREQUEST) {
999                 if (registration->msg_size != ch->msg_size) {
1000                         /* the local and remote sides aren't the same */
1001
1002                         /*
1003                          * Because XPC_DISCONNECT_CHANNEL() can block we're
1004                          * forced to up the registration sema before we unlock
1005                          * the channel lock. But that's okay here because we're
1006                          * done with the part that required the registration
1007                          * sema. XPC_DISCONNECT_CHANNEL() requires that the
1008                          * channel lock be locked and will unlock and relock
1009                          * the channel lock as needed.
1010                          */
1011                         up(&registration->sema);
1012                         XPC_DISCONNECT_CHANNEL(ch, xpcUnequalMsgSizes,
1013                                                                 &irq_flags);
1014                         spin_unlock_irqrestore(&ch->lock, irq_flags);
1015                         return xpcUnequalMsgSizes;
1016                 }
1017         } else {
1018                 ch->msg_size = registration->msg_size;
1019
1020                 XPC_SET_REASON(ch, 0, 0);
1021                 ch->flags &= ~XPC_C_DISCONNECTED;
1022
1023                 atomic_inc(&xpc_partitions[ch->partid].nchannels_active);
1024         }
1025
1026         up(&registration->sema);
1027
1028
1029         /* initiate the connection */
1030
1031         ch->flags |= (XPC_C_OPENREQUEST | XPC_C_CONNECTING);
1032         xpc_IPI_send_openrequest(ch, &irq_flags);
1033
1034         xpc_process_connect(ch, &irq_flags);
1035
1036         spin_unlock_irqrestore(&ch->lock, irq_flags);
1037
1038         return xpcSuccess;
1039 }
1040
1041
1042 /*
1043  * Notify those who wanted to be notified upon delivery of their message.
1044  */
1045 static void
1046 xpc_notify_senders(struct xpc_channel *ch, enum xpc_retval reason, s64 put)
1047 {
1048         struct xpc_notify *notify;
1049         u8 notify_type;
1050         s64 get = ch->w_remote_GP.get - 1;
1051
1052
1053         while (++get < put && atomic_read(&ch->n_to_notify) > 0) {
1054
1055                 notify = &ch->notify_queue[get % ch->local_nentries];
1056
1057                 /*
1058                  * See if the notify entry indicates it was associated with
1059                  * a message who's sender wants to be notified. It is possible
1060                  * that it is, but someone else is doing or has done the
1061                  * notification.
1062                  */
1063                 notify_type = notify->type;
1064                 if (notify_type == 0 ||
1065                                 cmpxchg(&notify->type, notify_type, 0) !=
1066                                                                 notify_type) {
1067                         continue;
1068                 }
1069
1070                 DBUG_ON(notify_type != XPC_N_CALL);
1071
1072                 atomic_dec(&ch->n_to_notify);
1073
1074                 if (notify->func != NULL) {
1075                         dev_dbg(xpc_chan, "notify->func() called, notify=0x%p, "
1076                                 "msg_number=%ld, partid=%d, channel=%d\n",
1077                                 (void *) notify, get, ch->partid, ch->number);
1078
1079                         notify->func(reason, ch->partid, ch->number,
1080                                                                 notify->key);
1081
1082                         dev_dbg(xpc_chan, "notify->func() returned, "
1083                                 "notify=0x%p, msg_number=%ld, partid=%d, "
1084                                 "channel=%d\n", (void *) notify, get,
1085                                 ch->partid, ch->number);
1086                 }
1087         }
1088 }
1089
1090
1091 /*
1092  * Clear some of the msg flags in the local message queue.
1093  */
1094 static inline void
1095 xpc_clear_local_msgqueue_flags(struct xpc_channel *ch)
1096 {
1097         struct xpc_msg *msg;
1098         s64 get;
1099
1100
1101         get = ch->w_remote_GP.get;
1102         do {
1103                 msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
1104                                 (get % ch->local_nentries) * ch->msg_size);
1105                 msg->flags = 0;
1106         } while (++get < (volatile s64) ch->remote_GP.get);
1107 }
1108
1109
1110 /*
1111  * Clear some of the msg flags in the remote message queue.
1112  */
1113 static inline void
1114 xpc_clear_remote_msgqueue_flags(struct xpc_channel *ch)
1115 {
1116         struct xpc_msg *msg;
1117         s64 put;
1118
1119
1120         put = ch->w_remote_GP.put;
1121         do {
1122                 msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
1123                                 (put % ch->remote_nentries) * ch->msg_size);
1124                 msg->flags = 0;
1125         } while (++put < (volatile s64) ch->remote_GP.put);
1126 }
1127
1128
1129 static void
1130 xpc_process_msg_IPI(struct xpc_partition *part, int ch_number)
1131 {
1132         struct xpc_channel *ch = &part->channels[ch_number];
1133         int nmsgs_sent;
1134
1135
1136         ch->remote_GP = part->remote_GPs[ch_number];
1137
1138
1139         /* See what, if anything, has changed for each connected channel */
1140
1141         xpc_msgqueue_ref(ch);
1142
1143         if (ch->w_remote_GP.get == ch->remote_GP.get &&
1144                                 ch->w_remote_GP.put == ch->remote_GP.put) {
1145                 /* nothing changed since GPs were last pulled */
1146                 xpc_msgqueue_deref(ch);
1147                 return;
1148         }
1149
1150         if (!(ch->flags & XPC_C_CONNECTED)){
1151                 xpc_msgqueue_deref(ch);
1152                 return;
1153         }
1154
1155
1156         /*
1157          * First check to see if messages recently sent by us have been
1158          * received by the other side. (The remote GET value will have
1159          * changed since we last looked at it.)
1160          */
1161
1162         if (ch->w_remote_GP.get != ch->remote_GP.get) {
1163
1164                 /*
1165                  * We need to notify any senders that want to be notified
1166                  * that their sent messages have been received by their
1167                  * intended recipients. We need to do this before updating
1168                  * w_remote_GP.get so that we don't allocate the same message
1169                  * queue entries prematurely (see xpc_allocate_msg()).
1170                  */
1171                 if (atomic_read(&ch->n_to_notify) > 0) {
1172                         /*
1173                          * Notify senders that messages sent have been
1174                          * received and delivered by the other side.
1175                          */
1176                         xpc_notify_senders(ch, xpcMsgDelivered,
1177                                                         ch->remote_GP.get);
1178                 }
1179
1180                 /*
1181                  * Clear msg->flags in previously sent messages, so that
1182                  * they're ready for xpc_allocate_msg().
1183                  */
1184                 xpc_clear_local_msgqueue_flags(ch);
1185
1186                 ch->w_remote_GP.get = ch->remote_GP.get;
1187
1188                 dev_dbg(xpc_chan, "w_remote_GP.get changed to %ld, partid=%d, "
1189                         "channel=%d\n", ch->w_remote_GP.get, ch->partid,
1190                         ch->number);
1191
1192                 /*
1193                  * If anyone was waiting for message queue entries to become
1194                  * available, wake them up.
1195                  */
1196                 if (atomic_read(&ch->n_on_msg_allocate_wq) > 0) {
1197                         wake_up(&ch->msg_allocate_wq);
1198                 }
1199         }
1200
1201
1202         /*
1203          * Now check for newly sent messages by the other side. (The remote
1204          * PUT value will have changed since we last looked at it.)
1205          */
1206
1207         if (ch->w_remote_GP.put != ch->remote_GP.put) {
1208                 /*
1209                  * Clear msg->flags in previously received messages, so that
1210                  * they're ready for xpc_get_deliverable_msg().
1211                  */
1212                 xpc_clear_remote_msgqueue_flags(ch);
1213
1214                 ch->w_remote_GP.put = ch->remote_GP.put;
1215
1216                 dev_dbg(xpc_chan, "w_remote_GP.put changed to %ld, partid=%d, "
1217                         "channel=%d\n", ch->w_remote_GP.put, ch->partid,
1218                         ch->number);
1219
1220                 nmsgs_sent = ch->w_remote_GP.put - ch->w_local_GP.get;
1221                 if (nmsgs_sent > 0) {
1222                         dev_dbg(xpc_chan, "msgs waiting to be copied and "
1223                                 "delivered=%d, partid=%d, channel=%d\n",
1224                                 nmsgs_sent, ch->partid, ch->number);
1225
1226                         if (ch->flags & XPC_C_CONNECTCALLOUT) {
1227                                 xpc_activate_kthreads(ch, nmsgs_sent);
1228                         }
1229                 }
1230         }
1231
1232         xpc_msgqueue_deref(ch);
1233 }
1234
1235
1236 void
1237 xpc_process_channel_activity(struct xpc_partition *part)
1238 {
1239         unsigned long irq_flags;
1240         u64 IPI_amo, IPI_flags;
1241         struct xpc_channel *ch;
1242         int ch_number;
1243
1244
1245         IPI_amo = xpc_get_IPI_flags(part);
1246
1247         /*
1248          * Initiate channel connections for registered channels.
1249          *
1250          * For each connected channel that has pending messages activate idle
1251          * kthreads and/or create new kthreads as needed.
1252          */
1253
1254         for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
1255                 ch = &part->channels[ch_number];
1256
1257
1258                 /*
1259                  * Process any open or close related IPI flags, and then deal
1260                  * with connecting or disconnecting the channel as required.
1261                  */
1262
1263                 IPI_flags = XPC_GET_IPI_FLAGS(IPI_amo, ch_number);
1264
1265                 if (XPC_ANY_OPENCLOSE_IPI_FLAGS_SET(IPI_flags)) {
1266                         xpc_process_openclose_IPI(part, ch_number, IPI_flags);
1267                 }
1268
1269
1270                 if (ch->flags & XPC_C_DISCONNECTING) {
1271                         spin_lock_irqsave(&ch->lock, irq_flags);
1272                         xpc_process_disconnect(ch, &irq_flags);
1273                         spin_unlock_irqrestore(&ch->lock, irq_flags);
1274                         continue;
1275                 }
1276
1277                 if (part->act_state == XPC_P_DEACTIVATING) {
1278                         continue;
1279                 }
1280
1281                 if (!(ch->flags & XPC_C_CONNECTED)) {
1282                         if (!(ch->flags & XPC_C_OPENREQUEST)) {
1283                                 DBUG_ON(ch->flags & XPC_C_SETUP);
1284                                 (void) xpc_connect_channel(ch);
1285                         } else {
1286                                 spin_lock_irqsave(&ch->lock, irq_flags);
1287                                 xpc_process_connect(ch, &irq_flags);
1288                                 spin_unlock_irqrestore(&ch->lock, irq_flags);
1289                         }
1290                         continue;
1291                 }
1292
1293
1294                 /*
1295                  * Process any message related IPI flags, this may involve the
1296                  * activation of kthreads to deliver any pending messages sent
1297                  * from the other partition.
1298                  */
1299
1300                 if (XPC_ANY_MSG_IPI_FLAGS_SET(IPI_flags)) {
1301                         xpc_process_msg_IPI(part, ch_number);
1302                 }
1303         }
1304 }
1305
1306
1307 /*
1308  * XPC's heartbeat code calls this function to inform XPC that a partition has
1309  * gone down.  XPC responds by tearing down the XPartition Communication
1310  * infrastructure used for the just downed partition.
1311  *
1312  * XPC's heartbeat code will never call this function and xpc_partition_up()
1313  * at the same time. Nor will it ever make multiple calls to either function
1314  * at the same time.
1315  */
1316 void
1317 xpc_partition_down(struct xpc_partition *part, enum xpc_retval reason)
1318 {
1319         unsigned long irq_flags;
1320         int ch_number;
1321         struct xpc_channel *ch;
1322
1323
1324         dev_dbg(xpc_chan, "deactivating partition %d, reason=%d\n",
1325                 XPC_PARTID(part), reason);
1326
1327         if (!xpc_part_ref(part)) {
1328                 /* infrastructure for this partition isn't currently set up */
1329                 return;
1330         }
1331
1332
1333         /* disconnect all channels associated with the downed partition */
1334
1335         for (ch_number = 0; ch_number < part->nchannels; ch_number++) {
1336                 ch = &part->channels[ch_number];
1337
1338
1339                 xpc_msgqueue_ref(ch);
1340                 spin_lock_irqsave(&ch->lock, irq_flags);
1341
1342                 XPC_DISCONNECT_CHANNEL(ch, reason, &irq_flags);
1343
1344                 spin_unlock_irqrestore(&ch->lock, irq_flags);
1345                 xpc_msgqueue_deref(ch);
1346         }
1347
1348         xpc_wakeup_channel_mgr(part);
1349
1350         xpc_part_deref(part);
1351 }
1352
1353
1354 /*
1355  * Teardown the infrastructure necessary to support XPartition Communication
1356  * between the specified remote partition and the local one.
1357  */
1358 void
1359 xpc_teardown_infrastructure(struct xpc_partition *part)
1360 {
1361         partid_t partid = XPC_PARTID(part);
1362
1363
1364         /*
1365          * We start off by making this partition inaccessible to local
1366          * processes by marking it as no longer setup. Then we make it
1367          * inaccessible to remote processes by clearing the XPC per partition
1368          * specific variable's magic # (which indicates that these variables
1369          * are no longer valid) and by ignoring all XPC notify IPIs sent to
1370          * this partition.
1371          */
1372
1373         DBUG_ON(atomic_read(&part->nchannels_active) != 0);
1374         DBUG_ON(part->setup_state != XPC_P_SETUP);
1375         part->setup_state = XPC_P_WTEARDOWN;
1376
1377         xpc_vars_part[partid].magic = 0;
1378
1379
1380         free_irq(SGI_XPC_NOTIFY, (void *) (u64) partid);
1381
1382
1383         /*
1384          * Before proceding with the teardown we have to wait until all
1385          * existing references cease.
1386          */
1387         wait_event(part->teardown_wq, (atomic_read(&part->references) == 0));
1388
1389
1390         /* now we can begin tearing down the infrastructure */
1391
1392         part->setup_state = XPC_P_TORNDOWN;
1393
1394         /* in case we've still got outstanding timers registered... */
1395         del_timer_sync(&part->dropped_IPI_timer);
1396
1397         kfree(part->remote_openclose_args_base);
1398         part->remote_openclose_args = NULL;
1399         kfree(part->local_openclose_args_base);
1400         part->local_openclose_args = NULL;
1401         kfree(part->remote_GPs_base);
1402         part->remote_GPs = NULL;
1403         kfree(part->local_GPs_base);
1404         part->local_GPs = NULL;
1405         kfree(part->channels);
1406         part->channels = NULL;
1407         part->local_IPI_amo_va = NULL;
1408 }
1409
1410
1411 /*
1412  * Called by XP at the time of channel connection registration to cause
1413  * XPC to establish connections to all currently active partitions.
1414  */
1415 void
1416 xpc_initiate_connect(int ch_number)
1417 {
1418         partid_t partid;
1419         struct xpc_partition *part;
1420         struct xpc_channel *ch;
1421
1422
1423         DBUG_ON(ch_number < 0 || ch_number >= XPC_NCHANNELS);
1424
1425         for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
1426                 part = &xpc_partitions[partid];
1427
1428                 if (xpc_part_ref(part)) {
1429                         ch = &part->channels[ch_number];
1430
1431                         if (!(ch->flags & XPC_C_DISCONNECTING)) {
1432                                 DBUG_ON(ch->flags & XPC_C_OPENREQUEST);
1433                                 DBUG_ON(ch->flags & XPC_C_CONNECTED);
1434                                 DBUG_ON(ch->flags & XPC_C_SETUP);
1435
1436                                 /*
1437                                  * Initiate the establishment of a connection
1438                                  * on the newly registered channel to the
1439                                  * remote partition.
1440                                  */
1441                                 xpc_wakeup_channel_mgr(part);
1442                         }
1443
1444                         xpc_part_deref(part);
1445                 }
1446         }
1447 }
1448
1449
1450 void
1451 xpc_connected_callout(struct xpc_channel *ch)
1452 {
1453         unsigned long irq_flags;
1454
1455
1456         /* let the registerer know that a connection has been established */
1457
1458         if (ch->func != NULL) {
1459                 dev_dbg(xpc_chan, "ch->func() called, reason=xpcConnected, "
1460                         "partid=%d, channel=%d\n", ch->partid, ch->number);
1461
1462                 ch->func(xpcConnected, ch->partid, ch->number,
1463                                 (void *) (u64) ch->local_nentries, ch->key);
1464
1465                 dev_dbg(xpc_chan, "ch->func() returned, reason=xpcConnected, "
1466                         "partid=%d, channel=%d\n", ch->partid, ch->number);
1467         }
1468
1469         spin_lock_irqsave(&ch->lock, irq_flags);
1470         ch->flags |= XPC_C_CONNECTCALLOUT;
1471         spin_unlock_irqrestore(&ch->lock, irq_flags);
1472 }
1473
1474
1475 /*
1476  * Called by XP at the time of channel connection unregistration to cause
1477  * XPC to teardown all current connections for the specified channel.
1478  *
1479  * Before returning xpc_initiate_disconnect() will wait until all connections
1480  * on the specified channel have been closed/torndown. So the caller can be
1481  * assured that they will not be receiving any more callouts from XPC to the
1482  * function they registered via xpc_connect().
1483  *
1484  * Arguments:
1485  *
1486  *      ch_number - channel # to unregister.
1487  */
1488 void
1489 xpc_initiate_disconnect(int ch_number)
1490 {
1491         unsigned long irq_flags;
1492         partid_t partid;
1493         struct xpc_partition *part;
1494         struct xpc_channel *ch;
1495
1496
1497         DBUG_ON(ch_number < 0 || ch_number >= XPC_NCHANNELS);
1498
1499         /* initiate the channel disconnect for every active partition */
1500         for (partid = 1; partid < XP_MAX_PARTITIONS; partid++) {
1501                 part = &xpc_partitions[partid];
1502
1503                 if (xpc_part_ref(part)) {
1504                         ch = &part->channels[ch_number];
1505                         xpc_msgqueue_ref(ch);
1506
1507                         spin_lock_irqsave(&ch->lock, irq_flags);
1508
1509                         XPC_DISCONNECT_CHANNEL(ch, xpcUnregistering,
1510                                                                 &irq_flags);
1511
1512                         spin_unlock_irqrestore(&ch->lock, irq_flags);
1513
1514                         xpc_msgqueue_deref(ch);
1515                         xpc_part_deref(part);
1516                 }
1517         }
1518
1519         xpc_disconnect_wait(ch_number);
1520 }
1521
1522
1523 /*
1524  * To disconnect a channel, and reflect it back to all who may be waiting.
1525  *
1526  * >>> An OPEN is not allowed until XPC_C_DISCONNECTING is cleared by
1527  * >>> xpc_free_msgqueues().
1528  *
1529  * THE CHANNEL IS TO BE LOCKED BY THE CALLER AND WILL REMAIN LOCKED UPON RETURN.
1530  */
1531 void
1532 xpc_disconnect_channel(const int line, struct xpc_channel *ch,
1533                         enum xpc_retval reason, unsigned long *irq_flags)
1534 {
1535         u32 flags;
1536
1537
1538         DBUG_ON(!spin_is_locked(&ch->lock));
1539
1540         if (ch->flags & (XPC_C_DISCONNECTING | XPC_C_DISCONNECTED)) {
1541                 return;
1542         }
1543         DBUG_ON(!(ch->flags & (XPC_C_CONNECTING | XPC_C_CONNECTED)));
1544
1545         dev_dbg(xpc_chan, "reason=%d, line=%d, partid=%d, channel=%d\n",
1546                 reason, line, ch->partid, ch->number);
1547
1548         XPC_SET_REASON(ch, reason, line);
1549
1550         flags = ch->flags;
1551         /* some of these may not have been set */
1552         ch->flags &= ~(XPC_C_OPENREQUEST | XPC_C_OPENREPLY |
1553                         XPC_C_ROPENREQUEST | XPC_C_ROPENREPLY |
1554                         XPC_C_CONNECTING | XPC_C_CONNECTED);
1555
1556         ch->flags |= (XPC_C_CLOSEREQUEST | XPC_C_DISCONNECTING);
1557         xpc_IPI_send_closerequest(ch, irq_flags);
1558
1559         if (flags & XPC_C_CONNECTED) {
1560                 ch->flags |= XPC_C_WASCONNECTED;
1561         }
1562
1563         if (atomic_read(&ch->kthreads_idle) > 0) {
1564                 /* wake all idle kthreads so they can exit */
1565                 wake_up_all(&ch->idle_wq);
1566         }
1567
1568         spin_unlock_irqrestore(&ch->lock, *irq_flags);
1569
1570
1571         /* wake those waiting to allocate an entry from the local msg queue */
1572
1573         if (atomic_read(&ch->n_on_msg_allocate_wq) > 0) {
1574                 wake_up(&ch->msg_allocate_wq);
1575         }
1576
1577         /* wake those waiting for notify completion */
1578
1579         if (atomic_read(&ch->n_to_notify) > 0) {
1580                 xpc_notify_senders(ch, reason, ch->w_local_GP.put);
1581         }
1582
1583         spin_lock_irqsave(&ch->lock, *irq_flags);
1584 }
1585
1586
1587 void
1588 xpc_disconnected_callout(struct xpc_channel *ch)
1589 {
1590         /*
1591          * Let the channel's registerer know that the channel is now
1592          * disconnected. We don't want to do this if the registerer was never
1593          * informed of a connection being made, unless the disconnect was for
1594          * abnormal reasons.
1595          */
1596
1597         if (ch->func != NULL) {
1598                 dev_dbg(xpc_chan, "ch->func() called, reason=%d, partid=%d, "
1599                         "channel=%d\n", ch->reason, ch->partid, ch->number);
1600
1601                 ch->func(ch->reason, ch->partid, ch->number, NULL, ch->key);
1602
1603                 dev_dbg(xpc_chan, "ch->func() returned, reason=%d, partid=%d, "
1604                         "channel=%d\n", ch->reason, ch->partid, ch->number);
1605         }
1606 }
1607
1608
1609 /*
1610  * Wait for a message entry to become available for the specified channel,
1611  * but don't wait any longer than 1 jiffy.
1612  */
1613 static enum xpc_retval
1614 xpc_allocate_msg_wait(struct xpc_channel *ch)
1615 {
1616         enum xpc_retval ret;
1617
1618
1619         if (ch->flags & XPC_C_DISCONNECTING) {
1620                 DBUG_ON(ch->reason == xpcInterrupted);  // >>> Is this true?
1621                 return ch->reason;
1622         }
1623
1624         atomic_inc(&ch->n_on_msg_allocate_wq);
1625         ret = interruptible_sleep_on_timeout(&ch->msg_allocate_wq, 1);
1626         atomic_dec(&ch->n_on_msg_allocate_wq);
1627
1628         if (ch->flags & XPC_C_DISCONNECTING) {
1629                 ret = ch->reason;
1630                 DBUG_ON(ch->reason == xpcInterrupted);  // >>> Is this true?
1631         } else if (ret == 0) {
1632                 ret = xpcTimeout;
1633         } else {
1634                 ret = xpcInterrupted;
1635         }
1636
1637         return ret;
1638 }
1639
1640
1641 /*
1642  * Allocate an entry for a message from the message queue associated with the
1643  * specified channel.
1644  */
1645 static enum xpc_retval
1646 xpc_allocate_msg(struct xpc_channel *ch, u32 flags,
1647                         struct xpc_msg **address_of_msg)
1648 {
1649         struct xpc_msg *msg;
1650         enum xpc_retval ret;
1651         s64 put;
1652
1653
1654         /* this reference will be dropped in xpc_send_msg() */
1655         xpc_msgqueue_ref(ch);
1656
1657         if (ch->flags & XPC_C_DISCONNECTING) {
1658                 xpc_msgqueue_deref(ch);
1659                 return ch->reason;
1660         }
1661         if (!(ch->flags & XPC_C_CONNECTED)) {
1662                 xpc_msgqueue_deref(ch);
1663                 return xpcNotConnected;
1664         }
1665
1666
1667         /*
1668          * Get the next available message entry from the local message queue.
1669          * If none are available, we'll make sure that we grab the latest
1670          * GP values.
1671          */
1672         ret = xpcTimeout;
1673
1674         while (1) {
1675
1676                 put = (volatile s64) ch->w_local_GP.put;
1677                 if (put - (volatile s64) ch->w_remote_GP.get <
1678                                                         ch->local_nentries) {
1679
1680                         /* There are available message entries. We need to try
1681                          * to secure one for ourselves. We'll do this by trying
1682                          * to increment w_local_GP.put as long as someone else
1683                          * doesn't beat us to it. If they do, we'll have to
1684                          * try again.
1685                          */
1686                         if (cmpxchg(&ch->w_local_GP.put, put, put + 1) ==
1687                                                                         put) {
1688                                 /* we got the entry referenced by put */
1689                                 break;
1690                         }
1691                         continue;       /* try again */
1692                 }
1693
1694
1695                 /*
1696                  * There aren't any available msg entries at this time.
1697                  *
1698                  * In waiting for a message entry to become available,
1699                  * we set a timeout in case the other side is not
1700                  * sending completion IPIs. This lets us fake an IPI
1701                  * that will cause the IPI handler to fetch the latest
1702                  * GP values as if an IPI was sent by the other side.
1703                  */
1704                 if (ret == xpcTimeout) {
1705                         xpc_IPI_send_local_msgrequest(ch);
1706                 }
1707
1708                 if (flags & XPC_NOWAIT) {
1709                         xpc_msgqueue_deref(ch);
1710                         return xpcNoWait;
1711                 }
1712
1713                 ret = xpc_allocate_msg_wait(ch);
1714                 if (ret != xpcInterrupted && ret != xpcTimeout) {
1715                         xpc_msgqueue_deref(ch);
1716                         return ret;
1717                 }
1718         }
1719
1720
1721         /* get the message's address and initialize it */
1722         msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
1723                                 (put % ch->local_nentries) * ch->msg_size);
1724
1725
1726         DBUG_ON(msg->flags != 0);
1727         msg->number = put;
1728
1729         dev_dbg(xpc_chan, "w_local_GP.put changed to %ld; msg=0x%p, "
1730                 "msg_number=%ld, partid=%d, channel=%d\n", put + 1,
1731                 (void *) msg, msg->number, ch->partid, ch->number);
1732
1733         *address_of_msg = msg;
1734
1735         return xpcSuccess;
1736 }
1737
1738
1739 /*
1740  * Allocate an entry for a message from the message queue associated with the
1741  * specified channel. NOTE that this routine can sleep waiting for a message
1742  * entry to become available. To not sleep, pass in the XPC_NOWAIT flag.
1743  *
1744  * Arguments:
1745  *
1746  *      partid - ID of partition to which the channel is connected.
1747  *      ch_number - channel #.
1748  *      flags - see xpc.h for valid flags.
1749  *      payload - address of the allocated payload area pointer (filled in on
1750  *                return) in which the user-defined message is constructed.
1751  */
1752 enum xpc_retval
1753 xpc_initiate_allocate(partid_t partid, int ch_number, u32 flags, void **payload)
1754 {
1755         struct xpc_partition *part = &xpc_partitions[partid];
1756         enum xpc_retval ret = xpcUnknownReason;
1757         struct xpc_msg *msg;
1758
1759
1760         DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
1761         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
1762
1763         *payload = NULL;
1764
1765         if (xpc_part_ref(part)) {
1766                 ret = xpc_allocate_msg(&part->channels[ch_number], flags, &msg);
1767                 xpc_part_deref(part);
1768
1769                 if (msg != NULL) {
1770                         *payload = &msg->payload;
1771                 }
1772         }
1773
1774         return ret;
1775 }
1776
1777
1778 /*
1779  * Now we actually send the messages that are ready to be sent by advancing
1780  * the local message queue's Put value and then send an IPI to the recipient
1781  * partition.
1782  */
1783 static void
1784 xpc_send_msgs(struct xpc_channel *ch, s64 initial_put)
1785 {
1786         struct xpc_msg *msg;
1787         s64 put = initial_put + 1;
1788         int send_IPI = 0;
1789
1790
1791         while (1) {
1792
1793                 while (1) {
1794                         if (put == (volatile s64) ch->w_local_GP.put) {
1795                                 break;
1796                         }
1797
1798                         msg = (struct xpc_msg *) ((u64) ch->local_msgqueue +
1799                                (put % ch->local_nentries) * ch->msg_size);
1800
1801                         if (!(msg->flags & XPC_M_READY)) {
1802                                 break;
1803                         }
1804
1805                         put++;
1806                 }
1807
1808                 if (put == initial_put) {
1809                         /* nothing's changed */
1810                         break;
1811                 }
1812
1813                 if (cmpxchg_rel(&ch->local_GP->put, initial_put, put) !=
1814                                                                 initial_put) {
1815                         /* someone else beat us to it */
1816                         DBUG_ON((volatile s64) ch->local_GP->put < initial_put);
1817                         break;
1818                 }
1819
1820                 /* we just set the new value of local_GP->put */
1821
1822                 dev_dbg(xpc_chan, "local_GP->put changed to %ld, partid=%d, "
1823                         "channel=%d\n", put, ch->partid, ch->number);
1824
1825                 send_IPI = 1;
1826
1827                 /*
1828                  * We need to ensure that the message referenced by
1829                  * local_GP->put is not XPC_M_READY or that local_GP->put
1830                  * equals w_local_GP.put, so we'll go have a look.
1831                  */
1832                 initial_put = put;
1833         }
1834
1835         if (send_IPI) {
1836                 xpc_IPI_send_msgrequest(ch);
1837         }
1838 }
1839
1840
1841 /*
1842  * Common code that does the actual sending of the message by advancing the
1843  * local message queue's Put value and sends an IPI to the partition the
1844  * message is being sent to.
1845  */
1846 static enum xpc_retval
1847 xpc_send_msg(struct xpc_channel *ch, struct xpc_msg *msg, u8 notify_type,
1848                         xpc_notify_func func, void *key)
1849 {
1850         enum xpc_retval ret = xpcSuccess;
1851         struct xpc_notify *notify = NULL;   // >>> to keep the compiler happy!!
1852         s64 put, msg_number = msg->number;
1853
1854
1855         DBUG_ON(notify_type == XPC_N_CALL && func == NULL);
1856         DBUG_ON((((u64) msg - (u64) ch->local_msgqueue) / ch->msg_size) !=
1857                                         msg_number % ch->local_nentries);
1858         DBUG_ON(msg->flags & XPC_M_READY);
1859
1860         if (ch->flags & XPC_C_DISCONNECTING) {
1861                 /* drop the reference grabbed in xpc_allocate_msg() */
1862                 xpc_msgqueue_deref(ch);
1863                 return ch->reason;
1864         }
1865
1866         if (notify_type != 0) {
1867                 /*
1868                  * Tell the remote side to send an ACK interrupt when the
1869                  * message has been delivered.
1870                  */
1871                 msg->flags |= XPC_M_INTERRUPT;
1872
1873                 atomic_inc(&ch->n_to_notify);
1874
1875                 notify = &ch->notify_queue[msg_number % ch->local_nentries];
1876                 notify->func = func;
1877                 notify->key = key;
1878                 notify->type = notify_type;
1879
1880                 // >>> is a mb() needed here?
1881
1882                 if (ch->flags & XPC_C_DISCONNECTING) {
1883                         /*
1884                          * An error occurred between our last error check and
1885                          * this one. We will try to clear the type field from
1886                          * the notify entry. If we succeed then
1887                          * xpc_disconnect_channel() didn't already process
1888                          * the notify entry.
1889                          */
1890                         if (cmpxchg(&notify->type, notify_type, 0) ==
1891                                                                 notify_type) {
1892                                 atomic_dec(&ch->n_to_notify);
1893                                 ret = ch->reason;
1894                         }
1895
1896                         /* drop the reference grabbed in xpc_allocate_msg() */
1897                         xpc_msgqueue_deref(ch);
1898                         return ret;
1899                 }
1900         }
1901
1902         msg->flags |= XPC_M_READY;
1903
1904         /*
1905          * The preceding store of msg->flags must occur before the following
1906          * load of ch->local_GP->put.
1907          */
1908         mb();
1909
1910         /* see if the message is next in line to be sent, if so send it */
1911
1912         put = ch->local_GP->put;
1913         if (put == msg_number) {
1914                 xpc_send_msgs(ch, put);
1915         }
1916
1917         /* drop the reference grabbed in xpc_allocate_msg() */
1918         xpc_msgqueue_deref(ch);
1919         return ret;
1920 }
1921
1922
1923 /*
1924  * Send a message previously allocated using xpc_initiate_allocate() on the
1925  * specified channel connected to the specified partition.
1926  *
1927  * This routine will not wait for the message to be received, nor will
1928  * notification be given when it does happen. Once this routine has returned
1929  * the message entry allocated via xpc_initiate_allocate() is no longer
1930  * accessable to the caller.
1931  *
1932  * This routine, although called by users, does not call xpc_part_ref() to
1933  * ensure that the partition infrastructure is in place. It relies on the
1934  * fact that we called xpc_msgqueue_ref() in xpc_allocate_msg().
1935  *
1936  * Arguments:
1937  *
1938  *      partid - ID of partition to which the channel is connected.
1939  *      ch_number - channel # to send message on.
1940  *      payload - pointer to the payload area allocated via
1941  *                      xpc_initiate_allocate().
1942  */
1943 enum xpc_retval
1944 xpc_initiate_send(partid_t partid, int ch_number, void *payload)
1945 {
1946         struct xpc_partition *part = &xpc_partitions[partid];
1947         struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
1948         enum xpc_retval ret;
1949
1950
1951         dev_dbg(xpc_chan, "msg=0x%p, partid=%d, channel=%d\n", (void *) msg,
1952                 partid, ch_number);
1953
1954         DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
1955         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
1956         DBUG_ON(msg == NULL);
1957
1958         ret = xpc_send_msg(&part->channels[ch_number], msg, 0, NULL, NULL);
1959
1960         return ret;
1961 }
1962
1963
1964 /*
1965  * Send a message previously allocated using xpc_initiate_allocate on the
1966  * specified channel connected to the specified partition.
1967  *
1968  * This routine will not wait for the message to be sent. Once this routine
1969  * has returned the message entry allocated via xpc_initiate_allocate() is no
1970  * longer accessable to the caller.
1971  *
1972  * Once the remote end of the channel has received the message, the function
1973  * passed as an argument to xpc_initiate_send_notify() will be called. This
1974  * allows the sender to free up or re-use any buffers referenced by the
1975  * message, but does NOT mean the message has been processed at the remote
1976  * end by a receiver.
1977  *
1978  * If this routine returns an error, the caller's function will NOT be called.
1979  *
1980  * This routine, although called by users, does not call xpc_part_ref() to
1981  * ensure that the partition infrastructure is in place. It relies on the
1982  * fact that we called xpc_msgqueue_ref() in xpc_allocate_msg().
1983  *
1984  * Arguments:
1985  *
1986  *      partid - ID of partition to which the channel is connected.
1987  *      ch_number - channel # to send message on.
1988  *      payload - pointer to the payload area allocated via
1989  *                      xpc_initiate_allocate().
1990  *      func - function to call with asynchronous notification of message
1991  *                receipt. THIS FUNCTION MUST BE NON-BLOCKING.
1992  *      key - user-defined key to be passed to the function when it's called.
1993  */
1994 enum xpc_retval
1995 xpc_initiate_send_notify(partid_t partid, int ch_number, void *payload,
1996                                 xpc_notify_func func, void *key)
1997 {
1998         struct xpc_partition *part = &xpc_partitions[partid];
1999         struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
2000         enum xpc_retval ret;
2001
2002
2003         dev_dbg(xpc_chan, "msg=0x%p, partid=%d, channel=%d\n", (void *) msg,
2004                 partid, ch_number);
2005
2006         DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
2007         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
2008         DBUG_ON(msg == NULL);
2009         DBUG_ON(func == NULL);
2010
2011         ret = xpc_send_msg(&part->channels[ch_number], msg, XPC_N_CALL,
2012                                                                 func, key);
2013         return ret;
2014 }
2015
2016
2017 static struct xpc_msg *
2018 xpc_pull_remote_msg(struct xpc_channel *ch, s64 get)
2019 {
2020         struct xpc_partition *part = &xpc_partitions[ch->partid];
2021         struct xpc_msg *remote_msg, *msg;
2022         u32 msg_index, nmsgs;
2023         u64 msg_offset;
2024         enum xpc_retval ret;
2025
2026
2027         if (down_interruptible(&ch->msg_to_pull_sema) != 0) {
2028                 /* we were interrupted by a signal */
2029                 return NULL;
2030         }
2031
2032         while (get >= ch->next_msg_to_pull) {
2033
2034                 /* pull as many messages as are ready and able to be pulled */
2035
2036                 msg_index = ch->next_msg_to_pull % ch->remote_nentries;
2037
2038                 DBUG_ON(ch->next_msg_to_pull >=
2039                                         (volatile s64) ch->w_remote_GP.put);
2040                 nmsgs =  (volatile s64) ch->w_remote_GP.put -
2041                                                 ch->next_msg_to_pull;
2042                 if (msg_index + nmsgs > ch->remote_nentries) {
2043                         /* ignore the ones that wrap the msg queue for now */
2044                         nmsgs = ch->remote_nentries - msg_index;
2045                 }
2046
2047                 msg_offset = msg_index * ch->msg_size;
2048                 msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
2049                                                                 msg_offset);
2050                 remote_msg = (struct xpc_msg *) (ch->remote_msgqueue_pa +
2051                                                                 msg_offset);
2052
2053                 if ((ret = xpc_pull_remote_cachelines(part, msg, remote_msg,
2054                                 nmsgs * ch->msg_size)) != xpcSuccess) {
2055
2056                         dev_dbg(xpc_chan, "failed to pull %d msgs starting with"
2057                                 " msg %ld from partition %d, channel=%d, "
2058                                 "ret=%d\n", nmsgs, ch->next_msg_to_pull,
2059                                 ch->partid, ch->number, ret);
2060
2061                         XPC_DEACTIVATE_PARTITION(part, ret);
2062
2063                         up(&ch->msg_to_pull_sema);
2064                         return NULL;
2065                 }
2066
2067                 mb();   /* >>> this may not be needed, we're not sure */
2068
2069                 ch->next_msg_to_pull += nmsgs;
2070         }
2071
2072         up(&ch->msg_to_pull_sema);
2073
2074         /* return the message we were looking for */
2075         msg_offset = (get % ch->remote_nentries) * ch->msg_size;
2076         msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue + msg_offset);
2077
2078         return msg;
2079 }
2080
2081
2082 /*
2083  * Get a message to be delivered.
2084  */
2085 static struct xpc_msg *
2086 xpc_get_deliverable_msg(struct xpc_channel *ch)
2087 {
2088         struct xpc_msg *msg = NULL;
2089         s64 get;
2090
2091
2092         do {
2093                 if ((volatile u32) ch->flags & XPC_C_DISCONNECTING) {
2094                         break;
2095                 }
2096
2097                 get = (volatile s64) ch->w_local_GP.get;
2098                 if (get == (volatile s64) ch->w_remote_GP.put) {
2099                         break;
2100                 }
2101
2102                 /* There are messages waiting to be pulled and delivered.
2103                  * We need to try to secure one for ourselves. We'll do this
2104                  * by trying to increment w_local_GP.get and hope that no one
2105                  * else beats us to it. If they do, we'll we'll simply have
2106                  * to try again for the next one.
2107                  */
2108
2109                 if (cmpxchg(&ch->w_local_GP.get, get, get + 1) == get) {
2110                         /* we got the entry referenced by get */
2111
2112                         dev_dbg(xpc_chan, "w_local_GP.get changed to %ld, "
2113                                 "partid=%d, channel=%d\n", get + 1,
2114                                 ch->partid, ch->number);
2115
2116                         /* pull the message from the remote partition */
2117
2118                         msg = xpc_pull_remote_msg(ch, get);
2119
2120                         DBUG_ON(msg != NULL && msg->number != get);
2121                         DBUG_ON(msg != NULL && (msg->flags & XPC_M_DONE));
2122                         DBUG_ON(msg != NULL && !(msg->flags & XPC_M_READY));
2123
2124                         break;
2125                 }
2126
2127         } while (1);
2128
2129         return msg;
2130 }
2131
2132
2133 /*
2134  * Deliver a message to its intended recipient.
2135  */
2136 void
2137 xpc_deliver_msg(struct xpc_channel *ch)
2138 {
2139         struct xpc_msg *msg;
2140
2141
2142         if ((msg = xpc_get_deliverable_msg(ch)) != NULL) {
2143
2144                 /*
2145                  * This ref is taken to protect the payload itself from being
2146                  * freed before the user is finished with it, which the user
2147                  * indicates by calling xpc_initiate_received().
2148                  */
2149                 xpc_msgqueue_ref(ch);
2150
2151                 atomic_inc(&ch->kthreads_active);
2152
2153                 if (ch->func != NULL) {
2154                         dev_dbg(xpc_chan, "ch->func() called, msg=0x%p, "
2155                                 "msg_number=%ld, partid=%d, channel=%d\n",
2156                                 (void *) msg, msg->number, ch->partid,
2157                                 ch->number);
2158
2159                         /* deliver the message to its intended recipient */
2160                         ch->func(xpcMsgReceived, ch->partid, ch->number,
2161                                         &msg->payload, ch->key);
2162
2163                         dev_dbg(xpc_chan, "ch->func() returned, msg=0x%p, "
2164                                 "msg_number=%ld, partid=%d, channel=%d\n",
2165                                 (void *) msg, msg->number, ch->partid,
2166                                 ch->number);
2167                 }
2168
2169                 atomic_dec(&ch->kthreads_active);
2170         }
2171 }
2172
2173
2174 /*
2175  * Now we actually acknowledge the messages that have been delivered and ack'd
2176  * by advancing the cached remote message queue's Get value and if requested
2177  * send an IPI to the message sender's partition.
2178  */
2179 static void
2180 xpc_acknowledge_msgs(struct xpc_channel *ch, s64 initial_get, u8 msg_flags)
2181 {
2182         struct xpc_msg *msg;
2183         s64 get = initial_get + 1;
2184         int send_IPI = 0;
2185
2186
2187         while (1) {
2188
2189                 while (1) {
2190                         if (get == (volatile s64) ch->w_local_GP.get) {
2191                                 break;
2192                         }
2193
2194                         msg = (struct xpc_msg *) ((u64) ch->remote_msgqueue +
2195                                (get % ch->remote_nentries) * ch->msg_size);
2196
2197                         if (!(msg->flags & XPC_M_DONE)) {
2198                                 break;
2199                         }
2200
2201                         msg_flags |= msg->flags;
2202                         get++;
2203                 }
2204
2205                 if (get == initial_get) {
2206                         /* nothing's changed */
2207                         break;
2208                 }
2209
2210                 if (cmpxchg_rel(&ch->local_GP->get, initial_get, get) !=
2211                                                                 initial_get) {
2212                         /* someone else beat us to it */
2213                         DBUG_ON((volatile s64) ch->local_GP->get <=
2214                                                                 initial_get);
2215                         break;
2216                 }
2217
2218                 /* we just set the new value of local_GP->get */
2219
2220                 dev_dbg(xpc_chan, "local_GP->get changed to %ld, partid=%d, "
2221                         "channel=%d\n", get, ch->partid, ch->number);
2222
2223                 send_IPI = (msg_flags & XPC_M_INTERRUPT);
2224
2225                 /*
2226                  * We need to ensure that the message referenced by
2227                  * local_GP->get is not XPC_M_DONE or that local_GP->get
2228                  * equals w_local_GP.get, so we'll go have a look.
2229                  */
2230                 initial_get = get;
2231         }
2232
2233         if (send_IPI) {
2234                 xpc_IPI_send_msgrequest(ch);
2235         }
2236 }
2237
2238
2239 /*
2240  * Acknowledge receipt of a delivered message.
2241  *
2242  * If a message has XPC_M_INTERRUPT set, send an interrupt to the partition
2243  * that sent the message.
2244  *
2245  * This function, although called by users, does not call xpc_part_ref() to
2246  * ensure that the partition infrastructure is in place. It relies on the
2247  * fact that we called xpc_msgqueue_ref() in xpc_deliver_msg().
2248  *
2249  * Arguments:
2250  *
2251  *      partid - ID of partition to which the channel is connected.
2252  *      ch_number - channel # message received on.
2253  *      payload - pointer to the payload area allocated via
2254  *                      xpc_initiate_allocate().
2255  */
2256 void
2257 xpc_initiate_received(partid_t partid, int ch_number, void *payload)
2258 {
2259         struct xpc_partition *part = &xpc_partitions[partid];
2260         struct xpc_channel *ch;
2261         struct xpc_msg *msg = XPC_MSG_ADDRESS(payload);
2262         s64 get, msg_number = msg->number;
2263
2264
2265         DBUG_ON(partid <= 0 || partid >= XP_MAX_PARTITIONS);
2266         DBUG_ON(ch_number < 0 || ch_number >= part->nchannels);
2267
2268         ch = &part->channels[ch_number];
2269
2270         dev_dbg(xpc_chan, "msg=0x%p, msg_number=%ld, partid=%d, channel=%d\n",
2271                 (void *) msg, msg_number, ch->partid, ch->number);
2272
2273         DBUG_ON((((u64) msg - (u64) ch->remote_msgqueue) / ch->msg_size) !=
2274                                         msg_number % ch->remote_nentries);
2275         DBUG_ON(msg->flags & XPC_M_DONE);
2276
2277         msg->flags |= XPC_M_DONE;
2278
2279         /*
2280          * The preceding store of msg->flags must occur before the following
2281          * load of ch->local_GP->get.
2282          */
2283         mb();
2284
2285         /*
2286          * See if this message is next in line to be acknowledged as having
2287          * been delivered.
2288          */
2289         get = ch->local_GP->get;
2290         if (get == msg_number) {
2291                 xpc_acknowledge_msgs(ch, get, msg->flags);
2292         }
2293
2294         /* the call to xpc_msgqueue_ref() was done by xpc_deliver_msg()  */
2295         xpc_msgqueue_deref(ch);
2296 }
2297