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         switch(msg->any.head.cmd & DMSGF_TRANSMASK) {
 436         case DMSG_LNK_CONN | DMSGF_CREATE:
 437         case DMSG_LNK_CONN | DMSGF_CREATE | DMSGF_DELETE:
 438                 /*
 439                  * On transaction start we allocate a new h2span_conn and
 440                  * acknowledge the request, leaving the transaction open.
 441                  * We then relay priority-selected SPANs.
 442                  */
 443                 fprintf(stderr, "LNK_CONN(%08x): %s/%s/%s\n",
 444                         (uint32_t)msg->any.head.msgid,
 445                         dmsg_uuid_to_str(&msg->any.lnk_conn.pfs_clid,
 446                                             &alloc),
 447                         msg->any.lnk_conn.cl_label,
 448                         msg->any.lnk_conn.fs_label);
 449                 free(alloc);
 450
 451                 conn = dmsg_alloc(sizeof(*conn));
 452
 453                 RB_INIT(&conn->tree);
 454                 state->iocom->conn = conn;      /* XXX only one */
 455                 conn->state = state;
 456                 state->func = dmsg_lnk_conn;
 457                 state->any.conn = conn;
 458                 TAILQ_INSERT_TAIL(&connq, conn, entry);
 459
 460                 /*
 461                  * Set up media
 462                  */
 463                 TAILQ_FOREACH(media, &mediaq, entry) {
 464                         if (uuid_compare(&msg->any.lnk_conn.mediaid,
 465                                          &media->mediaid, NULL) == 0) {
 466                                 break;
 467                         }
 468                 }
 469                 if (media == NULL) {
 470                         media = dmsg_alloc(sizeof(*media));
 471                         media->mediaid = msg->any.lnk_conn.mediaid;
 472                         TAILQ_INSERT_TAIL(&mediaq, media, entry);
 473                 }
 474                 conn->media = media;
 475                 ++media->refs;
 476
 477                 if ((msg->any.head.cmd & DMSGF_DELETE) == 0) {
 478                         dmsg_msg_result(msg, 0);
 479                         dmsg_iocom_signal(msg->iocom);
 480                         break;
 481                 }
 482                 /* FALL THROUGH */
 483         case DMSG_LNK_CONN | DMSGF_DELETE:
 484         case DMSG_LNK_ERROR | DMSGF_DELETE:
 485 deleteconn:
 486                 /*
 487                  * On transaction terminate we clean out our h2span_conn
 488                  * and acknowledge the request, closing the transaction.
 489                  */
 490                 fprintf(stderr, "LNK_CONN: Terminated\n");
 491                 conn = state->any.conn;
 492                 assert(conn);
 493
 494                 /*
 495                  * Clean out the media structure. If refs drops to zero we
 496                  * also clean out the media config threads.  These threads
 497                  * maintain span connections to other hammer2 service daemons.
 498                  */
 499                 media = conn->media;
 500                 if (--media->refs == 0) {
 501                         fprintf(stderr, "Shutting down media spans\n");
 502                         for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
 503                                 conf = &media->config[i];
 504
 505                                 if (conf->thread == NULL)
 506                                         continue;
 507                                 conf->ctl = H2CONFCTL_STOP;
 508                                 pthread_cond_signal(&conf->cond);
 509                         }
 510                         for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
 511                                 conf = &media->config[i];
 512
 513                                 if (conf->thread == NULL)
 514                                         continue;
 515                                 pthread_mutex_unlock(&cluster_mtx);
 516                                 pthread_join(conf->thread, NULL);
 517                                 pthread_mutex_lock(&cluster_mtx);
 518                                 conf->thread = NULL;
 519                                 pthread_cond_destroy(&conf->cond);
 520                         }
 521                         fprintf(stderr, "Media shutdown complete\n");
 522                         TAILQ_REMOVE(&mediaq, media, entry);
 523                         dmsg_free(media);
 524                 }
 525
 526                 /*
 527                  * Clean out all relays.  This requires terminating each
 528                  * relay transaction.
 529                  */
 530                 while ((relay = RB_ROOT(&conn->tree)) != NULL) {
 531                         dmsg_relay_delete(relay);
 532                 }
 533
 534                 /*
 535                  * Clean out conn
 536                  */
 537                 conn->media = NULL;
 538                 conn->state = NULL;
 539                 msg->state->any.conn = NULL;
 540                 msg->state->iocom->conn = NULL;
 541                 TAILQ_REMOVE(&connq, conn, entry);
 542                 dmsg_free(conn);
 543
 544                 dmsg_msg_reply(msg, 0);
 545                 /* state invalid after reply */
 546                 break;
 547         case DMSG_LNK_VOLCONF:
 548                 /*
 549                  * One-way volume-configuration message is transmitted
 550                  * over the open LNK_CONN transaction.
 551                  */
 552                 fprintf(stderr, "RECEIVED VOLCONF\n");
 553                 if (msg->any.lnk_volconf.index < 0 ||
 554                     msg->any.lnk_volconf.index >= DMSG_COPYID_COUNT) {
 555                         fprintf(stderr, "VOLCONF: ILLEGAL INDEX %d\n",
 556                                 msg->any.lnk_volconf.index);
 557                         break;
 558                 }
 559                 if (msg->any.lnk_volconf.copy.path[sizeof(msg->any.lnk_volconf.copy.path) - 1] != 0 ||
 560                     msg->any.lnk_volconf.copy.path[0] == 0) {
 561                         fprintf(stderr, "VOLCONF: ILLEGAL PATH %d\n",
 562                                 msg->any.lnk_volconf.index);
 563                         break;
 564                 }
 565                 conn = msg->state->any.conn;
 566                 if (conn == NULL) {
 567                         fprintf(stderr, "VOLCONF: LNK_CONN is missing\n");
 568                         break;
 569                 }
 570                 conf = &conn->media->config[msg->any.lnk_volconf.index];
 571                 conf->copy_pend = msg->any.lnk_volconf.copy;
 572                 conf->ctl |= H2CONFCTL_UPDATE;
 573                 if (conf->thread == NULL) {
 574                         fprintf(stderr, "VOLCONF THREAD STARTED\n");
 575                         pthread_cond_init(&conf->cond, NULL);
 576                         pthread_create(&conf->thread, NULL,
 577                                        dmsg_volconf_thread, (void *)conf);
 578                 }
 579                 pthread_cond_signal(&conf->cond);
 580                 break;
 581         default:
 582                 /*
 583                  * Failsafe
 584                  */
 585                 if (msg->any.head.cmd & DMSGF_DELETE)
 586                         goto deleteconn;
 587                 dmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
 588                 break;
 589         }
 590         pthread_mutex_unlock(&cluster_mtx);
 591 }
 592
 593 /*
 594  * LNK_SPAN - Spanning tree protocol message reception
 595  *            (incoming iocom lock not held)
 596  *
 597  * Receive a spanning tree transactional message, creating or destroying
 598  * a SPAN and propagating it to other iocoms.
 599  */
 600 void
 601 dmsg_lnk_span(dmsg_msg_t *msg)
 602 {
 603         dmsg_state_t *state = msg->state;
 604         h2span_cluster_t dummy_cls;
 605         h2span_node_t dummy_node;
 606         h2span_cluster_t *cls;
 607         h2span_node_t *node;
 608         h2span_link_t *slink;
 609         h2span_relay_t *relay;
 610         char *alloc = NULL;
 611
 612         assert((msg->any.head.cmd & DMSGF_REPLY) == 0);
 613
 614         pthread_mutex_lock(&cluster_mtx);
 615
 616         /*
 617          * On transaction start we initialize the tracking infrastructure
 618          */
 619         if (msg->any.head.cmd & DMSGF_CREATE) {
 620                 assert(state->func == NULL);
 621                 state->func = dmsg_lnk_span;
 622
 623                 dmsg_termstr(msg->any.lnk_span.cl_label);
 624                 dmsg_termstr(msg->any.lnk_span.fs_label);
 625
 626                 /*
 627                  * Find the cluster
 628                  */
 629                 dummy_cls.pfs_clid = msg->any.lnk_span.pfs_clid;
 630                 dummy_cls.peer_type = msg->any.lnk_span.peer_type;
 631                 bcopy(msg->any.lnk_span.cl_label,
 632                       dummy_cls.cl_label,
 633                       sizeof(dummy_cls.cl_label));
 634                 cls = RB_FIND(h2span_cluster_tree, &cluster_tree, &dummy_cls);
 635                 if (cls == NULL) {
 636                         cls = dmsg_alloc(sizeof(*cls));
 637                         cls->pfs_clid = msg->any.lnk_span.pfs_clid;
 638                         cls->peer_type = msg->any.lnk_span.peer_type;
 639                         bcopy(msg->any.lnk_span.cl_label,
 640                               cls->cl_label,
 641                               sizeof(cls->cl_label));
 642                         RB_INIT(&cls->tree);
 643                         RB_INSERT(h2span_cluster_tree, &cluster_tree, cls);
 644                 }
 645
 646                 /*
 647                  * Find the node
 648                  */
 649                 dummy_node.pfs_fsid = msg->any.lnk_span.pfs_fsid;
 650                 bcopy(msg->any.lnk_span.fs_label, dummy_node.fs_label,
 651                       sizeof(dummy_node.fs_label));
 652                 node = RB_FIND(h2span_node_tree, &cls->tree, &dummy_node);
 653                 if (node == NULL) {
 654                         node = dmsg_alloc(sizeof(*node));
 655                         node->pfs_fsid = msg->any.lnk_span.pfs_fsid;
 656                         node->pfs_type = msg->any.lnk_span.pfs_type;
 657                         bcopy(msg->any.lnk_span.fs_label,
 658                               node->fs_label,
 659                               sizeof(node->fs_label));
 660                         node->cls = cls;
 661                         RB_INIT(&node->tree);
 662                         RB_INSERT(h2span_node_tree, &cls->tree, node);
 663                         if (dmsg_node_handler) {
 664                                 dmsg_node_handler(&node->opaque, msg,
 665                                                   DMSG_NODEOP_ADD);
 666                         }
 667                 }
 668
 669                 /*
 670                  * Create the link
 671                  */
 672                 assert(state->any.link == NULL);
 673                 slink = dmsg_alloc(sizeof(*slink));
 674                 TAILQ_INIT(&slink->relayq);
 675                 slink->node = node;
 676                 slink->dist = msg->any.lnk_span.dist;
 677                 slink->rnss = msg->any.lnk_span.rnss;
 678                 slink->state = state;
 679                 state->any.link = slink;
 680
 681                 RB_INSERT(h2span_link_tree, &node->tree, slink);
 682
 683                 fprintf(stderr,
 684                         "LNK_SPAN(thr %p): %p %s cl=%s fs=%s dist=%d\n",
 685                         msg->iocom,
 686                         slink,
 687                         dmsg_uuid_to_str(&msg->any.lnk_span.pfs_clid, &alloc),
 688                         msg->any.lnk_span.cl_label,
 689                         msg->any.lnk_span.fs_label,
 690                         msg->any.lnk_span.dist);
 691                 free(alloc);
 692 #if 0
 693                 dmsg_relay_scan(NULL, node);
 694 #endif
 695                 dmsg_iocom_signal(msg->iocom);
 696         }
 697
 698         /*
 699          * On transaction terminate we remove the tracking infrastructure.
 700          */
 701         if (msg->any.head.cmd & DMSGF_DELETE) {
 702                 slink = state->any.link;
 703                 assert(slink != NULL);
 704                 node = slink->node;
 705                 cls = node->cls;
 706
 707                 fprintf(stderr, "LNK_DELE(thr %p): %p %s cl=%s fs=%s dist=%d\n",
 708                         msg->iocom,
 709                         slink,
 710                         dmsg_uuid_to_str(&cls->pfs_clid, &alloc),
 711                         state->msg->any.lnk_span.cl_label,
 712                         state->msg->any.lnk_span.fs_label,
 713                         state->msg->any.lnk_span.dist);
 714                 free(alloc);
 715
 716                 /*
 717                  * Clean out all relays.  This requires terminating each
 718                  * relay transaction.
 719                  */
 720                 while ((relay = TAILQ_FIRST(&slink->relayq)) != NULL) {
 721                         dmsg_relay_delete(relay);
 722                 }
 723
 724                 /*
 725                  * Clean out the topology
 726                  */
 727                 RB_REMOVE(h2span_link_tree, &node->tree, slink);
 728                 if (RB_EMPTY(&node->tree)) {
 729                         RB_REMOVE(h2span_node_tree, &cls->tree, node);
 730                         if (dmsg_node_handler) {
 731                                 dmsg_node_handler(&node->opaque, msg,
 732                                                   DMSG_NODEOP_DEL);
 733                         }
 734                         if (RB_EMPTY(&cls->tree) && cls->refs == 0) {
 735                                 RB_REMOVE(h2span_cluster_tree,
 736                                           &cluster_tree, cls);
 737                                 dmsg_free(cls);
 738                         }
 739                         node->cls = NULL;
 740                         dmsg_free(node);
 741                         node = NULL;
 742                 }
 743                 state->any.link = NULL;
 744                 slink->state = NULL;
 745                 slink->node = NULL;
 746                 dmsg_free(slink);
 747
 748                 /*
 749                  * We have to terminate the transaction
 750                  */
 751                 dmsg_state_reply(state, 0);
 752                 /* state invalid after reply */
 753
 754                 /*
 755                  * If the node still exists issue any required updates.  If
 756                  * it doesn't then all related relays have already been
 757                  * removed and there's nothing left to do.
 758                  */
 759 #if 0
 760                 if (node)
 761                         dmsg_relay_scan(NULL, node);
 762 #endif
 763                 if (node)
 764                         dmsg_iocom_signal(msg->iocom);
 765         }
 766
 767         pthread_mutex_unlock(&cluster_mtx);
 768 }
 769
 770 /*
 771  * LNK_CIRC - Virtual circuit protocol message reception
 772  *            (incoming iocom lock not held)
 773  *
 774  * Handles all cases.
 775  */
 776 void
 777 dmsg_lnk_circ(dmsg_msg_t *msg)
 778 {
 779         dmsg_circuit_t *circA;
 780         dmsg_circuit_t *circB;
 781         dmsg_state_t *rx_state;
 782         dmsg_state_t *tx_state;
 783         dmsg_state_t *state;
 784         dmsg_state_t dummy;
 785         dmsg_msg_t *fwd_msg;
 786         dmsg_iocom_t *iocomA;
 787         dmsg_iocom_t *iocomB;
 788
 789         /*pthread_mutex_lock(&cluster_mtx);*/
 790
 791         switch (msg->any.head.cmd & (DMSGF_CREATE |
 792                                      DMSGF_DELETE |
 793                                      DMSGF_REPLY)) {
 794         case DMSGF_CREATE:
 795         case DMSGF_CREATE | DMSGF_DELETE:
 796                 /*
 797                  * (A) wishes to establish a virtual circuit through us to (B).
 798                  * (B) is specified by lnk_circ.target (the message id for
 799                  * a LNK_SPAN that (A) received from us which represents (B)).
 800                  *
 801                  * Designate the originator of the circuit (the current
 802                  * remote end) as (A) and the other side as (B).
 803                  *
 804                  * Accept the VC but do not reply.  We will wait for the end-
 805                  * to-end reply to propagate back.
 806                  */
 807                 iocomA = msg->iocom;
 808
 809                 /*
 810                  * Locate the open transaction state that the other end
 811                  * specified in <target>.  This will be an open SPAN
 812                  * transaction that we transmitted (h2span_relay) over
 813                  * the interface the LNK_CIRC is being received on.
 814                  *
 815                  * (all LNK_CIRC's that we transmit are on circuit0)
 816                  */
 817                 pthread_mutex_lock(&iocomA->mtx);
 818                 dummy.msgid = msg->any.lnk_circ.target;
 819                 tx_state = RB_FIND(dmsg_state_tree,
 820                                    &iocomA->circuit0.statewr_tree,
 821                                    &dummy);
 822                 /* XXX state refs */
 823                 assert(tx_state);
 824                 pthread_mutex_unlock(&iocomA->mtx);
 825
 826                 /* locate h2span_link */
 827                 rx_state = tx_state->any.relay->source_rt;
 828
 829                 /*
 830                  * A wishes to establish a VC through us to the
 831                  * specified target.
 832                  *
 833                  * A sends us the msgid of an open SPAN transaction
 834                  * it received from us as <target>.
 835                  */
 836                 circA = dmsg_alloc(sizeof(*circA));
 837                 circA->iocom = iocomA;
 838                 circA->state = msg->state;      /* LNK_CIRC state */
 839                 circA->msgid = msg->state->msgid;
 840                 circA->span_state = tx_state;   /* H2SPAN_RELAY state */
 841                 circA->is_relay = 1;
 842                 circA->refs = 2;                /* state and peer */
 843                 msg->state->any.circ = circA;
 844
 845                 iocomB = rx_state->iocom;
 846
 847                 circB = dmsg_alloc(sizeof(*circB));
 848
 849                 /*
 850                  * Create a LNK_CIRC transaction on B
 851                  */
 852                 fwd_msg = dmsg_msg_alloc(&iocomB->circuit0,
 853                                          0, DMSG_LNK_CIRC | DMSGF_CREATE,
 854                                          dmsg_lnk_circ, circB);
 855                 fwd_msg->state->any.circ = circB;
 856                 circB->iocom = iocomB;
 857                 circB->state = fwd_msg->state;  /* LNK_CIRC state */
 858                 circB->msgid = fwd_msg->any.head.msgid;
 859                 circB->span_state = rx_state;   /* H2SPAN_LINK state */
 860                 circB->is_relay = 0;
 861                 circB->refs = 2;                /* state and peer */
 862
 863                 /*
 864                  * Link the two circuits together.
 865                  */
 866                 circA->peer = circB;
 867                 circB->peer = circA;
 868
 869                 if (RB_INSERT(dmsg_circuit_tree, &iocomA->circuit_tree, circA))
 870                         assert(0);
 871                 if (RB_INSERT(dmsg_circuit_tree, &iocomB->circuit_tree, circB))
 872                         assert(0);
 873
 874                 dmsg_msg_write(fwd_msg);
 875
 876                 if ((msg->any.head.cmd & DMSGF_DELETE) == 0)
 877                         break;
 878                 /* FALL THROUGH TO DELETE */
 879         case DMSGF_DELETE:
 880                 /*
 881                  * (A) Is deleting the virtual circuit, propogate closure
 882                  * to (B).
 883                  */
 884                 iocomA = msg->iocom;
 885                 circA = msg->state->any.circ;
 886                 circB = circA->peer;
 887                 assert(msg->state == circA->state);
 888
 889                 /*
 890                  * If we are closing A and the peer B is closed, disconnect.
 891                  */
 892                 if (circB && (state = circB->state) != NULL) {
 893                         if (state->rxcmd & DMSGF_DELETE) {
 894                                 circB->state = NULL;
 895                                 state->any.circ = NULL;
 896                                 dmsg_circuit_drop(circB);
 897                         }
 898                         dmsg_state_reply(state, msg->any.head.error);
 899                 }
 900
 901                 /*
 902                  * If both sides now closed terminate the peer association
 903                  * and the state association.  This may drop up to two refs
 904                  * on circA and one on circB.
 905                  */
 906                 if (circA->state->txcmd & DMSGF_DELETE) {
 907                         if (circB) {
 908                                 circA->peer = NULL;
 909                                 circB->peer = NULL;
 910                                 dmsg_circuit_drop(circA);
 911                                 dmsg_circuit_drop(circB); /* XXX SMP */
 912                         }
 913                         circA->state->any.circ = NULL;
 914                         circA->state = NULL;
 915                         dmsg_circuit_drop(circA);
 916                 }
 917                 break;
 918         case DMSGF_REPLY | DMSGF_CREATE:
 919         case DMSGF_REPLY | DMSGF_CREATE | DMSGF_DELETE:
 920                 /*
 921                  * (B) is acknowledging the creation of the virtual
 922                  * circuit.  This propagates all the way back to (A), though
 923                  * it should be noted that (A) can start issuing commands
 924                  * via the virtual circuit before seeing this reply.
 925                  */
 926                 circB = msg->state->any.circ;
 927                 circA = circB->peer;
 928                 assert(msg->state == circB->state);
 929                 if (circA && (msg->any.head.cmd & DMSGF_DELETE) == 0) {
 930                         dmsg_state_result(circA->state, msg->any.head.error);
 931                         break;
 932                 }
 933                 /* FALL THROUGH TO DELETE */
 934         case DMSGF_REPLY | DMSGF_DELETE:
 935                 /*
 936                  * (B) Is deleting the virtual circuit or acknowledging
 937                  * our deletion of the virtual circuit, propogate closure
 938                  * to (A).
 939                  */
 940                 iocomB = msg->iocom;
 941                 circB = msg->state->any.circ;
 942                 circA = circB->peer;
 943                 assert(msg->state == circB->state);
 944
 945                 /*
 946                  * If we are closing A and the peer B is closed, disconnect.
 947                  */
 948                 if (circA && (state = circA->state) != NULL) {
 949                         if (state->rxcmd & DMSGF_DELETE) {
 950                                 circA->state = NULL;
 951                                 state->any.circ = NULL;
 952                                 dmsg_circuit_drop(circA);
 953                         }
 954                         dmsg_state_reply(state, msg->any.head.error);
 955                 }
 956
 957                 /*
 958                  * If both sides now closed terminate the peer association
 959                  * and the state association.  This may drop up to two refs
 960                  * on circA and one on circB.
 961                  */
 962                 if (circB->state->txcmd & DMSGF_DELETE) {
 963                         if (circA) {
 964                                 circB->peer = NULL;
 965                                 circA->peer = NULL;
 966                                 dmsg_circuit_drop(circB);
 967                                 dmsg_circuit_drop(circA); /* XXX SMP */
 968                         }
 969                         circB->state->any.circ = NULL;
 970                         circB->state = NULL;
 971                         dmsg_circuit_drop(circB);
 972                 }
 973                 break;
 974         }
 975
 976         /*pthread_mutex_lock(&cluster_mtx);*/
 977 }
 978
 979 /*
 980  * Update relay transactions for SPANs.
 981  *
 982  * Called with cluster_mtx held.
 983  */
 984 static void dmsg_relay_scan_specific(h2span_node_t *node,
 985                                         h2span_conn_t *conn);
 986
 987 static void
 988 dmsg_relay_scan(h2span_conn_t *conn, h2span_node_t *node)
 989 {
 990         h2span_cluster_t *cls;
 991
 992         if (node) {
 993                 /*
 994                  * Iterate specific node
 995                  */
 996                 TAILQ_FOREACH(conn, &connq, entry)
 997                         dmsg_relay_scan_specific(node, conn);
 998         } else {
 999                 /*
1000                  * Full iteration.
1001                  *
1002                  * Iterate cluster ids, nodes, and either a specific connection
1003                  * or all connections.
1004                  */
1005                 RB_FOREACH(cls, h2span_cluster_tree, &cluster_tree) {
1006                         /*
1007                          * Iterate node ids
1008                          */
1009                         RB_FOREACH(node, h2span_node_tree, &cls->tree) {
1010                                 /*
1011                                  * Synchronize the node's link (received SPANs)
1012                                  * with each connection's relays.
1013                                  */
1014                                 if (conn) {
1015                                         dmsg_relay_scan_specific(node, conn);
1016                                 } else {
1017                                         TAILQ_FOREACH(conn, &connq, entry) {
1018                                             dmsg_relay_scan_specific(node,
1019                                                                         conn);
1020                                         }
1021                                         assert(conn == NULL);
1022                                 }
1023                         }
1024                 }
1025         }
1026 }
1027
1028 /*
1029  * Update the relay'd SPANs for this (node, conn).
1030  *
1031  * Iterate links and adjust relays to match.  We only propagate the top link
1032  * for now (XXX we want to propagate the top two).
1033  *
1034  * The dmsg_relay_scan_cmp() function locates the first relay element
1035  * for any given node.  The relay elements will be sub-sorted by dist.
1036  */
1037 struct relay_scan_info {
1038         h2span_node_t *node;
1039         h2span_relay_t *relay;
1040 };
1041
1042 static int
1043 dmsg_relay_scan_cmp(h2span_relay_t *relay, void *arg)
1044 {
1045         struct relay_scan_info *info = arg;
1046
1047         if ((intptr_t)relay->source_rt->any.link->node < (intptr_t)info->node)
1048                 return(-1);
1049         if ((intptr_t)relay->source_rt->any.link->node > (intptr_t)info->node)
1050                 return(1);
1051         return(0);
1052 }
1053
1054 static int
1055 dmsg_relay_scan_callback(h2span_relay_t *relay, void *arg)
1056 {
1057         struct relay_scan_info *info = arg;
1058
1059         info->relay = relay;
1060         return(-1);
1061 }
1062
1063 static void
1064 dmsg_relay_scan_specific(h2span_node_t *node, h2span_conn_t *conn)
1065 {
1066         struct relay_scan_info info;
1067         h2span_relay_t *relay;
1068         h2span_relay_t *next_relay;
1069         h2span_link_t *slink;
1070         dmsg_lnk_conn_t *lconn;
1071         dmsg_lnk_span_t *lspan;
1072         int count;
1073         int maxcount = 2;
1074         uint32_t lastdist = DMSG_SPAN_MAXDIST;
1075         uint32_t lastrnss = 0;
1076
1077         info.node = node;
1078         info.relay = NULL;
1079
1080         /*
1081          * Locate the first related relay for the node on this connection.
1082          * relay will be NULL if there were none.
1083          */
1084         RB_SCAN(h2span_relay_tree, &conn->tree,
1085                 dmsg_relay_scan_cmp, dmsg_relay_scan_callback, &info);
1086         relay = info.relay;
1087         info.relay = NULL;
1088         if (relay)
1089                 assert(relay->source_rt->any.link->node == node);
1090
1091         if (DMsgDebugOpt > 8)
1092                 fprintf(stderr, "relay scan for connection %p\n", conn);
1093
1094         /*
1095          * Iterate the node's links (received SPANs) in distance order,
1096          * lowest (best) dist first.
1097          *
1098          * PROPAGATE THE BEST LINKS OVER THE SPECIFIED CONNECTION.
1099          *
1100          * Track relays while iterating the best links and construct
1101          * missing relays when necessary.
1102          *
1103          * (If some prior better link was removed it would have also
1104          *  removed the relay, so the relay can only match exactly or
1105          *  be worse).
1106          */
1107         count = 0;
1108         RB_FOREACH(slink, h2span_link_tree, &node->tree) {
1109                 /*
1110                  * Increment count of successful relays.  This isn't
1111                  * quite accurate if we break out but nothing after
1112                  * the loop uses (count).
1113                  *
1114                  * If count exceeds the maximum number of relays we desire
1115                  * we normally want to break out.  However, in order to
1116                  * guarantee a symmetric path we have to continue if both
1117                  * (dist) and (rnss) continue to match.  Otherwise the SPAN
1118                  * propagation in the reverse direction may choose different
1119                  * routes and we will not have a symmetric path.
1120                  *
1121                  * NOTE: Spanning tree does not have to be symmetrical so
1122                  *       this code is not currently enabled.
1123                  */
1124                 if (++count >= maxcount) {
1125 #ifdef REQUIRE_SYMMETRICAL
1126                         if (lastdist != slink->dist || lastrnss != slink->rnss)
1127                                 break;
1128 #else
1129                         break;
1130 #endif
1131                         /* go beyond the nominal maximum desired relays */
1132                 }
1133
1134                 /*
1135                  * Match, relay already in-place, get the next
1136                  * relay to match against the next slink.
1137                  */
1138                 if (relay && relay->source_rt->any.link == slink) {
1139                         relay = RB_NEXT(h2span_relay_tree, &conn->tree, relay);
1140                         continue;
1141                 }
1142
1143                 /*
1144                  * We might want this SLINK, if it passes our filters.
1145                  *
1146                  * The spanning tree can cause closed loops so we have
1147                  * to limit slink->dist.
1148                  */
1149                 if (slink->dist > DMSG_SPAN_MAXDIST)
1150                         break;
1151
1152                 /*
1153                  * Don't bother transmitting a LNK_SPAN out the same
1154                  * connection it came in on.  Trivial optimization.
1155                  */
1156                 if (slink->state->iocom == conn->state->iocom)
1157                         break;
1158
1159                 /*
1160                  * NOTE ON FILTERS: The protocol spec allows non-requested
1161                  * SPANs to be transmitted, the other end is expected to
1162                  * leave their transactions open but otherwise ignore them.
1163                  *
1164                  * Don't bother transmitting if the remote connection
1165                  * is not accepting this SPAN's peer_type.
1166                  *
1167                  * pfs_mask is typically used so pure clients can filter
1168                  * out receiving SPANs for other pure clients.
1169                  */
1170                 lspan = &slink->state->msg->any.lnk_span;
1171                 lconn = &conn->state->msg->any.lnk_conn;
1172                 if (((1LLU << lspan->peer_type) & lconn->peer_mask) == 0)
1173                         break;
1174                 if (((1LLU << lspan->pfs_type) & lconn->pfs_mask) == 0)
1175                         break;
1176
1177                 /*
1178                  * Do not give pure clients visibility to other pure clients
1179                  */
1180                 if (lconn->pfs_type == DMSG_PFSTYPE_CLIENT &&
1181                     lspan->pfs_type == DMSG_PFSTYPE_CLIENT) {
1182                         break;
1183                 }
1184
1185                 /*
1186                  * Connection filter, if cluster uuid is not NULL it must
1187                  * match the span cluster uuid.  Only applies when the
1188                  * peer_type matches.
1189                  */
1190                 if (lspan->peer_type == lconn->peer_type &&
1191                     !uuid_is_nil(&lconn->pfs_clid, NULL) &&
1192                     uuid_compare(&slink->node->cls->pfs_clid,
1193                                  &lconn->pfs_clid, NULL)) {
1194                         break;
1195                 }
1196
1197                 /*
1198                  * Connection filter, if cluster label is not empty it must
1199                  * match the span cluster label.  Only applies when the
1200                  * peer_type matches.
1201                  */
1202                 if (lspan->peer_type == lconn->peer_type &&
1203                     lconn->cl_label[0] &&
1204                     strcmp(lconn->cl_label, slink->node->cls->cl_label)) {
1205                         break;
1206                 }
1207
1208                 /*
1209                  * NOTE! pfs_fsid differentiates nodes within the same cluster
1210                  *       so we obviously don't want to match those.  Similarly
1211                  *       for fs_label.
1212                  */
1213
1214                 /*
1215                  * Ok, we've accepted this SPAN for relaying.
1216                  */
1217                 assert(relay == NULL ||
1218                        relay->source_rt->any.link->node != slink->node ||
1219                        relay->source_rt->any.link->dist >= slink->dist);
1220                 relay = dmsg_generate_relay(conn, slink);
1221                 lastdist = slink->dist;
1222                 lastrnss = slink->rnss;
1223
1224                 /*
1225                  * Match (created new relay), get the next relay to
1226                  * match against the next slink.
1227                  */
1228                 relay = RB_NEXT(h2span_relay_tree, &conn->tree, relay);
1229         }
1230
1231         /*
1232          * Any remaining relay's belonging to this connection which match
1233          * the node are in excess of the current aggregate spanning state
1234          * and should be removed.
1235          */
1236         while (relay && relay->source_rt->any.link->node == node) {
1237                 next_relay = RB_NEXT(h2span_relay_tree, &conn->tree, relay);
1238                 dmsg_relay_delete(relay);
1239                 relay = next_relay;
1240         }
1241 }
1242
1243 /*
1244  * Helper function to generate missing relay.
1245  *
1246  * cluster_mtx must be held
1247  */
1248 static
1249 h2span_relay_t *
1250 dmsg_generate_relay(h2span_conn_t *conn, h2span_link_t *slink)
1251 {
1252         h2span_relay_t *relay;
1253         h2span_node_t *node;
1254         dmsg_msg_t *msg;
1255
1256         node = slink->node;
1257
1258         relay = dmsg_alloc(sizeof(*relay));
1259         relay->conn = conn;
1260         relay->source_rt = slink->state;
1261         /* relay->source_rt->any.link = slink; */
1262
1263         /*
1264          * NOTE: relay->target_rt->any.relay set to relay by alloc.
1265          */
1266         msg = dmsg_msg_alloc(&conn->state->iocom->circuit0,
1267                              0, DMSG_LNK_SPAN | DMSGF_CREATE,
1268                              dmsg_lnk_relay, relay);
1269         relay->target_rt = msg->state;
1270
1271         msg->any.lnk_span = slink->state->msg->any.lnk_span;
1272         msg->any.lnk_span.dist = slink->dist + 1;
1273         msg->any.lnk_span.rnss = slink->rnss + dmsg_rnss();
1274
1275         RB_INSERT(h2span_relay_tree, &conn->tree, relay);
1276         TAILQ_INSERT_TAIL(&slink->relayq, relay, entry);
1277
1278         dmsg_msg_write(msg);
1279
1280         return (relay);
1281 }
1282
1283 /*
1284  * Messages received on relay SPANs.  These are open transactions so it is
1285  * in fact possible for the other end to close the transaction.
1286  *
1287  * XXX MPRACE on state structure
1288  */
1289 static void
1290 dmsg_lnk_relay(dmsg_msg_t *msg)
1291 {
1292         dmsg_state_t *state = msg->state;
1293         h2span_relay_t *relay;
1294
1295         assert(msg->any.head.cmd & DMSGF_REPLY);
1296
1297         if (msg->any.head.cmd & DMSGF_DELETE) {
1298                 pthread_mutex_lock(&cluster_mtx);
1299                 if ((relay = state->any.relay) != NULL) {
1300                         dmsg_relay_delete(relay);
1301                 } else {
1302                         dmsg_state_reply(state, 0);
1303                 }
1304                 pthread_mutex_unlock(&cluster_mtx);
1305         }
1306 }
1307
1308
1309 static
1310 void
1311 dmsg_relay_delete(h2span_relay_t *relay)
1312 {
1313         fprintf(stderr,
1314                 "RELAY DELETE %p RELAY %p ON CLS=%p NODE=%p DIST=%d FD %d STATE %p\n",
1315                 relay->source_rt->any.link,
1316                 relay,
1317                 relay->source_rt->any.link->node->cls, relay->source_rt->any.link->node,
1318                 relay->source_rt->any.link->dist,
1319                 relay->conn->state->iocom->sock_fd, relay->target_rt);
1320
1321         RB_REMOVE(h2span_relay_tree, &relay->conn->tree, relay);
1322         TAILQ_REMOVE(&relay->source_rt->any.link->relayq, relay, entry);
1323
1324         if (relay->target_rt) {
1325                 relay->target_rt->any.relay = NULL;
1326                 dmsg_state_reply(relay->target_rt, 0);
1327                 /* state invalid after reply */
1328                 relay->target_rt = NULL;
1329         }
1330         relay->conn = NULL;
1331         relay->source_rt = NULL;
1332         dmsg_free(relay);
1333 }
1334
1335 static void *
1336 dmsg_volconf_thread(void *info)
1337 {
1338         h2span_media_config_t *conf = info;
1339
1340         pthread_mutex_lock(&cluster_mtx);
1341         while ((conf->ctl & H2CONFCTL_STOP) == 0) {
1342                 if (conf->ctl & H2CONFCTL_UPDATE) {
1343                         fprintf(stderr, "VOLCONF UPDATE\n");
1344                         conf->ctl &= ~H2CONFCTL_UPDATE;
1345                         if (bcmp(&conf->copy_run, &conf->copy_pend,
1346                                  sizeof(conf->copy_run)) == 0) {
1347                                 fprintf(stderr, "VOLCONF: no changes\n");
1348                                 continue;
1349                         }
1350                         /*
1351                          * XXX TODO - auto reconnect on lookup failure or
1352                          *              connect failure or stream failure.
1353                          */
1354
1355                         pthread_mutex_unlock(&cluster_mtx);
1356                         dmsg_volconf_stop(conf);
1357                         conf->copy_run = conf->copy_pend;
1358                         if (conf->copy_run.copyid != 0 &&
1359                             strncmp(conf->copy_run.path, "span:", 5) == 0) {
1360                                 dmsg_volconf_start(conf,
1361                                                       conf->copy_run.path + 5);
1362                         }
1363                         pthread_mutex_lock(&cluster_mtx);
1364                         fprintf(stderr, "VOLCONF UPDATE DONE state %d\n", conf->state);
1365                 }
1366                 if (conf->state == H2MC_CONNECT) {
1367                         dmsg_volconf_start(conf, conf->copy_run.path + 5);
1368                         pthread_mutex_unlock(&cluster_mtx);
1369                         sleep(5);
1370                         pthread_mutex_lock(&cluster_mtx);
1371                 } else {
1372                         pthread_cond_wait(&conf->cond, &cluster_mtx);
1373                 }
1374         }
1375         pthread_mutex_unlock(&cluster_mtx);
1376         dmsg_volconf_stop(conf);
1377         return(NULL);
1378 }
1379
1380 static
1381 void
1382 dmsg_volconf_stop(h2span_media_config_t *conf)
1383 {
1384         switch(conf->state) {
1385         case H2MC_STOPPED:
1386                 break;
1387         case H2MC_CONNECT:
1388                 conf->state = H2MC_STOPPED;
1389                 break;
1390         case H2MC_RUNNING:
1391                 shutdown(conf->fd, SHUT_WR);
1392                 pthread_join(conf->iocom_thread, NULL);
1393                 conf->iocom_thread = NULL;
1394                 break;
1395         }
1396 }
1397
1398 static
1399 void
1400 dmsg_volconf_start(h2span_media_config_t *conf, const char *hostname)
1401 {
1402         dmsg_master_service_info_t *info;
1403
1404         switch(conf->state) {
1405         case H2MC_STOPPED:
1406         case H2MC_CONNECT:
1407                 conf->fd = dmsg_connect(hostname);
1408                 if (conf->fd < 0) {
1409                         fprintf(stderr, "Unable to connect to %s\n", hostname);
1410                         conf->state = H2MC_CONNECT;
1411                 } else {
1412                         info = malloc(sizeof(*info));
1413                         bzero(info, sizeof(*info));
1414                         info->fd = conf->fd;
1415                         info->detachme = 0;
1416                         conf->state = H2MC_RUNNING;
1417                         pthread_create(&conf->iocom_thread, NULL,
1418                                        dmsg_master_service, info);
1419                 }
1420                 break;
1421         case H2MC_RUNNING:
1422                 break;
1423         }
1424 }
1425
1426 /************************************************************************
1427  *                      MESSAGE ROUTING AND SOURCE VALIDATION           *
1428  ************************************************************************/
1429
1430 int
1431 dmsg_circuit_relay(dmsg_msg_t *msg)
1432 {
1433         dmsg_iocom_t *iocom = msg->iocom;
1434         dmsg_circuit_t *circ;
1435         dmsg_circuit_t *peer;
1436         dmsg_circuit_t dummy;
1437         int error = 0;
1438
1439         /*
1440          * Relay occurs before any state processing, msg state should always
1441          * be NULL.
1442          */
1443         assert(msg->state == NULL);
1444
1445         /*
1446          * Lookup the circuit on the incoming iocom.
1447          */
1448         pthread_mutex_lock(&cluster_mtx);
1449
1450         dummy.msgid = msg->any.head.circuit;
1451         circ = RB_FIND(dmsg_circuit_tree, &iocom->circuit_tree, &dummy);
1452         assert(circ);
1453         peer = circ->peer;
1454
1455         msg->iocom = peer->iocom;
1456         msg->any.head.circuit = peer->msgid;
1457
1458         pthread_mutex_unlock(&cluster_mtx);
1459
1460         fprintf(stderr, "ROUTE MESSAGE VC %08x to %08x\n",
1461                 (uint32_t)circ->msgid, (uint32_t)peer->msgid);  /* brevity */
1462         dmsg_msg_write(msg);
1463         error = DMSG_IOQ_ERROR_ROUTED;
1464
1465         return error;
1466 }
1467
1468 /************************************************************************
1469  *                      ROUTER AND MESSAGING HANDLES                    *
1470  ************************************************************************
1471  *
1472  * Basically the idea here is to provide a stable data structure which
1473  * can be localized to the caller for higher level protocols to work with.
1474  * Depends on the context, these dmsg_handle's can be pooled by use-case
1475  * and remain persistent through a client (or mount point's) life.
1476  */
1477
1478 #if 0
1479 /*
1480  * Obtain a stable handle on a cluster given its uuid.  This ties directly
1481  * into the global cluster topology, creating the structure if necessary
1482  * (even if the uuid does not exist or does not exist yet), and preventing
1483  * the structure from getting ripped out from under us while we hold a
1484  * pointer to it.
1485  */
1486 h2span_cluster_t *
1487 dmsg_cluster_get(uuid_t *pfs_clid)
1488 {
1489         h2span_cluster_t dummy_cls;
1490         h2span_cluster_t *cls;
1491
1492         dummy_cls.pfs_clid = *pfs_clid;
1493         pthread_mutex_lock(&cluster_mtx);
1494         cls = RB_FIND(h2span_cluster_tree, &cluster_tree, &dummy_cls);
1495         if (cls)
1496                 ++cls->refs;
1497         pthread_mutex_unlock(&cluster_mtx);
1498         return (cls);
1499 }
1500
1501 void
1502 dmsg_cluster_put(h2span_cluster_t *cls)
1503 {
1504         pthread_mutex_lock(&cluster_mtx);
1505         assert(cls->refs > 0);
1506         --cls->refs;
1507         if (RB_EMPTY(&cls->tree) && cls->refs == 0) {
1508                 RB_REMOVE(h2span_cluster_tree,
1509                           &cluster_tree, cls);
1510                 dmsg_free(cls);
1511         }
1512         pthread_mutex_unlock(&cluster_mtx);
1513 }
1514
1515 /*
1516  * Obtain a stable handle to a specific cluster node given its uuid.
1517  * This handle does NOT lock in the route to the node and is typically
1518  * used as part of the dmsg_handle_*() API to obtain a set of
1519  * stable nodes.
1520  */
1521 h2span_node_t *
1522 dmsg_node_get(h2span_cluster_t *cls, uuid_t *pfs_fsid)
1523 {
1524 }
1525
1526 #endif
1527
1528 /*
1529  * Dumps the spanning tree
1530  */
1531 void
1532 dmsg_shell_tree(dmsg_circuit_t *circuit, char *cmdbuf __unused)
1533 {
1534         h2span_cluster_t *cls;
1535         h2span_node_t *node;
1536         h2span_link_t *slink;
1537         char *uustr = NULL;
1538
1539         pthread_mutex_lock(&cluster_mtx);
1540         RB_FOREACH(cls, h2span_cluster_tree, &cluster_tree) {
1541                 dmsg_circuit_printf(circuit, "Cluster %s %s (%s)\n",
1542                                   dmsg_peer_type_to_str(cls->peer_type),
1543                                   dmsg_uuid_to_str(&cls->pfs_clid, &uustr),
1544                                   cls->cl_label);
1545                 RB_FOREACH(node, h2span_node_tree, &cls->tree) {
1546                         dmsg_circuit_printf(circuit, "    Node %s %s (%s)\n",
1547                                 dmsg_pfs_type_to_str(node->pfs_type),
1548                                 dmsg_uuid_to_str(&node->pfs_fsid, &uustr),
1549                                 node->fs_label);
1550                         RB_FOREACH(slink, h2span_link_tree, &node->tree) {
1551                                 dmsg_circuit_printf(circuit,
1552                                             "\tLink dist=%d via %d\n",
1553                                             slink->dist,
1554                                             slink->state->iocom->sock_fd);
1555                         }
1556                 }
1557         }
1558         pthread_mutex_unlock(&cluster_mtx);
1559         if (uustr)
1560                 free(uustr);
1561 #if 0
1562         TAILQ_FOREACH(conn, &connq, entry) {
1563         }
1564 #endif
1565 }
1566
1567 /*
1568  * Random number sub-sort value to add to SPAN rnss fields on relay.
1569  * This allows us to differentiate spans with the same <dist> field
1570  * for relaying purposes.  We must normally limit the number of relays
1571  * for any given SPAN origination but we must also guarantee that a
1572  * symmetric reverse path exists, so we use the rnss field as a sub-sort
1573  * (since there can be thousands or millions if we only match on <dist>),
1574  * and if there STILL too many spans we go past the limit.
1575  */
1576 static
1577 uint32_t
1578 dmsg_rnss(void)
1579 {
1580         if (DMsgRNSS == 0) {
1581                 pthread_mutex_lock(&cluster_mtx);
1582                 while (DMsgRNSS == 0) {
1583                         srandomdev();
1584                         DMsgRNSS = random();
1585                 }
1586                 pthread_mutex_unlock(&cluster_mtx);
1587         }
1588         return(DMsgRNSS);
1589 }