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

Code before:

/* Send a message to all the nodes that are part of the cluster having
 * a connected link.
 *
 * It is guaranteed that this function will never have as a side effect
 * some node->link to be invalidated, so it is safe to call this function
 * from event handlers that will do stuff with node links later. */
void clusterBroadcastMessage(clusterMsgSendBlock *msgblock) {
    dictIterator *di;
    dictEntry *de;

    uint16_t type = ntohs(msgblock->msg.type);
    int is_msg_type_publish = (type == CLUSTERMSG_TYPE_PUBLISH || type == CLUSTERMSG_TYPE_PUBLISHSHARD ||
                               type == CLUSTERMSG_TYPE_PUBLISH_LIGHT || type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT);
    int nodes_not_supporting_light_header = 0;

    di = dictGetSafeIterator(server.cluster->nodes);
    while ((de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);
        if (!nodeSupportsLightMsgHdr(node)) nodes_not_supporting_light_header++;
        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        if (!is_msg_type_publish) {
            clusterSendMessage(node->link, msgblock);
        } else {
            clusterSendPublishMessage(node, type, msgblock);
        }
    }
    dictReleaseIterator(di);

    if (nodes_not_supporting_light_header) {
        server.cluster->is_light_hdr_supported = 0;
    } else {
        server.cluster->is_light_hdr_supported = 1;
    }
}

/* Build the message header. hdr must point to a buffer at least
 * sizeof(clusterMsg) in bytes. */
static void clusterBuildMessageHdr(clusterMsg *hdr, int type, size_t msglen) {
    hdr->ver = htons(CLUSTER_PROTO_VER);
    hdr->sig[0] = 'R';
    hdr->sig[1] = 'C';
    hdr->sig[2] = 'm';
    hdr->sig[3] = 'b';
    hdr->type = htons(type);
    hdr->totlen = htonl(msglen);

    if (type == CLUSTERMSG_TYPE_PUBLISH_LIGHT || type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT) {
        clusterMsgLight *hdr_light = (clusterMsgLight *)(void *)hdr;
        hdr_light->notused1 = 0;
        return;

Review comment (refactor):
I don't like this branching and pubsub-specific logic in clusterBroadcastMessage, and the fact that it doesn't broadcast the message to all nodes anymore.

The broadcasting is just a loop over the nodes and a send to each. We can duplicate this loop in clusterPropagatePublish or in a helper function just for pubsub.

Code after:
 * the link to be invalidated, so it is safe to call this function
 * from event handlers that will do stuff with the same link later. */
void clusterSendMessage(clusterLink *link, clusterMsgSendBlock *msgblock) {
    if (!link) {
        return;
    }
    if (listLength(link->send_msg_queue) == 0 && msgblock->msg.totlen != 0)
        connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);

    listAddNodeTail(link->send_msg_queue, msgblock);
    msgblock->refcount++;

    /* Update memory tracking */
    link->send_msg_queue_mem += sizeof(listNode) + msgblock->totlen;
    server.stat_cluster_links_memory += sizeof(listNode);

    /* Populate sent messages stats. */
    uint16_t type = ntohs(msgblock->msg.type);
    if (type < CLUSTERMSG_TYPE_COUNT) server.cluster->stats_bus_messages_sent[type]++;
}

/* Send a message to all the nodes that are part of the cluster having
 * a connected link.
 *
 * It is guaranteed that this function will never have as a side effect
 * some node->link to be invalidated, so it is safe to call this function
 * from event handlers that will do stuff with node links later. */
void clusterBroadcastMessage(clusterMsgSendBlock *msgblock) {
    dictIterator *di;
    dictEntry *de;

    di = dictGetSafeIterator(server.cluster->nodes);
    while ((de = dictNext(di)) != NULL) {
        clusterNode *node = dictGetVal(de);

        if (node->flags & (CLUSTER_NODE_MYSELF | CLUSTER_NODE_HANDSHAKE)) continue;
        clusterSendMessage(node->link, msgblock);
    }
    dictReleaseIterator(di);
}

static void clusterBuildMessageHdrLight(clusterMsgLight *hdr, int type, size_t msglen) {
    hdr->ver = htons(CLUSTER_PROTO_VER);
    hdr->sig[0] = 'R';
    hdr->sig[1] = 'C';
    hdr->sig[2] = 'm';
    hdr->sig[3] = 'b';
    hdr->type = htons(type);
    hdr->notused1 = 0;
    hdr->notused2 = 0;
    hdr->totlen = htonl(msglen);