git.oblomov.eu Git - linux-2.6/blob - drivers/misc/sgi-xp/xpc_channel.c

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