]> git.sur5r.net Git - openldap/blob - servers/slurpd/rq.c
Don't defer abandon due to pending
[openldap] / servers / slurpd / rq.c
1 /* $OpenLDAP$ */
2 /* This work is part of OpenLDAP Software <http://www.openldap.org/>.
3  *
4  * Copyright 1998-2004 The OpenLDAP Foundation.
5  * All rights reserved.
6  *
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted only as authorized by the OpenLDAP
9  * Public License.
10  *
11  * A copy of this license is available in file LICENSE in the
12  * top-level directory of the distribution or, alternatively, at
13  * <http://www.OpenLDAP.org/license.html>.
14  */
15 /* Portions Copyright (c) 1996 Regents of the University of Michigan.
16  * All rights reserved.
17  *
18  * Redistribution and use in source and binary forms are permitted
19  * provided that this notice is preserved and that due credit is given
20  * to the University of Michigan at Ann Arbor. The name of the University
21  * may not be used to endorse or promote products derived from this
22  * software without specific prior written permission. This software
23  * is provided ``as is'' without express or implied warranty.
24  */
25 /* ACKNOWLEDGEMENTS:
26  * This work was originally developed by the University of Michigan
27  * (as part of U-MICH LDAP).
28  */
29
30 /*
31  * rq.c - routines used to manage the queue of replication entries.
32  * An Rq (Replication queue) struct contains a linked list of Re
33  * (Replication entry) structures.
34  *
35  * Routines wishing to access the replication queue should do so through
36  * the Rq struct's member functions, e.g. rq->rq_gethead() and friends.
37  * For example, Re structs should be added to the queue by calling 
38  * the rq_add() member function.
39  *
40  * Access to the queue is serialized by a mutex.  Member functions which do
41  * not do their own locking should only be called after locking the queue
42  * using the rq_lock() member function.  The queue should be unlocked with
43  * the rq_unlock() member function.
44  *
45  * Note that some member functions handle their own locking internally.
46  * Callers should not lock the queue before calling these functions.
47  * See the comment block for each function below.
48  *
49  */
50
51 #include "portable.h"
52
53 #include <stdio.h>
54 #include <sys/stat.h>
55
56 #include <ac/stdlib.h>
57 #include <ac/string.h>
58 #include <ac/unistd.h>          /* get ftruncate() */
59
60 #ifdef HAVE_SYS_TYPES_H
61 #include <sys/types.h>
62 #endif
63 #ifdef HAVE_FCNTL_H
64 #include <fcntl.h>
65 #endif
66
67 #include "slurp.h"
68 #include "globals.h"
69
70 /*
71  * Lock the replication queue.
72  */
73 static int
74 Rq_lock(
75     Rq  *rq
76 )
77 {
78     return( ldap_pvt_thread_mutex_lock( &rq->rq_mutex ));
79 }
80
81
82 /*
83  * Unlock the replication queue.
84  */
85 static int
86 Rq_unlock(
87     Rq  *rq
88 )
89 {
90     return( ldap_pvt_thread_mutex_unlock( &rq->rq_mutex ));
91 }
92
93
94
95 /*
96  * Return the head of the queue.  Callers should lock the queue before
97  * calling this routine.
98  */
99 static Re *
100 Rq_gethead(
101     Rq  *rq
102 )
103 {
104     return( rq == NULL ? NULL : rq->rq_head );
105 }
106
107
108 /*
109  * Return the next item in the queue.  Callers should lock the queue before
110  * calling this routine.
111  */
112 static Re *
113 Rq_getnext(
114     Re  *re
115 )
116 {
117     if ( re == NULL ) {
118         return NULL;
119     } else {
120         return( re->re_getnext( re ));
121     }
122 }
123
124
125 /*
126  * Delete the item at the head of the list.  The queue should be locked
127  * by the caller before calling this routine.
128  */
129 static int
130 Rq_delhead(
131     Rq  *rq
132 )
133 {
134     Re  *savedhead;
135     int rc;
136
137     if ( rq == NULL ) {
138         return( -1 );
139     }
140
141     savedhead = rq->rq_head;
142     if ( savedhead == NULL ) {
143         return( 0 );
144     }
145
146     if ( savedhead->re_getrefcnt( savedhead ) != 0 ) {
147 #ifdef NEW_LOGGING
148         LDAP_LOG ( SLURPD, WARNING, "Rq_delhead: "
149                 "Warning: attempt to delete when refcnt != 0\n", 0, 0, 0 );
150 #else
151         Debug( LDAP_DEBUG_ANY, "Warning: attempt to delete when refcnt != 0\n",
152                 0, 0, 0 );
153 #endif
154         return( -1 );
155     }
156
157     rq->rq_head = rq->rq_head->re_getnext( rq->rq_head );
158     rc = savedhead->re_free( savedhead );
159     rq->rq_nre--;       /* decrement count of Re's in queue */
160     return( rc );
161 }
162
163
164 /* 
165  * Add an entry to the tail of the replication queue.  Locking is handled
166  * internally.  When items are added to the queue, this routine wakes
167  * up any threads which are waiting for more work by signaling on the
168  * rq->rq_more condition variable.
169  */
170 static int
171 Rq_add(
172     Rq          *rq,
173     char        *buf
174 )
175 {
176     Re  *re;
177     int wasempty = 0;
178
179     /* Lock the queue */
180     rq->rq_lock( rq );
181
182     /* Create a new Re */
183     if ( Re_init( &re ) < 0 ) {
184         rq->rq_unlock( rq );
185         return -1;
186     }
187
188     /* parse buf and fill in the re struct */
189     if ( re->re_parse( re, buf ) < 0 ) {
190         re->re_free( re );
191         rq->rq_unlock( rq );
192         return -1;
193     }
194
195     /* Insert into queue */
196     if ( rq->rq_head == NULL ) {
197         rq->rq_head = re;
198         rq->rq_tail = re;
199         wasempty = 1;
200     } else {
201         rq->rq_tail->re_next = re;
202     }
203
204     /* set the sequence number */
205     re->re_seq = 0;
206     if ( !wasempty && ( rq->rq_tail->re_timestamp == re->re_timestamp )) {
207         /*
208          * Our new re has the same timestamp as the tail's timestamp.
209          * Increment the seq number in the tail and use it as our seq number.
210          */
211         re->re_seq = rq->rq_tail->re_seq + 1;
212     }
213     rq->rq_tail = re;
214
215     /* Increment count of items in queue */
216     rq->rq_nre++;
217     /* wake up any threads waiting for more work */
218     ldap_pvt_thread_cond_broadcast( &rq->rq_more );
219
220     /* ... and unlock the queue */
221     rq->rq_unlock( rq );
222
223     return 0;
224 }
225
226
227 /*
228  * Garbage-collect the replication queue.  Locking is handled internally.
229  */
230 static void
231 Rq_gc(
232     Rq  *rq
233 )
234 {
235     if ( rq == NULL ) {
236 #ifdef NEW_LOGGING
237         LDAP_LOG ( SLURPD, DETAIL1, "Rq_gc: rq is NULL!\n", 0, 0, 0 );
238 #else
239         Debug( LDAP_DEBUG_ANY, "Rq_gc: rq is NULL!\n", 0, 0, 0 );
240 #endif
241         return;
242     }
243     rq->rq_lock( rq ); 
244     while (( rq->rq_head != NULL ) &&
245             ( rq->rq_head->re_getrefcnt( rq->rq_head ) == 0 )) {
246         rq->rq_delhead( rq );
247         rq->rq_ndel++;  /* increment count of deleted entries */
248     }
249     rq->rq_unlock( rq ); 
250     return;
251 }
252
253
254 /*
255  * For debugging: dump the contents of the replication queue to a file.
256  * Locking is handled internally.
257  */
258 static void
259 Rq_dump(
260     Rq  *rq
261 )
262 {
263     Re          *re;
264     FILE        *fp;
265     int         tmpfd;
266
267     if ( rq == NULL ) {
268 #ifdef NEW_LOGGING
269         LDAP_LOG ( SLURPD, ARGS, "Rq_dump: rq is NULL!\n", 0, 0, 0 );
270 #else
271         Debug( LDAP_DEBUG_ANY, "Rq_dump: rq is NULL!\n", 0, 0, 0 );
272 #endif
273         return;
274     }
275
276     if (unlink(SLURPD_DUMPFILE) == -1 && errno != ENOENT) {
277 #ifdef NEW_LOGGING
278         LDAP_LOG ( SLURPD, ERR, "Rq_dump: "
279                 "\"%s\" exists, cannot unlink\n", SLURPD_DUMPFILE, 0, 0 );
280 #else
281         Debug( LDAP_DEBUG_ANY, "Rq_dump: \"%s\" exists, and cannot unlink\n",
282                 SLURPD_DUMPFILE, 0, 0 );
283 #endif
284         return;
285     }
286     if (( tmpfd = open(SLURPD_DUMPFILE, O_CREAT|O_RDWR|O_EXCL, 0600)) == -1) {
287 #ifdef NEW_LOGGING
288         LDAP_LOG ( SLURPD, ERR, "Rq_dump: "
289                 "cannot open \"%s\" for write\n", SLURPD_DUMPFILE, 0, 0 );
290 #else
291         Debug( LDAP_DEBUG_ANY, "Rq_dump: cannot open \"%s\" for write\n",
292                 SLURPD_DUMPFILE, 0, 0 );
293 #endif
294         return;
295     }
296     if (( fp = fdopen( tmpfd, "w" )) == NULL ) {
297 #ifdef NEW_LOGGING
298         LDAP_LOG ( SLURPD, ERR, "Rq_dump: "
299                 "cannot fdopen \"%s\" for write\n", SLURPD_DUMPFILE, 0, 0 );
300 #else
301         Debug( LDAP_DEBUG_ANY, "Rq_dump: cannot fdopen \"%s\" for write\n",
302                 SLURPD_DUMPFILE, 0, 0 );
303 #endif
304         return;
305     }
306
307     rq->rq_lock( rq );
308     for ( re = rq->rq_gethead( rq ); re != NULL; re = rq->rq_getnext( re )) {
309         re->re_dump( re, fp );
310     }
311     rq->rq_unlock( rq );
312     fclose( fp );
313     return;
314 }
315
316
317 /*
318  * Write the contents of a replication queue to a file.  Returns zero if
319  * successful, -1 if not.  Handles queue locking internally.  Callers should
320  * provide an open file pointer, which should refer to a locked file.
321  */
322 static int
323 Rq_write(
324     Rq          *rq,
325     FILE        *fp
326 )
327 {
328     Re          *re;
329     time_t      now;
330
331     if ( rq == NULL ) {
332         return -1;
333     }
334
335 #ifdef NEW_LOGGING
336         LDAP_LOG ( SLURPD, ENTRY, "Rq_write: "
337                 "re-write on-disk replication log\n", 0, 0, 0 );
338 #else
339     Debug( LDAP_DEBUG_ARGS, "re-write on-disk replication log\n",
340             0, 0, 0 );
341 #endif
342 #ifndef SEEK_SET
343 #define SEEK_SET 0
344 #endif
345     fseek( fp, 0L, SEEK_SET );  /* Go to beginning of file */
346     rq->rq_lock( rq );
347
348     for ( re = rq->rq_gethead( rq ); re != NULL; re = rq->rq_getnext( re )) {
349         if ( re->re_write( NULL, re, fp ) < 0 ) {
350             fflush( fp );
351             rq->rq_unlock( rq );
352             return -1;
353         }
354     }
355     fflush( fp );
356     sglob->srpos = ftell( fp ); /* update replog file position */
357     /* and truncate to correct len */
358     if ( ftruncate( fileno( fp ), sglob->srpos ) < 0 ) {
359 #ifdef NEW_LOGGING
360         LDAP_LOG ( SLURPD, ERR, "Rq_write: "
361                 "Error truncating replication log: %s\n", sys_errlist[ errno ], 0, 0 );
362 #else
363         Debug( LDAP_DEBUG_ANY, "Error truncating replication log: %s\n",
364                 sys_errlist[ errno ], 0, 0 );
365 #endif
366     }
367     rq->rq_ndel = 0;    /* reset count of deleted re's */
368     time( &now );
369     rq->rq_lasttrim = now;      /* reset last trim time */
370     rq->rq_unlock( rq );
371     return 0;
372 }
373
374
375 /*
376  * Check to see if the private slurpd replication log needs trimming.
377  * The current criteria are:
378  *  - The last trim was more than 5 minutes ago, *and*
379  *  - We've finished with at least 50 replication log entries since the
380  *    last time we re-wrote the replication log.
381  *
382  * Return 1 if replogfile should be trimmed, 0 if not.
383  * Any different policy should be implemented by replacing this function.
384  */
385 static int
386 Rq_needtrim(
387     Rq  *rq
388 )
389 {
390     int         rc = 0;
391     time_t      now;
392
393     if ( rq == NULL ) {
394         return 0;
395     }
396
397     rq->rq_lock( rq );
398
399     time( &now );
400
401     if ( now > ( rq->rq_lasttrim + TRIMCHECK_INTERVAL )) {
402         rc = ( rq->rq_ndel >= 50 );
403     } else {
404         rc = 0;
405     }
406     rq->rq_unlock( rq );
407     return rc;
408 }
409
410
411 /*
412  * Return counts of Re structs in the queue.
413  */
414 static int
415 Rq_getcount(
416     Rq  *rq,
417     int type
418 )
419 {
420     int count = 0;
421     Re  *re;
422
423     if ( rq == NULL ) {
424         return 0;
425     }
426
427     rq->rq_lock( rq );
428     if ( type == RQ_COUNT_ALL ) {
429         count = rq->rq_nre;
430     } else {
431         for ( re = rq->rq_gethead( rq ); re != NULL;
432                 re = rq->rq_getnext( re )) {
433             if ( type == RQ_COUNT_NZRC ) {
434                 if ( re->re_getrefcnt( re ) > 0 ) {
435                     count++;
436                 }
437             }
438         }
439     }
440     rq->rq_unlock( rq );
441     return count;
442 }
443
444
445 /* 
446  * Allocate and initialize an Rq object.
447  */
448 int
449 Rq_init(
450     Rq  **rq
451 )
452 {
453     /* Instantiate the struct */
454     (*rq) = (Rq *) malloc( sizeof( Rq ));
455     if ( *rq == NULL ) {
456         return -1;
457     }
458
459     /* Fill in all the function pointers */
460     (*rq)->rq_gethead = Rq_gethead;
461     (*rq)->rq_getnext = Rq_getnext;
462     (*rq)->rq_delhead = Rq_delhead;
463     (*rq)->rq_add = Rq_add;
464     (*rq)->rq_gc = Rq_gc;
465     (*rq)->rq_lock = Rq_lock;
466     (*rq)->rq_unlock = Rq_unlock;
467     (*rq)->rq_dump = Rq_dump;
468     (*rq)->rq_needtrim = Rq_needtrim;
469     (*rq)->rq_write = Rq_write;
470     (*rq)->rq_getcount = Rq_getcount;
471
472     /* Initialize private data */
473     ldap_pvt_thread_mutex_init( &((*rq)->rq_mutex) );
474     ldap_pvt_thread_cond_init( &((*rq)->rq_more) );
475     (*rq)->rq_head = NULL;
476     (*rq)->rq_tail = NULL;
477     (*rq)->rq_nre = 0;
478     (*rq)->rq_ndel = 0;
479     (*rq)->rq_lasttrim = (time_t) 0L;
480
481     return 0;
482 }