func (c *cluster) Score(keyMembers []common.KeyMember) (map[common.KeyMember]Presence, error) {
	m := map[int][]common.KeyMember{}
	for _, keyMember := range keyMembers {
		index := c.pool.Index(keyMember.Key)
		m[index] = append(m[index], keyMember)
	}
	type response struct {
		presenceMap map[common.KeyMember]Presence
		err         error
	}
	responseChan := make(chan response, len(m))
	for index, keyMembers := range m {
		go func(index int, keyMembers []common.KeyMember) {
			var presenceMap map[common.KeyMember]Presence
			err := c.pool.WithIndex(index, func(conn redis.Conn) (err error) {
				presenceMap, err = pipelineScore(conn, keyMembers)
				return
			})
			if err != nil {
				log.Printf("cluster: Score: %q: %s", c.pool.ID(index), err)
			}
			responseChan <- response{presenceMap, err}
		}(index, keyMembers)
	}
	presenceMap := map[common.KeyMember]Presence{}
	for i := 0; i < cap(responseChan); i++ {
		response := <-responseChan
		if response.err != nil {
			continue
		}
		for keyMember, presence := range response.presenceMap {
			presenceMap[keyMember] = presence
		}
	}
	return presenceMap, nil
}
Score returns the presence statistics of each passed key-member.
That is, whether the key-member exists in this cluster, if it's in
an insert set, and its score.
package cluster
import (
	"fmt"
	"log"
	"math/rand"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/garyburd/redigo/redis"
	"github.com/soundcloud/roshi/common"
	"github.com/soundcloud/roshi/instrumentation"
	"github.com/soundcloud/roshi/pool"
)
func init() {
	rand.Seed(time.Now().UnixNano())
}
type Cluster interface {
	Inserter
	Selecter
	Deleter
	Scorer
	Scanner
}
type Inserter interface {
	Insert(tuples []common.KeyScoreMember) error
}
type Selecter interface {
	SelectOffset(keys []string, offset, limit int) <-chan Element
	SelectRange(keys []string, start, stop common.Cursor, limit int) <-chan Element
}
type Deleter interface {
	Delete(tuples []common.KeyScoreMember) error
}
type Scorer interface {
	Score([]common.KeyMember) (map[common.KeyMember]Presence, error)
}
type Scanner interface {
	Keys(batchSize int) <-chan []string
}
const (
	insertSuffix = "+"
	deleteSuffix = "-"
)
var (
	genericScript = `
		local addKey = KEYS[1] .. 'ADDSUFFIX'
		local remKey = KEYS[1] .. 'REMSUFFIX'
		local maxSize = tonumber(ARGV[3])
		local atCapacity = tonumber(redis.call('ZCARD', addKey)) >= maxSize
		if atCapacity then
			local oldestTs = redis.call('ZRANGE', addKey, 0, 0, 'WITHSCORES')[2]
			if oldestTs and tonumber(ARGV[1]) < tonumber(oldestTs) then
				return -1
			end
		end
		local insertTs = redis.call('ZSCORE', KEYS[1] .. 'INSERTSUFFIX', ARGV[2])
		local deleteTs = redis.call('ZSCORE', KEYS[1] .. 'DELETESUFFIX', ARGV[2])
		if insertTs and tonumber(ARGV[1]) < tonumber(insertTs) then
			return -1
		elseif deleteTs and tonumber(ARGV[1]) <= tonumber(deleteTs) then
			return -1
		end
		redis.call('ZREM', remKey, ARGV[2])
		local n = redis.call('ZADD', addKey, ARGV[1], ARGV[2])
		redis.call('ZREMRANGEBYRANK', addKey, 0, -(maxSize+1))
		return n
	`
	insertScript *redis.Script
	deleteScript *redis.Script
)
func init() {
	genericScript = strings.NewReplacer(
		"INSERTSUFFIX", insertSuffix,
		"DELETESUFFIX", deleteSuffix,
	).Replace(genericScript)
	insertScript = redis.NewScript(1, strings.NewReplacer(
		"REMSUFFIX", deleteSuffix,  
		"ADDSUFFIX", insertSuffix,  
	).Replace(genericScript))
	deleteScript = redis.NewScript(1, strings.NewReplacer(
		"REMSUFFIX", insertSuffix,  
		"ADDSUFFIX", deleteSuffix,  
	).Replace(genericScript))
}
type cluster struct {
	pool            *pool.Pool
	maxSize         int
	selectGap       time.Duration
	instrumentation instrumentation.Instrumentation
}
func New(pool *pool.Pool, maxSize int, selectGap time.Duration, instr instrumentation.Instrumentation) Cluster {
	if instr == nil {
		instr = instrumentation.NopInstrumentation{}
	}
	return &cluster{
		pool:            pool,
		maxSize:         maxSize,
		selectGap:       selectGap,
		instrumentation: instr,
	}
}
func (c *cluster) Insert(keyScoreMembers []common.KeyScoreMember) error {
	m := map[int][]common.KeyScoreMember{}
	for _, tuple := range keyScoreMembers {
		index := c.pool.Index(tuple.Key)
		m[index] = append(m[index], tuple)
	}
	errChan := make(chan error, len(m))
	for index, keyScoreMembers := range m {
		go func(index int, keyScoreMembers []common.KeyScoreMember) {
			errChan <- c.pool.WithIndex(index, func(conn redis.Conn) error {
				return pipelineInsert(conn, keyScoreMembers, c.maxSize)
			})
		}(index, keyScoreMembers)
	}
	for _ = range m {
		if err := <-errChan; err != nil {
			return err
		}
	}
	return nil
}
func (c *cluster) SelectOffset(keys []string, offset, limit int) <-chan Element {
	return c.selectCommon(keys, func(conn redis.Conn, myKeys []string) (map[string][]common.KeyScoreMember, error) {
		return pipelineRange(conn, myKeys, offset, limit)
	})
}
func (c *cluster) SelectRange(keys []string, start, stop common.Cursor, limit int) <-chan Element {
	return c.selectCommon(keys, func(conn redis.Conn, myKeys []string) (map[string][]common.KeyScoreMember, error) {
		return pipelineRangeByScore(conn, myKeys, start, stop, limit)
	})
}
func (c *cluster) selectCommon(
	keys []string,
	fn func(redis.Conn, []string) (map[string][]common.KeyScoreMember, error),
) <-chan Element {
	out := make(chan Element)
	go func() {
		m := map[int][]string{}
		for _, key := range keys {
			index := c.pool.Index(key)
			m[index] = append(m[index], key)
		}
		wg := sync.WaitGroup{}
		wg.Add(len(m))
		delay := time.Duration(0)
		for index, keys := range m {
			go func(index int, keys []string, delay time.Duration) {
				defer wg.Done()
				time.Sleep(delay)
				var elements []Element
				var result map[string][]common.KeyScoreMember
				if err := c.pool.WithIndex(index, func(conn redis.Conn) (err error) {
					result, err = fn(conn, keys)
					return
				}); err != nil {
					elements = errorElements(keys, err)
				} else {
					elements = successElements(result)
				}
				for _, element := range elements {
					out <- element
				}
			}(index, keys, delay)
			delay += c.selectGap
		}
		wg.Wait()
		close(out)
	}()
	return out
}
func (c *cluster) Delete(keyScoreMembers []common.KeyScoreMember) error {
	m := map[int][]common.KeyScoreMember{}
	for _, keyScoreMember := range keyScoreMembers {
		index := c.pool.Index(keyScoreMember.Key)
		m[index] = append(m[index], keyScoreMember)
	}
	errChan := make(chan error, len(m))
	for index, keyScoreMembers := range m {
		go func(index int, keyScoreMembers []common.KeyScoreMember) {
			errChan <- c.pool.WithIndex(index, func(conn redis.Conn) error {
				return pipelineDelete(conn, keyScoreMembers, c.maxSize)
			})
		}(index, keyScoreMembers)
	}
	for _ = range m {
		if err := <-errChan; err != nil {
			return err
		}
	}
	return nil
}