import errno
import os
import sys
import time

from rpython.rlib import jit, rgc, rthread
from rpython.rlib.rarithmetic import intmask, r_uint
from rpython.rtyper.lltypesystem import rffi, lltype
from rpython.rtyper.tool import rffi_platform as platform
from rpython.translator.tool.cbuild import ExternalCompilationInfo

from pypy.interpreter.baseobjspace import W_Root
from pypy.interpreter.error import oefmt, wrap_oserror
from pypy.interpreter.gateway import interp2app, unwrap_spec
from pypy.interpreter.typedef import GetSetProperty, TypeDef
from pypy.module._multiprocessing.interp_connection import w_handle

RECURSIVE_MUTEX, SEMAPHORE = range(2)

if sys.platform == 'win32':
    from rpython.rlib import rwin32
    from pypy.module._multiprocessing.interp_win32 import (
        _GetTickCount, handle_w)

    SEM_VALUE_MAX = int(2**31-1) # max rffi.LONG

    _CreateSemaphore = rwin32.winexternal(
        'CreateSemaphoreA', [rffi.VOIDP, rffi.LONG, rffi.LONG, rwin32.LPCSTR],
        rwin32.HANDLE,
        save_err=rffi.RFFI_FULL_LASTERROR)
    _CloseHandle_no_errno = rwin32.winexternal('CloseHandle', [rwin32.HANDLE],
        rwin32.BOOL, releasegil=False)
    _ReleaseSemaphore = rwin32.winexternal(
        'ReleaseSemaphore', [rwin32.HANDLE, rffi.LONG, rffi.LONGP],
        rwin32.BOOL,
        save_err=rffi.RFFI_SAVE_LASTERROR, releasegil=False)

