su_select_port.c 16.7 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27
/*
 * 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_select_port.c
 *
28
 * Port implementation using select().
29 30 31 32
 *
 * @author Pekka Pessi <Pekka.Pessi@nokia.com>
 * @author Kai Vehmanen <kai.vehmanen@nokia.com>
 *
33
 * @date Created: Fri Jan 26 17:56:34 2007 ppessi
34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53
 */

#include "config.h"

#if HAVE_SELECT

#define su_port_s su_select_port_s

#include "sofia-sip/su.h"
#include "su_port.h"
#include "sofia-sip/su_alloc.h"

#include <stdlib.h>
#include <assert.h>
#include <stdarg.h>
#include <stdio.h>
#include <string.h>
#include <limits.h>
#include <errno.h>

54 55 56
#if HAVE_WIN32
#error winsock select() not supported yet
#else
57 58 59 60 61 62
#if HAVE_SYS_SELECT_H
#include <sys/select.h>
#elif HAVE_SYS_TIME_H
#include <sys/time.h>
#endif

63 64 65 66 67 68 69 70 71 72 73 74
#include <sys/types.h>
#include <unistd.h>

#ifndef __NFDBITS
#define __NFDBITS (8 * sizeof (long int))
#endif

#undef FDSETSIZE
/* Size of fd set in bytes */
#define FDSETSIZE(n) (((n) + __NFDBITS - 1) / __NFDBITS * (__NFDBITS / 8))
#endif

75 76 77
/** Port based on select(). */

struct su_select_port_s {
78
  su_socket_port_t sup_base[1];
79

80 81 82 83
#define sup_home sup_base->sup_base->sup_base->sup_home

  /** epoll fd */
  int              sup_epoll;
84 85
  unsigned         sup_multishot; /**< Multishot operation? */

86 87
  unsigned         sup_registers; /** Counter incremented by
				      su_port_register() or
88 89
				      su_port_unregister()
				   */
90 91 92 93 94 95 96
  int              sup_n_registrations;
  int              sup_max_index; /**< Indexes are equal or smaller than this */
  int              sup_size_indices; /**< Size of allocated index table */

  /** Structure containing registration data */
  struct su_select_register {
    struct su_select_register *ser_next; /* Next in free list */
97 98 99
    su_wakeup_f     ser_cb;
    su_wakeup_arg_t*ser_arg;
    su_root_t      *ser_root;
100 101 102 103 104
    int             ser_id; /** registration identifier */
    su_wait_t       ser_wait[1];
  } **sup_indices;

  int               sup_maxfd, sup_allocfd;
105

106 107
  fd_set           *sup_readfds, *sup_readfds2;
  fd_set           *sup_writefds, *sup_writefds2;
108 109
};

110 111 112
static void su_select_port_decref(su_port_t *self,
				 int blocking,
				 char const *who);
113
static int su_select_port_register(su_port_t *self,
114 115
				  su_root_t *root,
				  su_wait_t *wait,
116 117 118
				  su_wakeup_f callback,
				  su_wakeup_arg_t *arg,
				  int priority);
119
static int su_select_port_unregister(su_port_t *port,
120 121 122
				    su_root_t *root,
				    su_wait_t *wait,
				    su_wakeup_f callback,
123
				    su_wakeup_arg_t *arg);
124 125
static int su_select_port_deregister(su_port_t *self, int i);
static int su_select_port_unregister_all(su_port_t *self, su_root_t *root);
126
static int su_select_port_eventmask(su_port_t *self,
127 128 129
				   int index,
				   int socket,
				   int events);
130 131
static int su_select_port_multishot(su_port_t *self, int multishot);
static int su_select_port_wait_events(su_port_t *self, su_duration_t tout);
132
static char const *su_select_port_name(su_port_t const *self);
133 134 135 136 137 138 139 140 141

