@@ -1,4 +1,5 @@ from threading import local +import logging from hash_ring import HashRing @@ -16,6 +17,7 @@ LOOKUP_LOOKS = 2 USE_LOCAL_CACHE = True default_node = TyrantNode +default_system = 'default' SYSTEMS = {} LOCAL_CACHE = local() @@ -25,10 +27,12 @@ LOCAL_CACHE = local() def init(lookup_nodes, storage_nodes, system='default', node_type=None): name_to_s_node = {} name_to_l_node = {} - + global default_system + default_system = system lookup_ring = generate_ring(lookup_nodes, name_to_l_node, node_type) storage_ring = generate_ring(storage_nodes, name_to_s_node, node_type) - + + print("init() system = %s" % (system)) SYSTEMS[system] = (lookup_ring, storage_ring, name_to_l_node, name_to_s_node) @@ -48,6 +52,8 @@ def generate_nodes(lc_config): #--- System accessors ---------------------------------------------- def get_lookup_ring(system='default'): + #if system == 'default' and default_system != "default" + # return SYSTEMS[default_system][0] return SYSTEMS[system][0] def get_storage_ring(system='default'): @@ -185,6 +191,7 @@ def set(key, value, system='default'): If it is, same storage node is used. If it isn't, then the storage node is determinted by using hash_ring. """ + storage_node = locate_node_or_init(key, system) if USE_LOCAL_CACHE: lightcloud/__init__.py @@ -1,7 +1,14 @@ #!/usr/bin/python - """ redis.py - A client for the Redis daemon. + +Changes from Plurk Version: + +:connections: the *get_connection* function which returned a Redis + instance object, now is managed internally in Redis.connect(). + So, the connect() automatically uses local threads to connection pool + + """ __author__ = "Ludovico Magnocavallo <ludo\x40qix\x2eit>" @@ -33,14 +40,16 @@ class InvalidData(RedisError): pass connections = threading.local() def get_connection(host=None, port=None, timeout=None, db=None): + """Deprecated""" + raise Exception("Deprecated") key = '%s:%s' % (host, port) cur_connection = hasattr(connections, key) - + if not cur_connection: client = Redis(host, port, timeout, db) client.cache_key = key setattr(connections, key, client) - + return getattr(connections, key) def close_open_connections(): @@ -52,18 +61,20 @@ def close_open_connections(): class Redis(object): - """The main Redis client. + """The main Redis client. Internally manages connection + pooling automatically using threadlocal() """ - def __init__(self, host=None, port=None, timeout=None, db=None): self.host = host or 'localhost' self.port = port or 6379 + self.key = '%s:%s' % (self.host, self.port) if timeout: socket.setdefaulttimeout(timeout) self._sock = None self._fp = None self.db = db - + self._conn = None + def _write(self, s): """ >>> r = Redis(db=9) @@ -83,14 +94,17 @@ class Redis(object): if e.args[0] == 32: # broken pipe self.disconnect() - raise ConnectionError("Error %s while writing to socket. %s." % tuple(e.args)) + if self.ping() == "PONG": + pass # reconnected + else: + raise ConnectionError("Error %s while writing to socket. %s." % tuple(e.args)) def _read(self): try: return self._fp.readline() except socket.error, e: - if e.args and e.args[0] == errno.EAGAIN: - return + #if e.args and e.args[0] == errno.EAGAIN: + # return self.disconnect() raise ConnectionError("Error %s while reading from socket. %s." % tuple(e.args)) if not data: @@ -106,8 +120,7 @@ class Redis(object): >>> """ self.connect() - self._write('PING\r\n') - return self.get_response() + return self.get_response(write='PING\r\n') def set(self, name, value, preserve=False): """ @@ -133,12 +146,11 @@ class Redis(object): # convert back to unicode? and what about ints, or pickled values? try: value = value if isinstance(value, basestring) else str(value) - self._write('%s %s %s\r\n%s\r\n' % ( - 'SETNX' if preserve else 'SET', name, len(value), value - )) + write = '%s %s %s\r\n%s\r\n' % ('SETNX' if preserve else 'SET', name, len(value), value) + except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for key '%s': %s." % (name, e)) - return self.get_response() + return self.get_response(write=write) def get(self, name): """ @@ -161,8 +173,7 @@ class Redis(object): >>> """ self.connect() - self._write('GET %s\r\n' % name) - return self.get_response() + return self.get_response(write='GET %s\r\n' % name) def mget(self, *args): """ @@ -174,8 +185,7 @@ class Redis(object): >>> """ self.connect() - self._write('MGET %s\r\n' % ' '.join(args)) - return self.get_response() + return self.get_response(write='MGET %s\r\n' % ' '.join(args)) def incr(self, name, amount=1): """ @@ -192,10 +202,9 @@ class Redis(object): """ self.connect() if amount == 1: - self._write('INCR %s\r\n' % name) + return self.get_response(write='INCR %s\r\n' % name) else: - self._write('INCRBY %s %s\r\n' % (name, amount)) - return self.get_response() + return self.get_response(write='INCRBY %s %s\r\n' % (name, amount)) def decr(self, name, amount=1): """ @@ -215,10 +224,9 @@ class Redis(object): """ self.connect() if amount == 1: - self._write('DECR %s\r\n' % name) + return self.get_response(write='DECR %s\r\n' % name) else: - self._write('DECRBY %s %s\r\n' % (name, amount)) - return self.get_response() + return self.get_response(write='DECRBY %s %s\r\n' % (name, amount)) def exists(self, name): """ @@ -232,8 +240,7 @@ class Redis(object): >>> """ self.connect() - self._write('EXISTS %s\r\n' % name) - return self.get_response() + return self.get_response(write='EXISTS %s\r\n' % name) def delete(self, name): """ @@ -251,8 +258,7 @@ class Redis(object): >>> """ self.connect() - self._write('DEL %s\r\n' % name) - return self.get_response() + return self.get_response(write='DEL %s\r\n' % name) def get_type(self, name): """ @@ -265,8 +271,7 @@ class Redis(object): >>> """ self.connect() - self._write('TYPE %s\r\n' % name) - res = self.get_response() + res = self.get_response(write='TYPE %s\r\n' % name) return None if res == 'none' else res def keys(self, pattern): @@ -289,8 +294,7 @@ class Redis(object): >>> """ self.connect() - self._write('KEYS %s\r\n' % pattern) - return self.get_response().split() + return self.get_response(write='KEYS %s\r\n' % pattern).split() def randomkey(self): """ @@ -303,8 +307,7 @@ class Redis(object): """ #raise NotImplementedError("Implemented but buggy, do not use.") self.connect() - self._write('RANDOMKEY\r\n') - return self.get_response() + return self.get_response(write='RANDOMKEY\r\n') def rename(self, src, dst, preserve=False): """ @@ -370,12 +373,12 @@ class Redis(object): # same considerations on unicode as in set() apply here try: value = value if isinstance(value, basestring) else str(value) - self._write('%s %s %s\r\n%s\r\n' % ( + write = '%s %s %s\r\n%s\r\n' % ( 'LPUSH' if tail else 'RPUSH', name, len(value), value - )) + ) except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for element in list '%s': %s." % (name, e)) - return self.get_response() + return self.get_response(write=write) def llen(self, name): """ @@ -393,8 +396,7 @@ class Redis(object): >>> """ self.connect() - self._write('LLEN %s\r\n' % name) - return self.get_response() + return self.get_response(write='LLEN %s\r\n' % name) def lrange(self, name, start, end): """ @@ -420,8 +422,7 @@ class Redis(object): >>> """ self.connect() - self._write('LRANGE %s %s %s\r\n' % (name, start, end)) - return self.get_response() + return self.get_response(write='LRANGE %s %s %s\r\n' % (name, start, end)) def ltrim(self, name, start, end): """ @@ -450,8 +451,7 @@ class Redis(object): >>> """ self.connect() - self._write('LTRIM %s %s %s\r\n' % (name, start, end)) - return self.get_response() + return self.get_response(write='LTRIM %s %s %s\r\n' % (name, start, end)) def lindex(self, name, index): """ @@ -472,8 +472,7 @@ class Redis(object): >>> """ self.connect() - self._write('LINDEX %s %s\r\n' % (name, index)) - return self.get_response() + return self.get_response(write='LINDEX %s %s\r\n' % (name, index)) def pop(self, name, tail=False): """ @@ -502,8 +501,7 @@ class Redis(object): >>> """ self.connect() - self._write('%s %s\r\n' % ('RPOP' if tail else 'LPOP', name)) - return self.get_response() + return self.get_response(write='%s %s\r\n' % ('RPOP' if tail else 'LPOP', name)) def lset(self, name, index, value): """ @@ -531,13 +529,13 @@ class Redis(object): self.connect() try: value = value if isinstance(value, basestring) else str(value) - self._write('LSET %s %s %s\r\n%s\r\n' % ( + write = 'LSET %s %s %s\r\n%s\r\n' % ( name, index, len(value), value - )) + ) except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for element %s in list '%s': %s." % (index, name, e)) - return self.get_response() - + return self.get_response(write=write) + def lrem(self, name, value, num=0): """ >>> r = Redis(db=9) @@ -574,7 +572,7 @@ class Redis(object): except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for element %s in list '%s': %s." % (index, name, e)) return self.get_response() - + def sort(self, name, by=None, get=None, start=None, num=None, desc=False, alpha=False): """ >>> r = Redis(db=9) @@ -635,7 +633,7 @@ class Redis(object): self.connect() self._write(' '.join(stmt + ["\r\n"])) return self.get_response() - + def sadd(self, name, value): """ >>> r = Redis(db=9) @@ -656,7 +654,7 @@ class Redis(object): except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for element in set '%s': %s." % (name, e)) return self.get_response() - + def srem(self, name, value): """ >>> r = Redis(db=9) @@ -682,7 +680,7 @@ class Redis(object): except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for element in set '%s': %s." % (name, e)) return self.get_response() - + def sismember(self, name, value): """ >>> r = Redis(db=9) @@ -708,7 +706,7 @@ class Redis(object): except UnicodeEncodeError, e: raise InvalidData("Error encoding unicode value for element in set '%s': %s." % (name, e)) return self.get_response() - + def sinter(self, *args): """ >>> r = Redis(db=9) @@ -740,7 +738,7 @@ class Redis(object): self.connect() self._write('SINTER %s\r\n' % ' '.join(args)) return set(self.get_response()) - + def sinterstore(self, dest, *args): """ >>> r = Redis(db=9) @@ -764,7 +762,7 @@ class Redis(object): self.connect() self._write('SINTERSTORE %s %s\r\n' % (dest, ' '.join(args))) return self.get_response() - + def smembers(self, name): """ >>> r = Redis(db=9) @@ -786,7 +784,7 @@ class Redis(object): self.connect() self._write('SMEMBERS %s\r\n' % name) return set(self.get_response()) - + def select(self, db): """ >>> r = Redis(db=9) @@ -804,7 +802,7 @@ class Redis(object): self.connect() self._write('SELECT %s\r\n' % db) return self.get_response() - + def move(self, name, db): """ >>> r = Redis(db=9) @@ -833,7 +831,7 @@ class Redis(object): self.connect() self._write('MOVE %s %s\r\n' % (name, db)) return self.get_response() - + def save(self, background=False): """ >>> r = Redis(db=9) @@ -849,11 +847,10 @@ class Redis(object): """ self.connect() if background: - self._write('BGSAVE\r\n') + return self.get_response(write='BGSAVE\r\n') else: - self._write('SAVE\r\n') - return self.get_response() - + return self.get_response(write='SAVE\r\n') + def lastsave(self): """ >>> import time @@ -868,7 +865,7 @@ class Redis(object): self.connect() self._write('LASTSAVE\r\n') return self.get_response() - + def flush(self, all_dbs=False): """ >>> r = Redis(db=9) @@ -880,7 +877,7 @@ class Redis(object): self.connect() self._write('%s\r\n' % ('FLUSHALL' if all_dbs else 'FLUSHDB')) return self.get_response() - + def info(self): """ >>> r = Redis(db=9) @@ -892,20 +889,31 @@ class Redis(object): >>> """ self.connect() - self._write('INFO\r\n') info = dict() - for l in self.get_response().split('\r\n'): + for l in self.get_response(write='INFO\r\n').split('\r\n'): if not l: continue k, v = l.split(':', 1) info[k] = int(v) if v.isdigit() else v return info - - def get_response(self): - data = self._read().strip() - if not data: - self.disconnect() - raise ConnectionError("Socket closed on remote end") + + def get_response(self,write=None): + if write is not None: + self._write(write) + data = self._read().strip() + if not data: + self.disconnect() + print("915: reconnecting, bad connection") + self.connect() + self._write(write) + data = self._read().strip() + if not data: + raise ConnectionError("Socket closed on remote end") + else: + data = self._read().strip() + if not data: + self.disconnect() + raise ConnectionError("Socket closed on remote end") c = data[0] if c == '-': raise ResponseError(data[5:] if data[:5] == '-ERR ' else data[1:]) @@ -921,7 +929,7 @@ class Redis(object): result.append(self._get_value()) return result return self._get_value(data) - + def _get_value(self, data=None): data = data or self._read().strip() if data == '$-1': @@ -942,8 +950,9 @@ class Redis(object): if i < 0: break return ''.join(buf)[:-2] - + def disconnect(self): + print("Redis: Disconnect") if isinstance(self._sock, socket.socket): try: self._sock.close() @@ -951,7 +960,12 @@ class Redis(object): pass self._sock = None self._fp = None - + if hasattr(self,"_conn"): + self._conn.disconnect() + self._conn = None + elif hasattr(connections,self.key ): + getattr(connections, self.key ).disconnect() + def connect(self): """ >>> r = Redis(db=9) @@ -961,19 +975,50 @@ class Redis(object): >>> r.disconnect() >>> """ - if isinstance(self._sock, socket.socket): + class socket_connection(object): + def __init__(self,host,port,key): + self.key = key + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.connect((host, port)) + sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1) + except socket.error, e: + raise ConnectionError("Error %s connecting to %s:%s. %s." % (e.args[0], host, port, e.args[1])) + else: + # no exceptions + self.sock = sock + self.fp = self.sock.makefile('r') + + def disconnect(self): + try: + self.sock.close() + except socket.error: + pass + self.fp = None + self.sock = None + delattr(connections, self.key) + + + if isinstance(self._conn, socket_connection): return try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.connect((self.host, self.port)) - sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1) + # connection pooling to thread local + if not hasattr(connections, self.key): + print("1023: connecting") + conn = socket_connection(self.host,self.port,self.key) + setattr(connections,self.key,conn) + else: + conn = getattr(connections, self.key) except socket.error, e: raise ConnectionError("Error %s connecting to %s:%s. %s." % (e.args[0], self.host, self.port, e.args[1])) else: - self._sock = sock - self._fp = self._sock.makefile('r') + # no exceptions + self._conn = conn + self._sock = conn.sock + self._fp = conn.fp if self.db: self.select(self.db) + if __name__ == '__main__': lightcloud/redis.py 1e89a835fc794cce213900a2cbebd07004ae576b Aaron Raddon araddon@yahoo.com http://github.com/araddon/lightcloud/commit/f066d0f5caa872e0d24a47c46a741f1565f9ff7e f066d0f5caa872e0d24a47c46a741f1565f9ff7e 2009-11-02T22:02:19-08:00 2009-11-02T22:02:19-08:00 moving connection pooling internal to Redis class, self manages connection pools with threadlocals 8e56165d0ce53417340d317819587b17eb1bd40f Aaron Raddon araddon@yahoo.com