kenmirrors
/
freeswitch
mirror of https://github.com/signalwire/freeswitch.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
freeswitch/libs/sofia-sip/libsofia-sip-ua/su/su_osx_runloop.c

1788 lines
46 KiB
C
Raw Blame History

/*
* This file is part of the Sofia-SIP package
*
* Copyright (C) 2005 Nokia Corporation.
*
* Contact: Pekka Pessi <pekka.pessi@nokia.com>
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public License
* as published by the Free Software Foundation; either version 2.1 of
* the License, or (at your option) any later version.
*
* This library 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
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public
* License along with this library; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
* 02110-1301 USA
*
*/
/**@ingroup su_wait
* @CFILE su_osx_runloop.c
*
* OS-Independent Socket Syncronization Interface.
*
* This looks like nth reincarnation of "reactor". It implements the
* poll/select/WaitForMultipleObjects and message passing functionality.
*
* @author Pekka Pessi <Pekka.Pessi@nokia.com>
* @author Kai Vehmanen <kai.vehmanen@nokia.com>
*
* @date Created: Tue Sep 14 15:51:04 1999 ppessi
*/
#include "config.h"
/* React to multiple events per one poll() to make sure
* that high-priority events can never completely mask other events.
* Enabled by default on all platforms except WIN32 */
#ifndef WIN32
#define SU_ENABLE_MULTISHOT_POLL 1
#else
#define SU_ENABLE_MULTISHOT_POLL 0
#endif
#include <stdlib.h>
#include <assert.h>
#include <stdarg.h>
#include <stdio.h>
#include <string.h>
#include <limits.h>
#include <errno.h>
#define SU_PORT_IMPLEMENTATION 1
#include "sofia-sip/su.h"
#include "su_port.h"
#include "sofia-sip/su_osx_runloop.h"
#include "sofia-sip/su_alloc.h"
#include "sofia-sip/su_debug.h"
#if HAVE_FUNC
#define enter (void)SU_DEBUG_9(("%s: entering\n", __func__))
#elif HAVE_FUNCTION
#define enter (void)SU_DEBUG_9(("%s: entering\n", __FUNCTION__))
#else
#define enter (void)0
#endif
#if SU_HAVE_PTHREADS
/* Pthread implementation */
#include <pthread.h>
#define SU_HAVE_MBOX 1
#else
#define SU_HAVE_MBOX 0
#endif
#if HAVE_SOCKETPAIR
#define MBOX_SEND 1
#else
#define MBOX_SEND 0
#endif
static su_port_t *su_osx_runloop_create(void) __attribute__((__malloc__));
/* Callback for CFObserver and CFSocket */
static
void cf_observer_cb(CFRunLoopObserverRef observer,
CFRunLoopActivity activity,
void *info);
static void su_port_osx_socket_cb(CFSocketRef s,
CFSocketCallBackType callbackType,
CFDataRef address,
const void *data,
void *info);
static CFSocketCallBackType map_poll_event_to_cf_event(int events);
static void su_port_osx_lock(su_port_t *self, char const *who);
static void su_port_osx_unlock(su_port_t *self, char const *who);
static void su_port_osx_incref(su_port_t *self, char const *who);
static void su_port_osx_decref(su_port_t *self, int blocking, char const *who);
static int su_port_osx_send(su_port_t *self, su_msg_r rmsg);
static int su_port_osx_register(su_port_t *self,
su_root_t *root,
su_wait_t *wait,
su_wakeup_f callback,
su_wakeup_arg_t *arg,
int priority);
static int su_port_osx_unregister(su_port_t *port,
su_root_t *root,
su_wait_t *wait,
su_wakeup_f callback,
su_wakeup_arg_t *arg);
static int su_port_osx_deregister(su_port_t *self, int i);
static int su_port_osx_unregister_all(su_port_t *self,
su_root_t *root);
static int su_port_osx_getmsgs(su_port_t *self);
static
int su_port_osx_eventmask(su_port_t *self, int index, int socket, int events);
static
void su_port_osx_run(su_port_t *self);
static
void su_port_osx_break(su_port_t *self);
static
su_duration_t su_port_osx_step(su_port_t *self, su_duration_t tout);
#if 0
unsigned su_port_query(su_port_t *, su_wait_t *, unsigned n_waits);
void su_port_event(su_port_t *, su_wait_t *waitobj);
#endif
static
int su_port_osx_own_thread(su_port_t const *port);
static
int su_port_osx_add_prepoll(su_port_t *port,
su_root_t *root,
su_prepoll_f *,
su_prepoll_magic_t *);
static
int su_port_osx_remove_prepoll(su_port_t *port,
su_root_t *root);
static
su_timer_t **su_port_osx_timers(su_port_t *port);
static
int su_port_osx_multishot(su_port_t *port, int multishot);
static
int su_port_osx_threadsafe(su_port_t *port);
static
int su_port_osx_yield(su_port_t *port);
su_port_vtable_t const su_port_osx_vtable[1] =
{{
/* su_vtable_size: */ sizeof su_port_osx_vtable,
su_port_osx_lock,
su_port_osx_unlock,
su_port_osx_incref,
su_port_osx_decref,
NULL /* su_port_osx_runloop */, /* XXX - was: gsource, */
su_port_osx_send,
su_port_osx_register,
su_port_osx_unregister,
su_port_osx_deregister,
su_port_osx_unregister_all,
su_port_osx_eventmask,
su_port_osx_run,
su_port_osx_break,
su_port_osx_step,
su_port_osx_own_thread,
su_port_osx_add_prepoll,
su_port_osx_remove_prepoll,
su_port_osx_timers,
su_port_osx_multishot,
su_port_osx_threadsafe,
su_port_osx_yield
}};
static int su_port_osx_wait_events(su_port_t *self, su_duration_t tout);
/* *
* Port is a per-thread reactor.
*
* Multiple root objects executed by single thread share a su_port_t object.
*/
struct su_port_s {
su_home_t sup_home[1];
su_port_vtable_t const *sup_vtable;
unsigned sup_running;
unsigned sup_source_fired;
#if SU_HAVE_PTHREADS
pthread_t sup_tid;
pthread_mutex_t sup_mutex[1];
#if __CYGWIN__
pthread_mutex_t sup_reflock[1];
int sup_ref;
#else
pthread_rwlock_t sup_ref[1];
#endif
#else
int sup_ref;
#endif
#if SU_HAVE_MBOX
su_socket_t sup_mbox[MBOX_SEND + 1];
su_wait_t sup_mbox_wait;
#endif
CFRunLoopRef sup_main_loop;
CFRunLoopSourceRef *sup_sources;
CFSocketRef *sup_sockets;
unsigned sup_multishot; /**< Multishot operation? */
unsigned sup_registers; /** Counter incremented by
su_port_register() or
su_port_unregister()
*/
int sup_n_waits; /**< Active su_wait_t in su_waits */
int sup_size_waits; /**< Size of allocate su_waits */
int sup_pri_offset; /**< Offset to prioritized waits */
#define INDEX_MAX (0x7fffffff)
/** Indices from index returned by su_root_register() to tables below.
*
* Free elements are negative. Free elements form a list, value of free
* element is (0 - index of next free element).
*
* First element sup_indices[0] points to first free element.
*/
int *sup_indices;
int *sup_reverses; /** Reverse index */
su_wakeup_f *sup_wait_cbs;
su_wakeup_arg_t**sup_wait_args;
su_root_t **sup_wait_roots;
su_wait_t *sup_waits;
/* Pre-poll callback */
su_prepoll_f *sup_prepoll;
su_prepoll_magic_t *sup_pp_magic;
su_root_t *sup_pp_root;
/* Timer list */
su_timer_t *sup_timers;
/* Message list - this is protected by lock */
su_msg_t *sup_head;
su_msg_t **sup_tail;
#if 0
int sup_free_index; /**< Number of first free index */
int *sup_indices; /** Indices to registrations */
int *sup_reverses; /** Reverse index */
su_wakeup_f *sup_wait_cbs;
su_wakeup_arg_t**sup_wait_args;
su_root_t **sup_wait_roots;
su_wait_t *sup_waits;
/* Pre-poll callback */
su_prepoll_f *sup_prepoll;
su_prepoll_magic_t *sup_pp_magic;
su_root_t *sup_pp_root;
/* Timer list */
su_timer_t *sup_timers;
/* Message list - this is protected by lock */
su_msg_t *sup_head;
su_msg_t **sup_tail;
#endif
};
/* Struct for CFSocket callbacks; contains current CFSource index */
typedef struct {
su_port_t *o_port;
int o_current;
int o_count;
} osx_magic_t;
#if SU_HAVE_PTHREADS
#define SU_PORT_OSX_OWN_THREAD(p) (pthread_equal((p)->sup_tid, pthread_self()))
#if __CYGWIN__
/* Debugging versions */
#define SU_PORT_OSX_INITREF(p) (pthread_mutex_init((p)->sup_reflock, NULL), printf("initref(%p)\n", (p)))
#define SU_PORT_OSX_INCREF(p, f) (pthread_mutex_lock(p->sup_reflock), p->sup_ref++, pthread_mutex_unlock(p->sup_reflock), printf("incref(%p) by %s\n", (p), f))
#define SU_PORT_OSX_DECREF(p, f) do { \
pthread_mutex_lock(p->sup_reflock); p->sup_ref--; pthread_mutex_unlock(p->sup_reflock); \
if ((p->sup_ref) == 0) { \
printf("decref(%p) to 0 by %s\n", (p), f); su_port_osx_destroy(p); } \
else { printf("decref(%p) to %u by %s\n", (p), p->sup_ref, f); } } while(0)
#define SU_PORT_OSX_ZAPREF(p, f) do { printf("zapref(%p) by %s\n", (p), f), \
pthread_mutex_lock(p->sup_reflock); p->sup_ref--; pthread_mutex_unlock(p->sup_reflock); \
if ((p->sup_ref) != 0) { \
assert(!"SU_PORT_OSX_ZAPREF"); } \
su_port_osx_destroy(p); } while(0)
#define SU_PORT_OSX_INITLOCK(p) \
(pthread_mutex_init((p)->sup_mutex, NULL), printf("init_lock(%p)\n", p))
#define SU_PORT_OSX_LOCK(p, f) \
(printf("%ld at %s locking(%p)...", pthread_self(), f, p), pthread_mutex_lock((p)->sup_mutex), printf(" ...%ld at %s locked(%p)...", pthread_self(), f, p))
#define SU_PORT_OSX_UNLOCK(p, f) \
(pthread_mutex_unlock((p)->sup_mutex), printf(" ...%ld at %s unlocked(%p)\n", pthread_self(), f, p))
#elif 1
#define SU_PORT_OSX_INITREF(p) (pthread_rwlock_init(p->sup_ref, NULL))
#define SU_PORT_OSX_INCREF(p, f) (pthread_rwlock_rdlock(p->sup_ref))
#define SU_PORT_OSX_DECREF(p, f) do { pthread_rwlock_unlock(p->sup_ref); \
if (pthread_rwlock_trywrlock(p->sup_ref) == 0) su_port_osx_destroy(p); } while(0)
#define SU_PORT_OSX_ZAPREF(p, f) do { pthread_rwlock_unlock(p->sup_ref); \
if (pthread_rwlock_trywrlock(p->sup_ref) != 0) { \
assert(!"SU_PORT_OSX_ZAPREF"); pthread_rwlock_wrlock(p->sup_ref); } \
su_port_osx_destroy(p); } while(0)
#define SU_PORT_OSX_INITLOCK(p) (pthread_mutex_init((p)->sup_mutex, NULL))
#define SU_PORT_OSX_LOCK(p, f) (pthread_mutex_lock((p)->sup_mutex))
#define SU_PORT_OSX_UNLOCK(p, f) (pthread_mutex_unlock((p)->sup_mutex))
#else
/* Debugging versions */
#define SU_PORT_OSX_INITREF(p) (pthread_rwlock_init((p)->sup_ref, NULL), printf("initref(%p)\n", (p)))
#define SU_PORT_OSX_INCREF(p, f) (pthread_rwlock_rdlock(p->sup_ref), printf("incref(%p) by %s\n", (p), f))
#define SU_PORT_OSX_DECREF(p, f) do { \
pthread_rwlock_unlock(p->sup_ref); \
if (pthread_rwlock_trywrlock(p->sup_ref) == 0) { \
printf("decref(%p) to 0 by %s\n", (p), f); su_port_osx_destroy(p); } \
else { printf("decref(%p) by %s\n", (p), f); } } while(0)
#define SU_PORT_OSX_ZAPREF(p, f) do { printf("zapref(%p) by %s\n", (p), f), \
pthread_rwlock_unlock(p->sup_ref); \
if (pthread_rwlock_trywrlock(p->sup_ref) != 0) { \
assert(!"SU_PORT_OSX_ZAPREF"); pthread_rwlock_wrlock(p->sup_ref); } \
su_port_osx_destroy(p); } while(0)
#define SU_PORT_OSX_INITLOCK(p) \
(pthread_mutex_init((p)->sup_mutex, NULL), printf("init_lock(%p)\n", p))
#define SU_PORT_OSX_LOCK(p, f) \
(printf("%ld at %s locking(%p)...", pthread_self(), f, p), pthread_mutex_lock((p)->sup_mutex), printf(" ...%ld at %s locked(%p)...", pthread_self(), f, p))
#define SU_PORT_OSX_UNLOCK(p, f) \
(pthread_mutex_unlock((p)->sup_mutex), printf(" ...%ld at %s unlocked(%p)\n", pthread_self(), f, p))
#endif
#else /* !SU_HAVE_PTHREADS */
#define SU_PORT_OSX_OWN_THREAD(p) 1
#define SU_PORT_OSX_INITLOCK(p) (void)(p)
#define SU_PORT_OSX_LOCK(p, f) (void)(p)
#define SU_PORT_OSX_UNLOCK(p, f) (void)(p)
#define SU_PORT_OSX_ZAPREF(p, f) ((p)->sup_ref--)
#define SU_PORT_OSX_INITREF(p) ((p)->sup_ref = 1)
#define SU_PORT_OSX_INCREF(p, f) ((p)->sup_ref++)
#define SU_PORT_OSX_DECREF(p, f) \
do { if (--((p)->sup_ref) == 0) su_port_osx_destroy(p); } while (0);
#endif
#if SU_HAVE_MBOX
static int su_port_osx_wakeup(su_root_magic_t *magic,
su_wait_t *w,
su_wakeup_arg_t *arg);
#endif
static void su_port_osx_destroy(su_port_t *self);
/** Create a reactor object.
*
* Allocate and initialize the instance of su_root_t.
*
* @param magic pointer to user data
*
* @return A pointer to allocated su_root_t instance, NULL on error.
*
* @NEW_1_12_4.
*/
su_root_t *su_root_osx_runloop_create(su_root_magic_t *magic)
{
return su_root_create_with_port(magic, su_osx_runloop_create());
}
void osx_enabler_cb(CFSocketRef s,
CFSocketCallBackType type,
CFDataRef address,
const void *data,
void *info)
{
CFRunLoopRef rl;
osx_magic_t *magic = (osx_magic_t *) info;
su_port_t *self = magic->o_port;
su_duration_t tout = 0;
su_time_t now = su_now();
rl = CFRunLoopGetCurrent();
if (self->sup_running) {
if (self->sup_prepoll)
self->sup_prepoll(self->sup_pp_magic, self->sup_pp_root);
if (self->sup_head)
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, now);
}
CFRunLoopWakeUp(rl);
}
/**@internal
*
* Allocates and initializes a message port. It creates a mailbox used to.
* wake up the tasks waiting on the port if needed. Currently, the
* mailbox is simply an UDP socket connected to itself.
*
* @return
* If successful a pointer to the new message port is returned, otherwise
* NULL is returned.
*/
su_port_t *su_osx_runloop_create(void)
{
su_port_t *self;
SU_DEBUG_9(("su_osx_runloop_create() called\n"));
self = su_home_clone(NULL, sizeof(*self));
if (self) {
#if SU_HAVE_MBOX
int af;
su_socket_t mb = INVALID_SOCKET;
char const *why;
#endif
CFRunLoopObserverRef cf_observer;
osx_magic_t *osx_magic = NULL;
CFRunLoopObserverContext cf_observer_cntx[1] = {{0, NULL, NULL,
NULL, NULL}};
self->sup_vtable = su_port_osx_vtable;
SU_PORT_OSX_INITREF(self);
SU_PORT_OSX_INITLOCK(self);
self->sup_tail = &self->sup_head;
self->sup_multishot = 0; /* XXX (SU_ENABLE_MULTISHOT_POLL) != 0; */
#if SU_HAVE_PTHREADS
self->sup_tid = pthread_self();
#endif
#if SU_HAVE_MBOX
#if HAVE_SOCKETPAIR
#if defined(AF_LOCAL)
af = AF_LOCAL;
#else
af = AF_UNIX;
#endif
if (socketpair(af, SOCK_STREAM, 0, self->sup_mbox) == -1) {
why = "su_port_osx_init: socketpair"; goto error;
}
mb = self->sup_mbox[0];
su_setblocking(self->sup_mbox[0], 0);
su_setblocking(self->sup_mbox[1], 0);
#else
{
struct sockaddr_in sin = { sizeof(struct sockaddr_in), 0 };
socklen_t sinsize = sizeof sin;
struct sockaddr *sa = (struct sockaddr *)&sin;
af = PF_INET;
self->sup_mbox[0] = mb = su_socket(af, SOCK_DGRAM, IPPROTO_UDP);
if (mb == INVALID_SOCKET) {
why = "su_port_osx_init: socket"; goto error;
}
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = htonl(INADDR_LOOPBACK); /* 127.1 */
/* Get a port for us */
if (bind(mb, sa, sizeof sin) == -1) {
why = "su_port_osx_init: bind"; goto error;
}
if (getsockname(mb, sa, &sinsize) == -1) {
why = "su_port_osx_init: getsockname"; goto error;
}
if (connect(mb, sa, sinsize) == -1) {
why = "su_port_osx_init: connect"; goto error;
}
}
#endif
if (su_wait_create(&self->sup_mbox_wait, mb, SU_WAIT_IN) == -1) {
why = "su_port_osx_init: su_wait_create"; goto error;
}
if (su_port_osx_register(self, NULL, &self->sup_mbox_wait,
su_port_osx_wakeup,
(su_wakeup_arg_t *)self->sup_mbox, 0)
== -1) {
why = "su_port_osx_create: su_port_osx_register"; goto error;
}
osx_magic = calloc(1, sizeof(*osx_magic));
osx_magic->o_port = self;
cf_observer_cntx->info = osx_magic;
cf_observer =
CFRunLoopObserverCreate(NULL, kCFRunLoopAfterWaiting | kCFRunLoopBeforeWaiting,
TRUE, 0, cf_observer_cb, cf_observer_cntx);
CFRunLoopAddObserver(CFRunLoopGetCurrent(),
cf_observer,
kCFRunLoopDefaultMode);
SU_DEBUG_9(("su_port_osx_create() returns %p\n", self));
return self;
error:
su_perror(why);
su_port_osx_destroy(self), self = NULL;
#endif
}
SU_DEBUG_9(("su_port_osx_create() returns %p\n", self));
return self;
}
static
void cf_observer_cb(CFRunLoopObserverRef observer,
CFRunLoopActivity activity,
void *info)
{
CFRunLoopRef rl;
osx_magic_t *magic = (osx_magic_t *) info;
su_port_t *self = magic->o_port;
su_duration_t tout = 0;
su_time_t now = su_now();
rl = CFRunLoopGetCurrent();
if (self->sup_running) {
if (self->sup_prepoll)
self->sup_prepoll(self->sup_pp_magic, self->sup_pp_root);
if (self->sup_head)
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, now);
} else
SU_DEBUG_9(("cf_observer_cb(): PORT IS NOT RUNNING!\n"));
CFRunLoopWakeUp(rl);
return;
}
/** @internal Destroy a port. */
void su_port_osx_destroy(su_port_t *self)
{
assert(self);
SU_DEBUG_9(("su_port_osx_destroy() called\n"));
#if SU_HAVE_MBOX
if (self->sup_mbox[0] != INVALID_SOCKET) {
su_port_osx_unregister(self, NULL, &self->sup_mbox_wait, NULL,
(su_wakeup_arg_t *)self->sup_mbox);
su_wait_destroy(&self->sup_mbox_wait);
su_close(self->sup_mbox[0]); self->sup_mbox[0] = INVALID_SOCKET;
#if HAVE_SOCKETPAIR
su_close(self->sup_mbox[1]); self->sup_mbox[1] = INVALID_SOCKET;
#endif
SU_DEBUG_9(("su_port_osx_destroy() close mailbox\n"));
}
#endif
if (self->sup_sources)
free(self->sup_sources), self->sup_sources = NULL;
if (self->sup_sockets)
free(self->sup_sockets), self->sup_sockets = NULL;
if (self->sup_waits)
free(self->sup_waits), self->sup_waits = NULL;
if (self->sup_wait_cbs)
free(self->sup_wait_cbs), self->sup_wait_cbs = NULL;
if (self->sup_wait_args)
free(self->sup_wait_args), self->sup_wait_args = NULL;
if (self->sup_wait_roots)
free(self->sup_wait_roots), self->sup_wait_roots = NULL;
if (self->sup_reverses)
free(self->sup_reverses), self->sup_reverses = NULL;
if (self->sup_indices)
free(self->sup_indices), self->sup_indices = NULL;
SU_DEBUG_9(("su_port_osx_destroy() freed registrations\n"));
su_home_zap(self->sup_home);
SU_DEBUG_9(("su_port_osx_destroy() returns\n"));
}
static void su_port_osx_lock(su_port_t *self, char const *who)
{
SU_PORT_OSX_LOCK(self, who);
}
static void su_port_osx_unlock(su_port_t *self, char const *who)
{
SU_PORT_OSX_UNLOCK(self, who);
}
static void su_port_osx_incref(su_port_t *self, char const *who)
{
SU_PORT_OSX_INCREF(self, who);
}
static void su_port_osx_decref(su_port_t *self, int blocking, char const *who)
{
if (blocking)
SU_PORT_OSX_ZAPREF(self, who);
else
SU_PORT_OSX_DECREF(self, who);
}
static
CFSocketCallBackType map_poll_event_to_cf_event(int events)
{
CFSocketCallBackType type = 0;
if (events & SU_WAIT_IN)
type |= kCFSocketReadCallBack;
if (events & SU_WAIT_OUT)
type |= kCFSocketWriteCallBack;
#if 0
if (events & SU_WAIT_CONNECT)
type |= kCFSocketConnectCallBack;
if (events & SU_WAIT_ACCEPT)
type |= kCFSocketAcceptCallBack;
#endif
return type;
}
static
int map_cf_event_to_poll_event(CFSocketCallBackType type)
{
int event = 0;
if (type & kCFSocketReadCallBack)
event |= SU_WAIT_IN;
if (type & kCFSocketWriteCallBack)
event |= SU_WAIT_OUT;
if (type & kCFSocketConnectCallBack)
event |= SU_WAIT_CONNECT;
if (type & kCFSocketAcceptCallBack)
event |= SU_WAIT_ACCEPT;
return event;
}
static
void su_port_osx_socket_cb(CFSocketRef s,
CFSocketCallBackType type,
CFDataRef address,
const void *data,
void *info)
{
osx_magic_t *magic = (osx_magic_t *) info;
su_port_t *self = magic->o_port;
int curr = magic->o_current;
su_duration_t tout = 0;
#if SU_HAVE_POLL
{
su_root_t *root;
su_wait_t *waits = self->sup_waits;
int n = self->sup_indices[curr];
assert(self->sup_reverses[n] == curr);
SU_DEBUG_9(("socket_cb(%p): count %u index %d\n", self->sup_sources[n], magic->o_count, curr));
waits[n].revents = map_poll_event_to_cf_event(type);
root = self->sup_wait_roots[n];
self->sup_wait_cbs[n](root ? su_root_magic(root) : NULL,
&waits[n],
self->sup_wait_args[n]);
if (self->sup_running) {
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, su_now());
if (self->sup_head)
tout = 0;
/* CFRunLoopWakeUp(CFRunLoopGetCurrent()); */
}
/* Tell to run loop an su socket fired */
self->sup_source_fired = 1;
}
#endif
}
#if SU_HAVE_MBOX
/** @internal Message box wakeup function. */
static int su_port_osx_wakeup(su_root_magic_t *magic, /* NULL */
su_wait_t *w,
su_wakeup_arg_t *arg)
{
char buf[32];
su_socket_t s = *(su_socket_t *)arg;
su_wait_events(w, s);
recv(s, buf, sizeof(buf), 0);
/* CFRunLoopWakeUp(CFRunLoopGetCurrent()); */
return 0;
}
#endif
/** @internal Send a message to the port. */
int su_port_osx_send(su_port_t *self, su_msg_r rmsg)
{
CFRunLoopRef rl;
if (self) {
int wakeup;
SU_PORT_OSX_LOCK(self, "su_port_osx_send");
wakeup = self->sup_head == NULL;
*self->sup_tail = rmsg[0]; rmsg[0] = NULL;
self->sup_tail = &(*self->sup_tail)->sum_next;
#if SU_HAVE_MBOX
/* if (!pthread_equal(pthread_self(), self->sup_tid)) */
if (wakeup)
{
assert(self->sup_mbox[MBOX_SEND] != INVALID_SOCKET);
if (send(self->sup_mbox[MBOX_SEND], "X", 1, 0) == -1) {
#if HAVE_SOCKETPAIR
if (su_errno() != EWOULDBLOCK)
#endif
su_perror("su_msg_send: send()");
}
}
#endif
SU_PORT_OSX_UNLOCK(self, "su_port_osx_send");
rl = CFRunLoopGetCurrent();
CFRunLoopWakeUp(rl);
return 0;
}
else {
su_msg_destroy(rmsg);
return -1;
}
}
/** @internal
* Execute the messages in the incoming queue until the queue is empty..
*
* @param self - pointer to a port object
*
* @retval Number of messages sent
*/
int su_port_osx_getmsgs(su_port_t *self)
{
int n = 0;
if (self->sup_head) {
su_msg_f f;
su_msg_t *msg, *queue;
SU_PORT_OSX_LOCK(self, "su_port_getmsgs");
queue = self->sup_head;
self->sup_tail = &self->sup_head;
self->sup_head = NULL;
SU_PORT_OSX_UNLOCK(self, "su_port_osx_getmsgs");
for (msg = queue; msg; msg = queue) {
queue = msg->sum_next;
msg->sum_next = NULL;
f = msg->sum_func;
if (f)
f(SU_ROOT_MAGIC(msg->sum_to->sut_root), &msg, msg->sum_data);
su_msg_delivery_report(&msg);
n++;
}
/* Check for wait events that may have been generated by messages */
su_port_osx_wait_events(self, 0);
}
return n;
}
static int o_count;
/** @internal
*
* Register a @c su_wait_t object. The wait object, a callback function and
* a argument pointer is stored in the port object. The callback function
* will be called when the wait object is signaled.
*
* Please note if identical wait objects are inserted, only first one is
* ever signalled.
*
* @param self pointer to port
* @param root pointer to root object
* @param waits pointer to wait object
* @param callback callback function pointer
* @param arg argument given to callback function when it is invoked
* @param priority relative priority of the wait object
* (0 is normal, 1 important, 2 realtime)
*
* @return
* The function @su_port_osx_register returns nonzero index of the wait object,
* or -1 upon an error. */
int su_port_osx_register(su_port_t *self,
su_root_t *root,
su_wait_t *wait,
su_wakeup_f callback,
su_wakeup_arg_t *arg,
int priority)
{
int i, j, n;
CFRunLoopRef rl;
CFRunLoopSourceRef *sources, source;
CFSocketRef cf_socket, *sockets;
int events = 0;
osx_magic_t *osx_magic = NULL;
CFSocketContext cf_socket_cntx[1] = {{0, NULL, NULL, NULL, NULL}};
CFOptionFlags flags = 0;
assert(SU_PORT_OSX_OWN_THREAD(self));
n = self->sup_n_waits;
if (n >= SU_WAIT_MAX)
return su_seterrno(ENOMEM);
if (n >= self->sup_size_waits) {
/* Reallocate size arrays */
int size;
int *indices;
int *reverses;
su_wait_t *waits;
su_wakeup_f *wait_cbs;
su_wakeup_arg_t **wait_args;
su_root_t **wait_tasks;
if (self->sup_size_waits == 0)
size = su_root_size_hint;
else
size = 2 * self->sup_size_waits;
if (size < SU_WAIT_MIN)
size = SU_WAIT_MIN;
/* Too large */
if (-3 - size > 0)
return (errno = ENOMEM), -1;
indices = realloc(self->sup_indices, (size + 1) * sizeof(*indices));
if (indices) {
self->sup_indices = indices;
for (i = self->sup_size_waits; i <= size; i++)
indices[i] = -1 - i;
}
reverses = realloc(self->sup_reverses, size * sizeof(*waits));
if (reverses) {
for (i = self->sup_size_waits; i < size; i++)
reverses[i] = -1;
self->sup_reverses = reverses;
}
sources = realloc(self->sup_sources, size * sizeof(*sources));
if (sources)
self->sup_sources = sources;
sockets = realloc(self->sup_sockets, size * sizeof(*sockets));
if (sockets)
self->sup_sockets = sockets;
waits = realloc(self->sup_waits, size * sizeof(*waits));
if (waits)
self->sup_waits = waits;
wait_cbs = realloc(self->sup_wait_cbs, size * sizeof(*wait_cbs));
if (wait_cbs)
self->sup_wait_cbs = wait_cbs;
wait_args = realloc(self->sup_wait_args, size * sizeof(*wait_args));
if (wait_args)
self->sup_wait_args = wait_args;
/* Add sup_wait_roots array, if needed */
wait_tasks = realloc(self->sup_wait_roots, size * sizeof(*wait_tasks));
if (wait_tasks)
self->sup_wait_roots = wait_tasks;
if (!(indices &&
reverses && sources && sockets && waits && wait_cbs && wait_args && wait_tasks)) {
return -1;
}
self->sup_size_waits = size;
}
i = -self->sup_indices[0]; assert(i <= self->sup_size_waits);
if (priority > 0) {
/* Insert */
for (n = self->sup_n_waits; n > 0; n--) {
j = self->sup_reverses[n-1]; assert(self->sup_indices[j] == n - 1);
self->sup_indices[j] = n;
self->sup_reverses[n] = self->sup_reverses[n-1];
self->sup_sources[n] = self->sup_sources[n-1];
self->sup_sockets[n] = self->sup_sockets[n-1];
self->sup_waits[n] = self->sup_waits[n-1];
self->sup_wait_cbs[n] = self->sup_wait_cbs[n-1];
self->sup_wait_args[n] = self->sup_wait_args[n-1];
self->sup_wait_roots[n] = self->sup_wait_roots[n-1];
}
self->sup_pri_offset++;
}
else {
/* Append - no need to move anything */
n = self->sup_n_waits;
}
self->sup_n_waits++;
self->sup_indices[0] = self->sup_indices[i]; /* Free index */
self->sup_indices[i] = n;
self->sup_reverses[n] = i;
self->sup_waits[n] = *wait;
self->sup_wait_cbs[n] = callback;
self->sup_wait_args[n] = arg;
self->sup_wait_roots[n] = root;
self->sup_registers++;
/* XXX -- mela: leak, leak -- free() somewheeeere */
osx_magic = calloc(1, sizeof(*osx_magic));
osx_magic->o_port = self;
osx_magic->o_current = i;
osx_magic->o_count = ++o_count;
cf_socket_cntx->info = osx_magic;
events = map_poll_event_to_cf_event(wait->events);
cf_socket = CFSocketCreateWithNative(NULL,
(CFSocketNativeHandle) su_wait_socket(wait),
events, su_port_osx_socket_cb, cf_socket_cntx);
flags = CFSocketGetSocketFlags(cf_socket);
flags &= ~kCFSocketCloseOnInvalidate;
CFSocketSetSocketFlags(cf_socket, flags);
CFRetain(cf_socket);
source = CFSocketCreateRunLoopSource(NULL, cf_socket, 0);
SU_DEBUG_9(("source(%p): count %u index %d\n", source, o_count, i));
rl = CFRunLoopGetCurrent();
CFRunLoopAddSource(rl, source, kCFRunLoopDefaultMode);
CFRetain(source);
self->sup_sources[n] = source;
self->sup_sockets[n] = cf_socket;
CFRunLoopWakeUp(rl);
/* Just like epoll, we return -1 or positive integer */
return i;
}
/** Deregister a su_wait_t object. */
static
int su_port_osx_deregister0(su_port_t *self, int i)
{
CFRunLoopRef rl;
int n, N, *indices, *reverses;
indices = self->sup_indices;
reverses = self->sup_reverses;
n = indices[i]; assert(n >= 0); assert(i == reverses[n]);
N = --self->sup_n_waits;
rl = CFRunLoopGetCurrent();
CFSocketInvalidate(self->sup_sockets[n]);
CFRelease(self->sup_sockets[n]);
CFRunLoopRemoveSource(rl, self->sup_sources[n], kCFRunLoopDefaultMode);
CFRelease(self->sup_sources[n]);
CFRunLoopWakeUp(rl);
if (n < self->sup_pri_offset) {
int j = --self->sup_pri_offset;
if (n != j) {
assert(reverses[j] > 0);
assert(indices[reverses[j]] == j);
indices[reverses[j]] = n;
reverses[n] = reverses[j];
self->sup_sources[n] = self->sup_sources[j];
self->sup_sockets[n] = self->sup_sockets[j];
self->sup_waits[n] = self->sup_waits[j];
self->sup_wait_cbs[n] = self->sup_wait_cbs[j];
self->sup_wait_args[n] = self->sup_wait_args[j];
self->sup_wait_roots[n] = self->sup_wait_roots[j];
n = j;
}
}
if (n < N) {
assert(reverses[N] > 0);
assert(indices[reverses[N]] == N);
indices[reverses[N]] = n;
reverses[n] = reverses[N];
self->sup_sources[n] = self->sup_sources[N];
self->sup_sockets[n] = self->sup_sockets[N];
self->sup_waits[n] = self->sup_waits[N];
self->sup_wait_cbs[n] = self->sup_wait_cbs[N];
self->sup_wait_args[n] = self->sup_wait_args[N];
self->sup_wait_roots[n] = self->sup_wait_roots[N];
n = N;
}
reverses[n] = -1;
memset(&self->sup_waits[n], 0, sizeof self->sup_waits[n]);
self->sup_sources[n] = NULL;
self->sup_sockets[n] = NULL;
self->sup_wait_cbs[n] = NULL;
self->sup_wait_args[n] = NULL;
self->sup_wait_roots[n] = NULL;
indices[i] = indices[0];
indices[0] = -i;
self->sup_registers++;
return i;
}
/** Unregister a su_wait_t object.
*
* The function su_port_osx_unregister() unregisters a su_wait_t object. The
* wait object, a callback function and a argument are removed from the
* port object.
*
* @param self - pointer to port object
* @param root - pointer to root object
* @param wait - pointer to wait object
* @param callback - callback function pointer (may be NULL)
* @param arg - argument given to callback function when it is invoked
* (may be NULL)
*
* @return Nonzero index of the wait object, or -1 upon an error.
*/
int su_port_osx_unregister(su_port_t *self,
su_root_t *root,
su_wait_t *wait,
su_wakeup_f callback, /* XXX - ignored */
su_wakeup_arg_t *arg)
{
int n, N;
assert(self);
assert(SU_PORT_OSX_OWN_THREAD(self));
N = self->sup_n_waits;
for (n = 0; n < N; n++) {
if (SU_WAIT_CMP(wait[0], self->sup_waits[n]) == 0) {
return su_port_osx_deregister0(self, self->sup_reverses[n]);
}
}
su_seterrno(ENOENT);
return -1;
}
/** Deregister a su_wait_t object.
*
* The function su_port_osx_deregister() deregisters a su_wait_t registrattion.
* The wait object, a callback function and a argument are removed from the
* port object.
*
* @param self - pointer to port object
* @param i - registration index
*
* @return Index of the wait object, or -1 upon an error.
*/
int su_port_osx_deregister(su_port_t *self, int i)
{
su_wait_t wait[1] = { SU_WAIT_INIT };
int retval;
assert(self);
assert(SU_PORT_OSX_OWN_THREAD(self));
if (i <= 0 || i > self->sup_size_waits)
return su_seterrno(EBADF);
if (self->sup_indices[i] < 0)
return su_seterrno(EBADF);
retval = su_port_osx_deregister0(self, i);
su_wait_destroy(wait);
return retval;
}
/** @internal
* Unregister all su_wait_t objects.
*
* The function su_port_osx_unregister_all() unregisters all su_wait_t objects
* and destroys all queued timers associated with given root object.
*
* @param self - pointer to port object
* @param root - pointer to root object
*
* @return Number of wait objects removed.
*/
int su_port_osx_unregister_all(su_port_t *self,
su_root_t *root)
{
int i, j, index, N;
int *indices, *reverses;
su_wait_t *waits;
su_wakeup_f *wait_cbs;
su_wakeup_arg_t **wait_args;
su_root_t **wait_roots;
CFRunLoopRef rl;
CFRunLoopSourceRef *sources;
CFSocketRef *sockets;
assert(SU_PORT_OSX_OWN_THREAD(self));
N = self->sup_n_waits;
indices = self->sup_indices;
reverses = self->sup_reverses;
sources = self->sup_sources;
sockets = self->sup_sockets;
waits = self->sup_waits;
wait_cbs = self->sup_wait_cbs;
wait_args = self->sup_wait_args;
wait_roots = self->sup_wait_roots;
rl = CFRunLoopGetCurrent();
for (i = j = 0; i < N; i++) {
index = reverses[i]; assert(index > 0 && indices[index] == i);
if (wait_roots[i] == root) {
if (i < self->sup_pri_offset)
self->sup_pri_offset--;
indices[index] = indices[0];
indices[0] = -index;
continue;
}
if (i != j) {
indices[index] = j;
CFSocketInvalidate(self->sup_sockets[j]);
CFRelease(self->sup_sockets[j]);
CFRunLoopRemoveSource(rl, sources[j], kCFRunLoopDefaultMode);
CFRelease(sources[j]);
reverses[j] = reverses[i];
sources[j] = sources[i];
sockets[j] = sockets[i];
waits[j] = waits[i];
wait_cbs[j] = wait_cbs[i];
wait_args[j] = wait_args[i];
wait_roots[j] = wait_roots[i];
}
j++;
}
/* Prepare for removing CFSources */
for (i = j; i < N; i++) {
reverses[i] = -1;
CFSocketInvalidate(self->sup_sockets[i]);
CFRelease(self->sup_sockets[i]);
CFRunLoopRemoveSource(rl, sources[i], kCFRunLoopDefaultMode);
CFRunLoopSourceInvalidate(sources[i]);
sources[i] = NULL;
sockets[i] = NULL;
wait_cbs[i] = NULL;
wait_args[i] = NULL;
wait_roots[i] = NULL;
}
memset(&waits[j], 0, (char *)&waits[N] - (char *)&waits[j]);
/* Tell run loop things have changed */
CFRunLoopWakeUp(rl);
self->sup_n_waits = j;
self->sup_registers++;
return N - j;
}
/**Set mask for a registered event. @internal
*
* The function su_port_osx_eventmask() sets the mask describing events that can
* signal the registered callback.
*
* @param port pointer to port object
* @param index registration index
* @param socket socket
* @param events new event mask
*
* @retval 0 when successful,
* @retval -1 upon an error.
*/
int su_port_osx_eventmask(su_port_t *self, int index, int socket, int events)
{
int n, ret;
assert(self);
assert(SU_PORT_OSX_OWN_THREAD(self));
if (index <= 0 || index > self->sup_size_waits)
return su_seterrno(EBADF);
n = self->sup_indices[index];
if (n < 0)
return su_seterrno(EBADF);
ret = su_wait_mask(&self->sup_waits[n], socket, events);
CFSocketSetSocketFlags(self->sup_sockets[n],
map_poll_event_to_cf_event(events));
return ret;
}
/** @internal
*
* Copies the su_wait_t objects from the port. The number of wait objects
* can be found out by calling su_port_osx_query() with @a n_waits as zero.
*
* @note This function is called only by friends.
*
* @param self - pointer to port object
* @param waits - pointer to array to which wait objects are copied
* @param n_waits - number of wait objects fitting in array waits
*
* @return Number of wait objects, or 0 upon an error.
*/
unsigned su_port_osx_query(su_port_t *self, su_wait_t *waits, unsigned n_waits)
{
unsigned n;
assert(SU_PORT_OSX_OWN_THREAD(self));
n = self->sup_n_waits;
if (n_waits != 0) {
if (waits && n_waits >= n)
memcpy(waits, self->sup_waits, n * sizeof(*waits));
else
n = 0;
}
return n;
}
/** @internal Enable multishot mode.
*
* The function su_port_osx_multishot() enables, disables or queries the
* multishot mode for the port. The multishot mode determines how the events
* are scheduled by port. If multishot mode is enabled, port serves all the
* sockets that have received network events. If it is disables, only first
* socket event is served.
*
* @param self pointer to port object
* @param multishot multishot mode (0 => disables, 1 => enables, -1 => query)
*
* @retval 0 multishot mode is disabled
* @retval 1 multishot mode is enabled
* @retval -1 an error occurred
*/
int su_port_osx_multishot(su_port_t *self, int multishot)
{
if (multishot < 0)
return self->sup_multishot;
else if (multishot == 0 || multishot == 1)
return self->sup_multishot = multishot;
else
return (errno = EINVAL), -1;
}
/** @internal Enable threadsafe operation. */
static
int su_port_osx_threadsafe(su_port_t *port)
{
return su_home_threadsafe(port->sup_home);
}
/** Prepare root to be run on OSX Run Loop.
*
* Sets #su_root_t object to be callable by the application's run loop. This
* function is to be used instead of su_root_run() for OSX applications
* using Core Foundation's Run Loop.
*
* The function su_root_osx_prepare_run() returns immmediately.
*
* @param root pointer to root object
*
* @NEW_1_12_4.
*/
void su_root_osx_prepare_run(su_root_t *root)
{
su_port_t *self = root->sur_task->sut_port;
CFRunLoopRef rl;
su_duration_t tout = 0;
assert(SU_PORT_OSX_OWN_THREAD(self));
enter;
self->sup_running = 1;
rl = CFRunLoopGetCurrent();
if (self->sup_prepoll)
self->sup_prepoll(self->sup_pp_magic, self->sup_pp_root);
if (self->sup_head)
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, su_now());
if (!self->sup_running)
return;
CFRetain(rl);
self->sup_main_loop = rl;
return;
}
/** @internal Main loop.
*
* The function @c su_port_osx_run() waits for wait objects and the timers
* associated with the port object. When any wait object is signaled or
* timer is expired, it invokes the callbacks, and returns waiting.
*
* The function @c su_port_osx_run() runs until @c su_port_osx_break() is called
* from a callback.
*
* @param self pointer to port object
*
*/
void su_port_osx_run(su_port_t *self)
{
CFRunLoopRef rl;
su_duration_t tout = 0;
assert(SU_PORT_OSX_OWN_THREAD(self));
enter;
self->sup_running = 1;
rl = CFRunLoopGetCurrent();
if (self->sup_prepoll)
self->sup_prepoll(self->sup_pp_magic, self->sup_pp_root);
if (self->sup_head)
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, su_now());
if (!self->sup_running)
return;
CFRetain(rl);
self->sup_main_loop = rl;
/* if there are messages do a quick wait */
if (self->sup_head)
tout = 0;
CFRunLoopRun();
self->sup_main_loop = NULL;
}
#if tuning
/* This version can help tuning... */
void su_port_osx_run_tune(su_port_t *self)
{
int i;
int timers = 0, messages = 0, events = 0;
su_duration_t tout = 0, tout0;
su_time_t started = su_now(), woken = started, bedtime = woken;
assert(SU_PORT_OSX_OWN_THREAD(self));
for (self->sup_running = 1; self->sup_running;) {
tout0 = tout, tout = 2000;
timers = 0, messages = 0;
if (self->sup_prepoll)
self->sup_prepoll(self->sup_pp_magic, self->sup_pp_root);
if (self->sup_head)
messages = su_port_osx_getmsgs(self);
if (self->sup_timers)
timers = su_timer_expire(&self->sup_timers, &tout, su_now());
if (!self->sup_running)
break;
if (self->sup_head) /* if there are messages do a quick wait */
tout = 0;
bedtime = su_now();
events = su_port_osx_wait_events(self, tout);
woken = su_now();
if (messages || timers || events)
SU_DEBUG_1(("su_port_osx_run(%p): %.6f: %u messages %u timers %u "
"events slept %.6f/%.3f\n",
self, su_time_diff(woken, started), messages, timers, events,
su_time_diff(woken, bedtime), tout0 * 1e-3));
if (!self->sup_running)
break;
}
}
#endif
/** @internal
* The function @c su_port_osx_break() is used to terminate execution of @c
* su_port_osx_run(). It can be called from a callback function.
*
* @param self pointer to port
*
*/
void su_port_osx_break(su_port_t *self)
{
if (self->sup_main_loop)
CFRunLoopStop(self->sup_main_loop);
self->sup_running = 0;
}
/** @internal
* The function @c su_port_osx_wait_events() is used to poll() for wait objects
*
* @param self pointer to port
* @param tout timeout in milliseconds
*
* @return number of events handled
*/
static
int su_port_osx_wait_events(su_port_t *self, su_duration_t tout)
{
int i, events = 0;
su_wait_t *waits = self->sup_waits;
unsigned n = self->sup_n_waits;
#if HAVE_POLL
unsigned version = self->sup_registers;
#endif
su_root_t *root;
i = su_wait(waits, n, tout);
if (i >= 0 && (unsigned)i < n) {
#if HAVE_POLL
/* poll() can return events for multiple wait objects */
if (self->sup_multishot) {
for (; i < n; i++) {
if (waits[i].revents) {
root = self->sup_wait_roots[i];
self->sup_wait_cbs[i](root ? su_root_magic(root) : NULL,
&waits[i],
self->sup_wait_args[i]);
events++;
/* Callback function used su_register()/su_deregister() */
if (version != self->sup_registers)
break;
}
}
}
#else /* !HAVE_POLL */
if (0) {
}
#endif
else {
root = self->sup_wait_roots[i];
self->sup_wait_cbs[i](root ? su_root_magic(root) : NULL,
&self->sup_waits[i],
self->sup_wait_args[i]);
events++;
}
}
return events;
}
/** @internal
* Used to check wait events in callbacks that take lots of time
*
* This function does a timeout 0 poll() and runs wait objects.
*
* @param port pointer to port
*
* @return number of events handled
*/
static
int su_port_osx_yield(su_port_t *port)
{
return su_port_osx_wait_events(port, 0);
}
/** @internal Block until wait object is signaled or timeout.
*
* This function waits for wait objects and the timers associated with
* the root object. When any wait object is signaled or timer is
* expired, it invokes the callbacks.
*
* This function returns when a callback has been invoked or @c tout
* milliseconds is elapsed.
*
* @param self pointer to port
* @param tout timeout in milliseconds
*
* @return
* Milliseconds to the next invocation of timer, or @c SU_WAIT_FOREVER if
* there are no active timers.
*/
su_duration_t su_port_osx_step(su_port_t *self, su_duration_t tout)
{
CFRunLoopRef rl;
su_time_t now = su_now();
CFAbsoluteTime start;
int ret, timeout = tout > INT32_MAX ? INT32_MAX : tout;
assert(SU_PORT_OSX_OWN_THREAD(self));
rl = CFRunLoopGetCurrent();
if (!rl)
return -1;
CFRunLoopWakeUp(rl);
if (tout < timeout)
timeout = tout;
if (self->sup_prepoll)
self->sup_prepoll(self->sup_pp_magic, self->sup_pp_root);
if (self->sup_head)
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, now);
/* if there are messages do a quick wait */
if (self->sup_head)
tout = 0;
start = CFAbsoluteTimeGetCurrent();
for (;;) {
/* Run loop with only one pass, indicate if a source was processed */
ret = CFRunLoopRunInMode(kCFRunLoopDefaultMode,
0,
TRUE);
/* Ok, one of our sources was fired */
if (self->sup_source_fired == 1) {
self->sup_source_fired = 0;
break;
}
/* Check how long to run this loop */
if (CFAbsoluteTimeGetCurrent() >= start + timeout / 1000)
break;
}
if (self->sup_head)
su_port_osx_getmsgs(self);
if (self->sup_timers)
su_timer_expire(&self->sup_timers, &tout, su_now());
if (self->sup_head)
tout = 0;
return tout;
}
/** @internal
* Checks if the calling thread owns the port object.
*
* @param self pointer to a port object
*
* @retval true (nonzero) if the calling thread owns the port,
* @retval false (zero) otherwise.
*/
int su_port_osx_own_thread(su_port_t const *self)
{
return self == NULL || SU_PORT_OSX_OWN_THREAD(self);
}
#if 0
/** @internal
* Prints out the contents of the port.
*
* @param self pointer to a port
* @param f pointer to a file (if @c NULL, uses @c stdout).
*/
void su_port_osx_dump(su_port_t const *self, FILE *f)
{
int i;
#define IS_WAIT_IN(x) (((x)->events & SU_WAIT_IN) ? "IN" : "")
#define IS_WAIT_OUT(x) (((x)->events & SU_WAIT_OUT) ? "OUT" : "")
#define IS_WAIT_ACCEPT(x) (((x)->events & SU_WAIT_ACCEPT) ? "ACCEPT" : "")
if (f == NULL)
f = stdout;
fprintf(f, "su_port_osx_t at %p:\n", self);
fprintf(f, "\tport is%s running\n", self->sup_running ? "" : "not ");
#if SU_HAVE_PTHREADS
fprintf(f, "\tport tid %p\n", (void *)self->sup_tid);
#endif
#if SU_HAVE_MBOX
fprintf(f, "\tport mbox %d (%s%s%s)\n", self->sup_mbox[0],
IS_WAIT_IN(&self->sup_mbox_wait),
IS_WAIT_OUT(&self->sup_mbox_wait),
IS_WAIT_ACCEPT(&self->sup_mbox_wait));
#endif
fprintf(f, "\t%d wait objects\n", self->sup_n_waits);
for (i = 0; i < self->sup_n_waits; i++) {
}
}
#endif
/* =========================================================================
* Pre-poll() callback
*/
int su_port_osx_add_prepoll(su_port_t *port,
su_root_t *root,
su_prepoll_f *callback,
su_prepoll_magic_t *magic)
{
if (port->sup_prepoll)
return -1;
port->sup_prepoll = callback;
port->sup_pp_magic = magic;
port->sup_pp_root = root;
return 0;
}
int su_port_osx_remove_prepoll(su_port_t *port,
su_root_t *root)
{
if (port->sup_pp_root != root)
return -1;
port->sup_prepoll = NULL;
port->sup_pp_magic = NULL;
port->sup_pp_root = NULL;
return 0;
}
/* =========================================================================
* Timers
*/
static
su_timer_t **su_port_osx_timers(su_port_t *self)
{
return &self->sup_timers;
}
Reference in New Issue View Git Blame Copy Permalink