files.new_contents: 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 = clean_app_id(data.app())
        namespace = data.name_space()
  
      return self._SEPARATOR.join([app_id, namespace])
  
    @staticmethod
    def get_ancestor_key_from_ent_key(ent_key):
      """ Get the key string for the ancestor portion of a composite key.
  
      Args:
        ent_key: A string of the entire path of an entity.
      Returns:
        A str of the path of the ancestor.
      """
      ancestor = ""
      tokens = str(ent_key).split(dbconstants.KIND_SEPARATOR)
      # Strip off the empty placeholder and also do not include the last kind.
      for token in tokens[:-2]:
        ancestor += token + dbconstants.KIND_SEPARATOR
      return ancestor
  
    @staticmethod
    def get_composite_index_key(index, entity, position_list=None, 
      filters=None):
      """ Creates a key to the composite index table for a given entity
      for a composite cursor.
  
      Keys are built as such: 
        app_id/ns/composite_id/ancestor/valuevaluevalue..../entity_key
      Components explained:
      ns: The namespace of the entity.
      composite_id: The composite ID assigned to this index upon creation.
      ancestor: The root ancestor path (only if the query this index is for 
        has an ancestor)
      value(s): The string representation of mulitiple properties.
      entity_key: The entity key (full path) used as a means of having a unique
        identifier. This prevents two entities with the same values from
        colliding.
  
      Args:
        index: A datstore_pb.CompositeIndex.
        entity: A entity_pb.EntityProto.
        position_list: A list of datastore_pb.CompiledCursor_Position items.
          Contains values for property items from a cursor.
        filters: A list of datastore_pb.Query_Filters, used to attain equality
          values not present in position_list.
      Returns:
        A string representing a key to the composite table.
      """ 
      composite_id = index.id()
      definition = index.definition()
      app_id = clean_app_id(entity.key().app())
      name_space = entity.key().name_space()
      ent_key = encode_index_pb(entity.key().path())
      pre_comp_index_key = "{0}{1}{2}{4}{3}{4}".format(app_id, 
        DatastoreDistributed._NAMESPACE_SEPARATOR, name_space, composite_id,
        DatastoreDistributed._SEPARATOR)
      if definition.ancestor() == 1:
        ancestor = DatastoreDistributed.get_ancestor_key_from_ent_key(ent_key)
        pre_comp_index_key += "{0}{1}".format(ancestor,
          DatastoreDistributed._SEPARATOR) 
  
      value_dict = {}
      for prop in entity.property_list():
        value_dict[prop.name()] = \
          str(encode_index_pb(prop.value()))
  
      # Position list and filters are used if we're creating a composite
      # key for a cursor.
      if position_list:
        for indexvalue in position_list[0].indexvalue_list():
          value_dict[indexvalue.property()] = \
            str(encode_index_pb(indexvalue.value()))
      if filters:
        for filt in filters:
          if filt.op() == datastore_pb.Query_Filter.EQUAL:
            value_dict[filt.property(0).name()] = \
              str(encode_index_pb(filt.property(0).value()))
  
      index_value = ""
      for prop in definition.property_list():
        name = prop.name()
        value = ''
        if name in value_dict:
          value = value_dict[name]
        elif name == "__key__":
          value = ent_key
        else:
          logging.warning("Given entity {0} is missing a property value {1}.".\
            format(entity, prop.name()))
        if prop.direction() == entity_pb.Index_Property.DESCENDING:
          value = helper_functions.reverse_lex(value)
  
        index_value += str(value) + DatastoreDistributed._SEPARATOR
  
      # We append the ent key to have unique keys if entities happen
      # to share the same index values (and ancestor).
      composite_key = "{0}{1}{2}".format(pre_comp_index_key, index_value,
        ent_key)
      return composite_key
    
    def insert_composite_indexes(self, entities, composite_indexes):
      """ Creates composite indexes for a set of entities.
  
      Args:
        entities: A list entities.
        composite_indexes: A list of datastore_pb.CompositeIndex.
      """
      if not composite_indexes:
        return
      row_keys = []
      row_values = {}
      # Create default composite index for all entities. Here we take each
      # of the properties in one 
      for ent in entities:
        for index_def in composite_indexes:
          # Skip any indexes if the kind does not match.
          kind = get_entity_kind(ent.key())
          if index_def.definition().entity_type() != kind:
            continue
  
          # Make sure the entity contains the required entities for the composite
          # definition. 
          prop_name_def_list = [index_prop.name() for index_prop in \
            index_def.definition().property_list()]
          all_prop_names_in_ent = [prop.name() for prop in \
            ent.property_list()]
          has_values = True 
          for index_prop_name in prop_name_def_list:
            if index_prop_name not in all_prop_names_in_ent:
              has_values = False
            # Special property name which does not show up in the list but 
            # is apart of the key of the entity.
            if index_prop_name == "__key__":
              has_values = True
          if not has_values:
            continue
  
          # Get the composite index key.
          composite_index_keys = get_composite_index_keys(index_def, ent)
          row_keys.extend(composite_index_keys)
  
          # Get the reference value for the composite table.
          entity_key = str(encode_index_pb(ent.key().path()))
          prefix = self.get_table_prefix(ent.key())
          reference = "{0}{1}{2}".format(prefix, self._SEPARATOR,  entity_key)
          for composite_key in composite_index_keys:
            row_values[composite_key] = {'reference': reference}
  
      self.datastore_batch.batch_put_entity(dbconstants.COMPOSITE_TABLE, 
                                            row_keys, 
                                            dbconstants.COMPOSITE_SCHEMA,
                                            row_values)
  
    def delete_composite_index_metadata(self, app_id, index):
      """ Deletes a index for the given application identifier.
    
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      """
      self.logger.info('Deleting composite index:\n{}'.format(index))
      index_keys = []
      composite_id = str(index.id())
      index_keys.append(self._SEPARATOR.join([app_id, 'index', composite_id]))
      self.datastore_batch.batch_delete(dbconstants.METADATA_TABLE,
                                        index_keys, 
                                        column_names=dbconstants.METADATA_TABLE)
  
    def create_composite_index(self, app_id, index):
      """ Stores a new index for the given application identifier.
    
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      Returns:
        A unique number representing the composite index ID.
      """
      # Generate a random number based on time of creation.
      rand = int(str(int(time.time())) + str(random.randint(0, 999999)))
      index.set_id(rand)
      encoded_entity = index.Encode()
      row_key = self._SEPARATOR.join([app_id, 'index', str(rand)])
      row_keys = [row_key]
      row_values = {}
      row_values[row_key] = {dbconstants.METADATA_SCHEMA[0]: encoded_entity}
      self.datastore_batch.batch_put_entity(dbconstants.METADATA_TABLE, 
                                            row_keys, 
                                            dbconstants.METADATA_SCHEMA, 
                                            row_values)    
      return rand 
  
    def update_composite_index(self, app_id, index):
      """ Updates an index for a given app ID.
  
      Args:
        app_id: A string containing the app ID.
        index: An entity_pb.CompositeIndex object.
      """
      self.logger.info('Updating index: {}'.format(index))
      entries_updated = 0
      entity_type = index.definition().entity_type()
  
      # TODO: Adjust prefix based on ancestor.
      prefix = '{app}{delimiter}{entity_type}{kind_separator}'.format(
        app=app_id,
        delimiter=self._SEPARATOR * 2,
        entity_type=entity_type,
        kind_separator=dbconstants.KIND_SEPARATOR,
      )
      start_row = prefix
      end_row = prefix + self._TERM_STRING
      start_inclusive = True
  
      while True:
        # Fetch references from the kind table since entity keys can have a
        # parent prefix.
        references = self.datastore_batch.range_query(
          table_name=dbconstants.APP_KIND_TABLE,
          column_names=dbconstants.APP_KIND_SCHEMA,
          start_key=start_row,
          end_key=end_row,
          limit=self.BATCH_SIZE,
          offset=0,
          start_inclusive=start_inclusive,
        )
  
        pb_entities = self.__fetch_entities(references)
        entities = [entity_pb.EntityProto(entity) for entity in pb_entities]
  
        self.insert_composite_indexes(entities, [index])
        entries_updated += len(entities)
  
        # If we fetched fewer references than we asked for, we're done.
        if len(references) < self.BATCH_SIZE:
          break
  
        start_row = references[-1].keys()[0]
        start_inclusive = self._DISABLE_INCLUSIVITY
  
      self.logger.info('Updated {} index entries.'.format(entries_updated))
  
    def allocate_size(self, project, size):
      """ Allocates a block of IDs for a project.
  
      Args:
        project: A string specifying the project ID.
        size: An integer specifying the number of IDs to reserve.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      return allocator.allocate_size(size)
  
    def allocate_max(self, project, max_id):
      """ Reserves all IDs up to the one given.
  
      Args:
        project: A string specifying the project ID.
        max_id: An integer specifying the maximum ID to allocated.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      return allocator.allocate_max(max_id)
  
    def reserve_ids(self, project_id, ids):
      """ Ensures the given IDs are not re-allocated.
  
      Args:
        project_id: A string specifying the project ID.
        ids: An iterable of integers specifying entity IDs.
      """
      if project_id not in self.sequential_allocators:
        self.sequential_allocators[project_id] = EntityIDAllocator(
          self.datastore_batch.session, project_id)
  
      if project_id not in self.scattered_allocators:
        self.scattered_allocators[project_id] = ScatteredAllocator(
          self.datastore_batch.session, project_id)
  
      for id_ in ids:
        counter, space = IdToCounter(id_)
        if space == SEQUENTIAL:
          allocator = self.sequential_allocators[project_id]
        else:
          allocator = self.scattered_allocators[project_id]
  
        allocator.set_min_counter(counter)
  
    def put_entities(self, app, entities, composite_indexes=()):
      """ Updates indexes of existing entities, inserts new entities and 
          indexes for them.
  
      Args:
        app: A string containing the application ID.
        entities: List of entities.
        composite_indexes: A list or tuple of CompositeIndex objects.
      """
      self.logger.debug('Inserting {} entities'.format(len(entities)))
      entity_keys = []
      for entity in entities:
        prefix = self.get_table_prefix(entity)
        entity_keys.append(get_entity_key(prefix, entity.key().path()))
  
      current_values = self.datastore_batch.batch_get_entity(
        dbconstants.APP_ENTITY_TABLE, entity_keys, APP_ENTITY_SCHEMA)
  
      by_group = {}
      for entity in entities:
        group_key = group_for_key(entity.key()).Encode()
        if group_key not in by_group:
          by_group[group_key] = []
        by_group[group_key].append(entity)
  
      for encoded_group_key, entity_list in by_group.iteritems():
        group_key = entity_pb.Reference(encoded_group_key)
  
        txid = self.transaction_manager.create_transaction_id(app, xg=False)
        self.transaction_manager.set_groups(app, txid, [group_key])
  
        # Allow the lock to stick around if there is an issue applying the batch.
        lock = entity_lock.EntityLock(self.zookeeper.handle, [group_key], txid)
        try:
          lock.acquire()
        except entity_lock.LockTimeout:
          raise TimeoutError('Unable to acquire entity group lock')
  
        batch = []
        entity_changes = []
        for entity in entity_list:
          prefix = self.get_table_prefix(entity)
          entity_key = get_entity_key(prefix, entity.key().path())
  
          current_value = None
          if current_values[entity_key]:
            current_value = entity_pb.EntityProto(
              current_values[entity_key][APP_ENTITY_SCHEMA[0]])
  
          batch.extend(mutations_for_entity(entity, txid, current_value,
                                            composite_indexes))
  
          batch.append({'table': 'group_updates',
                        'key': bytearray(encoded_group_key),
                        'last_update': txid})
  
          entity_changes.append(
            {'key': entity.key(), 'old': current_value, 'new': entity})
        self.datastore_batch.batch_mutate(app, batch, entity_changes, txid)
  
        lock.release()
        self.transaction_manager.delete_transaction_id(app, txid)
  
    def delete_entities(self, group, txid, keys, composite_indexes=()):
      """ Deletes the entities and the indexes associated with them.
  
      Args:
        group: An entity group Reference object.
        txid: An integer specifying a transaction ID.
        keys: An interable containing entity Reference objects.
        composite_indexes: A list or tuple of CompositeIndex objects.
      """
      entity_keys = []
      for key in keys:
        prefix = self.get_table_prefix(key)
        entity_keys.append(get_entity_key(prefix, key.path()))
  
      # Must fetch the entities to get the keys of indexes before deleting.
      current_values = self.datastore_batch.batch_get_entity(
        dbconstants.APP_ENTITY_TABLE, entity_keys, APP_ENTITY_SCHEMA)
  
      for key in entity_keys:
        if not current_values[key]:
          continue
  
        current_value = entity_pb.EntityProto(
          current_values[key][APP_ENTITY_SCHEMA[0]])
        batch = deletions_for_entity(current_value, composite_indexes)
  
        batch.append({'table': 'group_updates',
                      'key': bytearray(group.Encode()),
                      'last_update': txid})
  
        self.datastore_batch._normal_batch(batch, txid)
  
    def dynamic_put(self, app_id, put_request, put_response):
      """ Stores and entity and its indexes in the datastore.
      
      Args:
        app_id: Application ID.
        put_request: Request with entities to store.
        put_response: The response sent back to the app server.
      Raises:
        ZKTransactionException: If we are unable to acquire/release ZooKeeper locks.
      """
      if app_id not in self.scattered_allocators:
        self.scattered_allocators[app_id] = ScatteredAllocator(
          self.datastore_batch.session, app_id)
      allocator = self.scattered_allocators[app_id]
  
      entities = put_request.entity_list()
  
      for entity in entities:
        self.validate_key(entity.key())
  
        for prop in itertools.chain(entity.property_list(),
                                    entity.raw_property_list()):
          if prop.value().has_uservalue():
            uid = md5.new(prop.value().uservalue().email().lower()).digest()
            uid = '1' + ''.join(['%02d' % ord(x) for x in uid])[:20]
            prop.mutable_value().mutable_uservalue().set_obfuscated_gaiaid(uid)
  
        last_path = entity.key().path().element_list()[-1]
        if last_path.id() == 0 and not last_path.has_name():
          last_path.set_id(allocator.next())
          group = entity.mutable_entity_group()
          root = entity.key().path().element(0)
          group.add_element().CopyFrom(root)
  
      if put_request.has_transaction():
        self.datastore_batch.put_entities_tx(
          app_id, put_request.transaction().handle(), entities)
      else:
        self.put_entities(app_id, entities, put_request.composite_index_list())
        self.logger.debug('Updated {} entities'.format(len(entities)))
  
      put_response.key_list().extend([e.key() for e in entities])
  
    def get_root_key_from_entity_key(self, entity_key):
      """ Extract the root key from an entity key. We 
          remove any excess children from a string to get to
          the root key.
      
      Args:
        entity_key: A string or Key object representing a row key.
      Returns:
        The root key extracted from the row key.
      Raises:
        TypeError: If the type is not supported.
      """
      if isinstance(entity_key, str):
        tokens = entity_key.split(dbconstants.KIND_SEPARATOR)
        return tokens[0] + dbconstants.KIND_SEPARATOR
      elif isinstance(entity_key, entity_pb.Reference):
        app_id = clean_app_id(entity_key.app())
        path = entity_key.path()
        element_list = path.element_list()
        return self.get_root_key(app_id, entity_key.name_space(), element_list)
      else:
        raise TypeError("Unable to get root key from given type of %s" % \
                        entity_key.__class__)
  
    def get_root_key(self, app_id, ns, ancestor_list):
      """ Gets the root key string from an ancestor listing.
     
      Args:
        app_id: The app ID of the listing.
        ns: The namespace of the entity.
        ancestor_list: The ancestry of a given entity.
      Returns:
        A string representing the root key of an entity.
      """
      prefix = self.get_table_prefix((app_id, ns))
      first_ent = ancestor_list[0]
      if first_ent.has_name():
        key_id = first_ent.name()
      elif first_ent.has_id():
        # Make sure ids are ordered lexigraphically by making sure they 
        # are of set size i.e. 2 > 0003 but 0002 < 0003.
        key_id = str(first_ent.id()).zfill(ID_KEY_LENGTH)
      return "{0}{1}{2}:{3}{4}".format(prefix, self._NAMESPACE_SEPARATOR, 
        first_ent.type(), key_id, dbconstants.KIND_SEPARATOR)
  
    def is_instance_wrapper(self, obj, expected_type):
      """ A wrapper for isinstance for mocking purposes. 
  
      Return whether an object is an instance of a class or of a subclass thereof.
      With a type as second argument, return whether that is the object's type.
  
      Args:
        obj: The object to check.
        expected_type: A instance type we are comparing obj's type to.
      Returns: 
        True if obj is of type expected_type, False otherwise. 
      """
      return isinstance(obj, expected_type) 
   
    def acquire_locks_for_trans(self, entities, txnid):
      """ Acquires locks for entities for one particular entity group. 
   
      Args:
        entities: A list of entities (entity_pb.EntityProto or entity_pb.Reference)
                  for which are are getting a lock for.
        txnid: The transaction ID handler.
      Returns:
        A hash mapping root keys to transaction IDs.
      Raises:
        ZKTransactionException: If lock is not obtainable.
        TypeError: If args are of incorrect types.
      """
      # Key tuples are the prefix and the root key for which we're getting locks.
      root_keys = []
      txn_hash = {}
      if not self.is_instance_wrapper(entities, list):
        raise TypeError("Expected a list and got {0}".format(entities.__class__))
      for ent in entities:
        if self.is_instance_wrapper(ent, entity_pb.Reference):
          root_keys.append(self.get_root_key_from_entity_key(ent))
        elif self.is_instance_wrapper(ent, entity_pb.EntityProto):
          root_keys.append(self.get_root_key_from_entity_key(ent.key()))
        else:
          raise TypeError("Excepted either a reference or an EntityProto"
             "got {0}".format(ent.__class__))
  
      if entities == []:
        return {}
  
      if self.is_instance_wrapper(entities[0], entity_pb.Reference):
        app_id = entities[0].app()
      else:
        app_id = entities[0].key().app()
      app_id = clean_app_id(app_id)
      # Remove all duplicate root keys.
      root_keys = list(set(root_keys))
      try:
        for root_key in root_keys:
          txn_hash[root_key] = txnid
          self.zookeeper.acquire_lock(app_id, txnid, root_key)
      except zktransaction.ZKTransactionException as zkte:
        self.logger.warning('Concurrent transaction: {}'.format(txnid))
        for root_key in txn_hash:
          self.zookeeper.notify_failed_transaction(app_id, txn_hash[root_key])
        raise zkte
  
      return txn_hash
  
    def release_locks_for_nontrans(self, app_id, entities, txn_hash):
      """  Releases locks for non-transactional puts.
    
      Args:
        entities: List of entities for which we are releasing locks. Can
                  be either entity_pb.EntityProto or entity_pb.Reference.
        txn_hash: A hash mapping root keys to transaction IDs.
      Raises:
        ZKTransactionException: If we are unable to release locks.
      """
      root_keys = []
      for ent in entities:
        if isinstance(ent, entity_pb.EntityProto):
          ent = ent.key()
        root_keys.append(self.get_root_key_from_entity_key(ent))
  
      # Remove all duplicate root keys
      root_keys = list(set(root_keys))
      for root_key in root_keys: 
        txnid = txn_hash[root_key]
        self.zookeeper.release_lock(app_id, txnid)
  
    def fetch_keys(self, key_list):
      """ Given a list of keys fetch the entities.
      
      Args:
        key_list: A list of keys to fetch.
      Returns:
        A tuple of entities from the datastore and key list.
      """
      row_keys = []
      for key in key_list:
        self.validate_app_id(key.app())
        index_key = str(encode_index_pb(key.path()))
        prefix = self.get_table_prefix(key)
        row_keys.append(self._SEPARATOR.join([prefix, index_key]))
      result = self.datastore_batch.batch_get_entity(
        dbconstants.APP_ENTITY_TABLE,
        row_keys,
        APP_ENTITY_SCHEMA)
      return result, row_keys
  
    def dynamic_get(self, app_id, get_request, get_response):
      """ Fetch keys from the datastore.
      
      Args: 
         app_id: The application ID.
         get_request: Request with list of keys.
         get_response: Response to application server.
      Raises:
        ZKTransactionException: If a lock was unable to get acquired.
      """ 
      keys = get_request.key_list()
      if len(keys) < 5:
        self.logger.debug('Get:\n{}'.format(get_request))
      else:
        self.logger.debug('Get: {} keys'.format(len(keys)))
  
      if get_request.has_transaction():
        results, row_keys = self.fetch_keys(keys)
        fetched_groups = {group_for_key(key).Encode() for key in keys}
        self.datastore_batch.record_reads(
          app_id, get_request.transaction().handle(), fetched_groups)
      else:
        results, row_keys = self.fetch_keys(keys)
  
      result_count = 0
      for r in row_keys:
        group = get_response.add_entity()
        if r in results and APP_ENTITY_SCHEMA[0] in results[r]:
          result_count += 1
          group.mutable_entity().CopyFrom(
            entity_pb.EntityProto(results[r][APP_ENTITY_SCHEMA[0]]))
      self.logger.debug('Returning {} results'.format(result_count))
  
    def dynamic_delete(self, app_id, delete_request):
      """ Deletes a set of rows.
      
      Args: 
        app_id: The application ID.
        delete_request: Request with a list of keys.
      """
      keys = delete_request.key_list()
      if not keys:
        return
  
      ent_kinds = []
      for key in delete_request.key_list():
        last_path = key.path().element_list()[-1]
        if last_path.type() not in ent_kinds:
          ent_kinds.append(last_path.type())
  
      # We use the marked changes field to signify if we should 
      # look up composite indexes because delete request do not
      # include that information.
      composite_indexes = []
      filtered_indexes = []
      if delete_request.has_mark_changes():
        all_composite_indexes = self.datastore_batch.get_indices(app_id)
        for index in all_composite_indexes:
          new_index = entity_pb.CompositeIndex()
          new_index.ParseFromString(index)
          composite_indexes.append(new_index)
        # Only get composites of the correct kinds.
        for index in composite_indexes:
          if index.definition().entity_type() in ent_kinds:
            filtered_indexes.append(index)
  
      if delete_request.has_transaction():
        txid = delete_request.transaction().handle()
        self.datastore_batch.delete_entities_tx(app_id, txid, keys)
      else:
        by_group = {}
        for key in keys:
          group_key = group_for_key(key).Encode()
          if group_key not in by_group:
            by_group[group_key] = []
          by_group[group_key].append(key)
  
        for encoded_group_key, key_list in by_group.iteritems():
          group_key = entity_pb.Reference(encoded_group_key)
  
          txid = self.transaction_manager.create_transaction_id(app_id, xg=False)
          self.transaction_manager.set_groups(app_id, txid, [group_key])
  
          # Allow the lock to stick around if there is an issue applying the batch.
          lock = entity_lock
  
  [... the rest is omitted ...]
