wtf.ext.memcache

134 """ 135 Memcache cluster proxy 136 137 :CVariables: 138 - `DEFAULT_GRACE_TIME`: Default grace time in seconds. See `__init__` for 139 further details. 140 - `DEFAULT_RETRY_TIME`: Default retry time in seconds. See `__init__` for 141 details. 142 - `DEFAULT_COMPRESS_THRESHOLD`: Default minimum size for compressing 143 values 144 - `DEFAULT_PADDED`: Default padding behavior 145 - `DEFAULT_SPLIT`: Default splitting behaviour 146 - `DEFAULT_LARGEST_SLAB`: Default maximum slab size 147 - `_TYPEMAP`: typemap 148 149 :IVariables: 150 - `_pools`: List of available pools 151 - `_weighted`: Weighted list of available pools 152 - `_grace_time`: Grace time for this instance 153 - `_retry_time`: Retry time for this instance 154 - `_compress_threshold`: Minimal size for compression 155 - `_padded`: Pad small values (<16 byte)? 156 - `_split`: Allow large value splitting? 157 - `_prefix`: Key prefix to use 158 - `_largest_slab`: Largest SLAB size 159 160 :Types: 161 - `DEFAULT_GRACE_TIME`: ``int`` 162 - `DEFAULT_RETRY_TIME`: ``int`` 163 - `DEFAULT_COMPRESS_THRESHOLD`: ``int`` 164 - `DEFAULT_PADDED`: ``bool`` 165 - `DEFAULT_SPLIT`: ``bool`` 166 - `DEFAULT_LARGEST_SLAB`: ``int`` 167 - `_TYPEMAP`: ``dict`` 168 - `_pools`: ``tuple`` 169 - `_weighted`: ``tuple`` 170 - `_grace_time`: ``int`` 171 - `_retry_time`: ``int`` 172 - `_compress_threshold`: ``int`` 173 - `_padded`: ``bool`` 174 - `_split`: ``bool`` 175 - `_prefix`: ``str`` 176 - `_largest_slab`: ``int`` 177 """ 178 DEFAULT_GRACE_TIME = 30 179 DEFAULT_RETRY_TIME = 60 180 DEFAULT_COMPRESS_THRESHOLD = 128 181 DEFAULT_PADDED = True 182 DEFAULT_SPLIT = True 183 DEFAULT_LARGEST_SLAB = 1048576 # POWER_BLOCK in slabs.c 184 _TYPEMAP = TYPEMAP 185

186 - def __init__(self, pools, prepare=None, grace_time=None, retry_time=None, 187 compress_threshold=None, padded=None, split=None, 188 prefix=None, largest_slab=None):

189 """ 190 Initialization 191 192 `grace_time` and `retry_time` describe the behaviour of 193 the dispatcher in case one or more dead pools. The algorithm works 194 as follows: 195 196 If a server is detected to be unreachable, it is marked dead. Now the 197 grace counter starts to run. Now if the server stays dead until 198 `grace_time` is reached requests for the server (read and write) are 199 discarded. This gives short memcache outages (like restarts) 200 a chance to recover without adding load to the other caches 201 of the cluster. However, when the grace time threshold is reached, 202 the server is considered completely dead and the requests are 203 dispatched to the other ones. The now completely-declared-dead 204 server will be retried every `retry_time` seconds from now on until 205 it's vivified again. 206 207 :Parameters: 208 - `pools`: List of memcache connection pools 209 (``[MemcacheConnectionPool, ...]``) 210 - `prepare`: Key preparation function 211 - `grace_time`: Grace time in seconds 212 - `retry_time`: Retry time in seconds 213 - `compress_threshold`: Minimum size for compression. If omitted or 214 ``None``, `DEFAULT_COMPRESS_THRESHOLD` is applied. 215 - `padded`: Pad small values (< 16 byte)? If omitted or ``None``, 216 `DEFAULT_PADDED` is applied 217 - `split`: Split large values? If omitted or ``None``, 218 `DEFAULT_SPLIT` is applied 219 - `prefix`: Prefix for keys. Empty by default 220 - `largest_slab`: Largest SLAB item size of the server, if omitted or 221 ``None``, `DEFAULT_LARGEST_SLAB` is applied. 222 223 :Types: 224 - `pools`: ``iterable`` 225 - `prepare`: ``callable`` 226 - `grace_time`: ``int`` 227 - `retry_time`: ``int`` 228 - `compress_threshold`: ``int`` 229 - `padded`: ``bool`` 230 - `split`: ``bool`` 231 - `prefix`: ``str`` 232 - `largest_slab`: ``int`` 233 """ 234 # Key config 235 if prepare is None: 236 if prefix: 237 prepare = lambda x: prefix + x 238 else: 239 prepare = lambda x: x 240 elif prefix: 241 _prepare = prepare 242 prepare = lambda x: prefix + _prepare(x) 243 self._prepare_key = prepare 244 245 # Value config 246 if compress_threshold is None: 247 compress_threshold = self.DEFAULT_COMPRESS_THRESHOLD 248 self._compress_threshold = compress_threshold 249 self._padded = padded 250 self._split = split 251 self._largest_slab = \ 252 [largest_slab, self.DEFAULT_LARGEST_SLAB][largest_slab is None] 253 254 # Pool config 255 self._pools = tuple(pools) 256 self._weighted = tuple(_it.chain(*[[pool] * pool.weight 257 for pool in self._pools])) 258 self._grace_time = int( 259 [grace_time, self.DEFAULT_GRACE_TIME][grace_time is None]) 260 self._retry_time = int( 261 [retry_time, self.DEFAULT_RETRY_TIME][retry_time is None])

