path: root/src/pybind/cephfs/cephfs.pyx

"""
This module is a thin wrapper around libcephfs.
"""

from cpython cimport PyObject, ref, exc
from libc.stdint cimport *
from libc.stdlib cimport malloc, realloc, free

from types cimport *
IF BUILD_DOC:
    include "mock_cephfs.pxi"
    cdef class Rados:
        cdef:
            rados_t cluster
ELSE:
    from c_cephfs cimport *
    from rados cimport Rados

from collections import namedtuple
from datetime import datetime
import os
import time

AT_SYMLINK_NOFOLLOW = 0x0100
AT_STATX_SYNC_TYPE  = 0x6000
AT_STATX_SYNC_AS_STAT = 0x0000
AT_STATX_FORCE_SYNC = 0x2000
AT_STATX_DONT_SYNC = 0x4000
cdef int AT_SYMLINK_NOFOLLOW_CDEF = AT_SYMLINK_NOFOLLOW
CEPH_STATX_BASIC_STATS = 0x7ff
cdef int CEPH_STATX_BASIC_STATS_CDEF = CEPH_STATX_BASIC_STATS
CEPH_STATX_MODE = 0x1
CEPH_STATX_NLINK = 0x2
CEPH_STATX_UID = 0x4
CEPH_STATX_GID = 0x8
CEPH_STATX_RDEV = 0x10
CEPH_STATX_ATIME = 0x20
CEPH_STATX_MTIME = 0x40
CEPH_STATX_CTIME = 0x80
CEPH_STATX_INO = 0x100
CEPH_STATX_SIZE = 0x200
CEPH_STATX_BLOCKS = 0x400
CEPH_STATX_BTIME = 0x800
CEPH_STATX_VERSION = 0x1000

FALLOC_FL_KEEP_SIZE = 0x01
FALLOC_FL_PUNCH_HOLE = 0x02
FALLOC_FL_NO_HIDE_STALE = 0x04

CEPH_SETATTR_MODE = 0x1
CEPH_SETATTR_UID = 0x2
CEPH_SETATTR_GID = 0x4
CEPH_SETATTR_MTIME = 0x8
CEPH_SETATTR_ATIME = 0x10
CEPH_SETATTR_SIZE  = 0x20
CEPH_SETATTR_CTIME = 0x40
CEPH_SETATTR_BTIME = 0x200

# errno definitions
cdef enum:
    CEPHFS_EBLOCKLISTED = 108
    CEPHFS_EPERM = 1
    CEPHFS_ESTALE = 116
    CEPHFS_ENOSPC = 28
    CEPHFS_ETIMEDOUT = 110
    CEPHFS_EIO = 5
    CEPHFS_ENOTCONN = 107
    CEPHFS_EEXIST = 17
    CEPHFS_EINTR = 4
    CEPHFS_EINVAL = 22
    CEPHFS_EBADF = 9
    CEPHFS_EROFS = 30
    CEPHFS_EAGAIN = 11
    CEPHFS_EACCES = 13
    CEPHFS_ELOOP = 40
    CEPHFS_EISDIR = 21
    CEPHFS_ENOENT = 2
    CEPHFS_ENOTDIR = 20
    CEPHFS_ENAMETOOLONG = 36
    CEPHFS_EBUSY = 16
    CEPHFS_EDQUOT = 122
    CEPHFS_EFBIG = 27
    CEPHFS_ERANGE = 34
    CEPHFS_ENXIO = 6
    CEPHFS_ECANCELED = 125
    CEPHFS_ENODATA = 61
    CEPHFS_EOPNOTSUPP = 95
    CEPHFS_EXDEV = 18
    CEPHFS_ENOMEM = 12
    CEPHFS_ENOTRECOVERABLE = 131
    CEPHFS_ENOSYS = 38
    CEPHFS_EWOULDBLOCK = CEPHFS_EAGAIN
    CEPHFS_ENOTEMPTY = 39
    CEPHFS_EDEADLK = 35
    CEPHFS_EDEADLOCK = CEPHFS_EDEADLK
    CEPHFS_EDOM = 33
    CEPHFS_EMLINK = 31
    CEPHFS_ETIME = 62
    CEPHFS_EOLDSNAPC = 85

cdef extern from "Python.h":
    # These are in cpython/string.pxd, but use "object" types instead of
    # PyObject*, which invokes assumptions in cpython that we need to
    # legitimately break to implement zero-copy string buffers in Image.read().
    # This is valid use of the Python API and documented as a special case.
    PyObject *PyBytes_FromStringAndSize(char *v, Py_ssize_t len) except NULL
    char* PyBytes_AsString(PyObject *string) except NULL
    int _PyBytes_Resize(PyObject **string, Py_ssize_t newsize) except -1
    void PyEval_InitThreads()


class Error(Exception):
    def get_error_code(self):
        return 1


class LibCephFSStateError(Error):
    pass


class OSError(Error):
    def __init__(self, errno, strerror):
        super(OSError, self).__init__(errno, strerror)
        self.errno = errno
        self.strerror = "%s: %s" % (strerror, os.strerror(errno))

    def __str__(self):
        return '{} [Errno {}]'.format(self.strerror, self.errno)

    def get_error_code(self):
        return self.errno


class PermissionError(OSError):
    pass


class ObjectNotFound(OSError):
    pass


class NoData(OSError):
    pass


class ObjectExists(OSError):
    pass


class IOError(OSError):
    pass


class NoSpace(OSError):
    pass


class InvalidValue(OSError):
    pass


class OperationNotSupported(OSError):
    pass


class WouldBlock(OSError):
    pass


class OutOfRange(OSError):
    pass


class ObjectNotEmpty(OSError):
    pass

class NotDirectory(OSError):
    pass

class DiskQuotaExceeded(OSError):
    pass
class PermissionDenied(OSError):
    pass

cdef errno_to_exception =  {
    CEPHFS_EPERM      : PermissionError,
    CEPHFS_ENOENT     : ObjectNotFound,
    CEPHFS_EIO        : IOError,
    CEPHFS_ENOSPC     : NoSpace,
    CEPHFS_EEXIST     : ObjectExists,
    CEPHFS_ENODATA    : NoData,
    CEPHFS_EINVAL     : InvalidValue,
    CEPHFS_EOPNOTSUPP : OperationNotSupported,
    CEPHFS_ERANGE     : OutOfRange,
    CEPHFS_EWOULDBLOCK: WouldBlock,
    CEPHFS_ENOTEMPTY  : ObjectNotEmpty,
    CEPHFS_ENOTDIR    : NotDirectory,
    CEPHFS_EDQUOT     : DiskQuotaExceeded,
    CEPHFS_EACCES     : PermissionDenied,
}


cdef make_ex(ret, msg):
    """
    Translate a libcephfs return code into an exception.

    :param ret: the return code
    :type ret: int
    :param msg: the error message to use
    :type msg: str
    :returns: a subclass of :class:`Error`
    """
    ret = abs(ret)
    if ret in errno_to_exception:
        return errno_to_exception[ret](ret, msg)
    else:
        return OSError(ret, msg)


class DirEntry(namedtuple('DirEntry',
               ['d_ino', 'd_off', 'd_reclen', 'd_type', 'd_name'])):
    DT_DIR = 0x4
    DT_REG = 0x8
    DT_LNK = 0xA
    def is_dir(self):
        return self.d_type == self.DT_DIR

    def is_symbol_file(self):
        return self.d_type == self.DT_LNK

    def is_file(self):
        return self.d_type == self.DT_REG

StatResult = namedtuple('StatResult',
                        ["st_dev", "st_ino", "st_mode", "st_nlink", "st_uid",
                         "st_gid", "st_rdev", "st_size", "st_blksize",
                         "st_blocks", "st_atime", "st_mtime", "st_ctime"])

cdef class DirResult(object):
    cdef LibCephFS lib
    cdef ceph_dir_result* handle

# Bug in older Cython instances prevents this from being a static method.
#    @staticmethod
#    cdef create(LibCephFS lib, ceph_dir_result* handle):
#        d = DirResult()
#        d.lib = lib
#        d.handle = handle
#        return d

    def __dealloc__(self):
        self.close()

    def __enter__(self):
        if not self.handle:
            raise make_ex(CEPHFS_EBADF, "dir is not open")
        self.lib.require_state("mounted")
        with nogil:
            ceph_rewinddir(self.lib.cluster, self.handle)
        return self

    def __exit__(self, type_, value, traceback):
        self.close()
        return False

    def readdir(self):
        self.lib.require_state("mounted")

        with nogil:
            dirent = ceph_readdir(self.lib.cluster, self.handle)
        if not dirent:
            return None

        IF UNAME_SYSNAME == "FreeBSD" or UNAME_SYSNAME == "Darwin":
            return DirEntry(d_ino=dirent.d_ino,
                            d_off=0,
                            d_reclen=dirent.d_reclen,
                            d_type=dirent.d_type,
                            d_name=dirent.d_name)
        ELSE:
            return DirEntry(d_ino=dirent.d_ino,
                            d_off=dirent.d_off,
                            d_reclen=dirent.d_reclen,
                            d_type=dirent.d_type,
                            d_name=dirent.d_name)

    def close(self):
        if self.handle:
            self.lib.require_state("mounted")
            with nogil:
                ret = ceph_closedir(self.lib.cluster, self.handle)
            if ret < 0:
                raise make_ex(ret, "closedir failed")
            self.handle = NULL

    def rewinddir(self):
        if not self.handle:
            raise make_ex(CEPHFS_EBADF, "dir is not open")
        self.lib.require_state("mounted")
        with nogil:
            ceph_rewinddir(self.lib.cluster, self.handle)

    def telldir(self):
        if not self.handle:
            raise make_ex(CEPHFS_EBADF, "dir is not open")
        self.lib.require_state("mounted")
        with nogil:
            ret = ceph_telldir(self.lib.cluster, self.handle)
        if ret < 0:
            raise make_ex(ret, "telldir failed")
        return ret

    def seekdir(self, offset):
        if not self.handle:
            raise make_ex(CEPHFS_EBADF, "dir is not open")
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')
        self.lib.require_state("mounted")
        cdef int64_t _offset = offset
        with nogil:
            ceph_seekdir(self.lib.cluster, self.handle, _offset)