su_port_vtable_t const su_select_port_vtable[1] =
  {{
      /* su_vtable_size: */ sizeof su_select_port_vtable,
      su_pthread_port_lock,
      su_pthread_port_unlock,
      su_base_port_incref,
      su_select_port_decref,
      su_base_port_gsource,
142
      su_socket_port_send,
143 144 145 146 147 148 149 150
      su_select_port_register,
      su_select_port_unregister,
      su_select_port_deregister,
      su_select_port_unregister_all,
      su_select_port_eventmask,
      su_base_port_run,
      su_base_port_break,
      su_base_port_step,
151
      su_pthread_port_thread,
152 153 154 155 156
      su_base_port_add_prepoll,
      su_base_port_remove_prepoll,
      su_base_port_timers,
      su_select_port_multishot,
      su_select_port_wait_events,
157
      su_base_port_getmsgs,
158 159 160
      su_base_port_getmsgs_from,
      su_select_port_name,
      su_base_port_start_shared,
161
      su_pthread_port_wait,
162
      su_pthread_port_execute,
163 164
    }};

165
static char const *su_select_port_name(su_port_t const *self)
166
{
167
  return "select";
168 169
}

170 171 172 173 174
static void su_select_port_decref(su_port_t *self, int blocking, char const *who)
{
  (void)su_base_port_decref(self, blocking, who);
}

175 176 177 178
static void su_select_port_deinit(void *arg)
{
  su_port_t *self = arg;

179
  SU_DEBUG_9(("%s(%p) called\n", "su_select_port_deinit", (void *)self));
180

181
  su_socket_port_deinit(self->sup_base);
182 183 184 185
}

/** @internal
 *
186
 *  Register a #su_wait_t object. The wait object, a callback function and
187 188 189 190 191
 *  an 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.
192
 *
193 194 195 196 197
 * @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
198
 * @param priority   relative priority of the wait object
199
 *              (0 is normal, 1 important, 2 realtime)
200
 *
201
 * @return
202
 *   Positive index of the wait object,
203 204 205
 *   or -1 upon an error.
 */
int su_select_port_register(su_port_t *self,
206 207
			   su_root_t *root,
			   su_wait_t *wait,
208 209 210
			   su_wakeup_f callback,
			   su_wakeup_arg_t *arg,
			   int priority)
211 212
{
  int i, j, n;
213 214 215 216 217
  struct su_select_register *ser;
  struct su_select_register **indices = self->sup_indices;
  int allocfd = self->sup_allocfd;
  fd_set *readfds = self->sup_readfds, *readfds2 = self->sup_readfds2;
  fd_set *writefds = self->sup_writefds, *writefds2 = self->sup_writefds2;
218 219 220

  assert(su_port_own_thread(self));

221
  n = self->sup_size_indices;
222 223 224 225

  if (n >= SU_WAIT_MAX)
    return su_seterrno(ENOMEM);

226
  self->sup_registers++;
227

228
  if (wait->fd >= allocfd)
229
    allocfd += __NFDBITS;		/* long at a time */
230

231 232 233 234
  if (allocfd >= self->sup_allocfd) {
    size_t bytes = FDSETSIZE(allocfd);
    size_t bytes0 = FDSETSIZE(self->sup_allocfd);
    /* (Re)allocate fd_sets  */
235

236 237 238 239 240
    readfds = su_realloc(self->sup_home, readfds, bytes);
    if (readfds) self->sup_readfds = readfds;
    readfds2 = su_realloc(self->sup_home, readfds2, bytes);
    if (readfds2) self->sup_readfds2 = readfds2;
    if (!readfds || !readfds2)
241 242
      return -1;

243 244 245 246 247 248 249 250 251 252 253
    writefds = su_realloc(self->sup_home, writefds, bytes);
    if (writefds) self->sup_writefds = writefds;
    writefds2 = su_realloc(self->sup_home, writefds2, bytes);
    if (writefds2) self->sup_writefds2 = writefds2;
    if (!writefds || !writefds2)
      return -1;

    memset((char *)readfds + bytes0, 0, bytes - bytes0);
    memset((char *)writefds + bytes0, 0, bytes - bytes0);

    self->sup_allocfd = allocfd;
254 255
  }

256 257 258 259 260 261
  ser = indices[0];

  if (!ser) {
    su_home_t *h = su_port_home(self);

    i = self->sup_max_index, j = i == 0 ? 15 : i + 16;
262

263 264 265 266 267 268 269 270
    if (j >= self->sup_size_indices) {
      /* Reallocate index table */
      n = n < 1024 ? 2 * n : n + 1024;
      indices = su_realloc(h, indices, n * sizeof(indices[0]));
      if (!indices)
	return -1;
      self->sup_indices = indices;
      self->sup_size_indices = n;
271 272
    }

273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288
    /* Allocate registrations */
    ser = su_zalloc(h, (j - i) * (sizeof *ser));
    if (!ser)
      return -1;

    indices[0] = ser;

    for (i++; i <= j; i++) {
      ser->ser_id = i;
      ser->ser_next = i < j ? ser + 1 : NULL;
      indices[i] = ser++;
    }

    self->sup_max_index = j;

    ser = indices[0];
289 290
  }

291
  i = ser->ser_id;
292

293
  indices[0] = ser->ser_next;
294

295 296 297 298 299
  ser->ser_next = NULL;
  *ser->ser_wait = *wait;
  ser->ser_cb = callback;
  ser->ser_arg = arg;
  ser->ser_root = root;
300

301 302 303 304
  if (wait->events & SU_WAIT_IN)
    FD_SET(wait->fd, readfds);
  if (wait->events & SU_WAIT_OUT)
    FD_SET(wait->fd, writefds);
305

306 307
  if (wait->fd >= self->sup_maxfd)
    self->sup_maxfd = wait->fd + 1;
308

309
  self->sup_n_registrations++;
310

311
  return i;			/* return index */
312 313
}

