A119: Slicer LB Policy

[easwars]

No description provided.

[easwars]

@markdroth @ejona86 @dfawley
FYI: There are still some sections with TBDs (mostly around load reporting and xDS integration). I will have them filled out as early as I possibly can. But I believe that there is still enough in here for reviews to begin. Thanks.

[easwars]

I'm going to be making some structural changes to the spec based on our review today. Will ping here once it is ready for review. @markdroth @ejona86 @dfawley

@shivaspeaks @pawbhard Please expect some changes to the spec, but not wholesale changes. So, you could start looking at it from an implementation pov. Thanks.

[easwars]

The PR is ready to be looked at.
@markdroth @ejona86 @dfawley

[markdroth]

This looks really good!

I have a lot of comments here, but they're mostly fine details or clarifications -- the high-level shape of this is solid, and the doc does a good job of describing it.

Please let me know if you have any questions. Thanks!

[markdroth]

This is getting closer!

Please let me know if you have any questions. Thanks!

[markdroth]

It doesn't really matter whether it's defined here or in A121. I guess it should be in whichever of the two gRFCs is going to be merged last.

Please coordinate with @mbissa.

[markdroth]

Okay. @pawbhard, let's consider this for the C-core implementation.

[markdroth]

It looks like you're proposing a single field for the entire picker, but I don't think that's what we want. I think we need a separate field for each slice, not just a single field for the entire LB policy. Otherwise, if we trigger a connection attempt for one slice and then we get a pick request for a separate slice, we won't trigger a connection attempt for the second slice, and in that case I think we should.

[easwars]

Yes, I was thinking about it as I wrote it last night, and I thought I will go through it again this morning before you make a pass, but that didn't happen. I'll revisit this in a bit. Thanks.

[easwars]

Thought about this a little more. There seem to a few options here:

• Do we want the LB policy to swallow endpoint connectivity state transitions from IDLE -> CONNECTING?
  • If the LB policy swallows updates, we will have to mutate the below bit in the picker
• Do we want to maintain the connection_attempted bit on a per-slice or per-endpoint basis?

Let's examine the cases:

• No update swallowing + per-slice connection_attempted bit
  • At startup, RPC comes in, picks random endpoint, starts connection
  • Endpoint transitions from IDLE to CONNECTING
  • LB policy sends a new picker (this will have connection_attempted=true for all slices that contain the above endpoint)
  • RPC is retried and whatever endpoint it picks from the slice, it will see that there is an ongoing connection, so the RPC will get queued
  • At the same time, if another RPC comes in that hits a different slice (but one that contains the above endpoint), this will also get queued without a connection attempt
  • At the same time, if another RPC comes in that hits a different slice that does not contain the above endpoint, it will trigger a connection and get queued
  • Pros:
    • Number of connections scale with the number of RPCs
    • Picker does not mutate any state
  • Cons:
    • More retries because of more picker updates from the LB policy
• No update swallowing + per-endpoint connection_attempted bit
  • At startup, RPC comes in, picks random endpoint, starts connection
  • Endpoint transitions from IDLE to CONNECTING
  • LB policy sends a new picker (this will have connection_attempted=true for the above endpoint)
  • RPC is retried and will result in a connection until it picks an already connecting endpoint
    • Birthday paradox
  • At the same time, if another RPC comes in that hits a different slice (but one that contains the above endpoint)
    • this will also result in new connections until the already connecting endpoint is picked
  • At the same time, if another RPC comes in that hits a different slice that does not contain the above endpoint, it will trigger a connection and get queued
  • Pros:
    • Number of connections scale with the sq.root of the number of endpoints in the slice (according to birthday paradox, IIUC)
    • Picker does not mutate any state
  • Cons:
    • More retries because of more picker updates from the LB policy
• State swallowing + per-slice connection_attempted bit
  • It is not possible for the picker to update a per-slice bit atomically
  • So, this option is ruled out
