91ef036105072289facc0949836cafe07eb44124
[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 for each open transaction, then the ioq and iocom
246  * will be errored out and a non-transactional HAMMER2_LNK_ERROR
247  * msg will be returned as the final message.  The caller should not call
248  * us again after the final message is returned.
249  */
250 hammer2_msg_t *
251 hammer2_ioq_read(hammer2_iocom_t *iocom)
252 {
253         hammer2_ioq_t *ioq = &iocom->ioq_rx;
254         hammer2_msg_t *msg;
255         hammer2_msg_hdr_t *head;
256         hammer2_state_t *state;
257         ssize_t n;
258         size_t bytes;
259         size_t nmax;
260         uint16_t xcrc16;
261         uint32_t xcrc32;
262         int error;
263
264 again:
265         /*
266          * If a message is already pending we can just remove and
267          * return it.  Message state has already been processed.
268          */
269         if ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
270                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
271                 return (msg);
272         }
273
274         /*
275          * Message read in-progress (msg is NULL at the moment).  We don't
276          * allocate a msg until we have its core header.
277          */
278         bytes = ioq->fifo_end - ioq->fifo_beg;
279         nmax = sizeof(ioq->buf) - ioq->fifo_end;
280         msg = ioq->msg;
281
282         switch(ioq->state) {
283         case HAMMER2_MSGQ_STATE_HEADER1:
284                 /*
285                  * Load the primary header, fail on any non-trivial read
286                  * error or on EOF.  Since the primary header is the same
287                  * size is the message alignment it will never straddle
288                  * the end of the buffer.
289                  */
290                 if (bytes < (int)sizeof(msg->any.head)) {
291                         n = read(iocom->sock_fd,
292                                  ioq->buf + ioq->fifo_end,
293                                  nmax);
294                         if (n <= 0) {
295                                 if (n == 0) {
296                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
297                                         break;
298                                 }
299                                 if (errno != EINTR &&
300                                     errno != EINPROGRESS &&
301                                     errno != EAGAIN) {
302                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
303                                         break;
304                                 }
305                                 n = 0;
306                                 /* fall through */
307                         }
308                         ioq->fifo_end += n;
309                         bytes += n;
310                         nmax -= n;
311                 }
312
313                 /*
314                  * Insufficient data accumulated (msg is NULL, caller will
315                  * retry on event).
316                  */
317                 assert(msg == NULL);
318                 if (bytes < (int)sizeof(msg->any.head))
319                         break;
320
321                 /*
322                  * Calculate the header, decrypt data received so far.
323                  * Data will be decrypted in-place.  Partial blocks are
324                  * not immediately decrypted.
325                  */
326                 hammer2_crypto_decrypt(iocom, ioq);
327                 head = (void *)(ioq->buf + ioq->fifo_beg);
328
329                 /*
330                  * Check and fixup the core header.  Note that the icrc
331                  * has to be calculated before any fixups, but the crc
332                  * fields in the msg may have to be swapped like everything
333                  * else.
334                  */
335                 if (head->magic != HAMMER2_MSGHDR_MAGIC &&
336                     head->magic != HAMMER2_MSGHDR_MAGIC_REV) {
337                         ioq->error = HAMMER2_IOQ_ERROR_SYNC;
338                         break;
339                 }
340
341                 xcrc32 = hammer2_icrc32((char *)head + HAMMER2_MSGHDR_CRCOFF,
342                                         HAMMER2_MSGHDR_CRCBYTES);
343                 if (head->magic == HAMMER2_MSGHDR_MAGIC_REV) {
344                         hammer2_bswap_head(head);
345                 }
346                 xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
347                 if (xcrc16 != head->icrc1) {
348                         ioq->error = HAMMER2_IOQ_ERROR_HCRC;
349                         break;
350                 }
351
352                 /*
353                  * Calculate the full header size and aux data size
354                  */
355                 ioq->hbytes = (head->cmd & HAMMER2_MSGF_SIZE) *
356                               HAMMER2_MSG_ALIGN;
357                 ioq->abytes = head->aux_bytes * HAMMER2_MSG_ALIGN;
358                 if (ioq->hbytes < sizeof(msg->any.head) ||
359                     ioq->hbytes > sizeof(msg->any) ||
360                     ioq->abytes > HAMMER2_MSGAUX_MAX) {
361                         ioq->error = HAMMER2_IOQ_ERROR_FIELD;
362                         break;
363                 }
364
365                 /*
366                  * Finally allocate the message and copy the core header
367                  * to the embedded extended header.
368                  *
369                  * Initialize msg->aux_size to 0 and use it to track
370                  * the amount of data copied from the stream.
371                  */
372                 msg = hammer2_msg_alloc(iocom, ioq->abytes, 0);
373                 ioq->msg = msg;
374
375                 /*
376                  * We are either done or we fall-through
377                  */
378                 if (ioq->hbytes == sizeof(msg->any.head) && ioq->abytes == 0) {
379                         bcopy(head, &msg->any.head, sizeof(msg->any.head));
380                         ioq->fifo_beg += ioq->hbytes;
381                         break;
382                 }
383
384                 /*
385                  * Fall through to the next state.  Make sure that the
386                  * extended header does not straddle the end of the buffer.
387                  * We still want to issue larger reads into our buffer,
388                  * book-keeping is easier if we don't bcopy() yet.
389                  */
390                 if (bytes + nmax < ioq->hbytes) {
391                         bcopy(ioq->buf + ioq->fifo_beg, ioq->buf, bytes);
392                         ioq->fifo_cdx -= ioq->fifo_beg;
393                         ioq->fifo_beg = 0;
394                         ioq->fifo_end = bytes;
395                         nmax = sizeof(ioq->buf) - ioq->fifo_end;
396                 }
397                 ioq->state = HAMMER2_MSGQ_STATE_HEADER2;
398                 /* fall through */
399         case HAMMER2_MSGQ_STATE_HEADER2:
400                 /*
401                  * Fill out the extended header.
402                  */
403                 assert(msg != NULL);
404                 if (bytes < ioq->hbytes) {
405                         n = read(iocom->sock_fd,
406                                  msg->any.buf + ioq->fifo_end,
407                                  nmax);
408                         if (n <= 0) {
409                                 if (n == 0) {
410                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
411                                         break;
412                                 }
413                                 if (errno != EINTR &&
414                                     errno != EINPROGRESS &&
415                                     errno != EAGAIN) {
416                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
417                                         break;
418                                 }
419                                 n = 0;
420                                 /* fall through */
421                         }
422                         ioq->fifo_end += n;
423                         bytes += n;
424                         nmax -= n;
425                 }
426
427                 /*
428                  * Insufficient data accumulated (set msg NULL so caller will
429                  * retry on event).
430                  */
431                 if (bytes < ioq->hbytes) {
432                         msg = NULL;
433                         break;
434                 }
435
436                 /*
437                  * Calculate the extended header, decrypt data received
438                  * so far.
439                  */
440                 hammer2_crypto_decrypt(iocom, ioq);
441                 head = (void *)(ioq->buf + ioq->fifo_beg);
442
443                 /*
444                  * Check the crc on the extended header
445                  */
446                 if (ioq->hbytes > sizeof(hammer2_msg_hdr_t)) {
447                         xcrc32 = hammer2_icrc32(head + 1,
448                                                 ioq->hbytes - sizeof(*head));
449                         xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
450                         if (head->icrc2 != xcrc16) {
451                                 ioq->error = HAMMER2_IOQ_ERROR_XCRC;
452                                 break;
453                         }
454                 }
455
456                 /*
457                  * Copy the extended header into the msg and adjust the
458                  * FIFO.
459                  */
460                 bcopy(head, &msg->any, ioq->hbytes);
461
462                 /*
463                  * We are either done or we fall-through.
464                  */
465                 if (ioq->abytes == 0) {
466                         ioq->fifo_beg += ioq->hbytes;
467                         break;
468                 }
469
470                 /*
471                  * Must adjust nmax and bytes (and the state) when falling
472                  * through.
473                  */
474                 ioq->fifo_beg += ioq->hbytes;
475                 nmax -= ioq->hbytes;
476                 bytes -= ioq->hbytes;
477                 ioq->state = HAMMER2_MSGQ_STATE_AUXDATA1;
478                 /* fall through */
479         case HAMMER2_MSGQ_STATE_AUXDATA1:
480                 /*
481                  * Copy the partial or complete payload from remaining
482                  * bytes in the FIFO.  We have to fall-through either
483                  * way so we can check the crc.
484                  *
485                  * Adjust msg->aux_size to the final actual value.
486                  */
487                 ioq->already = ioq->fifo_cdx - ioq->fifo_beg;
488                 if (ioq->already > ioq->abytes)
489                         ioq->already = ioq->abytes;
490                 if (bytes >= ioq->abytes) {
491                         bcopy(ioq->buf + ioq->fifo_beg, msg->aux_data,
492                               ioq->abytes);
493                         msg->aux_size = ioq->abytes;
494                         ioq->fifo_beg += ioq->abytes;
495                         if (ioq->fifo_cdx < ioq->fifo_beg)
496                                 ioq->fifo_cdx = ioq->fifo_beg;
497                         bytes -= ioq->abytes;
498                 } else if (bytes) {
499                         bcopy(ioq->buf + ioq->fifo_beg, msg->aux_data,
500                               bytes);
501                         msg->aux_size = bytes;
502                         ioq->fifo_beg += bytes;
503                         if (ioq->fifo_cdx < ioq->fifo_beg)
504                                 ioq->fifo_cdx = ioq->fifo_beg;
505                         bytes = 0;
506                 } else {
507                         msg->aux_size = 0;
508                 }
509                 ioq->state = HAMMER2_MSGQ_STATE_AUXDATA2;
510                 /* fall through */
511         case HAMMER2_MSGQ_STATE_AUXDATA2:
512                 /*
513                  * Read the remainder of the payload directly into the
514                  * msg->aux_data buffer.
515                  */
516                 assert(msg);
517                 if (msg->aux_size < ioq->abytes) {
518                         assert(bytes == 0);
519                         n = read(iocom->sock_fd,
520                                  msg->aux_data + msg->aux_size,
521                                  ioq->abytes - msg->aux_size);
522                         if (n <= 0) {
523                                 if (n == 0) {
524                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
525                                         break;
526                                 }
527                                 if (errno != EINTR &&
528                                     errno != EINPROGRESS &&
529                                     errno != EAGAIN) {
530                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
531                                         break;
532                                 }
533                                 n = 0;
534                                 /* fall through */
535                         }
536                         msg->aux_size += n;
537                 }
538
539                 /*
540                  * Insufficient data accumulated (set msg NULL so caller will
541                  * retry on event).
542                  */
543                 if (msg->aux_size < ioq->abytes) {
544                         msg = NULL;
545                         break;
546                 }
547                 assert(msg->aux_size == ioq->abytes);
548                 hammer2_crypto_decrypt_aux(iocom, ioq, msg, ioq->already);
549
550                 /*
551                  * Check aux_icrc, then we are done.
552                  */
553                 xcrc32 = hammer2_icrc32(msg->aux_data, msg->aux_size);
554                 if (xcrc32 != msg->any.head.aux_icrc) {
555                         ioq->error = HAMMER2_IOQ_ERROR_ACRC;
556                         break;
557                 }
558                 break;
559         case HAMMER2_MSGQ_STATE_ERROR:
560                 /*
561                  * Continued calls to drain recorded transactions (returning
562                  * a LNK_ERROR for each one), before we return the final
563                  * LNK_ERROR.
564                  */
565                 assert(msg == NULL);
566                 break;
567         default:
568                 /*
569                  * We don't double-return errors, the caller should not
570                  * have called us again after getting an error msg.
571                  */
572                 assert(0);
573                 break;
574         }
575
576         /*
577          * Check the message sequence.  The iv[] should prevent any
578          * possibility of a replay but we add this check anyway.
579          */
580         if (msg && ioq->error == 0) {
581                 if ((msg->any.head.salt & 255) != (ioq->seq & 255)) {
582                         ioq->error = HAMMER2_IOQ_ERROR_MSGSEQ;
583                 } else {
584                         ++ioq->seq;
585                 }
586         }
587
588         /*
589          * Process transactional state for the message.
590          */
591         if (msg && ioq->error == 0) {
592                 error = hammer2_state_msgrx(iocom, msg);
593                 if (error) {
594                         if (error == HAMMER2_IOQ_ERROR_EALREADY) {
595                                 hammer2_msg_free(iocom, msg);
596                                 goto again;
597                         }
598                         ioq->error = error;
599                 }
600         }
601
602         /*
603          * Handle error, RREQ, or completion
604          *
605          * NOTE: nmax and bytes are invalid at this point, we don't bother
606          *       to update them when breaking out.
607          */
608         if (ioq->error) {
609                 /*
610                  * An unrecoverable error causes all active receive
611                  * transactions to be terminated with a LNK_ERROR message.
612                  *
613                  * Once all active transactions are exhausted we set the
614                  * iocom ERROR flag and return a non-transactional LNK_ERROR
615                  * message, which should cause master processing loops to
616                  * terminate.
617                  */
618                 assert(ioq->msg == msg);
619                 if (msg) {
620                         hammer2_msg_free(iocom, msg);
621                         ioq->msg = NULL;
622                 }
623
624                 /*
625                  * No more I/O read processing
626                  */
627                 ioq->state = HAMMER2_MSGQ_STATE_ERROR;
628
629                 /*
630                  * Return LNK_ERROR for any open transaction, and finally
631                  * as a non-transactional message when no transactions are
632                  * left.
633                  */
634                 msg = hammer2_msg_alloc(iocom, 0, 0);
635                 bzero(&msg->any.head, sizeof(msg->any.head));
636                 msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
637                 msg->any.head.cmd = HAMMER2_LNK_ERROR;
638                 msg->any.head.error = ioq->error;
639
640                 if ((state = RB_ROOT(&iocom->staterd_tree)) != NULL) {
641                         /*
642                          * Active transactions are still present.  Simulate
643                          * the other end sending us a DELETE.
644                          */
645                         state->txcmd |= HAMMER2_MSGF_DELETE;
646                         msg->state = state;
647                         msg->any.head.source = state->source;
648                         msg->any.head.target = state->target;
649                         msg->any.head.cmd |= HAMMER2_MSGF_ABORT |
650                                              HAMMER2_MSGF_DELETE;
651                 } else {
652                         /*
653                          * No active transactions remain
654                          */
655                         msg->state = NULL;
656                         iocom->flags |= HAMMER2_IOCOMF_EOF;
657                 }
658         } else if (msg == NULL) {
659                 /*
660                  * Insufficient data received to finish building the message,
661                  * set RREQ and return NULL.
662                  *
663                  * Leave ioq->msg intact.
664                  * Leave the FIFO intact.
665                  */
666                 iocom->flags |= HAMMER2_IOCOMF_RREQ;
667 #if 0
668                 ioq->fifo_cdx = 0;
669                 ioq->fifo_beg = 0;
670                 ioq->fifo_end = 0;
671 #endif
672         } else {
673                 /*
674                  * Return msg, clear the FIFO if it is now empty.
675                  * Flag RREQ if the caller needs to wait for a read-event
676                  * or not.
677                  *
678                  * The fifo has already been advanced past the message.
679                  * Trivially reset the FIFO indices if possible.
680                  */
681                 if (ioq->fifo_beg == ioq->fifo_end) {
682                         iocom->flags |= HAMMER2_IOCOMF_RREQ;
683                         ioq->fifo_cdx = 0;
684                         ioq->fifo_beg = 0;
685                         ioq->fifo_end = 0;
686                 } else {
687                         iocom->flags &= ~HAMMER2_IOCOMF_RREQ;
688                 }
689                 ioq->state = HAMMER2_MSGQ_STATE_HEADER1;
690                 ioq->msg = NULL;
691         }
692         return (msg);
693 }
694
695 /*
696  * Calculate the header and data crc's and write a low-level message to
697  * the connection.  If aux_icrc is non-zero the aux_data crc is already
698  * assumed to have been set.
699  *
700  * A non-NULL msg is added to the queue but not necessarily flushed.
701  * Calling this function with msg == NULL will get a flush going.
702  */
703 void
704 hammer2_ioq_write(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
705 {
706         hammer2_ioq_t *ioq = &iocom->ioq_tx;
707         uint16_t xcrc16;
708         uint32_t xcrc32;
709         int hbytes;
710         int error;
711
712         assert(msg);
713
714         /*
715          * Process transactional state.
716          */
717         if (ioq->error == 0) {
718                 error = hammer2_state_msgtx(iocom, msg);
719                 if (error) {
720                         if (error == HAMMER2_IOQ_ERROR_EALREADY) {
721                                 hammer2_msg_free(iocom, msg);
722                         } else {
723                                 ioq->error = error;
724                         }
725                 }
726         }
727
728         /*
729          * Process terminal connection errors.
730          */
731         if (ioq->error) {
732                 TAILQ_INSERT_TAIL(&ioq->msgq, msg, qentry);
733                 ++ioq->msgcount;
734                 hammer2_iocom_drain(iocom);
735                 return;
736         }
737
738         /*
739          * Finish populating the msg fields.  The salt ensures that the iv[]
740          * array is ridiculously randomized and we also re-seed our PRNG
741          * every 32768 messages just to be sure.
742          */
743         msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
744         msg->any.head.salt = (random() << 8) | (ioq->seq & 255);
745         ++ioq->seq;
746         if ((ioq->seq & 32767) == 0)
747                 srandomdev();
748
749         /*
750          * Calculate aux_icrc if 0, calculate icrc2, and finally
751          * calculate icrc1.
752          */
753         if (msg->aux_size && msg->any.head.aux_icrc == 0) {
754                 assert((msg->aux_size & HAMMER2_MSG_ALIGNMASK) == 0);
755                 xcrc32 = hammer2_icrc32(msg->aux_data, msg->aux_size);
756                 msg->any.head.aux_icrc = xcrc32;
757         }
758         msg->any.head.aux_bytes = msg->aux_size / HAMMER2_MSG_ALIGN;
759         assert((msg->aux_size & HAMMER2_MSG_ALIGNMASK) == 0);
760
761         if ((msg->any.head.cmd & HAMMER2_MSGF_SIZE) >
762             sizeof(msg->any.head) / HAMMER2_MSG_ALIGN) {
763                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
764                         HAMMER2_MSG_ALIGN;
765                 hbytes -= sizeof(msg->any.head);
766                 xcrc32 = hammer2_icrc32(&msg->any.head + 1, hbytes);
767                 xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
768                 msg->any.head.icrc2 = xcrc16;
769         } else {
770                 msg->any.head.icrc2 = 0;
771         }
772         xcrc32 = hammer2_icrc32(msg->any.buf + HAMMER2_MSGHDR_CRCOFF,
773                                 HAMMER2_MSGHDR_CRCBYTES);
774         xcrc16 = (uint16_t)xcrc32 ^ (uint16_t)(xcrc32 >> 16);
775         msg->any.head.icrc1 = xcrc16;
776
777         /*
778          * XXX Encrypt the message
779          */
780
781         /*
782          * Enqueue the message.
783          */
784         TAILQ_INSERT_TAIL(&ioq->msgq, msg, qentry);
785         ++ioq->msgcount;
786         iocom->flags &= ~HAMMER2_IOCOMF_WIDLE;
787
788         /*
789          * Flush if we know we can write (WREQ not set) and if
790          * sufficient messages have accumulated.  Otherwise hold
791          * off to avoid piecemeal system calls.
792          */
793         if (iocom->flags & HAMMER2_IOCOMF_WREQ)
794                 return;
795         if (ioq->msgcount < HAMMER2_IOQ_MAXIOVEC / 2)
796                 return;
797         hammer2_iocom_flush(iocom);
798 }
799
800 void
801 hammer2_iocom_flush(hammer2_iocom_t *iocom)
802 {
803         hammer2_ioq_t *ioq = &iocom->ioq_tx;
804         hammer2_msg_t *msg;
805         ssize_t nmax;
806         ssize_t nact;
807         struct iovec iov[HAMMER2_IOQ_MAXIOVEC];
808         size_t hbytes;
809         size_t abytes;
810         int hoff;
811         int aoff;
812         int n;
813
814         /*
815          * Pump messages out the connection by building an iovec.
816          */
817         n = 0;
818         nmax = 0;
819
820         TAILQ_FOREACH(msg, &ioq->msgq, qentry) {
821                 hoff = 0;
822                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
823                          HAMMER2_MSG_ALIGN;
824                 aoff = 0;
825                 abytes = msg->aux_size;
826                 if (n == 0) {
827                         hoff += ioq->hbytes;
828                         aoff += ioq->abytes;
829                 }
830                 if (hbytes - hoff > 0) {
831                         iov[n].iov_base = (char *)&msg->any.head + hoff;
832                         iov[n].iov_len = hbytes - hoff;
833                         nmax += hbytes - hoff;
834                         ++n;
835                         if (n == HAMMER2_IOQ_MAXIOVEC)
836                                 break;
837                 }
838                 if (abytes - aoff > 0) {
839                         assert(msg->aux_data != NULL);
840                         iov[n].iov_base = msg->aux_data + aoff;
841                         iov[n].iov_len = abytes - aoff;
842                         nmax += abytes - aoff;
843                         ++n;
844                         if (n == HAMMER2_IOQ_MAXIOVEC)
845                                 break;
846                 }
847         }
848         if (n == 0)
849                 return;
850
851         /*
852          * Encrypt and write the data.  The crypto code will move the
853          * data into the fifo and adjust the iov as necessary.  If
854          * encryption is disabled the iov is left alone.
855          *
856          * hammer2_crypto_encrypt_wrote()
857          */
858         n = hammer2_crypto_encrypt(iocom, ioq, iov, n);
859
860         /*
861          * Execute the writev() then figure out what happened.
862          */
863         nact = writev(iocom->sock_fd, iov, n);
864         if (nact < 0) {
865                 if (errno != EINTR &&
866                     errno != EINPROGRESS &&
867                     errno != EAGAIN) {
868                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
869                         hammer2_iocom_drain(iocom);
870                 } else {
871                         iocom->flags |= HAMMER2_IOCOMF_WREQ;
872                 }
873                 return;
874         }
875         hammer2_crypto_encrypt_wrote(iocom, ioq, nact);
876         if (nact == nmax)
877                 iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
878         else
879                 iocom->flags |= HAMMER2_IOCOMF_WREQ;
880
881         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
882                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
883                          HAMMER2_MSG_ALIGN;
884                 abytes = msg->aux_size;
885
886                 if ((size_t)nact < hbytes - ioq->hbytes) {
887                         ioq->hbytes += nact;
888                         break;
889                 }
890                 nact -= hbytes - ioq->hbytes;
891                 ioq->hbytes = hbytes;
892                 if ((size_t)nact < abytes - ioq->abytes) {
893                         ioq->abytes += nact;
894                         break;
895                 }
896                 nact -= abytes - ioq->abytes;
897
898                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
899                 --ioq->msgcount;
900                 ioq->hbytes = 0;
901                 ioq->abytes = 0;
902
903                 hammer2_state_cleanuptx(iocom, msg);
904         }
905         if (msg == NULL) {
906                 iocom->flags |= HAMMER2_IOCOMF_WIDLE;
907                 iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
908         }
909         if (ioq->error) {
910                 iocom->flags |= HAMMER2_IOCOMF_EOF |
911                                 HAMMER2_IOCOMF_WIDLE;
912                 iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
913         }
914 }
915
916 /*
917  * Kill pending msgs on ioq_tx and adjust the flags such that no more
918  * write events will occur.  We don't kill read msgs because we want
919  * the caller to pull off our contrived terminal error msg to detect
920  * the connection failure.
921  */
922 void
923 hammer2_iocom_drain(hammer2_iocom_t *iocom)
924 {
925         hammer2_ioq_t *ioq = &iocom->ioq_tx;
926         hammer2_msg_t *msg;
927
928         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
929                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
930                 --ioq->msgcount;
931                 hammer2_msg_free(iocom, msg);
932         }
933         iocom->flags |= HAMMER2_IOCOMF_WIDLE;
934         iocom->flags &= ~HAMMER2_IOCOMF_WREQ;
935 }
936
937 /*
938  * This is a shortcut to formulate a reply to msg with a simple error code.
939  * It can reply to transaction or one-way messages, or terminate one side
940  * of a stream.  A HAMMER2_LNK_ERROR command code is utilized to encode
941  * the error code (which can be 0).
942  *
943  * Replies to one-way messages are a bit of an oxymoron but the feature
944  * is used by the debug (DBG) protocol.
945  *
946  * The reply contains no data.
947  */
948 void
949 hammer2_msg_reply(hammer2_iocom_t *iocom, hammer2_msg_t *msg, uint16_t error)
950 {
951         hammer2_msg_t *nmsg;
952         uint32_t cmd;
953
954         cmd = HAMMER2_LNK_ERROR;
955         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
956                 /*
957                  * Reply to received reply, reply direction uses txcmd.
958                  * txcmd will be updated by hammer2_ioq_write().
959                  */
960                 if (msg->state) {
961                         if ((msg->state->rxcmd & HAMMER2_MSGF_CREATE) == 0)
962                                 cmd |= HAMMER2_MSGF_CREATE;
963                         cmd |= HAMMER2_MSGF_DELETE;
964                 }
965         } else {
966                 /*
967                  * Reply to received command, reply direction uses rxcmd.
968                  * txcmd will be updated by hammer2_ioq_write().
969                  */
970                 cmd |= HAMMER2_MSGF_REPLY;
971                 if (msg->state) {
972                         if ((msg->state->rxcmd & HAMMER2_MSGF_CREATE) == 0)
973                                 cmd |= HAMMER2_MSGF_CREATE;
974                         cmd |= HAMMER2_MSGF_DELETE;
975                 }
976         }
977         nmsg = hammer2_msg_alloc(iocom, 0, cmd);
978         nmsg->any.head.error = error;
979         hammer2_ioq_write(iocom, nmsg);
980 }
981
982 /************************************************************************
983  *                      TRANSACTION STATE HANDLING                      *
984  ************************************************************************
985  *
986  */
987
988 RB_GENERATE(hammer2_state_tree, hammer2_state, rbnode, hammer2_state_cmp);
989
990 /*
991  * Process state tracking for a message after reception, prior to
992  * execution.
993  *
994  * Called with msglk held and the msg dequeued.
995  *
996  * All messages are called with dummy state and return actual state.
997  * (One-off messages often just return the same dummy state).
998  *
999  * May request that caller discard the message by setting *discardp to 1.
1000  * The returned state is not used in this case and is allowed to be NULL.
1001  *
1002  * --
1003  *
1004  * These routines handle persistent and command/reply message state via the
1005  * CREATE and DELETE flags.  The first message in a command or reply sequence
1006  * sets CREATE, the last message in a command or reply sequence sets DELETE.
1007  *
1008  * There can be any number of intermediate messages belonging to the same
1009  * sequence sent inbetween the CREATE message and the DELETE message,
1010  * which set neither flag.  This represents a streaming command or reply.
1011  *
1012  * Any command message received with CREATE set expects a reply sequence to
1013  * be returned.  Reply sequences work the same as command sequences except the
1014  * REPLY bit is also sent.  Both the command side and reply side can
1015  * degenerate into a single message with both CREATE and DELETE set.  Note
1016  * that one side can be streaming and the other side not, or neither, or both.
1017  *
1018  * The msgid is unique for the initiator.  That is, two sides sending a new
1019  * message can use the same msgid without colliding.
1020  *
1021  * --
1022  *
1023  * ABORT sequences work by setting the ABORT flag along with normal message
1024  * state.  However, ABORTs can also be sent on half-closed messages, that is
1025  * even if the command or reply side has already sent a DELETE, as long as
1026  * the message has not been fully closed it can still send an ABORT+DELETE
1027  * to terminate the half-closed message state.
1028  *
1029  * Since ABORT+DELETEs can race we silently discard ABORT's for message
1030  * state which has already been fully closed.  REPLY+ABORT+DELETEs can
1031  * also race, and in this situation the other side might have already
1032  * initiated a new unrelated command with the same message id.  Since
1033  * the abort has not set the CREATE flag the situation can be detected
1034  * and the message will also be discarded.
1035  *
1036  * Non-blocking requests can be initiated with ABORT+CREATE[+DELETE].
1037  * The ABORT request is essentially integrated into the command instead
1038  * of being sent later on.  In this situation the command implementation
1039  * detects that CREATE and ABORT are both set (vs ABORT alone) and can
1040  * special-case non-blocking operation for the command.
1041  *
1042  * NOTE!  Messages with ABORT set without CREATE or DELETE are considered
1043  *        to be mid-stream aborts for command/reply sequences.  ABORTs on
1044  *        one-way messages are not supported.
1045  *
1046  * NOTE!  If a command sequence does not support aborts the ABORT flag is
1047  *        simply ignored.
1048  *
1049  * --
1050  *
1051  * One-off messages (no reply expected) are sent with neither CREATE or DELETE
1052  * set.  One-off messages cannot be aborted and typically aren't processed
1053  * by these routines.  The REPLY bit can be used to distinguish whether a
1054  * one-off message is a command or reply.  For example, one-off replies
1055  * will typically just contain status updates.
1056  */
1057 static int
1058 hammer2_state_msgrx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1059 {
1060         hammer2_state_t *state;
1061         hammer2_state_t dummy;
1062         int error;
1063
1064         /*
1065          * Lock RB tree and locate existing persistent state, if any.
1066          *
1067          * If received msg is a command state is on staterd_tree.
1068          * If received msg is a reply state is on statewr_tree.
1069          */
1070         /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1071
1072         dummy.msgid = msg->any.head.msgid;
1073         dummy.source = msg->any.head.source;
1074         dummy.target = msg->any.head.target;
1075         iocom_printf(iocom, msg->any.head.cmd,
1076                      "received msg %08x msgid %u source=%u target=%u\n",
1077                       msg->any.head.cmd, msg->any.head.msgid,
1078                       msg->any.head.source, msg->any.head.target);
1079         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1080                 state = RB_FIND(hammer2_state_tree,
1081                                 &iocom->statewr_tree, &dummy);
1082         } else {
1083                 state = RB_FIND(hammer2_state_tree,
1084                                 &iocom->staterd_tree, &dummy);
1085         }
1086         msg->state = state;
1087
1088         /*
1089          * Short-cut one-off or mid-stream messages (state may be NULL).
1090          */
1091         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1092                                   HAMMER2_MSGF_ABORT)) == 0) {
1093                 /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1094                 return(0);
1095         }
1096
1097         /*
1098          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
1099          * inside the case statements.
1100          */
1101         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1102                                     HAMMER2_MSGF_REPLY)) {
1103         case HAMMER2_MSGF_CREATE:
1104         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1105                 /*
1106                  * New persistant command received.
1107                  */
1108                 if (state) {
1109                         iocom_printf(iocom, msg->any.head.cmd,
1110                                      "hammer2_state_msgrx: "
1111                                      "duplicate transaction\n");
1112                         error = HAMMER2_IOQ_ERROR_TRANS;
1113                         break;
1114                 }
1115                 state = malloc(sizeof(*state));
1116                 bzero(state, sizeof(*state));
1117                 state->iocom = iocom;
1118                 state->flags = HAMMER2_STATE_DYNAMIC;
1119                 state->msg = msg;
1120                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1121                 RB_INSERT(hammer2_state_tree, &iocom->staterd_tree, state);
1122                 state->flags |= HAMMER2_STATE_INSERTED;
1123                 msg->state = state;
1124                 error = 0;
1125                 break;
1126         case HAMMER2_MSGF_DELETE:
1127                 /*
1128                  * Persistent state is expected but might not exist if an
1129                  * ABORT+DELETE races the close.
1130                  */
1131                 if (state == NULL) {
1132                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1133                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1134                         } else {
1135                                 iocom_printf(iocom, msg->any.head.cmd,
1136                                              "hammer2_state_msgrx: "
1137                                              "no state for DELETE\n");
1138                                 error = HAMMER2_IOQ_ERROR_TRANS;
1139                         }
1140                         break;
1141                 }
1142
1143                 /*
1144                  * Handle another ABORT+DELETE case if the msgid has already
1145                  * been reused.
1146                  */
1147                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1148                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1149                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1150                         } else {
1151                                 iocom_printf(iocom, msg->any.head.cmd,
1152                                              "hammer2_state_msgrx: "
1153                                              "state reused for DELETE\n");
1154                                 error = HAMMER2_IOQ_ERROR_TRANS;
1155                         }
1156                         break;
1157                 }
1158                 error = 0;
1159                 break;
1160         default:
1161                 /*
1162                  * Check for mid-stream ABORT command received, otherwise
1163                  * allow.
1164                  */
1165                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1166                         if (state == NULL ||
1167                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1168                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1169                                 break;
1170                         }
1171                 }
1172                 error = 0;
1173                 break;
1174         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
1175         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1176                 /*
1177                  * When receiving a reply with CREATE set the original
1178                  * persistent state message should already exist.
1179                  */
1180                 if (state == NULL) {
1181                         iocom_printf(iocom, msg->any.head.cmd,
1182                                      "hammer2_state_msgrx: "
1183                                      "no state match for REPLY cmd=%08x\n",
1184                                      msg->any.head.cmd);
1185                         error = HAMMER2_IOQ_ERROR_TRANS;
1186                         break;
1187                 }
1188                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1189                 error = 0;
1190                 break;
1191         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
1192                 /*
1193                  * Received REPLY+ABORT+DELETE in case where msgid has
1194                  * already been fully closed, ignore the message.
1195                  */
1196                 if (state == NULL) {
1197                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1198                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1199                         } else {
1200                                 iocom_printf(iocom, msg->any.head.cmd,
1201                                              "hammer2_state_msgrx: "
1202                                              "no state match for "
1203                                              "REPLY|DELETE\n");
1204                                 error = HAMMER2_IOQ_ERROR_TRANS;
1205                         }
1206                         break;
1207                 }
1208
1209                 /*
1210                  * Received REPLY+ABORT+DELETE in case where msgid has
1211                  * already been reused for an unrelated message,
1212                  * ignore the message.
1213                  */
1214                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1215                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1216                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1217                         } else {
1218                                 iocom_printf(iocom, msg->any.head.cmd,
1219                                              "hammer2_state_msgrx: "
1220                                              "state reused for REPLY|DELETE\n");
1221                                 error = HAMMER2_IOQ_ERROR_TRANS;
1222                         }
1223                         break;
1224                 }
1225                 error = 0;
1226                 break;
1227         case HAMMER2_MSGF_REPLY:
1228                 /*
1229                  * Check for mid-stream ABORT reply received to sent command.
1230                  */
1231                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1232                         if (state == NULL ||
1233                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1234                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1235                                 break;
1236                         }
1237                 }
1238                 error = 0;
1239                 break;
1240         }
1241         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1242         return (error);
1243 }
1244
1245 void
1246 hammer2_state_cleanuprx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1247 {
1248         hammer2_state_t *state;
1249
1250         if ((state = msg->state) == NULL) {
1251                 /*
1252                  * Free a non-transactional message, there is no state
1253                  * to worry about.
1254                  */
1255                 hammer2_msg_free(iocom, msg);
1256         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
1257                 /*
1258                  * Message terminating transaction, destroy the related
1259                  * state, the original message, and this message (if it
1260                  * isn't the original message due to a CREATE|DELETE).
1261                  */
1262                 /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1263                 state->rxcmd |= HAMMER2_MSGF_DELETE;
1264                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
1265                         if (state->msg == msg)
1266                                 state->msg = NULL;
1267                         assert(state->flags & HAMMER2_STATE_INSERTED);
1268                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1269                                 RB_REMOVE(hammer2_state_tree,
1270                                           &iocom->statewr_tree, state);
1271                         } else {
1272                                 RB_REMOVE(hammer2_state_tree,
1273                                           &iocom->staterd_tree, state);
1274                         }
1275                         state->flags &= ~HAMMER2_STATE_INSERTED;
1276                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1277                         hammer2_state_free(state);
1278                 } else {
1279                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1280                 }
1281                 hammer2_msg_free(iocom, msg);
1282         } else if (state->msg != msg) {
1283                 /*
1284                  * Message not terminating transaction, leave state intact
1285                  * and free message if it isn't the CREATE message.
1286                  */
1287                 hammer2_msg_free(iocom, msg);
1288         }
1289 }
1290
1291 /*
1292  * Process state tracking for a message prior to transmission.
1293  *
1294  * Called with msglk held and the msg dequeued.
1295  *
1296  * One-off messages are usually with dummy state and msg->state may be NULL
1297  * in this situation.
1298  *
1299  * New transactions (when CREATE is set) will insert the state.
1300  *
1301  * May request that caller discard the message by setting *discardp to 1.
1302  * A NULL state may be returned in this case.
1303  */
1304 static int
1305 hammer2_state_msgtx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1306 {
1307         hammer2_state_t *state;
1308         int error;
1309
1310         /*
1311          * Lock RB tree.  If persistent state is present it will have already
1312          * been assigned to msg.
1313          */
1314         /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1315         state = msg->state;
1316
1317         /*
1318          * Short-cut one-off or mid-stream messages (state may be NULL).
1319          */
1320         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1321                                   HAMMER2_MSGF_ABORT)) == 0) {
1322                 /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1323                 return(0);
1324         }
1325
1326
1327         /*
1328          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
1329          * inside the case statements.
1330          */
1331         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1332                                     HAMMER2_MSGF_REPLY)) {
1333         case HAMMER2_MSGF_CREATE:
1334         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1335                 /*
1336                  * Insert the new persistent message state and mark
1337                  * half-closed if DELETE is set.  Since this is a new
1338                  * message it isn't possible to transition into the fully
1339                  * closed state here.
1340                  *
1341                  * XXX state must be assigned and inserted by
1342                  *     hammer2_msg_write().  txcmd is assigned by us
1343                  *     on-transmit.
1344                  */
1345                 assert(state != NULL);
1346 #if 0
1347                 if (state == NULL) {
1348                         state = pmp->freerd_state;
1349                         pmp->freerd_state = NULL;
1350                         msg->state = state;
1351                         state->msg = msg;
1352                         state->msgid = msg->any.head.msgid;
1353                         state->source = msg->any.head.source;
1354                         state->target = msg->any.head.target;
1355                 }
1356                 assert((state->flags & HAMMER2_STATE_INSERTED) == 0);
1357                 if (RB_INSERT(hammer2_state_tree, &pmp->staterd_tree, state)) {
1358                         iocom_printf(iocom, msg->any.head.cmd,
1359                                     "hammer2_state_msgtx: "
1360                                     "duplicate transaction\n");
1361                         error = HAMMER2_IOQ_ERROR_TRANS;
1362                         break;
1363                 }
1364                 state->flags |= HAMMER2_STATE_INSERTED;
1365 #endif
1366                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1367                 error = 0;
1368                 break;
1369         case HAMMER2_MSGF_DELETE:
1370                 /*
1371                  * Sent ABORT+DELETE in case where msgid has already
1372                  * been fully closed, ignore the message.
1373                  */
1374                 if (state == NULL) {
1375                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1376                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1377                         } else {
1378                                 iocom_printf(iocom, msg->any.head.cmd,
1379                                              "hammer2_state_msgtx: "
1380                                              "no state match for DELETE\n");
1381                                 error = HAMMER2_IOQ_ERROR_TRANS;
1382                         }
1383                         break;
1384                 }
1385
1386                 /*
1387                  * Sent ABORT+DELETE in case where msgid has
1388                  * already been reused for an unrelated message,
1389                  * ignore the message.
1390                  */
1391                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1392                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1393                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1394                         } else {
1395                                 iocom_printf(iocom, msg->any.head.cmd,
1396                                              "hammer2_state_msgtx: "
1397                                              "state reused for DELETE\n");
1398                                 error = HAMMER2_IOQ_ERROR_TRANS;
1399                         }
1400                         break;
1401                 }
1402                 error = 0;
1403                 break;
1404         default:
1405                 /*
1406                  * Check for mid-stream ABORT command sent
1407                  */
1408                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1409                         if (state == NULL ||
1410                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1411                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1412                                 break;
1413                         }
1414                 }
1415                 error = 0;
1416                 break;
1417         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
1418         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1419                 /*
1420                  * When transmitting a reply with CREATE set the original
1421                  * persistent state message should already exist.
1422                  */
1423                 if (state == NULL) {
1424                         iocom_printf(iocom, msg->any.head.cmd,
1425                                      "hammer2_state_msgtx: no state match "
1426                                      "for REPLY | CREATE\n");
1427                         error = HAMMER2_IOQ_ERROR_TRANS;
1428                         break;
1429                 }
1430                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1431                 error = 0;
1432                 break;
1433         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
1434                 /*
1435                  * When transmitting a reply with DELETE set the original
1436                  * persistent state message should already exist.
1437                  *
1438                  * This is very similar to the REPLY|CREATE|* case except
1439                  * txcmd is already stored, so we just add the DELETE flag.
1440                  *
1441                  * Sent REPLY+ABORT+DELETE in case where msgid has
1442                  * already been fully closed, ignore the message.
1443                  */
1444                 if (state == NULL) {
1445                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1446                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1447                         } else {
1448                                 iocom_printf(iocom, msg->any.head.cmd,
1449                                              "hammer2_state_msgtx: "
1450                                              "no state match for "
1451                                              "REPLY | DELETE\n");
1452                                 error = HAMMER2_IOQ_ERROR_TRANS;
1453                         }
1454                         break;
1455                 }
1456
1457                 /*
1458                  * Sent REPLY+ABORT+DELETE in case where msgid has already
1459                  * been reused for an unrelated message, ignore the message.
1460                  */
1461                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1462                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1463                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1464                         } else {
1465                                 iocom_printf(iocom, msg->any.head.cmd,
1466                                              "hammer2_state_msgtx: "
1467                                              "state reused for "
1468                                              "REPLY | DELETE\n");
1469                                 error = HAMMER2_IOQ_ERROR_TRANS;
1470                         }
1471                         break;
1472                 }
1473                 error = 0;
1474                 break;
1475         case HAMMER2_MSGF_REPLY:
1476                 /*
1477                  * Check for mid-stream ABORT reply sent.
1478                  *
1479                  * One-off REPLY messages are allowed for e.g. status updates.
1480                  */
1481                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1482                         if (state == NULL ||
1483                             (state->txcmd & HAMMER2_MSGF_CREATE) == 0) {
1484                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1485                                 break;
1486                         }
1487                 }
1488                 error = 0;
1489                 break;
1490         }
1491         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1492         return (error);
1493 }
1494
1495 static void
1496 hammer2_state_cleanuptx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1497 {
1498         hammer2_state_t *state;
1499
1500         if ((state = msg->state) == NULL) {
1501                 hammer2_msg_free(iocom, msg);
1502         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
1503                 /*lockmgr(&pmp->msglk, LK_EXCLUSIVE);*/
1504                 state->txcmd |= HAMMER2_MSGF_DELETE;
1505                 if (state->rxcmd & HAMMER2_MSGF_DELETE) {
1506                         if (state->msg == msg)
1507                                 state->msg = NULL;
1508                         assert(state->flags & HAMMER2_STATE_INSERTED);
1509                         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1510                                 RB_REMOVE(hammer2_state_tree,
1511                                           &iocom->staterd_tree, state);
1512                         } else {
1513                                 RB_REMOVE(hammer2_state_tree,
1514                                           &iocom->statewr_tree, state);
1515                         }
1516                         state->flags &= ~HAMMER2_STATE_INSERTED;
1517                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1518                         hammer2_state_free(state);
1519                 } else {
1520                         /*lockmgr(&pmp->msglk, LK_RELEASE);*/
1521                 }
1522                 hammer2_msg_free(iocom, msg);
1523         } else if (state->msg != msg) {
1524                 hammer2_msg_free(iocom, msg);
1525         }
1526 }
1527
1528 void
1529 hammer2_state_free(hammer2_state_t *state)
1530 {
1531         hammer2_iocom_t *iocom = state->iocom;
1532         hammer2_msg_t *msg;
1533
1534         msg = state->msg;
1535         state->msg = NULL;
1536         if (msg)
1537                 hammer2_msg_free(iocom, msg);
1538         free(state);
1539 }
1540
1541 /*
1542  * Indexed messages are stored in a red-black tree indexed by their
1543  * msgid.  Only persistent messages are indexed.
1544  */
1545 int
1546 hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
1547 {
1548         if (state1->source < state2->source)
1549                 return(-1);
1550         if (state1->source > state2->source)
1551                 return(1);
1552         if (state1->target < state2->target)
1553                 return(-1);
1554         if (state1->target > state2->target)
1555                 return(1);
1556         if (state1->msgid < state2->msgid)
1557                 return(-1);
1558         if (state1->msgid > state2->msgid)
1559                 return(1);
1560         return(0);
1561 }