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->source = msg->any.head.source;
 134         state->target = msg->any.head.target;
 135         kprintf("received msg %08x msgid %u source=%u target=%u\n",
 136                 msg->any.head.cmd, msg->any.head.msgid, msg->any.head.source,
 137                 msg->any.head.target);
 138         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY)
 139                 state = RB_FIND(hammer2_state_tree, &pmp->statewr_tree, state);
 140         else
 141                 state = RB_FIND(hammer2_state_tree, &pmp->staterd_tree, state);
 142         msg->state = state;
 143
 144         /*
 145          * Short-cut one-off or mid-stream messages (state may be NULL).
 146          */
 147         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 148                                   HAMMER2_MSGF_ABORT)) == 0) {
 149                 lockmgr(&pmp->msglk, LK_RELEASE);
 150                 return(0);
 151         }
 152
 153         /*
 154          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
 155          * inside the case statements.
 156          */
 157         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 158                                     HAMMER2_MSGF_REPLY)) {
 159         case HAMMER2_MSGF_CREATE:
 160         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 161                 /*
 162                  * New persistant command received.
 163                  */
 164                 if (state) {
 165                         kprintf("hammer2_state_msgrx: duplicate transaction\n");
 166                         error = EINVAL;
 167                         break;
 168                 }
 169                 state = pmp->freerd_state;
 170                 pmp->freerd_state = NULL;
 171                 msg->state = state;
 172                 state->msg = msg;
 173                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 174                 RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state);
 175                 state->flags |= HAMMER2_STATE_INSERTED;
 176                 error = 0;
 177                 break;
 178         case HAMMER2_MSGF_DELETE:
 179                 /*
 180                  * Persistent state is expected but might not exist if an
 181                  * ABORT+DELETE races the close.
 182                  */
 183                 if (state == NULL) {
 184                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 185                                 error = EALREADY;
 186                         } else {
 187                                 kprintf("hammer2_state_msgrx: no state "
 188                                         "for DELETE\n");
 189                                 error = EINVAL;
 190                         }
 191                         break;
 192                 }
 193
 194                 /*
 195                  * Handle another ABORT+DELETE case if the msgid has already
 196                  * been reused.
 197                  */
 198                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 199                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 200                                 error = EALREADY;
 201                         } else {
 202                                 kprintf("hammer2_state_msgrx: state reused "
 203                                         "for DELETE\n");
 204                                 error = EINVAL;
 205                         }
 206                         break;
 207                 }
 208                 error = 0;
 209                 break;
 210         default:
 211                 /*
 212                  * Check for mid-stream ABORT command received, otherwise
 213                  * allow.
 214                  */
 215                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 216                         if (state == NULL ||
 217                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 218                                 error = EALREADY;
 219                                 break;
 220                         }
 221                 }
 222                 error = 0;
 223                 break;
 224         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
 225         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 226                 /*
 227                  * When receiving a reply with CREATE set the original
 228                  * persistent state message should already exist.
 229                  */
 230                 if (state == NULL) {
 231                         kprintf("hammer2_state_msgrx: no state match for "
 232                                 "REPLY cmd=%08x\n", msg->any.head.cmd);
 233                         error = EINVAL;
 234                         break;
 235                 }
 236                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 237                 error = 0;
 238                 break;
 239         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
 240                 /*
 241                  * Received REPLY+ABORT+DELETE in case where msgid has
 242                  * already been fully closed, ignore the message.
 243                  */
 244                 if (state == NULL) {
 245                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 246                                 error = EALREADY;
 247                         } else {
 248                                 kprintf("hammer2_state_msgrx: no state match "
 249                                         "for REPLY|DELETE\n");
 250                                 error = EINVAL;
 251                         }
 252                         break;
 253                 }
 254
 255                 /*
 256                  * Received REPLY+ABORT+DELETE in case where msgid has
 257                  * already been reused for an unrelated message,
 258                  * ignore the message.
 259                  */
 260                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 261                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 262                                 error = EALREADY;
 263                         } else {
 264                                 kprintf("hammer2_state_msgrx: state reused "
 265                                         "for REPLY|DELETE\n");
 266                                 error = EINVAL;
 267                         }
 268                         break;
 269                 }
 270                 error = 0;
 271                 break;
 272         case HAMMER2_MSGF_REPLY:
 273                 /*
 274                  * Check for mid-stream ABORT reply received to sent command.
 275                  */
 276                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 277                         if (state == NULL ||
 278                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
 279                                 error = EALREADY;
 280                                 break;
 281                         }
 282                 }
 283                 error = 0;
 284                 break;
 285         }
 286         lockmgr(&pmp->msglk, LK_RELEASE);
 287         return (error);
 288 }
 289
 290 void
 291 hammer2_state_cleanuprx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 292 {
 293         hammer2_state_t *state;
 294
 295         if ((state = msg->state) == NULL) {
 296                 hammer2_msg_free(pmp, msg);
 297         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
 298                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 299                 state->rxcmd |= HAMMER2_MSGF_DELETE;
 300                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
 301                         if (state->msg == msg)
 302                                 state->msg = NULL;
 303                         KKASSERT(state->flags & HAMMER2_STATE_INSERTED);
 304                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
 305                                 RB_REMOVE(hammer2_state_tree,
 306                                           &pmp->statewr_tree, state);
 307                         } else {
 308                                 RB_REMOVE(hammer2_state_tree,
 309                                           &pmp->staterd_tree, state);
 310                         }
 311                         state->flags &= ~HAMMER2_STATE_INSERTED;
 312                         lockmgr(&pmp->msglk, LK_RELEASE);
 313                         hammer2_state_free(state);
 314                 } else {
 315                         lockmgr(&pmp->msglk, LK_RELEASE);
 316                 }
 317                 hammer2_msg_free(pmp, msg);
 318         } else if (state->msg != msg) {
 319                 hammer2_msg_free(pmp, msg);
 320         }
 321 }
 322
 323 /*
 324  * Process state tracking for a message prior to transmission.
 325  *
 326  * Called with msglk held and the msg dequeued.
 327  *
 328  * One-off messages are usually with dummy state and msg->state may be NULL
 329  * in this situation.
 330  *
 331  * New transactions (when CREATE is set) will insert the state.
 332  *
 333  * May request that caller discard the message by setting *discardp to 1.
 334  * A NULL state may be returned in this case.
 335  */
 336 int
 337 hammer2_state_msgtx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 338 {
 339         hammer2_state_t *state;
 340         int error;
 341
 342         /*
 343          * Make sure a state structure is ready to go in case we need a new
 344          * one.  This is the only routine which uses freewr_state so no
 345          * races are possible.
 346          */
 347         if ((state = pmp->freewr_state) == NULL) {
 348                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
 349                 state->pmp = pmp;
 350                 state->flags = HAMMER2_STATE_DYNAMIC;
 351                 pmp->freewr_state = state;
 352         }
 353
 354         /*
 355          * Lock RB tree.  If persistent state is present it will have already
 356          * been assigned to msg.
 357          */
 358         lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 359         state = msg->state;
 360
 361         /*
 362          * Short-cut one-off or mid-stream messages (state may be NULL).
 363          */
 364         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 365                                   HAMMER2_MSGF_ABORT)) == 0) {
 366                 lockmgr(&pmp->msglk, LK_RELEASE);
 367                 return(0);
 368         }
 369
 370
 371         /*
 372          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
 373          * inside the case statements.
 374          */
 375         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
 376                                     HAMMER2_MSGF_REPLY)) {
 377         case HAMMER2_MSGF_CREATE:
 378         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 379                 /*
 380                  * Insert the new persistent message state and mark
 381                  * half-closed if DELETE is set.  Since this is a new
 382                  * message it isn't possible to transition into the fully
 383                  * closed state here.
 384                  *
 385                  * XXX state must be assigned and inserted by
 386                  *     hammer2_msg_write().  txcmd is assigned by us
 387                  *     on-transmit.
 388                  */
 389                 KKASSERT(state != NULL);
 390 #if 0
 391                 if (state == NULL) {
 392                         state = pmp->freerd_state;
 393                         pmp->freerd_state = NULL;
 394                         msg->state = state;
 395                         state->msg = msg;
 396                         state->msgid = msg->any.head.msgid;
 397                         state->source = msg->any.head.source;
 398                         state->target = msg->any.head.target;
 399                 }
 400                 KKASSERT((state->flags & HAMMER2_STATE_INSERTED) == 0);
 401                 if (RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state)) {
 402                         kprintf("hammer2_state_msgtx: duplicate transaction\n");
 403                         error = EINVAL;
 404                         break;
 405                 }
 406                 state->flags |= HAMMER2_STATE_INSERTED;
 407 #endif
 408                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 409                 error = 0;
 410                 break;
 411         case HAMMER2_MSGF_DELETE:
 412                 /*
 413                  * Sent ABORT+DELETE in case where msgid has already
 414                  * been fully closed, ignore the message.
 415                  */
 416                 if (state == NULL) {
 417                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 418                                 error = EALREADY;
 419                         } else {
 420                                 kprintf("hammer2_state_msgtx: no state match "
 421                                         "for DELETE\n");
 422                                 error = EINVAL;
 423                         }
 424                         break;
 425                 }
 426
 427                 /*
 428                  * Sent ABORT+DELETE in case where msgid has
 429                  * already been reused for an unrelated message,
 430                  * ignore the message.
 431                  */
 432                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 433                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 434                                 error = EALREADY;
 435                         } else {
 436                                 kprintf("hammer2_state_msgtx: state reused "
 437                                         "for DELETE\n");
 438                                 error = EINVAL;
 439                         }
 440                         break;
 441                 }
 442                 error = 0;
 443                 break;
 444         default:
 445                 /*
 446                  * Check for mid-stream ABORT command sent
 447                  */
 448                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 449                         if (state == NULL ||
 450                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 451                                 error = EALREADY;
 452                                 break;
 453                         }
 454                 }
 455                 error = 0;
 456                 break;
 457         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
 458         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
 459                 /*
 460                  * When transmitting a reply with CREATE set the original
 461                  * persistent state message should already exist.
 462                  */
 463                 if (state == NULL) {
 464                         kprintf("hammer2_state_msgtx: no state match "
 465                                 "for REPLY | CREATE\n");
 466                         error = EINVAL;
 467                         break;
 468                 }
 469                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
 470                 error = 0;
 471                 break;
 472         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
 473                 /*
 474                  * When transmitting a reply with DELETE set the original
 475                  * persistent state message should already exist.
 476                  *
 477                  * This is very similar to the REPLY|CREATE|* case except
 478                  * txcmd is already stored, so we just add the DELETE flag.
 479                  *
 480                  * Sent REPLY+ABORT+DELETE in case where msgid has
 481                  * already been fully closed, ignore the message.
 482                  */
 483                 if (state == NULL) {
 484                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 485                                 error = EALREADY;
 486                         } else {
 487                                 kprintf("hammer2_state_msgtx: no state match "
 488                                         "for REPLY | DELETE\n");
 489                                 error = EINVAL;
 490                         }
 491                         break;
 492                 }
 493
 494                 /*
 495                  * Sent REPLY+ABORT+DELETE in case where msgid has already
 496                  * been reused for an unrelated message, ignore the message.
 497                  */
 498                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 499                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 500                                 error = EALREADY;
 501                         } else {
 502                                 kprintf("hammer2_state_msgtx: state reused "
 503                                         "for REPLY | DELETE\n");
 504                                 error = EINVAL;
 505                         }
 506                         break;
 507                 }
 508                 error = 0;
 509                 break;
 510         case HAMMER2_MSGF_REPLY:
 511                 /*
 512                  * Check for mid-stream ABORT reply sent.
 513                  *
 514                  * One-off REPLY messages are allowed for e.g. status updates.
 515                  */
 516                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
 517                         if (state == NULL ||
 518                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
 519                                 error = EALREADY;
 520                                 break;
 521                         }
 522                 }
 523                 error = 0;
 524                 break;
 525         }
 526         lockmgr(&pmp->msglk, LK_RELEASE);
 527         return (error);
 528 }
 529
 530 void
 531 hammer2_state_cleanuptx(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 532 {
 533         hammer2_state_t *state;
 534
 535         if ((state = msg->state) == NULL) {
 536                 hammer2_msg_free(pmp, msg);
 537         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
 538                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 539                 state->txcmd |= HAMMER2_MSGF_DELETE;
 540                 if (state->rxcmd & HAMMER2_MSGF_DELETE) {
 541                         if (state->msg == msg)
 542                                 state->msg = NULL;
 543                         KKASSERT(state->flags & HAMMER2_STATE_INSERTED);
 544                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
 545                                 RB_REMOVE(hammer2_state_tree,
 546                                           &pmp->staterd_tree, state);
 547                         } else {
 548                                 RB_REMOVE(hammer2_state_tree,
 549                                           &pmp->statewr_tree, state);
 550                         }
 551                         state->flags &= ~HAMMER2_STATE_INSERTED;
 552                         lockmgr(&pmp->msglk, LK_RELEASE);
 553                         hammer2_state_free(state);
 554                 } else {
 555                         lockmgr(&pmp->msglk, LK_RELEASE);
 556                 }
 557                 hammer2_msg_free(pmp, msg);
 558         } else if (state->msg != msg) {
 559                 hammer2_msg_free(pmp, msg);
 560         }
 561 }
 562
 563 void
 564 hammer2_state_free(hammer2_state_t *state)
 565 {
 566         hammer2_pfsmount_t *pmp = state->pmp;
 567         hammer2_msg_t *msg;
 568
 569         msg = state->msg;
 570         state->msg = NULL;
 571         kfree(state, pmp->mmsg);
 572         if (msg)
 573                 hammer2_msg_free(pmp, msg);
 574 }
 575
 576 hammer2_msg_t *
 577 hammer2_msg_alloc(hammer2_pfsmount_t *pmp, uint16_t source, uint16_t target,
 578                   uint32_t cmd)
 579 {
 580         hammer2_msg_t *msg;
 581         size_t hbytes;
 582
 583         hbytes = (cmd & HAMMER2_MSGF_SIZE) * HAMMER2_MSG_ALIGN;
 584         msg = kmalloc(offsetof(struct hammer2_msg, any) + hbytes,
 585                       pmp->mmsg, M_WAITOK | M_ZERO);
 586         msg->hdr_size = hbytes;
 587         msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
 588         msg->any.head.source = source;
 589         msg->any.head.target = target;
 590         msg->any.head.cmd = cmd;
 591
 592         return (msg);
 593 }
 594
 595 void
 596 hammer2_msg_free(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg)
 597 {
 598         if (msg->aux_data && msg->aux_size) {
 599                 kfree(msg->aux_data, pmp->mmsg);
 600                 msg->aux_data = NULL;
 601                 msg->aux_size = 0;
 602         }
 603         kfree(msg, pmp->mmsg);
 604 }
 605
 606 /*
 607  * Indexed messages are stored in a red-black tree indexed by their
 608  * msgid.  Only persistent messages are indexed.
 609  */
 610 int
 611 hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
 612 {
 613         if (state1->source < state2->source)
 614                 return(-1);
 615         if (state1->source > state2->source)
 616                 return(1);
 617         if (state1->target < state2->target)
 618                 return(-1);
 619         if (state1->target > state2->target)
 620                 return(1);
 621         if (state1->msgid < state2->msgid)
 622                 return(-1);
 623         if (state1->msgid > state2->msgid)
 624                 return(1);
 625         return(0);
 626 }
 627
 628 /*
 629  * Write a message.  {source, target, cmd} have been set.  This function
 630  * merely queues the message to the management thread, it does not write
 631  * to the message socket/pipe.
 632  *
 633  * If CREATE is set we allocate the state and msgid and do the insertion.
 634  * If CREATE is not set the state and msgid must already be assigned.
 635  */
 636 hammer2_state_t *
 637 hammer2_msg_write(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg,
 638                   int (*func)(hammer2_pfsmount_t *, hammer2_msg_t *))
 639 {
 640         hammer2_state_t *state;
 641         uint16_t xcrc16;
 642         uint32_t xcrc32;
 643
 644         /*
 645          * Setup transaction (if applicable).  One-off messages always
 646          * use a msgid of 0.
 647          */
 648         if (msg->any.head.cmd & HAMMER2_MSGF_CREATE) {
 649                 /*
 650                  * New transaction, requires tracking state and a unique
 651                  * msgid.
 652                  */
 653                 KKASSERT(msg->state == NULL);
 654                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
 655                 state->pmp = pmp;
 656                 state->flags = HAMMER2_STATE_DYNAMIC;
 657                 state->func = func;
 658                 state->msg = msg;
 659                 state->source = msg->any.head.source;
 660                 state->target = msg->any.head.target;
 661                 msg->state = state;
 662
 663                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 664                 if ((state->msgid = pmp->msgid_iterator++) == 0)
 665                         state->msgid = pmp->msgid_iterator++;
 666                 while (RB_INSERT(hammer2_state_tree,
 667                                  &pmp->statewr_tree, state)) {
 668                         if ((state->msgid = pmp->msgid_iterator++) == 0)
 669                                 state->msgid = pmp->msgid_iterator++;
 670                 }
 671                 msg->any.head.msgid = state->msgid;
 672         } else if (msg->state) {
 673                 /*
 674                  * Continuance or termination
 675                  */
 676                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 677         } else {
 678                 /*
 679                  * One-off message (always uses msgid 0)
 680                  */
 681                 msg->any.head.msgid = 0;
 682                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
 683         }
 684
 685         /*
 686          * Set icrc2 and icrc1
 687          */
 688         if (msg->hdr_size > sizeof(msg->any.head)) {
 689                 xcrc32 = hammer2_icrc32(&msg->any.head + 1,
 690                                         msg->hdr_size - sizeof(msg->any.head));
 691                 xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
 692                 msg->any.head.icrc2 = xcrc16;
 693         }
 694         xcrc32 = hammer2_icrc32(msg->any.buf + HAMMER2_MSGHDR_CRCOFF,
 695                                 HAMMER2_MSGHDR_CRCBYTES);
 696         xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
 697         msg->any.head.icrc1 = xcrc16;
 698
 699         TAILQ_INSERT_TAIL(&pmp->msgq, msg, qentry);
 700         lockmgr(&pmp->msglk, LK_RELEASE);
 701
 702         return (msg->state);
 703 }