<commit_before>#!/usr/bin/pyhon

"""
   Copyright 2013 The Trustees of Princeton University

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
"""


import storage.storagetypes as storagetypes

import protobufs.ms_pb2 as ms_pb2
import logging

import random
import os

import types
import errno
import time
import datetime
import collections

from volume import Volume
from gateway import Gateway, GATEWAY_CAP_COORDINATE

from common.msconfig import *

MSENTRY_TYPE_FILE = ms_pb2.ms_entry.MS_ENTRY_TYPE_FILE
MSENTRY_TYPE_DIR = ms_pb2.ms_entry.MS_ENTRY_TYPE_DIR
MSENTRY_TYPE_NONE = ms_pb2.ms_entry.MS_ENTRY_TYPE_NONE

def is_readable( user_id, volume_owner_id, ent_owner_id, ent_mode ):
   return ent_owner_id == user_id or volume_owner_id == user_id or (ent_mode & 0044) != 0

def is_writable( user_id, volume_owner_id, ent_owner_id, ent_mode ):
   return ent_owner_id == user_id or volume_owner_id == user_id or (ent_mode & 0022) != 0


def latest_time( mtime_sec, mtime_nsec ):
   now_sec, now_nsec = storagetypes.clock_gettime()

   if mtime_sec < now_sec or (mtime_sec == now_sec and mtime_nsec < now_nsec):
      mtime_sec = now_sec
      mtime_nsec = now_nsec + 1
      if mtime_nsec > 1e9:
         mtime_nsec = 0
         mtime_sec += 1

   return (mtime_sec, mtime_nsec)

   
class MSEntryShard(storagetypes.Object):
   # Sharded component of an MSEntry.
   # NOTE: size doesn't change for directories, but the mtime_sec, mtime_nsec fields will.
   # They don't have to increase monotonically for directories---all the UG needs to determine is that the directory
   # has changed at all.  The concept of a "latest" shard is applicable only for files.
   mtime_sec = storagetypes.Integer(default=0, indexed=False)
   mtime_nsec = storagetypes.Integer(default=0, indexed=False)
   manifest_mtime_sec = storagetypes.Integer(default=0, indexed=False)
   manifest_mtime_nsec = storagetypes.Integer(default=0, indexed=False)
   size = storagetypes.Integer(default=0, indexed=False )
   write_nonce = storagetypes.Integer( default=0, indexed=False )
   xattr_nonce = storagetypes.Integer( default=0, indexed=False )
   nonce_ts = storagetypes.Integer( default=0, indexed=False )          # for finding the latest write or xattr nonces
   
   # version of the MSEntry we're associated with
   msentry_version = storagetypes.Integer(default=0, indexed=False )

   # volume ID of the MSEntry we're associated with
   msentry_volume_id = storagetypes.Integer()
   
   
   @classmethod
   def manifest_modtime_max( cls, m1, m2 ):
      # return the later of two MSEntryShards, based on manifest_modtime
      if m1.manifest_mtime_sec < m2.manifest_mtime_sec:
         return m2

      elif m1.manifest_mtime_sec > m2.manifest_mtime_sec:
         return m1

      elif m1.manifest_mtime_nsec < m2.manifest_mtime_nsec:
         return m2

      elif m1.manifest_mtime_nsec > m2.manifest_mtime_nsec:
         return m1

      return m1

   @classmethod
   def size_max( cls, sz1, sz2 ):
      if sz1.size > sz2.size:
         return sz1

      return sz2
   
   @classmethod 
   def nonce_max( cls, n1, n2 ):
      if n1.nonce_ts > n2.nonce_ts:
         return n1 
   
      return n2 
   
   @classmethod
   def get_latest_attr( cls, ent, shards, attr ):
      mm = MSEntryShard.get_latest_shard( ent, shards )
      if mm != None:
         return getattr( mm, attr )
      else:
         return None
   
   @classmethod
   def get_latest_shard( cls, ent, shards ):
      s = None
      latest_version = ent.version
      latest_shard = None
      for shard in shards:
         if s == None:
            s = shard
            continue

         if shard == None:
            continue

         if shard.msentry_version != latest_version:
            continue
         
         s = MSEntryShard.nonce_max( s, shard )
      
      return s

   @classmethod
   def get_mtime_from_shards( cls, ent, shards ):
      mm = MSEntryShard.get_latest_shard( ent, shards )
      if mm is not None:
         return (mm.mtime_sec, mm.mtime_nsec)
      else:
         return (None, None)

   @classmethod
   def get_size_from_shards( cls, ent, shards ):
      # NOTE: size can never decrease for a given version.
      # truncate() will re-version a file
      sz = None
      latest_version = ent.version
      for shard in shards:
         if sz == None:
            sz = shard
            continue

         if shard == None:
            continue

         if shard.msentry_version != latest_version:
            continue

         sz = MSEntryShard.size_max( sz, shard )

      return sz.size
   
   @classmethod 
   def get_manifest_mtime_from_shards( cls, ent, shards ):
      # get the latest manifest modification time
      mm = None
      latest_version = ent.version
      latest_shard = None
      for shard in shards:
         if mm == None:
            mm = shard
            continue

         if shard == None:
            continue

         if shard.msentry_version != latest_version:
            continue
         
         mm = MSEntryShard.manifest_modtime_max( mm, shard )
      
      if mm is not None:
         return (mm.manifest_mtime_sec, mm.manifest_mtime_nsec)
      else:
         return (None, None)
   
   @classmethod 
   def Delete_ByVolume( cls, volume_id, async=False ):
      """
      Asynchronously delete by Volume
      """
      def __delete_shard_mapper( ent_shard_key ):
         ent_shard_key.delete()
      
      return MSEntryShard.ListAll( {"MSEntryShard.msentry_volume_id ==": volume_id}, map_func=__delete_shard_mapper, keys_only=True, async=async )
   
   
class MSEntryNameHolder( storagetypes.Object ):
   """
   Exists to prove that an MSEntry record exists in a parent under a given name.
   There is one of these per MSEntry.
   """
   volume_id = storagetypes.Integer( default=-1 )
   parent_id = storagetypes.String( default="None", indexed=False )
   file_id = storagetypes.String( default="None", indexed=False )
   name = storagetypes.String( default="", indexed=False )
   
   @classmethod
   def make_key_name( cls, volume_id, parent_id, name ):
      return "MSEntryNameHolder: volume_id=%s,parent_id=%s,name=%s" % (volume_id, parent_id, name)
   
   @classmethod
   def create_async( cls, _volume_id, _parent_id, _file_id, _name ):
      return MSEntryNameHolder.get_or_insert_async( MSEntryNameHolder.make_key_name( _volume_id, _parent_id, _name ), volume_id=_volume_id, parent_id=_parent_id, file_id=_file_id, name=_name )
   
   @classmethod
   def Delete_ByVolume( cls, volume_id, async=False ):
      """
      Asynchronously delete by volume.
      """
      def __delete_nameholder_mapper( ent_nameholder_key ):
         ent_nameholder_key.delete()
         
      return MSEntryNameHolder.ListAll( {"MSEntryNameHolder.volume_id ==": volume_id}, map_func=__delete_nameholder_mapper, keys_only=True, async=async )
      
   
   
