/src/opensips/io_wait.h

Source (jump to first uncovered line)
/*
 * Copyright (C) 2014-2015 OpenSIPS Solutions
 * Copyright (C) 2005 iptelorg GmbH
 *
 * This file is part of opensips, a free SIP server.
 *
 * opensips is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version
 *
 * opensips 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 General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301  USA
 *
 * History:
 * --------
 *  2005-06-13  created by andrei
 *  2005-06-26  added kqueue (andrei)
 *  2005-07-01  added /dev/poll (andrei)
 *  2014-08-25  looping functions moved to io_wait_loop.h (bogdan)
 */

/*!
 * \file
 * \brief tcp io wait common stuff used by tcp_main.c & tcp_read.c
 * - \ref TCPiowait
 */

/*! \page TCPiowait TCP io wait common stuff used by tcp_main.c & tcp_read.c
 * All the functions are inline because of speed reasons and because they are
 * used only from 2 places.
 * You also have to define:
 *   -  int handle_io(struct fd_map* fm, int idx) (see below)
 *     (this could be trivially replaced by a callback pointer entry attached
 *      to the io_wait handler if more flexibility rather then performance
 *      is needed)
 *   -   fd_type - define to some enum of you choice and define also
 *                FD_TYPE_DEFINED (if you don't do it fd_type will be defined
 *                to int). 0 has a special not set/not init. meaning
 *                (a lot of sanity checks and the sigio_rt code are based on
 *                 this assumption)
 *   -  local_malloc (defaults to pkg_malloc)
 *   -  local_free   (defaults to pkg_free)
 *
 */


#ifndef _io_wait_h
#define _io_wait_h

#include <errno.h>
#include <string.h>
#ifdef HAVE_SIGIO_RT
#define __USE_GNU /* or else F_SETSIG won't be included */
#define _GNU_SOURCE /* define this as well */
#include <sys/types.h> /* recv */
#include <sys/socket.h> /* recv */
#include <signal.h> /* sigprocmask, sigwait a.s.o */
#endif
#ifdef HAVE_EPOLL
#include <sys/epoll.h>
#endif
#ifdef HAVE_KQUEUE
#include <sys/types.h> /* needed on freebsd */
#include <sys/event.h>
#include <sys/time.h>
#endif
#ifdef HAVE_DEVPOLL
#include <sys/devpoll.h>
#endif
#ifdef HAVE_SELECT
/* needed on openbsd for select*/
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
/* needed according to POSIX for select*/
#include <sys/select.h>
#endif
#include <sys/poll.h>
#include <fcntl.h>

#include "dprint.h"

#include "poll_types.h" /* poll_types*/
#include "pt.h" /* mypid() */
#include "error.h"

#ifdef __OS_linux
#include <features.h>     /* for GLIBC version testing */
#endif
#include "lib/dbg/backtrace.h"

#ifndef FD_TYPE_DEFINED
typedef int fd_type;
#define FD_TYPE_DEFINED
#endif

/*! \brief maps a fd to some other structure; used in almost all cases
 * except epoll and maybe kqueue or /dev/poll */
struct fd_map {
  int fd;               /* fd no */
  fd_type type;         /* "data" type */
  void* data;           /* pointer to the corresponding structure */
  int flags;            /* so far used to indicate whether we should 
                         * read, write or both ; last 4 are reserved for 
                         * internal usage */
  int app_flags;        /* flags to be used by upper layer apps, not by
                         * the reactor */
  unsigned int timeout;
};


#ifdef HAVE_KQUEUE
#ifndef KQ_CHANGES_ARRAY_SIZE
#define KQ_CHANGES_ARRAY_SIZE 128

#ifdef __OS_netbsd
#define KEV_UDATA_CAST (intptr_t)
#else
#define KEV_UDATA_CAST
#endif

#endif
#endif


#define IO_FD_CLOSING 16

