1 /* 2 * A type which wraps a pipe handle in message oriented mode 3 * 4 * pipe_connection.c 5 * 6 * Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt 7 */ 8 9 #include "multiprocessing.h" 10 11 #define CLOSE(h) CloseHandle(h) 12 13 /* 14 * Send string to the pipe; assumes in message oriented mode 15 */ 16 17 static Py_ssize_t 18 conn_send_string(ConnectionObject *conn, char *string, size_t length) 19 { 20 DWORD amount_written; 21 BOOL ret; 22 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; 33 } 34 35 /* 36 * Attempts to read into buffer, or if buffer too small into *newbuffer. 37 * 38 * Returns number of bytes read. Assumes in message oriented mode. 39 */ 40 41 static Py_ssize_t 42 conn_recv_string(ConnectionObject *conn, char *buffer, 43 size_t buflength, char **newbuffer, size_t maxlength) 44 { 45 DWORD left, length, full_length, err; 46 BOOL ret; 47 *newbuffer = NULL; 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; 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 } 62 63 if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, NULL, &left)) 64 return MP_STANDARD_ERROR; 65 66 full_length = length + left; 67 if (full_length > maxlength) 68 return MP_BAD_MESSAGE_LENGTH; 69 70 *newbuffer = PyMem_Malloc(full_length); 71 if (*newbuffer == NULL) 72 return MP_MEMORY_ERROR; 73 74 memcpy(*newbuffer, buffer, length); 75 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 } 86 } 87 88 /* 89 * Check whether any data is available for reading 90 */ 91 92 static int 93 conn_poll(ConnectionObject *conn, double timeout, PyThreadState *_save) 94 { 95 DWORD bytes, deadline, delay; 96 int difference, res; 97 BOOL block = FALSE; 98 99 if (!PeekNamedPipe(conn->handle, NULL, 0, NULL, &bytes, NULL)) 100 return MP_STANDARD_ERROR; 101 102 if (timeout == 0.0) 103 return bytes > 0; 104 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); 110 111 Sleep(0); 112 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; 118 119 if (!block) { 120 difference = deadline - GetTickCount(); 121 if (difference < 0) 122 return FALSE; 123 if ((int)delay > difference) 124 delay = difference; 125 } 126 127 if (delay > 20) 128 delay = 20; 129 130 Sleep(delay); 131 132 /* check for signals */ 133 Py_BLOCK_THREADS 134 res = PyErr_CheckSignals(); 135 Py_UNBLOCK_THREADS 136 137 if (res) 138 return MP_EXCEPTION_HAS_BEEN_SET; 139 } 140 } 141 142 /* 143 * "connection.h" defines the PipeConnection type using the definitions above 144 */ 145 146 #define CONNECTION_NAME "PipeConnection" 147 #define CONNECTION_TYPE PipeConnectionType 148 149 #include "connection.h" 150