su_osx_runloop.c 32.6 KB
Newer Older
Martti Mela's avatar
Martti Mela committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24
/*
 * 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
 *
 */

25 26 27 28 29 30 31 32 33
/**@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>
34
 * @author Martti Mela <martti.mela@nokia.com>
35 36
 *
 * @date Created: Tue Sep 14 15:51:04 1999 ppessi
Martti Mela's avatar
Martti Mela committed
37 38 39 40 41 42
 */

#include "config.h"

#include <stdlib.h>
#include <assert.h>
43
#include <stdarg.h>
Martti Mela's avatar
Martti Mela committed
44 45 46
#include <stdio.h>
#include <string.h>
#include <limits.h>
47
#include <errno.h>
Martti Mela's avatar
Martti Mela committed
48

49
#define su_port_s su_osx_port_s 
Martti Mela's avatar
Martti Mela committed
50 51

#include "su_port.h"
52
#include "sofia-sip/su_osx_runloop.h"
Martti Mela's avatar
Martti Mela committed
53
#include "sofia-sip/su_alloc.h"
54 55 56 57 58 59 60 61 62 63
#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

Martti Mela's avatar
Martti Mela committed
64 65
static su_port_t *su_osx_runloop_create(void) __attribute__((__malloc__));

66
/* Callback for CFObserver and CFSocket */
67 68 69 70 71
static void cf_observer_cb(CFRunLoopObserverRef observer, 
			   CFRunLoopActivity activity, 
			   void *info);

static void su_osx_port_socket_cb(CFSocketRef s, 
72 73 74 75 76
				  CFSocketCallBackType callbackType, 
				  CFDataRef address, 
				  const void *data, 
				  void *info);

77 78 79 80 81 82 83 84 85
static void su_osx_port_deinit(void *arg);

static void su_osx_port_decref(su_port_t *self, int blocking, char const *who)
{
  (void)su_base_port_decref(self, blocking, who);
}


static CFSocketCallBackType map_poll_event_to_cf_event(int events);
Martti Mela's avatar
Martti Mela committed
86

87
static int su_osx_port_send(su_port_t *self, su_msg_r rmsg);
Martti Mela's avatar
Martti Mela committed
88

89
static int su_osx_port_register(su_port_t *self,
Martti Mela's avatar
Martti Mela committed
90 91 92 93 94
			    su_root_t *root, 
			    su_wait_t *wait, 
			    su_wakeup_f callback,
			    su_wakeup_arg_t *arg,
			    int priority);
95
static int su_osx_port_unregister(su_port_t *port,
Martti Mela's avatar
Martti Mela committed
96 97 98 99
			      su_root_t *root, 
			      su_wait_t *wait,	
			      su_wakeup_f callback, 
			      su_wakeup_arg_t *arg);
100

101
static int su_osx_port_deregister(su_port_t *self, int i);
102

103
static int su_osx_port_unregister_all(su_port_t *self,
104 105
			   su_root_t *root);

106 107 108 109
static int su_osx_port_eventmask(su_port_t *, int , int, int );
static void su_osx_port_run(su_port_t *self);
static void su_osx_port_break(su_port_t *self);
static su_duration_t su_osx_port_step(su_port_t *self, su_duration_t tout);
110

111
static int su_osx_port_multishot(su_port_t *port, int multishot);
112

113
static int su_osx_port_wait_events(su_port_t *self, su_duration_t tout);
114

115 116 117 118
static char const *su_osx_port_name(su_port_t const *self)
{
  return "CFRunLoop";
}
119

120
su_port_vtable_t const su_osx_port_vtable[1] =
Martti Mela's avatar
Martti Mela committed
121
  {{
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150
      /* su_vtable_size: */ sizeof su_osx_port_vtable,
      su_pthread_port_lock,
      su_pthread_port_unlock,
      su_base_port_incref,
      su_osx_port_decref,
      su_base_port_gsource,
      su_osx_port_send,
      su_osx_port_register,
      su_osx_port_unregister,
      su_osx_port_deregister,
      su_osx_port_unregister_all,
      su_osx_port_eventmask,
      su_osx_port_run,
      su_osx_port_break,
      su_osx_port_step,
      su_pthread_port_own_thread,
      su_base_port_add_prepoll,
      su_base_port_remove_prepoll,
      su_base_port_timers,
      su_osx_port_multishot,
      su_base_port_threadsafe,
      su_base_port_yield,
      su_osx_port_wait_events,
      su_base_port_getmsgs,
      su_base_port_getmsgs_from,
      su_osx_port_name,
      su_base_port_start_shared,
      su_pthread_port_wait,
      su_pthread_port_execute,
Martti Mela's avatar
Martti Mela committed
151 152
    }};

