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_unmount(ceph_mount_info *cmount)
 114     int ceph_fstatx(ceph_mount_info *cmount, int fd, statx *stx, unsigned want, unsigned flags)
 115     int ceph_statx(ceph_mount_info *cmount, const char *path, statx *stx, unsigned want, unsigned flags)
 116     int ceph_statfs(ceph_mount_info *cmount, const char *path, statvfs *stbuf)
 117
 118     int ceph_mds_command(ceph_mount_info *cmount, const char *mds_spec, const char **cmd, size_t cmdlen,
 119                          const char *inbuf, size_t inbuflen, char **outbuf, size_t *outbuflen,
 120                          char **outs, size_t *outslen)
 121     int ceph_rename(ceph_mount_info *cmount, const char *from_, const char *to)
 122     int ceph_link(ceph_mount_info *cmount, const char *existing, const char *newname)
 123     int ceph_unlink(ceph_mount_info *cmount, const char *path)
 124     int ceph_symlink(ceph_mount_info *cmount, const char *existing, const char *newname)
 125     int ceph_readlink(ceph_mount_info *cmount, const char *path, char *buf, int64_t size)
 126     int ceph_setxattr(ceph_mount_info *cmount, const char *path, const char *name,
 127                       const void *value, size_t size, int flags)
 128     int ceph_getxattr(ceph_mount_info *cmount, const char *path, const char *name,
 129                       void *value, size_t size)
 130     int ceph_write(ceph_mount_info *cmount, int fd, const char *buf, int64_t size, int64_t offset)
 131     int ceph_read(ceph_mount_info *cmount, int fd, char *buf, int64_t size, int64_t offset)
 132     int ceph_flock(ceph_mount_info *cmount, int fd, int operation, uint64_t owner)
 133     int ceph_close(ceph_mount_info *cmount, int fd)
 134     int ceph_open(ceph_mount_info *cmount, const char *path, int flags, mode_t mode)
 135     int ceph_mkdir(ceph_mount_info *cmount, const char *path, mode_t mode)
 136     int ceph_mkdirs(ceph_mount_info *cmount, const char *path, mode_t mode)
 137     int ceph_closedir(ceph_mount_info *cmount, ceph_dir_result *dirp)
 138     int ceph_opendir(ceph_mount_info *cmount, const char *name, ceph_dir_result **dirpp)
 139     int ceph_chdir(ceph_mount_info *cmount, const char *path)
 140     dirent * ceph_readdir(ceph_mount_info *cmount, ceph_dir_result *dirp)
 141     int ceph_rmdir(ceph_mount_info *cmount, const char *path)
 142     const char* ceph_getcwd(ceph_mount_info *cmount)
 143     int ceph_sync_fs(ceph_mount_info *cmount)
 144     int ceph_fsync(ceph_mount_info *cmount, int fd, int syncdataonly)
 145     int ceph_conf_parse_argv(ceph_mount_info *cmount, int argc, const char **argv)
 146     void ceph_buffer_free(char *buf)
 147
 148
 149
 150 class Error(Exception):
 151     pass
 152
 153
 154 class OSError(Error):
 155     def __init__(self, errno, strerror):
 156         self.errno = errno
 157         self.strerror = strerror
 158
 159     def __str__(self):
 160         return '[Errno {0}] {1}'.format(self.errno, self.strerror)
 161
 162
 163 class PermissionError(OSError):
 164     pass
 165
 166
 167 class ObjectNotFound(OSError):
 168     pass
 169
 170
 171 class NoData(OSError):
 172     pass
 173
 174
 175 class ObjectExists(OSError):
 176     pass
 177
 178
 179 class IOError(OSError):
 180     pass
 181
 182
 183 class NoSpace(OSError):
 184     pass
 185
 186
 187 class InvalidValue(OSError):
 188     pass
 189
 190
 191 class OperationNotSupported(OSError):
 192     pass
 193
 194
 195 class LibCephFSStateError(Error):
 196     pass
 197
 198
 199 class WouldBlock(OSError):
 200     pass
 201
 202
 203 class OutOfRange(OSError):
 204     pass
 205
 206
 207 IF UNAME_SYSNAME == "FreeBSD":
 208     cdef errno_to_exception =  {
 209         errno.EPERM      : PermissionError,
 210         errno.ENOENT     : ObjectNotFound,
 211         errno.EIO        : IOError,
 212         errno.ENOSPC     : NoSpace,
 213         errno.EEXIST     : ObjectExists,
 214         errno.ENOATTR    : NoData,
 215         errno.EINVAL     : InvalidValue,
 216         errno.EOPNOTSUPP : OperationNotSupported,
 217         errno.ERANGE     : OutOfRange,
 218         errno.EWOULDBLOCK: WouldBlock,
 219     }
 220 ELSE:
 221     cdef errno_to_exception =  {
 222         errno.EPERM      : PermissionError,
 223         errno.ENOENT     : ObjectNotFound,
 224         errno.EIO        : IOError,
 225         errno.ENOSPC     : NoSpace,
 226         errno.EEXIST     : ObjectExists,
 227         errno.ENODATA    : NoData,
 228         errno.EINVAL     : InvalidValue,
 229         errno.EOPNOTSUPP : OperationNotSupported,
 230         errno.ERANGE     : OutOfRange,
 231         errno.EWOULDBLOCK: WouldBlock,
 232     }
 233
 234
 235 cdef make_ex(ret, msg):
 236     """
 237     Translate a librados return code into an exception.
 238
 239     :param ret: the return code
 240     :type ret: int
 241     :param msg: the error message to use
 242     :type msg: str
 243     :returns: a subclass of :class:`Error`
 244     """
 245     ret = abs(ret)
 246     if ret in errno_to_exception:
 247         return errno_to_exception[ret](ret, msg)
 248     else:
 249         return Error(ret, msg + (": error code %d" % ret))
 250
 251
 252 class DirEntry(namedtuple('DirEntry',
 253                ['d_ino', 'd_off', 'd_reclen', 'd_type', 'd_name'])):
 254     DT_DIR = 0x4
 255     DT_REG = 0xA
 256     DT_LNK = 0xC
 257     def is_dir(self):
 258         return self.d_type == self.DT_DIR
 259
 260     def is_symbol_file(self):
 261         return self.d_type == self.DT_LNK
 262
 263     def is_file(self):
 264         return self.d_type == self.DT_REG
 265
 266 StatResult = namedtuple('StatResult',
 267                         ["st_dev", "st_ino", "st_mode", "st_nlink", "st_uid",
 268                          "st_gid", "st_rdev", "st_size", "st_blksize",
 269                          "st_blocks", "st_atime", "st_mtime", "st_ctime"])
 270
 271 cdef class DirResult(object):
 272     cdef ceph_dir_result *handler
 273
 274
 275 def cstr(val, name, encoding="utf-8", opt=False):
 276     """
 277     Create a byte string from a Python string
 278
 279     :param basestring val: Python string
 280     :param str name: Name of the string parameter, for exceptions
 281     :param str encoding: Encoding to use
 282     :param bool opt: If True, None is allowed
 283     :rtype: bytes
 284     :raises: :class:`InvalidArgument`
 285     """
 286     if opt and val is None:
 287         return None
 288     if isinstance(val, bytes):
 289         return val
 290     elif isinstance(val, unicode):
 291         return val.encode(encoding)
 292     else:
 293         raise TypeError('%s must be a string' % name)
 294
 295
 296 def cstr_list(list_str, name, encoding="utf-8"):
 297     return [cstr(s, name) for s in list_str]
 298
 299
 300 def decode_cstr(val, encoding="utf-8"):
 301     """
 302     Decode a byte string into a Python string.
 303
 304     :param bytes val: byte string
 305     :rtype: unicode or None
 306     """
 307     if val is None:
 308         return None
 309
 310     return val.decode(encoding)
 311
 312
 313 cdef char* opt_str(s) except? NULL:
 314     if s is None:
 315         return NULL
 316     return s
 317
 318
 319 cdef char ** to_bytes_array(list_bytes):
 320     cdef char **ret = <char **>malloc(len(list_bytes) * sizeof(char *))
 321     if ret == NULL:
 322         raise MemoryError("malloc failed")
 323     for i in xrange(len(list_bytes)):
 324         ret[i] = <char *>list_bytes[i]
 325     return ret
 326
 327
 328 cdef void* realloc_chk(void* ptr, size_t size) except NULL:
 329     cdef void *ret = realloc(ptr, size)
 330     if ret == NULL:
 331         raise MemoryError("realloc failed")
 332     return ret
 333
 334
 335 cdef class LibCephFS(object):
 336     """libcephfs python wrapper"""
 337
 338     cdef public object state
 339     cdef ceph_mount_info *cluster
 340
 341     def require_state(self, *args):
 342         if self.state in args:
 343             return
 344         raise LibCephFSStateError("You cannot perform that operation on a "
 345                                   "CephFS object in state %s." % (self.state))
 346
 347     def __cinit__(self, conf=None, conffile=None, auth_id=None, rados_inst=None):
 348         """Create a libcephfs wrapper
 349
 350         :param conf dict opt: settings overriding the default ones and conffile
 351         :param conffile str opt: the path to ceph.conf to override the default settings
 352         :auth_id str opt: the id used to authenticate the client entity
 353         :rados_inst Rados opt: a rados.Rados instance
 354         """
 355         PyEval_InitThreads()
 356         self.state = "uninitialized"
 357         if rados_inst is not None:
 358             if auth_id is not None or conffile is not None or conf is not None:
 359                 raise make_ex(errno.EINVAL,
 360                               "May not pass RADOS instance as well as other configuration")
 361
 362             self.create_with_rados(rados_inst)
 363         else:
 364             self.create(conf, conffile, auth_id)
 365
 366     def create_with_rados(self, rados.Rados rados_inst):
 367         cdef int ret
 368         with nogil:
 369             ret = ceph_create_from_rados(&self.cluster, rados_inst.cluster)
 370         if ret != 0:
 371             raise Error("libcephfs_initialize failed with error code: %d" % ret)
 372         self.state = "configuring"
 373
 374     def create(self, conf=None, conffile=None, auth_id=None):
 375         if conf is not None and not isinstance(conf, dict):
 376             raise TypeError("conf must be dict or None")
 377         cstr(conffile, 'configfile', opt=True)
 378         auth_id = cstr(auth_id, 'auth_id', opt=True)
 379
 380         cdef:
 381             char* _auth_id = opt_str(auth_id)
 382             int ret
 383
 384         with nogil:
 385             ret = ceph_create(&self.cluster, <const char*>_auth_id)
 386         if ret != 0:
 387             raise Error("libcephfs_initialize failed with error code: %d" % ret)
 388
 389         self.state = "configuring"
 390         if conffile is not None:
 391             # read the default conf file when '' is given
 392             if conffile == '':
 393                 conffile = None
 394             self.conf_read_file(conffile)
 395         if conf is not None:
 396             for key, value in conf.iteritems():
 397                 self.conf_set(key, value)
 398
 399     def conf_read_file(self, conffile=None):
 400         conffile = cstr(conffile, 'conffile', opt=True)
 401         cdef:
 402             char *_conffile = opt_str(conffile)
 403         with nogil:
 404             ret = ceph_conf_read_file(self.cluster, <const char*>_conffile)
 405         if ret != 0:
 406             raise make_ex(ret, "error calling conf_read_file")
 407
 408     def conf_parse_argv(self, argv):
 409         self.require_state("configuring")
 410         cargv = cstr_list(argv, 'argv')
 411         cdef:
 412             int _argc = len(argv)
 413             char **_argv = to_bytes_array(cargv)
 414
 415         try:
 416             with nogil:
 417                 ret = ceph_conf_parse_argv(self.cluster, _argc,
 418                                            <const char **>_argv)
 419             if ret != 0:
 420                 raise make_ex(ret, "error calling conf_parse_argv")
 421         finally:
 422             free(_argv)
 423
 424     def shutdown(self):
 425         """
 426         Unmount and destroy the ceph mount handle.
 427         """
 428         if self.state in ["initialized", "mounted"]:
 429             with nogil:
 430                 ceph_shutdown(self.cluster)
 431             self.state = "shutdown"
 432
 433     def __enter__(self):
 434         self.mount()
 435         return self
 436
 437     def __exit__(self, type_, value, traceback):
 438         self.shutdown()
 439         return False
 440
 441     def __dealloc__(self):
 442         self.shutdown()
 443
 444     def version(self):
 445         """
 446         Get the version number of the ``libcephfs`` C library.
 447
 448         :returns: a tuple of ``(major, minor, extra)`` components of the
 449                   libcephfs version
 450         """
 451         cdef:
 452             int major = 0
 453             int minor = 0
 454             int extra = 0
 455         with nogil:
 456             ceph_version(&major, &minor, &extra)
 457         return (major, minor, extra)
 458
 459     def conf_get(self, option):
 460         self.require_state("configuring", "initialized", "mounted")
 461
 462         option = cstr(option, 'option')
 463         cdef:
 464             char *_option = option
 465             size_t length = 20
 466             char *ret_buf = NULL
 467
 468         try:
 469             while True:
 470                 ret_buf = <char *>realloc_chk(ret_buf, length)
 471                 with nogil:
 472                     ret = ceph_conf_get(self.cluster, _option, ret_buf, length)
 473                 if ret == 0:
 474                     return decode_cstr(ret_buf)
 475                 elif ret == -errno.ENAMETOOLONG:
 476                     length = length * 2
 477                 elif ret == -errno.ENOENT:
 478                     return None
 479                 else:
 480                     raise make_ex(ret, "error calling conf_get")
 481         finally:
 482             free(ret_buf)
 483
 484     def conf_set(self, option, val):
 485         self.require_state("configuring", "initialized", "mounted")
 486
 487         option = cstr(option, 'option')
 488         val = cstr(val, 'val')
 489         cdef:
 490             char *_option = option
 491             char *_val = val
 492
 493         with nogil:
 494             ret = ceph_conf_set(self.cluster, _option, _val)
 495         if ret != 0:
 496             raise make_ex(ret, "error calling conf_set")
 497
 498     def init(self):
 499         self.require_state("configuring")
 500         with nogil:
 501             ret = ceph_init(self.cluster)
 502         if ret != 0:
 503             raise make_ex(ret, "error calling ceph_init")
 504         self.state = "initialized"
 505
 506     def mount(self):
 507         if self.state == "configuring":
 508             self.init()
 509         self.require_state("initialized")
 510         with nogil:
 511             ret = ceph_mount(self.cluster, "/")
 512         if ret != 0:
 513             raise make_ex(ret, "error calling ceph_mount")
 514         self.state = "mounted"
 515
 516     def unmount(self):
 517         self.require_state("mounted")
 518         with nogil:
 519             ret = ceph_unmount(self.cluster)
 520         if ret != 0:
 521             raise make_ex(ret, "error calling ceph_unmount")
 522         self.state = "initialized"
 523
 524     def statfs(self, path):
 525         self.require_state("mounted")
 526         path = cstr(path, 'path')
 527         cdef:
 528             char* _path = path
 529             statvfs statbuf
 530
 531         with nogil:
 532             ret = ceph_statfs(self.cluster, _path, &statbuf)
 533         if ret < 0:
 534             raise make_ex(ret, "statfs failed: %s" % path)
 535         return {'f_bsize': statbuf.f_bsize,
 536                 'f_frsize': statbuf.f_frsize,
 537                 'f_blocks': statbuf.f_blocks,
 538                 'f_bfree': statbuf.f_bfree,
 539                 'f_bavail': statbuf.f_bavail,
 540                 'f_files': statbuf.f_files,
 541                 'f_ffree': statbuf.f_ffree,
 542                 'f_favail': statbuf.f_favail,
 543                 'f_fsid': statbuf.f_fsid,
 544                 'f_flag': statbuf.f_flag,
 545                 'f_namemax': statbuf.f_namemax}
 546
 547     def sync_fs(self):
 548         self.require_state("mounted")
 549         with nogil:
 550             ret = ceph_sync_fs(self.cluster)
 551         if ret < 0:
 552             raise make_ex(ret, "sync_fs failed")
 553
 554     def fsync(self, int fd, int syncdataonly):
 555         self.require_state("mounted")
 556         with nogil:
 557             ret = ceph_fsync(self.cluster, fd, syncdataonly)
 558         if ret < 0:
 559             raise make_ex(ret, "fsync failed")
 560
 561     def getcwd(self):
 562         self.require_state("mounted")
 563         with nogil:
 564             ret = ceph_getcwd(self.cluster)
 565         return ret
 566
 567     def chdir(self, path):
 568         self.require_state("mounted")
 569
 570         path = cstr(path, 'path')
 571         cdef char* _path = path
 572         with nogil:
 573             ret = ceph_chdir(self.cluster, _path)
 574         if ret < 0:
 575             raise make_ex(ret, "chdir failed")
 576
 577     def opendir(self, path):
 578         self.require_state("mounted")
 579
 580         path = cstr(path, 'path')
 581         cdef:
 582             char* _path = path
 583             ceph_dir_result *dir_handler
 584         with nogil:
 585             ret = ceph_opendir(self.cluster, _path, &dir_handler);
 586         if ret < 0:
 587             raise make_ex(ret, "opendir failed")
 588         d = DirResult()
 589         d.handler = dir_handler
 590         return d
 591
 592     def readdir(self, DirResult dir_handler):
 593         self.require_state("mounted")
 594
 595         cdef ceph_dir_result *_dir_handler = dir_handler.handler
 596         with nogil:
 597             dirent = ceph_readdir(self.cluster, _dir_handler)
 598         if not dirent:
 599             return None
 600
 601         d_name = dirent.d_name if sys.version[0:2] == '2.' else dirent.d_name.\
 602                  decode()
 603         return DirEntry(d_ino=dirent.d_ino,
 604                         d_off=dirent.d_off,
 605                         d_reclen=dirent.d_reclen,
 606                         d_type=dirent.d_type,
 607                         d_name=d_name)
 608
 609     def closedir(self, DirResult dir_handler):
 610         self.require_state("mounted")
 611         cdef:
 612             ceph_dir_result *_dir_handler = dir_handler.handler
 613
 614         with nogil:
 615             ret = ceph_closedir(self.cluster, _dir_handler)
 616         if ret < 0:
 617             raise make_ex(ret, "closedir failed")
 618
 619     def mkdir(self, path, mode):
 620         self.require_state("mounted")
 621         path = cstr(path, 'path')
 622         if not isinstance(mode, int):
 623             raise TypeError('mode must be an int')
 624         cdef:
 625             char* _path = path
 626             int _mode = mode
 627         with nogil:
 628             ret = ceph_mkdir(self.cluster, _path, _mode)
 629         if ret < 0:
 630             raise make_ex(ret, "error in mkdir '%s'" % path)
 631
 632     def mkdirs(self, path, mode):
 633         self.require_state("mounted")
 634         path = cstr(path, 'path')
 635         if not isinstance(mode, int):
 636             raise TypeError('mode must be an int')
 637         cdef:
 638             char* _path = path
 639             int _mode = mode
 640
 641         with nogil:
 642             ret = ceph_mkdirs(self.cluster, _path, _mode)
 643         if ret < 0:
 644             raise make_ex(ret, "error in mkdirs '%s'" % path)
 645
 646     def rmdir(self, path):
 647         self.require_state("mounted")
 648         path = cstr(path, 'path')
 649         cdef char* _path = path
 650         ret = ceph_rmdir(self.cluster, _path)
 651         if ret < 0:
 652             raise make_ex(ret, "error in rmdir '%s'" % path)
 653
 654     def open(self, path, flags, mode=0):
 655         self.require_state("mounted")
 656         path = cstr(path, 'path')
 657
 658         if not isinstance(mode, int):
 659             raise TypeError('mode must be an int')
 660         if isinstance(flags, str):
 661             cephfs_flags = 0
 662             if flags == '':
 663                 cephfs_flags = os.O_RDONLY
 664             else:
 665                 access_flags = 0;
 666                 for c in flags:
 667                     if c == 'r':
 668                         access_flags = 1;
 669                     elif c == 'w':
 670                         access_flags = 2;
 671                         cephfs_flags |= os.O_TRUNC | os.O_CREAT
 672                     elif access_flags > 0 and c == '+':
 673                         access_flags = 3;
 674                     else:
 675                         raise make_ex(errno.EOPNOTSUPP,
 676                                       "open flags doesn't support %s" % c)
 677
 678                 if access_flags == 1:
 679                     cephfs_flags |= os.O_RDONLY;
 680                 elif access_flags == 2:
 681                     cephfs_flags |= os.O_WRONLY;
 682                 else:
 683                     cephfs_flags |= os.O_RDWR;
 684
 685         elif isinstance(flags, int):
 686             cephfs_flags = flags
 687         else:
 688             raise TypeError("flags must be a string or an integer")
 689
 690         cdef:
 691             char* _path = path
 692             int _flags = cephfs_flags
 693             int _mode = mode
 694
 695         with nogil:
 696             ret = ceph_open(self.cluster, _path, _flags, _mode)
 697         if ret < 0:
 698             raise make_ex(ret, "error in open '%s'" % path)
 699         return ret
 700
 701     def close(self, fd):
 702         self.require_state("mounted")
 703         if not isinstance(fd, int):
 704             raise TypeError('fd must be an int')
 705         cdef int _fd = fd
 706         with nogil:
 707             ret = ceph_close(self.cluster, _fd)
 708         if ret < 0:
 709             raise make_ex(ret, "error in close")
 710
 711     def read(self, fd, offset, l):
 712         self.require_state("mounted")
 713         if not isinstance(offset, int):
 714             raise TypeError('offset must be an int')
 715         if not isinstance(l, int):
 716             raise TypeError('l must be an int')
 717         if not isinstance(fd, int):
 718             raise TypeError('fd must be an int')
 719         cdef:
 720             int _fd = fd
 721             int64_t _offset = offset
 722             int64_t _length = l
 723
 724             char *ret_buf
 725             PyObject* ret_s = NULL
 726
 727         ret_s = PyBytes_FromStringAndSize(NULL, _length)
 728         try:
 729             ret_buf = PyBytes_AsString(ret_s)
 730             with nogil:
 731                 ret = ceph_read(self.cluster, _fd, ret_buf, _length, _offset)
 732             if ret < 0:
 733                 raise make_ex(ret, "error in read")
 734
 735             if ret != _length:
 736                 _PyBytes_Resize(&ret_s, ret)
 737
 738             return <object>ret_s
 739         finally:
 740             # We DECREF unconditionally: the cast to object above will have
 741             # INCREFed if necessary. This also takes care of exceptions,
 742             # including if _PyString_Resize fails (that will free the string
 743             # itself and set ret_s to NULL, hence XDECREF).
 744             ref.Py_XDECREF(ret_s)
 745
 746     def write(self, fd, buf, offset):
 747         self.require_state("mounted")
 748         if not isinstance(fd, int):
 749             raise TypeError('fd must be an int')
 750         if not isinstance(buf, bytes):
 751             raise TypeError('buf must be a bytes')
 752         if not isinstance(offset, int):
 753             raise TypeError('offset must be an int')
 754
 755         cdef:
 756             int _fd = fd
 757             char *_data = buf
 758             int64_t _offset = offset
 759
 760             size_t length = len(buf)
 761
 762         with nogil:
 763             ret = ceph_write(self.cluster, _fd, _data, length, _offset)
 764         if ret < 0:
 765             raise make_ex(ret, "error in write")
 766         return ret
 767
 768     def flock(self, fd, operation, owner):
 769         self.require_state("mounted")
 770         if not isinstance(fd, int):
 771             raise TypeError('fd must be an int')
 772         if not isinstance(operation, int):
 773             raise TypeError('operation must be an int')
 774
 775         cdef:
 776             int _fd = fd
 777             int _op = operation
 778             uint64_t _owner = owner
 779
 780         with nogil:
 781             ret = ceph_flock(self.cluster, _fd, _op, _owner)
 782         if ret < 0:
 783             raise make_ex(ret, "error in write")
 784         return ret
 785
 786     def getxattr(self, path, name, size=255):
 787         self.require_state("mounted")
 788
 789         path = cstr(path, 'path')
 790         name = cstr(name, 'name')
 791
 792         cdef:
 793             char* _path = path
 794             char* _name = name
 795
 796             size_t ret_length = size
 797             char *ret_buf = NULL
 798
 799         try:
 800             ret_buf = <char *>realloc_chk(ret_buf, ret_length)
 801             with nogil:
 802                 ret = ceph_getxattr(self.cluster, _path, _name, ret_buf,
 803                                     ret_length)
 804
 805             if ret < 0:
 806                 raise make_ex(ret, "error in getxattr")
 807
 808             return ret_buf[:ret]
 809         finally:
 810             free(ret_buf)
 811
 812     def setxattr(self, path, name, value, flags):
 813         self.require_state("mounted")
 814
 815         name = cstr(name, 'name')
 816         path = cstr(path, 'path')
 817         if not isinstance(flags, int):
 818             raise TypeError('flags must be a int')
 819         if not isinstance(value, bytes):
 820             raise TypeError('value must be a bytes')
 821
 822         cdef:
 823             char *_path = path
 824             char *_name = name
 825             char *_value = value
 826             size_t _value_len = len(value)
 827             int _flags = flags
 828
 829         with nogil:
 830             ret = ceph_setxattr(self.cluster, _path, _name,
 831                                 _value, _value_len, _flags)
 832         if ret < 0:
 833             raise make_ex(ret, "error in setxattr")
 834
 835     def stat(self, path):
 836         self.require_state("mounted")
 837         path = cstr(path, 'path')
 838
 839         cdef:
 840             char* _path = path
 841             statx stx
 842
 843         with nogil:
 844             # FIXME: replace magic number with CEPH_STATX_BASIC_STATS
 845             ret = ceph_statx(self.cluster, _path, &stx, 0x7ffu, 0)
 846         if ret < 0:
 847             raise make_ex(ret, "error in stat: %s" % path)
 848         return StatResult(st_dev=stx.stx_dev, st_ino=stx.stx_ino,
 849                           st_mode=stx.stx_mode, st_nlink=stx.stx_nlink,
 850                           st_uid=stx.stx_uid, st_gid=stx.stx_gid,
 851                           st_rdev=stx.stx_rdev, st_size=stx.stx_size,
 852                           st_blksize=stx.stx_blksize,
 853                           st_blocks=stx.stx_blocks,
 854                           st_atime=datetime.fromtimestamp(stx.stx_atime.tv_sec),
 855                           st_mtime=datetime.fromtimestamp(stx.stx_mtime.tv_sec),
 856                           st_ctime=datetime.fromtimestamp(stx.stx_ctime.tv_sec))
 857
 858     def fstat(self, fd):
 859         self.require_state("mounted")
 860         if not isinstance(fd, int):
 861             raise TypeError('fd must be an int')
 862
 863         cdef:
 864             int _fd = fd
 865             statx stx
 866
 867         with nogil:
 868             # FIXME: replace magic number with CEPH_STATX_BASIC_STATS
 869             ret = ceph_fstatx(self.cluster, _fd, &stx, 0x7ffu, 0)
 870         if ret < 0:
 871             raise make_ex(ret, "error in fsat")
 872         return StatResult(st_dev=stx.stx_dev, st_ino=stx.stx_ino,
 873                           st_mode=stx.stx_mode, st_nlink=stx.stx_nlink,
 874                           st_uid=stx.stx_uid, st_gid=stx.stx_gid,
 875                           st_rdev=stx.stx_rdev, st_size=stx.stx_size,
 876                           st_blksize=stx.stx_blksize,
 877                           st_blocks=stx.stx_blocks,
 878                           st_atime=datetime.fromtimestamp(stx.stx_atime.tv_sec),
 879                           st_mtime=datetime.fromtimestamp(stx.stx_mtime.tv_sec),
 880                           st_ctime=datetime.fromtimestamp(stx.stx_ctime.tv_sec))
 881
 882     def symlink(self, existing, newname):
 883         self.require_state("mounted")
 884         existing = cstr(existing, 'existing')
 885         newname = cstr(newname, 'newname')
 886         cdef:
 887             char* _existing = existing
 888             char* _newname = newname
 889
 890         with nogil:
 891             ret = ceph_symlink(self.cluster, _existing, _newname)
 892         if ret < 0:
 893             raise make_ex(ret, "error in symlink")
 894
 895     def link(self, existing, newname):
 896         self.require_state("mounted")
 897         existing = cstr(existing, 'existing')
 898         newname = cstr(newname, 'newname')
 899         cdef:
 900             char* _existing = existing
 901             char* _newname = newname
 902
 903         with nogil:
 904             ret = ceph_link(self.cluster, _existing, _newname)
 905         if ret < 0:
 906             raise make_ex(ret, "error in link")
 907
 908     def readlink(self, path, size):
 909         self.require_state("mounted")
 910         path = cstr(path, 'path')
 911
 912         cdef:
 913             char* _path = path
 914             int64_t _size = size
 915             char *buf = NULL
 916
 917         try:
 918             buf = <char *>realloc_chk(buf, _size)
 919             with nogil:
 920                 ret = ceph_readlink(self.cluster, _path, buf, _size)
 921             if ret < 0:
 922                 raise make_ex(ret, "error in readlink")
 923             return buf
 924         finally:
 925             free(buf)
 926
 927     def unlink(self, path):
 928         self.require_state("mounted")
 929         path = cstr(path, 'path')
 930         cdef char* _path = path
 931         with nogil:
 932             ret = ceph_unlink(self.cluster, _path)
 933         if ret < 0:
 934             raise make_ex(ret, "error in unlink: %s" % path)
 935
 936     def rename(self, src, dst):
 937         self.require_state("mounted")
 938
 939         src = cstr(src, 'source')
 940         dst = cstr(dst, 'destination')
 941
 942         cdef:
 943             char* _src = src
 944             char* _dst = dst
 945
 946         with nogil:
 947             ret = ceph_rename(self.cluster, _src, _dst)
 948         if ret < 0:
 949             raise make_ex(ret, "error in rename '%s' to '%s'" % (src, dst))
 950
 951     def mds_command(self, mds_spec, args, input_data):
 952         """
 953         :return 3-tuple of output status int, output status string, output data
 954         """
 955         mds_spec = cstr(mds_spec, 'mds_spec')
 956         args = cstr_list(args, 'args')
 957         input_data = cstr(input_data, 'input_data')
 958
 959         cdef:
 960             char *_mds_spec = opt_str(mds_spec)
 961             char **_cmd = to_bytes_array(args)
 962             size_t _cmdlen = len(args)
 963
 964             char *_inbuf = input_data
 965             size_t _inbuf_len = len(input_data)
 966
 967             char *_outbuf = NULL
 968             size_t _outbuf_len = 0
 969             char *_outs = NULL
 970             size_t _outs_len = 0
 971
 972         try:
 973             with nogil:
 974                 ret = ceph_mds_command(self.cluster, _mds_spec,
 975                                        <const char **>_cmd, _cmdlen,
 976                                        <const char*>_inbuf, _inbuf_len,
 977                                        &_outbuf, &_outbuf_len,
 978                                        &_outs, &_outs_len)
 979             my_outs = decode_cstr(_outs[:_outs_len])
 980             my_outbuf = _outbuf[:_outbuf_len]
 981             if _outs_len:
 982                 ceph_buffer_free(_outs)
 983             if _outbuf_len:
 984                 ceph_buffer_free(_outbuf)
 985             return (ret, my_outbuf, my_outs)
 986         finally:
 987             free(_cmd)