files: import array
  import datetime
  import itertools
  import logging
  import md5
  import random
  import sys
  import threading
  import time
  
  import dbconstants
  import helper_functions
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from .dbconstants import APP_ENTITY_SCHEMA
  from .dbconstants import ID_KEY_LENGTH
  from .dbconstants import MAX_TX_DURATION
  from .dbconstants import TimeoutError
  from .cassandra_env import cassandra_interface
  from .cassandra_env.entity_id_allocator import EntityIDAllocator
  from .cassandra_env.entity_id_allocator import ScatteredAllocator
  from .cassandra_env.utils import deletions_for_entity
  from .cassandra_env.utils import mutations_for_entity
  from .utils import clean_app_id
  from .utils import encode_entity_table_key
  from .utils import encode_index_pb
  from .utils import encode_path_from_filter
  from .utils import get_composite_index_keys
  from .utils import get_entity_key
  from .utils import get_entity_kind
  from .utils import get_index_key_from_params
  from .utils import get_kind_key
  from .utils import group_for_key
  from .utils import reference_property_to_reference
  from .utils import UnprocessedQueryCursor
  from .zkappscale import entity_lock
  from .zkappscale import zktransaction
  from .zkappscale.transaction_manager import TransactionManager
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api import datastore_errors
  from google.appengine.api.datastore_distributed import _MAX_ACTIONS_PER_TXN
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.api.taskqueue.taskqueue_distributed import\
    TaskQueueServiceStub
  from google.appengine.datastore import appscale_stub_util
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_index
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import sortable_pb_encoder
  from google.appengine.datastore.datastore_stub_util import IdToCounter
  from google.appengine.datastore.datastore_stub_util import SEQUENTIAL
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext import db
  from google.appengine.ext.db.metadata import Namespace
  from google.net.proto.ProtocolBuffer import ProtocolBufferDecodeError
  
  
  class DatastoreDistributed():
    """ AppScale persistent layer for the datastore API. It is the 
        replacement for the AppServers to persist their data into 
        a distributed datastore instead of a flat file.
    """
    # Max number of results for a query
    _MAXIMUM_RESULTS = 10000
  
    # The number of entries looked at when doing a composite query
    # It will keep looking at this size window when getting the result
    _MAX_COMPOSITE_WINDOW = 10000
  
    # Maximum amount of filter and orderings allowed within a query
    _MAX_QUERY_COMPONENTS = 63
  
    # For enabling and disabling range inclusivity
    _ENABLE_INCLUSIVITY = True
    _DISABLE_INCLUSIVITY = False
  
    # Delimiter between app names and namespace and the rest of an entity key
    _NAMESPACE_SEPARATOR = dbconstants.KEY_DELIMITER
  
    # Delimiter between parameters in index keys.
    _SEPARATOR = dbconstants.KEY_DELIMITER
  
    # This is the terminating string for range queries
    _TERM_STRING = dbconstants.TERMINATING_STRING
  
    # Smallest possible value that is considered non-null and indexable.
    MIN_INDEX_VALUE = '\x01'
  
    # When assigning the first allocated ID, give this value
    _FIRST_VALID_ALLOCATED_ID = 1
  
    # The key we use to lock for allocating new IDs
    _ALLOCATE_ROOT_KEY = "__allocate__"
  
    # Number of times to retry acquiring a lock for non transactions.
    NON_TRANS_LOCK_RETRY_COUNT = 5
  
    # How long to wait before retrying to grab a lock
    LOCK_RETRY_TIME = .5
  
    # Maximum number of allowed composite indexes any one application can
    # register.
    _MAX_NUM_INDEXES = dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES
  
    # The position of the prop name when splitting an index entry by the
    # delimiter.
    PROP_NAME_IN_SINGLE_PROP_INDEX = 3
  
    # The cassandra index column that stores the reference to the entity.
    INDEX_REFERENCE_COLUMN = 'reference'
  
    # The number of entities to fetch at a time when updating indices.
    BATCH_SIZE = 100
  
    def __init__(self, datastore_batch, transaction_manager, zookeeper=None,
                 log_level=logging.INFO):
      """
         Constructor.
       
       Args:
         datastore_batch: A reference to the batch datastore interface.
         zookeeper: A reference to the zookeeper interface.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
  
      assert datastore_batch.valid_data_version()
  
      self.logger.info('Starting {}'.format(class_name))
  
      # datastore accessor used by this class to do datastore operations.
      self.datastore_batch = datastore_batch 
  
      # zookeeper instance for accesing ZK functionality.
      self.zookeeper = zookeeper
  
      # Maintain a stub object for each project using transactional tasks.
      self.taskqueue_stubs = {}
  
      # Maintain a scattered allocator for each project.
      self.scattered_allocators = {}
  
      # Maintain a sequential allocator for each project.
      self.sequential_allocators = {}
  
      self.transaction_manager = transaction_manager
      self.zookeeper.handle.add_listener(self._zk_state_listener)
  
    def get_limit(self, query):
      """ Returns the limit that should be used for the given query.
    
      Args:
        query: A datastore_pb.Query.
      Returns:
        An int, the limit to be used when accessing the datastore.
      """
      limit = self._MAXIMUM_RESULTS
      if query.has_count():
        limit = min(query.count(), self._MAXIMUM_RESULTS)
  
      if query.has_limit():
        limit = min(query.limit(), limit)
  
      if query.has_offset():
        limit = limit + min(query.offset(), self._MAXIMUM_RESULTS)
      # We can not scan with 0 or less, hence we set it to one.
      if limit <= 0:
        limit = 1
      return limit
    
    @staticmethod
    def __decode_index_str(value, prop_value):
      """ Takes an encoded string and converts it to a PropertyValue.
  
      Args:
        value: An encoded str.
        prop_value: PropertyValue to fill in.
      """
      value = str(value).replace('\x01\x01', '\x00').replace('\x01\x02', '\x01')
      decoded_value = sortable_pb_encoder.Decoder(
        array.array('B', str(value)))
      prop_value.Merge(decoded_value)
  
    @staticmethod
    def validate_app_id(app_id):
      """ Verify that this is the stub for app_id.
  
      Args:
        app_id: An application ID.
      Raises:
        AppScaleBadArg: If the application id is not set.
      """
      if not app_id: 
        raise dbconstants.AppScaleBadArg("Application name must be set")
  
    @staticmethod
    def validate_key(key):
      """ Validate this key by checking to see if it has a name or id.
  
      Args:
        key: entity_pb.Reference
      Raises:
        datastore_errors.BadRequestError: if the key is invalid
        TypeError: if key is not of entity_pb.Reference
      """
  
      if not isinstance(key, entity_pb.Reference): 
        raise TypeError("Expected type Reference")
  
      DatastoreDistributed.validate_app_id(key.app())
  
      for elem in key.path().element_list():
        if elem.has_id() and elem.has_name():
          raise datastore_errors.BadRequestError(
              'Each key path element should have id or name but not both: {0}' \
              .format(key))
  
    def get_table_prefix(self, data):
      """ Returns the namespace prefix for a query.
  
      Args:
        data: An Entity, Key or Query PB, or an (app_id, ns) tuple.
      Returns:
        A valid table prefix.
      """
      if isinstance(data, entity_pb.EntityProto):
        app_id = clean_app_id(data.key().app())
        namespace = data.key().name_space()
      elif isinstance(data, tuple):
        app_id = data[0]
        namespace = data[1]
      else:
        app_id
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/datastore_distributed.py
files: import array
  import datetime
  import itertools
  import logging
  import md5
  import random
  import sys
  import threading
  import time
  
  import dbconstants
  import helper_functions
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from .dbconstants import APP_ENTITY_SCHEMA
  from .dbconstants import ID_KEY_LENGTH
  from .dbconstants import MAX_TX_DURATION
  from .cassandra_env import cassandra_interface
  from .cassandra_env.entity_id_allocator import EntityIDAllocator
  from .cassandra_env.entity_id_allocator import ScatteredAllocator
  from .cassandra_env.utils import deletions_for_entity
  from .cassandra_env.utils import mutations_for_entity
  from .utils import clean_app_id
  from .utils import encode_entity_table_key
  from .utils import encode_index_pb
  from .utils import encode_path_from_filter
  from .utils import get_composite_index_keys
  from .utils import get_entity_key
  from .utils import get_entity_kind
  from .utils import get_index_key_from_params
  from .utils import get_kind_key
  from .utils import group_for_key
  from .utils import reference_property_to_reference
  from .utils import UnprocessedQueryCursor
  from .zkappscale import entity_lock
  from .zkappscale import zktransaction
  from .zkappscale.transaction_manager import TransactionManager
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api import datastore_errors
  from google.appengine.api.datastore_distributed import _MAX_ACTIONS_PER_TXN
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.api.taskqueue.taskqueue_distributed import\
    TaskQueueServiceStub
  from google.appengine.datastore import appscale_stub_util
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_index
  from google.appengine.datastore import entity_pb
  from google.appengine.datastore import sortable_pb_encoder
  from google.appengine.datastore.datastore_stub_util import IdToCounter
  from google.appengine.datastore.datastore_stub_util import SEQUENTIAL
  from google.appengine.runtime import apiproxy_errors
  from google.appengine.ext import db
  from google.appengine.ext.db.metadata import Namespace
  from google.net.proto.ProtocolBuffer import ProtocolBufferDecodeError
  
  
  class DatastoreDistributed():
    """ AppScale persistent layer for the datastore API. It is the 
        replacement for the AppServers to persist their data into 
        a distributed datastore instead of a flat file.
    """
    # Max number of results for a query
    _MAXIMUM_RESULTS = 10000
  
    # The number of entries looked at when doing a composite query
    # It will keep looking at this size window when getting the result
    _MAX_COMPOSITE_WINDOW = 10000
  
    # Maximum amount of filter and orderings allowed within a query
    _MAX_QUERY_COMPONENTS = 63
  
    # For enabling and disabling range inclusivity
    _ENABLE_INCLUSIVITY = True
    _DISABLE_INCLUSIVITY = False
  
    # Delimiter between app names and namespace and the rest of an entity key
    _NAMESPACE_SEPARATOR = dbconstants.KEY_DELIMITER
  
    # Delimiter between parameters in index keys.
    _SEPARATOR = dbconstants.KEY_DELIMITER
  
    # This is the terminating string for range queries
    _TERM_STRING = dbconstants.TERMINATING_STRING
  
    # Smallest possible value that is considered non-null and indexable.
    MIN_INDEX_VALUE = '\x01'
  
    # When assigning the first allocated ID, give this value
    _FIRST_VALID_ALLOCATED_ID = 1
  
    # The key we use to lock for allocating new IDs
    _ALLOCATE_ROOT_KEY = "__allocate__"
  
    # Number of times to retry acquiring a lock for non transactions.
    NON_TRANS_LOCK_RETRY_COUNT = 5
  
    # How long to wait before retrying to grab a lock
    LOCK_RETRY_TIME = .5
  
    # Maximum number of allowed composite indexes any one application can
    # register.
    _MAX_NUM_INDEXES = dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES
  
    # The position of the prop name when splitting an index entry by the
    # delimiter.
    PROP_NAME_IN_SINGLE_PROP_INDEX = 3
  
    # The cassandra index column that stores the reference to the entity.
    INDEX_REFERENCE_COLUMN = 'reference'
  
    # The number of entities to fetch at a time when updating indices.
    BATCH_SIZE = 100
  
    def __init__(self, datastore_batch, transaction_manager, zookeeper=None,
                 log_level=logging.INFO):
      """
         Constructor.
       
       Args:
         datastore_batch: A reference to the batch datastore interface.
         zookeeper: A reference to the zookeeper interface.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
  
      assert datastore_batch.valid_data_version()
  
      self.logger.info('Starting {}'.format(class_name))
  
      # datastore accessor used by this class to do datastore operations.
      self.datastore_batch = datastore_batch 
  
      # zookeeper instance for accesing ZK functionality.
      self.zookeeper = zookeeper
  
      # Maintain a stub object for each project using transactional tasks.
      self.taskqueue_stubs = {}
  
      # Maintain a scattered allocator for each project.
      self.scattered_allocators = {}
  
      # Maintain a sequential allocator for each project.
      self.sequential_allocators = {}
  
      self.transaction_manager = transaction_manager
      self.zookeeper.handle.add_listener(self._zk_state_listener)
  
    def get_limit(self, query):
      """ Returns the limit that should be used for the given query.
    
      Args:
        query: A datastore_pb.Query.
      Returns:
        An int, the limit to be used when accessing the datastore.
      """
      limit = self._MAXIMUM_RESULTS
      if query.has_count():
        limit = min(query.count(), self._MAXIMUM_RESULTS)
  
      if query.has_limit():
        limit = min(query.limit(), limit)
  
      if query.has_offset():
        limit = limit + min(query.offset(), self._MAXIMUM_RESULTS)
      # We can not scan with 0 or less, hence we set it to one.
      if limit <= 0:
        limit = 1
      return limit
    
    @staticmethod
    def __decode_index_str(value, prop_value):
      """ Takes an encoded string and converts it to a PropertyValue.
  
      Args:
        value: An encoded str.
        prop_value: PropertyValue to fill in.
      """
      value = str(value).replace('\x01\x01', '\x00').replace('\x01\x02', '\x01')
      decoded_value = sortable_pb_encoder.Decoder(
        array.array('B', str(value)))
      prop_value.Merge(decoded_value)
  
    @staticmethod
    def validate_app_id(app_id):
      """ Verify that this is the stub for app_id.
  
      Args:
        app_id: An application ID.
      Raises:
        AppScaleBadArg: If the application id is not set.
      """
      if not app_id: 
        raise dbconstants.AppScaleBadArg("Application name must be set")
  
    @staticmethod
    def validate_key(key):
      """ Validate this key by checking to see if it has a name or id.
  
      Args:
        key: entity_pb.Reference
      Raises:
        datastore_errors.BadRequestError: if the key is invalid
        TypeError: if key is not of entity_pb.Reference
      """
  
      if not isinstance(key, entity_pb.Reference): 
        raise TypeError("Expected type Reference")
  
      DatastoreDistributed.validate_app_id(key.app())
  
      for elem in key.path().element_list():
        if elem.has_id() and elem.has_name():
          raise datastore_errors.BadRequestError(
              'Each key path element should have id or name but not both: {0}' \
              .format(key))
  
    def get_table_prefix(self, data):
      """ Returns the namespace prefix for a query.
  
      Args:
        data: An Entity, Key or Query PB, or an (app_id, ns) tuple.
      Returns:
        A valid table prefix.
      """
      if isinstance(data, entity_pb.EntityProto):
        app_id = clean_app_id(data.key().app())
        namespace = data.key().name_space()
      elif isinstance(data, tuple):
        app_id = data[0]
        namespace = data[1]
      else:
        app_id = clean_app_id(data.app())
        namespace
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/datastore_distributed.py
files: """
   Datastore Constants
  """
  import cassandra.cluster
  
  SECRET_LOCATION = "/etc/appscale/secret.key"
  
  ERROR_DEFAULT = "DB_ERROR:"
  NONEXISTANT_TRANSACTION = "0"
  
  # The datastore's default HTTP port.
  DEFAULT_PORT = 4000
  
  # The lowest character to separate different fields in a row key.
  KEY_DELIMITER = '\x00'
  
  # The character used to separate kinds in an ancestry.
  KIND_SEPARATOR = '\x01'
  
  # HTTP code to indicate that the request is invalid.
  HTTP_BAD_REQUEST = 400
  
  # The length of an ID string. A constant length allows lexicographic ordering.
  ID_KEY_LENGTH = 10
  
  # The character between the kind and the ID/name of an entity.
  ID_SEPARATOR = ":"
  
  # Maximum number of groups allowed in cross group transactions.
  MAX_GROUPS_FOR_XG = 25
  
  # The maximum number of composite indexes an application can have.
  MAX_NUMBER_OF_COMPOSITE_INDEXES = 1000
  
  # The maximum number of seconds a transaction can take. In GAE, transactions
  # "have a maximum duration of 60 seconds with a 10 second idle expiration time
  # after 30 seconds." The 10-second idle check is not yet implemented.
  MAX_TX_DURATION = 60
  
  # A string used to create end keys when doing range queries.
  TERMINATING_STRING = chr(255) * 500
  
  # Tombstone value for soft deletes.
  TOMBSTONE = "APPSCALE_SOFT_DELETE"
  
  TRANSIENT_CASSANDRA_ERRORS = (
    cassandra.Unavailable, cassandra.Timeout, cassandra.CoordinationFailure,
    cassandra.OperationTimedOut, cassandra.cluster.NoHostAvailable)
  
  # The database backends supported by the AppScale datastore.
  VALID_DATASTORES = ['cassandra']
  
  # Table names
  USERS_TABLE = "USERS__"
  JOURNAL_TABLE = "JOURNAL__"
  
  ASC_PROPERTY_TABLE = "ASC_PROPERTY__"
  DSC_PROPERTY_TABLE = "DSC_PROPERTY__"
  COMPOSITE_TABLE = "COMPOSITE_INDEXES__"
  APP_ID_TABLE = "APP_IDS__"
  APP_ENTITY_TABLE = "ENTITIES__"
  APP_KIND_TABLE = "KINDS__"
  METADATA_TABLE = "METADATA__"
  DATASTORE_METADATA_TABLE = "DATASTORE_METADATA__"
  SCHEMA_TABLE = '__key__'
  
  INITIAL_TABLES = [ASC_PROPERTY_TABLE,
                    DSC_PROPERTY_TABLE,
                    APP_ID_TABLE,
                    APP_ENTITY_TABLE,
                    APP_KIND_TABLE,
                    COMPOSITE_TABLE,
                    METADATA_TABLE,
                    USERS_TABLE,
                    SCHEMA_TABLE,
                    DATASTORE_METADATA_TABLE]
  
  ###########################################
  # DB schemas for version 1 of the datastore
  ###########################################
  JOURNAL_SCHEMA = [
    "Encoded_Entity"]
  
  ENTITY_TABLE_SCHEMA = [
    "Encoded_Entity",
    "Txn_Num"]
  
  ###########################################
  # DB schema for version 2 of the datastore
  ###########################################
  
  # The schema of the table which holds the encoded entities
  APP_ENTITY_SCHEMA = [
    "entity",
    "txnID"]
  
  # Index tables store references are to entity table
  PROPERTY_SCHEMA = [
    "reference" ]
  APP_ID_SCHEMA = [
    "next_id" ]
  APP_KIND_SCHEMA = [
    "reference" ]
  COMPOSITE_SCHEMA = [
    "reference" ]
  METADATA_SCHEMA = [
    "data" ]
  
  USERS_SCHEMA = [
    "email",
    "pw",
    "date_creation", 
    "date_change",
    "date_last_login",
    "applications",
    "appdrop_rem_token",
    "appdrop_rem_token_exp",
    "visit_cnt",
    "cookie",
    "cookie_ip",
    "cookie_exp",
    "cksum",
    "enabled",
    "type",
    "is_cloud_admin",
    "capabilities" ]
  
  DATASTORE_METADATA_SCHEMA = [
    "version"]
  
  # All schema information for the keyspace is stored in the schema table.
  SCHEMA_TABLE_SCHEMA = ['schema']
  
  
  class TxnActions(object):
    """ Possible values in the 'action' column of the transaction table. """
    START = 0
    GET = 1
    MUTATE = 2
    ENQUEUE_TASK = 3
  
  
  class Operations(object):
    """ Possible datastore operations on entities. """
    PUT = 'put'
    DELETE = 'delete'
  
  
  ###############################
  # Generic Datastore Exceptions
  ###############################
  class AppScaleDBError(Exception):
    """ Tossed for generic datastore errors
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
  
    def __str__(self):
      return repr(self.value)
  
  class AppScaleDBConnectionError(Exception):
    """ Tossed when there is a bad connection
    """ 
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleMisconfiguredQuery(Exception):
    """ Tossed when a query is misconfigured
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleBadArg(Exception):
    """ Bad Argument given for a function
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class BadRequest(Exception):
    """ Indicates that a client provided invalid parameters for a request. """
  
  class ConcurrentModificationException(Exception):
    """ Indicates that an entity fetched during a transaction has changed. """
    pass
  
  class InternalError(Exception):
    """ Indicates that the datastore was unable to perform an operation. """
    pass
  
  class TimeoutError(Exception):
    """ Indicates that the datastore timed out while performing an operation. """
    pass
  
  class TooManyGroupsException(Exception):
    """ Indicates that there are too many groups involved in a transaction. """
    pass
  
  class ExcessiveTasks(Exception):
    """ Indicates that there are too many tasks for a transaction. """
    pass
  
  class TxTimeoutException(Exception):
    """ Indicates that the transaction started too long ago. """
    pass
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/dbconstants.py
files: """
   Datastore Constants
  """
  import cassandra.cluster
  
  SECRET_LOCATION = "/etc/appscale/secret.key"
  
  ERROR_DEFAULT = "DB_ERROR:"
  NONEXISTANT_TRANSACTION = "0"
  
  # The datastore's default HTTP port.
  DEFAULT_PORT = 4000
  
  # The lowest character to separate different fields in a row key.
  KEY_DELIMITER = '\x00'
  
  # The character used to separate kinds in an ancestry.
  KIND_SEPARATOR = '\x01'
  
  # HTTP code to indicate that the request is invalid.
  HTTP_BAD_REQUEST = 400
  
  # The length of an ID string. A constant length allows lexicographic ordering.
  ID_KEY_LENGTH = 10
  
  # The character between the kind and the ID/name of an entity.
  ID_SEPARATOR = ":"
  
  # Maximum number of groups allowed in cross group transactions.
  MAX_GROUPS_FOR_XG = 25
  
  # The maximum number of composite indexes an application can have.
  MAX_NUMBER_OF_COMPOSITE_INDEXES = 1000
  
  # The maximum number of seconds a transaction can take. In GAE, transactions
  # "have a maximum duration of 60 seconds with a 10 second idle expiration time
  # after 30 seconds." The 10-second idle check is not yet implemented.
  MAX_TX_DURATION = 60
  
  # A string used to create end keys when doing range queries.
  TERMINATING_STRING = chr(255) * 500
  
  # Tombstone value for soft deletes.
  TOMBSTONE = "APPSCALE_SOFT_DELETE"
  
  TRANSIENT_CASSANDRA_ERRORS = (
    cassandra.Unavailable, cassandra.Timeout, cassandra.CoordinationFailure,
    cassandra.OperationTimedOut, cassandra.cluster.NoHostAvailable)
  
  # The database backends supported by the AppScale datastore.
  VALID_DATASTORES = ['cassandra']
  
  # Table names
  USERS_TABLE = "USERS__"
  JOURNAL_TABLE = "JOURNAL__"
  
  ASC_PROPERTY_TABLE = "ASC_PROPERTY__"
  DSC_PROPERTY_TABLE = "DSC_PROPERTY__"
  COMPOSITE_TABLE = "COMPOSITE_INDEXES__"
  APP_ID_TABLE = "APP_IDS__"
  APP_ENTITY_TABLE = "ENTITIES__"
  APP_KIND_TABLE = "KINDS__"
  METADATA_TABLE = "METADATA__"
  DATASTORE_METADATA_TABLE = "DATASTORE_METADATA__"
  SCHEMA_TABLE = '__key__'
  
  INITIAL_TABLES = [ASC_PROPERTY_TABLE,
                    DSC_PROPERTY_TABLE,
                    APP_ID_TABLE,
                    APP_ENTITY_TABLE,
                    APP_KIND_TABLE,
                    COMPOSITE_TABLE,
                    METADATA_TABLE,
                    USERS_TABLE,
                    SCHEMA_TABLE,
                    DATASTORE_METADATA_TABLE]
  
  ###########################################
  # DB schemas for version 1 of the datastore
  ###########################################
  JOURNAL_SCHEMA = [
    "Encoded_Entity"]
  
  ENTITY_TABLE_SCHEMA = [
    "Encoded_Entity",
    "Txn_Num"]
  
  ###########################################
  # DB schema for version 2 of the datastore
  ###########################################
  
  # The schema of the table which holds the encoded entities
  APP_ENTITY_SCHEMA = [
    "entity",
    "txnID"]
  
  # Index tables store references are to entity table
  PROPERTY_SCHEMA = [
    "reference" ]
  APP_ID_SCHEMA = [
    "next_id" ]
  APP_KIND_SCHEMA = [
    "reference" ]
  COMPOSITE_SCHEMA = [
    "reference" ]
  METADATA_SCHEMA = [
    "data" ]
  
  USERS_SCHEMA = [
    "email",
    "pw",
    "date_creation", 
    "date_change",
    "date_last_login",
    "applications",
    "appdrop_rem_token",
    "appdrop_rem_token_exp",
    "visit_cnt",
    "cookie",
    "cookie_ip",
    "cookie_exp",
    "cksum",
    "enabled",
    "type",
    "is_cloud_admin",
    "capabilities" ]
  
  DATASTORE_METADATA_SCHEMA = [
    "version"]
  
  # All schema information for the keyspace is stored in the schema table.
  SCHEMA_TABLE_SCHEMA = ['schema']
  
  
  class TxnActions(object):
    """ Possible values in the 'action' column of the transaction table. """
    START = 0
    GET = 1
    MUTATE = 2
    ENQUEUE_TASK = 3
  
  
  class Operations(object):
    """ Possible datastore operations on entities. """
    PUT = 'put'
    DELETE = 'delete'
  
  
  ###############################
  # Generic Datastore Exceptions
  ###############################
  class AppScaleDBError(Exception):
    """ Tossed for generic datastore errors
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
  
    def __str__(self):
      return repr(self.value)
  
  class AppScaleDBConnectionError(Exception):
    """ Tossed when there is a bad connection
    """ 
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleMisconfiguredQuery(Exception):
    """ Tossed when a query is misconfigured
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class AppScaleBadArg(Exception):
    """ Bad Argument given for a function
    """
    def __init__(self, value):
      Exception.__init__(self, value)
      self.value = value
    def __str__(self):
      return repr(self.value)
  
  class BadRequest(Exception):
    """ Indicates that a client provided invalid parameters for a request. """
  
  class ConcurrentModificationException(Exception):
    """ Indicates that an entity fetched during a transaction has changed. """
    pass
  
  class InternalError(Exception):
    """ Indicates that the datastore was unable to perform an operation. """
    pass
  
  class TooManyGroupsException(Exception):
    """ Indicates that there are too many groups involved in a transaction. """
    pass
  
  class ExcessiveTasks(Exception):
    """ Indicates that there are too many tasks for a transaction. """
    pass
  
  class TxTimeoutException(Exception):
    """ Indicates that the transaction started too long ago. """
    pass
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/dbconstants.py
files: # See LICENSE file
  #
  """
  This web service interfaces with the datastore. It takes protocol buffer
  requests from AppServers and responds according to the type of request its
  given (Put, Get, Delete, Query, etc).
  """
  import argparse
  import json
  import logging
  import os
  import sys
  import threading
  import time
  import tornado.httpserver
  import tornado.web
  
  from appscale.common import appscale_info
  from appscale.common.async_retrying import retry_data_watch_coroutine
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from kazoo.exceptions import NodeExistsError
  from tornado import gen
  from tornado.httpclient import AsyncHTTPClient
  from tornado.ioloop import IOLoop
  from tornado.options import options
  from .. import dbconstants
  from ..appscale_datastore_batch import DatastoreFactory
  from ..datastore_distributed import DatastoreDistributed
  from ..utils import (clean_app_id,
                       logger,
                       UnprocessedQueryResult)
  from ..zkappscale import zktransaction
  from ..zkappscale.transaction_manager import TransactionManager
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_v4_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.ext.remote_api import remote_api_pb
  
  # Global for accessing the datastore. An instance of DatastoreDistributed.
  datastore_access = None
  
  # A record of active datastore servers.
  datastore_servers = set()
  
  # The ZooKeeper path where this server registers its availability.
  server_node = None
  
  # ZooKeeper global variable for locking
  zookeeper = None
  
  # Determines whether or not to allow datastore writes. Note: After enabling,
  # datastore processes must be restarted and the groomer must be stopped.
  READ_ONLY = False
  
  # Global stats.
  STATS = {}
  
  # The ZooKeeper path where a list of active datastore servers is stored.
  DATASTORE_SERVERS_NODE = '/appscale/datastore/servers'
  
  
  class ClearHandler(tornado.web.RequestHandler):
    """ Defines what to do when the webserver receives a /clear HTTP request. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handles POST requests for clearing datastore server stats. """
      global STATS
      STATS = {}
      self.write({"message": "Statistics for this server cleared."})
      self.finish()
  
  
  class ReadOnlyHandler(tornado.web.RequestHandler):
    """ Handles requests to check or set read-only mode. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handle requests to turn read-only mode on or off. """
      global READ_ONLY
  
      payload = self.request.body
      data = json.loads(payload)
      if 'readOnly' not in data:
        self.set_status(dbconstants.HTTP_BAD_REQUEST)
  
      if data['readOnly']:
        READ_ONLY = True
        message = 'Write operations now disabled.'
      else:
        READ_ONLY = False
        message = 'Write operations now enabled.'
  
      logger.info(message)
      self.write({'message': message})
      self.finish()
  
  
  class ReserveKeysHandler(tornado.web.RequestHandler):
    """ Handles v4 AllocateIds requests from other servers. """
    def post(self):
      """ Prevents the provided IDs from being re-allocated. """
      project_id = self.request.headers['appdata']
      request = datastore_v4_pb.AllocateIdsRequest(self.request.body)
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      datastore_access.reserve_ids(project_id, ids)
  
  
  class MainHandler(tornado.web.RequestHandler):
    """
    Defines what to do when the webserver receives different types of 
    HTTP requests.
    """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    def unknown_request(self, app_id, http_request_data, pb_type):
      """ Function which handles unknown protocol buffers.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer
      Raises:
        Raises exception.
      """
      raise NotImplementedError("Unknown request of operation {0}" \
        .format(pb_type))
  
    @gen.coroutine
    def post(self):
      """ Function which handles POST requests. Data of the request is
          the request from the AppServer in an encoded protocol buffer
          format.
      """
      request = self.request
      http_request_data = request.body
      pb_type = request.headers['protocolbuffertype']
      app_data = request.headers['appdata']
      app_data = app_data.split(':')
  
      if len(app_data) == 4:
        app_id, user_email, nick_name, auth_domain = app_data
        os.environ['AUTH_DOMAIN'] = auth_domain
        os.environ['USER_EMAIL'] = user_email
        os.environ['USER_NICKNAME'] = nick_name
        os.environ['APPLICATION_ID'] = app_id
      elif len(app_data) == 1:
        app_id = app_data[0]
        os.environ['APPLICATION_ID'] = app_id
      else:
        raise gen.Return()
  
      # If the application identifier has the HRD string prepened, remove it.
      app_id = clean_app_id(app_id)
  
      if pb_type == "Request":
        yield self.remote_request(app_id, http_request_data,
                                  service_id=request.headers.get('Module'),
                                  version_id=request.headers.get('Version'))
      else:
        self.unknown_request(app_id, http_request_data, pb_type)
  
    @tornado.web.asynchronous
    def get(self):
      """ Handles get request for the web server. Returns that it is currently
          up in json.
      """
      self.write(json.dumps(STATS))
      self.finish()
  
    @gen.coroutine
    def remote_request(self, app_id, http_request_data, service_id, version_id):
      """ Receives a remote request to which it should give the correct 
          response. The http_request_data holds an encoded protocol buffer
          of a certain type. Each type has a particular response type. 
      
      Args:
        app_id: The application ID that is sending this request.
        http_request_data: Encoded protocol buffer.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      """
      apirequest = remote_api_pb.Request()
      apirequest.ParseFromString(http_request_data)
      apiresponse = remote_api_pb.Response()
      response = None
      errcode = 0
      errdetail = ""
      apperror_pb = None
      if not apirequest.has_method():
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Method was not set in request"
        apirequest.set_method("NOT_FOUND")
      if not apirequest.has_request():
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Request missing in call"
        apirequest.set_method("NOT_FOUND")
        apirequest.clear_request()
      method = apirequest.method()
      http_request_data = apirequest.request()
      start = time.time()
  
      request_log = method
      if apirequest.has_request_id():
        request_log += ': {}'.format(apirequest.request_id())
      logger.debug(request_log)
  
      if method == "Put":
        response, errcode, errdetail = self.put_request(app_id,
                                                   http_request_data)
      elif method == "Get":
        response, errcode, errdetail = self.get_request(app_id,
                                                   http_request_data)
      elif method == "Delete":
        response, errcode, errdetail = self.delete_request(app_id,
                                                      http_request_data)
      elif method == "RunQuery":
        response, errcode, errdetail = self.run_query(http_request_data)
      elif method == "BeginTransaction":
        response, errcode, errdetail = self.begin_transaction_request(
                                                        app_id, http_request_data)
      elif method == "Commit":
        response, errcode, errdetail = self.commit_transaction_request(
                                                        app_id,
                                                        http_request_data)
      elif method == "Rollback":
        response, errcode, errdetail = self.rollback_transaction_request(
                                                          app_id,
                                                          http_request_data)
      elif method == "AllocateIds":
        response, errcode,
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/scripts/datastore.py
files: # See LICENSE file
  #
  """
  This web service interfaces with the datastore. It takes protocol buffer
  requests from AppServers and responds according to the type of request its
  given (Put, Get, Delete, Query, etc).
  """
  import argparse
  import json
  import logging
  import os
  import sys
  import threading
  import time
  import tornado.httpserver
  import tornado.web
  
  from appscale.common import appscale_info
  from appscale.common.async_retrying import retry_data_watch_coroutine
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from kazoo.client import KazooState
  from kazoo.exceptions import NodeExistsError
  from tornado import gen
  from tornado.httpclient import AsyncHTTPClient
  from tornado.ioloop import IOLoop
  from tornado.options import options
  from .. import dbconstants
  from ..appscale_datastore_batch import DatastoreFactory
  from ..datastore_distributed import DatastoreDistributed
  from ..utils import (clean_app_id,
                       logger,
                       UnprocessedQueryResult)
  from ..zkappscale import zktransaction
  from ..zkappscale.transaction_manager import TransactionManager
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api import api_base_pb
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import datastore_pb
  from google.appengine.datastore import datastore_v4_pb
  from google.appengine.datastore import entity_pb
  from google.appengine.ext.remote_api import remote_api_pb
  
  # Global for accessing the datastore. An instance of DatastoreDistributed.
  datastore_access = None
  
  # A record of active datastore servers.
  datastore_servers = set()
  
  # The ZooKeeper path where this server registers its availability.
  server_node = None
  
  # ZooKeeper global variable for locking
  zookeeper = None
  
  # Determines whether or not to allow datastore writes. Note: After enabling,
  # datastore processes must be restarted and the groomer must be stopped.
  READ_ONLY = False
  
  # Global stats.
  STATS = {}
  
  # The ZooKeeper path where a list of active datastore servers is stored.
  DATASTORE_SERVERS_NODE = '/appscale/datastore/servers'
  
  
  class ClearHandler(tornado.web.RequestHandler):
    """ Defines what to do when the webserver receives a /clear HTTP request. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handles POST requests for clearing datastore server stats. """
      global STATS
      STATS = {}
      self.write({"message": "Statistics for this server cleared."})
      self.finish()
  
  
  class ReadOnlyHandler(tornado.web.RequestHandler):
    """ Handles requests to check or set read-only mode. """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    @tornado.web.asynchronous
    def post(self):
      """ Handle requests to turn read-only mode on or off. """
      global READ_ONLY
  
      payload = self.request.body
      data = json.loads(payload)
      if 'readOnly' not in data:
        self.set_status(dbconstants.HTTP_BAD_REQUEST)
  
      if data['readOnly']:
        READ_ONLY = True
        message = 'Write operations now disabled.'
      else:
        READ_ONLY = False
        message = 'Write operations now enabled.'
  
      logger.info(message)
      self.write({'message': message})
      self.finish()
  
  
  class ReserveKeysHandler(tornado.web.RequestHandler):
    """ Handles v4 AllocateIds requests from other servers. """
    def post(self):
      """ Prevents the provided IDs from being re-allocated. """
      project_id = self.request.headers['appdata']
      request = datastore_v4_pb.AllocateIdsRequest(self.request.body)
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      datastore_access.reserve_ids(project_id, ids)
  
  
  class MainHandler(tornado.web.RequestHandler):
    """
    Defines what to do when the webserver receives different types of 
    HTTP requests.
    """
    def set_default_headers(self):
      """ Instructs clients to close the connection after each response. """
      self.set_header('Connection', 'close')
  
    def unknown_request(self, app_id, http_request_data, pb_type):
      """ Function which handles unknown protocol buffers.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer
      Raises:
        Raises exception.
      """
      raise NotImplementedError("Unknown request of operation {0}" \
        .format(pb_type))
  
    @gen.coroutine
    def post(self):
      """ Function which handles POST requests. Data of the request is
          the request from the AppServer in an encoded protocol buffer
          format.
      """
      request = self.request
      http_request_data = request.body
      pb_type = request.headers['protocolbuffertype']
      app_data = request.headers['appdata']
      app_data = app_data.split(':')
  
      if len(app_data) == 4:
        app_id, user_email, nick_name, auth_domain = app_data
        os.environ['AUTH_DOMAIN'] = auth_domain
        os.environ['USER_EMAIL'] = user_email
        os.environ['USER_NICKNAME'] = nick_name
        os.environ['APPLICATION_ID'] = app_id
      elif len(app_data) == 1:
        app_id = app_data[0]
        os.environ['APPLICATION_ID'] = app_id
      else:
        raise gen.Return()
  
      # If the application identifier has the HRD string prepened, remove it.
      app_id = clean_app_id(app_id)
  
      if pb_type == "Request":
        yield self.remote_request(app_id, http_request_data,
                                  service_id=request.headers.get('Module'),
                                  version_id=request.headers.get('Version'))
      else:
        self.unknown_request(app_id, http_request_data, pb_type)
  
    @tornado.web.asynchronous
    def get(self):
      """ Handles get request for the web server. Returns that it is currently
          up in json.
      """
      self.write(json.dumps(STATS))
      self.finish()
  
    @gen.coroutine
    def remote_request(self, app_id, http_request_data, service_id, version_id):
      """ Receives a remote request to which it should give the correct 
          response. The http_request_data holds an encoded protocol buffer
          of a certain type. Each type has a particular response type. 
      
      Args:
        app_id: The application ID that is sending this request.
        http_request_data: Encoded protocol buffer.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      """
      apirequest = remote_api_pb.Request()
      apirequest.ParseFromString(http_request_data)
      apiresponse = remote_api_pb.Response()
      response = None
      errcode = 0
      errdetail = ""
      apperror_pb = None
      if not apirequest.has_method():
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Method was not set in request"
        apirequest.set_method("NOT_FOUND")
      if not apirequest.has_request():
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Request missing in call"
        apirequest.set_method("NOT_FOUND")
        apirequest.clear_request()
      method = apirequest.method()
      http_request_data = apirequest.request()
      start = time.time()
  
      request_log = method
      if apirequest.has_request_id():
        request_log += ': {}'.format(apirequest.request_id())
      logger.debug(request_log)
  
      if method == "Put":
        response, errcode, errdetail = self.put_request(app_id,
                                                   http_request_data)
      elif method == "Get":
        response, errcode, errdetail = self.get_request(app_id,
                                                   http_request_data)
      elif method == "Delete":
        response, errcode, errdetail = self.delete_request(app_id,
                                                      http_request_data)
      elif method == "RunQuery":
        response, errcode, errdetail = self.run_query(http_request_data)
      elif method == "BeginTransaction":
        response, errcode, errdetail = self.begin_transaction_request(
                                                        app_id, http_request_data)
      elif method == "Commit":
        response, errcode, errdetail = self.commit_transaction_request(
                                                        app_id,
                                                        http_request_data)
      elif method == "Rollback":
        response, errcode, errdetail = self.rollback_transaction_request(
                                                          app_id,
                                                          http_request_data)
      elif method == "AllocateIds":
        response, errcode,
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/scripts/datastore.py
files: import base64
  import uuid
  
  from kazoo.exceptions import (
    CancelledError,
    KazooException,
    LockTimeout,
    NoNodeError,
    NotEmptyError
  )
  
  from kazoo.retry import (
    ForceRetryError,
    KazooRetry,
    RetryFailedError
  )
  
  # The ZooKeeper node that contains lock entries for an entity group.
  LOCK_PATH_TEMPLATE = u'/appscale/apps/{project}/locks/{namespace}/{group}'
  
  # The number of seconds to wait for a lock before raising a timeout error.
  LOCK_TIMEOUT = 10
  
  
  def zk_group_path(key):
    """ Retrieve the ZooKeeper lock path for a given entity key.
  
    Args:
      key: An entity Reference object.
    Returns:
      A string containing the location of a ZooKeeper path.
    """
    project = key.app().decode('utf-8')
    if key.name_space():
      namespace = key.name_space().decode('utf-8')
    else:
      namespace = u':default'
  
    first_element = key.path().element(0)
    kind = first_element.type().decode('utf-8')
  
    # Differentiate between types of identifiers.
    if first_element.has_id():
      group = u'{}:{}'.format(kind, first_element.id())
    else:
      # Kazoo does not accept certain characters (eg. newlines) that Cloud
      # Datastore allows.
      encoded_id = base64.b64encode(first_element.name())
      group = u'{}::{}'.format(kind, encoded_id.decode('utf-8').rstrip('='))
  
    return LOCK_PATH_TEMPLATE.format(project=project, namespace=namespace,
                                     group=group)
  
  
  class EntityLock(object):
    """ A ZooKeeper-based entity lock that allows test-and-set operations.
  
    This is based on kazoo's lock recipe, and has been modified to lock multiple
    entity groups. This lock is not re-entrant. Repeated calls after already
    acquired will block.
    """
    _NODE_NAME = '__lock__'
  
    def __init__(self, client, keys, txid=None):
      """ Create an entity lock.
  
      Args:
        client: A kazoo client.
        keys: A list of entity Reference objects.
        txid: An integer specifying the transaction ID.
      """
      self.client = client
      self.paths = [zk_group_path(key) for key in keys]
  
      # The txid is written to the contender nodes for deadlock resolution.
      self.data = str(txid or '')
  
      self.wake_event = client.handler.event_object()
  
      # Give the contender nodes a uniquely identifiable prefix in case its
      # existence is in question.
      self.prefix = uuid.uuid4().hex + self._NODE_NAME
  
      self.create_paths = [path + '/' + self.prefix for path in self.paths]
  
      self.create_tried = False
      self.is_acquired = False
      self.cancelled = False
      self._retry = KazooRetry(max_tries=None,
                               sleep_func=client.handler.sleep_func)
      self._lock = client.handler.lock_object()
  
    def _ensure_path(self):
      """ Make sure the ZooKeeper lock paths have been created. """
      for path in self.paths:
        self.client.ensure_path(path)
  
    def cancel(self):
      """ Cancel a pending lock acquire. """
      self.cancelled = True
      self.wake_event.set()
  
    def acquire(self):
      """ Acquire the lock. By default blocks and waits forever.
  
      Returns:
        A boolean indicating whether or not the lock was acquired.
      """
  
      def _acquire_lock():
        """ Acquire a kazoo thread lock. """
        got_it = self._lock.acquire(False)
        if not got_it:
          raise ForceRetryError()
        return True
  
      retry = self._retry.copy()
      retry.deadline = LOCK_TIMEOUT
  
      # Prevent other threads from acquiring the lock at the same time.
      locked = self._lock.acquire(False)
      if not locked:
        try:
          retry(_acquire_lock)
        except RetryFailedError:
          return False
  
      already_acquired = self.is_acquired
      try:
        gotten = False
        try:
          gotten = retry(self._inner_acquire)
        except RetryFailedError:
          if not already_acquired:
            self._best_effort_cleanup()
        except KazooException:
          if not already_acquired:
            self._best_effort_cleanup()
            self.cancelled = False
          raise
        if gotten:
          self.is_acquired = gotten
        if not gotten and not already_acquired:
          self._delete_nodes(self.nodes)
        return gotten
      finally:
        self._lock.release()
  
    def _watch_session(self, state):
      """ A callback function for handling connection state changes.
  
      Args:
        state: The new connection state.
      """
      self.wake_event.set()
      return True
  
    def _resolve_deadlocks(self, children_list):
      """ Check if there are any concurrent cross-group locks.
  
      Args:
        children_list: A list of current transactions for each group.
      """
      current_txid = int(self.data)
      for index, children in enumerate(children_list):
        our_index = children.index(self.nodes[index])
  
        # Skip groups where this lock already has the earliest contender.
        if our_index == 0:
          continue
  
        # Get transaction IDs for earlier contenders.
        for child in children[:our_index - 1]:
          try:
            data, _ = self.client.get(
              self.paths[index] + '/' + child)
          except NoNodeError:
            continue
  
          # If data is not set, it doesn't belong to a cross-group
          # transaction.
          if not data:
            continue
  
          child_txid = int(data)
          # As an arbitrary rule, require later transactions to
          # resolve deadlocks.
          if current_txid > child_txid:
            # TODO: Implement a more graceful deadlock detection.
            self.client.retry(self._delete_nodes(self.nodes))
            raise ForceRetryError()
  
    def _inner_acquire(self):
      """ Create contender node(s) and wait until the lock is acquired. """
  
      # Make sure the group lock node exists.
      self._ensure_path()
  
      nodes = [None for _ in self.paths]
      if self.create_tried:
        nodes = self._find_nodes()
      else:
        self.create_tried = True
  
      for index, node in enumerate(nodes):
        if node is not None:
          continue
  
        # The entity group lock root may have been deleted, so try a few times.
        try_num = 0
        while True:
          try:
            node = self.client.create(
              self.create_paths[index], self.data, sequence=True)
            break
          except NoNodeError:
            self.client.ensure_path(self.paths[index])
            if try_num > 3:
              raise ForceRetryError()
          try_num += 1
  
        # Strip off path to node.
        node = node[len(self.paths[index]) + 1:]
        nodes[index] = node
  
      self.nodes = nodes
  
      while True:
        self.wake_event.clear()
  
        # Bail out with an exception if cancellation has been requested.
        if self.cancelled:
          raise CancelledError()
  
        children_list = self._get_sorted_children()
  
        predecessors = []
        for index, children in enumerate(children_list):
          try:
            our_index = children.index(nodes[index])
          except ValueError:
            raise ForceRetryError()
  
          # If the lock for this group hasn't been acquired, get the predecessor.
          if our_index != 0:
            predecessors.append(
              self.paths[index] + "/" + children[our_index - 1])
  
        if not predecessors:
          return True
  
        if len(nodes) > 1:
          self._resolve_deadlocks(children_list)
  
        # Wait for predecessor to be removed.
        # TODO: Listen for all at the same time.
        for index, predecessor in enumerate(predecessors):
          self.client.add_listener(self._watch_session)
          try:
            if self.client.exists(predecessor, self._watch_predecessor):
              self.wake_event.wait(LOCK_TIMEOUT)
              if not self.wake_event.isSet():
                error = 'Failed to acquire lock on {} after {} '\
                  'seconds'.format(self.paths, LOCK_TIMEOUT * (index + 1))
                raise LockTimeout(error)
          finally:
            self.client.remove_listener(self._watch_session)
  
    def _watch_predecessor(self, event):
      """ A callback function for handling contender deletions.
  
      Args:
        event: A ZooKeeper event.
      """
      self.wake_event.set()
  
    def _get_sorted_children(self):
      """ Retrieve a list of sorted contenders for each group.
  
      Returns:
        A list of contenders for each group.
      """
      children = []
      for path in self.paths:
        try:
          children.append(self.client.get_children(path))
        except NoNodeError:
          children.append([])
  
     
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/zkappscale/entity_lock.py
files: import base64
  import uuid
  
  from kazoo.exceptions import (
    CancelledError,
    KazooException,
    LockTimeout,
    NoNodeError,
    NotEmptyError
  )
  
  from kazoo.retry import (
    ForceRetryError,
    KazooRetry,
    RetryFailedError
  )
  
  # The ZooKeeper node that contains lock entries for an entity group.
  LOCK_PATH_TEMPLATE = u'/appscale/apps/{project}/locks/{namespace}/{group}'
  
  # The number of seconds to wait for a lock before raising a timeout error.
  LOCK_TIMEOUT = 10
  
  
  def zk_group_path(key):
    """ Retrieve the ZooKeeper lock path for a given entity key.
  
    Args:
      key: An entity Reference object.
    Returns:
      A string containing the location of a ZooKeeper path.
    """
    project = key.app().decode('utf-8')
    if key.name_space():
      namespace = key.name_space().decode('utf-8')
    else:
      namespace = u':default'
  
    first_element = key.path().element(0)
    kind = first_element.type().decode('utf-8')
  
    # Differentiate between types of identifiers.
    if first_element.has_id():
      group = u'{}:{}'.format(kind, first_element.id())
    else:
      # Kazoo does not accept certain characters (eg. newlines) that Cloud
      # Datastore allows.
      encoded_id = base64.b64encode(first_element.name())
      group = u'{}::{}'.format(kind, encoded_id.decode('utf-8').rstrip('='))
  
    return LOCK_PATH_TEMPLATE.format(project=project, namespace=namespace,
                                     group=group)
  
  
  class EntityLock(object):
    """ A ZooKeeper-based entity lock that allows test-and-set operations.
  
    This is based on kazoo's lock recipe, and has been modified to lock multiple
    entity groups. This lock is not re-entrant. Repeated calls after already
    acquired will block.
    """
    _NODE_NAME = '__lock__'
  
    def __init__(self, client, keys, txid=None):
      """ Create an entity lock.
  
      Args:
        client: A kazoo client.
        keys: A list of entity Reference objects.
        txid: An integer specifying the transaction ID.
      """
      self.client = client
      self.paths = [zk_group_path(key) for key in keys]
  
      # The txid is written to the contender nodes for deadlock resolution.
      self.data = str(txid or '')
  
      self.wake_event = client.handler.event_object()
  
      # Give the contender nodes a uniquely identifiable prefix in case its
      # existence is in question.
      self.prefix = uuid.uuid4().hex + self._NODE_NAME
  
      self.create_paths = [path + '/' + self.prefix for path in self.paths]
  
      self.create_tried = False
      self.is_acquired = False
      self.cancelled = False
      self._retry = KazooRetry(max_tries=None,
                               sleep_func=client.handler.sleep_func)
      self._lock = client.handler.lock_object()
  
    def _ensure_path(self):
      """ Make sure the ZooKeeper lock paths have been created. """
      for path in self.paths:
        self.client.ensure_path(path)
  
    def cancel(self):
      """ Cancel a pending lock acquire. """
      self.cancelled = True
      self.wake_event.set()
  
    def acquire(self):
      """ Acquire the lock. By default blocks and waits forever.
  
      Returns:
        A boolean indicating whether or not the lock was acquired.
      """
  
      def _acquire_lock():
        """ Acquire a kazoo thread lock. """
        got_it = self._lock.acquire(False)
        if not got_it:
          raise ForceRetryError()
        return True
  
      retry = self._retry.copy()
      retry.deadline = LOCK_TIMEOUT
  
      # Prevent other threads from acquiring the lock at the same time.
      locked = self._lock.acquire(False)
      if not locked:
        try:
          retry(_acquire_lock)
        except RetryFailedError:
          return False
  
      already_acquired = self.is_acquired
      try:
        gotten = False
        try:
          gotten = retry(self._inner_acquire)
        except RetryFailedError:
          if not already_acquired:
            self._best_effort_cleanup()
        except KazooException:
          if not already_acquired:
            self._best_effort_cleanup()
            self.cancelled = False
          raise
        if gotten:
          self.is_acquired = gotten
        if not gotten and not already_acquired:
          self._delete_nodes(self.nodes)
        return gotten
      finally:
        self._lock.release()
  
    def _watch_session(self, state):
      """ A callback function for handling connection state changes.
  
      Args:
        state: The new connection state.
      """
      self.wake_event.set()
      return True
  
    def _resolve_deadlocks(self, children_list):
      """ Check if there are any concurrent cross-group locks.
  
      Args:
        children_list: A list of current transactions for each group.
      """
      current_txid = int(self.data)
      for index, children in enumerate(children_list):
        our_index = children.index(self.nodes[index])
  
        # Skip groups where this lock already has the earliest contender.
        if our_index == 0:
          continue
  
        # Get transaction IDs for earlier contenders.
        for child in children[:our_index - 1]:
          try:
            data, _ = self.client.get(
              self.paths[index] + '/' + child)
          except NoNodeError:
            continue
  
          # If data is not set, it doesn't belong to a cross-group
          # transaction.
          if not data:
            continue
  
          child_txid = int(data)
          # As an arbitrary rule, require later transactions to
          # resolve deadlocks.
          if current_txid > child_txid:
            # TODO: Implement a more graceful deadlock detection.
            self.client.retry(self._delete_nodes(self.nodes))
            raise ForceRetryError()
  
    def _inner_acquire(self):
      """ Create contender node(s) and wait until the lock is acquired. """
  
      # Make sure the group lock node exists.
      self._ensure_path()
  
      nodes = [None for _ in self.paths]
      if self.create_tried:
        nodes = self._find_nodes()
      else:
        self.create_tried = True
  
      for index, node in enumerate(nodes):
        if node is not None:
          continue
  
        # The entity group lock root may have been deleted, so try a few times.
        try_num = 0
        while True:
          try:
            node = self.client.create(
              self.create_paths[index], self.data, ephemeral=True,
              sequence=True)
            break
          except NoNodeError:
            self.client.ensure_path(self.paths[index])
            if try_num > 3:
              raise ForceRetryError()
          try_num += 1
  
        # Strip off path to node.
        node = node[len(self.paths[index]) + 1:]
        nodes[index] = node
  
      self.nodes = nodes
  
      while True:
        self.wake_event.clear()
  
        # Bail out with an exception if cancellation has been requested.
        if self.cancelled:
          raise CancelledError()
  
        children_list = self._get_sorted_children()
  
        predecessors = []
        for index, children in enumerate(children_list):
          try:
            our_index = children.index(nodes[index])
          except ValueError:
            raise ForceRetryError()
  
          # If the lock for this group hasn't been acquired, get the predecessor.
          if our_index != 0:
            predecessors.append(
              self.paths[index] + "/" + children[our_index - 1])
  
        if not predecessors:
          return True
  
        if len(nodes) > 1:
          self._resolve_deadlocks(children_list)
  
        # Wait for predecessor to be removed.
        # TODO: Listen for all at the same time.
        for index, predecessor in enumerate(predecessors):
          self.client.add_listener(self._watch_session)
          try:
            if self.client.exists(predecessor, self._watch_predecessor):
              self.wake_event.wait(LOCK_TIMEOUT)
              if not self.wake_event.isSet():
                error = 'Failed to acquire lock on {} after {} '\
                  'seconds'.format(self.paths, LOCK_TIMEOUT * (index + 1))
                raise LockTimeout(error)
          finally:
            self.client.remove_listener(self._watch_session)
  
    def _watch_predecessor(self, event):
      """ A callback function for handling contender deletions.
  
      Args:
        event: A ZooKeeper event.
      """
      self.wake_event.set()
  
    def _get_sorted_children(self):
      """ Retrieve a list of sorted contenders for each group.
  
      Returns:
        A list of contenders for each group.
      """
      children = []
      for path in self.paths:
        try:
          children.append(self.client.get_children(path))
        except NoNodeError:
         
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/zkappscale/entity_lock.py