314
static void su_select_port_update_maxfd(su_port_t *self)
315
{
316 317
  int i;
  su_socket_t maxfd = 0;
318

319 320 321 322 323 324
  for (i = 1; i <= self->sup_max_index; i++) {
    if (!self->sup_indices[i]->ser_cb)
      continue;
    if (maxfd <= self->sup_indices[i]->ser_wait->fd)
      maxfd = self->sup_indices[i]->ser_wait->fd + 1;
  }
325

326 327
  self->sup_maxfd = maxfd;
}
328

329 330 331 332 333 334 335 336 337 338
/** Deregister a su_wait_t object. */
static int su_select_port_deregister0(su_port_t *self, int i, int destroy_wait)
{
  struct su_select_register **indices = self->sup_indices;
  struct su_select_register *ser;

  ser = self->sup_indices[i];
  if (ser == NULL || ser->ser_cb == NULL) {
    su_seterrno(ENOENT);
    return -1;
339 340
  }

341
  assert(ser->ser_id == i);
342

343 344
  FD_CLR(ser->ser_wait->fd, self->sup_readfds);
  FD_CLR(ser->ser_wait->fd, self->sup_writefds);
345

346 347
  if (ser->ser_wait->fd + 1 >= self->sup_maxfd)
    self->sup_maxfd = 0;
348

349 350 351
  memset(ser, 0, sizeof *ser);
  ser->ser_id = i;
  ser->ser_next = indices[0], indices[0] = ser;
352

353
  self->sup_n_registrations--;
354 355 356 357 358 359 360
  self->sup_registers++;

  return i;
}


/** Unregister a su_wait_t object.
361 362
 *
 * The function su_select_port_unregister() unregisters a su_wait_t object.
363 364
 * The registration defined by the wait object, the callback function and
 * the argument pointer are removed from the port object.
365
 *
366 367 368 369
 * @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)
370
 * @param arg      - argument given to callback function when it is invoked
371 372
 *                   (may be NULL)
 *
373
 * @deprecated Use su_select_port_deregister() instead.
374 375 376 377
 *
 * @return Nonzero index of the wait object, or -1 upon an error.
 */
int su_select_port_unregister(su_port_t *self,
378 379
			     su_root_t *root,
			     su_wait_t *wait,
380 381
			     su_wakeup_f callback, /* XXX - ignored */
			     su_wakeup_arg_t *arg)