class MSEntryXAttr( storagetypes.Object ):
   """
   A single extended attribute for an MSEntry.
   There are many of these per MSEntry.
   """
   
   file_id = storagetypes.String( default="None" )              # has to be a string, since this is an unsigned 64-bit int (and GAE only supports signed 64-bit int)
   volume_id = storagetypes.Integer( default=-1 )
   
   xattr_name = storagetypes.String( default="None" )
   xattr_value = storagetypes.Text()
   
   nonce = storagetypes.Integer( default=0 )
   
   owner = storagetypes.Integer( default=0 )     # 0 means "anyone"
   mode = storagetypes.Integer(default=0777)
   
   XATTR_OWNER_ANYONE = 0
   
   @classmethod 
   def make_key_name( cls, volume_id, file_id, name ):
      return "MSEntryXAttr: volume_id=%s,file_id=%s,name=%s" % (volume_id, file_id, name)
   
   @classmethod 
   def cache_listing_key_name( cls, volume_id, file_id ):
      return "MSEntryXAttr: volume_id=%s,file_id=%s" % (volume_id, file_id)
   
   @classmethod
   def ListXAttrs( cls, volume, msent, requester_owner_id, caller_is_admin=False ):
      """
      Get the visible names of the xattrs for this MSEntry.
      """
      cached_listing_name = MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id )
      names_and_metadata = storagetypes.memcache.get( cached_listing_name )
      if names_and_metadata is None:
            
         names_and_metadata = cls.ListAll( {"MSEntryXAttr.file_id ==": msent.file_id, "MSEntryXAttr.volume_id ==": msent.volume_id}, projection=["xattr_name", "owner", "mode"] )
         
         storagetypes.memcache.set( cached_listing_name, names_and_metadata );
      
      
      # return the list of attributes that we can access
      def is_accessible( xattr ):
         if xattr.owner != cls.XATTR_OWNER_ANYONE and xattr.owner != requester_owner_id and (xattr.mode & 0066) == 0:
            return None
         else:
            return xattr.xattr_name
      
      visible_names = filter( lambda x: x is not None, map( is_accessible, names_and_metadata ))
      
      return 0, visible_names
   
   @classmethod
   def ReadXAttr( cls, volume_id, file_id, xattr_name ):
      """
      Get an extended attribute record from the datastore
      Fail with ENOATTR (ENODATA) if there is no such attribute.
      """
      xattr_key_name = MSEntryXAttr.make_key_name( volume_id, file_id, xattr_name )
      xattr = storagetypes.memcache.get( xattr_key_name )
      rc = 0
     
      if xattr == None:
         xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
         xattr = xattr_key.get()
         
         if xattr != None:
            # cache for later 
            storagetypes.memcache.set( xattr_key_name, xattr )
            
      
      if xattr == None:
         # not found 
         return (-errno.ENODATA, None)
         
      else:
         return (0, xattr)
      
   
   @classmethod 
   def GetXAttr( cls, volume, msent, xattr_name, requester_owner_id, caller_is_admin=False ):
      """
      Get an extended attribute's value.
      Fail with ENOATTR (ENODATA) if there is no such attribute
      """
      
      rc, xattr = cls.ReadXAttr( volume.volume_id, msent.file_id, xattr_name )
      if rc == 0:
         
         if cls.XAttrReadable( requester_owner_id, xattr, caller_is_admin ):
            return (rc, xattr.xattr_value)
         else:
            return (-errno.EACCES, None)
         
      else:
         return (rc, None)
      
   
   @classmethod 
   def SetXAttr( cls, volume, msent, xattr_name, xattr_value, create=False, replace=False, owner=XATTR_OWNER_ANYONE, mode=0777, caller_is_admin=False ):
      """
      Set an extended attribute for the MSEntry msent.
      If create == True, then replace only if it does NOT exist.  Otherwise fail with EEXIST.
      If replace == True, then replace only if it already exists.  Otherwise fail with ENOATTR (ENODATA)
      Setting create = replace = True fails with EINVAL.
      
      The caller must ensure that the msent is writable by the requester
      """
      
      if create == replace and create == True:
         return -errno.EINVAL
      
      xattr_key_name = MSEntryXAttr.make_key_name( msent.volume_id, msent.file_id, xattr_name )
      rc = 0
      
      if not replace:
         # put the attribute...
         nonce = random.randint( -2**63, 2**63 - 1 )
         xattr = MSEntryXAttr.get_or_insert( xattr_key_name, file_id=msent.file_id, volume_id=msent.volume_id, xattr_name=xattr_name, xattr_value=xattr_value, nonce=nonce, owner=owner, mode=mode )
         
         if xattr.nonce != nonce and create:
            # can't already exist
            # check permissions and return the appropriate error code 
            if not cls.XAttrWritable( owner, xattr, caller_is_admin ):
               # don't even reveal its existence
               log.error("XAttr %s is not writable by %s" % (xattr.name, owner))
               rc = -errno.EACCES
            else:
               # already existed
               rc = -errno.EEXIST
                  
         elif xattr.nonce != nonce:
            # didn't create; got back existing xattr
            # check permissions 
            if not cls.XAttrWritable( owner, xattr, caller_is_admin ):
               rc = -errno.EACCES
               
            else:
               # put the new attribute over it
               xattr = MSEntryXAttr( key=storagetypes.make_key( MSEntryXAttr, xattr_key_name ), file_id=msent.file_id, volume_id=volume.volume_id,
                                       xattr_name=xattr_name, xattr_value=xattr_value, mode=mode, owner=owner, nonce=nonce )
            
               xattr.put()
               
         else:
            # created; clear the cached listing for this entry
            cached_listing_name = MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id )
            storagetypes.memcache.delete( cached_listing_name )
         
      else:
         # only replace if it exists
         def put_if_exists():
            xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
            xattr = xattr_key.get()
            
            # sanity check 
            if xattr is None:
               return -errno.ENODATA
            
            # check permissions 
            if not cls.XAttrWritable( owner, xattr, caller_is_admin ):
               return -errno.EACCES
         
            xattr.xattr_value = xattr_value 
            xattr.put()
            
            return 0
         
         try:
            rc = storagetypes.transaction( lambda: put_if_exists() )
         except Exception, e:
            log.exception(e)
            rc = -errno.EAGAIN
         
      if rc == 0:
         # successfully put xattr
         # update the xattr_nonce in the msentry
         new_shard = MSEntry.update_shard( volume.num_shards, msent, xattr_nonce=random.randint( -2**63, 2**63 - 1 ) )
         new_shard.put()
         
         # clear cached xattr value, xattr listing, msentry, and msentry listing
         ent_cache_key_name = MSEntry.cache_key_name( msent.volume_id, msent.file_id )
         storagetypes.memcache.delete_multi( [xattr_key_name, MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id ), ent_cache_key_name] )
         
      return rc
      
   
   @classmethod 
   def XAttrIsOwner( cls, owner_id, xattr, caller_is_admin=False ):
      return xattr.owner == cls.XATTR_OWNER_ANYONE or owner_id == xattr.owner or caller_is_admin
   
   @classmethod 
   def XAttrReadable( cls, owner_id, xattr, caller_is_admin=False ):
      return cls.XAttrIsOwner( owner_id, xattr, caller_is_admin ) or (xattr.mode & 0044) != 0
   
   @classmethod
   def XAttrWritable( cls, owner_id, xattr, caller_is_admin=False ):
      return cls.XAttrIsOwner( owner_id, xattr, caller_is_admin ) or (xattr.mode & 0022) != 0
   
   @classmethod 
   def ChmodXAttr( cls, volume, file_id, xattr_name, new_mode, requester_owner_id, caller_is_admin=False ):
      """
      Set an extended attribute's access mode, atomically.
      Only the owner of the xattr can set the mode, unless caller_is_admin is True
      """
      def update_mode_txn( xattr_key_name, m ):
         
         xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
         xattr = xattr_key.get()
         
         if xattr is None:
            return -errno.ENOENT 
         
         # verify that the owner is "anyone", or that it's the owner of the msent 
         if not cls.XAttrIsOwner( requester_owner_id, xattr, caller_is_admin ):
            return -errno.EACCES
         
         xattr.mode = m 
         
         xattr.put()
         
         return 0
         
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
      try:
         rc = storagetypes.transaction( lambda: update_mode_txn( xattr_key_name, new_mode ) )
      except Exception, e:
         log.exception(e)
         rc = -errno.EAGAIN
         
      if rc == 0:
         # clear cached metadata
         cached_listing_name = MSEntryXAttr.cache_listing_key_name( volume.volume_id, file_id )
         cached_xattr_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
         
         storagetypes.memcache.delete_multi( [cached_xattr_name, cached_listing_name] )
         
      return rc
         
   
   @classmethod 
   def ChownXAttr( cls, volume, file_id, xattr_name, new_owner, requester_owner_id, caller_is_admin=False ):
      """
      Set an extended attribute's owner, atomically.
      Only the owner of the xattr can change the mode, unless caller_is_admin is true.
      """
      def update_owner_txn( xattr_key_name, o ):
         
         xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
         xattr = xattr_key.get()
         
         if xattr is None:
            return -errno.ENOENT 
         
         # verify that the owner is "anyone", or that it's the owner of msent 
         if not cls.XAttrIsOwner( requester_owner_id, xattr, caller_is_admin ):
            return -errno.EACCES 
         
         xattr.owner = o
         
         xattr.put()
         
         return 0
         
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
      try:
         rc = storagetypes.transaction( lambda: update_owner_txn( xattr_key_name, new_owner ) )
      except Exception, e:
         log.exception(e)
         rc = -errno.EAGAIN
      
      if rc == 0:
         # clear cached metadata
         cached_listing_name = MSEntryXAttr.cache_listing_key_name( volume.volume_id, file_id )
         cached_xattr_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
         
         storagetypes.memcache.delete_multi( [cached_xattr_name, cached_listing_name] )
         
      return rc
      
   @classmethod 
   def remove_and_uncache( cls, volume, msent, xattr_name ):
      """
      Remove and uncache an xattr.
      Updates the MSEntry shard.
      Use this with removexattr()
      """
      
      # get the xattr
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, msent.file_id, xattr_name )
      xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
      
      # get a new shard 
      new_shard = MSEntry.update_shard( volume.num_shards, msent, xattr_nonce=random.randint( -2**63, 2**63 - 1 ) )
      
      # delete the xattr and put the new shard
      delete_fut = xattr_key.delete_async()
      shard_fut = new_shard.put_async()
      
      storagetypes.wait_futures( [delete_fut, shard_fut] )
      
      # clear caches
      listing_key = MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id )
      ent_cache_key_name = MSEntry.cache_key_name( msent.volume_id, msent.file_id )
      storagetypes.memcache.delete_multi( [xattr_key_name, listing_key, ent_cache_key_name] )
      
   
   @classmethod 
   def delete_and_uncache( cls, volume_id, file_id, xattr_name ):
      """
      Delete and uncache an xattr.
      Does NOT update the associated MSEntry shard.
      Only use this when deleting an MSEntry.
      """
      # get the xattr
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, msent.file_id, xattr_name )
      xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
      
      # delete the xattr
      xattr_key.delete()
      
      # clear caches
      storagetypes.memcache.delete( xattr_key_name )
      
   @classmethod 
   def RemoveXAttr( cls, volume, msent, xattr_name, requester_owner_id, caller_is_admin=False ):
      """
      Delete an extended attribute from the MSEntry msent.
      """
      
      rc, xattr = cls.ReadXAttr( volume.volume_id, msent.file_id, xattr_name )
      
      if rc != 0:
         return rc
      
      if xattr is None:
         return -errno.ENODATA
      
      # xattr must be writable 
      if not cls.XAttrWritable( requester_owner_id, xattr, caller_is_admin ):
         return -errno.EACCES
      
      storagetypes.deferred.defer( MSEntryXAttr.remove_and_uncache, volume, msent, xattr_name )
      
      return 0
      
   
   @classmethod
   def Delete_ByFile( cls, volume_id, file_id, async=False ):
      """
      Delete all xattrs for a given file asynchronously
      """
      
      def __xattr_deferred_delete( xattr ):
         MSEntryXAttr.delete_and_uncache( volume_id, file_id, xattr.xattr_name )
         
      return cls.ListAll( {"MSEntryXAttr.file_id ==": file_id, "MSEntryXAttr.volume_id ==": volume_id}, projection=['xattr_name'], map_func=__xattr_deferred_delete, async=async )
   
   
   @classmethod
   def Delete_ByVolume( cls, volume_id, async=False ):
      """
      Delete all xattrs for a given volume asynchronously
      """
      def __xattr_deferred_delete( xattr ):
         MSEntryXAttr.delete_and_uncache( volume_id, xattr.file_id, xattr.xattr_name )
      
      return cls.ListAll( {"MSEntryXAttr.volume_id ==": volume_id}, projection=['xattr_name', 'file_id'], map_func=__xattr_deferred_delete, async=async )


