]> git.sur5r.net Git - openldap/blob - servers/slurpd/fm.c
Don't defer abandon due to pending
[openldap] / servers / slurpd / fm.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  * fm.c - file management routines.
32  */
33
34 #include "portable.h"
35
36 #include <stdio.h>
37
38 #include <ac/stdlib.h>
39 #include <ac/string.h>
40 #include <ac/signal.h>
41 #include <ac/socket.h>
42 #include <ac/unistd.h>
43
44 #include "slurp.h"
45 #include "globals.h"
46
47
48 /*
49  * Forward references
50  */
51 static char *get_record LDAP_P(( FILE * ));
52 static void populate_queue LDAP_P(( char *f ));
53
54
55 /*
56  * Main file manager routine.  Watches for new data to be appended to the
57  * slapd replication log.  When new data is appended, fm does the following:
58  *  - appends the data to slurpd's private copy of the replication log.
59  *  - truncates the slapd replog
60  *  - adds items to the internal queue of replication work to do
61  *  - signals the replication threads to let them know new work has arrived.
62  */
63 void *
64 fm(
65     void *arg
66 )
67 {
68     int rc;
69     int i;
70     fd_set readfds;
71
72     /* Set up our signal handlers:
73      * SIG{TERM,INT,HUP} causes a shutdown
74      */
75     (void) SIGNAL( SIGTERM, slurp_set_shutdown );
76     (void) SIGNAL( SIGINT, slurp_set_shutdown );
77 #ifdef SIGHUP
78     (void) SIGNAL( SIGHUP, slurp_set_shutdown );
79 #endif
80 #if defined(SIGBREAK) && defined(HAVE_NT_SERVICE_MANAGER)
81     (void) SIGNAL( SIGBREAK, do_nothing );
82 #endif
83
84     if ( sglob->one_shot_mode ) {
85         if ( file_nonempty( sglob->slapd_replogfile )) {
86             populate_queue( sglob->slapd_replogfile );
87         }
88         printf( "Processing in one-shot mode:\n" );
89         printf( "%d total replication records in file,\n",
90                 sglob->rq->rq_getcount( sglob->rq, RQ_COUNT_ALL ));
91         printf( "%d replication records to process.\n",
92                 sglob->rq->rq_getcount( sglob->rq, RQ_COUNT_NZRC ));
93         return NULL;
94     }
95     /*
96      * There may be some leftover replication records in our own
97      * copy of the replication log.  If any exist, add them to the
98      * queue.
99      */
100     if ( file_nonempty( sglob->slurpd_replogfile )) {
101         populate_queue( sglob->slurpd_replogfile );
102     }
103
104     FD_ZERO( &readfds );
105
106     while ( !sglob->slurpd_shutdown ) {
107         if ( file_nonempty( sglob->slapd_replogfile )) {
108             /* New work found - copy to slurpd replog file */
109 #ifdef NEW_LOGGING
110         LDAP_LOG ( SLURPD, ARGS, 
111                         "fm: new work in %s\n", sglob->slapd_replogfile, 0, 0 );
112 #else
113             Debug( LDAP_DEBUG_ARGS, "new work in %s\n",
114                     sglob->slapd_replogfile, 0, 0 );
115 #endif
116             if (( rc = copy_replog( sglob->slapd_replogfile,
117                     sglob->slurpd_replogfile )) == 0 )  {
118                 populate_queue( sglob->slurpd_replogfile );
119             } else {
120                 if ( rc < 0 ) {
121 #ifdef NEW_LOGGING
122                 LDAP_LOG ( SLURPD, CRIT, 
123                                 "fm: Fatal error while copying replication log\n" , 0, 0, 0);
124 #else
125                     Debug( LDAP_DEBUG_ANY,
126                             "Fatal error while copying replication log\n",
127                             0, 0, 0 );
128 #endif
129                     sglob->slurpd_shutdown = 1;
130                 }
131             }
132         } else {
133             struct timeval tv;
134
135             FD_SET( sglob->wake_sds[0], &readfds );
136             tv.tv_sec = sglob->no_work_interval;
137             tv.tv_usec = 0;
138
139             rc = select( sglob->wake_sds[0]+1, &readfds, NULL, NULL, &tv );
140         }
141
142         /* Garbage-collect queue */
143         sglob->rq->rq_gc( sglob->rq );
144
145         /* Trim replication log file, if needed */
146         if ( sglob->rq->rq_needtrim( sglob->rq )) {
147             FILE *fp, *lfp;
148             if (( rc = acquire_lock( sglob->slurpd_replogfile, &fp,
149                     &lfp )) < 0 ) {
150 #ifdef NEW_LOGGING
151                 LDAP_LOG ( SLURPD, ERR, 
152                         "fm: Error: cannot acquire lock on \"%s\" for trimming\n",
153                         sglob->slurpd_replogfile, 0, 0 );
154 #else
155                 Debug( LDAP_DEBUG_ANY,
156                         "Error: cannot acquire lock on \"%s\" for trimming\n",
157                         sglob->slurpd_replogfile, 0, 0 );
158 #endif
159             } else {
160                 sglob->rq->rq_write( sglob->rq, fp );
161                 (void) relinquish_lock( sglob->slurpd_replogfile, fp, lfp );
162             }
163         }
164     }
165     sglob->rq->rq_lock( sglob->rq );                    /* lock queue */
166     ldap_pvt_thread_cond_broadcast( &(sglob->rq->rq_more) );    /* wake repl threads */
167     for ( i = 0; i < sglob->num_replicas; i++ ) {
168         (sglob->replicas[ i ])->ri_wake( sglob->replicas[ i ]);
169     }
170     sglob->rq->rq_unlock( sglob->rq );                  /* unlock queue */
171 #ifdef NEW_LOGGING
172         LDAP_LOG ( SLURPD, RESULTS, "fm: exiting\n", 0, 0, 0 );
173 #else
174     Debug( LDAP_DEBUG_ARGS, "fm: exiting\n", 0, 0, 0 );
175 #endif
176     return NULL;
177 }
178
179
180
181
182 /*
183  * Set a global flag which signals that we're shutting down.
184  */
185 RETSIGTYPE
186 slurp_set_shutdown(int sig)
187 {
188     sglob->slurpd_shutdown = 1;                         /* set flag */
189     tcp_write( sglob->wake_sds[1], "0", 1);             /* wake up file mgr */
190
191     (void) SIGNAL_REINSTALL( sig, slurp_set_shutdown ); /* reinstall handlers */
192 }
193
194
195
196
197 /*
198  * A do-nothing signal handler.
199  */
200 RETSIGTYPE
201 do_nothing(int sig)
202 {
203     (void) SIGNAL_REINSTALL( sig, do_nothing );
204 }
205
206
207
208
209 /*
210  * Open the slurpd replication log, seek to our last known position, and
211  * process any pending replication entries.
212  */
213 static void
214 populate_queue(
215     char *f
216 )
217 {
218     FILE        *fp, *lfp;
219     char        *p;
220
221     if ( acquire_lock( f, &fp, &lfp ) < 0 ) {
222 #ifdef NEW_LOGGING
223         LDAP_LOG ( SLURPD, ERR, 
224                 "populate_queue: error: can't lock file \"%s\": %s\n", 
225                 f, sys_errlist[ errno ], 0 );
226 #else
227         Debug( LDAP_DEBUG_ANY,
228                 "error: can't lock file \"%s\": %s\n",
229                 f, sys_errlist[ errno ], 0 );
230 #endif
231         return;
232     }
233
234     /*
235      * Read replication records from fp and append them the
236      * the queue.
237      */
238     if ( fseek( fp, sglob->srpos, 0 ) < 0 ) {
239 #ifdef NEW_LOGGING
240         LDAP_LOG ( SLURPD, ERR, 
241                 "populate_queue: error: can't seek to offset %ld in file \"%s\"\n", 
242                 sglob->srpos, f, 0 );
243 #else
244         Debug( LDAP_DEBUG_ANY,
245                 "error: can't seek to offset %ld in file \"%s\"\n",
246                 sglob->srpos, f, 0 );
247 #endif
248     } else {
249     while (( p = get_record( fp )) != NULL ) {
250         if ( sglob->rq->rq_add( sglob->rq, p ) < 0 ) {
251             char *t;
252             /* Print an error message.  Only print first line.  */
253             if (( t = strchr( p, '\n' )) != NULL ) {
254                 *t = '\0';
255             }
256 #ifdef NEW_LOGGING
257                 LDAP_LOG ( SLURPD, ERR, 
258                         "populate_queue: error: malformed replog entry "
259                         "(begins with \"%s\")\n", p, 0, 0 );
260 #else
261             Debug( LDAP_DEBUG_ANY,
262                     "error: malformed replog entry (begins with \"%s\")\n",
263                     p, 0, 0 );
264 #endif
265         }
266         free( p );
267         ldap_pvt_thread_yield();
268     }
269     sglob->srpos = ftell( fp );
270     }
271     (void) relinquish_lock( f, fp, lfp );
272 }
273     
274
275
276
277 /*
278  * Get the next "record" from the file pointed to by fp.  A "record"
279  * is delimited by two consecutive newlines.  Returns NULL on EOF.
280  */
281 static char *
282 get_record(
283     FILE *fp
284 )
285 {
286     int         len;
287     static char line[BUFSIZ];
288     char        *buf = NULL;
289     static int  lcur, lmax;
290
291     lcur = lmax = 0;
292
293     while (( fgets( line, sizeof(line), fp ) != NULL ) &&
294             (( len = strlen( line )) > 1 )) {
295
296                 while ( lcur + len + 1 > lmax ) {
297                     lmax += BUFSIZ;
298                     buf = (char *) ch_realloc( buf, lmax );
299                 }
300                 strcpy( buf + lcur, line );
301                 lcur += len;
302     }
303     return( buf );
304 }