/*! \brief handler structure */
struct io_wait_handler{
  char *name;
  int max_prio;
#ifdef HAVE_EPOLL
  struct epoll_event* ep_array;
  int epfd; /* epoll ctrl fd */
#endif
#ifdef HAVE_SIGIO_RT
  sigset_t sset; /* signal mask for sigio & sigrtmin */
  int signo;     /* real time signal used */
#endif
#ifdef HAVE_KQUEUE
  struct kevent* kq_array;   /* used for the eventlist*/
  struct kevent* kq_changes; /* used for the changelist */
  size_t kq_nchanges;
  size_t kq_changes_size; /* size of the changes array */
  int kq_fd;
#endif
#ifdef HAVE_DEVPOLL
  int dpoll_fd;
  struct pollfd* dp_changes;
#endif
#ifdef HAVE_SELECT
  fd_set master_set;
  int max_fd_select; /* maximum select used fd */
#endif
  /* common stuff for POLL, SIGIO_RT and SELECT
   * since poll support is always compiled => this will always be compiled */
  int *prio_idx; /* size of max_prio - idxs in fd_array where prio changes*/
  struct fd_map* fd_hash;
  struct pollfd* fd_array;
  int fd_no; /*  current index used in fd_array */
  int max_fd_no; /* maximum fd no, is also the size of fd_array,
                   fd_hash  and ep_array*/
  enum poll_types poll_method;
  int flags;
};

typedef struct io_wait_handler io_wait_h;


/*! \brief get the corresponding fd_map structure pointer */
#define get_fd_map(h, fd)   (&(h)->fd_hash[(fd)])

/*! \brief remove a fd_map structure from the hash;
 * the pointer must be returned by get_fd_map or hash_fd_map
 */
#define unhash_fd_map(pfm,c_flags,sock_flags,erase) \
  do{ \
    if ((c_flags & IO_FD_CLOSING) || (pfm->flags&IO_WATCH_PRV_FILTER)==sock_flags) { \
      (pfm)->type=0 /*F_NONE */; \
      (pfm)->fd=-1; \
      (pfm)->flags = 0; \
      (pfm)->data = NULL; \
      erase = 1; \
    } else { \
      (pfm)->flags &= ~sock_flags; \
      erase = 0; \
    } \
  }while(0)

#define unhash_fd_map2(pfm,c_flags,sock_flags,erase)  \
  do{ \
    if ((c_flags & IO_FD_CLOSING) || (pfm->flags&IO_WATCH_PRV_FILTER)==sock_flags) { \
      rla_log("erase case detected with sflags=%x\n",sock_flags); \
      (pfm)->type=0 /*F_NONE */; \
      (pfm)->fd=-1; \
      (pfm)->flags = 0; \
      (pfm)->data = NULL; \
      erase = 1; \
    } else { \
      rla_log("not erasing flags=%x, flags=%x\n",(pfm)->flags,sock_flags); \
      (pfm)->flags &= ~sock_flags; \
      erase = 0; \
    } \
  }while(0)


/*! \brief add a fd_map structure to the fd hash */
static inline struct fd_map* hash_fd_map( io_wait_h* h,
            int fd,
            fd_type type,
            void* data,
            int flags,
            unsigned int timeout,
            int *already)
{
  if (h->fd_hash[fd].fd <= 0) {
    *already = 0;
  } else {
    *already = 1;
  }

  h->fd_hash[fd].fd=fd;
  h->fd_hash[fd].type=type;
  h->fd_hash[fd].data=data;

  h->fd_hash[fd].flags|=flags;

  h->fd_hash[fd].timeout = timeout;

  return &h->fd_hash[fd];
}


#ifdef HAVE_KQUEUE
/*
 * kqueue specific function: register a change
 * (adds a change to the kevent change array, and if full flushes it first)
 * returns: -1 on error, 0 on success
 */
static inline int kq_ev_change(io_wait_h* h, int fd, int filter, int flag,
                void* data)
{
  int n;
  struct timespec tspec;

  if (h->kq_nchanges>=h->kq_changes_size){
    /* changes array full ! */
    LM_WARN("[%s] kqueue changes array full trying to flush...\n",
      h->name);
    tspec.tv_sec=0;
    tspec.tv_nsec=0;
again:
    n=kevent(h->kq_fd, h->kq_changes, h->kq_nchanges, 0, 0, &tspec);
    if (n==-1){
      if (errno==EINTR) goto again;
      LM_ERR("[%s] kevent flush changes failed: %s [%d]\n",
        h->name, strerror(errno), errno);
      return -1;
    }
    h->kq_nchanges=0; /* changes array is empty */
  }
  EV_SET(&h->kq_changes[h->kq_nchanges], fd, filter, flag, 0, 0,
      KEV_UDATA_CAST data);
  h->kq_nchanges++;
  return 0;
}
#endif


