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
6 * Copyright (c) 2004-2008 Silicon Graphics, Inc. All Rights Reserved.
10 * Cross Partition Communication (XPC) support - standard version.
12 * XPC provides a message passing capability that crosses partition
13 * boundaries. This module is made up of two parts:
15 * partition This part detects the presence/absence of other
16 * partitions. It provides a heartbeat and monitors
17 * the heartbeats of other partitions.
19 * channel This part manages the channels and sends/receives
20 * messages across them to/from other partitions.
22 * There are a couple of additional functions residing in XP, which
23 * provide an interface to XPC for its users.
28 * . We currently have no way to determine which nasid an IPI came
29 * from. Thus, xpc_IPI_send() does a remote AMO write followed by
30 * an IPI. The AMO indicates where data is to be pulled from, so
31 * after the IPI arrives, the remote partition checks the AMO word.
32 * The IPI can actually arrive before the AMO however, so other code
33 * must periodically check for this case. Also, remote AMO operations
34 * do not reliably time out. Thus we do a remote PIO read solely to
35 * know whether the remote partition is down and whether we should
36 * stop sending IPIs to it. This remote PIO read operation is set up
37 * in a special nofault region so SAL knows to ignore (and cleanup)
38 * any errors due to the remote AMO write, PIO read, and/or PIO
41 * If/when new hardware solves this IPI problem, we should abandon
42 * the current approach.
46 #include <linux/kernel.h>
47 #include <linux/module.h>
48 #include <linux/init.h>
49 #include <linux/cache.h>
50 #include <linux/interrupt.h>
51 #include <linux/delay.h>
52 #include <linux/reboot.h>
53 #include <linux/completion.h>
54 #include <linux/kdebug.h>
55 #include <linux/kthread.h>
56 #include <linux/uaccess.h>
57 #include <asm/sn/intr.h>
58 #include <asm/sn/sn_sal.h>
61 /* define two XPC debug device structures to be used with dev_dbg() et al */
63 struct device_driver xpc_dbg_name = {
67 struct device xpc_part_dbg_subname = {
68 .bus_id = {0}, /* set to "part" at xpc_init() time */
69 .driver = &xpc_dbg_name
72 struct device xpc_chan_dbg_subname = {
73 .bus_id = {0}, /* set to "chan" at xpc_init() time */
74 .driver = &xpc_dbg_name
77 struct device *xpc_part = &xpc_part_dbg_subname;
78 struct device *xpc_chan = &xpc_chan_dbg_subname;
80 static int xpc_kdebug_ignore;
82 /* systune related variables for /proc/sys directories */
84 static int xpc_hb_interval = XPC_HB_DEFAULT_INTERVAL;
85 static int xpc_hb_min_interval = 1;
86 static int xpc_hb_max_interval = 10;
88 static int xpc_hb_check_interval = XPC_HB_CHECK_DEFAULT_INTERVAL;
89 static int xpc_hb_check_min_interval = 10;
90 static int xpc_hb_check_max_interval = 120;
92 int xpc_disengage_request_timelimit = XPC_DISENGAGE_REQUEST_DEFAULT_TIMELIMIT;
93 static int xpc_disengage_request_min_timelimit; /* = 0 */
94 static int xpc_disengage_request_max_timelimit = 120;
96 static ctl_table xpc_sys_xpc_hb_dir[] = {
98 .ctl_name = CTL_UNNUMBERED,
99 .procname = "hb_interval",
100 .data = &xpc_hb_interval,
101 .maxlen = sizeof(int),
103 .proc_handler = &proc_dointvec_minmax,
104 .strategy = &sysctl_intvec,
105 .extra1 = &xpc_hb_min_interval,
106 .extra2 = &xpc_hb_max_interval},
108 .ctl_name = CTL_UNNUMBERED,
109 .procname = "hb_check_interval",
110 .data = &xpc_hb_check_interval,
111 .maxlen = sizeof(int),
113 .proc_handler = &proc_dointvec_minmax,
114 .strategy = &sysctl_intvec,
115 .extra1 = &xpc_hb_check_min_interval,
116 .extra2 = &xpc_hb_check_max_interval},
119 static ctl_table xpc_sys_xpc_dir[] = {
121 .ctl_name = CTL_UNNUMBERED,
124 .child = xpc_sys_xpc_hb_dir},
126 .ctl_name = CTL_UNNUMBERED,
127 .procname = "disengage_request_timelimit",
128 .data = &xpc_disengage_request_timelimit,
129 .maxlen = sizeof(int),
131 .proc_handler = &proc_dointvec_minmax,
132 .strategy = &sysctl_intvec,
133 .extra1 = &xpc_disengage_request_min_timelimit,
134 .extra2 = &xpc_disengage_request_max_timelimit},
137 static ctl_table xpc_sys_dir[] = {
139 .ctl_name = CTL_UNNUMBERED,
142 .child = xpc_sys_xpc_dir},
145 static struct ctl_table_header *xpc_sysctl;
147 /* non-zero if any remote partition disengage request was timed out */
148 int xpc_disengage_request_timedout;
150 /* #of activate IRQs received */
151 atomic_t xpc_activate_IRQ_rcvd = ATOMIC_INIT(0);
153 /* IRQ handler notifies this wait queue on receipt of an IRQ */
154 DECLARE_WAIT_QUEUE_HEAD(xpc_activate_IRQ_wq);
156 static unsigned long xpc_hb_check_timeout;
157 static struct timer_list xpc_hb_timer;
158 void *xpc_heartbeating_to_mask;
160 /* notification that the xpc_hb_checker thread has exited */
161 static DECLARE_COMPLETION(xpc_hb_checker_exited);
163 /* notification that the xpc_discovery thread has exited */
164 static DECLARE_COMPLETION(xpc_discovery_exited);
166 static void xpc_kthread_waitmsgs(struct xpc_partition *, struct xpc_channel *);
168 static int xpc_system_reboot(struct notifier_block *, unsigned long, void *);
169 static struct notifier_block xpc_reboot_notifier = {
170 .notifier_call = xpc_system_reboot,
173 static int xpc_system_die(struct notifier_block *, unsigned long, void *);
174 static struct notifier_block xpc_die_notifier = {
175 .notifier_call = xpc_system_die,
178 enum xp_retval (*xpc_rsvd_page_init) (struct xpc_rsvd_page *rp);
179 void (*xpc_heartbeat_init) (void);
180 void (*xpc_heartbeat_exit) (void);
181 void (*xpc_increment_heartbeat) (void);
182 void (*xpc_offline_heartbeat) (void);
183 void (*xpc_online_heartbeat) (void);
184 void (*xpc_check_remote_hb) (void);
186 enum xp_retval (*xpc_make_first_contact) (struct xpc_partition *part);
187 u64 (*xpc_get_IPI_flags) (struct xpc_partition *part);
188 struct xpc_msg *(*xpc_get_deliverable_msg) (struct xpc_channel *ch);
190 void (*xpc_initiate_partition_activation) (struct xpc_rsvd_page *remote_rp,
191 u64 remote_rp_pa, int nasid);
193 void (*xpc_process_activate_IRQ_rcvd) (int n_IRQs_expected);
194 enum xp_retval (*xpc_setup_infrastructure) (struct xpc_partition *part);
195 void (*xpc_teardown_infrastructure) (struct xpc_partition *part);
197 void (*xpc_mark_partition_engaged) (struct xpc_partition *part);
198 void (*xpc_mark_partition_disengaged) (struct xpc_partition *part);
199 void (*xpc_request_partition_disengage) (struct xpc_partition *part);
200 void (*xpc_cancel_partition_disengage_request) (struct xpc_partition *part);
201 u64 (*xpc_partition_engaged) (u64 partid_mask);
202 u64 (*xpc_partition_disengage_requested) (u64 partid_mask);
203 void (*xpc_clear_partition_engaged) (u64 partid_mask);
204 void (*xpc_clear_partition_disengage_request) (u64 partid_mask);
206 void (*xpc_IPI_send_local_activate) (int from_nasid);
207 void (*xpc_IPI_send_activated) (struct xpc_partition *part);
208 void (*xpc_IPI_send_local_reactivate) (int from_nasid);
209 void (*xpc_IPI_send_disengage) (struct xpc_partition *part);
211 void (*xpc_IPI_send_closerequest) (struct xpc_channel *ch,
212 unsigned long *irq_flags);
213 void (*xpc_IPI_send_closereply) (struct xpc_channel *ch,
214 unsigned long *irq_flags);
215 void (*xpc_IPI_send_openrequest) (struct xpc_channel *ch,
216 unsigned long *irq_flags);
217 void (*xpc_IPI_send_openreply) (struct xpc_channel *ch,
218 unsigned long *irq_flags);
220 enum xp_retval (*xpc_send_msg) (struct xpc_channel *ch, u32 flags,
221 void *payload, u16 payload_size, u8 notify_type,
222 xpc_notify_func func, void *key);
223 void (*xpc_received_msg) (struct xpc_channel *ch, struct xpc_msg *msg);
226 * Timer function to enforce the timelimit on the partition disengage request.
229 xpc_timeout_partition_disengage_request(unsigned long data)
231 struct xpc_partition *part = (struct xpc_partition *)data;
233 DBUG_ON(time_is_after_jiffies(part->disengage_request_timeout));
235 (void)xpc_partition_disengaged(part);
237 DBUG_ON(part->disengage_request_timeout != 0);
238 DBUG_ON(xpc_partition_engaged(1UL << XPC_PARTID(part)) != 0);
242 * Timer to produce the heartbeat. The timer structures function is
243 * already set when this is initially called. A tunable is used to
244 * specify when the next timeout should occur.
247 xpc_hb_beater(unsigned long dummy)
249 xpc_increment_heartbeat();
251 if (time_is_before_eq_jiffies(xpc_hb_check_timeout))
252 wake_up_interruptible(&xpc_activate_IRQ_wq);
254 xpc_hb_timer.expires = jiffies + (xpc_hb_interval * HZ);
255 add_timer(&xpc_hb_timer);
259 xpc_start_hb_beater(void)
261 xpc_heartbeat_init();
262 init_timer(&xpc_hb_timer);
263 xpc_hb_timer.function = xpc_hb_beater;
268 xpc_stop_hb_beater(void)
270 del_timer_sync(&xpc_hb_timer);
271 xpc_heartbeat_exit();
275 * This thread is responsible for nearly all of the partition
276 * activation/deactivation.
279 xpc_hb_checker(void *ignore)
281 int last_IRQ_count = 0;
285 /* this thread was marked active by xpc_hb_init() */
287 set_cpus_allowed_ptr(current, &cpumask_of_cpu(XPC_HB_CHECK_CPU));
289 /* set our heartbeating to other partitions into motion */
290 xpc_hb_check_timeout = jiffies + (xpc_hb_check_interval * HZ);
291 xpc_start_hb_beater();
293 while (!xpc_exiting) {
295 dev_dbg(xpc_part, "woke up with %d ticks rem; %d IRQs have "
297 (int)(xpc_hb_check_timeout - jiffies),
298 atomic_read(&xpc_activate_IRQ_rcvd) - last_IRQ_count);
300 /* checking of remote heartbeats is skewed by IRQ handling */
301 if (time_is_before_eq_jiffies(xpc_hb_check_timeout)) {
302 dev_dbg(xpc_part, "checking remote heartbeats\n");
303 xpc_check_remote_hb();
306 * We need to periodically recheck to ensure no
307 * IPI/AMO pairs have been missed. That check
308 * must always reset xpc_hb_check_timeout.
313 /* check for outstanding IRQs */
314 new_IRQ_count = atomic_read(&xpc_activate_IRQ_rcvd);
315 if (last_IRQ_count < new_IRQ_count || force_IRQ != 0) {
318 dev_dbg(xpc_part, "found an IRQ to process; will be "
319 "resetting xpc_hb_check_timeout\n");
321 xpc_process_activate_IRQ_rcvd(new_IRQ_count -
323 last_IRQ_count = new_IRQ_count;
325 xpc_hb_check_timeout = jiffies +
326 (xpc_hb_check_interval * HZ);
329 /* wait for IRQ or timeout */
330 (void)wait_event_interruptible(xpc_activate_IRQ_wq,
331 (last_IRQ_count < atomic_read(
332 &xpc_activate_IRQ_rcvd)
333 || time_is_before_eq_jiffies(
334 xpc_hb_check_timeout) ||
338 xpc_stop_hb_beater();
340 dev_dbg(xpc_part, "heartbeat checker is exiting\n");
342 /* mark this thread as having exited */
343 complete(&xpc_hb_checker_exited);
348 * This thread will attempt to discover other partitions to activate
349 * based on info provided by SAL. This new thread is short lived and
350 * will exit once discovery is complete.
353 xpc_initiate_discovery(void *ignore)
357 dev_dbg(xpc_part, "discovery thread is exiting\n");
359 /* mark this thread as having exited */
360 complete(&xpc_discovery_exited);
365 * The first kthread assigned to a newly activated partition is the one
366 * created by XPC HB with which it calls xpc_activating(). XPC hangs on to
367 * that kthread until the partition is brought down, at which time that kthread
368 * returns back to XPC HB. (The return of that kthread will signify to XPC HB
369 * that XPC has dismantled all communication infrastructure for the associated
370 * partition.) This kthread becomes the channel manager for that partition.
372 * Each active partition has a channel manager, who, besides connecting and
373 * disconnecting channels, will ensure that each of the partition's connected
374 * channels has the required number of assigned kthreads to get the work done.
377 xpc_channel_mgr(struct xpc_partition *part)
379 while (part->act_state != XPC_P_DEACTIVATING ||
380 atomic_read(&part->nchannels_active) > 0 ||
381 !xpc_partition_disengaged(part)) {
383 xpc_process_channel_activity(part);
386 * Wait until we've been requested to activate kthreads or
387 * all of the channel's message queues have been torn down or
388 * a signal is pending.
390 * The channel_mgr_requests is set to 1 after being awakened,
391 * This is done to prevent the channel mgr from making one pass
392 * through the loop for each request, since he will
393 * be servicing all the requests in one pass. The reason it's
394 * set to 1 instead of 0 is so that other kthreads will know
395 * that the channel mgr is running and won't bother trying to
398 atomic_dec(&part->channel_mgr_requests);
399 (void)wait_event_interruptible(part->channel_mgr_wq,
400 (atomic_read(&part->channel_mgr_requests) > 0 ||
401 part->local_IPI_amo != 0 ||
402 (part->act_state == XPC_P_DEACTIVATING &&
403 atomic_read(&part->nchannels_active) == 0 &&
404 xpc_partition_disengaged(part))));
405 atomic_set(&part->channel_mgr_requests, 1);
410 * When XPC HB determines that a partition has come up, it will create a new
411 * kthread and that kthread will call this function to attempt to set up the
412 * basic infrastructure used for Cross Partition Communication with the newly
415 * The kthread that was created by XPC HB and which setup the XPC
416 * infrastructure will remain assigned to the partition becoming the channel
417 * manager for that partition until the partition is deactivating, at which
418 * time the kthread will teardown the XPC infrastructure and then exit.
421 xpc_activating(void *__partid)
423 short partid = (u64)__partid;
424 struct xpc_partition *part = &xpc_partitions[partid];
425 unsigned long irq_flags;
427 DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
429 spin_lock_irqsave(&part->act_lock, irq_flags);
431 if (part->act_state == XPC_P_DEACTIVATING) {
432 part->act_state = XPC_P_INACTIVE;
433 spin_unlock_irqrestore(&part->act_lock, irq_flags);
434 part->remote_rp_pa = 0;
438 /* indicate the thread is activating */
439 DBUG_ON(part->act_state != XPC_P_ACTIVATION_REQ);
440 part->act_state = XPC_P_ACTIVATING;
442 XPC_SET_REASON(part, 0, 0);
443 spin_unlock_irqrestore(&part->act_lock, irq_flags);
445 dev_dbg(xpc_part, "activating partition %d\n", partid);
447 xpc_allow_hb(partid);
449 if (xpc_setup_infrastructure(part) == xpSuccess) {
450 (void)xpc_part_ref(part); /* this will always succeed */
452 if (xpc_make_first_contact(part) == xpSuccess) {
453 xpc_mark_partition_active(part);
454 xpc_channel_mgr(part);
455 /* won't return until partition is deactivating */
458 xpc_part_deref(part);
459 xpc_teardown_infrastructure(part);
462 xpc_disallow_hb(partid);
463 xpc_mark_partition_inactive(part);
465 if (part->reason == xpReactivating) {
466 /* interrupting ourselves results in activating partition */
467 xpc_IPI_send_local_reactivate(part->reactivate_nasid);
474 xpc_activate_partition(struct xpc_partition *part)
476 short partid = XPC_PARTID(part);
477 unsigned long irq_flags;
478 struct task_struct *kthread;
480 spin_lock_irqsave(&part->act_lock, irq_flags);
482 DBUG_ON(part->act_state != XPC_P_INACTIVE);
484 part->act_state = XPC_P_ACTIVATION_REQ;
485 XPC_SET_REASON(part, xpCloneKThread, __LINE__);
487 spin_unlock_irqrestore(&part->act_lock, irq_flags);
489 kthread = kthread_run(xpc_activating, (void *)((u64)partid), "xpc%02d",
491 if (IS_ERR(kthread)) {
492 spin_lock_irqsave(&part->act_lock, irq_flags);
493 part->act_state = XPC_P_INACTIVE;
494 XPC_SET_REASON(part, xpCloneKThreadFailed, __LINE__);
495 spin_unlock_irqrestore(&part->act_lock, irq_flags);
500 * Check to see if there is any channel activity to/from the specified
504 xpc_check_for_channel_activity(struct xpc_partition *part)
507 unsigned long irq_flags;
509 /* this needs to be uncommented, but I'm thinking this function and the */
510 /* ones that call it need to be moved into xpc_sn2.c... */
511 IPI_amo = 0; /* = xpc_IPI_receive(part->local_IPI_amo_va); */
515 spin_lock_irqsave(&part->IPI_lock, irq_flags);
516 part->local_IPI_amo |= IPI_amo;
517 spin_unlock_irqrestore(&part->IPI_lock, irq_flags);
519 dev_dbg(xpc_chan, "received IPI from partid=%d, IPI_amo=0x%lx\n",
520 XPC_PARTID(part), IPI_amo);
522 xpc_wakeup_channel_mgr(part);
526 * Handle the receipt of a SGI_XPC_NOTIFY IRQ by seeing whether the specified
527 * partition actually sent it. Since SGI_XPC_NOTIFY IRQs may be shared by more
528 * than one partition, we use an AMO_t structure per partition to indicate
529 * whether a partition has sent an IPI or not. If it has, then wake up the
530 * associated kthread to handle it.
532 * All SGI_XPC_NOTIFY IRQs received by XPC are the result of IPIs sent by XPC
533 * running on other partitions.
535 * Noteworthy Arguments:
537 * irq - Interrupt ReQuest number. NOT USED.
539 * dev_id - partid of IPI's potential sender.
542 xpc_notify_IRQ_handler(int irq, void *dev_id)
544 short partid = (short)(u64)dev_id;
545 struct xpc_partition *part = &xpc_partitions[partid];
547 DBUG_ON(partid < 0 || partid >= xp_max_npartitions);
549 if (xpc_part_ref(part)) {
550 xpc_check_for_channel_activity(part);
552 xpc_part_deref(part);
558 * Check to see if xpc_notify_IRQ_handler() dropped any IPIs on the floor
559 * because the write to their associated IPI amo completed after the IRQ/IPI
563 xpc_dropped_IPI_check(struct xpc_partition *part)
565 if (xpc_part_ref(part)) {
566 xpc_check_for_channel_activity(part);
568 part->dropped_IPI_timer.expires = jiffies +
569 XPC_P_DROPPED_IPI_WAIT_INTERVAL;
570 add_timer(&part->dropped_IPI_timer);
571 xpc_part_deref(part);
576 xpc_activate_kthreads(struct xpc_channel *ch, int needed)
578 int idle = atomic_read(&ch->kthreads_idle);
579 int assigned = atomic_read(&ch->kthreads_assigned);
582 DBUG_ON(needed <= 0);
585 wakeup = (needed > idle) ? idle : needed;
588 dev_dbg(xpc_chan, "wakeup %d idle kthreads, partid=%d, "
589 "channel=%d\n", wakeup, ch->partid, ch->number);
591 /* only wakeup the requested number of kthreads */
592 wake_up_nr(&ch->idle_wq, wakeup);
598 if (needed + assigned > ch->kthreads_assigned_limit) {
599 needed = ch->kthreads_assigned_limit - assigned;
604 dev_dbg(xpc_chan, "create %d new kthreads, partid=%d, channel=%d\n",
605 needed, ch->partid, ch->number);
607 xpc_create_kthreads(ch, needed, 0);
611 * This function is where XPC's kthreads wait for messages to deliver.
614 xpc_kthread_waitmsgs(struct xpc_partition *part, struct xpc_channel *ch)
617 /* deliver messages to their intended recipients */
619 while (ch->w_local_GP.get < ch->w_remote_GP.put &&
620 !(ch->flags & XPC_C_DISCONNECTING)) {
624 if (atomic_inc_return(&ch->kthreads_idle) >
625 ch->kthreads_idle_limit) {
626 /* too many idle kthreads on this channel */
627 atomic_dec(&ch->kthreads_idle);
631 dev_dbg(xpc_chan, "idle kthread calling "
632 "wait_event_interruptible_exclusive()\n");
634 (void)wait_event_interruptible_exclusive(ch->idle_wq,
635 (ch->w_local_GP.get < ch->w_remote_GP.put ||
636 (ch->flags & XPC_C_DISCONNECTING)));
638 atomic_dec(&ch->kthreads_idle);
640 } while (!(ch->flags & XPC_C_DISCONNECTING));
644 xpc_kthread_start(void *args)
646 short partid = XPC_UNPACK_ARG1(args);
647 u16 ch_number = XPC_UNPACK_ARG2(args);
648 struct xpc_partition *part = &xpc_partitions[partid];
649 struct xpc_channel *ch;
651 unsigned long irq_flags;
653 dev_dbg(xpc_chan, "kthread starting, partid=%d, channel=%d\n",
656 ch = &part->channels[ch_number];
658 if (!(ch->flags & XPC_C_DISCONNECTING)) {
660 /* let registerer know that connection has been established */
662 spin_lock_irqsave(&ch->lock, irq_flags);
663 if (!(ch->flags & XPC_C_CONNECTEDCALLOUT)) {
664 ch->flags |= XPC_C_CONNECTEDCALLOUT;
665 spin_unlock_irqrestore(&ch->lock, irq_flags);
667 xpc_connected_callout(ch);
669 spin_lock_irqsave(&ch->lock, irq_flags);
670 ch->flags |= XPC_C_CONNECTEDCALLOUT_MADE;
671 spin_unlock_irqrestore(&ch->lock, irq_flags);
674 * It is possible that while the callout was being
675 * made that the remote partition sent some messages.
676 * If that is the case, we may need to activate
677 * additional kthreads to help deliver them. We only
678 * need one less than total #of messages to deliver.
680 n_needed = ch->w_remote_GP.put - ch->w_local_GP.get - 1;
681 if (n_needed > 0 && !(ch->flags & XPC_C_DISCONNECTING))
682 xpc_activate_kthreads(ch, n_needed);
685 spin_unlock_irqrestore(&ch->lock, irq_flags);
688 xpc_kthread_waitmsgs(part, ch);
691 /* let registerer know that connection is disconnecting */
693 spin_lock_irqsave(&ch->lock, irq_flags);
694 if ((ch->flags & XPC_C_CONNECTEDCALLOUT_MADE) &&
695 !(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) {
696 ch->flags |= XPC_C_DISCONNECTINGCALLOUT;
697 spin_unlock_irqrestore(&ch->lock, irq_flags);
699 xpc_disconnect_callout(ch, xpDisconnecting);
701 spin_lock_irqsave(&ch->lock, irq_flags);
702 ch->flags |= XPC_C_DISCONNECTINGCALLOUT_MADE;
704 spin_unlock_irqrestore(&ch->lock, irq_flags);
706 if (atomic_dec_return(&ch->kthreads_assigned) == 0) {
707 if (atomic_dec_return(&part->nchannels_engaged) == 0) {
708 xpc_mark_partition_disengaged(part);
709 xpc_IPI_send_disengage(part);
713 xpc_msgqueue_deref(ch);
715 dev_dbg(xpc_chan, "kthread exiting, partid=%d, channel=%d\n",
718 xpc_part_deref(part);
723 * For each partition that XPC has established communications with, there is
724 * a minimum of one kernel thread assigned to perform any operation that
725 * may potentially sleep or block (basically the callouts to the asynchronous
726 * functions registered via xpc_connect()).
728 * Additional kthreads are created and destroyed by XPC as the workload
731 * A kthread is assigned to one of the active channels that exists for a given
735 xpc_create_kthreads(struct xpc_channel *ch, int needed,
736 int ignore_disconnecting)
738 unsigned long irq_flags;
739 u64 args = XPC_PACK_ARGS(ch->partid, ch->number);
740 struct xpc_partition *part = &xpc_partitions[ch->partid];
741 struct task_struct *kthread;
743 while (needed-- > 0) {
746 * The following is done on behalf of the newly created
747 * kthread. That kthread is responsible for doing the
748 * counterpart to the following before it exits.
750 if (ignore_disconnecting) {
751 if (!atomic_inc_not_zero(&ch->kthreads_assigned)) {
752 /* kthreads assigned had gone to zero */
754 XPC_C_DISCONNECTINGCALLOUT_MADE));
758 } else if (ch->flags & XPC_C_DISCONNECTING) {
761 } else if (atomic_inc_return(&ch->kthreads_assigned) == 1) {
762 if (atomic_inc_return(&part->nchannels_engaged) == 1)
763 xpc_mark_partition_engaged(part);
765 (void)xpc_part_ref(part);
766 xpc_msgqueue_ref(ch);
768 kthread = kthread_run(xpc_kthread_start, (void *)args,
769 "xpc%02dc%d", ch->partid, ch->number);
770 if (IS_ERR(kthread)) {
771 /* the fork failed */
774 * NOTE: if (ignore_disconnecting &&
775 * !(ch->flags & XPC_C_DISCONNECTINGCALLOUT)) is true,
776 * then we'll deadlock if all other kthreads assigned
777 * to this channel are blocked in the channel's
778 * registerer, because the only thing that will unblock
779 * them is the xpDisconnecting callout that this
780 * failed kthread_run() would have made.
783 if (atomic_dec_return(&ch->kthreads_assigned) == 0 &&
784 atomic_dec_return(&part->nchannels_engaged) == 0) {
785 xpc_mark_partition_disengaged(part);
786 xpc_IPI_send_disengage(part);
788 xpc_msgqueue_deref(ch);
789 xpc_part_deref(part);
791 if (atomic_read(&ch->kthreads_assigned) <
792 ch->kthreads_idle_limit) {
794 * Flag this as an error only if we have an
795 * insufficient #of kthreads for the channel
798 spin_lock_irqsave(&ch->lock, irq_flags);
799 XPC_DISCONNECT_CHANNEL(ch, xpLackOfResources,
801 spin_unlock_irqrestore(&ch->lock, irq_flags);
809 xpc_disconnect_wait(int ch_number)
811 unsigned long irq_flags;
813 struct xpc_partition *part;
814 struct xpc_channel *ch;
815 int wakeup_channel_mgr;
817 /* now wait for all callouts to the caller's function to cease */
818 for (partid = 0; partid < xp_max_npartitions; partid++) {
819 part = &xpc_partitions[partid];
821 if (!xpc_part_ref(part))
824 ch = &part->channels[ch_number];
826 if (!(ch->flags & XPC_C_WDISCONNECT)) {
827 xpc_part_deref(part);
831 wait_for_completion(&ch->wdisconnect_wait);
833 spin_lock_irqsave(&ch->lock, irq_flags);
834 DBUG_ON(!(ch->flags & XPC_C_DISCONNECTED));
835 wakeup_channel_mgr = 0;
837 if (ch->delayed_IPI_flags) {
838 if (part->act_state != XPC_P_DEACTIVATING) {
839 spin_lock(&part->IPI_lock);
840 XPC_SET_IPI_FLAGS(part->local_IPI_amo,
842 ch->delayed_IPI_flags);
843 spin_unlock(&part->IPI_lock);
844 wakeup_channel_mgr = 1;
846 ch->delayed_IPI_flags = 0;
849 ch->flags &= ~XPC_C_WDISCONNECT;
850 spin_unlock_irqrestore(&ch->lock, irq_flags);
852 if (wakeup_channel_mgr)
853 xpc_wakeup_channel_mgr(part);
855 xpc_part_deref(part);
860 xpc_do_exit(enum xp_retval reason)
863 int active_part_count, printed_waiting_msg = 0;
864 struct xpc_partition *part;
865 unsigned long printmsg_time, disengage_request_timeout = 0;
867 /* a 'rmmod XPC' and a 'reboot' cannot both end up here together */
868 DBUG_ON(xpc_exiting == 1);
871 * Let the heartbeat checker thread and the discovery thread
872 * (if one is running) know that they should exit. Also wake up
873 * the heartbeat checker thread in case it's sleeping.
876 wake_up_interruptible(&xpc_activate_IRQ_wq);
878 /* wait for the discovery thread to exit */
879 wait_for_completion(&xpc_discovery_exited);
881 /* wait for the heartbeat checker thread to exit */
882 wait_for_completion(&xpc_hb_checker_exited);
884 /* sleep for a 1/3 of a second or so */
885 (void)msleep_interruptible(300);
887 /* wait for all partitions to become inactive */
889 printmsg_time = jiffies + (XPC_DISENGAGE_PRINTMSG_INTERVAL * HZ);
890 xpc_disengage_request_timedout = 0;
893 active_part_count = 0;
895 for (partid = 0; partid < xp_max_npartitions; partid++) {
896 part = &xpc_partitions[partid];
898 if (xpc_partition_disengaged(part) &&
899 part->act_state == XPC_P_INACTIVE) {
905 XPC_DEACTIVATE_PARTITION(part, reason);
907 if (part->disengage_request_timeout >
908 disengage_request_timeout) {
909 disengage_request_timeout =
910 part->disengage_request_timeout;
914 if (xpc_partition_engaged(-1UL)) {
915 if (time_is_before_jiffies(printmsg_time)) {
916 dev_info(xpc_part, "waiting for remote "
917 "partitions to disengage, timeout in "
919 (disengage_request_timeout - jiffies)
921 printmsg_time = jiffies +
922 (XPC_DISENGAGE_PRINTMSG_INTERVAL * HZ);
923 printed_waiting_msg = 1;
926 } else if (active_part_count > 0) {
927 if (printed_waiting_msg) {
928 dev_info(xpc_part, "waiting for local partition"
930 printed_waiting_msg = 0;
934 if (!xpc_disengage_request_timedout) {
935 dev_info(xpc_part, "all partitions have "
941 /* sleep for a 1/3 of a second or so */
942 (void)msleep_interruptible(300);
946 DBUG_ON(xpc_partition_engaged(-1UL));
947 DBUG_ON(xpc_any_hbs_allowed() != 0);
949 /* indicate to others that our reserved page is uninitialized */
950 xpc_rsvd_page->stamp = 0;
952 if (reason == xpUnloading) {
953 (void)unregister_die_notifier(&xpc_die_notifier);
954 (void)unregister_reboot_notifier(&xpc_reboot_notifier);
957 /* clear the interface to XPC's functions */
958 xpc_clear_interface();
961 unregister_sysctl_table(xpc_sysctl);
963 kfree(xpc_partitions);
964 kfree(xpc_remote_copy_buffer_base);
973 * This function is called when the system is being rebooted.
976 xpc_system_reboot(struct notifier_block *nb, unsigned long event, void *unused)
978 enum xp_retval reason;
982 reason = xpSystemReboot;
985 reason = xpSystemHalt;
988 reason = xpSystemPoweroff;
991 reason = xpSystemGoingDown;
999 * Notify other partitions to disengage from all references to our memory.
1002 xpc_die_disengage(void)
1004 struct xpc_partition *part;
1006 unsigned long engaged;
1007 long time, printmsg_time, disengage_request_timeout;
1009 /* keep xpc_hb_checker thread from doing anything (just in case) */
1012 xpc_disallow_all_hbs(); /*indicate we're deactivated */
1014 for (partid = 0; partid < xp_max_npartitions; partid++) {
1015 part = &xpc_partitions[partid];
1017 if (!XPC_SUPPORTS_DISENGAGE_REQUEST(part->
1018 remote_vars_version)) {
1020 /* just in case it was left set by an earlier XPC */
1021 xpc_clear_partition_engaged(1UL << partid);
1025 if (xpc_partition_engaged(1UL << partid) ||
1026 part->act_state != XPC_P_INACTIVE) {
1027 xpc_request_partition_disengage(part);
1028 xpc_mark_partition_disengaged(part);
1029 xpc_IPI_send_disengage(part);
1034 printmsg_time = time +
1035 (XPC_DISENGAGE_PRINTMSG_INTERVAL * sn_rtc_cycles_per_second);
1036 disengage_request_timeout = time +
1037 (xpc_disengage_request_timelimit * sn_rtc_cycles_per_second);
1039 /* wait for all other partitions to disengage from us */
1042 engaged = xpc_partition_engaged(-1UL);
1044 dev_info(xpc_part, "all partitions have disengaged\n");
1049 if (time >= disengage_request_timeout) {
1050 for (partid = 0; partid < xp_max_npartitions;
1052 if (engaged & (1UL << partid)) {
1053 dev_info(xpc_part, "disengage from "
1054 "remote partition %d timed "
1061 if (time >= printmsg_time) {
1062 dev_info(xpc_part, "waiting for remote partitions to "
1063 "disengage, timeout in %ld seconds\n",
1064 (disengage_request_timeout - time) /
1065 sn_rtc_cycles_per_second);
1066 printmsg_time = time +
1067 (XPC_DISENGAGE_PRINTMSG_INTERVAL *
1068 sn_rtc_cycles_per_second);
1074 * This function is called when the system is being restarted or halted due
1075 * to some sort of system failure. If this is the case we need to notify the
1076 * other partitions to disengage from all references to our memory.
1077 * This function can also be called when our heartbeater could be offlined
1078 * for a time. In this case we need to notify other partitions to not worry
1079 * about the lack of a heartbeat.
1082 xpc_system_die(struct notifier_block *nb, unsigned long event, void *unused)
1085 case DIE_MACHINE_RESTART:
1086 case DIE_MACHINE_HALT:
1087 xpc_die_disengage();
1090 case DIE_KDEBUG_ENTER:
1091 /* Should lack of heartbeat be ignored by other partitions? */
1092 if (!xpc_kdebug_ignore)
1096 case DIE_MCA_MONARCH_ENTER:
1097 case DIE_INIT_MONARCH_ENTER:
1098 xpc_offline_heartbeat();
1101 case DIE_KDEBUG_LEAVE:
1102 /* Is lack of heartbeat being ignored by other partitions? */
1103 if (!xpc_kdebug_ignore)
1107 case DIE_MCA_MONARCH_LEAVE:
1108 case DIE_INIT_MONARCH_LEAVE:
1109 xpc_online_heartbeat();
1121 struct xpc_partition *part;
1122 struct task_struct *kthread;
1127 * The ia64-sn2 architecture supports at most 64 partitions.
1128 * And the inability to unregister remote AMOs restricts us
1129 * further to only support exactly 64 partitions on this
1130 * architecture, no less.
1132 if (xp_max_npartitions != 64)
1135 ret = xpc_init_sn2();
1139 } else if (is_uv()) {
1146 snprintf(xpc_part->bus_id, BUS_ID_SIZE, "part");
1147 snprintf(xpc_chan->bus_id, BUS_ID_SIZE, "chan");
1149 buf_size = max(XPC_RP_VARS_SIZE,
1150 XPC_RP_HEADER_SIZE + XP_NASID_MASK_BYTES);
1151 xpc_remote_copy_buffer = xpc_kmalloc_cacheline_aligned(buf_size,
1153 &xpc_remote_copy_buffer_base);
1154 if (xpc_remote_copy_buffer == NULL) {
1155 dev_err(xpc_part, "can't get memory for remote copy buffer\n");
1160 xpc_partitions = kzalloc(sizeof(struct xpc_partition) *
1161 xp_max_npartitions, GFP_KERNEL);
1162 if (xpc_partitions == NULL) {
1163 dev_err(xpc_part, "can't get memory for partition structure\n");
1169 * The first few fields of each entry of xpc_partitions[] need to
1170 * be initialized now so that calls to xpc_connect() and
1171 * xpc_disconnect() can be made prior to the activation of any remote
1172 * partition. NOTE THAT NONE OF THE OTHER FIELDS BELONGING TO THESE
1173 * ENTRIES ARE MEANINGFUL UNTIL AFTER AN ENTRY'S CORRESPONDING
1174 * PARTITION HAS BEEN ACTIVATED.
1176 for (partid = 0; partid < xp_max_npartitions; partid++) {
1177 part = &xpc_partitions[partid];
1179 DBUG_ON((u64)part != L1_CACHE_ALIGN((u64)part));
1181 part->activate_IRQ_rcvd = 0;
1182 spin_lock_init(&part->act_lock);
1183 part->act_state = XPC_P_INACTIVE;
1184 XPC_SET_REASON(part, 0, 0);
1186 init_timer(&part->disengage_request_timer);
1187 part->disengage_request_timer.function =
1188 xpc_timeout_partition_disengage_request;
1189 part->disengage_request_timer.data = (unsigned long)part;
1191 part->setup_state = XPC_P_UNSET;
1192 init_waitqueue_head(&part->teardown_wq);
1193 atomic_set(&part->references, 0);
1196 xpc_sysctl = register_sysctl_table(xpc_sys_dir);
1199 * Fill the partition reserved page with the information needed by
1200 * other partitions to discover we are alive and establish initial
1203 xpc_rsvd_page = xpc_setup_rsvd_page();
1204 if (xpc_rsvd_page == NULL) {
1205 dev_err(xpc_part, "can't setup our reserved page\n");
1210 /* add ourselves to the reboot_notifier_list */
1211 ret = register_reboot_notifier(&xpc_reboot_notifier);
1213 dev_warn(xpc_part, "can't register reboot notifier\n");
1215 /* add ourselves to the die_notifier list */
1216 ret = register_die_notifier(&xpc_die_notifier);
1218 dev_warn(xpc_part, "can't register die notifier\n");
1221 * The real work-horse behind xpc. This processes incoming
1222 * interrupts and monitors remote heartbeats.
1224 kthread = kthread_run(xpc_hb_checker, NULL, XPC_HB_CHECK_THREAD_NAME);
1225 if (IS_ERR(kthread)) {
1226 dev_err(xpc_part, "failed while forking hb check thread\n");
1232 * Startup a thread that will attempt to discover other partitions to
1233 * activate based on info provided by SAL. This new thread is short
1234 * lived and will exit once discovery is complete.
1236 kthread = kthread_run(xpc_initiate_discovery, NULL,
1237 XPC_DISCOVERY_THREAD_NAME);
1238 if (IS_ERR(kthread)) {
1239 dev_err(xpc_part, "failed while forking discovery thread\n");
1241 /* mark this new thread as a non-starter */
1242 complete(&xpc_discovery_exited);
1244 xpc_do_exit(xpUnloading);
1248 /* set the interface to point at XPC's functions */
1249 xpc_set_interface(xpc_initiate_connect, xpc_initiate_disconnect,
1250 xpc_initiate_send, xpc_initiate_send_notify,
1251 xpc_initiate_received, xpc_initiate_partid_to_nasids);
1255 /* initialization was not successful */
1257 /* indicate to others that our reserved page is uninitialized */
1258 xpc_rsvd_page->stamp = 0;
1260 (void)unregister_die_notifier(&xpc_die_notifier);
1261 (void)unregister_reboot_notifier(&xpc_reboot_notifier);
1264 unregister_sysctl_table(xpc_sysctl);
1265 kfree(xpc_partitions);
1267 kfree(xpc_remote_copy_buffer_base);
1276 module_init(xpc_init);
1281 xpc_do_exit(xpUnloading);
1284 module_exit(xpc_exit);
1286 MODULE_AUTHOR("Silicon Graphics, Inc.");
1287 MODULE_DESCRIPTION("Cross Partition Communication (XPC) support");
1288 MODULE_LICENSE("GPL");
1290 module_param(xpc_hb_interval, int, 0);
1291 MODULE_PARM_DESC(xpc_hb_interval, "Number of seconds between "
1292 "heartbeat increments.");
1294 module_param(xpc_hb_check_interval, int, 0);
1295 MODULE_PARM_DESC(xpc_hb_check_interval, "Number of seconds between "
1296 "heartbeat checks.");
1298 module_param(xpc_disengage_request_timelimit, int, 0);
1299 MODULE_PARM_DESC(xpc_disengage_request_timelimit, "Number of seconds to wait "
1300 "for disengage request to complete.");
1302 module_param(xpc_kdebug_ignore, int, 0);
1303 MODULE_PARM_DESC(xpc_kdebug_ignore, "Should lack of heartbeat be ignored by "
1304 "other partitions when dropping into kdebug.");