153
/*
Martti Mela's avatar
Martti Mela committed
154 155 156 157
 * Port is a per-thread reactor.  
 *
 * Multiple root objects executed by single thread share a su_port_t object. 
 */
158 159
struct su_osx_port_s {
  su_socket_port_t sup_socket[1];
160

161 162 163
#define sup_pthread sup_socket->sup_base
#define sup_base sup_socket->sup_base->sup_base
#define sup_home sup_socket->sup_base->sup_base->sup_home
Martti Mela's avatar
Martti Mela committed
164

165
  unsigned         sup_source_fired;
166

167 168 169
  CFRunLoopRef        sup_main_loop;
  CFRunLoopSourceRef *sup_sources;
  CFSocketRef        *sup_sockets;
170 171 172 173 174 175 176 177 178

  CFRunLoopObserverRef sup_observer;
  CFRunLoopObserverContext sup_observer_cntx[1];
  /* Struct for CFSocket callbacks; contains current CFSource index */
  struct osx_magic {
    su_port_t *o_port;
    int        o_current;
    int        o_count;
  } osx_magic[1];
Martti Mela's avatar
Martti Mela committed
179
  
180
  unsigned         sup_multishot; /**< Multishot operation? */
Martti Mela's avatar
Martti Mela committed
181 182 183 184

  unsigned         sup_registers; /** Counter incremented by 
				      su_port_register() or 
				      su_port_unregister()
185 186 187
				   */
  int              sup_n_waits; /**< Active su_wait_t in su_waits */
  int              sup_size_waits; /**< Size of allocate su_waits */
188

189
  int              sup_pri_offset; /**< Offset to prioritized waits */
190

191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207
#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; 
Martti Mela's avatar
Martti Mela committed
208 209 210
};


211
/* XXX - mela static void su_osx_port_destroy(su_port_t *self); */
Martti Mela's avatar
Martti Mela committed
212

213 214 215 216 217 218 219
/** 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.
220 221
 *
 * @NEW_1_12_4.
222
 */
Martti Mela's avatar
Martti Mela committed
223 224 225 226 227
su_root_t *su_root_osx_runloop_create(su_root_magic_t *magic)
{
  return su_root_create_with_port(magic, su_osx_runloop_create());
}

228 229 230 231 232 233 234
void osx_enabler_cb(CFSocketRef s, 
		    CFSocketCallBackType type, 
		    CFDataRef address, 
		    const void *data, 
		    void *info)
{
  CFRunLoopRef  rl;
235
  struct osx_magic  *magic = (struct osx_magic *) info;
236 237 238 239 240 241
  su_port_t    *self = magic->o_port;
  su_duration_t tout = 0;
  su_time_t     now = su_now();
  
  rl = CFRunLoopGetCurrent();

242
  if (self->sup_base->sup_running) {
243
    
244 245
    if (self->sup_base->sup_prepoll)
      self->sup_base->sup_prepoll(self->sup_base->sup_pp_magic, self->sup_base->sup_pp_root);
246
    
247 248
    if (self->sup_base->sup_head)
      su_base_port_getmsgs(self);
249
    
250 251
    if (self->sup_base->sup_timers)
      su_timer_expire(&self->sup_base->sup_timers, &tout, now);
252 253 254 255 256 257
  }
  
  CFRunLoopWakeUp(rl);
}


Martti Mela's avatar
Martti Mela committed
258 259
/**@internal
 *
260
 * Allocates and initializes a message port.
Martti Mela's avatar
Martti Mela committed
261 262 263 264 265 266 267
 *
 * @return
 *   If successful a pointer to the new message port is returned, otherwise
 *   NULL is returned.  
 */
