summaryrefslogtreecommitdiffstats
path: root/scripts/external_libs/platform/cel59/zmq/backend/cython/message.pyx
blob: 312ae120df6f8e9ebcf58df8388d844616c85817 (plain)
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
"""0MQ Message related classes."""

#
#    Copyright (c) 2013 Brian E. Granger & Min Ragan-Kelley
#
#    This file is part of pyzmq.
#
#    pyzmq is free software; you can redistribute it and/or modify it under
#    the terms of the Lesser GNU General Public License as published by
#    the Free Software Foundation; either version 3 of the License, or
#    (at your option) any later version.
#
#    pyzmq is distributed in the hope that it will be useful,
#    but WITHOUT ANY WARRANTY; without even the implied warranty of
#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#    Lesser GNU General Public License for more details.
#
#    You should have received a copy of the Lesser GNU General Public License
#    along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

#-----------------------------------------------------------------------------
# Imports
#-----------------------------------------------------------------------------

# get version-independent aliases:
cdef extern from "pyversion_compat.h":
    pass

from cpython cimport Py_DECREF, Py_INCREF

from buffers cimport asbuffer_r, viewfromobject_r

cdef extern from "Python.h":
    ctypedef int Py_ssize_t

from libzmq cimport *

from libc.stdio cimport fprintf, stderr as cstderr
from libc.stdlib cimport malloc, free
from libc.string cimport memcpy

import time

try:
    # below 3.3
    from threading import _Event as Event
except (ImportError, AttributeError):
    # python throws ImportError, cython throws AttributeError
    from threading import Event

import zmq
from zmq.error import _check_version
from zmq.backend.cython.checkrc cimport _check_rc
from zmq.utils.strtypes import bytes,unicode,basestring

#-----------------------------------------------------------------------------
# Code
#-----------------------------------------------------------------------------

ctypedef struct zhint:
    void *ctx
    size_t id

cdef void free_python_msg(void *data, void *vhint) nogil:
    """A pure-C function for DECREF'ing Python-owned message data.
    
    Sends a message on a PUSH socket
    
    The hint is a `zhint` struct with two values:
    
    ctx (void *): pointer to the Garbage Collector's context
    id (size_t): the id to be used to construct a zmq_msg_t that should be sent on a PUSH socket,
       signaling the Garbage Collector to remove its reference to the object.
    
    - A PUSH socket is created in the context,
    - it is connected to the garbage collector inproc channel,
    - it sends the gc message
    - the PUSH socket is closed
    
    When the Garbage Collector's PULL socket receives the message,
    it deletes its reference to the object,
    allowing Python to free the memory.
    """
    cdef void *push
    cdef zmq_msg_t msg
    cdef zhint *hint = <zhint *> vhint
    if hint != NULL:
        zmq_msg_init_size(&msg, sizeof(size_t))
        memcpy(zmq_msg_data(&msg), &hint.id, sizeof(size_t))
        
        push = zmq_socket(hint.ctx, ZMQ_PUSH)
        if push == NULL:
            # this will happen if the context has been terminated
            return
        rc = zmq_connect(push, "inproc://pyzmq.gc.01")
        if rc < 0:
            fprintf(cstderr, "pyzmq-gc connect failed: %s\n", zmq_strerror(zmq_errno()))
            return
        
        rc = zmq_msg_send(&msg, push, 0)
        if rc < 0:
            fprintf(cstderr, "pyzmq-gc send failed: %s\n", zmq_strerror(zmq_errno()))
        
        zmq_msg_close(&msg)
        zmq_close(push)
        free(hint)

gc = None

