Synnefo

#PITHOS_BACKEND_BLOCK_PATH = '/tmp/pithos-data/'

# Enable and configure secondary rados storage for pithos

#PITHOS_RADOS_STORAGE = False

#PITHOS_RADOS_POOL_BLOCKS = 'blocks'

#PITHOS_RADOS_POOL_MAPS = 'maps'

from fileblocker import FileBlocker

def intersect(a, b):

    """ return the intersection of two lists """

    return list(set(a) & set(b))

def union(a, b):

    """ return the union of two lists """

    return list(set(a) | set(b))

        self.rblocker = None

        try:

            if params['blockpool']:

                from radosblocker import RadosBlocker

                self.rblocker = RadosBlocker(**params)

        except KeyError:

            pass

        self.fblocker = FileBlocker(**params)

        self.hashlen = self.fblocker.hashlen

        return self.fblocker.block_hash(data)

        r = []

        if self.rblocker:

            r = self.rblocker.block_ping(hashes)

        f = self.fblocker.block_ping(hashes)

        return union(r, f)

        return self.fblocker.block_retr(hashes)

        r_missing = []

        (hashes, f_missing) = self.fblocker.block_stor(blocklist)

        if self.rblocker:

            (_, r_missing) = self.rblocker.block_stor(blocklist)

        return (hashes, union(r_missing, f_missing))

        r_hash = None

        r_existed = True

        (f_hash, f_existed) = self.fblocker.block_delta(blkhash, offset, data)

        if self.rblocker:

            (r_hash, r_existed) = self.rblocker.block_delta(blkhash, offset,

                                                            data)

        if not r_hash and not f_hash:

        if self.rblocker and not r_hash:

            block = self.fblocker.block_retr((blkhash,))

            r_hash, r_existed = self.rblocker.block_stor((newblock,))

        return f_hash, 1 if r_existed and f_existed else 0

# Copyright 2011-2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from os import SEEK_CUR, SEEK_SET

from errno import ENOENT, EROFS

_zeros = ''

def zeros(nr):

    global _zeros

    size = len(_zeros)

    if nr == size:

        return _zeros

    if nr > size:

        _zeros += '\0' * (nr - size)

        return _zeros

    if nr < size:

        _zeros = _zeros[:nr]

        return _zeros

def file_sync_write_chunks(openfile, chunksize, offset, chunks, size=None):

    """Write given chunks to the given buffered file object.

       Writes never span across chunk boundaries.

       If size is given stop after or pad until size bytes have been written.

    """

    fwrite = openfile.write

    seek = openfile.seek

    padding = 0

    try:

        seek(offset * chunksize)

    except IOError:

        seek = None

        for x in xrange(offset):

            fwrite(zeros(chunksize))

    cursize = offset * chunksize

    for chunk in chunks:

        if padding:

            if seek:

                seek(padding - 1, SEEK_CUR)

                fwrite("\x00")

            else:

                fwrite(buffer(zeros(chunksize), 0, padding))

        if size is not None and cursize + chunksize >= size:

            chunk = chunk[:chunksize - (cursize - size)]

            fwrite(chunk)

            cursize += len(chunk)

            break

        fwrite(chunk)

        padding = chunksize - len(chunk)

    padding = size - cursize if size is not None else 0

    if padding <= 0:

        return

    q, r = divmod(padding, chunksize)

    for x in xrange(q):

        fwrite(zeros(chunksize))

    fwrite(buffer(zeros(chunksize), 0, r))

def file_sync_read_chunks(openfile, chunksize, nr, offset=0):

    """Read and yield groups of chunks from a buffered file object at offset.

       Reads never span accros chunksize boundaries.

    """

    fread = openfile.read

    remains = offset * chunksize

    seek = openfile.seek

    try:

        seek(remains)

    except IOError:

        seek = None

        while 1:

            s = fread(remains)

            remains -= len(s)

            if remains <= 0:

                break

    while nr:

        remains = chunksize

        chunk = ''

        while 1:

            s = fread(remains)

            if not s:

                if chunk:

                    yield chunk

                return

            chunk += s

            remains -= len(s)

            if remains <= 0:

                break

        yield chunk

        nr -= 1