• State swallowing + per-endpoint connection_attempted bit
  • At startup, RPC comes in, picks random endpoint, starts connection
  • Endpoint transitions from IDLE to CONNECTING
  • No picker update from the LB
  • Picker mutates atomic state in endpoint to set connection_attempted=true
  • No more retries until the endpoint moves to READY or TF
  • At the same time, another RPC hits another slice (that contains the above endpoint)
    • Will result in a connection unless it picks the above endpoint
  • At the same time, another RPC hits another slice (that does not contain the above endpoint)
    • Will result in the same flow as the first RPC (one connection attempt at max, and no retries)
  • Pros:
    • Connections scale with the number of RPCs
    • Reduced number of retries when the policy is starting out
  • Cons:
    • A little bit more complexity since we are going to have the picker mutate some state, but the LB policy is not going to have to access this state, so there is no synchronization required

I'm OK with going with either option 1 or option 4. What do you think? Can you see something that I missed?

[markdroth]

In ring_hash, the reason for having the hasEndpointInConnectingState bit in the picker is to minimize the number of endpoints on which we trigger potentially unnecessary connection attempts. Note that we can't strictly guarantee that this will never happen; we're really just minimizing the probability: even if we make that bit an atomic and have the picker mutate it, it's still possible that the LB policy has already returned a new picker to the channel between the time that the old picker triggers a connection attempt and updates its atomic and the time that the LB policy sees the request for the connection attempt, and that intermediate new picker could still trigger another connection attempt. Given that the atomic wouldn't provide any real guarantee, it seems better to not accept the overhead of using an atomic. But we do still want to minimize the number of unnecessary connections in ring_hash, so if we're not going to use an atomic, we definitely don't want the LB policy swallow the IDLE -> CONNECTING updates, because that would increase the window during which the picker might trigger duplicate updates. I think the same general logic applies here for those two specific questions.

Stepping back a bit, though, I think there are a couple of key differences between ring_hash and what we're trying to do here:

• In ring_hash, we have one giant ring to iterate over, which means that every single RPC may wind up iterating over every single endpoint, so the number of endpoints we could wind up connecting to could be quite large. But in slicer, that's not the case: we have endpoints divided into separate shards, so any given RPC will at worst iterate over only the set of endpoints in the chosen slice, thus bounding the number of potentially wasteful connections we might establish. (In fallback mode, we might iterate over all endpoints, but that should be a rare case, and we'll probably eventually wind up connecting to all endpoints in fallback mode anyway, no matter what we do.)
• The reason for lazily connecting in ring_hash is slightly different than in slicer. In ring_hash, each individual RPC may be for a different hash and thus wind up on a different endpoint, but we have no way of knowing a priori what the distribution of RPCs across endpoints is actually going to look like. It's quite possible that for a given client, it will be only a small subset of endpoints, in which case any connection we establish to any other endpoint is essentially wasted, and we want to avoid that. In contrast, the reason we are doing lazy connecting in slicer is that in the non-AFE use case, we expect a given client to access only a subset of slices -- but we can expect that if a client hits a given slice, then there will likely be additional traffic for that slice. As a result, I think it is reasonable to accept that once we hit a slice, we can be more proactive about establishing additional connections for endpoints in that slice.

Given those differences, I would suggest that we don't bother with the hasEndpointInConnectingState bit at all. I think the algorithm should just be the following:

def pick_from_assigned_endpoints(pool, pick_args):
  if not pool.endpoints:
    # This can be true only when the matching entry is in fallback mode
    # (because of not having any endpoints) *and* fallback is disabled.
    return PICK_FAILED

  # Pick a random endpoint within the slice.
  first_index = pick_random_index(pool.endpoints)

  # Loop through all endpoints in the slice starting at the above random index.
  requested_connection = False
  found_connecting = False
  for i in range(len(pool.endpoints)):
    index = (first_index + i) % len(pool.endpoints)
    endpoint = pool.endpoints[index]
    # If the endpoint is READY, use it.
    if endpoint.state == READY:
      return endpoint.picker.Pick(pick_args)
    # Record whether we see a CONNECTING endpoint.
    if endpoint.state == CONNECTING:
      found_connecting = True
    # If we see an IDLE endpoint and have not yet triggered a
    # connection attempt, do so now.
    if not requested_connection and endpoint.state == IDLE:
      endpoint.request_connection()
      requested_connection = True

  # If we did not find any READY endpoint but we either requested
  # a connection or saw a subchannel in CONNECTING state,
  # queue the pick.
  if requested_connection or found_connecting:
    return PICK_QUEUE

  # All endpoints are in TF.  Fail the pick.
  # We do this by delegating to the original endpoint, which will
  # yield a better error message than failing ourselves.
  return pool.endpoints[first_index].picker.Pick(pick_args)