files.new_path: AppDB/appscale/datastore/datastore_distributed.py
files.old_contents: 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 = data.name_space()
  
      return self._SEPARATOR.join([app_id, namespace])
  
    @staticmethod
    def get_ancestor_key_from_ent_key(ent_key):
      """ Get the key string for the ancestor portion of a composite key.
  
      Args:
        ent_key: A string of the entire path of an entity.
      Returns:
        A str of the path of the ancestor.
      """
      ancestor = ""
      tokens = str(ent_key).split(dbconstants.KIND_SEPARATOR)
      # Strip off the empty placeholder and also do not include the last kind.
      for token in tokens[:-2]:
        ancestor += token + dbconstants.KIND_SEPARATOR
      return ancestor
  
    @staticmethod
    def get_composite_index_key(index, entity, position_list=None, 
      filters=None):
      """ Creates a key to the composite index table for a given entity
      for a composite cursor.
  
      Keys are built as such: 
        app_id/ns/composite_id/ancestor/valuevaluevalue..../entity_key
      Components explained:
      ns: The namespace of the entity.
      composite_id: The composite ID assigned to this index upon creation.
      ancestor: The root ancestor path (only if the query this index is for 
        has an ancestor)
      value(s): The string representation of mulitiple properties.
      entity_key: The entity key (full path) used as a means of having a unique
        identifier. This prevents two entities with the same values from
        colliding.
  
      Args:
        index: A datstore_pb.CompositeIndex.
        entity: A entity_pb.EntityProto.
        position_list: A list of datastore_pb.CompiledCursor_Position items.
          Contains values for property items from a cursor.
        filters: A list of datastore_pb.Query_Filters, used to attain equality
          values not present in position_list.
      Returns:
        A string representing a key to the composite table.
      """ 
      composite_id = index.id()
      definition = index.definition()
      app_id = clean_app_id(entity.key().app())
      name_space = entity.key().name_space()
      ent_key = encode_index_pb(entity.key().path())
      pre_comp_index_key = "{0}{1}{2}{4}{3}{4}".format(app_id, 
        DatastoreDistributed._NAMESPACE_SEPARATOR, name_space, composite_id,
        DatastoreDistributed._SEPARATOR)
      if definition.ancestor() == 1:
        ancestor = DatastoreDistributed.get_ancestor_key_from_ent_key(ent_key)
        pre_comp_index_key += "{0}{1}".format(ancestor,
          DatastoreDistributed._SEPARATOR) 
  
      value_dict = {}
      for prop in entity.property_list():
        value_dict[prop.name()] = \
          str(encode_index_pb(prop.value()))
  
      # Position list and filters are used if we're creating a composite
      # key for a cursor.
      if position_list:
        for indexvalue in position_list[0].indexvalue_list():
          value_dict[indexvalue.property()] = \
            str(encode_index_pb(indexvalue.value()))
      if filters:
        for filt in filters:
          if filt.op() == datastore_pb.Query_Filter.EQUAL:
            value_dict[filt.property(0).name()] = \
              str(encode_index_pb(filt.property(0).value()))
  
      index_value = ""
      for prop in definition.property_list():
        name = prop.name()
        value = ''
        if name in value_dict:
          value = value_dict[name]
        elif name == "__key__":
          value = ent_key
        else:
          logging.warning("Given entity {0} is missing a property value {1}.".\
            format(entity, prop.name()))
        if prop.direction() == entity_pb.Index_Property.DESCENDING:
          value = helper_functions.reverse_lex(value)
  
        index_value += str(value) + DatastoreDistributed._SEPARATOR
  
      # We append the ent key to have unique keys if entities happen
      # to share the same index values (and ancestor).
      composite_key = "{0}{1}{2}".format(pre_comp_index_key, index_value,
        ent_key)
      return composite_key
    
    def insert_composite_indexes(self, entities, composite_indexes):
      """ Creates composite indexes for a set of entities.
  
      Args:
        entities: A list entities.
        composite_indexes: A list of datastore_pb.CompositeIndex.
      """
      if not composite_indexes:
        return
      row_keys = []
      row_values = {}
      # Create default composite index for all entities. Here we take each
      # of the properties in one 
      for ent in entities:
        for index_def in composite_indexes:
          # Skip any indexes if the kind does not match.
          kind = get_entity_kind(ent.key())
          if index_def.definition().entity_type() != kind:
            continue
  
          # Make sure the entity contains the required entities for the composite
          # definition. 
          prop_name_def_list = [index_prop.name() for index_prop in \
            index_def.definition().property_list()]
          all_prop_names_in_ent = [prop.name() for prop in \
            ent.property_list()]
          has_values = True 
          for index_prop_name in prop_name_def_list:
            if index_prop_name not in all_prop_names_in_ent:
              has_values = False
            # Special property name which does not show up in the list but 
            # is apart of the key of the entity.
            if index_prop_name == "__key__":
              has_values = True
          if not has_values:
            continue
  
          # Get the composite index key.
          composite_index_keys = get_composite_index_keys(index_def, ent)
          row_keys.extend(composite_index_keys)
  
          # Get the reference value for the composite table.
          entity_key = str(encode_index_pb(ent.key().path()))
          prefix = self.get_table_prefix(ent.key())
          reference = "{0}{1}{2}".format(prefix, self._SEPARATOR,  entity_key)
          for composite_key in composite_index_keys:
            row_values[composite_key] = {'reference': reference}
  
      self.datastore_batch.batch_put_entity(dbconstants.COMPOSITE_TABLE, 
                                            row_keys, 
                                            dbconstants.COMPOSITE_SCHEMA,
                                            row_values)
  
    def delete_composite_index_metadata(self, app_id, index):
      """ Deletes a index for the given application identifier.
    
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      """
      self.logger.info('Deleting composite index:\n{}'.format(index))
      index_keys = []
      composite_id = str(index.id())
      index_keys.append(self._SEPARATOR.join([app_id, 'index', composite_id]))
      self.datastore_batch.batch_delete(dbconstants.METADATA_TABLE,
                                        index_keys, 
                                        column_names=dbconstants.METADATA_TABLE)
  
    def create_composite_index(self, app_id, index):
      """ Stores a new index for the given application identifier.
    
      Args:
        app_id: A string representing the application identifier.
        index: A entity_pb.CompositeIndex object.
      Returns:
        A unique number representing the composite index ID.
      """
      # Generate a random number based on time of creation.
      rand = int(str(int(time.time())) + str(random.randint(0, 999999)))
      index.set_id(rand)
      encoded_entity = index.Encode()
      row_key = self._SEPARATOR.join([app_id, 'index', str(rand)])
      row_keys = [row_key]
      row_values = {}
      row_values[row_key] = {dbconstants.METADATA_SCHEMA[0]: encoded_entity}
      self.datastore_batch.batch_put_entity(dbconstants.METADATA_TABLE, 
                                            row_keys, 
                                            dbconstants.METADATA_SCHEMA, 
                                            row_values)    
      return rand 
  
    def update_composite_index(self, app_id, index):
      """ Updates an index for a given app ID.
  
      Args:
        app_id: A string containing the app ID.
        index: An entity_pb.CompositeIndex object.
      """
      self.logger.info('Updating index: {}'.format(index))
      entries_updated = 0
      entity_type = index.definition().entity_type()
  
      # TODO: Adjust prefix based on ancestor.
      prefix = '{app}{delimiter}{entity_type}{kind_separator}'.format(
        app=app_id,
        delimiter=self._SEPARATOR * 2,
        entity_type=entity_type,
        kind_separator=dbconstants.KIND_SEPARATOR,
      )
      start_row = prefix
      end_row = prefix + self._TERM_STRING
      start_inclusive = True
  
      while True:
        # Fetch references from the kind table since entity keys can have a
        # parent prefix.
        references = self.datastore_batch.range_query(
          table_name=dbconstants.APP_KIND_TABLE,
          column_names=dbconstants.APP_KIND_SCHEMA,
          start_key=start_row,
          end_key=end_row,
          limit=self.BATCH_SIZE,
          offset=0,
          start_inclusive=start_inclusive,
        )
  
        pb_entities = self.__fetch_entities(references)
        entities = [entity_pb.EntityProto(entity) for entity in pb_entities]
  
        self.insert_composite_indexes(entities, [index])
        entries_updated += len(entities)
  
        # If we fetched fewer references than we asked for, we're done.
        if len(references) < self.BATCH_SIZE:
          break
  
        start_row = references[-1].keys()[0]
        start_inclusive = self._DISABLE_INCLUSIVITY
  
      self.logger.info('Updated {} index entries.'.format(entries_updated))
  
    def allocate_size(self, project, size):
      """ Allocates a block of IDs for a project.
  
      Args:
        project: A string specifying the project ID.
        size: An integer specifying the number of IDs to reserve.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      return allocator.allocate_size(size)
  
    def allocate_max(self, project, max_id):
      """ Reserves all IDs up to the one given.
  
      Args:
        project: A string specifying the project ID.
        max_id: An integer specifying the maximum ID to allocated.
      Returns:
        A tuple of integers specifying the start and end ID.
      """
      if project not in self.sequential_allocators:
        self.sequential_allocators[project] = EntityIDAllocator(
          self.datastore_batch.session, project)
  
      allocator = self.sequential_allocators[project]
      return allocator.allocate_max(max_id)
  
    def reserve_ids(self, project_id, ids):
      """ Ensures the given IDs are not re-allocated.
  
      Args:
        project_id: A string specifying the project ID.
        ids: An iterable of integers specifying entity IDs.
      """
      if project_id not in self.sequential_allocators:
        self.sequential_allocators[project_id] = EntityIDAllocator(
          self.datastore_batch.session, project_id)
  
      if project_id not in self.scattered_allocators:
        self.scattered_allocators[project_id] = ScatteredAllocator(
          self.datastore_batch.session, project_id)
  
      for id_ in ids:
        counter, space = IdToCounter(id_)
        if space == SEQUENTIAL:
          allocator = self.sequential_allocators[project_id]
        else:
          allocator = self.scattered_allocators[project_id]
  
        allocator.set_min_counter(counter)
  
    def put_entities(self, app, entities, composite_indexes=()):
      """ Updates indexes of existing entities, inserts new entities and 
          indexes for them.
  
      Args:
        app: A string containing the application ID.
        entities: List of entities.
        composite_indexes: A list or tuple of CompositeIndex objects.
      """
      self.logger.debug('Inserting {} entities'.format(len(entities)))
      entity_keys = []
      for entity in entities:
        prefix = self.get_table_prefix(entity)
        entity_keys.append(get_entity_key(prefix, entity.key().path()))
  
      current_values = self.datastore_batch.batch_get_entity(
        dbconstants.APP_ENTITY_TABLE, entity_keys, APP_ENTITY_SCHEMA)
  
      by_group = {}
      for entity in entities:
        group_key = group_for_key(entity.key()).Encode()
        if group_key not in by_group:
          by_group[group_key] = []
        by_group[group_key].append(entity)
  
      for encoded_group_key, entity_list in by_group.iteritems():
        group_key = entity_pb.Reference(encoded_group_key)
  
        txid = self.transaction_manager.create_transaction_id(app, xg=False)
        self.transaction_manager.set_groups(app, txid, [group_key])
        lock = entity_lock.EntityLock(self.zookeeper.handle, [group_key], txid)
        try:
          with lock:
            batch = []
            entity_changes = []
            for entity in entity_list:
              prefix = self.get_table_prefix(entity)
              entity_key = get_entity_key(prefix, entity.key().path())
  
              current_value = None
              if current_values[entity_key]:
                current_value = entity_pb.EntityProto(
                  current_values[entity_key][APP_ENTITY_SCHEMA[0]])
  
              batch.extend(mutations_for_entity(entity, txid, current_value,
                                                composite_indexes))
  
              batch.append({'table': 'group_updates',
                            'key': bytearray(encoded_group_key),
                            'last_update': txid})
  
              entity_changes.append(
                {'key': entity.key(), 'old': current_value, 'new': entity})
            self.datastore_batch.batch_mutate(app, batch, entity_changes, txid)
        finally:
          self.transaction_manager.delete_transaction_id(app, txid)
  
    def delete_entities(self, group, txid, keys, composite_indexes=()):
      """ Deletes the entities and the indexes associated with them.
  
      Args:
        group: An entity group Reference object.
        txid: An integer specifying a transaction ID.
        keys: An interable containing entity Reference objects.
        composite_indexes: A list or tuple of CompositeIndex objects.
      """
      entity_keys = []
      for key in keys:
        prefix = self.get_table_prefix(key)
        entity_keys.append(get_entity_key(prefix, key.path()))
  
      # Must fetch the entities to get the keys of indexes before deleting.
      current_values = self.datastore_batch.batch_get_entity(
        dbconstants.APP_ENTITY_TABLE, entity_keys, APP_ENTITY_SCHEMA)
  
      for key in entity_keys:
        if not current_values[key]:
          continue
  
        current_value = entity_pb.EntityProto(
          current_values[key][APP_ENTITY_SCHEMA[0]])
        batch = deletions_for_entity(current_value, composite_indexes)
  
        batch.append({'table': 'group_updates',
                      'key': bytearray(group.Encode()),
                      'last_update': txid})
  
        self.datastore_batch._normal_batch(batch, txid)
  
    def dynamic_put(self, app_id, put_request, put_response):
      """ Stores and entity and its indexes in the datastore.
      
      Args:
        app_id: Application ID.
        put_request: Request with entities to store.
        put_response: The response sent back to the app server.
      Raises:
        ZKTransactionException: If we are unable to acquire/release ZooKeeper locks.
      """
      if app_id not in self.scattered_allocators:
        self.scattered_allocators[app_id] = ScatteredAllocator(
          self.datastore_batch.session, app_id)
      allocator = self.scattered_allocators[app_id]
  
      entities = put_request.entity_list()
  
      for entity in entities:
        self.validate_key(entity.key())
  
        for prop in itertools.chain(entity.property_list(),
                                    entity.raw_property_list()):
          if prop.value().has_uservalue():
            uid = md5.new(prop.value().uservalue().email().lower()).digest()
            uid = '1' + ''.join(['%02d' % ord(x) for x in uid])[:20]
            prop.mutable_value().mutable_uservalue().set_obfuscated_gaiaid(uid)
  
        last_path = entity.key().path().element_list()[-1]
        if last_path.id() == 0 and not last_path.has_name():
          last_path.set_id(allocator.next())
          group = entity.mutable_entity_group()
          root = entity.key().path().element(0)
          group.add_element().CopyFrom(root)
  
      if put_request.has_transaction():
        self.datastore_batch.put_entities_tx(
          app_id, put_request.transaction().handle(), entities)
      else:
        try:
          self.put_entities(app_id, entities, put_request.composite_index_list())
        except entity_lock.LockTimeout as timeout_error:
          raise dbconstants.AppScaleDBConnectionError(str(timeout_error))
        self.logger.debug('Updated {} entities'.format(len(entities)))
  
      put_response.key_list().extend([e.key() for e in entities])
  
    def get_root_key_from_entity_key(self, entity_key):
      """ Extract the root key from an entity key. We 
          remove any excess children from a string to get to
          the root key.
      
      Args:
        entity_key: A string or Key object representing a row key.
      Returns:
        The root key extracted from the row key.
      Raises:
        TypeError: If the type is not supported.
      """
      if isinstance(entity_key, str):
        tokens = entity_key.split(dbconstants.KIND_SEPARATOR)
        return tokens[0] + dbconstants.KIND_SEPARATOR
      elif isinstance(entity_key, entity_pb.Reference):
        app_id = clean_app_id(entity_key.app())
        path = entity_key.path()
        element_list = path.element_list()
        return self.get_root_key(app_id, entity_key.name_space(), element_list)
      else:
        raise TypeError("Unable to get root key from given type of %s" % \
                        entity_key.__class__)
  
    def get_root_key(self, app_id, ns, ancestor_list):
      """ Gets the root key string from an ancestor listing.
     
      Args:
        app_id: The app ID of the listing.
        ns: The namespace of the entity.
        ancestor_list: The ancestry of a given entity.
      Returns:
        A string representing the root key of an entity.
      """
      prefix = self.get_table_prefix((app_id, ns))
      first_ent = ancestor_list[0]
      if first_ent.has_name():
        key_id = first_ent.name()
      elif first_ent.has_id():
        # Make sure ids are ordered lexigraphically by making sure they 
        # are of set size i.e. 2 > 0003 but 0002 < 0003.
        key_id = str(first_ent.id()).zfill(ID_KEY_LENGTH)
      return "{0}{1}{2}:{3}{4}".format(prefix, self._NAMESPACE_SEPARATOR, 
        first_ent.type(), key_id, dbconstants.KIND_SEPARATOR)
  
    def is_instance_wrapper(self, obj, expected_type):
      """ A wrapper for isinstance for mocking purposes. 
  
      Return whether an object is an instance of a class or of a subclass thereof.
      With a type as second argument, return whether that is the object's type.
  
      Args:
        obj: The object to check.
        expected_type: A instance type we are comparing obj's type to.
      Returns: 
        True if obj is of type expected_type, False otherwise. 
      """
      return isinstance(obj, expected_type) 
   
    def acquire_locks_for_trans(self, entities, txnid):
      """ Acquires locks for entities for one particular entity group. 
   
      Args:
        entities: A list of entities (entity_pb.EntityProto or entity_pb.Reference)
                  for which are are getting a lock for.
        txnid: The transaction ID handler.
      Returns:
        A hash mapping root keys to transaction IDs.
      Raises:
        ZKTransactionException: If lock is not obtainable.
        TypeError: If args are of incorrect types.
      """
      # Key tuples are the prefix and the root key for which we're getting locks.
      root_keys = []
      txn_hash = {}
      if not self.is_instance_wrapper(entities, list):
        raise TypeError("Expected a list and got {0}".format(entities.__class__))
      for ent in entities:
        if self.is_instance_wrapper(ent, entity_pb.Reference):
          root_keys.append(self.get_root_key_from_entity_key(ent))
        elif self.is_instance_wrapper(ent, entity_pb.EntityProto):
          root_keys.append(self.get_root_key_from_entity_key(ent.key()))
        else:
          raise TypeError("Excepted either a reference or an EntityProto"
             "got {0}".format(ent.__class__))
  
      if entities == []:
        return {}
  
      if self.is_instance_wrapper(entities[0], entity_pb.Reference):
        app_id = entities[0].app()
      else:
        app_id = entities[0].key().app()
      app_id = clean_app_id(app_id)
      # Remove all duplicate root keys.
      root_keys = list(set(root_keys))
      try:
        for root_key in root_keys:
          txn_hash[root_key] = txnid
          self.zookeeper.acquire_lock(app_id, txnid, root_key)
      except zktransaction.ZKTransactionException as zkte:
        self.logger.warning('Concurrent transaction: {}'.format(txnid))
        for root_key in txn_hash:
          self.zookeeper.notify_failed_transaction(app_id, txn_hash[root_key])
        raise zkte
  
      return txn_hash
  
    def release_locks_for_nontrans(self, app_id, entities, txn_hash):
      """  Releases locks for non-transactional puts.
    
      Args:
        entities: List of entities for which we are releasing locks. Can
                  be either entity_pb.EntityProto or entity_pb.Reference.
        txn_hash: A hash mapping root keys to transaction IDs.
      Raises:
        ZKTransactionException: If we are unable to release locks.
      """
      root_keys = []
      for ent in entities:
        if isinstance(ent, entity_pb.EntityProto):
          ent = ent.key()
        root_keys.append(self.get_root_key_from_entity_key(ent))
  
      # Remove all duplicate root keys
      root_keys = list(set(root_keys))
      for root_key in root_keys: 
        txnid = txn_hash[root_key]
        self.zookeeper.release_lock(app_id, txnid)
  
    def fetch_keys(self, key_list):
      """ Given a list of keys fetch the entities.
      
      Args:
        key_list: A list of keys to fetch.
      Returns:
        A tuple of entities from the datastore and key list.
      """
      row_keys = []
      for key in key_list:
        self.validate_app_id(key.app())
        index_key = str(encode_index_pb(key.path()))
        prefix = self.get_table_prefix(key)
        row_keys.append(self._SEPARATOR.join([prefix, index_key]))
      result = self.datastore_batch.batch_get_entity(
        dbconstants.APP_ENTITY_TABLE,
        row_keys,
        APP_ENTITY_SCHEMA)
      return result, row_keys
  
    def dynamic_get(self, app_id, get_request, get_response):
      """ Fetch keys from the datastore.
      
      Args: 
         app_id: The application ID.
         get_request: Request with list of keys.
         get_response: Response to application server.
      Raises:
        ZKTransactionException: If a lock was unable to get acquired.
      """ 
      keys = get_request.key_list()
      if len(keys) < 5:
        self.logger.debug('Get:\n{}'.format(get_request))
      else:
        self.logger.debug('Get: {} keys'.format(len(keys)))
  
      if get_request.has_transaction():
        results, row_keys = self.fetch_keys(keys)
        fetched_groups = {group_for_key(key).Encode() for key in keys}
        self.datastore_batch.record_reads(
          app_id, get_request.transaction().handle(), fetched_groups)
      else:
        results, row_keys = self.fetch_keys(keys)
  
      result_count = 0
      for r in row_keys:
        group = get_response.add_entity()
        if r in results and APP_ENTITY_SCHEMA[0] in results[r]:
          result_count += 1
          group.mutable_entity().CopyFrom(
            entity_pb.EntityProto(results[r][APP_ENTITY_SCHEMA[0]]))
      self.logger.debug('Returning {} results'.format(result_count))
  
    def dynamic_delete(self, app_id, delete_request):
      """ Deletes a set of rows.
      
      Args: 
        app_id: The application ID.
        delete_request: Request with a list of keys.
      """
      keys = delete_request.key_list()
      if not keys:
        return
  
      ent_kinds = []
      for key in delete_request.key_list():
        last_path = key.path().element_list()[-1]
        if last_path.type() not in ent_kinds:
          ent_kinds.append(last_path.type())
  
      # We use the marked changes field to signify if we should 
      # look up composite indexes because delete request do not
      # include that information.
      composite_indexes = []
      filtered_indexes = []
      if delete_request.has_mark_changes():
        all_composite_indexes = self.datastore_batch.get_indices(app_id)
        for index in all_composite_indexes:
          new_index = entity_pb.CompositeIndex()
          new_index.ParseFromString(index)
          composite_indexes.append(new_index)
        # Only get composites of the correct kinds.
        for index in composite_indexes:
          if index.definition().entity_type() in ent_kinds:
            filtered_indexes.append(index)
  
      if delete_request.has_transaction():
        txid = delete_request.transaction().handle()
        self.datastore_batch.delete_entities_tx(app_id, txid, keys)
      else:
        by_group = {}
        for key in keys:
          group_key = group_for_key(key).Encode()
          if group_key not in by_group:
            by_group[group_key] = []
          by_group[group_key].append(key)
  
        for encoded_group_key, key_list in by_group.iteritems():
          group_key = entity_pb.Reference(encoded_group_key)
  
          txid = self.transaction_manager.create_transaction_id(app_id, xg=False)
          self.transaction_manager.set_groups(app_id, txid, [group_key])
          lock = entity_lock.EntityLock(self.zookeeper.handle, [group_key], txid)
          try:
            with lock:
              self.delete_entities(
                group_key,
                txid,
                key_list
  
  [... the rest is omitted ...]
files.old_path: AppDB/appscale/datastore/datastore_distributed.py
files.new_contents: """
   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.new_path: AppDB/appscale/datastore/dbconstants.py