else:
    from rpython.rlib import rposix

    if sys.platform == 'darwin':
        libraries = []
    else:
        libraries = ['rt']

    eci = ExternalCompilationInfo(
        includes = ['sys/time.h',
                    'limits.h',
                    'semaphore.h',
                    ],
        libraries = libraries,
        )

    class CConfig:
        _compilation_info_ = eci
        TIMEVAL = platform.Struct('struct timeval', [('tv_sec', rffi.LONG),
                                                     ('tv_usec', rffi.LONG)])
        TIMESPEC = platform.Struct('struct timespec', [('tv_sec', rffi.TIME_T),
                                                       ('tv_nsec', rffi.LONG)])
        TIMEZONE = platform.Struct('struct timezone', [
                                            ('tz_minuteswest', rffi.INT_real),
                                            ('tz_dsttime', rffi.INT_real),
                                  ])
        SEM_FAILED = platform.ConstantInteger('SEM_FAILED')
        SEM_VALUE_MAX = platform.DefinedConstantInteger('SEM_VALUE_MAX')
        SEM_TIMED_WAIT = platform.Has('sem_timedwait')
        SEM_T_SIZE = platform.SizeOf('sem_t')

    config = platform.configure(CConfig)
    TIMEVAL        = config['TIMEVAL']
    TIMESPEC       = config['TIMESPEC']
    TIMEZONE       = config['TIMEZONE']
    TIMEVALP       = rffi.CArrayPtr(TIMEVAL)
    TIMESPECP      = rffi.CArrayPtr(TIMESPEC)
    TIMEZONEP      = rffi.CArrayPtr(TIMEZONE)
    SEM_T          = rffi.COpaquePtr('sem_t', compilation_info=eci)
    #                rffi.cast(SEM_T, config['SEM_FAILED'])
    SEM_FAILED     = config['SEM_FAILED']
    SEM_VALUE_MAX  = config['SEM_VALUE_MAX']
    if SEM_VALUE_MAX is None:      # on Hurd
        SEM_VALUE_MAX = sys.maxint
    SEM_TIMED_WAIT = config['SEM_TIMED_WAIT']
    SEM_T_SIZE = config['SEM_T_SIZE']
    if sys.platform == 'darwin':
        HAVE_BROKEN_SEM_GETVALUE = True
    else:
        HAVE_BROKEN_SEM_GETVALUE = False

    def external(name, args, result, **kwargs):
        return rffi.llexternal(name, args, result,
                               compilation_info=eci, **kwargs)

    _sem_open = external('sem_open',
                         [rffi.CCHARP, rffi.INT, rffi.INT, rffi.UINT],
                         SEM_T, save_err=rffi.RFFI_SAVE_ERRNO)
    # sem_close is releasegil=False to be able to use it in the __del__
    _sem_close_no_errno = external('sem_close', [SEM_T], rffi.INT,
                                   releasegil=False)
    _sem_close = external('sem_close', [SEM_T], rffi.INT, releasegil=False,
                          save_err=rffi.RFFI_SAVE_ERRNO)
    _sem_unlink = external('sem_unlink', [rffi.CCHARP], rffi.INT,
                           save_err=rffi.RFFI_SAVE_ERRNO)
    _sem_wait = external('sem_wait', [SEM_T], rffi.INT,
                         save_err=rffi.RFFI_SAVE_ERRNO)
    _sem_trywait = external('sem_trywait', [SEM_T], rffi.INT,
                            save_err=rffi.RFFI_SAVE_ERRNO)
    _sem_post = external('sem_post', [SEM_T], rffi.INT, releasegil=False,
                         save_err=rffi.RFFI_SAVE_ERRNO)
    _sem_getvalue = external('sem_getvalue', [SEM_T, rffi.INT_realP], rffi.INT,
                             save_err=rffi.RFFI_SAVE_ERRNO)

    _gettimeofday = external('gettimeofday', [TIMEVALP, rffi.VOIDP], rffi.INT,
                             save_err=rffi.RFFI_SAVE_ERRNO)

    _select = external('select', [rffi.INT, rffi.VOIDP, rffi.VOIDP, rffi.VOIDP,
                                                          TIMEVALP], rffi.INT,
                       save_err=rffi.RFFI_SAVE_ERRNO)

    @jit.dont_look_inside
    def sem_open(name, oflag, mode, value):
        res = _sem_open(name, oflag, mode, value)
        if res == rffi.cast(SEM_T, SEM_FAILED):
            raise OSError(rposix.get_saved_errno(), "sem_open failed")
        return res

    def sem_close(handle):
        res = _sem_close(handle)
        if res < 0:
            raise OSError(rposix.get_saved_errno(), "sem_close failed")

    def sem_unlink(name):
        res = _sem_unlink(name)
        if res < 0:
            raise OSError(rposix.get_saved_errno(), "sem_unlink failed")

    def sem_wait(sem):
        res = _sem_wait(sem)
        if res < 0:
            raise OSError(rposix.get_saved_errno(), "sem_wait failed")

    def sem_trywait(sem):
        res = _sem_trywait(sem)
        if res < 0:
            raise OSError(rposix.get_saved_errno(), "sem_trywait failed")

    def sem_timedwait(sem, deadline):
        res = _sem_timedwait(sem, deadline)
        if res < 0:
            raise OSError(rposix.get_saved_errno(), "sem_timedwait failed")

    def _sem_timedwait_save(sem, deadline):
        delay = 0
        void = lltype.nullptr(rffi.VOIDP.TO)
        with lltype.scoped_alloc(TIMEVALP.TO, 1) as tvdeadline:
            while True:
                # poll
                if _sem_trywait(sem) == 0:
                    return 0
                elif rposix.get_saved_errno() != errno.EAGAIN:
                    return -1

                now = gettimeofday()
                c_tv_sec = rffi.getintfield(deadline[0], 'c_tv_sec')
                c_tv_nsec = rffi.getintfield(deadline[0], 'c_tv_nsec')
                if (c_tv_sec < now[0] or
                    (c_tv_sec == now[0] and c_tv_nsec <= now[1])):
                    rposix.set_saved_errno(errno.ETIMEDOUT)
                    return -1


                # calculate how much time is left
                difference = ((c_tv_sec - now[0]) * 1000000 +
                                    (c_tv_nsec - now[1]))

                # check delay not too long -- maximum is 20 msecs
                if delay > 20000:
                    delay = 20000
                if delay > difference:
                    delay = difference
                delay += 1000

                # sleep
                rffi.setintfield(tvdeadline[0], 'c_tv_sec', delay / 1000000)
                rffi.setintfield(tvdeadline[0], 'c_tv_usec', delay % 1000000)
                if _select(0, void, void, void, tvdeadline) < 0:
                    return -1

    if SEM_TIMED_WAIT:
        _sem_timedwait = external('sem_timedwait', [SEM_T, TIMESPECP],
                                  rffi.INT, save_err=rffi.RFFI_SAVE_ERRNO)
    else:
        _sem_timedwait = _sem_timedwait_save

    def sem_post(sem):
        res = _sem_post(sem)
        if res < 0:
            raise OSError(rposix.get_saved_errno(), "sem_post failed")

    def sem_getvalue(sem):
        sval_ptr = lltype.malloc(rffi.INT_realP.TO, 1, flavor='raw')
        try:
            res = _sem_getvalue(sem, sval_ptr)
            if res < 0:
                raise OSError(rposix.get_saved_errno(), "sem_getvalue failed")
            return rffi.cast(lltype.Signed, sval_ptr[0])
        finally:
            lltype.free(sval_ptr, flavor='raw')

    def gettimeofday():
        now = lltype.malloc(TIMEVALP.TO, 1, flavor='raw')
        void = lltype.nullptr(rffi.VOIDP.TO)
        try:
            res = _gettimeofday(now, void)
            if res < 0:
                raise OSError(rposix.get_saved_errno(), "gettimeofday failed")
            return (rffi.getintfield(now[0], 'c_tv_sec'),
                    rffi.getintfield(now[0], 'c_tv_usec'))
        finally:
            lltype.free(now, flavor='raw')

    def handle_w(space, w_handle):
        return rffi.cast(SEM_T, space.int_w(w_handle))