class MSEntryVacuumLog( storagetypes.Object ):
   """
   Log of manifest metadata of an MSEntry, for use in garbage collection.
   UGs poll these in order to garbage-collect stale manifests and blocks from the RGs for an MSEntry.
   The MS remembers which previous manifests (and thus blocks) exist through a set of these records.
   """
   
   file_id = storagetypes.String( default="None" )              # has to be a string, since this is an unsigned 64-bit int (and GAE only supports signed 64-bit int)
   volume_id = storagetypes.Integer( default=-1 )
   
   version = storagetypes.Integer( default=-1, indexed=False )
   manifest_mtime_sec = storagetypes.Integer(default=0, indexed=False)
   manifest_mtime_nsec = storagetypes.Integer(default=0, indexed=False)
   
   affected_blocks = storagetypes.Integer( repeated=True )
   
   nonce = storagetypes.Integer( default=0, indexed=False )
   
   @classmethod
   def make_key_name( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec ):
      return "MSEntryVacuumLog: volume_id=%s,file_id=%s,version=%s,manifest_mtime_sec=%s,manifest_mtime_nsec=%s" % (volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec)
   
   @classmethod
   def create_async( cls, _volume_id, _file_id, _version, _manifest_mtime_sec, _manifest_mtime_nsec, _affected_blocks, nonce=None ):
      
      if _affected_blocks is None:
         _affected_blocks = []
         
      key_name = MSEntryVacuumLog.make_key_name( _volume_id, _file_id, _version, _manifest_mtime_sec, _manifest_mtime_nsec )
      
      if nonce is None:
         nonce = random.randint(-2**63, 2**63 - 1)
      
      return MSEntryVacuumLog.get_or_insert_async( key_name,
                                                   volume_id=_volume_id,
                                                   file_id=_file_id,
                                                   version=_version,
                                                   manifest_mtime_sec=_manifest_mtime_sec,
                                                   manifest_mtime_nsec=_manifest_mtime_nsec,
                                                   affected_blocks=_affected_blocks,
                                                   nonce=nonce )
   
   @classmethod 
   def delete_async( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec ):
      
      key_name = MSEntryVacuumLog.make_key_name( volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec )
      record_key = storagetypes.make_key( MSEntryVacuumLog, key_name )
      
      storagetypes.deferred.defer( MSEntryVacuumLog.delete_all, [record_key] )
      
      return True
   
   
   @classmethod 
   def Peek( cls, volume_id, file_id, async=False ):
      """
      Get the head of the log for this file.
      That is, the oldest manifest record.
      """
      
      log_head = MSEntryVacuumLog.ListAll( {"MSEntryVacuumLog.volume_id ==": volume_id, "MSEntryVacuumLog.file_id ==": file_id},
                                           limit=1,
                                           async=async )
      
      print "log head of /%s/%s" % (volume_id, file_id)
      print log_head 
      
      return log_head
   
   @classmethod 
   def Insert( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec, affected_blocks ):
      """
      Add a new manifest log record.
      Verify that the MSEntry exists afterwards, and undo (delete) the newly-inserted record if not
      """
      
      nonce = random.randint(-2**63, 2**63 - 1)
      
      rec_fut = cls.create_async( volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec, affected_blocks, nonce=nonce )
      
      rec = rec_fut.get_result()
      
      msent = MSEntry.ReadBase( volume_id, file_id )
      
      if msent is None or msent.deleted:
         # undo, if we created 
         if rec.nonce == nonce:
            cls.delete_async( rec.volume_id, rec.file_id, rec.version, rec.manifest_mtime_sec, rec.manifest_mtime_nsec )
            
      return rec
   
   
   @classmethod 
   def Remove( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec ):
      """
      Remove a manifest log record.
      Return -errno.ENOENT if it doesn't exist 
      """
      
      print "remove log head /%s/%s (version = %s, manifest_mtime_sec = %s, manifest_mtime_nsec = %s)" % (volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec)
      
      key_name = MSEntryVacuumLog.make_key_name( volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec )
      log_ent_key = storagetypes.make_key( MSEntryVacuumLog, key_name )
      
      log_ent = log_ent_key.get()
      
      if log_ent is None:
         return -errno.ENOENT 
      
      else:
         
         MSEntryVacuumLog.delete_all( [log_ent_key] )
         return 0
   