su_port_t *su_osx_runloop_create(void)
{
268
  su_port_t *self = su_home_new(sizeof *self);
269

270 271
  if (!self)
    return self;
272

273
  enter;
274

275 276
  if (su_home_destructor(su_port_home(self), su_osx_port_deinit) < 0)
    return su_home_unref(su_port_home(self)), NULL;
277

278
  self->sup_multishot = SU_ENABLE_MULTISHOT_POLL;
279

280 281 282 283 284 285 286
  if (su_socket_port_init(self->sup_base, su_osx_port_vtable) == 0) {
    self->osx_magic->o_port = self;
    self->sup_observer_cntx->info = self->osx_magic;
    self->sup_observer =
      CFRunLoopObserverCreate(NULL, 
			      kCFRunLoopAfterWaiting | kCFRunLoopBeforeWaiting,
			      TRUE, 0, cf_observer_cb, self->sup_observer_cntx);
287
#if 0
288
    CFRunLoopAddObserver(CFRunLoopGetCurrent(),
289
			 self->sup_observer,
290
			 kCFRunLoopDefaultMode);
291
#endif
Martti Mela's avatar
Martti Mela committed
292
  }
293 294 295
  else
    return su_home_unref(su_port_home(self)), NULL;

296
  return self;
Martti Mela's avatar
Martti Mela committed
297 298
}

299 300 301 302 303 304
static
void cf_observer_cb(CFRunLoopObserverRef observer, 
		    CFRunLoopActivity activity, 
		    void *info)
{
  CFRunLoopRef  rl;
305
  struct osx_magic  *magic = (struct osx_magic *) info;
306 307 308 309 310 311
  su_port_t    *self = magic->o_port;
  su_duration_t tout = 0;
  su_time_t     now = su_now();

  rl = CFRunLoopGetCurrent();

312
  if (self->sup_base->sup_running) {
313

314 315
    if (self->sup_base->sup_prepoll)
      self->sup_base->sup_prepoll(self->sup_base->sup_pp_magic, self->sup_base->sup_pp_root);
316
    
317 318
    if (self->sup_base->sup_head)
      su_port_getmsgs(self);
319
    
320 321
    if (self->sup_base->sup_timers)
      su_timer_expire(&self->sup_base->sup_timers, &tout, now);
322 323 324 325 326 327 328 329
  } else
    SU_DEBUG_9(("cf_observer_cb(): PORT IS NOT RUNNING!\n"));

  CFRunLoopWakeUp(rl);
  
  return;
}

Martti Mela's avatar
Martti Mela committed
330
/** @internal Destroy a port. */
331
static void su_osx_port_deinit(void *arg)
Martti Mela's avatar
Martti Mela committed
332
{
333
  su_port_t *self = arg;
Martti Mela's avatar
Martti Mela committed
334

335
  SU_DEBUG_9(("%s(%p) called\n", "su_osx_port_deinit", (void *)self));
Martti Mela's avatar
Martti Mela committed
336

337
  su_socket_port_deinit(self->sup_base);
338 339
}

340 341
static
CFSocketCallBackType map_poll_event_to_cf_event(int events)
342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359
{
  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;
Martti Mela's avatar
Martti Mela committed
360 361
}

362

363
#if 0
364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382
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;
}
383
#endif
384

385
static
386
void su_osx_port_socket_cb(CFSocketRef s, 
387 388 389 390
			   CFSocketCallBackType type, 
			   CFDataRef address, 
			   const void *data, 
			   void *info)