262

263 - def delete(self, key, block_time=None, all_pools=False):

264 """ 265 Delete a key/value pair from the cache 266 267 :Parameters: 268 - `key`: The key to identify the item to delete 269 - `block_time`: Time to block add and replace requests for this key 270 in seconds. If omitted or ``None``, the blocking time is ``0``. 271 - `all_pools`: Issue delete to each pool? This may be useful to 272 enforce the deletion on backup pools, too. However, it won't 273 delete the key from currently dead pools. So, it might be not 274 that useful after all, but it's the best we can do from this 275 side of the ocean. 276 277 :Types: 278 - `key`: ``str`` 279 - `block_time`: ``int`` 280 - `all_pools`: ``bool`` 281 282 :return: Whether it was really deleted from the main pool (or the 283 current backup pool) of this key (i.e. whether it existed 284 before) 285 :rtype: ``bool`` 286 """ 287 result = False 288 key = self._prepare_key(key) 289 block_time = max(0, int(block_time or 0)) 290 mainpool = None 291 try: 292 conns = self._get_conn(key) 293 if conns is not None: 294 try: 295 conn = conns.keys()[0] 296 mainpool = conn.pool 297 conn.write("delete %s %s%s" % (key, block_time, CRLF)) 298 conn.flush() 299 line = self._error(conn.readline()) 300 finally: 301 conns = conns.keys() 302 while conns: 303 conns.pop().close() 304 result = line == "DELETED" 305 except _socket.error: 306 pass 307 308 if all_pools: 309 for pool in self._pools: 310 if pool == mainpool or pool.dead: 311 continue 312 try: 313 conn = pool.get_conn() 314 try: 315 conn.write("delete %s %s%s" % (key, block_time, CRLF)) 316 conn.flush() 317 conn.readline() # don't care about the response 318 finally: 319 conn.close() 320 except _socket.error: 321 pass 322 return result

323

324 - def get(self, *keys):

325 """ 326 Get a list of key/value pairs from the cache (if applicable) 327 328 The returned dict contains all pairs it could get. But keys maybe 329 missing or the dict might be completely empty (of course). 330 331 :Parameters: 332 - `keys`: The keys to fetch 333 334 :Types: 335 - `keys`: ``tuple`` 336 337 :return: The dict of key/value pairs 338 :rtype: ``dict`` 339 """ 340 # pylint: disable = R0912, R0915 341 342 result = {} 343 if not keys: 344 return result 345 keymap = dict((self._prepare_key(key), key) for key in keys) 346 try: 347 conns = self._get_conn(*keymap.keys()) 348 if not conns: 349 return result 350 conns = conns.items() 351 try: 352 while conns: 353 conn, keys = conns.pop() 354 try: 355 conn.write("get %s%s" % (" ".join(keys), CRLF)) 356 conn.flush() 357 while True: 358 line = self._error(conn.readline()) 359 if line == "END": 360 break 361 elif line.startswith("VALUE "): 362 _, key, flags, length = line.split() 363 flags, length = int(flags), int(length) 364 value = _stream.read_exact(conn, length) 365 if _stream.read_exact(conn, 2) != CRLF: 366 # sync error? 367 conn, _ = None, conn.destroy() 368 return {} 369 try: 370 result[keymap[key]] = \ 371 self._decode_value(flags, value) 372 except (TypeError, ValueError): 373 pass # wrong flags or something 374 except KeyError: 375 raise KeyError('%r, %s: %r' % ( 376 line, key, keymap 377 )) 378 except (SystemExit, KeyboardInterrupt): 379 raise 380 except: 381 import sys 382 e = sys.exc_info() 383 try: 384 msg = "%s:: %r, %s, %r" % ( 385 str(e[1]), line, flags, value 386 ) 387 e = (e[0], msg, e[2]) 388 finally: 389 try: 390 raise e[0], e[1], e[2] 391 finally: 392 del e 393 else: 394 # something else we don't know. Better close 395 # the connection. 396 conn, _ = None, conn.destroy() 397 return {} 398 finally: 399 if conn is not None: 400 conn.close() 401 finally: 402 while conns: 403 conns.pop()[0].close() 404 except _socket.error: 405 pass 406 return result