files.old_contents: """
   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.old_path: AppDB/appscale/datastore/dbconstants.py
files.new_contents: # 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, errdetail = self.allocate_ids_request(
                                                          app_id,
                                                          http_request_data)
      elif method == "CreateIndex":
        response, errcode, errdetail = self.create_index_request(app_id,
                                                          http_request_data)
      elif method == "GetIndices":
        response, errcode, errdetail = self.get_indices_request(app_id)
      elif method == "UpdateIndex":
        response, errcode, errdetail = self.update_index_request(app_id,
          http_request_data)
      elif method == "DeleteIndex":
        response, errcode, errdetail = self.delete_index_request(app_id,
                                                         http_request_data)
      elif method == 'AddActions':
        response, errcode, errdetail = self.add_actions_request(
          app_id, http_request_data, service_id, version_id)
      elif method == 'datastore_v4.AllocateIds':
        response, errcode, errdetail = yield self.v4_allocate_ids_request(
          app_id, http_request_data)
      else:
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Unknown datastore message"
  
      time_taken = time.time() - start
      if method in STATS:
        if errcode in STATS[method]:
          prev_req, pre_time = STATS[method][errcode]
          STATS[method][errcode] = prev_req + 1, pre_time + time_taken
        else:
          STATS[method][errcode] = (1, time_taken)
      else:
        STATS[method] = {}
        STATS[method][errcode] = (1, time_taken)
  
      apiresponse.set_response(response)
      if errcode != 0:
        apperror_pb = apiresponse.mutable_application_error()
        apperror_pb.set_code(errcode)
        apperror_pb.set_detail(errdetail)
  
      self.write(apiresponse.Encode())
  
    def begin_transaction_request(self, app_id, http_request_data):
      """ Handles the intial request to start a transaction. Replies with 
          a unique identifier to handle this transaction in future requests.
    
      Args:
        app_id: The application ID requesting the transaction.
        http_request_data: The encoded request.
      Returns:
        An encoded transaction protocol buffer with a unique handler.
      """
      global datastore_access
      begin_transaction_req_pb = datastore_pb.BeginTransactionRequest(
        http_request_data)
      multiple_eg = False
      if begin_transaction_req_pb.has_allow_multiple_eg():
        multiple_eg = bool(begin_transaction_req_pb.allow_multiple_eg())
  
      handle = None
      transaction_pb = datastore_pb.Transaction()
  
      if READ_ONLY:
        logger.warning('Unable to begin transaction in read-only mode: {}'.
          format(begin_transaction_req_pb))
        return (transaction_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        handle = datastore_access.setup_transaction(app_id, multiple_eg)
      except dbconstants.InternalError as error:
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except dbconstants.BadRequest as error:
        return '', datastore_pb.Error.BAD_REQUEST, str(error)
      except (zktransaction.ZKInternalException,
              dbconstants.AppScaleDBConnectionError) as error:
        logger.exception('Unable to begin transaction')
        return (transaction_pb.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                str(error))
  
      transaction_pb.set_app(app_id)
      transaction_pb.set_handle(handle)
      return (transaction_pb.Encode(), 0, "")
  
    def commit_transaction_request(self, app_id, http_request_data):
      """ Handles the commit phase of a transaction.
  
      Args:
        app_id: The application ID requesting the transaction commit.
        http_request_data: The encoded request of datastore_pb.Transaction.
      Returns:
        An encoded protocol buffer commit response.
      """
      global datastore_access
  
      if READ_ONLY:
        commitres_pb = datastore_pb.CommitResponse()
        transaction_pb = datastore_pb.Transaction(http_request_data)
        logger.warning('Unable to commit in read-only mode: {}'.
          format(transaction_pb))
        return (commitres_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      return datastore_access.commit_transaction(app_id, http_request_data)
  
    def rollback_transaction_request(self, app_id, http_request_data):
      """ Handles the rollback phase of a transaction.
  
      Args:
        app_id: The application ID requesting the rollback.
        http_request_data: The encoded request.
      Returns:
        An encoded protocol buffer void response.
      """
      global datastore_access
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to rollback in read-only mode: {}'.
          format(http_request_data))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        return datastore_access.rollback_transaction(app_id, http_request_data)
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {} for {}'.
          format(http_request_data, app_id))
        return (response.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except Exception:
        logger.exception('Unable to rollback transaction')
        return(response.Encode(),
               datastore_pb.Error.INTERNAL_ERROR,
               "Unable to rollback for this transaction")
  
    def run_query(self, http_request_data):
      """ High level function for running queries.
  
      Args:
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        Returns an encoded query response.
      """
      global datastore_access
      query = datastore_pb.Query(http_request_data)
      clone_qr_pb = UnprocessedQueryResult()
      try:
        datastore_access._dynamic_run_query(query, clone_qr_pb)
      except zktransaction.ZKBadRequest, zkie:
        logger.exception('Illegal arguments in transaction during {}'.
          format(query))
        return (clone_qr_pb.Encode(),
                datastore_pb.Error.BAD_REQUEST,
                "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {}'.format(query))
        clone_qr_pb.set_more_results(False)
        return (clone_qr_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.format(query))
        clone_qr_pb.set_more_results(False)
        return (clone_qr_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on put.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during query')
        clone_qr_pb.set_more_results(False)
        return (clone_qr_pb.Encode(),
               datastore_pb.Error.INTERNAL_ERROR,
               "Datastore connection error on run_query request.")
      return clone_qr_pb.Encode(), 0, ""
  
    def create_index_request(self, app_id, http_request_data):
      """ High level function for creating composite indexes.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.Integer64Proto()
  
      if READ_ONLY:
        logger.warning('Unable to create in read-only mode: {}'.
          format(request))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        index_id = datastore_access.create_composite_index(app_id, request)
        response.set_value(index_id)
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during index creation')
        response.set_value(0)
        return (response.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on create index request.")
      return response.Encode(), 0, ""
  
    def update_index_request(self, app_id, http_request_data):
      """ High level function for updating a composite index.
  
      Args:
        app_id: A string containing the application ID.
        http_request_data: A string containing the protocol buffer request
          from the AppServer.
      Returns:
         A tuple containing an encoded response, error code, and error details.
      """
      global datastore_access
      index = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to update in read-only mode: {}'.
          format(index))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      state = index.state()
      if state not in [index.READ_WRITE, index.WRITE_ONLY]:
        state_name = entity_pb.CompositeIndex.State_Name(state)
        error_message = 'Unable to update index because state is {}. '\
          'Index: {}'.format(state_name, index)
        logger.error(error_message)
        return response.Encode(), datastore_pb.Error.PERMISSION_DENIED,\
          error_message
      else:
        # Updating index asynchronously so we can return a response quickly.
        threading.Thread(target=datastore_access.update_composite_index,
          args=(app_id, index)).start()
  
      return response.Encode(), 0, ""
  
    def delete_index_request(self, app_id, http_request_data):
      """ Deletes a composite index for a given application.
    
      Args:
        app_id: Name of the application.
        http_request_data: A serialized CompositeIndices item
      Returns:
        A Tuple of an encoded entity_pb.VoidProto, error code, and 
        error explanation.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to delete in read-only mode: {}'.
          format(request))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.delete_composite_index_metadata(app_id, request)
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during index deletion')
        return (response.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on delete index request.")
      return response.Encode(), 0, ""
  
    def get_indices_request(self, app_id):
      """ Gets the indices of the given application.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
        A Tuple of an encoded response, error code, and error explanation.
      """
      global datastore_access
      response = datastore_pb.CompositeIndices()
      try:
        indices = datastore_access.datastore_batch.get_indices(app_id)
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error while fetching indices for '
          '{}'.format(app_id))
        return (response.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on get indices request.")
      for index in indices:
        new_index = response.add_index()
        new_index.ParseFromString(index)
      return response.Encode(), 0, ""
  
    def allocate_ids_request(self, app_id, http_request_data):
      """ High level function for getting unique identifiers for entities.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      Raises:
         NotImplementedError: when requesting a max id.
      """
      request = datastore_pb.AllocateIdsRequest(http_request_data)
      response = datastore_pb.AllocateIdsResponse()
  
      if request.has_max() and request.has_size():
        return (response.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Both size and max cannot be set.')
      if not (request.has_max() or request.has_size()):
        return (response.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Either size or max must be set.')
  
      if request.has_size():
        try:
          start, end = datastore_access.allocate_size(app_id, request.size())
        except dbconstants.AppScaleBadArg as error:
          return response.Encode(), datastore_pb.Error.BAD_REQUEST, str(error)
        except dbconstants.AppScaleDBConnectionError as error:
          return response.Encode(), datastore_pb.Error.INTERNAL_ERROR, str(error)
      else:
        try:
          start, end = datastore_access.allocate_max(app_id, request.max())
        except dbconstants.AppScaleBadArg as error:
          return response.Encode(), datastore_pb.Error.BAD_REQUEST, str(error)
        except dbconstants.AppScaleDBConnectionError as error:
          return response.Encode(), datastore_pb.Error.INTERNAL_ERROR, str(error)
  
      response.set_start(start)
      response.set_end(end)
      return response.Encode(), 0, ""
  
    @staticmethod
    @gen.coroutine
    def v4_allocate_ids_request(app_id, http_request_data):
      """ Reserves entity IDs so that they will not be re-allocated.
  
      Args:
        app_id: Name of the application.
        http_request_data: The protocol buffer request from the AppServer.
      Returns:
         Returns an encoded response.
      """
      request = datastore_v4_pb.AllocateIdsRequest(http_request_data)
      response = datastore_v4_pb.AllocateIdsResponse()
  
      if not request.reserve_list():
        raise gen.Return((response.Encode(), datastore_v4_pb.Error.BAD_REQUEST,
                          'Request must include reserve list'))
  
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      datastore_access.reserve_ids(app_id, ids)
  
      # Forward request to other datastore servers in order to adjust any blocks
      # they've already allocated.
      client = AsyncHTTPClient()
      headers = {'appdata': app_id}
  
      futures = []
      for server in datastore_servers:
        ip, port = server.split(':')
        port = int(port)
        if ip == options.private_ip and port == options.port:
          continue
  
        url = 'http://{}:{}/reserve-keys'.format(ip, port)
        future = client.fetch(url, method='POST', headers=headers,
                              body=http_request_data)
        futures.append(future)
  
      for future in futures:
        yield future
  
      raise gen.Return((response.Encode(), 0, ''))
  
    def put_request(self, app_id, http_request_data):
      """ High level function for doing puts.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        Returns an encoded put response.
      """
      global datastore_access
  
      putreq_pb = datastore_pb.PutRequest(http_request_data)
      putresp_pb = datastore_pb.PutResponse()
  
      if READ_ONLY:
        logger.warning('Unable to put in read-only mode: {}'.
          format(putreq_pb))
        return (putresp_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.dynamic_put(app_id, putreq_pb, putresp_pb)
        return (putresp_pb.Encode(), 0, "")
      except dbconstants.InternalError as error:
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except dbconstants.TimeoutError as error:
        return '', datastore_pb.Error.TIMEOUT, str(error)
      except dbconstants.BadRequest as error:
        return '', datastore_pb.Error.BAD_REQUEST, str(error)
      except zktransaction.ZKBadRequest as zkie:
        logger.exception('Illegal argument during {}'.format(putreq_pb))
        return (putresp_pb.Encode(),
              datastore_pb.Error.BAD_REQUEST,
              "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException as error:
        logger.exception('ZKInternalException during put')
        return (putresp_pb.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                str(error))
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.
          format(putreq_pb))
        return (putresp_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on put.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during put')
        return (putresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on put.")
  
  
    def get_request(self, app_id, http_request_data):
      """ High level function for doing gets.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        An encoded get response.
      """
      global datastore_access
      getreq_pb = datastore_pb.GetRequest(http_request_data)
      getresp_pb = datastore_pb.GetResponse()
      try:
        datastore_access.dynamic_get(app_id, getreq_pb, getresp_pb)
      except zktransaction.ZKBadRequest as zkie:
        logger.exception('Illegal argument during {}'.format(getreq_pb))
        return (getresp_pb.Encode(),
                datastore_pb.Error.BAD_REQUEST,
                "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {}'.format(getreq_pb))
        return (getresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.
          format(getreq_pb))
        return (getresp_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on get.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during get')
        return (getresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on get.")
  
      return getresp_pb.Encode(), 0, ""
  
    def delete_request(self, app_id, http_request_data):
      """ High level function for doing deletes.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        An encoded delete response.
      """
      global datastore_access
  
      delreq_pb = datastore_pb.DeleteRequest( http_request_data )
      delresp_pb = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to delete in read-only mode: {}'.
          format(delreq_pb))
        return (delresp_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.dynamic_delete(app_id, delreq_pb)
        return (delresp_pb.Encode(), 0, "")
      except dbconstants.InternalError as error:
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except dbconstants.TimeoutError as error:
        return '', datastore_pb.Error.TIMEOUT, str(error)
      except dbconstants.BadRequest as error:
        return '', datastore_pb.Error.BAD_REQUEST, str(error)
      except zktransaction.ZKBadRequest as zkie:
        logger.exception('Illegal argument during {}'.format(delreq_pb))
        return (delresp_pb.Encode(),
                datastore_pb.Error.BAD_REQUEST,
                "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {}'.format(delreq_pb))
        return (delresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.
          format(delreq_pb))
        return (delresp_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on delete.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during delete')
        return (delresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on delete.")
  
    def add_actions_request(self, app_id, http_request_data, service_id,
                            version_id):
      """ High level function for adding transactional tasks.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      Returns:
        An encoded AddActions response.
      """
      global datastore_access
  
      req_pb = taskqueue_service_pb.TaskQueueBulkAddRequest(http_request_data)
      resp_pb = taskqueue_service_pb.TaskQueueBulkAddResponse()
  
      if service_id is None:
        return (resp_pb.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Module header must be defined')
  
      if version_id is None:
        return (resp_pb.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Version header must be defined')
  
      if READ_ONLY:
        logger.warning('Unable to add transactional tasks in read-only mode')
        return (resp_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.dynamic_add_actions(app_id, req_pb, service_id,
                                             version_id)
        return resp_pb.Encode(), 0, ""
      except dbconstants.ExcessiveTasks as error:
        return (resp_pb.Encode(), datastore_pb.Error.BAD_REQUEST, str(error))
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error')
        return (resp_pb.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                'Datastore connection error when adding transaction tasks.')
  
  
  def create_server_node():
    """ Creates a server registration entry in ZooKeeper. """
    try:
      zookeeper.handle.create(server_node, ephemeral=True)
    except NodeExistsError:
      # If the server gets restarted, the old node may exist for a short time.
      zookeeper.handle.delete(server_node)
      zookeeper.handle.create(server_node, ephemeral=True)
  
    logger.info('Datastore registered at {}'.format(server_node))
  
  
  def zk_state_listener(state):
    """ Handles changes to ZooKeeper connection state.
  
    Args:
      state: A string specifying the new ZooKeeper connection state.
    """
    if state == KazooState.CONNECTED:
      persistent_create_server_node = retry_data_watch_coroutine(
        server_node, create_server_node)
      IOLoop.instance().add_callback(persistent_create_server_node)
  
  
  def update_servers(new_servers):
    """ Updates the record of active datastore servers.
  
    Args:
      new_servers: A list of strings identifying server locations.
    """
    to_remove = [server for server in datastore_servers
                 if server not in new_servers]
    for old_server in to_remove:
      datastore_servers.remove(old_server)
  
    for new_server in new_servers:
      if new_server not in datastore_servers:
        datastore_servers.add(new_server)
  
  
  def update_servers_watch(new_servers):
    """ Updates the record of active datastore servers.
  
    Args:
      new_servers: A list of strings identifying server locations.
    """
    main_io_loop = IOLoop.instance()
    main_io_loop.add_callback(update_servers, new_servers)
  
  
  pb_application = tornado.web.Application([
    ('/clear', ClearHandler),
    ('/read-only', ReadOnlyHandler),
    ('/reserve-keys', ReserveKeysHandler),
    (r'/*', MainHandler),
  ])
  
  
  def main():
    """ Starts a web service for handing datastore requests. """
  
    global datastore_access
    global server_node
    global zookeeper
    zookeeper_locations = appscale_info.get_zk_locations_string()
  
    parser = argparse.ArgumentParser()
    parser.add_argument('-t', '--type', choices=dbconstants.VALID_DATASTORES,
                        default=dbconstants.VALID_DATASTORES[0],
                        help='Database type')
    parser.add_argument('-p', '--port', type=int,
                        default=dbconstants.DEFAULT_PORT,
                        help='Datastore server port')
    parser.add_argument('-v', '--verbose', action='store_true',
                        help='Output debug-level logging')
    args = parser.parse_args()
  
    if args.verbose:
      logger.setLevel(logging.DEBUG)
  
    options.define('private_ip', appscale_info.get_private_ip())
    options.define('port', args.port)
  
    server_node = '{}/{}:{}'.format(DATASTORE_SERVERS_NODE, options.private_ip,
                                    options.port)
  
    datastore_batch = DatastoreFactory.getDatastore(
      args.type, log_level=logger.getEffectiveLevel())
    zookeeper = zktransaction.ZKTransaction(
      host=zookeeper_locations, db_access=datastore_batch,
      log_level=logger.getEffectiveLevel())
  
    zookeeper.handle.add_listener(zk_state_listener)
    zookeeper.handle
  
  [... the rest is omitted ...]
files.new_path: AppDB/appscale/datastore/scripts/datastore.py
files.old_contents: # 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, errdetail = self.allocate_ids_request(
                                                          app_id,
                                                          http_request_data)
      elif method == "CreateIndex":
        response, errcode, errdetail = self.create_index_request(app_id,
                                                          http_request_data)
      elif method == "GetIndices":
        response, errcode, errdetail = self.get_indices_request(app_id)
      elif method == "UpdateIndex":
        response, errcode, errdetail = self.update_index_request(app_id,
          http_request_data)
      elif method == "DeleteIndex":
        response, errcode, errdetail = self.delete_index_request(app_id,
                                                         http_request_data)
      elif method == 'AddActions':
        response, errcode, errdetail = self.add_actions_request(
          app_id, http_request_data, service_id, version_id)
      elif method == 'datastore_v4.AllocateIds':
        response, errcode, errdetail = yield self.v4_allocate_ids_request(
          app_id, http_request_data)
      else:
        errcode = datastore_pb.Error.BAD_REQUEST
        errdetail = "Unknown datastore message"
  
      time_taken = time.time() - start
      if method in STATS:
        if errcode in STATS[method]:
          prev_req, pre_time = STATS[method][errcode]
          STATS[method][errcode] = prev_req + 1, pre_time + time_taken
        else:
          STATS[method][errcode] = (1, time_taken)
      else:
        STATS[method] = {}
        STATS[method][errcode] = (1, time_taken)
  
      apiresponse.set_response(response)
      if errcode != 0:
        apperror_pb = apiresponse.mutable_application_error()
        apperror_pb.set_code(errcode)
        apperror_pb.set_detail(errdetail)
  
      self.write(apiresponse.Encode())
  
    def begin_transaction_request(self, app_id, http_request_data):
      """ Handles the intial request to start a transaction. Replies with 
          a unique identifier to handle this transaction in future requests.
    
      Args:
        app_id: The application ID requesting the transaction.
        http_request_data: The encoded request.
      Returns:
        An encoded transaction protocol buffer with a unique handler.
      """
      global datastore_access
      begin_transaction_req_pb = datastore_pb.BeginTransactionRequest(
        http_request_data)
      multiple_eg = False
      if begin_transaction_req_pb.has_allow_multiple_eg():
        multiple_eg = bool(begin_transaction_req_pb.allow_multiple_eg())
  
      handle = None
      transaction_pb = datastore_pb.Transaction()
  
      if READ_ONLY:
        logger.warning('Unable to begin transaction in read-only mode: {}'.
          format(begin_transaction_req_pb))
        return (transaction_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        handle = datastore_access.setup_transaction(app_id, multiple_eg)
      except dbconstants.InternalError as error:
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except dbconstants.BadRequest as error:
        return '', datastore_pb.Error.BAD_REQUEST, str(error)
      except (zktransaction.ZKInternalException,
              dbconstants.AppScaleDBConnectionError) as error:
        logger.exception('Unable to begin transaction')
        return (transaction_pb.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                str(error))
  
      transaction_pb.set_app(app_id)
      transaction_pb.set_handle(handle)
      return (transaction_pb.Encode(), 0, "")
  
    def commit_transaction_request(self, app_id, http_request_data):
      """ Handles the commit phase of a transaction.
  
      Args:
        app_id: The application ID requesting the transaction commit.
        http_request_data: The encoded request of datastore_pb.Transaction.
      Returns:
        An encoded protocol buffer commit response.
      """
      global datastore_access
  
      if READ_ONLY:
        commitres_pb = datastore_pb.CommitResponse()
        transaction_pb = datastore_pb.Transaction(http_request_data)
        logger.warning('Unable to commit in read-only mode: {}'.
          format(transaction_pb))
        return (commitres_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      return datastore_access.commit_transaction(app_id, http_request_data)
  
    def rollback_transaction_request(self, app_id, http_request_data):
      """ Handles the rollback phase of a transaction.
  
      Args:
        app_id: The application ID requesting the rollback.
        http_request_data: The encoded request.
      Returns:
        An encoded protocol buffer void response.
      """
      global datastore_access
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to rollback in read-only mode: {}'.
          format(http_request_data))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        return datastore_access.rollback_transaction(app_id, http_request_data)
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {} for {}'.
          format(http_request_data, app_id))
        return (response.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except Exception:
        logger.exception('Unable to rollback transaction')
        return(response.Encode(),
               datastore_pb.Error.INTERNAL_ERROR,
               "Unable to rollback for this transaction")
  
    def run_query(self, http_request_data):
      """ High level function for running queries.
  
      Args:
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        Returns an encoded query response.
      """
      global datastore_access
      query = datastore_pb.Query(http_request_data)
      clone_qr_pb = UnprocessedQueryResult()
      try:
        datastore_access._dynamic_run_query(query, clone_qr_pb)
      except zktransaction.ZKBadRequest, zkie:
        logger.exception('Illegal arguments in transaction during {}'.
          format(query))
        return (clone_qr_pb.Encode(),
                datastore_pb.Error.BAD_REQUEST,
                "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {}'.format(query))
        clone_qr_pb.set_more_results(False)
        return (clone_qr_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.format(query))
        clone_qr_pb.set_more_results(False)
        return (clone_qr_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on put.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during query')
        clone_qr_pb.set_more_results(False)
        return (clone_qr_pb.Encode(),
               datastore_pb.Error.INTERNAL_ERROR,
               "Datastore connection error on run_query request.")
      return clone_qr_pb.Encode(), 0, ""
  
    def create_index_request(self, app_id, http_request_data):
      """ High level function for creating composite indexes.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.Integer64Proto()
  
      if READ_ONLY:
        logger.warning('Unable to create in read-only mode: {}'.
          format(request))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        index_id = datastore_access.create_composite_index(app_id, request)
        response.set_value(index_id)
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during index creation')
        response.set_value(0)
        return (response.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on create index request.")
      return response.Encode(), 0, ""
  
    def update_index_request(self, app_id, http_request_data):
      """ High level function for updating a composite index.
  
      Args:
        app_id: A string containing the application ID.
        http_request_data: A string containing the protocol buffer request
          from the AppServer.
      Returns:
         A tuple containing an encoded response, error code, and error details.
      """
      global datastore_access
      index = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to update in read-only mode: {}'.
          format(index))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      state = index.state()
      if state not in [index.READ_WRITE, index.WRITE_ONLY]:
        state_name = entity_pb.CompositeIndex.State_Name(state)
        error_message = 'Unable to update index because state is {}. '\
          'Index: {}'.format(state_name, index)
        logger.error(error_message)
        return response.Encode(), datastore_pb.Error.PERMISSION_DENIED,\
          error_message
      else:
        # Updating index asynchronously so we can return a response quickly.
        threading.Thread(target=datastore_access.update_composite_index,
          args=(app_id, index)).start()
  
      return response.Encode(), 0, ""
  
    def delete_index_request(self, app_id, http_request_data):
      """ Deletes a composite index for a given application.
    
      Args:
        app_id: Name of the application.
        http_request_data: A serialized CompositeIndices item
      Returns:
        A Tuple of an encoded entity_pb.VoidProto, error code, and 
        error explanation.
      """
      global datastore_access
      request = entity_pb.CompositeIndex(http_request_data)
      response = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to delete in read-only mode: {}'.
          format(request))
        return (response.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.delete_composite_index_metadata(app_id, request)
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during index deletion')
        return (response.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on delete index request.")
      return response.Encode(), 0, ""
  
    def get_indices_request(self, app_id):
      """ Gets the indices of the given application.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
        A Tuple of an encoded response, error code, and error explanation.
      """
      global datastore_access
      response = datastore_pb.CompositeIndices()
      try:
        indices = datastore_access.datastore_batch.get_indices(app_id)
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error while fetching indices for '
          '{}'.format(app_id))
        return (response.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on get indices request.")
      for index in indices:
        new_index = response.add_index()
        new_index.ParseFromString(index)
      return response.Encode(), 0, ""
  
    def allocate_ids_request(self, app_id, http_request_data):
      """ High level function for getting unique identifiers for entities.
  
      Args:
         app_id: Name of the application.
         http_request_data: Stores the protocol buffer request from the 
                 AppServer.
      Returns: 
         Returns an encoded response.
      Raises:
         NotImplementedError: when requesting a max id.
      """
      request = datastore_pb.AllocateIdsRequest(http_request_data)
      response = datastore_pb.AllocateIdsResponse()
  
      if request.has_max() and request.has_size():
        return (response.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Both size and max cannot be set.')
      if not (request.has_max() or request.has_size()):
        return (response.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Either size or max must be set.')
  
      if request.has_size():
        try:
          start, end = datastore_access.allocate_size(app_id, request.size())
        except dbconstants.AppScaleBadArg as error:
          return response.Encode(), datastore_pb.Error.BAD_REQUEST, str(error)
        except dbconstants.AppScaleDBConnectionError as error:
          return response.Encode(), datastore_pb.Error.INTERNAL_ERROR, str(error)
      else:
        try:
          start, end = datastore_access.allocate_max(app_id, request.max())
        except dbconstants.AppScaleBadArg as error:
          return response.Encode(), datastore_pb.Error.BAD_REQUEST, str(error)
        except dbconstants.AppScaleDBConnectionError as error:
          return response.Encode(), datastore_pb.Error.INTERNAL_ERROR, str(error)
  
      response.set_start(start)
      response.set_end(end)
      return response.Encode(), 0, ""
  
    @staticmethod
    @gen.coroutine
    def v4_allocate_ids_request(app_id, http_request_data):
      """ Reserves entity IDs so that they will not be re-allocated.
  
      Args:
        app_id: Name of the application.
        http_request_data: The protocol buffer request from the AppServer.
      Returns:
         Returns an encoded response.
      """
      request = datastore_v4_pb.AllocateIdsRequest(http_request_data)
      response = datastore_v4_pb.AllocateIdsResponse()
  
      if not request.reserve_list():
        raise gen.Return((response.Encode(), datastore_v4_pb.Error.BAD_REQUEST,
                          'Request must include reserve list'))
  
      ids = [key.path_element_list()[-1].id() for key in request.reserve_list()]
      datastore_access.reserve_ids(app_id, ids)
  
      # Forward request to other datastore servers in order to adjust any blocks
      # they've already allocated.
      client = AsyncHTTPClient()
      headers = {'appdata': app_id}
  
      futures = []
      for server in datastore_servers:
        ip, port = server.split(':')
        port = int(port)
        if ip == options.private_ip and port == options.port:
          continue
  
        url = 'http://{}:{}/reserve-keys'.format(ip, port)
        future = client.fetch(url, method='POST', headers=headers,
                              body=http_request_data)
        futures.append(future)
  
      for future in futures:
        yield future
  
      raise gen.Return((response.Encode(), 0, ''))
  
    def put_request(self, app_id, http_request_data):
      """ High level function for doing puts.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        Returns an encoded put response.
      """
      global datastore_access
  
      putreq_pb = datastore_pb.PutRequest(http_request_data)
      putresp_pb = datastore_pb.PutResponse()
  
      if READ_ONLY:
        logger.warning('Unable to put in read-only mode: {}'.
          format(putreq_pb))
        return (putresp_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.dynamic_put(app_id, putreq_pb, putresp_pb)
        return (putresp_pb.Encode(), 0, "")
      except dbconstants.InternalError as error:
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except dbconstants.BadRequest as error:
        return '', datastore_pb.Error.BAD_REQUEST, str(error)
      except zktransaction.ZKBadRequest as zkie:
        logger.exception('Illegal argument during {}'.format(putreq_pb))
        return (putresp_pb.Encode(),
              datastore_pb.Error.BAD_REQUEST,
              "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException as error:
        logger.exception('ZKInternalException during put')
        return (putresp_pb.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                str(error))
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.
          format(putreq_pb))
        return (putresp_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on put.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during put')
        return (putresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on put.")
  
  
    def get_request(self, app_id, http_request_data):
      """ High level function for doing gets.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        An encoded get response.
      """
      global datastore_access
      getreq_pb = datastore_pb.GetRequest(http_request_data)
      getresp_pb = datastore_pb.GetResponse()
      try:
        datastore_access.dynamic_get(app_id, getreq_pb, getresp_pb)
      except zktransaction.ZKBadRequest as zkie:
        logger.exception('Illegal argument during {}'.format(getreq_pb))
        return (getresp_pb.Encode(),
                datastore_pb.Error.BAD_REQUEST,
                "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {}'.format(getreq_pb))
        return (getresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.
          format(getreq_pb))
        return (getresp_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on get.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during get')
        return (getresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on get.")
  
      return getresp_pb.Encode(), 0, ""
  
    def delete_request(self, app_id, http_request_data):
      """ High level function for doing deletes.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
      Returns:
        An encoded delete response.
      """
      global datastore_access
  
      delreq_pb = datastore_pb.DeleteRequest( http_request_data )
      delresp_pb = api_base_pb.VoidProto()
  
      if READ_ONLY:
        logger.warning('Unable to delete in read-only mode: {}'.
          format(delreq_pb))
        return (delresp_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.dynamic_delete(app_id, delreq_pb)
        return (delresp_pb.Encode(), 0, "")
      except dbconstants.InternalError as error:
        return '', datastore_pb.Error.INTERNAL_ERROR, str(error)
      except dbconstants.BadRequest as error:
        return '', datastore_pb.Error.BAD_REQUEST, str(error)
      except zktransaction.ZKBadRequest as zkie:
        logger.exception('Illegal argument during {}'.format(delreq_pb))
        return (delresp_pb.Encode(),
                datastore_pb.Error.BAD_REQUEST,
                "Illegal arguments for transaction. {0}".format(str(zkie)))
      except zktransaction.ZKInternalException:
        logger.exception('ZKInternalException during {}'.format(delreq_pb))
        return (delresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Internal error with ZooKeeper connection.")
      except zktransaction.ZKTransactionException:
        logger.exception('Concurrent transaction during {}'.
          format(delreq_pb))
        return (delresp_pb.Encode(),
                datastore_pb.Error.CONCURRENT_TRANSACTION,
                "Concurrent transaction exception on delete.")
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error during delete')
        return (delresp_pb.Encode(),
                datastore_pb.Error.INTERNAL_ERROR,
                "Datastore connection error on delete.")
  
    def add_actions_request(self, app_id, http_request_data, service_id,
                            version_id):
      """ High level function for adding transactional tasks.
  
      Args:
        app_id: Name of the application.
        http_request_data: Stores the protocol buffer request from the AppServer.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      Returns:
        An encoded AddActions response.
      """
      global datastore_access
  
      req_pb = taskqueue_service_pb.TaskQueueBulkAddRequest(http_request_data)
      resp_pb = taskqueue_service_pb.TaskQueueBulkAddResponse()
  
      if service_id is None:
        return (resp_pb.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Module header must be defined')
  
      if version_id is None:
        return (resp_pb.Encode(), datastore_pb.Error.BAD_REQUEST,
                'Version header must be defined')
  
      if READ_ONLY:
        logger.warning('Unable to add transactional tasks in read-only mode')
        return (resp_pb.Encode(), datastore_pb.Error.CAPABILITY_DISABLED,
          'Datastore is in read-only mode.')
  
      try:
        datastore_access.dynamic_add_actions(app_id, req_pb, service_id,
                                             version_id)
        return resp_pb.Encode(), 0, ""
      except dbconstants.ExcessiveTasks as error:
        return (resp_pb.Encode(), datastore_pb.Error.BAD_REQUEST, str(error))
      except dbconstants.AppScaleDBConnectionError:
        logger.exception('DB connection error')
        return (resp_pb.Encode(), datastore_pb.Error.INTERNAL_ERROR,
                'Datastore connection error when adding transaction tasks.')
  
  
  def create_server_node():
    """ Creates a server registration entry in ZooKeeper. """
    try:
      zookeeper.handle.create(server_node, ephemeral=True)
    except NodeExistsError:
      # If the server gets restarted, the old node may exist for a short time.
      zookeeper.handle.delete(server_node)
      zookeeper.handle.create(server_node, ephemeral=True)
  
    logger.info('Datastore registered at {}'.format(server_node))
  
  
  def zk_state_listener(state):
    """ Handles changes to ZooKeeper connection state.
  
    Args:
      state: A string specifying the new ZooKeeper connection state.
    """
    if state == KazooState.CONNECTED:
      persistent_create_server_node = retry_data_watch_coroutine(
        server_node, create_server_node)
      IOLoop.instance().add_callback(persistent_create_server_node)
  
  
  def update_servers(new_servers):
    """ Updates the record of active datastore servers.
  
    Args:
      new_servers: A list of strings identifying server locations.
    """
    to_remove = [server for server in datastore_servers
                 if server not in new_servers]
    for old_server in to_remove:
      datastore_servers.remove(old_server)
  
    for new_server in new_servers:
      if new_server not in datastore_servers:
        datastore_servers.add(new_server)
  
  
  def update_servers_watch(new_servers):
    """ Updates the record of active datastore servers.
  
    Args:
      new_servers: A list of strings identifying server locations.
    """
    main_io_loop = IOLoop.instance()
    main_io_loop.add_callback(update_servers, new_servers)
  
  
  pb_application = tornado.web.Application([
    ('/clear', ClearHandler),
    ('/read-only', ReadOnlyHandler),
    ('/reserve-keys', ReserveKeysHandler),
    (r'/*', MainHandler),
  ])
  
  
  def main():
    """ Starts a web service for handing datastore requests. """
  
    global datastore_access
    global server_node
    global zookeeper
    zookeeper_locations = appscale_info.get_zk_locations_string()
  
    parser = argparse.ArgumentParser()
    parser.add_argument('-t', '--type', choices=dbconstants.VALID_DATASTORES,
                        default=dbconstants.VALID_DATASTORES[0],
                        help='Database type')
    parser.add_argument('-p', '--port', type=int,
                        default=dbconstants.DEFAULT_PORT,
                        help='Datastore server port')
    parser.add_argument('-v', '--verbose', action='store_true',
                        help='Output debug-level logging')
    args = parser.parse_args()
  
    if args.verbose:
      logger.setLevel(logging.DEBUG)
  
    options.define('private_ip', appscale_info.get_private_ip())
    options.define('port', args.port)
  
    server_node = '{}/{}:{}'.format(DATASTORE_SERVERS_NODE, options.private_ip,
                                    options.port)
  
    datastore_batch = DatastoreFactory.getDatastore(
      args.type, log_level=logger.getEffectiveLevel())
    zookeeper = zktransaction.ZKTransaction(
      host=zookeeper_locations, db_access=datastore_batch,
      log_level=logger.getEffectiveLevel())
  
    zookeeper.handle.add_listener(zk_state_listener)
    zookeeper.handle.ensure_path(DATASTORE_SERVERS_NODE)
    # Since the client was started before adding the listener, make sure the
    # server node gets created.
    zk_state_listener(zookeeper.handle.state)
    zookeeper.handle.
  
  [... the rest is omitted ...]
files.old_path: AppDB/appscale/datastore/scripts/datastore.py
files.new_contents: 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([])
  
      # Ignore lock path prefix when sorting contenders.
      lockname = self._NODE_NAME
      for child_list in children:
        child_list.sort(key=lambda c: c[c.find(lockname) + len(lockname):])
      return children
  
    def _find_nodes(self):
      """ Retrieve a list of paths this lock has created.
  
      Returns:
        A list of ZooKeeper paths.
      """
      nodes = []
      for path in self.paths:
        try:
          children = self.client.get_children(path)
        except NoNodeError:
          children = []
  
        node = None
        for child in children:
          if child.startswith(self.prefix):
            node = child
        nodes.append(node)
      return nodes
  
    def _delete_nodes(self, nodes):
      """ Remove ZooKeeper nodes.
  
      Args:
        nodes: A list of nodes to delete.
      """
      for index, node in enumerate(nodes):
        if node is None:
          continue
        self.client.delete(self.paths[index] + "/" + node)
  
    def _best_effort_cleanup(self):
      """ Attempt to delete nodes that this lock has created. """
      try:
        nodes = self._find_nodes()
        self._delete_nodes(nodes)
      except KazooException:
        pass
  
    def release(self):
      """ Release the lock immediately. """
      self.client.retry(self._inner_release)
  
      # Try to clean up the group lock path.
      for path in self.paths:
        try:
          self.client.delete(path)
        except (NotEmptyError, NoNodeError):
          pass
      return
  
    def _inner_release(self):
      """ Release the lock by removing created nodes. """
      if not self.is_acquired:
        return False
  
      try:
        self._delete_nodes(self.nodes)
      except NoNodeError:
        pass
  
      self.is_acquired = False
      self.nodes = [None for _ in self.paths]
      return True
  
    def __enter__(self):
      self.acquire()
  
    def __exit__(self, exc_type, exc_value, traceback):
      self.release()
  
  
  [... the rest is omitted ...]
files.new_path: AppDB/appscale/datastore/zkappscale/entity_lock.py
files.old_contents: 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:
          children.append([])
  
      # Ignore lock path prefix when sorting contenders.
      lockname = self._NODE_NAME
      for child_list in children:
        child_list.sort(key=lambda c: c[c.find(lockname) + len(lockname):])
      return children
  
    def _find_nodes(self):
      """ Retrieve a list of paths this lock has created.
  
      Returns:
        A list of ZooKeeper paths.
      """
      nodes = []
      for path in self.paths:
        try:
          children = self.client.get_children(path)
        except NoNodeError:
          children = []
  
        node = None
        for child in children:
          if child.startswith(self.prefix):
            node = child
        nodes.append(node)
      return nodes
  
    def _delete_nodes(self, nodes):
      """ Remove ZooKeeper nodes.
  
      Args:
        nodes: A list of nodes to delete.
      """
      for index, node in enumerate(nodes):
        if node is None:
          continue
        self.client.delete(self.paths[index] + "/" + node)
  
    def _best_effort_cleanup(self):
      """ Attempt to delete nodes that this lock has created. """
      try:
        nodes = self._find_nodes()
        self._delete_nodes(nodes)
      except KazooException:
        pass
  
    def release(self):
      """ Release the lock immediately. """
      self.client.retry(self._inner_release)
  
      # Try to clean up the group lock path.
      for path in self.paths:
        try:
          self.client.delete(path)
        except (NotEmptyError, NoNodeError):
          pass
      return
  
    def _inner_release(self):
      """ Release the lock by removing created nodes. """
      if not self.is_acquired:
        return False
  
      try:
        self._delete_nodes(self.nodes)
      except NoNodeError:
        pass
  
      self.is_acquired = False
      self.nodes = [None for _ in self.paths]
      return True
  
    def __enter__(self):
      self.acquire()
  
    def __exit__(self, exc_type, exc_value, traceback):
      self.release()
  
  
  [... the rest is omitted ...]
files.old_path: AppDB/appscale/datastore/zkappscale/entity_lock.py