Changeset 391 for python/trunk/Modules/_multiprocessing
- Timestamp:
- Mar 19, 2014, 11:31:01 PM (11 years ago)
- Location:
- python/trunk
- Files:
-
- 8 edited
Legend:
- Unmodified
- Added
- Removed
-
python/trunk
-
Property svn:mergeinfo
set to
/python/vendor/Python-2.7.6 merged eligible /python/vendor/current merged eligible
-
Property svn:mergeinfo
set to
-
python/trunk/Modules/_multiprocessing/connection.h
r2 r391 1 1 /* 2 * Definition of a `Connection` type. 2 * Definition of a `Connection` type. 3 3 * Used by `socket_connection.c` and `pipe_connection.c`. 4 4 * … … 20 20 #define CHECK_READABLE(self) \ 21 21 if (!(self->flags & READABLE)) { \ 22 23 22 PyErr_SetString(PyExc_IOError, "connection is write-only"); \ 23 return NULL; \ 24 24 } 25 25 26 26 #define CHECK_WRITABLE(self) \ 27 27 if (!(self->flags & WRITABLE)) { \ 28 29 28 PyErr_SetString(PyExc_IOError, "connection is read-only"); \ 29 return NULL; \ 30 30 } 31 31 … … 37 37 connection_new(PyTypeObject *type, PyObject *args, PyObject *kwds) 38 38 { 39 40 41 42 43 44 45 if (!PyArg_ParseTupleAndKeywords(args, kwds, F_HANDLE "|ii", kwlist, 46 47 48 49 50 51 52 53 54 55 56 PyErr_SetString(PyExc_ValueError, 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 39 ConnectionObject *self; 40 HANDLE handle; 41 BOOL readable = TRUE, writable = TRUE; 42 43 static char *kwlist[] = {"handle", "readable", "writable", NULL}; 44 45 if (!PyArg_ParseTupleAndKeywords(args, kwds, F_HANDLE "|ii", kwlist, 46 &handle, &readable, &writable)) 47 return NULL; 48 49 if (handle == INVALID_HANDLE_VALUE || (Py_ssize_t)handle < 0) { 50 PyErr_Format(PyExc_IOError, "invalid handle %zd", 51 (Py_ssize_t)handle); 52 return NULL; 53 } 54 55 if (!readable && !writable) { 56 PyErr_SetString(PyExc_ValueError, 57 "either readable or writable must be true"); 58 return NULL; 59 } 60 61 self = PyObject_New(ConnectionObject, type); 62 if (self == NULL) 63 return NULL; 64 65 self->weakreflist = NULL; 66 self->handle = handle; 67 self->flags = 0; 68 69 if (readable) 70 self->flags |= READABLE; 71 if (writable) 72 self->flags |= WRITABLE; 73 assert(self->flags >= 1 && self->flags <= 3); 74 75 return (PyObject*)self; 76 76 } 77 77 … … 79 79 connection_dealloc(ConnectionObject* self) 80 80 { 81 82 83 84 85 86 87 88 89 81 if (self->weakreflist != NULL) 82 PyObject_ClearWeakRefs((PyObject*)self); 83 84 if (self->handle != INVALID_HANDLE_VALUE) { 85 Py_BEGIN_ALLOW_THREADS 86 CLOSE(self->handle); 87 Py_END_ALLOW_THREADS 88 } 89 PyObject_Del(self); 90 90 } 91 91 … … 97 97 connection_sendbytes(ConnectionObject *self, PyObject *args) 98 98 { 99 char *buffer; 100 Py_ssize_t length, offset=0, size=PY_SSIZE_T_MIN; 101 int res; 102 103 if (!PyArg_ParseTuple(args, F_RBUFFER "#|" F_PY_SSIZE_T F_PY_SSIZE_T, 104 &buffer, &length, &offset, &size)) 105 return NULL; 106 107 CHECK_WRITABLE(self); 108 109 if (offset < 0) { 110 PyErr_SetString(PyExc_ValueError, "offset is negative"); 111 return NULL; 112 } 113 if (length < offset) { 114 PyErr_SetString(PyExc_ValueError, "buffer length < offset"); 115 return NULL; 116 } 117 118 if (size == PY_SSIZE_T_MIN) { 119 size = length - offset; 120 } else { 121 if (size < 0) { 122 PyErr_SetString(PyExc_ValueError, "size is negative"); 123 return NULL; 124 } 125 if (offset + size > length) { 126 PyErr_SetString(PyExc_ValueError, 127 "buffer length < offset + size"); 128 return NULL; 129 } 130 } 131 132 res = conn_send_string(self, buffer + offset, size); 133 134 if (res < 0) 135 return mp_SetError(PyExc_IOError, res); 136 137 Py_RETURN_NONE; 138 } 139 140 static PyObject * 141 connection_recvbytes(ConnectionObject *self, PyObject *args) 142 { 143 char *freeme = NULL; 144 Py_ssize_t res, maxlength = PY_SSIZE_T_MAX; 145 PyObject *result = NULL; 146 147 if (!PyArg_ParseTuple(args, "|" F_PY_SSIZE_T, &maxlength)) 148 return NULL; 149 150 CHECK_READABLE(self); 151 152 if (maxlength < 0) { 153 PyErr_SetString(PyExc_ValueError, "maxlength < 0"); 154 return NULL; 155 } 156 157 res = conn_recv_string(self, self->buffer, CONNECTION_BUFFER_SIZE, 158 &freeme, maxlength); 159 160 if (res < 0) { 161 if (res == MP_BAD_MESSAGE_LENGTH) { 162 if ((self->flags & WRITABLE) == 0) { 163 Py_BEGIN_ALLOW_THREADS 164 CLOSE(self->handle); 165 Py_END_ALLOW_THREADS 166 self->handle = INVALID_HANDLE_VALUE; 167 } else { 168 self->flags = WRITABLE; 169 } 170 } 171 mp_SetError(PyExc_IOError, res); 172 } else { 173 if (freeme == NULL) { 174 result = PyString_FromStringAndSize(self->buffer, res); 175 } else { 176 result = PyString_FromStringAndSize(freeme, res); 177 PyMem_Free(freeme); 178 } 179 } 180 181 return result; 182 } 183 184 static PyObject * 185 connection_recvbytes_into(ConnectionObject *self, PyObject *args) 186 { 187 char *freeme = NULL, *buffer = NULL; 188 Py_ssize_t res, length, offset = 0; 189 PyObject *result = NULL; 190 Py_buffer pbuf; 191 192 CHECK_READABLE(self); 193 194 if (!PyArg_ParseTuple(args, "w*|" F_PY_SSIZE_T, 195 &pbuf, &offset)) 196 return NULL; 197 198 buffer = pbuf.buf; 199 length = pbuf.len; 200 201 if (offset < 0) { 202 PyErr_SetString(PyExc_ValueError, "negative offset"); 203 goto _error; 204 } 205 206 if (offset > length) { 207 PyErr_SetString(PyExc_ValueError, "offset too large"); 208 goto _error; 209 } 210 211 res = conn_recv_string(self, buffer+offset, length-offset, 212 &freeme, PY_SSIZE_T_MAX); 213 214 if (res < 0) { 215 if (res == MP_BAD_MESSAGE_LENGTH) { 216 if ((self->flags & WRITABLE) == 0) { 217 Py_BEGIN_ALLOW_THREADS 218 CLOSE(self->handle); 219 Py_END_ALLOW_THREADS 220 self->handle = INVALID_HANDLE_VALUE; 221 } else { 222 self->flags = WRITABLE; 223 } 224 } 225 mp_SetError(PyExc_IOError, res); 226 } else { 227 if (freeme == NULL) { 228 result = PyInt_FromSsize_t(res); 229 } else { 230 result = PyObject_CallFunction(BufferTooShort, 231 F_RBUFFER "#", 232 freeme, res); 233 PyMem_Free(freeme); 234 if (result) { 235 PyErr_SetObject(BufferTooShort, result); 236 Py_DECREF(result); 237 } 238 goto _error; 239 } 240 } 99 char *buffer; 100 Py_ssize_t length, offset=0, size=PY_SSIZE_T_MIN; 101 int res; 102 103 if (!PyArg_ParseTuple(args, F_RBUFFER "#|" F_PY_SSIZE_T F_PY_SSIZE_T, 104 &buffer, &length, &offset, &size)) 105 return NULL; 106 107 CHECK_WRITABLE(self); 108 109 if (offset < 0) { 110 PyErr_SetString(PyExc_ValueError, "offset is negative"); 111 return NULL; 112 } 113 if (length < offset) { 114 PyErr_SetString(PyExc_ValueError, "buffer length < offset"); 115 return NULL; 116 } 117 118 if (size == PY_SSIZE_T_MIN) { 119 size = length - offset; 120 } else { 121 if (size < 0) { 122 PyErr_SetString(PyExc_ValueError, "size is negative"); 123 return NULL; 124 } 125 if (offset + size > length) { 126 PyErr_SetString(PyExc_ValueError, 127 "buffer length < offset + size"); 128 return NULL; 129 } 130 } 131 132 res = conn_send_string(self, buffer + offset, size); 133 134 if (res < 0) { 135 if (PyErr_Occurred()) 136 return NULL; 137 else 138 return mp_SetError(PyExc_IOError, res); 139 } 140 141 Py_RETURN_NONE; 142 } 143 144 static PyObject * 145 connection_recvbytes(ConnectionObject *self, PyObject *args) 146 { 147 char *freeme = NULL; 148 Py_ssize_t res, maxlength = PY_SSIZE_T_MAX; 149 PyObject *result = NULL; 150 151 if (!PyArg_ParseTuple(args, "|" F_PY_SSIZE_T, &maxlength)) 152 return NULL; 153 154 CHECK_READABLE(self); 155 156 if (maxlength < 0) { 157 PyErr_SetString(PyExc_ValueError, "maxlength < 0"); 158 return NULL; 159 } 160 161 res = conn_recv_string(self, self->buffer, CONNECTION_BUFFER_SIZE, 162 &freeme, maxlength); 163 164 if (res < 0) { 165 if (res == MP_BAD_MESSAGE_LENGTH) { 166 if ((self->flags & WRITABLE) == 0) { 167 Py_BEGIN_ALLOW_THREADS 168 CLOSE(self->handle); 169 Py_END_ALLOW_THREADS 170 self->handle = INVALID_HANDLE_VALUE; 171 } else { 172 self->flags = WRITABLE; 173 } 174 } 175 mp_SetError(PyExc_IOError, res); 176 } else { 177 if (freeme == NULL) { 178 result = PyString_FromStringAndSize(self->buffer, res); 179 } else { 180 result = PyString_FromStringAndSize(freeme, res); 181 PyMem_Free(freeme); 182 } 183 } 184 185 return result; 186 } 187 188 static PyObject * 189 connection_recvbytes_into(ConnectionObject *self, PyObject *args) 190 { 191 char *freeme = NULL, *buffer = NULL; 192 Py_ssize_t res, length, offset = 0; 193 PyObject *result = NULL; 194 Py_buffer pbuf; 195 196 CHECK_READABLE(self); 197 198 if (!PyArg_ParseTuple(args, "w*|" F_PY_SSIZE_T, 199 &pbuf, &offset)) 200 return NULL; 201 202 buffer = pbuf.buf; 203 length = pbuf.len; 204 205 if (offset < 0) { 206 PyErr_SetString(PyExc_ValueError, "negative offset"); 207 goto _error; 208 } 209 210 if (offset > length) { 211 PyErr_SetString(PyExc_ValueError, "offset too large"); 212 goto _error; 213 } 214 215 res = conn_recv_string(self, buffer+offset, length-offset, 216 &freeme, PY_SSIZE_T_MAX); 217 218 if (res < 0) { 219 if (res == MP_BAD_MESSAGE_LENGTH) { 220 if ((self->flags & WRITABLE) == 0) { 221 Py_BEGIN_ALLOW_THREADS 222 CLOSE(self->handle); 223 Py_END_ALLOW_THREADS 224 self->handle = INVALID_HANDLE_VALUE; 225 } else { 226 self->flags = WRITABLE; 227 } 228 } 229 mp_SetError(PyExc_IOError, res); 230 } else { 231 if (freeme == NULL) { 232 result = PyInt_FromSsize_t(res); 233 } else { 234 result = PyObject_CallFunction(BufferTooShort, 235 F_RBUFFER "#", 236 freeme, res); 237 PyMem_Free(freeme); 238 if (result) { 239 PyErr_SetObject(BufferTooShort, result); 240 Py_DECREF(result); 241 } 242 goto _error; 243 } 244 } 241 245 242 246 _cleanup: 243 244 247 PyBuffer_Release(&pbuf); 248 return result; 245 249 246 250 _error: 247 248 251 result = NULL; 252 goto _cleanup; 249 253 } 250 254 … … 256 260 connection_send_obj(ConnectionObject *self, PyObject *obj) 257 261 { 258 259 260 261 262 263 264 265 pickled_string = PyObject_CallFunctionObjArgs(pickle_dumps, obj, 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 262 char *buffer; 263 int res; 264 Py_ssize_t length; 265 PyObject *pickled_string = NULL; 266 267 CHECK_WRITABLE(self); 268 269 pickled_string = PyObject_CallFunctionObjArgs(pickle_dumps, obj, 270 pickle_protocol, NULL); 271 if (!pickled_string) 272 goto failure; 273 274 if (PyString_AsStringAndSize(pickled_string, &buffer, &length) < 0) 275 goto failure; 276 277 res = conn_send_string(self, buffer, (int)length); 278 279 if (res < 0) { 280 mp_SetError(PyExc_IOError, res); 281 goto failure; 282 } 283 284 Py_XDECREF(pickled_string); 285 Py_RETURN_NONE; 282 286 283 287 failure: 284 285 288 Py_XDECREF(pickled_string); 289 return NULL; 286 290 } 287 291 … … 289 293 connection_recv_obj(ConnectionObject *self) 290 294 { 291 292 293 294 295 296 297 res = conn_recv_string(self, self->buffer, CONNECTION_BUFFER_SIZE, 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 } else { 313 314 315 316 317 318 319 320 321 322 result = PyObject_CallFunctionObjArgs(pickle_loads, 323 324 325 295 char *freeme = NULL; 296 Py_ssize_t res; 297 PyObject *temp = NULL, *result = NULL; 298 299 CHECK_READABLE(self); 300 301 res = conn_recv_string(self, self->buffer, CONNECTION_BUFFER_SIZE, 302 &freeme, PY_SSIZE_T_MAX); 303 304 if (res < 0) { 305 if (res == MP_BAD_MESSAGE_LENGTH) { 306 if ((self->flags & WRITABLE) == 0) { 307 Py_BEGIN_ALLOW_THREADS 308 CLOSE(self->handle); 309 Py_END_ALLOW_THREADS 310 self->handle = INVALID_HANDLE_VALUE; 311 } else { 312 self->flags = WRITABLE; 313 } 314 } 315 mp_SetError(PyExc_IOError, res); 316 } else { 317 if (freeme == NULL) { 318 temp = PyString_FromStringAndSize(self->buffer, res); 319 } else { 320 temp = PyString_FromStringAndSize(freeme, res); 321 PyMem_Free(freeme); 322 } 323 } 324 325 if (temp) 326 result = PyObject_CallFunctionObjArgs(pickle_loads, 327 temp, NULL); 328 Py_XDECREF(temp); 329 return result; 326 330 } 327 331 … … 333 337 connection_poll(ConnectionObject *self, PyObject *args) 334 338 { 335 336 337 338 339 340 341 342 343 344 345 346 347 timeout = -1.0;/* block forever */348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 339 PyObject *timeout_obj = NULL; 340 double timeout = 0.0; 341 int res; 342 343 CHECK_READABLE(self); 344 345 if (!PyArg_ParseTuple(args, "|O", &timeout_obj)) 346 return NULL; 347 348 if (timeout_obj == NULL) { 349 timeout = 0.0; 350 } else if (timeout_obj == Py_None) { 351 timeout = -1.0; /* block forever */ 352 } else { 353 timeout = PyFloat_AsDouble(timeout_obj); 354 if (PyErr_Occurred()) 355 return NULL; 356 if (timeout < 0.0) 357 timeout = 0.0; 358 } 359 360 Py_BEGIN_ALLOW_THREADS 361 res = conn_poll(self, timeout, _save); 362 Py_END_ALLOW_THREADS 363 364 switch (res) { 365 case TRUE: 366 Py_RETURN_TRUE; 367 case FALSE: 368 Py_RETURN_FALSE; 369 default: 370 return mp_SetError(PyExc_IOError, res); 371 } 368 372 } 369 373 … … 371 375 connection_fileno(ConnectionObject* self) 372 376 { 373 374 375 376 377 377 if (self->handle == INVALID_HANDLE_VALUE) { 378 PyErr_SetString(PyExc_IOError, "handle is invalid"); 379 return NULL; 380 } 381 return PyInt_FromLong((long)self->handle); 378 382 } 379 383 … … 381 385 connection_close(ConnectionObject *self) 382 386 { 383 384 385 386 387 388 389 390 387 if (self->handle != INVALID_HANDLE_VALUE) { 388 Py_BEGIN_ALLOW_THREADS 389 CLOSE(self->handle); 390 Py_END_ALLOW_THREADS 391 self->handle = INVALID_HANDLE_VALUE; 392 } 393 394 Py_RETURN_NONE; 391 395 } 392 396 … … 394 398 connection_repr(ConnectionObject *self) 395 399 { 396 397 398 399 return FROM_FORMAT("<%s %s, handle %zd>", 400 401 400 static char *conn_type[] = {"read-only", "write-only", "read-write"}; 401 402 assert(self->flags >= 1 && self->flags <= 3); 403 return FROM_FORMAT("<%s %s, handle %zd>", 404 conn_type[self->flags - 1], 405 CONNECTION_NAME, (Py_ssize_t)self->handle); 402 406 } 403 407 … … 409 413 connection_closed(ConnectionObject *self, void *closure) 410 414 { 411 415 return PyBool_FromLong((long)(self->handle == INVALID_HANDLE_VALUE)); 412 416 } 413 417 … … 415 419 connection_readable(ConnectionObject *self, void *closure) 416 420 { 417 421 return PyBool_FromLong((long)(self->flags & READABLE)); 418 422 } 419 423 … … 421 425 connection_writable(ConnectionObject *self, void *closure) 422 426 { 423 427 return PyBool_FromLong((long)(self->flags & WRITABLE)); 424 428 } 425 429 … … 429 433 430 434 static PyMethodDef connection_methods[] = { 431 {"send_bytes", (PyCFunction)connection_sendbytes, METH_VARARGS, 432 433 {"recv_bytes", (PyCFunction)connection_recvbytes, METH_VARARGS, 434 435 436 437 438 439 {"send", (PyCFunction)connection_send_obj, METH_O, 440 441 {"recv", (PyCFunction)connection_recv_obj, METH_NOARGS, 442 443 444 {"poll", (PyCFunction)connection_poll, METH_VARARGS, 445 446 447 448 449 450 451 435 {"send_bytes", (PyCFunction)connection_sendbytes, METH_VARARGS, 436 "send the byte data from a readable buffer-like object"}, 437 {"recv_bytes", (PyCFunction)connection_recvbytes, METH_VARARGS, 438 "receive byte data as a string"}, 439 {"recv_bytes_into",(PyCFunction)connection_recvbytes_into,METH_VARARGS, 440 "receive byte data into a writeable buffer-like object\n" 441 "returns the number of bytes read"}, 442 443 {"send", (PyCFunction)connection_send_obj, METH_O, 444 "send a (picklable) object"}, 445 {"recv", (PyCFunction)connection_recv_obj, METH_NOARGS, 446 "receive a (picklable) object"}, 447 448 {"poll", (PyCFunction)connection_poll, METH_VARARGS, 449 "whether there is any input available to be read"}, 450 {"fileno", (PyCFunction)connection_fileno, METH_NOARGS, 451 "file descriptor or handle of the connection"}, 452 {"close", (PyCFunction)connection_close, METH_NOARGS, 453 "close the connection"}, 454 455 {NULL} /* Sentinel */ 452 456 }; 453 457 454 458 static PyGetSetDef connection_getset[] = { 455 {"closed", (getter)connection_closed, NULL, 456 457 {"readable", (getter)connection_readable, NULL, 458 459 {"writable", (getter)connection_writable, NULL, 460 461 459 {"closed", (getter)connection_closed, NULL, 460 "True if the connection is closed", NULL}, 461 {"readable", (getter)connection_readable, NULL, 462 "True if the connection is readable", NULL}, 463 {"writable", (getter)connection_writable, NULL, 464 "True if the connection is writable", NULL}, 465 {NULL} 462 466 }; 463 467 … … 467 471 468 472 PyDoc_STRVAR(connection_doc, 469 470 471 473 "Connection type whose constructor signature is\n\n" 474 " Connection(handle, readable=True, writable=True).\n\n" 475 "The constructor does *not* duplicate the handle."); 472 476 473 477 PyTypeObject CONNECTION_TYPE = { 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 /* tp_flags */ Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 478 PyVarObject_HEAD_INIT(NULL, 0) 479 /* tp_name */ "_multiprocessing." CONNECTION_NAME, 480 /* tp_basicsize */ sizeof(ConnectionObject), 481 /* tp_itemsize */ 0, 482 /* tp_dealloc */ (destructor)connection_dealloc, 483 /* tp_print */ 0, 484 /* tp_getattr */ 0, 485 /* tp_setattr */ 0, 486 /* tp_compare */ 0, 487 /* tp_repr */ (reprfunc)connection_repr, 488 /* tp_as_number */ 0, 489 /* tp_as_sequence */ 0, 490 /* tp_as_mapping */ 0, 491 /* tp_hash */ 0, 492 /* tp_call */ 0, 493 /* tp_str */ 0, 494 /* tp_getattro */ 0, 495 /* tp_setattro */ 0, 496 /* tp_as_buffer */ 0, 497 /* tp_flags */ Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | 498 Py_TPFLAGS_HAVE_WEAKREFS, 499 /* tp_doc */ connection_doc, 500 /* tp_traverse */ 0, 501 /* tp_clear */ 0, 502 /* tp_richcompare */ 0, 503 /* tp_weaklistoffset */ offsetof(ConnectionObject, weakreflist), 504 /* tp_iter */ 0, 505 /* tp_iternext */ 0, 506 /* tp_methods */ connection_methods, 507 /* tp_members */ 0, 508 /* tp_getset */ connection_getset, 509 /* tp_base */ 0, 510 /* tp_dict */ 0, 511 /* tp_descr_get */ 0, 512 /* tp_descr_set */ 0, 513 /* tp_dictoffset */ 0, 514 /* tp_init */ 0, 515 /* tp_alloc */ 0, 516 /* tp_new */ connection_new, 513 517 }; 514 518 -
python/trunk/Modules/_multiprocessing/multiprocessing.c
r10 r391 9 9 #include "multiprocessing.h" 10 10 11 #if (defined(CMSG_LEN) && defined(SCM_RIGHTS)) 12 #define HAVE_FD_TRANSFER 1 13 #else 14 #define HAVE_FD_TRANSFER 0 15 #endif 16 11 17 PyObject *create_win32_namespace(void); 12 18 … … 21 27 mp_SetError(PyObject *Type, int num) 22 28 { 23 29 switch (num) { 24 30 #ifdef MS_WINDOWS 25 case MP_STANDARD_ERROR: 26 27 28 29 30 31 32 33 34 31 case MP_STANDARD_ERROR: 32 if (Type == NULL) 33 Type = PyExc_WindowsError; 34 PyErr_SetExcFromWindowsErr(Type, 0); 35 break; 36 case MP_SOCKET_ERROR: 37 if (Type == NULL) 38 Type = PyExc_WindowsError; 39 PyErr_SetExcFromWindowsErr(Type, WSAGetLastError()); 40 break; 35 41 #else /* !MS_WINDOWS */ 36 37 38 39 40 41 42 case MP_STANDARD_ERROR: 43 case MP_SOCKET_ERROR: 44 if (Type == NULL) 45 Type = PyExc_OSError; 46 PyErr_SetFromErrno(Type); 47 break; 42 48 #endif /* !MS_WINDOWS */ 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 "unkown error number %d", num);61 62 49 case MP_MEMORY_ERROR: 50 PyErr_NoMemory(); 51 break; 52 case MP_END_OF_FILE: 53 PyErr_SetNone(PyExc_EOFError); 54 break; 55 case MP_EARLY_END_OF_FILE: 56 PyErr_SetString(PyExc_IOError, 57 "got end of file during message"); 58 break; 59 case MP_BAD_MESSAGE_LENGTH: 60 PyErr_SetString(PyExc_IOError, "bad message length"); 61 break; 62 case MP_EXCEPTION_HAS_BEEN_SET: 63 break; 64 default: 65 PyErr_Format(PyExc_RuntimeError, 66 "unknown error number %d", num); 67 } 68 return NULL; 63 69 } 64 70 … … 77 83 ProcessingCtrlHandler(DWORD dwCtrlType) 78 84 { 79 80 85 SetEvent(sigint_event); 86 return FALSE; 81 87 } 82 88 … … 92 98 Reimplements some of the functionality of the fdcred 93 99 module at http://www.mca-ltd.com/resources/fdcred_1.tgz. */ 100 /* Based in http://resin.csoft.net/cgi-bin/man.cgi?section=3&topic=CMSG_DATA */ 94 101 95 102 static PyObject * 96 103 multiprocessing_sendfd(PyObject *self, PyObject *args) 97 104 { 98 int conn, fd, res; 99 char dummy_char; 100 char buf[CMSG_SPACE(sizeof(int))]; 101 struct msghdr msg = {0}; 102 struct iovec dummy_iov; 103 struct cmsghdr *cmsg; 104 105 if (!PyArg_ParseTuple(args, "ii", &conn, &fd)) 106 return NULL; 107 108 dummy_iov.iov_base = &dummy_char; 109 dummy_iov.iov_len = 1; 110 msg.msg_control = buf; 111 msg.msg_controllen = sizeof(buf); 112 msg.msg_iov = &dummy_iov; 113 msg.msg_iovlen = 1; 114 cmsg = CMSG_FIRSTHDR(&msg); 115 cmsg->cmsg_level = SOL_SOCKET; 116 cmsg->cmsg_type = SCM_RIGHTS; 117 cmsg->cmsg_len = CMSG_LEN(sizeof(int)); 118 msg.msg_controllen = cmsg->cmsg_len; 119 *(int*)CMSG_DATA(cmsg) = fd; 120 121 Py_BEGIN_ALLOW_THREADS 122 res = sendmsg(conn, &msg, 0); 123 Py_END_ALLOW_THREADS 124 125 if (res < 0) 126 return PyErr_SetFromErrno(PyExc_OSError); 127 Py_RETURN_NONE; 105 int conn, fd, res; 106 struct iovec dummy_iov; 107 char dummy_char; 108 struct msghdr msg; 109 struct cmsghdr *cmsg; 110 union { 111 struct cmsghdr hdr; 112 unsigned char buf[CMSG_SPACE(sizeof(int))]; 113 } cmsgbuf; 114 115 if (!PyArg_ParseTuple(args, "ii", &conn, &fd)) 116 return NULL; 117 118 dummy_iov.iov_base = &dummy_char; 119 dummy_iov.iov_len = 1; 120 121 memset(&msg, 0, sizeof(msg)); 122 msg.msg_control = &cmsgbuf.buf; 123 msg.msg_controllen = sizeof(cmsgbuf.buf); 124 msg.msg_iov = &dummy_iov; 125 msg.msg_iovlen = 1; 126 127 cmsg = CMSG_FIRSTHDR(&msg); 128 cmsg->cmsg_len = CMSG_LEN(sizeof(int)); 129 cmsg->cmsg_level = SOL_SOCKET; 130 cmsg->cmsg_type = SCM_RIGHTS; 131 * (int *) CMSG_DATA(cmsg) = fd; 132 133 Py_BEGIN_ALLOW_THREADS 134 res = sendmsg(conn, &msg, 0); 135 Py_END_ALLOW_THREADS 136 137 if (res < 0) 138 return PyErr_SetFromErrno(PyExc_OSError); 139 Py_RETURN_NONE; 128 140 } 129 141 … … 131 143 multiprocessing_recvfd(PyObject *self, PyObject *args) 132 144 { 133 int conn, fd, res; 134 char dummy_char; 135 char buf[CMSG_SPACE(sizeof(int))]; 136 struct msghdr msg = {0}; 137 struct iovec dummy_iov; 138 struct cmsghdr *cmsg; 139 140 if (!PyArg_ParseTuple(args, "i", &conn)) 141 return NULL; 142 143 dummy_iov.iov_base = &dummy_char; 144 dummy_iov.iov_len = 1; 145 msg.msg_control = buf; 146 msg.msg_controllen = sizeof(buf); 147 msg.msg_iov = &dummy_iov; 148 msg.msg_iovlen = 1; 149 cmsg = CMSG_FIRSTHDR(&msg); 150 cmsg->cmsg_level = SOL_SOCKET; 151 cmsg->cmsg_type = SCM_RIGHTS; 152 cmsg->cmsg_len = CMSG_LEN(sizeof(int)); 153 msg.msg_controllen = cmsg->cmsg_len; 154 155 Py_BEGIN_ALLOW_THREADS 156 res = recvmsg(conn, &msg, 0); 157 Py_END_ALLOW_THREADS 158 159 if (res < 0) 160 return PyErr_SetFromErrno(PyExc_OSError); 161 162 fd = *(int*)CMSG_DATA(cmsg); 163 return Py_BuildValue("i", fd); 145 int conn, fd, res; 146 char dummy_char; 147 struct iovec dummy_iov; 148 struct msghdr msg = {0}; 149 struct cmsghdr *cmsg; 150 union { 151 struct cmsghdr hdr; 152 unsigned char buf[CMSG_SPACE(sizeof(int))]; 153 } cmsgbuf; 154 155 if (!PyArg_ParseTuple(args, "i", &conn)) 156 return NULL; 157 158 dummy_iov.iov_base = &dummy_char; 159 dummy_iov.iov_len = 1; 160 161 memset(&msg, 0, sizeof(msg)); 162 msg.msg_control = &cmsgbuf.buf; 163 msg.msg_controllen = sizeof(cmsgbuf.buf); 164 msg.msg_iov = &dummy_iov; 165 msg.msg_iovlen = 1; 166 167 cmsg = CMSG_FIRSTHDR(&msg); 168 cmsg->cmsg_level = SOL_SOCKET; 169 cmsg->cmsg_type = SCM_RIGHTS; 170 cmsg->cmsg_len = CMSG_LEN(sizeof(int)); 171 msg.msg_controllen = cmsg->cmsg_len; 172 173 Py_BEGIN_ALLOW_THREADS 174 res = recvmsg(conn, &msg, 0); 175 Py_END_ALLOW_THREADS 176 177 if (res < 0) 178 return PyErr_SetFromErrno(PyExc_OSError); 179 180 if (msg.msg_controllen < CMSG_LEN(sizeof(int)) || 181 (cmsg = CMSG_FIRSTHDR(&msg)) == NULL || 182 cmsg->cmsg_level != SOL_SOCKET || 183 cmsg->cmsg_type != SCM_RIGHTS || 184 cmsg->cmsg_len < CMSG_LEN(sizeof(int))) { 185 /* If at least one control message is present, there should be 186 no room for any further data in the buffer. */ 187 PyErr_SetString(PyExc_RuntimeError, "No file descriptor received"); 188 return NULL; 189 } 190 191 fd = * (int *) CMSG_DATA(cmsg); 192 return Py_BuildValue("i", fd); 164 193 } 165 194 … … 176 205 multiprocessing_address_of_buffer(PyObject *self, PyObject *obj) 177 206 { 178 179 180 181 182 183 184 return Py_BuildValue("N" F_PY_SSIZE_T, 185 207 void *buffer; 208 Py_ssize_t buffer_len; 209 210 if (PyObject_AsWriteBuffer(obj, &buffer, &buffer_len) < 0) 211 return NULL; 212 213 return Py_BuildValue("N" F_PY_SSIZE_T, 214 PyLong_FromVoidPtr(buffer), buffer_len); 186 215 } 187 216 … … 192 221 193 222 static PyMethodDef module_methods[] = { 194 {"address_of_buffer", multiprocessing_address_of_buffer, METH_O, 195 "address_of_buffer(obj) -> int\n" 196 223 {"address_of_buffer", multiprocessing_address_of_buffer, METH_O, 224 "address_of_buffer(obj) -> int\n" 225 "Return address of obj assuming obj supports buffer inteface"}, 197 226 #if HAVE_FD_TRANSFER 198 {"sendfd", multiprocessing_sendfd, METH_VARARGS, 199 200 201 202 203 204 205 206 #endif 207 227 {"sendfd", multiprocessing_sendfd, METH_VARARGS, 228 "sendfd(sockfd, fd) -> None\n" 229 "Send file descriptor given by fd over the unix domain socket\n" 230 "whose file decriptor is sockfd"}, 231 {"recvfd", multiprocessing_recvfd, METH_VARARGS, 232 "recvfd(sockfd) -> fd\n" 233 "Receive a file descriptor over a unix domain socket\n" 234 "whose file decriptor is sockfd"}, 235 #endif 236 {NULL} 208 237 }; 209 238 … … 216 245 init_multiprocessing(void) 217 246 { 218 PyObject *module, *temp, *value; 219 220 /* Initialize module */ 221 module = Py_InitModule("_multiprocessing", module_methods); 222 if (!module) 223 return; 224 225 /* Get copy of objects from pickle */ 226 temp = PyImport_ImportModule(PICKLE_MODULE); 227 if (!temp) 228 return; 229 pickle_dumps = PyObject_GetAttrString(temp, "dumps"); 230 pickle_loads = PyObject_GetAttrString(temp, "loads"); 231 pickle_protocol = PyObject_GetAttrString(temp, "HIGHEST_PROTOCOL"); 232 Py_XDECREF(temp); 233 234 /* Get copy of BufferTooShort */ 235 temp = PyImport_ImportModule("multiprocessing"); 236 if (!temp) 237 return; 238 BufferTooShort = PyObject_GetAttrString(temp, "BufferTooShort"); 239 Py_XDECREF(temp); 240 241 /* Add connection type to module */ 242 if (PyType_Ready(&ConnectionType) < 0) 243 return; 244 Py_INCREF(&ConnectionType); 245 PyModule_AddObject(module, "Connection", (PyObject*)&ConnectionType); 246 247 #if defined(MS_WINDOWS) || HAVE_SEM_OPEN 248 /* Add SemLock type to module */ 249 if (PyType_Ready(&SemLockType) < 0) 250 return; 251 Py_INCREF(&SemLockType); 252 PyDict_SetItemString(SemLockType.tp_dict, "SEM_VALUE_MAX", 253 Py_BuildValue("i", SEM_VALUE_MAX)); 254 PyModule_AddObject(module, "SemLock", (PyObject*)&SemLockType); 247 PyObject *module, *temp, *value; 248 249 /* Initialize module */ 250 module = Py_InitModule("_multiprocessing", module_methods); 251 if (!module) 252 return; 253 254 /* Get copy of objects from pickle */ 255 temp = PyImport_ImportModule(PICKLE_MODULE); 256 if (!temp) 257 return; 258 pickle_dumps = PyObject_GetAttrString(temp, "dumps"); 259 pickle_loads = PyObject_GetAttrString(temp, "loads"); 260 pickle_protocol = PyObject_GetAttrString(temp, "HIGHEST_PROTOCOL"); 261 Py_XDECREF(temp); 262 263 /* Get copy of BufferTooShort */ 264 temp = PyImport_ImportModule("multiprocessing"); 265 if (!temp) 266 return; 267 BufferTooShort = PyObject_GetAttrString(temp, "BufferTooShort"); 268 Py_XDECREF(temp); 269 270 /* Add connection type to module */ 271 if (PyType_Ready(&ConnectionType) < 0) 272 return; 273 Py_INCREF(&ConnectionType); 274 PyModule_AddObject(module, "Connection", (PyObject*)&ConnectionType); 275 276 #if defined(MS_WINDOWS) || \ 277 (defined(HAVE_SEM_OPEN) && !defined(POSIX_SEMAPHORES_NOT_ENABLED)) 278 /* Add SemLock type to module */ 279 if (PyType_Ready(&SemLockType) < 0) 280 return; 281 Py_INCREF(&SemLockType); 282 { 283 PyObject *py_sem_value_max; 284 /* Some systems define SEM_VALUE_MAX as an unsigned value that 285 * causes it to be negative when used as an int (NetBSD). */ 286 if ((int)(SEM_VALUE_MAX) < 0) 287 py_sem_value_max = PyLong_FromLong(INT_MAX); 288 else 289 py_sem_value_max = PyLong_FromLong(SEM_VALUE_MAX); 290 if (py_sem_value_max == NULL) 291 return; 292 PyDict_SetItemString(SemLockType.tp_dict, "SEM_VALUE_MAX", 293 py_sem_value_max); 294 } 295 PyModule_AddObject(module, "SemLock", (PyObject*)&SemLockType); 255 296 #endif 256 297 257 298 #ifdef MS_WINDOWS 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 #endif 282 283 284 285 286 287 #define ADD_FLAG(name) 288 value = Py_BuildValue("i", name);\289 if (value == NULL) { Py_DECREF(temp); return; }\290 if (PyDict_SetItemString(temp, #name, value) < 0) {\291 Py_DECREF(temp); Py_DECREF(value); return; }\292 293 294 #if def HAVE_SEM_OPEN295 299 /* Add PipeConnection to module */ 300 if (PyType_Ready(&PipeConnectionType) < 0) 301 return; 302 Py_INCREF(&PipeConnectionType); 303 PyModule_AddObject(module, "PipeConnection", 304 (PyObject*)&PipeConnectionType); 305 306 /* Initialize win32 class and add to multiprocessing */ 307 temp = create_win32_namespace(); 308 if (!temp) 309 return; 310 PyModule_AddObject(module, "win32", temp); 311 312 /* Initialize the event handle used to signal Ctrl-C */ 313 sigint_event = CreateEvent(NULL, TRUE, FALSE, NULL); 314 if (!sigint_event) { 315 PyErr_SetFromWindowsErr(0); 316 return; 317 } 318 if (!SetConsoleCtrlHandler(ProcessingCtrlHandler, TRUE)) { 319 PyErr_SetFromWindowsErr(0); 320 return; 321 } 322 #endif 323 324 /* Add configuration macros */ 325 temp = PyDict_New(); 326 if (!temp) 327 return; 328 #define ADD_FLAG(name) \ 329 value = Py_BuildValue("i", name); \ 330 if (value == NULL) { Py_DECREF(temp); return; } \ 331 if (PyDict_SetItemString(temp, #name, value) < 0) { \ 332 Py_DECREF(temp); Py_DECREF(value); return; } \ 333 Py_DECREF(value) 334 335 #if defined(HAVE_SEM_OPEN) && !defined(POSIX_SEMAPHORES_NOT_ENABLED) 336 ADD_FLAG(HAVE_SEM_OPEN); 296 337 #endif 297 338 #ifdef HAVE_SEM_TIMEDWAIT 298 339 ADD_FLAG(HAVE_SEM_TIMEDWAIT); 299 340 #endif 300 341 #ifdef HAVE_FD_TRANSFER 301 342 ADD_FLAG(HAVE_FD_TRANSFER); 302 343 #endif 303 344 #ifdef HAVE_BROKEN_SEM_GETVALUE 304 345 ADD_FLAG(HAVE_BROKEN_SEM_GETVALUE); 305 346 #endif 306 347 #ifdef HAVE_BROKEN_SEM_UNLINK 307 308 #endif 309 310 311 } 348 ADD_FLAG(HAVE_BROKEN_SEM_UNLINK); 349 #endif 350 if (PyModule_AddObject(module, "flags", temp) < 0) 351 return; 352 } -
python/trunk/Modules/_multiprocessing/multiprocessing.h
r10 r391 3 3 4 4 #define PY_SSIZE_T_CLEAN 5 6 #ifdef __sun 7 /* The control message API is only available on Solaris 8 if XPG 4.2 or later is requested. */ 9 #define _XOPEN_SOURCE 500 10 #endif 5 11 6 12 #include "Python.h" … … 16 22 # include <windows.h> 17 23 # include <winsock2.h> 18 # include <process.h> 24 # include <process.h> /* getpid() */ 19 25 # ifdef Py_DEBUG 20 26 # include <crtdbg.h> … … 28 34 # include <sys/uio.h> 29 35 # include <arpa/inet.h> /* htonl() and ntohl() */ 30 #ifndef __ EMX__31 # if HAVE_SEM_OPEN36 #ifndef __OS2__ 37 # if defined(HAVE_SEM_OPEN) && !defined(POSIX_SEMAPHORES_NOT_ENABLED) 32 38 # include <semaphore.h> 33 39 typedef sem_t *SEM_HANDLE; … … 50 56 */ 51 57 #ifndef SEM_VALUE_MAX 52 # ifdef _SEM_VALUE_MAX 53 # define SEM_VALUE_MAX _SEM_VALUE_MAX 54 # else 55 # define SEM_VALUE_MAX INT_MAX 56 # endif 58 #if defined(HAVE_SYSCONF) && defined(_SC_SEM_VALUE_MAX) 59 # define SEM_VALUE_MAX sysconf(_SC_SEM_VALUE_MAX) 60 #elif defined(_SEM_VALUE_MAX) 61 # define SEM_VALUE_MAX _SEM_VALUE_MAX 62 #elif defined(_POSIX_SEM_VALUE_MAX) 63 # define SEM_VALUE_MAX _POSIX_SEM_VALUE_MAX 64 #else 65 # define SEM_VALUE_MAX INT_MAX 66 #endif 57 67 #endif 68 58 69 59 70 /* … … 162 173 163 174 typedef struct { 164 165 166 167 168 175 PyObject_HEAD 176 HANDLE handle; 177 int flags; 178 PyObject *weakreflist; 179 char buffer[CONNECTION_BUFFER_SIZE]; 169 180 } ConnectionObject; 170 181 -
python/trunk/Modules/_multiprocessing/pipe_connection.c
r2 r391 18 18 conn_send_string(ConnectionObject *conn, char *string, size_t length) 19 19 { 20 21 20 DWORD amount_written; 21 BOOL ret; 22 22 23 Py_BEGIN_ALLOW_THREADS 24 ret = WriteFile(conn->handle, string, length, &amount_written, NULL); 25 Py_END_ALLOW_THREADS 26 return ret ? MP_SUCCESS : MP_STANDARD_ERROR; 23 Py_BEGIN_ALLOW_THREADS 24 ret = WriteFile(conn->handle, string, length, &amount_written, NULL); 25 Py_END_ALLOW_THREADS 26 27 if (ret == 0 && GetLastError() == ERROR_NO_SYSTEM_RESOURCES) { 28 PyErr_Format(PyExc_ValueError, "Cannnot send %" PY_FORMAT_SIZE_T "d bytes over connection", length); 29 return MP_STANDARD_ERROR; 30 } 31 32 return ret ? MP_SUCCESS : MP_STANDARD_ERROR; 27 33 } 28 34 … … 34 40 35 41 static Py_ssize_t 36 conn_recv_string(ConnectionObject *conn, char *buffer, 37 42 conn_recv_string(ConnectionObject *conn, char *buffer, 43 size_t buflength, char **newbuffer, size_t maxlength) 38 44 { 39 40 41 45 DWORD left, length, full_length, err; 46 BOOL ret; 47 *newbuffer = NULL; 42 48 43 44 ret = ReadFile(conn->handle, buffer, MIN(buflength, maxlength), 45 46 47 48 49 Py_BEGIN_ALLOW_THREADS 50 ret = ReadFile(conn->handle, buffer, MIN(buflength, maxlength), 51 &length, NULL); 52 Py_END_ALLOW_THREADS 53 if (ret) 54 return length; 49 55 50 51 52 53 54 55 56 err = GetLastError(); 57 if (err != ERROR_MORE_DATA) { 58 if (err == ERROR_BROKEN_PIPE) 59 return MP_END_OF_FILE; 60 return MP_STANDARD_ERROR; 61 } 56 62 57 58 63 if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, NULL, &left)) 64 return MP_STANDARD_ERROR; 59 65 60 61 62 66 full_length = length + left; 67 if (full_length > maxlength) 68 return MP_BAD_MESSAGE_LENGTH; 63 69 64 65 66 70 *newbuffer = PyMem_Malloc(full_length); 71 if (*newbuffer == NULL) 72 return MP_MEMORY_ERROR; 67 73 68 74 memcpy(*newbuffer, buffer, length); 69 75 70 71 72 73 74 75 76 77 78 79 76 Py_BEGIN_ALLOW_THREADS 77 ret = ReadFile(conn->handle, *newbuffer+length, left, &length, NULL); 78 Py_END_ALLOW_THREADS 79 if (ret) { 80 assert(length == left); 81 return full_length; 82 } else { 83 PyMem_Free(*newbuffer); 84 return MP_STANDARD_ERROR; 85 } 80 86 } 81 87 … … 87 93 conn_poll(ConnectionObject *conn, double timeout, PyThreadState *_save) 88 94 { 89 90 91 95 DWORD bytes, deadline, delay; 96 int difference, res; 97 BOOL block = FALSE; 92 98 93 94 99 if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL)) 100 return MP_STANDARD_ERROR; 95 101 96 97 102 if (timeout == 0.0) 103 return bytes > 0; 98 104 99 100 101 102 103 105 if (timeout < 0.0) 106 block = TRUE; 107 else 108 /* XXX does not check for overflow */ 109 deadline = GetTickCount() + (DWORD)(1000 * timeout + 0.5); 104 110 105 111 Sleep(0); 106 112 107 108 109 110 111 113 for (delay = 1 ; ; delay += 1) { 114 if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL)) 115 return MP_STANDARD_ERROR; 116 else if (bytes > 0) 117 return TRUE; 112 118 113 114 115 116 117 118 119 119 if (!block) { 120 difference = deadline - GetTickCount(); 121 if (difference < 0) 122 return FALSE; 123 if ((int)delay > difference) 124 delay = difference; 125 } 120 126 121 122 127 if (delay > 20) 128 delay = 20; 123 129 124 130 Sleep(delay); 125 131 126 127 Py_BLOCK_THREADS 128 129 132 /* check for signals */ 133 Py_BLOCK_THREADS 134 res = PyErr_CheckSignals(); 135 Py_UNBLOCK_THREADS 130 136 131 132 133 137 if (res) 138 return MP_EXCEPTION_HAS_BEEN_SET; 139 } 134 140 } 135 141 -
python/trunk/Modules/_multiprocessing/semaphore.c
r2 r391 12 12 13 13 typedef struct { 14 15 16 17 18 19 14 PyObject_HEAD 15 SEM_HANDLE handle; 16 long last_tid; 17 int count; 18 int maxvalue; 19 int kind; 20 20 } SemLockObject; 21 21 … … 41 41 _GetSemaphoreValue(HANDLE handle, long *value) 42 42 { 43 44 45 46 47 48 49 50 51 52 53 54 55 56 43 long previous; 44 45 switch (WaitForSingleObject(handle, 0)) { 46 case WAIT_OBJECT_0: 47 if (!ReleaseSemaphore(handle, 1, &previous)) 48 return MP_STANDARD_ERROR; 49 *value = previous + 1; 50 return 0; 51 case WAIT_TIMEOUT: 52 *value = 0; 53 return 0; 54 default: 55 return MP_STANDARD_ERROR; 56 } 57 57 } 58 58 … … 60 60 semlock_acquire(SemLockObject *self, PyObject *args, PyObject *kwds) 61 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 62 int blocking = 1; 63 double timeout; 64 PyObject *timeout_obj = Py_None; 65 DWORD res, full_msecs, msecs, start, ticks; 66 67 static char *kwlist[] = {"block", "timeout", NULL}; 68 69 if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iO", kwlist, 70 &blocking, &timeout_obj)) 71 return NULL; 72 73 /* calculate timeout */ 74 if (!blocking) { 75 full_msecs = 0; 76 } else if (timeout_obj == Py_None) { 77 full_msecs = INFINITE; 78 } else { 79 timeout = PyFloat_AsDouble(timeout_obj); 80 if (PyErr_Occurred()) 81 return NULL; 82 timeout *= 1000.0; /* convert to millisecs */ 83 if (timeout < 0.0) { 84 timeout = 0.0; 85 } else if (timeout >= 0.5 * INFINITE) { /* 25 days */ 86 PyErr_SetString(PyExc_OverflowError, 87 "timeout is too large"); 88 return NULL; 89 } 90 full_msecs = (DWORD)(timeout + 0.5); 91 } 92 93 /* check whether we already own the lock */ 94 if (self->kind == RECURSIVE_MUTEX && ISMINE(self)) { 95 ++self->count; 96 Py_RETURN_TRUE; 97 } 98 99 /* check whether we can acquire without blocking */ 100 if (WaitForSingleObject(self->handle, 0) == WAIT_OBJECT_0) { 101 self->last_tid = GetCurrentThreadId(); 102 ++self->count; 103 Py_RETURN_TRUE; 104 } 105 106 msecs = full_msecs; 107 start = GetTickCount(); 108 109 for ( ; ; ) { 110 HANDLE handles[2] = {self->handle, sigint_event}; 111 112 /* do the wait */ 113 Py_BEGIN_ALLOW_THREADS 114 ResetEvent(sigint_event); 115 res = WaitForMultipleObjects(2, handles, FALSE, msecs); 116 Py_END_ALLOW_THREADS 117 118 /* handle result */ 119 if (res != WAIT_OBJECT_0 + 1) 120 break; 121 122 /* got SIGINT so give signal handler a chance to run */ 123 Sleep(1); 124 125 /* if this is main thread let KeyboardInterrupt be raised */ 126 if (PyErr_CheckSignals()) 127 return NULL; 128 129 /* recalculate timeout */ 130 if (msecs != INFINITE) { 131 ticks = GetTickCount(); 132 if ((DWORD)(ticks - start) >= full_msecs) 133 Py_RETURN_FALSE; 134 msecs = full_msecs - (ticks - start); 135 } 136 } 137 138 /* handle result */ 139 switch (res) { 140 case WAIT_TIMEOUT: 141 Py_RETURN_FALSE; 142 case WAIT_OBJECT_0: 143 self->last_tid = GetCurrentThreadId(); 144 ++self->count; 145 Py_RETURN_TRUE; 146 case WAIT_FAILED: 147 return PyErr_SetFromWindowsErr(0); 148 default: 149 PyErr_Format(PyExc_RuntimeError, "WaitForSingleObject() or " 150 "WaitForMultipleObjects() gave unrecognized " 151 "value %d", res); 152 return NULL; 153 } 154 154 } 155 155 … … 157 157 semlock_release(SemLockObject *self, PyObject *args) 158 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 159 if (self->kind == RECURSIVE_MUTEX) { 160 if (!ISMINE(self)) { 161 PyErr_SetString(PyExc_AssertionError, "attempt to " 162 "release recursive lock not owned " 163 "by thread"); 164 return NULL; 165 } 166 if (self->count > 1) { 167 --self->count; 168 Py_RETURN_NONE; 169 } 170 assert(self->count == 1); 171 } 172 173 if (!ReleaseSemaphore(self->handle, 1, NULL)) { 174 if (GetLastError() == ERROR_TOO_MANY_POSTS) { 175 PyErr_SetString(PyExc_ValueError, "semaphore or lock " 176 "released too many times"); 177 return NULL; 178 } else { 179 return PyErr_SetFromWindowsErr(0); 180 } 181 } 182 183 --self->count; 184 Py_RETURN_NONE; 185 185 } 186 186 … … 198 198 #define SEM_UNLINK(name) sem_unlink(name) 199 199 200 #if HAVE_BROKEN_SEM_UNLINK 200 /* OS X 10.4 defines SEM_FAILED as -1 instead of (sem_t *)-1; this gives 201 compiler warnings, and (potentially) undefined behaviour. */ 202 #ifdef __APPLE__ 203 # undef SEM_FAILED 204 # define SEM_FAILED ((sem_t *)-1) 205 #endif 206 207 #ifndef HAVE_SEM_UNLINK 201 208 # define sem_unlink(name) 0 202 209 #endif 203 210 204 #if !HAVE_SEM_TIMEDWAIT211 #ifndef HAVE_SEM_TIMEDWAIT 205 212 # define sem_timedwait(sem,deadline) sem_timedwait_save(sem,deadline,_save) 206 213 … … 208 215 sem_timedwait_save(sem_t *sem, struct timespec *deadline, PyThreadState *_save) 209 216 { 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 if (tvdeadline.tv_sec < now.tv_sec || 231 (tvdeadline.tv_sec == now.tv_sec && 232 233 234 235 236 237 238 difference = (tvdeadline.tv_sec - now.tv_sec) * 1000000 + 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 Py_BLOCK_THREADS 255 256 257 258 259 260 261 262 217 int res; 218 unsigned long delay, difference; 219 struct timeval now, tvdeadline, tvdelay; 220 221 errno = 0; 222 tvdeadline.tv_sec = deadline->tv_sec; 223 tvdeadline.tv_usec = deadline->tv_nsec / 1000; 224 225 for (delay = 0 ; ; delay += 1000) { 226 /* poll */ 227 if (sem_trywait(sem) == 0) 228 return 0; 229 else if (errno != EAGAIN) 230 return MP_STANDARD_ERROR; 231 232 /* get current time */ 233 if (gettimeofday(&now, NULL) < 0) 234 return MP_STANDARD_ERROR; 235 236 /* check for timeout */ 237 if (tvdeadline.tv_sec < now.tv_sec || 238 (tvdeadline.tv_sec == now.tv_sec && 239 tvdeadline.tv_usec <= now.tv_usec)) { 240 errno = ETIMEDOUT; 241 return MP_STANDARD_ERROR; 242 } 243 244 /* calculate how much time is left */ 245 difference = (tvdeadline.tv_sec - now.tv_sec) * 1000000 + 246 (tvdeadline.tv_usec - now.tv_usec); 247 248 /* check delay not too long -- maximum is 20 msecs */ 249 if (delay > 20000) 250 delay = 20000; 251 if (delay > difference) 252 delay = difference; 253 254 /* sleep */ 255 tvdelay.tv_sec = delay / 1000000; 256 tvdelay.tv_usec = delay % 1000000; 257 if (select(0, NULL, NULL, NULL, &tvdelay) < 0) 258 return MP_STANDARD_ERROR; 259 260 /* check for signals */ 261 Py_BLOCK_THREADS 262 res = PyErr_CheckSignals(); 263 Py_UNBLOCK_THREADS 264 265 if (res) { 266 errno = EINTR; 267 return MP_EXCEPTION_HAS_BEEN_SET; 268 } 269 } 263 270 } 264 271 … … 268 275 semlock_acquire(SemLockObject *self, PyObject *args, PyObject *kwds) 269 276 { 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 277 int blocking = 1, res; 278 double timeout; 279 PyObject *timeout_obj = Py_None; 280 struct timespec deadline = {0}; 281 struct timeval now; 282 long sec, nsec; 283 284 static char *kwlist[] = {"block", "timeout", NULL}; 285 286 if (!PyArg_ParseTupleAndKeywords(args, kwds, "|iO", kwlist, 287 &blocking, &timeout_obj)) 288 return NULL; 289 290 if (self->kind == RECURSIVE_MUTEX && ISMINE(self)) { 291 ++self->count; 292 Py_RETURN_TRUE; 293 } 294 295 if (timeout_obj != Py_None) { 296 timeout = PyFloat_AsDouble(timeout_obj); 297 if (PyErr_Occurred()) 298 return NULL; 299 if (timeout < 0.0) 300 timeout = 0.0; 301 302 if (gettimeofday(&now, NULL) < 0) { 303 PyErr_SetFromErrno(PyExc_OSError); 304 return NULL; 305 } 306 sec = (long) timeout; 307 nsec = (long) (1e9 * (timeout - sec) + 0.5); 308 deadline.tv_sec = now.tv_sec + sec; 309 deadline.tv_nsec = now.tv_usec * 1000 + nsec; 310 deadline.tv_sec += (deadline.tv_nsec / 1000000000); 311 deadline.tv_nsec %= 1000000000; 312 } 313 314 do { 315 Py_BEGIN_ALLOW_THREADS 316 if (blocking && timeout_obj == Py_None) 317 res = sem_wait(self->handle); 318 else if (!blocking) 319 res = sem_trywait(self->handle); 320 else 321 res = sem_timedwait(self->handle, &deadline); 322 Py_END_ALLOW_THREADS 323 if (res == MP_EXCEPTION_HAS_BEEN_SET) 324 break; 325 } while (res < 0 && errno == EINTR && !PyErr_CheckSignals()); 326 327 if (res < 0) { 328 if (errno == EAGAIN || errno == ETIMEDOUT) 329 Py_RETURN_FALSE; 330 else if (errno == EINTR) 331 return NULL; 332 else 333 return PyErr_SetFromErrno(PyExc_OSError); 334 } 335 336 ++self->count; 337 self->last_tid = PyThread_get_thread_ident(); 338 339 Py_RETURN_TRUE; 333 340 } 334 341 … … 336 343 semlock_release(SemLockObject *self, PyObject *args) 337 344 { 338 339 340 341 342 343 344 345 346 347 348 349 350 351 #if HAVE_BROKEN_SEM_GETVALUE352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 345 if (self->kind == RECURSIVE_MUTEX) { 346 if (!ISMINE(self)) { 347 PyErr_SetString(PyExc_AssertionError, "attempt to " 348 "release recursive lock not owned " 349 "by thread"); 350 return NULL; 351 } 352 if (self->count > 1) { 353 --self->count; 354 Py_RETURN_NONE; 355 } 356 assert(self->count == 1); 357 } else { 358 #ifdef HAVE_BROKEN_SEM_GETVALUE 359 /* We will only check properly the maxvalue == 1 case */ 360 if (self->maxvalue == 1) { 361 /* make sure that already locked */ 362 if (sem_trywait(self->handle) < 0) { 363 if (errno != EAGAIN) { 364 PyErr_SetFromErrno(PyExc_OSError); 365 return NULL; 366 } 367 /* it is already locked as expected */ 368 } else { 369 /* it was not locked so undo wait and raise */ 370 if (sem_post(self->handle) < 0) { 371 PyErr_SetFromErrno(PyExc_OSError); 372 return NULL; 373 } 374 PyErr_SetString(PyExc_ValueError, "semaphore " 375 "or lock released too many " 376 "times"); 377 return NULL; 378 } 379 } 373 380 #else 374 375 376 377 378 379 380 381 382 383 384 381 int sval; 382 383 /* This check is not an absolute guarantee that the semaphore 384 does not rise above maxvalue. */ 385 if (sem_getvalue(self->handle, &sval) < 0) { 386 return PyErr_SetFromErrno(PyExc_OSError); 387 } else if (sval >= self->maxvalue) { 388 PyErr_SetString(PyExc_ValueError, "semaphore or lock " 389 "released too many times"); 390 return NULL; 391 } 385 392 #endif 386 387 388 389 390 391 392 393 } 394 395 if (sem_post(self->handle) < 0) 396 return PyErr_SetFromErrno(PyExc_OSError); 397 398 --self->count; 399 Py_RETURN_NONE; 393 400 } 394 401 … … 402 409 newsemlockobject(PyTypeObject *type, SEM_HANDLE handle, int kind, int maxvalue) 403 410 { 404 405 406 407 408 409 410 411 412 413 414 411 SemLockObject *self; 412 413 self = PyObject_New(SemLockObject, type); 414 if (!self) 415 return NULL; 416 self->handle = handle; 417 self->kind = kind; 418 self->count = 0; 419 self->last_tid = 0; 420 self->maxvalue = maxvalue; 421 return (PyObject*)self; 415 422 } 416 423 … … 418 425 semlock_new(PyTypeObject *type, PyObject *args, PyObject *kwds) 419 426 { 420 421 422 423 424 425 426 427 if (!PyArg_ParseTupleAndKeywords(args, kwds, "iii", kwlist, 428 429 430 431 432 433 434 435 436 PyOS_snprintf(buffer, sizeof(buffer), "/mp%d-%d",getpid(), counter++);437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 427 char buffer[256]; 428 SEM_HANDLE handle = SEM_FAILED; 429 int kind, maxvalue, value; 430 PyObject *result; 431 static char *kwlist[] = {"kind", "value", "maxvalue", NULL}; 432 static int counter = 0; 433 434 if (!PyArg_ParseTupleAndKeywords(args, kwds, "iii", kwlist, 435 &kind, &value, &maxvalue)) 436 return NULL; 437 438 if (kind != RECURSIVE_MUTEX && kind != SEMAPHORE) { 439 PyErr_SetString(PyExc_ValueError, "unrecognized kind"); 440 return NULL; 441 } 442 443 PyOS_snprintf(buffer, sizeof(buffer), "/mp%ld-%d", (long)getpid(), counter++); 444 445 SEM_CLEAR_ERROR(); 446 handle = SEM_CREATE(buffer, value, maxvalue); 447 /* On Windows we should fail if GetLastError()==ERROR_ALREADY_EXISTS */ 448 if (handle == SEM_FAILED || SEM_GET_LAST_ERROR() != 0) 449 goto failure; 450 451 if (SEM_UNLINK(buffer) < 0) 452 goto failure; 453 454 result = newsemlockobject(type, handle, kind, maxvalue); 455 if (!result) 456 goto failure; 457 458 return result; 452 459 453 460 failure: 454 455 456 457 461 if (handle != SEM_FAILED) 462 SEM_CLOSE(handle); 463 mp_SetError(NULL, MP_STANDARD_ERROR); 464 return NULL; 458 465 } 459 466 … … 461 468 semlock_rebuild(PyTypeObject *type, PyObject *args) 462 469 { 463 464 465 466 if (!PyArg_ParseTuple(args, F_SEM_HANDLE "ii", 467 468 469 470 470 SEM_HANDLE handle; 471 int kind, maxvalue; 472 473 if (!PyArg_ParseTuple(args, F_SEM_HANDLE "ii", 474 &handle, &kind, &maxvalue)) 475 return NULL; 476 477 return newsemlockobject(type, handle, kind, maxvalue); 471 478 } 472 479 … … 474 481 semlock_dealloc(SemLockObject* self) 475 482 { 476 477 478 483 if (self->handle != SEM_FAILED) 484 SEM_CLOSE(self->handle); 485 PyObject_Del(self); 479 486 } 480 487 … … 482 489 semlock_count(SemLockObject *self) 483 490 { 484 491 return PyInt_FromLong((long)self->count); 485 492 } 486 493 … … 488 495 semlock_ismine(SemLockObject *self) 489 496 { 490 491 497 /* only makes sense for a lock */ 498 return PyBool_FromLong(ISMINE(self)); 492 499 } 493 500 … … 495 502 semlock_getvalue(SemLockObject *self) 496 503 { 497 #if HAVE_BROKEN_SEM_GETVALUE498 499 504 #ifdef HAVE_BROKEN_SEM_GETVALUE 505 PyErr_SetNone(PyExc_NotImplementedError); 506 return NULL; 500 507 #else 501 502 503 504 /* some posix implementations use negative numbers to indicate 505 506 507 508 508 int sval; 509 if (SEM_GETVALUE(self->handle, &sval) < 0) 510 return mp_SetError(NULL, MP_STANDARD_ERROR); 511 /* some posix implementations use negative numbers to indicate 512 the number of waiting threads */ 513 if (sval < 0) 514 sval = 0; 515 return PyInt_FromLong((long)sval); 509 516 #endif 510 517 } … … 513 520 semlock_iszero(SemLockObject *self) 514 521 { 515 #if HAVE_BROKEN_SEM_GETVALUE516 517 518 519 520 521 522 523 524 522 #ifdef HAVE_BROKEN_SEM_GETVALUE 523 if (sem_trywait(self->handle) < 0) { 524 if (errno == EAGAIN) 525 Py_RETURN_TRUE; 526 return mp_SetError(NULL, MP_STANDARD_ERROR); 527 } else { 528 if (sem_post(self->handle) < 0) 529 return mp_SetError(NULL, MP_STANDARD_ERROR); 530 Py_RETURN_FALSE; 531 } 525 532 #else 526 527 528 529 533 int sval; 534 if (SEM_GETVALUE(self->handle, &sval) < 0) 535 return mp_SetError(NULL, MP_STANDARD_ERROR); 536 return PyBool_FromLong((long)sval == 0); 530 537 #endif 531 538 } … … 534 541 semlock_afterfork(SemLockObject *self) 535 542 { 536 537 543 self->count = 0; 544 Py_RETURN_NONE; 538 545 } 539 546 … … 543 550 544 551 static PyMethodDef semlock_methods[] = { 545 546 547 {"release", (PyCFunction)semlock_release, METH_NOARGS, 548 552 {"acquire", (PyCFunction)semlock_acquire, METH_VARARGS | METH_KEYWORDS, 553 "acquire the semaphore/lock"}, 554 {"release", (PyCFunction)semlock_release, METH_NOARGS, 555 "release the semaphore/lock"}, 549 556 {"__enter__", (PyCFunction)semlock_acquire, METH_VARARGS | METH_KEYWORDS, 550 551 {"__exit__", (PyCFunction)semlock_release, METH_VARARGS, 552 553 {"_count", (PyCFunction)semlock_count, METH_NOARGS, 554 555 {"_is_mine", (PyCFunction)semlock_ismine, METH_NOARGS, 556 557 {"_get_value", (PyCFunction)semlock_getvalue, METH_NOARGS, 558 "get the value of the semaphore"}, 559 {"_is_zero", (PyCFunction)semlock_iszero, METH_NOARGS, 560 "returns whether semaphore has value zero"}, 561 {"_rebuild", (PyCFunction)semlock_rebuild, METH_VARARGS | METH_CLASS, 562 ""}, 563 564 565 557 "enter the semaphore/lock"}, 558 {"__exit__", (PyCFunction)semlock_release, METH_VARARGS, 559 "exit the semaphore/lock"}, 560 {"_count", (PyCFunction)semlock_count, METH_NOARGS, 561 "num of `acquire()`s minus num of `release()`s for this process"}, 562 {"_is_mine", (PyCFunction)semlock_ismine, METH_NOARGS, 563 "whether the lock is owned by this thread"}, 564 {"_get_value", (PyCFunction)semlock_getvalue, METH_NOARGS, 565 "get the value of the semaphore"}, 566 {"_is_zero", (PyCFunction)semlock_iszero, METH_NOARGS, 567 "returns whether semaphore has value zero"}, 568 {"_rebuild", (PyCFunction)semlock_rebuild, METH_VARARGS | METH_CLASS, 569 ""}, 570 {"_after_fork", (PyCFunction)semlock_afterfork, METH_NOARGS, 571 "rezero the net acquisition count after fork()"}, 572 {NULL} 566 573 }; 567 574 … … 571 578 572 579 static PyMemberDef semlock_members[] = { 573 {"handle", T_SEM_HANDLE, offsetof(SemLockObject, handle), READONLY, 574 575 {"kind", T_INT, offsetof(SemLockObject, kind), READONLY, 576 577 {"maxvalue", T_INT, offsetof(SemLockObject, maxvalue), READONLY, 578 579 580 {"handle", T_SEM_HANDLE, offsetof(SemLockObject, handle), READONLY, 581 ""}, 582 {"kind", T_INT, offsetof(SemLockObject, kind), READONLY, 583 ""}, 584 {"maxvalue", T_INT, offsetof(SemLockObject, maxvalue), READONLY, 585 ""}, 586 {NULL} 580 587 }; 581 588 … … 585 592 586 593 PyTypeObject SemLockType = { 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 594 PyVarObject_HEAD_INIT(NULL, 0) 595 /* tp_name */ "_multiprocessing.SemLock", 596 /* tp_basicsize */ sizeof(SemLockObject), 597 /* tp_itemsize */ 0, 598 /* tp_dealloc */ (destructor)semlock_dealloc, 599 /* tp_print */ 0, 600 /* tp_getattr */ 0, 601 /* tp_setattr */ 0, 602 /* tp_compare */ 0, 603 /* tp_repr */ 0, 604 /* tp_as_number */ 0, 605 /* tp_as_sequence */ 0, 606 /* tp_as_mapping */ 0, 607 /* tp_hash */ 0, 608 /* tp_call */ 0, 609 /* tp_str */ 0, 610 /* tp_getattro */ 0, 611 /* tp_setattro */ 0, 612 /* tp_as_buffer */ 0, 613 /* tp_flags */ Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE, 614 /* tp_doc */ "Semaphore/Mutex type", 615 /* tp_traverse */ 0, 616 /* tp_clear */ 0, 617 /* tp_richcompare */ 0, 618 /* tp_weaklistoffset */ 0, 619 /* tp_iter */ 0, 620 /* tp_iternext */ 0, 621 /* tp_methods */ semlock_methods, 622 /* tp_members */ semlock_members, 623 /* tp_getset */ 0, 624 /* tp_base */ 0, 625 /* tp_dict */ 0, 626 /* tp_descr_get */ 0, 627 /* tp_descr_set */ 0, 628 /* tp_dictoffset */ 0, 629 /* tp_init */ 0, 630 /* tp_alloc */ 0, 631 /* tp_new */ semlock_new, 625 632 }; -
python/trunk/Modules/_multiprocessing/socket_connection.c
r2 r391 8 8 9 9 #include "multiprocessing.h" 10 11 #if defined(HAVE_POLL) && !defined(HAVE_BROKEN_POLL) 12 # include "poll.h" 13 #endif 10 14 11 15 #ifdef MS_WINDOWS … … 20 24 21 25 /* 26 * Wrapper for PyErr_CheckSignals() which can be called without the GIL 27 */ 28 29 static int 30 check_signals(void) 31 { 32 PyGILState_STATE state; 33 int res; 34 state = PyGILState_Ensure(); 35 res = PyErr_CheckSignals(); 36 PyGILState_Release(state); 37 return res; 38 } 39 40 /* 22 41 * Send string to file descriptor 23 42 */ … … 26 45 _conn_sendall(HANDLE h, char *string, size_t length) 27 46 { 28 char *p = string; 29 Py_ssize_t res; 30 31 while (length > 0) { 32 res = WRITE(h, p, length); 33 if (res < 0) 34 return MP_SOCKET_ERROR; 35 length -= res; 36 p += res; 37 } 38 39 return MP_SUCCESS; 40 } 41 42 /* 43 * Receive string of exact length from file descriptor 47 char *p = string; 48 Py_ssize_t res; 49 50 while (length > 0) { 51 res = WRITE(h, p, length); 52 if (res < 0) { 53 if (errno == EINTR) { 54 if (check_signals() < 0) 55 return MP_EXCEPTION_HAS_BEEN_SET; 56 continue; 57 } 58 return MP_SOCKET_ERROR; 59 } 60 length -= res; 61 p += res; 62 } 63 64 return MP_SUCCESS; 65 } 66 67 /* 68 * Receive string of exact length from file descriptor 44 69 */ 45 70 … … 47 72 _conn_recvall(HANDLE h, char *buffer, size_t length) 48 73 { 49 size_t remaining = length; 50 Py_ssize_t temp; 51 char *p = buffer; 52 53 while (remaining > 0) { 54 temp = READ(h, p, remaining); 55 if (temp <= 0) { 56 if (temp == 0) 57 return remaining == length ? 58 MP_END_OF_FILE : MP_EARLY_END_OF_FILE; 59 else 60 return temp; 61 } 62 remaining -= temp; 63 p += temp; 64 } 65 66 return MP_SUCCESS; 74 size_t remaining = length; 75 Py_ssize_t temp; 76 char *p = buffer; 77 78 while (remaining > 0) { 79 temp = READ(h, p, remaining); 80 if (temp < 0) { 81 if (errno == EINTR) { 82 if (check_signals() < 0) 83 return MP_EXCEPTION_HAS_BEEN_SET; 84 continue; 85 } 86 return temp; 87 } 88 else if (temp == 0) { 89 return remaining == length ? MP_END_OF_FILE : MP_EARLY_END_OF_FILE; 90 } 91 remaining -= temp; 92 p += temp; 93 } 94 95 return MP_SUCCESS; 67 96 } 68 97 … … 74 103 conn_send_string(ConnectionObject *conn, char *string, size_t length) 75 104 { 76 77 78 79 80 81 82 83 84 85 86 87 88 89 *(UINT32*)message = htonl((UINT32)length); 90 91 92 93 94 95 96 97 98 99 100 101 102 103 res = _conn_sendall(conn->handle, (char*)&lenbuff, 4) || 104 105 106 107 105 Py_ssize_t res; 106 /* The "header" of the message is a 32 bit unsigned number (in 107 network order) which specifies the length of the "body". If 108 the message is shorter than about 16kb then it is quicker to 109 combine the "header" and the "body" of the message and send 110 them at once. */ 111 if (length < (16*1024)) { 112 char *message; 113 114 message = PyMem_Malloc(length+4); 115 if (message == NULL) 116 return MP_MEMORY_ERROR; 117 118 *(UINT32*)message = htonl((UINT32)length); 119 memcpy(message+4, string, length); 120 Py_BEGIN_ALLOW_THREADS 121 res = _conn_sendall(conn->handle, message, length+4); 122 Py_END_ALLOW_THREADS 123 PyMem_Free(message); 124 } else { 125 UINT32 lenbuff; 126 127 if (length > MAX_MESSAGE_LENGTH) 128 return MP_BAD_MESSAGE_LENGTH; 129 130 lenbuff = htonl((UINT32)length); 131 Py_BEGIN_ALLOW_THREADS 132 res = _conn_sendall(conn->handle, (char*)&lenbuff, 4) || 133 _conn_sendall(conn->handle, string, length); 134 Py_END_ALLOW_THREADS 135 } 136 return res; 108 137 } 109 138 … … 115 144 116 145 static Py_ssize_t 117 conn_recv_string(ConnectionObject *conn, char *buffer, 118 size_t buflength, char **newbuffer, size_t maxlength) 119 { 120 int res; 121 UINT32 ulength; 122 123 *newbuffer = NULL; 124 125 Py_BEGIN_ALLOW_THREADS 126 res = _conn_recvall(conn->handle, (char*)&ulength, 4); 127 Py_END_ALLOW_THREADS 128 if (res < 0) 129 return res; 130 131 ulength = ntohl(ulength); 132 if (ulength > maxlength) 133 return MP_BAD_MESSAGE_LENGTH; 134 135 if (ulength <= buflength) { 136 Py_BEGIN_ALLOW_THREADS 137 res = _conn_recvall(conn->handle, buffer, (size_t)ulength); 138 Py_END_ALLOW_THREADS 139 return res < 0 ? res : ulength; 140 } else { 141 *newbuffer = PyMem_Malloc((size_t)ulength); 142 if (*newbuffer == NULL) 143 return MP_MEMORY_ERROR; 144 Py_BEGIN_ALLOW_THREADS 145 res = _conn_recvall(conn->handle, *newbuffer, (size_t)ulength); 146 Py_END_ALLOW_THREADS 147 return res < 0 ? (Py_ssize_t)res : (Py_ssize_t)ulength; 148 } 146 conn_recv_string(ConnectionObject *conn, char *buffer, 147 size_t buflength, char **newbuffer, size_t maxlength) 148 { 149 Py_ssize_t res; 150 UINT32 ulength; 151 152 *newbuffer = NULL; 153 154 Py_BEGIN_ALLOW_THREADS 155 res = _conn_recvall(conn->handle, (char*)&ulength, 4); 156 Py_END_ALLOW_THREADS 157 if (res < 0) 158 return res; 159 160 ulength = ntohl(ulength); 161 if (ulength > maxlength) 162 return MP_BAD_MESSAGE_LENGTH; 163 164 if (ulength > buflength) { 165 *newbuffer = buffer = PyMem_Malloc((size_t)ulength); 166 if (buffer == NULL) 167 return MP_MEMORY_ERROR; 168 } 169 170 Py_BEGIN_ALLOW_THREADS 171 res = _conn_recvall(conn->handle, buffer, (size_t)ulength); 172 Py_END_ALLOW_THREADS 173 174 if (res >= 0) { 175 res = (Py_ssize_t)ulength; 176 } else if (*newbuffer != NULL) { 177 PyMem_Free(*newbuffer); 178 *newbuffer = NULL; 179 } 180 return res; 149 181 } 150 182 … … 156 188 conn_poll(ConnectionObject *conn, double timeout, PyThreadState *_save) 157 189 { 158 int res; 159 fd_set rfds; 160 161 /* 162 * Verify the handle, issue 3321. Not required for windows. 163 */ 164 #ifndef MS_WINDOWS 165 if (((int)conn->handle) < 0 || ((int)conn->handle) >= FD_SETSIZE) { 166 Py_BLOCK_THREADS 167 PyErr_SetString(PyExc_IOError, "handle out of range in select()"); 168 Py_UNBLOCK_THREADS 169 return MP_EXCEPTION_HAS_BEEN_SET; 170 } 171 #endif 172 173 FD_ZERO(&rfds); 174 FD_SET((SOCKET)conn->handle, &rfds); 175 176 if (timeout < 0.0) { 177 res = select((int)conn->handle+1, &rfds, NULL, NULL, NULL); 178 } else { 179 struct timeval tv; 180 tv.tv_sec = (long)timeout; 181 tv.tv_usec = (long)((timeout - tv.tv_sec) * 1e6 + 0.5); 182 res = select((int)conn->handle+1, &rfds, NULL, NULL, &tv); 183 } 184 185 if (res < 0) { 186 return MP_SOCKET_ERROR; 187 } else if (FD_ISSET(conn->handle, &rfds)) { 188 return TRUE; 189 } else { 190 assert(res == 0); 191 return FALSE; 192 } 190 #if defined(HAVE_POLL) && !defined(HAVE_BROKEN_POLL) 191 int res; 192 struct pollfd p; 193 194 p.fd = (int)conn->handle; 195 p.events = POLLIN | POLLPRI; 196 p.revents = 0; 197 198 if (timeout < 0) { 199 do { 200 res = poll(&p, 1, -1); 201 } while (res < 0 && errno == EINTR); 202 } else { 203 res = poll(&p, 1, (int)(timeout * 1000 + 0.5)); 204 if (res < 0 && errno == EINTR) { 205 /* We were interrupted by a signal. Just indicate a 206 timeout even though we are early. */ 207 return FALSE; 208 } 209 } 210 211 if (res < 0) { 212 return MP_SOCKET_ERROR; 213 } else if (p.revents & (POLLNVAL|POLLERR)) { 214 Py_BLOCK_THREADS 215 PyErr_SetString(PyExc_IOError, "poll() gave POLLNVAL or POLLERR"); 216 Py_UNBLOCK_THREADS 217 return MP_EXCEPTION_HAS_BEEN_SET; 218 } else if (p.revents != 0) { 219 return TRUE; 220 } else { 221 assert(res == 0); 222 return FALSE; 223 } 224 #else 225 int res; 226 fd_set rfds; 227 228 /* 229 * Verify the handle, issue 3321. Not required for windows. 230 */ 231 #ifndef MS_WINDOWS 232 if (((int)conn->handle) < 0 || ((int)conn->handle) >= FD_SETSIZE) { 233 Py_BLOCK_THREADS 234 PyErr_SetString(PyExc_IOError, "handle out of range in select()"); 235 Py_UNBLOCK_THREADS 236 return MP_EXCEPTION_HAS_BEEN_SET; 237 } 238 #endif 239 240 FD_ZERO(&rfds); 241 FD_SET((SOCKET)conn->handle, &rfds); 242 243 if (timeout < 0.0) { 244 do { 245 res = select((int)conn->handle+1, &rfds, NULL, NULL, NULL); 246 } while (res < 0 && errno == EINTR); 247 } else { 248 struct timeval tv; 249 tv.tv_sec = (long)timeout; 250 tv.tv_usec = (long)((timeout - tv.tv_sec) * 1e6 + 0.5); 251 res = select((int)conn->handle+1, &rfds, NULL, NULL, &tv); 252 if (res < 0 && errno == EINTR) { 253 /* We were interrupted by a signal. Just indicate a 254 timeout even though we are early. */ 255 return FALSE; 256 } 257 } 258 259 if (res < 0) { 260 return MP_SOCKET_ERROR; 261 } else if (FD_ISSET(conn->handle, &rfds)) { 262 return TRUE; 263 } else { 264 assert(res == 0); 265 return FALSE; 266 } 267 #endif 193 268 } 194 269 -
python/trunk/Modules/_multiprocessing/win32_functions.c
r2 r391 20 20 win32_CloseHandle(PyObject *self, PyObject *args) 21 21 { 22 23 24 25 26 27 28 29 success = CloseHandle(hObject); 30 31 32 33 34 35 22 HANDLE hObject; 23 BOOL success; 24 25 if (!PyArg_ParseTuple(args, F_HANDLE, &hObject)) 26 return NULL; 27 28 Py_BEGIN_ALLOW_THREADS 29 success = CloseHandle(hObject); 30 Py_END_ALLOW_THREADS 31 32 if (!success) 33 return PyErr_SetFromWindowsErr(0); 34 35 Py_RETURN_NONE; 36 36 } 37 37 … … 39 39 win32_ConnectNamedPipe(PyObject *self, PyObject *args) 40 40 { 41 42 43 44 45 if (!PyArg_ParseTuple(args, F_HANDLE F_POINTER, 46 47 48 49 50 51 52 53 54 55 56 41 HANDLE hNamedPipe; 42 LPOVERLAPPED lpOverlapped; 43 BOOL success; 44 45 if (!PyArg_ParseTuple(args, F_HANDLE F_POINTER, 46 &hNamedPipe, &lpOverlapped)) 47 return NULL; 48 49 Py_BEGIN_ALLOW_THREADS 50 success = ConnectNamedPipe(hNamedPipe, lpOverlapped); 51 Py_END_ALLOW_THREADS 52 53 if (!success) 54 return PyErr_SetFromWindowsErr(0); 55 56 Py_RETURN_NONE; 57 57 } 58 58 … … 60 60 win32_CreateFile(PyObject *self, PyObject *args) 61 61 { 62 63 64 65 66 67 68 69 70 71 if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_POINTER 72 73 &lpFileName, &dwDesiredAccess, &dwShareMode, 74 &lpSecurityAttributes, &dwCreationDisposition, 75 76 77 78 79 handle = CreateFile(lpFileName, dwDesiredAccess, 80 dwShareMode, lpSecurityAttributes, 81 dwCreationDisposition, 82 83 84 85 86 87 88 62 LPCTSTR lpFileName; 63 DWORD dwDesiredAccess; 64 DWORD dwShareMode; 65 LPSECURITY_ATTRIBUTES lpSecurityAttributes; 66 DWORD dwCreationDisposition; 67 DWORD dwFlagsAndAttributes; 68 HANDLE hTemplateFile; 69 HANDLE handle; 70 71 if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_POINTER 72 F_DWORD F_DWORD F_HANDLE, 73 &lpFileName, &dwDesiredAccess, &dwShareMode, 74 &lpSecurityAttributes, &dwCreationDisposition, 75 &dwFlagsAndAttributes, &hTemplateFile)) 76 return NULL; 77 78 Py_BEGIN_ALLOW_THREADS 79 handle = CreateFile(lpFileName, dwDesiredAccess, 80 dwShareMode, lpSecurityAttributes, 81 dwCreationDisposition, 82 dwFlagsAndAttributes, hTemplateFile); 83 Py_END_ALLOW_THREADS 84 85 if (handle == INVALID_HANDLE_VALUE) 86 return PyErr_SetFromWindowsErr(0); 87 88 return Py_BuildValue(F_HANDLE, handle); 89 89 } 90 90 … … 92 92 win32_CreateNamedPipe(PyObject *self, PyObject *args) 93 93 { 94 95 96 97 98 99 100 101 102 103 104 if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_DWORD 105 106 &lpName, &dwOpenMode, &dwPipeMode, 107 &nMaxInstances, &nOutBufferSize, 108 109 110 111 112 113 handle = CreateNamedPipe(lpName, dwOpenMode, dwPipeMode, 114 nMaxInstances, nOutBufferSize, 115 116 117 118 119 120 121 122 94 LPCTSTR lpName; 95 DWORD dwOpenMode; 96 DWORD dwPipeMode; 97 DWORD nMaxInstances; 98 DWORD nOutBufferSize; 99 DWORD nInBufferSize; 100 DWORD nDefaultTimeOut; 101 LPSECURITY_ATTRIBUTES lpSecurityAttributes; 102 HANDLE handle; 103 104 if (!PyArg_ParseTuple(args, "s" F_DWORD F_DWORD F_DWORD 105 F_DWORD F_DWORD F_DWORD F_POINTER, 106 &lpName, &dwOpenMode, &dwPipeMode, 107 &nMaxInstances, &nOutBufferSize, 108 &nInBufferSize, &nDefaultTimeOut, 109 &lpSecurityAttributes)) 110 return NULL; 111 112 Py_BEGIN_ALLOW_THREADS 113 handle = CreateNamedPipe(lpName, dwOpenMode, dwPipeMode, 114 nMaxInstances, nOutBufferSize, 115 nInBufferSize, nDefaultTimeOut, 116 lpSecurityAttributes); 117 Py_END_ALLOW_THREADS 118 119 if (handle == INVALID_HANDLE_VALUE) 120 return PyErr_SetFromWindowsErr(0); 121 122 return Py_BuildValue(F_HANDLE, handle); 123 123 } 124 124 … … 126 126 win32_ExitProcess(PyObject *self, PyObject *args) 127 127 { 128 129 130 131 132 133 134 135 136 137 138 139 140 141 128 UINT uExitCode; 129 130 if (!PyArg_ParseTuple(args, "I", &uExitCode)) 131 return NULL; 132 133 #if defined(Py_DEBUG) 134 SetErrorMode(SEM_FAILCRITICALERRORS|SEM_NOALIGNMENTFAULTEXCEPT|SEM_NOGPFAULTERRORBOX|SEM_NOOPENFILEERRORBOX); 135 _CrtSetReportMode(_CRT_ASSERT, _CRTDBG_MODE_DEBUG); 136 #endif 137 138 139 ExitProcess(uExitCode); 140 141 return NULL; 142 142 } 143 143 … … 145 145 win32_GetLastError(PyObject *self, PyObject *args) 146 146 { 147 147 return Py_BuildValue(F_DWORD, GetLastError()); 148 148 } 149 149 … … 151 151 win32_OpenProcess(PyObject *self, PyObject *args) 152 152 { 153 154 155 156 157 158 if (!PyArg_ParseTuple(args, F_DWORD "i" F_DWORD, 159 160 161 162 handle = OpenProcess(dwDesiredAccess, bInheritHandle, dwProcessId); 163 164 165 166 153 DWORD dwDesiredAccess; 154 BOOL bInheritHandle; 155 DWORD dwProcessId; 156 HANDLE handle; 157 158 if (!PyArg_ParseTuple(args, F_DWORD "i" F_DWORD, 159 &dwDesiredAccess, &bInheritHandle, &dwProcessId)) 160 return NULL; 161 162 handle = OpenProcess(dwDesiredAccess, bInheritHandle, dwProcessId); 163 if (handle == NULL) 164 return PyErr_SetFromWindowsErr(0); 165 166 return Py_BuildValue(F_HANDLE, handle); 167 167 } 168 168 … … 170 170 win32_SetNamedPipeHandleState(PyObject *self, PyObject *args) 171 171 { 172 173 174 175 176 177 if (!PyArg_ParseTuple(args, F_HANDLE "OOO", 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 172 HANDLE hNamedPipe; 173 PyObject *oArgs[3]; 174 DWORD dwArgs[3], *pArgs[3] = {NULL, NULL, NULL}; 175 int i; 176 177 if (!PyArg_ParseTuple(args, F_HANDLE "OOO", 178 &hNamedPipe, &oArgs[0], &oArgs[1], &oArgs[2])) 179 return NULL; 180 181 PyErr_Clear(); 182 183 for (i = 0 ; i < 3 ; i++) { 184 if (oArgs[i] != Py_None) { 185 dwArgs[i] = PyInt_AsUnsignedLongMask(oArgs[i]); 186 if (PyErr_Occurred()) 187 return NULL; 188 pArgs[i] = &dwArgs[i]; 189 } 190 } 191 192 if (!SetNamedPipeHandleState(hNamedPipe, pArgs[0], pArgs[1], pArgs[2])) 193 return PyErr_SetFromWindowsErr(0); 194 195 Py_RETURN_NONE; 196 196 } 197 197 … … 199 199 win32_WaitNamedPipe(PyObject *self, PyObject *args) 200 200 { 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 201 LPCTSTR lpNamedPipeName; 202 DWORD nTimeOut; 203 BOOL success; 204 205 if (!PyArg_ParseTuple(args, "s" F_DWORD, &lpNamedPipeName, &nTimeOut)) 206 return NULL; 207 208 Py_BEGIN_ALLOW_THREADS 209 success = WaitNamedPipe(lpNamedPipeName, nTimeOut); 210 Py_END_ALLOW_THREADS 211 212 if (!success) 213 return PyErr_SetFromWindowsErr(0); 214 215 Py_RETURN_NONE; 216 216 } 217 217 218 218 static PyMethodDef win32_methods[] = { 219 220 221 222 223 224 225 226 227 228 219 WIN32_FUNCTION(CloseHandle), 220 WIN32_FUNCTION(GetLastError), 221 WIN32_FUNCTION(OpenProcess), 222 WIN32_FUNCTION(ExitProcess), 223 WIN32_FUNCTION(ConnectNamedPipe), 224 WIN32_FUNCTION(CreateFile), 225 WIN32_FUNCTION(CreateNamedPipe), 226 WIN32_FUNCTION(SetNamedPipeHandleState), 227 WIN32_FUNCTION(WaitNamedPipe), 228 {NULL} 229 229 }; 230 230 231 231 232 232 PyTypeObject Win32Type = { 233 233 PyVarObject_HEAD_INIT(NULL, 0) 234 234 }; 235 235 … … 238 238 create_win32_namespace(void) 239 239 { 240 Win32Type.tp_name = "_multiprocessing.win32"; 241 Win32Type.tp_methods = win32_methods; 242 if (PyType_Ready(&Win32Type) < 0) 243 return NULL; 244 Py_INCREF(&Win32Type); 245 246 WIN32_CONSTANT(F_DWORD, ERROR_ALREADY_EXISTS); 247 WIN32_CONSTANT(F_DWORD, ERROR_PIPE_BUSY); 248 WIN32_CONSTANT(F_DWORD, ERROR_PIPE_CONNECTED); 249 WIN32_CONSTANT(F_DWORD, ERROR_SEM_TIMEOUT); 250 WIN32_CONSTANT(F_DWORD, GENERIC_READ); 251 WIN32_CONSTANT(F_DWORD, GENERIC_WRITE); 252 WIN32_CONSTANT(F_DWORD, INFINITE); 253 WIN32_CONSTANT(F_DWORD, NMPWAIT_WAIT_FOREVER); 254 WIN32_CONSTANT(F_DWORD, OPEN_EXISTING); 255 WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_DUPLEX); 256 WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_INBOUND); 257 WIN32_CONSTANT(F_DWORD, PIPE_READMODE_MESSAGE); 258 WIN32_CONSTANT(F_DWORD, PIPE_TYPE_MESSAGE); 259 WIN32_CONSTANT(F_DWORD, PIPE_UNLIMITED_INSTANCES); 260 WIN32_CONSTANT(F_DWORD, PIPE_WAIT); 261 WIN32_CONSTANT(F_DWORD, PROCESS_ALL_ACCESS); 262 263 WIN32_CONSTANT("i", NULL); 264 265 return (PyObject*)&Win32Type; 266 } 240 Win32Type.tp_name = "_multiprocessing.win32"; 241 Win32Type.tp_methods = win32_methods; 242 if (PyType_Ready(&Win32Type) < 0) 243 return NULL; 244 Py_INCREF(&Win32Type); 245 246 WIN32_CONSTANT(F_DWORD, ERROR_ALREADY_EXISTS); 247 WIN32_CONSTANT(F_DWORD, ERROR_NO_DATA); 248 WIN32_CONSTANT(F_DWORD, ERROR_PIPE_BUSY); 249 WIN32_CONSTANT(F_DWORD, ERROR_PIPE_CONNECTED); 250 WIN32_CONSTANT(F_DWORD, ERROR_SEM_TIMEOUT); 251 WIN32_CONSTANT(F_DWORD, GENERIC_READ); 252 WIN32_CONSTANT(F_DWORD, GENERIC_WRITE); 253 WIN32_CONSTANT(F_DWORD, INFINITE); 254 WIN32_CONSTANT(F_DWORD, NMPWAIT_WAIT_FOREVER); 255 WIN32_CONSTANT(F_DWORD, OPEN_EXISTING); 256 WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_DUPLEX); 257 WIN32_CONSTANT(F_DWORD, PIPE_ACCESS_INBOUND); 258 WIN32_CONSTANT(F_DWORD, PIPE_READMODE_MESSAGE); 259 WIN32_CONSTANT(F_DWORD, PIPE_TYPE_MESSAGE); 260 WIN32_CONSTANT(F_DWORD, PIPE_UNLIMITED_INSTANCES); 261 WIN32_CONSTANT(F_DWORD, PIPE_WAIT); 262 WIN32_CONSTANT(F_DWORD, PROCESS_ALL_ACCESS); 263 264 WIN32_CONSTANT("i", NULL); 265 266 return (PyObject*)&Win32Type; 267 }
Note:
See TracChangeset
for help on using the changeset viewer.