# the page information we cache
VersionedPage = collections.namedtuple( "VersionedPage", ["write_nonce", "file_ids", "have_more"] )

class MSEntry( storagetypes.Object ):
   """
   Syndicate metadata entry.
   Each entry is named by its file ID and volume ID.
   Each entry is in its own entity group, and is named by its file ID and volume ID.   
   """

   # stuff that can't be encrypted
   ftype = storagetypes.Integer( default=-1, indexed=False )
   file_id = storagetypes.String( default="None" )              # has to be a string, not an int, since file IDs are *unsigned* 64-bit numbers
   owner_id = storagetypes.Integer( default=-1 )
   coordinator_id = storagetypes.Integer( default=1 )
   volume_id = storagetypes.Integer( default=-1 )
   mode = storagetypes.Integer( default=0 )
   is_readable_in_list = storagetypes.Boolean( default=False )    # used to simplify queries
   parent_id = storagetypes.String( default="-1" )                # file_id value of parent directory
   deleted = storagetypes.Boolean( default=False, indexed=False ) # whether or not this directory is considered to be deleted
   version = storagetypes.Integer( default=0, indexed=False )
   
   # stuff that can be encrypted
   name = storagetypes.String( default="", indexed=False )
   ctime_sec = storagetypes.Integer( default=0, indexed=False )
   ctime_nsec = storagetypes.Integer( default=0, indexed=False )
   max_read_freshness = storagetypes.Integer( default=0, indexed=False )
   max_write_freshness = storagetypes.Integer( default=0, indexed=False )

   # stuff that can be filled in from a shard (but can't be encrypted)
   mtime_sec = storagetypes.Integer( default=-1, indexed=False )
   mtime_nsec = storagetypes.Integer( default=-1, indexed=False )
   manifest_mtime_sec = storagetypes.Integer( default=-1, indexed=False )               # NOTE: only used for files
   manifest_mtime_nsec = storagetypes.Integer( default=-1, indexed=False )              # NOTE: only used for files
   size = storagetypes.Integer( default=0, indexed=False )
   write_nonce = storagetypes.Integer( default=0, indexed=False )
   xattr_nonce = storagetypes.Integer( default=0, indexed=False )
   nonce_ts = storagetypes.Integer( default=0, indexed=False )          # internal monotonically increasing timestamp from coordinator for nonces

   # attributes that must be supplied on creation
   required_attrs = [
      "ftype",
      "file_id",
      "name",
      "version",
      "owner_id",
      "coordinator_id",
      "volume_id",
      "mtime_sec",
      "mtime_nsec",
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "ctime_sec",
      "ctime_nsec",
      "mode",
      "size"
   ]

   # attributes that uniquely identify this entry
   key_attrs = [
      "volume_id",
      "file_id",
   ]

   # required for create/update/delete calls
   call_attrs = [
      "volume_id",
      "file_id",
      "name",
      "parent_id",
      "coordinator_id",
      "manifest_mtime_sec",
      "manifest_mtime_nsec"
   ]

   # methods for generating default values for attributes (sharded or not)
   default_values = {
      "max_read_freshness": (lambda cls, attrs: 0),
      "max_write_freshness": (lambda cls, attrs: 0),
      "file_id": (lambda cls, attrs: "None")
   }

   # publicly readable attributes, sharded or not
   read_attrs = [
      "ftype",
      "version",
      "name",
      "ctime_sec",
      "ctime_nsec",
      "mtime_sec",
      "mtime_nsec",
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "owner_id",
      "coordinator_id",
      "volume_id",
      "mode",
      "size",
      "max_read_freshness",
      "max_write_freshness",
      "write_nonce",
      "xattr_nonce"
   ]

   # publicly writable attributes, sharded or not
   write_attrs = [
      "name",
      "version",
      "owner_id",
      "mode",
      "size",
      "mtime_sec",
      "mtime_nsec",
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "max_read_freshness",
      "max_write_freshness"
   ]

   # shard class
   shard_class = MSEntryShard
   
   # sharded fields
   shard_fields = [
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "mtime_sec",
      "mtime_nsec",
      "size",
      "msentry_version",
      "msentry_volume_id",
      "write_nonce",
      "xattr_nonce",
      "nonce_ts"
   ]

   # functions that read a sharded value from shards for an instance of this ent
   shard_readers = {
      "manifest_mtime_sec": (lambda ent, shards: MSEntryShard.get_manifest_mtime_from_shards( ent, shards )[0]),
      "manifest_mtime_nsec": (lambda ent, shards: MSEntryShard.get_manifest_mtime_from_shards( ent, shards )[1]),
      "mtime_sec": (lambda ent, shards: MSEntryShard.get_mtime_from_shards( ent, shards )[0]),
      "mtime_nsec": (lambda ent, shards: MSEntryShard.get_mtime_from_shards( ent, shards )[1]),
      "size": (lambda ent, shards: MSEntryShard.get_size_from_shards( ent, shards )),
      "write_nonce": (lambda ent, shards: MSEntryShard.get_latest_attr( ent, shards, "write_nonce" )),
      "xattr_nonce": (lambda ent, shards: MSEntryShard.get_latest_attr( ent, shards, "xattr_nonce" )),
      "nonce_ts": (lambda ent, shards: MSEntryShard.get_latest_attr( ent, shards, "nonce_ts" ))
   }

   # functions that write a sharded value, given this ent
   shard_writers = {
      "manifest_mtime_sec": (lambda ent: ent.manifest_mtime_sec),
      "manifest_mtime_nsec": (lambda ent: ent.manifest_mtime_nsec),
      "mtime_sec": (lambda ent: ent.mtime_sec),
      "mtime_nsec": (lambda ent: ent.mtime_nsec),
      "msentry_version": (lambda ent: ent.version),
      "msentry_volume_id": (lambda ent: ent.volume_id),
      "size": (lambda ent: ent.size),
      "write_nonce": (lambda ent: ent.write_nonce),
      "xattr_nonce": (lambda ent: ent.xattr_nonce),
      "nonce_ts": (lambda ent: ent.nonce_ts)
   }
   
   @classmethod 
   def unserialize_id( self, i ):
      """
      Convert a numeric file_id to a string
      """
      return '{:016X}'.format(i)
   
   @classmethod
   def serialize_id( self, i_str ):
      """
      Convert a file_id as a hex string to an int.
      """
      return int(i_str, 16)
   
   def protobuf( self, pbent, **kwargs ):
      """
      Return an ms_entry instance containing this entry's data
      """

      pbent.type = kwargs.get( 'ftype', self.ftype )
      pbent.ctime_sec = kwargs.get( 'ctime_sec', self.ctime_sec )
      pbent.ctime_nsec = kwargs.get( 'ctime_nsec', self.ctime_nsec )
      pbent.mtime_sec = kwargs.get( 'mtime_sec', self.mtime_sec )
      pbent.mtime_nsec = kwargs.get( 'mtime_nsec', self.mtime_nsec )<commit_msg>Added a method for creating an empty protobuf<commit_after>#!/usr/bin/pyhon

"""
   Copyright 2013 The Trustees of Princeton University

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
"""


import storage.storagetypes as storagetypes

import protobufs.ms_pb2 as ms_pb2
import logging

import random
import os

import types
import errno
import time
import datetime
import collections

from volume import Volume
from gateway import Gateway, GATEWAY_CAP_COORDINATE

from common.msconfig import *

