Merge branches 'hammer2' and 'master' of ssh://crater.dragonflybsd.org/repository...
[dragonfly.git] / sbin / hammer2 / msg.c
1 /*
2  * Copyright (c) 2011-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@dragonflybsd.org>
6  * by Venkatesh Srinivas <vsrinivas@dragonflybsd.org>
7  *
8  * Redistribution and use in source and binary forms, with or without
9  * modification, are permitted provided that the following conditions
10  * are met:
11  *
12  * 1. Redistributions of source code must retain the above copyright
13  *    notice, this list of conditions and the following disclaimer.
14  * 2. Redistributions in binary form must reproduce the above copyright
15  *    notice, this list of conditions and the following disclaimer in
16  *    the documentation and/or other materials provided with the
17  *    distribution.
18  * 3. Neither the name of The DragonFly Project nor the names of its
19  *    contributors may be used to endorse or promote products derived
20  *    from this software without specific, prior written permission.
21  *
22  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
23  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
24  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
25  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
26  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
27  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
28  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
29  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
30  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
32  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
33  * SUCH DAMAGE.
34  */
35
36 #include "hammer2.h"
37
38 static int hammer2_state_msgrx(hammer2_iocom_t *iocom, hammer2_msg_t *msg);
39 static int hammer2_state_msgtx(hammer2_iocom_t *iocom, hammer2_msg_t *msg);
40 static void hammer2_state_cleanuptx(hammer2_iocom_t *iocom, hammer2_msg_t *msg);
41
42 /*
43  * Initialize a low-level ioq
44  */
45 void
46 hammer2_ioq_init(hammer2_iocom_t *iocom __unused, hammer2_ioq_t *ioq)
47 {
48         bzero(ioq, sizeof(*ioq));
49         ioq->state = HAMMER2_MSGQ_STATE_HEADER1;
50         TAILQ_INIT(&ioq->msgq);
51 }
52
53 void
54 hammer2_ioq_done(hammer2_iocom_t *iocom __unused, hammer2_ioq_t *ioq)
55 {
56         hammer2_msg_t *msg;
57
58         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
59                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
60                 hammer2_msg_free(iocom, msg);
61         }
62         if ((msg = ioq->msg) != NULL) {
63                 ioq->msg = NULL;
64                 hammer2_msg_free(iocom, msg);
65         }
66 }
67
68 /*
69  * Initialize a low-level communications channel
70  */
71 void
72 hammer2_iocom_init(hammer2_iocom_t *iocom, int sock_fd, int alt_fd)
73 {
74         bzero(iocom, sizeof(*iocom));
75
76         RB_INIT(&iocom->staterd_tree);
77         RB_INIT(&iocom->statewr_tree);
78         TAILQ_INIT(&iocom->freeq);
79         TAILQ_INIT(&iocom->freeq_aux);
80         iocom->sock_fd = sock_fd;
81         iocom->alt_fd = alt_fd;
82         iocom->flags = HAMMER2_IOCOMF_RREQ | HAMMER2_IOCOMF_WIDLE;
83         hammer2_ioq_init(iocom, &iocom->ioq_rx);
84         hammer2_ioq_init(iocom, &iocom->ioq_tx);
85
86         /*
87          * Negotiate session crypto synchronously.  This will mark the
88          * connection as error'd if it fails.
89          */
90         hammer2_crypto_negotiate(iocom);
91
92         /*
93          * Make sure our fds are set to non-blocking for the iocom core.
94          */
95         if (sock_fd >= 0)
96                 fcntl(sock_fd, F_SETFL, O_NONBLOCK);
97 #if 0
98         /* if line buffered our single fgets() should be fine */
99         if (alt_fd >= 0)
100                 fcntl(alt_fd, F_SETFL, O_NONBLOCK);
101 #endif
102 }
103
104 void
105 hammer2_iocom_done(hammer2_iocom_t *iocom)
106 {
107         hammer2_msg_t *msg;
108
109         iocom->sock_fd = -1;
110         hammer2_ioq_done(iocom, &iocom->ioq_rx);
111         hammer2_ioq_done(iocom, &iocom->ioq_tx);
112         if ((msg = TAILQ_FIRST(&iocom->freeq)) != NULL) {
113                 TAILQ_REMOVE(&iocom->freeq, msg, qentry);
114                 free(msg);
115         }
116         if ((msg = TAILQ_FIRST(&iocom->freeq_aux)) != NULL) {
117                 TAILQ_REMOVE(&iocom->freeq_aux, msg, qentry);
118                 free(msg->aux_data);
119                 msg->aux_data = NULL;
120                 free(msg);
121         }
122 }
123
124 /*
125  * Allocate a new one-way message.
126  */
127 hammer2_msg_t *
128 hammer2_msg_alloc(hammer2_iocom_t *iocom, size_t aux_size, uint32_t cmd)
129 {
130         hammer2_msg_t *msg;
131         int hbytes;
132
133         if (aux_size) {
134                 aux_size = (aux_size + HAMMER2_MSG_ALIGNMASK) &
135                            ~HAMMER2_MSG_ALIGNMASK;
136                 if ((msg = TAILQ_FIRST(&iocom->freeq_aux)) != NULL)
137                         TAILQ_REMOVE(&iocom->freeq_aux, msg, qentry);
138         } else {
139                 if ((msg = TAILQ_FIRST(&iocom->freeq)) != NULL)
140                         TAILQ_REMOVE(&iocom->freeq, msg, qentry);
141         }
142         if (msg == NULL) {
143                 msg = malloc(sizeof(*msg));
144                 bzero(msg, sizeof(*msg));
145                 msg->aux_data = NULL;
146                 msg->aux_size = 0;
147         }
148         if (msg->aux_size != aux_size) {
149                 if (msg->aux_data) {
150                         free(msg->aux_data);
151                         msg->aux_data = NULL;
152                         msg->aux_size = 0;
153                 }
154                 if (aux_size) {
155                         msg->aux_data = malloc(aux_size);
156                         msg->aux_size = aux_size;
157                 }
158         }
159         hbytes = (cmd & HAMMER2_MSGF_SIZE) * HAMMER2_MSG_ALIGN;
160         if (hbytes)
161                 bzero(&msg->any.head, hbytes);
162         msg->hdr_size = hbytes;
163         msg->any.head.aux_icrc = 0;
164         msg->any.head.cmd = cmd;
165
166         return (msg);
167 }
168
169 /*
170  * Free a message so it can be reused afresh.
171  *
172  * NOTE: aux_size can be 0 with a non-NULL aux_data.
173  */
174 void
175 hammer2_msg_free(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
176 {
177         if (msg->aux_data)
178                 TAILQ_INSERT_TAIL(&iocom->freeq_aux, msg, qentry);
179         else
180                 TAILQ_INSERT_TAIL(&iocom->freeq, msg, qentry);
181 }
182
183 /*
184  * I/O core loop for an iocom.
185  */
186 void
187 hammer2_iocom_core(hammer2_iocom_t *iocom,
188                    void (*recvmsg_func)(hammer2_iocom_t *),
189                    void (*sendmsg_func)(hammer2_iocom_t *),
190                    void (*altmsg_func)(hammer2_iocom_t *))
191 {
192         struct pollfd fds[2];
193         int timeout;
194
195         iocom->recvmsg_callback = recvmsg_func;
196         iocom->sendmsg_callback = sendmsg_func;
197         iocom->altmsg_callback = altmsg_func;
198
199         while ((iocom->flags & HAMMER2_IOCOMF_EOF) == 0) {
200                 timeout = 5000;
201
202                 fds[0].fd = iocom->sock_fd;
203                 fds[0].events = 0;
204                 fds[0].revents = 0;
205
206                 if (iocom->flags & HAMMER2_IOCOMF_RREQ)
207                         fds[0].events |= POLLIN;
208                 else
209                         timeout = 0;
210                 if ((iocom->flags & HAMMER2_IOCOMF_WIDLE) == 0) {
211                         if (iocom->flags & HAMMER2_IOCOMF_WREQ)
212                                 fds[0].events |= POLLOUT;
213                         else
214                                 timeout = 0;
215                 }
216
217                 if (iocom->alt_fd >= 0) {
218                         fds[1].fd = iocom->alt_fd;
219                         fds[1].events |= POLLIN;
220                         fds[1].revents = 0;
221                         poll(fds, 2, timeout);
222                 } else {
223                         poll(fds, 1, timeout);
224                 }
225                 if ((fds[0].revents & POLLIN) ||
226                     (iocom->flags & HAMMER2_IOCOMF_RREQ) == 0) {
227                         iocom->recvmsg_callback(iocom);
228                 }
229                 if ((iocom->flags & HAMMER2_IOCOMF_WIDLE) == 0) {
230                         if ((fds[0].revents & POLLOUT) ||
231                             (iocom->flags & HAMMER2_IOCOMF_WREQ) == 0) {
232                                 iocom->sendmsg_callback(iocom);
233                         }
234                 }
235                 if (iocom->alt_fd >= 0 && (fds[1].revents & POLLIN))
236                         iocom->altmsg_callback(iocom);
237         }
238 }
239
240 /*
241  * Read the next ready message from the ioq, issuing I/O if needed.
242  * Caller should retry on a read-event when NULL is returned.
243  *
244  * If an error occurs during reception a HAMMER2_LNK_ERROR msg will
245  * be returned (and the caller must not call us again after that).
246  */
247 hammer2_msg_t *
248 hammer2_ioq_read(hammer2_iocom_t *iocom)
249 {
250         hammer2_ioq_t *ioq = &iocom->ioq_rx;
251         hammer2_msg_t *msg;
252         hammer2_msg_hdr_t *head;
253         ssize_t n;
254         size_t bytes;
255         size_t nmax;
256         uint16_t xcrc16;
257         uint32_t xcrc32;
258         int error;
259
260 again:
261         /*
262          * If a message is already pending we can just remove and
263          * return it.  Message state has already been processed.
264          */
265         if ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
266                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
267                 return (msg);
268         }
269
270         /*
271          * Message read in-progress (msg is NULL at the moment).  We don't
272          * allocate a msg until we have its core header.
273          */
274         bytes = ioq->fifo_end - ioq->fifo_beg;
275         nmax = sizeof(ioq->buf) - ioq->fifo_end;
276         msg = ioq->msg;
277
278         switch(ioq->state) {
279         case HAMMER2_MSGQ_STATE_HEADER1:
280                 /*
281                  * Load the primary header, fail on any non-trivial read
282                  * error or on EOF.  Since the primary header is the same
283                  * size is the message alignment it will never straddle
284                  * the end of the buffer.
285                  */
286                 if (bytes < (int)sizeof(msg->any.head)) {
287                         n = read(iocom->sock_fd,
288                                  ioq->buf + ioq->fifo_end,
289                                  nmax);
290                         if (n <= 0) {
291                                 if (n == 0) {
292                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
293                                         break;
294                                 }
295                                 if (errno != EINTR &&
296                                     errno != EINPROGRESS &&
297                                     errno != EAGAIN) {
298                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
299                                         break;
300                                 }
301                                 n = 0;
302                                 /* fall through */
303                         }
304                         ioq->fifo_end += n;
305                         bytes += n;
306                         nmax -= n;
307                 }
308
309                 /*
310                  * Insufficient data accumulated (msg is NULL, caller will
311                  * retry on event).
312                  */
313                 assert(msg == NULL);
314                 if (bytes < (int)sizeof(msg->any.head))
315                         break;
316
317                 /*
318                  * Calculate the header, decrypt data received so far.
319                  * Data will be decrypted in-place.  Partial blocks are
320                  * not immediately decrypted.
321                  */
322                 hammer2_crypto_decrypt(iocom, ioq);
323                 head = (void *)(ioq->buf + ioq->fifo_beg);
324
325                 /*
326                  * Check and fixup the core header.  Note that the icrc
327                  * has to be calculated before any fixups, but the crc
328                  * fields in the msg may have to be swapped like everything
329                  * else.
330                  */
331                 if (head->magic != HAMMER2_MSGHDR_MAGIC &&
332                     head->magic != HAMMER2_MSGHDR_MAGIC_REV) {
333                         ioq->error = HAMMER2_IOQ_ERROR_SYNC;
334                         break;
335                 }
336
337                 xcrc32 = hammer2_icrc32((char *)head + HAMMER2_MSGHDR_CRCOFF,
338                                         HAMMER2_MSGHDR_CRCBYTES);
339                 if (head->magic == HAMMER2_MSGHDR_MAGIC_REV) {
340                         hammer2_bswap_head(head);
341                 }
342                 xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
343                 if (xcrc16 != head->icrc1) {
344                         ioq->error = HAMMER2_IOQ_ERROR_HCRC;
345                         break;
346                 }
347
348                 /*
349                  * Calculate the full header size and aux data size
350                  */
351                 ioq->hbytes = (head->cmd & HAMMER2_MSGF_SIZE) *
352                               HAMMER2_MSG_ALIGN;
353                 ioq->abytes = head->aux_bytes * HAMMER2_MSG_ALIGN;
354                 if (ioq->hbytes < sizeof(msg->any.head) ||
355                     ioq->hbytes > sizeof(msg->any) ||
356                     ioq->abytes > HAMMER2_MSGAUX_MAX) {
357                         ioq->error = HAMMER2_IOQ_ERROR_FIELD;
358                         break;
359                 }
360
361                 /*
362                  * Finally allocate the message and copy the core header
363                  * to the embedded extended header.
364                  *
365                  * Initialize msg->aux_size to 0 and use it to track
366                  * the amount of data copied from the stream.
367                  */
368                 msg = hammer2_msg_alloc(iocom, ioq->abytes, 0);
369                 ioq->msg = msg;
370
371                 /*
372                  * We are either done or we fall-through
373                  */
374                 if (ioq->hbytes == sizeof(msg->any.head) && ioq->abytes == 0) {
375                         bcopy(head, &msg->any.head, sizeof(msg->any.head));
376                         ioq->fifo_beg += ioq->hbytes;
377                         break;
378                 }
379
380                 /*
381                  * Fall through to the next state.  Make sure that the
382                  * extended header does not straddle the end of the buffer.
383                  * We still want to issue larger reads into our buffer,
384                  * book-keeping is easier if we don't bcopy() yet.
385                  */
386                 if (bytes + nmax < ioq->hbytes) {
387                         bcopy(ioq->buf + ioq->fifo_beg, ioq->buf, bytes);
388                         ioq->fifo_cdx -= ioq->fifo_beg;
389                         ioq->fifo_beg = 0;
390                         ioq->fifo_end = bytes;
391                         nmax = sizeof(ioq->buf) - ioq->fifo_end;
392                 }
393                 ioq->state = HAMMER2_MSGQ_STATE_HEADER2;
394                 /* fall through */
395         case HAMMER2_MSGQ_STATE_HEADER2:
396                 /*
397                  * Fill out the extended header.
398                  */
399                 assert(msg != NULL);
400                 if (bytes < ioq->hbytes) {
401                         n = read(iocom->sock_fd,
402                                  msg->any.buf + ioq->fifo_end,
403                                  nmax);
404                         if (n <= 0) {
405                                 if (n == 0) {
406                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
407                                         break;
408                                 }
409                                 if (errno != EINTR &&
410                                     errno != EINPROGRESS &&
411                                     errno != EAGAIN) {
412                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
413                                         break;
414                                 }
415                                 n = 0;
416                                 /* fall through */
417                         }
418                         ioq->fifo_end += n;
419                         bytes += n;
420                         nmax -= n;
421                 }
422
423                 /*
424                  * Insufficient data accumulated (set msg NULL so caller will
425                  * retry on event).
426                  */
427                 if (bytes < ioq->hbytes) {
428                         msg = NULL;
429                         break;
430                 }
431
432                 /*
433                  * Calculate the extended header, decrypt data received
434                  * so far.
435                  */
436                 hammer2_crypto_decrypt(iocom, ioq);
437                 head = (void *)(ioq->buf + ioq->fifo_beg);
438
439                 /*
440                  * Check the crc on the extended header
441                  */
442                 if (ioq->hbytes > sizeof(hammer2_msg_hdr_t)) {
443                         xcrc32 = hammer2_icrc32(head + 1,
444                                                 ioq->hbytes - sizeof(*head));
445                         xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
446                         if (head->icrc2 != xcrc16) {
447                                 ioq->error = HAMMER2_IOQ_ERROR_XCRC;
448                                 break;
449                         }
450                 }
451
452                 /*
453                  * Copy the extended header into the msg and adjust the
454                  * FIFO.
455                  */
456                 bcopy(head, &msg->any, ioq->hbytes);
457
458                 /*
459                  * We are either done or we fall-through.
460                  */
461                 if (ioq->abytes == 0) {
462                         ioq->fifo_beg += ioq->hbytes;
463                         break;
464                 }
465
466                 /*
467                  * Must adjust nmax and bytes (and the state) when falling
468                  * through.
469                  */
470                 ioq->fifo_beg += ioq->hbytes;
471                 nmax -= ioq->hbytes;
472                 bytes -= ioq->hbytes;
473                 ioq->state = HAMMER2_MSGQ_STATE_AUXDATA1;
474                 /* fall through */
475         case HAMMER2_MSGQ_STATE_AUXDATA1:
476                 /*
477                  * Copy the partial or complete payload from remaining
478                  * bytes in the FIFO.  We have to fall-through either
479                  * way so we can check the crc.
480                  *
481                  * Adjust msg->aux_size to the final actual value.
482                  */
483                 ioq->already = ioq->fifo_cdx - ioq->fifo_beg;
484                 if (ioq->already > ioq->abytes)
485                         ioq->already = ioq->abytes;
486                 if (bytes >= ioq->abytes) {
487                         bcopy(ioq->buf + ioq->fifo_beg, msg->aux_data,
488                               ioq->abytes);
489                         msg->aux_size = ioq->abytes;
490                         ioq->fifo_beg += ioq->abytes;
491                         if (ioq->fifo_cdx < ioq->fifo_beg)
492                                 ioq->fifo_cdx = ioq->fifo_beg;
493                         bytes -= ioq->abytes;
494                 } else if (bytes) {
495                         bcopy(ioq->buf + ioq->fifo_beg, msg->aux_data,
496                               bytes);
497                         msg->aux_size = bytes;
498                         ioq->fifo_beg += bytes;
499                         if (ioq->fifo_cdx < ioq->fifo_beg)
500                                 ioq->fifo_cdx = ioq->fifo_beg;
501                         bytes = 0;
502                 } else {
503                         msg->aux_size = 0;
504                 }
505                 ioq->state = HAMMER2_MSGQ_STATE_AUXDATA2;
506                 /* fall through */
507         case HAMMER2_MSGQ_STATE_AUXDATA2:
508                 /*
509                  * Read the remainder of the payload directly into the
510                  * msg->aux_data buffer.
511                  */
512                 assert(msg);
513                 if (msg->aux_size < ioq->abytes) {
514                         assert(bytes == 0);
515                         n = read(iocom->sock_fd,
516                                  msg->aux_data + msg->aux_size,
517                                  ioq->abytes - msg->aux_size);
518                         if (n <= 0) {
519                                 if (n == 0) {
520                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
521                                         break;
522                                 }
523                                 if (errno != EINTR &&
524                                     errno != EINPROGRESS &&
525                                     errno != EAGAIN) {
526                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
527                                         break;
528                                 }
529                                 n = 0;
530                                 /* fall through */
531                         }
532                         msg->aux_size += n;
533                 }
534
535                 /*
536                  * Insufficient data accumulated (set msg NULL so caller will
537                  * retry on event).
538                  */
539                 if (msg->aux_size < ioq->abytes) {
540                         msg = NULL;
541                         break;
542                 }
543                 assert(msg->aux_size == ioq->abytes);
544                 hammer2_crypto_decrypt_aux(iocom, ioq, msg, ioq->already);
545
546                 /*
547                  * Check aux_icrc, then we are done.
548                  */
549                 xcrc32 = hammer2_icrc32(msg->aux_data, msg->aux_size);
550                 if (xcrc32 != msg->any.head.aux_icrc) {
551                         ioq->error = HAMMER2_IOQ_ERROR_ACRC;
552                         break;
553                 }
554                 break;
555         case HAMMER2_MSGQ_STATE_ERROR:
556         default:
557                 /*
558                  * We don't double-return errors, the caller should not
559                  * have called us again after getting an error msg.
560                  */
561                 assert(0);
562                 break;
563         }
564
565         /*
566          * Check the message sequence.  The iv[] should prevent any
567          * possibility of a replay but we add this check anyway.
568          */
569         if (msg && ioq->error == 0) {
570                 if ((msg->any.head.salt & 255) != (ioq->seq & 255)) {
571                         ioq->error = HAMMER2_IOQ_ERROR_MSGSEQ;
572                 } else {
573                         ++ioq->seq;
574                 }
575         }
576
577         /*
578          * Process transactional state for the message.
579          */
580         if (msg && ioq->error == 0) {
581                 error = hammer2_state_msgrx(iocom, msg);
582                 if (error) {
583                         if (error == HAMMER2_IOQ_ERROR_EALREADY) {
584                                 hammer2_msg_free(iocom, msg);
585                                 goto again;
586                         }
587                         ioq->error = error;
588                 }
589         }
590
591         /*
592          * Handle error, RREQ, or completion
593          *
594          * NOTE: nmax and bytes are invalid at this point, we don't bother
595          *       to update them when breaking out.
596          */
597         if (ioq->error) {
598                 /*
599                  * An unrecoverable error occured during processing,
600                  * return a special error message.  Try to leave the
601                  * ioq state alone for post-mortem debugging.
602                  *
603                  * Link error messages are returned as one-way messages,
604                  * so no flags get set.  Source and target is 0 (link-level),
605                  * msgid is 0 (link-level).  All we really need to do is
606                  * set up magic, cmd, and error.
607                  */
608                 assert(ioq->msg == msg);
609                 if (msg == NULL)
610                         msg = hammer2_msg_alloc(iocom, 0, 0);
611                 else
612                         ioq->msg = NULL;
613
614                 if (msg->aux_data) {
615                         free(msg->aux_data);
616                         msg->aux_data = NULL;
617                         msg->aux_size = 0;
618                 }
619                 bzero(&msg->any.head, sizeof(msg->any.head));
620                 msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
621                 msg->any.head.cmd = HAMMER2_LNK_ERROR;
622                 msg->any.head.error = ioq->error;
623                 ioq->state = HAMMER2_MSGQ_STATE_ERROR;
624                 iocom->flags |= HAMMER2_IOCOMF_EOF;
625         } else if (msg == NULL) {
626                 /*
627                  * Insufficient data received to finish building the message,
628                  * set RREQ and return NULL.
629                  *
630                  * Leave ioq->msg intact.
631                  * Leave the FIFO intact.
632                  */
633                 iocom->flags |= HAMMER2_IOCOMF_RREQ;
634 #if 0
635                 ioq->fifo_cdx = 0;
636                 ioq->fifo_beg = 0;
637                 ioq->fifo_end = 0;
638 #endif
639         } else {
640                 /*
641                  * Return msg, clear the FIFO if it is now empty.
642                  * Flag RREQ if the caller needs to wait for a read-event
643                  * or not.
644                  *
645                  * The fifo has already been advanced past the message.
646                  * Trivially reset the FIFO indices if possible.
647                  */
648                 if (ioq->fifo_beg == ioq->fifo_end) {
649                         iocom->flags |= HAMMER2_IOCOMF_RREQ;
650                         ioq->fifo_cdx = 0;
651                         ioq->fifo_beg = 0;
652                         ioq->fifo_end = 0;
653                 } else {
654                         iocom->flags &= ~HAMMER2_IOCOMF_RREQ;
655                 }
656                 ioq->state = HAMMER2_MSGQ_STATE_HEADER1;
657                 ioq->msg = NULL;
658         }
659         return (msg);
660 }
661
662 /*
663  * Calculate the header and data crc's and write a low-level message to
664  * the connection.  If aux_icrc is non-zero the aux_data crc is already
665  * assumed to have been set.
666  *
667  * A non-NULL msg is added to the queue but not necessarily flushed.
668  * Calling this function with msg == NULL will get a flush going.
669  */
670 void
671 hammer2_ioq_write(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
672 {
673         hammer2_ioq_t *ioq = &iocom->ioq_tx;
674         uint16_t xcrc16;
675         uint32_t xcrc32;
676         int hbytes;
677         int error;
678
679         assert(msg);
680
681         /*
682          * Process transactional state.
683          */
684         if (ioq->error == 0) {
685                 error = hammer2_state_msgtx(iocom, msg);
686                 if (error) {
687                         if (error == HAMMER2_IOQ_ERROR_EALREADY) {
688                                 hammer2_msg_free(iocom, msg);
689                         } else {
690                                 ioq->error = error;
691                         }
692                 }
693         }
694
695         /*
696          * Process terminal connection errors.
697          */
698         if (ioq->error) {
699                 TAILQ_INSERT_TAIL(&ioq->msgq, msg, qentry);
700                 ++ioq->msgcount;
701                 hammer2_iocom_drain(iocom);
702                 return;
703         }
704
705         /*
706          * Finish populating the msg fields.  The salt ensures that the iv[]
707          * array is ridiculously randomized and we also re-seed our PRNG
708          * every 32768 messages just to be sure.
709          */
710         msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
711         msg->any.head.salt = (random() << 8) | (ioq->seq & 255);
712         ++ioq->seq;
713         if ((ioq->seq & 32767) == 0)
714                 srandomdev();
715
716         /*
717          * Calculate aux_icrc if 0, calculate icrc2, and finally
718          * calculate icrc1.
719          */
720         if (msg->aux_size && msg->any.head.aux_icrc == 0) {
721                 assert((msg->aux_size & HAMMER2_MSG_ALIGNMASK) == 0);
722                 xcrc32 = hammer2_icrc32(msg->aux_data, msg->aux_size);
723                 msg->any.head.aux_icrc = xcrc32;
724         }
725         msg->any.head.aux_bytes = msg->aux_size / HAMMER2_MSG_ALIGN;
726         assert((msg->aux_size & HAMMER2_MSG_ALIGNMASK) == 0);
727
728         if ((msg->any.head.cmd & HAMMER2_MSGF_SIZE) >
729             sizeof(msg->any.head) / HAMMER2_MSG_ALIGN) {
730                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
731                         HAMMER2_MSG_ALIGN;
732                 hbytes -= sizeof(msg->any.head);
733                 xcrc32 = hammer2_icrc32(&msg->any.head + 1, hbytes);
734                 xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
735                 msg->any.head.icrc2 = xcrc16;
736         } else {
737                 msg->any.head.icrc2 = 0;
738         }
739         xcrc32 = hammer2_icrc32(msg->any.buf + HAMMER2_MSGHDR_CRCOFF,
740                                 HAMMER2_MSGHDR_CRCBYTES);
741         xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
742         msg->any.head.icrc1 = xcrc16;
743
744         /*
745          * XXX Encrypt the message
746          */
747
748         /*
749          * Enqueue the message.
750          */
751         TAILQ_INSERT_TAIL(&ioq->msgq, msg, qentry);
752         ++ioq->msgcount;
753         iocom->flags &= ~HAMMER2_IOCOMF_WIDLE;
754
755         /*
756          * Flush if we know we can write (WREQ not set) and if
757          * sufficient messages have accumulated.  Otherwise hold
758          * off to avoid piecemeal system calls.
759          */
760         if (iocom->flags & HAMMER2_IOCOMF_WREQ)
761                 return;
762         if (ioq->msgcount < HAMMER2_IOQ_MAXIOVEC / 2)
763                 return;
764         hammer2_iocom_flush(iocom);
765 }
766
767 void
768 hammer2_iocom_flush(hammer2_iocom_t *iocom)
769 {
770         hammer2_ioq_t *ioq = &iocom->ioq_tx;
771         hammer2_msg_t *msg;
772         ssize_t nmax;
773         ssize_t nact;
774         struct iovec iov[HAMMER2_IOQ_MAXIOVEC];
775         size_t hbytes;
776         size_t abytes;
777         int hoff;
778         int aoff;
779         int n;
780
781         /*
782          * Pump messages out the connection by building an iovec.
783          */
784         n = 0;
785         nmax = 0;
786
787         TAILQ_FOREACH(msg, &ioq->msgq, qentry) {
788                 hoff = 0;
789                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
790                          HAMMER2_MSG_ALIGN;
791                 aoff = 0;
792                 abytes = msg->aux_size;
793                 if (n == 0) {
794                         hoff += ioq->hbytes;
795                         aoff += ioq->abytes;
796                 }
797                 if (hbytes - hoff > 0) {
798                         iov[n].iov_base = (char *)&msg->any.head + hoff;
799                         iov[n].iov_len = hbytes - hoff;
800                         nmax += hbytes - hoff;
801                         ++n;
802                         if (n == HAMMER2_IOQ_MAXIOVEC)
803                                 break;
804                 }
805                 if (abytes - aoff > 0) {
806                         assert(msg->aux_data != NULL);
807                         iov[n].iov_base = msg->aux_data + aoff;
808                         iov[n].iov_len = abytes - aoff;
809                         nmax += abytes - aoff;
810                         ++n;
811                         if (n == HAMMER2_IOQ_MAXIOVEC)
812                                 break;
813                 }
814         }
815         if (n == 0)
816                 return;
817
818         /*
819          * Encrypt and write the data.  The crypto code will move the
820          * data into the fifo and adjust the iov as necessary.  If
821          * encryption is disabled the iov is left alone.
822          *
823          * hammer2_crypto_encrypt_wrote()
824          */
825         n = hammer2_crypto_encrypt(iocom, ioq, iov, n);
826
827         /*
828          * Execute the writev() then figure out what happened.
829          */
830         nact = writev(iocom->sock_fd, iov, n);
831         if (nact < 0) {
832                 if (errno != EINTR &&
833                     errno != EINPROGRESS &&
834                     errno != EAGAIN) {
835                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
836                         hammer2_iocom_drain(iocom);
837                 } else {
838                         iocom->flags |= HAMMER2_IOCOMF_WREQ;
839                 }
840                 return;
841         }
842         hammer2_crypto_encrypt_wrote(iocom, ioq, nact);
843         if (nact == nmax)
844                 iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
845         else
846                 iocom->flags |= HAMMER2_IOCOMF_WREQ;
847
848         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
849                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
850                          HAMMER2_MSG_ALIGN;
851                 abytes = msg->aux_size;
852
853                 if ((size_t)nact < hbytes - ioq->hbytes) {
854                         ioq->hbytes += nact;
855                         break;
856                 }
857                 nact -= hbytes - ioq->hbytes;
858                 ioq->hbytes = hbytes;
859                 if ((size_t)nact < abytes - ioq->abytes) {
860                         ioq->abytes += nact;
861                         break;
862                 }
863                 nact -= abytes - ioq->abytes;
864
865                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
866                 --ioq->msgcount;
867                 ioq->hbytes = 0;
868                 ioq->abytes = 0;
869
870                 hammer2_state_cleanuptx(iocom, msg);
871         }
872         if (msg == NULL) {
873                 iocom->flags |= HAMMER2_IOCOMF_WIDLE;
874                 iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
875         }
876         if (ioq->error) {
877                 iocom->flags |= HAMMER2_IOCOMF_EOF |
878                                 HAMMER2_IOCOMF_WIDLE;
879                 iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
880         }
881 }
882
883 /*
884  * Kill pending msgs on ioq_tx and adjust the flags such that no more
885  * write events will occur.  We don't kill read msgs because we want
886  * the caller to pull off our contrived terminal error msg to detect
887  * the connection failure.
888  */
889 void
890 hammer2_iocom_drain(hammer2_iocom_t *iocom)
891 {
892         hammer2_ioq_t *ioq = &iocom->ioq_tx;
893         hammer2_msg_t *msg;
894
895         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
896                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
897                 --ioq->msgcount;
898                 hammer2_msg_free(iocom, msg);
899         }
900         iocom->flags |= HAMMER2_IOCOMF_WIDLE;
901         iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
902 }
903
904 /*
905  * This is a shortcut to formulate a reply to msg with a simple error code.
906  * It can reply to transaction or one-way messages, or terminate one side
907  * of a stream.  A HAMMER2_LNK_ERROR command code is utilized to encode
908  * the error code (which can be 0).
909  *
910  * Replies to one-way messages are a bit of an oxymoron but the feature
911  * is used by the debug (DBG) protocol.
912  *
913  * The reply contains no data.
914  */
915 void
916 hammer2_msg_reply(hammer2_iocom_t *iocom, hammer2_msg_t *msg, uint16_t error)
917 {
918         hammer2_msg_t *nmsg;
919         uint32_t cmd;
920
921         cmd = HAMMER2_LNK_ERROR;
922         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
923                 /*
924                  * Reply to received reply, reply direction uses txcmd.
925                  * txcmd will be updated by hammer2_ioq_write().
926                  */
927                 if (msg->state) {
928                         if ((msg->state->rxcmd & HAMMER2_MSGF_CREATE) == 0)
929                                 cmd |= HAMMER2_MSGF_CREATE;
930                         cmd |= HAMMER2_MSGF_DELETE;
931                 }
932         } else {
933                 /*
934                  * Reply to received command, reply direction uses rxcmd.
935                  * txcmd will be updated by hammer2_ioq_write().
936                  */
937                 cmd |= HAMMER2_MSGF_REPLY;
938                 if (msg->state) {
939                         if ((msg->state->rxcmd & HAMMER2_MSGF_CREATE) == 0)
940                                 cmd |= HAMMER2_MSGF_CREATE;
941                         cmd |= HAMMER2_MSGF_DELETE;
942                 }
943         }
944         nmsg = hammer2_msg_alloc(iocom, 0, cmd);
945         nmsg->any.head.error = error;
946         hammer2_ioq_write(iocom, nmsg);
947 }
948
949 /************************************************************************
950  *                      TRANSACTION STATE HANDLING                      *
951  ************************************************************************
952  *
953  */
954
955 RB_GENERATE(hammer2_state_tree, hammer2_state, rbnode, hammer2_state_cmp);
956
957 /*
958  * Process state tracking for a message after reception, prior to
959  * execution.
960  *
961  * Called with msglk held and the msg dequeued.
962  *
963  * All messages are called with dummy state and return actual state.
964  * (One-off messages often just return the same dummy state).
965  *
966  * May request that caller discard the message by setting *discardp to 1.
967  * The returned state is not used in this case and is allowed to be NULL.
968  *
969  * --
970  *
971  * These routines handle persistent and command/reply message state via the
972  * CREATE and DELETE flags.  The first message in a command or reply sequence
973  * sets CREATE, the last message in a command or reply sequence sets DELETE.
974  *
975  * There can be any number of intermediate messages belonging to the same
976  * sequence sent inbetween the CREATE message and the DELETE message,
977  * which set neither flag.  This represents a streaming command or reply.
978  *
979  * Any command message received with CREATE set expects a reply sequence to
980  * be returned.  Reply sequences work the same as command sequences except the
981  * REPLY bit is also sent.  Both the command side and reply side can
982  * degenerate into a single message with both CREATE and DELETE set.  Note
983  * that one side can be streaming and the other side not, or neither, or both.
984  *
985  * The msgid is unique for the initiator.  That is, two sides sending a new
986  * message can use the same msgid without colliding.
987  *
988  * --
989  *
990  * ABORT sequences work by setting the ABORT flag along with normal message
991  * state.  However, ABORTs can also be sent on half-closed messages, that is
992  * even if the command or reply side has already sent a DELETE, as long as
993  * the message has not been fully closed it can still send an ABORT+DELETE
994  * to terminate the half-closed message state.
995  *
996  * Since ABORT+DELETEs can race we silently discard ABORT's for message
997  * state which has already been fully closed.  REPLY+ABORT+DELETEs can
998  * also race, and in this situation the other side might have already
999  * initiated a new unrelated command with the same message id.  Since
1000  * the abort has not set the CREATE flag the situation can be detected
1001  * and the message will also be discarded.
1002  *
1003  * Non-blocking requests can be initiated with ABORT+CREATE[+DELETE].
1004  * The ABORT request is essentially integrated into the command instead
1005  * of being sent later on.  In this situation the command implementation
1006  * detects that CREATE and ABORT are both set (vs ABORT alone) and can
1007  * special-case non-blocking operation for the command.
1008  *
1009  * NOTE!  Messages with ABORT set without CREATE or DELETE are considered
1010  *        to be mid-stream aborts for command/reply sequences.  ABORTs on
1011  *        one-way messages are not supported.
1012  *
1013  * NOTE!  If a command sequence does not support aborts the ABORT flag is
1014  *        simply ignored.
1015  *
1016  * --
1017  *
1018  * One-off messages (no reply expected) are sent with neither CREATE or DELETE
1019  * set.  One-off messages cannot be aborted and typically aren't processed
1020  * by these routines.  The REPLY bit can be used to distinguish whether a
1021  * one-off message is a command or reply.  For example, one-off replies
1022  * will typically just contain status updates.
1023  */
1024 static int
1025 hammer2_state_msgrx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1026 {
1027         hammer2_state_t *state;
1028         hammer2_state_t dummy;
1029         int error;
1030
1031         /*
1032          * Lock RB tree and locate existing persistent state, if any.
1033          *
1034          * If received msg is a command state is on staterd_tree.
1035          * If received msg is a reply state is on statewr_tree.
1036          */
1037         /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1038
1039         dummy.msgid = msg->any.head.msgid;
1040         dummy.source = msg->any.head.source;
1041         dummy.target = msg->any.head.target;
1042         iocom_printf(iocom, msg->any.head.cmd,
1043                      "received msg %08x msgid %u source=%u target=%u\n",
1044                       msg->any.head.cmd, msg->any.head.msgid,
1045                       msg->any.head.source, msg->any.head.target);
1046         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1047                 state = RB_FIND(hammer2_state_tree,
1048                                 &iocom->statewr_tree, &dummy);
1049         } else {
1050                 state = RB_FIND(hammer2_state_tree,
1051                                 &iocom->staterd_tree, &dummy);
1052         }
1053         msg->state = state;
1054
1055         /*
1056          * Short-cut one-off or mid-stream messages (state may be NULL).
1057          */
1058         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1059                                   HAMMER2_MSGF_ABORT)) == 0) {
1060                 /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1061                 return(0);
1062         }
1063
1064         /*
1065          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
1066          * inside the case statements.
1067          */
1068         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1069                                     HAMMER2_MSGF_REPLY)) {
1070         case HAMMER2_MSGF_CREATE:
1071         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1072                 /*
1073                  * New persistant command received.
1074                  */
1075                 if (state) {
1076                         iocom_printf(iocom, msg->any.head.cmd,
1077                                      "hammer2_state_msgrx: "
1078                                      "duplicate transaction\n");
1079                         error = HAMMER2_IOQ_ERROR_TRANS;
1080                         break;
1081                 }
1082                 state = malloc(sizeof(*state));
1083                 bzero(state, sizeof(*state));
1084                 state->iocom = iocom;
1085                 state->flags = HAMMER2_STATE_DYNAMIC;
1086                 state->msg = msg;
1087                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1088                 RB_INSERT(hammer2_state_tree, &iocom->staterd_tree, state);
1089                 state->flags |= HAMMER2_STATE_INSERTED;
1090                 msg->state = state;
1091                 error = 0;
1092                 break;
1093         case HAMMER2_MSGF_DELETE:
1094                 /*
1095                  * Persistent state is expected but might not exist if an
1096                  * ABORT+DELETE races the close.
1097                  */
1098                 if (state == NULL) {
1099                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1100                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1101                         } else {
1102                                 iocom_printf(iocom, msg->any.head.cmd,
1103                                              "hammer2_state_msgrx: "
1104                                              "no state for DELETE\n");
1105                                 error = HAMMER2_IOQ_ERROR_TRANS;
1106                         }
1107                         break;
1108                 }
1109
1110                 /*
1111                  * Handle another ABORT+DELETE case if the msgid has already
1112                  * been reused.
1113                  */
1114                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1115                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1116                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1117                         } else {
1118                                 iocom_printf(iocom, msg->any.head.cmd,
1119                                              "hammer2_state_msgrx: "
1120                                              "state reused for DELETE\n");
1121                                 error = HAMMER2_IOQ_ERROR_TRANS;
1122                         }
1123                         break;
1124                 }
1125                 error = 0;
1126                 break;
1127         default:
1128                 /*
1129                  * Check for mid-stream ABORT command received, otherwise
1130                  * allow.
1131                  */
1132                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1133                         if (state == NULL ||
1134                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1135                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1136                                 break;
1137                         }
1138                 }
1139                 error = 0;
1140                 break;
1141         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
1142         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1143                 /*
1144                  * When receiving a reply with CREATE set the original
1145                  * persistent state message should already exist.
1146                  */
1147                 if (state == NULL) {
1148                         iocom_printf(iocom, msg->any.head.cmd,
1149                                      "hammer2_state_msgrx: "
1150                                      "no state match for REPLY cmd=%08x\n",
1151                                      msg->any.head.cmd);
1152                         error = HAMMER2_IOQ_ERROR_TRANS;
1153                         break;
1154                 }
1155                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1156                 error = 0;
1157                 break;
1158         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
1159                 /*
1160                  * Received REPLY+ABORT+DELETE in case where msgid has
1161                  * already been fully closed, ignore the message.
1162                  */
1163                 if (state == NULL) {
1164                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1165                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1166                         } else {
1167                                 iocom_printf(iocom, msg->any.head.cmd,
1168                                              "hammer2_state_msgrx: "
1169                                              "no state match for "
1170                                              "REPLY|DELETE\n");
1171                                 error = HAMMER2_IOQ_ERROR_TRANS;
1172                         }
1173                         break;
1174                 }
1175
1176                 /*
1177                  * Received REPLY+ABORT+DELETE in case where msgid has
1178                  * already been reused for an unrelated message,
1179                  * ignore the message.
1180                  */
1181                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1182                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1183                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1184                         } else {
1185                                 iocom_printf(iocom, msg->any.head.cmd,
1186                                              "hammer2_state_msgrx: "
1187                                              "state reused for REPLY|DELETE\n");
1188                                 error = HAMMER2_IOQ_ERROR_TRANS;
1189                         }
1190                         break;
1191                 }
1192                 error = 0;
1193                 break;
1194         case HAMMER2_MSGF_REPLY:
1195                 /*
1196                  * Check for mid-stream ABORT reply received to sent command.
1197                  */
1198                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1199                         if (state == NULL ||
1200                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1201                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1202                                 break;
1203                         }
1204                 }
1205                 error = 0;
1206                 break;
1207         }
1208         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1209         return (error);
1210 }
1211
1212 void
1213 hammer2_state_cleanuprx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1214 {
1215         hammer2_state_t *state;
1216
1217         if ((state = msg->state) == NULL) {
1218                 hammer2_msg_free(iocom, msg);
1219         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
1220                 /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1221                 state->rxcmd |= HAMMER2_MSGF_DELETE;
1222                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
1223                         if (state->msg == msg)
1224                                 state->msg = NULL;
1225                         assert(state->flags & HAMMER2_STATE_INSERTED);
1226                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1227                                 RB_REMOVE(hammer2_state_tree,
1228                                           &iocom->statewr_tree, state);
1229                         } else {
1230                                 RB_REMOVE(hammer2_state_tree,
1231                                           &iocom->staterd_tree, state);
1232                         }
1233                         state->flags &= ~HAMMER2_STATE_INSERTED;
1234                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1235                         hammer2_state_free(state);
1236                 } else {
1237                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1238                 }
1239                 hammer2_msg_free(iocom, msg);
1240         } else if (state->msg != msg) {
1241                 hammer2_msg_free(iocom, msg);
1242         }
1243 }
1244
1245 /*
1246  * Process state tracking for a message prior to transmission.
1247  *
1248  * Called with msglk held and the msg dequeued.
1249  *
1250  * One-off messages are usually with dummy state and msg->state may be NULL
1251  * in this situation.
1252  *
1253  * New transactions (when CREATE is set) will insert the state.
1254  *
1255  * May request that caller discard the message by setting *discardp to 1.
1256  * A NULL state may be returned in this case.
1257  */
1258 static int
1259 hammer2_state_msgtx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1260 {
1261         hammer2_state_t *state;
1262         int error;
1263
1264         /*
1265          * Lock RB tree.  If persistent state is present it will have already
1266          * been assigned to msg.
1267          */
1268         /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1269         state = msg->state;
1270
1271         /*
1272          * Short-cut one-off or mid-stream messages (state may be NULL).
1273          */
1274         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1275                                   HAMMER2_MSGF_ABORT)) == 0) {
1276                 /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1277                 return(0);
1278         }
1279
1280
1281         /*
1282          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
1283          * inside the case statements.
1284          */
1285         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1286                                     HAMMER2_MSGF_REPLY)) {
1287         case HAMMER2_MSGF_CREATE:
1288         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1289                 /*
1290                  * Insert the new persistent message state and mark
1291                  * half-closed if DELETE is set.  Since this is a new
1292                  * message it isn't possible to transition into the fully
1293                  * closed state here.
1294                  *
1295                  * XXX state must be assigned and inserted by
1296                  *     hammer2_msg_write().  txcmd is assigned by us
1297                  *     on-transmit.
1298                  */
1299                 assert(state != NULL);
1300 #if 0
1301                 if (state == NULL) {
1302                         state = pmp->freerd_state;
1303                         pmp->freerd_state = NULL;
1304                         msg->state = state;
1305                         state->msg = msg;
1306                         state->msgid = msg->any.head.msgid;
1307                         state->source = msg->any.head.source;
1308                         state->target = msg->any.head.target;
1309                 }
1310                 assert((state->flags & HAMMER2_STATE_INSERTED) == 0);
1311                 if (RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state)) {
1312                         iocom_printf(iocom, msg->any.head.cmd,
1313                                     "hammer2_state_msgtx: "
1314                                     "duplicate transaction\n");
1315                         error = HAMMER2_IOQ_ERROR_TRANS;
1316                         break;
1317                 }
1318                 state->flags |= HAMMER2_STATE_INSERTED;
1319 #endif
1320                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1321                 error = 0;
1322                 break;
1323         case HAMMER2_MSGF_DELETE:
1324                 /*
1325                  * Sent ABORT+DELETE in case where msgid has already
1326                  * been fully closed, ignore the message.
1327                  */
1328                 if (state == NULL) {
1329                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1330                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1331                         } else {
1332                                 iocom_printf(iocom, msg->any.head.cmd,
1333                                              "hammer2_state_msgtx: "
1334                                              "no state match for DELETE\n");
1335                                 error = HAMMER2_IOQ_ERROR_TRANS;
1336                         }
1337                         break;
1338                 }
1339
1340                 /*
1341                  * Sent ABORT+DELETE in case where msgid has
1342                  * already been reused for an unrelated message,
1343                  * ignore the message.
1344                  */
1345                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1346                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1347                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1348                         } else {
1349                                 iocom_printf(iocom, msg->any.head.cmd,
1350                                              "hammer2_state_msgtx: "
1351                                              "state reused for DELETE\n");
1352                                 error = HAMMER2_IOQ_ERROR_TRANS;
1353                         }
1354                         break;
1355                 }
1356                 error = 0;
1357                 break;
1358         default:
1359                 /*
1360                  * Check for mid-stream ABORT command sent
1361                  */
1362                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1363                         if (state == NULL ||
1364                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1365                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1366                                 break;
1367                         }
1368                 }
1369                 error = 0;
1370                 break;
1371         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
1372         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1373                 /*
1374                  * When transmitting a reply with CREATE set the original
1375                  * persistent state message should already exist.
1376                  */
1377                 if (state == NULL) {
1378                         iocom_printf(iocom, msg->any.head.cmd,
1379                                      "hammer2_state_msgtx: no state match "
1380                                      "for REPLY | CREATE\n");
1381                         error = HAMMER2_IOQ_ERROR_TRANS;
1382                         break;
1383                 }
1384                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1385                 error = 0;
1386                 break;
1387         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
1388                 /*
1389                  * When transmitting a reply with DELETE set the original
1390                  * persistent state message should already exist.
1391                  *
1392                  * This is very similar to the REPLY|CREATE|* case except
1393                  * txcmd is already stored, so we just add the DELETE flag.
1394                  *
1395                  * Sent REPLY+ABORT+DELETE in case where msgid has
1396                  * already been fully closed, ignore the message.
1397                  */
1398                 if (state == NULL) {
1399                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1400                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1401                         } else {
1402                                 iocom_printf(iocom, msg->any.head.cmd,
1403                                              "hammer2_state_msgtx: "
1404                                              "no state match for "
1405                                              "REPLY | DELETE\n");
1406                                 error = HAMMER2_IOQ_ERROR_TRANS;
1407                         }
1408                         break;
1409                 }
1410
1411                 /*
1412                  * Sent REPLY+ABORT+DELETE in case where msgid has already
1413                  * been reused for an unrelated message, ignore the message.
1414                  */
1415                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1416                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1417                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1418                         } else {
1419                                 iocom_printf(iocom, msg->any.head.cmd,
1420                                              "hammer2_state_msgtx: "
1421                                              "state reused for "
1422                                              "REPLY | DELETE\n");
1423                                 error = HAMMER2_IOQ_ERROR_TRANS;
1424                         }
1425                         break;
1426                 }
1427                 error = 0;
1428                 break;
1429         case HAMMER2_MSGF_REPLY:
1430                 /*
1431                  * Check for mid-stream ABORT reply sent.
1432                  *
1433                  * One-off REPLY messages are allowed for e.g. status updates.
1434                  */
1435                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1436                         if (state == NULL ||
1437                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1438                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1439                                 break;
1440                         }
1441                 }
1442                 error = 0;
1443                 break;
1444         }
1445         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1446         return (error);
1447 }
1448
1449 static void
1450 hammer2_state_cleanuptx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1451 {
1452         hammer2_state_t *state;
1453
1454         if ((state = msg->state) == NULL) {
1455                 hammer2_msg_free(iocom, msg);
1456         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
1457                 /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1458                 state->txcmd |= HAMMER2_MSGF_DELETE;
1459                 if (state->rxcmd & HAMMER2_MSGF_DELETE) {
1460                         if (state->msg == msg)
1461                                 state->msg = NULL;
1462                         assert(state->flags & HAMMER2_STATE_INSERTED);
1463                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1464                                 RB_REMOVE(hammer2_state_tree,
1465                                           &iocom->staterd_tree, state);
1466                         } else {
1467                                 RB_REMOVE(hammer2_state_tree,
1468                                           &iocom->statewr_tree, state);
1469                         }
1470                         state->flags &= ~HAMMER2_STATE_INSERTED;
1471                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1472                         hammer2_state_free(state);
1473                 } else {
1474                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1475                 }
1476                 hammer2_msg_free(iocom, msg);
1477         } else if (state->msg != msg) {
1478                 hammer2_msg_free(iocom, msg);
1479         }
1480 }
1481
1482 void
1483 hammer2_state_free(hammer2_state_t *state)
1484 {
1485         hammer2_iocom_t *iocom = state->iocom;
1486         hammer2_msg_t *msg;
1487
1488         msg = state->msg;
1489         state->msg = NULL;
1490         free(state);
1491         if (msg)
1492                 hammer2_msg_free(iocom, msg);
1493         free(state);
1494 }
1495
1496 /*
1497  * Indexed messages are stored in a red-black tree indexed by their
1498  * msgid.  Only persistent messages are indexed.
1499  */
1500 int
1501 hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
1502 {
1503         if (state1->source < state2->source)
1504                 return(-1);
1505         if (state1->source > state2->source)
1506                 return(1);
1507         if (state1->target < state2->target)
1508                 return(-1);
1509         if (state1->target > state2->target)
1510                 return(1);
1511         if (state1->msgid < state2->msgid)
1512                 return(-1);
1513         if (state1->msgid > state2->msgid)
1514                 return(1);
1515         return(0);
1516 }