class CounterState:
    def __init__(self, space):
        self.counter = 0

    def _cleanup_(self):
        self.counter = 0

    def getCount(self):
        value = self.counter
        self.counter += 1
        return value

# These functions may raise bare OSError or WindowsError,
# don't forget to wrap them into OperationError

if sys.platform == 'win32':
    def create_semaphore(space, name, val, max):
        rwin32.SetLastError_saved(0)
        handle = _CreateSemaphore(rffi.NULL, val, max, rffi.NULL)
        # On Windows we should fail on ERROR_ALREADY_EXISTS
        err = rwin32.GetLastError_saved()
        if err != 0:
            raise WindowsError(err, "CreateSemaphore")
        return handle

    def delete_semaphore(handle):
        _CloseHandle_no_errno(handle)

    def semlock_acquire(self, space, block, w_timeout):
        if not block:
            full_msecs = 0
        elif space.is_none(w_timeout):
            full_msecs = rwin32.INFINITE
        else:
            timeout = space.float_w(w_timeout)
            timeout *= 1000.0
            if timeout < 0.0:
                timeout = 0.0
            elif timeout >= 0.5 * rwin32.INFINITE: # 25 days
                raise oefmt(space.w_OverflowError, "timeout is too large")
            full_msecs = r_uint(int(timeout + 0.5))

        # check whether we can acquire without blocking
        res = rwin32.WaitForSingleObject(self.handle, 0)

        if res != rwin32.WAIT_TIMEOUT:
            self.last_tid = rthread.get_ident()
            self.count += 1
            return True

        msecs = full_msecs
        start = _GetTickCount()

        while True:
            from pypy.module.time.interp_time import State
            interrupt_event = space.fromcache(State).get_interrupt_event()
            handles = [self.handle, interrupt_event]

            # do the wait
            rwin32.ResetEvent(interrupt_event)
            res = rwin32.WaitForMultipleObjects(handles, timeout=msecs)

            if res != rwin32.WAIT_OBJECT_0 + 1:
                break

            # got SIGINT so give signal handler a chance to run
            time.sleep(0.001)

            # if this is main thread let KeyboardInterrupt be raised
            _check_signals(space)

            # recalculate timeout
            if msecs != rwin32.INFINITE:
                ticks = _GetTickCount()
                if r_uint(ticks - start) >= full_msecs:
                    return False
                msecs = full_msecs - r_uint(ticks - start)

        # handle result
        if res != rwin32.WAIT_TIMEOUT:
            self.last_tid = rthread.get_ident()
            self.count += 1
            return True
        return False

    def semlock_release(self, space):
        if not _ReleaseSemaphore(self.handle, 1,
                                 lltype.nullptr(rffi.LONGP.TO)):
            err = rwin32.GetLastError_saved()
            if err == 0x0000012a: # ERROR_TOO_MANY_POSTS
                raise oefmt(space.w_ValueError,
                            "semaphore or lock released too many times")
            else:
                raise WindowsError(err, "ReleaseSemaphore")

    def semlock_getvalue(self, space):
        if rwin32.WaitForSingleObject(self.handle, 0) == rwin32.WAIT_TIMEOUT:
            return 0
        previous_ptr = lltype.malloc(rffi.LONGP.TO, 1, flavor='raw')
        try:
            if not _ReleaseSemaphore(self.handle, 1, previous_ptr):
                raise rwin32.lastSavedWindowsError("ReleaseSemaphore")
            return intmask(previous_ptr[0]) + 1
        finally:
            lltype.free(previous_ptr, flavor='raw')

    def semlock_iszero(self, space):
        return semlock_getvalue(self, space) == 0

