ceph/src/pybind/cephfs/cephfs.pyx

   1 """
   2 This module is a thin wrapper around libcephfs.
   3 """
   4
   5 from cpython cimport PyObject, ref, exc
   6 from libc cimport errno
   7 from libc.stdint cimport *
   8 from libc.stdlib cimport malloc, realloc, free
   9
  10 cimport rados
  11
  12 from collections import namedtuple
  13 from datetime import datetime
  14 import errno
  15 import os
  16 import sys
  17
  18 # Are we running Python 2.x
  19 if sys.version_info[0] < 3:
  20     str_type = basestring
  21 else:
  22     str_type = str
  23
  24
  25 cdef extern from "Python.h":
  26     # These are in cpython/string.pxd, but use "object" types instead of
  27     # PyObject*, which invokes assumptions in cpython that we need to
  28     # legitimately break to implement zero-copy string buffers in Image.read().
  29     # This is valid use of the Python API and documented as a special case.
  30     PyObject *PyBytes_FromStringAndSize(char *v, Py_ssize_t len) except NULL
  31     char* PyBytes_AsString(PyObject *string) except NULL
  32     int _PyBytes_Resize(PyObject **string, Py_ssize_t newsize) except -1
  33     void PyEval_InitThreads()
  34
  35
  36 cdef extern from "sys/statvfs.h":
  37     cdef struct statvfs:
  38         unsigned long int f_bsize
  39         unsigned long int f_frsize
  40         unsigned long int f_blocks
  41         unsigned long int f_bfree
  42         unsigned long int f_bavail
  43         unsigned long int f_files
  44         unsigned long int f_ffree
  45         unsigned long int f_favail
  46         unsigned long int f_fsid
  47         unsigned long int f_flag
  48         unsigned long int f_namemax
  49         unsigned long int f_padding[32]
  50
  51
  52 cdef extern from "dirent.h":
  53     cdef struct dirent:
  54         long int d_ino
  55         unsigned long int d_off
  56         unsigned short int d_reclen
  57         unsigned char d_type
  58         char d_name[256]
  59
  60
  61 cdef extern from "time.h":
  62     ctypedef long int time_t
  63
  64 cdef extern from "time.h":
  65     cdef struct timespec:
  66         time_t      tv_sec
  67         long int    tv_nsec
  68
  69 cdef extern from "sys/types.h":
  70     ctypedef unsigned long mode_t
  71
  72 cdef extern from "cephfs/ceph_statx.h":
  73     cdef struct statx "ceph_statx":
  74         uint32_t    stx_mask
  75         uint32_t    stx_blksize
  76         uint32_t    stx_nlink
  77         uint32_t    stx_uid
  78         uint32_t    stx_gid
  79         uint16_t    stx_mode
  80         uint64_t    stx_ino
  81         uint64_t    stx_size
  82         uint64_t    stx_blocks
  83         uint64_t    stx_dev
  84         uint64_t    stx_rdev
  85         timespec    stx_atime
  86         timespec    stx_ctime
  87         timespec    stx_mtime
  88         timespec    stx_btime
  89         uint64_t    stx_version
  90
  91 cdef extern from "cephfs/libcephfs.h" nogil:
  92     cdef struct ceph_mount_info:
  93         pass
  94
  95     cdef struct ceph_dir_result:
  96         pass
  97
  98     ctypedef void* rados_t
  99
 100     const char *ceph_version(int *major, int *minor, int *patch)
 101
 102     int ceph_create(ceph_mount_info **cmount, const char * const id)
 103     int ceph_create_from_rados(ceph_mount_info **cmount, rados_t cluster)
 104     int ceph_init(ceph_mount_info *cmount)
 105     void ceph_shutdown(ceph_mount_info *cmount)
 106
 107     int ceph_conf_read_file(ceph_mount_info *cmount, const char *path_list)
 108     int ceph_conf_parse_argv(ceph_mount_info *cmount, int argc, const char **argv)
 109     int ceph_conf_get(ceph_mount_info *cmount, const char *option, char *buf, size_t len)
 110     int ceph_conf_set(ceph_mount_info *cmount, const char *option, const char *value)
 111
 112     int ceph_mount(ceph_mount_info *cmount, const char *root)
 113     int ceph_select_filesystem(ceph_mount_info *cmount, const char *fs_name)
 114     int ceph_unmount(ceph_mount_info *cmount)
 115     int ceph_abort_conn(ceph_mount_info *cmount)
 116     uint64_t ceph_get_instance_id(ceph_mount_info *cmount)
 117     int ceph_fstatx(ceph_mount_info *cmount, int fd, statx *stx, unsigned want, unsigned flags)
 118     int ceph_statx(ceph_mount_info *cmount, const char *path, statx *stx, unsigned want, unsigned flags)
 119     int ceph_statfs(ceph_mount_info *cmount, const char *path, statvfs *stbuf)
 120
 121     int ceph_mds_command(ceph_mount_info *cmount, const char *mds_spec, const char **cmd, size_t cmdlen,
 122                          const char *inbuf, size_t inbuflen, char **outbuf, size_t *outbuflen,
 123                          char **outs, size_t *outslen)
 124     int ceph_rename(ceph_mount_info *cmount, const char *from_, const char *to)
 125     int ceph_link(ceph_mount_info *cmount, const char *existing, const char *newname)
 126     int ceph_unlink(ceph_mount_info *cmount, const char *path)
 127     int ceph_symlink(ceph_mount_info *cmount, const char *existing, const char *newname)
 128     int ceph_readlink(ceph_mount_info *cmount, const char *path, char *buf, int64_t size)
 129     int ceph_setxattr(ceph_mount_info *cmount, const char *path, const char *name,
 130                       const void *value, size_t size, int flags)
 131     int ceph_getxattr(ceph_mount_info *cmount, const char *path, const char *name,
 132                       void *value, size_t size)
 133     int ceph_write(ceph_mount_info *cmount, int fd, const char *buf, int64_t size, int64_t offset)
 134     int ceph_read(ceph_mount_info *cmount, int fd, char *buf, int64_t size, int64_t offset)
 135     int ceph_flock(ceph_mount_info *cmount, int fd, int operation, uint64_t owner)
 136     int ceph_close(ceph_mount_info *cmount, int fd)
 137     int ceph_open(ceph_mount_info *cmount, const char *path, int flags, mode_t mode)
 138     int ceph_mkdir(ceph_mount_info *cmount, const char *path, mode_t mode)
 139     int ceph_mkdirs(ceph_mount_info *cmount, const char *path, mode_t mode)
 140     int ceph_closedir(ceph_mount_info *cmount, ceph_dir_result *dirp)
 141     int ceph_opendir(ceph_mount_info *cmount, const char *name, ceph_dir_result **dirpp)
 142     int ceph_chdir(ceph_mount_info *cmount, const char *path)
 143     dirent * ceph_readdir(ceph_mount_info *cmount, ceph_dir_result *dirp)
 144     int ceph_rmdir(ceph_mount_info *cmount, const char *path)
 145     const char* ceph_getcwd(ceph_mount_info *cmount)
 146     int ceph_sync_fs(ceph_mount_info *cmount)
 147     int ceph_fsync(ceph_mount_info *cmount, int fd, int syncdataonly)
 148     int ceph_conf_parse_argv(ceph_mount_info *cmount, int argc, const char **argv)
 149     int ceph_chmod(ceph_mount_info *cmount, const char *path, mode_t mode)
 150     void ceph_buffer_free(char *buf)
 151     mode_t ceph_umask(ceph_mount_info *cmount, mode_t mode)
 152
 153
 154 class Error(Exception):
 155     pass
 156
 157
 158 class OSError(Error):
 159     def __init__(self, errno, strerror):
 160         self.errno = errno
 161         self.strerror = strerror
 162
 163     def __str__(self):
 164         return '[Errno {0}] {1}'.format(self.errno, self.strerror)
 165
 166
 167 class PermissionError(OSError):
 168     pass
 169
 170
 171 class ObjectNotFound(OSError):
 172     pass
 173
 174
 175 class NoData(OSError):
 176     pass
 177
 178
 179 class ObjectExists(OSError):
 180     pass
 181
 182
 183 class IOError(OSError):
 184     pass
 185
 186
 187 class NoSpace(OSError):
 188     pass
 189
 190
 191 class InvalidValue(OSError):
 192     pass
 193
 194
 195 class OperationNotSupported(OSError):
 196     pass
 197
 198
 199 class LibCephFSStateError(Error):
 200     pass
 201
 202
 203 class WouldBlock(OSError):
 204     pass
 205
 206
 207 class OutOfRange(OSError):
 208     pass
 209
 210
 211 IF UNAME_SYSNAME == "FreeBSD":
 212     cdef errno_to_exception =  {
 213         errno.EPERM      : PermissionError,
 214         errno.ENOENT     : ObjectNotFound,
 215         errno.EIO        : IOError,
 216         errno.ENOSPC     : NoSpace,
 217         errno.EEXIST     : ObjectExists,
 218         errno.ENOATTR    : NoData,
 219         errno.EINVAL     : InvalidValue,
 220         errno.EOPNOTSUPP : OperationNotSupported,
 221         errno.ERANGE     : OutOfRange,
 222         errno.EWOULDBLOCK: WouldBlock,
 223     }
 224 ELSE:
 225     cdef errno_to_exception =  {
 226         errno.EPERM      : PermissionError,
 227         errno.ENOENT     : ObjectNotFound,
 228         errno.EIO        : IOError,
 229         errno.ENOSPC     : NoSpace,
 230         errno.EEXIST     : ObjectExists,
 231         errno.ENODATA    : NoData,
 232         errno.EINVAL     : InvalidValue,
 233         errno.EOPNOTSUPP : OperationNotSupported,
 234         errno.ERANGE     : OutOfRange,
 235         errno.EWOULDBLOCK: WouldBlock,
 236     }
 237
 238
 239 cdef make_ex(ret, msg):
 240     """
 241     Translate a librados return code into an exception.
 242
 243     :param ret: the return code
 244     :type ret: int
 245     :param msg: the error message to use
 246     :type msg: str
 247     :returns: a subclass of :class:`Error`
 248     """
 249     ret = abs(ret)
 250     if ret in errno_to_exception:
 251         return errno_to_exception[ret](ret, msg)
 252     else:
 253         return Error(ret, msg + (": error code %d" % ret))
 254
 255
 256 class DirEntry(namedtuple('DirEntry',
 257                ['d_ino', 'd_off', 'd_reclen', 'd_type', 'd_name'])):
 258     DT_DIR = 0x4
 259     DT_REG = 0xA
 260     DT_LNK = 0xC
 261     def is_dir(self):
 262         return self.d_type == self.DT_DIR
 263
 264     def is_symbol_file(self):
 265         return self.d_type == self.DT_LNK
 266
 267     def is_file(self):
 268         return self.d_type == self.DT_REG
 269
 270 StatResult = namedtuple('StatResult',
 271                         ["st_dev", "st_ino", "st_mode", "st_nlink", "st_uid",
 272                          "st_gid", "st_rdev", "st_size", "st_blksize",
 273                          "st_blocks", "st_atime", "st_mtime", "st_ctime"])
 274
 275 cdef class DirResult(object):
 276     cdef ceph_dir_result *handler
 277
 278
 279 def cstr(val, name, encoding="utf-8", opt=False):
 280     """
 281     Create a byte string from a Python string
 282
 283     :param basestring val: Python string
 284     :param str name: Name of the string parameter, for exceptions
 285     :param str encoding: Encoding to use
 286     :param bool opt: If True, None is allowed
 287     :rtype: bytes
 288     :raises: :class:`InvalidArgument`
 289     """
 290     if opt and val is None:
 291         return None
 292     if isinstance(val, bytes):
 293         return val
 294     elif isinstance(val, unicode):
 295         return val.encode(encoding)
 296     else:
 297         raise TypeError('%s must be a string' % name)
 298
 299
 300 def cstr_list(list_str, name, encoding="utf-8"):
 301     return [cstr(s, name) for s in list_str]
 302
 303
 304 def decode_cstr(val, encoding="utf-8"):
 305     """
 306     Decode a byte string into a Python string.
 307
 308     :param bytes val: byte string
 309     :rtype: unicode or None
 310     """
 311     if val is None:
 312         return None
 313
 314     return val.decode(encoding)
 315
 316
 317 cdef char* opt_str(s) except? NULL:
 318     if s is None:
 319         return NULL
 320     return s
 321
 322
 323 cdef char ** to_bytes_array(list_bytes):
 324     cdef char **ret = <char **>malloc(len(list_bytes) * sizeof(char *))
 325     if ret == NULL:
 326         raise MemoryError("malloc failed")
 327     for i in xrange(len(list_bytes)):
 328         ret[i] = <char *>list_bytes[i]
 329     return ret
 330
 331
 332 cdef void* realloc_chk(void* ptr, size_t size) except NULL:
 333     cdef void *ret = realloc(ptr, size)
 334     if ret == NULL:
 335         raise MemoryError("realloc failed")
 336     return ret
 337
 338
 339 cdef class LibCephFS(object):
 340     """libcephfs python wrapper"""
 341
 342     cdef public object state
 343     cdef ceph_mount_info *cluster
 344
 345     def require_state(self, *args):
 346         if self.state in args:
 347             return
 348         raise LibCephFSStateError("You cannot perform that operation on a "
 349                                   "CephFS object in state %s." % (self.state))
 350
 351     def __cinit__(self, conf=None, conffile=None, auth_id=None, rados_inst=None):
 352         """Create a libcephfs wrapper
 353
 354         :param conf dict opt: settings overriding the default ones and conffile
 355         :param conffile str opt: the path to ceph.conf to override the default settings
 356         :auth_id str opt: the id used to authenticate the client entity
 357         :rados_inst Rados opt: a rados.Rados instance
 358         """
 359         PyEval_InitThreads()
 360         self.state = "uninitialized"
 361         if rados_inst is not None:
 362             if auth_id is not None or conffile is not None or conf is not None:
 363                 raise make_ex(errno.EINVAL,
 364                               "May not pass RADOS instance as well as other configuration")
 365
 366             self.create_with_rados(rados_inst)
 367         else:
 368             self.create(conf, conffile, auth_id)
 369
 370     def create_with_rados(self, rados.Rados rados_inst):
 371         cdef int ret
 372         with nogil:
 373             ret = ceph_create_from_rados(&self.cluster, rados_inst.cluster)
 374         if ret != 0:
 375             raise Error("libcephfs_initialize failed with error code: %d" % ret)
 376         self.state = "configuring"
 377
 378     def create(self, conf=None, conffile=None, auth_id=None):
 379         """
 380         Create a mount handle for interacting with Ceph.  All libcephfs
 381         functions operate on a mount info handle.
 382
 383         :param conf dict opt: settings overriding the default ones and conffile
 384         :param conffile str opt: the path to ceph.conf to override the default settings
 385         :auth_id str opt: the id used to authenticate the client entity
 386         """
 387         if conf is not None and not isinstance(conf, dict):
 388             raise TypeError("conf must be dict or None")
 389         cstr(conffile, 'configfile', opt=True)
 390         auth_id = cstr(auth_id, 'auth_id', opt=True)
 391
 392         cdef:
 393             char* _auth_id = opt_str(auth_id)
 394             int ret
 395
 396         with nogil:
 397             ret = ceph_create(&self.cluster, <const char*>_auth_id)
 398         if ret != 0:
 399             raise Error("libcephfs_initialize failed with error code: %d" % ret)
 400
 401         self.state = "configuring"
 402         if conffile is not None:
 403             # read the default conf file when '' is given
 404             if conffile == '':
 405                 conffile = None
 406             self.conf_read_file(conffile)
 407         if conf is not None:
 408             for key, value in conf.iteritems():
 409                 self.conf_set(key, value)
 410
 411     def conf_read_file(self, conffile=None):
 412         """
 413         Load the ceph configuration from the specified config file.
 414
 415         :param conffile str opt: the path to ceph.conf to override the default settings
 416         """
 417         conffile = cstr(conffile, 'conffile', opt=True)
 418         cdef:
 419             char *_conffile = opt_str(conffile)
 420         with nogil:
 421             ret = ceph_conf_read_file(self.cluster, <const char*>_conffile)
 422         if ret != 0:
 423             raise make_ex(ret, "error calling conf_read_file")
 424
 425     def conf_parse_argv(self, argv):
 426         """
 427         Parse the command line arguments and load the configuration parameters.
 428
 429         :param argv: the argument list
 430         """
 431         self.require_state("configuring")
 432         cargv = cstr_list(argv, 'argv')
 433         cdef:
 434             int _argc = len(argv)
 435             char **_argv = to_bytes_array(cargv)
 436
 437         try:
 438             with nogil:
 439                 ret = ceph_conf_parse_argv(self.cluster, _argc,
 440                                            <const char **>_argv)
 441             if ret != 0:
 442                 raise make_ex(ret, "error calling conf_parse_argv")
 443         finally:
 444             free(_argv)
 445
 446     def shutdown(self):
 447         """
 448         Unmount and destroy the ceph mount handle.
 449         """
 450         if self.state in ["initialized", "mounted"]:
 451             with nogil:
 452                 ceph_shutdown(self.cluster)
 453             self.state = "shutdown"
 454
 455     def __enter__(self):
 456         self.mount()
 457         return self
 458
 459     def __exit__(self, type_, value, traceback):
 460         self.shutdown()
 461         return False
 462
 463     def __dealloc__(self):
 464         self.shutdown()
 465
 466     def version(self):
 467         """
 468         Get the version number of the ``libcephfs`` C library.
 469
 470         :returns: a tuple of ``(major, minor, extra)`` components of the
 471                   libcephfs version
 472         """
 473         cdef:
 474             int major = 0
 475             int minor = 0
 476             int extra = 0
 477         with nogil:
 478             ceph_version(&major, &minor, &extra)
 479         return (major, minor, extra)
 480
 481     def conf_get(self, option):
 482         """
 483         Gets the configuration value as a string.
 484
 485         :param option: the config option to get
 486         """
 487         self.require_state("configuring", "initialized", "mounted")
 488
 489         option = cstr(option, 'option')
 490         cdef:
 491             char *_option = option
 492             size_t length = 20
 493             char *ret_buf = NULL
 494
 495         try:
 496             while True:
 497                 ret_buf = <char *>realloc_chk(ret_buf, length)
 498                 with nogil:
 499                     ret = ceph_conf_get(self.cluster, _option, ret_buf, length)
 500                 if ret == 0:
 501                     return decode_cstr(ret_buf)
 502                 elif ret == -errno.ENAMETOOLONG:
 503                     length = length * 2
 504                 elif ret == -errno.ENOENT:
 505                     return None
 506                 else:
 507                     raise make_ex(ret, "error calling conf_get")
 508         finally:
 509             free(ret_buf)
 510
 511     def conf_set(self, option, val):
 512         """
 513         Sets a configuration value from a string.
 514
 515         :param option: the configuration option to set
 516         :param value: the value of the configuration option to set
 517         """
 518         self.require_state("configuring", "initialized", "mounted")
 519
 520         option = cstr(option, 'option')
 521         val = cstr(val, 'val')
 522         cdef:
 523             char *_option = option
 524             char *_val = val
 525
 526         with nogil:
 527             ret = ceph_conf_set(self.cluster, _option, _val)
 528         if ret != 0:
 529             raise make_ex(ret, "error calling conf_set")
 530
 531     def init(self):
 532         """
 533         Initialize the filesystem client (but do not mount the filesystem yet)
 534         """
 535         self.require_state("configuring")
 536         with nogil:
 537             ret = ceph_init(self.cluster)
 538         if ret != 0:
 539             raise make_ex(ret, "error calling ceph_init")
 540         self.state = "initialized"
 541
 542     def mount(self, mount_root=None, filesystem_name=None):
 543         """
 544         Perform a mount using the path for the root of the mount.
 545         """
 546         if self.state == "configuring":
 547             self.init()
 548         self.require_state("initialized")
 549
 550         # Configure which filesystem to mount if one was specified
 551         if filesystem_name is None:
 552             filesystem_name = b""
 553         cdef:
 554             char *_filesystem_name = filesystem_name
 555         if filesystem_name:
 556             with nogil:
 557                 ret = ceph_select_filesystem(self.cluster,
 558                         _filesystem_name)
 559             if ret != 0:
 560                 raise make_ex(ret, "error calling ceph_select_filesystem")
 561
 562         # Prepare mount_root argument, default to "/"
 563         root = b"/" if mount_root is None else mount_root
 564         cdef:
 565             char *_mount_root = root
 566
 567         with nogil:
 568             ret = ceph_mount(self.cluster, _mount_root)
 569         if ret != 0:
 570             raise make_ex(ret, "error calling ceph_mount")
 571         self.state = "mounted"
 572
 573     def unmount(self):
 574         """
 575         Unmount a mount handle.
 576         """
 577         self.require_state("mounted")
 578         with nogil:
 579             ret = ceph_unmount(self.cluster)
 580         if ret != 0:
 581             raise make_ex(ret, "error calling ceph_unmount")
 582         self.state = "initialized"
 583
 584     def abort_conn(self):
 585         """
 586         Abort mds connections.
 587         """
 588         self.require_state("mounted")
 589         with nogil:
 590             ret = ceph_abort_conn(self.cluster)
 591         if ret != 0:
 592             raise make_ex(ret, "error calling ceph_abort_conn")
 593         self.state = "initialized"
 594
 595     def get_instance_id(self):
 596         """
 597         Get a global id for current instance
 598         """
 599         self.require_state("initialized", "mounted")
 600         with nogil:
 601             ret = ceph_get_instance_id(self.cluster)
 602         return ret;
 603
 604     def statfs(self, path):
 605         """
 606         Perform a statfs on the ceph file system.  This call fills in file system wide statistics
 607         into the passed in buffer.
 608
 609         :param path: any path within the mounted filesystem
 610         """
 611         self.require_state("mounted")
 612         path = cstr(path, 'path')
 613         cdef:
 614             char* _path = path
 615             statvfs statbuf
 616
 617         with nogil:
 618             ret = ceph_statfs(self.cluster, _path, &statbuf)
 619         if ret < 0:
 620             raise make_ex(ret, "statfs failed: %s" % path)
 621         return {'f_bsize': statbuf.f_bsize,
 622                 'f_frsize': statbuf.f_frsize,
 623                 'f_blocks': statbuf.f_blocks,
 624                 'f_bfree': statbuf.f_bfree,
 625                 'f_bavail': statbuf.f_bavail,
 626                 'f_files': statbuf.f_files,
 627                 'f_ffree': statbuf.f_ffree,
 628                 'f_favail': statbuf.f_favail,
 629                 'f_fsid': statbuf.f_fsid,
 630                 'f_flag': statbuf.f_flag,
 631                 'f_namemax': statbuf.f_namemax}
 632
 633     def sync_fs(self):
 634         """
 635         Synchronize all filesystem data to persistent media
 636         """
 637         self.require_state("mounted")
 638         with nogil:
 639             ret = ceph_sync_fs(self.cluster)
 640         if ret < 0:
 641             raise make_ex(ret, "sync_fs failed")
 642
 643     def fsync(self, int fd, int syncdataonly):
 644         """
 645         Synchronize an open file to persistent media.
 646
 647         :param fd: the file descriptor of the file to sync.
 648         :param syncdataonly: a boolean whether to synchronize metadata and data (0)
 649                              or just data (1).
 650         """
 651         self.require_state("mounted")
 652         with nogil:
 653             ret = ceph_fsync(self.cluster, fd, syncdataonly)
 654         if ret < 0:
 655             raise make_ex(ret, "fsync failed")
 656
 657     def getcwd(self):
 658         """
 659         Get the current working directory.
 660
 661         :rtype the path to the current working directory
 662         """
 663         self.require_state("mounted")
 664         with nogil:
 665             ret = ceph_getcwd(self.cluster)
 666         return ret
 667
 668     def chdir(self, path):
 669         """
 670         Change the current working directory.
 671
 672         :param path the path to the working directory to change into.
 673         """
 674         self.require_state("mounted")
 675
 676         path = cstr(path, 'path')
 677         cdef char* _path = path
 678         with nogil:
 679             ret = ceph_chdir(self.cluster, _path)
 680         if ret < 0:
 681             raise make_ex(ret, "chdir failed")
 682
 683     def opendir(self, path):
 684         """
 685         Open the given directory.
 686
 687         :param path: the path name of the directory to open.  Must be either an absolute path
 688                      or a path relative to the current working directory.
 689         :param dir_handler: the directory result pointer structure to fill in.
 690         """
 691         self.require_state("mounted")
 692
 693         path = cstr(path, 'path')
 694         cdef:
 695             char* _path = path
 696             ceph_dir_result *dir_handler
 697         with nogil:
 698             ret = ceph_opendir(self.cluster, _path, &dir_handler);
 699         if ret < 0:
 700             raise make_ex(ret, "opendir failed")
 701         d = DirResult()
 702         d.handler = dir_handler
 703         return d
 704
 705     def readdir(self, DirResult dir_handler):
 706         """
 707         Get the next entry in an open directory.
 708
 709         :param dir_handler: the directory stream pointer from an opendir holding the state of the
 710                             next entry to return.
 711         :rtype dir_handler: the next directory entry or NULL if at the end of the directory (or the directory is empty.
 712                             This pointer should not be freed by the caller, and is only safe to access between return and
 713                             the next call to readdir or closedir.
 714         """
 715         self.require_state("mounted")
 716
 717         cdef ceph_dir_result *_dir_handler = dir_handler.handler
 718         with nogil:
 719             dirent = ceph_readdir(self.cluster, _dir_handler)
 720         if not dirent:
 721             return None
 722
 723         d_name = dirent.d_name if sys.version[0:2] == '2.' else dirent.d_name.\
 724                  decode()
 725         return DirEntry(d_ino=dirent.d_ino,
 726                         d_off=dirent.d_off,
 727                         d_reclen=dirent.d_reclen,
 728                         d_type=dirent.d_type,
 729                         d_name=d_name)
 730
 731     def closedir(self, DirResult dir_handler):
 732         """
 733         Close the open directory.
 734
 735         :param dir_handler: the directory result pointer (set by ceph_opendir) to close
 736         """
 737         self.require_state("mounted")
 738         cdef:
 739             ceph_dir_result *_dir_handler = dir_handler.handler
 740
 741         with nogil:
 742             ret = ceph_closedir(self.cluster, _dir_handler)
 743         if ret < 0:
 744             raise make_ex(ret, "closedir failed")
 745
 746     def mkdir(self, path, mode):
 747         """
 748         Create a directory.
 749
 750         :param path: the path of the directory to create.  This must be either an
 751                      absolute path or a relative path off of the current working directory.
 752         :param mode the permissions the directory should have once created.
 753         """
 754
 755         self.require_state("mounted")
 756         path = cstr(path, 'path')
 757         if not isinstance(mode, int):
 758             raise TypeError('mode must be an int')
 759         cdef:
 760             char* _path = path
 761             int _mode = mode
 762         with nogil:
 763             ret = ceph_mkdir(self.cluster, _path, _mode)
 764         if ret < 0:
 765             raise make_ex(ret, "error in mkdir '%s'" % path)
 766
 767     def chmod(self, path, mode) :
 768         """
 769         Change directory mode.
 770         :param path: the path of the directory to create.  This must be either an
 771                     absolute path or a relative path off of the current working directory.
 772         :param mode the permissions the directory should have once created.
 773         """
 774         self.require_state("mounted")
 775         path = cstr(path, 'path')
 776         if not isinstance(mode, int):
 777             raise TypeError('mode must be an int')
 778         cdef:
 779             char* _path = path
 780             int _mode = mode
 781         with nogil:
 782             ret = ceph_chmod(self.cluster, _path, _mode)
 783         if ret < 0:
 784             raise make_ex(ret, "error in chmod '%s'" % path)
 785
 786     def mkdirs(self, path, mode):
 787         """
 788         Create multiple directories at once.
 789
 790         :param path: the full path of directories and sub-directories that should
 791                      be created.
 792         :param mode the permissions the directory should have once created
 793         """
 794         self.require_state("mounted")
 795         path = cstr(path, 'path')
 796         if not isinstance(mode, int):
 797             raise TypeError('mode must be an int')
 798         cdef:
 799             char* _path = path
 800             int _mode = mode
 801
 802         with nogil:
 803             ret = ceph_mkdirs(self.cluster, _path, _mode)
 804         if ret < 0:
 805             raise make_ex(ret, "error in mkdirs '%s'" % path)
 806
 807     def rmdir(self, path):
 808         """
 809         Remove a directory.
 810
 811         :param path: the path of the directory to remove.
 812         """
 813         self.require_state("mounted")
 814         path = cstr(path, 'path')
 815         cdef char* _path = path
 816         ret = ceph_rmdir(self.cluster, _path)
 817         if ret < 0:
 818             raise make_ex(ret, "error in rmdir '%s'" % path)
 819
 820     def open(self, path, flags, mode=0):
 821         """
 822         Create and/or open a file.
 823
 824         :param path: the path of the file to open.  If the flags parameter includes O_CREAT,
 825                      the file will first be created before opening.
 826         :param flags: set of option masks that control how the file is created/opened.
 827         :param mode: the permissions to place on the file if the file does not exist and O_CREAT
 828                      is specified in the flags.
 829         """
 830         self.require_state("mounted")
 831         path = cstr(path, 'path')
 832
 833         if not isinstance(mode, int):
 834             raise TypeError('mode must be an int')
 835         if isinstance(flags, str):
 836             cephfs_flags = 0
 837             if flags == '':
 838                 cephfs_flags = os.O_RDONLY
 839             else:
 840                 access_flags = 0;
 841                 for c in flags:
 842                     if c == 'r':
 843                         access_flags = 1;
 844                     elif c == 'w':
 845                         access_flags = 2;
 846                         cephfs_flags |= os.O_TRUNC | os.O_CREAT
 847                     elif access_flags > 0 and c == '+':
 848                         access_flags = 3;
 849                     else:
 850                         raise make_ex(errno.EOPNOTSUPP,
 851                                       "open flags doesn't support %s" % c)
 852
 853                 if access_flags == 1:
 854                     cephfs_flags |= os.O_RDONLY;
 855                 elif access_flags == 2:
 856                     cephfs_flags |= os.O_WRONLY;
 857                 else:
 858                     cephfs_flags |= os.O_RDWR;
 859
 860         elif isinstance(flags, int):
 861             cephfs_flags = flags
 862         else:
 863             raise TypeError("flags must be a string or an integer")
 864
 865         cdef:
 866             char* _path = path
 867             int _flags = cephfs_flags
 868             int _mode = mode
 869
 870         with nogil:
 871             ret = ceph_open(self.cluster, _path, _flags, _mode)
 872         if ret < 0:
 873             raise make_ex(ret, "error in open '%s'" % path)
 874         return ret
 875
 876     def close(self, fd):
 877         """
 878         Close the open file.
 879
 880         :param fd: the file descriptor referring to the open file.
 881         """
 882
 883         self.require_state("mounted")
 884         if not isinstance(fd, int):
 885             raise TypeError('fd must be an int')
 886         cdef int _fd = fd
 887         with nogil:
 888             ret = ceph_close(self.cluster, _fd)
 889         if ret < 0:
 890             raise make_ex(ret, "error in close")
 891
 892     def read(self, fd, offset, l):
 893         """
 894         Read data from the file.
 895
 896         :param fd : the file descriptor of the open file to read from.
 897         :param offset : the offset in the file to read from.  If this value is negative, the
 898                         function reads from the current offset of the file descriptor.
 899         :param l : the flag to indicate what type of seeking to perform
 900         """
 901         self.require_state("mounted")
 902         if not isinstance(offset, int):
 903             raise TypeError('offset must be an int')
 904         if not isinstance(l, int):
 905             raise TypeError('l must be an int')
 906         if not isinstance(fd, int):
 907             raise TypeError('fd must be an int')
 908         cdef:
 909             int _fd = fd
 910             int64_t _offset = offset
 911             int64_t _length = l
 912
 913             char *ret_buf
 914             PyObject* ret_s = NULL
 915
 916         ret_s = PyBytes_FromStringAndSize(NULL, _length)
 917         try:
 918             ret_buf = PyBytes_AsString(ret_s)
 919             with nogil:
 920                 ret = ceph_read(self.cluster, _fd, ret_buf, _length, _offset)
 921             if ret < 0:
 922                 raise make_ex(ret, "error in read")
 923
 924             if ret != _length:
 925                 _PyBytes_Resize(&ret_s, ret)
 926
 927             return <object>ret_s
 928         finally:
 929             # We DECREF unconditionally: the cast to object above will have
 930             # INCREFed if necessary. This also takes care of exceptions,
 931             # including if _PyString_Resize fails (that will free the string
 932             # itself and set ret_s to NULL, hence XDECREF).
 933             ref.Py_XDECREF(ret_s)
 934
 935     def write(self, fd, buf, offset):
 936         """
 937         Write data to a file.
 938
 939         :param fd : the file descriptor of the open file to write to
 940         :param buf : the bytes to write to the file
 941         :param offset : the offset of the file write into.  If this value is negative, the
 942                         function writes to the current offset of the file descriptor.
 943         """
 944         self.require_state("mounted")
 945         if not isinstance(fd, int):
 946             raise TypeError('fd must be an int')
 947         if not isinstance(buf, bytes):
 948             raise TypeError('buf must be a bytes')
 949         if not isinstance(offset, int):
 950             raise TypeError('offset must be an int')
 951
 952         cdef:
 953             int _fd = fd
 954             char *_data = buf
 955             int64_t _offset = offset
 956
 957             size_t length = len(buf)
 958
 959         with nogil:
 960             ret = ceph_write(self.cluster, _fd, _data, length, _offset)
 961         if ret < 0:
 962             raise make_ex(ret, "error in write")
 963         return ret
 964
 965     def flock(self, fd, operation, owner):
 966         """
 967         Apply or remove an advisory lock.
 968
 969         :param fd: the open file descriptor to change advisory lock.
 970         :param operation: the advisory lock operation to be performed on the file
 971         :param owner: the user-supplied owner identifier (an arbitrary integer)
 972         """
 973         self.require_state("mounted")
 974         if not isinstance(fd, int):
 975             raise TypeError('fd must be an int')
 976         if not isinstance(operation, int):
 977             raise TypeError('operation must be an int')
 978
 979         cdef:
 980             int _fd = fd
 981             int _op = operation
 982             uint64_t _owner = owner
 983
 984         with nogil:
 985             ret = ceph_flock(self.cluster, _fd, _op, _owner)
 986         if ret < 0:
 987             raise make_ex(ret, "error in write")
 988         return ret
 989
 990     def getxattr(self, path, name, size=255):
 991         """
 992          Get an extended attribute.
 993
 994          :param path: the path to the file
 995          :param name: the name of the extended attribute to get
 996          :param size: the size of the pre-allocated buffer
 997         """
 998         self.require_state("mounted")
 999
