Shard aware connections v2

[mmaslankaprv]

Made inter-node connections shard assignment deterministic.

Client side

Seastar in posix-stack for outgoing connections uses following port_assignment implementation:

        static thread_local std::uniform_int_distribution<uint16_t> u(49152/smp::count + 1, 65535/smp::count - 1);

        uint16_t port = attempts++ < 5 && requested_port == 0 && proto == transport::TCP ? u(random_engine) * smp::count + this_shard_id() : requested_port;

This algorithm makes outgoing port to be shard dependent.

Server side:

Incoming connections are assigned to shards using one of the following policies:

  enum class load_balancing_algorithm {
        // This algorithm tries to distribute all connections equally between all shards.
        // It does this by sending new connections to a shard with smallest amount of connections.
        connection_distribution,
        // This algorithm distributes new connection based on peer's tcp port. Destination shard
        // is calculated as a port number modulo number of shards. This allows a client to connect
        // to a specific shard in a server given it knows how many shards server has by choosing
        // src port number accordingly.
        port,
        // This algorithm distributes all new connections to listen_options::fixed_cpu shard only.
        fixed,
        default_ = connection_distribution
    };

By default connections are distributed using simple round robin algorithm. In worst case scenario it can lead to the situation in which all active connections are handled on single cores. (Multiple re-connections are required).

We can leverage the fact that all the nodes in the cluster have unique ids and seastar outgoing port assignment being shard dependent to assign connection to shard deterministically on target node.

It is important to note that proposed solution is simple but it will provide perfect connection-shard assignment only on cluster with uniform number of shards per node.

Implemented solution

• update client shard assignment to include current node id. I.e. add node _id to shard received from jump_consistent_hash assignment
• use load_balancing_algorithm::port on the server to assign connection to shard.

Changes since v1:

• using boost::hash_combine instead of simple modulo
• changed optional load_balancing_algorithm to default

[emaxerrno]

THe only question I have here, is whether or not this actually makes a difference in the node balancing.

We have 3 sources of entropy

1. total shards
2. source shard
3. destination node_id

and this adds the

4. source node_id

What does the distribution look like

[mmaslankaprv]

Here is an example for 3 nodes cluster 16 core each:

Here number represent shards:

0 -> 1 = outgoing: [{1, 12, 13}] incoming: [{1, 12, 13}]
0 -> 2 = outgoing: [{0, 15}] incoming: [{0, 15}]
1 -> 0 = outgoing: [{8, 10, 3}] incoming: [{8, 10, 3}]
1 -> 2 = outgoing: [{0, 8, 15}] incoming: [{0, 8, 15}]
2 -> 0 = outgoing: [{10, 5}] incoming: [{10, 5}]
2 -> 1 = outgoing: [{15, 6, 7}] incoming: [{15, 6, 7}]

This is connections count per shard:

Node 0 clients shards {1: 1, 12: 1, 13: 1, 0: 1, 15: 1}
Node 1 clients shards {8: 2, 10: 1, 3: 1, 0: 1, 15: 1}
Node 2 clients shards {10: 1, 5: 1, 15: 1, 6: 1, 7: 1}
Node 1 server shards {1: 1, 12: 1, 13: 1, 15: 1, 6: 1, 7: 1}
Node 2 server shards {0: 2, 15: 2, 8: 1}
Node 0 server shards {8: 1, 10: 2, 3: 1, 5: 1}

[emaxerrno]

I don't think i understand the output.

[emaxerrno]

what is 2-1? or 0->1 ?

[emaxerrno]

I guess i'm looking for a before and after

[mmaslankaprv]

0 -> 1 are connections from node 0 to node 1, outgoing are shards on node 0, incoming are shards on node 1

[mmaslankaprv]

Oh I didn't notice that you are interested in change that is made
Assignment before:

0 -> 1 = outgoing: [{11, 15, 7}] incoming: [{11, 15, 7}]
0 -> 2 = outgoing: [{1, 10, 5}] incoming: [{1, 10, 5}]
1 -> 0 = outgoing: [{3, 5}] incoming: [{3, 5}]
1 -> 2 = outgoing: [{1, 10, 5}] incoming: [{1, 10, 5}]
2 -> 0 = outgoing: [{3, 5}] incoming: [{3, 5}]
2 -> 1 = outgoing: [{11, 15, 7}] incoming: [{11, 15, 7}]

After:

0 -> 1 = outgoing: [{1, 12, 13}] incoming: [{1, 12, 13}]
0 -> 2 = outgoing: [{0, 15}] incoming: [{0, 15}]
1 -> 0 = outgoing: [{8, 10, 3}] incoming: [{8, 10, 3}]
1 -> 2 = outgoing: [{0, 8, 15}] incoming: [{0, 8, 15}]
2 -> 0 = outgoing: [{10, 5}] incoming: [{10, 5}]
2 -> 1 = outgoing: [{15, 6, 7}] incoming: [{15, 6, 7}]

Notice that before on all nodes connections targeting the same node were created on the same shards, they would end up being processed by the same shards on the target node

[mmaslankaprv]

This would cause all the connections comming to node being handled on the same shard

[emaxerrno]

makes sense.

[BenPope]

test approval

test dismissal

[BenPope]

test reapproval

test redismissal