#define IO_WATCH_READ            (1<<0)
#define IO_WATCH_WRITE           (1<<1)
#define IO_WATCH_ERROR           (1<<2)
#define IO_WATCH_TIMEOUT         (1<<3)
/* 24 starting are reserved, do not attempt to use */
#define IO_WATCH_PRV_FILTER      ((1<<24)-1)
#define IO_WATCH_PRV_CHECKED     (1<<29)
#define IO_WATCH_PRV_TRIG_READ   (1<<30)
#define IO_WATCH_PRV_TRIG_WRITE  (1<<31)

#define fd_array_print \
  do { \
    int k;\
    LM_DBG("[%s] size=%d, fd array is",h->name,h->fd_no);\
    for(k=0;k<h->fd_no;k++) LM_GEN1(L_DBG," %d flags = %d",h->fd_array[k].fd,h->fd_hash[h->fd_array[k].fd].flags);\
    LM_GEN1(L_DBG,"\n"); \
    LM_DBG("[%s] size=%d, prio array is",h->name,h->max_prio);\
    for(k=0;k<h->max_prio;k++) LM_GEN1(L_DBG," %d",h->prio_idx[k]);\
    LM_GEN1(L_DBG,"\n"); \
  }while(0)


#define check_io_data() \
  do { \
    struct fd_map* _e;\
    int _t,k;\
    check_error = 0;\
    /* iterate the fd_array and check if fd_hash is properly set for each */ \
    for(k=0;k<h->fd_no;k++) {\
      _e = get_fd_map(h, h->fd_array[k].fd); \
      if (_e->type==0 || _e->fd<=0 || \
      (_e->flags&(IO_WATCH_READ|IO_WATCH_WRITE))==0 ) {\
        LM_BUG("fd_array idx %d (fd=%d) points to bogus map "\
          "(fd=%d,type=%d,flags=%x,data=%p)\n",k,h->fd_array[k].fd,\
          _e->fd, _e->type, _e->flags, _e->data);\
          check_error = 1;\
      }\
      _e->flags |= IO_WATCH_PRV_CHECKED;\
    }\
    /* iterate the fd_map and see if all records are checked */ \
    _t = 0; \
    for(k=0;k<h->max_fd_no;k++) {\
      _e = get_fd_map(h, k); \
      if (_e->type==0) { \
        /* fd not in used, everything should be on zero */ \
        if (_e->fd>0 || _e->data!=NULL || _e->flags!=0 ) {\
          LM_BUG("unused fd_map fd=%d has bogus data "\
          "(fd=%d,flags=%x,data=%p)\n",k,\
          _e->fd, _e->flags, _e->data);\
          check_error = 1;\
        }\
      } else {\
        /* fd in used, check if in checked */ \
        if (_e->fd<=0 || \
        (_e->flags&(IO_WATCH_READ|IO_WATCH_WRITE))==0 ) {\
        LM_BUG("used fd map fd=%d has bogus data "\
          "(fd=%d,type=%d,flags=%x,data=%p)\n",k,\
          _e->fd, _e->type, _e->flags, _e->data);\
          check_error = 1;\
        }\
        /* the map is valid */ \
        if ((_e->flags&IO_WATCH_PRV_CHECKED)==0) {\
          LM_BUG("used fd map fd=%d is not present in fd_array "\
            "(fd=%d,type=%d,flags=%x,data=%p)\n",k,\
            _e->fd, _e->type, _e->flags, _e->data);\
            check_error = 1;\
        }\
        _e->flags &= ~IO_WATCH_PRV_CHECKED;\
        _t++;\
      }\
    }\
    if (_t!=h->fd_no) { \
      LM_BUG("fd_map versus fd_array size mismatch: %d versus %d\n",\
        _t, h->fd_no);\
      check_error = 1;\
    }\
  } while(0)