382
{
383 384 385
  int i, I;

  struct su_select_register *ser;
386 387 388 389

  assert(self);
  assert(su_port_own_thread(self));

390
  I = self->sup_max_index;
391

392 393 394 395 396 397 398
  for (i = 1; i <= I; i++) {
    ser = self->sup_indices[i];

    if (ser->ser_cb &&
	arg == ser->ser_arg &&
	SU_WAIT_CMP(wait[0], ser->ser_wait[0]) == 0)
      return su_select_port_deregister0(self, ser->ser_id, 0);
399 400 401 402 403 404 405 406
  }

  su_seterrno(ENOENT);

  return -1;
}

/** Deregister a su_wait_t object.
407
 *
408 409 410
 *  Deregisters a registration by index. The wait object, a callback
 *  function and a argument are removed from the port object. The wait
 *  object is destroyed.
411
 *
412 413
 * @param self     - pointer to port object
 * @param i        - registration index
414
 *
415 416 417 418
 * @return Index of the wait object, or -1 upon an error.
 */
int su_select_port_deregister(su_port_t *self, int i)
{
419
  struct su_select_register *ser;
420

421
  if (i <= 0 || i > self->sup_max_index)
422 423
    return su_seterrno(EBADF);

424 425
  ser = self->sup_indices[i];
  if (!ser->ser_cb)
426 427
    return su_seterrno(EBADF);

428
  return su_select_port_deregister0(self, i, 1);
429 430 431 432
}


/** @internal
433
 * Unregister all su_wait_t objects of given su_root_t instance.
434
 *
435 436
 * The function su_select_port_unregister_all() unregisters all su_wait_t
 * objects associated with given root object.
437
 *
438 439
 * @param  self     - pointer to port object
 * @param  root     - pointer to root object
440
 *
441 442
 * @return Number of wait objects removed.
 */
443
int su_select_port_unregister_all(su_port_t *self, su_root_t *root)
444
{
445
  int i, I, n;
446

447
  struct su_select_register *ser;
448

449 450
  assert(self); assert(root);
  assert(su_port_own_thread(self));
451

452
  I = self->sup_max_index;
453

454 455 456
  for (i = 1, n = 0; i <= I; i++) {
    ser = self->sup_indices[i];
    if (ser->ser_root != root)
457
      continue;
458 459
    su_select_port_deregister0(self, ser->ser_id, 0);
    n++;
460 461
  }

462
  return n;
463 464 465 466
}

/**Set mask for a registered event. @internal
 *
467 468
 * The function su_select_port_eventmask() sets the mask describing events
 * that can signal the registered callback.
469 470 471 472 473 474 475 476 477
 *
 * @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.
 */
478 479 480
int su_select_port_eventmask(su_port_t *self,
			     int index,
			     int socket, int events)
481
{
482 483 484 485
  struct su_select_register *ser;

  if (index <= 0 || index > self->sup_max_index)
    return su_seterrno(EBADF);
486

487 488
  ser = self->sup_indices[index];
  if (!ser->ser_cb)
489
    return su_seterrno(EBADF);
490 491 492 493 494

  if (self->sup_maxfd == 0)
    su_select_port_update_maxfd(self);

  if (socket >= self->sup_maxfd)
495 496
    return su_seterrno(EBADF);

497 498 499 500
  if (su_wait_mask(ser->ser_wait, socket, events) < 0)
    return -1;

  assert(socket < self->sup_maxfd);
501

502 503 504 505 506 507 508 509 510 511 512
  if (events & SU_WAIT_IN)
    FD_SET(socket, self->sup_readfds);
  else
    FD_CLR(socket, self->sup_readfds);

  if (events & SU_WAIT_OUT)
    FD_SET(socket, self->sup_writefds);
  else
    FD_CLR(socket, self->sup_writefds);

  return 0;
513 514 515 516 517 518 519 520 521 522 523
}

/** @internal Enable multishot mode.
 *
 * 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 disabled, only first socket event is served.
 *
 * @param self      pointer to port object
 * @param multishot multishot mode (0 => disables, 1 => enables, -1 => query)
524
 *
525 526 527 528 529 530 531 532 533 534 535
 * @retval 0 multishot mode is disabled
 * @retval 1 multishot mode is enabled
 * @retval -1 an error occurred
 */
