Handle pipes on platforms where they are non-existant of broken.

We do this by creating a TCP socket pair instead of a normal pipe. Since
Windows isn't UNIX-y enough to support read()/write() on sockets, we also
need a wrapper to handle read() vs recv() and write() vs send().


git-svn-id: file:///home/lennart/svn/public/pulseaudio/trunk@840 fefdeb5f-60dc-0310-8127-8f9354f1896f

9 files changed, 249 insertions, 107 deletions

diff --git a/configure.ac b/configure.ac
index 22ab3962..360249ea 100644
--- a/configure.ac
+++ b/configure.ac
@@ -220,7 +220,7 @@ AC_FUNC_FORK
 AC_FUNC_GETGROUPS
 AC_FUNC_SELECT_ARGTYPES
 AC_CHECK_FUNCS([chmod chown getaddrinfo getgrgid_r getpwuid_r gettimeofday \
-    getuid inet_ntop inet_pton nanosleep setpgid setsid sigaction sleep])
+    getuid inet_ntop inet_pton nanosleep pipe setpgid setsid sigaction sleep])
 AC_CHECK_FUNCS([mkfifo], [HAVE_MKFIFO=1], [HAVE_MKFIFO=0])
 
 AM_CONDITIONAL(HAVE_MKFIFO, test "x$HAVE_MKFIFO" = "x1")
diff --git a/src/Makefile.am b/src/Makefile.am
index ee7b2325..72e3c1eb 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -356,6 +356,7 @@ libpolyp_@PA_MAJORMINOR@_la_SOURCES = \
 		polyp/mainloop.c polyp/mainloop.h \
 		polyp/mainloop-signal.c polyp/mainloop-signal.h \
 		polyp/thread-mainloop.c polyp/thread-mainloop.h \
+		polypcore/pipe.c polypcore/pipe.h \
 		polypcore/poll.c polypcore/poll.h
 
 # Internal stuff that is shared with libpolypcore
@@ -518,6 +519,7 @@ libpolypcore_la_SOURCES += \
 		polypcore/module.c polypcore/module.h \
 		polypcore/namereg.c polypcore/namereg.h \
 		polypcore/pid.c polypcore/pid.h \
+		polypcore/pipe.c polypcore/pipe.h \
 		polypcore/play-memchunk.c polypcore/play-memchunk.h \
 		polypcore/poll.c polypcore/poll.h \
 		polypcore/props.c polypcore/props.h \