Martti Mela's avatar
Martti Mela committed
391
{
392 393 394
  struct osx_magic *magic = (struct osx_magic *) info;
  su_port_t        *self = magic->o_port;
  int               curr = magic->o_current;
395 396
  su_duration_t tout = 0;
  
397 398 399 400 401
#if SU_HAVE_POLL
  {
    su_root_t *root;
    su_wait_t *waits = self->sup_waits;
    int n = self->sup_indices[curr];
402
    
403
    assert(self->sup_reverses[n] == curr);
404 405 406
    
    SU_DEBUG_9(("socket_cb(%p): count %u index %d\n", self->sup_sources[n], magic->o_count, curr));
    
407 408
    waits[n].revents = map_poll_event_to_cf_event(type);

409 410 411 412
    root = self->sup_wait_roots[n];
    self->sup_wait_cbs[n](root ? su_root_magic(root) : NULL, 
			  &waits[n], 
			  self->sup_wait_args[n]);
413
    
414 415
    if (self->sup_base->sup_running) {
      su_port_getmsgs(self);
416
      
417 418
      if (self->sup_base->sup_timers)
	su_timer_expire(&self->sup_base->sup_timers, &tout, su_now());
419

420
      if (self->sup_base->sup_head)
421 422
	tout = 0;

423
      /* CFRunLoopWakeUp(CFRunLoopGetCurrent()); */
424 425
    }
    
426 427 428 429
    /* Tell to run loop an su socket fired */
    self->sup_source_fired = 1;
  }
#endif
430
  
Martti Mela's avatar
Martti Mela committed
431 432 433
}

/** @internal Send a message to the port. */
434
int su_osx_port_send(su_port_t *self, su_msg_r rmsg)
Martti Mela's avatar
Martti Mela committed
435
{
436 437
  CFRunLoopRef rl;

Martti Mela's avatar
Martti Mela committed
438
  if (self) {
439
    int wakeup;
Martti Mela's avatar
Martti Mela committed
440

441
    //XXX - mela SU_OSX_PORT_LOCK(self, "su_osx_port_send");
442
    
443
    wakeup = self->sup_base->sup_head == NULL;
Martti Mela's avatar
Martti Mela committed
444

445 446
    *self->sup_base->sup_tail = rmsg[0]; rmsg[0] = NULL;
    self->sup_base->sup_tail = &(*self->sup_base->sup_tail)->sum_next;
447 448 449 450 451 452 453 454 455 456 457 458 459 460 461

#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
Martti Mela's avatar
Martti Mela committed
462

463
    //XXX - mela SU_OSX_PORT_UNLOCK(self, "su_osx_port_send");
Martti Mela's avatar
Martti Mela committed
464 465

    rl = CFRunLoopGetCurrent();
466
    CFRunLoopWakeUp(rl);
Martti Mela's avatar
Martti Mela committed
467 468 469 470 471 472 473 474

    return 0;
  }
  else {
    su_msg_destroy(rmsg);
    return -1;
  }
}
475 476
static int o_count;

Martti Mela's avatar
Martti Mela committed
477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494
/** @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
495
 *   The function @su_osx_port_register returns nonzero index of the wait object, 
Martti Mela's avatar
Martti Mela committed
496
 *   or -1 upon an error.  */
497
int su_osx_port_register(su_port_t *self,
498 499 500 501 502
			 su_root_t *root, 
			 su_wait_t *wait, 
			 su_wakeup_f callback,
			 su_wakeup_arg_t *arg,
			 int priority)
Martti Mela's avatar
Martti Mela committed
503
{
504
  int i, j, n;
Martti Mela's avatar
Martti Mela committed
505
  CFRunLoopRef rl;
506 507
  CFRunLoopSourceRef *sources, source;
  CFSocketRef cf_socket, *sockets;
508
  int events = 0;
509
  struct osx_magic *osx_magic = NULL;
510 511
  CFSocketContext cf_socket_cntx[1] = {{0, NULL, NULL, NULL, NULL}};
  CFOptionFlags flags = 0;
Martti Mela's avatar
Martti Mela committed
512

513
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));
Martti Mela's avatar
Martti Mela committed
514 515 516

  n = self->sup_n_waits;

517 518 519
  if (n >= SU_WAIT_MAX)
    return su_seterrno(ENOMEM);

Martti Mela's avatar
Martti Mela committed
520 521 522
  if (n >= self->sup_size_waits) {
    /* Reallocate size arrays */
    int size;
523 524
    int *indices;
    int *reverses;
Martti Mela's avatar
Martti Mela committed
525 526 527 528 529 530
    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)
531
      size = su_root_size_hint;
Martti Mela's avatar
Martti Mela committed
532 533 534
    else 
      size = 2 * self->sup_size_waits;

535 536
    if (size < SU_WAIT_MIN)
      size = SU_WAIT_MIN;