MSENTRY_TYPE_FILE = ms_pb2.ms_entry.MS_ENTRY_TYPE_FILE
MSENTRY_TYPE_DIR = ms_pb2.ms_entry.MS_ENTRY_TYPE_DIR
MSENTRY_TYPE_NONE = ms_pb2.ms_entry.MS_ENTRY_TYPE_NONE

def is_readable( user_id, volume_owner_id, ent_owner_id, ent_mode ):
   return ent_owner_id == user_id or volume_owner_id == user_id or (ent_mode & 0044) != 0

def is_writable( user_id, volume_owner_id, ent_owner_id, ent_mode ):
   return ent_owner_id == user_id or volume_owner_id == user_id or (ent_mode & 0022) != 0


def latest_time( mtime_sec, mtime_nsec ):
   now_sec, now_nsec = storagetypes.clock_gettime()

   if mtime_sec < now_sec or (mtime_sec == now_sec and mtime_nsec < now_nsec):
      mtime_sec = now_sec
      mtime_nsec = now_nsec + 1
      if mtime_nsec > 1e9:
         mtime_nsec = 0
         mtime_sec += 1

   return (mtime_sec, mtime_nsec)

   
class MSEntryShard(storagetypes.Object):
   # Sharded component of an MSEntry.
   # NOTE: size doesn't change for directories, but the mtime_sec, mtime_nsec fields will.
   # They don't have to increase monotonically for directories---all the UG needs to determine is that the directory
   # has changed at all.  The concept of a "latest" shard is applicable only for files.
   mtime_sec = storagetypes.Integer(default=0, indexed=False)
   mtime_nsec = storagetypes.Integer(default=0, indexed=False)
   manifest_mtime_sec = storagetypes.Integer(default=0, indexed=False)
   manifest_mtime_nsec = storagetypes.Integer(default=0, indexed=False)
   size = storagetypes.Integer(default=0, indexed=False )
   write_nonce = storagetypes.Integer( default=0, indexed=False )
   xattr_nonce = storagetypes.Integer( default=0, indexed=False )
   nonce_ts = storagetypes.Integer( default=0, indexed=False )          # for finding the latest write or xattr nonces
   
   # version of the MSEntry we're associated with
   msentry_version = storagetypes.Integer(default=0, indexed=False )

   # volume ID of the MSEntry we're associated with
   msentry_volume_id = storagetypes.Integer()
   
   
   @classmethod
   def manifest_modtime_max( cls, m1, m2 ):
      # return the later of two MSEntryShards, based on manifest_modtime
      if m1.manifest_mtime_sec < m2.manifest_mtime_sec:
         return m2

      elif m1.manifest_mtime_sec > m2.manifest_mtime_sec:
         return m1

      elif m1.manifest_mtime_nsec < m2.manifest_mtime_nsec:
         return m2

      elif m1.manifest_mtime_nsec > m2.manifest_mtime_nsec:
         return m1

      return m1

   @classmethod
   def size_max( cls, sz1, sz2 ):
      if sz1.size > sz2.size:
         return sz1

      return sz2
   
   @classmethod 
   def nonce_max( cls, n1, n2 ):
      if n1.nonce_ts > n2.nonce_ts:
         return n1 
   
      return n2 
   
   @classmethod
   def get_latest_attr( cls, ent, shards, attr ):
      mm = MSEntryShard.get_latest_shard( ent, shards )
      if mm != None:
         return getattr( mm, attr )
      else:
         return None
   
   @classmethod
   def get_latest_shard( cls, ent, shards ):
      s = None
      latest_version = ent.version
      latest_shard = None
      for shard in shards:
         if s == None:
            s = shard
            continue

         if shard == None:
            continue

         if shard.msentry_version != latest_version:
            continue
         
         s = MSEntryShard.nonce_max( s, shard )
      
      return s

   @classmethod
   def get_mtime_from_shards( cls, ent, shards ):
      mm = MSEntryShard.get_latest_shard( ent, shards )
      if mm is not None:
         return (mm.mtime_sec, mm.mtime_nsec)
      else:
         return (None, None)

   @classmethod
   def get_size_from_shards( cls, ent, shards ):
      # NOTE: size can never decrease for a given version.
      # truncate() will re-version a file
      sz = None
      latest_version = ent.version
      for shard in shards:
         if sz == None:
            sz = shard
            continue

         if shard == None:
            continue

         if shard.msentry_version != latest_version:
            continue

         sz = MSEntryShard.size_max( sz, shard )

      return sz.size
   
   @classmethod 
   def get_manifest_mtime_from_shards( cls, ent, shards ):
      # get the latest manifest modification time
      mm = None
      latest_version = ent.version
      latest_shard = None
      for shard in shards:
         if mm == None:
            mm = shard
            continue

         if shard == None:
            continue

         if shard.msentry_version != latest_version:
            continue
         
         mm = MSEntryShard.manifest_modtime_max( mm, shard )
      
      if mm is not None:
         return (mm.manifest_mtime_sec, mm.manifest_mtime_nsec)
      else:
         return (None, None)
   
   @classmethod 
   def Delete_ByVolume( cls, volume_id, async=False ):
      """
      Asynchronously delete by Volume
      """
      def __delete_shard_mapper( ent_shard_key ):
         ent_shard_key.delete()
      
      return MSEntryShard.ListAll( {"MSEntryShard.msentry_volume_id ==": volume_id}, map_func=__delete_shard_mapper, keys_only=True, async=async )
   
   
class MSEntryNameHolder( storagetypes.Object ):
   """
   Exists to prove that an MSEntry record exists in a parent under a given name.
   There is one of these per MSEntry.
   """
   volume_id = storagetypes.Integer( default=-1 )
   parent_id = storagetypes.String( default="None", indexed=False )
   file_id = storagetypes.String( default="None", indexed=False )
   name = storagetypes.String( default="", indexed=False )
   
   @classmethod
   def make_key_name( cls, volume_id, parent_id, name ):
      return "MSEntryNameHolder: volume_id=%s,parent_id=%s,name=%s" % (volume_id, parent_id, name)
   
   @classmethod
   def create_async( cls, _volume_id, _parent_id, _file_id, _name ):
      return MSEntryNameHolder.get_or_insert_async( MSEntryNameHolder.make_key_name( _volume_id, _parent_id, _name ), volume_id=_volume_id, parent_id=_parent_id, file_id=_file_id, name=_name )
   
   @classmethod
   def Delete_ByVolume( cls, volume_id, async=False ):
      """
      Asynchronously delete by volume.
      """
      def __delete_nameholder_mapper( ent_nameholder_key ):
         ent_nameholder_key.delete()
         
      return MSEntryNameHolder.ListAll( {"MSEntryNameHolder.volume_id ==": volume_id}, map_func=__delete_nameholder_mapper, keys_only=True, async=async )
      
   
   
