Repository: valkey-io/valkey
File: src/cluster_legacy.c (C)

Code before:
 *
 * If `sharded` is 0:
 * For now we do very little, just propagating [S]PUBLISH messages across the whole
 * cluster. In the future we'll try to get smarter and avoiding propagating those
 * messages to hosts without receives for a given channel.
 * Otherwise:
 * Publish this message across the slot (primary/replica).
 * -------------------------------------------------------------------------- */
void clusterPropagatePublish(robj *channel, robj **messages, int count, int sharded) {
    /* Currently publish command does not support multiple payload. */
    serverAssert(count == 1);

    clusterMsgSendBlock *msgblock, *msgblock_light;
    int nodes_not_supporting_light_header = 0;

    msgblock_light = clusterCreatePublishLightMsgBlock(
        channel, messages, count, sharded ? CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT : CLUSTERMSG_TYPE_PUBLISH_LIGHT);

    if (server.cluster->all_nodes_are_known_to_support_light_hdr) {
        /* If the cluster is homogeneous broadcast to all nodes */
        if (sharded) {
            clusterShardBroadcastMessage(msgblock_light);
        } else {
            clusterBroadcastMessage(msgblock_light);
        }
    } else {
        msgblock = clusterCreatePublishMsgBlock(channel, messages[0],
                                                sharded ? CLUSTERMSG_TYPE_PUBLISHSHARD : CLUSTERMSG_TYPE_PUBLISH);
        DictOrListIterator iter;
        if (sharded) {
            list *nodes_for_slot = clusterGetNodesInMyShard(server.cluster->myself);
            serverAssert(nodes_for_slot != NULL);
            iterInitForList(&iter, nodes_for_slot);
        } else {
            iterInitForDict(&iter, server.cluster->nodes);
        }
        clusterNode *node;
        while ((node = iterNext(&iter)) != NULL) {
            /* If the cluster is not marked as homogeneous send respective message blocks to all nodes. */
            if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
            if (!nodeSupportsLightMsgHdr(node)) {
                clusterSendMessage(node->link, msgblock);
                nodes_not_supporting_light_header++;
                continue;
            }
            clusterSendMessage(node->link, msgblock_light);
        }
        iterReset(&iter);
        /* If the cluster is not heterogeneous anymore, update the all_nodes_are_known_to_support_light_hdr to 1 */
        if (!nodes_not_supporting_light_header) {
            server.cluster->all_nodes_are_known_to_support_light_hdr = 1;

Review comment (performance):
IMO this just adds extra complexity and duplicated logic. Broadcast is just a loop over the nodes anyway, so we can just do our own loop in all cases. We don't need `clusterShardBroadcastMessage` either.

The other benefits that we can remove `server.cluster->all_nodes_are_known_to_support_light_hdr` variable and the logic for keeping it updated. This is not a performance critical code path in any way, so it's better to keep it simple.

Lazy initialization of msgblock would be good though, to only create it when we see a node that needs it.

Code after:

/* This function sends a FAILOVER_AUTH_REQUEST message to every node in order to
 * see if there is the quorum for this replica instance to failover its failing
 * primary.
 *
 * Note that we send the failover request to everybody, primary and replica nodes,
 * but only the primaries are supposed to reply to our query. */
void clusterRequestFailoverAuth(void) {
    uint32_t msglen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST, msglen);

    clusterMsg *hdr = &msgblock->msg;
    /* If this is a manual failover, set the CLUSTERMSG_FLAG0_FORCEACK bit
     * in the header to communicate the nodes receiving the message that
     * they should authorized the failover even if the primary is working. */
    if (server.cluster->mf_end) hdr->mflags[0] |= CLUSTERMSG_FLAG0_FORCEACK;
    clusterBroadcastMessage(msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* Send a FAILOVER_AUTH_ACK message to the specified node. */
void clusterSendFailoverAuth(clusterNode *node) {
    if (!node->link) return;

    uint32_t msglen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK, msglen);

    clusterSendMessage(node->link, msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* Send a MFSTART message to the specified node. */
void clusterSendMFStart(clusterNode *node) {
    if (!node->link) return;

    uint32_t msglen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
    clusterMsgSendBlock *msgblock = createClusterMsgSendBlock(CLUSTERMSG_TYPE_MFSTART, msglen);

    clusterSendMessage(node->link, msgblock);
    clusterMsgSendBlockDecrRefCount(msgblock);
}

/* Vote for the node asking for our vote if there are the conditions. */
void clusterSendFailoverAuthIfNeeded(clusterNode *node, clusterMsg *request) {
    clusterNode *primary = node->replicaof;
    uint64_t requestCurrentEpoch = ntohu64(request->currentEpoch);
    uint64_t requestConfigEpoch = ntohu64(request->configEpoch);
    unsigned char *claimed_slots = request->myslots;
    int force_ack = request->mflags[0] & CLUSTERMSG_FLAG0_FORCEACK;
    int j;