class ContextFile(object):

    __slots__ = ("name", "fdesc", "create")

    def __init__(self, name, create=0):

        self.name = name

        self.fdesc = None

        self.create = create

        #self.dirty = 0

    def __enter__(self):

        name = self.name

        try:

            fdesc = open(name, 'rb+')

        except IOError, e:

            if self.create and e.errno == ENOENT:

                fdesc = open(name, 'w+')

            elif not self.create and e.errno == EROFS:

                fdesc = open(name, 'rb')

            else:

                raise

        self.fdesc = fdesc

        return self

    def __exit__(self, exc, arg, trace):

        fdesc = self.fdesc

        if fdesc is not None:

            #if self.dirty:

            #    fsync(fdesc.fileno())

            fdesc.close()

        return False  # propagate exceptions

    def seek(self, offset, whence=SEEK_SET):

        return self.fdesc.seek(offset, whence)

    def tell(self):

        return self.fdesc.tell()

    def truncate(self, size):

        self.fdesc.truncate(size)

    def sync_write(self, data):

        #self.dirty = 1

        self.fdesc.write(data)

    def sync_write_chunks(self, chunksize, offset, chunks, size=None):

        #self.dirty = 1

        return file_sync_write_chunks(self.fdesc, chunksize, offset, chunks,

                                      size)

    def sync_read(self, size):

        read = self.fdesc.read

        data = ''

        while 1:

            s = read(size)

            if not s:

                break

            data += s

        return data

    def sync_read_chunks(self, chunksize, nr, offset=0):

        return file_sync_read_chunks(self.fdesc, chunksize, nr, offset)

# Copyright 2011-2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from os import SEEK_CUR, SEEK_SET

from rados import ObjectNotFound

_zeros = ''

def zeros(nr):

    global _zeros

    size = len(_zeros)

    if nr == size:

        return _zeros

    if nr > size:

        _zeros += '\0' * (nr - size)

        return _zeros

    if nr < size:

        _zeros = _zeros[:nr]

        return _zeros

def file_sync_write_chunks(radosobject, chunksize, offset, chunks, size=None):

    """Write given chunks to the given buffered file object.

       Writes never span across chunk boundaries.

       If size is given stop after or pad until size bytes have been written.

    """

    padding = 0

    cursize = chunksize * offset

    radosobject.seek(cursize)

    for chunk in chunks:

        if padding:

            radosobject.sync_write(buffer(zeros(chunksize), 0, padding))

        if size is not None and cursize + chunksize >= size:

            chunk = chunk[:chunksize - (cursize - size)]

            radosobject.sync_write(chunk)

            cursize += len(chunk)

            break

        radosobject.sync_write(chunk)

        padding = chunksize - len(chunk)

    padding = size - cursize if size is not None else 0

    if padding <= 0:

        return

    q, r = divmod(padding, chunksize)

    for x in xrange(q):

        radosobject.sunc_write(zeros(chunksize))

    radosobject.sync_write(buffer(zeros(chunksize), 0, r))

def file_sync_read_chunks(radosobject, chunksize, nr, offset=0):

    """Read and yield groups of chunks from a buffered file object at offset.

       Reads never span accros chunksize boundaries.

    """

    radosobject.seek(offset * chunksize)

    while nr:

        remains = chunksize

        chunk = ''

        while 1:

            s = radosobject.sync_read(remains)

            if not s:

                if chunk:

                    yield chunk

                return

            chunk += s

            remains -= len(s)

            if remains <= 0:

                break

        yield chunk

        nr -= 1