class MSEntryXAttr( storagetypes.Object ):
   """
   A single extended attribute for an MSEntry.
   There are many of these per MSEntry.
   """
   
   file_id = storagetypes.String( default="None" )              # has to be a string, since this is an unsigned 64-bit int (and GAE only supports signed 64-bit int)
   volume_id = storagetypes.Integer( default=-1 )
   
   xattr_name = storagetypes.String( default="None" )
   xattr_value = storagetypes.Text()
   
   nonce = storagetypes.Integer( default=0 )
   
   owner = storagetypes.Integer( default=0 )     # 0 means "anyone"
   mode = storagetypes.Integer(default=0777)
   
   XATTR_OWNER_ANYONE = 0
   
   @classmethod 
   def make_key_name( cls, volume_id, file_id, name ):
      return "MSEntryXAttr: volume_id=%s,file_id=%s,name=%s" % (volume_id, file_id, name)
   
   @classmethod 
   def cache_listing_key_name( cls, volume_id, file_id ):
      return "MSEntryXAttr: volume_id=%s,file_id=%s" % (volume_id, file_id)
   
   @classmethod
   def ListXAttrs( cls, volume, msent, requester_owner_id, caller_is_admin=False ):
      """
      Get the visible names of the xattrs for this MSEntry.
      """
      cached_listing_name = MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id )
      names_and_metadata = storagetypes.memcache.get( cached_listing_name )
      if names_and_metadata is None:
            
         names_and_metadata = cls.ListAll( {"MSEntryXAttr.file_id ==": msent.file_id, "MSEntryXAttr.volume_id ==": msent.volume_id}, projection=["xattr_name", "owner", "mode"] )
         
         storagetypes.memcache.set( cached_listing_name, names_and_metadata );
      
      
      # return the list of attributes that we can access
      def is_accessible( xattr ):
         if xattr.owner != cls.XATTR_OWNER_ANYONE and xattr.owner != requester_owner_id and (xattr.mode & 0066) == 0:
            return None
         else:
            return xattr.xattr_name
      
      visible_names = filter( lambda x: x is not None, map( is_accessible, names_and_metadata ))
      
      return 0, visible_names
   
   @classmethod
   def ReadXAttr( cls, volume_id, file_id, xattr_name ):
      """
      Get an extended attribute record from the datastore
      Fail with ENOATTR (ENODATA) if there is no such attribute.
      """
      xattr_key_name = MSEntryXAttr.make_key_name( volume_id, file_id, xattr_name )
      xattr = storagetypes.memcache.get( xattr_key_name )
      rc = 0
     
      if xattr == None:
         xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
         xattr = xattr_key.get()
         
         if xattr != None:
            # cache for later 
            storagetypes.memcache.set( xattr_key_name, xattr )
            
      
      if xattr == None:
         # not found 
         return (-errno.ENODATA, None)
         
      else:
         return (0, xattr)
      
   
   @classmethod 
   def GetXAttr( cls, volume, msent, xattr_name, requester_owner_id, caller_is_admin=False ):
      """
      Get an extended attribute's value.
      Fail with ENOATTR (ENODATA) if there is no such attribute
      """
      
      rc, xattr = cls.ReadXAttr( volume.volume_id, msent.file_id, xattr_name )
      if rc == 0:
         
         if cls.XAttrReadable( requester_owner_id, xattr, caller_is_admin ):
            return (rc, xattr.xattr_value)
         else:
            return (-errno.EACCES, None)
         
      else:
         return (rc, None)
      
   
   @classmethod 
   def SetXAttr( cls, volume, msent, xattr_name, xattr_value, create=False, replace=False, owner=XATTR_OWNER_ANYONE, mode=0777, caller_is_admin=False ):
      """
      Set an extended attribute for the MSEntry msent.
      If create == True, then replace only if it does NOT exist.  Otherwise fail with EEXIST.
      If replace == True, then replace only if it already exists.  Otherwise fail with ENOATTR (ENODATA)
      Setting create = replace = True fails with EINVAL.
      
      The caller must ensure that the msent is writable by the requester
      """
      
      if create == replace and create == True:
         return -errno.EINVAL
      
      xattr_key_name = MSEntryXAttr.make_key_name( msent.volume_id, msent.file_id, xattr_name )
      rc = 0
      
      if not replace:
         # put the attribute...
         nonce = random.randint( -2**63, 2**63 - 1 )
         xattr = MSEntryXAttr.get_or_insert( xattr_key_name, file_id=msent.file_id, volume_id=msent.volume_id, xattr_name=xattr_name, xattr_value=xattr_value, nonce=nonce, owner=owner, mode=mode )
         
         if xattr.nonce != nonce and create:
            # can't already exist
            # check permissions and return the appropriate error code 
            if not cls.XAttrWritable( owner, xattr, caller_is_admin ):
               # don't even reveal its existence
               log.error("XAttr %s is not writable by %s" % (xattr.name, owner))
               rc = -errno.EACCES
            else:
               # already existed
               rc = -errno.EEXIST
                  
         elif xattr.nonce != nonce:
            # didn't create; got back existing xattr
            # check permissions 
            if not cls.XAttrWritable( owner, xattr, caller_is_admin ):
               rc = -errno.EACCES
               
            else:
               # put the new attribute over it
               xattr = MSEntryXAttr( key=storagetypes.make_key( MSEntryXAttr, xattr_key_name ), file_id=msent.file_id, volume_id=volume.volume_id,
                                       xattr_name=xattr_name, xattr_value=xattr_value, mode=mode, owner=owner, nonce=nonce )
            
               xattr.put()
               
         else:
            # created; clear the cached listing for this entry
            cached_listing_name = MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id )
            storagetypes.memcache.delete( cached_listing_name )
         
      else:
         # only replace if it exists
         def put_if_exists():
            xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
            xattr = xattr_key.get()
            
            # sanity check 
            if xattr is None:
               return -errno.ENODATA
            
            # check permissions 
            if not cls.XAttrWritable( owner, xattr, caller_is_admin ):
               return -errno.EACCES
         
            xattr.xattr_value = xattr_value 
            xattr.put()
            
            return 0
         
         try:
            rc = storagetypes.transaction( lambda: put_if_exists() )
         except Exception, e:
            log.exception(e)
            rc = -errno.EAGAIN
         
      if rc == 0:
         # successfully put xattr
         # update the xattr_nonce in the msentry
         new_shard = MSEntry.update_shard( volume.num_shards, msent, xattr_nonce=random.randint( -2**63, 2**63 - 1 ) )
         new_shard.put()
         
         # clear cached xattr value, xattr listing, msentry, and msentry listing
         ent_cache_key_name = MSEntry.cache_key_name( msent.volume_id, msent.file_id )
         storagetypes.memcache.delete_multi( [xattr_key_name, MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id ), ent_cache_key_name] )
         
      return rc
      
   
   @classmethod 
   def XAttrIsOwner( cls, owner_id, xattr, caller_is_admin=False ):
      return xattr.owner == cls.XATTR_OWNER_ANYONE or owner_id == xattr.owner or caller_is_admin
   
   @classmethod 
   def XAttrReadable( cls, owner_id, xattr, caller_is_admin=False ):
      return cls.XAttrIsOwner( owner_id, xattr, caller_is_admin ) or (xattr.mode & 0044) != 0
   
   @classmethod
   def XAttrWritable( cls, owner_id, xattr, caller_is_admin=False ):
      return cls.XAttrIsOwner( owner_id, xattr, caller_is_admin ) or (xattr.mode & 0022) != 0
   
   @classmethod 
   def ChmodXAttr( cls, volume, file_id, xattr_name, new_mode, requester_owner_id, caller_is_admin=False ):
      """
      Set an extended attribute's access mode, atomically.
      Only the owner of the xattr can set the mode, unless caller_is_admin is True
      """
      def update_mode_txn( xattr_key_name, m ):
         
         xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
         xattr = xattr_key.get()
         
         if xattr is None:
            return -errno.ENOENT 
         
         # verify that the owner is "anyone", or that it's the owner of the msent 
         if not cls.XAttrIsOwner( requester_owner_id, xattr, caller_is_admin ):
            return -errno.EACCES
         
         xattr.mode = m 
         
         xattr.put()
         
         return 0
         
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
      try:
         rc = storagetypes.transaction( lambda: update_mode_txn( xattr_key_name, new_mode ) )
      except Exception, e:
         log.exception(e)
         rc = -errno.EAGAIN
         
      if rc == 0:
         # clear cached metadata
         cached_listing_name = MSEntryXAttr.cache_listing_key_name( volume.volume_id, file_id )
         cached_xattr_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
         
         storagetypes.memcache.delete_multi( [cached_xattr_name, cached_listing_name] )
         
      return rc
         
   
   @classmethod 
   def ChownXAttr( cls, volume, file_id, xattr_name, new_owner, requester_owner_id, caller_is_admin=False ):
      """
      Set an extended attribute's owner, atomically.
      Only the owner of the xattr can change the mode, unless caller_is_admin is true.
      """
      def update_owner_txn( xattr_key_name, o ):
         
         xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
         xattr = xattr_key.get()
         
         if xattr is None:
            return -errno.ENOENT 
         
         # verify that the owner is "anyone", or that it's the owner of msent 
         if not cls.XAttrIsOwner( requester_owner_id, xattr, caller_is_admin ):
            return -errno.EACCES 
         
         xattr.owner = o
         
         xattr.put()
         
         return 0
         
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
      try:
         rc = storagetypes.transaction( lambda: update_owner_txn( xattr_key_name, new_owner ) )
      except Exception, e:
         log.exception(e)
         rc = -errno.EAGAIN
      
      if rc == 0:
         # clear cached metadata
         cached_listing_name = MSEntryXAttr.cache_listing_key_name( volume.volume_id, file_id )
         cached_xattr_name = MSEntryXAttr.make_key_name( volume.volume_id, file_id, xattr_name )
         
         storagetypes.memcache.delete_multi( [cached_xattr_name, cached_listing_name] )
         
      return rc
      
   @classmethod 
   def remove_and_uncache( cls, volume, msent, xattr_name ):
      """
      Remove and uncache an xattr.
      Updates the MSEntry shard.
      Use this with removexattr()
      """
      
      # get the xattr
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, msent.file_id, xattr_name )
      xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
      
      # get a new shard 
      new_shard = MSEntry.update_shard( volume.num_shards, msent, xattr_nonce=random.randint( -2**63, 2**63 - 1 ) )
      
      # delete the xattr and put the new shard
      delete_fut = xattr_key.delete_async()
      shard_fut = new_shard.put_async()
      
      storagetypes.wait_futures( [delete_fut, shard_fut] )
      
      # clear caches
      listing_key = MSEntryXAttr.cache_listing_key_name( msent.volume_id, msent.file_id )
      ent_cache_key_name = MSEntry.cache_key_name( msent.volume_id, msent.file_id )
      storagetypes.memcache.delete_multi( [xattr_key_name, listing_key, ent_cache_key_name] )
      
   
   @classmethod 
   def delete_and_uncache( cls, volume_id, file_id, xattr_name ):
      """
      Delete and uncache an xattr.
      Does NOT update the associated MSEntry shard.
      Only use this when deleting an MSEntry.
      """
      # get the xattr
      xattr_key_name = MSEntryXAttr.make_key_name( volume.volume_id, msent.file_id, xattr_name )
      xattr_key = storagetypes.make_key( MSEntryXAttr, xattr_key_name )
      
      # delete the xattr
      xattr_key.delete()
      
      # clear caches
      storagetypes.memcache.delete( xattr_key_name )
      
   @classmethod 
   def RemoveXAttr( cls, volume, msent, xattr_name, requester_owner_id, caller_is_admin=False ):
      """
      Delete an extended attribute from the MSEntry msent.
      """
      
      rc, xattr = cls.ReadXAttr( volume.volume_id, msent.file_id, xattr_name )
      
      if rc != 0:
         return rc
      
      if xattr is None:
         return -errno.ENODATA
      
      # xattr must be writable 
      if not cls.XAttrWritable( requester_owner_id, xattr, caller_is_admin ):
         return -errno.EACCES
      
      storagetypes.deferred.defer( MSEntryXAttr.remove_and_uncache, volume, msent, xattr_name )
      
      return 0
      
   
   @classmethod
   def Delete_ByFile( cls, volume_id, file_id, async=False ):
      """
      Delete all xattrs for a given file asynchronously
      """
      
      def __xattr_deferred_delete( xattr ):
         MSEntryXAttr.delete_and_uncache( volume_id, file_id, xattr.xattr_name )
         
      return cls.ListAll( {"MSEntryXAttr.file_id ==": file_id, "MSEntryXAttr.volume_id ==": volume_id}, projection=['xattr_name'], map_func=__xattr_deferred_delete, async=async )
   
   
   @classmethod
   def Delete_ByVolume( cls, volume_id, async=False ):
      """
      Delete all xattrs for a given volume asynchronously
      """
      def __xattr_deferred_delete( xattr ):
         MSEntryXAttr.delete_and_uncache( volume_id, xattr.file_id, xattr.xattr_name )
      
      return cls.ListAll( {"MSEntryXAttr.volume_id ==": volume_id}, projection=['xattr_name', 'file_id'], map_func=__xattr_deferred_delete, async=async )