This is basically the same as A76, with only two differences:

1. There is no hasEndpointInConnectingState bit in the picker.
2. It also queues if there is an endpoint already in CONNECTING state. (This actually looks like a bug in A76 -- I think it should check for CONNECTING too.)

As I was typing this, I also remembered that we had actually discussed this internally at one point in the past, and I dug up the notes from that discussion. It looks like our conclusion then was basically what I just suggested.

@ejona86 or @dfawley may also want to weigh in here.

[markdroth]

(As per discussion today, the CONNECTING thing isn't actually a bug in A76 -- the CONNECTING case is handled by the fact that the hasEndpointInConnectingState is set to true when the picker is created.)

[easwars]

Yes, this is definitely simpler and works. Updated this section.

[markdroth]

Meant to tag @pawbhard on this comment.

[markdroth]

It occurs to me that we should have a similar section describing what actions we take when we get a connectivity state update for an endpoint from the pick_first child policy.

In particular, we should consider the fact that connectivity state updates may be fairly common, and we may not want to have to completely rebuild the entire slice map when this happens. We should maybe figure out how to rebuild only the part that we actually need to.

Note that we already do something like this in ring_hash. We rebuild the ring only when we get an updated endpoint list. When we get a connectivity state update from an endpoint, we rebuild the picker, but in that case, the picker just grabs a new ref to the most recent ring, so it doesn't have to completely rebuild the ring.

[easwars]

Currently, we've erased the hostname information when building the SliceMap, because we've found the mapping from the assignment to the endpointMap, and therefore store endpoint state directly in the SliceMap. If we want to support a lightweight update as you suggest, we need to retain this information in the SliceMap.

This is what I came up with. Let me know what you think:

• Add a field in the SliceMap that maps hostname to endpoint index in the sliceMap.allEndpoints field. Let's call this new field name_to_index.
• Also retain the hostname in the endpointState struct
• Whenever the LB policy receives a state update from one of its children, it will make a copy of the existing SliceMap
  • Then, it will locate the endpoint corresponding to the hostname of the child sending the update
  • Update the endpoint state
  • Re-evaluate inFallback for all slices that refer to this endpoint index
  • All existing indices in the different slices (and fallbackPool) will continue to be valid
  • Create a new Picker with the newly created SliceMap

[markdroth]

I think we can do this without storing the hostnames in the SliceMap.

While thinking about this, it occurred to me that we can't use the EndpointState object directly in the SliceMap, because we don't want to share that object between the picker and the LB policy. The LB policy is going to update the child policy's connectivity state and picker, but we don't want existing picker objects to see those updates immediately; instead, we want the picker to contain its own immutable copy of that data, and we want to return a new picker to start actually using that new information to route RPCs. We also don't actually need the child policy itself in the picker, only in the LB policy itself.

So I think we need to start by splitting out a separate object for endpoint state within the picker, like this:

# Endpoint state used in the LB policy, not the picker.
# Stored in the LB policy in a map keyed by hostname.
class EndpointState:
  # Index within the NR update.
  index:   int

  # List of addresses for this endpoint.
  # Used when lazily creating child policy.
  addresses: list[Address]

  # Child policy.  Initially None, will be created lazily.
  child_lb: Balancer

  # Most recent connectivity state and picker.
  state:    ConnectivityState
  picker:   Picker

  def __init_(self, index, addresses):
    self.index = index
    self.addresses = addresses

  # Called to request a connection.
  # Lazily creates the LB policy as needed.
  def RequestConnection(self):
    if self.child_lb is None:
      # ...create child policy...
    self.child_lb.ExitIdle()

# Endpoint state used in the picker.
# Whenever we construct a new picker, the picker builds a list
# of PickerEndpoint objects from the LB policy's endpoint map.
# Each endpoint's index in this list will be the EndpointState.index
# field.
class PickerEndpoint:
  state:    ConnectivityState
  picker:   Picker
  # A ref to the EndpointState object.
  # We hop into the WorkSerializer to call this to request a connection.
  endpoint: EndpointState

Next, let's move the list of endpoints (the all_endpoints field) out of SliceMap and into a separate list of PickerEndpoint objects. In this approach, the SliceMap is essentially just a data structure constructed by binding a LogicalAssignment to a particular ordered list of endpoints -- in other words, it must be used in conjunction with a list of PickerEndpoint objects, but we can swap out which specific instance of the list we use it with, as long as the number and order of endpoints remains the same.

When the LB policy gets an NR update, we do something like this:

def OnResolverUpdate(self, endpoints):
  # Build endpoint map.
  self.endpoint_map = {}
  for i, endpoint in enumerate(endpoints):
    self.endpoint_map[endpoint.hostname] = EndpointState(i, endpoint.addresses)
  # Build new SliceMap.
  self.slice_map = self.BuildSliceMap()
  # Determine aggregated connectivity state and generate new picker.
  self.UpdateStateAndPicker()  

Building a new SliceMap would look like basically like you already have it, except that instead of populating the all_endpoints list, it will use EndpointState.index as the index for each endpoint. Whenever we get a new slicer assignment, we build a new SliceMap and then create a new picker.

When we construct a new picker, the picker will construct its own list of PickerEndpoint objects from the LB policy's endpoint map, like this:

class Picker:
  endpoint_list: list[PickerEndpoint]
  slice_map:     SliceMap

  def __init__(self, endpoint_map, slice_map):
    for endpoint in sorted(endpoint_map.values(), key=lambda e: e.index):
      self.endpoint_list.add(PickerEndpoint(endpoint, endpoint.state, endpoint.picker))
    self.slice_map = slice_map

When the picker does a pick, it can then use self.slice_map and self.endpoint_list together.

When the LB policy gets a state update from the child policy for an endpoint, it can update the state and picker in the EndpointState object. It can then generate a new picker without regenerating the SliceMap, because the state update has not changed the ordered list of endpoints. The new picker will wind up with the same ordered list, so the same SliceMap can be used with it.

What I've just described closely mirrors how ring_hash works in the C-core impl, so I think it should work just as well here.

[easwars]

Still thinking through this and trying to wrap my head around this. Will get back asap.

[easwars]

I was able to validate that this works. I had to make some tweaks to the existing data structures and the picker pseudo-code (since whether a sliceEntry is in fallback can change without the SliceMap changing). PTAL.

[nanassito]

Wouldn't the slicer refresh rate usually be significantly larger than the WRR refresh rate (default to 1m iirc). Because the slicer likely requires starting new instances, moving data around, etc.

[nanassito]

I'd also be fine having WRR built inside of slicer_experimental. but I could see how someone else would want leastReq or RoundRobin

[nanassito]

Here is an example use case we have at Reddit for this. A service caches a significant dataset in local memory, but this will soon be too big to fit on a single instance, so we were hoping to define shards/slices. Each instance gets assigned to a few slices, preload the data and start serving.

But also, not all instances have the same capacity due to a mix of multi-tenancy and mixed hardware. So if we randomly load balance we end up with instances that are severely overloaded while others have spare capacity. We've been using orca + wrr to address this issue with great success.

An alternative for us would be to define an xds cluster per slice and route based on a header but this feels like a poor man's version of this proposal.

[markdroth]

I don't think it does.

[easwars]

Confirmed with slicer folks and removed that check.

[markdroth]

You removed the comment but didn't actually remove the check. :)

[easwars]

🤦

Done

[markdroth]

I think we can do this without storing the hostnames in the SliceMap.

While thinking about this, it occurred to me that we can't use the EndpointState object directly in the SliceMap, because we don't want to share that object between the picker and the LB policy. The LB policy is going to update the child policy's connectivity state and picker, but we don't want existing picker objects to see those updates immediately; instead, we want the picker to contain its own immutable copy of that data, and we want to return a new picker to start actually using that new information to route RPCs. We also don't actually need the child policy itself in the picker, only in the LB policy itself.

So I think we need to start by splitting out a separate object for endpoint state within the picker, like this:

# Endpoint state used in the LB policy, not the picker.
# Stored in the LB policy in a map keyed by hostname.
class EndpointState:
  # Index within the NR update.
  index:   int

  # List of addresses for this endpoint.
  # Used when lazily creating child policy.
  addresses: list[Address]

  # Child policy.  Initially None, will be created lazily.
  child_lb: Balancer

  # Most recent connectivity state and picker.
  state:    ConnectivityState
  picker:   Picker

  def __init_(self, index, addresses):
    self.index = index
    self.addresses = addresses

  # Called to request a connection.
  # Lazily creates the LB policy as needed.
  def RequestConnection(self):
    if self.child_lb is None:
      # ...create child policy...
    self.child_lb.ExitIdle()

# Endpoint state used in the picker.
# Whenever we construct a new picker, the picker builds a list
# of PickerEndpoint objects from the LB policy's endpoint map.
# Each endpoint's index in this list will be the EndpointState.index
# field.
class PickerEndpoint:
  state:    ConnectivityState
  picker:   Picker
  # A ref to the EndpointState object.
  # We hop into the WorkSerializer to call this to request a connection.
  endpoint: EndpointState

Next, let's move the list of endpoints (the all_endpoints field) out of SliceMap and into a separate list of PickerEndpoint objects. In this approach, the SliceMap is essentially just a data structure constructed by binding a LogicalAssignment to a particular ordered list of endpoints -- in other words, it must be used in conjunction with a list of PickerEndpoint objects, but we can swap out which specific instance of the list we use it with, as long as the number and order of endpoints remains the same.

When the LB policy gets an NR update, we do something like this:

def OnResolverUpdate(self, endpoints):
  # Build endpoint map.
  self.endpoint_map = {}
  for i, endpoint in enumerate(endpoints):
    self.endpoint_map[endpoint.hostname] = EndpointState(i, endpoint.addresses)
  # Build new SliceMap.
  self.slice_map = self.BuildSliceMap()
  # Determine aggregated connectivity state and generate new picker.
  self.UpdateStateAndPicker()  

Building a new SliceMap would look like basically like you already have it, except that instead of populating the all_endpoints list, it will use EndpointState.index as the index for each endpoint. Whenever we get a new slicer assignment, we build a new SliceMap and then create a new picker.

When we construct a new picker, the picker will construct its own list of PickerEndpoint objects from the LB policy's endpoint map, like this:

class Picker:
  endpoint_list: list[PickerEndpoint]
  slice_map:     SliceMap

  def __init__(self, endpoint_map, slice_map):
    for endpoint in sorted(endpoint_map.values(), key=lambda e: e.index):
      self.endpoint_list.add(PickerEndpoint(endpoint, endpoint.state, endpoint.picker))
    self.slice_map = slice_map

When the picker does a pick, it can then use self.slice_map and self.endpoint_list together.

When the LB policy gets a state update from the child policy for an endpoint, it can update the state and picker in the EndpointState object. It can then generate a new picker without regenerating the SliceMap, because the state update has not changed the ordered list of endpoints. The new picker will wind up with the same ordered list, so the same SliceMap can be used with it.

What I've just described closely mirrors how ring_hash works in the C-core impl, so I think it should work just as well here.

[markdroth]

s/start_Key/start_key/

[easwars]

Done

[markdroth]

s/start_Key/start_key/

[easwars]

Done

[markdroth]

You removed the comment but didn't actually remove the check. :)

[markdroth]

Do we actually need this sentence here? It seems like this is actually unrelated to the backoff and stream handling behavior, which is what this section is talking about.

[easwars]

We actually don't. This is probably a relic from the past when I didnt have enough details in the fallback section.

[markdroth]

For readability reasons, I would remove the ConnectivityState. prefix before the state name throughout this pseudocode.

[easwars]

Done

[markdroth]

There shouldn't be a break here, because we want to keep going in case we find one that is already in CONNECTING state.

[easwars]

Yes, you are right. There is no break in the psuedo-code in ring_hash.

[markdroth]

s/Indefenitely/Indefinitely/

[easwars]

Done.

[markdroth]

Suggest converting the picker pseudo-code to python as well.

[easwars]

Done

[markdroth]

In ring_hash, the reason for having the hasEndpointInConnectingState bit in the picker is to minimize the number of endpoints on which we trigger potentially unnecessary connection attempts. Note that we can't strictly guarantee that this will never happen; we're really just minimizing the probability: even if we make that bit an atomic and have the picker mutate it, it's still possible that the LB policy has already returned a new picker to the channel between the time that the old picker triggers a connection attempt and updates its atomic and the time that the LB policy sees the request for the connection attempt, and that intermediate new picker could still trigger another connection attempt. Given that the atomic wouldn't provide any real guarantee, it seems better to not accept the overhead of using an atomic. But we do still want to minimize the number of unnecessary connections in ring_hash, so if we're not going to use an atomic, we definitely don't want the LB policy swallow the IDLE -> CONNECTING updates, because that would increase the window during which the picker might trigger duplicate updates. I think the same general logic applies here for those two specific questions.

Stepping back a bit, though, I think there are a couple of key differences between ring_hash and what we're trying to do here:

• In ring_hash, we have one giant ring to iterate over, which means that every single RPC may wind up iterating over every single endpoint, so the number of endpoints we could wind up connecting to could be quite large. But in slicer, that's not the case: we have endpoints divided into separate shards, so any given RPC will at worst iterate over only the set of endpoints in the chosen slice, thus bounding the number of potentially wasteful connections we might establish. (In fallback mode, we might iterate over all endpoints, but that should be a rare case, and we'll probably eventually wind up connecting to all endpoints in fallback mode anyway, no matter what we do.)
• The reason for lazily connecting in ring_hash is slightly different than in slicer. In ring_hash, each individual RPC may be for a different hash and thus wind up on a different endpoint, but we have no way of knowing a priori what the distribution of RPCs across endpoints is actually going to look like. It's quite possible that for a given client, it will be only a small subset of endpoints, in which case any connection we establish to any other endpoint is essentially wasted, and we want to avoid that. In contrast, the reason we are doing lazy connecting in slicer is that in the non-AFE use case, we expect a given client to access only a subset of slices -- but we can expect that if a client hits a given slice, then there will likely be additional traffic for that slice. As a result, I think it is reasonable to accept that once we hit a slice, we can be more proactive about establishing additional connections for endpoints in that slice.

