cluster - Major kernel component work (diskiocom, xdisk, kdmsg)
[dragonfly.git] / lib / libdmsg / msg_lnk.c
1 /*
2  * Copyright (c) 2012 The DragonFly Project.  All rights reserved.
3  *
4  * This code is derived from software contributed to The DragonFly Project
5  * by Matthew Dillon <dillon@dragonflybsd.org>
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  *
11  * 1. Redistributions of source code must retain the above copyright
12  *    notice, this list of conditions and the following disclaimer.
13  * 2. Redistributions in binary form must reproduce the above copyright
14  *    notice, this list of conditions and the following disclaimer in
15  *    the documentation and/or other materials provided with the
16  *    distribution.
17  * 3. Neither the name of The DragonFly Project nor the names of its
18  *    contributors may be used to endorse or promote products derived
19  *    from this software without specific, prior written permission.
20  *
21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
32  * SUCH DAMAGE.
33  */
34 /*
35  * LNK_SPAN PROTOCOL SUPPORT FUNCTIONS - Please see sys/dmsg.h for an
36  * involved explanation of the protocol.
37  */
38
39 #include "dmsg_local.h"
40
41 void (*dmsg_node_handler)(void **opaquep, struct dmsg_msg *msg, int op);
42
43 /*
44  * Maximum spanning tree distance.  This has the practical effect of
45  * stopping tail-chasing closed loops when a feeder span is lost.
46  */
47 #define DMSG_SPAN_MAXDIST       16
48
49 /*
50  * RED-BLACK TREE DEFINITIONS
51  *
52  * We need to track:
53  *
54  * (1) shared fsid's (a cluster).
55  * (2) unique fsid's (a node in a cluster) <--- LNK_SPAN transactions.
56  *
57  * We need to aggegate all active LNK_SPANs, aggregate, and create our own
58  * outgoing LNK_SPAN transactions on each of our connections representing
59  * the aggregated state.
60  *
61  * h2span_conn          - list of iocom connections who wish to receive SPAN
62  *                        propagation from other connections.  Might contain
63  *                        a filter string.  Only iocom's with an open
64  *                        LNK_CONN transactions are applicable for SPAN
65  *                        propagation.
66  *
67  * h2span_relay         - List of links relayed (via SPAN).  Essentially
68  *                        each relay structure represents a LNK_SPAN
69  *                        transaction that we initiated, verses h2span_link
70  *                        which is a LNK_SPAN transaction that we received.
71  *
72  * --
73  *
74  * h2span_cluster       - Organizes the shared fsid's.  One structure for
75  *                        each cluster.
76  *
77  * h2span_node          - Organizes the nodes in a cluster.  One structure
78  *                        for each unique {cluster,node}, aka {fsid, pfs_fsid}.
79  *
80  * h2span_link          - Organizes all incoming and outgoing LNK_SPAN message
81  *                        transactions related to a node.
82  *
83  *                        One h2span_link structure for each incoming LNK_SPAN
84  *                        transaction.  Links selected for propagation back
85  *                        out are also where the outgoing LNK_SPAN messages
86  *                        are indexed into (so we can propagate changes).
87  *
88  *                        The h2span_link's use a red-black tree to sort the
89  *                        distance hop metric for the incoming LNK_SPAN.  We
90  *                        then select the top N for outgoing.  When the
91  *                        topology changes the top N may also change and cause
92  *                        new outgoing LNK_SPAN transactions to be opened
93  *                        and less desireable ones to be closed, causing
94  *                        transactional aborts within the message flow in
95  *                        the process.
96  *
97  * Also note            - All outgoing LNK_SPAN message transactions are also
98  *                        entered into a red-black tree for use by the routing
99  *                        function.  This is handled by msg.c in the state
100  *                        code, not here.
101  */
102
103 struct h2span_link;
104 struct h2span_relay;
105 TAILQ_HEAD(h2span_media_queue, h2span_media);
106 TAILQ_HEAD(h2span_conn_queue, h2span_conn);
107 TAILQ_HEAD(h2span_relay_queue, h2span_relay);
108
109 RB_HEAD(h2span_cluster_tree, h2span_cluster);
110 RB_HEAD(h2span_node_tree, h2span_node);
111 RB_HEAD(h2span_link_tree, h2span_link);
112 RB_HEAD(h2span_relay_tree, h2span_relay);
113 uint32_t DMsgRNSS;
114
115 /*
116  * This represents a media
117  */
118 struct h2span_media {
119         TAILQ_ENTRY(h2span_media) entry;
120         uuid_t  mediaid;
121         int     refs;
122         struct h2span_media_config {
123                 dmsg_vol_data_t         copy_run;
124                 dmsg_vol_data_t         copy_pend;
125                 pthread_t               thread;
126                 pthread_cond_t          cond;
127                 int                     ctl;
128                 int                     fd;
129                 dmsg_iocom_t            iocom;
130                 pthread_t               iocom_thread;
131                 enum { H2MC_STOPPED, H2MC_CONNECT, H2MC_RUNNING } state;
132         } config[DMSG_COPYID_COUNT];
133 };
134
135 typedef struct h2span_media_config h2span_media_config_t;
136
137 #define H2CONFCTL_STOP          0x00000001
138 #define H2CONFCTL_UPDATE        0x00000002
139
140 /*
141  * Received LNK_CONN transaction enables SPAN protocol over connection.
142  * (may contain filter).  Typically one for each mount and several may
143  * share the same media.
144  */
145 struct h2span_conn {
146         TAILQ_ENTRY(h2span_conn) entry;
147         struct h2span_relay_tree tree;
148         struct h2span_media *media;
149         dmsg_state_t *state;
150 };
151
152 /*
153  * All received LNK_SPANs are organized by cluster (pfs_clid),
154  * node (pfs_fsid), and link (received LNK_SPAN transaction).
155  */
156 struct h2span_cluster {
157         RB_ENTRY(h2span_cluster) rbnode;
158         struct h2span_node_tree tree;
159         uuid_t  pfs_clid;               /* shared fsid */
160         uint8_t peer_type;
161         char    cl_label[128];          /* cluster label (typ PEER_BLOCK) */
162         int     refs;                   /* prevents destruction */
163 };
164
165 struct h2span_node {
166         RB_ENTRY(h2span_node) rbnode;
167         struct h2span_link_tree tree;
168         struct h2span_cluster *cls;
169         uint8_t pfs_type;
170         uuid_t  pfs_fsid;               /* unique fsid */
171         char    fs_label[128];          /* fs label (typ PEER_HAMMER2) */
172         void    *opaque;
173 };
174
175 struct h2span_link {
176         RB_ENTRY(h2span_link) rbnode;
177         dmsg_state_t    *state;         /* state<->link */
178         struct h2span_node *node;       /* related node */
179         uint32_t        dist;
180         uint32_t        rnss;
181         struct h2span_relay_queue relayq; /* relay out */
182 };
183
184 /*
185  * Any LNK_SPAN transactions we receive which are relayed out other
186  * connections utilize this structure to track the LNK_SPAN transactions
187  * we initiate (relay out) on other connections.  We only relay out
188  * LNK_SPANs on connections we have an open CONN transaction for.
189  *
190  * The relay structure points to the outgoing LNK_SPAN trans (out_state)
191  * and to the incoming LNK_SPAN transaction (in_state).  The relay
192  * structure holds refs on the related states.
193  *
194  * In many respects this is the core of the protocol... actually figuring
195  * out what LNK_SPANs to relay.  The spanid used for relaying is the
196  * address of the 'state' structure, which is why h2span_relay has to
197  * be entered into a RB-TREE based at h2span_conn (so we can look
198  * up the spanid to validate it).
199  */
200 struct h2span_relay {
201         TAILQ_ENTRY(h2span_relay) entry;        /* from link */
202         RB_ENTRY(h2span_relay) rbnode;          /* from h2span_conn */
203         struct h2span_conn      *conn;          /* related CONN transaction */
204         dmsg_state_t            *source_rt;     /* h2span_link state */
205         dmsg_state_t            *target_rt;     /* h2span_relay state */
206 };
207
208 typedef struct h2span_media h2span_media_t;
209 typedef struct h2span_conn h2span_conn_t;
210 typedef struct h2span_cluster h2span_cluster_t;
211 typedef struct h2span_node h2span_node_t;
212 typedef struct h2span_link h2span_link_t;
213 typedef struct h2span_relay h2span_relay_t;
214
215 #define dmsg_termstr(array)     _dmsg_termstr((array), sizeof(array))
216
217 static h2span_relay_t *dmsg_generate_relay(h2span_conn_t *conn,
218                                         h2span_link_t *slink);
219 static uint32_t dmsg_rnss(void);
220
221 static __inline
222 void
223 _dmsg_termstr(char *base, size_t size)
224 {
225         base[size-1] = 0;
226 }
227
228 /*
229  * Cluster peer_type, uuid, AND label must match for a match
230  */
231 static
232 int
233 h2span_cluster_cmp(h2span_cluster_t *cls1, h2span_cluster_t *cls2)
234 {
235         int r;
236
237         if (cls1->peer_type < cls2->peer_type)
238                 return(-1);
239         if (cls1->peer_type > cls2->peer_type)
240                 return(1);
241         r = uuid_compare(&cls1->pfs_clid, &cls2->pfs_clid, NULL);
242         if (r == 0)
243                 r = strcmp(cls1->cl_label, cls2->cl_label);
244
245         return r;
246 }
247
248 /*
249  * Match against fs_label/pfs_fsid.  Together these two items represent a
250  * unique node.  In most cases the primary differentiator is pfs_fsid but
251  * we also string-match fs_label.
252  */
253 static
254 int
255 h2span_node_cmp(h2span_node_t *node1, h2span_node_t *node2)
256 {
257         int r;
258
259         r = strcmp(node1->fs_label, node2->fs_label);
260         if (r == 0)
261                 r = uuid_compare(&node1->pfs_fsid, &node2->pfs_fsid, NULL);
262         return (r);
263 }
264
265 /*
266  * Sort/subsort must match h2span_relay_cmp() under any given node
267  * to make the aggregation algorithm easier, so the best links are
268  * in the same sorted order as the best relays.
269  *
270  * NOTE: We cannot use link*->state->msgid because this msgid is created
271  *       by each remote host and thus might wind up being the same.
272  */
273 static
274 int
275 h2span_link_cmp(h2span_link_t *link1, h2span_link_t *link2)
276 {
277         if (link1->dist < link2->dist)
278                 return(-1);
279         if (link1->dist > link2->dist)
280                 return(1);
281         if (link1->rnss < link2->rnss)
282                 return(-1);
283         if (link1->rnss > link2->rnss)
284                 return(1);
285 #if 1
286         if ((uintptr_t)link1->state < (uintptr_t)link2->state)
287                 return(-1);
288         if ((uintptr_t)link1->state > (uintptr_t)link2->state)
289                 return(1);
290 #else
291         if (link1->state->msgid < link2->state->msgid)
292                 return(-1);
293         if (link1->state->msgid > link2->state->msgid)
294                 return(1);
295 #endif
296         return(0);
297 }
298
299 /*
300  * Relay entries are sorted by node, subsorted by distance and link
301  * address (so we can match up the conn->tree relay topology with
302  * a node's link topology).
303  */
304 static
305 int
306 h2span_relay_cmp(h2span_relay_t *relay1, h2span_relay_t *relay2)
307 {
308         h2span_link_t *link1 = relay1->source_rt->any.link;
309         h2span_link_t *link2 = relay2->source_rt->any.link;
310
311         if ((intptr_t)link1->node < (intptr_t)link2->node)
312                 return(-1);
313         if ((intptr_t)link1->node > (intptr_t)link2->node)
314                 return(1);
315         if (link1->dist < link2->dist)
316                 return(-1);
317         if (link1->dist > link2->dist)
318                 return(1);
319         if (link1->rnss < link2->rnss)
320                 return(-1);
321         if (link1->rnss > link2->rnss)
322                 return(1);
323 #if 1
324         if ((uintptr_t)link1->state < (uintptr_t)link2->state)
325                 return(-1);
326         if ((uintptr_t)link1->state > (uintptr_t)link2->state)
327                 return(1);
328 #else
329         if (link1->state->msgid < link2->state->msgid)
330                 return(-1);
331         if (link1->state->msgid > link2->state->msgid)
332                 return(1);
333 #endif
334         return(0);
335 }
336
337 RB_PROTOTYPE_STATIC(h2span_cluster_tree, h2span_cluster,
338              rbnode, h2span_cluster_cmp);
339 RB_PROTOTYPE_STATIC(h2span_node_tree, h2span_node,
340              rbnode, h2span_node_cmp);
341 RB_PROTOTYPE_STATIC(h2span_link_tree, h2span_link,
342              rbnode, h2span_link_cmp);
343 RB_PROTOTYPE_STATIC(h2span_relay_tree, h2span_relay,
344              rbnode, h2span_relay_cmp);
345
346 RB_GENERATE_STATIC(h2span_cluster_tree, h2span_cluster,
347              rbnode, h2span_cluster_cmp);
348 RB_GENERATE_STATIC(h2span_node_tree, h2span_node,
349              rbnode, h2span_node_cmp);
350 RB_GENERATE_STATIC(h2span_link_tree, h2span_link,
351              rbnode, h2span_link_cmp);
352 RB_GENERATE_STATIC(h2span_relay_tree, h2span_relay,
353              rbnode, h2span_relay_cmp);
354
355 /*
356  * Global mutex protects cluster_tree lookups, connq, mediaq.
357  */
358 static pthread_mutex_t cluster_mtx;
359 static struct h2span_cluster_tree cluster_tree = RB_INITIALIZER(cluster_tree);
360 static struct h2span_conn_queue connq = TAILQ_HEAD_INITIALIZER(connq);
361 static struct h2span_media_queue mediaq = TAILQ_HEAD_INITIALIZER(mediaq);
362
363 static void dmsg_lnk_span(dmsg_msg_t *msg);
364 static void dmsg_lnk_conn(dmsg_msg_t *msg);
365 static void dmsg_lnk_circ(dmsg_msg_t *msg);
366 static void dmsg_lnk_relay(dmsg_msg_t *msg);
367 static void dmsg_relay_scan(h2span_conn_t *conn, h2span_node_t *node);
368 static void dmsg_relay_delete(h2span_relay_t *relay);
369
370 static void *dmsg_volconf_thread(void *info);
371 static void dmsg_volconf_stop(h2span_media_config_t *conf);
372 static void dmsg_volconf_start(h2span_media_config_t *conf,
373                                 const char *hostname);
374
375 void
376 dmsg_msg_lnk_signal(dmsg_iocom_t *iocom __unused)
377 {
378         pthread_mutex_lock(&cluster_mtx);
379         dmsg_relay_scan(NULL, NULL);
380         pthread_mutex_unlock(&cluster_mtx);
381 }
382
383 /*
384  * DMSG_PROTO_LNK - Generic DMSG_PROTO_LNK.
385  *            (incoming iocom lock not held)
386  *
387  * This function is typically called for one-way and opening-transactions
388  * since state->func is assigned after that, but it will also be called
389  * if no state->func is assigned on transaction-open.
390  */
391 void
392 dmsg_msg_lnk(dmsg_msg_t *msg)
393 {
394         uint32_t icmd = msg->state ? msg->state->icmd : msg->any.head.cmd;
395
396         switch(icmd & DMSGF_BASECMDMASK) {
397         case DMSG_LNK_CONN:
398                 dmsg_lnk_conn(msg);
399                 break;
400         case DMSG_LNK_SPAN:
401                 dmsg_lnk_span(msg);
402                 break;
403         case DMSG_LNK_CIRC:
404                 dmsg_lnk_circ(msg);
405                 break;
406         default:
407                 fprintf(stderr,
408                         "MSG_PROTO_LNK: Unknown msg %08x\n", msg->any.head.cmd);
409                 dmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
410                 /* state invalid after reply */
411                 break;
412         }
413 }
414
415 /*
416  * LNK_CONN - iocom identify message reception.
417  *            (incoming iocom lock not held)
418  *
419  * Remote node identifies itself to us, sets up a SPAN filter, and gives us
420  * the ok to start transmitting SPANs.
421  */
422 void
423 dmsg_lnk_conn(dmsg_msg_t *msg)
424 {
425         dmsg_state_t *state = msg->state;
426         h2span_media_t *media;
427         h2span_media_config_t *conf;
428         h2span_conn_t *conn;
429         h2span_relay_t *relay;
430         char *alloc = NULL;
431         int i;
432
433         pthread_mutex_lock(&cluster_mtx);
434
435         fprintf(stderr, "dmsg_lnk_conn: msg %p cmd %08x state %p txcmd %08x rxcmd %08x\n",
436                 msg, msg->any.head.cmd, state, state->txcmd, state->rxcmd);
437
438         switch(msg->any.head.cmd & DMSGF_TRANSMASK) {
439         case DMSG_LNK_CONN | DMSGF_CREATE:
440         case DMSG_LNK_CONN | DMSGF_CREATE | DMSGF_DELETE:
441                 /*
442                  * On transaction start we allocate a new h2span_conn and
443                  * acknowledge the request, leaving the transaction open.
444                  * We then relay priority-selected SPANs.
445                  */
446                 fprintf(stderr, "LNK_CONN(%08x): %s/%s/%s\n",
447                         (uint32_t)msg->any.head.msgid,
448                         dmsg_uuid_to_str(&msg->any.lnk_conn.pfs_clid,
449                                             &alloc),
450                         msg->any.lnk_conn.cl_label,
451                         msg->any.lnk_conn.fs_label);
452                 free(alloc);
453
454                 conn = dmsg_alloc(sizeof(*conn));
455
456                 RB_INIT(&conn->tree);
457                 state->iocom->conn = conn;      /* XXX only one */
458                 conn->state = state;
459                 state->func = dmsg_lnk_conn;
460                 state->any.conn = conn;
461                 TAILQ_INSERT_TAIL(&connq, conn, entry);
462
463                 /*
464                  * Set up media
465                  */
466                 TAILQ_FOREACH(media, &mediaq, entry) {
467                         if (uuid_compare(&msg->any.lnk_conn.mediaid,
468                                          &media->mediaid, NULL) == 0) {
469                                 break;
470                         }
471                 }
472                 if (media == NULL) {
473                         media = dmsg_alloc(sizeof(*media));
474                         media->mediaid = msg->any.lnk_conn.mediaid;
475                         TAILQ_INSERT_TAIL(&mediaq, media, entry);
476                 }
477                 conn->media = media;
478                 ++media->refs;
479
480                 if ((msg->any.head.cmd & DMSGF_DELETE) == 0) {
481                         dmsg_msg_result(msg, 0);
482                         dmsg_iocom_signal(msg->iocom);
483                         break;
484                 }
485                 /* FALL THROUGH */
486         case DMSG_LNK_CONN | DMSGF_DELETE:
487         case DMSG_LNK_ERROR | DMSGF_DELETE:
488 deleteconn:
489                 /*
490                  * On transaction terminate we clean out our h2span_conn
491                  * and acknowledge the request, closing the transaction.
492                  */
493                 fprintf(stderr, "LNK_CONN: Terminated\n");
494                 conn = state->any.conn;
495                 assert(conn);
496
497                 /*
498                  * Clean out the media structure. If refs drops to zero we
499                  * also clean out the media config threads.  These threads
500                  * maintain span connections to other hammer2 service daemons.
501                  */
502                 media = conn->media;
503                 if (--media->refs == 0) {
504                         fprintf(stderr, "Shutting down media spans\n");
505                         for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
506                                 conf = &media->config[i];
507
508                                 if (conf->thread == NULL)
509                                         continue;
510                                 conf->ctl = H2CONFCTL_STOP;
511                                 pthread_cond_signal(&conf->cond);
512                         }
513                         for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
514                                 conf = &media->config[i];
515
516                                 if (conf->thread == NULL)
517                                         continue;
518                                 pthread_mutex_unlock(&cluster_mtx);
519                                 pthread_join(conf->thread, NULL);
520                                 pthread_mutex_lock(&cluster_mtx);
521                                 conf->thread = NULL;
522                                 pthread_cond_destroy(&conf->cond);
523                         }
524                         fprintf(stderr, "Media shutdown complete\n");
525                         TAILQ_REMOVE(&mediaq, media, entry);
526                         dmsg_free(media);
527                 }
528
529                 /*
530                  * Clean out all relays.  This requires terminating each
531                  * relay transaction.
532                  */
533                 while ((relay = RB_ROOT(&conn->tree)) != NULL) {
534                         dmsg_relay_delete(relay);
535                 }
536
537                 /*
538                  * Clean out conn
539                  */
540                 conn->media = NULL;
541                 conn->state = NULL;
542                 msg->state->any.conn = NULL;
543                 msg->state->iocom->conn = NULL;
544                 TAILQ_REMOVE(&connq, conn, entry);
545                 dmsg_free(conn);
546
547                 dmsg_msg_reply(msg, 0);
548                 /* state invalid after reply */
549                 break;
550         case DMSG_LNK_VOLCONF:
551                 /*
552                  * One-way volume-configuration message is transmitted
553                  * over the open LNK_CONN transaction.
554                  */
555                 fprintf(stderr, "RECEIVED VOLCONF\n");
556                 if (msg->any.lnk_volconf.index < 0 ||
557                     msg->any.lnk_volconf.index >= DMSG_COPYID_COUNT) {
558                         fprintf(stderr, "VOLCONF: ILLEGAL INDEX %d\n",
559                                 msg->any.lnk_volconf.index);
560                         break;
561                 }
562                 if (msg->any.lnk_volconf.copy.path[sizeof(msg->any.lnk_volconf.copy.path) - 1] != 0 ||
563                     msg->any.lnk_volconf.copy.path[0] == 0) {
564                         fprintf(stderr, "VOLCONF: ILLEGAL PATH %d\n",
565                                 msg->any.lnk_volconf.index);
566                         break;
567                 }
568                 conn = msg->state->any.conn;
569                 if (conn == NULL) {
570                         fprintf(stderr, "VOLCONF: LNK_CONN is missing\n");
571                         break;
572                 }
573                 conf = &conn->media->config[msg->any.lnk_volconf.index];
574                 conf->copy_pend = msg->any.lnk_volconf.copy;
575                 conf->ctl |= H2CONFCTL_UPDATE;
576                 if (conf->thread == NULL) {
577                         fprintf(stderr, "VOLCONF THREAD STARTED\n");
578                         pthread_cond_init(&conf->cond, NULL);
579                         pthread_create(&conf->thread, NULL,
580                                        dmsg_volconf_thread, (void *)conf);
581                 }
582                 pthread_cond_signal(&conf->cond);
583                 break;
584         default:
585                 /*
586                  * Failsafe
587                  */
588                 if (msg->any.head.cmd & DMSGF_DELETE)
589                         goto deleteconn;
590                 dmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
591                 break;
592         }
593         pthread_mutex_unlock(&cluster_mtx);
594 }
595
596 /*
597  * LNK_SPAN - Spanning tree protocol message reception
598  *            (incoming iocom lock not held)
599  *
600  * Receive a spanning tree transactional message, creating or destroying
601  * a SPAN and propagating it to other iocoms.
602  */
603 void
604 dmsg_lnk_span(dmsg_msg_t *msg)
605 {
606         dmsg_state_t *state = msg->state;
607         h2span_cluster_t dummy_cls;
608         h2span_node_t dummy_node;
609         h2span_cluster_t *cls;
610         h2span_node_t *node;
611         h2span_link_t *slink;
612         h2span_relay_t *relay;
613         char *alloc = NULL;
614
615         assert((msg->any.head.cmd & DMSGF_REPLY) == 0);
616
617         pthread_mutex_lock(&cluster_mtx);
618
619         /*
620          * On transaction start we initialize the tracking infrastructure
621          */
622         if (msg->any.head.cmd & DMSGF_CREATE) {
623                 assert(state->func == NULL);
624                 state->func = dmsg_lnk_span;
625
626                 dmsg_termstr(msg->any.lnk_span.cl_label);
627                 dmsg_termstr(msg->any.lnk_span.fs_label);
628
629                 /*
630                  * Find the cluster
631                  */
632                 dummy_cls.pfs_clid = msg->any.lnk_span.pfs_clid;
633                 dummy_cls.peer_type = msg->any.lnk_span.peer_type;
634                 bcopy(msg->any.lnk_span.cl_label,
635                       dummy_cls.cl_label,
636                       sizeof(dummy_cls.cl_label));
637                 cls = RB_FIND(h2span_cluster_tree, &cluster_tree, &dummy_cls);
638                 if (cls == NULL) {
639                         cls = dmsg_alloc(sizeof(*cls));
640                         cls->pfs_clid = msg->any.lnk_span.pfs_clid;
641                         cls->peer_type = msg->any.lnk_span.peer_type;
642                         bcopy(msg->any.lnk_span.cl_label,
643                               cls->cl_label,
644                               sizeof(cls->cl_label));
645                         RB_INIT(&cls->tree);
646                         RB_INSERT(h2span_cluster_tree, &cluster_tree, cls);
647                 }
648
649                 /*
650                  * Find the node
651                  */
652                 dummy_node.pfs_fsid = msg->any.lnk_span.pfs_fsid;
653                 bcopy(msg->any.lnk_span.fs_label, dummy_node.fs_label,
654                       sizeof(dummy_node.fs_label));
655                 node = RB_FIND(h2span_node_tree, &cls->tree, &dummy_node);
656                 if (node == NULL) {
657                         node = dmsg_alloc(sizeof(*node));
658                         node->pfs_fsid = msg->any.lnk_span.pfs_fsid;
659                         node->pfs_type = msg->any.lnk_span.pfs_type;
660                         bcopy(msg->any.lnk_span.fs_label,
661                               node->fs_label,
662                               sizeof(node->fs_label));
663                         node->cls = cls;
664                         RB_INIT(&node->tree);
665                         RB_INSERT(h2span_node_tree, &cls->tree, node);
666                         if (dmsg_node_handler) {
667                                 dmsg_node_handler(&node->opaque, msg,
668                                                   DMSG_NODEOP_ADD);
669                         }
670                 }
671
672                 /*
673                  * Create the link
674                  */
675                 assert(state->any.link == NULL);
676                 slink = dmsg_alloc(sizeof(*slink));
677                 TAILQ_INIT(&slink->relayq);
678                 slink->node = node;
679                 slink->dist = msg->any.lnk_span.dist;
680                 slink->rnss = msg->any.lnk_span.rnss;
681                 slink->state = state;
682                 state->any.link = slink;
683
684                 RB_INSERT(h2span_link_tree, &node->tree, slink);
685
686                 fprintf(stderr,
687                         "LNK_SPAN(thr %p): %p %s cl=%s fs=%s dist=%d\n",
688                         msg->iocom,
689                         slink,
690                         dmsg_uuid_to_str(&msg->any.lnk_span.pfs_clid, &alloc),
691                         msg->any.lnk_span.cl_label,
692                         msg->any.lnk_span.fs_label,
693                         msg->any.lnk_span.dist);
694                 free(alloc);
695 #if 0
696                 dmsg_relay_scan(NULL, node);
697 #endif
698                 dmsg_iocom_signal(msg->iocom);
699         }
700
701         /*
702          * On transaction terminate we remove the tracking infrastructure.
703          */
704         if (msg->any.head.cmd & DMSGF_DELETE) {
705                 slink = state->any.link;
706                 assert(slink != NULL);
707                 node = slink->node;
708                 cls = node->cls;
709
710                 fprintf(stderr, "LNK_DELE(thr %p): %p %s cl=%s fs=%s dist=%d\n",
711                         msg->iocom,
712                         slink,
713                         dmsg_uuid_to_str(&cls->pfs_clid, &alloc),
714                         state->msg->any.lnk_span.cl_label,
715                         state->msg->any.lnk_span.fs_label,
716                         state->msg->any.lnk_span.dist);
717                 free(alloc);
718
719                 /*
720                  * Clean out all relays.  This requires terminating each
721                  * relay transaction.
722                  */
723                 while ((relay = TAILQ_FIRST(&slink->relayq)) != NULL) {
724                         dmsg_relay_delete(relay);
725                 }
726
727                 /*
728                  * Clean out the topology
729                  */
730                 RB_REMOVE(h2span_link_tree, &node->tree, slink);
731                 if (RB_EMPTY(&node->tree)) {
732                         RB_REMOVE(h2span_node_tree, &cls->tree, node);
733                         if (dmsg_node_handler) {
734                                 dmsg_node_handler(&node->opaque, msg,
735                                                   DMSG_NODEOP_DEL);
736                         }
737                         if (RB_EMPTY(&cls->tree) && cls->refs == 0) {
738                                 RB_REMOVE(h2span_cluster_tree,
739                                           &cluster_tree, cls);
740                                 dmsg_free(cls);
741                         }
742                         node->cls = NULL;
743                         dmsg_free(node);
744                         node = NULL;
745                 }
746                 state->any.link = NULL;
747                 slink->state = NULL;
748                 slink->node = NULL;
749                 dmsg_free(slink);
750
751                 /*
752                  * We have to terminate the transaction
753                  */
754                 dmsg_state_reply(state, 0);
755                 /* state invalid after reply */
756
757                 /*
758                  * If the node still exists issue any required updates.  If
759                  * it doesn't then all related relays have already been
760                  * removed and there's nothing left to do.
761                  */
762 #if 0
763                 if (node)
764                         dmsg_relay_scan(NULL, node);
765 #endif
766                 if (node)
767                         dmsg_iocom_signal(msg->iocom);
768         }
769
770         pthread_mutex_unlock(&cluster_mtx);
771 }
772
773 /*
774  * LNK_CIRC - Virtual circuit protocol message reception
775  *            (incoming iocom lock not held)
776  *
777  * Handles all cases.
778  */
779 void
780 dmsg_lnk_circ(dmsg_msg_t *msg)
781 {
782         dmsg_circuit_t *circA;
783         dmsg_circuit_t *circB;
784         dmsg_state_t *rx_state;
785         dmsg_state_t *tx_state;
786         dmsg_state_t *state;
787         dmsg_state_t dummy;
788         dmsg_msg_t *fwd_msg;
789         dmsg_iocom_t *iocomA;
790         dmsg_iocom_t *iocomB;
791         int disconnect;
792
793         /*pthread_mutex_lock(&cluster_mtx);*/
794
795         switch (msg->any.head.cmd & (DMSGF_CREATE |
796                                      DMSGF_DELETE |
797                                      DMSGF_REPLY)) {
798         case DMSGF_CREATE:
799         case DMSGF_CREATE | DMSGF_DELETE:
800                 /*
801                  * (A) wishes to establish a virtual circuit through us to (B).
802                  * (B) is specified by lnk_circ.target (the message id for
803                  * a LNK_SPAN that (A) received from us which represents (B)).
804                  *
805                  * Designate the originator of the circuit (the current
806                  * remote end) as (A) and the other side as (B).
807                  *
808                  * Accept the VC but do not reply.  We will wait for the end-
809                  * to-end reply to propagate back.
810                  */
811                 iocomA = msg->iocom;
812
813                 /*
814                  * Locate the open transaction state that the other end
815                  * specified in <target>.  This will be an open SPAN
816                  * transaction that we transmitted (h2span_relay) over
817                  * the interface the LNK_CIRC is being received on.
818                  *
819                  * (all LNK_CIRC's that we transmit are on circuit0)
820                  */
821                 pthread_mutex_lock(&iocomA->mtx);
822                 dummy.msgid = msg->any.lnk_circ.target;
823                 tx_state = RB_FIND(dmsg_state_tree,
824                                    &iocomA->circuit0.statewr_tree,
825                                    &dummy);
826                 pthread_mutex_unlock(&iocomA->mtx);
827                 if (tx_state == NULL) {
828                         fprintf(stderr, "dmsg_lnk_circ: no circuit\n");
829                         dmsg_msg_reply(msg, DMSG_ERR_CANTCIRC);
830                         break;
831                 }
832
833                 /* locate h2span_link */
834                 rx_state = tx_state->any.relay->source_rt;
835
836                 /*
837                  * A wishes to establish a VC through us to the
838                  * specified target.
839                  *
840                  * A sends us the msgid of an open SPAN transaction
841                  * it received from us as <target>.
842                  */
843                 circA = dmsg_alloc(sizeof(*circA));
844                 circA->iocom = iocomA;
845                 circA->state = msg->state;      /* LNK_CIRC state */
846                 circA->msgid = msg->state->msgid;
847                 circA->span_state = tx_state;   /* H2SPAN_RELAY state */
848                 circA->is_relay = 1;
849                 circA->refs = 2;                /* state and peer */
850                 msg->state->any.circ = circA;
851
852                 iocomB = rx_state->iocom;
853
854                 circB = dmsg_alloc(sizeof(*circB));
855
856                 /*
857                  * Create a LNK_CIRC transaction on B
858                  */
859                 fwd_msg = dmsg_msg_alloc(&iocomB->circuit0,
860                                          0, DMSG_LNK_CIRC | DMSGF_CREATE,
861                                          dmsg_lnk_circ, circB);
862                 fwd_msg->state->any.circ = circB;
863                 circB->iocom = iocomB;
864                 circB->state = fwd_msg->state;  /* LNK_CIRC state */
865                 circB->msgid = fwd_msg->any.head.msgid;
866                 circB->span_state = rx_state;   /* H2SPAN_LINK state */
867                 circB->is_relay = 0;
868                 circB->refs = 2;                /* state and peer */
869
870                 /*
871                  * Link the two circuits together.
872                  */
873                 circA->peer = circB;
874                 circB->peer = circA;
875
876                 if (RB_INSERT(dmsg_circuit_tree, &iocomA->circuit_tree, circA))
877                         assert(0);
878                 if (RB_INSERT(dmsg_circuit_tree, &iocomB->circuit_tree, circB))
879                         assert(0);
880
881                 dmsg_msg_write(fwd_msg);
882
883                 if ((msg->any.head.cmd & DMSGF_DELETE) == 0)
884                         break;
885                 /* FALL THROUGH TO DELETE */
886         case DMSGF_DELETE:
887                 /*
888                  * (A) Is deleting the virtual circuit, propogate closure
889                  * to (B).
890                  */
891                 iocomA = msg->iocom;
892                 if (msg->state->any.circ == NULL) {
893                         /* already returned an error/deleted */
894                         break;
895                 }
896                 circA = msg->state->any.circ;
897                 circB = circA->peer;
898                 assert(msg->state == circA->state);
899
900                 /*
901                  * We are closing B's send side.  If B's receive side is
902                  * already closed we disconnect the circuit from B's state.
903                  */
904                 disconnect = 0;
905                 if (circB && (state = circB->state) != NULL) {
906                         if (state->rxcmd & DMSGF_DELETE) {
907                                 circB->state = NULL;
908                                 state->any.circ = NULL;
909                                 dmsg_circuit_drop(circB);
910                         }
911                         dmsg_state_reply(state, msg->any.head.error);
912                         disconnect = 1;
913                 }
914
915                 /*
916                  * We received a close on A.  If A's send side is already
917                  * closed we disconnect the circuit from A's state.
918                  */
919                 if (circA && (state = circA->state) != NULL) {
920                         if (state->txcmd & DMSGF_DELETE) {
921                                 circA->state = NULL;
922                                 state->any.circ = NULL;
923                                 dmsg_circuit_drop(circA);
924                         }
925                         disconnect = 1;
926                 }
927
928                 /*
929                  * Disconnect the peer<->peer association
930                  */
931                 if (disconnect) {
932                         if (circB) {
933                                 circA->peer = NULL;
934                                 circB->peer = NULL;
935                                 dmsg_circuit_drop(circA);
936                                 dmsg_circuit_drop(circB); /* XXX SMP */
937                         }
938                 }
939                 break;
940         case DMSGF_REPLY | DMSGF_CREATE:
941         case DMSGF_REPLY | DMSGF_CREATE | DMSGF_DELETE:
942                 /*
943                  * (B) is acknowledging the creation of the virtual
944                  * circuit.  This propagates all the way back to (A), though
945                  * it should be noted that (A) can start issuing commands
946                  * via the virtual circuit before seeing this reply.
947                  */
948                 circB = msg->state->any.circ;
949                 assert(circB);
950                 circA = circB->peer;
951                 assert(msg->state == circB->state);
952                 assert(circA);
953                 if ((msg->any.head.cmd & DMSGF_DELETE) == 0) {
954                         dmsg_state_result(circA->state, msg->any.head.error);
955                         break;
956                 }
957                 /* FALL THROUGH TO DELETE */
958         case DMSGF_REPLY | DMSGF_DELETE:
959                 /*
960                  * (B) Is deleting the virtual circuit or acknowledging
961                  * our deletion of the virtual circuit, propogate closure
962                  * to (A).
963                  */
964                 iocomB = msg->iocom;
965                 circB = msg->state->any.circ;
966                 circA = circB->peer;
967                 assert(msg->state == circB->state);
968
969                 /*
970                  * We received a close on (B), propagate to (A).  If we have
971                  * already received the close from (A) we disconnect the state.
972                  */
973                 disconnect = 0;
974                 if (circA && (state = circA->state) != NULL) {
975                         if (state->rxcmd & DMSGF_DELETE) {
976                                 circA->state = NULL;
977                                 state->any.circ = NULL;
978                                 dmsg_circuit_drop(circA);
979                         }
980                         dmsg_state_reply(state, msg->any.head.error);
981                         disconnect = 1;
982                 }
983
984                 /*
985                  * We received a close on (B).  If (B)'s send side is already
986                  * closed we disconnect the state.
987                  */
988                 if (circB && (state = circB->state) != NULL) {
989                         if (state->txcmd & DMSGF_DELETE) {
990                                 circB->state = NULL;
991                                 state->any.circ = NULL;
992                                 dmsg_circuit_drop(circB);
993                         }
994                         disconnect = 1;
995                 }
996
997                 /*
998                  * Disconnect the peer<->peer association
999                  */
1000                 if (disconnect) {
1001                         if (circA) {
1002                                 circB->peer = NULL;
1003                                 circA->peer = NULL;
1004                                 dmsg_circuit_drop(circB);
1005                                 dmsg_circuit_drop(circA); /* XXX SMP */
1006                         }
1007                 }
1008                 break;
1009         }
1010
1011         /*pthread_mutex_lock(&cluster_mtx);*/
1012 }
1013
1014 /*
1015  * Update relay transactions for SPANs.
1016  *
1017  * Called with cluster_mtx held.
1018  */
1019 static void dmsg_relay_scan_specific(h2span_node_t *node,
1020                                         h2span_conn_t *conn);
1021
1022 static void
1023 dmsg_relay_scan(h2span_conn_t *conn, h2span_node_t *node)
1024 {
1025         h2span_cluster_t *cls;
1026
1027         if (node) {
1028                 /*
1029                  * Iterate specific node
1030                  */
1031                 TAILQ_FOREACH(conn, &connq, entry)
1032                         dmsg_relay_scan_specific(node, conn);
1033         } else {
1034                 /*
1035                  * Full iteration.
1036                  *
1037                  * Iterate cluster ids, nodes, and either a specific connection
1038                  * or all connections.
1039                  */
1040                 RB_FOREACH(cls, h2span_cluster_tree, &cluster_tree) {
1041                         /*
1042                          * Iterate node ids
1043                          */
1044                         RB_FOREACH(node, h2span_node_tree, &cls->tree) {
1045                                 /*
1046                                  * Synchronize the node's link (received SPANs)
1047                                  * with each connection's relays.
1048                                  */
1049                                 if (conn) {
1050                                         dmsg_relay_scan_specific(node, conn);
1051                                 } else {
1052                                         TAILQ_FOREACH(conn, &connq, entry) {
1053                                             dmsg_relay_scan_specific(node,
1054                                                                         conn);
1055                                         }
1056                                         assert(conn == NULL);
1057                                 }
1058                         }
1059                 }
1060         }
1061 }
1062
1063 /*
1064  * Update the relay'd SPANs for this (node, conn).
1065  *
1066  * Iterate links and adjust relays to match.  We only propagate the top link
1067  * for now (XXX we want to propagate the top two).
1068  *
1069  * The dmsg_relay_scan_cmp() function locates the first relay element
1070  * for any given node.  The relay elements will be sub-sorted by dist.
1071  */
1072 struct relay_scan_info {
1073         h2span_node_t *node;
1074         h2span_relay_t *relay;
1075 };
1076
1077 static int
1078 dmsg_relay_scan_cmp(h2span_relay_t *relay, void *arg)
1079 {
1080         struct relay_scan_info *info = arg;
1081
1082         if ((intptr_t)relay->source_rt->any.link->node < (intptr_t)info->node)
1083                 return(-1);
1084         if ((intptr_t)relay->source_rt->any.link->node > (intptr_t)info->node)
1085                 return(1);
1086         return(0);
1087 }
1088
1089 static int
1090 dmsg_relay_scan_callback(h2span_relay_t *relay, void *arg)
1091 {
1092         struct relay_scan_info *info = arg;
1093
1094         info->relay = relay;
1095         return(-1);
1096 }
1097
1098 static void
1099 dmsg_relay_scan_specific(h2span_node_t *node, h2span_conn_t *conn)
1100 {
1101         struct relay_scan_info info;
1102         h2span_relay_t *relay;
1103         h2span_relay_t *next_relay;
1104         h2span_link_t *slink;
1105         dmsg_lnk_conn_t *lconn;
1106         dmsg_lnk_span_t *lspan;
1107         int count;
1108         int maxcount = 2;
1109         uint32_t lastdist = DMSG_SPAN_MAXDIST;
1110         uint32_t lastrnss = 0;
1111
1112         info.node = node;
1113         info.relay = NULL;
1114
1115         /*
1116          * Locate the first related relay for the node on this connection.
1117          * relay will be NULL if there were none.
1118          */
1119         RB_SCAN(h2span_relay_tree, &conn->tree,
1120                 dmsg_relay_scan_cmp, dmsg_relay_scan_callback, &info);
1121         relay = info.relay;
1122         info.relay = NULL;
1123         if (relay)
1124                 assert(relay->source_rt->any.link->node == node);
1125
1126         if (DMsgDebugOpt > 8)
1127                 fprintf(stderr, "relay scan for connection %p\n", conn);
1128
1129         /*
1130          * Iterate the node's links (received SPANs) in distance order,
1131          * lowest (best) dist first.
1132          *
1133          * PROPAGATE THE BEST LINKS OVER THE SPECIFIED CONNECTION.
1134          *
1135          * Track relays while iterating the best links and construct
1136          * missing relays when necessary.
1137          *
1138          * (If some prior better link was removed it would have also
1139          *  removed the relay, so the relay can only match exactly or
1140          *  be worse).
1141          */
1142         count = 0;
1143         RB_FOREACH(slink, h2span_link_tree, &node->tree) {
1144                 /*
1145                  * Increment count of successful relays.  This isn't
1146                  * quite accurate if we break out but nothing after
1147                  * the loop uses (count).
1148                  *
1149                  * If count exceeds the maximum number of relays we desire
1150                  * we normally want to break out.  However, in order to
1151                  * guarantee a symmetric path we have to continue if both
1152                  * (dist) and (rnss) continue to match.  Otherwise the SPAN
1153                  * propagation in the reverse direction may choose different
1154                  * routes and we will not have a symmetric path.
1155                  *
1156                  * NOTE: Spanning tree does not have to be symmetrical so
1157                  *       this code is not currently enabled.
1158                  */
1159                 if (++count >= maxcount) {
1160 #ifdef REQUIRE_SYMMETRICAL
1161                         if (lastdist != slink->dist || lastrnss != slink->rnss)
1162                                 break;
1163 #else
1164                         break;
1165 #endif
1166                         /* go beyond the nominal maximum desired relays */
1167                 }
1168
1169                 /*
1170                  * Match, relay already in-place, get the next
1171                  * relay to match against the next slink.
1172                  */
1173                 if (relay && relay->source_rt->any.link == slink) {
1174                         relay = RB_NEXT(h2span_relay_tree, &conn->tree, relay);
1175                         continue;
1176                 }
1177
1178                 /*
1179                  * We might want this SLINK, if it passes our filters.
1180                  *
1181                  * The spanning tree can cause closed loops so we have
1182                  * to limit slink->dist.
1183                  */
1184                 if (slink->dist > DMSG_SPAN_MAXDIST)
1185                         break;
1186
1187                 /*
1188                  * Don't bother transmitting a LNK_SPAN out the same
1189                  * connection it came in on.  Trivial optimization.
1190                  */
1191                 if (slink->state->iocom == conn->state->iocom)
1192                         break;
1193
1194                 /*
1195                  * NOTE ON FILTERS: The protocol spec allows non-requested
1196                  * SPANs to be transmitted, the other end is expected to
1197                  * leave their transactions open but otherwise ignore them.
1198                  *
1199                  * Don't bother transmitting if the remote connection
1200                  * is not accepting this SPAN's peer_type.
1201                  *
1202                  * pfs_mask is typically used so pure clients can filter
1203                  * out receiving SPANs for other pure clients.
1204                  */
1205                 lspan = &slink->state->msg->any.lnk_span;
1206                 lconn = &conn->state->msg->any.lnk_conn;
1207                 if (((1LLU << lspan->peer_type) & lconn->peer_mask) == 0)
1208                         break;
1209                 if (((1LLU << lspan->pfs_type) & lconn->pfs_mask) == 0)
1210                         break;
1211
1212                 /*
1213                  * Do not give pure clients visibility to other pure clients
1214                  */
1215                 if (lconn->pfs_type == DMSG_PFSTYPE_CLIENT &&
1216                     lspan->pfs_type == DMSG_PFSTYPE_CLIENT) {
1217                         break;
1218                 }
1219
1220                 /*
1221                  * Connection filter, if cluster uuid is not NULL it must
1222                  * match the span cluster uuid.  Only applies when the
1223                  * peer_type matches.
1224                  */
1225                 if (lspan->peer_type == lconn->peer_type &&
1226                     !uuid_is_nil(&lconn->pfs_clid, NULL) &&
1227                     uuid_compare(&slink->node->cls->pfs_clid,
1228                                  &lconn->pfs_clid, NULL)) {
1229                         break;
1230                 }
1231
1232                 /*
1233                  * Connection filter, if cluster label is not empty it must
1234                  * match the span cluster label.  Only applies when the
1235                  * peer_type matches.
1236                  */
1237                 if (lspan->peer_type == lconn->peer_type &&
1238                     lconn->cl_label[0] &&
1239                     strcmp(lconn->cl_label, slink->node->cls->cl_label)) {
1240                         break;
1241                 }
1242
1243                 /*
1244                  * NOTE! pfs_fsid differentiates nodes within the same cluster
1245                  *       so we obviously don't want to match those.  Similarly
1246                  *       for fs_label.
1247                  */
1248
1249                 /*
1250                  * Ok, we've accepted this SPAN for relaying.
1251                  */
1252                 assert(relay == NULL ||
1253                        relay->source_rt->any.link->node != slink->node ||
1254                        relay->source_rt->any.link->dist >= slink->dist);
1255                 relay = dmsg_generate_relay(conn, slink);
1256                 lastdist = slink->dist;
1257                 lastrnss = slink->rnss;
1258
1259                 /*
1260                  * Match (created new relay), get the next relay to
1261                  * match against the next slink.
1262                  */
1263                 relay = RB_NEXT(h2span_relay_tree, &conn->tree, relay);
1264         }
1265
1266         /*
1267          * Any remaining relay's belonging to this connection which match
1268          * the node are in excess of the current aggregate spanning state
1269          * and should be removed.
1270          */
1271         while (relay && relay->source_rt->any.link->node == node) {
1272                 next_relay = RB_NEXT(h2span_relay_tree, &conn->tree, relay);
1273                 dmsg_relay_delete(relay);
1274                 relay = next_relay;
1275         }
1276 }
1277
1278 /*
1279  * Helper function to generate missing relay.
1280  *
1281  * cluster_mtx must be held
1282  */
1283 static
1284 h2span_relay_t *
1285 dmsg_generate_relay(h2span_conn_t *conn, h2span_link_t *slink)
1286 {
1287         h2span_relay_t *relay;
1288         h2span_node_t *node;
1289         dmsg_msg_t *msg;
1290
1291         node = slink->node;
1292
1293         relay = dmsg_alloc(sizeof(*relay));
1294         relay->conn = conn;
1295         relay->source_rt = slink->state;
1296         /* relay->source_rt->any.link = slink; */
1297
1298         /*
1299          * NOTE: relay->target_rt->any.relay set to relay by alloc.
1300          */
1301         msg = dmsg_msg_alloc(&conn->state->iocom->circuit0,
1302                              0, DMSG_LNK_SPAN | DMSGF_CREATE,
1303                              dmsg_lnk_relay, relay);
1304         relay->target_rt = msg->state;
1305
1306         msg->any.lnk_span = slink->state->msg->any.lnk_span;
1307         msg->any.lnk_span.dist = slink->dist + 1;
1308         msg->any.lnk_span.rnss = slink->rnss + dmsg_rnss();
1309
1310         RB_INSERT(h2span_relay_tree, &conn->tree, relay);
1311         TAILQ_INSERT_TAIL(&slink->relayq, relay, entry);
1312
1313         dmsg_msg_write(msg);
1314
1315         return (relay);
1316 }
1317
1318 /*
1319  * Messages received on relay SPANs.  These are open transactions so it is
1320  * in fact possible for the other end to close the transaction.
1321  *
1322  * XXX MPRACE on state structure
1323  */
1324 static void
1325 dmsg_lnk_relay(dmsg_msg_t *msg)
1326 {
1327         dmsg_state_t *state = msg->state;
1328         h2span_relay_t *relay;
1329
1330         assert(msg->any.head.cmd & DMSGF_REPLY);
1331
1332         if (msg->any.head.cmd & DMSGF_DELETE) {
1333                 pthread_mutex_lock(&cluster_mtx);
1334                 if ((relay = state->any.relay) != NULL) {
1335                         dmsg_relay_delete(relay);
1336                 } else {
1337                         dmsg_state_reply(state, 0);
1338                 }
1339                 pthread_mutex_unlock(&cluster_mtx);
1340         }
1341 }
1342
1343
1344 static
1345 void
1346 dmsg_relay_delete(h2span_relay_t *relay)
1347 {
1348         fprintf(stderr,
1349                 "RELAY DELETE %p RELAY %p ON CLS=%p NODE=%p DIST=%d FD %d STATE %p\n",
1350                 relay->source_rt->any.link,
1351                 relay,
1352                 relay->source_rt->any.link->node->cls, relay->source_rt->any.link->node,
1353                 relay->source_rt->any.link->dist,
1354                 relay->conn->state->iocom->sock_fd, relay->target_rt);
1355
1356         RB_REMOVE(h2span_relay_tree, &relay->conn->tree, relay);
1357         TAILQ_REMOVE(&relay->source_rt->any.link->relayq, relay, entry);
1358
1359         if (relay->target_rt) {
1360                 relay->target_rt->any.relay = NULL;
1361                 dmsg_state_reply(relay->target_rt, 0);
1362                 /* state invalid after reply */
1363                 relay->target_rt = NULL;
1364         }
1365         relay->conn = NULL;
1366         relay->source_rt = NULL;
1367         dmsg_free(relay);
1368 }
1369
1370 static void *
1371 dmsg_volconf_thread(void *info)
1372 {
1373         h2span_media_config_t *conf = info;
1374
1375         pthread_mutex_lock(&cluster_mtx);
1376         while ((conf->ctl & H2CONFCTL_STOP) == 0) {
1377                 if (conf->ctl & H2CONFCTL_UPDATE) {
1378                         fprintf(stderr, "VOLCONF UPDATE\n");
1379                         conf->ctl &= ~H2CONFCTL_UPDATE;
1380                         if (bcmp(&conf->copy_run, &conf->copy_pend,
1381                                  sizeof(conf->copy_run)) == 0) {
1382                                 fprintf(stderr, "VOLCONF: no changes\n");
1383                                 continue;
1384                         }
1385                         /*
1386                          * XXX TODO - auto reconnect on lookup failure or
1387                          *              connect failure or stream failure.
1388                          */
1389
1390                         pthread_mutex_unlock(&cluster_mtx);
1391                         dmsg_volconf_stop(conf);
1392                         conf->copy_run = conf->copy_pend;
1393                         if (conf->copy_run.copyid != 0 &&
1394                             strncmp(conf->copy_run.path, "span:", 5) == 0) {
1395                                 dmsg_volconf_start(conf,
1396                                                       conf->copy_run.path + 5);
1397                         }
1398                         pthread_mutex_lock(&cluster_mtx);
1399                         fprintf(stderr, "VOLCONF UPDATE DONE state %d\n", conf->state);
1400                 }
1401                 if (conf->state == H2MC_CONNECT) {
1402                         dmsg_volconf_start(conf, conf->copy_run.path + 5);
1403                         pthread_mutex_unlock(&cluster_mtx);
1404                         sleep(5);
1405                         pthread_mutex_lock(&cluster_mtx);
1406                 } else {
1407                         pthread_cond_wait(&conf->cond, &cluster_mtx);
1408                 }
1409         }
1410         pthread_mutex_unlock(&cluster_mtx);
1411         dmsg_volconf_stop(conf);
1412         return(NULL);
1413 }
1414
1415 static
1416 void
1417 dmsg_volconf_stop(h2span_media_config_t *conf)
1418 {
1419         switch(conf->state) {
1420         case H2MC_STOPPED:
1421                 break;
1422         case H2MC_CONNECT:
1423                 conf->state = H2MC_STOPPED;
1424                 break;
1425         case H2MC_RUNNING:
1426                 shutdown(conf->fd, SHUT_WR);
1427                 pthread_join(conf->iocom_thread, NULL);
1428                 conf->iocom_thread = NULL;
1429                 break;
1430         }
1431 }
1432
1433 static
1434 void
1435 dmsg_volconf_start(h2span_media_config_t *conf, const char *hostname)
1436 {
1437         dmsg_master_service_info_t *info;
1438
1439         switch(conf->state) {
1440         case H2MC_STOPPED:
1441         case H2MC_CONNECT:
1442                 conf->fd = dmsg_connect(hostname);
1443                 if (conf->fd < 0) {
1444                         fprintf(stderr, "Unable to connect to %s\n", hostname);
1445                         conf->state = H2MC_CONNECT;
1446                 } else {
1447                         info = malloc(sizeof(*info));
1448                         bzero(info, sizeof(*info));
1449                         info->fd = conf->fd;
1450                         info->detachme = 0;
1451                         conf->state = H2MC_RUNNING;
1452                         pthread_create(&conf->iocom_thread, NULL,
1453                                        dmsg_master_service, info);
1454                 }
1455                 break;
1456         case H2MC_RUNNING:
1457                 break;
1458         }
1459 }
1460
1461 /************************************************************************
1462  *                      MESSAGE ROUTING AND SOURCE VALIDATION           *
1463  ************************************************************************/
1464
1465 int
1466 dmsg_circuit_relay(dmsg_msg_t *msg)
1467 {
1468         dmsg_iocom_t *iocom = msg->iocom;
1469         dmsg_circuit_t *circ;
1470         dmsg_circuit_t *peer;
1471         dmsg_circuit_t dummy;
1472         int error = 0;
1473
1474         /*
1475          * Relay occurs before any state processing, msg state should always
1476          * be NULL.
1477          */
1478         assert(msg->state == NULL);
1479
1480         /*
1481          * Lookup the circuit on the incoming iocom.
1482          */
1483         pthread_mutex_lock(&cluster_mtx);
1484
1485         dummy.msgid = msg->any.head.circuit;
1486         circ = RB_FIND(dmsg_circuit_tree, &iocom->circuit_tree, &dummy);
1487         assert(circ);
1488         peer = circ->peer;
1489
1490         msg->iocom = peer->iocom;
1491         msg->any.head.circuit = peer->msgid;
1492
1493         pthread_mutex_unlock(&cluster_mtx);
1494
1495         dmsg_msg_write(msg);
1496         error = DMSG_IOQ_ERROR_ROUTED;
1497
1498         return error;
1499 }
1500
1501 /************************************************************************
1502  *                      ROUTER AND MESSAGING HANDLES                    *
1503  ************************************************************************
1504  *
1505  * Basically the idea here is to provide a stable data structure which
1506  * can be localized to the caller for higher level protocols to work with.
1507  * Depends on the context, these dmsg_handle's can be pooled by use-case
1508  * and remain persistent through a client (or mount point's) life.
1509  */
1510
1511 #if 0
1512 /*
1513  * Obtain a stable handle on a cluster given its uuid.  This ties directly
1514  * into the global cluster topology, creating the structure if necessary
1515  * (even if the uuid does not exist or does not exist yet), and preventing
1516  * the structure from getting ripped out from under us while we hold a
1517  * pointer to it.
1518  */
1519 h2span_cluster_t *
1520 dmsg_cluster_get(uuid_t *pfs_clid)
1521 {
1522         h2span_cluster_t dummy_cls;
1523         h2span_cluster_t *cls;
1524
1525         dummy_cls.pfs_clid = *pfs_clid;
1526         pthread_mutex_lock(&cluster_mtx);
1527         cls = RB_FIND(h2span_cluster_tree, &cluster_tree, &dummy_cls);
1528         if (cls)
1529                 ++cls->refs;
1530         pthread_mutex_unlock(&cluster_mtx);
1531         return (cls);
1532 }
1533
1534 void
1535 dmsg_cluster_put(h2span_cluster_t *cls)
1536 {
1537         pthread_mutex_lock(&cluster_mtx);
1538         assert(cls->refs > 0);
1539         --cls->refs;
1540         if (RB_EMPTY(&cls->tree) && cls->refs == 0) {
1541                 RB_REMOVE(h2span_cluster_tree,
1542                           &cluster_tree, cls);
1543                 dmsg_free(cls);
1544         }
1545         pthread_mutex_unlock(&cluster_mtx);
1546 }
1547
1548 /*
1549  * Obtain a stable handle to a specific cluster node given its uuid.
1550  * This handle does NOT lock in the route to the node and is typically
1551  * used as part of the dmsg_handle_*() API to obtain a set of
1552  * stable nodes.
1553  */
1554 h2span_node_t *
1555 dmsg_node_get(h2span_cluster_t *cls, uuid_t *pfs_fsid)
1556 {
1557 }
1558
1559 #endif
1560
1561 /*
1562  * Dumps the spanning tree
1563  */
1564 void
1565 dmsg_shell_tree(dmsg_circuit_t *circuit, char *cmdbuf __unused)
1566 {
1567         h2span_cluster_t *cls;
1568         h2span_node_t *node;
1569         h2span_link_t *slink;
1570         char *uustr = NULL;
1571
1572         pthread_mutex_lock(&cluster_mtx);
1573         RB_FOREACH(cls, h2span_cluster_tree, &cluster_tree) {
1574                 dmsg_circuit_printf(circuit, "Cluster %s %s (%s)\n",
1575                                   dmsg_peer_type_to_str(cls->peer_type),
1576                                   dmsg_uuid_to_str(&cls->pfs_clid, &uustr),
1577                                   cls->cl_label);
1578                 RB_FOREACH(node, h2span_node_tree, &cls->tree) {
1579                         dmsg_circuit_printf(circuit, "    Node %s %s (%s)\n",
1580                                 dmsg_pfs_type_to_str(node->pfs_type),
1581                                 dmsg_uuid_to_str(&node->pfs_fsid, &uustr),
1582                                 node->fs_label);
1583                         RB_FOREACH(slink, h2span_link_tree, &node->tree) {
1584                                 dmsg_circuit_printf(circuit,
1585                                             "\tLink dist=%d via %d\n",
1586                                             slink->dist,
1587                                             slink->state->iocom->sock_fd);
1588                         }
1589                 }
1590         }
1591         pthread_mutex_unlock(&cluster_mtx);
1592         if (uustr)
1593                 free(uustr);
1594 #if 0
1595         TAILQ_FOREACH(conn, &connq, entry) {
1596         }
1597 #endif
1598 }
1599
1600 /*
1601  * Random number sub-sort value to add to SPAN rnss fields on relay.
1602  * This allows us to differentiate spans with the same <dist> field
1603  * for relaying purposes.  We must normally limit the number of relays
1604  * for any given SPAN origination but we must also guarantee that a
1605  * symmetric reverse path exists, so we use the rnss field as a sub-sort
1606  * (since there can be thousands or millions if we only match on <dist>),
1607  * and if there STILL too many spans we go past the limit.
1608  */
1609 static
1610 uint32_t
1611 dmsg_rnss(void)
1612 {
1613         if (DMsgRNSS == 0) {
1614                 pthread_mutex_lock(&cluster_mtx);
1615                 while (DMsgRNSS == 0) {
1616                         srandomdev();
1617                         DMsgRNSS = random();
1618                 }
1619                 pthread_mutex_unlock(&cluster_mtx);
1620         }
1621         return(DMsgRNSS);
1622 }