pypy / pypy / module / _multiprocessing / interp_semaphore.py

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
import errno
import os
import sys
import time

from rpython.rlib import rgc, rthread
from rpython.rlib.rarithmetic import 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 OperationError, 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 = sys.maxint

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

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)])
        SEM_FAILED = platform.ConstantInteger('SEM_FAILED')
        SEM_VALUE_MAX = platform.ConstantInteger('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']
    TIMEVALP       = rffi.CArrayPtr(TIMEVAL)
    TIMESPECP      = rffi.CArrayPtr(TIMESPEC)
    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']
    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)
    # sem_close is releasegil=False to be able to use it in the __del__
    _sem_close = external('sem_close', [SEM_T], rffi.INT, releasegil=False)
    _sem_unlink = external('sem_unlink', [rffi.CCHARP], rffi.INT)
    _sem_wait = external('sem_wait', [SEM_T], rffi.INT)
    _sem_trywait = external('sem_trywait', [SEM_T], rffi.INT)
    _sem_post = external('sem_post', [SEM_T], rffi.INT)
    _sem_getvalue = external('sem_getvalue', [SEM_T, rffi.INTP], rffi.INT)

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

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

    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_errno(), "sem_open failed")
        return res

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

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

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

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

    def sem_timedwait(sem, deadline):
        res = _sem_timedwait(sem, deadline)
        if res < 0:
            raise OSError(rposix.get_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_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_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)
    else:
        _sem_timedwait = _sem_timedwait_save

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

    def sem_getvalue(sem):
        sval_ptr = lltype.malloc(rffi.INTP.TO, 1, flavor='raw')
        try:
            res = _sem_getvalue(sem, sval_ptr)
            if res < 0:
                raise OSError(rposix.get_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')
        try:
            res = _gettimeofday(now, None)
            if res < 0:
                raise OSError(rposix.get_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(0)
        handle = _CreateSemaphore(rffi.NULL, val, max, rffi.NULL)
        # On Windows we should fail on ERROR_ALREADY_EXISTS
        err = rwin32.GetLastError()
        if err != 0:
            raise WindowsError(err, "CreateSemaphore")
        return handle

    def delete_semaphore(handle):
        if not _CloseHandle(handle):
            err = rwin32.GetLastError()
            raise WindowsError(err, "CloseHandle")

    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 OperationError(space.w_OverflowError,
                                     space.wrap("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:
            return True

        msecs = full_msecs
        start = _GetTickCount()

        while True:
            from pypy.module.rctime.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:
            return True
        return False

    def semlock_release(self, space):
        if not _ReleaseSemaphore(self.handle, 1,
                                 lltype.nullptr(rffi.LONGP.TO)):
            err = rwin32.GetLastError()
            if err == 0x0000012a: # ERROR_TOO_MANY_POSTS
                raise OperationError(
                    space.w_ValueError,
                    space.wrap("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.lastWindowsError("ReleaseSemaphore")
            return 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(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, e:
                    if e.errno == errno.EINTR:
                        # again
                        continue
                    elif e.errno in (errno.EAGAIN, errno.ETIMEDOUT):
                        return False
                    raise
                _check_signals(space)

                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, 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 OperationError(
                        space.w_ValueError, space.wrap(
                            "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 OperationError(
                    space.w_ValueError, space.wrap(
                    "semaphore or lock released too many times"))

        sem_post(self.handle)

    def semlock_getvalue(self, space):
        if HAVE_BROKEN_SEM_GETVALUE:
            raise OperationError(space.w_NotImplementedError, space.wrap(
                        '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, 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, handle, kind, maxvalue):
        self.handle = handle
        self.kind = kind
        self.count = 0
        self.maxvalue = maxvalue

    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.wrap(self.count)

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

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

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

    def get_value(self, space):
        try:
            val = semlock_getvalue(self, space)
        except OSError, e:
            raise wrap_oserror(space, e)
        return space.wrap(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:
            got = semlock_acquire(self, space, block, w_timeout)
        except OSError, e:
            raise wrap_oserror(space, e)

        if got:
            self.last_tid = rthread.get_ident()
            self.count += 1
            return space.w_True
        else:
            return space.w_False

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

        try:
            semlock_release(self, space)
        except OSError, e:
            raise wrap_oserror(space, e)

        self.count -= 1

    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__(handle_w(space, w_handle), kind, maxvalue)
        return space.wrap(self)

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

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

    def __del__(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 OperationError(space.w_ValueError,
                             space.wrap("unrecognized kind"))

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

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

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

    return space.wrap(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()
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.