git.sur5r.net Git - openldap/blob - libraries/libldap_r/rq.c

   1 /* $OpenLDAP$ */
   2 /* This work is part of OpenLDAP Software <http://www.openldap.org/>.
   3  *
   4  * Copyright 2003-2012 The OpenLDAP Foundation.
   5  * Portions Copyright 2003 IBM Corporation.
   6  * All rights reserved.
   7  *
   8  * Redistribution and use in source and binary forms, with or without
   9  * modification, are permitted only as authorized by the OpenLDAP
  10  * Public License.
  11  *
  12  * A copy of this license is available in file LICENSE in the
  13  * top-level directory of the distribution or, alternatively, at
  14  * <http://www.OpenLDAP.org/license.html>.
  15  */
  16 /* This work was initially developed by Jong Hyuk Choi for inclusion
  17  * in OpenLDAP Software.
  18  */
  19
  20 #include "portable.h"
  21
  22 #include <stdio.h>
  23
  24 #include <ac/stdarg.h>
  25 #include <ac/stdlib.h>
  26 #include <ac/errno.h>
  27 #include <ac/socket.h>
  28 #include <ac/string.h>
  29 #include <ac/time.h>
  30
  31 #include "ldap-int.h"
  32 #include "ldap_pvt_thread.h"
  33 #include "ldap_queue.h"
  34 #include "ldap_rq.h"
  35
  36 struct re_s *
  37 ldap_pvt_runqueue_insert(
  38         struct runqueue_s* rq,
  39         time_t interval,
  40         ldap_pvt_thread_start_t *routine,
  41         void *arg,
  42         char *tname,
  43         char *tspec
  44 )
  45 {
  46         struct re_s* entry;
  47
  48         entry = (struct re_s *) LDAP_CALLOC( 1, sizeof( struct re_s ));
  49         if ( entry ) {
  50                 entry->interval.tv_sec = interval;
  51                 entry->interval.tv_usec = 0;
  52                 entry->next_sched.tv_sec = time( NULL );
  53                 entry->next_sched.tv_usec = 0;
  54                 entry->routine = routine;
  55                 entry->arg = arg;
  56                 entry->tname = tname;
  57                 entry->tspec = tspec;
  58                 LDAP_STAILQ_INSERT_HEAD( &rq->task_list, entry, tnext );
  59         }
  60         return entry;
  61 }
  62
  63 struct re_s *
  64 ldap_pvt_runqueue_find(
  65         struct runqueue_s *rq,
  66         ldap_pvt_thread_start_t *routine,
  67         void *arg
  68 )
  69 {
  70         struct re_s* e;
  71
  72         LDAP_STAILQ_FOREACH( e, &rq->task_list, tnext ) {
  73                 if ( e->routine == routine && e->arg == arg )
  74                         return e;
  75         }
  76         return NULL;
  77 }
  78
  79 void
  80 ldap_pvt_runqueue_remove(
  81         struct runqueue_s* rq,
  82         struct re_s* entry
  83 )
  84 {
  85         struct re_s* e;
  86
  87         LDAP_STAILQ_FOREACH( e, &rq->task_list, tnext ) {
  88                 if ( e == entry)
  89                         break;
  90         }
  91
  92         assert( e == entry );
  93
  94         LDAP_STAILQ_REMOVE( &rq->task_list, entry, re_s, tnext );
  95
  96         LDAP_FREE( entry );
  97 }
  98
  99 struct re_s*
 100 ldap_pvt_runqueue_next_sched(
 101         struct runqueue_s* rq,
 102         struct timeval* next_run
 103 )
 104 {
 105         struct re_s* entry;
 106
 107         entry = LDAP_STAILQ_FIRST( &rq->task_list );
 108         if ( entry == NULL || entry->next_sched.tv_sec == 0 ) {
 109                 return NULL;
 110         } else {
 111                 *next_run = entry->next_sched;
 112                 return entry;
 113         }
 114 }
 115
 116 void
 117 ldap_pvt_runqueue_runtask(
 118         struct runqueue_s* rq,
 119         struct re_s* entry
 120 )
 121 {
 122         LDAP_STAILQ_INSERT_TAIL( &rq->run_list, entry, rnext );
 123 }
 124
 125 void
 126 ldap_pvt_runqueue_stoptask(
 127         struct runqueue_s* rq,
 128         struct re_s* entry
 129 )
 130 {
 131         LDAP_STAILQ_REMOVE( &rq->run_list, entry, re_s, rnext );
 132 }
 133
 134 int
 135 ldap_pvt_runqueue_isrunning(
 136         struct runqueue_s* rq,
 137         struct re_s* entry
 138 )
 139 {
 140         struct re_s* e;
 141
 142         LDAP_STAILQ_FOREACH( e, &rq->run_list, rnext ) {
 143                 if ( e == entry ) {
 144                         return 1;
 145                 }
 146         }
 147         return 0;
 148 }
 149
 150 void
 151 ldap_pvt_runqueue_resched(
 152         struct runqueue_s* rq,
 153         struct re_s* entry,
 154         int defer
 155 )
 156 {
 157         struct re_s* prev;
 158         struct re_s* e;
 159
 160         LDAP_STAILQ_FOREACH( e, &rq->task_list, tnext ) {
 161                 if ( e == entry )
 162                         break;
 163         }
 164
 165         assert ( e == entry );
 166
 167         LDAP_STAILQ_REMOVE( &rq->task_list, entry, re_s, tnext );
 168
 169         if ( !defer ) {
 170                 entry->next_sched.tv_sec = time( NULL ) + entry->interval.tv_sec;
 171         } else {
 172                 entry->next_sched.tv_sec = 0;
 173         }
 174
 175         if ( LDAP_STAILQ_EMPTY( &rq->task_list )) {
 176                 LDAP_STAILQ_INSERT_HEAD( &rq->task_list, entry, tnext );
 177         } else if ( entry->next_sched.tv_sec == 0 ) {
 178                 LDAP_STAILQ_INSERT_TAIL( &rq->task_list, entry, tnext );
 179         } else {
 180                 prev = NULL;
 181                 LDAP_STAILQ_FOREACH( e, &rq->task_list, tnext ) {
 182                         if ( e->next_sched.tv_sec == 0 ) {
 183                                 if ( prev == NULL ) {
 184                                         LDAP_STAILQ_INSERT_HEAD( &rq->task_list, entry, tnext );
 185                                 } else {
 186                                         LDAP_STAILQ_INSERT_AFTER( &rq->task_list, prev, entry, tnext );
 187                                 }
 188                                 return;
 189                         } else if ( e->next_sched.tv_sec > entry->next_sched.tv_sec ) {
 190                                 if ( prev == NULL ) {
 191                                         LDAP_STAILQ_INSERT_HEAD( &rq->task_list, entry, tnext );
 192                                 } else {
 193                                         LDAP_STAILQ_INSERT_AFTER( &rq->task_list, prev, entry, tnext );
 194                                 }
 195                                 return;
 196                         }
 197                         prev = e;
 198                 }
 199                 LDAP_STAILQ_INSERT_TAIL( &rq->task_list, entry, tnext );
 200         }
 201 }
 202
 203 int
 204 ldap_pvt_runqueue_persistent_backload(
 205         struct runqueue_s* rq
 206 )
 207 {
 208         struct re_s* e;
 209         int count = 0;
 210
 211         ldap_pvt_thread_mutex_lock( &rq->rq_mutex );
 212         if ( !LDAP_STAILQ_EMPTY( &rq->task_list )) {
 213                 LDAP_STAILQ_FOREACH( e, &rq->task_list, tnext ) {
 214                         if ( e->next_sched.tv_sec == 0 )
 215                                 count++;
 216                 }
 217         }
 218         ldap_pvt_thread_mutex_unlock( &rq->rq_mutex );
 219         return count;
 220 }
 221