537 538 539 540 541

    /* Too large */
    if (-3 - size > 0)
      return (errno = ENOMEM), -1;

542
    indices = realloc(self->sup_indices, (size + 1) * sizeof(*indices));
Martti Mela's avatar
Martti Mela committed
543 544 545
    if (indices) {
      self->sup_indices = indices;

546 547
      for (i = self->sup_size_waits; i <= size; i++)
	indices[i] = -1 - i;
548
    }
Martti Mela's avatar
Martti Mela committed
549

550 551 552 553 554 555
    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;
    }
Martti Mela's avatar
Martti Mela committed
556
      
557 558 559 560
    sources = realloc(self->sup_sources, size * sizeof(*sources));
    if (sources)
      self->sup_sources = sources;

561 562 563 564 565 566 567 568
    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;

Martti Mela's avatar
Martti Mela committed
569 570 571 572 573 574 575 576 577 578 579 580 581
    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;

582
    if (!(indices && 
583
	  reverses && sources && sockets && waits && wait_cbs && wait_args && wait_tasks)) {
Martti Mela's avatar
Martti Mela committed
584 585 586 587 588 589
      return -1;
    }

    self->sup_size_waits = size;
  }

590
  i = -self->sup_indices[0]; assert(i <= self->sup_size_waits);
591

Martti Mela's avatar
Martti Mela committed
592 593
  if (priority > 0) {
    /* Insert */
594 595 596
    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;
597 598
      self->sup_reverses[n] = self->sup_reverses[n-1];
      self->sup_sources[n] = self->sup_sources[n-1];
599 600
      self->sup_sockets[n] = self->sup_sockets[n-1];
      self->sup_waits[n] = self->sup_waits[n-1];
Martti Mela's avatar
Martti Mela committed
601 602 603 604
      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];	
    }
605

606
    self->sup_pri_offset++;
Martti Mela's avatar
Martti Mela committed
607 608 609
  }
  else {
    /* Append - no need to move anything */
610
    n = self->sup_n_waits;
Martti Mela's avatar
Martti Mela committed
611 612
  }

613 614 615 616
  self->sup_n_waits++;

  self->sup_indices[0] = self->sup_indices[i];  /* Free index */
  self->sup_indices[i] = n;
617 618

  self->sup_reverses[n] = i;
Martti Mela's avatar
Martti Mela committed
619 620 621 622 623
  self->sup_waits[n] = *wait;
  self->sup_wait_cbs[n] = callback;
  self->sup_wait_args[n] = arg;
  self->sup_wait_roots[n] = root;

624
  self->sup_registers++;
Martti Mela's avatar
Martti Mela committed
625

626 627 628 629 630 631 632 633 634 635 636
  /* 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),
637
				       events, su_osx_port_socket_cb, cf_socket_cntx);
638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661

  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;
662
}
Martti Mela's avatar
Martti Mela committed
663

664 665
/** Deregister a su_wait_t object. */
static
666
int su_osx_port_deregister0(su_port_t *self, int i)
667 668
{
  CFRunLoopRef rl;
669
  int n, N, *indices, *reverses;
Martti Mela's avatar
Martti Mela committed
670

671 672 673
  indices = self->sup_indices;
  reverses = self->sup_reverses;

674
  n = indices[i]; assert(n >= 0); assert(i == reverses[n]);
675

676
  N = --self->sup_n_waits;
677 678
  
  rl = CFRunLoopGetCurrent();
679 680 681 682 683
  CFSocketInvalidate(self->sup_sockets[n]);
  CFRelease(self->sup_sockets[n]);
  CFRunLoopRemoveSource(rl, self->sup_sources[n], kCFRunLoopDefaultMode);
  CFRelease(self->sup_sources[n]);

684
  CFRunLoopWakeUp(rl);
685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700

  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;
701 702 703
    }
  }

704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730
  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;
Martti Mela's avatar
Martti Mela committed
731 732 733

  self->sup_registers++;

734
  return i;
Martti Mela's avatar
Martti Mela committed
735 736
}

737

Martti Mela's avatar
Martti Mela committed
738 739
/** Unregister a su_wait_t object.
 *  
740
 *  The function su_osx_port_unregister() unregisters a su_wait_t object. The
Martti Mela's avatar
Martti Mela committed
741 742 743 744 745 746 747 748 749 750 751 752
 *  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.
 */