class RadosObject(object):

    __slots__ = ("name", "ioctx", "create", "offset")

    def __init__(self, name, ioctx, create=0):

        self.name = name

        self.ioctx = ioctx

        self.create = create

        self.offset = 0

        #self.dirty = 0

    def __enter__(self):

        return self

    def __exit__(self, exc, arg, trace):

        return False

    def seek(self, offset, whence=SEEK_SET):

        if whence == SEEK_CUR:

            offset += self.offset

        self.offset = offset

        return offset

    def tell(self):

        return self.offset

    def truncate(self, size):

        self.ioctx.trunc(self.name, size)

    def sync_write(self, data):

        #self.dirty = 1

        self.ioctx.write(self.name, data, self.offset)

        self.offset += len(data)

    def sync_write_chunks(self, chunksize, offset, chunks, size=None):

        #self.dirty = 1

        return file_sync_write_chunks(self, chunksize, offset, chunks, size)

    def sync_read(self, size):

        read = self.ioctx.read

        data = ''

        datalen = 0

        while 1:

            try:

                s = read(self.name, size - datalen, self.offset)

            except ObjectNotFound:

                s = None

            if not s:

                break

            data += s

            datalen += len(s)

            self.offset += len(s)

            if datalen >= size:

                break

        return data

    def sync_read_chunks(self, chunksize, nr, offset=0):

        return file_sync_read_chunks(self, chunksize, nr, offset)

# Copyright 2011-2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from os import makedirs

from os.path import isdir, realpath, exists, join

from hashlib import new as newhasher

from binascii import hexlify

from context_file import ContextFile, file_sync_read_chunks

class FileBlocker(object):

    """Blocker.

       Required constructor parameters: blocksize, blockpath, hashtype.

    """

    blocksize = None

    blockpath = None

    hashtype = None

    def __init__(self, **params):

        blocksize = params['blocksize']

        blockpath = params['blockpath']

        blockpath = realpath(blockpath)

        if not isdir(blockpath):

            if not exists(blockpath):

                makedirs(blockpath)

            else:

                raise ValueError("Variable blockpath '%s' is not a directory" %

                                 (blockpath,))

        hashtype = params['hashtype']

        try:

            hasher = newhasher(hashtype)

        except ValueError:

            msg = "Variable hashtype '%s' is not available from hashlib"

            raise ValueError(msg % (hashtype,))

        hasher.update("")

        emptyhash = hasher.digest()

        self.blocksize = blocksize

        self.blockpath = blockpath

        self.hashtype = hashtype

        self.hashlen = len(emptyhash)

        self.emptyhash = emptyhash

    def _pad(self, block):

        return block + ('\x00' * (self.blocksize - len(block)))

    def _get_rear_block(self, blkhash, create=0):

        filename = hexlify(blkhash)

        dir = join(self.blockpath, filename[0:2], filename[2:4], filename[4:6])

        if not exists(dir):

            makedirs(dir)

        name = join(dir, filename)

        return ContextFile(name, create)

    def _check_rear_block(self, blkhash):

        filename = hexlify(blkhash)

        dir = join(self.blockpath, filename[0:2], filename[2:4], filename[4:6])

        name = join(dir, filename)

        return exists(name)

    def block_hash(self, data):

        """Hash a block of data"""

        hasher = newhasher(self.hashtype)

        hasher.update(data.rstrip('\x00'))

        return hasher.digest()

    def block_ping(self, hashes):

        """Check hashes for existence and

           return those missing from block storage.

        """

        notfound = []

        append = notfound.append

        for h in hashes:

            if h not in notfound and not self._check_rear_block(h):

                append(h)

        return notfound

    def block_retr(self, hashes):

        """Retrieve blocks from storage by their hashes."""

        blocksize = self.blocksize

        blocks = []

        append = blocks.append

        block = None

        for h in hashes:

            if h == self.emptyhash:

                append(self._pad(''))

                continue

            with self._get_rear_block(h, 0) as rbl:

                if not rbl:

                    break

                for block in rbl.sync_read_chunks(blocksize, 1, 0):

                    break  # there should be just one block there

            if not block:

                break

            append(self._pad(block))

        return blocks

    def block_stor(self, blocklist):

        """Store a bunch of blocks and return (hashes, missing).

           Hashes is a list of the hashes of the blocks,

           missing is a list of indices in that list indicating

           which blocks were missing from the store.

        """

        block_hash = self.block_hash

        hashlist = [block_hash(b) for b in blocklist]

        missing = [i for i, h in enumerate(hashlist) if not

                   self._check_rear_block(h)]

        for i in missing:

            with self._get_rear_block(hashlist[i], 1) as rbl:

                rbl.sync_write(blocklist[i])  # XXX: verify?

        return hashlist, missing

    def block_delta(self, blkhash, offset, data):

        """Construct and store a new block from a given block

           and a data 'patch' applied at offset. Return:

           (the hash of the new block, if the block already existed)

        """

        blocksize = self.blocksize

        if offset >= blocksize or not data:

            return None, None

        block = self.block_retr((blkhash,))

        if not block:

            return None, None

        block = block[0]

        newblock = block[:offset] + data

        if len(newblock) > blocksize:

            newblock = newblock[:blocksize]

        elif len(newblock) < blocksize:

            newblock += block[len(newblock):]

        h, a = self.block_stor((newblock,))

        return h[0], 1 if a else 0

    def block_hash_file(self, openfile):

        """Return the list of hashes (hashes map)

           for the blocks in a buffered file.

           Helper method, does not affect store.

        """

        hashes = []

        append = hashes.append

        block_hash = self.block_hash

        for block in file_sync_read_chunks(openfile, self.blocksize, 1, 0):

            append(block_hash(block))

        return hashes

    def block_stor_file(self, openfile):

        """Read blocks from buffered file object and store them. Return:

           (bytes read, list of hashes, list of hashes that were missing)

        """

        blocksize = self.blocksize

        block_stor = self.block_stor

        hashlist = []

        hextend = hashlist.extend

        storedlist = []

        sextend = storedlist.extend

        lastsize = 0

        for block in file_sync_read_chunks(openfile, blocksize, 1, 0):

            hl, sl = block_stor((block,))

            hextend(hl)

            sextend(sl)

            lastsize = len(block)

        size = (len(hashlist) - 1) * blocksize + lastsize if hashlist else 0

        return size, hashlist, storedlist