407

408 - def set(self, key, value, max_age):

409 """ 410 Set a key/value pair unconditionally 411 412 :Parameters: 413 - `key`: The key to store under 414 - `value`: The value to store (should be picklable) 415 - `max_age`: Maximum age in seconds 416 417 :Types: 418 - `key`: ``str`` 419 - `value`: any 420 - `max_age`: ``int`` 421 422 :return: Stored successfully? 423 :rtype: ``bool`` 424 """ 425 return self.store("set", key, value, max_age)

426

427 - def add(self, key, value, max_age):

428 """ 429 Set a key/value pair if the key does not exist yet 430 431 :Parameters: 432 - `key`: The key to store under 433 - `value`: The value to store (should be picklable) 434 - `max_age`: Maximum age in seconds 435 436 :Types: 437 - `key`: ``str`` 438 - `value`: any 439 - `max_age`: ``int`` 440 441 :return: Stored successfully? 442 :rtype: ``bool`` 443 """ 444 return self.store("add", key, value, max_age)

445

446 - def replace(self, key, value, max_age):

447 """ 448 Set a key/value pair only if the key does exist already 449 450 :Parameters: 451 - `key`: The key to store under 452 - `value`: The value to store (should be picklable) 453 - `max_age`: Maximum age in seconds 454 455 :Types: 456 - `key`: ``str`` 457 - `value`: any 458 - `max_age`: ``int`` 459 460 :return: Stored successfully? 461 :rtype: ``bool`` 462 """ 463 return self.store("replace", key, value, max_age)

464

465 - def store(self, method, key, value, max_age, compress=True):

466 """ 467 Store the value under the given key expiring now + expiry 468 469 :Parameters: 470 - `method`: Actual method to call (``set``, ``add`` or ``replace``) 471 - `key`: The key to store under 472 - `value`: The value to store (should be picklable) 473 - `max_age`: Max age of the entry in seconds 474 - `compress`: Compress the value? 475 476 :Types: 477 - `method`: ``str`` 478 - `key`: ``str`` 479 - `value`: any 480 - `max_age`: ``int`` 481 - `compress`: ``bool`` 482 483 :return: Stored successfully? 484 :rtype: ``bool`` 485 """ 486 conn = None 487 key = self._prepare_key(key) 488 try: 489 try: 490 flags, value = self._encode_value( 491 key, value, max_age, compress 492 ) 493 conns = self._get_conn(key) 494 if not conns: 495 return False 496 conn, conns = conns.keys()[0], None 497 expiry = int(_time.time()) + max_age - conn.pool.timediff 498 cmd = "%(cmd)s %(key)s %(flags)s %(exp)s %(len)s%(nl)s" % \ 499 dict( 500 cmd=method, 501 key=key, 502 flags=flags, 503 exp=expiry, 504 len=len(value), 505 nl=CRLF, 506 ) 507 conn.write(cmd) 508 conn.write(value) 509 conn.write(CRLF) 510 conn.flush() 511 return self._error(conn.readline()) == "STORED" 512 except _socket.error: 513 conn, _ = None, conn.destroy() 514 return False 515 finally: 516 if conn is not None: 517 conn.close()

518

519 - def _error(self, line):

520 """ 521 Convert a response line into an error or pass it through 522 523 :Parameters: 524 - `line`: The response line to inspect 525 526 :Types: 527 - `line`: ``str`` 528 529 :return: The stripped response line 530 :rtype: ``str`` 531 532 :Exceptions: 533 - `CommandError`: Command error 534 - `ClientError`: Client error 535 - `ServerError`: Server error 536 """ 537 line = line.strip() 538 if "ERROR" in line: 539 if line == "ERROR": 540 raise CommandError() 541 elif line.startswith("CLIENT_ERROR "): 542 raise ClientError(line[13:]) 543 elif line.startswith("SERVER_ERROR "): 544 raise ServerError(line[13:]) 545 else: 546 pos = line.find(' ') 547 if pos > 0: 548 raise UnknownError(line[pos + 1:]) 549 raise UnknownError() 550 return line

