git.sur5r.net Git - openldap/blob - servers/slurpd/ri.c

   1 /*
   2  * Copyright (c) 1996 Regents of the University of Michigan.
   3  * All rights reserved.
   4  *
   5  * Redistribution and use in source and binary forms are permitted
   6  * provided that this notice is preserved and that due credit is given
   7  * to the University of Michigan at Ann Arbor. The name of the University
   8  * may not be used to endorse or promote products derived from this
   9  * software without specific prior written permission. This software
  10  * is provided ``as is'' without express or implied warranty.
  11  */
  12
  13 /*
  14  * ri.c - routines used to manipulate Ri structures.  An Ri (Replica
  15  * information) struct contains all information about one replica
  16  * instance.  The Ri struct is defined in slurp.h
  17  */
  18
  19
  20 #include "portable.h"
  21
  22 #include <stdio.h>
  23 #include <stdlib.h>
  24 #include <ac/signal.h>
  25
  26 #include "slurp.h"
  27 #include "globals.h"
  28
  29
  30 /* Forward references */
  31 static int ismine LDAP_P(( Ri  *, Re  * ));
  32 static int isnew LDAP_P(( Ri  *, Re  * ));
  33
  34
  35 /*
  36  * Process any unhandled replication entries in the queue.
  37  */
  38 static int
  39 Ri_process(
  40     Ri *ri
  41 )
  42 {
  43     Rq          *rq = sglob->rq;
  44     Re          *re, *new_re;
  45     int         rc ;
  46     char        *errmsg;
  47
  48 #ifdef HAVE_LINUX_THREADS
  49     (void) SIGNAL( SIGSTKFLT, do_nothing );
  50 #else
  51     (void) SIGNAL( SIGUSR1, do_nothing );
  52 #endif
  53     (void) SIGNAL( SIGPIPE, SIG_IGN );
  54     if ( ri == NULL ) {
  55         Debug( LDAP_DEBUG_ANY, "Error: Ri_process: ri == NULL!\n", 0, 0, 0 );
  56         return -1;
  57     }
  58
  59     /*
  60      * Startup code.  See if there's any work to do.  If not, wait on the
  61      * rq->rq_more condition variable.
  62      */
  63     rq->rq_lock( rq );
  64     while ( !sglob->slurpd_shutdown &&
  65             (( re = rq->rq_gethead( rq )) == NULL )) {
  66         /* No work - wait on condition variable */
  67         pthread_cond_wait( &rq->rq_more, &rq->rq_mutex );
  68     }
  69
  70     /*
  71      * When we get here, there's work in the queue, and we have the
  72      * queue locked.  re should be pointing to the head of the queue.
  73      */
  74     rq->rq_unlock( rq );
  75     while ( !sglob->slurpd_shutdown ) {
  76         if ( re != NULL ) {
  77             if ( !ismine( ri, re )) {
  78                 /* The Re doesn't list my host:port */
  79                 Debug( LDAP_DEBUG_TRACE,
  80                         "Replica %s:%d, skip repl record for %s (not mine)\n",
  81                         ri->ri_hostname, ri->ri_port, re->re_dn );
  82             } else if ( !isnew( ri, re )) {
  83                 /* This Re is older than my saved status information */
  84                 Debug( LDAP_DEBUG_TRACE,
  85                         "Replica %s:%d, skip repl record for %s (old)\n",
  86                         ri->ri_hostname, ri->ri_port, re->re_dn );
  87             } else {
  88                 rc = do_ldap( ri, re, &errmsg );
  89                 switch ( rc ) {
  90                 case DO_LDAP_ERR_RETRYABLE:
  91                     tsleep( RETRY_SLEEP_TIME );
  92                     Debug( LDAP_DEBUG_ANY,
  93                             "Retrying operation for DN %s on replica %s:%d\n",
  94                             re->re_dn, ri->ri_hostname, ri->ri_port );
  95                     continue;
  96                     break;
  97                 case DO_LDAP_ERR_FATAL: {
  98                     /* Non-retryable error.  Write rejection log. */
  99                         int ld_errno = 0;
 100                         ldap_get_option(ri->ri_ldp, LDAP_OPT_ERROR_NUMBER, &ld_errno);
 101                     write_reject( ri, re, ld_errno, errmsg );
 102                     /* Update status ... */
 103                     (void) sglob->st->st_update( sglob->st, ri->ri_stel, re );
 104                     /* ... and write to disk */
 105                     (void) sglob->st->st_write( sglob->st );
 106                     } break;
 107                 default:
 108                     /* LDAP op completed ok - update status... */
 109                     (void) sglob->st->st_update( sglob->st, ri->ri_stel, re );
 110                     /* ... and write to disk */
 111                     (void) sglob->st->st_write( sglob->st );
 112                     break;
 113                 }
 114             }
 115         } else {
 116             Debug( LDAP_DEBUG_ANY, "Error: re is null in Ri_process\n",
 117                     0, 0, 0 );
 118         }
 119         rq->rq_lock( rq );
 120         while ( !sglob->slurpd_shutdown &&
 121                 ((new_re = re->re_getnext( re )) == NULL )) {
 122             if ( sglob->one_shot_mode ) {
 123                 rq->rq_unlock( rq );
 124                 return 0;
 125             }
 126             /* No work - wait on condition variable */
 127             pthread_cond_wait( &rq->rq_more, &rq->rq_mutex );
 128         }
 129         re->re_decrefcnt( re );
 130         re = new_re;
 131         rq->rq_unlock( rq );
 132         if ( sglob->slurpd_shutdown ) {
 133             return 0;
 134         }
 135     }
 136     return 0;
 137 }
 138
 139
 140 /*
 141  * Wake a replication thread which may be sleeping.
 142  * Send it a SIG(STKFLT|USR1).
 143  */
 144 static void
 145 Ri_wake(
 146     Ri *ri
 147 )
 148 {
 149     if ( ri == NULL ) {
 150         return;
 151     }
 152 #ifdef HAVE_LINUX_THREADS
 153     pthread_kill( ri->ri_tid, SIGSTKFLT );
 154     (void) SIGNAL( SIGSTKFLT, do_nothing );
 155 #else
 156     pthread_kill( ri->ri_tid, SIGUSR1 );
 157     (void) SIGNAL( SIGUSR1, do_nothing );
 158 #endif
 159 }
 160
 161
 162
 163 /*
 164  * Allocate and initialize an Ri struct.
 165  */
 166 int
 167 Ri_init(
 168     Ri  **ri
 169 )
 170 {
 171     (*ri) = ( Ri * ) malloc( sizeof( Ri ));
 172     if ( *ri == NULL ) {
 173         return -1;
 174     }
 175
 176     /* Initialize member functions */
 177     (*ri)->ri_process = Ri_process;
 178     (*ri)->ri_wake = Ri_wake;
 179
 180     /* Initialize private data */
 181     (*ri)->ri_hostname = NULL;
 182     (*ri)->ri_port = 0;
 183     (*ri)->ri_ldp = NULL;
 184     (*ri)->ri_bind_method = 0;
 185     (*ri)->ri_bind_dn = NULL;
 186     (*ri)->ri_password = NULL;
 187     (*ri)->ri_principal = NULL;
 188     (*ri)->ri_srvtab = NULL;
 189     (*ri)->ri_curr = NULL;
 190
 191     return 0;
 192 }
 193
 194
 195
 196
 197 /*
 198  * Return 1 if the hostname and port in re match the hostname and port
 199  * in ri, otherwise return zero.
 200  */
 201 static int
 202 ismine(
 203     Ri  *ri,
 204     Re  *re
 205 )
 206 {
 207     Rh  *rh;
 208     int i;
 209
 210     if ( ri == NULL || re == NULL || ri->ri_hostname == NULL ||
 211             re->re_replicas == NULL ) {
 212         return 0;
 213     }
 214     rh = re->re_replicas;
 215     for ( i = 0; rh[ i ].rh_hostname != NULL; i++ ) {
 216         if ( !strcmp( rh[ i ].rh_hostname, ri->ri_hostname) &&
 217                 rh[ i ].rh_port == ri->ri_port ) {
 218             return 1;
 219         }
 220     }
 221     return 0;
 222 }
 223
 224
 225
 226
 227 /*
 228  * Return 1 if the Re's timestamp/seq combination are greater than the
 229  * timestamp and seq in the Ri's ri_stel member.  In other words, if we
 230  * find replication entries in the log which we've already processed,
 231  * don't process them.  If the re is "old," return 0.
 232  * No check for NULL pointers is done.
 233  */
 234 static int
 235 isnew(
 236     Ri  *ri,
 237     Re  *re
 238 )
 239 {
 240     int x;
 241     int ret;
 242
 243     /* Lock the St struct to avoid a race */
 244     sglob->st->st_lock( sglob->st );
 245     x = strcmp( re->re_timestamp, ri->ri_stel->last );
 246     if ( x > 0 ) {
 247         /* re timestamp is newer */
 248         ret = 1;
 249     } else if ( x < 0 ) {
 250         ret = 0;
 251     } else {
 252         /* timestamps were equal */
 253         if ( re->re_seq > ri->ri_stel->seq ) {
 254             /* re seq is newer */
 255             ret = 1;
 256         } else {
 257             ret = 0;
 258         }
 259     }
 260     sglob->st->st_unlock( sglob->st );
 261     return ret;
 262 }