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
|
from cython.cimports import cqueue
from cython import cast
@cython.cclass
class Queue:
"""A queue class for C integer values.
>>> q = Queue()
>>> q.append(5)
>>> q.peek()
5
>>> q.pop()
5
"""
_c_queue = cython.declare(cython.pointer(cqueue.Queue))
def __cinit__(self):
self._c_queue = cqueue.queue_new()
if self._c_queue is cython.NULL:
raise MemoryError()
def __dealloc__(self):
if self._c_queue is not cython.NULL:
cqueue.queue_free(self._c_queue)
@cython.ccall
def append(self, value: cython.int):
if not cqueue.queue_push_tail(self._c_queue,
cast(cython.p_void, cast(cython.Py_ssize_t, value))):
raise MemoryError()
# The `cpdef` feature is obviously not available for the original "extend()"
# method, as the method signature is incompatible with Python argument
# types (Python does not have pointers). However, we can rename
# the C-ish "extend()" method to e.g. "extend_ints()", and write
# a new "extend()" method that provides a suitable Python interface by
# accepting an arbitrary Python iterable.
@cython.ccall
def extend(self, values):
for value in values:
self.append(value)
@cython.cfunc
def extend_ints(self, values: cython.p_int, count: cython.size_t):
value: cython.int
for value in values[:count]: # Slicing pointer to limit the iteration boundaries.
self.append(value)
@cython.ccall
@cython.exceptval(-1, check=True)
def peek(self) -> cython.int:
value: cython.int = cast(cython.Py_ssize_t, cqueue.queue_peek_head(self._c_queue))
if value == 0:
# this may mean that the queue is empty,
# or that it happens to contain a 0 value
if cqueue.queue_is_empty(self._c_queue):
raise IndexError("Queue is empty")
return value
@cython.ccall
@cython.exceptval(-1, check=True)
def pop(self) -> cython.int:
if cqueue.queue_is_empty(self._c_queue):
raise IndexError("Queue is empty")
return cast(cython.Py_ssize_t, cqueue.queue_pop_head(self._c_queue))
def __bool__(self):
return not cqueue.queue_is_empty(self._c_queue)
|
