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
|
# -*- coding: utf-8 -*-
#
# File: portato/ipc.pyx
# This file is part of the Portato-Project, a graphical portage-frontend.
#
# Copyright (C) 2006-2010 René 'Necoro' Neumann
# This is free software. You may redistribute copies of it under the terms of
# the GNU General Public License version 2.
# There is NO WARRANTY, to the extent permitted by law.
#
# Written by René 'Necoro' Neumann <necoro@necoro.net>
class MessageQueueError(Exception):
"""
Base class for different queue errors.
"""
pass
class MessageQueueRemovedError (MessageQueueError):
"""
This class is used iff the queue is already removed.
"""
pass
cdef class MessageQueue (object):
"""
A simple interface to the SysV message queues.
"""
cdef int msgid
cdef readonly key_t key
def __init__ (self, key = None, create = False, exclusive = False):
"""
Create a new MessageQueue instance. Depending on the passed in flags,
different behavior occurs. See man msgget for the details.
If key is None, a random key is created.
"""
cdef int flags = 0600 # start mode
if exclusive and not create:
raise ValueError("'exclusive' must be combined with 'create'.")
if key is None and not exclusive:
raise ValueError("The key can only be None if 'exclusive' is set.")
if create:
flags |= IPC_CREAT
if exclusive:
flags |= IPC_EXCL
if key is None:
check = True
while check:
self.key = self.random_key()
self.msgid = msgget(self.key, flags)
check = (self.msgid == -1 and errno == EEXIST)
else:
self.key = key
self.msgid = msgget(key, flags)
if self.msgid == -1:
if errno == EACCES:
raise MessageQueueError("Permission denied.")
elif errno == EEXIST:
raise MessageQueueError("Queue already exists.")
elif errno == ENOENT:
raise MessageQueueError("Queue does not exist and 'create' is not set.")
elif errno == ENOMEM or errno == ENOSPC:
PyErr_NoMemory()
else:
raise OSError(errno, strerror(errno))
def remove (self):
"""
Removes the message queue.
"""
cdef msqid_ds info
cdef int ret
ret = msgctl(self.msgid, IPC_RMID, &info)
if ret == -1:
if errno == EIDRM or errno == EINVAL:
raise MessageQueueRemovedError("Queue already removed.")
elif errno == EPERM:
raise MessageQueueError("Permission denied.")
else:
raise OSError(errno, strerror(errno))
def send (self, message, int type = 1):
"""
Sends a message with a specific type.
The type must be larger zero.
Also note, that this is always blocking.
"""
cdef msg_data * msg
cdef int ret
cdef long size = len(message)
if type <= 0:
raise ValueError("type must be > 0")
if size >= MAX_MESSAGE_SIZE:
raise ValueError("Message must be smaller than %d" % MAX_MESSAGE_SIZE)
msg = <msg_data*>PyMem_Malloc(sizeof(msg_data) + size)
if msg is NULL:
PyErr_NoMemory()
memcpy(msg.mtext, <char*>message, size)
msg.mtype = type
with nogil:
ret = msgsnd(self.msgid, msg, size, 0)
PyMem_Free(msg)
if ret == -1:
if errno == EIDRM or errno == EINVAL:
raise MessageQueueRemovedError("Queue was removed.")
elif errno == EINTR:
raise MessageQueueError("Signaled while waiting.")
elif errno == EACCES:
raise MessageQueueError("Permission denied.")
else:
raise OSError(errno, strerror(errno))
def receive (self):
"""
Receives a message from the queue and returns the (msg, type) pair.
Note that this method is always blocking.
"""
cdef msg_data * msg
cdef int ret
cdef object retTuple
msg = <msg_data*>PyMem_Malloc(sizeof(msg_data) + MAX_MESSAGE_SIZE)
if msg is NULL:
PyErr_NoMemory()
msg.mtype = 0
with nogil:
ret = msgrcv(self.msgid, msg, <size_t>MAX_MESSAGE_SIZE, 0, 0)
try:
if ret == -1:
if errno == EIDRM or errno == EINVAL:
raise MessageQueueRemovedError("Queue was removed.")
elif errno == EINTR:
raise MessageQueueError("Signaled while waiting.")
elif errno == EACCES:
raise MessageQueueError("Permission denied.")
else:
raise OSError(errno, strerror(errno))
else:
return (PyString_FromStringAndSize(msg.mtext, ret), msg.mtype)
finally:
PyMem_Free(msg)
cdef inline key_t random_key (self):
return <int>(<double>rand() / (<double>RAND_MAX + 1) * INT_MAX)
|
