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

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