551

552 - def _encode_value(self, key, value, max_age, compress):

553 """ 554 Encode a value for the memcache 555 556 :Parameters: 557 - `key`: The store key 558 - `value`: The value to encode 559 - `max_age`: Maxc age of this item 560 - `compress`: Allow value compression? 561 562 :Types: 563 - `key`: ``str`` 564 - `value`: any 565 - `max_age`: ``int`` 566 - `compress`: ``bool`` 567 568 :return: The flags and the encoded value (``(int, str)``) 569 :rtype: ``tuple`` 570 """ 571 flags, vtype = 0, type(value) 572 for type_id, (kind, encoder, _) in self._TYPEMAP.iteritems(): 573 if type_id == 0: 574 continue 575 if vtype is kind: 576 flags, value = type_id, encoder(value) 577 break 578 else: 579 value = self._TYPEMAP[0][1](value) 580 if compress and \ 581 len(value) >= self._compress_threshold and _zlib is not None: 582 value = _zlib.compress(value, 9) 583 flags |= FLAG_COMPRESSED 584 if self._padded and len(value) < 16: 585 value = value + "\0" * 16 586 flags |= FLAG_PADDED 587 if (len(value) + len(key) + 100) > self._largest_slab: 588 tpl = "split:%s:%s-%%s" % ( 589 _md5.md5(_os.urandom(20)).hexdigest(), key 590 ) 591 blocklen = self._largest_slab - len(tpl) - 100 592 skeys, idx = [], 0 593 while value: 594 skey = tpl % idx 595 skeys.append(skey) 596 idx += 1 597 buf, value = value[:blocklen], value[blocklen:] 598 self.store("set", skey, buf, max_age, compress=False) 599 flags |= FLAG_SPLIT 600 value = ' '.join(skeys) 601 return flags, value

602

603 - def _decode_value(self, flags, value):

604 """ 605 Decode a value depending on its flags 606 607 :Parameters: 608 - `flags`: Flag bit field 609 - `value`: Value to decode 610 611 :Types: 612 - `flags`: ``int`` 613 - `value`: ``str`` 614 615 :return: The decoded value 616 :rtype: any 617 618 :Exceptions: 619 - `ValueError`: Bad flags or bad value 620 """ 621 type_id = flags & 255 622 flags = flags & 65280 623 if type_id not in self._TYPEMAP or flags & NO_FLAGS: 624 raise ValueError() 625 626 if flags & FLAG_SPLIT: 627 keys = str(value).split() 628 value = self.get(*keys) 629 try: 630 value = ''.join([value[key] for key in keys]) 631 except KeyError: 632 raise ValueError() 633 if flags & FLAG_PADDED: 634 if len(value) < 16: 635 raise ValueError() 636 value = value[:-16] 637 if flags & FLAG_COMPRESSED: 638 if _zlib is None: 639 raise ValueError() 640 try: 641 value = _zlib.decompress(value) 642 except _zlib.error, e: 643 raise ValueError(str(e)) 644 value = self._TYPEMAP[type_id][2](value) 645 return value

646

647 - def _get_conn(self, key, *keys):

648 """ 649 Retrieve memcache connection 650 651 The actual memcache connection is selected by the key. 652 The algorithm is a simple 653 ``hashfunc(key) % weighted_selectable_pools`` 654 655 :Parameters: 656 - `key`: The key to use for selection 657 658 :Types: 659 - `key`: ``str`` 660 661 :return: The connection or ``None`` 662 :rtype: `MemcacheConnection` 663 """ 664 pools, conns, seen = self._weighted, {}, {} 665 for key in (key,) + keys: 666 conn, hashed = None, int(abs(hashfunc(key))) 667 while conn is None and pools: 668 pool = pools[hashed % len(pools)] 669 if pool in seen: 670 conns[seen[pool]].append(key) 671 break 672 state, retry = pool.state 673 if state == STATE_RETRY and not retry: 674 pools = pool.backup 675 continue 676 677 try: 678 conn = pool.get_conn() 679 except MemcacheConnectError: 680 if state == STATE_RETRY: 681 pools = pool.backup 682 continue 683 elif state != STATE_GRACE: 684 pool.mark_dead( 685 self._grace_time, self._retry_time, self._pools 686 ) 687 break 688 else: 689 if pool.dead: 690 pool.mark_alive() 691 seen[pool] = conn 692 conns.setdefault(conn, []).append(key) 693 break 694 return conns

Source Code for Module wtf.ext.memcache