Feature: support connection lifetime

[terut]

Hi, thank you for writing great library. It is working well for now !

Recently I noticed that request is unbalanced when its replica failover on a cloud service if the connection keeps. So I consider about connection lifetime to reconnect to redis endpoint because existing connection are not rerouted when a node reintroduced.

Here is the document about archtecture and connection balance manegement.

https://cloud.google.com/memorystore/docs/redis/about-read-replicas#architecture
https://cloud.google.com/memorystore/docs/redis/about-read-replicas#connection_balance_management

Other library like go-redis and redigo has a option such as MaxConnectionLifetime to drop expired connection from connection pool. So I tried to writing the implement of connection lifetime, but rueidis has auto pipelining is working background and detecting to finish all commands is difficult for me. Background worker for auto piplining keep going to received commands until error occurred, right?

I'm glad if I know the wire is safe to close like the following.

# mux.go

func (m *mux) _pipe(i uint16) (w wire, err error) {
	if w = m.wire[i].Load().(wire); w != m.init && !w.IsExpired() {
		return w, nil
	}
        if w.IsExpired() {
                 m.wire[i].CompareAndSwap(w, w.init)
                 ew := w
                 go func() {
                      <- ew.Done()
                      ew.close()
                 }
        }
      //...
}

Please tell me you have any solution about that. Thanks!

[rueian]

Hi @terut,

I believe it is safe to do .IsExpired() and swap the connection out, but that requires comparing the current time for each operation which is too costly.

Maybe scheduling a timer to close the pipe is a better way. The timer can be scheduled before this line

rueidis/pipe.go

Line 329 in bd17103

return p, nil

[terut]

Thank you for reply. That sounds good and it's better idea than mine. I tried that approach.

Probably is it like this?

// pipe.go
func _newPipe(connFn func() (net.Conn, error), option *ClientOption, r2ps, nobg bool) (p *pipe, err error) {
  // ...
  go func() {
    <-time.After(10 * time.Second)
    p.Close()
  }()
}

This approach sometimes occurred error such "rueidis client is closing or unable to connect redis" and stopped the process keeps going to call commands. Read-only command is no problem because it are retryable, but write command cannot retry. It maybe be harder than I thought.

I completely thought that p.Close stops background worker gracefully after receiving responses of all commands. p.Close() changes p.state to 2, that meaning is "stop to accept new request", right? However, the pipe seems to pick up from m.wire even if the state is 2.

Shoud I check p.state of wire here ?

rueidis/mux.go

Line 136 in c586d3c

if w = m.wire[i].Load().(wire); w != m.init {

[rueian]

Checking p.state there doesn’t help. We may need a special error and let writes can retry on it.

[terut]

Okay. Thank you for your advice. How is this?

// pipe.go
func _newPipe(connFn func() (net.Conn, error), option *ClientOption, r2ps, nobg bool) (p *pipe, err error) {
  // ...
  go func() {
    <-time.After(5 * time.Second) // depends on MaxConnLifetime option
    p.expired()
  }()
}

func (p *pipe) expired() {
  p.error.CompareAndSwap(nil, errExpired)
  p.Close()
}
var errExpired = &errs{error: ErrExpiredConnection}

// client.go
func (c *singleClient) Do(ctx context.Context, cmd Completed) (resp RedisResult) {
  attempts := 1
retry:
  resp = c.conn.Do(ctx, cmd)
  if resp.Error() == ErrExpiredConnection { // force retry
    goto retry
  }
  if c.retry && cmd.IsReadOnly() && c.isRetryable(resp.Error(), ctx) {
    // ...
  }
  //...
}

For now, I feel like it's working well on concurrent processing about both reads and writes.

[rueian]

Looks good to me generally. Some additional thoughts:

• Use p.lifeTm = time.AfterFunc(...) instead, because we need more fine-grained control over the timer:
• Stop p.lifeTm early if there is a network error or the connection is closed manually.
• For dedicated and blocking usages, we should stop p.lifeTm after acquiring the connection from the dpool or spool and reset it when putting the connection back.
• ErrExpiredConnection is a bit too long and it does not need to be public. errConnExpired could be better.

[rueian]

Read-only command is no problem because it are retryable, but write command cannot retry.

BTW, I think setting the MaxConnectionLifetime on the client connected to the GCP primary endpoint is unnecessary and not recommended.

[terut]

I agree with you about your additional point. I'll try it when creating pull request.

I'm not sure that GCP says refreshing connection to the primary endpoint is not recommended, but as you said I think it is unnecessary and meaningless because client connections to the primary endpoint disconnected during a failover.

but write command cannot retry

Sorry I made a mistake. It may be enough that reads only retry because the lifetime option needs when only using the replica endpoint.

[proost]

@rueian @terut
Hi! Tell me If i'm wrong.

Recently I noticed that request is unbalanced when its replica failover on a cloud service if the connection keeps. So I consider about connection lifetime to reconnect to redis endpoint because existing connection are not rerouted when a node reintroduced.

If i'm correct, We should fix three things first.

1. refresh quickly topology cache and connections if some nodes fail.
2. not connecting to failed nodes.
3. detecting new nodes.

I don't understand how MaxConnectionLifetime can solve this situation. If topology is changed, refreshing topology cache and connections are still needed. Pruning stale connections in asynchronous way can be a good to client side. But we do in _refresh.

In some respects, closing connections and swapping new connection act like ShardsRefreshInterval. Refreshing topology is not good enough?

[rueian]

Hi @proost, we are discussing the singleClient not the clusterClient.

The topology here is the two fixed endpoints GCP provides: 1) the primary endpoint and 2) the read endpoint. These endpoints are fixed, i.e. their DNS names or IP addresses will not change.

When the client starts TCP handshake with the read endpoint, GCP will pick one hidden replica to serve the connection and the connection will be kept alive if no error occurs. Therefore the only way to balance connections among new replicas behind the read endpoint is for the client to do reconnection.

[proost]

Oh, I'm really sorry there is misunderstanding.

Overall sounds good to me.

Below is my opinion that can be ignored.
Purpose "MaxConnectionLifetime" and "ShardsRefreshInterval" is almost same, Refreshing node cache and connections. How about renaming MaxConnectionLifetime to ClientRefreshInterval? If not only clusterClient but also singleClient need refresh option, it looks reasonable to me. After @terut work, Integrate ClientRefreshInterval and ShardsRefreshInterval(maybe deprecationShardsRefreshInterval is needed). If agree, It's okay to take charge of it to me.

[terut]

Hi, @proost. Thank you for your propose. I'm sorry, honestly I'm not familiar with cluster client. It's okay if also reasonable for @rueian.

[rueian]

No, ClientRefreshInterval looks confusing to me. MaxConnectionLifetime makes more sense but probably removing the Max, just ConnectionLifetime is better.