basic threading model

git-svn-id: file:///home/lennart/svn/public/libsydney/trunk@27 9ba3c220-e4d3-45a2-8aa3-73fcc9aff6ce

1 files changed, 95 insertions, 2 deletions

diff --git a/oss.c b/oss.c
index b1b1168..838e0f4 100644
--- a/oss.c
+++ b/oss.c
@@ -1,9 +1,13 @@
+#include <signal.h>
 #include <string.h>
 #include <unistd.h>
 #include <fcntl.h>
 #include <errno.h>
 #include <sys/ioctl.h>
 #include <sys/soundcard.h>
+#include <poll.h>
+#include <sys/types.h>
+#include <sys/socket.h>
 
 #include "sydney.h"
 #include "common.h"
@@ -11,6 +15,7 @@
 #include "malloc.h"
 #include "converter.h"
 #include "driver.h"
+#include "thread.h"
 
 #define DEFAULT_DEVICE "/dev/dsp"
 #define DRIVER_NAME "oss"
@@ -25,6 +30,8 @@ struct oss_stream {
     converter_t converter_read, converter_write;
     size_t read_fragment_size, write_fragment_size;
     unsigned read_nfragments, write_nfragments;
+    sa_thread_t *thread;
+    int socket_fds[2];
 };
 
 static int simple_log2(int v) {
@@ -73,6 +80,7 @@ int driver_open(sa_stream_t *s) {
         return SA_ERROR_OOM;
 
     oss->parent = s;
+    oss->socket_fds[0] = oss->socket_fds[1] = -1;
 
     n = s->device ? s->device : DEFAULT_DEVICE;
     if (!s->codec)
@@ -556,20 +564,105 @@ int driver_destroy(sa_stream_t *s) {
     oss_stream_t *oss = OSS_STREAM(s);
 
     if (oss) {
+
+        if (oss->thread)
+            driver_stop_thread(s);
+        
         if (oss->fd >= 0)
             close(oss->fd);
         
         sa_free(oss->real_pcm_attrs.channel_map);
         converter_done(&oss->converter_read);
         converter_done(&oss->converter_write);
+
         sa_free(oss);
     }
     
     return SA_SUCCESS;
 }
 
-int driver_start_thread(sa_stream_t *s, sa_event_callback_t *callback) {
-    return SA_ERROR_NOT_SUPPORTED;
+enum {
+    POLLFD_OSS_FD,
+    POLLFD_SOCKET_FD,
+    POLLFD_MAX
+};
+
+static void thread_func(void *data) {
+    struct pollfd pollfds[POLLFD_MAX];
+    sa_stream_t *s = data;
+    oss_stream_t *oss = OSS_STREAM(s);
+    sigset_t mask;
+
+    sigfillset(&mask);
+    pthread_sigmask(SIG_BLOCK, &mask, NULL);
+    
+    s->event = SA_EVENT_ERROR;
+    if (s->callback(s, SA_EVENT_INIT_THREAD) < 0)
+        return;
+    
+    memset(pollfds, 0, sizeof(pollfds));
+
+    pollfds[POLLFD_SOCKET_FD].fd = oss->socket_fds[0];
+    pollfds[POLLFD_SOCKET_FD].events = POLLIN;
+    
+    pollfds[POLLFD_OSS_FD].fd = oss->fd;
+    pollfds[POLLFD_OSS_FD].events = ((s->mode & SA_MODE_RDONLY) ? POLLIN : 0) | ((s->mode & SA_MODE_WRONLY) ? POLLOUT : 0);
+    
+    for (;;) {
+        if (poll(pollfds, POLLFD_MAX, -1) < 0) {
+            if (errno == EINTR)
+                continue;
+        }
+
+        if (pollfds[POLLFD_SOCKET_FD].revents)
+            break;
+
+        if (pollfds[POLLFD_OSS_FD].revents & (POLLERR|POLLHUP|POLLNVAL)) {
+            s->event = SA_EVENT_ERROR;
+            s->error = SA_ERROR_SYSTEM;
+            errno = EIO;
+            s->callback(s, SA_EVENT_ERROR);
+            break;
+        }
+
+        s->event = SA_EVENT_REQUEST_IO;
+        s->callback(s, SA_EVENT_REQUEST_IO);
+    }
+}
+
+int driver_start_thread(sa_stream_t *s, sa_event_callback_t callback) {
+    oss_stream_t *oss = OSS_STREAM(s);
+    sa_return_val_if_fail(!oss->thread, SA_ERROR_STATE);
+
+    s->callback = callback;
+
+    if ((socketpair(AF_UNIX, SOCK_DGRAM, 0, oss->socket_fds)) < 0)
+        return SA_ERROR_SYSTEM;
+
+    if (!(oss->thread = sa_thread_new(thread_func, s)))
+        return SA_ERROR_OOM;
+    
+    return SA_SUCCESS;
+}
+
+int driver_stop_thread(sa_stream_t *s) {
+    oss_stream_t *oss = OSS_STREAM(s);
+    sa_return_val_if_fail(oss->thread, SA_ERROR_STATE);
+    sa_return_val_if_fail(oss->thread != sa_thread_self(), SA_ERROR_STATE);
+
+    if (oss->socket_fds[0] >= 0)
+        close(oss->socket_fds[0]);
+    
+    if (oss->socket_fds[1] >= 0)
+        close(oss->socket_fds[1]);
+    
+    if (oss->thread)
+        sa_thread_free(oss->thread);
+    
+    oss->thread = NULL;
+    oss->socket_fds[0] = oss->socket_fds[1] = -1;
+
+    return SA_SUCCESS;
 }
 
 int driver_change_read_volume(sa_stream_t *s, const int32_t vol[]) {