hammer2 - SPAN protocol work
[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 void hammer2_state_cleanuptx(hammer2_iocom_t *iocom, hammer2_msg_t *msg);
40
41 /*
42  * Indexed messages are stored in a red-black tree indexed by their
43  * msgid.  Only persistent messages are indexed.
44  */
45 int
46 hammer2_state_cmp(hammer2_state_t *state1, hammer2_state_t *state2)
47 {
48         if (state1->spanid < state2->spanid)
49                 return(-1);
50         if (state1->spanid > state2->spanid)
51                 return(1);
52         if (state1->msgid < state2->msgid)
53                 return(-1);
54         if (state1->msgid > state2->msgid)
55                 return(1);
56         return(0);
57 }
58
59 RB_GENERATE(hammer2_state_tree, hammer2_state, rbnode, hammer2_state_cmp);
60
61 /*
62  * Initialize a low-level ioq
63  */
64 void
65 hammer2_ioq_init(hammer2_iocom_t *iocom __unused, hammer2_ioq_t *ioq)
66 {
67         bzero(ioq, sizeof(*ioq));
68         ioq->state = HAMMER2_MSGQ_STATE_HEADER1;
69         TAILQ_INIT(&ioq->msgq);
70 }
71
72 /*
73  * Cleanup queue.
74  *
75  * caller holds iocom->mtx.
76  */
77 void
78 hammer2_ioq_done(hammer2_iocom_t *iocom __unused, hammer2_ioq_t *ioq)
79 {
80         hammer2_msg_t *msg;
81
82         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
83                 assert(0);      /* shouldn't happen */
84                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
85                 hammer2_msg_free(iocom, msg);
86         }
87         if ((msg = ioq->msg) != NULL) {
88                 ioq->msg = NULL;
89                 hammer2_msg_free(iocom, msg);
90         }
91 }
92
93 /*
94  * Initialize a low-level communications channel.
95  *
96  * NOTE: The state_func() is called at least once from the loop and can be
97  *       re-armed via hammer2_iocom_restate().
98  */
99 void
100 hammer2_iocom_init(hammer2_iocom_t *iocom, int sock_fd, int alt_fd,
101                    void (*state_func)(hammer2_iocom_t *),
102                    void (*rcvmsg_func)(hammer2_iocom_t *, hammer2_msg_t *msg),
103                    void (*altmsg_func)(hammer2_iocom_t *))
104 {
105         bzero(iocom, sizeof(*iocom));
106
107         iocom->state_callback = state_func;
108         iocom->rcvmsg_callback = rcvmsg_func;
109         iocom->altmsg_callback = altmsg_func;
110
111         pthread_mutex_init(&iocom->mtx, NULL);
112         RB_INIT(&iocom->staterd_tree);
113         RB_INIT(&iocom->statewr_tree);
114         TAILQ_INIT(&iocom->freeq);
115         TAILQ_INIT(&iocom->freeq_aux);
116         TAILQ_INIT(&iocom->addrq);
117         TAILQ_INIT(&iocom->txmsgq);
118         iocom->sock_fd = sock_fd;
119         iocom->alt_fd = alt_fd;
120         iocom->flags = HAMMER2_IOCOMF_RREQ;
121         if (state_func)
122                 iocom->flags |= HAMMER2_IOCOMF_SWORK;
123         hammer2_ioq_init(iocom, &iocom->ioq_rx);
124         hammer2_ioq_init(iocom, &iocom->ioq_tx);
125         if (pipe(iocom->wakeupfds) < 0)
126                 assert(0);
127         fcntl(iocom->wakeupfds[0], F_SETFL, O_NONBLOCK);
128         fcntl(iocom->wakeupfds[1], F_SETFL, O_NONBLOCK);
129
130         /*
131          * Negotiate session crypto synchronously.  This will mark the
132          * connection as error'd if it fails.
133          */
134         hammer2_crypto_negotiate(iocom);
135
136         /*
137          * Make sure our fds are set to non-blocking for the iocom core.
138          */
139         if (sock_fd >= 0)
140                 fcntl(sock_fd, F_SETFL, O_NONBLOCK);
141 #if 0
142         /* if line buffered our single fgets() should be fine */
143         if (alt_fd >= 0)
144                 fcntl(alt_fd, F_SETFL, O_NONBLOCK);
145 #endif
146 }
147
148 /*
149  * May only be called from a callback from iocom_core.
150  *
151  * Adjust state machine functions, set flags to guarantee that both
152  * the recevmsg_func and the sendmsg_func is called at least once.
153  */
154 void
155 hammer2_iocom_restate(hammer2_iocom_t *iocom,
156                    void (*state_func)(hammer2_iocom_t *),
157                    void (*rcvmsg_func)(hammer2_iocom_t *, hammer2_msg_t *msg),
158                    void (*altmsg_func)(hammer2_iocom_t *))
159 {
160         iocom->state_callback = state_func;
161         iocom->rcvmsg_callback = rcvmsg_func;
162         iocom->altmsg_callback = altmsg_func;
163         if (state_func)
164                 iocom->flags |= HAMMER2_IOCOMF_SWORK;
165         else
166                 iocom->flags &= ~HAMMER2_IOCOMF_SWORK;
167 }
168
169 /*
170  * Cleanup a terminating iocom.
171  *
172  * Caller should not hold iocom->mtx.  The iocom has already been disconnected
173  * from all possible references to it.
174  */
175 void
176 hammer2_iocom_done(hammer2_iocom_t *iocom)
177 {
178         hammer2_msg_t *msg;
179
180         if (iocom->sock_fd >= 0) {
181                 close(iocom->sock_fd);
182                 iocom->sock_fd = -1;
183         }
184         if (iocom->alt_fd >= 0) {
185                 close(iocom->alt_fd);
186                 iocom->alt_fd = -1;
187         }
188         hammer2_ioq_done(iocom, &iocom->ioq_rx);
189         hammer2_ioq_done(iocom, &iocom->ioq_tx);
190         if ((msg = TAILQ_FIRST(&iocom->freeq)) != NULL) {
191                 TAILQ_REMOVE(&iocom->freeq, msg, qentry);
192                 free(msg);
193         }
194         if ((msg = TAILQ_FIRST(&iocom->freeq_aux)) != NULL) {
195                 TAILQ_REMOVE(&iocom->freeq_aux, msg, qentry);
196                 free(msg->aux_data);
197                 msg->aux_data = NULL;
198                 free(msg);
199         }
200         if (iocom->wakeupfds[0] >= 0) {
201                 close(iocom->wakeupfds[0]);
202                 iocom->wakeupfds[0] = -1;
203         }
204         if (iocom->wakeupfds[1] >= 0) {
205                 close(iocom->wakeupfds[1]);
206                 iocom->wakeupfds[1] = -1;
207         }
208         pthread_mutex_destroy(&iocom->mtx);
209 }
210
211 /*
212  * Allocate a new one-way message.
213  */
214 hammer2_msg_t *
215 hammer2_msg_alloc(hammer2_iocom_t *iocom, size_t aux_size, uint32_t cmd)
216 {
217         hammer2_msg_t *msg;
218         int hbytes;
219
220         pthread_mutex_lock(&iocom->mtx);
221         if (aux_size) {
222                 aux_size = (aux_size + HAMMER2_MSG_ALIGNMASK) &
223                            ~HAMMER2_MSG_ALIGNMASK;
224                 if ((msg = TAILQ_FIRST(&iocom->freeq_aux)) != NULL)
225                         TAILQ_REMOVE(&iocom->freeq_aux, msg, qentry);
226         } else {
227                 if ((msg = TAILQ_FIRST(&iocom->freeq)) != NULL)
228                         TAILQ_REMOVE(&iocom->freeq, msg, qentry);
229         }
230         pthread_mutex_unlock(&iocom->mtx);
231         if (msg == NULL) {
232                 msg = malloc(sizeof(*msg));
233                 bzero(msg, sizeof(*msg));
234                 msg->aux_data = NULL;
235                 msg->aux_size = 0;
236         }
237         if (msg->aux_size != aux_size) {
238                 if (msg->aux_data) {
239                         free(msg->aux_data);
240                         msg->aux_data = NULL;
241                         msg->aux_size = 0;
242                 }
243                 if (aux_size) {
244                         msg->aux_data = malloc(aux_size);
245                         msg->aux_size = aux_size;
246                 }
247         }
248         hbytes = (cmd & HAMMER2_MSGF_SIZE) * HAMMER2_MSG_ALIGN;
249         if (hbytes)
250                 bzero(&msg->any.head, hbytes);
251         msg->hdr_size = hbytes;
252         msg->any.head.cmd = cmd;
253         msg->any.head.aux_descr = 0;
254         msg->any.head.aux_crc = 0;
255
256         return (msg);
257 }
258
259 /*
260  * Free a message so it can be reused afresh.
261  *
262  * NOTE: aux_size can be 0 with a non-NULL aux_data.
263  */
264 static
265 void
266 hammer2_msg_free_locked(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
267 {
268         msg->state = NULL;
269         if (msg->aux_data)
270                 TAILQ_INSERT_TAIL(&iocom->freeq_aux, msg, qentry);
271         else
272                 TAILQ_INSERT_TAIL(&iocom->freeq, msg, qentry);
273 }
274
275 void
276 hammer2_msg_free(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
277 {
278         pthread_mutex_lock(&iocom->mtx);
279         hammer2_msg_free_locked(iocom, msg);
280         pthread_mutex_unlock(&iocom->mtx);
281 }
282
283 /*
284  * I/O core loop for an iocom.
285  *
286  * Thread localized, iocom->mtx not held.
287  */
288 void
289 hammer2_iocom_core(hammer2_iocom_t *iocom)
290 {
291         struct pollfd fds[3];
292         char dummybuf[256];
293         hammer2_msg_t *msg;
294         int timeout;
295         int count;
296         int wi; /* wakeup pipe */
297         int si; /* socket */
298         int ai; /* alt bulk path socket */
299
300         while ((iocom->flags & HAMMER2_IOCOMF_EOF) == 0) {
301                 if ((iocom->flags & (HAMMER2_IOCOMF_RWORK |
302                                      HAMMER2_IOCOMF_WWORK |
303                                      HAMMER2_IOCOMF_PWORK |
304                                      HAMMER2_IOCOMF_SWORK |
305                                      HAMMER2_IOCOMF_ARWORK |
306                                      HAMMER2_IOCOMF_AWWORK)) == 0) {
307                         /*
308                          * Only poll if no immediate work is pending.
309                          * Otherwise we are just wasting our time calling
310                          * poll.
311                          */
312                         timeout = 5000;
313
314                         count = 0;
315                         wi = -1;
316                         si = -1;
317                         ai = -1;
318
319                         /*
320                          * Always check the inter-thread pipe, e.g.
321                          * for iocom->txmsgq work.
322                          */
323                         wi = count++;
324                         fds[wi].fd = iocom->wakeupfds[0];
325                         fds[wi].events = POLLIN;
326                         fds[wi].revents = 0;
327
328                         /*
329                          * Check the socket input/output direction as
330                          * requested
331                          */
332                         if (iocom->flags & (HAMMER2_IOCOMF_RREQ |
333                                             HAMMER2_IOCOMF_WREQ)) {
334                                 si = count++;
335                                 fds[si].fd = iocom->sock_fd;
336                                 fds[si].events = 0;
337                                 fds[si].revents = 0;
338
339                                 if (iocom->flags & HAMMER2_IOCOMF_RREQ)
340                                         fds[si].events |= POLLIN;
341                                 if (iocom->flags & HAMMER2_IOCOMF_WREQ)
342                                         fds[si].events |= POLLOUT;
343                         }
344
345                         /*
346                          * Check the alternative fd for work.
347                          */
348                         if (iocom->alt_fd >= 0) {
349                                 ai = count++;
350                                 fds[ai].fd = iocom->alt_fd;
351                                 fds[ai].events = POLLIN;
352                                 fds[ai].revents = 0;
353                         }
354                         poll(fds, count, timeout);
355
356                         if (wi >= 0 && (fds[wi].revents & POLLIN))
357                                 iocom->flags |= HAMMER2_IOCOMF_PWORK;
358                         if (si >= 0 && (fds[si].revents & POLLIN))
359                                 iocom->flags |= HAMMER2_IOCOMF_RWORK;
360                         if (si >= 0 && (fds[si].revents & POLLOUT))
361                                 iocom->flags |= HAMMER2_IOCOMF_WWORK;
362                         if (wi >= 0 && (fds[wi].revents & POLLOUT))
363                                 iocom->flags |= HAMMER2_IOCOMF_WWORK;
364                         if (ai >= 0 && (fds[ai].revents & POLLIN))
365                                 iocom->flags |= HAMMER2_IOCOMF_ARWORK;
366                 } else {
367                         /*
368                          * Always check the pipe
369                          */
370                         iocom->flags |= HAMMER2_IOCOMF_PWORK;
371                 }
372
373                 if (iocom->flags & HAMMER2_IOCOMF_SWORK) {
374                         iocom->flags &= ~HAMMER2_IOCOMF_SWORK;
375                         iocom->state_callback(iocom);
376                 }
377
378                 /*
379                  * Pending message queues from other threads wake us up
380                  * with a write to the wakeupfds[] pipe.  We have to clear
381                  * the pipe with a dummy read.
382                  */
383                 if (iocom->flags & HAMMER2_IOCOMF_PWORK) {
384                         iocom->flags &= ~HAMMER2_IOCOMF_PWORK;
385                         read(iocom->wakeupfds[0], dummybuf, sizeof(dummybuf));
386                         iocom->flags |= HAMMER2_IOCOMF_RWORK;
387                         iocom->flags |= HAMMER2_IOCOMF_WWORK;
388                         if (TAILQ_FIRST(&iocom->txmsgq))
389                                 hammer2_iocom_flush1(iocom);
390                 }
391
392                 /*
393                  * Message write sequencing
394                  */
395                 if (iocom->flags & HAMMER2_IOCOMF_WWORK)
396                         hammer2_iocom_flush1(iocom);
397
398                 /*
399                  * Message read sequencing.  Run this after the write
400                  * sequencing in case the write sequencing allowed another
401                  * auto-DELETE to occur on the read side.
402                  */
403                 if (iocom->flags & HAMMER2_IOCOMF_RWORK) {
404                         while ((iocom->flags & HAMMER2_IOCOMF_EOF) == 0 &&
405                                (msg = hammer2_ioq_read(iocom)) != NULL) {
406                                 if (DebugOpt) {
407                                         fprintf(stderr, "receive %s\n",
408                                                 hammer2_msg_str(msg));
409                                 }
410                                 iocom->rcvmsg_callback(iocom, msg);
411                                 hammer2_state_cleanuprx(iocom, msg);
412                         }
413                 }
414
415                 if (iocom->flags & HAMMER2_IOCOMF_ARWORK) {
416                         iocom->flags &= ~HAMMER2_IOCOMF_ARWORK;
417                         iocom->altmsg_callback(iocom);
418                 }
419         }
420 }
421
422 /*
423  * Read the next ready message from the ioq, issuing I/O if needed.
424  * Caller should retry on a read-event when NULL is returned.
425  *
426  * If an error occurs during reception a HAMMER2_LNK_ERROR msg will
427  * be returned for each open transaction, then the ioq and iocom
428  * will be errored out and a non-transactional HAMMER2_LNK_ERROR
429  * msg will be returned as the final message.  The caller should not call
430  * us again after the final message is returned.
431  *
432  * Thread localized, iocom->mtx not held.
433  */
434 hammer2_msg_t *
435 hammer2_ioq_read(hammer2_iocom_t *iocom)
436 {
437         hammer2_ioq_t *ioq = &iocom->ioq_rx;
438         hammer2_msg_t *msg;
439         hammer2_msg_hdr_t *head;
440         hammer2_state_t *state;
441         ssize_t n;
442         size_t bytes;
443         size_t nmax;
444         uint32_t xcrc32;
445         int error;
446
447 again:
448         iocom->flags &= ~(HAMMER2_IOCOMF_RREQ | HAMMER2_IOCOMF_RWORK);
449
450         /*
451          * If a message is already pending we can just remove and
452          * return it.  Message state has already been processed.
453          */
454         if ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
455                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
456                 return (msg);
457         }
458
459         if (ioq->error)
460                 goto skip;
461
462         /*
463          * Message read in-progress (msg is NULL at the moment).  We don't
464          * allocate a msg until we have its core header.
465          */
466         bytes = ioq->fifo_end - ioq->fifo_beg;
467         nmax = sizeof(ioq->buf) - ioq->fifo_end;
468         msg = ioq->msg;
469
470         switch(ioq->state) {
471         case HAMMER2_MSGQ_STATE_HEADER1:
472                 /*
473                  * Load the primary header, fail on any non-trivial read
474                  * error or on EOF.  Since the primary header is the same
475                  * size is the message alignment it will never straddle
476                  * the end of the buffer.
477                  */
478                 if (bytes < (int)sizeof(msg->any.head)) {
479                         n = read(iocom->sock_fd,
480                                  ioq->buf + ioq->fifo_end,
481                                  nmax);
482                         if (n <= 0) {
483                                 if (n == 0) {
484                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
485                                         break;
486                                 }
487                                 if (errno != EINTR &&
488                                     errno != EINPROGRESS &&
489                                     errno != EAGAIN) {
490                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
491                                         break;
492                                 }
493                                 n = 0;
494                                 /* fall through */
495                         }
496                         ioq->fifo_end += n;
497                         bytes += n;
498                         nmax -= n;
499                 }
500
501                 /*
502                  * Insufficient data accumulated (msg is NULL, caller will
503                  * retry on event).
504                  */
505                 assert(msg == NULL);
506                 if (bytes < (int)sizeof(msg->any.head))
507                         break;
508
509                 /*
510                  * Calculate the header, decrypt data received so far.
511                  * Data will be decrypted in-place.  Partial blocks are
512                  * not immediately decrypted.
513                  *
514                  * WARNING!  The header might be in the wrong endian, we
515                  *           do not fix it up until we get the entire
516                  *           extended header.
517                  */
518                 hammer2_crypto_decrypt(iocom, ioq);
519                 head = (void *)(ioq->buf + ioq->fifo_beg);
520
521                 /*
522                  * Check and fixup the core header.  Note that the icrc
523                  * has to be calculated before any fixups, but the crc
524                  * fields in the msg may have to be swapped like everything
525                  * else.
526                  */
527                 if (head->magic != HAMMER2_MSGHDR_MAGIC &&
528                     head->magic != HAMMER2_MSGHDR_MAGIC_REV) {
529                         ioq->error = HAMMER2_IOQ_ERROR_SYNC;
530                         break;
531                 }
532
533                 /*
534                  * Calculate the full header size and aux data size
535                  */
536                 if (head->magic == HAMMER2_MSGHDR_MAGIC_REV) {
537                         ioq->hbytes = (bswap32(head->cmd) & HAMMER2_MSGF_SIZE) *
538                                       HAMMER2_MSG_ALIGN;
539                         ioq->abytes = bswap32(head->aux_bytes) *
540                                       HAMMER2_MSG_ALIGN;
541                 } else {
542                         ioq->hbytes = (head->cmd & HAMMER2_MSGF_SIZE) *
543                                       HAMMER2_MSG_ALIGN;
544                         ioq->abytes = head->aux_bytes * HAMMER2_MSG_ALIGN;
545                 }
546                 if (ioq->hbytes < sizeof(msg->any.head) ||
547                     ioq->hbytes > sizeof(msg->any) ||
548                     ioq->abytes > HAMMER2_MSGAUX_MAX) {
549                         ioq->error = HAMMER2_IOQ_ERROR_FIELD;
550                         break;
551                 }
552
553                 /*
554                  * Allocate the message, the next state will fill it in.
555                  */
556                 msg = hammer2_msg_alloc(iocom, ioq->abytes, 0);
557                 ioq->msg = msg;
558
559                 /*
560                  * Fall through to the next state.  Make sure that the
561                  * extended header does not straddle the end of the buffer.
562                  * We still want to issue larger reads into our buffer,
563                  * book-keeping is easier if we don't bcopy() yet.
564                  */
565                 if (bytes + nmax < ioq->hbytes) {
566                         bcopy(ioq->buf + ioq->fifo_beg, ioq->buf, bytes);
567                         ioq->fifo_cdx -= ioq->fifo_beg;
568                         ioq->fifo_beg = 0;
569                         ioq->fifo_end = bytes;
570                         nmax = sizeof(ioq->buf) - ioq->fifo_end;
571                 }
572                 ioq->state = HAMMER2_MSGQ_STATE_HEADER2;
573                 /* fall through */
574         case HAMMER2_MSGQ_STATE_HEADER2:
575                 /*
576                  * Fill out the extended header.
577                  */
578                 assert(msg != NULL);
579                 if (bytes < ioq->hbytes) {
580                         n = read(iocom->sock_fd,
581                                  ioq->buf + ioq->fifo_end,
582                                  nmax);
583                         if (n <= 0) {
584                                 if (n == 0) {
585                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
586                                         break;
587                                 }
588                                 if (errno != EINTR &&
589                                     errno != EINPROGRESS &&
590                                     errno != EAGAIN) {
591                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
592                                         break;
593                                 }
594                                 n = 0;
595                                 /* fall through */
596                         }
597                         ioq->fifo_end += n;
598                         bytes += n;
599                         nmax -= n;
600                 }
601
602                 /*
603                  * Insufficient data accumulated (set msg NULL so caller will
604                  * retry on event).
605                  */
606                 if (bytes < ioq->hbytes) {
607                         msg = NULL;
608                         break;
609                 }
610
611                 /*
612                  * Calculate the extended header, decrypt data received
613                  * so far.  Handle endian-conversion for the entire extended
614                  * header.
615                  */
616                 hammer2_crypto_decrypt(iocom, ioq);
617                 head = (void *)(ioq->buf + ioq->fifo_beg);
618
619                 /*
620                  * Check the CRC.
621                  */
622                 if (head->magic == HAMMER2_MSGHDR_MAGIC_REV)
623                         xcrc32 = bswap32(head->hdr_crc);
624                 else
625                         xcrc32 = head->hdr_crc;
626                 head->hdr_crc = 0;
627                 if (hammer2_icrc32(head, ioq->hbytes) != xcrc32) {
628                         ioq->error = HAMMER2_IOQ_ERROR_XCRC;
629                         fprintf(stderr, "BAD-XCRC(%08x,%08x) %s\n",
630                                 xcrc32, hammer2_icrc32(head, ioq->hbytes),
631                                 hammer2_msg_str(msg));
632                         assert(0);
633                         break;
634                 }
635                 head->hdr_crc = xcrc32;
636
637                 if (head->magic == HAMMER2_MSGHDR_MAGIC_REV) {
638                         hammer2_bswap_head(head);
639                 }
640
641                 /*
642                  * Copy the extended header into the msg and adjust the
643                  * FIFO.
644                  */
645                 bcopy(head, &msg->any, ioq->hbytes);
646
647                 /*
648                  * We are either done or we fall-through.
649                  */
650                 if (ioq->abytes == 0) {
651                         ioq->fifo_beg += ioq->hbytes;
652                         break;
653                 }
654
655                 /*
656                  * Must adjust nmax and bytes (and the state) when falling
657                  * through.
658                  */
659                 ioq->fifo_beg += ioq->hbytes;
660                 nmax -= ioq->hbytes;
661                 bytes -= ioq->hbytes;
662                 ioq->state = HAMMER2_MSGQ_STATE_AUXDATA1;
663                 /* fall through */
664         case HAMMER2_MSGQ_STATE_AUXDATA1:
665                 /*
666                  * Copy the partial or complete payload from remaining
667                  * bytes in the FIFO.  We have to fall-through either
668                  * way so we can check the crc.
669                  *
670                  * Adjust msg->aux_size to the final actual value.
671                  */
672                 ioq->already = ioq->fifo_cdx - ioq->fifo_beg;
673                 if (ioq->already > ioq->abytes)
674                         ioq->already = ioq->abytes;
675                 if (bytes >= ioq->abytes) {
676                         bcopy(ioq->buf + ioq->fifo_beg, msg->aux_data,
677                               ioq->abytes);
678                         msg->aux_size = ioq->abytes;
679                         ioq->fifo_beg += ioq->abytes;
680                         if (ioq->fifo_cdx < ioq->fifo_beg)
681                                 ioq->fifo_cdx = ioq->fifo_beg;
682                         bytes -= ioq->abytes;
683                 } else if (bytes) {
684                         bcopy(ioq->buf + ioq->fifo_beg, msg->aux_data,
685                               bytes);
686                         msg->aux_size = bytes;
687                         ioq->fifo_beg += bytes;
688                         if (ioq->fifo_cdx < ioq->fifo_beg)
689                                 ioq->fifo_cdx = ioq->fifo_beg;
690                         bytes = 0;
691                 } else {
692                         msg->aux_size = 0;
693                 }
694                 ioq->state = HAMMER2_MSGQ_STATE_AUXDATA2;
695                 /* fall through */
696         case HAMMER2_MSGQ_STATE_AUXDATA2:
697                 /*
698                  * Read the remainder of the payload directly into the
699                  * msg->aux_data buffer.
700                  */
701                 assert(msg);
702                 if (msg->aux_size < ioq->abytes) {
703                         assert(bytes == 0);
704                         n = read(iocom->sock_fd,
705                                  msg->aux_data + msg->aux_size,
706                                  ioq->abytes - msg->aux_size);
707                         if (n <= 0) {
708                                 if (n == 0) {
709                                         ioq->error = HAMMER2_IOQ_ERROR_EOF;
710                                         break;
711                                 }
712                                 if (errno != EINTR &&
713                                     errno != EINPROGRESS &&
714                                     errno != EAGAIN) {
715                                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
716                                         break;
717                                 }
718                                 n = 0;
719                                 /* fall through */
720                         }
721                         msg->aux_size += n;
722                 }
723
724                 /*
725                  * Insufficient data accumulated (set msg NULL so caller will
726                  * retry on event).
727                  */
728                 if (msg->aux_size < ioq->abytes) {
729                         msg = NULL;
730                         break;
731                 }
732                 assert(msg->aux_size == ioq->abytes);
733                 hammer2_crypto_decrypt_aux(iocom, ioq, msg, ioq->already);
734
735                 /*
736                  * Check aux_crc, then we are done.
737                  */
738                 xcrc32 = hammer2_icrc32(msg->aux_data, msg->aux_size);
739                 if (xcrc32 != msg->any.head.aux_crc) {
740                         ioq->error = HAMMER2_IOQ_ERROR_ACRC;
741                         break;
742                 }
743                 break;
744         case HAMMER2_MSGQ_STATE_ERROR:
745                 /*
746                  * Continued calls to drain recorded transactions (returning
747                  * a LNK_ERROR for each one), before we return the final
748                  * LNK_ERROR.
749                  */
750                 assert(msg == NULL);
751                 break;
752         default:
753                 /*
754                  * We don't double-return errors, the caller should not
755                  * have called us again after getting an error msg.
756                  */
757                 assert(0);
758                 break;
759         }
760
761         /*
762          * Check the message sequence.  The iv[] should prevent any
763          * possibility of a replay but we add this check anyway.
764          */
765         if (msg && ioq->error == 0) {
766                 if ((msg->any.head.salt & 255) != (ioq->seq & 255)) {
767                         ioq->error = HAMMER2_IOQ_ERROR_MSGSEQ;
768                 } else {
769                         ++ioq->seq;
770                 }
771         }
772
773         /*
774          * Process transactional state for the message.
775          */
776         if (msg && ioq->error == 0) {
777                 error = hammer2_state_msgrx(iocom, msg);
778                 if (error) {
779                         if (error == HAMMER2_IOQ_ERROR_EALREADY) {
780                                 hammer2_msg_free(iocom, msg);
781                                 goto again;
782                         }
783                         ioq->error = error;
784                 }
785         }
786
787         /*
788          * Handle error, RREQ, or completion
789          *
790          * NOTE: nmax and bytes are invalid at this point, we don't bother
791          *       to update them when breaking out.
792          */
793         if (ioq->error) {
794 skip:
795                 /*
796                  * An unrecoverable error causes all active receive
797                  * transactions to be terminated with a LNK_ERROR message.
798                  *
799                  * Once all active transactions are exhausted we set the
800                  * iocom ERROR flag and return a non-transactional LNK_ERROR
801                  * message, which should cause master processing loops to
802                  * terminate.
803                  */
804                 assert(ioq->msg == msg);
805                 if (msg) {
806                         hammer2_msg_free(iocom, msg);
807                         ioq->msg = NULL;
808                 }
809
810                 /*
811                  * No more I/O read processing
812                  */
813                 ioq->state = HAMMER2_MSGQ_STATE_ERROR;
814
815                 /*
816                  * Simulate a remote LNK_ERROR DELETE msg for any open
817                  * transactions, ending with a final non-transactional
818                  * LNK_ERROR (that the session can detect) when no
819                  * transactions remain.
820                  */
821                 msg = hammer2_msg_alloc(iocom, 0, 0);
822                 bzero(&msg->any.head, sizeof(msg->any.head));
823                 msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
824                 msg->any.head.cmd = HAMMER2_LNK_ERROR;
825                 msg->any.head.error = ioq->error;
826
827                 pthread_mutex_lock(&iocom->mtx);
828                 hammer2_iocom_drain(iocom);
829                 if ((state = RB_ROOT(&iocom->staterd_tree)) != NULL) {
830                         /*
831                          * Active remote transactions are still present.
832                          * Simulate the other end sending us a DELETE.
833                          */
834                         if (state->rxcmd & HAMMER2_MSGF_DELETE) {
835                                 hammer2_msg_free(iocom, msg);
836                                 msg = NULL;
837                         } else {
838                                 /*state->txcmd |= HAMMER2_MSGF_DELETE;*/
839                                 msg->state = state;
840                                 msg->any.head.spanid = state->spanid;
841                                 msg->any.head.msgid = state->msgid;
842                                 msg->any.head.cmd |= HAMMER2_MSGF_ABORT |
843                                                      HAMMER2_MSGF_DELETE;
844                         }
845                 } else if ((state = RB_ROOT(&iocom->statewr_tree)) != NULL) {
846                         /*
847                          * Active local transactions are still present.
848                          * Simulate the other end sending us a DELETE.
849                          */
850                         if (state->rxcmd & HAMMER2_MSGF_DELETE) {
851                                 hammer2_msg_free(iocom, msg);
852                                 msg = NULL;
853                         } else {
854                                 msg->state = state;
855                                 msg->any.head.spanid = state->spanid;
856                                 msg->any.head.msgid = state->msgid;
857                                 msg->any.head.cmd |= HAMMER2_MSGF_ABORT |
858                                                      HAMMER2_MSGF_DELETE |
859                                                      HAMMER2_MSGF_REPLY;
860                                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
861                                         msg->any.head.cmd |=
862                                                      HAMMER2_MSGF_CREATE;
863                                 }
864                         }
865                 } else {
866                         /*
867                          * No active local or remote transactions remain.
868                          * Generate a final LNK_ERROR and flag EOF.
869                          */
870                         msg->state = NULL;
871                         iocom->flags |= HAMMER2_IOCOMF_EOF;
872                         fprintf(stderr, "EOF ON SOCKET %d\n", iocom->sock_fd);
873                 }
874                 pthread_mutex_unlock(&iocom->mtx);
875
876                 /*
877                  * For the iocom error case we want to set RWORK to indicate
878                  * that more messages might be pending.
879                  *
880                  * It is possible to return NULL when there is more work to
881                  * do because each message has to be DELETEd in both
882                  * directions before we continue on with the next (though
883                  * this could be optimized).  The transmit direction will
884                  * re-set RWORK.
885                  */
886                 if (msg)
887                         iocom->flags |= HAMMER2_IOCOMF_RWORK;
888         } else if (msg == NULL) {
889                 /*
890                  * Insufficient data received to finish building the message,
891                  * set RREQ and return NULL.
892                  *
893                  * Leave ioq->msg intact.
894                  * Leave the FIFO intact.
895                  */
896                 iocom->flags |= HAMMER2_IOCOMF_RREQ;
897         } else {
898                 /*
899                  * Return msg.
900                  *
901                  * The fifo has already been advanced past the message.
902                  * Trivially reset the FIFO indices if possible.
903                  *
904                  * clear the FIFO if it is now empty and set RREQ to wait
905                  * for more from the socket.  If the FIFO is not empty set
906                  * TWORK to bypass the poll so we loop immediately.
907                  */
908                 if (ioq->fifo_beg == ioq->fifo_end) {
909                         iocom->flags |= HAMMER2_IOCOMF_RREQ;
910                         ioq->fifo_cdx = 0;
911                         ioq->fifo_beg = 0;
912                         ioq->fifo_end = 0;
913                 } else {
914                         iocom->flags |= HAMMER2_IOCOMF_RWORK;
915                 }
916                 ioq->state = HAMMER2_MSGQ_STATE_HEADER1;
917                 ioq->msg = NULL;
918         }
919         return (msg);
920 }
921
922 /*
923  * Calculate the header and data crc's and write a low-level message to
924  * the connection.  If aux_crc is non-zero the aux_data crc is already
925  * assumed to have been set.
926  *
927  * A non-NULL msg is added to the queue but not necessarily flushed.
928  * Calling this function with msg == NULL will get a flush going.
929  *
930  * Caller must hold iocom->mtx.
931  */
932 void
933 hammer2_iocom_flush1(hammer2_iocom_t *iocom)
934 {
935         hammer2_ioq_t *ioq = &iocom->ioq_tx;
936         hammer2_msg_t *msg;
937         uint32_t xcrc32;
938         int hbytes;
939         hammer2_msg_queue_t tmpq;
940
941         iocom->flags &= ~(HAMMER2_IOCOMF_WREQ | HAMMER2_IOCOMF_WWORK);
942         TAILQ_INIT(&tmpq);
943         pthread_mutex_lock(&iocom->mtx);
944         while ((msg = TAILQ_FIRST(&iocom->txmsgq)) != NULL) {
945                 TAILQ_REMOVE(&iocom->txmsgq, msg, qentry);
946                 TAILQ_INSERT_TAIL(&tmpq, msg, qentry);
947         }
948         pthread_mutex_unlock(&iocom->mtx);
949
950         while ((msg = TAILQ_FIRST(&tmpq)) != NULL) {
951                 /*
952                  * Process terminal connection errors.
953                  */
954                 TAILQ_REMOVE(&tmpq, msg, qentry);
955                 if (ioq->error) {
956                         TAILQ_INSERT_TAIL(&ioq->msgq, msg, qentry);
957                         ++ioq->msgcount;
958                         continue;
959                 }
960
961                 /*
962                  * Finish populating the msg fields.  The salt ensures that
963                  * the iv[] array is ridiculously randomized and we also
964                  * re-seed our PRNG every 32768 messages just to be sure.
965                  */
966                 msg->any.head.magic = HAMMER2_MSGHDR_MAGIC;
967                 msg->any.head.salt = (random() << 8) | (ioq->seq & 255);
968                 ++ioq->seq;
969                 if ((ioq->seq & 32767) == 0)
970                         srandomdev();
971
972                 /*
973                  * Calculate aux_crc if 0, then calculate hdr_crc.
974                  */
975                 if (msg->aux_size && msg->any.head.aux_crc == 0) {
976                         assert((msg->aux_size & HAMMER2_MSG_ALIGNMASK) == 0);
977                         xcrc32 = hammer2_icrc32(msg->aux_data, msg->aux_size);
978                         msg->any.head.aux_crc = xcrc32;
979                 }
980                 msg->any.head.aux_bytes = msg->aux_size / HAMMER2_MSG_ALIGN;
981                 assert((msg->aux_size & HAMMER2_MSG_ALIGNMASK) == 0);
982
983                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
984                          HAMMER2_MSG_ALIGN;
985                 msg->any.head.hdr_crc = 0;
986                 msg->any.head.hdr_crc = hammer2_icrc32(&msg->any.head, hbytes);
987
988                 /*
989                  * Enqueue the message (the flush codes handles stream
990                  * encryption).
991                  */
992                 TAILQ_INSERT_TAIL(&ioq->msgq, msg, qentry);
993                 ++ioq->msgcount;
994         }
995         hammer2_iocom_flush2(iocom);
996 }
997
998 /*
999  * Thread localized, iocom->mtx not held by caller.
1000  */
1001 void
1002 hammer2_iocom_flush2(hammer2_iocom_t *iocom)
1003 {
1004         hammer2_ioq_t *ioq = &iocom->ioq_tx;
1005         hammer2_msg_t *msg;
1006         ssize_t nmax;
1007         ssize_t omax;
1008         ssize_t nact;
1009         struct iovec iov[HAMMER2_IOQ_MAXIOVEC];
1010         size_t hbytes;
1011         size_t abytes;
1012         size_t hoff;
1013         size_t aoff;
1014         int n;
1015
1016         if (ioq->error) {
1017                 hammer2_iocom_drain(iocom);
1018                 return;
1019         }
1020
1021         /*
1022          * Pump messages out the connection by building an iovec.
1023          *
1024          * ioq->hbytes/ioq->abytes tracks how much of the first message
1025          * in the queue has been successfully written out, so we can
1026          * resume writing.
1027          */
1028         n = 0;
1029         nmax = 0;
1030         hoff = ioq->hbytes;
1031         aoff = ioq->abytes;
1032
1033         TAILQ_FOREACH(msg, &ioq->msgq, qentry) {
1034                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
1035                          HAMMER2_MSG_ALIGN;
1036                 abytes = msg->aux_size;
1037                 assert(hoff <= hbytes && aoff <= abytes);
1038
1039                 if (hoff < hbytes) {
1040                         iov[n].iov_base = (char *)&msg->any.head + hoff;
1041                         iov[n].iov_len = hbytes - hoff;
1042                         nmax += hbytes - hoff;
1043                         ++n;
1044                         if (n == HAMMER2_IOQ_MAXIOVEC)
1045                                 break;
1046                 }
1047                 if (aoff < abytes) {
1048                         assert(msg->aux_data != NULL);
1049                         iov[n].iov_base = (char *)msg->aux_data + aoff;
1050                         iov[n].iov_len = abytes - aoff;
1051                         nmax += abytes - aoff;
1052                         ++n;
1053                         if (n == HAMMER2_IOQ_MAXIOVEC)
1054                                 break;
1055                 }
1056                 hoff = 0;
1057                 aoff = 0;
1058         }
1059         if (n == 0)
1060                 return;
1061
1062         /*
1063          * Encrypt and write the data.  The crypto code will move the
1064          * data into the fifo and adjust the iov as necessary.  If
1065          * encryption is disabled the iov is left alone.
1066          *
1067          * May return a smaller iov (thus a smaller n), with aggregated
1068          * chunks.  May reduce nmax to what fits in the FIFO.
1069          */
1070         omax = nmax;
1071         n = hammer2_crypto_encrypt(iocom, ioq, iov, n, &nmax);
1072
1073         /*
1074          * Execute the writev() then figure out what happened.
1075          */
1076         nact = writev(iocom->sock_fd, iov, n);
1077         if (nact < 0) {
1078                 if (errno != EINTR &&
1079                     errno != EINPROGRESS &&
1080                     errno != EAGAIN) {
1081                         /*
1082                          * Fatal write error
1083                          */
1084                         ioq->error = HAMMER2_IOQ_ERROR_SOCK;
1085                         hammer2_iocom_drain(iocom);
1086                 } else {
1087                         /*
1088                          * Wait for socket buffer space
1089                          */
1090                         iocom->flags |= HAMMER2_IOCOMF_WREQ;
1091                 }
1092                 return;
1093         }
1094
1095         /*
1096          * Indicate bytes written successfully.
1097          *
1098          * If we were unable to write the entire iov array then set WREQ
1099          * to wait for more socket buffer space.
1100          *
1101          * If the FIFO space was insufficient to fully drain all messages
1102          * set WWORK to cause the core to call us again for the next batch.
1103          */
1104         hammer2_crypto_encrypt_wrote(iocom, ioq, nact);
1105         if (nact != nmax)
1106                 iocom->flags |= HAMMER2_IOCOMF_WREQ;
1107
1108         /*
1109          * Clean out the transmit queue based on what we successfully
1110          * sent.  ioq->hbytes/abytes represents the portion of the first
1111          * message previously sent.
1112          */
1113         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
1114                 hbytes = (msg->any.head.cmd & HAMMER2_MSGF_SIZE) *
1115                          HAMMER2_MSG_ALIGN;
1116                 abytes = msg->aux_size;
1117
1118                 if ((size_t)nact < hbytes - ioq->hbytes) {
1119                         ioq->hbytes += nact;
1120                         /* nact = 0; */
1121                         break;
1122                 }
1123                 nact -= hbytes - ioq->hbytes;
1124                 ioq->hbytes = hbytes;
1125                 if ((size_t)nact < abytes - ioq->abytes) {
1126                         ioq->abytes += nact;
1127                         /* nact = 0; */
1128                         break;
1129                 }
1130                 nact -= abytes - ioq->abytes;
1131
1132                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
1133                 --ioq->msgcount;
1134                 ioq->hbytes = 0;
1135                 ioq->abytes = 0;
1136
1137                 hammer2_state_cleanuptx(iocom, msg);
1138         }
1139
1140         /*
1141          * If more messages are pending on WREQ wasn't set we must
1142          * ensure that WWORK gets set.
1143          */
1144         if (msg && (iocom->flags & HAMMER2_IOCOMF_WREQ) == 0)
1145                 iocom->flags |= HAMMER2_IOCOMF_WWORK;
1146         assert(nact == 0);
1147         if (ioq->error) {
1148                 hammer2_iocom_drain(iocom);
1149         }
1150 }
1151
1152 /*
1153  * Kill pending msgs on ioq_tx and adjust the flags such that no more
1154  * write events will occur.  We don't kill read msgs because we want
1155  * the caller to pull off our contrived terminal error msg to detect
1156  * the connection failure.
1157  *
1158  * Thread localized, iocom->mtx not held by caller.
1159  */
1160 void
1161 hammer2_iocom_drain(hammer2_iocom_t *iocom)
1162 {
1163         hammer2_ioq_t *ioq = &iocom->ioq_tx;
1164         hammer2_msg_t *msg;
1165
1166         iocom->flags &= ~(HAMMER2_IOCOMF_WREQ | HAMMER2_IOCOMF_WWORK);
1167         ioq->hbytes = 0;
1168         ioq->abytes = 0;
1169
1170         while ((msg = TAILQ_FIRST(&ioq->msgq)) != NULL) {
1171                 TAILQ_REMOVE(&ioq->msgq, msg, qentry);
1172                 --ioq->msgcount;
1173                 hammer2_state_cleanuptx(iocom, msg);
1174         }
1175 }
1176
1177 /*
1178  * Write a message to an iocom, with additional state processing.
1179  */
1180 void
1181 hammer2_msg_write(hammer2_iocom_t *iocom, hammer2_msg_t *msg,
1182                   void (*func)(hammer2_state_t *, hammer2_msg_t *),
1183                   void *data,
1184                   hammer2_state_t **statep)
1185 {
1186         hammer2_state_t *state;
1187         char dummy;
1188
1189         /*
1190          * Handle state processing, create state if necessary.
1191          */
1192         pthread_mutex_lock(&iocom->mtx);
1193         if ((state = msg->state) != NULL) {
1194                 /*
1195                  * Existing transaction (could be reply).  It is also
1196                  * possible for this to be the first reply (CREATE is set),
1197                  * in which case we populate state->txcmd.
1198                  */
1199                 msg->any.head.msgid = state->msgid;
1200                 msg->any.head.spanid = state->spanid;
1201                 if (func) {
1202                         state->func = func;
1203                         state->any.any = data;
1204                 }
1205                 assert(((state->txcmd ^ msg->any.head.cmd) &
1206                         HAMMER2_MSGF_REPLY) == 0);
1207                 if (msg->any.head.cmd & HAMMER2_MSGF_CREATE)
1208                         state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1209         } else if (msg->any.head.cmd & HAMMER2_MSGF_CREATE) {
1210                 /*
1211                  * No existing state and CREATE is set, create new
1212                  * state for outgoing command.  This can't happen if
1213                  * REPLY is set as the state would already exist for
1214                  * a transaction reply.
1215                  */
1216                 assert((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0);
1217
1218                 state = malloc(sizeof(*state));
1219                 bzero(state, sizeof(*state));
1220                 state->iocom = iocom;
1221                 state->flags = HAMMER2_STATE_DYNAMIC;
1222                 state->msg = msg;
1223                 state->msgid = (uint64_t)(uintptr_t)state;
1224                 state->spanid = msg->any.head.spanid;
1225                 state->txcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1226                 state->rxcmd = HAMMER2_MSGF_REPLY;
1227                 state->func = func;
1228                 state->any.any = data;
1229                 pthread_mutex_lock(&iocom->mtx);
1230                 RB_INSERT(hammer2_state_tree, &iocom->statewr_tree, state);
1231                 pthread_mutex_unlock(&iocom->mtx);
1232                 state->flags |= HAMMER2_STATE_INSERTED;
1233                 msg->state = state;
1234                 msg->any.head.msgid = state->msgid;
1235                 /* spanid set by caller */
1236         } else {
1237                 msg->any.head.msgid = 0;
1238                 /* spanid set by caller */
1239         }
1240
1241         if (statep)
1242                 *statep = state;
1243
1244         /*
1245          * Queue it for output, wake up the I/O pthread.  Note that the
1246          * I/O thread is responsible for generating the CRCs and encryption.
1247          */
1248         TAILQ_INSERT_TAIL(&iocom->txmsgq, msg, qentry);
1249         dummy = 0;
1250         write(iocom->wakeupfds[1], &dummy, 1);  /* XXX optimize me */
1251         pthread_mutex_unlock(&iocom->mtx);
1252 }
1253
1254 /*
1255  * This is a shortcut to formulate a reply to msg with a simple error code,
1256  * It can reply to and terminate a transaction, or it can reply to a one-way
1257  * messages.  A HAMMER2_LNK_ERROR command code is utilized to encode
1258  * the error code (which can be 0).  Not all transactions are terminated
1259  * with HAMMER2_LNK_ERROR status (the low level only cares about the
1260  * MSGF_DELETE flag), but most are.
1261  *
1262  * Replies to one-way messages are a bit of an oxymoron but the feature
1263  * is used by the debug (DBG) protocol.
1264  *
1265  * The reply contains no extended data.
1266  */
1267 void
1268 hammer2_msg_reply(hammer2_iocom_t *iocom, hammer2_msg_t *msg, uint32_t error)
1269 {
1270         hammer2_state_t *state = msg->state;
1271         hammer2_msg_t *nmsg;
1272         uint32_t cmd;
1273
1274
1275         /*
1276          * Reply with a simple error code and terminate the transaction.
1277          */
1278         cmd = HAMMER2_LNK_ERROR;
1279
1280         /*
1281          * Check if our direction has even been initiated yet, set CREATE.
1282          *
1283          * Check what direction this is (command or reply direction).  Note
1284          * that txcmd might not have been initiated yet.
1285          *
1286          * If our direction has already been closed we just return without
1287          * doing anything.
1288          */
1289         if (state) {
1290                 if (state->txcmd & HAMMER2_MSGF_DELETE)
1291                         return;
1292                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
1293                         cmd |= HAMMER2_MSGF_CREATE;
1294                 if (state->txcmd & HAMMER2_MSGF_REPLY)
1295                         cmd |= HAMMER2_MSGF_REPLY;
1296                 cmd |= HAMMER2_MSGF_DELETE;
1297         } else {
1298                 if ((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0)
1299                         cmd |= HAMMER2_MSGF_REPLY;
1300         }
1301
1302         nmsg = hammer2_msg_alloc(iocom, 0, cmd);
1303         nmsg->any.head.error = error;
1304         nmsg->state = msg->state;
1305         hammer2_msg_write(iocom, nmsg, NULL, NULL, NULL);
1306 }
1307
1308 /*
1309  * Similar to hammer2_msg_reply() but leave the transaction open.  That is,
1310  * we are generating a streaming reply or an intermediate acknowledgement
1311  * of some sort as part of the higher level protocol, with more to come
1312  * later.
1313  */
1314 void
1315 hammer2_msg_result(hammer2_iocom_t *iocom, hammer2_msg_t *msg, uint32_t error)
1316 {
1317         hammer2_state_t *state = msg->state;
1318         hammer2_msg_t *nmsg;
1319         uint32_t cmd;
1320
1321
1322         /*
1323          * Reply with a simple error code and terminate the transaction.
1324          */
1325         cmd = HAMMER2_LNK_ERROR;
1326
1327         /*
1328          * Check if our direction has even been initiated yet, set CREATE.
1329          *
1330          * Check what direction this is (command or reply direction).  Note
1331          * that txcmd might not have been initiated yet.
1332          *
1333          * If our direction has already been closed we just return without
1334          * doing anything.
1335          */
1336         if (state) {
1337                 if (state->txcmd & HAMMER2_MSGF_DELETE)
1338                         return;
1339                 if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
1340                         cmd |= HAMMER2_MSGF_CREATE;
1341                 if (state->txcmd & HAMMER2_MSGF_REPLY)
1342                         cmd |= HAMMER2_MSGF_REPLY;
1343                 /* continuing transaction, do not set MSGF_DELETE */
1344         } else {
1345                 if ((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0)
1346                         cmd |= HAMMER2_MSGF_REPLY;
1347         }
1348
1349         nmsg = hammer2_msg_alloc(iocom, 0, cmd);
1350         nmsg->any.head.error = error;
1351         nmsg->state = state;
1352         hammer2_msg_write(iocom, nmsg, NULL, NULL, NULL);
1353 }
1354
1355 /*
1356  * Terminate a transaction given a state structure by issuing a DELETE.
1357  */
1358 void
1359 hammer2_state_reply(hammer2_state_t *state, uint32_t error)
1360 {
1361         hammer2_msg_t *nmsg;
1362         uint32_t cmd = HAMMER2_LNK_ERROR | HAMMER2_MSGF_DELETE;
1363
1364         /*
1365          * Nothing to do if we already transmitted a delete
1366          */
1367         if (state->txcmd & HAMMER2_MSGF_DELETE)
1368                 return;
1369
1370         /*
1371          * We must also set CREATE if this is our first response to a
1372          * remote command.
1373          */
1374         if ((state->txcmd & HAMMER2_MSGF_CREATE) == 0)
1375                 cmd |= HAMMER2_MSGF_CREATE;
1376
1377         /*
1378          * Set REPLY if the other end initiated the command.  Otherwise
1379          * we are the command direction.
1380          */
1381         if (state->txcmd & HAMMER2_MSGF_REPLY)
1382                 cmd |= HAMMER2_MSGF_REPLY;
1383
1384         nmsg = hammer2_msg_alloc(state->iocom, 0, cmd);
1385         nmsg->any.head.error = error;
1386         nmsg->state = state;
1387         hammer2_msg_write(state->iocom, nmsg, NULL, NULL, NULL);
1388 }
1389
1390 /************************************************************************
1391  *                      TRANSACTION STATE HANDLING                      *
1392  ************************************************************************
1393  *
1394  */
1395
1396 /*
1397  * Process state tracking for a message after reception, prior to
1398  * execution.
1399  *
1400  * Called with msglk held and the msg dequeued.
1401  *
1402  * All messages are called with dummy state and return actual state.
1403  * (One-off messages often just return the same dummy state).
1404  *
1405  * May request that caller discard the message by setting *discardp to 1.
1406  * The returned state is not used in this case and is allowed to be NULL.
1407  *
1408  * --
1409  *
1410  * These routines handle persistent and command/reply message state via the
1411  * CREATE and DELETE flags.  The first message in a command or reply sequence
1412  * sets CREATE, the last message in a command or reply sequence sets DELETE.
1413  *
1414  * There can be any number of intermediate messages belonging to the same
1415  * sequence sent inbetween the CREATE message and the DELETE message,
1416  * which set neither flag.  This represents a streaming command or reply.
1417  *
1418  * Any command message received with CREATE set expects a reply sequence to
1419  * be returned.  Reply sequences work the same as command sequences except the
1420  * REPLY bit is also sent.  Both the command side and reply side can
1421  * degenerate into a single message with both CREATE and DELETE set.  Note
1422  * that one side can be streaming and the other side not, or neither, or both.
1423  *
1424  * The msgid is unique for the initiator.  That is, two sides sending a new
1425  * message can use the same msgid without colliding.
1426  *
1427  * --
1428  *
1429  * ABORT sequences work by setting the ABORT flag along with normal message
1430  * state.  However, ABORTs can also be sent on half-closed messages, that is
1431  * even if the command or reply side has already sent a DELETE, as long as
1432  * the message has not been fully closed it can still send an ABORT+DELETE
1433  * to terminate the half-closed message state.
1434  *
1435  * Since ABORT+DELETEs can race we silently discard ABORT's for message
1436  * state which has already been fully closed.  REPLY+ABORT+DELETEs can
1437  * also race, and in this situation the other side might have already
1438  * initiated a new unrelated command with the same message id.  Since
1439  * the abort has not set the CREATE flag the situation can be detected
1440  * and the message will also be discarded.
1441  *
1442  * Non-blocking requests can be initiated with ABORT+CREATE[+DELETE].
1443  * The ABORT request is essentially integrated into the command instead
1444  * of being sent later on.  In this situation the command implementation
1445  * detects that CREATE and ABORT are both set (vs ABORT alone) and can
1446  * special-case non-blocking operation for the command.
1447  *
1448  * NOTE!  Messages with ABORT set without CREATE or DELETE are considered
1449  *        to be mid-stream aborts for command/reply sequences.  ABORTs on
1450  *        one-way messages are not supported.
1451  *
1452  * NOTE!  If a command sequence does not support aborts the ABORT flag is
1453  *        simply ignored.
1454  *
1455  * --
1456  *
1457  * One-off messages (no reply expected) are sent with neither CREATE or DELETE
1458  * set.  One-off messages cannot be aborted and typically aren't processed
1459  * by these routines.  The REPLY bit can be used to distinguish whether a
1460  * one-off message is a command or reply.  For example, one-off replies
1461  * will typically just contain status updates.
1462  */
1463 static int
1464 hammer2_state_msgrx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1465 {
1466         hammer2_state_t *state;
1467         hammer2_state_t dummy;
1468         int error;
1469
1470         /*
1471          * Lock RB tree and locate existing persistent state, if any.
1472          *
1473          * If received msg is a command state is on staterd_tree.
1474          * If received msg is a reply state is on statewr_tree.
1475          */
1476
1477         dummy.msgid = msg->any.head.msgid;
1478         dummy.spanid = msg->any.head.spanid;
1479         pthread_mutex_lock(&iocom->mtx);
1480         if (msg->any.head.cmd & HAMMER2_MSGF_REPLY) {
1481                 state = RB_FIND(hammer2_state_tree,
1482                                 &iocom->statewr_tree, &dummy);
1483         } else {
1484                 state = RB_FIND(hammer2_state_tree,
1485                                 &iocom->staterd_tree, &dummy);
1486         }
1487         msg->state = state;
1488         pthread_mutex_unlock(&iocom->mtx);
1489
1490         /*
1491          * Short-cut one-off or mid-stream messages (state may be NULL).
1492          */
1493         if ((msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1494                                   HAMMER2_MSGF_ABORT)) == 0) {
1495                 return(0);
1496         }
1497
1498         /*
1499          * Switch on CREATE, DELETE, REPLY, and also handle ABORT from
1500          * inside the case statements.
1501          */
1502         switch(msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1503                                     HAMMER2_MSGF_REPLY)) {
1504         case HAMMER2_MSGF_CREATE:
1505         case HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1506                 /*
1507                  * New persistant command received.
1508                  */
1509                 if (state) {
1510                         fprintf(stderr, "duplicate-trans %s\n",
1511                                 hammer2_msg_str(msg));
1512                         error = HAMMER2_IOQ_ERROR_TRANS;
1513                         assert(0);
1514                         break;
1515                 }
1516                 state = malloc(sizeof(*state));
1517                 bzero(state, sizeof(*state));
1518                 state->iocom = iocom;
1519                 state->flags = HAMMER2_STATE_DYNAMIC;
1520                 state->msg = msg;
1521                 state->txcmd = HAMMER2_MSGF_REPLY;
1522                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1523                 state->flags |= HAMMER2_STATE_INSERTED;
1524                 state->msgid = msg->any.head.msgid;
1525                 state->spanid = msg->any.head.spanid;
1526                 msg->state = state;
1527                 pthread_mutex_lock(&iocom->mtx);
1528                 RB_INSERT(hammer2_state_tree, &iocom->staterd_tree, state);
1529                 pthread_mutex_unlock(&iocom->mtx);
1530                 error = 0;
1531                 if (DebugOpt) {
1532                         fprintf(stderr, "create state %p id=%08x on iocom staterd %p\n",
1533                                 state, (uint32_t)state->msgid, iocom);
1534                 }
1535                 break;
1536         case HAMMER2_MSGF_DELETE:
1537                 /*
1538                  * Persistent state is expected but might not exist if an
1539                  * ABORT+DELETE races the close.
1540                  */
1541                 if (state == NULL) {
1542                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1543                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1544                         } else {
1545                                 fprintf(stderr, "missing-state %s\n",
1546                                         hammer2_msg_str(msg));
1547                                 error = HAMMER2_IOQ_ERROR_TRANS;
1548                         assert(0);
1549                         }
1550                         break;
1551                 }
1552
1553                 /*
1554                  * Handle another ABORT+DELETE case if the msgid has already
1555                  * been reused.
1556                  */
1557                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1558                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1559                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1560                         } else {
1561                                 fprintf(stderr, "reused-state %s\n",
1562                                         hammer2_msg_str(msg));
1563                                 error = HAMMER2_IOQ_ERROR_TRANS;
1564                         assert(0);
1565                         }
1566                         break;
1567                 }
1568                 error = 0;
1569                 break;
1570         default:
1571                 /*
1572                  * Check for mid-stream ABORT command received, otherwise
1573                  * allow.
1574                  */
1575                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1576                         if (state == NULL ||
1577                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1578                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1579                                 break;
1580                         }
1581                 }
1582                 error = 0;
1583                 break;
1584         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE:
1585         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE:
1586                 /*
1587                  * When receiving a reply with CREATE set the original
1588                  * persistent state message should already exist.
1589                  */
1590                 if (state == NULL) {
1591                         fprintf(stderr, "no-state(r) %s\n",
1592                                 hammer2_msg_str(msg));
1593                         error = HAMMER2_IOQ_ERROR_TRANS;
1594                         assert(0);
1595                         break;
1596                 }
1597                 assert(((state->rxcmd ^ msg->any.head.cmd) &
1598                         HAMMER2_MSGF_REPLY) == 0);
1599                 state->rxcmd = msg->any.head.cmd & ~HAMMER2_MSGF_DELETE;
1600                 error = 0;
1601                 break;
1602         case HAMMER2_MSGF_REPLY | HAMMER2_MSGF_DELETE:
1603                 /*
1604                  * Received REPLY+ABORT+DELETE in case where msgid has
1605                  * already been fully closed, ignore the message.
1606                  */
1607                 if (state == NULL) {
1608                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1609                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1610                         } else {
1611                                 fprintf(stderr, "no-state(r,d) %s\n",
1612                                         hammer2_msg_str(msg));
1613                                 error = HAMMER2_IOQ_ERROR_TRANS;
1614                         assert(0);
1615                         }
1616                         break;
1617                 }
1618
1619                 /*
1620                  * Received REPLY+ABORT+DELETE in case where msgid has
1621                  * already been reused for an unrelated message,
1622                  * ignore the message.
1623                  */
1624                 if ((state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1625                         if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1626                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1627                         } else {
1628                                 fprintf(stderr, "reused-state(r,d) %s\n",
1629                                         hammer2_msg_str(msg));
1630                                 error = HAMMER2_IOQ_ERROR_TRANS;
1631                         assert(0);
1632                         }
1633                         break;
1634                 }
1635                 error = 0;
1636                 break;
1637         case HAMMER2_MSGF_REPLY:
1638                 /*
1639                  * Check for mid-stream ABORT reply received to sent command.
1640                  */
1641                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT) {
1642                         if (state == NULL ||
1643                             (state->rxcmd & HAMMER2_MSGF_CREATE) == 0) {
1644                                 error = HAMMER2_IOQ_ERROR_EALREADY;
1645                                 break;
1646                         }
1647                 }
1648                 error = 0;
1649                 break;
1650         }
1651         return (error);
1652 }
1653
1654 void
1655 hammer2_state_cleanuprx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1656 {
1657         hammer2_state_t *state;
1658
1659         if ((state = msg->state) == NULL) {
1660                 /*
1661                  * Free a non-transactional message, there is no state
1662                  * to worry about.
1663                  */
1664                 hammer2_msg_free(iocom, msg);
1665         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
1666                 /*
1667                  * Message terminating transaction, destroy the related
1668                  * state, the original message, and this message (if it
1669                  * isn't the original message due to a CREATE|DELETE).
1670                  */
1671                 pthread_mutex_lock(&iocom->mtx);
1672                 state->rxcmd |= HAMMER2_MSGF_DELETE;
1673                 if (state->txcmd & HAMMER2_MSGF_DELETE) {
1674                         if (state->msg == msg)
1675                                 state->msg = NULL;
1676                         assert(state->flags & HAMMER2_STATE_INSERTED);
1677                         if (state->rxcmd & HAMMER2_MSGF_REPLY) {
1678                                 assert(msg->any.head.cmd & HAMMER2_MSGF_REPLY);
1679                                 RB_REMOVE(hammer2_state_tree,
1680                                           &iocom->statewr_tree, state);
1681                         } else {
1682                                 assert((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0);
1683                                 RB_REMOVE(hammer2_state_tree,
1684                                           &iocom->staterd_tree, state);
1685                         }
1686                         state->flags &= ~HAMMER2_STATE_INSERTED;
1687                         hammer2_state_free(state);
1688                 } else {
1689                         ;
1690                 }
1691                 pthread_mutex_unlock(&iocom->mtx);
1692                 hammer2_msg_free(iocom, msg);
1693         } else if (state->msg != msg) {
1694                 /*
1695                  * Message not terminating transaction, leave state intact
1696                  * and free message if it isn't the CREATE message.
1697                  */
1698                 hammer2_msg_free(iocom, msg);
1699         }
1700 }
1701
1702 static void
1703 hammer2_state_cleanuptx(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
1704 {
1705         hammer2_state_t *state;
1706
1707         if ((state = msg->state) == NULL) {
1708                 hammer2_msg_free(iocom, msg);
1709         } else if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
1710                 pthread_mutex_lock(&iocom->mtx);
1711                 state->txcmd |= HAMMER2_MSGF_DELETE;
1712                 if (state->rxcmd & HAMMER2_MSGF_DELETE) {
1713                         if (state->msg == msg)
1714                                 state->msg = NULL;
1715                         assert(state->flags & HAMMER2_STATE_INSERTED);
1716                         if (state->txcmd & HAMMER2_MSGF_REPLY) {
1717                                 assert(msg->any.head.cmd & HAMMER2_MSGF_REPLY);
1718                                 RB_REMOVE(hammer2_state_tree,
1719                                           &iocom->staterd_tree, state);
1720                         } else {
1721                                 assert((msg->any.head.cmd & HAMMER2_MSGF_REPLY) == 0);
1722                                 RB_REMOVE(hammer2_state_tree,
1723                                           &iocom->statewr_tree, state);
1724                         }
1725                         state->flags &= ~HAMMER2_STATE_INSERTED;
1726                         hammer2_state_free(state);
1727                 } else {
1728                         ;
1729                 }
1730                 pthread_mutex_unlock(&iocom->mtx);
1731                 hammer2_msg_free(iocom, msg);
1732         } else if (state->msg != msg) {
1733                 hammer2_msg_free(iocom, msg);
1734         }
1735 }
1736
1737 /*
1738  * Called with iocom locked
1739  */
1740 void
1741 hammer2_state_free(hammer2_state_t *state)
1742 {
1743         hammer2_iocom_t *iocom = state->iocom;
1744         hammer2_msg_t *msg;
1745         char dummy;
1746
1747         if (DebugOpt) {
1748                 fprintf(stderr, "terminate state %p id=%08x\n",
1749                         state, (uint32_t)state->msgid);
1750         }
1751         assert(state->any.any == NULL);
1752         msg = state->msg;
1753         state->msg = NULL;
1754         if (msg)
1755                 hammer2_msg_free_locked(iocom, msg);
1756         free(state);
1757
1758         /*
1759          * When an iocom error is present we are trying to close down the
1760          * iocom, but we have to wait for all states to terminate before
1761          * we can do so.  The iocom rx code will terminate the receive side
1762          * for all transactions by simulating incoming DELETE messages,
1763          * but the state doesn't go away until both sides are terminated.
1764          *
1765          * We may have to wake up the rx code.
1766          */
1767         if (iocom->ioq_rx.error &&
1768             RB_EMPTY(&iocom->staterd_tree) &&
1769             RB_EMPTY(&iocom->statewr_tree)) {
1770                 dummy = 0;
1771                 write(iocom->wakeupfds[1], &dummy, 1);
1772         }
1773 }
1774
1775 const char *
1776 hammer2_basecmd_str(uint32_t cmd)
1777 {
1778         static char buf[64];
1779         char protobuf[32];
1780         char cmdbuf[32];
1781         const char *protostr;
1782         const char *cmdstr;
1783
1784         switch(cmd & HAMMER2_MSGF_PROTOS) {
1785         case HAMMER2_MSG_PROTO_LNK:
1786                 protostr = "LNK_";
1787                 break;
1788         case HAMMER2_MSG_PROTO_DBG:
1789                 protostr = "DBG_";
1790                 break;
1791         case HAMMER2_MSG_PROTO_DOM:
1792                 protostr = "DOM_";
1793                 break;
1794         case HAMMER2_MSG_PROTO_CAC:
1795                 protostr = "CAC_";
1796                 break;
1797         case HAMMER2_MSG_PROTO_QRM:
1798                 protostr = "QRM_";
1799                 break;
1800         case HAMMER2_MSG_PROTO_BLK:
1801                 protostr = "BLK_";
1802                 break;
1803         case HAMMER2_MSG_PROTO_VOP:
1804                 protostr = "VOP_";
1805                 break;
1806         default:
1807                 snprintf(protobuf, sizeof(protobuf), "%x_",
1808                         (cmd & HAMMER2_MSGF_PROTOS) >> 20);
1809                 protostr = protobuf;
1810                 break;
1811         }
1812
1813         switch(cmd & (HAMMER2_MSGF_PROTOS |
1814                       HAMMER2_MSGF_CMDS |
1815                       HAMMER2_MSGF_SIZE)) {
1816         case HAMMER2_LNK_PAD:
1817                 cmdstr = "PAD";
1818                 break;
1819         case HAMMER2_LNK_PING:
1820                 cmdstr = "PING";
1821                 break;
1822         case HAMMER2_LNK_AUTH:
1823                 cmdstr = "AUTH";
1824                 break;
1825         case HAMMER2_LNK_CONN:
1826                 cmdstr = "CONN";
1827                 break;
1828         case HAMMER2_LNK_SPAN:
1829                 cmdstr = "SPAN";
1830                 break;
1831         case HAMMER2_LNK_ERROR:
1832                 if (cmd & HAMMER2_MSGF_DELETE)
1833                         cmdstr = "RETURN";
1834                 else
1835                         cmdstr = "RESULT";
1836                 break;
1837         case HAMMER2_DBG_SHELL:
1838                 cmdstr = "SHELL";
1839                 break;
1840         default:
1841                 snprintf(cmdbuf, sizeof(cmdbuf),
1842                          "%06x", (cmd & (HAMMER2_MSGF_PROTOS |
1843                                          HAMMER2_MSGF_CMDS |
1844                                          HAMMER2_MSGF_SIZE)));
1845                 cmdstr = cmdbuf;
1846                 break;
1847         }
1848         snprintf(buf, sizeof(buf), "%s%s", protostr, cmdstr);
1849         return (buf);
1850 }
1851
1852 const char *
1853 hammer2_msg_str(hammer2_msg_t *msg)
1854 {
1855         hammer2_state_t *state;
1856         static char buf[256];
1857         char errbuf[16];
1858         char statebuf[64];
1859         char flagbuf[64];
1860         const char *statestr;
1861         const char *errstr;
1862         uint32_t basecmd;
1863         int i;
1864
1865         /*
1866          * Parse the state
1867          */
1868         if ((state = msg->state) != NULL) {
1869                 basecmd = (state->rxcmd & HAMMER2_MSGF_REPLY) ?
1870                           state->txcmd : state->rxcmd;
1871                 snprintf(statebuf, sizeof(statebuf),
1872                          " %s=%s,L=%s%s,R=%s%s",
1873                          ((state->txcmd & HAMMER2_MSGF_REPLY) ?
1874                                 "rcvcmd" : "sndcmd"),
1875                          hammer2_basecmd_str(basecmd),
1876                          ((state->txcmd & HAMMER2_MSGF_CREATE) ? "C" : ""),
1877                          ((state->txcmd & HAMMER2_MSGF_DELETE) ? "D" : ""),
1878                          ((state->rxcmd & HAMMER2_MSGF_CREATE) ? "C" : ""),
1879                          ((state->rxcmd & HAMMER2_MSGF_DELETE) ? "D" : "")
1880                 );
1881                 statestr = statebuf;
1882         } else {
1883                 statestr = "";
1884         }
1885
1886         /*
1887          * Parse the error
1888          */
1889         switch(msg->any.head.error) {
1890         case 0:
1891                 errstr = "";
1892                 break;
1893         case HAMMER2_IOQ_ERROR_SYNC:
1894                 errstr = "err=IOQ:NOSYNC";
1895                 break;
1896         case HAMMER2_IOQ_ERROR_EOF:
1897                 errstr = "err=IOQ:STREAMEOF";
1898                 break;
1899         case HAMMER2_IOQ_ERROR_SOCK:
1900                 errstr = "err=IOQ:SOCKERR";
1901                 break;
1902         case HAMMER2_IOQ_ERROR_FIELD:
1903                 errstr = "err=IOQ:BADFIELD";
1904                 break;
1905         case HAMMER2_IOQ_ERROR_HCRC:
1906                 errstr = "err=IOQ:BADHCRC";
1907                 break;
1908         case HAMMER2_IOQ_ERROR_XCRC:
1909                 errstr = "err=IOQ:BADXCRC";
1910                 break;
1911         case HAMMER2_IOQ_ERROR_ACRC:
1912                 errstr = "err=IOQ:BADACRC";
1913                 break;
1914         case HAMMER2_IOQ_ERROR_STATE:
1915                 errstr = "err=IOQ:BADSTATE";
1916                 break;
1917         case HAMMER2_IOQ_ERROR_NOPEER:
1918                 errstr = "err=IOQ:PEERCONFIG";
1919                 break;
1920         case HAMMER2_IOQ_ERROR_NORKEY:
1921                 errstr = "err=IOQ:BADRKEY";
1922                 break;
1923         case HAMMER2_IOQ_ERROR_NOLKEY:
1924                 errstr = "err=IOQ:BADLKEY";
1925                 break;
1926         case HAMMER2_IOQ_ERROR_KEYXCHGFAIL:
1927                 errstr = "err=IOQ:BADKEYXCHG";
1928                 break;
1929         case HAMMER2_IOQ_ERROR_KEYFMT:
1930                 errstr = "err=IOQ:BADFMT";
1931                 break;
1932         case HAMMER2_IOQ_ERROR_BADURANDOM:
1933                 errstr = "err=IOQ:BADRANDOM";
1934                 break;
1935         case HAMMER2_IOQ_ERROR_MSGSEQ:
1936                 errstr = "err=IOQ:BADSEQ";
1937                 break;
1938         case HAMMER2_IOQ_ERROR_EALREADY:
1939                 errstr = "err=IOQ:DUPMSG";
1940                 break;
1941         case HAMMER2_IOQ_ERROR_TRANS:
1942                 errstr = "err=IOQ:BADTRANS";
1943                 break;
1944         case HAMMER2_MSG_ERR_NOSUPP:
1945                 errstr = "err=NOSUPPORT";
1946                 break;
1947         default:
1948                 snprintf(errbuf, sizeof(errbuf),
1949                          " err=%d", msg->any.head.error);
1950                 errstr = errbuf;
1951                 break;
1952         }
1953
1954         /*
1955          * Message flags
1956          */
1957         i = 0;
1958         if (msg->any.head.cmd & (HAMMER2_MSGF_CREATE | HAMMER2_MSGF_DELETE |
1959                                  HAMMER2_MSGF_ABORT | HAMMER2_MSGF_REPLY)) {
1960                 flagbuf[i++] = '|';
1961                 if (msg->any.head.cmd & HAMMER2_MSGF_CREATE)
1962                         flagbuf[i++] = 'C';
1963                 if (msg->any.head.cmd & HAMMER2_MSGF_DELETE)
1964                         flagbuf[i++] = 'D';
1965                 if (msg->any.head.cmd & HAMMER2_MSGF_REPLY)
1966                         flagbuf[i++] = 'R';
1967                 if (msg->any.head.cmd & HAMMER2_MSGF_ABORT)
1968                         flagbuf[i++] = 'A';
1969         }
1970         flagbuf[i] = 0;
1971
1972         /*
1973          * Generate the buf
1974          */
1975         snprintf(buf, sizeof(buf),
1976                 "msg=%s%s %s id=%08x span=%08x %s",
1977                  hammer2_basecmd_str(msg->any.head.cmd),
1978                  flagbuf,
1979                  errstr,
1980                  (uint32_t)(intmax_t)msg->any.head.msgid,   /* for brevity */
1981                  (uint32_t)(intmax_t)msg->any.head.spanid,  /* for brevity */
1982                  statestr);
1983
1984         return(buf);
1985 }