else:
    def create_semaphore(space, name, val, max):
        sem = sem_open(name, os.O_CREAT | os.O_EXCL, 0600, val)
        try:
            sem_unlink(name)
        except OSError:
            pass
        else:
            rgc.add_memory_pressure(SEM_T_SIZE)
        return sem

    def delete_semaphore(handle):
        _sem_close_no_errno(handle)

    def semlock_acquire(self, space, block, w_timeout):
        if not block:
            deadline = lltype.nullptr(TIMESPECP.TO)
        elif space.is_none(w_timeout):
            deadline = lltype.nullptr(TIMESPECP.TO)
        else:
            timeout = space.float_w(w_timeout)
            sec = int(timeout)
            nsec = int(1e9 * (timeout - sec) + 0.5)

            now_sec, now_usec = gettimeofday()

            deadline = lltype.malloc(TIMESPECP.TO, 1, flavor='raw')
            rffi.setintfield(deadline[0], 'c_tv_sec', now_sec + sec)
            rffi.setintfield(deadline[0], 'c_tv_nsec', now_usec * 1000 + nsec)
            val = (rffi.getintfield(deadline[0], 'c_tv_sec') +
                   rffi.getintfield(deadline[0], 'c_tv_nsec') / 1000000000)
            rffi.setintfield(deadline[0], 'c_tv_sec', val)
            val = rffi.getintfield(deadline[0], 'c_tv_nsec') % 1000000000
            rffi.setintfield(deadline[0], 'c_tv_nsec', val)
        try:
            while True:
                try:
                    if not block:
                        sem_trywait(self.handle)
                    elif not deadline:
                        sem_wait(self.handle)
                    else:
                        sem_timedwait(self.handle, deadline)
                except OSError as e:
                    if e.errno == errno.EINTR:
                        # again
                        _check_signals(space)
                        continue
                    elif e.errno in (errno.EAGAIN, errno.ETIMEDOUT):
                        return False
                    raise
                _check_signals(space)
                self.last_tid = rthread.get_ident()
                self.count += 1
                return True
        finally:
            if deadline:
                lltype.free(deadline, flavor='raw')


    def semlock_release(self, space):
        if self.kind == RECURSIVE_MUTEX:
            sem_post(self.handle)
            return
        if HAVE_BROKEN_SEM_GETVALUE:
            # We will only check properly the maxvalue == 1 case
            if self.maxvalue == 1:
                # make sure that already locked
                try:
                    sem_trywait(self.handle)
                except OSError as e:
                    if e.errno != errno.EAGAIN:
                        raise
                    # it is already locked as expected
                else:
                    # it was not locked so undo wait and raise
                    sem_post(self.handle)
                    raise oefmt(space.w_ValueError,
                                "semaphore or lock released too many times")
        else:
            # This check is not an absolute guarantee that the semaphore does
            # not rise above maxvalue.
            if sem_getvalue(self.handle) >= self.maxvalue:
                raise oefmt(space.w_ValueError,
                            "semaphore or lock released too many times")

        sem_post(self.handle)

    def semlock_getvalue(self, space):
        if HAVE_BROKEN_SEM_GETVALUE:
            raise oefmt(space.w_NotImplementedError,
                        "sem_getvalue is not implemented on this system")
        else:
            val = sem_getvalue(self.handle)
            # some posix implementations use negative numbers to indicate
            # the number of waiting threads
            if val < 0:
                val = 0
            return val

    def semlock_iszero(self, space):
        if HAVE_BROKEN_SEM_GETVALUE:
            try:
                sem_trywait(self.handle)
            except OSError as e:
                if e.errno != errno.EAGAIN:
                    raise
                return True
            else:
                sem_post(self.handle)
                return False
        else:
            return semlock_getvalue(self, space) == 0