# Copyright 2011-2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from os import makedirs

from os.path import isdir, realpath, exists, join

from binascii import hexlify

from context_file import ContextFile

class FileMapper(object):

    """Mapper.

       Required constructor parameters: mappath, namelen.

    """

    mappath = None

    namelen = None

    def __init__(self, **params):

        self.params = params

        self.namelen = params['namelen']

        mappath = realpath(params['mappath'])

        if not isdir(mappath):

            if not exists(mappath):

                makedirs(mappath)

            else:

                raise ValueError("Variable mappath '%s' is not a directory" %

                                 (mappath,))

        self.mappath = mappath

    def _get_rear_map(self, maphash, create=0):

        filename = hexlify(maphash)

        dir = join(self.mappath, filename[0:2], filename[2:4], filename[4:6])

        if not exists(dir):

            makedirs(dir)

        name = join(dir, filename)

        return ContextFile(name, create)

    def _check_rear_map(self, maphash):

        filename = hexlify(maphash)

        dir = join(self.mappath, filename[0:2], filename[2:4], filename[4:6])

        name = join(dir, filename)

        return exists(name)

    def map_retr(self, maphash, blkoff=0, nr=100000000000000):

        """Return as a list, part of the hashes map of an object

           at the given block offset.

           By default, return the whole hashes map.

        """

        namelen = self.namelen

        hashes = ()

        with self._get_rear_map(maphash, 0) as rmap:

            if rmap:

                hashes = list(rmap.sync_read_chunks(namelen, nr, blkoff))

        return hashes

    def map_stor(self, maphash, hashes=(), blkoff=0, create=1):

        """Store hashes in the given hashes map."""

        namelen = self.namelen

        if self._check_rear_map(maphash):

            return

        with self._get_rear_map(maphash, 1) as rmap:

            rmap.sync_write_chunks(namelen, blkoff, hashes, None)