/*! \brief generic io_watch_add function
 * \return 0 on success, -1 on error
 *
 * this version should be faster than pointers to poll_method specific
 * functions (it avoids functions calls, the overhead being only an extra
 *  switch())
*/
inline static int io_watch_add( io_wait_h* h, // lgtm [cpp/use-of-goto]
                int fd,
                fd_type type,
                void* data,
                int prio,
                unsigned int timeout,
                int flags)
{

  /* helper macros */
#define fd_array_setup \
  do{ \
    n = h->prio_idx[prio]; \
    if (n<h->fd_no)\
      memmove( &h->fd_array[n+1], &h->fd_array[n],\
        (h->fd_no-n)*sizeof(*(h->fd_array)) ); \
    h->fd_array[n].fd=fd; \
    h->fd_array[n].events=0; \
    if (flags & IO_WATCH_READ) \
      h->fd_array[n].events|=POLLIN; /* useless for select */ \
    if (flags & IO_WATCH_WRITE) \
      h->fd_array[n].events|=POLLOUT; /* useless for select */ \
    h->fd_array[n].revents=0;     /* useless for select */ \
    for( n=prio ; n<h->max_prio ; n++) \
      h->prio_idx[n]++; \
    h->fd_no++; \
  }while(0)

#define set_fd_flags(f) \
  do{ \
      ctl_flags=fcntl(fd, F_GETFL); \
      if (ctl_flags==-1){ \
        LM_ERR("[%s] fcntl: GETFL failed:" \
          " %s [%d]\n", h->name, strerror(errno), errno); \
        goto error; \
      } \
      if (fcntl(fd, F_SETFL, ctl_flags|(f))==-1){ \
        LM_ERR("[%s] fcntl: SETFL" \
          " failed: %s [%d]\n", h->name, strerror(errno), errno);\
        goto error; \
      } \
  }while(0)


  struct fd_map* e;
  int already=-1;
#ifdef HAVE_EPOLL
  struct epoll_event ep_event;
#endif
#ifdef HAVE_DEVPOLL
  struct pollfd pfd;
#endif
  int ctl_flags;
  int n;  //FIXME
  int check_error;
#if 0 //defined(HAVE_SIGIO_RT) || defined (HAVE_EPOLL) FIXME
  int n;
  int idx;
  int check_io;
  struct pollfd pf;

  check_io=0; /* set to 1 if we need to check for pre-existing queued
           io/data on the fd */
  idx=-1;
#endif
  e=0;

  if (fd==-1){
    LM_CRIT("fd is -1!\n");
    goto error0;
  }
  /* check if not too big */
  if (h->fd_no >= h->max_fd_no || fd >= h->max_fd_no) {
    LM_CRIT("[%s] maximum fd number exceeded: %d, %d/%d\n",
      h->name, fd, h->fd_no, h->max_fd_no);
    goto error0;
  }
  if (prio > h->max_prio) {
    LM_BUG("[%s] priority %d requested (max is %d)\n",
      h->name, prio, h->max_prio);
    goto error0;
  }
#if defined (HAVE_EPOLL)
  LM_DBG("[%s] io_watch_add op (%d on %d) (%p, %d, %d, %p,%d), fd_no=%d/%d\n",
      h->name,fd,h->epfd, h,fd,type,data,flags,h->fd_no,h->max_fd_no);
#else
  LM_DBG("[%s] io_watch_add op (%d) (%p, %d, %d, %p,%d), fd_no=%d/%d\n",
      h->name,fd, h,fd,type,data,flags,h->fd_no,h->max_fd_no);
#endif
  //fd_array_print;
  /*  hash sanity check */
  e=get_fd_map(h, fd);

  check_io_data();
  if (check_error) {
    LM_CRIT("[%s] check failed before fd add "
      "(fd=%d,type=%d,data=%p,flags=%x) already=%d\n",h->name,
      fd, type, data, flags, already);
  }

  if (e->flags & flags){
    if (e->data != data) {
      LM_BUG("[%s] BUG trying to overwrite entry %d"
          " in the hash(%d, %d, %p,%d) with (%d, %d, %p,%d)\n",
          h->name,fd, e->fd, e->type, e->data,e->flags, fd, type, data,flags);
      goto error0;
    }
    LM_DBG("[%s] Socket %d is already being listened on for flags %d\n",
         h->name,fd,flags);
    return 0;
  }

  if (timeout)
    timeout+=get_ticks();

  if ((e=hash_fd_map(h, fd, type, data,flags, timeout, &already))==0){
    LM_ERR("[%s] failed to hash the fd %d\n",h->name, fd);
    goto error0;
  }
  switch(h->poll_method){ /* faster then pointer to functions */
    case POLL_POLL:
      set_fd_flags(O_NONBLOCK);
      break;
#ifdef HAVE_SELECT
    case POLL_SELECT:
      FD_SET(fd, &h->master_set);
      if (h->max_fd_select<fd) h->max_fd_select=fd;
      break;
#endif
#ifdef HAVE_SIGIO_RT
    case POLL_SIGIO_RT:
      /* re-set O_ASYNC might be needed, if not done from
       * io_watch_del (or if somebody wants to add a fd which has
       * already O_ASYNC/F_SETSIG set on a dupplicate)
       */
      /* set async & signal */
      if (fcntl(fd, F_SETOWN, my_pid())==-1){
        LM_ERR("[%s] fcntl: SETOWN"
        " failed: %s [%d]\n",h->name, strerror(errno), errno);
        goto error;
      }
      if (fcntl(fd, F_SETSIG, h->signo)==-1){
        LM_ERR("[%s] fcntl: SETSIG"
          " failed: %s [%d]\n",h->name, strerror(errno), errno);
        goto error;
      }
      /* set both non-blocking and async */
      set_fd_flags(O_ASYNC| O_NONBLOCK);
#ifdef EXTRA_DEBUG
      LM_DBG("[%s] sigio_rt on f %d, signal %d to pid %d\n",
          h->name,fd,  h->signo, my_pid());
#endif
      /* empty socket receive buffer, if buffer is already full
       * no more space to put packets
       * => no more signals are ever generated
       * also when moving fds, the freshly moved fd might have
       *  already some bytes queued, we want to get them now
       *  and not later -- andrei */
      //idx=h->fd_no;  FIXME
      //check_io=1;
      break;
#endif
#ifdef HAVE_EPOLL
    case POLL_EPOLL:
      ep_event.data.ptr=e;
      ep_event.events=0;
      if (e->flags & IO_WATCH_READ)
        ep_event.events|=EPOLLIN;
      if (e->flags & IO_WATCH_WRITE)
        ep_event.events|=EPOLLOUT;
      if (!already) {
again1:
#if 0
/* This is currently broken, because when using EPOLLEXCLUSIVE, the OS will
 * send sequential events to the same process - thus our pseudo-dispatcher
 * will no longer work, since events on a pipe will be queued by a single
 * process. - razvanc
 */
#if (defined __OS_linux) && (__GLIBC__ >= 2) && (__GLIBC_MINOR__ >= 24)
        if (e->flags & IO_WATCH_READ)
          ep_event.events|=EPOLLEXCLUSIVE;
#endif
#endif
        n=epoll_ctl(h->epfd, EPOLL_CTL_ADD, fd, &ep_event);
        if (n==-1){
          if (errno==EAGAIN) goto again1;
          LM_ERR("[%s] epoll_ctl ADD failed: %s [%d]\n",
            h->name,strerror(errno), errno);
          goto error;
        }
      } else {
again11:
        n=epoll_ctl(h->epfd, EPOLL_CTL_MOD, fd, &ep_event);
        if (n==-1){
          if (errno==EAGAIN) goto again11;
          LM_ERR("[%s] epoll_ctl MOD failed: %s [%d]\n",
            h->name,strerror(errno), errno);
          goto error;
        }
      }
      break;
#endif
#ifdef HAVE_KQUEUE
    case POLL_KQUEUE:
      if (kq_ev_change(h, fd, EVFILT_READ, EV_ADD, e)==-1)
        goto error;
      break;
#endif
#ifdef HAVE_DEVPOLL
    case POLL_DEVPOLL:
      pfd.fd=fd;
      pfd.events=POLLIN;
      pfd.revents=0;
again_devpoll:
      if (write(h->dpoll_fd, &pfd, sizeof(pfd))==-1){
        if (errno==EAGAIN) goto again_devpoll;
        LM_ERR("[%s] /dev/poll write failed:"
          "%s [%d]\n",h->name, strerror(errno), errno);
        goto error;
      }
      break;
#endif

    default:
      LM_CRIT("[%s] no support for poll method "
        " %s (%d)\n",h->name, poll_method_str[h->poll_method],
        h->poll_method);
      goto error;
  }

  if (!already) {
    fd_array_setup;
  }

#if 0 //defined(HAVE_SIGIO_RT) || defined (HAVE_EPOLL) FIXME !!!
  if (check_io){
    /* handle possible pre-existing events */
    pf.fd=fd;
    pf.events=POLLIN;
check_io_again:
    while( ((n=poll(&pf, 1, 0))>0) && (handle_io(e, idx,IO_WATCH_READ)>0));
    if (n==-1){
      if (errno==EINTR) goto check_io_again;
      LM_ERR("check_io poll: %s [%d]\n",
            strerror(errno), errno);
    }
  }
#endif
  //fd_array_print;
  check_io_data();
  if (check_error) {
    LM_CRIT("[%s] check failed after successful fd add "
      "(fd=%d,type=%d,data=%p,flags=%x) already=%d\n",h->name,
      fd, type, data, flags, already);
  }
  return 0;
error:
  if (e) unhash_fd_map(e,0,flags,already);
error0:
  check_io_data();
  if (check_error) {
    LM_CRIT("[%s] check failed after failed fd add "
      "(fd=%d,type=%d,data=%p,flags=%x) already=%d\n",h->name,
      fd, type, data, flags, already);
  }
  return -1;
#undef fd_array_setup
#undef set_fd_flags
}