class W_SemLock(W_Root):
    def __init__(self, space, handle, kind, maxvalue):
        self.handle = handle
        self.kind = kind
        self.count = 0
        self.maxvalue = maxvalue
        self.register_finalizer(space)
        self.last_tid = -1

    def kind_get(self, space):
        return space.newint(self.kind)
    def maxvalue_get(self, space):
        return space.newint(self.maxvalue)
    def handle_get(self, space):
        return w_handle(space, self.handle)

    def get_count(self, space):
        return space.newint(self.count)

    def _ismine(self):
        return self.count > 0 and rthread.get_ident() == self.last_tid

    def is_mine(self, space):
        return space.newbool(self._ismine())

    def is_zero(self, space):
        try:
            res = semlock_iszero(self, space)
        except OSError as e:
            raise wrap_oserror(space, e)
        return space.newbool(res)

    def get_value(self, space):
        try:
            val = semlock_getvalue(self, space)
        except OSError as e:
            raise wrap_oserror(space, e)
        return space.newint(val)

    @unwrap_spec(block=bool)
    def acquire(self, space, block=True, w_timeout=None):
        # check whether we already own the lock
        if self.kind == RECURSIVE_MUTEX and self._ismine():
            self.count += 1
            return space.w_True
        try:
            # sets self.last_tid and increments self.count
            # those steps need to be as close as possible to
            # acquiring the semlock for self._ismine() to support
            # multiple threads
            got = semlock_acquire(self, space, block, w_timeout)
        except OSError as e:
            raise wrap_oserror(space, e)
        if got:
            return space.w_True
        else:
            return space.w_False

    def release(self, space):
        if self.kind == RECURSIVE_MUTEX:
            if not self._ismine():
                raise oefmt(space.w_AssertionError,
                            "attempt to release recursive lock not owned by "
                            "thread")
            if self.count > 1:
                self.count -= 1
                return

        try:
            # Note: a succesful semlock_release() must not release the GIL,
            # otherwise there is a race condition on self.count
            semlock_release(self, space)
            self.count -= 1
        except OSError as e:
            raise wrap_oserror(space, e)


    def after_fork(self):
        self.count = 0

    @unwrap_spec(kind=int, maxvalue=int)
    def rebuild(space, w_cls, w_handle, kind, maxvalue):
        self = space.allocate_instance(W_SemLock, w_cls)
        self.__init__(space, handle_w(space, w_handle), kind, maxvalue)
        return self

    def enter(self, space):
        return self.acquire(space, w_timeout=space.w_None)

    def exit(self, space, __args__):
        self.release(space)

    def _finalize_(self):
        delete_semaphore(self.handle)

@unwrap_spec(kind=int, value=int, maxvalue=int)
def descr_new(space, w_subtype, kind, value, maxvalue):
    if kind != RECURSIVE_MUTEX and kind != SEMAPHORE:
        raise oefmt(space.w_ValueError, "unrecognized kind")

    counter = space.fromcache(CounterState).getCount()
    name = "/mp%d-%d" % (os.getpid(), counter)

    try:
        handle = create_semaphore(space, name, value, maxvalue)
    except OSError as e:
        raise wrap_oserror(space, e)

    self = space.allocate_instance(W_SemLock, w_subtype)
    self.__init__(space, handle, kind, maxvalue)

    return self

W_SemLock.typedef = TypeDef(
    "SemLock",
    __new__ = interp2app(descr_new),
    kind = GetSetProperty(W_SemLock.kind_get),
    maxvalue = GetSetProperty(W_SemLock.maxvalue_get),
    handle = GetSetProperty(W_SemLock.handle_get),
    _count = interp2app(W_SemLock.get_count),
    _is_mine = interp2app(W_SemLock.is_mine),
    _is_zero = interp2app(W_SemLock.is_zero),
    _get_value = interp2app(W_SemLock.get_value),
    acquire = interp2app(W_SemLock.acquire),
    release = interp2app(W_SemLock.release),
    _rebuild = interp2app(W_SemLock.rebuild.im_func, as_classmethod=True),
    _after_fork = interp2app(W_SemLock.after_fork),
    __enter__=interp2app(W_SemLock.enter),
    __exit__=interp2app(W_SemLock.exit),
    SEM_VALUE_MAX=SEM_VALUE_MAX,
    )

def _check_signals(space):
    space.getexecutioncontext().checksignals()
