| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| |
| from __future__ import print_function |
| from __future__ import with_statement |
| |
| import os |
| import logging.config |
| import six |
| |
| from thrift.Thrift import TMessageType, TApplicationException |
| from twisted.internet import defer |
| from twisted.internet import reactor |
| from twisted.internet.defer import inlineCallbacks, succeed, fail |
| from twisted.internet.protocol import ClientCreator |
| |
| from pypegasus import rrdb, replication |
| from pypegasus.base.ttypes import * |
| from pypegasus.operate.packet import * |
| from pypegasus.replication.ttypes import query_cfg_request |
| from pypegasus.rrdb import * |
| from pypegasus.rrdb.ttypes import scan_request, get_scanner_request, update_request, key_value, multi_put_request, \ |
| multi_get_request, multi_remove_request |
| from pypegasus.transport.protocol import * |
| from pypegasus.utils.tools import restore_key, get_ttl, bytes_cmp, ScanOptions |
| |
| try: |
| from thrift.protocol import fastbinary |
| except: |
| fastbinary = None |
| |
| |
| logging.config.fileConfig(os.path.dirname(__file__)+"/logger.conf") |
| logger = logging.getLogger("pgclient") |
| |
| DEFAULT_TIMEOUT = 2000 # ms |
| META_CHECK_INTERVAL = 2 # s |
| MAX_TIMEOUT_THRESHOLD = 5 # times |
| |
| |
| class BaseSession(object): |
| |
| def __init__(self, transport, oprot_factory, container, timeout): |
| self._transport = transport # TPegasusTransport |
| self._oprot_factory = oprot_factory |
| self._container = container |
| self._seqid = 0 |
| self._requests = {} |
| self._default_timeout = timeout |
| |
| def __del__(self): |
| self.close() |
| |
| def get_peer_addr(self): |
| return self._transport.get_peer_addr() |
| |
| def cb_send(self, _, seqid): |
| return self._requests[seqid] |
| |
| def eb_send(self, f, seqid): |
| d = self._requests.pop(seqid) |
| d.errback(f) |
| logger.warning('peer: %s, failure: %s', |
| self.get_peer_addr(), f) |
| return d |
| |
| def eb_recv(self, f): |
| logger.warning('peer: %s, failure: %s', |
| self.get_peer_addr(), f) |
| |
| def on_timeout(self, _, seconds): |
| ec = error_types.ERR_TIMEOUT |
| self._container.update_state(ec) |
| logger.warning('peer: %s, time: %s s, timeout', |
| self.get_peer_addr(), seconds) |
| return ec.value, None |
| |
| def operate(self, op, timeout=None): |
| if not isinstance(timeout, int) or timeout <= 0: |
| timeout = self._default_timeout |
| |
| seqid = self._seqid = self._seqid + 1 # TODO should keep atomic |
| dr = defer.Deferred() |
| dr.addErrback(self.eb_recv) |
| self._requests[seqid] = dr |
| |
| # ds(deferred send) will wait dr(deferred receive) |
| ds = defer.maybeDeferred(self.send_req, op, seqid) |
| ds.addCallbacks( |
| callback=self.cb_send, |
| callbackArgs=(seqid,), |
| errback=self.eb_send, |
| errbackArgs=(seqid,)) |
| ds.addTimeout(timeout/1000.0, reactor, self.on_timeout) |
| return ds |
| |
| def send_req(self, op, seqid): |
| oprot = self._oprot_factory.getProtocol(self._transport) |
| oprot.trans.seek(ThriftHeader.HEADER_LENGTH) # skip header |
| op.send_data(oprot, seqid) |
| body_length = oprot.trans.tell() - ThriftHeader.HEADER_LENGTH |
| oprot.trans.seek(0) # back to header |
| oprot.trans.write(op.prepare_thrift_header(body_length)) |
| oprot.trans.flush() |
| |
| def recv_ACK(self, iprot, mtype, rseqid, errno, result_type, parser): |
| if rseqid not in self._requests: |
| logger.warning('peer: %s rseqid: %s not found', |
| self.get_peer_addr(), rseqid) |
| return |
| d = self._requests.pop(rseqid) |
| if errno != 'ERR_OK': |
| rc = error_code.value_of(errno) |
| self._container.update_state(rc) |
| return d.callback(rc) |
| else: |
| if mtype == TMessageType.EXCEPTION: |
| x = TApplicationException() |
| x.read(iprot) |
| iprot.readMessageEnd() |
| return d.errback(x) |
| |
| result = result_type() |
| result.read(iprot) |
| if result.success: |
| return d.callback(parser(result.success)) |
| |
| return d.errback(TApplicationException(TApplicationException.MISSING_RESULT, |
| "%s failed: unknown result" % |
| getattr(result_type, '__name__'))) |
| |
| def close(self): |
| self._transport.close() |
| |
| |
| class MetaSession(BaseSession): |
| |
| def __init__(self, transport, oprot_factory, container, timeout): |
| BaseSession.__init__(self, transport, oprot_factory, container, timeout) |
| |
| def recv_RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| meta.query_cfg_result, |
| QueryCfgOperator.parse_result) |
| |
| |
| class ReplicaSession(BaseSession): |
| |
| def __init__(self, transport, oprot_factory, container, timeout): |
| BaseSession.__init__(self, transport, oprot_factory, container, timeout) |
| |
| def recv_RPC_RRDB_RRDB_PUT_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.put_result, RrdbPutOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_TTL_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.ttl_result, |
| RrdbTtlOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_GET_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.get_result, |
| RrdbGetOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_REMOVE_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.put_result, |
| RrdbRemoveOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_SORTKEY_COUNT_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.sortkey_count_result, |
| RrdbSortkeyCountOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_MULTI_PUT_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.put_result, |
| RrdbMultiPutOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_MULTI_GET_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.multi_get_result, |
| RrdbMultiGetOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_MULTI_REMOVE_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.multi_remove_result, |
| RrdbMultiRemoveOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_GET_SCANNER_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.get_scanner_result, |
| RrdbGetScannerOperator.parse_result) |
| |
| def recv_RPC_RRDB_RRDB_SCAN_ACK(self, iprot, mtype, rseqid, errno): |
| self.recv_ACK(iprot, mtype, rseqid, errno, |
| rrdb.scan_result, |
| RrdbScanOperator.parse_result) |
| |
| |
| class SessionManager(object): |
| def __init__(self, name, timeout): |
| self.name = name |
| self.session_dict = {} # rpc_addr => session |
| self.timeout = timeout |
| |
| def __del__(self): |
| self.close() |
| |
| def got_conn(self, conn): |
| addr = rpc_address() |
| addr.from_string(conn.transport.addr[0] + ':' + str(conn.transport.addr[1])) |
| self.session_dict[addr] = conn.client |
| return conn.client |
| |
| def got_err(self, err): |
| logger.error('table: %s, connect err: %s', |
| self.name, err) |
| |
| def close(self): |
| for session in self.session_dict.values(): |
| session.close() |
| |
| def update_state(self, ec): |
| pass |
| |
| |
| class MetaSessionManager(SessionManager): |
| |
| def __init__(self, table_name, timeout): |
| SessionManager.__init__(self, table_name, timeout) |
| self.addr_list = [] |
| |
| def add_meta_server(self, meta_addr): |
| rpc_addr = rpc_address() |
| if rpc_addr.from_string(meta_addr): |
| ip_port = meta_addr.split(':') |
| if not len(ip_port) == 2: |
| return False |
| |
| ip, port = ip_port[0], int(ip_port[1]) |
| self.addr_list.append((ip, port)) |
| |
| return True |
| else: |
| return False |
| |
| @inlineCallbacks |
| def query_one(self, session): |
| req = query_cfg_request(self.name, []) |
| op = QueryCfgOperator(gpid(0, 0), req, 0) |
| ret = yield session.operate(op) |
| defer.returnValue(ret) |
| |
| def got_results(self, res): |
| for (suc, result) in res: |
| if suc \ |
| and result.__class__.__name__ == "query_cfg_response" \ |
| and result.is_stateful: |
| logger.info('table: %s, partition result: %s', |
| self.name, result) |
| return result |
| |
| logger.error('query partition info err. table: %s err: %s', |
| self.name, res) |
| |
| def query(self): |
| ds = [] |
| for (ip, port) in self.addr_list: |
| rpc_addr = rpc_address() |
| rpc_addr.from_string(ip + ':' + str(port)) |
| if rpc_addr in self.session_dict: |
| self.session_dict[rpc_addr].close() |
| |
| d = ClientCreator(reactor, |
| TPegasusThriftClientProtocol, |
| MetaSession, |
| TBinaryProtocol.TBinaryProtocolFactory(), |
| None, |
| self, |
| self.timeout |
| ).connectTCP(ip, port, self.timeout) |
| d.addCallbacks(self.got_conn, self.got_err) |
| d.addCallbacks(self.query_one, self.got_err) |
| ds.append(d) |
| |
| dlist = defer.DeferredList(ds, consumeErrors=True) |
| dlist.addCallback(self.got_results) |
| return dlist |
| |
| |
| class Table(SessionManager): |
| def __init__(self, name, container, timeout): |
| SessionManager.__init__(self, name, timeout) |
| self.app_id = 0 |
| self.partition_count = 0 |
| self.query_cfg_response = None |
| self.partition_dict = {} # partition_index => rpc_addr |
| self.partition_ballot = {} # partition_index => ballot |
| self.container = container |
| |
| def got_results(self, res): |
| logger.info('table: %s, replica session: %s', |
| self.name, res) |
| return True |
| |
| def update_cfg(self, resp): |
| if resp.__class__.__name__ != "query_cfg_response": |
| logger.error('table: %s, query_cfg_response is error', |
| self.name) |
| return None |
| |
| self.query_cfg_response = resp |
| self.app_id = self.query_cfg_response.app_id |
| self.partition_count = self.query_cfg_response.partition_count |
| |
| ds = [] |
| connected_rpc_addrs = {} |
| for partition in self.query_cfg_response.partitions: |
| rpc_addr = partition.primary |
| self.partition_dict[partition.pid.get_pidx()] = rpc_addr |
| self.partition_ballot[partition.pid.get_pidx()] = partition.ballot |
| |
| # table is partition split, and child partition is not ready |
| # child requests should be redirected to its parent partition |
| # this will be happened when query meta is called during partition split |
| if partition.ballot < 0: |
| continue |
| |
| if rpc_addr in connected_rpc_addrs or rpc_addr.address == 0: |
| continue |
| |
| ip, port = rpc_addr.to_ip_port() |
| if rpc_addr in self.session_dict: |
| self.session_dict[rpc_addr].close() |
| |
| d = ClientCreator(reactor, |
| TPegasusThriftClientProtocol, |
| ReplicaSession, |
| TBinaryProtocol.TBinaryProtocolFactory(), |
| None, |
| self.container, |
| self.timeout |
| ).connectTCP(ip, port, self.timeout) |
| connected_rpc_addrs[rpc_addr] = 1 |
| d.addCallbacks(self.got_conn, self.got_err) |
| ds.append(d) |
| |
| dlist = defer.DeferredList(ds, consumeErrors=True) |
| dlist.addCallback(self.got_results) |
| return dlist |
| |
| def get_hashkey_hash(self, hash_key): |
| if six.PY3 and isinstance(hash_key, six.string_types): |
| hash_key = hash_key.encode("UTF-8") |
| return PegasusHash.default_hash(hash_key) |
| |
| def get_blob_hash(self, blob_key): |
| return PegasusHash.hash(blob_key) |
| |
| def get_gpid_by_hash(self, partition_hash): |
| pidx = partition_hash % self.get_partition_count() |
| if self.partition_ballot[pidx] < 0: |
| logger.warn("table[%s] partition[%d] is not ready, requests will send to parent partition[%d]", |
| self.name, |
| pidx, |
| pidx - int(self.partition_count / 2)) |
| pidx -= int(self.partition_count / 2) |
| return gpid(self.app_id, pidx) |
| |
| def get_all_gpid(self): |
| return [gpid(self.app_id, pidx) |
| for pidx in range(self.get_partition_count())] |
| |
| def get_partition_count(self): |
| return self.query_cfg_response.partition_count |
| |
| def get_session(self, peer_gpid): |
| pidx = peer_gpid.get_pidx() |
| if pidx in self.partition_dict.keys(): |
| replica_addr = self.partition_dict[pidx] |
| if replica_addr in self.session_dict.keys(): |
| return self.session_dict[replica_addr] |
| |
| self.container.update_state(error_types.ERR_OBJECT_NOT_FOUND) |
| return None |
| |
| |
| class PegasusScanner(object): |
| """ |
| Pegasus scanner class, used for scanning data in pegasus table. |
| """ |
| |
| CONTEXT_ID_VALID_MIN = 0 |
| CONTEXT_ID_COMPLETED = -1 |
| CONTEXT_ID_NOT_EXIST = -2 |
| |
| def __init__(self, table, gpid_list, scan_options, partition_hash_list, check_hash, |
| start_key=blob(b'\x00\x00'), stop_key=blob(b'\xFF\xFF')): |
| self._table = table |
| self._gpid = gpid(0) |
| self._gpid_list = gpid_list |
| self._scan_options = scan_options |
| self._start_key = start_key |
| self._stop_key = stop_key |
| self._p = -1 |
| self._context_id = self.CONTEXT_ID_COMPLETED |
| self._kvs = [] |
| self._partition_hash = 0 |
| self._partition_hash_list = partition_hash_list |
| self._check_hash = check_hash |
| |
| def __repr__(self): |
| lst = ['%s=%r' % (key, value) |
| for key, value in self.__dict__.items()] |
| return '%s(%s)' % (self.__class__.__name__, ', '.join(lst)) |
| |
| @inlineCallbacks |
| def get_next(self): |
| """ |
| scan the next k-v pair for the scanner. |
| :return: (tuple<tuple<hash_key, sort_key>, value> or None) |
| all the sort_keys returned by this API are in ascend order. |
| """ |
| self._p += 1 |
| while self._p >= len(self._kvs): |
| if self._context_id == self.CONTEXT_ID_COMPLETED: |
| # reach the end of one partition |
| if len(self._gpid_list) == 0: |
| defer.returnValue(None) |
| else: |
| self._gpid = self._gpid_list.pop() |
| self._partition_hash = self._partition_hash_list.pop() |
| self.split_reset() |
| elif self._context_id == self.CONTEXT_ID_NOT_EXIST: |
| # no valid context_id found |
| yield self.start_scan() |
| else: |
| ret = yield self.next_batch() |
| if ret == 1: |
| # context not found |
| self._context_id = self.CONTEXT_ID_NOT_EXIST |
| elif ret != 0: |
| raise Exception("Rocksdb error: " + ret) |
| |
| self._p += 1 |
| |
| defer.returnValue((restore_key(self._kvs[self._p].key.data), |
| self._kvs[self._p].value.data)) |
| |
| def split_reset(self): |
| self._kvs = [] |
| self._p = -1 |
| self._context_id = self.CONTEXT_ID_NOT_EXIST |
| |
| def scan_cb(self, ctx): |
| if isinstance(ctx, dict) and ctx['error'] == 0: |
| self._kvs = ctx['kvs'] |
| self._p = -1 |
| self._context_id = ctx['context_id'] |
| |
| return ctx['error'] |
| else: |
| raise Exception('operate error!') |
| |
| def scan_err_cb(self, err): |
| logger.error('scan table: %s start_key: %s stop_key: %s is error: %s', |
| self._table.name, self._start_key, self._stop_key, err) |
| return err |
| |
| def start_scan(self): |
| request = get_scanner_request() |
| if len(self._kvs) == 0: |
| request.start_key = self._start_key |
| request.start_inclusive = self._scan_options.start_inclusive |
| else: |
| request.start_key = self._kvs[-1].key |
| request.start_inclusive = False |
| |
| request.stop_key = self._stop_key |
| request.stop_inclusive = self._scan_options.stop_inclusive |
| request.batch_size = self._scan_options.batch_size |
| request.need_check_hash = self._check_hash |
| |
| op = RrdbGetScannerOperator(self._gpid, request, self._partition_hash) |
| session = self._table.get_session(self._gpid) |
| if not session or not op: |
| raise Exception('session or packet error!') |
| |
| ret = session.operate(op, self._scan_options.timeout_millis) |
| ret.addCallbacks(self.scan_cb, self.scan_err_cb) |
| return ret |
| |
| def next_batch(self): |
| request = scan_request(self._context_id) |
| op = RrdbScanOperator(self._gpid, request, self._partition_hash) |
| session = self._table.get_session(self._gpid) |
| if not session or not op: |
| raise Exception('session or packet error!') |
| |
| ret = session.operate(op, self._scan_options.timeout_millis) |
| ret.addCallbacks(self.scan_cb, self.scan_err_cb) |
| return ret |
| |
| def close(self): |
| if self._context_id >= self.CONTEXT_ID_VALID_MIN: |
| op = RrdbClearScannerOperator(self._gpid, self._context_id, self._partition_hash) |
| session = self._table.get_session(self._gpid) |
| self._context_id = self.CONTEXT_ID_COMPLETED |
| if not session or not op: |
| raise Exception('session or packet error!') |
| |
| session.operate(op, self._scan_options.timeout_millis) |
| |
| |
| class PegasusHash(object): |
| polynomial, = struct.unpack('<q', struct.pack('<Q', 0x9a6c9329ac4bc9b5)) |
| table_forward = [0] * 256 |
| |
| @classmethod |
| def unsigned_right_shift(cls, val, n): |
| if val >= 0: |
| val >>= n |
| else: |
| val = ((val + 0x10000000000000000) >> n) |
| return val |
| |
| @classmethod |
| def populate_table(cls): |
| for i in range(256): |
| crc = i |
| for j in range(8): |
| if crc & 1: |
| crc = cls.unsigned_right_shift(crc, 1) |
| crc ^= cls.polynomial |
| else: |
| crc = cls.unsigned_right_shift(crc, 1) |
| cls.table_forward[i] = crc |
| |
| @classmethod |
| def crc64(cls, data, offset, length): |
| crc = 0xffffffffffffffff |
| end = offset + length |
| |
| for c in data[offset:end:1]: |
| crc = cls.table_forward[(c ^ crc) & 0xFF] ^ cls.unsigned_right_shift(crc, 8) |
| return ~crc |
| |
| @classmethod |
| def default_hash(cls, hash_key): |
| return cls.crc64(hash_key, 0, len(hash_key)) |
| |
| @classmethod |
| def hash(cls, blob_key): |
| assert blob_key is not None and len(blob_key) >= 2, 'blob_key is invalid!' |
| |
| # hash_key_len is in big endian |
| s = struct.Struct('>H') |
| hash_key_len = s.unpack(blob_key.data[:2])[0] |
| |
| assert hash_key_len != 0xFFFF and (2 + hash_key_len <= len(blob_key)), 'blob_key hash_key_len is invalid!' |
| if hash_key_len == 0: |
| return cls.crc64(blob_key.data, 2, len(blob_key) - 2) |
| else: |
| return cls.crc64(blob_key.data, 2, hash_key_len) |
| |
| |
| class Pegasus(object): |
| """ |
| Pegasus client class. |
| """ |
| |
| @classmethod |
| def generate_key(cls, hash_key, sort_key): |
| # assert(len(hash_key) < sys.maxsize > 1) |
| if six.PY3 and isinstance(hash_key, six.string_types): |
| hash_key = hash_key.encode("UTF-8") |
| if six.PY3 and isinstance(sort_key, six.string_types): |
| sort_key = sort_key.encode("UTF-8") |
| |
| # hash_key_len is in big endian |
| hash_key_len = len(hash_key) |
| sort_key_len = len(sort_key) |
| |
| if sort_key_len > 0: |
| values = (hash_key_len, hash_key, sort_key) |
| s = struct.Struct('>H'+str(hash_key_len)+'s'+str(sort_key_len)+'s') |
| else: |
| values = (hash_key_len, hash_key) |
| s = struct.Struct('>H'+str(hash_key_len)+'s') |
| |
| buff = ctypes.create_string_buffer(s.size) |
| s.pack_into(buff, 0, *values) |
| |
| return blob(buff) |
| |
| @classmethod |
| def generate_next_bytes(cls, buff): |
| pos = len(buff) - 1 |
| found = False |
| while pos >= 0: |
| if ord(buff[pos]) != 0xFF: |
| buff[pos] += 1 |
| found = True |
| break |
| if found: |
| return buff |
| else: |
| return buff + chr(0) |
| |
| @classmethod |
| def generate_stop_key(cls, hash_key, stop_sort_key): |
| if stop_sort_key: |
| return cls.generate_key(hash_key, stop_sort_key), True |
| else: |
| return cls.generate_next_bytes(hash_key), False |
| |
| def __init__(self, meta_addrs=None, table_name='', |
| timeout=DEFAULT_TIMEOUT): |
| """ |
| :param meta_addrs: (list) pagasus meta servers list. |
| example: ['127.0.0.1:34601', '127.0.0.1:34602', '127.0.0.1:34603'] |
| :param table_name: (str) table name/app name used in pegasus. |
| :param timeout: (int) default timeout in milliseconds when communicate with meta sever and replica server. |
| """ |
| self.name = table_name |
| self.table = Table(table_name, self, timeout) |
| self.meta_session_manager = MetaSessionManager(table_name, timeout) |
| if isinstance(meta_addrs, list): |
| for meta_addr in meta_addrs: |
| self.meta_session_manager.add_meta_server(meta_addr) |
| PegasusHash.populate_table() |
| self.timeout_times = 0 |
| self.update_partition = False |
| self.timer = reactor.callLater(META_CHECK_INTERVAL, self.check_state) |
| |
| def init(self): |
| """ |
| Initialize the client before you can use it. |
| |
| :return: (DeferredList) True when initialized succeed, others when failed. |
| """ |
| dlist = self.meta_session_manager.query() |
| dlist.addCallback(self.table.update_cfg) |
| return dlist |
| |
| def close(self): |
| """ |
| Close the client. The client can not be used again after closed. |
| """ |
| self.timer.cancel() |
| self.table.close() |
| self.meta_session_manager.close() |
| |
| @inlineCallbacks |
| def check_state(self): |
| logger.info('table: %s, checking meta ...', |
| self.name) |
| if self.update_partition: |
| self.update_partition = False |
| yield self.init() |
| self.timer = reactor.callLater(META_CHECK_INTERVAL, self.check_state) |
| |
| def update_state(self, ec): |
| if ec == error_types.ERR_TIMEOUT: |
| self.timeout_times += 1 |
| if self.timeout_times >= MAX_TIMEOUT_THRESHOLD: |
| self.update_partition = True |
| self.timeout_times = 0 |
| elif ec == error_types.ERR_INVALID_DATA: |
| logger.error('table: %s, ignore ec: %s:%s', |
| self.name, ec.name, ec.value) # TODO when it happen? |
| elif ec == error_types.ERR_SESSION_RESET: |
| pass |
| elif (ec == error_types.ERR_OBJECT_NOT_FOUND |
| or ec == error_types.ERR_INACTIVE_STATE |
| or ec == error_types.ERR_INVALID_STATE |
| or ec == error_types.ERR_NOT_ENOUGH_MEMBER |
| or ec == error_types.ERR_PARENT_PARTITION_MISUSED): |
| self.update_partition = True |
| else: |
| logger.error('table: %s, ignore ec: %s:%s', |
| self.name, ec.name, ec.value) |
| |
| def ttl(self, hash_key, sort_key, timeout=0): |
| """ |
| Get ttl(time to live) of the data. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sort_key: (bytes) which sort key used for this API. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, int>) (code, ttl) |
| code: error_types.ERR_OK.value when data exist, error_types.ERR_OBJECT_NOT_FOUND.value when data not found. |
| ttl: in seconds, -1 means forever. |
| """ |
| blob_key = self.generate_key(hash_key, sort_key) |
| partition_hash = self.table.get_blob_hash(blob_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| op = RrdbTtlOperator(peer_gpid, blob_key, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def exist(self, hash_key, sort_key, timeout=0): |
| """ |
| Check value exist. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sort_key: (bytes) which sort key used for this API. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, None>) (code, ign) |
| code: error_types.ERR_OK.value when data exist, error_types.ERR_OBJECT_NOT_FOUND.value when data not found. |
| ign: useless, should be ignored. |
| """ |
| return self.ttl(hash_key, sort_key, timeout) |
| |
| def get(self, hash_key, sort_key, timeout=0): |
| """ |
| Get value stored in <hash_key, sort_key>. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sort_key: (bytes) which sort key used for this API. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, bytes>) (code, value). |
| code: error_types.ERR_OK.value when data got succeed, error_types.ERR_OBJECT_NOT_FOUND.value when data not found. |
| value: data stored in this <hash_key, sort_key> |
| """ |
| blob_key = self.generate_key(hash_key, sort_key) |
| partition_hash = self.table.get_blob_hash(blob_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| op = RrdbGetOperator(peer_gpid, blob_key, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def set(self, hash_key, sort_key, value, ttl=0, timeout=0): |
| """ |
| Set value to be stored in <hash_key, sort_key>. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sort_key: (bytes) which sort key used for this API. |
| :param value: (bytes) value to be stored under <hash_key, sort_key>. |
| :param ttl: (int) ttl(time to live) in seconds of this data. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, None>) (code, ign) |
| code: error_types.ERR_OK.value when data stored succeed. |
| ign: useless, should be ignored. |
| """ |
| blob_key = self.generate_key(hash_key, sort_key) |
| partition_hash = self.table.get_blob_hash(blob_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| op = RrdbPutOperator(peer_gpid, update_request(blob_key, blob(value), get_ttl(ttl)), partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def remove(self, hash_key, sort_key, timeout=0): |
| """ |
| Remove the entire <hash_key, sort_key>-value in pegasus. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sort_key: (bytes) which sort key used for this API. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, None>) (code, ign) |
| code: error_types.ERR_OK.value when data stored succeed. |
| ign: useless, should be ignored. |
| """ |
| blob_key = self.generate_key(hash_key, sort_key) |
| partition_hash = self.table.get_blob_hash(blob_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| op = RrdbRemoveOperator(peer_gpid, blob_key, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def sort_key_count(self, hash_key, timeout=0): |
| """ |
| Get the total sort key count under the hash_key. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, count>) (code, count) |
| code: error_types.ERR_OK.value when data got succeed, error_types.ERR_OBJECT_NOT_FOUND.value when data not found. |
| value: total sort key count under the hash_key. |
| """ |
| partition_hash = self.table.get_hashkey_hash(hash_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| op = RrdbSortkeyCountOperator(peer_gpid, blob(hash_key), partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def multi_set(self, hash_key, sortkey_value_dict, ttl=0, timeout=0): |
| """ |
| Set multiple sort_keys-values under hash_key to be stored. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sortkey_value_dict: (dict) <sort_key, value> pairs in dict. |
| :param ttl: (int) ttl(time to live) in seconds of these data. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, _>) (code, ign) |
| code: error_types.ERR_OK.value when data stored succeed. |
| ign: useless, should be ignored. |
| """ |
| partition_hash = self.table.get_hashkey_hash(hash_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| kvs = [key_value(blob(k), blob(v)) for k, v in sortkey_value_dict.items()] |
| ttl = get_ttl(ttl) |
| req = multi_put_request(blob(hash_key), kvs, ttl) |
| op = RrdbMultiPutOperator(peer_gpid, req, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def multi_get(self, hash_key, |
| sortkey_set, |
| max_kv_count=100, |
| max_kv_size=1000000, |
| no_value=False, |
| timeout=0): |
| """ |
| Get multiple values stored in <hash_key, sortkey> pairs. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sortkey_set: (set) sort keys in set. |
| :param max_kv_count: (int) max count of k-v pairs to be fetched. max_fetch_count <= 0 means no limit. |
| :param max_kv_size: (int) max total data size of k-v pairs to be fetched. max_fetch_size <= 0 means no limit. |
| :param no_value: (bool) whether to fetch value of these keys. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, dict>) (code, kvs) |
| code: error_types.ERR_OK.value when data got succeed. |
| kvs: <sort_key, value> pairs in dict. |
| """ |
| partition_hash = self.table.get_hashkey_hash(hash_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| ks = [] |
| if sortkey_set is None: |
| pass |
| elif isinstance(sortkey_set, set): |
| ks = [blob(k) for k in sortkey_set] |
| else: |
| return error_types.ERR_INVALID_PARAMETERS.value, 0 |
| |
| req = multi_get_request(blob(hash_key), ks, |
| max_kv_count, max_kv_size, |
| no_value) |
| op = RrdbMultiGetOperator(peer_gpid, req, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def multi_get_opt(self, hash_key, |
| start_sort_key, stop_sort_key, |
| multi_get_options, |
| max_kv_count=100, |
| max_kv_size=1000000, |
| timeout=0): |
| """ |
| Get multiple values stored in hash_key, and sort key range in [start_sort_key, stop_sort_key) as default. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param start_sort_key: (bytes) returned k-v pairs is start from start_sort_key. |
| :param stop_sort_key: (bytes) returned k-v pairs is stop at stop_sort_key. |
| :param multi_get_options: (MultiGetOptions) configurable multi_get options. |
| :param max_kv_count: (int) max count of k-v pairs to be fetched. max_fetch_count <= 0 means no limit. |
| :param max_kv_size: (int) max total data size of k-v pairs to be fetched. max_fetch_size <= 0 means no limit. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, dict>) (code, kvs) |
| code: error_types.ERR_OK.value when data got succeed. |
| kvs: <sort_key, value> pairs in dict. |
| """ |
| partition_hash = self.table.get_hashkey_hash(hash_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| req = multi_get_request(blob(hash_key), |
| None, |
| max_kv_count, |
| max_kv_size, |
| multi_get_options.no_value, |
| blob(start_sort_key), |
| blob(stop_sort_key), |
| multi_get_options.start_inclusive, |
| multi_get_options.stop_inclusive, |
| multi_get_options.sortkey_filter_type, |
| blob(multi_get_options.sortkey_filter_pattern), |
| multi_get_options.reverse) |
| op = RrdbMultiGetOperator(peer_gpid, req, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def get_sort_keys(self, hash_key, |
| max_kv_count=100, |
| max_kv_size=1000000, |
| timeout=0): |
| """ |
| Get multiple sort keys under hash_key. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param max_kv_count: (int) max count of k-v pairs to be fetched. max_fetch_count <= 0 means no limit. |
| :param max_kv_size: (int) max total data size of k-v pairs to be fetched. max_fetch_size <= 0 means no limit. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, set>) (code, ks) |
| code: error_types.ERR_OK.value when data got succeed. |
| ks: <sort_key, ign> pairs in dict, ign will always be empty str. |
| """ |
| return self.multi_get(hash_key, None, |
| max_kv_count, max_kv_size, |
| True, timeout) |
| |
| def multi_del(self, hash_key, sortkey_set, timeout=0): |
| """ |
| Remove multiple entire <hash_key, sort_key>-values in pegasus. |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param sortkey_set: (set) sort keys in set. |
| :param timeout: (int) how long will the operation timeout in milliseconds. |
| if timeout > 0, it is a timeout value for current operation, |
| else the timeout value specified to create the instance will be used. |
| :return: (tuple<error_types.code.value, int>) (code, count). |
| code: error_types.ERR_OK.value when data got succeed. |
| count: count of deleted k-v pairs. |
| """ |
| partition_hash = self.table.get_hashkey_hash(hash_key) |
| peer_gpid = self.table.get_gpid_by_hash(partition_hash) |
| session = self.table.get_session(peer_gpid) |
| ks = [] |
| if isinstance(sortkey_set, set): |
| ks = [blob(k) for k in sortkey_set] |
| else: |
| return error_types.ERR_INVALID_PARAMETERS.value, 0 |
| |
| req = multi_remove_request(blob(hash_key), ks) # 100 limit? |
| op = RrdbMultiRemoveOperator(peer_gpid, req, partition_hash) |
| if not session or not op: |
| return error_types.ERR_INVALID_STATE.value, 0 |
| |
| return session.operate(op, timeout) |
| |
| def get_scanner(self, hash_key, |
| start_sort_key, stop_sort_key, |
| scan_options): |
| """ |
| Get scanner for hash_key, start from start_sort_key, and stop at stop_sort_key. |
| Whether the scanner include the start_sort_key and stop_sort_key is configurable by scan_options |
| |
| :param hash_key: (bytes) which hash key used for this API. |
| :param start_sort_key: (bytes) returned scanner is start from start_sort_key. |
| :param stop_sort_key: (bytes) returned scanner is stop at stop_sort_key. |
| :param scan_options: (ScanOptions) configurable scan options. |
| :return: (PegasusScanner) scanner, instance of PegasusScanner. |
| """ |
| start_key = self.generate_key(hash_key, start_sort_key) |
| stop_key, stop_inclusive = self.generate_stop_key(hash_key, stop_sort_key) |
| if not stop_inclusive: |
| scan_options.stop_inclusive = stop_inclusive |
| gpid_list = [] |
| hash_list = [] |
| r = bytes_cmp(start_key.data, stop_key.data) |
| if r < 0 or \ |
| (r == 0 and scan_options.start_inclusive and scan_options.stop_inclusive): |
| partition_hash = self.table.get_blob_hash(start_key) |
| gpid_list.append(self.table.get_gpid_by_hash(partition_hash)) |
| hash_list.append(partition_hash) |
| |
| return PegasusScanner(self.table, gpid_list, scan_options, hash_list, False, start_key, stop_key) |
| |
| def get_unordered_scanners(self, max_split_count, scan_options): |
| """ |
| Get scanners for the whole pegasus table. |
| |
| :param max_split_count: (int) max count of scanners will be returned. |
| :param scan_options: (ScanOptions) configurable scan options. |
| :return: (list) instance of PegasusScanner list. |
| each scanner in this list can scan separate part of the whole pegasus table. |
| """ |
| if max_split_count <= 0: |
| return None |
| |
| all_gpid_list = self.table.get_all_gpid() |
| split = min(len(all_gpid_list), max_split_count) |
| count = len(all_gpid_list) |
| size = count // split |
| more = count % split |
| |
| opt = ScanOptions() |
| opt.timeout_millis = scan_options.timeout_millis |
| opt.batch_size = scan_options.batch_size |
| opt.snapshot = scan_options.snapshot |
| scanner_list = [] |
| for i in range(split): |
| gpid_list = [] |
| hash_list = [] |
| s = i < more and size + 1 or size |
| for j in range(s): |
| if all_gpid_list: |
| count -= 1 |
| gpid_list.append(all_gpid_list[count]) |
| hash_list.append(int(count)) |
| |
| scanner_list.append(PegasusScanner(self.table, gpid_list, opt, hash_list, True)) |
| |
| return scanner_list |