/*!
 * \brief
 * \param h handler
 * \param fd file descriptor
 * \param idx index in the fd_array if known, -1 if not
 *                    (if index==-1 fd_array will be searched for the
 *                     corresponding fd* entry -- slower but unavoidable in
 *                     some cases). index is not used (no fd_array) for epoll,
 *                     /dev/poll and kqueue
 * \param flags optimization flags, e.g. IO_FD_CLOSING, the fd was or will
 *                    shortly be closed, in some cases we can avoid extra
 *                    remove operations (e.g.: epoll, kqueue, sigio)
 * \return 0 if ok, -1 on error
 */
inline static int io_watch_del(io_wait_h* h, int fd, int idx,
          int flags,int sock_flags)
{
#define fix_fd_array \
  do{\
      if (idx==-1){ \
        /* fix idx if -1 and needed */ \
        for (idx=0; (idx<h->fd_no) && \
              (h->fd_array[idx].fd!=fd); idx++); \
      } \
      rla_log("fixing: final idx=%d out of %d, erase=%d\n",idx,idx<h->fd_no,erase); \
      if (idx<h->fd_no){ \
        if (erase) { \
          memmove(&h->fd_array[idx], &h->fd_array[idx+1], \
            (h->fd_no-(idx+1))*sizeof(*(h->fd_array))); \
          for( i=0 ; i<h->max_prio && h->prio_idx[i]<=idx ; i++ ); \
          for( ; i<h->max_prio ; i++ ) h->prio_idx[i]-- ; \
          h->fd_no--; \
        } else { \
          h->fd_array[idx].events = 0; \
          if (e->flags & IO_WATCH_READ) \
            h->fd_array[idx].events|=POLLIN; /* useless for select */ \
          if (flags & IO_WATCH_WRITE) \
            h->fd_array[idx].events|=POLLOUT; /* useless for select */ \
          h->fd_array[idx].revents = 0; \
        } \
      } \
  }while(0)

  struct fd_map* e;
#ifdef HAVE_EPOLL
  int n;
  struct epoll_event ep_event;
#endif
#ifdef HAVE_DEVPOLL
  struct pollfd pfd;
#endif
#ifdef HAVE_SIGIO_RT
  int fd_flags;
#endif
  int erase = 0;
  int check_error;
  int i;

  #define x_RL_NO  10
  #define x_RL_LEN 200
  static char rla[x_RL_NO][x_RL_LEN];
  int rla_idx=0;
  #define rla_log( _fmt, args...) \
    snprintf( rla[rla_idx++], x_RL_LEN, _fmt, ## args)
  #define rla_dump() \
    do { \
      int w; \
      for(w=0;w<rla_idx;w++) \
        LM_CRIT("[%d]-> [%s]",w,rla[w]); \
    } while(0)

  if ((fd<0) || (fd>=h->max_fd_no)){
    LM_CRIT("[%s] invalid fd %d, not in [0, %d)\n", h->name, fd, h->fd_no);
    goto error0;
  }
  LM_DBG("[%s] io_watch_del op on index %d %d (%p, %d, %d, 0x%x,0x%x) "
    "fd_no=%d called\n", h->name,idx,fd, h, fd, idx, flags,
    sock_flags,h->fd_no);
  rla_log("[%s] io_watch_del op on index %d %d (%p, %d, %d, 0x%x,0x%x) "
    "fd_no=%d called\n", h->name,idx,fd, h, fd, idx, flags,
    sock_flags,h->fd_no);
  //fd_array_print;

  e=get_fd_map(h, fd);
  /* more sanity checks */
  if (e==0){
    LM_CRIT("[%s] no corresponding hash entry for %d\n",h->name, fd);
    goto error0;
  }
  if (e->type==0 /*F_NONE*/){
    LM_ERR("[%s] trying to delete already erased"
        " entry %d in the hash(%d, %d, %p) )\n",
        h->name,fd, e->fd, e->type, e->data);
    goto error0;
  }

  if (idx != -1) {
    if (!(idx>=0 && idx<h->fd_no)) {
      LM_CRIT("[%s] FD index check failed, idx=%d, max=%d"
        " operating on %d\n",h->name, idx, h->fd_no, fd );
      log_backtrace();
      rla_dump();
      idx = -1;
    } else if (h->fd_array[idx].fd!=fd) {
      LM_CRIT("[%s] FD consistency check failed, idx=%d points to fd=%d,"
        " but operating on %d\n",h->name, idx, h->fd_array[idx].fd, fd );
      log_backtrace();
      rla_dump();
      idx = -1;
    }
  }

  if ((e->flags & sock_flags) == 0) {
    LM_ERR("BUG - [%s] trying to del fd %d with flags %d %d\n",
      h->name, fd, e->flags,sock_flags);
    goto error0;
  }

  unhash_fd_map2(e,flags,sock_flags,erase);

  switch(h->poll_method){
    case POLL_POLL:
      break;
#ifdef HAVE_SELECT
    case POLL_SELECT:
      FD_CLR(fd, &h->master_set);
      if (h->max_fd_select && (h->max_fd_select==fd))
        /* we don't know the prev. max, so we just decrement it */
        h->max_fd_select--;
      break;
#endif
#ifdef HAVE_SIGIO_RT
    case POLL_SIGIO_RT:
      /* the O_ASYNC flag must be reset all the time, the fd
       *  can be changed only if  O_ASYNC is reset (if not and
       *  the fd is a duplicate, you will get signals from the dup. fd
       *  and not from the original, even if the dup. fd was closed
       *  and the signals re-set on the original) -- andrei
       */
      /*if (!(flags & IO_FD_CLOSING)){*/
        /* reset ASYNC */
        fd_flags=fcntl(fd, F_GETFL);
        if (fd_flags==-1){
          LM_ERR("[%s] fcntl: GETFL failed:"
            " %s [%d]\n",h->name, strerror(errno), errno);
          goto error;
        }
        if (fcntl(fd, F_SETFL, fd_flags&(~O_ASYNC))==-1){
          LM_ERR("[%s] fcntl: SETFL"
            " failed: %s [%d]\n",h->name, strerror(errno), errno);
          goto error;
        }
      break;
#endif
#ifdef HAVE_EPOLL
    case POLL_EPOLL:
      /* epoll doesn't seem to automatically remove sockets,
       * if the socket is a dupplicate/moved and the original
       * is still open. The fd is removed from the epoll set
       * only when the original (and all the  copies?) is/are
       * closed. This is probably a bug in epoll. --andrei */
#ifdef EPOLL_NO_CLOSE_BUG
      if (!(flags & IO_FD_CLOSING)){
#endif
        if (erase) {
          n=epoll_ctl(h->epfd, EPOLL_CTL_DEL, fd, &ep_event);
          /*
           * in some cases (fds managed by external libraries),
           * the fd may have already been closed
           */
          if (n==-1 && errno != EBADF && errno != ENOENT) {
            LM_ERR("[%s] removing fd from epoll (%d from %d) "
              "list failed: %s [%d]\n",h->name, fd, h->epfd,
              strerror(errno), errno);
            goto error;
          }
        } else {
          ep_event.data.ptr=e;
          ep_event.events=0;
          if (e->flags & IO_WATCH_READ)
            ep_event.events|=EPOLLIN;
          if (e->flags & IO_WATCH_WRITE)
            ep_event.events|=EPOLLOUT;
          n=epoll_ctl(h->epfd, EPOLL_CTL_MOD, fd, &ep_event);
          if (n==-1){
            LM_ERR("[%s] epoll_ctl failed: %s [%d]\n",
              h->name,strerror(errno), errno);
            goto error;
          }
        }
#ifdef EPOLL_NO_CLOSE_BUG
      }
#endif
      break;
#endif
#ifdef HAVE_KQUEUE
    case POLL_KQUEUE:
      if (!(flags & IO_FD_CLOSING)){
        if (kq_ev_change(h, fd, EVFILT_READ, EV_DELETE, 0)==-1)
          goto error;
      }
      break;
#endif
#ifdef HAVE_DEVPOLL
    case POLL_DEVPOLL:
        /* for /dev/poll the closed fds _must_ be removed
           (they are not removed automatically on close()) */
        pfd.fd=fd;
        pfd.events=POLLREMOVE;
        pfd.revents=0;
again_devpoll:
        if (write(h->dpoll_fd, &pfd, sizeof(pfd))==-1){
          if (errno==EINTR) goto again_devpoll;
          LM_ERR("[%s] removing fd from /dev/poll failed: "
            "%s [%d]\n",h->name, strerror(errno), errno);
          goto error;
        }
        break;
#endif
    default:
      LM_CRIT("[%s] no support for poll method %s (%d)\n",
        h->name,poll_method_str[h->poll_method], h->poll_method);
      goto error;
  }
  rla_log("fixing fd array, idx=%d\n",idx); \

  fix_fd_array;
  //fd_array_print;

  check_io_data();
  if (check_error) {
    LM_CRIT("[%s] check failed after successful fd del "
      "(fd=%d,flags=%d, sflags=%d) over map "
      "(fd=%d,type=%d,data=%p,flags=%d) erase=%d\n",h->name,
      fd, flags, sock_flags,
      e->fd, e->type, e->data, e->flags,
      erase);
    rla_dump();
  }

  return 0;
error:
  /*
   * although the DEL operation failed, both
   * "fd_hash" and "fd_array" must remain consistent
   */
  fix_fd_array;

  check_io_data();
  if (check_error) {
    LM_CRIT("[%s] check failed after failed fd del "
      "(fd=%d,flags=%d, sflags=%d) over map "
      "(fd=%d,type=%d,data=%p,flags=%d) erase=%d\n",h->name,
      fd, flags, sock_flags,
      e->fd, e->type, e->data, e->flags,
      erase);
    rla_dump();
  }
error0:

  return -1;
#undef fix_fd_array
}


/* init */


/*! \brief initializes the static vars/arrays
 * \param h pointer to the io_wait_h that will be initialized
 * \param max_fd maximum allowed fd number
 * \param poll_method poll method (0 for automatic best fit)
 */
int init_io_wait(io_wait_h* h, char *name, int max_fd,
                enum poll_types poll_method, int max_prio);

/*! \brief destroys everything init_io_wait allocated */
void destroy_io_wait(io_wait_h* h);

int io_set_app_flag( io_wait_h *h , int type, int app_flag);

int io_check_app_flag( io_wait_h *h , int app_flag);


#endif

Coverage Report

Created: 2025-04-11 06:59