ef84f340a1401a54cac2f28fee7b5a645b2cddbb
[dragonfly.git] / 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_msg_t *msg)
108 {
109         hammer2_pfsmount_t *pmp;
110         hammer2_state_t *state;
111         int error;
112
113         pmp = msg->router->pmp;
114
115         /*
116          * XXX resolve msg->any.head.source and msg->any.head.target
117          *     into LNK_SPAN references.
118          *
119          * XXX replace msg->router
120          */
121
122         /*
123          * Make sure a state structure is ready to go in case we need a new
124          * one.  This is the only routine which uses freerd_state so no
125          * races are possible.
126          */
127         if ((state = pmp->freerd_state) == NULL) {
128                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
129                 state->pmp = pmp;
130                 state->flags = HAMMER2_STATE_DYNAMIC;
131                 pmp->freerd_state = state;
132         }
133
134         /*
135          * Lock RB tree and locate existing persistent state, if any.
136          *
137          * If received msg is a command state is on staterd_tree.
138          * If received msg is a reply state is on statewr_tree.
139          */
140         lockmgr(&pmp->msglk, LK_EXCLUSIVE);
141
142         state->msgid = msg->any.head.msgid;
143         state->router = &pmp->router;
144         kprintf("received msg %08x msgid %jx source=%jx target=%jx\n",
145                 msg->any.head.cmd,
146                 (intmax_t)msg->any.head.msgid,
147                 (intmax_t)msg->any.head.source,
148                 (intmax_t)msg->any.head.target);
149         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY)
150                 state = RB_FIND(hammer2_state_tree, &pmp->statewr_tree, state);
151         else
152                 state = RB_FIND(hammer2_state_tree, &pmp->staterd_tree, state);
153         msg->state = state;
154
155         /*
156          * Short-cut one-off or mid-stream messages (state may be NULL).
157          */
158         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
159                                   HAMMER2_MSGF_ABORT)) == 0) {
160                 lockmgr(&pmp->msglk, LK_RELEASE);
161                 return(0);
162         }
163
164         /*
165          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
166          * inside the case statements.
167          */
168         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
169                                     HAMMER2_MSGF_REPLY)) {
170         case HAMMER2_MSGF_CREATE:
171         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
172                 /*
173                  * New persistant command received.
174                  */
175                 if (state) {
176                         kprintf("hammer2_state_msgrx: duplicate transaction\n");
177                         error = EINVAL;
178                         break;
179                 }
180                 state = pmp->freerd_state;
181                 pmp->freerd_state = NULL;
182                 msg->state = state;
183                 state->router = msg->router;
184                 state->msg = msg;
185                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
186                 state->txcmd = HAMMER2_MSGF_REPLY;
187                 RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state);
188                 state->flags |= HAMMER2_STATE_INSERTED;
189                 error = 0;
190                 break;
191         case HAMMER2_MSGF_DELETE:
192                 /*
193                  * Persistent state is expected but might not exist if an
194                  * ABORT+DELETE races the close.
195                  */
196                 if (state == NULL) {
197                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
198                                 error = EALREADY;
199                         } else {
200                                 kprintf("hammer2_state_msgrx: no state "
201                                         "for DELETE\n");
202                                 error = EINVAL;
203                         }
204                         break;
205                 }
206
207                 /*
208                  * Handle another ABORT+DELETE case if the msgid has already
209                  * been reused.
210                  */
211                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
212                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
213                                 error = EALREADY;
214                         } else {
215                                 kprintf("hammer2_state_msgrx: state reused "
216                                         "for DELETE\n");
217                                 error = EINVAL;
218                         }
219                         break;
220                 }
221                 error = 0;
222                 break;
223         default:
224                 /*
225                  * Check for mid-stream ABORT command received, otherwise
226                  * allow.
227                  */
228                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
229                         if (state == NULL ||
230                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
231                                 error = EALREADY;
232                                 break;
233                         }
234                 }
235                 error = 0;
236                 break;
237         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
238         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
239                 /*
240                  * When receiving a reply with CREATE set the original
241                  * persistent state message should already exist.
242                  */
243                 if (state == NULL) {
244                         kprintf("hammer2_state_msgrx: no state match for "
245                                 "REPLY cmd=%08x msgid=%016jx\n",
246                                 msg->any.head.cmd,
247                                 (intmax_t)msg->any.head.msgid);
248                         error = EINVAL;
249                         break;
250                 }
251                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
252                 error = 0;
253                 break;
254         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
255                 /*
256                  * Received REPLY+ABORT+DELETE in case where msgid has
257                  * already been fully closed, ignore the message.
258                  */
259                 if (state == NULL) {
260                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
261                                 error = EALREADY;
262                         } else {
263                                 kprintf("hammer2_state_msgrx: no state match "
264                                         "for REPLY|DELETE\n");
265                                 error = EINVAL;
266                         }
267                         break;
268                 }
269
270                 /*
271                  * Received REPLY+ABORT+DELETE in case where msgid has
272                  * already been reused for an unrelated message,
273                  * ignore the message.
274                  */
275                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
276                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
277                                 error = EALREADY;
278                         } else {
279                                 kprintf("hammer2_state_msgrx: state reused "
280                                         "for REPLY|DELETE\n");
281                                 error = EINVAL;
282                         }
283                         break;
284                 }
285                 error = 0;
286                 break;
287         case HAMMER2_MSGF_REPLY:
288                 /*
289                  * Check for mid-stream ABORT reply received to sent command.
290                  */
291                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
292                         if (state == NULL ||
293                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
294                                 error = EALREADY;
295                                 break;
296                         }
297                 }
298                 error = 0;
299                 break;
300         }
301         lockmgr(&pmp->msglk, LK_RELEASE);
302         return (error);
303 }
304
305 void
306 hammer2_state_cleanuprx(hammer2_msg_t *msg)
307 {
308         hammer2_pfsmount_t *pmp = msg->router->pmp;
309         hammer2_state_t *state;
310
311         if ((state = msg->state) == NULL) {
312                 hammer2_msg_free(msg);
313         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
314                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
315                 state->rxcmd |= HAMMER2_MSGF_DELETE;
316                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
317                         if (state->msg == msg)
318                                 state->msg = NULL;
319                         KKASSERT(state->flags & HAMMER2_STATE_INSERTED);
320                         if (state->rxcmd & HAMMER2_MSGF_REPLY) {
321                                 KKASSERT(msg->any.head.cmd &
322                                          HAMMER2_MSGF_REPLY);
323                                 RB_REMOVE(hammer2_state_tree,
324                                           &pmp->statewr_tree, state);
325                         } else {
326                                 KKASSERT((msg->any.head.cmd &
327                                           HAMMER2_MSGF_REPLY) == 0);
328                                 RB_REMOVE(hammer2_state_tree,
329                                           &pmp->staterd_tree, state);
330                         }
331                         state->flags &= ~HAMMER2_STATE_INSERTED;
332                         lockmgr(&pmp->msglk, LK_RELEASE);
333                         hammer2_state_free(state);
334                 } else {
335                         lockmgr(&pmp->msglk, LK_RELEASE);
336                 }
337                 hammer2_msg_free(msg);
338         } else if (state->msg != msg) {
339                 hammer2_msg_free(msg);
340         }
341 }
342
343 /*
344  * Process state tracking for a message prior to transmission.
345  *
346  * Called with msglk held and the msg dequeued.
347  *
348  * One-off messages are usually with dummy state and msg->state may be NULL
349  * in this situation.
350  *
351  * New transactions (when CREATE is set) will insert the state.
352  *
353  * May request that caller discard the message by setting *discardp to 1.
354  * A NULL state may be returned in this case.
355  */
356 int
357 hammer2_state_msgtx(hammer2_msg_t *msg)
358 {
359         hammer2_pfsmount_t *pmp = msg->router->pmp;
360         hammer2_state_t *state;
361         int error;
362
363         /*
364          * Make sure a state structure is ready to go in case we need a new
365          * one.  This is the only routine which uses freewr_state so no
366          * races are possible.
367          */
368         if ((state = pmp->freewr_state) == NULL) {
369                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
370                 state->pmp = pmp;
371                 state->flags = HAMMER2_STATE_DYNAMIC;
372                 pmp->freewr_state = state;
373         }
374
375         /*
376          * Lock RB tree.  If persistent state is present it will have already
377          * been assigned to msg.
378          */
379         lockmgr(&pmp->msglk, LK_EXCLUSIVE);
380         state = msg->state;
381
382         /*
383          * Short-cut one-off or mid-stream messages (state may be NULL).
384          */
385         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
386                                   HAMMER2_MSGF_ABORT)) == 0) {
387                 lockmgr(&pmp->msglk, LK_RELEASE);
388                 return(0);
389         }
390
391
392         /*
393          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
394          * inside the case statements.
395          */
396         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
397                                     HAMMER2_MSGF_REPLY)) {
398         case HAMMER2_MSGF_CREATE:
399         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
400                 /*
401                  * Insert the new persistent message state and mark
402                  * half-closed if DELETE is set.  Since this is a new
403                  * message it isn't possible to transition into the fully
404                  * closed state here.
405                  *
406                  * XXX state must be assigned and inserted by
407                  *     hammer2_msg_write().  txcmd is assigned by us
408                  *     on-transmit.
409                  */
410                 KKASSERT(state != NULL);
411                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
412                 state->rxcmd = HAMMER2_MSGF_REPLY;
413                 error = 0;
414                 break;
415         case HAMMER2_MSGF_DELETE:
416                 /*
417                  * Sent ABORT+DELETE in case where msgid has already
418                  * been fully closed, ignore the message.
419                  */
420                 if (state == NULL) {
421                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
422                                 error = EALREADY;
423                         } else {
424                                 kprintf("hammer2_state_msgtx: no state match "
425                                         "for DELETE cmd=%08x msgid=%016jx\n",
426                                         msg->any.head.cmd,
427                                         (intmax_t)msg->any.head.msgid);
428                                 error = EINVAL;
429                         }
430                         break;
431                 }
432
433                 /*
434                  * Sent ABORT+DELETE in case where msgid has
435                  * already been reused for an unrelated message,
436                  * ignore the message.
437                  */
438                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
439                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
440                                 error = EALREADY;
441                         } else {
442                                 kprintf("hammer2_state_msgtx: state reused "
443                                         "for DELETE\n");
444                                 error = EINVAL;
445                         }
446                         break;
447                 }
448                 error = 0;
449                 break;
450         default:
451                 /*
452                  * Check for mid-stream ABORT command sent
453                  */
454                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
455                         if (state == NULL ||
456                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
457                                 error = EALREADY;
458                                 break;
459                         }
460                 }
461                 error = 0;
462                 break;
463         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
464         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
465                 /*
466                  * When transmitting a reply with CREATE set the original
467                  * persistent state message should already exist.
468                  */
469                 if (state == NULL) {
470                         kprintf("hammer2_state_msgtx: no state match "
471                                 "for REPLY | CREATE\n");
472                         error = EINVAL;
473                         break;
474                 }
475                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
476                 error = 0;
477                 break;
478         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
479                 /*
480                  * When transmitting a reply with DELETE set the original
481                  * persistent state message should already exist.
482                  *
483                  * This is very similar to the REPLY|CREATE|* case except
484                  * txcmd is already stored, so we just add the DELETE flag.
485                  *
486                  * Sent REPLY+ABORT+DELETE in case where msgid has
487                  * already been fully closed, ignore the message.
488                  */
489                 if (state == NULL) {
490                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
491                                 error = EALREADY;
492                         } else {
493                                 kprintf("hammer2_state_msgtx: no state match "
494                                         "for REPLY | DELETE\n");
495                                 error = EINVAL;
496                         }
497                         break;
498                 }
499
500                 /*
501                  * Sent REPLY+ABORT+DELETE in case where msgid has already
502                  * been reused for an unrelated message, ignore the message.
503                  */
504                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
505                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
506                                 error = EALREADY;
507                         } else {
508                                 kprintf("hammer2_state_msgtx: state reused "
509                                         "for REPLY | DELETE\n");
510                                 error = EINVAL;
511                         }
512                         break;
513                 }
514                 error = 0;
515                 break;
516         case HAMMER2_MSGF_REPLY:
517                 /*
518                  * Check for mid-stream ABORT reply sent.
519                  *
520                  * One-off REPLY messages are allowed for e.g. status updates.
521                  */
522                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
523                         if (state == NULL ||
524                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
525                                 error = EALREADY;
526                                 break;
527                         }
528                 }
529                 error = 0;
530                 break;
531         }
532         lockmgr(&pmp->msglk, LK_RELEASE);
533         return (error);
534 }
535
536 void
537 hammer2_state_cleanuptx(hammer2_msg_t *msg)
538 {
539         hammer2_pfsmount_t *pmp = msg->router->pmp;
540         hammer2_state_t *state;
541
542         if ((state = msg->state) == NULL) {
543                 hammer2_msg_free(msg);
544         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
545                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
546                 state->txcmd |= HAMMER2_MSGF_DELETE;
547                 if (state->rxcmd & HAMMER2_MSGF_DELETE) {
548                         if (state->msg == msg)
549                                 state->msg = NULL;
550                         KKASSERT(state->flags & HAMMER2_STATE_INSERTED);
551                         if (state->txcmd & HAMMER2_MSGF_REPLY) {
552                                 KKASSERT(msg->any.head.cmd &
553                                          HAMMER2_MSGF_REPLY);
554                                 RB_REMOVE(hammer2_state_tree,
555                                           &pmp->staterd_tree, state);
556                         } else {
557                                 KKASSERT((msg->any.head.cmd &
558                                           HAMMER2_MSGF_REPLY) == 0);
559                                 RB_REMOVE(hammer2_state_tree,
560                                           &pmp->statewr_tree, state);
561                         }
562                         state->flags &= ~HAMMER2_STATE_INSERTED;
563                         lockmgr(&pmp->msglk, LK_RELEASE);
564                         hammer2_state_free(state);
565                 } else {
566                         lockmgr(&pmp->msglk, LK_RELEASE);
567                 }
568                 hammer2_msg_free(msg);
569         } else if (state->msg != msg) {
570                 hammer2_msg_free(msg);
571         }
572 }
573
574 void
575 hammer2_state_free(hammer2_state_t *state)
576 {
577         hammer2_pfsmount_t *pmp = state->pmp;
578         hammer2_msg_t *msg;
579
580         msg = state->msg;
581         state->msg = NULL;
582         kfree(state, pmp->mmsg);
583         if (msg)
584                 hammer2_msg_free(msg);
585 }
586
587 hammer2_msg_t *
588 hammer2_msg_alloc(hammer2_router_t *router, uint32_t cmd)
589 {
590         hammer2_msg_t *msg;
591         size_t hbytes;
592
593         hbytes = (cmd & HAMMER2_MSGF_SIZE) * HAMMER2_MSG_ALIGN;
594         msg = kmalloc(offsetof(struct hammer2_msg, any) + hbytes,
595                       router->pmp->mmsg, M_WAITOK | M_ZERO);
596         msg->hdr_size = hbytes;
597         msg->router = router;
598         KKASSERT(router != NULL);
599         msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
600         msg->any.head.source = 0;
601         msg->any.head.target = router->target;
602         msg->any.head.cmd = cmd;
603
604         return (msg);
605 }
606
607 void
608 hammer2_msg_free(hammer2_msg_t *msg)
609 {
610         hammer2_pfsmount_t *pmp = msg->router->pmp;
611
612         if (msg->aux_data && msg->aux_size) {
613                 kfree(msg->aux_data, pmp->mmsg);
614                 msg->aux_data = NULL;
615                 msg->aux_size = 0;
616                 msg->router = NULL;
617         }
618         kfree(msg, pmp->mmsg);
619 }
620
621 /*
622  * Indexed messages are stored in a red-black tree indexed by their
623  * msgid.  Only persistent messages are indexed.
624  */
625 int
626 hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
627 {
628         if (state1->router < state2->router)
629                 return(-1);
630         if (state1->router > state2->router)
631                 return(1);
632         if (state1->msgid < state2->msgid)
633                 return(-1);
634         if (state1->msgid > state2->msgid)
635                 return(1);
636         return(0);
637 }
638
639 /*
640  * Write a message.  All requisit command flags have been set.
641  *
642  * If msg->state is non-NULL the message is written to the existing
643  * transaction.  msgid will be set accordingly.
644  *
645  * If msg->state is NULL and CREATE is set new state is allocated and
646  * (func, data) is installed.  A msgid is assigned.
647  *
648  * If msg->state is NULL and CREATE is not set the message is assumed
649  * to be a one-way message.  The originator must assign the msgid
650  * (or leave it 0, which is typical.
651  *
652  * This function merely queues the message to the management thread, it
653  * does not write to the message socket/pipe.
654  */
655 void
656 hammer2_msg_write(hammer2_msg_t *msg,
657                   int (*func)(hammer2_state_t *, hammer2_msg_t *), void *data)
658 {
659         hammer2_pfsmount_t *pmp = msg->router->pmp;
660         hammer2_state_t *state;
661
662         if (msg->state) {
663                 /*
664                  * Continuance or termination of existing transaction.
665                  * The transaction could have been initiated by either end.
666                  *
667                  * (Function callback and aux data for the receive side can
668                  * be replaced or left alone).
669                  */
670                 state = msg->state;
671                 msg->any.head.msgid = state->msgid;
672                 msg->any.head.source = 0;
673                 msg->any.head.target = state->router->target;
674                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
675                 if (func) {
676                         state->func = func;
677                         state->any.any = data;
678                 }
679         } else if (msg->any.head.cmd & HAMMER2_MSGF_CREATE) {
680                 /*
681                  * New transaction, requires tracking state and a unique
682                  * msgid to be allocated.
683                  */
684                 KKASSERT(msg->state == NULL);
685                 state = kmalloc(sizeof(*state), pmp->mmsg, M_WAITOK | M_ZERO);
686                 state->pmp = pmp;
687                 state->flags = HAMMER2_STATE_DYNAMIC;
688                 state->func = func;
689                 state->any.any = data;
690                 state->msg = msg;
691                 state->msgid = (uint64_t)(uintptr_t)state;
692                 state->router = msg->router;
693                 msg->state = state;
694                 msg->any.head.source = 0;
695                 msg->any.head.target = state->router->target;
696                 msg->any.head.msgid = state->msgid;
697
698                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
699                 if (RB_INSERT(hammer2_state_tree, &pmp->statewr_tree, state))
700                         panic("duplicate msgid allocated");
701                 msg->any.head.msgid = state->msgid;
702         } else {
703                 /*
704                  * One-off message (always uses msgid 0 to distinguish
705                  * between a possibly lost in-transaction message due to
706                  * competing aborts and a real one-off message?)
707                  */
708                 msg->any.head.msgid = 0;
709                 msg->any.head.source = 0;
710                 msg->any.head.target = msg->router->target;
711                 lockmgr(&pmp->msglk, LK_EXCLUSIVE);
712         }
713
714         /*
715          * Finish up the msg fields
716          */
717         msg->any.head.salt = /* (random << 8) | */ (pmp->msg_seq & 255);
718         ++pmp->msg_seq;
719
720         msg->any.head.hdr_crc = 0;
721         msg->any.head.hdr_crc = hammer2_icrc32(msg->any.buf, msg->hdr_size);
722
723         TAILQ_INSERT_TAIL(&pmp->msgq, msg, qentry);
724         lockmgr(&pmp->msglk, LK_RELEASE);
725 }
726
727 /*
728  * Reply to a message and terminate our side of the transaction.
729  *
730  * If msg->state is non-NULL we are replying to a one-way message.
731  */
732 void
733 hammer2_msg_reply(hammer2_msg_t *msg, uint32_t error)
734 {
735         hammer2_state_t *state = msg->state;
736         hammer2_msg_t *nmsg;
737         uint32_t cmd;
738
739         /*
740          * Reply with a simple error code and terminate the transaction.
741          */
742         cmd = HAMMER2_LNK_ERROR;
743
744         /*
745          * Check if our direction has even been initiated yet, set CREATE.
746          *
747          * Check what direction this is (command or reply direction).  Note
748          * that txcmd might not have been initiated yet.
749          *
750          * If our direction has already been closed we just return without
751          * doing anything.
752          */
753         if (state) {
754                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
755                         hammer2_msg_free(msg);
756                         return;
757                 }
758                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
759                         cmd |= HAMMER2_MSGF_CREATE;
760                 if (state->txcmd & HAMMER2_MSGF_REPLY)
761                         cmd |= HAMMER2_MSGF_REPLY;
762                 cmd |= HAMMER2_MSGF_DELETE;
763         } else {
764                 if ((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0)
765                         cmd |= HAMMER2_MSGF_REPLY;
766         }
767
768         nmsg = hammer2_msg_alloc(msg->router, cmd);
769         nmsg->any.head.error = error;
770         nmsg->state = state;
771         hammer2_msg_write(nmsg, NULL, NULL);
772 }
773
774 /*
775  * Reply to a message and continue our side of the transaction.
776  *
777  * If msg->state is non-NULL we are replying to a one-way message and this
778  * function degenerates into the same as hammer2_msg_reply().
779  */
780 void
781 hammer2_msg_result(hammer2_msg_t *msg, uint32_t error)
782 {
783         hammer2_state_t *state = msg->state;
784         hammer2_msg_t *nmsg;
785         uint32_t cmd;
786
787         /*
788          * Return a simple result code, do NOT terminate the transaction.
789          */
790         cmd = HAMMER2_LNK_ERROR;
791
792         /*
793          * Check if our direction has even been initiated yet, set CREATE.
794          *
795          * Check what direction this is (command or reply direction).  Note
796          * that txcmd might not have been initiated yet.
797          *
798          * If our direction has already been closed we just return without
799          * doing anything.
800          */
801         if (state) {
802                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
803                         hammer2_msg_free(msg);
804                         return;
805                 }
806                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
807                         cmd |= HAMMER2_MSGF_CREATE;
808                 if (state->txcmd & HAMMER2_MSGF_REPLY)
809                         cmd |= HAMMER2_MSGF_REPLY;
810                 /* continuing transaction, do not set MSGF_DELETE */
811         } else {
812                 if ((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0)
813                         cmd |= HAMMER2_MSGF_REPLY;
814         }
815
816         nmsg = hammer2_msg_alloc(msg->router, cmd);
817         nmsg->any.head.error = error;
818         nmsg->state = state;
819         hammer2_msg_write(nmsg, NULL, NULL);
820 }