Given those differences, I would suggest that we don't bother with the hasEndpointInConnectingState bit at all. I think the algorithm should just be the following:

def pick_from_assigned_endpoints(pool, pick_args):
  if not pool.endpoints:
    # This can be true only when the matching entry is in fallback mode
    # (because of not having any endpoints) *and* fallback is disabled.
    return PICK_FAILED

  # Pick a random endpoint within the slice.
  first_index = pick_random_index(pool.endpoints)

  # Loop through all endpoints in the slice starting at the above random index.
  requested_connection = False
  found_connecting = False
  for i in range(len(pool.endpoints)):
    index = (first_index + i) % len(pool.endpoints)
    endpoint = pool.endpoints[index]
    # If the endpoint is READY, use it.
    if endpoint.state == READY:
      return endpoint.picker.Pick(pick_args)
    # Record whether we see a CONNECTING endpoint.
    if endpoint.state == CONNECTING:
      found_connecting = True
    # If we see an IDLE endpoint and have not yet triggered a
    # connection attempt, do so now.
    if not requested_connection and endpoint.state == IDLE:
      endpoint.request_connection()
      requested_connection = True

  # If we did not find any READY endpoint but we either requested
  # a connection or saw a subchannel in CONNECTING state,
  # queue the pick.
  if requested_connection or found_connecting:
    return PICK_QUEUE

  # All endpoints are in TF.  Fail the pick.
  # We do this by delegating to the original endpoint, which will
  # yield a better error message than failing ourselves.
  return pool.endpoints[first_index].picker.Pick(pick_args)

This is basically the same as A76, with only two differences:

1. There is no hasEndpointInConnectingState bit in the picker.
2. It also queues if there is an endpoint already in CONNECTING state. (This actually looks like a bug in A76 -- I think it should check for CONNECTING too.)

As I was typing this, I also remembered that we had actually discussed this internally at one point in the past, and I dug up the notes from that discussion. It looks like our conclusion then was basically what I just suggested.

@ejona86 or @dfawley may also want to weigh in here.

[easwars]

@markdroth : I think I've addressed all your comments. Thanks for the detailed review.