Shard aware connections v1

[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.

[emaxerrno]

i don't get this. why not simply do another line of

boost::hash_combine(h, self())

You want a better bit distribution than modulo.

[mmaslankaprv]

good idea, will combine with node_id

[emaxerrno]

why not simply default it to port based? rather than std::optional

Also, i think we should. expose this via the configuration.hh

[mmaslankaprv]

We have to make it configurable as we do not want Kafka server to have shard aware connections as the client may use some random ports to connect to redpanda. This would cause uneven distribution of Kafka connections on shards.

[emaxerrno]

then, let's use the seastar default. My point was that we already have a default/optional value which is to use the default value. So let's default that so it is more explicit.