libgevent/python/gevent2.pyx at master · denik/libgevent · GitHub

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
import sys
from cpython cimport *


cdef extern from "../gevent.c":
    int GEVENT_COTHREAD_CURRENT
    int GEVENT_COTHREAD_NEW
    int GEVENT_COTHREAD_READY
    int GEVENT_COTHREAD_DEAD
    int GEVENT_COTHREAD_CHANNEL_R
    int GEVENT_COTHREAD_CHANNEL_S

    ctypedef struct gevent_cothread:
        int state
        void* exit_fn

    ctypedef struct gevent_hub:
        pass

    ctypedef struct gevent_channel:
        pass

    ctypedef struct gevent_semaphore:
        int counter

    gevent_hub* gevent_default_hub()
    void gevent_cothread_init(gevent_hub*, gevent_cothread*, void*)
    void gevent_cothread_spawn(gevent_cothread*)

    int gevent_sleep(gevent_hub* hub, long timeout)
    int gevent_wait(gevent_hub* hub, long timeout)

    void gevent_channel_init(gevent_hub* hub, gevent_channel* ch)
    void gevent_channel_send(gevent_channel* ch, object value)
    void gevent_channel_receive(gevent_channel* ch, void** result)

    void gevent_semaphore_init(gevent_hub* hub, gevent_semaphore* sem, int counter)
    int gevent_semaphore_acquire(gevent_semaphore* sem)
    int gevent_semaphore_release(gevent_semaphore* sem)


cdef extern from "socketmodule.c":
    object pygevent_getaddrinfo(object)


cdef extern from "callbacks.h":
    void cothread_init(gevent_cothread* t, object run, object args, object kwargs)
    void cothread_exit_fn(gevent_cothread* t)


cdef public class cothread [object PyGeventCothreadObject, type PyGeventCothread_Type]:
    cdef gevent_cothread data

    def __cinit__(self, run, *args, **kwargs):
        cothread_init(&self.data, run, args, kwargs)

    cpdef spawn(self):
        if self.data.state != GEVENT_COTHREAD_NEW:
            raise ValueError('This cothread already was spawned')
        Py_INCREF(self)
        self.data.exit_fn = <void*>cothread_exit_fn
        gevent_cothread_spawn(&self.data)

    property state:

        def __get__(self):
            return self.data.state


cdef public class channel [object PyGeventChannelObject, type PyGeventChannel_Type]:
    cdef gevent_channel data

    def __cinit__(self):
        gevent_channel_init(gevent_default_hub(), &self.data)

    def send(self, object item):
        gevent_channel_send(&self.data, item)

    def receive(self):
        cdef void* obj
        gevent_channel_receive(&self.data, &obj)
        return <object>obj


cdef public class semaphore [object PyGeventSemaphoreObject, type PyGeventSemaphore_Type]:
    cdef gevent_semaphore data

    def __cinit__(self, int count):
        gevent_semaphore_init(gevent_default_hub(), &self.data, count)

    def acquire(self):
        gevent_semaphore_acquire(&self.data)

    def release(self):
        gevent_semaphore_release(&self.data)

    property counter:

        def __get__(self):
            return self.data.counter

        def __set__(self, int counter):
            self.data.counter = counter


def spawn(*args, **kwargs):
    cdef cothread t = cothread(*args, **kwargs)
    t.spawn()
    return t


def sleep(timeout):
    cdef long c_timeout = timeout * 1000.0
    if gevent_sleep(gevent_default_hub(), c_timeout):
        raise RuntimeError


def wait(timeout=None):
    cdef long c_timeout
    if timeout is None:
        c_timeout = 0
    else:
        c_timeout = timeout
    if gevent_wait(gevent_default_hub(), c_timeout):
        raise MemoryError


def getaddrinfo(*args):
    # XXX useless wrapper
    return pygevent_getaddrinfo(args)