It Works Better with Thread Signaling

fried · fried · commit 127372ba0a6d · 2012-11-26T23:27:49.000-08:00
Signals are gone. I spawn a thread for notification of job done.
Then push the callback to python PendingCalls, I attempt to queue the
callback unlimited times, sleeping longer each time.

I have a hack in gevent.py to wake up the event loop if only one event
is running.

This works well on a gevent script with 32 workers all doing IO
diff --git a/README.markdown b/README.markdown
@@ -4,6 +4,10 @@ Python Asynchronous I/O bindings (aio.h)
 Version 0.3
 **Linux only**
 
+You should wait for the callback to finish before queuing more requests in
+a tight loop. Or pyaio could hang against python pending queue length
+
+
 Reading
 -------
 
diff --git a/pyaio/core.c b/pyaio/core.c
@@ -5,6 +5,7 @@
 #include <string.h>
 #include <fcntl.h>
 #include <errno.h>
+#include <unistd.h>
 
 typedef struct py_aio_callback {
     struct aiocb *cb;
@@ -18,93 +19,61 @@ PyDoc_STRVAR(pyaio_read_doc,
 PyDoc_STRVAR(pyaio_write_doc,
         "aio_write(fileno, buffer, offset, callback)\n");
 
-static void aio_read_completion_handler(int sig, siginfo_t *info, void *context)
+static int _async_callback(void *arg)
 {
-    Pyaio_cb *aio;
+    Pyaio_cb *aio = (Pyaio_cb *)arg;
     struct aiocb *cb;
-    PyObject *callback, *args;
+    PyObject *callback, *args, *result, *buffer;
     Py_ssize_t read_size = 0;
     Py_buffer pbuf;
 
-    PyGILState_STATE gstate;
-    gstate = PyGILState_Ensure();
-
-    aio = (Pyaio_cb*) info->si_value.sival_ptr;
-
     cb = aio->cb;
     callback = aio->callback;
+    buffer = aio->buffer;
 
-    /* Lets only let python know about how much was actually read */
-    if (aio_return(cb) > 0) {
-        read_size = aio_return(cb);
+    if (buffer == NULL) {
+        if (aio_return(cb) > 0) {
+            read_size = aio_return(cb);
+        }
+        /* Create a return buffer */
+        PyBuffer_FillInfo(&pbuf, 0, (void *)cb->aio_buf, read_size, 0,
+                            PyBUF_CONTIG);
+        args = Py_BuildValue("(Nni)", PyMemoryView_FromBuffer(&pbuf),
+                             aio_return(cb), aio_error(cb));
+    }
+    else { /* WRITE */
+        args = Py_BuildValue("(ni)", aio_return(cb), aio_error(cb));
     }
-
-    /* Create a buffer to return the data to python */
-    PyBuffer_FillInfo(&pbuf, 0, (void *)cb->aio_buf, read_size, 0, PyBUF_CONTIG);
-
-    args = Py_BuildValue("(Nni)", PyMemoryView_FromBuffer(&pbuf),
-                         aio_return(cb), aio_error(cb));
-
     Py_XINCREF(args);
-
-    //TODO add checking for null, kill program
-    PyObject_CallObject(callback, args);
-
+    result = PyObject_CallObject(callback, args);
+    if (result == NULL) {
+        printf("Exception in aio callback, dying!\n");
+        kill(getpid(), SIGKILL);  // DIE FAST
+    }
+    Py_XDECREF(result);
     Py_XDECREF(callback);
     Py_XDECREF(args);
-
+    if (buffer != NULL) {
+        Py_XDECREF(buffer);
+    }
     free((struct aiocb *)cb);
     free(aio);
-
-    PyGILState_Release(gstate);
+    return 0;
 }
 
-static void aio_write_completion_handler(int sig, siginfo_t *info, void *context)
+static void aio_completion_handler(sigval_t sigval)
 {
     Pyaio_cb *aio;
-    struct aiocb *cb;
-    PyObject *callback, *args;
-
-    PyGILState_STATE gstate;
-    gstate = PyGILState_Ensure();
-
-    aio = (Pyaio_cb*) info->si_value.sival_ptr;
-    cb = aio->cb;
-    callback = aio->callback;
-    args = Py_BuildValue("(ni)", aio_return(cb), aio_error(cb));
-
-    Py_XDECREF(aio->buffer);
-    Py_XINCREF(args);
-
-    //TODO add checking for return null, kill program
-    PyObject_CallObject(callback, args);
-
-    Py_XDECREF(callback);
-    Py_XDECREF(args);
+    int tries = 1;
+    aio = (Pyaio_cb*) sigval.sival_ptr;
 
-    free((struct aiocb *)cb);
-    free(aio);
-
-    PyGILState_Release(gstate);
-}
+    //We should set an upper limit like 50 retries or something
+    while(Py_AddPendingCall(&_async_callback, aio) < 0) {
+        usleep(500*(tries/2)); //Step off timer
+        tries += 1;
+    }
 
-static void init_sig_handlers(void)
-{
-    struct sigaction *sa;
-
-    /* Install Read Handler */
-    sa = malloc(sizeof(struct sigaction));
-    sigemptyset(&sa->sa_mask);
-    sa->sa_sigaction = aio_read_completion_handler;
-    sa->sa_flags = SA_SIGINFO | SA_RESTART;
-    sigaction(SIGRTMIN+1, sa, NULL);
-
-    /* Install Write Handler */
-    sa = malloc(sizeof(struct sigaction));
-    sigemptyset(&sa->sa_mask);
-    sa->sa_sigaction = aio_write_completion_handler;
-    sa->sa_flags = SA_SIGINFO | SA_RESTART;
-    sigaction(SIGRTMIN+2, sa, NULL);
+    return;
 }
 
 static PyObject *
@@ -115,7 +84,7 @@ pyaio_read(PyObject *dummy, PyObject *args) {
 
     Pyaio_cb *aio;
     PyObject *callback, *return_;
-
+    Py_XINCREF(args);
     if (PyArg_ParseTuple(args, "innO:set_callback", &fd, &offset, &numbytes,
                                  &callback)) {
         if (!PyCallable_Check(callback)) {
@@ -125,21 +94,22 @@ pyaio_read(PyObject *dummy, PyObject *args) {
         }
         Py_XINCREF(callback); /* Add a reference to new callback */
     }
-
+    Py_XDECREF(args);
     aio = malloc(sizeof(Pyaio_cb));
 
     aio->cb = malloc(sizeof(struct aiocb));
     bzero((char *) aio->cb, sizeof(struct aiocb));
 
     aio->callback = callback;
+    aio->buffer = NULL;
 
     aio->cb->aio_buf = malloc((numbytes) * sizeof(char));
     aio->cb->aio_fildes = fd;
     aio->cb->aio_nbytes = numbytes;
     aio->cb->aio_offset = offset;
-    aio->cb->aio_sigevent.sigev_notify =  SIGEV_SIGNAL;
-    aio->cb->aio_sigevent.sigev_signo = SIGRTMIN+1;
+    aio->cb->aio_sigevent.sigev_notify = SIGEV_THREAD;  /* EvIL */
     aio->cb->aio_sigevent.sigev_notify_attributes = NULL;
+    aio->cb->aio_sigevent.sigev_notify_function = aio_completion_handler;
     aio->cb->aio_sigevent.sigev_value.sival_ptr = aio;
 
     ret = aio_read(aio->cb);
@@ -165,7 +135,7 @@ pyaio_write(PyObject *dummy, PyObject *args) {
 
     Pyaio_cb *aio;
     PyObject *callback, *return_;
-
+    Py_XINCREF(args);
     if (PyArg_ParseTuple(args, "iOnO:set_callback", &fd, &buffer,
                           &offset, &callback)) {
         if (!PyCallable_Check(callback)) {
@@ -180,7 +150,7 @@ pyaio_write(PyObject *dummy, PyObject *args) {
         }
         Py_XINCREF(callback); /* Add a reference to new callback */
     }
-
+    Py_XDECREF(args);
     /* Get a Memoryview */
     if (!PyMemoryView_Check(buffer)) {
         buffer = PyMemoryView_GetContiguous(buffer, PyBUF_READ, 'C');
@@ -199,9 +169,9 @@ pyaio_write(PyObject *dummy, PyObject *args) {
     aio->cb->aio_fildes = fd;
     aio->cb->aio_nbytes = buffer_view->len;
     aio->cb->aio_offset = offset;
-    aio->cb->aio_sigevent.sigev_notify =  SIGEV_SIGNAL;
-    aio->cb->aio_sigevent.sigev_signo = SIGRTMIN+2;
+    aio->cb->aio_sigevent.sigev_notify = SIGEV_THREAD;
     aio->cb->aio_sigevent.sigev_notify_attributes = NULL;
+    aio->cb->aio_sigevent.sigev_notify_function = aio_completion_handler;
     aio->cb->aio_sigevent.sigev_value.sival_ptr = aio;
 
     ret = aio_write(aio->cb);
@@ -273,8 +243,6 @@ init_pyaio(void) {
         return NULL;
     }
 
-    init_sig_handlers();
-
     return m;
 }
 
diff --git a/pyaio/gevent.py b/pyaio/gevent.py
@@ -1,19 +1,24 @@
 from __future__ import absolute_import
 import os
 import pyaio
+import gevent
 from gevent.event import AsyncResult
+from gevent.coros import RLock
 
 class aioFile(object):
     """a buffered File like object that uses pyaio and gevent"""
     def __init__(self, filename, mode='r', buffer=16<<10):
         modes = os.O_LARGEFILE | os.O_CREAT
         self._offset = 0
         self._buffer_size = buffer
+        if buffer:
+            self._buffer_lock = RLock()
         self._read = False
         self._write = False
         self._read_buf = None
         self._write_buf = None
         self._eof = False   # Optimization to limit calls
+        self._append = False   # Append Mode writes ignore offset
 
         if mode.startswith('r') or '+' in mode:
             self._read = True
@@ -30,9 +35,10 @@ def __init__(self, filename, mode='r', buffer=16<<10):
                 modes |= os.O_WRONLY
         if '+' in mode:
             modes |= os.O_RDWR
-        self._fd = os.open(filename, modes)
         if mode.startswith('a'):
-            self.seek(0, os.SEEK_END)  # Append so goto end
+            modes |= os.O_APPEND
+            self._append = True
+        self._fd = os.open(filename, modes)
 
     def _clear_read_buf(self):
         if self._read:
@@ -70,7 +76,8 @@ def seek(self, pos, how=os.SEEK_SET):
             raise OSError(14, 'File Position invalid, less than 0')
         #Even if the pos didn't change fix the buffers and EOF
         self._clear_read_buf()
-        self.flush()
+        if not self._append:   # DON'T FLUSH on seek with append
+            self.flush()
         self._offset = offset
         return offset
 
@@ -95,30 +102,51 @@ def write(self, buf, offset=None):
         """write a buffer object to file"""
         if not self._write:
             raise IOError(9, 'Bad file descriptor')
-        if self._buffer_size and self._read_buf:  # We should clear read cache
+        if not self._append and self._buffer_size and self._read_buf:
+                # We should clear read cache
             self._clear_read_buf()
         if offset is None:
             offset = self._offset
         write_size = self._buffer_size
         if not self._buffer_size and buf:
             write_size = len(buf)
-        if offset != self._offset:
+        if not self._append and offset != self._offset:
             self.seek(offset)  # Makes sure we write our buffer
-        if buf:
-            self._write_buf.extend(buf)
-        while len(self._write_buf) >= self._buffer_size \
-                or (self._flush and self._write_buf):
+
+        #If we buffer we use the global buffer if not we use a local buffer
+        if self._buffer_size:
+            lbuf = self._write_buf
+            self._buffer_lock.acquire()
+            if buf:
+                                          # The a memoryview of the buffer
+                    lbuf.extend(buf)      # pushed to pyaio so we need to lock
+        else:
+            lbuf = buf
+
+        while lbuf and len(lbuf) >= self._buffer_size \
+                or (self._flush and lbuf):
             result = AsyncResult()
             def _write_results(rcode, errno):
                 result.set((rcode, errno))
-            pyaio.aio_write(self._fd, memoryview(self._write_buf)[0:write_size],
+            pyaio.aio_write(self._fd, memoryview(lbuf)[0:write_size],
                             offset, _write_results)
-            rcode, errno = result.get()  # gevent yield
+            #WARNING THIS IS A DIRTY DIRTY TRICK
+            def _no_op():
+                pass
+            #100 micro Seconds In the Future to be exact
+            gevent.spawn_later(0.0001, _no_op);  # WAKE THE EVENT LOOP
+            rcode, errno = result.get()  #SLEEP
+
             if rcode < 0:   # Some kind of error
                 raise IOError(errno, 'AIO Write Error %d' % errno)
             # Clean up buffer (of actually written bytes)
-            del self._write_buf[0:rcode]
+            if self._buffer_size:
+                del lbuf[0:rcode]
+            else:
+                lbuf = None
             self._offset = offset = offset + rcode  # Move the file offset
+        if self._buffer_size:
+            self._buffer_lock.release()
         if buf:
             return len(buf)
         else:
@@ -129,7 +157,7 @@ def read(self, size=0, offset=None):
         """for speed we assume EOF after first short read"""
         if not self._read:
             raise IOError(9, 'Bad file descriptor')
-        if self._buffer_size and self._write_buf:
+        if not self._append and self._buffer_size and self._write_buf:
             self.flush()
         if offset is None:
             offset = self._offset
@@ -138,7 +166,7 @@ def read(self, size=0, offset=None):
         if size == 0:  # Attempt to read entire file and return in a single return
             return self._read_file()
         else:
-            rbuf = bytearray()
+            rbuf = bytearray()  # Holding Place for multiple reads
             while len(rbuf) < size:  # People get what they ask for
                 # If we don't want to buffer then just read what they want
                 if len(self._read_buf) < size - len(rbuf) and not self._eof:
@@ -150,19 +178,28 @@ def _read_results(buf, rcode, errno):
                     if self._buffer_size:   # If we buffer read buffer instead
                         read_size = self._buffer_size
                     pyaio.aio_read(self._fd, offset, read_size, _read_results)
-                    buf, rcode, errno = result.get()  #gevent YIELD :)
+                    #WARNING THIS IS A DIRTY DIRTY TRICK
+                    def _no_op():
+                        pass
+                    #100 micro Seconds In the Future to be exact
+                    gevent.spawn_later(0.0001, _no_op);  # WAKE THE EVENT LOOP
+                    buf, rcode, errno = result.get()  #SLEEP
                     if rcode < 0:  # Some kind of error
                         raise IOError(errno, 'AIO Read Error %d' % errno)
                     #Rcode will be the bytes read so lets push the offset
                     self._offset = offset = offset + rcode
-                    self._read_buf.extend(buf)
+                    if self._buffer_size:
+                        self._read_buf.extend(buf)
+                    else:
+                        rbuf = buf  # Pass through because we are not buffering
                     if rcode == 0 or rcode < read_size:  # Good Enough
                         self._eof = True
                 #Do a buffer read
                 toread = size - len(rbuf)
-                rbuf.extend(memoryview(self._read_buf)[0:toread])
-                #Clean up read buffer
-                del self._read_buf[0:toread]
+                if self._buffer_size:
+                    rbuf.extend(memoryview(self._read_buf)[0:toread])
+                    #Clean up read buffer
+                    del self._read_buf[0:toread]
                 if not self._read_buf and self._eof:  # Empty buffer and eof
                     break
             if self._eof and not rbuf:
diff --git a/setup.py b/setup.py
@@ -2,8 +2,7 @@
 import sys
 
 if sys.version < '2.7':
-    print "pyaio requires python of at least version 2.7"
-    sys.exit(1)
+    print "pyaio requires python of at least version 2.7 to build correctly"
 
 version = '0.3'
 
diff --git a/tests/test_pyaio.py b/tests/test_pyaio.py
