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