Diff for /libaitsched/src/tasks.c between versions 1.13 and 1.30.6.2

version 1.13, 2012/08/21 12:54:39 version 1.30.6.2, 2023/02/24 16:21:23
Line 12  terms: Line 12  terms:
 All of the documentation and software included in the ELWIX and AITNET  All of the documentation and software included in the ELWIX and AITNET
 Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>  Releases is copyrighted by ELWIX - Sofia/Bulgaria <info@elwix.org>
   
Copyright 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012Copyright 2004 - 2023
         by Michael Pounov <misho@elwix.org>.  All rights reserved.          by Michael Pounov <misho@elwix.org>.  All rights reserved.
   
 Redistribution and use in source and binary forms, with or without  Redistribution and use in source and binary forms, with or without
Line 52  SUCH DAMAGE. Line 52  SUCH DAMAGE.
  * @root = root task   * @root = root task
  * return: NULL error or !=NULL prepared task   * return: NULL error or !=NULL prepared task
  */   */
inline sched_task_t *sched_task_t *
 sched_useTask(sched_root_task_t * __restrict root)  sched_useTask(sched_root_task_t * __restrict root)
 {  {
         sched_task_t *task, *tmp;          sched_task_t *task, *tmp;
   
           SCHED_QLOCK(root, taskUNUSE);
         TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {          TAILQ_FOREACH_SAFE(task, &root->root_unuse, task_node, tmp) {
                 if (!TASK_ISLOCKED(task)) {                  if (!TASK_ISLOCKED(task)) {
 #ifdef HAVE_LIBPTHREAD  
                         pthread_mutex_lock(&root->root_mtx[taskUNUSE]);  
 #endif  
                         TAILQ_REMOVE(&root->root_unuse, task, task_node);                          TAILQ_REMOVE(&root->root_unuse, task, task_node);
 #ifdef HAVE_LIBPTHREAD  
                         pthread_mutex_unlock(&root->root_mtx[taskUNUSE]);  
 #endif  
                         break;                          break;
                 }                  }
         }          }
           SCHED_QUNLOCK(root, taskUNUSE);
   
         if (!task) {          if (!task) {
                task = malloc(sizeof(sched_task_t));                task = e_malloc(sizeof(sched_task_t));
                 if (!task) {                  if (!task) {
                         LOGERR;                          LOGERR;
                         return NULL;                          return NULL;
Line 89  sched_useTask(sched_root_task_t * __restrict root) Line 85  sched_useTask(sched_root_task_t * __restrict root)
  * @task = task   * @task = task
  * return: always is NULL   * return: always is NULL
  */   */
inline sched_task_t *sched_task_t *
 sched_unuseTask(sched_task_t * __restrict task)  sched_unuseTask(sched_task_t * __restrict task)
 {  {
         TASK_UNLOCK(task);          TASK_UNLOCK(task);
   
         TASK_TYPE(task) = taskUNUSE;          TASK_TYPE(task) = taskUNUSE;
#ifdef HAVE_LIBPTHREAD        insert_task_to(task, &(TASK_ROOT(task))->root_unuse);
        pthread_mutex_lock(&TASK_ROOT(task)->root_mtx[taskUNUSE]); 
#endif 
        TAILQ_INSERT_TAIL(&TASK_ROOT(task)->root_unuse, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
        pthread_mutex_unlock(&TASK_ROOT(task)->root_mtx[taskUNUSE]); 
#endif 
        task = NULL; 
   
           task = NULL;
         return task;          return task;
 }  }
   
   /*
    * sched_taskExit() - Exit routine for scheduler task, explicit required for thread tasks
    *
    * @task = current task
    * @retcode = return code
    * return: return code
    */
   void *
   sched_taskExit(sched_task_t *task, intptr_t retcode)
   {
           if (!task || !TASK_ROOT(task))
                   return (void*) -1;
   
           if (TASK_ROOT(task)->root_hooks.hook_exec.exit)
                   TASK_ROOT(task)->root_hooks.hook_exec.exit(task, (void*) retcode);
   
           TASK_ROOT(task)->root_ret = (void*) retcode;
           return (void*) retcode;
   }
   
   
 /*  /*
  * schedRead() - Add READ I/O task to scheduler queue   * schedRead() - Add READ I/O task to scheduler queue
  *   *
Line 122  sched_task_t * Line 133  sched_task_t *
 schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedRead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 void *opt_data, size_t opt_dlen)                  void *opt_data, size_t opt_dlen)
 {  {
           return schedReadExt(root, func, arg, fd, opt_data, opt_dlen, 0);
   }
   
   /*
    * schedReadExt() - Add READ I/O task to scheduler queue with custom event mask
    *
    * @root = root task
    * @func = task execution function
    * @arg = 1st func argument
    * @fd = fd handle
    * @opt_data = Optional data
    * @opt_dlen = Optional data length
    * @mask = Event mask
    * return: NULL error or !=NULL new queued task
    */
   sched_task_t *
   schedReadExt(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                   void *opt_data, size_t opt_dlen, u_long mask)
   {
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 132  schedRead(sched_root_task_t * __restrict root, sched_t Line 162  schedRead(sched_root_task_t * __restrict root, sched_t
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskREAD;          TASK_TYPE(task) = taskREAD;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 142  schedRead(sched_root_task_t * __restrict root, sched_t Line 172  schedRead(sched_root_task_t * __restrict root, sched_t
         TASK_DATA(task) = opt_data;          TASK_DATA(task) = opt_data;
         TASK_DATLEN(task) = opt_dlen;          TASK_DATLEN(task) = opt_dlen;
   
           TASK_HARG(task) = mask;
   
         if (root->root_hooks.hook_add.read)          if (root->root_hooks.hook_add.read)
                ptr = root->root_hooks.hook_add.read(task, NULL);                ptr = root->root_hooks.hook_add.read(task, 
                                 (void*) task->task_harg);
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_read);
                pthread_mutex_lock(&root->root_mtx[taskREAD]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_read, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskREAD]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
Line 176  sched_task_t * Line 203  sched_task_t *
 schedWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 void *opt_data, size_t opt_dlen)                  void *opt_data, size_t opt_dlen)
 {  {
           return schedWriteExt(root, func, arg, fd, opt_data, opt_dlen, 0);
   }
   
   /*
    * schedWriteExt() - Add WRITE I/O task to scheduler queue with custom event mask
    *
    * @root = root task
    * @func = task execution function
    * @arg = 1st func argument
    * @fd = fd handle
    * @opt_data = Optional data
    * @opt_dlen = Optional data length
    * @mask = Event mask
    * return: NULL error or !=NULL new queued task
    */
   sched_task_t *
   schedWriteExt(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                   void *opt_data, size_t opt_dlen, u_long mask)
   {
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 186  schedWrite(sched_root_task_t * __restrict root, sched_ Line 232  schedWrite(sched_root_task_t * __restrict root, sched_
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskWRITE;          TASK_TYPE(task) = taskWRITE;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 196  schedWrite(sched_root_task_t * __restrict root, sched_ Line 242  schedWrite(sched_root_task_t * __restrict root, sched_
         TASK_DATA(task) = opt_data;          TASK_DATA(task) = opt_data;
         TASK_DATLEN(task) = opt_dlen;          TASK_DATLEN(task) = opt_dlen;
   
           TASK_HARG(task) = mask;
   
         if (root->root_hooks.hook_add.write)          if (root->root_hooks.hook_add.write)
                ptr = root->root_hooks.hook_add.write(task, NULL);                ptr = root->root_hooks.hook_add.write(task, 
                                 (void*) task->task_harg);
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_write);
                pthread_mutex_lock(&root->root_mtx[taskWRITE]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_write, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskWRITE]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
Line 230  sched_task_t * Line 273  sched_task_t *
 schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedNode(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 void *opt_data, size_t opt_dlen)                  void *opt_data, size_t opt_dlen)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 240  schedNode(sched_root_task_t * __restrict root, sched_t Line 287  schedNode(sched_root_task_t * __restrict root, sched_t
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskNODE;          TASK_TYPE(task) = taskNODE;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 255  schedNode(sched_root_task_t * __restrict root, sched_t Line 302  schedNode(sched_root_task_t * __restrict root, sched_t
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_node);
                pthread_mutex_lock(&root->root_mtx[taskNODE]);        else
                 task = sched_unuseTask(task);
 
         return task;
 #endif  /* KQ_SUPPORT */
 }
 
 /*
  * schedNode2() - Add NODE task with all events to scheduler queue
  *
  * @root = root task
  * @func = task execution function
  * @arg = 1st func argument
  * @fd = fd handle
  * @opt_data = Optional data
  * @opt_dlen = Optional data length
  * return: NULL error or !=NULL new queued task
  */
 sched_task_t *
 schedNode2(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 void *opt_data, size_t opt_dlen)
 {
 #if SUP_ENABLE != KQ_SUPPORT
         sched_SetErr(ENOTSUP, "disabled kqueue support");
         return NULL;
 #else
         sched_task_t *task;
         void *ptr;
 
         if (!root || !func)
                 return NULL;
 
         /* get new task */
         if (!(task = sched_useTask(root)))
                 return NULL;
 
         TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskNODE;
         TASK_ROOT(task) = root;
 
         TASK_ARG(task) = arg;
         TASK_FD(task) = fd;
 
         TASK_DATA(task) = opt_data;
         TASK_DATLEN(task) = opt_dlen;
 
         if (root->root_hooks.hook_add.node)
 #ifdef __FreeBSD__
                 ptr = root->root_hooks.hook_add.node(task, 
                                 (void*) (NOTE_READ | NOTE_CLOSE_WRITE | NOTE_CLOSE | NOTE_OPEN));
 #else
                 ptr = root->root_hooks.hook_add.node(task, NULL);
 #endif  #endif
                TAILQ_INSERT_TAIL(&root->root_node, TASK_ID(task), task_node);        else
#ifdef HAVE_LIBPTHREAD                ptr = NULL;
                pthread_mutex_unlock(&root->root_mtx[taskNODE]);
#endif        if (!ptr)
        } else                insert_task_to(task, &root->root_node);
         else
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
   #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 284  sched_task_t * Line 384  sched_task_t *
 schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid,   schedProc(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long pid, 
                 void *opt_data, size_t opt_dlen)                  void *opt_data, size_t opt_dlen)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 294  schedProc(sched_root_task_t * __restrict root, sched_t Line 398  schedProc(sched_root_task_t * __restrict root, sched_t
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskPROC;          TASK_TYPE(task) = taskPROC;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 309  schedProc(sched_root_task_t * __restrict root, sched_t Line 413  schedProc(sched_root_task_t * __restrict root, sched_t
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_proc);
                pthread_mutex_lock(&root->root_mtx[taskPROC]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_proc, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskPROC]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
   #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 338  sched_task_t * Line 437  sched_task_t *
 schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id,   schedUser(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, u_long id, 
                 void *opt_data, size_t opt_dlen)                  void *opt_data, size_t opt_dlen)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
 #ifndef EVFILT_USER  #ifndef EVFILT_USER
         sched_SetErr(ENOTSUP, "Not supported kevent() filter");          sched_SetErr(ENOTSUP, "Not supported kevent() filter");
         return NULL;          return NULL;
Line 352  schedUser(sched_root_task_t * __restrict root, sched_t Line 455  schedUser(sched_root_task_t * __restrict root, sched_t
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskUSER;          TASK_TYPE(task) = taskUSER;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 367  schedUser(sched_root_task_t * __restrict root, sched_t Line 470  schedUser(sched_root_task_t * __restrict root, sched_t
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_user);
                pthread_mutex_lock(&root->root_mtx[taskUSER]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_user, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskUSER]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
#endif#endif  /* EVFILT_USER */
 #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 407  schedSignal(sched_root_task_t * __restrict root, sched Line 505  schedSignal(sched_root_task_t * __restrict root, sched
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskSIGNAL;          TASK_TYPE(task) = taskSIGNAL;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 422  schedSignal(sched_root_task_t * __restrict root, sched Line 520  schedSignal(sched_root_task_t * __restrict root, sched
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_signal);
                pthread_mutex_lock(&root->root_mtx[taskSIGNAL]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_signal, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskSIGNAL]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
Line 443  schedSignal(sched_root_task_t * __restrict root, sched Line 535  schedSignal(sched_root_task_t * __restrict root, sched
  * @func = task execution function   * @func = task execution function
  * @arg = 1st func argument   * @arg = 1st func argument
  * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!   * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!
 * @opt_data = Optional data * @opt_data = Alarm timer ID
  * @opt_dlen = Optional data length   * @opt_dlen = Optional data length
  * return: NULL error or !=NULL new queued task   * return: NULL error or !=NULL new queued task
  */   */
Line 451  sched_task_t * Line 543  sched_task_t *
 schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts,   schedAlarm(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
                 void *opt_data, size_t opt_dlen)                  void *opt_data, size_t opt_dlen)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 461  schedAlarm(sched_root_task_t * __restrict root, sched_ Line 557  schedAlarm(sched_root_task_t * __restrict root, sched_
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskALARM;          TASK_TYPE(task) = taskALARM;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 476  schedAlarm(sched_root_task_t * __restrict root, sched_ Line 572  schedAlarm(sched_root_task_t * __restrict root, sched_
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_alarm);
                pthread_mutex_lock(&root->root_mtx[taskALARM]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_alarm, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskALARM]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
   #endif  /* KQ_SUPPORT */
 }  }
   
 #ifdef AIO_SUPPORT  #ifdef AIO_SUPPORT
Line 506  sched_task_t * Line 597  sched_task_t *
 schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,   schedAIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, 
                 struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)                  struct aiocb * __restrict acb, void *opt_data, size_t opt_dlen)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 516  schedAIO(sched_root_task_t * __restrict root, sched_ta Line 611  schedAIO(sched_root_task_t * __restrict root, sched_ta
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskAIO;          TASK_TYPE(task) = taskAIO;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 531  schedAIO(sched_root_task_t * __restrict root, sched_ta Line 626  schedAIO(sched_root_task_t * __restrict root, sched_ta
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_aio);
                pthread_mutex_lock(&root->root_mtx[taskAIO]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_aio, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskAIO]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
   #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 557  schedAIO(sched_root_task_t * __restrict root, sched_ta Line 647  schedAIO(sched_root_task_t * __restrict root, sched_ta
  * @offset = Offset from start of file, if =-1 from current position   * @offset = Offset from start of file, if =-1 from current position
  * return: NULL error or !=NULL new queued task   * return: NULL error or !=NULL new queued task
  */   */
inline sched_task_t *sched_task_t *
 schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedAIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 void *buffer, size_t buflen, off_t offset)                  void *buffer, size_t buflen, off_t offset)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         struct aiocb *acb;          struct aiocb *acb;
         off_t off;          off_t off;
   
Line 576  schedAIORead(sched_root_task_t * __restrict root, sche Line 670  schedAIORead(sched_root_task_t * __restrict root, sche
         } else          } else
                 off = offset;                  off = offset;
   
        if (!(acb = malloc(sizeof(struct aiocb)))) {        if (!(acb = e_malloc(sizeof(struct aiocb)))) {
                 LOGERR;                  LOGERR;
                 return NULL;                  return NULL;
         } else          } else
Line 592  schedAIORead(sched_root_task_t * __restrict root, sche Line 686  schedAIORead(sched_root_task_t * __restrict root, sche
   
         if (aio_read(acb)) {          if (aio_read(acb)) {
                 LOGERR;                  LOGERR;
                free(acb);                e_free(acb);
                 return NULL;                  return NULL;
         }          }
   
         return schedAIO(root, func, arg, acb, buffer, buflen);          return schedAIO(root, func, arg, acb, buffer, buflen);
   #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 611  schedAIORead(sched_root_task_t * __restrict root, sche Line 706  schedAIORead(sched_root_task_t * __restrict root, sche
  * @offset = Offset from start of file, if =-1 from current position   * @offset = Offset from start of file, if =-1 from current position
  * return: NULL error or !=NULL new queued task   * return: NULL error or !=NULL new queued task
  */   */
inline sched_task_t *sched_task_t *
 schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedAIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 void *buffer, size_t buflen, off_t offset)                  void *buffer, size_t buflen, off_t offset)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         struct aiocb *acb;          struct aiocb *acb;
         off_t off;          off_t off;
   
Line 630  schedAIOWrite(sched_root_task_t * __restrict root, sch Line 729  schedAIOWrite(sched_root_task_t * __restrict root, sch
         } else          } else
                 off = offset;                  off = offset;
   
        if (!(acb = malloc(sizeof(struct aiocb)))) {        if (!(acb = e_malloc(sizeof(struct aiocb)))) {
                 LOGERR;                  LOGERR;
                 return NULL;                  return NULL;
         } else          } else
Line 646  schedAIOWrite(sched_root_task_t * __restrict root, sch Line 745  schedAIOWrite(sched_root_task_t * __restrict root, sch
   
         if (aio_write(acb)) {          if (aio_write(acb)) {
                 LOGERR;                  LOGERR;
                free(acb);                e_free(acb);
                 return NULL;                  return NULL;
         }          }
   
         return schedAIO(root, func, arg, acb, buffer, buflen);          return schedAIO(root, func, arg, acb, buffer, buflen);
   #endif  /* KQ_SUPPORT */
 }  }
   
 #ifdef EVFILT_LIO  #ifdef EVFILT_LIO
Line 669  sched_task_t * Line 769  sched_task_t *
 schedLIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg,   schedLIO(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, 
                 struct aiocb ** __restrict acbs, void *opt_data, size_t opt_dlen)                  struct aiocb ** __restrict acbs, void *opt_data, size_t opt_dlen)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         sched_task_t *task;          sched_task_t *task;
         void *ptr;          void *ptr;
   
Line 679  schedLIO(sched_root_task_t * __restrict root, sched_ta Line 783  schedLIO(sched_root_task_t * __restrict root, sched_ta
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskLIO;          TASK_TYPE(task) = taskLIO;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 694  schedLIO(sched_root_task_t * __restrict root, sched_ta Line 798  schedLIO(sched_root_task_t * __restrict root, sched_ta
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_lio);
                pthread_mutex_lock(&root->root_mtx[taskLIO]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_lio, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskLIO]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
   #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 724  sched_task_t * Line 823  sched_task_t *
 schedLIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedLIORead(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 struct iovec *bufs, size_t nbufs, off_t offset)                  struct iovec *bufs, size_t nbufs, off_t offset)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         struct sigevent sig;          struct sigevent sig;
         struct aiocb **acb;          struct aiocb **acb;
         off_t off;          off_t off;
Line 741  schedLIORead(sched_root_task_t * __restrict root, sche Line 844  schedLIORead(sched_root_task_t * __restrict root, sche
         } else          } else
                 off = offset;                  off = offset;
   
        if (!(acb = calloc(sizeof(void*), nbufs))) {        if (!(acb = e_calloc(sizeof(void*), nbufs))) {
                 LOGERR;                  LOGERR;
                 return NULL;                  return NULL;
         } else          } else
                 memset(acb, 0, sizeof(void*) * nbufs);                  memset(acb, 0, sizeof(void*) * nbufs);
         for (i = 0; i < nbufs; off += bufs[i++].iov_len) {          for (i = 0; i < nbufs; off += bufs[i++].iov_len) {
                acb[i] = malloc(sizeof(struct aiocb));                acb[i] = e_malloc(sizeof(struct aiocb));
                 if (!acb[i]) {                  if (!acb[i]) {
                         LOGERR;                          LOGERR;
                         for (i = 0; i < nbufs; i++)                          for (i = 0; i < nbufs; i++)
                                 if (acb[i])                                  if (acb[i])
                                        free(acb[i]);                                        e_free(acb[i]);
                        free(acb);                        e_free(acb);
                         return NULL;                          return NULL;
                 } else                  } else
                         memset(acb[i], 0, sizeof(struct aiocb));                          memset(acb[i], 0, sizeof(struct aiocb));
Line 772  schedLIORead(sched_root_task_t * __restrict root, sche Line 875  schedLIORead(sched_root_task_t * __restrict root, sche
                 LOGERR;                  LOGERR;
                 for (i = 0; i < nbufs; i++)                  for (i = 0; i < nbufs; i++)
                         if (acb[i])                          if (acb[i])
                                free(acb[i]);                                e_free(acb[i]);
                free(acb);                e_free(acb);
                 return NULL;                  return NULL;
         }          }
   
         return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);          return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);
   #endif  /* KQ_SUPPORT */
 }  }
   
 /*  /*
Line 792  schedLIORead(sched_root_task_t * __restrict root, sche Line 896  schedLIORead(sched_root_task_t * __restrict root, sche
  * @offset = Offset from start of file, if =-1 from current position   * @offset = Offset from start of file, if =-1 from current position
  * return: NULL error or !=NULL new queued task   * return: NULL error or !=NULL new queued task
  */   */
inline sched_task_t *sched_task_t *
 schedLIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd,   schedLIOWrite(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int fd, 
                 struct iovec *bufs, size_t nbufs, off_t offset)                  struct iovec *bufs, size_t nbufs, off_t offset)
 {  {
   #if SUP_ENABLE != KQ_SUPPORT
           sched_SetErr(ENOTSUP, "disabled kqueue support");
           return NULL;
   #else
         struct sigevent sig;          struct sigevent sig;
         struct aiocb **acb;          struct aiocb **acb;
         off_t off;          off_t off;
Line 813  schedLIOWrite(sched_root_task_t * __restrict root, sch Line 921  schedLIOWrite(sched_root_task_t * __restrict root, sch
         } else          } else
                 off = offset;                  off = offset;
   
        if (!(acb = calloc(sizeof(void*), nbufs))) {        if (!(acb = e_calloc(sizeof(void*), nbufs))) {
                 LOGERR;                  LOGERR;
                 return NULL;                  return NULL;
         } else          } else
                 memset(acb, 0, sizeof(void*) * nbufs);                  memset(acb, 0, sizeof(void*) * nbufs);
         for (i = 0; i < nbufs; off += bufs[i++].iov_len) {          for (i = 0; i < nbufs; off += bufs[i++].iov_len) {
                acb[i] = malloc(sizeof(struct aiocb));                acb[i] = e_malloc(sizeof(struct aiocb));
                 if (!acb[i]) {                  if (!acb[i]) {
                         LOGERR;                          LOGERR;
                         for (i = 0; i < nbufs; i++)                          for (i = 0; i < nbufs; i++)
                                 if (acb[i])                                  if (acb[i])
                                        free(acb[i]);                                        e_free(acb[i]);
                        free(acb);                        e_free(acb);
                         return NULL;                          return NULL;
                 } else                  } else
                         memset(acb[i], 0, sizeof(struct aiocb));                          memset(acb[i], 0, sizeof(struct aiocb));
Line 844  schedLIOWrite(sched_root_task_t * __restrict root, sch Line 952  schedLIOWrite(sched_root_task_t * __restrict root, sch
                 LOGERR;                  LOGERR;
                 for (i = 0; i < nbufs; i++)                  for (i = 0; i < nbufs; i++)
                         if (acb[i])                          if (acb[i])
                                free(acb[i]);                                e_free(acb[i]);
                free(acb);                e_free(acb);
                 return NULL;                  return NULL;
         }          }
   
         return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);          return schedLIO(root, func, arg, (void*) acb, bufs, nbufs);
   #endif  /* KQ_SUPPORT */
 }  }
 #endif  /* EVFILT_LIO */  #endif  /* EVFILT_LIO */
 #endif  /* AIO_SUPPORT */  #endif  /* AIO_SUPPORT */
Line 880  schedTimer(sched_root_task_t * __restrict root, sched_ Line 989  schedTimer(sched_root_task_t * __restrict root, sched_
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskTIMER;          TASK_TYPE(task) = taskTIMER;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 908  schedTimer(sched_root_task_t * __restrict root, sched_ Line 1017  schedTimer(sched_root_task_t * __restrict root, sched_
                 ptr = NULL;                  ptr = NULL;
   
         if (!ptr) {          if (!ptr) {
#ifdef HAVE_LIBPTHREAD                SCHED_QLOCK(root, taskTIMER);
                pthread_mutex_lock(&root->root_mtx[taskTIMER]); 
#endif 
 #ifdef TIMER_WITHOUT_SORT  #ifdef TIMER_WITHOUT_SORT
                TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);                TAILQ_INSERT_TAIL(&root->root_timer, task, task_node);
 #else  #else
                 TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)                  TAILQ_FOREACH_SAFE(t, &root->root_timer, task_node, tmp)
                         if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)                          if (sched_timespeccmp(&TASK_TS(task), &TASK_TS(t), -) < 1)
                                 break;                                  break;
                 if (!t)                  if (!t)
                        TAILQ_INSERT_TAIL(&root->root_timer, TASK_ID(task), task_node);                        TAILQ_INSERT_TAIL(&root->root_timer, task, task_node);
                 else                  else
                        TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);                        TAILQ_INSERT_BEFORE(t, task, task_node);
 #endif  #endif
#ifdef HAVE_LIBPTHREAD                SCHED_QUNLOCK(root, taskTIMER);
                pthread_mutex_unlock(&root->root_mtx[taskTIMER]); 
#endif 
         } else          } else
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
Line 956  schedEvent(sched_root_task_t * __restrict root, sched_ Line 1061  schedEvent(sched_root_task_t * __restrict root, sched_
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskEVENT;          TASK_TYPE(task) = taskEVENT;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 971  schedEvent(sched_root_task_t * __restrict root, sched_ Line 1076  schedEvent(sched_root_task_t * __restrict root, sched_
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_event);
                pthread_mutex_lock(&root->root_mtx[taskEVENT]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_event, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskEVENT]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
Line 1011  schedTask(sched_root_task_t * __restrict root, sched_t Line 1110  schedTask(sched_root_task_t * __restrict root, sched_t
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskTASK;          TASK_TYPE(task) = taskTASK;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 1027  schedTask(sched_root_task_t * __restrict root, sched_t Line 1126  schedTask(sched_root_task_t * __restrict root, sched_t
                 ptr = NULL;                  ptr = NULL;
   
         if (!ptr) {          if (!ptr) {
#ifdef HAVE_LIBPTHREAD                SCHED_QLOCK(root, taskTASK);
                pthread_mutex_lock(&root->root_mtx[taskTASK]); 
#endif 
                 TAILQ_FOREACH_SAFE(t, &root->root_task, task_node, tmp)                  TAILQ_FOREACH_SAFE(t, &root->root_task, task_node, tmp)
                         if (TASK_VAL(task) < TASK_VAL(t))                          if (TASK_VAL(task) < TASK_VAL(t))
                                 break;                                  break;
                 if (!t)                  if (!t)
                        TAILQ_INSERT_TAIL(&root->root_task, TASK_ID(task), task_node);                        TAILQ_INSERT_TAIL(&root->root_task, task, task_node);
                 else                  else
                        TAILQ_INSERT_BEFORE(t, TASK_ID(task), task_node);                        TAILQ_INSERT_BEFORE(t, task, task_node);
#ifdef HAVE_LIBPTHREAD                SCHED_QUNLOCK(root, taskTASK);
                pthread_mutex_unlock(&root->root_mtx[taskTASK]); 
#endif 
         } else          } else
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
Line 1071  schedSuspend(sched_root_task_t * __restrict root, sche Line 1166  schedSuspend(sched_root_task_t * __restrict root, sche
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskSUSPEND;          TASK_TYPE(task) = taskSUSPEND;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 1086  schedSuspend(sched_root_task_t * __restrict root, sche Line 1181  schedSuspend(sched_root_task_t * __restrict root, sche
         else          else
                 ptr = NULL;                  ptr = NULL;
   
        if (!ptr) {        if (!ptr)
#ifdef HAVE_LIBPTHREAD                insert_task_to(task, &root->root_suspend);
                pthread_mutex_lock(&root->root_mtx[taskSUSPEND]);        else
#endif 
                TAILQ_INSERT_TAIL(&root->root_suspend, TASK_ID(task), task_node); 
#ifdef HAVE_LIBPTHREAD 
                pthread_mutex_unlock(&root->root_mtx[taskSUSPEND]); 
#endif 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
         return task;          return task;
Line 1125  schedCallOnce(sched_root_task_t * __restrict root, sch Line 1214  schedCallOnce(sched_root_task_t * __restrict root, sch
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskEVENT;          TASK_TYPE(task) = taskEVENT;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 1147  schedCallOnce(sched_root_task_t * __restrict root, sch Line 1236  schedCallOnce(sched_root_task_t * __restrict root, sch
  * @root = root task   * @root = root task
  * @func = task execution function   * @func = task execution function
  * @arg = 1st func argument   * @arg = 1st func argument
 * @detach = Detach thread from scheduler, if !=0 * @ss = stack size
  * @opt_data = Optional data   * @opt_data = Optional data
  * @opt_dlen = Optional data length   * @opt_dlen = Optional data length
  * return: NULL error or !=NULL new queued task   * return: NULL error or !=NULL new queued task
  */   */
 sched_task_t *  sched_task_t *
schedThread(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, int detach, schedThread(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, 
                void *opt_data, size_t opt_dlen)                size_t ss, void *opt_data, size_t opt_dlen)
 {  {
 #ifndef HAVE_LIBPTHREAD  #ifndef HAVE_LIBPTHREAD
         sched_SetErr(ENOTSUP, "Not supported thread tasks");          sched_SetErr(ENOTSUP, "Not supported thread tasks");
         return NULL;          return NULL;
 #endif  #endif
         sched_task_t *task;          sched_task_t *task;
         void *ptr;  
         pthread_attr_t attr;          pthread_attr_t attr;
           void *ptr;
   
         if (!root || !func)          if (!root || !func)
                 return NULL;                  return NULL;
Line 1171  schedThread(sched_root_task_t * __restrict root, sched Line 1260  schedThread(sched_root_task_t * __restrict root, sched
         if (!(task = sched_useTask(root)))          if (!(task = sched_useTask(root)))
                 return NULL;                  return NULL;
   
        task->task_func = func;        TASK_FUNC(task) = func;
         TASK_TYPE(task) = taskTHREAD;          TASK_TYPE(task) = taskTHREAD;
         TASK_ROOT(task) = root;          TASK_ROOT(task) = root;
   
Line 1181  schedThread(sched_root_task_t * __restrict root, sched Line 1270  schedThread(sched_root_task_t * __restrict root, sched
         TASK_DATLEN(task) = opt_dlen;          TASK_DATLEN(task) = opt_dlen;
   
         pthread_attr_init(&attr);          pthread_attr_init(&attr);
        pthread_attr_setdetachstate(&attr, detach ?         pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
                        PTHREAD_CREATE_DETACHED : PTHREAD_CREATE_JOINABLE);        if (ss && (errno = pthread_attr_setstacksize(&attr, ss))) {
                 LOGERR;
                 pthread_attr_destroy(&attr);
                 return sched_unuseTask(task);
         }
         if ((errno = pthread_attr_getstacksize(&attr, &ss))) {
                 LOGERR;
                 pthread_attr_destroy(&attr);
                 return sched_unuseTask(task);
         } else
                 TASK_FLAG(task) = ss;
 
 #ifdef SCHED_RR
         pthread_attr_setschedpolicy(&attr, SCHED_RR);
 #else
         pthread_attr_setschedpolicy(&attr, SCHED_OTHER);
 #endif
 
         if (root->root_hooks.hook_add.thread)          if (root->root_hooks.hook_add.thread)
                 ptr = root->root_hooks.hook_add.thread(task, &attr);                  ptr = root->root_hooks.hook_add.thread(task, &attr);
         else          else
                 ptr = NULL;                  ptr = NULL;
         pthread_attr_destroy(&attr);  
   
        if (!ptr) {        if (!ptr)
                pthread_mutex_lock(&root->root_mtx[taskTHREAD]);                insert_task_to(task, &root->root_thread);
                TAILQ_INSERT_TAIL(&root->root_thread, TASK_ID(task), task_node);        else
                pthread_mutex_unlock(&root->root_mtx[taskTHREAD]); 
        } else 
                 task = sched_unuseTask(task);                  task = sched_unuseTask(task);
   
           pthread_attr_destroy(&attr);
         return task;          return task;
 }  }
   
   /*
    * schedRTC() - Add RTC task to scheduler queue
    *
    * @root = root task
    * @func = task execution function
    * @arg = 1st func argument
    * @ts = timeout argument structure, minimum alarm timer resolution is 1msec!
    * @opt_data = Optional RTC ID
    * @opt_dlen = Optional data length
    * return: NULL error or !=NULL new queued task
    */
   sched_task_t *
   schedRTC(sched_root_task_t * __restrict root, sched_task_func_t func, void *arg, struct timespec ts, 
                   void *opt_data, size_t opt_dlen)
   {
   #if defined(HAVE_LIBRT) && defined(HAVE_TIMER_CREATE) && \
           defined(HAVE_TIMER_SETTIME) && defined(HAVE_TIMER_DELETE)
           sched_task_t *task;
           void *ptr;
   
           if (!root || !func)
                   return NULL;
   
           /* get new task */
           if (!(task = sched_useTask(root)))
                   return NULL;
   
           TASK_FUNC(task) = func;
           TASK_TYPE(task) = taskRTC;
           TASK_ROOT(task) = root;
   
           TASK_ARG(task) = arg;
           TASK_TS(task) = ts;
   
           TASK_DATA(task) = opt_data;
           TASK_DATLEN(task) = opt_dlen;
   
           if (root->root_hooks.hook_add.rtc)
                   ptr = root->root_hooks.hook_add.rtc(task, NULL);
           else
                   ptr = NULL;
   
           if (!ptr)
                   insert_task_to(task, &root->root_rtc);
           else
                   task = sched_unuseTask(task);
   
           return task;
   #else
           sched_SetErr(ENOTSUP, "Not supported realtime clock extensions");
           return NULL;
   #endif
   }

Removed from v.1.13  
changed lines
  Added in v.1.30.6.2


FreeBSD-CVSweb <freebsd-cvsweb@FreeBSD.org>