1000         path = cstr(path, 'path')
1001         name = cstr(name, 'name')
1002
1003         cdef:
1004             char* _path = path
1005             char* _name = name
1006
1007             size_t ret_length = size
1008             char *ret_buf = NULL
1009
1010         try:
1011             ret_buf = <char *>realloc_chk(ret_buf, ret_length)
1012             with nogil:
1013                 ret = ceph_getxattr(self.cluster, _path, _name, ret_buf,
1014                                     ret_length)
1015
1016             if ret < 0:
1017                 raise make_ex(ret, "error in getxattr")
1018
1019             return ret_buf[:ret]
1020         finally:
1021             free(ret_buf)
1022
1023     def setxattr(self, path, name, value, flags):
1024         """
1025         Set an extended attribute on a file.
1026
1027        :param path: the path to the file.
1028        :param name: the name of the extended attribute to set.
1029        :param value: the bytes of the extended attribute value
1030        """
1031         self.require_state("mounted")
1032
1033         name = cstr(name, 'name')
1034         path = cstr(path, 'path')
1035         if not isinstance(flags, int):
1036             raise TypeError('flags must be a int')
1037         if not isinstance(value, bytes):
1038             raise TypeError('value must be a bytes')
1039
1040         cdef:
1041             char *_path = path
1042             char *_name = name
1043             char *_value = value
1044             size_t _value_len = len(value)
1045             int _flags = flags
1046
1047         with nogil:
1048             ret = ceph_setxattr(self.cluster, _path, _name,
1049                                 _value, _value_len, _flags)
1050         if ret < 0:
1051             raise make_ex(ret, "error in setxattr")
1052
1053
1054     def stat(self, path):
1055         """
1056         Get a file's extended statistics and attributes.
1057
1058         :param path: the file or directory to get the statistics of.
1059         """
1060         self.require_state("mounted")
1061         path = cstr(path, 'path')
1062
1063         cdef:
1064             char* _path = path
1065             statx stx
1066
1067         with nogil:
1068             # FIXME: replace magic number with CEPH_STATX_BASIC_STATS
1069             ret = ceph_statx(self.cluster, _path, &stx, 0x7ffu, 0)
1070         if ret < 0:
1071             raise make_ex(ret, "error in stat: %s" % path)
1072         return StatResult(st_dev=stx.stx_dev, st_ino=stx.stx_ino,
1073                           st_mode=stx.stx_mode, st_nlink=stx.stx_nlink,
1074                           st_uid=stx.stx_uid, st_gid=stx.stx_gid,
1075                           st_rdev=stx.stx_rdev, st_size=stx.stx_size,
1076                           st_blksize=stx.stx_blksize,
1077                           st_blocks=stx.stx_blocks,
1078                           st_atime=datetime.fromtimestamp(stx.stx_atime.tv_sec),
1079                           st_mtime=datetime.fromtimestamp(stx.stx_mtime.tv_sec),
1080                           st_ctime=datetime.fromtimestamp(stx.stx_ctime.tv_sec))
1081
1082     def fstat(self, fd):
1083         """
1084         Get an open file's extended statistics and attributes.
1085
1086         :param fd: the file descriptor of the file to get statistics of.
1087          """
1088         self.require_state("mounted")
1089         if not isinstance(fd, int):
1090             raise TypeError('fd must be an int')
1091
1092         cdef:
1093             int _fd = fd
1094             statx stx
1095
1096         with nogil:
1097             # FIXME: replace magic number with CEPH_STATX_BASIC_STATS
1098             ret = ceph_fstatx(self.cluster, _fd, &stx, 0x7ffu, 0)
1099         if ret < 0:
1100             raise make_ex(ret, "error in fsat")
1101         return StatResult(st_dev=stx.stx_dev, st_ino=stx.stx_ino,
1102                           st_mode=stx.stx_mode, st_nlink=stx.stx_nlink,
1103                           st_uid=stx.stx_uid, st_gid=stx.stx_gid,
1104                           st_rdev=stx.stx_rdev, st_size=stx.stx_size,
1105                           st_blksize=stx.stx_blksize,
1106                           st_blocks=stx.stx_blocks,
1107                           st_atime=datetime.fromtimestamp(stx.stx_atime.tv_sec),
1108                           st_mtime=datetime.fromtimestamp(stx.stx_mtime.tv_sec),
1109                           st_ctime=datetime.fromtimestamp(stx.stx_ctime.tv_sec))
1110
1111     def symlink(self, existing, newname):
1112         """
1113         Creates a symbolic link.
1114
1115         :param existing: the path to the existing file/directory to link to.
1116         :param newname: the path to the new file/directory to link from.
1117         """
1118         self.require_state("mounted")
1119         existing = cstr(existing, 'existing')
1120         newname = cstr(newname, 'newname')
1121         cdef:
1122             char* _existing = existing
1123             char* _newname = newname
1124
1125         with nogil:
1126             ret = ceph_symlink(self.cluster, _existing, _newname)
1127         if ret < 0:
1128             raise make_ex(ret, "error in symlink")
1129
1130     def link(self, existing, newname):
1131         """
1132         Create a link.
1133
1134         :param existing: the path to the existing file/directory to link to.
1135         :param newname: the path to the new file/directory to link from.
1136         """
1137
1138         self.require_state("mounted")
1139         existing = cstr(existing, 'existing')
1140         newname = cstr(newname, 'newname')
1141         cdef:
1142             char* _existing = existing
1143             char* _newname = newname
1144
1145         with nogil:
1146             ret = ceph_link(self.cluster, _existing, _newname)
1147         if ret < 0:
1148             raise make_ex(ret, "error in link")
1149
1150     def readlink(self, path, size):
1151         """
1152         Read a symbolic link.
1153
1154         :param path: the path to the symlink to read
1155         :param size: the length of the buffer
1156         :rtype buf: buffer to hold the path of the file that the symlink points to.
1157         """
1158         self.require_state("mounted")
1159         path = cstr(path, 'path')
1160
1161         cdef:
1162             char* _path = path
1163             int64_t _size = size
1164             char *buf = NULL
1165
1166         try:
1167             buf = <char *>realloc_chk(buf, _size)
1168             with nogil:
1169                 ret = ceph_readlink(self.cluster, _path, buf, _size)
1170             if ret < 0:
1171                 raise make_ex(ret, "error in readlink")
1172             return buf
1173         finally:
1174             free(buf)
1175
1176     def unlink(self, path):
1177         """
1178         Removes a file, link, or symbolic link.  If the file/link has multiple links to it, the
1179         file will not disappear from the namespace until all references to it are removed.
1180
1181         :param path: the path of the file or link to unlink.
1182         """
1183         self.require_state("mounted")
1184         path = cstr(path, 'path')
1185         cdef char* _path = path
1186         with nogil:
1187             ret = ceph_unlink(self.cluster, _path)
1188         if ret < 0:
1189             raise make_ex(ret, "error in unlink: %s" % path)
1190
1191     def rename(self, src, dst):
1192         """
1193         Rename a file or directory.
1194
1195         :param src: the path to the existing file or directory.
1196         :param dst: the new name of the file or directory.
1197         """
1198
1199         self.require_state("mounted")
1200
1201         src = cstr(src, 'source')
1202         dst = cstr(dst, 'destination')
1203
1204         cdef:
1205             char* _src = src
1206             char* _dst = dst
1207
1208         with nogil:
1209             ret = ceph_rename(self.cluster, _src, _dst)
1210         if ret < 0:
1211             raise make_ex(ret, "error in rename '%s' to '%s'" % (src, dst))
1212
1213     def mds_command(self, mds_spec, args, input_data):
1214         """
1215         :return 3-tuple of output status int, output status string, output data
1216         """
1217         mds_spec = cstr(mds_spec, 'mds_spec')
1218         args = cstr_list(args, 'args')
1219         input_data = cstr(input_data, 'input_data')
1220
1221         cdef:
1222             char *_mds_spec = opt_str(mds_spec)
1223             char **_cmd = to_bytes_array(args)
1224             size_t _cmdlen = len(args)
1225
1226             char *_inbuf = input_data
1227             size_t _inbuf_len = len(input_data)
1228
1229             char *_outbuf = NULL
1230             size_t _outbuf_len = 0
1231             char *_outs = NULL
1232             size_t _outs_len = 0
1233
1234         try:
1235             with nogil:
1236                 ret = ceph_mds_command(self.cluster, _mds_spec,
1237                                        <const char **>_cmd, _cmdlen,
1238                                        <const char*>_inbuf, _inbuf_len,
1239                                        &_outbuf, &_outbuf_len,
1240                                        &_outs, &_outs_len)
1241             my_outs = decode_cstr(_outs[:_outs_len])
1242             my_outbuf = _outbuf[:_outbuf_len]
1243             if _outs_len:
1244                 ceph_buffer_free(_outs)
1245             if _outbuf_len:
1246                 ceph_buffer_free(_outbuf)
1247             return (ret, my_outbuf, my_outs)
1248         finally:
1249             free(_cmd)
1250
1251     def umask(self, mode) :
1252         self.require_state("mounted")
1253         cdef:
1254             mode_t _mode = mode
1255         with nogil:
1256             ret = ceph_umask(self.cluster, _mode)
1257         if ret < 0:
1258             raise make_ex(ret, "error in umask")
1259         return ret