cdef class Frame:
    """Frame(data=None, track=False)

    A zmq message Frame class for non-copy send/recvs.

    This class is only needed if you want to do non-copying send and recvs.
    When you pass a string to this class, like ``Frame(s)``, the 
    ref-count of `s` is increased by two: once because the Frame saves `s` as 
    an instance attribute and another because a ZMQ message is created that
    points to the buffer of `s`. This second ref-count increase makes sure
    that `s` lives until all messages that use it have been sent. Once 0MQ
    sends all the messages and it doesn't need the buffer of s, 0MQ will call
    ``Py_DECREF(s)``.

    Parameters
    ----------

    data : object, optional
        any object that provides the buffer interface will be used to
        construct the 0MQ message data.
    track : bool [default: False]
        whether a MessageTracker_ should be created to track this object.
        Tracking a message has a cost at creation, because it creates a threadsafe
        Event object.
    
    """

    def __cinit__(self, object data=None, track=False, **kwargs):
        cdef int rc
        cdef char *data_c = NULL
        cdef Py_ssize_t data_len_c=0
        cdef zhint *hint

        # init more as False
        self.more = False

        # Save the data object in case the user wants the the data as a str.
        self._data = data
        self._failed_init = True  # bool switch for dealloc
        self._buffer = None       # buffer view of data
        self._bytes = None        # bytes copy of data

        # Event and MessageTracker for monitoring when zmq is done with data:
        if track:
            evt = Event()
            self.tracker_event = evt
            self.tracker = zmq.MessageTracker(evt)
        else:
            self.tracker_event = None
            self.tracker = None

        if isinstance(data, unicode):
            raise TypeError("Unicode objects not allowed. Only: str/bytes, buffer interfaces.")

        if data is None:
            rc = zmq_msg_init(&self.zmq_msg)
            _check_rc(rc)
            self._failed_init = False
            return
        else:
            asbuffer_r(data, <void **>&data_c, &data_len_c)
        
        # create the hint for zmq_free_fn
        # two pointers: the gc context and a message to be sent to the gc PULL socket
        # allows libzmq to signal to Python when it is done with Python-owned memory.
        global gc
        if gc is None:
            from zmq.utils.garbage import gc
        
        hint = <zhint *> malloc(sizeof(zhint))
        hint.id = gc.store(data, self.tracker_event)
        hint.ctx = <void *> <size_t> gc._context.underlying
        
        rc = zmq_msg_init_data(
                &self.zmq_msg, <void *>data_c, data_len_c, 
                <zmq_free_fn *>free_python_msg, <void *>hint
            )
        if rc != 0:
            free(hint)
            _check_rc(rc)
        self._failed_init = False
    
    def __init__(self, object data=None, track=False):
        """Enforce signature"""
        pass

    def __dealloc__(self):
        cdef int rc
        if self._failed_init:
            return
        # This simply decreases the 0MQ ref-count of zmq_msg.
        with nogil:
            rc = zmq_msg_close(&self.zmq_msg)
        _check_rc(rc)
    
    # buffer interface code adapted from petsc4py by Lisandro Dalcin, a BSD project
    
    def __getbuffer__(self, Py_buffer* buffer, int flags):
        # new-style (memoryview) buffer interface
        buffer.buf = zmq_msg_data(&self.zmq_msg)
        buffer.len = zmq_msg_size(&self.zmq_msg)
        
        buffer.obj = self
        buffer.readonly = 1
        buffer.format = "B"
        buffer.ndim = 0
        buffer.shape = NULL
        buffer.strides = NULL
        buffer.suboffsets = NULL
        buffer.itemsize = 1
        buffer.internal = NULL
    
    def __getsegcount__(self, Py_ssize_t *lenp):
        # required for getreadbuffer
        if lenp != NULL:
            lenp[0] = zmq_msg_size(&self.zmq_msg)
        return 1
    
    def __getreadbuffer__(self, Py_ssize_t idx, void **p):
        # old-style (buffer) interface
        cdef char *data_c = NULL
        cdef Py_ssize_t data_len_c
        if idx != 0:
            raise SystemError("accessing non-existent buffer segment")
        # read-only, because we don't want to allow
        # editing of the message in-place
        data_c = <char *>zmq_msg_data(&self.zmq_msg)
        data_len_c = zmq_msg_size(&self.zmq_msg)
        if p != NULL:
            p[0] = <void*>data_c
        return data_len_c
    
    # end buffer interface
    
    def __copy__(self):
        """Create a shallow copy of the message.

        This does not copy the contents of the Frame, just the pointer.
        This will increment the 0MQ ref count of the message, but not
        the ref count of the Python object. That is only done once when
        the Python is first turned into a 0MQ message.
        """
        return self.fast_copy()

    cdef Frame fast_copy(self):
        """Fast, cdef'd version of shallow copy of the Frame."""
        cdef Frame new_msg
        new_msg = Frame()
        # This does not copy the contents, but just increases the ref-count 
        # of the zmq_msg by one.
        zmq_msg_copy(&new_msg.zmq_msg, &self.zmq_msg)
        # Copy the ref to data so the copy won't create a copy when str is
        # called.
        if self._data is not None:
            new_msg._data = self._data
        if self._buffer is not None:
            new_msg._buffer = self._buffer
        if self._bytes is not None:
            new_msg._bytes = self._bytes

        # Frame copies share the tracker and tracker_event
        new_msg.tracker_event = self.tracker_event
        new_msg.tracker = self.tracker

        return new_msg

    def __len__(self):
        """Return the length of the message in bytes."""
        cdef size_t sz
        sz = zmq_msg_size(&self.zmq_msg)
        return sz
        # return <int>zmq_msg_size(&self.zmq_msg)

    def __str__(self):
        """Return the str form of the message."""
        if isinstance(self._data, bytes):
            b = self._data
        else:
            b = self.bytes
        if str is unicode:
            return b.decode()
        else:
            return b

    cdef inline object _getbuffer(self):
        """Create a Python buffer/view of the message data.

        This will be called only once, the first time the `buffer` property
        is accessed. Subsequent calls use a cached copy.
        """
        if self._data is None:
            return viewfromobject_r(self)
        else:
            return viewfromobject_r(self._data)
    
    @property
    def buffer(self):
        """A read-only buffer view of the message contents."""
        if self._buffer is None:
            self._buffer = self._getbuffer()
        return self._buffer

    @property
    def bytes(self):
        """The message content as a Python bytes object.

        The first time this property is accessed, a copy of the message 
        contents is made. From then on that same copy of the message is
        returned.
        """
        if self._bytes is None:
            self._bytes = copy_zmq_msg_bytes(&self.zmq_msg)
        return self._bytes
    
    def set(self, int option, int value):
        """Frame.set(option, value)
        
        Set a Frame option.
        
        See the 0MQ API documentation for zmq_msg_set
        for details on specific options.
        
        .. versionadded:: libzmq-3.2
        .. versionadded:: 13.0
        """
        cdef int rc = zmq_msg_set(&self.zmq_msg, option, value)
        _check_rc(rc)

    def get(self, option):
        """Frame.get(option)

        Get a Frame option or property.

        See the 0MQ API documentation for zmq_msg_get and zmq_msg_gets
        for details on specific options.

        .. versionadded:: libzmq-3.2
        .. versionadded:: 13.0
        
        .. versionchanged:: 14.3
            add support for zmq_msg_gets (requires libzmq-4.1)
        """
        cdef int rc = 0
        cdef char *property_c = NULL
        cdef Py_ssize_t property_len_c = 0

        # zmq_msg_get
        if isinstance(option, int):
            rc = zmq_msg_get(&self.zmq_msg, option)
            _check_rc(rc)
            return rc

        # zmq_msg_gets
        _check_version((4,1), "get string properties")
        if isinstance(option, unicode):
            option = option.encode('utf8')
        
        if not isinstance(option, bytes):
            raise TypeError("expected str, got: %r" % option)
        
        property_c = option
        
        cdef const char *result = <char *>zmq_msg_gets(&self.zmq_msg, property_c)
        if result == NULL:
            _check_rc(-1)
        return result.decode('utf8')

# legacy Message name
Message = Frame

__all__ = ['Frame', 'Message']