def cstr(val, name, encoding="utf-8", opt=False):
    """
    Create a byte string from a Python string

    :param basestring val: Python string
    :param str name: Name of the string parameter, for exceptions
    :param str encoding: Encoding to use
    :param bool opt: If True, None is allowed
    :rtype: bytes
    :raises: :class:`InvalidArgument`
    """
    if opt and val is None:
        return None
    if isinstance(val, bytes):
        return val
    else:
        try:
            v = val.encode(encoding)
        except:
            raise TypeError('%s must be encodeable as a bytearray' % name)
        assert isinstance(v, bytes)
        return v

def cstr_list(list_str, name, encoding="utf-8"):
    return [cstr(s, name) for s in list_str]


def decode_cstr(val, encoding="utf-8"):
    """
    Decode a byte string into a Python string.

    :param bytes val: byte string
    :rtype: str or None
    """
    if val is None:
        return None

    return val.decode(encoding)

cdef timeval to_timeval(t):
    """
    return timeval equivalent from time
    """
    tt = int(t)
    cdef timeval buf = timeval(tt, (t - tt) * 1000000)
    return buf

cdef timespec to_timespec(t):
    """
    return timespec equivalent from time
    """
    tt = int(t)
    cdef timespec buf = timespec(tt, (t - tt) * 1000000000)
    return buf

cdef char* opt_str(s) except? NULL:
    if s is None:
        return NULL
    return s


cdef char ** to_bytes_array(list_bytes):
    cdef char **ret = <char **>malloc(len(list_bytes) * sizeof(char *))
    if ret == NULL:
        raise MemoryError("malloc failed")
    for i in range(len(list_bytes)):
        ret[i] = <char *>list_bytes[i]
    return ret


cdef void* realloc_chk(void* ptr, size_t size) except NULL:
    cdef void *ret = realloc(ptr, size)
    if ret == NULL:
        raise MemoryError("realloc failed")
    return ret


cdef iovec * to_iovec(buffers) except NULL:
    cdef iovec *iov = <iovec *>malloc(len(buffers) * sizeof(iovec))
    cdef char *s = NULL
    if iov == NULL:
        raise MemoryError("malloc failed")
    for i in xrange(len(buffers)):
        s = <char*>buffers[i]
        iov[i] = [<void*>s, len(buffers[i])]
    return iov


