Skip to content

W
wendelin.core
Commit d75d23bd authored by Kirill Smelkov's avatar Kirill Smelkov
Browse files
Options

.

parent 6f19c05d
Hide whitespace changes
Inline Side-by-side
Showing with 0 additions and 553 deletions
wcfs/internal/io.pyx deleted 100644 → 0
View file @ 6f19c05d
# Copyright (C) 2019-2020 Nexedi SA and Contributors.
# Kirill Smelkov <kirr@nexedi.com>
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
# cython: language_level=2
"""Package io complements IO facility provided by Python."""
from posix.unistd cimport pread
from cpython.exc cimport PyErr_SetFromErrno
# readat calls pread to read from fd@off into buf.
def readat(int fd, size_t off, unsigned char[::1] buf not None): # -> n
cdef void *dest = &buf[0]
cdef size_t size = buf.shape[0]
cdef ssize_t n
with nogil:
n = pread(fd, dest, size, off)
if n < 0:
PyErr_SetFromErrno(OSError)
return n
wcfs/internal/mm.pyx deleted 100644 → 0
View file @ 6f19c05d
# Copyright (C) 2019-2020 Nexedi SA and Contributors.
# Kirill Smelkov <kirr@nexedi.com>
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
# cython: language_level=2
"""Package mm provides access to OS memory management interfaces like mlock and mincore."""
from posix cimport mman
from cpython.exc cimport PyErr_SetFromErrno
#from libc.stdio cimport printf
# mlock2 is provided starting from glibc 2.27
cdef extern from *:
"""
#if defined(__GLIBC__)
#if !__GLIBC_PREREQ(2, 27)
#include <unistd.h>
#include <sys/syscall.h>
static int mlock2(const void *addr, size_t len, int flags) {
long err = syscall(SYS_mlock2, addr, len, flags);
if (err != 0) {
errno = -err;
return -1;
}
return 0;
}
#endif
#endif
#ifndef MLOCK_ONFAULT
# define MLOCK_ONFAULT 1
#endif
"""
pass
cdef extern from "<sys/user.h>":
cpdef enum:
PAGE_SIZE
cpdef enum:
PROT_EXEC = mman.PROT_EXEC
PROT_READ = mman.PROT_READ
PROT_WRITE = mman.PROT_WRITE
PROT_NONE = mman.PROT_NONE
MLOCK_ONFAULT = mman.MLOCK_ONFAULT
MCL_CURRENT = mman.MCL_CURRENT
MCL_FUTURE = mman.MCL_FUTURE
MCL_ONFAULT = mman.MCL_ONFAULT
MADV_NORMAL = mman.MADV_NORMAL
MADV_RANDOM = mman.MADV_RANDOM
MADV_SEQUENTIAL = mman.MADV_SEQUENTIAL
MADV_WILLNEED = mman.MADV_WILLNEED
MADV_DONTNEED = mman.MADV_DONTNEED
#MADV_FREE = mman.MADV_FREE
MADV_REMOVE = mman.MADV_REMOVE
MS_ASYNC = mman.MS_ASYNC
MS_SYNC = mman.MS_SYNC
MS_INVALIDATE = mman.MS_INVALIDATE
# incore returns bytearray vector indicating whether page of mem is in core or not.
#
# mem start must be page-aligned.
def incore(const unsigned char[::1] mem not None) -> bytearray:
cdef size_t size = mem.shape[0]
if size == 0:
return bytearray()
cdef const void *addr = &mem[0]
# size in pages; rounded up
cdef size_t pgsize = (size + (PAGE_SIZE-1)) // PAGE_SIZE
#printf("\n\n%p %ld\n", addr, size)
incore = bytearray(pgsize)
cdef unsigned char[::1] incorev = incore
cdef err = mman.mincore(<void *>addr, size, &incorev[0])
if err:
PyErr_SetFromErrno(OSError)
return incore
# lock locks mem pages to be resident in RAM.
#
# see mlock2(2) for description of flags.
def lock(const unsigned char[::1] mem not None, int flags):
cdef const void *addr = &mem[0]
cdef size_t size = mem.shape[0]
cdef err = mman.mlock2(addr, size, flags)
if err:
PyErr_SetFromErrno(OSError)
return
# unlock unlocks mem pages from being pinned in RAM.
def unlock(const unsigned char[::1] mem not None):
cdef const void *addr = &mem[0]
cdef size_t size = mem.shape[0]
cdef err = mman.munlock(addr, size)
if err:
PyErr_SetFromErrno(OSError)
return
from posix.types cimport off_t
# map_ro memory-maps fd[offset +size) as read-only.
# The mapping is created with MAP_SHARED.
def map_ro(int fd, off_t offset, size_t size):
cdef void *addr
addr = mman.mmap(NULL, size, mman.PROT_READ, mman.MAP_SHARED, fd, offset)
if addr == mman.MAP_FAILED:
PyErr_SetFromErrno(OSError)
return <unsigned char[:size:1]>addr
# map_into_ro is similar to map_ro, but mmaps fd[offset:...] into mem's memory.
def map_into_ro(unsigned char[::1] mem not None, int fd, off_t offset):
cdef void *addr = &mem[0]
cdef size_t size = mem.shape[0]
addr = mman.mmap(addr, size, mman.PROT_READ, mman.MAP_FIXED |
mman.MAP_SHARED, fd, offset)
if addr == mman.MAP_FAILED:
PyErr_SetFromErrno(OSError)
return
# unmap unmaps memory covered by mem.
def unmap(const unsigned char[::1] mem not None):
cdef const void *addr = &mem[0]
cdef size_t size = mem.shape[0]
cdef err = mman.munmap(<void *>addr, size)
if err:
PyErr_SetFromErrno(OSError)
return
# map_zero_ro creats new read-only mmaping that all reads as zero.
# created mapping, even after it is accessed, does not consume memory.
def map_zero_ro(size_t size):
cdef void *addr
# mmap /dev/zero with MAP_NORESERVE and MAP_SHARED
# this way the mapping will be able to be read, but no memory will be allocated to keep it.
f = open("/dev/zero", "rb")
addr = mman.mmap(NULL, size, mman.PROT_READ, mman.MAP_SHARED | mman.MAP_NORESERVE, f.fileno(), 0)
f.close()
if addr == mman.MAP_FAILED:
PyErr_SetFromErrno(OSError)
return
return <unsigned char[:size:1]>addr
# map_zero_into_ro is similar to map_zero_ro, but mmaps zeros into mem's memory.
def map_zero_into_ro(unsigned char[::1] mem not None):
cdef void *addr = &mem[0]
cdef size_t size = mem.shape[0]
f = open("/dev/zero", "rb")
addr = mman.mmap(addr, size, mman.PROT_READ, mman.MAP_FIXED |
mman.MAP_SHARED | mman.MAP_NORESERVE, f.fileno(), 0)
f.close()
if addr == mman.MAP_FAILED:
PyErr_SetFromErrno(OSError)
return
return
# advise advises kernel about use of mem's memory.
#
# see madvise(2) for details.
def advise(const unsigned char[::1] mem not None, int advice):
cdef const void *addr = &mem[0]
cdef size_t size = mem.shape[0]
cdef err = mman.madvise(<void *>addr, size, advice)
if err:
PyErr_SetFromErrno(OSError)
return
# protect sets protection on a region of memory.
#
# see mprotect(2) for details.
def protect(const unsigned char[::1] mem not None, int prot):
cdef const void *addr = &mem[0]
cdef size_t size = mem.shape[0]
cdef err = mman.mprotect(<void *>addr, size, prot)
if err:
PyErr_SetFromErrno(OSError)
return
# sync asks the kernel to synchronize the file with a memory map.
#
# see msync(2) for details.
def sync(const unsigned char[::1] mem not None, int flags):
cdef const void *addr = &mem[0]
cdef size_t size = mem.shape[0]
cdef err = mman.msync(<void *>addr, size, flags)
if err:
PyErr_SetFromErrno(OSError)
return
wcfs/internal/wcfs_test.pyx deleted 100644 → 0
View file @ 6f19c05d
# Copyright (C) 2019-2020 Nexedi SA and Contributors.
# Kirill Smelkov <kirr@nexedi.com>
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
# cython: language_level=2
# distutils: language=c++
"""Module wcfs_test.pyx complements wcfs_test.py with things that cannot be
implemented in Python."""
from posix.signal cimport sigaction, sigaction_t, siginfo_t, SA_SIGINFO, sigemptyset
from libc.signal cimport SIGBUS, SIGSEGV
from libc.setjmp cimport sigjmp_buf, sigsetjmp, siglongjmp
from libc.stdlib cimport abort
from libc.string cimport strlen
from posix.unistd cimport write, sleep
from posix.types cimport off_t
from cpython.exc cimport PyErr_SetFromErrno
from golang cimport chan, pychan, select, panic, topyexc, cbool
from golang cimport sync, time
# _tDB is pyx part of tDB.
cdef class _tDB:
cdef readonly pychan _closed # chan[structZ]
cdef readonly pychan _wcfuseaborted # chan[structZ]
def __cinit__(_tDB t):
t._closed = pychan(dtype='C.structZ')
t._wcfuseaborted = pychan(dtype='C.structZ')
# _abort_ontimeout sends abort to fuse control file if timeout happens
# before tDB is closed.
#
# It runs without GIL to avoid deadlock: e.g. if a code that is
# holding GIL will access wcfs-mmapped memory, and wcfs will send pin,
# but pin handler is failing one way or another - select will wake-up
# but, if _abort_ontimeout uses GIL, won't continue to run trying to lock
# GIL -> deadlock.
def _abort_ontimeout(_tDB t, double dt, pychan nogilready not None):
cdef chan[double] timeoutch = time.after(dt)
cdef int fdabort = t._wcfuseabort.fileno()
emsg1 = "\nC: test timed out after %.1fs\n" % (dt / time.second)
cdef char *_emsg1 = emsg1
with nogil:
# tell main thread that we entered nogil world
nogilready.chan_structZ().close()
t.__abort_ontimeout(dt, timeoutch, fdabort, _emsg1)
cdef void __abort_ontimeout(_tDB t, double dt, chan[double] timeoutch,
int fdabort, const char *emsg1) nogil except +topyexc:
_ = select([
timeoutch.recvs(), # 0
t._closed.chan_structZ().recvs(), # 1
])
if _ == 1:
return # tDB closed = testcase completed
# timeout -> force-umount wcfs
writeerr(emsg1)
writeerr("-> aborting wcfs fuse connection to unblock ...\n\n")
xwrite(fdabort, b"1\n")
t._wcfuseaborted.chan_structZ().close()
# read_nogil reads mem with GIL released and returns its content.
def read_nogil(const unsigned char[::1] mem not None) -> bytes:
assert len(mem) == 1, "read_nogil: only [1] mem is supported for now"
cdef unsigned char b
with nogil:
b = mem[0]
return bytes(bytearray([b]))
# read_mustfault verifies that read-access to mem causes SIGSEGV.
cdef sync.Mutex mustfaultMu # one at a time as sigaction is per-process
cdef sigjmp_buf mustfaultJmp
cdef cbool faultExpected = False
cdef cbool faultedOk = False
cdef extern from * nogil:
"""
volatile unsigned char mustfaultG; // global var for compiler not to optimize-out p[0] access
"""
unsigned char mustfaultG
cdef void mustfaultSighand(int sig) nogil:
global faultedOk
if not faultExpected:
panic("unexpected fault")
# just return from sighandler to proper place
faultedOk = True
siglongjmp(mustfaultJmp, 1)
cdef void _read_mustfault(const unsigned char *p) nogil except +topyexc:
global faultExpected, faultedOk, mustfaultG
cdef sigaction_t act, saveact
act.sa_handler = mustfaultSighand
act.sa_flags = 0
err = sigemptyset(&act.sa_mask)
if err != 0:
panic("sigemptyset: failed")
err = sigaction(SIGSEGV, &act, &saveact)
if err != 0:
panic("sigaction SIGSEGV -> mustfaultSighand: failed")
faultExpected = True
faultedOk = False
if sigsetjmp(mustfaultJmp, 1) == 0:
mustfaultG = p[0] # should pagefault -> sighandler does longjmp
panic("not faulted")
else:
# faulted
if not faultedOk:
panic("faulted, but !faultedOk")
faultExpected = False
err = sigaction(SIGSEGV, &saveact, NULL)
if err != 0:
panic("sigaction SIGSEGV <- restore: failed")
def read_mustfault(const unsigned char[::1] mem not None):
assert len(mem) == 1, "read_mustfault: only [1] mem is supported for now"
# somewhat dup of MUST_FAULT in test_virtmem.c
with nogil:
mustfaultMu.lock()
try:
with nogil:
_read_mustfault(&mem[0])
finally:
with nogil:
mustfaultMu.unlock()
# --------
cdef extern from "<fcntl.h>" nogil:
int posix_fadvise(int fd, off_t offset, off_t len, int advice);
enum: POSIX_FADV_DONTNEED
# fadvise_dontneed teels the kernel that file<fd>[offset +len) is not needed.
#
# see fadvise(2) for details.
def fadvise_dontneed(int fd, off_t offset, off_t len):
cdef int err = posix_fadvise(fd, offset, len, POSIX_FADV_DONTNEED)
if err:
PyErr_SetFromErrno(OSError)
# ---- signal handling ----
# TODO -> golang.signal ?
# install_sigbus_trap installs SIGBUS handler that prints python-level
# traceback before aborting.
#
# Such handler is useful, because when wcfs.go bugs/panics while handling file
# access from wcfs.py, wcfs.py receives SIGBUS signal and by default aborts.
def install_sigbus_trap():
cdef sigaction_t act
act.sa_sigaction = on_sigbus
act.sa_flags = SA_SIGINFO
cdef int err = sigaction(SIGBUS, &act, NULL)
if err:
PyErr_SetFromErrno(OSError)
cdef void on_sigbus(int sig, siginfo_t *si, void *_uc) nogil:
# - wait a bit to give time for other threads to complete their exception dumps
# (e.g. getting "Transport endpoint is not connected" after wcfs.go dying)
# - dump py-level traceback and abort.
# TODO turn SIGBUS into python-level exception? (see sigpanic in Go how to do).
writeerr("\nC: SIGBUS received; giving time to other threads "
"to dump their exceptions (if any) ...\n")
with gil:
pass
sleep(1)
writeerr("\nC: SIGBUS'ed thread traceback:\n")
with gil:
import traceback
traceback.print_stack()
writeerr("-> SIGBUS\n");
# FIXME nothing is printed if pytest stdout/stderr capture is on (no -s given)
abort()
# writeerr writes msg to stderr without depending on stdio buffering and locking.
cdef void writeerr(const char *msg) nogil:
xwrite(2, msg)
# xwrite writes msg to fd without depending on stdio buffering and locking.
cdef void xwrite(int fd, const char *msg) nogil:
cdef ssize_t n, left = strlen(msg)
while left > 0:
n = write(fd, msg, left)
if n == -1:
panic("write: failed")
left -= n
msg += n
wcfs/testprog/wcfs_readcancel.py deleted 100755 → 0
View file @ 6f19c05d
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright (C) 2019-2020 Nexedi SA and Contributors.
# Kirill Smelkov <kirr@nexedi.com>
#
# This program is free software: you can Use, Study, Modify and Redistribute
# it under the terms of the GNU General Public License version 3, or (at your
# option) any later version, as published by the Free Software Foundation.
#
# You can also Link and Combine this program with other software covered by
# the terms of any of the Free Software licenses or any of the Open Source
# Initiative approved licenses and Convey the resulting work. Corresponding
# source of such a combination shall include the source code for all other
# software used.
#
# This program is distributed WITHOUT ANY WARRANTY; without even the implied
# warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
# See COPYING file for full licensing terms.
# See https://www.nexedi.com/licensing for rationale and options.
"""Program wcfs_readcancel is helper for wcfs_test to verify that
sysread(/head/watch) is unblocked and canceled when kernel asks WCFS to cancel
that read request.
Without proper FUSE INTERRUPT handling on WCFS side, such reads are not
cancelled, which results in processes that were aborted or even `kill-9`ed being
stuck forever waiting for WCFS to release them.
"""
from __future__ import print_function, absolute_import
from golang import select, default
from golang import context, sync, time
import os, sys
def main():
wcfs_root = sys.argv[1]
f = open("%s/head/watch" % wcfs_root)
wg = sync.WorkGroup(context.background())
def _(ctx):
data = f.read() # should block forever
raise AssertionError("read: woken up: data=%r" % data)
wg.go(_)
def _(ctx):
time.sleep(100*time.millisecond)
_, _rx = select(
default, # 0
ctx.done().recv, # 1
)
if _ == 1:
raise ctx.err()
os._exit(0)
wg.go(_)
wg.wait()
raise AssertionError("should be unreachable")
if __name__ == '__main__':
main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7