753
int su_osx_port_unregister(su_port_t *self,
754 755 756 757
		       su_root_t *root, 
		       su_wait_t *wait,	
		       su_wakeup_f callback, /* XXX - ignored */
		       su_wakeup_arg_t *arg)
Martti Mela's avatar
Martti Mela committed
758
{
759
  int n, N;
Martti Mela's avatar
Martti Mela committed
760 761

  assert(self);
762
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));
Martti Mela's avatar
Martti Mela committed
763 764 765 766

  N = self->sup_n_waits;

  for (n = 0; n < N; n++) {
767
    if (SU_WAIT_CMP(wait[0], self->sup_waits[n]) == 0) {
768
      return su_osx_port_deregister0(self, self->sup_reverses[n]);
Martti Mela's avatar
Martti Mela committed
769 770 771
    }
  }

772
  su_seterrno(ENOENT);
Martti Mela's avatar
Martti Mela committed
773

774
  return -1;
Martti Mela's avatar
Martti Mela committed
775 776 777 778
}

/** Deregister a su_wait_t object.
 *  
779
 *  The function su_osx_port_deregister() deregisters a su_wait_t registrattion. 
Martti Mela's avatar
Martti Mela committed
780 781 782 783 784 785 786 787
 *  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.
 */
788
int su_osx_port_deregister(su_port_t *self, int i)
Martti Mela's avatar
Martti Mela committed
789
{
790 791
  su_wait_t wait[1] = { SU_WAIT_INIT };
  int retval;
Martti Mela's avatar
Martti Mela committed
792 793

  assert(self);
794
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));
Martti Mela's avatar
Martti Mela committed
795

796 797
  if (i <= 0 || i > self->sup_size_waits)
    return su_seterrno(EBADF);
Martti Mela's avatar
Martti Mela committed
798

799 800 801
  if (self->sup_indices[i] < 0)
    return su_seterrno(EBADF);
    
802
  retval = su_osx_port_deregister0(self, i);
Martti Mela's avatar
Martti Mela committed
803 804 805

  su_wait_destroy(wait);

806
  return retval;
Martti Mela's avatar
Martti Mela committed
807 808
}

809

Martti Mela's avatar
Martti Mela committed
810 811 812
/** @internal
 * Unregister all su_wait_t objects.
 *
813
 * The function su_osx_port_unregister_all() unregisters all su_wait_t objects
814
 * and destroys all queued timers associated with given root object.
Martti Mela's avatar
Martti Mela committed
815 816 817 818 819 820
 * 
 * @param  self     - pointer to port object
 * @param  root     - pointer to root object
 * 
 * @return Number of wait objects removed.
 */
821
int su_osx_port_unregister_all(su_port_t *self, 
Martti Mela's avatar
Martti Mela committed
822 823
			   su_root_t *root)
{
824 825 826 827 828 829 830
  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;
831
  CFRunLoopSourceRef *sources;
832
  CFSocketRef        *sockets;
Martti Mela's avatar
Martti Mela committed
833

834
  // XXX - assert(SU_OSX_PORT_OWN_THREAD(self));
Martti Mela's avatar
Martti Mela committed
835

836
  N          = self->sup_n_waits;
837 838 839
  indices    = self->sup_indices;
  reverses   = self->sup_reverses;
  sources    = self->sup_sources; 
840
  sockets    = self->sup_sockets; 
Martti Mela's avatar
Martti Mela committed
841 842 843 844 845
  waits      = self->sup_waits; 
  wait_cbs   = self->sup_wait_cbs; 
  wait_args  = self->sup_wait_args;
  wait_roots = self->sup_wait_roots; 
  
846 847 848 849 850
  rl = CFRunLoopGetCurrent();

  for (i = j = 0; i < N; i++) {
    index = reverses[i]; assert(index > 0 && indices[index] == i);

Martti Mela's avatar
Martti Mela committed
851
    if (wait_roots[i] == root) {
852 853 854 855 856
      if (i < self->sup_pri_offset)
	self->sup_pri_offset--;

      indices[index] = indices[0];
      indices[0] = -index;
Martti Mela's avatar
Martti Mela committed
857 858
      continue;
    }
859

860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875
    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];
    }