from filemapper import FileMapper

        self.rmap = None

        try:

            if params['mappool']:

                from radosmapper import RadosMapper

                self.rmap = RadosMapper(**params)

        except KeyError:

            pass

        self.fmap = FileMapper(**params)

        return self.fmap.map_retr(maphash, blkoff, nr)

        if self.rmap:

            self.rmap.map_stor(maphash, hashes, blkoff, create)

        self.fmap.map_stor(maphash, hashes, blkoff, create)

# Copyright 2011-2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from hashlib import new as newhasher

from binascii import hexlify

from rados import *

from context_object import RadosObject, file_sync_read_chunks

CEPH_CONF_FILE = "/etc/ceph/ceph.conf"

class RadosBlocker(object):

    """Blocker.

       Required constructor parameters: blocksize, blockpath, hashtype.

    """

    blocksize = None

    blockpool = None

    hashtype = None

    def __init__(self, **params):

        blocksize = params['blocksize']

        blockpool = params['blockpool']

        rados = Rados(conffile=CEPH_CONF_FILE)

        rados.connect()

        ioctx = rados.open_ioctx(blockpool)

        hashtype = params['hashtype']

        try:

            hasher = newhasher(hashtype)

        except ValueError:

            msg = "Variable hashtype '%s' is not available from hashlib"

            raise ValueError(msg % (hashtype,))

        hasher.update("")

        emptyhash = hasher.digest()

        self.blocksize = blocksize

        self.blockpool = blockpool

        self.rados = rados

        self.ioctx = ioctx

        self.hashtype = hashtype

        self.hashlen = len(emptyhash)

        self.emptyhash = emptyhash

    def _pad(self, block):

        return block + ('\x00' * (self.blocksize - len(block)))

    def _get_rear_block(self, blkhash, create=0):

        name = hexlify(blkhash)

        return RadosObject(name, self.ioctx, create)

    def _check_rear_block(self, blkhash):

        filename = hexlify(blkhash)

        try:

            self.ioctx.stat(filename)

            return True

        except ObjectNotFound:

            return False

    def block_hash(self, data):

        """Hash a block of data"""

        hasher = newhasher(self.hashtype)

        hasher.update(data.rstrip('\x00'))

        return hasher.digest()

    def block_ping(self, hashes):

        """Check hashes for existence and

           return those missing from block storage.

        """

        notfound = []

        append = notfound.append

        for h in hashes:

            if h not in notfound and not self._check_rear_block(h):

                append(h)

        return notfound

    def block_retr(self, hashes):

        """Retrieve blocks from storage by their hashes."""

        blocksize = self.blocksize

        blocks = []

        append = blocks.append

        block = None

        for h in hashes:

            if h == self.emptyhash:

                append(self._pad(''))

                continue

            with self._get_rear_block(h, 0) as rbl:

                if not rbl:

                    break

                for block in rbl.sync_read_chunks(blocksize, 1, 0):

                    break  # there should be just one block there

            if not block:

                break

            append(self._pad(block))

        return blocks

    def block_stor(self, blocklist):

        """Store a bunch of blocks and return (hashes, missing).

           Hashes is a list of the hashes of the blocks,

           missing is a list of indices in that list indicating

           which blocks were missing from the store.

        """

        block_hash = self.block_hash

        hashlist = [block_hash(b) for b in blocklist]

        missing = [i for i, h in enumerate(hashlist) if not

                   self._check_rear_block(h)]

        for i in missing:

            with self._get_rear_block(hashlist[i], 1) as rbl:

                rbl.sync_write(blocklist[i])  # XXX: verify?

        return hashlist, missing

    def block_delta(self, blkhash, offset, data):

        """Construct and store a new block from a given block

           and a data 'patch' applied at offset. Return:

           (the hash of the new block, if the block already existed)

        """

        blocksize = self.blocksize

        if offset >= blocksize or not data:

            return None, None

        block = self.block_retr((blkhash,))

        if not block:

            return None, None

        block = block[0]

        newblock = block[:offset] + data

        if len(newblock) > blocksize:

            newblock = newblock[:blocksize]

        elif len(newblock) < blocksize:

            newblock += block[len(newblock):]

        h, a = self.block_stor((newblock,))

        return h[0], 1 if a else 0

    def block_hash_file(self, radosobject):

        """Return the list of hashes (hashes map)

           for the blocks in a buffered file.

           Helper method, does not affect store.

        """

        hashes = []

        append = hashes.append

        block_hash = self.block_hash

        for block in file_sync_read_chunks(radosobject, self.blocksize, 1, 0):

            append(block_hash(block))

        return hashes

    def block_stor_file(self, radosobject):

        """Read blocks from buffered file object and store them. Return:

           (bytes read, list of hashes, list of hashes that were missing)

        """

        blocksize = self.blocksize

        block_stor = self.block_stor

        hashlist = []

        hextend = hashlist.extend

        storedlist = []

        sextend = storedlist.extend

        lastsize = 0

        for block in file_sync_read_chunks(radosobject, blocksize, 1, 0):

            hl, sl = block_stor((block,))

            hextend(hl)

            sextend(sl)

            lastsize = len(block)

        size = (len(hashlist) - 1) * blocksize + lastsize if hashlist else 0

        return size, hashlist, storedlist