diff --git a/src/polyp/mainloop-signal.c b/src/polyp/mainloop-signal.c
index 73de0002..6073dbb3 100644
--- a/src/polyp/mainloop-signal.c
+++ b/src/polyp/mainloop-signal.c
@@ -59,25 +59,17 @@ struct pa_signal_event {
 static pa_mainloop_api *api = NULL;
 static int signal_pipe[2] = { -1, -1 };
 static pa_io_event* io_event = NULL;
-static pa_time_event *time_event = NULL;
 static pa_signal_event *signals = NULL;
 
-#ifdef OS_IS_WIN32
-static unsigned int waiting_signals = 0;
-static CRITICAL_SECTION crit;
-#endif
-
 static void signal_handler(int sig) {
+    int result;
 #ifndef HAVE_SIGACTION
     signal(sig, signal_handler);
 #endif
-    write(signal_pipe[1], &sig, sizeof(sig));
-
-#ifdef OS_IS_WIN32
-    EnterCriticalSection(&crit);
-    waiting_signals++;
-    LeaveCriticalSection(&crit);
-#endif
+    pa_log(__FILE__": Got signal %d", sig);
+    result = pa_write(signal_pipe[1], &sig, sizeof(sig));
+    if (result != sizeof(sig))
+        pa_log(__FILE__": Bad write (%d, %d)", result, WSAGetLastError());
 }
 
 static void dispatch(pa_mainloop_api*a, int sig) {
@@ -91,46 +83,14 @@ static void dispatch(pa_mainloop_api*a, int sig) {
         }
 }
 
-#ifdef OS_IS_WIN32
-static void timer(pa_mainloop_api*a, pa_time_event*e, PA_GCC_UNUSED const struct timeval *tv, PA_GCC_UNUSED void *userdata) {
-    ssize_t r;
-    int sig;
-    unsigned int sigs;
-    struct timeval tvnext;
-
-    EnterCriticalSection(&crit);
-    sigs = waiting_signals;
-    waiting_signals = 0;
-    LeaveCriticalSection(&crit);
-
-    while (sigs) {
-        if ((r = read(signal_pipe[0], &sig, sizeof(sig))) < 0) {
-            pa_log(__FILE__": read(): %s", strerror(errno));
-            return;
-        }
-        
-        if (r != sizeof(sig)) {
-            pa_log(__FILE__": short read()");
-            return;
-        }
-
-        dispatch(a, sig);
-
-        sigs--;
-    }
-
-    pa_timeval_add(pa_gettimeofday(&tvnext), 100000);
-    a->time_restart(e, &tvnext);
-}
-#endif
-
 static void callback(pa_mainloop_api*a, pa_io_event*e, int fd, pa_io_event_flags_t f, PA_GCC_UNUSED void *userdata) {
     ssize_t r;
     int sig;
     assert(a && e && f == PA_IO_EVENT_INPUT && e == io_event && fd == signal_pipe[0]);
 
-        
-    if ((r = read(signal_pipe[0], &sig, sizeof(sig))) < 0) {
+    pa_log(__FILE__": Signal pipe callback");
+
+    if ((r = pa_read(signal_pipe[0], &sig, sizeof(sig))) < 0) {
         if (errno == EAGAIN)
             return;
 
@@ -147,17 +107,10 @@ static void callback(pa_mainloop_api*a, pa_io_event*e, int fd, pa_io_event_flags
 }
 
 int pa_signal_init(pa_mainloop_api *a) {
-#ifdef OS_IS_WIN32
-    struct timeval tv;
-#endif
 
-    assert(!api && a && signal_pipe[0] == -1 && signal_pipe[1] == -1 && !io_event && !time_event);
+    assert(!api && a && signal_pipe[0] == -1 && signal_pipe[1] == -1 && !io_event);
 
-#ifdef OS_IS_WIN32
-    if (_pipe(signal_pipe, 200, _O_BINARY) < 0) {
-#else
     if (pipe(signal_pipe) < 0) {
-#endif
         pa_log(__FILE__": pipe() failed: %s", strerror(errno));
         return -1;
     }
@@ -169,34 +122,20 @@ int pa_signal_init(pa_mainloop_api *a) {
 
     api = a;
 
-#ifndef OS_IS_WIN32
     io_event = api->io_new(api, signal_pipe[0], PA_IO_EVENT_INPUT, callback, NULL);
     assert(io_event);
-#else
-    time_event = api->time_new(api, pa_gettimeofday(&tv), timer, NULL);
-    assert(time_event);
-
-    InitializeCriticalSection(&crit);
-#endif
 
     return 0;
 }
 
 void pa_signal_done(void) {
-    assert(api && signal_pipe[0] >= 0 && signal_pipe[1] >= 0 && (io_event || time_event));
+    assert(api && signal_pipe[0] >= 0 && signal_pipe[1] >= 0 && io_event);
 
     while (signals)
         pa_signal_free(signals);
     
-#ifndef OS_IS_WIN32
     api->io_free(io_event);
     io_event = NULL;
-#else
-    api->time_free(time_event);
-    time_event = NULL;
-
-    DeleteCriticalSection(&crit);
-#endif
 
     close(signal_pipe[0]);
     close(signal_pipe[1]);
diff --git a/src/polyp/mainloop.c b/src/polyp/mainloop.c
index 98b3f3c6..589fe77e 100644
--- a/src/polyp/mainloop.c
+++ b/src/polyp/mainloop.c
@@ -40,6 +40,10 @@
 
 #include "../polypcore/winsock.h"
 
+#ifndef HAVE_PIPE
+#include "../polypcore/pipe.h"
+#endif
+
 #include <polypcore/util.h>
 #include <polypcore/idxset.h>
 #include <polypcore/xmalloc.h>
@@ -330,18 +334,14 @@ pa_mainloop *pa_mainloop_new(void) {
 
     m = pa_xmalloc(sizeof(pa_mainloop));
 
-#ifndef OS_IS_WIN32
     if (pipe(m->wakeup_pipe) < 0) {
+        pa_log_error(__FILE__": ERROR: cannot create wakeup pipe");
         pa_xfree(m);
         return NULL;
     }
 
     pa_make_nonblock_fd(m->wakeup_pipe[0]);
     pa_make_nonblock_fd(m->wakeup_pipe[1]);
-#else
-    m->wakeup_pipe[0] = -1;
-    m->wakeup_pipe[1] = -1;
-#endif
 
     m->io_events = pa_idxset_new(NULL, NULL);
     m->defer_events = pa_idxset_new(NULL, NULL);
@@ -622,7 +622,7 @@ void pa_mainloop_wakeup(pa_mainloop *m) {
     assert(m);
 
     if (m->wakeup_pipe[1] >= 0)
-        write(m->wakeup_pipe[1], &c, sizeof(c));
+        pa_write(m->wakeup_pipe[1], &c, sizeof(c));
 }
 
 static void clear_wakeup(pa_mainloop *m) {
@@ -633,7 +633,7 @@ static void clear_wakeup(pa_mainloop *m) {
     if (m->wakeup_pipe[0] < 0)
         return;
 
-    while (read(m->wakeup_pipe[0], &c, sizeof(c)) == sizeof(c));
+    while (pa_read(m->wakeup_pipe[0], &c, sizeof(c)) == sizeof(c));
 }
 
 int pa_mainloop_prepare(pa_mainloop *m, int timeout) {
diff --git a/src/polypcore/iochannel.c b/src/polypcore/iochannel.c
index 5da7a9a9..623925ac 100644
--- a/src/polypcore/iochannel.c
+++ b/src/polypcore/iochannel.c
@@ -202,19 +202,7 @@ ssize_t pa_iochannel_write(pa_iochannel*io, const void*data, size_t l) {
     assert(l);
     assert(io->ofd >= 0);
 
-#ifdef OS_IS_WIN32
-    r = send(io->ofd, data, l, 0);
-    if (r < 0) {
-        if (WSAGetLastError() != WSAENOTSOCK) {
-            errno = WSAGetLastError();
-            return r;
-        }
-    }
-
-    if (r < 0)
-#endif
-        r = write(io->ofd, data, l);
-
+    r = pa_write(io->ofd, data, l);
     if (r >= 0) {
         io->writable = 0;
         enable_mainloop_sources(io);
@@ -229,20 +217,8 @@ ssize_t pa_iochannel_read(pa_iochannel*io, void*data, size_t l) {
     assert(io);
     assert(data);
     assert(io->ifd >= 0);
-    
-#ifdef OS_IS_WIN32
-    r = recv(io->ifd, data, l, 0);
-    if (r < 0) {
-        if (WSAGetLastError() != WSAENOTSOCK) {
-            errno = WSAGetLastError();
-            return r;
-        }
-    }
 
-    if (r < 0)
-#endif
-        r = read(io->ifd, data, l);
-    
+    r = pa_read(io->ifd, data, l);
     if (r >= 0) {
         io->readable = 0;
         enable_mainloop_sources(io);
diff --git a/src/polypcore/pipe.c b/src/polypcore/pipe.c
new file mode 100644
index 00000000..eef6d533
--- /dev/null
+++ b/src/polypcore/pipe.c
@@ -0,0 +1,160 @@
+/* $Id$ */
+
+/***
+   This file is part of polypaudio.
+
+   polypaudio is free software; you can redistribute it and/or modify it
+   under the terms of the GNU Library General Public License as published
+   by the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   polypaudio is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Library General Public License for more details.
+
+   You should have received a copy of the GNU Library General Public
+   License along with polypaudio; If not, write to the Free Software
+   Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
+   USA.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <errno.h>
+#include <unistd.h>
+#include <fcntl.h>
+
+#include <sys/types.h>
+
+#ifdef HAVE_SYS_SOCKET_H
+#include <sys/socket.h>
+#endif
+
+#include "winsock.h"
+
+#include "pipe.h"
+
+#ifndef HAVE_PIPE
+
+static int set_block(int fd, int blocking) {
+#ifdef O_NONBLOCK
+
+    int v;
+
+    assert(fd >= 0);
+
+    if ((v = fcntl(fd, F_GETFL)) < 0)
+        return -1;
+
+    if (blocking)
+        v &= ~O_NONBLOCK;
+    else
+        v |= O_NONBLOCK;
+
+    if (fcntl(fd, F_SETFL, v) < 0)
+        return -1;
+
+    return 0;
+
+#elif defined(OS_IS_WIN32)
+
+    u_long arg;
+
+    arg = !blocking;
+
+    if (ioctlsocket(fd, FIONBIO, &arg) < 0)
+        return -1;
+
+    return 0;
+
+#else
+
+    return -1;
+
+#endif
+}
+
+int pipe(int filedes[2]) {
+    int listener;
+    struct sockaddr_in addr, peer;
+    socklen_t len;
+
+    listener = -1;
+    filedes[0] = -1;
+    filedes[1] = -1;
+
+    listener = socket(PF_INET, SOCK_STREAM, 0);
+    if (listener < 0)
+        goto error;
+
+    filedes[0] = socket(PF_INET, SOCK_STREAM, 0);
+    if (filedes[0] < 0)
+        goto error;
+
+    filedes[1] = socket(PF_INET, SOCK_STREAM, 0);
+    if (filedes[1] < 0)
+        goto error;
+
+    /* Make non-blocking so that connect() won't block */
+    if (set_block(filedes[0], 0) < 0)
+        goto error;
+
+    addr.sin_family = AF_INET;
+    addr.sin_port = 0;
+    addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+
+    if (bind(listener, (struct sockaddr*)&addr, sizeof(addr)) < 0)
+        goto error;
+
+    if (listen(listener, 1) < 0)
+        goto error;
+
+    len = sizeof(addr);
+    if (getsockname(listener, (struct sockaddr*)&addr, &len) < 0)
+        goto error;
+
+    if (connect(filedes[0], (struct sockaddr*)&addr, sizeof(addr)) < 0) {
+#ifdef OS_IS_WIN32
+        if (WSAGetLastError() != EWOULDBLOCK)
+#else
+        if (errno != EINPROGRESS)
+#endif
+            goto error;
+    }
+
+    len = sizeof(peer);
+    filedes[1] = accept(listener, (struct sockaddr*)&peer, &len);
+    if (filedes[1] < 0)
+        goto error;
+
+    /* Restore blocking */
+    if (set_block(filedes[0], 1) < 0)
+        goto error;
+
+    len = sizeof(addr);
+    if (getsockname(filedes[0], (struct sockaddr*)&addr, &len) < 0)
+        goto error;
+
+    /* Check that someone else didn't steal the connection */
+    if ((addr.sin_port != peer.sin_port) || (addr.sin_addr.s_addr != peer.sin_addr.s_addr))
+        goto error;
+
+    close(listener);
+
+    return 0;
+
+error:
+	if (listener >= 0)
+		close(listener);
+	if (filedes[0] >= 0)
+		close(filedes[0]);
+	if (filedes[1] >= 0)
+		close(filedes[0]);
+
+	return -1;
+}
+
+#endif /* HAVE_PIPE */
diff --git a/src/polypcore/pipe.h b/src/polypcore/pipe.h
new file mode 100644
index 00000000..a9b088b5
--- /dev/null
+++ b/src/polypcore/pipe.h
@@ -0,0 +1,22 @@
+/* $Id$ */
+
+/***
+   This file is part of polypaudio.
+
+   polypaudio is free software; you can redistribute it and/or modify it
+   under the terms of the GNU Library General Public License as published
+   by the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   polypaudio is distributed in the hope that it will be useful, but
+   WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+   Library General Public License for more details.
+
+   You should have received a copy of the GNU Library General Public
+   License along with polypaudio; If not, write to the Free Software
+   Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
+   USA.
+***/
+
+int pipe(int filedes[2]);
diff --git a/src/polypcore/util.c b/src/polypcore/util.c
index 191fa809..beb93f45 100644
--- a/src/polypcore/util.c
+++ b/src/polypcore/util.c
@@ -198,6 +198,47 @@ finish:
     return ret;
 }
 
+/** Platform independent read function. Necessary since not all systems
+ * treat all file descriptors equal. */
+ssize_t pa_read(int fd, void *buf, size_t count) {
+    ssize_t r;
+
+#ifdef OS_IS_WIN32
+    r = recv(fd, buf, count, 0);
+    if (r < 0) {
+        if (WSAGetLastError() != WSAENOTSOCK) {
+            errno = WSAGetLastError();
+            return r;
+        }
+    }
+
+    if (r < 0)
+#endif
+        r = read(fd, buf, count);
+
+    return r;
+}
+
+/** Similar to pa_read(), but handles writes */
+ssize_t pa_write(int fd, void *buf, size_t count) {
+    ssize_t r;
+
+#ifdef OS_IS_WIN32
+    r = send(fd, buf, count, 0);
+    if (r < 0) {
+        if (WSAGetLastError() != WSAENOTSOCK) {
+            errno = WSAGetLastError();
+            return r;
+        }
+    }
+
+    if (r < 0)
+#endif
+        r = write(fd, buf, count);
+
+    return r;
+}
+
 /** Calls read() in a loop. Makes sure that as much as 'size' bytes,
  * unless EOF is reached or an error occured */
 ssize_t pa_loop_read(int fd, void*data, size_t size) {
@@ -207,7 +248,7 @@ ssize_t pa_loop_read(int fd, void*data, size_t size) {
     while (size > 0) {
         ssize_t r;
 
-        if ((r = read(fd, data, size)) < 0)
+        if ((r = pa_read(fd, data, size)) < 0)
             return r;
 
         if (r == 0)
@@ -229,7 +270,7 @@ ssize_t pa_loop_write(int fd, const void*data, size_t size) {
     while (size > 0) {
         ssize_t r;
 
-        if ((r = write(fd, data, size)) < 0)
+        if ((r = pa_write(fd, data, size)) < 0)
             return r;
 
         if (r == 0)
diff --git a/src/polypcore/util.h b/src/polypcore/util.h
index ca81b229..a7c65381 100644
--- a/src/polypcore/util.h
+++ b/src/polypcore/util.h
@@ -37,6 +37,8 @@ void pa_make_nonblock_fd(int fd);
 int pa_make_secure_dir(const char* dir);
 int pa_make_secure_parent_dir(const char *fn);
 
+ssize_t pa_read(int fd, void *buf, size_t count);
+ssize_t pa_write(int fd, void *buf, size_t count);
 ssize_t pa_loop_read(int fd, void*data, size_t size);
 ssize_t pa_loop_write(int fd, const void*data, size_t size);