pgclient.py

from __future__ import print_function
from __future__ import with_statement

import logging.config

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 base.ttypes import *
from operate.packet import *
from replication.ttypes import *
from rrdb.ttypes import *
from utils.tools import *
from transport.protocol import TPegasusThriftClientProtocol

try:
    from thrift.protocol import fastbinary
except:
    fastbinary = None


logging.config.fileConfig("logger.conf")
logger = logging.getLogger("root")

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._reqs = {}
        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._reqs[seqid]

    def eb_send(self, f, seqid):
        d = self._reqs.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._reqs[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._reqs:
            logger.warning('peer: %s rseqid: %s not found',
                           self.get_peer_addr(), rseqid)
            return
        d = self._reqs.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):
            host_port_list = meta_addr.split(':')
            if not len(host_port_list) == 2:
                return False

            host, port = host_port_list[0], int(host_port_list[1])
            self.addr_list.append((host, port))

            return True
        else:
            return False

    @inlineCallbacks
    def query_one(self, session):
        req = query_cfg_request(self.name, [])
        op = QueryCfgOperator(gpid(0, 0), req)
        ret = yield session.operate(op)
        defer.returnValue(ret)

    def got_results(self, res):
        for (suc, result) in res:
            if suc                                                              \
               and isinstance(result, replication.ttypes.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 (host, port) in self.addr_list:
            rpc_addr = rpc_address()
            rpc_addr.from_string(host + ':' + 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(host, port, DEFAULT_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.query_cfg_response = None
        self.partition_dict = {}                                    # partition => rpc_addr
        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 not isinstance(resp, replication.ttypes.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

        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
            if rpc_addr in connected_rpc_addrs or rpc_addr.address == 0:
                continue

            host, port = rpc_addr.to_host_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(host, port, DEFAULT_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_hash_key_pid(self, hash_key):
        hash_value = PegasusHash.default_hash(hash_key)
        pidx = hash_value % self.get_partition_count()
        return gpid(self.app_id, pidx)

    def get_gpid(self, blob_key):
        hash_value = PegasusHash.hash(blob_key)
        pidx = hash_value % self.get_partition_count()
        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):

    CONTEXT_ID_VALID_MIN = 0
    CONTEXT_ID_COMPLETED = -1
    CONTEXT_ID_NOT_EXIST = -2

    def __init__(self, table, gpid_list, scan_options,
                 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 = []

    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):
        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.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
        op = RrdbGetScannerOperator(self._gpid, request)
        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)
        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)
            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 = 0x9a6c9329ac4bc9b5
    table_forward = [0] * 256

    @classmethod
    def populate_table(cls):
        for i in range(256):
            crc = i
            for j in range(8):
                if crc & 1:
                    crc >>= 1
                    crc ^= cls.polynomial
                else:
                    crc >>= 1
            cls.table_forward[i] = crc

    @classmethod
    def crc64(cls, data, offset, length):
        crc = -1
        end = offset + length
        for c in data[offset:end:1]:
            crc = cls.table_forward[(ord(c) ^ crc) & 0xFF] ^ (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):

    @classmethod
    def generate_key(cls, hash_key, sort_key):
        # assert(len(hash_key) < sys.maxsize > 1)

        # 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):
        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 host_port in meta_addrs:
                self.meta_session_manager.add_meta_server(host_port)
        PegasusHash.populate_table()
        self.timeout_times = 0
        self.update_partition = False
        self.timer = reactor.callLater(META_CHECK_INTERVAL, self.check_state)

    def init(self):
        dlist = self.meta_session_manager.query()
        dlist.addCallback(self.table.update_cfg)
        return dlist

    def close(self):
        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):
            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=DEFAULT_TIMEOUT):
        blob_key = self.generate_key(hash_key, sort_key)
        peer_gpid = self.table.get_gpid(blob_key)
        session = self.table.get_session(peer_gpid)
        op = RrdbTtlOperator(peer_gpid, blob_key)
        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=DEFAULT_TIMEOUT):
        return self.ttl(hash_key, sort_key, timeout)

    def get(self, hash_key, sort_key, timeout=DEFAULT_TIMEOUT):
        blob_key = self.generate_key(hash_key, sort_key)
        peer_gpid = self.table.get_gpid(blob_key)
        session = self.table.get_session(peer_gpid)
        op = RrdbGetOperator(peer_gpid, blob_key)
        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=DEFAULT_TIMEOUT):
        blob_key = self.generate_key(hash_key, sort_key)
        peer_gpid = self.table.get_gpid(blob_key)
        session = self.table.get_session(peer_gpid)
        op = RrdbPutOperator(peer_gpid, update_request(blob_key, blob(value), get_ttl(ttl)))
        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=DEFAULT_TIMEOUT):
        blob_key = self.generate_key(hash_key, sort_key)
        peer_gpid = self.table.get_gpid(blob_key)
        session = self.table.get_session(peer_gpid)
        op = RrdbRemoveOperator(peer_gpid, blob_key)
        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=DEFAULT_TIMEOUT):
        peer_gpid = self.table.get_hash_key_pid(hash_key)
        session = self.table.get_session(peer_gpid)
        op = RrdbSortkeyCountOperator(peer_gpid, blob(hash_key))
        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=DEFAULT_TIMEOUT):
        peer_gpid = self.table.get_hash_key_pid(hash_key)
        session = self.table.get_session(peer_gpid)
        kvs = [key_value(blob(str(k)), blob(str(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)
        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=DEFAULT_TIMEOUT):
        peer_gpid = self.table.get_hash_key_pid(hash_key)
        session = self.table.get_session(peer_gpid)
        ks = []
        if sortkey_set is None:
            pass
        elif isinstance(sortkey_set, set):
            ks = [blob(str(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)
        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=DEFAULT_TIMEOUT):
        return self.multi_get(hash_key, None,
                              max_kv_count, max_kv_size,
                              True, timeout)

    def multi_del(self, hash_key, sortkey_set, timeout=DEFAULT_TIMEOUT):
        peer_gpid = self.table.get_hash_key_pid(hash_key)
        session = self.table.get_session(peer_gpid)
        ks = []
        if isinstance(sortkey_set, set):
            ks = [blob(str(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)
        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):
        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 = []
        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):
            gpid_list.append(self.table.get_gpid(start_key))

        return PegasusScanner(self.table, gpid_list, scan_options, start_key, stop_key)

    def get_unordered_scanners(self, max_split_count, scan_options):    # max_split_count: max scanners count returned
        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 = []
            s = i < more and size + 1 or size
            for j in range(s):
                if all_gpid_list:
                    gpid_list.append(all_gpid_list.pop())

            scanner_list.append(PegasusScanner(self.table, gpid_list, opt))

        return scanner_list