class MSEntryVacuumLog( storagetypes.Object ):
   """
   Log of manifest metadata of an MSEntry, for use in garbage collection.
   UGs poll these in order to garbage-collect stale manifests and blocks from the RGs for an MSEntry.
   The MS remembers which previous manifests (and thus blocks) exist through a set of these records.
   """
   
   file_id = storagetypes.String( default="None" )              # has to be a string, since this is an unsigned 64-bit int (and GAE only supports signed 64-bit int)
   volume_id = storagetypes.Integer( default=-1 )
   
   version = storagetypes.Integer( default=-1, indexed=False )
   manifest_mtime_sec = storagetypes.Integer(default=0, indexed=False)
   manifest_mtime_nsec = storagetypes.Integer(default=0, indexed=False)
   
   affected_blocks = storagetypes.Integer( repeated=True )
   
   nonce = storagetypes.Integer( default=0, indexed=False )
   
   @classmethod
   def make_key_name( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec ):
      return "MSEntryVacuumLog: volume_id=%s,file_id=%s,version=%s,manifest_mtime_sec=%s,manifest_mtime_nsec=%s" % (volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec)
   
   @classmethod
   def create_async( cls, _volume_id, _file_id, _version, _manifest_mtime_sec, _manifest_mtime_nsec, _affected_blocks, nonce=None ):
      
      if _affected_blocks is None:
         _affected_blocks = []
         
      key_name = MSEntryVacuumLog.make_key_name( _volume_id, _file_id, _version, _manifest_mtime_sec, _manifest_mtime_nsec )
      
      if nonce is None:
         nonce = random.randint(-2**63, 2**63 - 1)
      
      return MSEntryVacuumLog.get_or_insert_async( key_name,
                                                   volume_id=_volume_id,
                                                   file_id=_file_id,
                                                   version=_version,
                                                   manifest_mtime_sec=_manifest_mtime_sec,
                                                   manifest_mtime_nsec=_manifest_mtime_nsec,
                                                   affected_blocks=_affected_blocks,
                                                   nonce=nonce )
   
   @classmethod 
   def delete_async( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec ):
      
      key_name = MSEntryVacuumLog.make_key_name( volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec )
      record_key = storagetypes.make_key( MSEntryVacuumLog, key_name )
      
      storagetypes.deferred.defer( MSEntryVacuumLog.delete_all, [record_key] )
      
      return True
   
   
   @classmethod 
   def Peek( cls, volume_id, file_id, async=False ):
      """
      Get the head of the log for this file.
      That is, the oldest manifest record.
      """
      
      log_head = MSEntryVacuumLog.ListAll( {"MSEntryVacuumLog.volume_id ==": volume_id, "MSEntryVacuumLog.file_id ==": file_id},
                                           limit=1,
                                           async=async )
      
      print "log head of /%s/%s" % (volume_id, file_id)
      print log_head 
      
      return log_head
   
   @classmethod 
   def Insert( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec, affected_blocks ):
      """
      Add a new manifest log record.
      Verify that the MSEntry exists afterwards, and undo (delete) the newly-inserted record if not
      """
      
      nonce = random.randint(-2**63, 2**63 - 1)
      
      rec_fut = cls.create_async( volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec, affected_blocks, nonce=nonce )
      
      rec = rec_fut.get_result()
      
      msent = MSEntry.ReadBase( volume_id, file_id )
      
      if msent is None or msent.deleted:
         # undo, if we created 
         if rec.nonce == nonce:
            cls.delete_async( rec.volume_id, rec.file_id, rec.version, rec.manifest_mtime_sec, rec.manifest_mtime_nsec )
            
      return rec
   
   
   @classmethod 
   def Remove( cls, volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec ):
      """
      Remove a manifest log record.
      Return -errno.ENOENT if it doesn't exist 
      """
      
      print "remove log head /%s/%s (version = %s, manifest_mtime_sec = %s, manifest_mtime_nsec = %s)" % (volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec)
      
      key_name = MSEntryVacuumLog.make_key_name( volume_id, file_id, version, manifest_mtime_sec, manifest_mtime_nsec )
      log_ent_key = storagetypes.make_key( MSEntryVacuumLog, key_name )
      
      log_ent = log_ent_key.get()
      
      if log_ent is None:
         return -errno.ENOENT 
      
      else:
         
         MSEntryVacuumLog.delete_all( [log_ent_key] )
         return 0
   

