Merge branches 'hammer2' and 'master' of ssh://crater.dragonflybsd.org/repository...
[dragonfly.git] / sys / vfs / hammer2 / hammer2_msg.c
CommitLineData
26bf1a36
MD
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
9b8b748f
MD
37RB_GENERATE(hammer2_state_tree, hammer2_state, rbnode, hammer2_state_cmp);
38
26bf1a36
MD
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 */
106int
107hammer2_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;
9b8b748f
MD
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);
26bf1a36
MD
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 "
9b8b748f 232 "REPLY cmd=%08x\n", msg->any.head.cmd);
26bf1a36
MD
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
290void
291hammer2_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 */
336int
337hammer2_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.
9b8b748f
MD
384 *
385 * XXX state must be assigned and inserted by
386 * hammer2_msg_write(). txcmd is assigned by us
387 * on-transmit.
26bf1a36 388 */
9b8b748f
MD
389 KKASSERT(state != NULL);
390#if 0
26bf1a36
MD
391 if (state == NULL) {
392 state = pmp->freerd_state;
393 pmp->freerd_state = NULL;
394 msg->state = state;
395 state->msg = msg;
9b8b748f
MD
396 state->msgid = msg->any.head.msgid;
397 state->source = msg->any.head.source;
398 state->target = msg->any.head.target;
26bf1a36
MD
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;
9b8b748f 407#endif
26bf1a36
MD
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
530void
531hammer2_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
563void
564hammer2_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
9b8b748f
MD
576hammer2_msg_t *
577hammer2_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
26bf1a36
MD
595void
596hammer2_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 */
610int
611hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
612{
9b8b748f
MD
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);
26bf1a36
MD
621 if (state1->msgid < state2->msgid)
622 return(-1);
623 if (state1->msgid > state2->msgid)
624 return(1);
625 return(0);
626}
627
628/*
42e2a62e
MD
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.
26bf1a36 632 *
9b8b748f
MD
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.
26bf1a36 635 */
9b8b748f
MD
636hammer2_state_t *
637hammer2_msg_write(hammer2_pfsmount_t *pmp, hammer2_msg_t *msg,
638 int (*func)(hammer2_pfsmount_t *, hammer2_msg_t *))
26bf1a36 639{
9b8b748f
MD
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);
26bf1a36 703}