cdef class LibCephFS(object):
    """libcephfs python wrapper"""

    cdef public object state
    cdef ceph_mount_info *cluster

    def require_state(self, *args):
        if self.state in args:
            return
        raise LibCephFSStateError("You cannot perform that operation on a "
                                  "CephFS object in state %s." % (self.state))

    def __cinit__(self, conf=None, conffile=None, auth_id=None, rados_inst=None):
        """Create a libcephfs wrapper

        :param conf dict opt: settings overriding the default ones and conffile
        :param conffile str opt: the path to ceph.conf to override the default settings
        :auth_id str opt: the id used to authenticate the client entity
        :rados_inst Rados opt: a rados.Rados instance
        """
        PyEval_InitThreads()
        self.state = "uninitialized"
        if rados_inst is not None:
            if auth_id is not None or conffile is not None or conf is not None:
                raise make_ex(CEPHFS_EINVAL,
                              "May not pass RADOS instance as well as other configuration")

            self.create_with_rados(rados_inst)
        else:
            self.create(conf, conffile, auth_id)

    def create_with_rados(self, Rados rados_inst):
        cdef int ret
        with nogil:
            ret = ceph_create_from_rados(&self.cluster, rados_inst.cluster)
        if ret != 0:
            raise Error("libcephfs_initialize failed with error code: %d" % ret)
        self.state = "configuring"

    NO_CONF_FILE = -1
    "special value that indicates no conffile should be read when creating a mount handle"
    DEFAULT_CONF_FILES = -2
    "special value that indicates the default conffiles should be read when creating a mount handle"

    def create(self, conf=None, conffile=NO_CONF_FILE, auth_id=None):
        """
        Create a mount handle for interacting with Ceph.  All libcephfs
        functions operate on a mount info handle.
        
        :param conf dict opt: settings overriding the default ones and conffile
        :param conffile Union[int,str], optional: the path to ceph.conf to override the default settings
        :auth_id str opt: the id used to authenticate the client entity
        """
        if conf is not None and not isinstance(conf, dict):
            raise TypeError("conf must be dict or None")
        cstr(conffile, 'configfile', opt=True)
        auth_id = cstr(auth_id, 'auth_id', opt=True)

        cdef:
            char* _auth_id = opt_str(auth_id)
            int ret

        with nogil:
            ret = ceph_create(&self.cluster, <const char*>_auth_id)
        if ret != 0:
            raise Error("libcephfs_initialize failed with error code: %d" % ret)

        self.state = "configuring"
        if conffile in (self.NO_CONF_FILE, None):
            pass
        elif conffile in (self.DEFAULT_CONF_FILES, ''):
            self.conf_read_file(None)
        else:
            self.conf_read_file(conffile)
        if conf is not None:
            for key, value in conf.items():
                self.conf_set(key, value)

    def get_fscid(self):
        """
        Return the file system id for this fs client.
        """
        self.require_state("mounted")
        with nogil:
            ret = ceph_get_fs_cid(self.cluster)
        if ret < 0:
            raise make_ex(ret, "error fetching fscid")
        return ret

    def get_addrs(self):
        """
        Get associated client addresses with this RADOS session.
        """
        self.require_state("mounted")

        cdef:
            char* addrs = NULL

        try:

            with nogil:
                ret = ceph_getaddrs(self.cluster, &addrs)
            if ret:
                raise make_ex(ret, "error calling getaddrs")

            return decode_cstr(addrs)
        finally:
            ceph_buffer_free(addrs)


    def conf_read_file(self, conffile=None):
        """
        Load the ceph configuration from the specified config file.
         
        :param conffile str opt: the path to ceph.conf to override the default settings
        """
        conffile = cstr(conffile, 'conffile', opt=True)
        cdef:
            char *_conffile = opt_str(conffile)
        with nogil:
            ret = ceph_conf_read_file(self.cluster, <const char*>_conffile)
        if ret != 0:
            raise make_ex(ret, "error calling conf_read_file")

    def conf_parse_argv(self, argv):
        """
        Parse the command line arguments and load the configuration parameters.
        
        :param argv: the argument list
        """
        self.require_state("configuring")
        cargv = cstr_list(argv, 'argv')
        cdef:
            int _argc = len(argv)
            char **_argv = to_bytes_array(cargv)

        try:
            with nogil:
                ret = ceph_conf_parse_argv(self.cluster, _argc,
                                           <const char **>_argv)
            if ret != 0:
                raise make_ex(ret, "error calling conf_parse_argv")
        finally:
            free(_argv)

    def shutdown(self):
        """
        Unmount and destroy the ceph mount handle.
        """
        if self.state in ["initialized", "mounted"]:
            with nogil:
                ceph_shutdown(self.cluster)
            self.state = "shutdown"

    def __enter__(self):
        self.mount()
        return self

    def __exit__(self, type_, value, traceback):
        self.shutdown()
        return False

    def __dealloc__(self):
        self.shutdown()

    def version(self):
        """
        Get the version number of the ``libcephfs`` C library.

        :returns: a tuple of ``(major, minor, extra)`` components of the
                  libcephfs version
        """
        cdef:
            int major = 0
            int minor = 0
            int extra = 0
        with nogil:
            ceph_version(&major, &minor, &extra)
        return (major, minor, extra)

    def conf_get(self, option):
        """
        Gets the configuration value as a string.
        
        :param option: the config option to get
        """
        self.require_state("configuring", "initialized", "mounted")

        option = cstr(option, 'option')
        cdef:
            char *_option = option
            size_t length = 20
            char *ret_buf = NULL

        try:
            while True:
                ret_buf = <char *>realloc_chk(ret_buf, length)
                with nogil:
                    ret = ceph_conf_get(self.cluster, _option, ret_buf, length)
                if ret == 0:
                    return decode_cstr(ret_buf)
                elif ret == -CEPHFS_ENAMETOOLONG:
                    length = length * 2
                elif ret == -CEPHFS_ENOENT:
                    return None
                else:
                    raise make_ex(ret, "error calling conf_get")
        finally:
            free(ret_buf)

    def conf_set(self, option, val):
        """
        Sets a configuration value from a string.
        
        :param option: the configuration option to set
        :param value: the value of the configuration option to set
        """
        self.require_state("configuring", "initialized", "mounted")

        option = cstr(option, 'option')
        val = cstr(val, 'val')
        cdef:
            char *_option = option
            char *_val = val

        with nogil:
            ret = ceph_conf_set(self.cluster, _option, _val)
        if ret != 0:
            raise make_ex(ret, "error calling conf_set")

    def set_mount_timeout(self, timeout):
        """
        Set mount timeout

        :param timeout: mount timeout
        """
        self.require_state("configuring", "initialized")
        if not isinstance(timeout, int):
            raise TypeError('timeout must be an integer')
        if timeout < 0:
            raise make_ex(CEPHFS_EINVAL, 'timeout must be greater than or equal to 0')
        cdef:
            uint32_t _timeout = timeout
        with nogil:
            ret = ceph_set_mount_timeout(self.cluster, _timeout)
        if ret != 0:
            raise make_ex(ret, "error setting mount timeout")

    def init(self):
        """
        Initialize the filesystem client (but do not mount the filesystem yet)
        """
        self.require_state("configuring")
        with nogil:
            ret = ceph_init(self.cluster)
        if ret != 0:
            raise make_ex(ret, "error calling ceph_init")
        self.state = "initialized"

    def mount(self, mount_root=None, filesystem_name=None):
        """
        Perform a mount using the path for the root of the mount.
        """
        if self.state == "configuring":
            self.init()
        self.require_state("initialized")

        # Configure which filesystem to mount if one was specified
        if filesystem_name is None:
            filesystem_name = b""
        else:
            filesystem_name = cstr(filesystem_name, 'filesystem_name')
        cdef:
            char *_filesystem_name = filesystem_name
        if filesystem_name:
            with nogil:
                ret = ceph_select_filesystem(self.cluster,
                        _filesystem_name)
            if ret != 0:
                raise make_ex(ret, "error calling ceph_select_filesystem")

        # Prepare mount_root argument, default to "/"
        root = b"/" if mount_root is None else mount_root
        cdef:
            char *_mount_root = root

        with nogil:
            ret = ceph_mount(self.cluster, _mount_root)
        if ret != 0:
            raise make_ex(ret, "error calling ceph_mount")
        self.state = "mounted"

    def unmount(self):
        """
        Unmount a mount handle.
        """
        self.require_state("mounted")
        with nogil:
            ret = ceph_unmount(self.cluster)
        if ret != 0:
            raise make_ex(ret, "error calling ceph_unmount")
        self.state = "initialized"

    def abort_conn(self):
        """
        Abort mds connections.
        """
        self.require_state("mounted")
        with nogil:
            ret = ceph_abort_conn(self.cluster)
        if ret != 0:
            raise make_ex(ret, "error calling ceph_abort_conn")
        self.state = "initialized"

    def get_instance_id(self):
        """
        Get a global id for current instance
        """
        self.require_state("initialized", "mounted")
        with nogil:
            ret = ceph_get_instance_id(self.cluster)
        return ret;

    def statfs(self, path):
        """
        Perform a statfs on the ceph file system.  This call fills in file system wide statistics
        into the passed in buffer.
        
        :param path: any path within the mounted filesystem
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        cdef:
            char* _path = path
            statvfs statbuf

        with nogil:
            ret = ceph_statfs(self.cluster, _path, &statbuf)
        if ret < 0:
            raise make_ex(ret, "statfs failed: %s" % path)
        return {'f_bsize': statbuf.f_bsize,
                'f_frsize': statbuf.f_frsize,
                'f_blocks': statbuf.f_blocks,
                'f_bfree': statbuf.f_bfree,
                'f_bavail': statbuf.f_bavail,
                'f_files': statbuf.f_files,
                'f_ffree': statbuf.f_ffree,
                'f_favail': statbuf.f_favail,
                'f_fsid': statbuf.f_fsid,
                'f_flag': statbuf.f_flag,
                'f_namemax': statbuf.f_namemax}

    def sync_fs(self):
        """
        Synchronize all filesystem data to persistent media
        """
        self.require_state("mounted")
        with nogil:
            ret = ceph_sync_fs(self.cluster)
        if ret < 0:
            raise make_ex(ret, "sync_fs failed")

    def fsync(self, int fd, int syncdataonly):
        """
        Synchronize an open file to persistent media.
        
        :param fd: the file descriptor of the file to sync.
        :param syncdataonly: a boolean whether to synchronize metadata and data (0)
                             or just data (1).
        """
        self.require_state("mounted")
        with nogil:
            ret = ceph_fsync(self.cluster, fd, syncdataonly)
        if ret < 0:
            raise make_ex(ret, "fsync failed")

    def lazyio(self, fd, enable):
        """
        Enable/disable lazyio for the file.

        :param fd: the file descriptor of the file for which to enable lazio.
        :param enable: a boolean to enable lazyio or disable lazyio.
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(enable, int):
            raise TypeError('enable must be an int')

        cdef:
            int _fd = fd
            int _enable = enable

        with nogil:
            ret = ceph_lazyio(self.cluster, _fd, _enable)
        if ret < 0:
            raise make_ex(ret, "lazyio failed")

    def lazyio_propagate(self, fd, offset, count):
        """
        Flushes the write buffer for the file thereby propogating the buffered write to the file.

        :param fd: the file descriptor of the file to sync.
        :param offset: the byte range starting.
        :param count: the number of bytes starting from offset.
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')
        if not isinstance(count, int):
            raise TypeError('count must be an int')

        cdef:
            int _fd = fd
            int64_t _offset = offset
            size_t _count = count

        with nogil:
            ret = ceph_lazyio_propagate(self.cluster, _fd, _offset, _count)
        if ret < 0:
            raise make_ex(ret, "lazyio_propagate failed")

    def lazyio_synchronize(self, fd, offset, count):
        """
        Flushes the write buffer for the file and invalidate the read cache. This allows a
        subsequent read operation to read and cache data directly from the file and hence
        everyone's propagated writes would be visible.

        :param fd: the file descriptor of the file to sync.
        :param offset: the byte range starting.
        :param count: the number of bytes starting from offset.
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')
        if not isinstance(count, int):
            raise TypeError('count must be an int')

        cdef:
            int _fd = fd
            int64_t _offset = offset
            size_t _count = count

        with nogil:
            ret = ceph_lazyio_synchronize(self.cluster, _fd, _offset, _count)
        if ret < 0:
            raise make_ex(ret, "lazyio_synchronize failed")

    def fallocate(self, fd, offset, length, mode=0):
        """
        Preallocate or release disk space for the file for the byte range.

        :param fd: the file descriptor of the file to fallocate.
        :param mode: the flags determines the operation to be performed on the given
                     range. default operation (0) allocate and initialize to zero
                     the file in the byte range, and the file size will be changed
                     if offset + length is greater than the file size. if the
                     FALLOC_FL_KEEP_SIZE flag is specified in the mode, the file size
                     will not be changed. if the FALLOC_FL_PUNCH_HOLE flag is specified
                     in the mode, the operation is deallocate space and zero the byte range.
        :param offset: the byte range starting.
        :param length: the length of the range.
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')
        if not isinstance(length, int):
            raise TypeError('length must be an int')

        cdef:
            int _fd = fd
            int _mode = mode
            int64_t _offset = offset
            int64_t _length = length

        with nogil:
            ret = ceph_fallocate(self.cluster, _fd, _mode, _offset, _length)
        if ret < 0:
            raise make_ex(ret, "fallocate failed")

    def getcwd(self):
        """
        Get the current working directory.
        
        :rtype the path to the current working directory
        """
        self.require_state("mounted")
        with nogil:
            ret = ceph_getcwd(self.cluster)
        return ret

    def chdir(self, path):
        """
        Change the current working directory.
        
        :param path the path to the working directory to change into.
        """
        self.require_state("mounted")

        path = cstr(path, 'path')
        cdef char* _path = path
        with nogil:
            ret = ceph_chdir(self.cluster, _path)
        if ret < 0:
            raise make_ex(ret, "chdir failed")

    def opendir(self, path):
        """
        Open the given directory.
        
        :param path: the path name of the directory to open.  Must be either an absolute path
                     or a path relative to the current working directory.
        :rtype handle: the open directory stream handle
        """
        self.require_state("mounted")

        path = cstr(path, 'path')
        cdef:
            char* _path = path
            ceph_dir_result* handle
        with nogil:
            ret = ceph_opendir(self.cluster, _path, &handle);
        if ret < 0:
            raise make_ex(ret, "opendir failed at {}".format(path.decode('utf-8')))
        d = DirResult()
        d.lib = self
        d.handle = handle
        return d

    def readdir(self, DirResult handle):
        """
        Get the next entry in an open directory.
        
        :param handle: the open directory stream handle
        :rtype dentry: the next directory entry or None if at the end of the
                       directory (or the directory is empty.  This pointer
                       should not be freed by the caller, and is only safe to
                       access between return and the next call to readdir or
                       closedir.
        """
        self.require_state("mounted")

        return handle.readdir()

    def closedir(self, DirResult handle):
        """
        Close the open directory.
        
        :param handle: the open directory stream handle
        """
        self.require_state("mounted")

        return handle.close()

    def rewinddir(self, DirResult handle):
        """
        Rewind the directory stream to the beginning of the directory.

        :param handle: the open directory stream handle
        """
        return handle.rewinddir()

    def telldir(self, DirResult handle):
        """
        Get the current position of a directory stream.

        :param handle: the open directory stream handle
        :return value: The position of the directory stream. Note that the offsets
                       returned by ceph_telldir do not have a particular order (cannot
                       be compared with inequality).
        """
        return handle.telldir()

    def seekdir(self, DirResult handle, offset):
        """
        Move the directory stream to a position specified by the given offset.

        :param handle: the open directory stream handle
        :param offset: the position to move the directory stream to. This offset should be
                       a value returned by telldir. Note that this value does not refer to
                       the nth entry in a directory, and can not be manipulated with plus
                       or minus.
        """
        return handle.seekdir(offset)

    def mkdir(self, path, mode):
        """
        Create a directory.
 
        :param path: the path of the directory to create.  This must be either an
                     absolute path or a relative path off of the current working directory.
        :param mode: the permissions the directory should have once created.
        """

        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        cdef:
            char* _path = path
            int _mode = mode
        with nogil:
            ret = ceph_mkdir(self.cluster, _path, _mode)
        if ret < 0:
            raise make_ex(ret, "error in mkdir {}".format(path.decode('utf-8')))

    def mksnap(self, path, name, mode, metadata={}):
        """
        Create a snapshot.

        :param path: path of the directory to snapshot.
        :param name: snapshot name
        :param mode: permission of the snapshot
        :param metadata: metadata key/value to store with the snapshot

        :raises: :class: `TypeError`
        :raises: :class: `Error`
        :returns: int: 0 on success
        """

        self.require_state("mounted")
        path = cstr(path, 'path')
        name = cstr(name, 'name')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        if not isinstance(metadata, dict):
            raise TypeError('metadata must be an dictionary')
        md = {}
        for key, value in metadata.items():
            if not isinstance(key, str) or not isinstance(value, str):
                raise TypeError('metadata key and values should be strings')
            md[key.encode('utf-8')] = value.encode('utf-8')
        cdef:
            char* _path = path
            char* _name = name
            int _mode = mode
            size_t nr = len(md)
            snap_metadata *_snap_meta = <snap_metadata *>malloc(nr * sizeof(snap_metadata))
        if nr and _snap_meta == NULL:
            raise MemoryError("malloc failed")
        i = 0
        for key, value in md.items():
            _snap_meta[i] = snap_metadata(<char*>key, <char*>value)
            i += 1
        with nogil:
            ret = ceph_mksnap(self.cluster, _path, _name, _mode, _snap_meta, nr)
        free(_snap_meta)
        if ret < 0:
            raise make_ex(ret, "mksnap error")
        return 0

    def rmsnap(self, path, name):
        """
        Remove a snapshot.

        :param path: path of the directory for removing snapshot
        :param name: snapshot name

        :raises: :class: `Error`
        :returns: int: 0 on success
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        name = cstr(name, 'name')
        cdef:
            char* _path = path
            char* _name = name
        ret = ceph_rmsnap(self.cluster, _path, _name)
        if ret < 0:
            raise make_ex(ret, "rmsnap error")
        return 0

    def snap_info(self, path):
        """
        Fetch sapshot info

        :param path: snapshot path

        :raises: :class: `Error`
        :returns: dict: snapshot metadata
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        cdef:
            char* _path = path
            snap_info info
        ret = ceph_get_snap_info(self.cluster, _path, &info)
        if ret < 0:
            raise make_ex(ret, "snap_info error")
        md = {}
        if info.nr_snap_metadata:
            md = {snap_meta.key.decode('utf-8'): snap_meta.value.decode('utf-8') for snap_meta in
                  info.snap_metadata[:info.nr_snap_metadata]}
            ceph_free_snap_info_buffer(&info)
        return {'id': info.id, 'metadata': md}

    def chmod(self, path, mode) :
        """
        Change directory mode.

        :param path: the path of the directory to create.  This must be either an
                     absolute path or a relative path off of the current working directory.
        :param mode: the permissions the directory should have once created.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        cdef:
            char* _path = path
            int _mode = mode
        with nogil:
            ret = ceph_chmod(self.cluster, _path, _mode)
        if ret < 0:
            raise make_ex(ret, "error in chmod {}".format(path.decode('utf-8')))

    def lchmod(self, path, mode) -> None:
        """
        Change file mode. If the path is a symbolic link, it won't be dereferenced.

        :param path: the path of the file. This must be either an absolute path or
                     a relative path off of the current working directory.
        :param mode: the permissions to be set .
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        cdef:
            char* _path = path
            int _mode = mode
        with nogil:
            ret = ceph_lchmod(self.cluster, _path, _mode)
        if ret < 0:
            raise make_ex(ret, "error in chmod {}".format(path.decode('utf-8')))

    def fchmod(self, fd, mode) :
        """
        Change file mode based on fd.
        :param fd: the file descriptor of the file to change file mode
        :param mode: the permissions to be set.
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        cdef:
            int _fd = fd
            int _mode = mode
        with nogil:
            ret = ceph_fchmod(self.cluster, _fd, _mode)
        if ret < 0:
            raise make_ex(ret, "error in fchmod")

    def chown(self, path, uid, gid, follow_symlink=True):
        """
        Change directory ownership

        :param path: the path of the directory to change.
        :param uid: the uid to set
        :param gid: the gid to set
        :param follow_symlink: perform the operation on the target file if @path
                               is a symbolic link (default)
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(uid, int):
            raise TypeError('uid must be an int')
        elif not isinstance(gid, int):
            raise TypeError('gid must be an int')

        cdef:
            char* _path = path
            int _uid = uid
            int _gid = gid
        if follow_symlink:
            with nogil:
                ret = ceph_chown(self.cluster, _path, _uid, _gid)
        else:
            with nogil:
                ret = ceph_lchown(self.cluster, _path, _uid, _gid)
        if ret < 0:
            raise make_ex(ret, "error in chown {}".format(path.decode('utf-8')))

    def lchown(self, path, uid, gid):
        """
        Change ownership of a symbolic link
        :param path: the path of the symbolic link to change.
        :param uid: the uid to set
        :param gid: the gid to set
        """
        self.chown(path, uid, gid, follow_symlink=False)

    def fchown(self, fd, uid, gid):
        """
        Change file ownership
        :param fd: the file descriptor of the file to change ownership
        :param uid: the uid to set
        :param gid: the gid to set
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(uid, int):
            raise TypeError('uid must be an int')
        elif not isinstance(gid, int):
            raise TypeError('gid must be an int')

        cdef:
            int _fd = fd
            int _uid = uid
            int _gid = gid
        with nogil:
            ret = ceph_fchown(self.cluster, _fd, _uid, _gid)
        if ret < 0:
            raise make_ex(ret, "error in fchown")

    def mkdirs(self, path, mode):
        """
        Create multiple directories at once.

        :param path: the full path of directories and sub-directories that should
                     be created.
        :param mode: the permissions the directory should have once created
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        cdef:
            char* _path = path
            int _mode = mode

        with nogil:
            ret = ceph_mkdirs(self.cluster, _path, _mode)
        if ret < 0:
            raise make_ex(ret, "error in mkdirs {}".format(path.decode('utf-8')))

    def rmdir(self, path):
        """
        Remove a directory.
         
        :param path: the path of the directory to remove.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        cdef char* _path = path
        ret = ceph_rmdir(self.cluster, _path)
        if ret < 0:
            raise make_ex(ret, "error in rmdir {}".format(path.decode('utf-8')))

    def open(self, path, flags, mode=0):
        """
        Create and/or open a file.
         
        :param path: the path of the file to open.  If the flags parameter includes O_CREAT,
                     the file will first be created before opening.
        :param flags: set of option masks that control how the file is created/opened.
        :param mode: the permissions to place on the file if the file does not exist and O_CREAT
                     is specified in the flags.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')

        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        if isinstance(flags, str):
            cephfs_flags = 0
            if flags == '':
                cephfs_flags = os.O_RDONLY
            else:
                access_flags = 0;
                for c in flags:
                    if c == 'r':
                        access_flags = 1;
                    elif c == 'w':
                        access_flags = 2;
                        cephfs_flags |= os.O_TRUNC | os.O_CREAT
                    elif access_flags > 0 and c == '+':
                        access_flags = 3;
                    else:
                        raise make_ex(CEPHFS_EOPNOTSUPP,
                                      "open flags doesn't support %s" % c)

                if access_flags == 1:
                    cephfs_flags |= os.O_RDONLY;
                elif access_flags == 2:
                    cephfs_flags |= os.O_WRONLY;
                else:
                    cephfs_flags |= os.O_RDWR;

        elif isinstance(flags, int):
            cephfs_flags = flags
        else:
            raise TypeError("flags must be a string or an integer")

        cdef:
            char* _path = path
            int _flags = cephfs_flags
            int _mode = mode

        with nogil:
            ret = ceph_open(self.cluster, _path, _flags, _mode)
        if ret < 0:
            raise make_ex(ret, "error in open {}".format(path.decode('utf-8')))
        return ret

    def close(self, fd):
        """
        Close the open file.
        
        :param fd: the file descriptor referring to the open file.
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        cdef int _fd = fd
        with nogil:
            ret = ceph_close(self.cluster, _fd)
        if ret < 0:
            raise make_ex(ret, "error in close")

    def read(self, fd, offset, l):
        """
        Read data from the file.
 
        :param fd: the file descriptor of the open file to read from.
        :param offset: the offset in the file to read from.  If this value is negative, the
                       function reads from the current offset of the file descriptor.
        :param l: the flag to indicate what type of seeking to perform
        """     
        self.require_state("mounted")
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')
        if not isinstance(l, int):
            raise TypeError('l must be an int')
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        cdef:
            int _fd = fd
            int64_t _offset = offset
            int64_t _length = l

            char *ret_buf
            PyObject* ret_s = NULL

        ret_s = PyBytes_FromStringAndSize(NULL, _length)
        try:
            ret_buf = PyBytes_AsString(ret_s)
            with nogil:
                ret = ceph_read(self.cluster, _fd, ret_buf, _length, _offset)
            if ret < 0:
                raise make_ex(ret, "error in read")

            if ret != _length:
                _PyBytes_Resize(&ret_s, ret)

            return <object>ret_s
        finally:
            # We DECREF unconditionally: the cast to object above will have
            # INCREFed if necessary. This also takes care of exceptions,
            # including if _PyString_Resize fails (that will free the string
            # itself and set ret_s to NULL, hence XDECREF).
            ref.Py_XDECREF(ret_s)

    def preadv(self, fd, buffers, offset):
        """
        Write data to a file.

        :param fd: the file descriptor of the open file to read from
        :param buffers: the list of byte object to read from the file
        :param offset: the offset of the file read from.  If this value is negative, the
                       function reads from the current offset of the file descriptor.
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(buffers, list):
            raise TypeError('buffers must be a list')
        for buf in buffers:
            if not isinstance(buf, bytearray):
                raise TypeError('buffers must be a list of bytes')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')

        cdef:
            int _fd = fd
            int _iovcnt = len(buffers)
            int64_t _offset = offset
            iovec *_iov = to_iovec(buffers)
        try:
            with nogil:
                ret = ceph_preadv(self.cluster, _fd, _iov, _iovcnt, _offset)
            if ret < 0:
                raise make_ex(ret, "error in preadv")
            return ret
        finally:
            free(_iov)

    def write(self, fd, buf, offset):
        """
        Write data to a file.
       
        :param fd: the file descriptor of the open file to write to
        :param buf: the bytes to write to the file
        :param offset: the offset of the file write into.  If this value is negative, the
                       function writes to the current offset of the file descriptor.
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(buf, bytes):
            raise TypeError('buf must be a bytes')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')

        cdef:
            int _fd = fd
            char *_data = buf
            int64_t _offset = offset

            size_t length = len(buf)

        with nogil:
            ret = ceph_write(self.cluster, _fd, _data, length, _offset)
        if ret < 0:
            raise make_ex(ret, "error in write")
        return ret

    def pwritev(self, fd, buffers, offset):
        """
        Write data to a file.

        :param fd: the file descriptor of the open file to write to
        :param buffers: the list of byte object to write to the file
        :param offset: the offset of the file write into.  If this value is negative, the
                       function writes to the current offset of the file descriptor.
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(buffers, list):
            raise TypeError('buffers must be a list')
        for buf in buffers:
            if not isinstance(buf, bytes):
                raise TypeError('buffers must be a list of bytes')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')

        cdef:
            int _fd = fd
            int _iovcnt = len(buffers)
            int64_t _offset = offset
            iovec *_iov = to_iovec(buffers)
        try:
            with nogil:
                ret = ceph_pwritev(self.cluster, _fd, _iov, _iovcnt, _offset)
            if ret < 0:
                raise make_ex(ret, "error in pwritev")
            return ret
        finally:
            free(_iov)

    def flock(self, fd, operation, owner):
        """
        Apply or remove an advisory lock.
        
        :param fd: the open file descriptor to change advisory lock.
        :param operation: the advisory lock operation to be performed on the file
        :param owner: the user-supplied owner identifier (an arbitrary integer)
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(operation, int):
            raise TypeError('operation must be an int')

        cdef:
            int _fd = fd
            int _op = operation
            uint64_t _owner = owner

        with nogil:
            ret = ceph_flock(self.cluster, _fd, _op, _owner)
        if ret < 0:
            raise make_ex(ret, "error in write")
        return ret

    def truncate(self, path, size):
        """
        Truncate the file to the given size.  If this operation causes the
        file to expand, the empty bytes will be filled in with zeros.

        :param path: the path to the file to truncate.
        :param size: the new size of the file.
        """

        if not isinstance(size, int):
            raise TypeError('size must be a int')

        statx_dict = dict()
        statx_dict["size"] = size
        self.setattrx(path, statx_dict, CEPH_SETATTR_SIZE, AT_SYMLINK_NOFOLLOW)

    def ftruncate(self, fd, size):
        """
        Truncate the file to the given size.  If this operation causes the
        file to expand, the empty bytes will be filled in with zeros.

        :param path: the path to the file to truncate.
        :param size: the new size of the file.
        """

        if not isinstance(size, int):
            raise TypeError('size must be a int')

        statx_dict = dict()
        statx_dict["size"] = size
        self.fsetattrx(fd, statx_dict, CEPH_SETATTR_SIZE)

    def mknod(self, path, mode, rdev=0):
        """
        Make a block or character special file.

        :param path: the path to the special file.
        :param mode: the permissions to use and the type of special file. The type can be
                     one of stat.S_IFREG, stat.S_IFCHR, stat.S_IFBLK, stat.S_IFIFO. Both
                     should be combined using bitwise OR.
        :param rdev: If the file type is stat.S_IFCHR or stat.S_IFBLK then this parameter
                     specifies the major and minor numbers of the newly created device
                     special file. Otherwise, it is ignored.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')

        if not isinstance(mode, int):
            raise TypeError('mode must be an int')
        if not isinstance(rdev, int):
            raise TypeError('rdev must be an int')

        cdef:
            char* _path = path
            mode_t _mode = mode
            dev_t _rdev = rdev

        with nogil:
            ret = ceph_mknod(self.cluster, _path, _mode, _rdev)
        if ret < 0:
            raise make_ex(ret, "error in mknod {}".format(path.decode('utf-8')))

    def getxattr(self, path, name, size=255, follow_symlink=True):
        """
         Get an extended attribute.
         
         :param path: the path to the file
         :param name: the name of the extended attribute to get
         :param size: the size of the pre-allocated buffer
        """ 
        self.require_state("mounted")

        path = cstr(path, 'path')
        name = cstr(name, 'name')

        cdef:
            char* _path = path
            char* _name = name

            size_t ret_length = size
            char *ret_buf = NULL

        try:
            ret_buf = <char *>realloc_chk(ret_buf, ret_length)
            if follow_symlink:
                with nogil:
                    ret = ceph_getxattr(self.cluster, _path, _name, ret_buf,
                                        ret_length)
            else:
                with nogil:
                    ret = ceph_lgetxattr(self.cluster, _path, _name, ret_buf,
                                         ret_length)

            if ret < 0:
                raise make_ex(ret, "error in getxattr")

            return ret_buf[:ret]
        finally:
            free(ret_buf)

    def fgetxattr(self, fd, name, size=255):
        """
         Get an extended attribute given the fd of a file.

         :param fd: the open file descriptor referring to the file
         :param name: the name of the extended attribute to get
         :param size: the size of the pre-allocated buffer
        """
        self.require_state("mounted")

        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        name = cstr(name, 'name')

        cdef:
            int _fd = fd
            char* _name = name

            size_t ret_length = size
            char *ret_buf = NULL

        try:
            ret_buf = <char *>realloc_chk(ret_buf, ret_length)
            with nogil:
                ret = ceph_fgetxattr(self.cluster, _fd, _name, ret_buf,
                                    ret_length)

            if ret < 0:
                raise make_ex(ret, "error in fgetxattr")

            return ret_buf[:ret]
        finally:
            free(ret_buf)

    def lgetxattr(self, path, name, size=255):
        """
         Get an extended attribute without following symbolic links. This
         function is identical to ceph_getxattr, but if the path refers to
         a symbolic link, we get the extended attributes of the symlink
         rather than the attributes of the file it points to.

         :param path: the path to the file
         :param name: the name of the extended attribute to get
         :param size: the size of the pre-allocated buffer
        """

        return self.getxattr(path, name, size=size, follow_symlink=False)

    def setxattr(self, path, name, value, flags, follow_symlink=True):
        """
        Set an extended attribute on a file.

       :param path: the path to the file.
       :param name: the name of the extended attribute to set.
       :param value: the bytes of the extended attribute value
       """
        self.require_state("mounted")

        name = cstr(name, 'name')
        path = cstr(path, 'path')
        if not isinstance(flags, int):
            raise TypeError('flags must be a int')
        if not isinstance(value, bytes):
            raise TypeError('value must be a bytes')

        cdef:
            char *_path = path
            char *_name = name
            char *_value = value
            size_t _value_len = len(value)
            int _flags = flags

        if follow_symlink:
            with nogil:
                ret = ceph_setxattr(self.cluster, _path, _name,
                                    _value, _value_len, _flags)
        else:
            with nogil:
                ret = ceph_lsetxattr(self.cluster, _path, _name,
                                    _value, _value_len, _flags)

        if ret < 0:
            raise make_ex(ret, "error in setxattr")

    def fsetxattr(self, fd, name, value, flags):
        """
        Set an extended attribute on a file.

        :param fd: the open file descriptor referring to the file.
        :param name: the name of the extended attribute to set.
        :param value: the bytes of the extended attribute value
        """
        self.require_state("mounted")

        name = cstr(name, 'name')
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(flags, int):
            raise TypeError('flags must be a int')
        if not isinstance(value, bytes):
            raise TypeError('value must be a bytes')

        cdef:
            int _fd = fd
            char *_name = name
            char *_value = value
            size_t _value_len = len(value)
            int _flags = flags

        with nogil:
            ret = ceph_fsetxattr(self.cluster, _fd, _name,
                                 _value, _value_len, _flags)
        if ret < 0:
            raise make_ex(ret, "error in fsetxattr")

    def lsetxattr(self, path, name, value, flags):
        """
        Set an extended attribute on a file but do not follow symbolic link.

        :param path: the path to the file.
        :param name: the name of the extended attribute to set.
        :param value: the bytes of the extended attribute value
        """

        self.setxattr(path, name, value, flags, follow_symlink=False)

    def removexattr(self, path, name, follow_symlink=True):
        """
        Remove an extended attribute of a file.

        :param path: path of the file.
        :param name: name of the extended attribute to remove.
        """
        self.require_state("mounted")

        name = cstr(name, 'name')
        path = cstr(path, 'path')

        cdef:
            char *_path = path
            char *_name = name

        if follow_symlink:
            with nogil:
                ret = ceph_removexattr(self.cluster, _path, _name)
        else:
            with nogil:
                ret = ceph_lremovexattr(self.cluster, _path, _name)

        if ret < 0:
            raise make_ex(ret, "error in removexattr")

    def fremovexattr(self, fd, name):
        """
        Remove an extended attribute of a file.

        :param fd: the open file descriptor referring to the file.
        :param name: name of the extended attribute to remove.
        """
        self.require_state("mounted")

        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        name = cstr(name, 'name')

        cdef:
            int _fd = fd
            char *_name = name

        with nogil:
            ret = ceph_fremovexattr(self.cluster, _fd, _name)
        if ret < 0:
            raise make_ex(ret, "error in fremovexattr")

    def lremovexattr(self, path, name):
        """
        Remove an extended attribute of a file but do not follow symbolic link.

        :param path: path of the file.
        :param name: name of the extended attribute to remove.
        """
        self.removexattr(path, name, follow_symlink=False)

    def listxattr(self, path, size=65536, follow_symlink=True):
        """
        List the extended attribute keys set on a file.

        :param path: path of the file.
        :param size: the size of list buffer to be filled with extended attribute keys.
        """
        self.require_state("mounted")

        path = cstr(path, 'path')

        cdef:
            char *_path = path
            char *ret_buf = NULL
            size_t ret_length = size

        try:
            ret_buf = <char *>realloc_chk(ret_buf, ret_length)
            if follow_symlink:
                with nogil:
                    ret = ceph_listxattr(self.cluster, _path, ret_buf, ret_length)
            else:
                with nogil:
                    ret = ceph_llistxattr(self.cluster, _path, ret_buf, ret_length)

            if ret < 0:
                raise make_ex(ret, "error in listxattr")

            return ret, ret_buf[:ret]
        finally:
            free(ret_buf)

    def flistxattr(self, fd, size=65536):
        """
        List the extended attribute keys set on a file.

        :param fd: the open file descriptor referring to the file.
        :param size: the size of list buffer to be filled with extended attribute keys.
        """
        self.require_state("mounted")

        if not isinstance(fd, int):
            raise TypeError('fd must be an int')

        cdef:
            int _fd = fd
            char *ret_buf = NULL
            size_t ret_length = size

        try:
            ret_buf = <char *>realloc_chk(ret_buf, ret_length)
            with nogil:
                ret = ceph_flistxattr(self.cluster, _fd, ret_buf, ret_length)

            if ret < 0:
                raise make_ex(ret, "error in flistxattr")

            return ret, ret_buf[:ret]
        finally:
            free(ret_buf)

    def llistxattr(self, path, size=65536):
        """
        List the extended attribute keys set on a file but do not follow symbolic link.

        :param path: path of the file.
        :param size: the size of list buffer to be filled with extended attribute keys.
        """

        return self.listxattr(path, size=size, follow_symlink=False)

    def stat(self, path, follow_symlink=True):
        """
        Get a file's extended statistics and attributes.
        
        :param path: the file or directory to get the statistics of.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')

        cdef:
            char* _path = path
            statx stx

        if follow_symlink:
            with nogil:
                ret = ceph_statx(self.cluster, _path, &stx,
                                 CEPH_STATX_BASIC_STATS_CDEF, 0)
        else:
            with nogil:
                ret = ceph_statx(self.cluster, _path, &stx,
                                 CEPH_STATX_BASIC_STATS_CDEF, AT_SYMLINK_NOFOLLOW_CDEF)

        if ret < 0:
            raise make_ex(ret, "error in stat: {}".format(path.decode('utf-8')))
        return StatResult(st_dev=stx.stx_dev, st_ino=stx.stx_ino,
                          st_mode=stx.stx_mode, st_nlink=stx.stx_nlink,
                          st_uid=stx.stx_uid, st_gid=stx.stx_gid,
                          st_rdev=stx.stx_rdev, st_size=stx.stx_size,
                          st_blksize=stx.stx_blksize,
                          st_blocks=stx.stx_blocks,
                          st_atime=datetime.fromtimestamp(stx.stx_atime.tv_sec),
                          st_mtime=datetime.fromtimestamp(stx.stx_mtime.tv_sec),
                          st_ctime=datetime.fromtimestamp(stx.stx_ctime.tv_sec))

    def lstat(self, path):
        """
        Get a file's extended statistics and attributes. If the file is a
        symbolic link, return the information of the link itself rather than
        the information of the file it points to.

        :param path: the file or directory to get the statistics of.
        """
        return self.stat(path, follow_symlink=False)

    def fstat(self, fd):
        """
        Get an open file's extended statistics and attributes.

        :param fd: the file descriptor of the file to get statistics of.
         """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')

        cdef:
            int _fd = fd
            statx stx

        with nogil:
            ret = ceph_fstatx(self.cluster, _fd, &stx,
                              CEPH_STATX_BASIC_STATS_CDEF, 0)
        if ret < 0:
            raise make_ex(ret, "error in fsat")
        return StatResult(st_dev=stx.stx_dev, st_ino=stx.stx_ino,
                          st_mode=stx.stx_mode, st_nlink=stx.stx_nlink,
                          st_uid=stx.stx_uid, st_gid=stx.stx_gid,
                          st_rdev=stx.stx_rdev, st_size=stx.stx_size,
                          st_blksize=stx.stx_blksize,
                          st_blocks=stx.stx_blocks,
                          st_atime=datetime.fromtimestamp(stx.stx_atime.tv_sec),
                          st_mtime=datetime.fromtimestamp(stx.stx_mtime.tv_sec),
                          st_ctime=datetime.fromtimestamp(stx.stx_ctime.tv_sec))
    
    def statx(self, path, mask, flag):
        """
        Get a file's extended statistics and attributes.

        :param path: the file or directory to get the statistics of.
        :param mask: want bitfield of CEPH_STATX_* flags showing designed attributes.
        :param flag: bitfield that can be used to set AT_* modifier flags (AT_STATX_SYNC_AS_STAT, AT_STATX_FORCE_SYNC, AT_STATX_DONT_SYNC and AT_SYMLINK_NOFOLLOW)
        """

        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(mask, int):
            raise TypeError('flag must be a int')
        if not isinstance(flag, int):
            raise TypeError('flag must be a int')

        cdef:
            char* _path = path
            statx stx
            int _mask = mask
            int _flag = flag
            dict_result = dict()

        with nogil:
            ret = ceph_statx(self.cluster, _path, &stx, _mask, _flag)
        if ret < 0:
            raise make_ex(ret, "error in stat: %s" % path)

        if (_mask & CEPH_STATX_MODE):
            dict_result["mode"] = stx.stx_mode
        if (_mask & CEPH_STATX_NLINK):
            dict_result["nlink"] = stx.stx_nlink
        if (_mask & CEPH_STATX_UID):
            dict_result["uid"] = stx.stx_uid
        if (_mask & CEPH_STATX_GID):
            dict_result["gid"] = stx.stx_gid
        if (_mask & CEPH_STATX_RDEV):
            dict_result["rdev"] = stx.stx_rdev
        if (_mask & CEPH_STATX_ATIME):
            dict_result["atime"] = datetime.fromtimestamp(stx.stx_atime.tv_sec)
        if (_mask & CEPH_STATX_MTIME):
            dict_result["mtime"] = datetime.fromtimestamp(stx.stx_mtime.tv_sec)
        if (_mask & CEPH_STATX_CTIME):
            dict_result["ctime"] = datetime.fromtimestamp(stx.stx_ctime.tv_sec)
        if (_mask & CEPH_STATX_INO):
            dict_result["ino"] = stx.stx_ino
        if (_mask & CEPH_STATX_SIZE):
            dict_result["size"] = stx.stx_size
        if (_mask & CEPH_STATX_BLOCKS):
            dict_result["blocks"] = stx.stx_blocks
        if (_mask & CEPH_STATX_BTIME):
            dict_result["btime"] = datetime.fromtimestamp(stx.stx_btime.tv_sec)
        if (_mask & CEPH_STATX_VERSION):
            dict_result["version"] = stx.stx_version

        return dict_result

    def setattrx(self, path, dict_stx, mask, flags):
        """
        Set a file's attributes.

        :param path: the path to the file/directory to set the attributes of.
        :param mask: a mask of all the CEPH_SETATTR_* values that have been set in the statx struct.
        :param stx: a dict of statx structure that must include attribute values to set on the file.
        :param flags: mask of AT_* flags (only AT_ATTR_NOFOLLOW is respected for now)
        """

        self.require_state("mounted")
        path = cstr(path, 'path')
        if not isinstance(dict_stx, dict):
            raise TypeError('dict_stx must be a dict')
        if not isinstance(mask, int):
            raise TypeError('mask must be a int')
        if not isinstance(flags, int):
            raise TypeError('flags must be a int')

        cdef statx stx

        if (mask & CEPH_SETATTR_MODE):
            stx.stx_mode = dict_stx["mode"]
        if (mask & CEPH_SETATTR_UID):
            stx.stx_uid = dict_stx["uid"]
        if (mask & CEPH_SETATTR_GID):
            stx.stx_gid = dict_stx["gid"]
        if (mask & CEPH_SETATTR_MTIME):
            stx.stx_mtime = to_timespec(dict_stx["mtime"].timestamp())
        if (mask & CEPH_SETATTR_ATIME):
            stx.stx_atime = to_timespec(dict_stx["atime"].timestamp())
        if (mask & CEPH_SETATTR_CTIME):
            stx.stx_ctime = to_timespec(dict_stx["ctime"].timestamp())
        if (mask & CEPH_SETATTR_SIZE):
            stx.stx_size = dict_stx["size"]
        if (mask & CEPH_SETATTR_BTIME):
            stx.stx_btime = to_timespec(dict_stx["btime"].timestamp())

        cdef:
            char* _path = path
            int _mask = mask
            int _flags = flags
            dict_result = dict()

        with nogil:
            ret = ceph_setattrx(self.cluster, _path, &stx, _mask, _flags)
        if ret < 0:
            raise make_ex(ret, "error in setattrx: %s" % path)

    def fsetattrx(self, fd, dict_stx, mask):
        """
        Set a file's attributes.

        :param path: the path to the file/directory to set the attributes of.
        :param mask: a mask of all the CEPH_SETATTR_* values that have been set in the statx struct.
        :param stx: a dict of statx structure that must include attribute values to set on the file.
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be a int')
        if not isinstance(dict_stx, dict):
            raise TypeError('dict_stx must be a dict')
        if not isinstance(mask, int):
            raise TypeError('mask must be a int')

        cdef statx stx

        if (mask & CEPH_SETATTR_MODE):
            stx.stx_mode = dict_stx["mode"]
        if (mask & CEPH_SETATTR_UID):
            stx.stx_uid = dict_stx["uid"]
        if (mask & CEPH_SETATTR_GID):
            stx.stx_gid = dict_stx["gid"]
        if (mask & CEPH_SETATTR_MTIME):
            stx.stx_mtime = to_timespec(dict_stx["mtime"].timestamp())
        if (mask & CEPH_SETATTR_ATIME):
            stx.stx_atime = to_timespec(dict_stx["atime"].timestamp())
        if (mask & CEPH_SETATTR_CTIME):
            stx.stx_ctime = to_timespec(dict_stx["ctime"].timestamp())
        if (mask & CEPH_SETATTR_SIZE):
            stx.stx_size = dict_stx["size"]
        if (mask & CEPH_SETATTR_BTIME):
            stx.stx_btime = to_timespec(dict_stx["btime"].timestamp())

        cdef:
            int _fd = fd
            int _mask = mask
            dict_result = dict()

        with nogil:
            ret = ceph_fsetattrx(self.cluster, _fd, &stx, _mask)
        if ret < 0:
            raise make_ex(ret, "error in fsetattrx")

    def symlink(self, existing, newname):
        """
        Creates a symbolic link.
       
        :param existing: the path to the existing file/directory to link to.
        :param newname: the path to the new file/directory to link from.
        """
        self.require_state("mounted")
        existing = cstr(existing, 'existing')
        newname = cstr(newname, 'newname')
        cdef:
            char* _existing = existing
            char* _newname = newname

        with nogil:
            ret = ceph_symlink(self.cluster, _existing, _newname)
        if ret < 0:
            raise make_ex(ret, "error in symlink")
    
    def link(self, existing, newname):
        """
        Create a link.
        
        :param existing: the path to the existing file/directory to link to.
        :param newname: the path to the new file/directory to link from.
        """

        self.require_state("mounted")
        existing = cstr(existing, 'existing')
        newname = cstr(newname, 'newname')
        cdef:
            char* _existing = existing
            char* _newname = newname
        
        with nogil:
            ret = ceph_link(self.cluster, _existing, _newname)
        if ret < 0:
            raise make_ex(ret, "error in link")    
    
    def readlink(self, path, size):
        """
        Read a symbolic link.
      
        :param path: the path to the symlink to read
        :param size: the length of the buffer
        :rtype buf: buffer to hold the path of the file that the symlink points to.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')

        cdef:
            char* _path = path
            int64_t _size = size
            char *buf = NULL

        try:
            buf = <char *>realloc_chk(buf, _size)
            with nogil:
                ret = ceph_readlink(self.cluster, _path, buf, _size)
            if ret < 0:
                raise make_ex(ret, "error in readlink")
            return buf[:ret]
        finally:
            free(buf)

    def unlink(self, path):
        """
        Removes a file, link, or symbolic link.  If the file/link has multiple links to it, the
        file will not disappear from the namespace until all references to it are removed.
        
        :param path: the path of the file or link to unlink.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')
        cdef char* _path = path
        with nogil:
            ret = ceph_unlink(self.cluster, _path)
        if ret < 0:
            raise make_ex(ret, "error in unlink: {}".format(path.decode('utf-8')))

    def rename(self, src, dst):
        """
        Rename a file or directory.
        
        :param src: the path to the existing file or directory.
        :param dst: the new name of the file or directory.
        """
        
        self.require_state("mounted")

        src = cstr(src, 'source')
        dst = cstr(dst, 'destination')

        cdef:
            char* _src = src
            char* _dst = dst

        with nogil:
            ret = ceph_rename(self.cluster, _src, _dst)
        if ret < 0:
            raise make_ex(ret, "error in rename {} to {}".format(src.decode(
                          'utf-8'), dst.decode('utf-8')))

    def mds_command(self, mds_spec, args, input_data):
        """
        :return 3-tuple of output status int, output status string, output data
        """
        mds_spec = cstr(mds_spec, 'mds_spec')
        args = cstr(args, 'args')
        input_data = cstr(input_data, 'input_data')

        cdef:
            char *_mds_spec = opt_str(mds_spec)
            char **_cmd = to_bytes_array([args])
            size_t _cmdlen = 1

            char *_inbuf = input_data
            size_t _inbuf_len = len(input_data)

            char *_outbuf = NULL
            size_t _outbuf_len = 0
            char *_outs = NULL
            size_t _outs_len = 0

        try:
            with nogil:
                ret = ceph_mds_command(self.cluster, _mds_spec,
                                       <const char **>_cmd, _cmdlen,
                                       <const char*>_inbuf, _inbuf_len,
                                       &_outbuf, &_outbuf_len,
                                       &_outs, &_outs_len)
            my_outs = decode_cstr(_outs[:_outs_len])
            my_outbuf = _outbuf[:_outbuf_len]
            if _outs_len:
                ceph_buffer_free(_outs)
            if _outbuf_len:
                ceph_buffer_free(_outbuf)
            return (ret, my_outbuf, my_outs)
        finally:
            free(_cmd)

    def umask(self, mode) :
        self.require_state("mounted")
        cdef:
            mode_t _mode = mode
        with nogil:
            ret = ceph_umask(self.cluster, _mode)
        if ret < 0:
            raise make_ex(ret, "error in umask")
        return ret

    def lseek(self, fd, offset, whence):
        """
        Set the file's current position.
 
        :param fd: the file descriptor of the open file to read from.
        :param offset: the offset in the file to read from.  If this value is negative, the
                       function reads from the current offset of the file descriptor.
        :param whence: the flag to indicate what type of seeking to performs:SEEK_SET, SEEK_CUR, SEEK_END
        """     
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if not isinstance(offset, int):
            raise TypeError('offset must be an int')
        if not isinstance(whence, int):
            raise TypeError('whence must be an int')

        cdef:
            int _fd = fd
            int64_t _offset = offset
            int64_t _whence = whence

        with nogil:
            ret = ceph_lseek(self.cluster, _fd, _offset, _whence)

        if ret < 0:
            raise make_ex(ret, "error in lseek")

        return ret      

    def utime(self, path, times=None):
        """
        Set access and modification time for path

        :param path: file path for which timestamps have to be changed
        :param times: if times is not None, it must be a tuple (atime, mtime)
        """

        self.require_state("mounted")
        path = cstr(path, 'path')
        if times:
            if not isinstance(times, tuple):
                raise TypeError('times must be a tuple')
            if not isinstance(times[0], int):
                raise TypeError('atime must be an int')
            if not isinstance(times[1], int):
                raise TypeError('mtime must be an int')
        actime = modtime = int(time.time())
        if times:
            actime = times[0]
            modtime = times[1]

        cdef:
            char *pth = path
            utimbuf buf = utimbuf(actime, modtime)
        with nogil:
            ret = ceph_utime(self.cluster, pth, &buf)
        if ret < 0:
            raise make_ex(ret, "error in utime {}".format(path.decode('utf-8')))

    def futime(self, fd, times=None):
        """
        Set access and modification time for a file pointed by descriptor

        :param fd: file descriptor of the open file
        :param times: if times is not None, it must be a tuple (atime, mtime)
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if times:
            if not isinstance(times, tuple):
                raise TypeError('times must be a tuple')
            if not isinstance(times[0], int):
                raise TypeError('atime must be an int')
            if not isinstance(times[1], int):
                raise TypeError('mtime must be an int')
        actime = modtime = int(time.time())
        if times:
            actime = times[0]
            modtime = times[1]

        cdef:
            int _fd = fd
            utimbuf buf = utimbuf(actime, modtime)
        with nogil:
            ret = ceph_futime(self.cluster, _fd, &buf)
        if ret < 0:
            raise make_ex(ret, "error in futime")

    def utimes(self, path, times=None, follow_symlink=True):
        """
        Set access and modification time for path

        :param path: file path for which timestamps have to be changed
        :param times: if times is not None, it must be a tuple (atime, mtime)
        :param follow_symlink: perform the operation on the target file if @path
                               is a symbolic link (default)
        """

        self.require_state("mounted")
        path = cstr(path, 'path')
        if times:
            if not isinstance(times, tuple):
                raise TypeError('times must be a tuple')
            if not isinstance(times[0], (int, float)):
                raise TypeError('atime must be an int or a float')
            if not isinstance(times[1], (int, float)):
                raise TypeError('mtime must be an int or a float')
        actime = modtime = time.time()
        if times:
            actime = float(times[0])
            modtime = float(times[1])

        cdef:
            char *pth = path
            timeval *buf = [to_timeval(actime), to_timeval(modtime)]
        if follow_symlink:
            with nogil:
                ret = ceph_utimes(self.cluster, pth, buf)
        else:
            with nogil:
                ret = ceph_lutimes(self.cluster, pth, buf)
        if ret < 0:
            raise make_ex(ret, "error in utimes {}".format(path.decode('utf-8')))

    def lutimes(self, path, times=None):
        """
        Set access and modification time for a file. If the file is a symbolic
        link do not follow to the target.

        :param path: file path for which timestamps have to be changed
        :param times: if times is not None, it must be a tuple (atime, mtime)
        """
        self.utimes(path, times=times, follow_symlink=False)

    def futimes(self, fd, times=None):
        """
        Set access and modification time for a file pointer by descriptor

        :param fd: file descriptor of the open file
        :param times: if times is not None, it must be a tuple (atime, mtime)
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if times:
            if not isinstance(times, tuple):
                raise TypeError('times must be a tuple')
            if not isinstance(times[0], (int, float)):
                raise TypeError('atime must be an int or a float')
            if not isinstance(times[1], (int, float)):
                raise TypeError('mtime must be an int or a float')
        actime = modtime = time.time()
        if times:
            actime = float(times[0])
            modtime = float(times[1])

        cdef:
            int _fd = fd
            timeval *buf = [to_timeval(actime), to_timeval(modtime)]
        with nogil:
                ret = ceph_futimes(self.cluster, _fd, buf)
        if ret < 0:
            raise make_ex(ret, "error in futimes")

    def futimens(self, fd, times=None):
        """
        Set access and modification time for a file pointer by descriptor

        :param fd: file descriptor of the open file
        :param times: if times is not None, it must be a tuple (atime, mtime)
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')
        if times:
            if not isinstance(times, tuple):
                raise TypeError('times must be a tuple')
            if not isinstance(times[0], (int, float)):
                raise TypeError('atime must be an int or a float')
            if not isinstance(times[1], (int, float)):
                raise TypeError('mtime must be an int or a float')
        actime = modtime = time.time()
        if times:
            actime = float(times[0])
            modtime = float(times[1])

        cdef:
            int _fd = fd
            timespec *buf = [to_timespec(actime), to_timespec(modtime)]
        with nogil:
                ret = ceph_futimens(self.cluster, _fd, buf)
        if ret < 0:
            raise make_ex(ret, "error in futimens")

    def get_file_replication(self, fd):
        """
        Get the file replication information from an open file descriptor.

        :param fd : the open file descriptor referring to the file to get
                    the replication information of.
        """
        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')

        cdef:
            int _fd = fd

        with nogil:
            ret = ceph_get_file_replication(self.cluster, _fd)
        if ret < 0:
            raise make_ex(ret, "error in get_file_replication")

        return ret

    def get_path_replication(self, path):
        """
        Get the file replication information given the path.

        :param path: the path of the file/directory to get the replication information of.
        """
        self.require_state("mounted")
        path = cstr(path, 'path')

        cdef:
            char* _path = path

        with nogil:
            ret = ceph_get_path_replication(self.cluster, _path)
        if ret < 0:
            raise make_ex(ret, "error in get_path_replication")

        return ret

    def get_pool_id(self, pool_name):
        """
        Get the id of the named pool.

        :param pool_name: the name of the pool.
        """

        self.require_state("mounted")
        pool_name = cstr(pool_name, 'pool_name')

        cdef:
            char* _pool_name = pool_name

        with nogil:
            ret = ceph_get_pool_id(self.cluster, _pool_name)
        if ret < 0:
            raise make_ex(ret, "error in get_pool_id")

        return ret

    def get_pool_replication(self, pool_id):
        """
        Get the pool replication factor.

        :param pool_id: the pool id to look up
        """

        self.require_state("mounted")
        if not isinstance(pool_id, int):
            raise TypeError('pool_id must be an int')

        cdef:
            int _pool_id = pool_id

        with nogil:
            ret = ceph_get_pool_replication(self.cluster, _pool_id)
        if ret < 0:
            raise make_ex(ret, "error in get_pool_replication")

        return ret

    def debug_get_fd_caps(self, fd):
        """
        Get the capabilities currently issued to the client given the fd.

        :param fd: the file descriptor to get issued
        """

        self.require_state("mounted")
        if not isinstance(fd, int):
            raise TypeError('fd must be an int')

        cdef:
            int _fd = fd

        with nogil:
            ret = ceph_debug_get_fd_caps(self.cluster, _fd)
        if ret < 0:
            raise make_ex(ret, "error in debug_get_fd_caps")

        return ret

    def debug_get_file_caps(self, path):
        """
        Get the capabilities currently issued to the client given the path.

        :param path: the path of the file/directory to get the capabilities of.
        """

        self.require_state("mounted")
        path = cstr(path, 'path')

        cdef:
            char* _path = path

        with nogil:
            ret = ceph_debug_get_file_caps(self.cluster, _path)
        if ret < 0:
            raise make_ex(ret, "error in debug_get_file_caps")

        return ret

    def get_cap_return_timeout(self):
        """
        Get the amount of time that the client has to return caps

        In the event that a client does not return its caps, the MDS may blocklist
        it after this timeout. Applications should check this value and ensure
        that they set the delegation timeout to a value lower than this.
        """

        self.require_state("mounted")

        with nogil:
            ret = ceph_get_cap_return_timeout(self.cluster)
        if ret < 0:
            raise make_ex(ret, "error in get_cap_return_timeout")

        return ret

    def set_uuid(self, uuid):
        """
        Set ceph client uuid. Must be called before mount.

        :param uuid: the uuid to set
        """

        uuid = cstr(uuid, 'uuid')

        cdef:
            char* _uuid = uuid

        with nogil:
            ceph_set_uuid(self.cluster, _uuid)

    def set_session_timeout(self, timeout):
        """
        Set ceph client session timeout. Must be called before mount.

        :param timeout: the timeout to set
        """

        if not isinstance(timeout, int):
            raise TypeError('timeout must be an int')

        cdef:
            int _timeout = timeout

        with nogil:
            ceph_set_session_timeout(self.cluster, _timeout)

    def get_layout(self, fd):
        """
        Set ceph client session timeout. Must be called before mount.

        :param fd: file descriptor of the file/directory for which to get the layout
        """

        if not isinstance(fd, int):
            raise TypeError('fd must be an int')

        cdef:
            int _fd = fd
            int stripe_unit
            int stripe_count
            int object_size
            int pool_id
            char *buf = NULL
            int buflen = 256
            dict_result = dict()

        with nogil:
            ret = ceph_get_file_layout(self.cluster, _fd, &stripe_unit, &stripe_count, &object_size, &pool_id)
        if ret < 0:
            raise make_ex(stripe_unit, "error in get_file_layout")
        dict_result["stripe_unit"] = stripe_unit
        dict_result["stripe_count"] = stripe_count
        dict_result["object_size"] = object_size
        dict_result["pool_id"] = pool_id

        try:
            while True:
                buf = <char *>realloc_chk(buf, buflen)
                with nogil:
                    ret = ceph_get_file_pool_name(self.cluster, _fd, buf, buflen)
                if ret > 0:
                    dict_result["pool_name"] = decode_cstr(buf)
                    return dict_result
                elif ret == -CEPHFS_ERANGE:
                    buflen = buflen * 2
                else:
                    raise make_ex(ret, "error in get_file_pool_name")
        finally:
           free(buf)


    def get_default_pool(self):
        """
        Get the default pool name and id of cephfs. This returns dict{pool_name, pool_id}.
        """

        cdef:
            char *buf = NULL
            int buflen = 256
            dict_result = dict()

        try:
            while True:
                buf = <char *>realloc_chk(buf, buflen)
                with nogil:
                    ret = ceph_get_default_data_pool_name(self.cluster, buf, buflen)
                if ret > 0:
                    dict_result["pool_name"] = decode_cstr(buf)
                    break
                elif ret == -CEPHFS_ERANGE:
                    buflen = buflen * 2
                else:
                    raise make_ex(ret, "error in get_default_data_pool_name")

            with nogil:
                ret = ceph_get_pool_id(self.cluster, buf)
            if ret < 0:
                raise make_ex(ret, "error in get_pool_id")
            dict_result["pool_id"] = ret
            return dict_result

        finally:
           free(buf)