# the page information we cache
VersionedPage = collections.namedtuple( "VersionedPage", ["write_nonce", "file_ids", "have_more"] )

class MSEntry( storagetypes.Object ):
   """
   Syndicate metadata entry.
   Each entry is named by its file ID and volume ID.
   Each entry is in its own entity group, and is named by its file ID and volume ID.   
   """

   # stuff that can't be encrypted
   ftype = storagetypes.Integer( default=-1, indexed=False )
   file_id = storagetypes.String( default="None" )              # has to be a string, not an int, since file IDs are *unsigned* 64-bit numbers
   owner_id = storagetypes.Integer( default=-1 )
   coordinator_id = storagetypes.Integer( default=1 )
   volume_id = storagetypes.Integer( default=-1 )
   mode = storagetypes.Integer( default=0 )
   is_readable_in_list = storagetypes.Boolean( default=False )    # used to simplify queries
   parent_id = storagetypes.String( default="-1" )                # file_id value of parent directory
   deleted = storagetypes.Boolean( default=False, indexed=False ) # whether or not this directory is considered to be deleted
   version = storagetypes.Integer( default=0, indexed=False )
   
   # stuff that can be encrypted
   name = storagetypes.String( default="", indexed=False )
   ctime_sec = storagetypes.Integer( default=0, indexed=False )
   ctime_nsec = storagetypes.Integer( default=0, indexed=False )
   max_read_freshness = storagetypes.Integer( default=0, indexed=False )
   max_write_freshness = storagetypes.Integer( default=0, indexed=False )

   # stuff that can be filled in from a shard (but can't be encrypted)
   mtime_sec = storagetypes.Integer( default=-1, indexed=False )
   mtime_nsec = storagetypes.Integer( default=-1, indexed=False )
   manifest_mtime_sec = storagetypes.Integer( default=-1, indexed=False )               # NOTE: only used for files
   manifest_mtime_nsec = storagetypes.Integer( default=-1, indexed=False )              # NOTE: only used for files
   size = storagetypes.Integer( default=0, indexed=False )
   write_nonce = storagetypes.Integer( default=0, indexed=False )
   xattr_nonce = storagetypes.Integer( default=0, indexed=False )
   nonce_ts = storagetypes.Integer( default=0, indexed=False )          # internal monotonically increasing timestamp from coordinator for nonces

   # attributes that must be supplied on creation
   required_attrs = [
      "ftype",
      "file_id",
      "name",
      "version",
      "owner_id",
      "coordinator_id",
      "volume_id",
      "mtime_sec",
      "mtime_nsec",
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "ctime_sec",
      "ctime_nsec",
      "mode",
      "size"
   ]

   # attributes that uniquely identify this entry
   key_attrs = [
      "volume_id",
      "file_id",
   ]

   # required for create/update/delete calls
   call_attrs = [
      "volume_id",
      "file_id",
      "name",
      "parent_id",
      "coordinator_id",
      "manifest_mtime_sec",
      "manifest_mtime_nsec"
   ]

   # methods for generating default values for attributes (sharded or not)
   default_values = {
      "max_read_freshness": (lambda cls, attrs: 0),
      "max_write_freshness": (lambda cls, attrs: 0),
      "file_id": (lambda cls, attrs: "None")
   }

   # publicly readable attributes, sharded or not
   read_attrs = [
      "ftype",
      "version",
      "name",
      "ctime_sec",
      "ctime_nsec",
      "mtime_sec",
      "mtime_nsec",
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "owner_id",
      "coordinator_id",
      "volume_id",
      "mode",
      "size",
      "max_read_freshness",
      "max_write_freshness",
      "write_nonce",
      "xattr_nonce"
   ]

   # publicly writable attributes, sharded or not
   write_attrs = [
      "name",
      "version",
      "owner_id",
      "mode",
      "size",
      "mtime_sec",
      "mtime_nsec",
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "max_read_freshness",
      "max_write_freshness"
   ]

   # shard class
   shard_class = MSEntryShard
   
   # sharded fields
   shard_fields = [
      "manifest_mtime_sec",
      "manifest_mtime_nsec",
      "mtime_sec",
      "mtime_nsec",
      "size",
      "msentry_version",
      "msentry_volume_id",
      "write_nonce",
      "xattr_nonce",
      "nonce_ts"
   ]

   # functions that read a sharded value from shards for an instance of this ent
   shard_readers = {
      "manifest_mtime_sec": (lambda ent, shards: MSEntryShard.get_manifest_mtime_from_shards( ent, shards )[0]),
      "manifest_mtime_nsec": (lambda ent, shards: MSEntryShard.get_manifest_mtime_from_shards( ent, shards )[1]),
      "mtime_sec": (lambda ent, shards: MSEntryShard.get_mtime_from_shards( ent, shards )[0]),
      "mtime_nsec": (lambda ent, shards: MSEntryShard.get_mtime_from_shards( ent, shards )[1]),
      "size": (lambda ent, shards: MSEntryShard.get_size_from_shards( ent, shards )),
      "write_nonce": (lambda ent, shards: MSEntryShard.get_latest_attr( ent, shards, "write_nonce" )),
      "xattr_nonce": (lambda ent, shards: MSEntryShard.get_latest_attr( ent, shards, "xattr_nonce" )),
      "nonce_ts": (lambda ent, shards: MSEntryShard.get_latest_attr( ent, shards, "nonce_ts" ))
   }

   # functions that write a sharded value, given this ent
   shard_writers = {
      "manifest_mtime_sec": (lambda ent: ent.manifest_mtime_sec),
      "manifest_mtime_nsec": (lambda ent: ent.manifest_mtime_nsec),
      "mtime_sec": (lambda ent: ent.mtime_sec),
      "mtime_nsec": (lambda ent: ent.mtime_nsec),
      "msentry_version": (lambda ent: ent.version),
      "msentry_volume_id": (lambda ent: ent.volume_id),
      "size": (lambda ent: ent.size),
      "write_nonce": (lambda ent: ent.write_nonce),
      "xattr_nonce": (lambda ent: ent.xattr_nonce),
      "nonce_ts": (lambda ent: ent.nonce_ts)
   }
   
   @classmethod 
   def unserialize_id( self, i ):
      """
      Convert a numeric file_id to a string
      """
      return '{:016X}'.format(i)
   
   @classmethod
   def serialize_id( self, i_str ):
      """
      Convert a file_id as a hex string to an int.
      """
      return int(i_str, 16)
   
   @classmethod 
   def protobuf_empty( cls, pbent, **kwargs ):
      """
      Create an empty protobuf, suitable as a placeholder.
      """
      
      ment = MSEntry()
      ment.protobuf( pbent )
      return
      
   
   def protobuf( self, pbent, **kwargs ):
      """
      Return an ms_entry instance containing this entry's data
      """

      pbent.type = kwargs.get( 'ftype', self.ftype )
      pbent.ctime_sec = kwargs.get( 'ctime_sec', self.ctime_sec )
      pbent.ctime_nsec = kwargs.get( 'ctime_nsec', self.ctime_nsec )
      pbent.mtime_sec = kwargs.get( 'mtime_sec', self.mtime_sec )
      pbent.mtime_nsec = kwargs.get( 'mtime_nsec', self.mtime_nsec )
{'fraction_non_alphanumeric': 0.08515903192709276, 'fraction_numerical': 0.0056732452169714525, 'mean_word_length': 2.4851448703791346, 'pattern_counts': {'":': 60, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 38, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1946594', 'n_tokens_mistral': 22283, 'n_tokens_neox': 21788, 'n_words': 5226}