sys/vfs/hammer2/hammer2_msg.c

   1 /*-
   2  * Copyright (c) 2012 The DragonFly Project.  All rights reserved.
   3  *
   4  * This code is derived from software contributed to The DragonFly Project
   5  * by Matthew Dillon <dillon@backplane.com>
   6  *
   7  * Redistribution and use in source and binary forms, with or without
   8  * modification, are permitted provided that the following conditions
   9  * are met:
  10  *
  11  * 1. Redistributions of source code must retain the above copyright
  12  *    notice, this list of conditions and the following disclaimer.
  13  * 2. Redistributions in binary form must reproduce the above copyright
  14  *    notice, this list of conditions and the following disclaimer in
  15  *    the documentation and/or other materials provided with the
  16  *    distribution.
  17  * 3. Neither the name of The DragonFly Project nor the names of its
  18  *    contributors may be used to endorse or promote products derived
  19  *    from this software without specific, prior written permission.
  20  *
  21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
  24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
  25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
  26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
  27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
  28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
  29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  32  * SUCH DAMAGE.
  33  */
  34
  35 #include "hammer2.h"
  36
  37 RB_GENERATE(hammer2_state_tree, hammer2_state, rbnode, hammer2_state_cmp);
  38
  39 /*
  40  * Process state tracking for a message after reception, prior to
  41  * execution.
  42  *
  43  * Called with msglk held and the msg dequeued.
  44  *
  45  * All messages are called with dummy state and return actual state.
  46  * (One-off messages often just return the same dummy state).
  47  *
  48  * May request that caller discard the message by setting *discardp to 1.
  49  * The returned state is not used in this case and is allowed to be NULL.
  50  *
  51  * --
  52  *
  53  * These routines handle persistent and command/reply message state via the
  54  * CREATE and DELETE flags.  The first message in a command or reply sequence
  55  * sets CREATE, the last message in a command or reply sequence sets DELETE.
  56  *
  57  * There can be any number of intermediate messages belonging to the same
  58  * sequence sent inbetween the CREATE message and the DELETE message,
  59  * which set neither flag.  This represents a streaming command or reply.
  60  *
  61  * Any command message received with CREATE set expects a reply sequence to
  62  * be returned.  Reply sequences work the same as command sequences except the
  63  * REPLY bit is also sent.  Both the command side and reply side can
  64  * degenerate into a single message with both CREATE and DELETE set.  Note
  65  * that one side can be streaming and the other side not, or neither, or both.
  66  *
  67  * The msgid is unique for the initiator.  That is, two sides sending a new
  68  * message can use the same msgid without colliding.
  69  *
  70  * --
  71  *
  72  * ABORT sequences work by setting the ABORT flag along with normal message
  73  * state.  However, ABORTs can also be sent on half-closed messages, that is
  74  * even if the command or reply side has already sent a DELETE, as long as
  75  * the message has not been fully closed it can still send an ABORT+DELETE
  76  * to terminate the half-closed message state.
  77  *
  78  * Since ABORT+DELETEs can race we silently discard ABORT's for message
  79  * state which has already been fully closed.  REPLY+ABORT+DELETEs can
  80  * also race, and in this situation the other side might have already
  81  * initiated a new unrelated command with the same message id.  Since
  82  * the abort has not set the CREATE flag the situation can be detected
  83  * and the message will also be discarded.
  84  *
  85  * Non-blocking requests can be initiated with ABORT+CREATE[+DELETE].
  86  * The ABORT request is essentially integrated into the command instead
  87  * of being sent later on.  In this situation the command implementation
  88  * detects that CREATE and ABORT are both set (vs ABORT alone) and can
  89  * special-case non-blocking operation for the command.
  90  *
  91  * NOTE!  Messages with ABORT set without CREATE or DELETE are considered
  92  *        to be mid-stream aborts for command/reply sequences.  ABORTs on
  93  *        one-way messages are not supported.
  94  *
  95  * NOTE!  If a command sequence does not support aborts the ABORT flag is
  96  *        simply ignored.
  97  *
  98  * --
  99  *
 100  * One-off messages (no reply expected) are sent with neither CREATE or DELETE
 101  * set.  One-off messages cannot be aborted and typically aren't processed
 102  * by these routines.  The REPLY bit can be used to distinguish whether a
 103  * one-off message is a command or reply.  For example, one-off replies
 104  * will typically just contain status updates.
 105  */
 106 int
 107 hammer2_state_msgrx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 108 {
 109         hammer2_state_t *state;
 110         int error;
 111
 112         /*
 113          * Make sure a state structure is ready to go in case we need a new
 114          * one.  This is the only routine which uses freerd_state so no
 115          * races are possible.
 116          */
 117         if ((state = pmp->freerd_state) == NULL) {
 118                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
 119                 state->pmp = pmp;
 120                 state->flags = HAMMER2_STATE_DYNAMIC;
 121                 pmp->freerd_state = state;
 122         }
 123
 124         /*
 125          * Lock RB tree and locate existing persistent state, if any.
 126          *
 127          * If received msg is a command state is on staterd_tree.
 128          * If received msg is a reply state is on statewr_tree.
 129          */
 130         lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 131
 132         state->msgid = msg->any.head.msgid;
 133         state->spanid = msg->any.head.spanid;
 134         kprintf("received msg %08x msgid %jx spanid=%jx\n",
 135                 msg->any.head.cmd,
 136                 (intmax_t)msg->any.head.msgid, (intmax_t)msg->any.head.spanid);
 137         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY)
 138                 state = RB_FIND(hammer2_state_tree, &pmp->statewr_tree, state);
 139         else
 140                 state = RB_FIND(hammer2_state_tree, &pmp->staterd_tree, state);
 141         msg->state = state;
 142
 143         /*
 144          * Short-cut one-off or mid-stream messages (state may be NULL).
 145          */
 146         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 147                                   HAMMER2_MSGF_ABORT)) == 0) {
 148                 lockmgr(&pmp->msglk, LK_RELEASE);
 149                 return(0);
 150         }
 151
 152         /*
 153          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
 154          * inside the case statements.
 155          */
 156         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 157                                     HAMMER2_MSGF_REPLY)) {
 158         case HAMMER2_MSGF_CREATE:
 159         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 160                 /*
 161                  * New persistant command received.
 162                  */
 163                 if (state) {
 164                         kprintf("hammer2_state_msgrx: duplicate transaction\n");
 165                         error = EINVAL;
 166                         break;
 167                 }
 168                 state = pmp->freerd_state;
 169                 pmp->freerd_state = NULL;
 170                 msg->state = state;
 171                 state->msg = msg;
 172                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 173                 RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state);
 174                 state->flags |= HAMMER2_STATE_INSERTED;
 175                 error = 0;
 176                 break;
 177         case HAMMER2_MSGF_DELETE:
 178                 /*
 179                  * Persistent state is expected but might not exist if an
 180                  * ABORT+DELETE races the close.
 181                  */
 182                 if (state == NULL) {
 183                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 184                                 error = EALREADY;
 185                         } else {
 186                                 kprintf("hammer2_state_msgrx: no state "
 187                                         "for DELETE\n");
 188                                 error = EINVAL;
 189                         }
 190                         break;
 191                 }
 192
 193                 /*
 194                  * Handle another ABORT+DELETE case if the msgid has already
 195                  * been reused.
 196                  */
 197                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 198                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 199                                 error = EALREADY;
 200                         } else {
 201                                 kprintf("hammer2_state_msgrx: state reused "
 202                                         "for DELETE\n");
 203                                 error = EINVAL;
 204                         }
 205                         break;
 206                 }
 207                 error = 0;
 208                 break;
 209         default:
 210                 /*
 211                  * Check for mid-stream ABORT command received, otherwise
 212                  * allow.
 213                  */
 214                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 215                         if (state == NULL ||
 216                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 217                                 error = EALREADY;
 218                                 break;
 219                         }
 220                 }
 221                 error = 0;
 222                 break;
 223         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
 224         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 225                 /*
 226                  * When receiving a reply with CREATE set the original
 227                  * persistent state message should already exist.
 228                  */
 229                 if (state == NULL) {
 230                         kprintf("hammer2_state_msgrx: no state match for "
 231                                 "REPLY cmd=%08x\n", msg->any.head.cmd);
 232                         error = EINVAL;
 233                         break;
 234                 }
 235                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 236                 error = 0;
 237                 break;
 238         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
 239                 /*
 240                  * Received REPLY+ABORT+DELETE in case where msgid has
 241                  * already been fully closed, ignore the message.
 242                  */
 243                 if (state == NULL) {
 244                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 245                                 error = EALREADY;
 246                         } else {
 247                                 kprintf("hammer2_state_msgrx: no state match "
 248                                         "for REPLY|DELETE\n");
 249                                 error = EINVAL;
 250                         }
 251                         break;
 252                 }
 253
 254                 /*
 255                  * Received REPLY+ABORT+DELETE in case where msgid has
 256                  * already been reused for an unrelated message,
 257                  * ignore the message.
 258                  */
 259                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 260                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 261                                 error = EALREADY;
 262                         } else {
 263                                 kprintf("hammer2_state_msgrx: state reused "
 264                                         "for REPLY|DELETE\n");
 265                                 error = EINVAL;
 266                         }
 267                         break;
 268                 }
 269                 error = 0;
 270                 break;
 271         case HAMMER2_MSGF_REPLY:
 272                 /*
 273                  * Check for mid-stream ABORT reply received to sent command.
 274                  */
 275                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 276                         if (state == NULL ||
 277                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 278                                 error = EALREADY;
 279                                 break;
 280                         }
 281                 }
 282                 error = 0;
 283                 break;
 284         }
 285         lockmgr(&pmp->msglk, LK_RELEASE);
 286         return (error);
 287 }
 288
 289 void
 290 hammer2_state_cleanuprx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 291 {
 292         hammer2_state_t *state;
 293
 294         if ((state = msg->state) == NULL) {
 295                 hammer2_msg_free(pmp, msg);
 296         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
 297                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 298                 state->rxcmd |= HAMMER2_MSGF_DELETE;
 299                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
 300                         if (state->msg == msg)
 301                                 state->msg = NULL;
 302                         KKASSERT(state->flags & HAMMER2_STATE_INSERTED);
 303                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
 304                                 RB_REMOVE(hammer2_state_tree,
 305                                           &pmp->statewr_tree, state);
 306                         } else {
 307                                 RB_REMOVE(hammer2_state_tree,
 308                                           &pmp->staterd_tree, state);
 309                         }
 310                         state->flags &= ~HAMMER2_STATE_INSERTED;
 311                         lockmgr(&pmp->msglk, LK_RELEASE);
 312                         hammer2_state_free(state);
 313                 } else {
 314                         lockmgr(&pmp->msglk, LK_RELEASE);
 315                 }
 316                 hammer2_msg_free(pmp, msg);
 317         } else if (state->msg != msg) {
 318                 hammer2_msg_free(pmp, msg);
 319         }
 320 }
 321
 322 /*
 323  * Process state tracking for a message prior to transmission.
 324  *
 325  * Called with msglk held and the msg dequeued.
 326  *
 327  * One-off messages are usually with dummy state and msg->state may be NULL
 328  * in this situation.
 329  *
 330  * New transactions (when CREATE is set) will insert the state.
 331  *
 332  * May request that caller discard the message by setting *discardp to 1.
 333  * A NULL state may be returned in this case.
 334  */
 335 int
 336 hammer2_state_msgtx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 337 {
 338         hammer2_state_t *state;
 339         int error;
 340
 341         /*
 342          * Make sure a state structure is ready to go in case we need a new
 343          * one.  This is the only routine which uses freewr_state so no
 344          * races are possible.
 345          */
 346         if ((state = pmp->freewr_state) == NULL) {
 347                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
 348                 state->pmp = pmp;
 349                 state->flags = HAMMER2_STATE_DYNAMIC;
 350                 pmp->freewr_state = state;
 351         }
 352
 353         /*
 354          * Lock RB tree.  If persistent state is present it will have already
 355          * been assigned to msg.
 356          */
 357         lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 358         state = msg->state;
 359
 360         /*
 361          * Short-cut one-off or mid-stream messages (state may be NULL).
 362          */
 363         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 364                                   HAMMER2_MSGF_ABORT)) == 0) {
 365                 lockmgr(&pmp->msglk, LK_RELEASE);
 366                 return(0);
 367         }
 368
 369
 370         /*
 371          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
 372          * inside the case statements.
 373          */
 374         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 375                                     HAMMER2_MSGF_REPLY)) {
 376         case HAMMER2_MSGF_CREATE:
 377         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 378                 /*
 379                  * Insert the new persistent message state and mark
 380                  * half-closed if DELETE is set.  Since this is a new
 381                  * message it isn't possible to transition into the fully
 382                  * closed state here.
 383                  *
 384                  * XXX state must be assigned and inserted by
 385                  *     hammer2_msg_write().  txcmd is assigned by us
 386                  *     on-transmit.
 387                  */
 388                 KKASSERT(state != NULL);
 389 #if 0
 390                 if (state == NULL) {
 391                         state = pmp->freerd_state;
 392                         pmp->freerd_state = NULL;
 393                         msg->state = state;
 394                         state->msg = msg;
 395                         state->msgid = msg->any.head.msgid;
 396                         state->spanid = msg->any.head.spanid;
 397                 }
 398                 KKASSERT((state->flags & HAMMER2_STATE_INSERTED) == 0);
 399                 if (RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state)) {
 400                         kprintf("hammer2_state_msgtx: duplicate transaction\n");
 401                         error = EINVAL;
 402                         break;
 403                 }
 404                 state->flags |= HAMMER2_STATE_INSERTED;
 405 #endif
 406                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 407                 error = 0;
 408                 break;
 409         case HAMMER2_MSGF_DELETE:
 410                 /*
 411                  * Sent ABORT+DELETE in case where msgid has already
 412                  * been fully closed, ignore the message.
 413                  */
 414                 if (state == NULL) {
 415                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 416                                 error = EALREADY;
 417                         } else {
 418                                 kprintf("hammer2_state_msgtx: no state match "
 419                                         "for DELETE\n");
 420                                 error = EINVAL;
 421                         }
 422                         break;
 423                 }
 424
 425                 /*
 426                  * Sent ABORT+DELETE in case where msgid has
 427                  * already been reused for an unrelated message,
 428                  * ignore the message.
 429                  */
 430                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 431                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 432                                 error = EALREADY;
 433                         } else {
 434                                 kprintf("hammer2_state_msgtx: state reused "
 435                                         "for DELETE\n");
 436                                 error = EINVAL;
 437                         }
 438                         break;
 439                 }
 440                 error = 0;
 441                 break;
 442         default:
 443                 /*
 444                  * Check for mid-stream ABORT command sent
 445                  */
 446                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 447                         if (state == NULL ||
 448                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 449                                 error = EALREADY;
 450                                 break;
 451                         }
 452                 }
 453                 error = 0;
 454                 break;
 455         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
 456         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 457                 /*
 458                  * When transmitting a reply with CREATE set the original
 459                  * persistent state message should already exist.
 460                  */
 461                 if (state == NULL) {
 462                         kprintf("hammer2_state_msgtx: no state match "
 463                                 "for REPLY | CREATE\n");
 464                         error = EINVAL;
 465                         break;
 466                 }
 467                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 468                 error = 0;
 469                 break;
 470         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
 471                 /*
 472                  * When transmitting a reply with DELETE set the original
 473                  * persistent state message should already exist.
 474                  *
 475                  * This is very similar to the REPLY|CREATE|* case except
 476                  * txcmd is already stored, so we just add the DELETE flag.
 477                  *
 478                  * Sent REPLY+ABORT+DELETE in case where msgid has
 479                  * already been fully closed, ignore the message.
 480                  */
 481                 if (state == NULL) {
 482                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 483                                 error = EALREADY;
 484                         } else {
 485                                 kprintf("hammer2_state_msgtx: no state match "
 486                                         "for REPLY | DELETE\n");
 487                                 error = EINVAL;
 488                         }
 489                         break;
 490                 }
 491
 492                 /*
 493                  * Sent REPLY+ABORT+DELETE in case where msgid has already
 494                  * been reused for an unrelated message, ignore the message.
 495                  */
 496                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 497                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 498                                 error = EALREADY;
 499                         } else {
 500                                 kprintf("hammer2_state_msgtx: state reused "
 501                                         "for REPLY | DELETE\n");
 502                                 error = EINVAL;
 503                         }
 504                         break;
 505                 }
 506                 error = 0;
 507                 break;
 508         case HAMMER2_MSGF_REPLY:
 509                 /*
 510                  * Check for mid-stream ABORT reply sent.
 511                  *
 512                  * One-off REPLY messages are allowed for e.g. status updates.
 513                  */
 514                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 515                         if (state == NULL ||
 516                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 517                                 error = EALREADY;
 518                                 break;
 519                         }
 520                 }
 521                 error = 0;
 522                 break;
 523         }
 524         lockmgr(&pmp->msglk, LK_RELEASE);
 525         return (error);
 526 }
 527
 528 void
 529 hammer2_state_cleanuptx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 530 {
 531         hammer2_state_t *state;
 532
 533         if ((state = msg->state) == NULL) {
 534                 hammer2_msg_free(pmp, msg);
 535         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
 536                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 537                 state->txcmd |= HAMMER2_MSGF_DELETE;
 538                 if (state->rxcmd & HAMMER2_MSGF_DELETE) {
 539                         if (state->msg == msg)
 540                                 state->msg = NULL;
 541                         KKASSERT(state->flags & HAMMER2_STATE_INSERTED);
 542                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
 543                                 RB_REMOVE(hammer2_state_tree,
 544                                           &pmp->staterd_tree, state);
 545                         } else {
 546                                 RB_REMOVE(hammer2_state_tree,
 547                                           &pmp->statewr_tree, state);
 548                         }
 549                         state->flags &= ~HAMMER2_STATE_INSERTED;
 550                         lockmgr(&pmp->msglk, LK_RELEASE);
 551                         hammer2_state_free(state);
 552                 } else {
 553                         lockmgr(&pmp->msglk, LK_RELEASE);
 554                 }
 555                 hammer2_msg_free(pmp, msg);
 556         } else if (state->msg != msg) {
 557                 hammer2_msg_free(pmp, msg);
 558         }
 559 }
 560
 561 void
 562 hammer2_state_free(hammer2_state_t *state)
 563 {
 564         hammer2_pfsmount_t *pmp = state->pmp;
 565         hammer2_msg_t *msg;
 566
 567         msg = state->msg;
 568         state->msg = NULL;
 569         kfree(state, pmp->mmsg);
 570         if (msg)
 571                 hammer2_msg_free(pmp, msg);
 572 }
 573
 574 hammer2_msg_t *
 575 hammer2_msg_alloc(hammer2_pfsmount_t *pmp, uint64_t spanid, uint32_t cmd)
 576 {
 577         hammer2_msg_t *msg;
 578         size_t hbytes;
 579
 580         hbytes = (cmd & HAMMER2_MSGF_SIZE) * HAMMER2_MSG_ALIGN;
 581         msg = kmalloc(offsetof(struct hammer2_msg, any) + hbytes,
 582                       pmp->mmsg, M_WAITOK | M_ZERO);
 583         msg->hdr_size = hbytes;
 584         msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
 585         msg->any.head.spanid = spanid;
 586         msg->any.head.cmd = cmd;
 587
 588         return (msg);
 589 }
 590
 591 void
 592 hammer2_msg_free(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 593 {
 594         if (msg->aux_data && msg->aux_size) {
 595                 kfree(msg->aux_data, pmp->mmsg);
 596                 msg->aux_data = NULL;
 597                 msg->aux_size = 0;
 598         }
 599         kfree(msg, pmp->mmsg);
 600 }
 601
 602 /*
 603  * Indexed messages are stored in a red-black tree indexed by their
 604  * msgid.  Only persistent messages are indexed.
 605  */
 606 int
 607 hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
 608 {
 609         if (state1->spanid < state2->spanid)
 610                 return(-1);
 611         if (state1->spanid > state2->spanid)
 612                 return(1);
 613         if (state1->msgid < state2->msgid)
 614                 return(-1);
 615         if (state1->msgid > state2->msgid)
 616                 return(1);
 617         return(0);
 618 }
 619
 620 /*
 621  * Write a message.  All requisit command flags have been set.
 622  *
 623  * If msg->state is non-NULL the message is written to the existing
 624  * transaction.  Both msgid and spanid will be set accordingly.
 625  *
 626  * If msg->state is NULL and CREATE is set new state is allocated and
 627  * (func, data) is installed.
 628  *
 629  * If msg->state is NULL and CREATE is not set the message is assumed
 630  * to be a one-way message.  The msgid and spanid must be set by the
 631  * caller (msgid is typically set to 0 for this case).
 632  *
 633  * This function merely queues the message to the management thread, it
 634  * does not write to the message socket/pipe.
 635  */
 636 void
 637 hammer2_msg_write(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg,
 638                   int (*func)(hammer2_state_t *, hammer2_msg_t *), void *data)
 639 {
 640         hammer2_state_t *state;
 641
 642         if (msg->state) {
 643                 /*
 644                  * Continuance or termination of existing transaction.
 645                  * The transaction could have been initiated by either end.
 646                  *
 647                  * (Function callback and aux data for the receive side can
 648                  * be replaced or left alone).
 649                  */
 650                 msg->any.head.msgid = msg->state->msgid;
 651                 msg->any.head.spanid = msg->state->spanid;
 652                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 653                 if (func) {
 654                         msg->state->func = func;
 655                         msg->state->any.any = data;
 656                 }
 657         } else if (msg->any.head.cmd & HAMMER2_MSGF_CREATE) {
 658                 /*
 659                  * New transaction, requires tracking state and a unique
 660                  * msgid to be allocated.
 661                  */
 662                 KKASSERT(msg->state == NULL);
 663                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
 664                 state->pmp = pmp;
 665                 state->flags = HAMMER2_STATE_DYNAMIC;
 666                 state->func = func;
 667                 state->any.any = data;
 668                 state->msg = msg;
 669                 state->msgid = (uint64_t)(uintptr_t)state;
 670                 state->spanid = msg->any.head.spanid;
 671                 msg->state = state;
 672
 673                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 674                 if (RB_INSERT(hammer2_state_tree, &pmp->statewr_tree, state))
 675                         panic("duplicate msgid allocated");
 676                 msg->any.head.msgid = state->msgid;
 677         } else {
 678                 /*
 679                  * One-off message (always uses msgid 0 to distinguish
 680                  * between a possibly lost in-transaction message due to
 681                  * competing aborts and a real one-off message?)
 682                  */
 683                 msg->any.head.msgid = 0;
 684                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 685         }
 686
 687         /*
 688          * Finish up the msg fields
 689          */
 690         msg->any.head.salt = /* (random << 8) | */ (pmp->msg_seq & 255);
 691         ++pmp->msg_seq;
 692
 693         msg->any.head.hdr_crc = 0;
 694         msg->any.head.hdr_crc = hammer2_icrc32(msg->any.buf, msg->hdr_size);
 695
 696         TAILQ_INSERT_TAIL(&pmp->msgq, msg, qentry);
 697         lockmgr(&pmp->msglk, LK_RELEASE);
 698 }
 699
 700 /*
 701  * Reply to a message and terminate our side of the transaction.
 702  *
 703  * If msg->state is non-NULL we are replying to a one-way message.
 704  */
 705 void
 706 hammer2_msg_reply(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg, uint32_t error)
 707 {
 708         hammer2_state_t *state = msg->state;
 709         uint32_t cmd;
 710
 711         /*
 712          * Reply with a simple error code and terminate the transaction.
 713          */
 714         cmd = HAMMER2_LNK_ERROR;
 715
 716         /*
 717          * Check if our direction has even been initiated yet, set CREATE.
 718          *
 719          * Check what direction this is (command or reply direction).  Note
 720          * that txcmd might not have been initiated yet.
 721          *
 722          * If our direction has already been closed we just return without
 723          * doing anything.
 724          */
 725         if (state) {
 726                 if (state->txcmd & HAMMER2_MSGF_DELETE)
 727                         return;
 728                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
 729                         cmd |= HAMMER2_MSGF_CREATE;
 730                 if ((state->rxcmd & HAMMER2_MSGF_REPLY) == 0)
 731                         cmd |= HAMMER2_MSGF_REPLY;
 732                 cmd |= HAMMER2_MSGF_DELETE;
 733         } else {
 734                 if ((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0)
 735                         cmd |= HAMMER2_MSGF_REPLY;
 736         }
 737
 738         msg = hammer2_msg_alloc(pmp, 0, cmd);
 739         msg->any.head.error = error;
 740         hammer2_msg_write(pmp, msg, NULL, NULL);
 741 }
 742
 743 /*
 744  * Reply to a message and continue our side of the transaction.
 745  *
 746  * If msg->state is non-NULL we are replying to a one-way message and this
 747  * function degenerates into the same as hammer2_msg_reply().
 748  */
 749 void
 750 hammer2_msg_result(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg, uint32_t error)
 751 {
 752         hammer2_state_t *state = msg->state;
 753         uint32_t cmd;
 754
 755         /*
 756          * Return a simple result code, do NOT terminate the transaction.
 757          */
 758         cmd = HAMMER2_LNK_ERROR;
 759
 760         /*
 761          * Check if our direction has even been initiated yet, set CREATE.
 762          *
 763          * Check what direction this is (command or reply direction).  Note
 764          * that txcmd might not have been initiated yet.
 765          *
 766          * If our direction has already been closed we just return without
 767          * doing anything.
 768          */
 769         if (state) {
 770                 if (state->txcmd & HAMMER2_MSGF_DELETE)
 771                         return;
 772                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
 773                         cmd |= HAMMER2_MSGF_CREATE;
 774                 if ((state->rxcmd & HAMMER2_MSGF_REPLY) == 0)
 775                         cmd |= HAMMER2_MSGF_REPLY;
 776                 /* continuing transaction, do not set MSGF_DELETE */
 777         } else {
 778                 if ((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0)
 779                         cmd |= HAMMER2_MSGF_REPLY;
 780         }
 781
 782         msg = hammer2_msg_alloc(pmp, 0, cmd);
 783         msg->any.head.error = error;
 784         hammer2_msg_write(pmp, msg, NULL, NULL);
 785 }