# Copyright 2011-2012 GRNET S.A. All rights reserved.

#

# Redistribution and use in source and binary forms, with or

# without modification, are permitted provided that the following

# conditions are met:

#

#   1. Redistributions of source code must retain the above

#      copyright notice, this list of conditions and the following

#      disclaimer.

#

#   2. Redistributions in binary form must reproduce the above

#      copyright notice, this list of conditions and the following

#      disclaimer in the documentation and/or other materials

#      provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS

# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR

# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF

# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED

# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT

# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN

# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

# POSSIBILITY OF SUCH DAMAGE.

#

# The views and conclusions contained in the software and

# documentation are those of the authors and should not be

# interpreted as representing official policies, either expressed

# or implied, of GRNET S.A.

from binascii import hexlify

from context_object import RadosObject

from rados import *

CEPH_CONF_FILE = "/etc/ceph/ceph.conf"

class RadosMapper(object):

    """Mapper.

       Required constructor parameters: mappath, namelen.

    """

    mappool = None

    namelen = None

    def __init__(self, **params):

        self.params = params

        self.namelen = params['namelen']

        mappool = params['mappool']

        rados = Rados(conffile=CEPH_CONF_FILE)

        rados.connect()

        ioctx = rados.open_ioctx(mappool)

        self.mappool = mappool

        self.rados = rados

        self.ioctx = ioctx

    def _get_rear_map(self, maphash, create=0):

        name = hexlify(maphash)

        return RadosObject(name, self.ioctx, create)

    def _check_rear_map(self, maphash):

        name = hexlify(maphash)

        try:

            self.ioctx.stat(name)

            return True

        except ObjectNotFound:

            return False

    def map_retr(self, maphash, blkoff=0, nr=100000000000000):

        """Return as a list, part of the hashes map of an object

           at the given block offset.

           By default, return the whole hashes map.

        """

        namelen = self.namelen

        hashes = ()

        with self._get_rear_map(maphash, 0) as rmap:

            if rmap:

                hashes = list(rmap.sync_read_chunks(namelen, nr, blkoff))

        return hashes

    def map_stor(self, maphash, hashes=(), blkoff=0, create=1):

        """Store hashes in the given hashes map."""

        namelen = self.namelen

        if self._check_rear_map(maphash):

            return

        with self._get_rear_map(maphash, 1) as rmap:

            rmap.sync_write_chunks(namelen, blkoff, hashes, None)

        path = params['path']

        if path and not os.path.exists(path):

            os.makedirs(path)

        if not os.path.isdir(path):

            raise RuntimeError("Cannot open path '%s'" % (path,))

        p = {'blocksize': params['block_size'],

             'blockpath': os.path.join(path + '/blocks'),

             'blockpool': params['blockpool']}

        self.blocker = Blocker(**p)

        p = {'mappath': os.path.join(path + '/maps'),

             'namelen': self.blocker.hashlen,

             'mappool': params['mappool']}

        self.mapper = Mapper(**p)