diff --git a/.gitignore b/.gitignore index 02c604d7..b4a2d1c8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ _* *.out *~ +.idea diff --git a/memcache/conn.go b/memcache/conn.go new file mode 100644 index 00000000..1c6eb273 --- /dev/null +++ b/memcache/conn.go @@ -0,0 +1,37 @@ +package memcache + +import ( + "bufio" + "net" + "time" +) + +// conn is a connection to a server. +type conn struct { + nc net.Conn + rw *bufio.ReadWriter + addr net.Addr + c *Client +} + +// release returns this connection back to the client's free pool +func (cn *conn) release() { + cn.c.putFreeConn(cn.addr, cn) +} + +func (cn *conn) extendDeadline() { + cn.nc.SetDeadline(time.Now().Add(cn.c.netTimeout())) +} + +// condRelease releases this connection if the error pointed to by err +// is nil (not an error) or is only a protocol level error (e.g. a +// cache miss). The purpose is to not recycle TCP connections that +// are bad. +func (cn *conn) condRelease(err *error) { + if *err == nil || resumableError(*err) { + cn.release() + } else { + cn.nc.Close() + } +} + diff --git a/memcache/errors.go b/memcache/errors.go new file mode 100644 index 00000000..eb8e4932 --- /dev/null +++ b/memcache/errors.go @@ -0,0 +1,46 @@ +package memcache + +import ( + "errors" + "net" +) + +var ( + // ErrCacheMiss means that a Get failed because the item wasn't present. + ErrCacheMiss = errors.New("memcache: cache miss") + + // ErrCASConflict means that a CompareAndSwap call failed due to the + // cached value being modified between the Get and the CompareAndSwap. + // If the cached value was simply evicted rather than replaced, + // ErrNotStored will be returned instead. + ErrCASConflict = errors.New("memcache: compare-and-swap conflict") + + // ErrNotStored means that a conditional write operation (i.e. Add or + // CompareAndSwap) failed because the condition was not satisfied. + ErrNotStored = errors.New("memcache: item not stored") + + // ErrServerError means that a server error occurred. + ErrServerError = errors.New("memcache: server error") + + // ErrNoStats means that no statistics were available. + ErrNoStats = errors.New("memcache: no statistics available") + + // ErrMalformedKey is returned when an invalid key is used. + // Keys must be at maximum 250 bytes long and not + // contain whitespace or control characters. + ErrMalformedKey = errors.New("malformed: key is too long or contains invalid characters") + + // ErrNoServers is returned when no servers are configured or available. + ErrNoServers = errors.New("memcache: no servers configured or available") +) + +// ConnectTimeoutError is the error type used when it takes +// too long to connect to the desired host. This level of +// detail can generally be ignored. +type ConnectTimeoutError struct { + Addr net.Addr +} + +func (cte *ConnectTimeoutError) Error() string { + return "memcache: connect timeout to " + cte.Addr.String() +} \ No newline at end of file diff --git a/memcache/helpers.go b/memcache/helpers.go new file mode 100644 index 00000000..46402cde --- /dev/null +++ b/memcache/helpers.go @@ -0,0 +1,79 @@ +package memcache + +import ( + "bufio" + "bytes" + "fmt" +) + +// resumableError returns true if err is only a protocol-level cache error. +// This is used to determine whether or not a server connection should +// be re-used or not. If an error occurs, by default we don't reuse the +// connection, unless it was just a cache error. +func resumableError(err error) bool { + switch err { + case ErrCacheMiss, ErrCASConflict, ErrNotStored, ErrMalformedKey: + return true + } + return false +} + +func legalKey(key string) bool { + if len(key) > 250 { + return false + } + for i := 0; i < len(key); i++ { + if key[i] <= ' ' || key[i] == 0x7f { + return false + } + } + return true +} + +// scanGetResponseLine populates it and returns the declared size of the item. +// It does not read the bytes of the item. +func scanGetResponseLine(line []byte, it *Item) (size int, err error) { + pattern := "VALUE %s %d %d %d\r\n" + dest := []interface{}{&it.Key, &it.Flags, &size, &it.casid} + if bytes.Count(line, space) == 3 { + pattern = "VALUE %s %d %d\r\n" + dest = dest[:3] + } + n, err := fmt.Sscanf(string(line), pattern, dest...) + if err != nil || n != len(dest) { + return -1, fmt.Errorf("memcache: unexpected line in get response: %q", line) + } + return size, nil +} + +func writeReadLine(rw *bufio.ReadWriter, format string, args ...interface{}) ([]byte, error) { + _, err := fmt.Fprintf(rw, format, args...) + if err != nil { + return nil, err + } + if err := rw.Flush(); err != nil { + return nil, err + } + line, err := rw.ReadSlice('\n') + return line, err +} + +func writeExpectf(rw *bufio.ReadWriter, expect []byte, format string, args ...interface{}) error { + line, err := writeReadLine(rw, format, args...) + if err != nil { + return err + } + switch { + case bytes.Equal(line, resultOK): + return nil + case bytes.Equal(line, expect): + return nil + case bytes.Equal(line, resultNotStored): + return ErrNotStored + case bytes.Equal(line, resultExists): + return ErrCASConflict + case bytes.Equal(line, resultNotFound): + return ErrCacheMiss + } + return fmt.Errorf("memcache: unexpected response line: %q", string(line)) +} diff --git a/memcache/item.go b/memcache/item.go new file mode 100644 index 00000000..69ec3cae --- /dev/null +++ b/memcache/item.go @@ -0,0 +1,22 @@ +package memcache + +// Item is an item to be got or stored in a memcached server. +type Item struct { + // Key is the Item's key (250 bytes maximum). + Key string + + // Value is the Item's value. + Value []byte + + // Flags are server-opaque flags whose semantics are entirely + // up to the app. + Flags uint32 + + // Expiration is the cache expiration time, in seconds: either a relative + // time from now (up to 1 month), or an absolute Unix epoch time. + // Zero means the Item has no expiration time. + Expiration int32 + + // Compare and swap ID. + casid uint64 +} diff --git a/memcache/memcache.go b/memcache/memcache.go index 28eccf03..8b6f8e3b 100644 --- a/memcache/memcache.go +++ b/memcache/memcache.go @@ -20,6 +20,7 @@ package memcache import ( "bufio" "bytes" + "context" "errors" "fmt" "io" @@ -34,35 +35,6 @@ import ( // Similar to: // https://godoc.org/google.golang.org/appengine/memcache -var ( - // ErrCacheMiss means that a Get failed because the item wasn't present. - ErrCacheMiss = errors.New("memcache: cache miss") - - // ErrCASConflict means that a CompareAndSwap call failed due to the - // cached value being modified between the Get and the CompareAndSwap. - // If the cached value was simply evicted rather than replaced, - // ErrNotStored will be returned instead. - ErrCASConflict = errors.New("memcache: compare-and-swap conflict") - - // ErrNotStored means that a conditional write operation (i.e. Add or - // CompareAndSwap) failed because the condition was not satisfied. - ErrNotStored = errors.New("memcache: item not stored") - - // ErrServer means that a server error occurred. - ErrServerError = errors.New("memcache: server error") - - // ErrNoStats means that no statistics were available. - ErrNoStats = errors.New("memcache: no statistics available") - - // ErrMalformedKey is returned when an invalid key is used. - // Keys must be at maximum 250 bytes long and not - // contain whitespace or control characters. - ErrMalformedKey = errors.New("malformed: key is too long or contains invalid characters") - - // ErrNoServers is returned when no servers are configured or available. - ErrNoServers = errors.New("memcache: no servers configured or available") -) - const ( // DefaultTimeout is the default socket read/write timeout. DefaultTimeout = 100 * time.Millisecond @@ -74,30 +46,6 @@ const ( const buffered = 8 // arbitrary buffered channel size, for readability -// resumableError returns true if err is only a protocol-level cache error. -// This is used to determine whether or not a server connection should -// be re-used or not. If an error occurs, by default we don't reuse the -// connection, unless it was just a cache error. -func resumableError(err error) bool { - switch err { - case ErrCacheMiss, ErrCASConflict, ErrNotStored, ErrMalformedKey: - return true - } - return false -} - -func legalKey(key string) bool { - if len(key) > 250 { - return false - } - for i := 0; i < len(key); i++ { - if key[i] <= ' ' || key[i] == 0x7f { - return false - } - } - return true -} - var ( crlf = []byte("\r\n") space = []byte(" ") @@ -150,54 +98,233 @@ type Client struct { freeconn map[string][]*conn } -// Item is an item to be got or stored in a memcached server. -type Item struct { - // Key is the Item's key (250 bytes maximum). - Key string - - // Value is the Item's value. - Value []byte +// FlushAll flushes all from all servers +func (c *Client) FlushAll() error { + return c.FlushAllWithContext(context.Background()) +} - // Flags are server-opaque flags whose semantics are entirely - // up to the app. - Flags uint32 +// FlushAllWithContext flushes all from all servers +func (c *Client) FlushAllWithContext(ctx context.Context) error { + return c.selector.Each(ctx, c.flushAllFromAddr) +} - // Expiration is the cache expiration time, in seconds: either a relative - // time from now (up to 1 month), or an absolute Unix epoch time. - // Zero means the Item has no expiration time. - Expiration int32 +// Get gets the item for the given key. ErrCacheMiss is returned for a +// memcache cache miss. The key must be at most 250 bytes in length. +func (c *Client) Get(key string) (item *Item, err error) { + return c.GetWithContext(context.Background(), key) +} - // Compare and swap ID. - casid uint64 +// GetWithContext gets the item for the given key. ErrCacheMiss is returned for a +// memcache cache miss. The key must be at most 250 bytes in length. +func (c *Client) GetWithContext(ctx context.Context, key string) (item *Item, err error) { + err = c.withKeyAddr(ctx, key, func(addr net.Addr) error { + return c.getFromAddr(ctx, addr, []string{key}, func(it *Item) { item = it }) + }) + if err == nil && item == nil { + err = ErrCacheMiss + } + return } -// conn is a connection to a server. -type conn struct { - nc net.Conn - rw *bufio.ReadWriter - addr net.Addr - c *Client +// Touch updates the expiry for the given key. The seconds parameter is either +// a Unix timestamp or, if seconds is less than 1 month, the number of seconds +// into the future at which time the item will expire. Zero means the item has +// no expiration time. ErrCacheMiss is returned if the key is not in the cache. +// The key must be at most 250 bytes in length. +func (c *Client) Touch(key string, seconds int32) (err error) { + return c.TouchWithContext(context.Background(), key, seconds) } -// release returns this connection back to the client's free pool -func (cn *conn) release() { - cn.c.putFreeConn(cn.addr, cn) +// TouchWithContext updates the expiry for the given key. The seconds parameter +// is either a Unix timestamp or, if seconds is less than 1 month, the number of +// seconds into the future at which time the item will expire. Zero means the item +// has no expiration time. ErrCacheMiss is returned if the key is not in the cache. +// The key must be at most 250 bytes in length. +func (c *Client) TouchWithContext(ctx context.Context, key string, seconds int32) (err error) { + return c.withKeyAddr(ctx, key, func(addr net.Addr) error { + return c.touchFromAddr(ctx, addr, []string{key}, seconds) + }) } -func (cn *conn) extendDeadline() { - cn.nc.SetDeadline(time.Now().Add(cn.c.netTimeout())) +// GetMulti is a batch version of Get. The returned map from keys to +// items may have fewer elements than the input slice, due to memcache +// cache misses. Each key must be at most 250 bytes in length. +// If no error is returned, the returned map will also be non-nil. +func (c *Client) GetMulti(keys []string) (map[string]*Item, error) { + return c.GetMultiWithContext(context.Background(), keys) } -// condRelease releases this connection if the error pointed to by err -// is nil (not an error) or is only a protocol level error (e.g. a -// cache miss). The purpose is to not recycle TCP connections that -// are bad. -func (cn *conn) condRelease(err *error) { - if *err == nil || resumableError(*err) { - cn.release() - } else { - cn.nc.Close() +// GetMultiWithContext is a batch version of Get. The returned map from keys to +// items may have fewer elements than the input slice, due to memcache +// cache misses. Each key must be at most 250 bytes in length. +// If no error is returned, the returned map will also be non-nil. +func (c *Client) GetMultiWithContext(ctx context.Context, keys []string) (map[string]*Item, error) { + var lk sync.Mutex + m := make(map[string]*Item) + addItemToMap := func(it *Item) { + lk.Lock() + defer lk.Unlock() + m[it.Key] = it } + + keyMap := make(map[net.Addr][]string) + for _, key := range keys { + if !legalKey(key) { + return nil, ErrMalformedKey + } + addr, err := c.selector.PickServer(ctx, key) + if err != nil { + return nil, err + } + keyMap[addr] = append(keyMap[addr], key) + } + + ch := make(chan error, buffered) + for addr, keys := range keyMap { + go func(addr net.Addr, keys []string) { + ch <- c.getFromAddr(ctx, addr, keys, addItemToMap) + }(addr, keys) + } + + var err error + for _ = range keyMap { + if ge := <-ch; ge != nil { + err = ge + } + } + return m, err +} + +// Set writes the given item, unconditionally. +func (c *Client) Set(item *Item) error { + return c.SetWithContext(context.Background(), item) +} + +// SetWithContext writes the given item, unconditionally. +func (c *Client) SetWithContext(ctx context.Context, item *Item) error { + return c.onItem(ctx, item, (*Client).set) +} + +// Add writes the given item, if no value already exists for its +// key. ErrNotStored is returned if that condition is not met. +func (c *Client) Add(item *Item) error { + return c.AddWithContext(context.Background(), item) +} + +// AddWithContext writes the given item, if no value already exists for its +// key. ErrNotStored is returned if that condition is not met. +func (c *Client) AddWithContext(ctx context.Context, item *Item) error { + return c.onItem(ctx, item, (*Client).add) +} + +// Replace writes the given item, but only if the server *does* +// already hold data for this key +func (c *Client) Replace(item *Item) error { + return c.ReplaceWithContext(context.Background(), item) +} + +// ReplaceWithContext writes the given item, but only if the server *does* +// already hold data for this key +func (c *Client) ReplaceWithContext(ctx context.Context, item *Item) error { + return c.onItem(ctx, item, (*Client).replace) +} + +// CompareAndSwap writes the given item that was previously returned +// by Get, if the value was neither modified or evicted between the +// Get and the CompareAndSwap calls. The item's Key should not change +// between calls but all other item fields may differ. ErrCASConflict +// is returned if the value was modified in between the +// calls. ErrNotStored is returned if the value was evicted in between +// the calls. +func (c *Client) CompareAndSwap(item *Item) error { + return c.CompareAndSwapWithContext(context.Background(), item) +} + +// CompareAndSwapWithContext writes the given item that was previously returned +// by Get, if the value was neither modified or evicted between the +// Get and the CompareAndSwap calls. The item's Key should not change +// between calls but all other item fields may differ. ErrCASConflict +// is returned if the value was modified in between the +// calls. ErrNotStored is returned if the value was evicted in between +// the calls. +func (c *Client) CompareAndSwapWithContext(ctx context.Context, item *Item) error { + return c.onItem(ctx, item, (*Client).cas) +} + +// Delete deletes the item with the provided key. The error ErrCacheMiss is +// returned if the item didn't already exist in the cache. +func (c *Client) Delete(key string) error { + return c.DeleteWithContext(context.Background(), key) +} + +// DeleteWithContext deletes the item with the provided key. The error ErrCacheMiss is +// returned if the item didn't already exist in the cache. +func (c *Client) DeleteWithContext(ctx context.Context, key string) error { + return c.withKeyRw(ctx, key, func(rw *bufio.ReadWriter) error { + return writeExpectf(rw, resultDeleted, "delete %s\r\n", key) + }) +} + +// DeleteAll deletes all items in the cache. +func (c *Client) DeleteAll() error { + return c.DeleteAllWithContext(context.Background()) +} + +// DeleteAllWithContext deletes all items in the cache. +func (c *Client) DeleteAllWithContext(ctx context.Context) error { + return c.withKeyRw(ctx, "", func(rw *bufio.ReadWriter) error { + return writeExpectf(rw, resultDeleted, "flush_all\r\n") + }) +} + +// Ping checks all instances if they are alive. Returns error if any +// of them is down. +func (c *Client) Ping() error { + return c.PingWithContext(context.Background()) +} + +// PingWithContext checks all instances if they are alive. Returns error if any +// of them is down. +func (c *Client) PingWithContext(ctx context.Context) error { + return c.selector.Each(ctx, c.ping) +} + +// Increment atomically increments key by delta. The return value is +// the new value after being incremented or an error. If the value +// didn't exist in memcached the error is ErrCacheMiss. The value in +// memcached must be an decimal number, or an error will be returned. +// On 64-bit overflow, the new value wraps around. +func (c *Client) Increment(key string, delta uint64) (newValue uint64, err error) { + return c.IncrementWithContext(context.Background(), key, delta) +} + +// IncrementWithContext atomically increments key by delta. The return value is +// the new value after being incremented or an error. If the value +// didn't exist in memcached the error is ErrCacheMiss. The value in +// memcached must be an decimal number, or an error will be returned. +// On 64-bit overflow, the new value wraps around. +func (c *Client) IncrementWithContext(ctx context.Context, key string, delta uint64) (newValue uint64, err error) { + return c.incrDecr(ctx, "incr", key, delta) +} + +// Decrement atomically decrements key by delta. The return value is +// the new value after being decremented or an error. If the value +// didn't exist in memcached the error is ErrCacheMiss. The value in +// memcached must be an decimal number, or an error will be returned. +// On underflow, the new value is capped at zero and does not wrap +// around. +func (c *Client) Decrement(key string, delta uint64) (newValue uint64, err error) { + return c.DecrementWithContext(context.Background(), key, delta) +} + +// DecrementWithContext atomically decrements key by delta. The return value is +// the new value after being decremented or an error. If the value +// didn't exist in memcached the error is ErrCacheMiss. The value in +// memcached must be an decimal number, or an error will be returned. +// On underflow, the new value is capped at zero and does not wrap +// around. +func (c *Client) DecrementWithContext(ctx context.Context, key string, delta uint64) (newValue uint64, err error) { + return c.incrDecr(ctx, "decr", key, delta) } func (c *Client) putFreeConn(addr net.Addr, cn *conn) { @@ -243,17 +370,6 @@ func (c *Client) maxIdleConns() int { return DefaultMaxIdleConns } -// ConnectTimeoutError is the error type used when it takes -// too long to connect to the desired host. This level of -// detail can generally be ignored. -type ConnectTimeoutError struct { - Addr net.Addr -} - -func (cte *ConnectTimeoutError) Error() string { - return "memcache: connect timeout to " + cte.Addr.String() -} - func (c *Client) dial(addr net.Addr) (net.Conn, error) { nc, err := net.DialTimeout(addr.Network(), addr.String(), c.netTimeout()) if err == nil { @@ -267,7 +383,14 @@ func (c *Client) dial(addr net.Addr) (net.Conn, error) { return nil, err } -func (c *Client) getConn(addr net.Addr) (*conn, error) { +func (c *Client) getConn(ctx context.Context, addr net.Addr) (*conn, error) { + // Check if the context is expired. + select { + default: + case <-ctx.Done(): + return nil, ctx.Err() + } + cn, ok := c.getFreeConn(addr) if ok { cn.extendDeadline() @@ -287,12 +410,12 @@ func (c *Client) getConn(addr net.Addr) (*conn, error) { return cn, nil } -func (c *Client) onItem(item *Item, fn func(*Client, *bufio.ReadWriter, *Item) error) error { - addr, err := c.selector.PickServer(item.Key) +func (c *Client) onItem(ctx context.Context, item *Item, fn func(*Client, *bufio.ReadWriter, *Item) error) error { + addr, err := c.selector.PickServer(ctx, item.Key) if err != nil { return err } - cn, err := c.getConn(addr) + cn, err := c.getConn(ctx, addr) if err != nil { return err } @@ -303,46 +426,19 @@ func (c *Client) onItem(item *Item, fn func(*Client, *bufio.ReadWriter, *Item) e return nil } -func (c *Client) FlushAll() error { - return c.selector.Each(c.flushAllFromAddr) -} - -// Get gets the item for the given key. ErrCacheMiss is returned for a -// memcache cache miss. The key must be at most 250 bytes in length. -func (c *Client) Get(key string) (item *Item, err error) { - err = c.withKeyAddr(key, func(addr net.Addr) error { - return c.getFromAddr(addr, []string{key}, func(it *Item) { item = it }) - }) - if err == nil && item == nil { - err = ErrCacheMiss - } - return -} - -// Touch updates the expiry for the given key. The seconds parameter is either -// a Unix timestamp or, if seconds is less than 1 month, the number of seconds -// into the future at which time the item will expire. Zero means the item has -// no expiration time. ErrCacheMiss is returned if the key is not in the cache. -// The key must be at most 250 bytes in length. -func (c *Client) Touch(key string, seconds int32) (err error) { - return c.withKeyAddr(key, func(addr net.Addr) error { - return c.touchFromAddr(addr, []string{key}, seconds) - }) -} - -func (c *Client) withKeyAddr(key string, fn func(net.Addr) error) (err error) { +func (c *Client) withKeyAddr(ctx context.Context, key string, fn func(net.Addr) error) (err error) { if !legalKey(key) { return ErrMalformedKey } - addr, err := c.selector.PickServer(key) + addr, err := c.selector.PickServer(ctx, key) if err != nil { return err } return fn(addr) } -func (c *Client) withAddrRw(addr net.Addr, fn func(*bufio.ReadWriter) error) (err error) { - cn, err := c.getConn(addr) +func (c *Client) withAddrRw(ctx context.Context, addr net.Addr, fn func(*bufio.ReadWriter) error) (err error) { + cn, err := c.getConn(ctx, addr) if err != nil { return err } @@ -350,14 +446,14 @@ func (c *Client) withAddrRw(addr net.Addr, fn func(*bufio.ReadWriter) error) (er return fn(cn.rw) } -func (c *Client) withKeyRw(key string, fn func(*bufio.ReadWriter) error) error { - return c.withKeyAddr(key, func(addr net.Addr) error { - return c.withAddrRw(addr, fn) +func (c *Client) withKeyRw(ctx context.Context, key string, fn func(*bufio.ReadWriter) error) error { + return c.withKeyAddr(ctx, key, func(addr net.Addr) error { + return c.withAddrRw(ctx, addr, fn) }) } -func (c *Client) getFromAddr(addr net.Addr, keys []string, cb func(*Item)) error { - return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { +func (c *Client) getFromAddr(ctx context.Context, addr net.Addr, keys []string, cb func(*Item)) error { + return c.withAddrRw(ctx, addr, func(rw *bufio.ReadWriter) error { if _, err := fmt.Fprintf(rw, "gets %s\r\n", strings.Join(keys, " ")); err != nil { return err } @@ -372,8 +468,8 @@ func (c *Client) getFromAddr(addr net.Addr, keys []string, cb func(*Item)) error } // flushAllFromAddr send the flush_all command to the given addr -func (c *Client) flushAllFromAddr(addr net.Addr) error { - return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { +func (c *Client) flushAllFromAddr(ctx context.Context, addr net.Addr) error { + return c.withAddrRw(ctx, addr, func(rw *bufio.ReadWriter) error { if _, err := fmt.Fprintf(rw, "flush_all\r\n"); err != nil { return err } @@ -395,8 +491,8 @@ func (c *Client) flushAllFromAddr(addr net.Addr) error { } // ping sends the version command to the given addr -func (c *Client) ping(addr net.Addr) error { - return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { +func (c *Client) ping(ctx context.Context, addr net.Addr) error { + return c.withAddrRw(ctx, addr, func(rw *bufio.ReadWriter) error { if _, err := fmt.Fprintf(rw, "version\r\n"); err != nil { return err } @@ -418,8 +514,8 @@ func (c *Client) ping(addr net.Addr) error { }) } -func (c *Client) touchFromAddr(addr net.Addr, keys []string, expiration int32) error { - return c.withAddrRw(addr, func(rw *bufio.ReadWriter) error { +func (c *Client) touchFromAddr(ctx context.Context, addr net.Addr, keys []string, expiration int32) error { + return c.withAddrRw(ctx, addr, func(rw *bufio.ReadWriter) error { for _, key := range keys { if _, err := fmt.Fprintf(rw, "touch %s %d\r\n", key, expiration); err != nil { return err @@ -444,47 +540,6 @@ func (c *Client) touchFromAddr(addr net.Addr, keys []string, expiration int32) e }) } -// GetMulti is a batch version of Get. The returned map from keys to -// items may have fewer elements than the input slice, due to memcache -// cache misses. Each key must be at most 250 bytes in length. -// If no error is returned, the returned map will also be non-nil. -func (c *Client) GetMulti(keys []string) (map[string]*Item, error) { - var lk sync.Mutex - m := make(map[string]*Item) - addItemToMap := func(it *Item) { - lk.Lock() - defer lk.Unlock() - m[it.Key] = it - } - - keyMap := make(map[net.Addr][]string) - for _, key := range keys { - if !legalKey(key) { - return nil, ErrMalformedKey - } - addr, err := c.selector.PickServer(key) - if err != nil { - return nil, err - } - keyMap[addr] = append(keyMap[addr], key) - } - - ch := make(chan error, buffered) - for addr, keys := range keyMap { - go func(addr net.Addr, keys []string) { - ch <- c.getFromAddr(addr, keys, addItemToMap) - }(addr, keys) - } - - var err error - for _ = range keyMap { - if ge := <-ch; ge != nil { - err = ge - } - } - return m, err -} - // parseGetResponse reads a GET response from r and calls cb for each // read and allocated Item func parseGetResponse(r *bufio.Reader, cb func(*Item)) error { @@ -516,62 +571,18 @@ func parseGetResponse(r *bufio.Reader, cb func(*Item)) error { } } -// scanGetResponseLine populates it and returns the declared size of the item. -// It does not read the bytes of the item. -func scanGetResponseLine(line []byte, it *Item) (size int, err error) { - pattern := "VALUE %s %d %d %d\r\n" - dest := []interface{}{&it.Key, &it.Flags, &size, &it.casid} - if bytes.Count(line, space) == 3 { - pattern = "VALUE %s %d %d\r\n" - dest = dest[:3] - } - n, err := fmt.Sscanf(string(line), pattern, dest...) - if err != nil || n != len(dest) { - return -1, fmt.Errorf("memcache: unexpected line in get response: %q", line) - } - return size, nil -} - -// Set writes the given item, unconditionally. -func (c *Client) Set(item *Item) error { - return c.onItem(item, (*Client).set) -} - func (c *Client) set(rw *bufio.ReadWriter, item *Item) error { return c.populateOne(rw, "set", item) } -// Add writes the given item, if no value already exists for its -// key. ErrNotStored is returned if that condition is not met. -func (c *Client) Add(item *Item) error { - return c.onItem(item, (*Client).add) -} - func (c *Client) add(rw *bufio.ReadWriter, item *Item) error { return c.populateOne(rw, "add", item) } -// Replace writes the given item, but only if the server *does* -// already hold data for this key -func (c *Client) Replace(item *Item) error { - return c.onItem(item, (*Client).replace) -} - func (c *Client) replace(rw *bufio.ReadWriter, item *Item) error { return c.populateOne(rw, "replace", item) } -// CompareAndSwap writes the given item that was previously returned -// by Get, if the value was neither modified or evicted between the -// Get and the CompareAndSwap calls. The item's Key should not change -// between calls but all other item fields may differ. ErrCASConflict -// is returned if the value was modified in between the -// calls. ErrNotStored is returned if the value was evicted in between -// the calls. -func (c *Client) CompareAndSwap(item *Item) error { - return c.onItem(item, (*Client).cas) -} - func (c *Client) cas(rw *bufio.ReadWriter, item *Item) error { return c.populateOne(rw, "cas", item) } @@ -617,81 +628,9 @@ func (c *Client) populateOne(rw *bufio.ReadWriter, verb string, item *Item) erro return fmt.Errorf("memcache: unexpected response line from %q: %q", verb, string(line)) } -func writeReadLine(rw *bufio.ReadWriter, format string, args ...interface{}) ([]byte, error) { - _, err := fmt.Fprintf(rw, format, args...) - if err != nil { - return nil, err - } - if err := rw.Flush(); err != nil { - return nil, err - } - line, err := rw.ReadSlice('\n') - return line, err -} - -func writeExpectf(rw *bufio.ReadWriter, expect []byte, format string, args ...interface{}) error { - line, err := writeReadLine(rw, format, args...) - if err != nil { - return err - } - switch { - case bytes.Equal(line, resultOK): - return nil - case bytes.Equal(line, expect): - return nil - case bytes.Equal(line, resultNotStored): - return ErrNotStored - case bytes.Equal(line, resultExists): - return ErrCASConflict - case bytes.Equal(line, resultNotFound): - return ErrCacheMiss - } - return fmt.Errorf("memcache: unexpected response line: %q", string(line)) -} - -// Delete deletes the item with the provided key. The error ErrCacheMiss is -// returned if the item didn't already exist in the cache. -func (c *Client) Delete(key string) error { - return c.withKeyRw(key, func(rw *bufio.ReadWriter) error { - return writeExpectf(rw, resultDeleted, "delete %s\r\n", key) - }) -} - -// DeleteAll deletes all items in the cache. -func (c *Client) DeleteAll() error { - return c.withKeyRw("", func(rw *bufio.ReadWriter) error { - return writeExpectf(rw, resultDeleted, "flush_all\r\n") - }) -} - -// Ping checks all instances if they are alive. Returns error if any -// of them is down. -func (c *Client) Ping() error { - return c.selector.Each(c.ping) -} - -// Increment atomically increments key by delta. The return value is -// the new value after being incremented or an error. If the value -// didn't exist in memcached the error is ErrCacheMiss. The value in -// memcached must be an decimal number, or an error will be returned. -// On 64-bit overflow, the new value wraps around. -func (c *Client) Increment(key string, delta uint64) (newValue uint64, err error) { - return c.incrDecr("incr", key, delta) -} - -// Decrement atomically decrements key by delta. The return value is -// the new value after being decremented or an error. If the value -// didn't exist in memcached the error is ErrCacheMiss. The value in -// memcached must be an decimal number, or an error will be returned. -// On underflow, the new value is capped at zero and does not wrap -// around. -func (c *Client) Decrement(key string, delta uint64) (newValue uint64, err error) { - return c.incrDecr("decr", key, delta) -} - -func (c *Client) incrDecr(verb, key string, delta uint64) (uint64, error) { +func (c *Client) incrDecr(ctx context.Context, verb, key string, delta uint64) (uint64, error) { var val uint64 - err := c.withKeyRw(key, func(rw *bufio.ReadWriter) error { + err := c.withKeyRw(ctx, key, func(rw *bufio.ReadWriter) error { line, err := writeReadLine(rw, "%s %s %d\r\n", verb, key, delta) if err != nil { return err diff --git a/memcache/memcache_test.go b/memcache/memcache_test.go index b94a30ca..47a0d36e 100644 --- a/memcache/memcache_test.go +++ b/memcache/memcache_test.go @@ -19,6 +19,7 @@ package memcache import ( "bufio" + "context" "fmt" "io" "io/ioutil" @@ -277,9 +278,10 @@ func BenchmarkOnItem(b *testing.B) { } }() + ctx := context.Background() addr := fakeServer.Addr() c := New(addr.String()) - if _, err := c.getConn(addr); err != nil { + if _, err := c.getConn(ctx, addr); err != nil { b.Fatal("failed to initialize connection to fake server") } @@ -287,6 +289,6 @@ func BenchmarkOnItem(b *testing.B) { dummyFn := func(_ *Client, _ *bufio.ReadWriter, _ *Item) error { return nil } b.ResetTimer() for i := 0; i < b.N; i++ { - c.onItem(&item, dummyFn) + c.onItem(ctx, &item, dummyFn) } } diff --git a/memcache/selector.go b/memcache/selector.go index 89ad81e0..47ded8fa 100644 --- a/memcache/selector.go +++ b/memcache/selector.go @@ -17,6 +17,7 @@ limitations under the License. package memcache import ( + "context" "hash/crc32" "net" "strings" @@ -31,8 +32,8 @@ import ( type ServerSelector interface { // PickServer returns the server address that a given item // should be shared onto. - PickServer(key string) (net.Addr, error) - Each(func(net.Addr) error) error + PickServer(ctx context.Context, key string) (net.Addr, error) + Each(ctx context.Context, fn func(context.Context, net.Addr) error) error } // ServerList is a simple ServerSelector. Its zero value is usable. @@ -90,11 +91,11 @@ func (ss *ServerList) SetServers(servers ...string) error { } // Each iterates over each server calling the given function -func (ss *ServerList) Each(f func(net.Addr) error) error { +func (ss *ServerList) Each(ctx context.Context, fn func(context.Context, net.Addr) error) error { ss.mu.RLock() defer ss.mu.RUnlock() for _, a := range ss.addrs { - if err := f(a); nil != err { + if err := fn(ctx, a); nil != err { return err } } @@ -111,7 +112,7 @@ var keyBufPool = sync.Pool{ }, } -func (ss *ServerList) PickServer(key string) (net.Addr, error) { +func (ss *ServerList) PickServer(_ context.Context, key string) (net.Addr, error) { ss.mu.RLock() defer ss.mu.RUnlock() if len(ss.addrs) == 0 { diff --git a/memcache/selector_test.go b/memcache/selector_test.go index 65a2c4dd..4241d560 100644 --- a/memcache/selector_test.go +++ b/memcache/selector_test.go @@ -16,7 +16,10 @@ limitations under the License. package memcache -import "testing" +import ( + "context" + "testing" +) func BenchmarkPickServer(b *testing.B) { // at least two to avoid 0 and 1 special cases: @@ -32,7 +35,7 @@ func benchPickServer(b *testing.B, servers ...string) { var ss ServerList ss.SetServers(servers...) for i := 0; i < b.N; i++ { - if _, err := ss.PickServer("some key"); err != nil { + if _, err := ss.PickServer(context.Background(), "some key"); err != nil { b.Fatal(err) } }