static
int su_select_port_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;
536
  else
537 538 539 540 541
    return (errno = EINVAL), -1;
}


/** @internal
542
 * Wait (poll()) for wait objects in port.
543 544 545 546 547 548 549 550 551
 *
 * @param self     pointer to port
 * @param tout     timeout in milliseconds
 *
 * @return number of events handled
 */
static
int su_select_port_wait_events(su_port_t *self, su_duration_t tout)
{
552 553 554 555 556
  int j, n, events = 0;
  unsigned version = self->sup_registers;
  size_t bytes;
  struct timeval tv;
  fd_set *rset = NULL, *wset = NULL;
557

558 559
  if (self->sup_maxfd == 0)
    su_select_port_update_maxfd(self);
560

561
  bytes = FDSETSIZE(self->sup_maxfd);
562 563 564 565

  if (bytes) {
    rset = memcpy(self->sup_readfds2, self->sup_readfds, bytes);
    wset = memcpy(self->sup_writefds2, self->sup_writefds, bytes);
566
  }
567 568 569 570 571 572

  tv.tv_sec = tout / 1000;
  tv.tv_usec = (tout % 1000) * 1000;

  n = select(self->sup_maxfd, rset, wset, NULL, &tv);

573
  if (n < 0) {
574 575 576
    SU_DEBUG_0(("su_select_port_wait_events(%p): %s (%d)\n",
		(void *)self, su_strerror(su_errno()), su_errno()));
    return 0;
577
  }
578 579
  else if (n == 0)
    return 0;
580

581
  for (j = 1; j <= self->sup_max_index; j++) {
582 583 584
    struct su_select_register *ser;
    su_root_magic_t *magic;
    int fd;
585

586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611
    ser = self->sup_indices[j];
    if (!ser->ser_cb)
      continue;

    fd = ser->ser_wait->fd;
    ser->ser_wait->revents = 0;

    if (ser->ser_wait->events & SU_WAIT_IN)
      if (FD_ISSET(fd, rset)) ser->ser_wait->revents |= SU_WAIT_IN, n--;
    if (ser->ser_wait->events & SU_WAIT_OUT)
      if (FD_ISSET(fd, wset)) ser->ser_wait->revents |= SU_WAIT_OUT, n--;

    if (ser->ser_wait->revents) {
      magic = ser->ser_root ? su_root_magic(ser->ser_root) : NULL;
      ser->ser_cb(magic, ser->ser_wait, ser->ser_arg);
      events++;
      if (version != self->sup_registers)
	/* Callback function used su_register()/su_deregister() */
	return events;
      if (!self->sup_multishot)
	/* Callback function used su_register()/su_deregister() */
	return events;
    }

    if (n == 0)
      break;
612
  }
613 614 615 616 617 618 619

  assert(n == 0);

  return events;
}

/** Create a port using epoll() or poll().
620 621 622
 */
su_port_t *su_select_port_create(void)
{
623
  su_port_t *self;
624

625
  self = su_home_new(sizeof *self);
626
  if (!self)
627 628 629
    return NULL;

  if (su_home_destructor(su_port_home(self), su_select_port_deinit) < 0 ||
630
      !(self->sup_indices =
631
	su_zalloc(su_port_home(self),
632
		  (sizeof self->sup_indices[0]) *
633 634 635 636
		  (self->sup_size_indices = __NFDBITS)))) {
    su_home_unref(su_port_home(self));
    return NULL;
  }
637 638 639

  self->sup_multishot = SU_ENABLE_MULTISHOT_POLL;

640
  if (su_socket_port_init(self->sup_base, su_select_port_vtable) < 0)
641 642 643 644 645 646
    return su_home_unref(su_port_home(self)), NULL;

  return self;
}

int su_select_clone_start(su_root_t *parent,
647 648 649 650
			 su_clone_r return_clone,
			 su_root_magic_t *magic,
			 su_root_init_f init,
			 su_root_deinit_f deinit)
651
{
652
  return su_pthreaded_port_start(su_select_port_create,
653 654 655
				 parent, return_clone, magic, init, deinit);
}

656

657
#endif