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
|
#define PY_SSIZE_T_CLEAN
#include <Python.h>
#include <stdint.h>
#include "structmember.h"
#include "module.h"
#include "tags.h"
// Constructors and destructors //////////////////////////////////////////////
static int
CBORTag_traverse(CBORTagObject *self, visitproc visit, void *arg)
{
Py_VISIT(self->value);
return 0;
}
static int
CBORTag_clear(CBORTagObject *self)
{
Py_CLEAR(self->value);
return 0;
}
// CBORTag.__del__(self)
static void
CBORTag_dealloc(CBORTagObject *self)
{
PyObject_GC_UnTrack(self);
CBORTag_clear(self);
Py_TYPE(self)->tp_free((PyObject *) self);
}
// CBORTag.__new__(cls, *args, **kwargs)
static PyObject *
CBORTag_new(PyTypeObject *type, PyObject *args, PyObject *kwargs)
{
CBORTagObject *self;
self = (CBORTagObject *) type->tp_alloc(type, 0);
if (self) {
self->tag = 0;
Py_INCREF(Py_None);
self->value = Py_None;
}
return (PyObject *) self;
}
// CBORTag.__init__(self, tag=None, value=None)
static int
CBORTag_init(CBORTagObject *self, PyObject *args, PyObject *kwargs)
{
static char *keywords[] = {"tag", "value", NULL};
PyObject *tmp, *value, *tmp_tag = NULL;
uint64_t tag = 0;
if (!PyArg_ParseTupleAndKeywords(args, kwargs, "OO", keywords,
&tmp_tag, &value))
return -1;
// Raises an overflow error if it doesn't work
tag = PyLong_AsUnsignedLongLong(tmp_tag);
if (tag == (uint64_t)-1) {
if (PyErr_Occurred()){
if (PyErr_ExceptionMatches(PyExc_OverflowError)) {
PyErr_Clear(); // clear the overflow error
PyErr_SetString(PyExc_TypeError, "CBORTag tags must be positive integers less than 2**64");
} // otherwise must be some other exception probably type err
return -1;
} // otherwise it's 2**64-1 which is fine :)
}
self->tag = tag;
if (value) {
tmp = self->value;
Py_INCREF(value);
self->value = value;
Py_XDECREF(tmp);
}
return 0;
}
// Special methods ///////////////////////////////////////////////////////////
static PyObject *
CBORTag_repr(CBORTagObject *self)
{
PyObject *ret = NULL;
if (Py_ReprEnter((PyObject *)self))
ret = PyUnicode_FromString("...");
else
ret = PyUnicode_FromFormat("CBORTag(%llu, %R)", self->tag, self->value);
Py_ReprLeave((PyObject *)self);
return ret;
}
static PyObject *
CBORTag_richcompare(PyObject *aobj, PyObject *bobj, int op)
{
PyObject *ret = NULL;
CBORTagObject *a, *b;
if (!(CBORTag_CheckExact(aobj) && CBORTag_CheckExact(bobj))) {
Py_RETURN_NOTIMPLEMENTED;
} else {
a = (CBORTagObject *)aobj;
b = (CBORTagObject *)bobj;
if (a == b) {
// Special case: both are the same object
switch (op) {
case Py_EQ: case Py_LE: case Py_GE: ret = Py_True; break;
case Py_NE: case Py_LT: case Py_GT: ret = Py_False; break;
default: assert(0);
}
Py_INCREF(ret);
} else if (a->tag == b->tag) {
// Tags are equal, rich-compare the value
ret = PyObject_RichCompare(a->value, b->value, op);
} else {
// Tags differ; simple integer comparison
switch (op) {
case Py_EQ: ret = Py_False; break;
case Py_NE: ret = Py_True; break;
case Py_LT: ret = a->tag < b->tag ? Py_True : Py_False; break;
case Py_LE: ret = a->tag <= b->tag ? Py_True : Py_False; break;
case Py_GE: ret = a->tag >= b->tag ? Py_True : Py_False; break;
case Py_GT: ret = a->tag > b->tag ? Py_True : Py_False; break;
default: assert(0);
}
Py_INCREF(ret);
}
}
return ret;
}
static Py_hash_t
CBORTag_hash(CBORTagObject *self)
{
if (!_CBOR2_thread_locals && _CBOR2_init_thread_locals() == -1)
return -1;
Py_hash_t ret = -1;
PyObject *running_hashes = NULL;
PyObject *tmp = NULL;
PyObject *self_id = PyLong_FromVoidPtr(self);
if (!self_id)
goto exit;
running_hashes = PyObject_GetAttrString(_CBOR2_thread_locals, "running_hashes");
if (!running_hashes) {
PyErr_Clear();
running_hashes = PySet_New(NULL);
if (PyObject_SetAttrString(_CBOR2_thread_locals, "running_hashes", running_hashes) == -1)
goto exit;
} else {
// Check if __hash__() is already being run against this object in this thread
switch (PySet_Contains(running_hashes, self_id)) {
case -1: // error
goto exit;
case 1: // this object is already in the set
PyErr_SetString(
PyExc_RuntimeError,
"This CBORTag is not hashable because it contains a reference to itself"
);
goto exit;
}
}
// Add id(self) to thread_locals.running_hashes
if (PySet_Add(running_hashes, self_id) == -1)
goto exit;
tmp = Py_BuildValue("(KO)", self->tag, self->value);
if (!tmp)
goto exit;
ret = PyObject_Hash(tmp);
if (ret == -1)
goto exit;
// Remove id(self) from thread_locals.running_hashes
if (PySet_Discard(running_hashes, self_id) == -1) {
ret = -1;
goto exit;
}
// Check how many more references there are in running_hashes
Py_ssize_t length = PySequence_Length(running_hashes);
if (length == -1) {
ret = -1;
goto exit;
}
// If this was the last reference, delete running_hashes from the thread-local variable
if (length == 0 && PyObject_DelAttrString(_CBOR2_thread_locals, "running_hashes") == -1) {
ret = -1;
goto exit;
}
exit:
Py_CLEAR(self_id);
Py_CLEAR(running_hashes);
Py_CLEAR(tmp);
return ret;
}
// C API /////////////////////////////////////////////////////////////////////
PyObject *
CBORTag_New(uint64_t tag)
{
CBORTagObject *ret = NULL;
ret = PyObject_GC_New(CBORTagObject, &CBORTagType);
if (ret) {
ret->tag = tag;
Py_INCREF(Py_None);
ret->value = Py_None;
}
return (PyObject *)ret;
}
int
CBORTag_SetValue(PyObject *tag, PyObject *value)
{
PyObject *tmp;
CBORTagObject *self;
if (!CBORTag_CheckExact(tag))
return -1;
if (!value)
return -1;
self = (CBORTagObject*)tag;
tmp = self->value;
Py_INCREF(value);
self->value = value;
Py_XDECREF(tmp);
return 0;
}
// Tag class definition //////////////////////////////////////////////////////
static PyMemberDef CBORTag_members[] = {
{"tag", T_ULONGLONG, offsetof(CBORTagObject, tag), 0,
"the semantic tag associated with the value"},
{"value", T_OBJECT_EX, offsetof(CBORTagObject, value), 0,
"the tagged value"},
{NULL}
};
PyDoc_STRVAR(CBORTag__doc__,
"The CBORTag class represents a semantically tagged value in a CBOR\n"
"encoded stream. The :attr:`tag` attribute holds the numeric tag\n"
"associated with the stored :attr:`value`.\n"
);
PyTypeObject CBORTagType = {
PyVarObject_HEAD_INIT(NULL, 0)
.tp_name = "_cbor2.CBORTag",
.tp_doc = CBORTag__doc__,
.tp_basicsize = sizeof(CBORTagObject),
.tp_itemsize = 0,
.tp_flags = Py_TPFLAGS_DEFAULT | Py_TPFLAGS_HAVE_GC,
.tp_new = CBORTag_new,
.tp_init = (initproc) CBORTag_init,
.tp_dealloc = (destructor) CBORTag_dealloc,
.tp_traverse = (traverseproc) CBORTag_traverse,
.tp_clear = (inquiry) CBORTag_clear,
.tp_members = CBORTag_members,
.tp_repr = (reprfunc) CBORTag_repr,
.tp_hash = (hashfunc) CBORTag_hash,
.tp_richcompare = CBORTag_richcompare,
};
|