876
    
Martti Mela's avatar
Martti Mela committed
877 878
    j++;
  }
879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899

  /* 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);

Martti Mela's avatar
Martti Mela committed
900 901 902
  self->sup_n_waits = j;
  self->sup_registers++;

903
  return N - j;
Martti Mela's avatar
Martti Mela committed
904 905 906 907
}

/**Set mask for a registered event. @internal
 *
908
 * The function su_osx_port_eventmask() sets the mask describing events that can
Martti Mela's avatar
Martti Mela committed
909 910 911 912 913 914 915 916 917 918
 * 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.
 */
919
int su_osx_port_eventmask(su_port_t *self, int index, int socket, int events)
Martti Mela's avatar
Martti Mela committed
920
{
921
  int n, ret;
Martti Mela's avatar
Martti Mela committed
922 923

  assert(self);
924
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));
Martti Mela's avatar
Martti Mela committed
925

926
  if (index <= 0 || index > self->sup_size_waits)
927 928
    return su_seterrno(EBADF);
  n = self->sup_indices[index];
929
  if (n < 0)
930
    return su_seterrno(EBADF);
Martti Mela's avatar
Martti Mela committed
931

932 933 934 935 936 937
  ret = su_wait_mask(&self->sup_waits[n], socket, events);

  CFSocketSetSocketFlags(self->sup_sockets[n],
			 map_poll_event_to_cf_event(events));
  
  return ret;
Martti Mela's avatar
Martti Mela committed
938 939
}

940 941 942
/** @internal
 *
 *  Copies the su_wait_t objects from the port. The number of wait objects
943
 *  can be found out by calling su_osx_port_query() with @a n_waits as zero.
944 945 946 947 948 949 950 951 952
 * 
 * @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.
 */
953
unsigned su_osx_port_query(su_port_t *self, su_wait_t *waits, unsigned n_waits)
954 955 956
{
  unsigned n;

957
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));
958 959 960 961 962 963 964 965 966 967 968 969 970 971 972

  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.
 *
973
 * The function su_osx_port_multishot() enables, disables or queries the
974 975 976 977 978 979 980 981 982 983 984 985
 * 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
 */
986
int su_osx_port_multishot(su_port_t *self, int multishot)
Martti Mela's avatar
Martti Mela committed
987
{
988 989
  if (multishot < 0)
    return self->sup_multishot;
Martti Mela's avatar
Martti Mela committed
990
  else if (multishot == 0 || multishot == 1)
991
    return self->sup_multishot = multishot;
Martti Mela's avatar
Martti Mela committed
992 993 994 995
  else 
    return (errno = EINVAL), -1;
}

996
#if 0
Martti Mela's avatar
Martti Mela committed
997 998
/** @internal Enable threadsafe operation. */
static
999
int su_osx_port_threadsafe(su_port_t *port)
Martti Mela's avatar
Martti Mela committed
1000 1001 1002
{
  return su_home_threadsafe(port->sup_home);
}
1003
#endif
Martti Mela's avatar
Martti Mela committed
1004

1005
/** Prepare root to be run on OSX Run Loop.
1006 1007 1008 1009 1010 1011
 *
 * 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.
1012
 * 
1013
 * @param root     pointer to root object
1014
 * 
1015
 * @NEW_1_12_4.
1016 1017 1018 1019 1020 1021 1022
 */
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;

1023
  // XXX - mela assert(SU_OSX_PORT_OWN_THREAD(self));
1024 1025 1026

  enter;

1027
  self->sup_base->sup_running = 1;
1028 1029
  rl = CFRunLoopGetCurrent();

1030 1031
  if (self->sup_base->sup_prepoll)
    self->sup_base->sup_prepoll(self->sup_base->sup_pp_magic, self->sup_base->sup_pp_root);
1032

1033 1034
  if (self->sup_base->sup_head)
    su_port_getmsgs(self);
1035
  
1036 1037
  if (self->sup_base->sup_timers)
    su_timer_expire(&self->sup_base->sup_timers, &tout, su_now());
1038

1039
  if (!self->sup_base->sup_running)