| 182 | } |
| 183 | |
| 184 | func (c *Memcached) fetchKeysBatched(ctx context.Context, keys []string) (found []string, bufs [][]byte, missed []string) { |
| 185 | resultsCh := make(chan *result) |
| 186 | batchSize := c.cfg.BatchSize |
| 187 | |
| 188 | go func() { |
| 189 | for i, j := 0, 0; i < len(keys); i += batchSize { |
| 190 | batchKeys := keys[i:min(i+batchSize, len(keys))] |
| 191 | select { |
| 192 | case <-c.quit: |
| 193 | return |
| 194 | case c.inputCh <- &work{ |
| 195 | keys: batchKeys, |
| 196 | ctx: ctx, |
| 197 | resultCh: resultsCh, |
| 198 | batchID: j, |
| 199 | }: |
| 200 | } |
| 201 | j++ |
| 202 | } |
| 203 | }() |
| 204 | |
| 205 | // Read all values from this channel to avoid blocking upstream. |
| 206 | numResults := len(keys) / batchSize |
| 207 | if len(keys)%batchSize != 0 { |
| 208 | numResults++ |
| 209 | } |
| 210 | |
| 211 | // We need to order found by the input keys order. |
| 212 | results := make([]*result, numResults) |
| 213 | loopResults: |
| 214 | for i := 0; i < numResults; i++ { |
| 215 | select { |
| 216 | case <-c.quit: |
| 217 | break loopResults |
| 218 | case result := <-resultsCh: |
| 219 | results[result.batchID] = result |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | for _, result := range results { |
| 224 | if result == nil { |
| 225 | continue |
| 226 | } |
| 227 | found = append(found, result.found...) |
| 228 | bufs = append(bufs, result.bufs...) |
| 229 | missed = append(missed, result.missed...) |
| 230 | } |
| 231 | |
| 232 | return |
| 233 | } |
| 234 | |
| 235 | // Store stores the key in the cache. |
| 236 | func (c *Memcached) Store(ctx context.Context, keys []string, bufs [][]byte) { |