Commit e88d35e3 authored by ale's avatar ale

Memcache writes returns as soon as the first write succeeds

parent d5da26d7
Pipeline #1601 passed with stages
in 1 minute and 28 seconds
......@@ -90,25 +90,32 @@ func (m *memcacheReplicatedClient) writeAll(key string, value []byte, ttl int) e
Expiration: int32(ttl),
}
// Write to the memcache servers. At least one write must succeed.
caches := m.clients()
ch := make(chan error, len(caches))
defer close(ch)
for _, c := range caches {
// Write to all memcache servers. We return as soon as the first
// write succeeds, keeping a goroutine around to close the channel
// when all the writes are done. This leaks goroutines in case of
// network issues, so keep a short timeout.
ch := make(chan error, 1)
var wg sync.WaitGroup
for _, c := range m.clients() {
wg.Add(1)
go func(c *memcache.Client) {
ch <- c.Set(item)
wg.Done()
}(c)
}
var ok bool
for i := 0; i < len(caches); i++ {
if err := <-ch; err == nil {
ok = true
go func() {
wg.Wait()
close(ch)
}()
for err := range ch {
if err == nil {
return nil
}
}
if !ok {
return errors.New("all memcache servers failed")
}
return nil
return errors.New("all memcache servers failed")
}
func (m *memcacheReplicatedClient) readAny(key string) ([]byte, bool) {
......@@ -125,14 +132,13 @@ func (m *memcacheReplicatedClient) readAny(key string) ([]byte, bool) {
for _, c := range m.clients() {
wg.Add(1)
go func(c *memcache.Client) {
defer wg.Done()
if item, err := c.Get(key); err == nil {
select {
case ch <- item.Value:
default:
}
}
wg.Done()
}(c)
}
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment