sbin/hammer2/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
  36  *
  37  * This code supports the LNK_SPAN protocol.  Essentially all PFS's
  38  * clients and services rendezvous with the userland hammer2 service and
  39  * open LNK_SPAN transactions using a message header linkid of 0,
  40  * registering any PFS's they have connectivity to with us.
  41  *
  42  * --
  43  *
  44  * Each registration maintains its own open LNK_SPAN message transaction.
  45  * The SPANs are collected, aggregated, and retransmitted over available
  46  * connections through the maintainance of additional LNK_SPAN message
  47  * transactions on each link.
  48  *
  49  * The msgid for each active LNK_SPAN transaction we receive allows us to
  50  * send a message to the target PFS (which might be one of many belonging
  51  * to the same cluster), by specifying that msgid as the linkid in any
  52  * message we send to the target PFS.
  53  *
  54  * Similarly the msgid we allocate for any LNK_SPAN transaction we transmit
  55  * (and remember we will maintain multiple open LNK_SPAN transactions on
  56  * each connection representing the topology span, so every node sees every
  57  * other node as a separate open transaction).  So, similarly the msgid for
  58  * these active transactions which we initiated can be used by the other
  59  * end to route messages through us to another node, ultimately winding up
  60  * at the identified hammer2 PFS.  We have to adjust the spanid in the message
  61  * header at each hop to be representative of the outgoing LNK_SPAN we
  62  * are forwarding the message through.
  63  *
  64  * --
  65  *
  66  * If we were to retransmit every LNK_SPAN transaction we receive it would
  67  * create a huge mess, so we have to aggregate all received LNK_SPAN
  68  * transactions, sort them by the fsid (the cluster) and sub-sort them by
  69  * the pfs_fsid (individual nodes in the cluster), and only retransmit
  70  * (create outgoing transactions) for a subset of the nearest weighted-hops
  71  * for each individual node.
  72  *
  73  * The higher level protocols can then issue transactions to the nodes making
  74  * up a cluster to perform all actions required.
  75  *
  76  * --
  77  *
  78  * Since this is a large topology and a spanning tree protocol, links can
  79  * go up and down all the time.  Any time a link goes down its transaction
  80  * is closed.  The transaction has to be closed on both ends before we can
  81  * delete (and potentially reuse) the related spanid.  The LNK_SPAN being
  82  * closed may have been propagated out to other connections and those related
  83  * LNK_SPANs are also closed.  Ultimately all routes via the lost LNK_SPAN
  84  * go away, ultimately reaching all sources and all targets.
  85  *
  86  * Any messages in-transit using a route that goes away will be thrown away.
  87  * Open transactions are only tracked at the two end-points.  When a link
  88  * failure propagates to an end-point the related open transactions lose
  89  * their spanid and are automatically aborted.
  90  *
  91  * It is important to note that internal route nodes cannot just associate
  92  * a lost LNK_SPAN transaction with another route to the same destination.
  93  * Message transactions MUST be serialized and MUST be ordered.  All messages
  94  * for a transaction must run over the same route.  So if the route used by
  95  * an active transaction is lost, the related messages will be fully aborted
  96  * and the higher protocol levels will retry as appropriate.
  97  *
  98  * It is also important to note that several paths to the same PFS can be
  99  * propagated along the same link, which allows concurrency and even
 100  * redundancy over several network interfaces or via different routes through
 101  * the topology.  Any given transaction will use only a single route but busy
 102  * servers will often have hundreds of transactions active simultaniously,
 103  * so having multiple active paths through the network topology for A<->B
 104  * will improve performance.
 105  *
 106  * --
 107  *
 108  * Most protocols consolidate operations rather than simply relaying them.
 109  * This is particularly true of LEAF protocols (such as strict HAMMER2
 110  * clients), of which there can be millions connecting into the cluster at
 111  * various points.  The SPAN protocol is not used for these LEAF elements.
 112  *
 113  * Instead the primary service they connect to implements a proxy for the
 114  * client protocols so the core topology only has to propagate a couple of
 115  * LNK_SPANs and not millions.  LNK_SPANs are meant to be used only for
 116  * core master nodes and satellite slaves and cache nodes.
 117  */
 118
 119 #include "hammer2.h"
 120
 121 /*
 122  * RED-BLACK TREE DEFINITIONS
 123  *
 124  * We need to track
 125  *
 126  * (1) shared fsid's (a cluster).
 127  * (2) unique fsid's (a node in a cluster) <--- LNK_SPAN transactions.
 128  *
 129  * We need to aggegate all active LNK_SPANs, aggregate, and create our own
 130  * outgoing LNK_SPAN transactions on each of our connections representing
 131  * the aggregated state.
 132  *
 133  * h2span_connect       - list of iocom connections who wish to receive SPAN
 134  *                        propagation from other connections.  Might contain
 135  *                        a filter string.  Only iocom's with an open
 136  *                        LNK_CONN transactions are applicable for SPAN
 137  *                        propagation.
 138  *
 139  * h2span_relay         - List of links relayed (via SPAN).  Essentially
 140  *                        each relay structure represents a LNK_SPAN
 141  *                        transaction that we initiated, verses h2span_link
 142  *                        which is a LNK_SPAN transaction that we received.
 143  *
 144  * --
 145  *
 146  * h2span_cluster       - Organizes the shared fsid's.  One structure for
 147  *                        each cluster.
 148  *
 149  * h2span_node          - Organizes the nodes in a cluster.  One structure
 150  *                        for each unique {cluster,node}, aka {fsid, pfs_fsid}.
 151  *
 152  * h2span_link          - Organizes all incoming and outgoing LNK_SPAN message
 153  *                        transactions related to a node.
 154  *
 155  *                        One h2span_link structure for each incoming LNK_SPAN
 156  *                        transaction.  Links selected for propagation back
 157  *                        out are also where the outgoing LNK_SPAN messages
 158  *                        are indexed into (so we can propagate changes).
 159  *
 160  *                        The h2span_link's use a red-black tree to sort the
 161  *                        weighted hop metric for the incoming LNK_SPAN.  We
 162  *                        then select the top N for outgoing.  When the
 163  *                        topology changes the top N may also change and cause
 164  *                        new outgoing LNK_SPAN transactions to be opened
 165  *                        and less desireable ones to be closed, causing
 166  *                        transactional aborts within the message flow in
 167  *                        the process.
 168  *
 169  * Also note            - All outgoing LNK_SPAN message transactions are also
 170  *                        entered into a red-black tree for use by the routing
 171  *                        function.  This is handled by msg.c in the state
 172  *                        code, not here.
 173  */
 174
 175 struct h2span_link;
 176 struct h2span_relay;
 177 TAILQ_HEAD(h2span_connect_queue, h2span_connect);
 178 TAILQ_HEAD(h2span_relay_queue, h2span_relay);
 179
 180 RB_HEAD(h2span_cluster_tree, h2span_cluster);
 181 RB_HEAD(h2span_node_tree, h2span_node);
 182 RB_HEAD(h2span_link_tree, h2span_link);
 183 RB_HEAD(h2span_relay_tree, h2span_relay);
 184
 185 /*
 186  * Received LNK_CONN transaction enables SPAN protocol over connection.
 187  * (may contain filter).
 188  */
 189 struct h2span_connect {
 190         TAILQ_ENTRY(h2span_connect) entry;
 191         struct h2span_relay_tree tree;
 192         hammer2_state_t *state;
 193 };
 194
 195 /*
 196  * All received LNK_SPANs are organized by cluster (pfs_clid),
 197  * node (pfs_fsid), and link (received LNK_SPAN transaction).
 198  */
 199 struct h2span_cluster {
 200         RB_ENTRY(h2span_cluster) rbnode;
 201         struct h2span_node_tree tree;
 202         uuid_t  pfs_clid;               /* shared fsid */
 203 };
 204
 205 struct h2span_node  {
 206         RB_ENTRY(h2span_node) rbnode;
 207         struct h2span_link_tree tree;
 208         struct h2span_cluster *cls;
 209         uuid_t  pfs_fsid;               /* unique fsid */
 210 };
 211
 212 struct h2span_link {
 213         RB_ENTRY(h2span_link) rbnode;
 214         hammer2_state_t *state;         /* state<->link */
 215         struct h2span_node *node;       /* related node */
 216         int32_t weight;
 217         struct h2span_relay_queue relayq; /* relay out */
 218 };
 219
 220 /*
 221  * Any LNK_SPAN transactions we receive which are relayed out other
 222  * connections utilize this structure to track the LNK_SPAN transaction
 223  * we initiate on the other connections, if selected for relay.
 224  *
 225  * In many respects this is the core of the protocol... actually figuring
 226  * out what LNK_SPANs to relay.  The spanid used for relaying is the
 227  * address of the 'state' structure, which is why h2span_relay has to
 228  * be entered into a RB-TREE based at h2span_connect (so we can look
 229  * up the spanid to validate it).
 230  */
 231 struct h2span_relay {
 232         RB_ENTRY(h2span_relay) rbnode;  /* from h2span_connect */
 233         TAILQ_ENTRY(h2span_relay) entry; /* from link */
 234         struct h2span_connect *conn;
 235         hammer2_state_t *state;         /* transmitted LNK_SPAN */
 236         struct h2span_link *link;       /* received LNK_SPAN */
 237 };
 238
 239
 240 typedef struct h2span_connect h2span_connect_t;
 241 typedef struct h2span_cluster h2span_cluster_t;
 242 typedef struct h2span_node h2span_node_t;
 243 typedef struct h2span_link h2span_link_t;
 244 typedef struct h2span_relay h2span_relay_t;
 245
 246 static
 247 int
 248 h2span_cluster_cmp(h2span_cluster_t *cls1, h2span_cluster_t *cls2)
 249 {
 250         return(uuid_compare(&cls1->pfs_clid, &cls2->pfs_clid, NULL));
 251 }
 252
 253 static
 254 int
 255 h2span_node_cmp(h2span_node_t *node1, h2span_node_t *node2)
 256 {
 257         return(uuid_compare(&node1->pfs_fsid, &node2->pfs_fsid, NULL));
 258 }
 259
 260 static
 261 int
 262 h2span_link_cmp(h2span_link_t *link1, h2span_link_t *link2)
 263 {
 264         if (link1->weight < link2->weight)
 265                 return(-1);
 266         if (link1->weight > link2->weight)
 267                 return(1);
 268         if ((intptr_t)link1 < (intptr_t)link2)
 269                 return(-1);
 270         if ((intptr_t)link1 > (intptr_t)link2)
 271                 return(1);
 272         return(0);
 273 }
 274
 275 static
 276 int
 277 h2span_relay_cmp(h2span_relay_t *relay1, h2span_relay_t *relay2)
 278 {
 279         if ((intptr_t)relay1->state < (intptr_t)relay2->state)
 280                 return(-1);
 281         if ((intptr_t)relay1->state > (intptr_t)relay2->state)
 282                 return(1);
 283         return(0);
 284 }
 285
 286 RB_PROTOTYPE_STATIC(h2span_cluster_tree, h2span_cluster,
 287              rbnode, h2span_cluster_cmp);
 288 RB_PROTOTYPE_STATIC(h2span_node_tree, h2span_node,
 289              rbnode, h2span_node_cmp);
 290 RB_PROTOTYPE_STATIC(h2span_link_tree, h2span_link,
 291              rbnode, h2span_link_cmp);
 292 RB_PROTOTYPE_STATIC(h2span_relay_tree, h2span_relay,
 293              rbnode, h2span_relay_cmp);
 294
 295 RB_GENERATE_STATIC(h2span_cluster_tree, h2span_cluster,
 296              rbnode, h2span_cluster_cmp);
 297 RB_GENERATE_STATIC(h2span_node_tree, h2span_node,
 298              rbnode, h2span_node_cmp);
 299 RB_GENERATE_STATIC(h2span_link_tree, h2span_link,
 300              rbnode, h2span_link_cmp);
 301 RB_GENERATE_STATIC(h2span_relay_tree, h2span_relay,
 302              rbnode, h2span_relay_cmp);
 303
 304 /*
 305  * Global mutex protects cluster_tree lookups.
 306  */
 307 static pthread_mutex_t cluster_mtx;
 308 static struct h2span_cluster_tree cluster_tree = RB_INITIALIZER(cluster_tree);
 309 static struct h2span_connect_queue connq = TAILQ_HEAD_INITIALIZER(connq);
 310
 311 static void hammer2_lnk_span(hammer2_state_t *state, hammer2_msg_t *msg);
 312 static void hammer2_lnk_conn(hammer2_state_t *state, hammer2_msg_t *msg);
 313 static void hammer2_lnk_conn_update(h2span_connect_t *conn);
 314
 315 /*
 316  * Receive a HAMMER2_MSG_PROTO_LNK message.  This only called for
 317  * one-way and opening-transactions since state->func will be assigned
 318  * in all other cases.
 319  */
 320 void
 321 hammer2_msg_lnk(hammer2_iocom_t *iocom, hammer2_msg_t *msg)
 322 {
 323         switch(msg->any.head.cmd & HAMMER2_MSGF_BASECMDMASK) {
 324         case HAMMER2_LNK_CONN:
 325                 hammer2_lnk_conn(msg->state, msg);
 326                 break;
 327         case HAMMER2_LNK_SPAN:
 328                 hammer2_lnk_span(msg->state, msg);
 329                 break;
 330         default:
 331                 fprintf(stderr,
 332                         "MSG_PROTO_LNK: Unknown msg %08x\n", msg->any.head.cmd);
 333                 hammer2_msg_reply(iocom, msg, HAMMER2_MSG_ERR_UNKNOWN);
 334                 /* state invalid after reply */
 335                 break;
 336         }
 337 }
 338
 339 void
 340 hammer2_lnk_conn(hammer2_state_t *state, hammer2_msg_t *msg)
 341 {
 342         h2span_connect_t *conn;
 343         h2span_relay_t *relay;
 344         char *alloc = NULL;
 345
 346         pthread_mutex_lock(&cluster_mtx);
 347
 348         /*
 349          * On transaction start we allocate a new h2span_connect and
 350          * acknowledge the request, leaving the transaction open.
 351          */
 352         if (msg->any.head.cmd & HAMMER2_MSGF_CREATE) {
 353                 state->func = hammer2_lnk_conn;
 354
 355                 fprintf(stderr, "LNK_CONN(%016jx): %s/%s\n",
 356                         (intmax_t)msg->any.head.msgid,
 357                         hammer2_uuid_to_str(&msg->any.lnk_conn.pfs_clid,
 358                                             &alloc),
 359                         msg->any.lnk_conn.label);
 360                 free(alloc);
 361
 362                 conn = hammer2_alloc(sizeof(*conn));
 363
 364                 RB_INIT(&conn->tree);
 365                 conn->state = state;
 366                 state->any.conn = conn;
 367                 TAILQ_INSERT_TAIL(&connq, conn, entry);
 368
 369                 hammer2_lnk_conn_update(conn);
 370                 hammer2_msg_result(state->iocom, msg, 0);
 371         }
 372
 373         /*
 374          * On transaction terminate we clean out our h2span_connect
 375          * and acknowledge the request, closing the transaction.
 376          */
 377         if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
 378                 fprintf(stderr, "LNK_CONN: Terminated\n");
 379                 conn = state->any.conn;
 380                 assert(conn);
 381                 while ((relay = RB_ROOT(&conn->tree)) != NULL) {
 382                         RB_REMOVE(h2span_relay_tree, &conn->tree, relay);
 383                         TAILQ_REMOVE(&relay->link->relayq, relay, entry);
 384
 385                         if (relay->state) {
 386                                 relay->state->any.relay = NULL;
 387                                 hammer2_state_reply(relay->state, 0);
 388                                 /* state invalid after reply */
 389                                 relay->state = NULL;
 390                         }
 391                         relay->conn = NULL;
 392                         relay->link = NULL;
 393                         hammer2_free(relay);
 394                 }
 395
 396                 /*
 397                  * Clean out conn
 398                  */
 399                 conn->state = NULL;
 400                 msg->state->any.conn = NULL;
 401                 TAILQ_REMOVE(&connq, conn, entry);
 402                 hammer2_free(conn);
 403
 404                 hammer2_msg_reply(state->iocom, msg, 0);
 405                 /* state invalid after reply */
 406         }
 407         pthread_mutex_unlock(&cluster_mtx);
 408 }
 409
 410 void
 411 hammer2_lnk_span(hammer2_state_t *state, hammer2_msg_t *msg)
 412 {
 413         h2span_cluster_t dummy_cls;
 414         h2span_node_t dummy_node;
 415         h2span_cluster_t *cls;
 416         h2span_node_t *node;
 417         h2span_link_t *slink;
 418         h2span_relay_t *relay;
 419         char *alloc = NULL;
 420
 421         pthread_mutex_lock(&cluster_mtx);
 422
 423         /*
 424          * On transaction start we initialize the tracking infrastructure
 425          */
 426         if (msg->any.head.cmd & HAMMER2_MSGF_CREATE) {
 427                 state->func = hammer2_lnk_span;
 428
 429                 fprintf(stderr, "LNK_SPAN: %s/%s\n",
 430                         hammer2_uuid_to_str(&msg->any.lnk_span.pfs_clid,
 431                                             &alloc),
 432                         msg->any.lnk_span.label);
 433                 free(alloc);
 434
 435                 /*
 436                  * Find the cluster
 437                  */
 438                 dummy_cls.pfs_clid = msg->any.lnk_span.pfs_clid;
 439                 cls = RB_FIND(h2span_cluster_tree, &cluster_tree, &dummy_cls);
 440                 if (cls == NULL) {
 441                         cls = hammer2_alloc(sizeof(*cls));
 442                         cls->pfs_clid = msg->any.lnk_span.pfs_clid;
 443                         RB_INIT(&cls->tree);
 444                         RB_INSERT(h2span_cluster_tree, &cluster_tree, cls);
 445                 }
 446
 447                 /*
 448                  * Find the node
 449                  */
 450                 dummy_node.pfs_fsid = msg->any.lnk_span.pfs_fsid;
 451                 node = RB_FIND(h2span_node_tree, &cls->tree, &dummy_node);
 452                 if (node == NULL) {
 453                         node = hammer2_alloc(sizeof(*node));
 454                         node->pfs_fsid = msg->any.lnk_span.pfs_fsid;
 455                         node->cls = cls;
 456                         RB_INIT(&node->tree);
 457                         RB_INSERT(h2span_node_tree, &cls->tree, node);
 458                 }
 459
 460                 /*
 461                  * Create the link
 462                  */
 463                 assert(state->any.link == NULL);
 464                 slink = hammer2_alloc(sizeof(*slink));
 465                 slink->node = node;
 466                 slink->weight = msg->any.lnk_span.weight;
 467                 slink->state = state;
 468                 state->any.link = slink;
 469                 RB_INSERT(h2span_link_tree, &node->tree, slink);
 470
 471                 /*
 472                  * Now filter and relay the span to all other iocoms. XXX
 473                  */
 474         }
 475
 476         /*
 477          * On transaction terminate we remove the tracking infrastructure.
 478          */
 479         if (msg->any.head.cmd & HAMMER2_MSGF_DELETE) {
 480                 slink = state->any.link;
 481                 assert(slink != NULL);
 482                 node = slink->node;
 483                 cls = node->cls;
 484
 485                 /*
 486                  * Clean out all relays
 487                  */
 488                 while ((relay = TAILQ_FIRST(&slink->relayq)) != NULL) {
 489                         RB_REMOVE(h2span_relay_tree, &relay->conn->tree, relay);
 490                         TAILQ_REMOVE(&slink->relayq, relay, entry);
 491
 492                         if (relay->state) {
 493                                 relay->state->any.relay = NULL;
 494                                 hammer2_state_reply(relay->state, 0);
 495                                 /* state invalid after reply */
 496                                 relay->state = NULL;
 497                         }
 498                         relay->conn = NULL;
 499                         relay->link = NULL;
 500                         hammer2_free(relay);
 501                 }
 502
 503                 /*
 504                  * Clean out the topology
 505                  */
 506                 RB_REMOVE(h2span_link_tree, &node->tree, slink);
 507                 if (RB_EMPTY(&node->tree)) {
 508                         RB_REMOVE(h2span_node_tree, &cls->tree, node);
 509                         if (RB_EMPTY(&cls->tree)) {
 510                                 RB_REMOVE(h2span_cluster_tree,
 511                                           &cluster_tree, cls);
 512                                 hammer2_free(cls);
 513                         }
 514                         node->cls = NULL;
 515                         hammer2_free(node);
 516                 }
 517                 state->any.link = NULL;
 518                 slink->state = NULL;
 519                 slink->node = NULL;
 520                 hammer2_free(slink);
 521         }
 522
 523         pthread_mutex_unlock(&cluster_mtx);
 524 }
 525
 526 /*
 527  * Initiate/Update the relayed spans associated with a connection.
 528  */
 529 static void
 530 hammer2_lnk_conn_update(h2span_connect_t *conn __unused)
 531 {
 532 }