Makefile.am | 1
Makefile.in | 1
ldap/admin/src/scripts/template-cleanallruv.pl.in | 186 +
ldap/schema/01core389.ldif | 17
ldap/servers/plugins/replication/cl5_api.c | 71
ldap/servers/plugins/replication/cl5_api.h | 9
ldap/servers/plugins/replication/repl5.h | 57
ldap/servers/plugins/replication/repl5_agmt.c | 110
ldap/servers/plugins/replication/repl5_agmtlist.c | 4
ldap/servers/plugins/replication/repl5_connection.c | 13
ldap/servers/plugins/replication/repl5_inc_protocol.c | 21
ldap/servers/plugins/replication/repl5_init.c | 46
ldap/servers/plugins/replication/repl5_plugins.c | 18
ldap/servers/plugins/replication/repl5_replica.c | 477 ++-
ldap/servers/plugins/replication/repl5_replica_config.c | 2252 +++++++++++-----
ldap/servers/plugins/replication/repl5_ruv.c | 58
ldap/servers/plugins/replication/repl5_ruv.h | 1
ldap/servers/plugins/replication/repl_extop.c | 446 +--
ldap/servers/plugins/replication/repl_globals.c | 5
ldap/servers/slapd/log.c | 27
ldap/servers/slapd/slapi-plugin.h | 3
ldap/servers/slapd/task.c | 56
22 files changed, 2839 insertions(+), 1040 deletions(-)
New commits:
commit 657efc69487cc1b04d82642dddd47f93b7a37cf1
Author: Mark Reynolds <mreynolds(a)redhat.com>
Date: Wed Aug 1 16:12:01 2012 -0400
Ticket 403 - CLEANALLRUV feature
This is a rewrite of the first version of CLEANALLRUV. The task is now an official
slapi task, but the
older/original method still works as well. You can run multiple tasks(4) at the same
time. The task can handle
situations where a replica is down, or "older"replicas are in the topology.
I've also added an "abort"
task to cancel a particular cleanallruv task, as the task can potentially run for a
long time while
waiting for a server to come online or be cleaned.
https://fedorahosted.org/389/ticket/403
Reviewed by: richm & noriko (Thanks!)
diff --git a/Makefile.am b/Makefile.am
index d90cd4b..000fef0 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -374,6 +374,7 @@ task_SCRIPTS = ldap/admin/src/scripts/template-bak2db \
ldap/admin/src/scripts/template-db2ldif.pl \
ldap/admin/src/scripts/template-fixup-linkedattrs.pl \
ldap/admin/src/scripts/template-fixup-memberof.pl \
+ ldap/admin/src/scripts/template-cleanallruv.pl \
ldap/admin/src/scripts/template-ldif2db.pl \
ldap/admin/src/scripts/template-ns-accountstatus.pl \
ldap/admin/src/scripts/template-ns-activate.pl \
diff --git a/Makefile.in b/Makefile.in
index 0de0465..3bd6ec6 100644
--- a/Makefile.in
+++ b/Makefile.in
@@ -1595,6 +1595,7 @@ task_SCRIPTS = ldap/admin/src/scripts/template-bak2db \
ldap/admin/src/scripts/template-db2ldif.pl \
ldap/admin/src/scripts/template-fixup-linkedattrs.pl \
ldap/admin/src/scripts/template-fixup-memberof.pl \
+ ldap/admin/src/scripts/template-cleanallruv.pl \
ldap/admin/src/scripts/template-ldif2db.pl \
ldap/admin/src/scripts/template-ns-accountstatus.pl \
ldap/admin/src/scripts/template-ns-activate.pl \
diff --git a/ldap/admin/src/scripts/template-cleanallruv.pl.in
b/ldap/admin/src/scripts/template-cleanallruv.pl.in
new file mode 100644
index 0000000..be95a6d
--- /dev/null
+++ b/ldap/admin/src/scripts/template-cleanallruv.pl.in
@@ -0,0 +1,186 @@
+#{{PERL-EXEC}}
+#
+# BEGIN COPYRIGHT BLOCK
+# This Program is free software; you can redistribute it and/or modify it under
+# the terms of the GNU General Public License as published by the Free Software
+# Foundation; version 2 of the License.
+#
+# This Program is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License along with
+# this Program; if not, write to the Free Software Foundation, Inc., 59 Temple
+# Place, Suite 330, Boston, MA 02111-1307 USA.
+#
+# In addition, as a special exception, Red Hat, Inc. gives You the additional
+# right to link the code of this Program with code not covered under the GNU
+# General Public License ("Non-GPL Code") and to distribute linked
combinations
+# including the two, subject to the limitations in this paragraph. Non-GPL Code
+# permitted under this exception must only link to the code of this Program
+# through those well defined interfaces identified in the file named EXCEPTION
+# found in the source code files (the "Approved Interfaces"). The files of
+# Non-GPL Code may instantiate templates or use macros or inline functions from
+# the Approved Interfaces without causing the resulting work to be covered by
+# the GNU General Public License. Only Red Hat, Inc. may make changes or
+# additions to the list of Approved Interfaces. You must obey the GNU General
+# Public License in all respects for all of the Program code and other code used
+# in conjunction with the Program except the Non-GPL Code covered by this
+# exception. If you modify this file, you may extend this exception to your
+# version of the file, but you are not obligated to do so. If you do not wish to
+# provide this exception without modification, you must delete this exception
+# statement from your version and license this file solely under the GPL without
+# exception.
+#
+#
+# Copyright (C) 2001 Sun Microsystems, Inc. Used by permission.
+# Copyright (C) 2012 Red Hat, Inc.
+# All rights reserved.
+# END COPYRIGHT BLOCK
+#
+
+sub usage {
+ print(STDERR "Usage: $0 [-v] -D rootdn { -w password | -w - | -j filename }
\n");
+ print(STDERR " [-b basedn | -r rid | -A]\n");
+ print(STDERR " Opts: -D rootdn - Directory Manager\n");
+ print(STDERR " : -w password - Directory Manager's
password\n");
+ print(STDERR " : -w - - Prompt for Directory Manager's
password\n");
+ print(STDERR " : -j filename - Read Directory Manager's password
from file\n");
+ print(STDERR " : -b basedn - DN of the replica root you want to
clean\n");
+ print(STDERR " : -r rid - The replica id that you want to
clean\n");
+ print(STDERR " : -A - Abort an existing cleanallruv
task(must use with -b and -r args\n");
+ print(STDERR " : -v - verbose\n");
+}
+
+$rootdn = "";
+$passwd = "";
+$passwdfile = "";
+$basedn = "";
+$rid = "";
+$abort = "";
+$verbose = 0;
+
+$prefix = "{{DS-ROOT}}";
+
+$ENV{'PATH'} =
"$prefix@ldaptool_bindir@:$prefix/usr/bin:@ldaptool_bindir@:/usr/bin";
+
+libpath_add("$prefix@nss_libdir@");
+libpath_add("$prefix/usr/lib");
+libpath_add("@nss_libdir@");
+libpath_add("/usr/lib");
+
+$ENV{'SHLIB_PATH'} = "$ENV{'LD_LIBRARY_PATH'}";
+
+$i = 0;
+while ($i <= $#ARGV)
+{
+ if ("$ARGV[$i]" eq "-b")
+ {
+ # Base DN
+ $i++; $basedn = $ARGV[$i];
+ }
+ elsif ("$ARGV[$i]" eq "-r")
+ {
+ # rid
+ $i++; $rid = $ARGV[$i];
+ }
+ elsif ("$ARGV[$i]" eq "-A")
+ {
+ # abort
+ $abort = "yes";
+ }
+ elsif ("$ARGV[$i]" eq "-D")
+ {
+ # Directory Manager
+ $i++; $rootdn = $ARGV[$i];
+ }
+ elsif ("$ARGV[$i]" eq "-w")
+ {
+ # Directory Manager's password
+ $i++; $passwd = $ARGV[$i];
+ }
+ elsif ("$ARGV[$i]" eq "-j")
+ {
+ # Read Directory Manager's password from a file
+ $i++; $passwdfile = $ARGV[$i];
+ }
+ elsif ("$ARGV[$i]" eq "-v")
+ {
+ # verbose
+ $verbose = 1;
+ }
+ else
+ {
+ &usage; exit(1);
+ }
+ $i++;
+}
+
+if ($passwdfile ne ""){
+# Open file and get the password
+ unless (open (RPASS, $passwdfile)) {
+ die "Error, cannot open password file $passwdfile\n";
+ }
+ $passwd = <RPASS>;
+ chomp($passwd);
+ close(RPASS);
+} elsif ($passwd eq "-"){
+# Read the password from terminal
+ print "Bind Password: ";
+ # Disable console echo
+ system("@sttyexec@ -echo") if -t STDIN;
+ # read the answer
+ $passwd = <STDIN>;
+ # Enable console echo
+ system("@sttyexec@ echo") if -t STDIN;
+ print "\n";
+ chop($passwd); # trim trailing newline
+}
+
+if ( $rootdn eq "" || $passwd eq "" || $basedn eq "" ||
$rid eq "")
+{
+ &usage;
+ exit(1);
+}
+
+$vstr = "";
+if ($verbose != 0)
+{
+ $vstr = "-v";
+}
+
+# Use a timestamp as part of the task entry name
+($s, $m, $h, $dy, $mn, $yr, $wdy, $ydy, $r) = localtime(time);
+$mn++; $yr += 1900;
+
+if($abort eq ""){
+ # Build the task entry to add
+ $taskname = "cleanallruv_${yr}_${mn}_${dy}_${h}_${m}_${s}";
+ $dn = "dn: cn=$taskname, cn=cleanallruv, cn=tasks, cn=config\n";
+} else {
+ $taskname = "abort_cleanallruv_${yr}_${mn}_${dy}_${h}_${m}_${s}";
+ $dn = "dn: cn=$taskname, cn=abort cleanallruv, cn=tasks, cn=config\n";
+}
+$misc = "changetype: add\nobjectclass: top\nobjectclass:
extensibleObject\n";
+$cn = "cn: $taskname\n";
+$basedn = "replica-base-dn: $basedn\n";
+$rid = "replica-id: $rid\n";
+
+
+$entry = "${dn}${misc}${cn}${basedn}${rid}";
+open(FOO, "| ldapmodify @ldaptool_opts@ $vstr -h {{SERVER-NAME}} -p {{SERVER-PORT}}
-D \"$rootdn\" -w \"$passwd\" -a" );
+print(FOO "$entry");
+close(FOO);
+
+sub libpath_add {
+ my $libpath = shift;
+
+ if ($libpath) {
+ if ($ENV{'LD_LIBRARY_PATH'}) {
+ $ENV{'LD_LIBRARY_PATH'} =
"$ENV{'LD_LIBRARY_PATH'}:$libpath";
+ } else {
+ $ENV{'LD_LIBRARY_PATH'} = "$libpath";
+ }
+ }
+}
+
diff --git a/ldap/schema/01core389.ldif b/ldap/schema/01core389.ldif
index def693d..58b4b76 100644
--- a/ldap/schema/01core389.ldif
+++ b/ldap/schema/01core389.ldif
@@ -134,15 +134,10 @@ attributeTypes: ( 2.16.840.1.113730.3.1.2096 NAME 'entryusn'
DESC 'Netscape defi
attributeTypes: ( 2.16.840.1.113730.3.1.2113 NAME 'internalModifiersName' DESC
'plugin dn' SYNTAX 1.3.6.1.4.1.1466.115.121.1.12 SINGLE-VALUE
NO-USER-MODIFICATION USAGE directoryOperation X-ORIGIN '389 Directory Server' )
attributeTypes: ( 2.16.840.1.113730.3.1.2114 NAME 'internalCreatorsName' DESC
'plugin dn' SYNTAX 1.3.6.1.4.1.1466.115.121.1.12 SINGLE-VALUE
NO-USER-MODIFICATION USAGE directoryOperation X-ORIGIN '389 Directory Server' )
attributeTypes: ( 2.16.840.1.113730.3.1.2133 NAME 'pwdUpdateTime' DESC 'Last
password update time' SYNTAX 1.3.6.1.4.1.1466.115.121.1.24 SINGLE-VALUE USAGE
directoryOperation X-ORIGIN '389 Directory Server' )
-attributeTypes: ( 2.16.840.1.113730.3.1.2111 NAME 'tombstoneNumSubordinates'
- DESC 'count of immediate subordinates for tombstone entries'
- EQUALITY integerMatch
- ORDERING integerOrderingMatch
- SYNTAX 1.3.6.1.4.1.1466.115.121.1.27
- SINGLE-VALUE
- NO-USER-MODIFICATION
- USAGE directoryOperation
- X-ORIGIN '389 directory server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2135 NAME 'nsds5ReplicaCleanRUV' DESC
'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 X-ORIGIN
'Netscape Directory Server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2136 NAME 'nsds5ReplicaCleanRUVNotified'
DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.27
X-ORIGIN 'Netscape Directory Server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2137 NAME 'nsds5ReplicaAbortCleanRUV'
DESC 'Netscape defined attribute type' SYNTAX 1.3.6.1.4.1.1466.115.121.1.15
X-ORIGIN 'Netscape Directory Server' )
+attributeTypes: ( 2.16.840.1.113730.3.1.2111 NAME 'tombstoneNumSubordinates' DESC
'count of immediate subordinates for tombstone entries' EQUALITY integerMatch
ORDERING integerOrderingMatch SYNTAX 1.3.6.1.4.1.1466.115.121.1.27 SINGLE-VALUE
NO-USER-MODIFICATION USAGE directoryOperation X-ORIGIN '389 directory server' )
#
# objectclasses
#
@@ -152,9 +147,9 @@ objectClasses: ( 2.16.840.1.113730.3.2.44 NAME 'nsIndex' DESC
'Netscape defined
objectClasses: ( 2.16.840.1.113730.3.2.109 NAME 'nsBackendInstance' DESC
'Netscape defined objectclass' SUP top MUST ( CN ) X-ORIGIN 'Netscape
Directory Server' )
objectClasses: ( 2.16.840.1.113730.3.2.110 NAME 'nsMappingTree' DESC
'Netscape defined objectclass' SUP top MUST ( CN ) X-ORIGIN 'Netscape
Directory Server' )
objectClasses: ( 2.16.840.1.113730.3.2.104 NAME 'nsContainer' DESC 'Netscape
defined objectclass' SUP top MUST ( CN ) X-ORIGIN 'Netscape Directory Server'
)
-objectClasses: ( 2.16.840.1.113730.3.2.108 NAME 'nsDS5Replica' DESC 'Netscape
defined objectclass' SUP top MUST ( nsDS5ReplicaRoot $ nsDS5ReplicaId ) MAY (cn $
nsDS5ReplicaType $ nsDS5ReplicaBindDN $ nsState $ nsDS5ReplicaName $ nsDS5Flags $
nsDS5Task $ nsDS5ReplicaReferral $ nsDS5ReplicaAutoReferral $ nsds5ReplicaPurgeDelay $
nsds5ReplicaTombstonePurgeInterval $ nsds5ReplicaChangeCount $ nsds5ReplicaLegacyConsumer)
X-ORIGIN 'Netscape Directory Server' )
+objectClasses: ( 2.16.840.1.113730.3.2.108 NAME 'nsDS5Replica' DESC 'Netscape
defined objectclass' SUP top MUST ( nsDS5ReplicaRoot $ nsDS5ReplicaId ) MAY (cn $
nsds5ReplicaCleanRUV $ nsds5ReplicaAbortCleanRUV $ nsDS5ReplicaType $ nsDS5ReplicaBindDN $
nsState $ nsDS5ReplicaName $ nsDS5Flags $ nsDS5Task $ nsDS5ReplicaReferral $
nsDS5ReplicaAutoReferral $ nsds5ReplicaPurgeDelay $ nsds5ReplicaTombstonePurgeInterval $
nsds5ReplicaChangeCount $ nsds5ReplicaLegacyConsumer) X-ORIGIN 'Netscape Directory
Server' )
objectClasses: ( 2.16.840.1.113730.3.2.113 NAME 'nsTombstone' DESC 'Netscape
defined objectclass' SUP top MAY ( nsParentUniqueId $ nscpEntryDN ) X-ORIGIN
'Netscape Directory Server' )
-objectClasses: ( 2.16.840.1.113730.3.2.103 NAME 'nsDS5ReplicationAgreement' DESC
'Netscape defined objectclass' SUP top MUST ( cn ) MAY ( nsDS5ReplicaHost $
nsDS5ReplicaPort $ nsDS5ReplicaTransportInfo $ nsDS5ReplicaBindDN $
nsDS5ReplicaCredentials $ nsDS5ReplicaBindMethod $ nsDS5ReplicaRoot $
nsDS5ReplicatedAttributeList $ nsDS5ReplicatedAttributeListTotal $
nsDS5ReplicaUpdateSchedule $ nsds5BeginReplicaRefresh $ description $ nsds50ruv $
nsruvReplicaLastModified $ nsds5ReplicaTimeout $ nsds5replicaChangesSentSinceStartup $
nsds5replicaLastUpdateEnd $ nsds5replicaLastUpdateStart $ nsds5replicaLastUpdateStatus $
nsds5replicaUpdateInProgress $ nsds5replicaLastInitEnd $ nsds5ReplicaEnabled $
nsds5replicaLastInitStart $ nsds5replicaLastInitStatus $ nsds5debugreplicatimeout $
nsds5replicaBusyWaitTime $ nsds5ReplicaStripAttrs $ nsds5replicaSessionPauseTime )
X-ORIGIN 'Netscape Directory Server' )
+objectClasses: ( 2.16.840.1.113730.3.2.103 NAME 'nsDS5ReplicationAgreement' DESC
'Netscape defined objectclass' SUP top MUST ( cn ) MAY (
nsds5ReplicaCleanRUVNotified $ nsDS5ReplicaHost $ nsDS5ReplicaPort $
nsDS5ReplicaTransportInfo $ nsDS5ReplicaBindDN $ nsDS5ReplicaCredentials $
nsDS5ReplicaBindMethod $ nsDS5ReplicaRoot $ nsDS5ReplicatedAttributeList $
nsDS5ReplicatedAttributeListTotal $ nsDS5ReplicaUpdateSchedule $ nsds5BeginReplicaRefresh
$ description $ nsds50ruv $ nsruvReplicaLastModified $ nsds5ReplicaTimeout $
nsds5replicaChangesSentSinceStartup $ nsds5replicaLastUpdateEnd $
nsds5replicaLastUpdateStart $ nsds5replicaLastUpdateStatus $ nsds5replicaUpdateInProgress
$ nsds5replicaLastInitEnd $ nsds5ReplicaEnabled $ nsds5replicaLastInitStart $
nsds5replicaLastInitStatus $ nsds5debugreplicatimeout $ nsds5replicaBusyWaitTime $
nsds5ReplicaStripAttrs $ nsds5replicaSessionPauseTime ) X-ORIGIN 'Netscape Directory
Server' )
objectClasses: ( 2.16.840.1.113730.3.2.39 NAME 'nsslapdConfig' DESC 'Netscape
defined objectclass' SUP top MAY ( cn ) X-ORIGIN 'Netscape Directory Server'
)
objectClasses: ( 2.16.840.1.113730.3.2.317 NAME 'nsSaslMapping' DESC
'Netscape defined objectclass' SUP top MUST ( cn $ nsSaslMapRegexString $
nsSaslMapBaseDNTemplate $ nsSaslMapFilterTemplate ) X-ORIGIN 'Netscape Directory
Server' )
objectClasses: ( 2.16.840.1.113730.3.2.43 NAME 'nsSNMP' DESC 'Netscape
defined objectclass' SUP top MUST ( cn $ nsSNMPEnabled ) MAY ( nsSNMPOrganization $
nsSNMPLocation $ nsSNMPContact $ nsSNMPDescription $ nsSNMPName $ nsSNMPMasterHost $
nsSNMPMasterPort ) X-ORIGIN 'Netscape Directory Server' )
diff --git a/ldap/servers/plugins/replication/cl5_api.c
b/ldap/servers/plugins/replication/cl5_api.c
index 08bf771..a347c57 100644
--- a/ldap/servers/plugins/replication/cl5_api.c
+++ b/ldap/servers/plugins/replication/cl5_api.c
@@ -345,15 +345,15 @@ static int _cl5CheckMissingCSN (const CSN *minCsn, const RUV
*supplierRUV, CL5DB
static int _cl5TrimInit ();
static void _cl5TrimCleanup ();
static int _cl5TrimMain (void *param);
-static void _cl5DoTrimming ();
-static void _cl5TrimFile (Object *obj, long *numToTrim);
+static void _cl5DoTrimming (ReplicaId rid);
+static void _cl5TrimFile (Object *obj, long *numToTrim, ReplicaId cleaned_rid);
static PRBool _cl5CanTrim (time_t time, long *numToTrim);
static int _cl5ReadRUV (const char *replGen, Object *obj, PRBool purge);
static int _cl5WriteRUV (CL5DBFile *file, PRBool purge);
static int _cl5ConstructRUV (const char *replGen, Object *obj, PRBool purge);
static int _cl5UpdateRUV (Object *obj, CSN *csn, PRBool newReplica, PRBool purge);
static int _cl5GetRUV2Purge2 (Object *fileObj, RUV **ruv);
-void trigger_cl_trimming_thread();
+void trigger_cl_trimming_thread(void *rid);
/* bakup/recovery, import/export */
static int _cl5LDIF2Operation (char *ldifEntry, slapi_operation_parameters *op,
@@ -699,12 +699,12 @@ int cl5DeleteDBSync (Object *replica)
}
/* Name: cl5GetUpperBoundRUV
- Description: retrieves vector for that represnts the upper bound of the changes for a
replica.
+ Description: retrieves vector for that represents the upper bound of the changes for a
replica.
Parameters: r - replica for which the purge vector is requested
ruv - contains a copy of the purge ruv if function is successful;
- unchanged otherwise. It is responsobility pf the caller to free
+ unchanged otherwise. It is responsibility of the caller to free
the ruv when it is no longer is in use
- Return: CL5_SUCCESS if function is successfull
+ Return: CL5_SUCCESS if function is successful
CL5_BAD_STATE if the changelog is not initialized;
CL5_BAD_DATA - if NULL id is supplied
CL5_NOTFOUND, if changelog file for replica is not found
@@ -1682,6 +1682,16 @@ cl5GetNextOperationToReplay (CL5ReplayIterator *iterator, CL5Entry
*entry)
return CL5_DB_ERROR;
}
+ if(is_cleaned_rid(csn_get_replicaid(csn))){
+ /*
+ * This operation is from a deleted replica. During the cleanallruv task the
+ * replicas are cleaned first before this instance is. This can cause the
+ * server to basically do a full update over and over. So we have to watch for
+ * this, and not send these operations out.
+ */
+ return CL5_IGNORE_OP;
+ }
+
/* there is an entry we should return */
/* Callers of this function should cl5_operation_parameters_done(op) */
if ( 0 != cl5DBData2Entry ( data, datalen, entry ) )
@@ -3383,7 +3393,7 @@ static int _cl5TrimMain (void *param)
{
/* time to trim */
timePrev = timeNow;
- _cl5DoTrimming ();
+ _cl5DoTrimming (0 /* there's no cleaned rid */);
}
if (NULL == s_cl5Desc.clLock)
{
@@ -3417,7 +3427,7 @@ static int _cl5TrimMain (void *param)
*/
-static void _cl5DoTrimming ()
+static void _cl5DoTrimming (ReplicaId rid)
{
Object *obj;
long numToTrim;
@@ -3430,7 +3440,7 @@ static void _cl5DoTrimming ()
obj = objset_first_obj (s_cl5Desc.dbFiles);
while (obj && _cl5CanTrim ((time_t)0, &numToTrim))
{
- _cl5TrimFile (obj, &numToTrim);
+ _cl5TrimFile (obj, &numToTrim, rid);
obj = objset_next_obj (s_cl5Desc.dbFiles, obj);
}
@@ -3447,12 +3457,13 @@ static void _cl5DoTrimming ()
*/
#define CL5_TRIM_MAX_PER_TRANSACTION 10
-static void _cl5TrimFile (Object *obj, long *numToTrim)
+static void _cl5TrimFile (Object *obj, long *numToTrim, ReplicaId cleaned_rid)
{
DB_TXN *txnid;
RUV *ruv = NULL;
CL5Entry entry;
slapi_operation_parameters op = {0};
+ ReplicaId csn_rid;
void *it;
int finished = 0, totalTrimmed = 0, count;
PRBool abort;
@@ -3476,7 +3487,6 @@ static void _cl5TrimFile (Object *obj, long *numToTrim)
count = 0;
txnid = NULL;
abort = PR_FALSE;
- ReplicaId rid;
/* DB txn lock accessed pages until the end of the transaction. */
@@ -3497,14 +3507,13 @@ static void _cl5TrimFile (Object *obj, long *numToTrim)
* This change can be trimmed if it exceeds purge
* parameters and has been seen by all consumers.
*/
- rid = csn_get_replicaid (op.csn);
+ csn_rid = csn_get_replicaid (op.csn);
if ( (*numToTrim > 0 || _cl5CanTrim (entry.time, numToTrim)) &&
ruv_covers_csn_strict (ruv, op.csn) )
{
rc = _cl5CurrentDeleteEntry (it);
- if ( rc == CL5_SUCCESS && !is_released_rid(rid))
+ if ( rc == CL5_SUCCESS && cleaned_rid != csn_rid)
{
- /* update purge vector, unless this is a released rid */
rc = _cl5UpdateRUV (obj, op.csn, PR_FALSE, PR_TRUE);
}
if ( rc == CL5_SUCCESS)
@@ -3530,8 +3539,7 @@ static void _cl5TrimFile (Object *obj, long *numToTrim)
* the trim forever.
*/
CSN *maxcsn = NULL;
- rid = csn_get_replicaid (op.csn);
- ruv_get_largest_csn_for_replica (ruv, rid, &maxcsn);
+ ruv_get_largest_csn_for_replica (ruv, csn_rid, &maxcsn);
if ( csn_compare (op.csn, maxcsn) != 0 )
{
/* op.csn is not anchor CSN */
@@ -3895,20 +3903,18 @@ static int _cl5UpdateRUV (Object *obj, CSN *csn, PRBool
newReplica, PRBool purge
/* update vector only if this replica is not yet part of RUV */
if (purge && newReplica)
{
- if (ruv_contains_replica (file->purgeRUV, rid))
+ if (ruv_contains_replica (file->purgeRUV, rid)){
return CL5_SUCCESS;
- else if(!is_cleaned_rid(rid))
- {
+ } else {
/* if the replica is not part of the purgeRUV yet, add it unless it's
from a cleaned rid */
ruv_add_replica (file->purgeRUV, rid, multimaster_get_local_purl());
}
}
else
{
- if (purge)
+ if (purge){
rc = ruv_set_csns(file->purgeRUV, csn, NULL);
- else if(!is_cleaned_rid(rid)){
- /* don't update maxRuv if rid is cleaned */
+ } else {
rc = ruv_set_csns(file->maxRUV, csn, NULL);
}
}
@@ -4520,17 +4526,9 @@ static int _cl5WriteOperationTxn(const char *replName, const char
*replGen,
CL5Entry entry;
CL5DBFile *file = NULL;
Object *file_obj = NULL;
- ReplicaId rid = csn_get_replicaid (op->csn);
DB_TXN *txnid = NULL;
DB_TXN *parent_txnid = (DB_TXN *)txn;
- /*
- * If the op csn contains the cleaned rid, don't write it
- */
- if(is_cleaned_rid(rid)){
- return CL5_SUCCESS;
- }
-
rc = _cl5GetDBFileByReplicaName (replName, replGen, &file_obj);
if (rc == CL5_NOTFOUND)
{
@@ -5089,7 +5087,7 @@ static int _cl5PositionCursorForReplay (ReplicaId consumerRID, const
RUV *consum
* legacy consumer. In this case the supplier
* and the consumer may have the same RID.
*/
- if (rid == consumerRID && rid != MAX_REPLICA_ID)
+ if ((rid == consumerRID && rid != MAX_REPLICA_ID) || (is_cleaned_rid(rid)) )
continue;
startCSN = csns[i];
@@ -6556,13 +6554,12 @@ cl5CleanRUV(ReplicaId rid){
}
}
-void trigger_cl_trimming(){
+void trigger_cl_trimming(ReplicaId rid){
PRThread *trim_tid = NULL;
- ReplicaId rid = get_released_rid();
slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name_cl, "trigger_cl_trimming: rid
(%d)\n",(int)rid);
trim_tid = PR_CreateThread(PR_USER_THREAD, (VFP)(void*)trigger_cl_trimming_thread,
- NULL, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ (void *)&rid, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
PR_UNJOINABLE_THREAD, DEFAULT_THREAD_STACKSIZE);
if (NULL == trim_tid){
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name_cl,
@@ -6575,7 +6572,9 @@ void trigger_cl_trimming(){
}
void
-trigger_cl_trimming_thread(){
+trigger_cl_trimming_thread(void *arg){
+ ReplicaId rid = *(ReplicaId *)arg;
+
/* make sure we have a change log, and we aren't closing it */
if(s_cl5Desc.dbState == CL5_STATE_CLOSED || s_cl5Desc.dbState == CL5_STATE_CLOSING){
return;
@@ -6585,6 +6584,6 @@ trigger_cl_trimming_thread(){
"trigger_cl_trimming: failed to increment thread count "
"NSPR error - %d\n", PR_GetError ());
}
- _cl5DoTrimming();
+ _cl5DoTrimming(rid);
_cl5RemoveThread();
}
diff --git a/ldap/servers/plugins/replication/cl5_api.h
b/ldap/servers/plugins/replication/cl5_api.h
index b9c3dd8..9b285ca 100644
--- a/ldap/servers/plugins/replication/cl5_api.h
+++ b/ldap/servers/plugins/replication/cl5_api.h
@@ -145,9 +145,10 @@ enum
CL5_CSN_ERROR, /* CSN API failed */
CL5_RUV_ERROR, /* RUV API failed */
CL5_OBJSET_ERROR, /* namedobjset api failed */
- CL5_PURGED_DATA, /* requested data has been purged */
- CL5_MISSING_DATA, /* data should be in the changelog, but is missing */
- CL5_UNKNOWN_ERROR /* unclassified error */
+ CL5_PURGED_DATA, /* requested data has been purged */
+ CL5_MISSING_DATA, /* data should be in the changelog, but is missing */
+ CL5_UNKNOWN_ERROR, /* unclassified error */
+ CL5_IGNORE_OP /* ignore this updated - used by CLEANALLRUV task */
};
/***** Module APIs *****/
@@ -489,6 +490,6 @@ int cl5WriteRUV();
int cl5DeleteRUV();
void cl5CleanRUV(ReplicaId rid);
void cl5NotifyCleanup(int rid);
-void trigger_cl_trimming();
+void trigger_cl_trimming(ReplicaId rid);
#endif
diff --git a/ldap/servers/plugins/replication/repl5.h
b/ldap/servers/plugins/replication/repl5.h
index d7ba6a6..fc756b1 100644
--- a/ldap/servers/plugins/replication/repl5.h
+++ b/ldap/servers/plugins/replication/repl5.h
@@ -94,10 +94,11 @@
* new set of start and response extops. */
#define REPL_START_NSDS90_REPLICATION_REQUEST_OID "2.16.840.1.113730.3.5.12"
#define REPL_NSDS90_REPLICATION_RESPONSE_OID "2.16.840.1.113730.3.5.13"
-/* cleanruv/releaseruv extended ops */
+/* cleanallruv extended ops */
#define REPL_CLEANRUV_OID "2.16.840.1.113730.3.6.5"
-#define REPL_RELEASERUV_OID "2.16.840.1.113730.3.6.6"
-
+#define REPL_ABORT_CLEANRUV_OID "2.16.840.1.113730.3.6.6"
+#define CLEANRUV_NOTIFIED 0
+#define CLEANRUV_RELEASED 1
/* DS 5.0 replication protocol error codes */
#define NSDS50_REPL_REPLICA_READY 0x00 /* Replica ready, go ahead */
@@ -157,6 +158,7 @@ extern const char *type_nsds5ReplicaBusyWaitTime;
extern const char *type_nsds5ReplicaSessionPauseTime;
extern const char *type_nsds5ReplicaEnabled;
extern const char *type_nsds5ReplicaStripAttrs;
+extern const char *type_nsds5ReplicaCleanRUVnotified;
/* Attribute names for windows replication agreements */
extern const char *type_nsds7WindowsReplicaArea;
@@ -185,6 +187,8 @@ extern const char *type_replicaPurgeDelay;
extern const char *type_replicaChangeCount;
extern const char *type_replicaTombstonePurgeInterval;
extern const char *type_replicaLegacyConsumer;
+extern const char *type_replicaCleanRUV;
+extern const char *type_replicaAbortCleanRUV;
extern const char *type_ruvElementUpdatetime;
/* multimaster plugin points */
@@ -224,7 +228,7 @@ char* get_repl_session_id (Slapi_PBlock *pb, char *id, CSN **opcsn);
int multimaster_extop_StartNSDS50ReplicationRequest(Slapi_PBlock *pb);
int multimaster_extop_EndNSDS50ReplicationRequest(Slapi_PBlock *pb);
int multimaster_extop_cleanruv(Slapi_PBlock *pb);
-int multimaster_extop_releaseruv(Slapi_PBlock *pb);
+int multimaster_extop_abort_cleanruv(Slapi_PBlock *pb);
int extop_noop(Slapi_PBlock *pb);
struct berval *NSDS50StartReplicationRequest_new(const char *protocol_oid,
const char *repl_root, char **extra_referrals, CSN *csn);
@@ -359,6 +363,10 @@ PRBool agmt_is_enabled(Repl_Agmt *ra);
int agmt_set_enabled_from_entry(Repl_Agmt *ra, Slapi_Entry *e);
char **agmt_get_attrs_to_strip(Repl_Agmt *ra);
int agmt_set_attrs_to_strip(Repl_Agmt *ra, Slapi_Entry *e);
+void agmt_set_cleanruv_notified_from_entry(Repl_Agmt *ra, Slapi_Entry *e);
+int agmt_set_cleanruv_data(Repl_Agmt *ra, ReplicaId rid, int op);
+int agmt_is_cleanruv_notified(Repl_Agmt *ra, ReplicaId rid);
+int agmt_set_timeout(Repl_Agmt *ra, long timeout);
typedef struct replica Replica;
@@ -369,7 +377,6 @@ void agmtlist_notify_all(Slapi_PBlock *pb);
Object* agmtlist_get_first_agreement_for_replica (Replica *r);
Object* agmtlist_get_next_agreement_for_replica (Replica *r, Object *prev);
-
/* In repl5_backoff.c */
typedef struct backoff_timer Backoff_Timer;
#define BACKOFF_FIXED 1
@@ -443,6 +450,7 @@ ConnResult conn_read_result_ex(Repl_Connection *conn, char **retoidp,
struct ber
LDAP * conn_get_ldap(Repl_Connection *conn);
void conn_lock(Repl_Connection *conn);
void conn_unlock(Repl_Connection *conn);
+void conn_delete_internal_ext(Repl_Connection *conn);
/* In repl5_protocol.c */
typedef struct repl_protocol Repl_Protocol;
@@ -554,6 +562,9 @@ void replica_set_tombstone_reap_interval (Replica *r, long interval);
void replica_update_ruv_consumer (Replica *r, RUV *supplier_ruv);
void replica_set_ruv_dirty (Replica *r);
void replica_write_ruv (Replica *r);
+char *replica_get_dn(Replica *r);
+void replica_check_for_tasks(Replica*r, Slapi_Entry *e);
+
/* The functions below handles the state flag */
/* Current internal state flags */
/* The replica can be busy and not other flag,
@@ -594,22 +605,40 @@ int replica_config_init();
void replica_config_destroy ();
int get_replica_type(Replica *r);
int replica_execute_cleanruv_task_ext(Object *r, ReplicaId rid);
-void set_cleaned_rid(ReplicaId rid);
-void delete_cleaned_rid();
+void add_cleaned_rid(ReplicaId rid, Replica *r, char *maxcsn);
int is_cleaned_rid(ReplicaId rid);
-int get_released_rid();
-void set_released_rid(int rid);
-int is_released_rid(int rid);
-int is_already_released_rid();
-void delete_released_rid();
-void replica_cleanallruv_monitor_thread(void *arg);
+int replica_cleanall_ruv_abort(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *eAfter,
+ int *returncode, char *returntext, void *arg);
+void replica_cleanallruv_thread_ext(void *arg);
+void stop_ruv_cleaning();
+int task_aborted();
+void replica_abort_task_thread(void *arg);
+void delete_cleaned_rid(Replica *r, ReplicaId rid, CSN *maxcsn);
+int process_repl_agmts(Replica *replica, int *agmt_info, char *oid, Slapi_Task *task,
struct berval *payload, int op);
+int decode_cleanruv_payload(struct berval *extop_value, char **payload);
+struct berval *create_ruv_payload(char *value);
+void replica_add_cleanruv_data(Replica *r, char *val);
+void replica_remove_cleanruv_data(Replica *r, char *val);
+CSN *replica_get_cleanruv_maxcsn(Replica *r, ReplicaId rid);
+void ruv_get_cleaned_rids(RUV *ruv, ReplicaId *rids);
+void add_aborted_rid(ReplicaId rid, Replica *r, char *repl_root);
+int is_task_aborted(ReplicaId rid);
+void delete_aborted_rid(Replica *replica, ReplicaId rid, char *repl_root);
+void set_cleaned_rid(ReplicaId rid);
+void cleanruv_log(Slapi_Task *task, char *task_type, char *fmt, ...);
-#define ALREADY_RELEASED -1
+#define CLEANRIDSIZ 4 /* maximum number for concurrent CLEANALLRUV tasks */
typedef struct _cleanruv_data
{
Object *repl_obj;
+ Replica *replica;
ReplicaId rid;
+ Slapi_Task *task;
+ struct berval *payload;
+ CSN *maxcsn;
+ char *repl_root;
+ Slapi_DN *sdn;
} cleanruv_data;
/* replutil.c */
diff --git a/ldap/servers/plugins/replication/repl5_agmt.c
b/ldap/servers/plugins/replication/repl5_agmt.c
index 2978aba..5a16083 100644
--- a/ldap/servers/plugins/replication/repl5_agmt.c
+++ b/ldap/servers/plugins/replication/repl5_agmt.c
@@ -141,6 +141,7 @@ typedef struct repl5agmt {
char **attrs_to_strip; /* for fractional replication, if a "mod" is empty,
strip out these attributes:
* modifiersname, modifytimestamp, internalModifiersname,
internalModifyTimestamp, etc */
int agreement_type;
+ int cleanruv_notified[CLEANRIDSIZ + 1]; /* specifies if the replica has been notified of
a CLEANALLRUV task */
} repl5agmt;
/* Forward declarations */
@@ -251,6 +252,7 @@ agmt_new_from_entry(Slapi_Entry *e)
Repl_Agmt *ra;
Slapi_Attr *sattr;
char *tmpstr;
+ char **clean_vals = NULL;
char **denied_attrs = NULL;
char *auto_initialize = NULL;
char *val_nsds5BeginReplicaRefresh = "start";
@@ -425,6 +427,19 @@ agmt_new_from_entry(Slapi_Entry *e)
ra->last_init_start_time = 0UL;
ra->last_init_status[0] = '\0';
+ /* cleanruv notification */
+ clean_vals = slapi_entry_attr_get_charray(e, type_nsds5ReplicaCleanRUVnotified);
+ if(clean_vals){
+ int i;
+ for (i = 0; i < CLEANRIDSIZ && clean_vals[i]; i++){
+ ra->cleanruv_notified[i] = atoi(clean_vals[i]);
+ }
+ ra->cleanruv_notified[i + 1] = 0;
+ slapi_ch_array_free(clean_vals);
+ } else {
+ ra->cleanruv_notified[0] = 0;
+ }
+
/* Fractional attributes */
slapi_entry_attr_find(e, type_nsds5ReplicatedAttributeList, &sattr);
@@ -1685,6 +1700,20 @@ agmt_set_timeout_from_entry(Repl_Agmt *ra, const Slapi_Entry *e)
return return_value;
}
+int
+agmt_set_timeout(Repl_Agmt *ra, long timeout)
+{
+ PR_Lock(ra->lock);
+ if (ra->stop_in_progress){
+ PR_Unlock(ra->lock);
+ return -1;
+ }
+ ra->timeout = timeout;
+ PR_Unlock(ra->lock);
+
+ return 0;
+}
+
/*
* Set or reset the busywaittime
*
@@ -2571,3 +2600,84 @@ agmt_set_attrs_to_strip(Repl_Agmt *ra, Slapi_Entry *e)
return -1;
}
+
+int
+agmt_is_cleanruv_notified(Repl_Agmt *ra, ReplicaId rid){
+ int notified = 0;
+ int i;
+
+ PR_Lock(ra->lock);
+ for(i = 0; i < CLEANRIDSIZ && ra->cleanruv_notified[i]; i++){
+ if(ra->cleanruv_notified[i] == rid){
+ notified = 1;
+ break;
+ }
+ }
+ PR_Unlock(ra->lock);
+
+ return notified;
+}
+
+/*
+ * This will trigger agmt_set_cleanruv_notified_from_entry() to be called,
+ * which will update the in memory agmt.
+ *
+ * op can be: CLEANRUV_NOTIFIED or CLEANRUV_RELEASED
+ */
+int
+agmt_set_cleanruv_data(Repl_Agmt *ra, ReplicaId rid, int op){
+ Slapi_PBlock *pb;
+ LDAPMod *mods[2];
+ LDAPMod mod;
+ struct berval *vals[2];
+ struct berval val;
+ char data[6];
+ int rc = 0;
+
+ if(ra == NULL){
+ return -1;
+ }
+
+ if(op == CLEANRUV_NOTIFIED){
+ /* add the cleanruv data */
+ mod.mod_op = LDAP_MOD_ADD|LDAP_MOD_BVALUES;
+ } else {
+ /* remove the cleanruv data */
+ mod.mod_op = LDAP_MOD_DELETE|LDAP_MOD_BVALUES;
+ }
+
+ pb = slapi_pblock_new();
+ val.bv_len = PR_snprintf(data, sizeof(data), "%d", (int)rid);
+ mod.mod_type = (char *)type_nsds5ReplicaCleanRUVnotified;
+ mod.mod_bvalues = vals;
+ vals [0] = &val;
+ vals [1] = NULL;
+ val.bv_val = data;
+ mods[0] = &mod;
+ mods[1] = NULL;
+
+ slapi_modify_internal_set_pb_ext (pb, ra->dn, mods, NULL, NULL,
+ repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), 0);
+ slapi_modify_internal_pb (pb);
+ slapi_pblock_destroy(pb);
+
+ return rc;
+}
+
+void
+agmt_set_cleanruv_notified_from_entry(Repl_Agmt *ra, Slapi_Entry *e){
+ char **attr_vals = NULL;
+ int i;
+
+ PR_Lock(ra->lock);
+ attr_vals = slapi_entry_attr_get_charray(e, type_nsds5ReplicaCleanRUVnotified);
+ if(attr_vals){
+ for (i = 0; i < CLEANRIDSIZ && attr_vals[i]; i++){
+ ra->cleanruv_notified[i] = atoi(attr_vals[i]);
+ }
+ ra->cleanruv_notified[i + 1] = 0;
+ } else {
+ ra->cleanruv_notified[0] = 0;
+ }
+ PR_Unlock(ra->lock);
+}
diff --git a/ldap/servers/plugins/replication/repl5_agmtlist.c
b/ldap/servers/plugins/replication/repl5_agmtlist.c
index 86f06bf..c0b9664 100644
--- a/ldap/servers/plugins/replication/repl5_agmtlist.c
+++ b/ldap/servers/plugins/replication/repl5_agmtlist.c
@@ -508,6 +508,10 @@ agmtlist_modify_callback(Slapi_PBlock *pb, Slapi_Entry *entryBefore,
Slapi_Entry
rc = SLAPI_DSE_CALLBACK_ERROR;
}
}
+ else if (slapi_attr_types_equivalent(mods[i]->mod_type,
type_nsds5ReplicaCleanRUVnotified))
+ {
+ agmt_set_cleanruv_notified_from_entry(agmt, e);
+ }
else if (0 == windows_handle_modify_agreement(agmt, mods[i]->mod_type, e))
{
slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
"agmtlist_modify_callback: "
diff --git a/ldap/servers/plugins/replication/repl5_connection.c
b/ldap/servers/plugins/replication/repl5_connection.c
index 5fdc601..51a2bc5 100644
--- a/ldap/servers/plugins/replication/repl5_connection.c
+++ b/ldap/servers/plugins/replication/repl5_connection.c
@@ -140,7 +140,7 @@ static void repl5_debug_timeout_callback(time_t when, void *arg);
static void close_connection_internal(Repl_Connection *conn);
/*
- * Create a new conenction object. Returns a pointer to the object, or
+ * Create a new connection object. Returns a pointer to the object, or
* NULL if an error occurs.
*/
Repl_Connection *
@@ -217,6 +217,17 @@ conn_delete_internal(Repl_Connection *conn)
}
/*
+ * Used by CLEANALLRUV - free it all!
+ */
+void
+conn_delete_internal_ext(Repl_Connection *conn)
+{
+ conn_delete_internal(conn);
+ PR_DestroyLock(conn->lock);
+ slapi_ch_free((void **)&conn);
+}
+
+/*
* Destroy a connection. It is an error to use the connection object
* after conn_delete() has been called.
*/
diff --git a/ldap/servers/plugins/replication/repl5_inc_protocol.c
b/ldap/servers/plugins/replication/repl5_inc_protocol.c
index 7388d62..025c23f 100644
--- a/ldap/servers/plugins/replication/repl5_inc_protocol.c
+++ b/ldap/servers/plugins/replication/repl5_inc_protocol.c
@@ -466,7 +466,7 @@ repl5_inc_waitfor_async_results(result_data *rd)
int done = 0;
int loops = 0;
/* Keep pulling results off the LDAP connection until we catch up to the last message id
stored in the rd */
- while (!done)
+ while (!done && !slapi_is_shutting_down())
{
/* Lock the structure to force memory barrier */
PR_Lock(rd->lock);
@@ -1473,10 +1473,10 @@ repl5_inc_update_from_op_result(Private_Repl_Protocol *prp,
ConnResult replay_cr
agmt_inc_last_update_changecount (prp->agmt, replica_id, 1 /*skipped*/);
}
slapi_log_error(*finished ? SLAPI_LOG_FATAL : slapi_log_urp, repl_plugin_name,
- "%s: Consumer failed to replay change (uniqueid %s, CSN %s): %s.
%s.\n",
+ "%s: Consumer failed to replay change (uniqueid %s, CSN %s): %s (%d).
%s.\n",
agmt_get_long_name(prp->agmt),
uniqueid, csn_str,
- ldap_err2string(connection_error),
+ ldap_err2string(connection_error), connection_error,
*finished ? "Will retry later" : "Skipping");
}
else if (CONN_NOT_CONNECTED == replay_crc)
@@ -1487,10 +1487,11 @@ repl5_inc_update_from_op_result(Private_Repl_Protocol *prp,
ConnResult replay_cr
*finished = 1;
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"%s: Consumer failed to replay change (uniqueid %s, CSN %s): "
- "%s. Will retry later.\n",
+ "%s(%d). Will retry later.\n",
agmt_get_long_name(prp->agmt),
uniqueid, csn_str,
- connection_error ? ldap_err2string(connection_error) : "Connection
lost");
+ connection_error ? ldap_err2string(connection_error) : "Connection
lost",
+ connection_error);
}
else if (CONN_TIMEOUT == replay_crc)
{
@@ -1533,7 +1534,7 @@ repl5_inc_update_from_op_result(Private_Repl_Protocol *prp,
ConnResult replay_cr
* has already been acquired, (2) that the consumer's update vector has
* been checked and (3) that it's ok to send incremental updates.
* Returns:
- * UPDATE_NO_MORE_UPDATES - all updates were sent succussfully
+ * UPDATE_NO_MORE_UPDATES - all updates were sent successfully
* UPDATE_TRANSIENT_ERROR - some non-permanent error occurred. Try again later.
* UPDATE_FATAL_ERROR - some bad, permanent error occurred.
* UPDATE_SCHEDULE_WINDOW_CLOSED - the schedule window closed on us.
@@ -1603,7 +1604,7 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector,
PRUint32 *nu
agmt_get_long_name(prp->agmt));
return_value = UPDATE_FATAL_ERROR;
break;
- case CL5_SYSTEM_ERROR: /* NSPR error occurred: use PR_GetError for furhter info */
+ case CL5_SYSTEM_ERROR: /* NSPR error occurred: use PR_GetError for further info */
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"%s: An NSPR error (%d) occurred\n",
agmt_get_long_name(prp->agmt), PR_GetError());
@@ -1642,7 +1643,7 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector,
PRUint32 *nu
break;
case CL5_UNKNOWN_ERROR: /* unclassified error */
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "%s: An unknown error was ecountered\n",
+ "%s: An unknown error was encountered\n",
agmt_get_long_name(prp->agmt));
return_value = UPDATE_TRANSIENT_ERROR;
break;
@@ -1834,6 +1835,8 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector,
PRUint32 *nu
agmt_get_long_name(prp->agmt));
return_value = UPDATE_FATAL_ERROR;
break;
+ case CL5_IGNORE_OP:
+ break;
default:
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"%s: Unknown error code (%d) returned from
cl5GetNextOperationToReplay\n",
@@ -1865,7 +1868,7 @@ send_updates(Private_Repl_Protocol *prp, RUV *remote_update_vector,
PRUint32 *nu
/* Terminate the results reading thread */
if (!prp->repl50consumer)
{
- /* We need to ensure that we wait until all the responses have been recived from our
operations */
+ /* We need to ensure that we wait until all the responses have been received from our
operations */
if (return_value != UPDATE_CONNECTION_LOST) {
/* if connection was lost/closed, there will be nothing to read */
repl5_inc_waitfor_async_results(rd);
diff --git a/ldap/servers/plugins/replication/repl5_init.c
b/ldap/servers/plugins/replication/repl5_init.c
index a878b19..7a80c6f 100644
--- a/ldap/servers/plugins/replication/repl5_init.c
+++ b/ldap/servers/plugins/replication/repl5_init.c
@@ -126,12 +126,12 @@ static char *cleanruv_name_list[] = {
NSDS_REPL_NAME_PREFIX " Cleanruv",
NULL
};
-static char *releaseruv_oid_list[] = {
- REPL_RELEASERUV_OID,
+static char *cleanruv_abort_oid_list[] = {
+ REPL_ABORT_CLEANRUV_OID,
NULL
};
-static char *releaseruv_name_list[] = {
- NSDS_REPL_NAME_PREFIX " Releaseruv",
+static char *cleanruv_abort_name_list[] = {
+ NSDS_REPL_NAME_PREFIX " Cleanruv Abort",
NULL
};
@@ -474,7 +474,7 @@ multimaster_cleanruv_extop_init( Slapi_PBlock *pb )
}
int
-multimaster_releaseruv_extop_init( Slapi_PBlock *pb )
+multimaster_cleanruv_abort_extop_init( Slapi_PBlock *pb )
{
int rc= 0; /* OK */
void *identity = NULL;
@@ -485,11 +485,11 @@ multimaster_releaseruv_extop_init( Slapi_PBlock *pb )
if (slapi_pblock_set( pb, SLAPI_PLUGIN_VERSION, SLAPI_PLUGIN_VERSION_01 ) != 0 ||
slapi_pblock_set( pb, SLAPI_PLUGIN_DESCRIPTION, (void *)&multimasterextopdesc ) !=
0 ||
- slapi_pblock_set( pb, SLAPI_PLUGIN_EXT_OP_OIDLIST, (void *)releaseruv_oid_list ) != 0
||
- slapi_pblock_set( pb, SLAPI_PLUGIN_EXT_OP_NAMELIST, (void *)releaseruv_name_list ) != 0
||
- slapi_pblock_set( pb, SLAPI_PLUGIN_EXT_OP_FN, (void *)multimaster_extop_releaseruv ))
+ slapi_pblock_set( pb, SLAPI_PLUGIN_EXT_OP_OIDLIST, (void *)cleanruv_abort_oid_list ) !=
0 ||
+ slapi_pblock_set( pb, SLAPI_PLUGIN_EXT_OP_NAMELIST, (void *)cleanruv_abort_name_list )
!= 0 ||
+ slapi_pblock_set( pb, SLAPI_PLUGIN_EXT_OP_FN, (void *)multimaster_extop_abort_cleanruv
))
{
- slapi_log_error( SLAPI_LOG_PLUGIN, repl_plugin_name,
"multimaster_releaseruv_extop_init failed\n" );
+ slapi_log_error( SLAPI_LOG_PLUGIN, repl_plugin_name,
"multimaster_cleanruv_abort_extop_init failed\n" );
rc= -1;
}
@@ -605,24 +605,22 @@ multimaster_stop( Slapi_PBlock *pb )
int rc= 0; /* OK */
if (!multimaster_stopped_flag)
- {
- if (!is_ldif_dump)
- {
- agmtlist_shutdown(); /* Shut down replication agreements */
- }
-
+ {
+ if (!is_ldif_dump)
+ {
+ /* Shut down replication agreements */
+ agmtlist_shutdown();
+ }
+ /* if we are cleaning a ruv, stop */
+ stop_ruv_cleaning();
/* unregister backend state change notification */
slapi_unregister_backend_state_change((void *)multimaster_be_state_change);
-
- changelog5_cleanup(); /* Shut down the changelog */
- multimaster_mtnode_extension_destroy(); /* Destroy mapping tree node exts */
+ changelog5_cleanup(); /* Shut down the changelog */
+ multimaster_mtnode_extension_destroy(); /* Destroy mapping tree node exts */
replica_destroy_name_hash(); /* destroy the hash and its remaining content */
replica_config_destroy (); /* Destroy replica config info */
- multimaster_stopped_flag = 1;
- /* JCMREPL - Wait for all our threads to stop */
- /* JCMREPL - Shut down the replication plugin */
- /* JCMREPL - Mark all the replication plugin interfaces at not enabled. */
- }
+ multimaster_stopped_flag = 1;
+ }
return rc;
}
@@ -680,7 +678,7 @@ int replication_multimaster_plugin_init(Slapi_PBlock *pb)
rc= slapi_register_plugin("extendedop", 1 /* Enabled */,
"multimaster_total_extop_init", multimaster_total_extop_init, "Multimaster
replication total update extended operation plugin", NULL, identity);
rc= slapi_register_plugin("extendedop", 1 /* Enabled */,
"multimaster_response_extop_init", multimaster_response_extop_init,
"Multimaster replication extended response plugin", NULL, identity);
rc= slapi_register_plugin("extendedop", 1 /* Enabled */,
"multimaster_cleanruv_extop_init", multimaster_cleanruv_extop_init,
"Multimaster replication cleanruv extended operation plugin", NULL, identity);
- rc= slapi_register_plugin("extendedop", 1 /* Enabled */,
"multimaster_releaseruv_extop_init", multimaster_releaseruv_extop_init,
"Multimaster replication releaserid extended response plugin", NULL, identity);
+ rc= slapi_register_plugin("extendedop", 1 /* Enabled */,
"multimaster_cleanruv_abort_extop_init", multimaster_cleanruv_abort_extop_init,
"Multimaster replication cleanruv abort extended operation plugin", NULL,
identity);
if (0 == rc)
{
multimaster_initialised = 1;
diff --git a/ldap/servers/plugins/replication/repl5_plugins.c
b/ldap/servers/plugins/replication/repl5_plugins.c
index 2b4d0c9..33b197d 100644
--- a/ldap/servers/plugins/replication/repl5_plugins.c
+++ b/ldap/servers/plugins/replication/repl5_plugins.c
@@ -1020,15 +1020,6 @@ write_changelog_and_ruv (Slapi_PBlock *pb)
r = (Replica*)object_get_data (repl_obj);
PR_ASSERT (r);
- /*
- * In case we had to run cleanruv, we don't want to continue to write
- * updates to the changelog/database ruv from that replica(rid).
- */
- if( is_cleaned_rid(replica_get_rid(r))){
- /* this RID has been cleaned, just goto done */
- goto done;
- }
-
if (replica_is_flag_set (r, REPLICA_LOG_CHANGES) &&
(cl5GetState () == CL5_STATE_OPEN))
{
@@ -1083,6 +1074,12 @@ write_changelog_and_ruv (Slapi_PBlock *pb)
op_params->target_address.uniqueid = slapi_ch_strdup (uniqueid);
}
+ if( is_cleaned_rid(csn_get_replicaid(op_params->csn))){
+ /* this RID has been cleaned */
+ object_release (repl_obj);
+ return 0;
+ }
+
/* we might have stripped all the mods - in that case we do not
log the operation */
if (op_params->operation_type != SLAPI_OPERATION_MODIFY ||
@@ -1135,7 +1132,6 @@ write_changelog_and_ruv (Slapi_PBlock *pb)
update_ruv_component(r, opcsn, pb);
}
-done:
object_release (repl_obj);
return return_value;
}
@@ -1338,7 +1334,7 @@ process_operation (Slapi_PBlock *pb, const CSN *csn)
ruv = (RUV*)object_get_data (ruv_obj);
PR_ASSERT (ruv);
- rc = ruv_add_csn_inprogress (ruv, csn);
+ rc = ruv_add_csn_inprogress (ruv, csn);
object_release (ruv_obj);
object_release (r_obj);
diff --git a/ldap/servers/plugins/replication/repl5_replica.c
b/ldap/servers/plugins/replication/repl5_replica.c
index ccaa29d..27cdacd 100644
--- a/ldap/servers/plugins/replication/repl5_replica.c
+++ b/ldap/servers/plugins/replication/repl5_replica.c
@@ -61,20 +61,20 @@
*/
struct replica {
Slapi_DN *repl_root; /* top of the replicated area */
- char *repl_name; /* unique replica name */
- PRBool new_name; /* new name was generated - need to be saved */
+ char *repl_name; /* unique replica name */
+ PRBool new_name; /* new name was generated - need to be saved */
ReplicaUpdateDNList updatedn_list; /* list of dns with which a supplier should bind
to update this replica */
ReplicaType repl_type; /* is this replica read-only ? */
- PRBool legacy_consumer; /* if true, this replica is supplied by 4.0 consumer
*/
- char* legacy_purl; /* partial url of the legacy supplier */
+ PRBool legacy_consumer; /* if true, this replica is supplied by 4.0 consumer */
+ char* legacy_purl; /* partial url of the legacy supplier */
ReplicaId repl_rid; /* replicaID */
Object *repl_ruv; /* replica update vector */
PRBool repl_ruv_dirty; /* Dirty flag for ruv */
CSNPL *min_csn_pl; /* Pending list for minimal CSN */
void *csn_pl_reg_id; /* registration assignment for csn callbacks */
unsigned long repl_state_flags; /* state flags */
- PRUint32 repl_flags; /* persistent, externally visible flags */
+ PRUint32 repl_flags; /* persistent, externally visible flags */
PRLock *repl_lock; /* protects entire structure */
Slapi_Eq_Context repl_eqcxt_rs; /* context to cancel event that saves ruv */
Slapi_Eq_Context repl_eqcxt_tr; /* context to cancel event that reaps tombstones */
@@ -85,9 +85,10 @@ struct replica {
PRBool tombstone_reap_active; /* TRUE when the tombstone reaper is running */
long tombstone_reap_interval; /* Time in seconds between tombstone reaping */
Slapi_ValueSet *repl_referral; /* A list of administrator provided referral URLs */
- PRBool state_update_inprogress; /* replica state is being updated */
- PRLock *agmt_lock; /* protects agreement creation, start and stop */
+ PRBool state_update_inprogress; /* replica state is being updated */
+ PRLock *agmt_lock; /* protects agreement creation, start and stop */
char *locking_purl; /* supplier who has exclusive access */
+ char *repl_cleanruv_data[CLEANRIDSIZ + 1];
};
@@ -123,6 +124,7 @@ static int replica_log_ruv_elements_nolock (const Replica *r);
static void replica_replace_ruv_tombstone(Replica *r);
static void start_agreements_for_replica (Replica *r, PRBool start);
static void _delete_tombstone(const char *tombstone_dn, const char *uniqueid, int
ext_op_flags);
+static void replica_strip_cleaned_rids(Replica *r);
/* Allocates new replica and reads its state and state of its component from
* various parts of the DIT.
@@ -270,6 +272,8 @@ replica_new_from_entry (Slapi_Entry *e, char *errortext, PRBool
is_add_operation
slapi_sdn_get_dn(r->repl_root));
}
+ replica_check_for_tasks(r, e);
+
done:
if (rc != 0 && r)
{
@@ -306,6 +310,7 @@ replica_destroy(void **arg)
{
Replica *r;
void *repl_name;
+ int i;
if (arg == NULL)
return;
@@ -392,6 +397,10 @@ replica_destroy(void **arg)
csnplFree(&r->min_csn_pl);;
}
+ for(i = 0;r->repl_cleanruv_data[i] != NULL; i++){
+ slapi_ch_free_string(&r->repl_cleanruv_data[i]);
+ }
+
slapi_ch_free((void **)arg);
}
@@ -1302,7 +1311,7 @@ replica_reload_ruv (Replica *r)
}
/* check if there is a changelog and whether this replica logs changes */
- if (cl5GetState () == CL5_STATE_OPEN && r->repl_flags &
REPLICA_LOG_CHANGES)
+ if (cl5GetState () == CL5_STATE_OPEN && (r->repl_flags &
REPLICA_LOG_CHANGES))
{
/* Compare new ruv to the changelog's upper bound ruv. We could only keep
@@ -1424,7 +1433,7 @@ int replica_check_for_data_reload (Replica *r, void *arg)
PR_ASSERT (r);
/* check that we have a changelog and if this replica logs changes */
- if (cl5GetState () == CL5_STATE_OPEN && r->repl_flags &
REPLICA_LOG_CHANGES)
+ if (cl5GetState () == CL5_STATE_OPEN && (r->repl_flags &
REPLICA_LOG_CHANGES))
{
/* Compare new ruv to the purge ruv. If the new contains csns which
are smaller than those in purge ruv, we need to remove old and
@@ -1558,6 +1567,12 @@ _replica_get_config_entry (const Slapi_DN *root)
return e;
}
+char *
+replica_get_dn(Replica *r)
+{
+ return _replica_get_config_dn (r->repl_root);
+}
+
static int
_replica_check_validity (const Replica *r)
{
@@ -1785,6 +1800,177 @@ _replica_init_from_config (Replica *r, Slapi_Entry *e, char
*errortext)
return (_replica_check_validity (r));
}
+void
+replica_check_for_tasks(Replica *r, Slapi_Entry *e)
+{
+ char **clean_vals;
+
+ if(e == NULL){
+ return;
+ }
+ /*
+ * check if we are in the middle of a CLEANALLRUV task,
+ * if so set the cleaned rid, and fire off the thread
+ */
+ if ((clean_vals = slapi_entry_attr_get_charray(e, type_replicaCleanRUV)) != NULL)
+ {
+ PRThread *thread = NULL;
+ struct berval *payload = NULL;
+ CSN *maxcsn = NULL;
+ char *csnpart;
+ char *iter;
+ char csnstr[CSN_STRSIZE];
+ char *ridstr;
+ char *token = NULL;
+ ReplicaId rid;
+ int i;
+
+ for(i = 0; clean_vals[i]; i++){
+ cleanruv_data *data = NULL;
+
+ /*
+ * Set the cleanruv data, and add the cleaned rid
+ */
+ r->repl_cleanruv_data[i] = slapi_ch_strdup(clean_vals[i]);
+ token = ldap_utf8strtok_r(clean_vals[i], ":", &iter);
+ if(token){
+ rid = atoi(token);
+ if(rid <= 0 || rid >= READ_ONLY_REPLICA_ID){
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "CleanAllRUV
Task: invalid replica id(%d) "
+ "aborting task.\n", rid);
+ goto done;
+ }
+ } else {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "CleanAllRUV
Task: unable to parse cleanallruv "
+ "data (%s), aborting task.\n",clean_vals[i]);
+ goto done;
+ }
+ csnpart = ldap_utf8strtok_r(iter, ":", &iter);
+ maxcsn = csn_new();
+ csn_init_by_string(maxcsn, csnpart);
+ csn_as_string(maxcsn, PR_FALSE, csnstr);
+ add_cleaned_rid(rid, r, csnstr);
+
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "CleanAllRUV Task:
cleanAllRUV task found, "
+ "resuming the cleaning of rid(%d)...\n", rid);
+ /*
+ * Create payload
+ */
+ ridstr = slapi_ch_smprintf("%d:%s:%s", rid,
slapi_sdn_get_dn(replica_get_root(r)), csnstr);
+ payload = create_ruv_payload(ridstr);
+ slapi_ch_free_string(&ridstr);
+
+ if(payload == NULL){
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "CleanAllRUV
Task: Startup: Failed to "
+ "create extended op payload, aborting task");
+ return;
+ }
+ /*
+ * Setup the data struct, and fire off the thread.
+ */
+ data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
+ if (data == NULL) {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV:
failed to allocate cleanruv_data.\n");
+ csn_free(&maxcsn);
+ } else {
+ /* setup our data */
+ data->repl_obj = NULL;
+ data->replica = NULL;
+ data->rid = rid;
+ data->task = NULL;
+ data->maxcsn = maxcsn;
+ data->sdn = slapi_sdn_dup(r->repl_root);
+ data->payload = payload;
+
+ thread = PR_CreateThread(PR_USER_THREAD, replica_cleanallruv_thread_ext,
+ (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+ if (thread == NULL) {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name,
"cleanAllRUV: unable to create cleanAllRUV "
+ "thread for rid(%d)\n", (int)data->rid);
+ }
+ }
+ }
+ r->repl_cleanruv_data[i] = NULL;
+ }
+
+ if ((clean_vals = slapi_entry_attr_get_charray(e, type_replicaAbortCleanRUV)) !=
NULL)
+ {
+ PRThread *thread = NULL;
+ struct berval *payload;
+ CSN *maxcsn = NULL;
+ char *iter;
+ char *ridstr = NULL;
+ char *repl_root;
+ char *token = NULL;
+ ReplicaId rid;
+ int i;
+
+ for(i = 0; clean_vals[i]; i++){
+ cleanruv_data *data = NULL;
+
+ token = ldap_utf8strtok_r(clean_vals[i], ":", &iter);
+ if(token){
+ rid = atoi(token);
+ if(rid <= 0 || rid >= READ_ONLY_REPLICA_ID){
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "Abort
CleanAllRUV Task: invalid replica id(%d) "
+ "aborting task.\n", rid);
+ goto done;
+ }
+ } else {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "Abort
CleanAllRUV Task: unable to parse cleanallruv "
+ "data (%s), aborting task.\n",clean_vals[i]);
+ goto done;
+ }
+
+ repl_root = ldap_utf8strtok_r(iter, ":", &iter);
+ stop_ruv_cleaning();
+ maxcsn = replica_get_cleanruv_maxcsn(r, rid);
+ delete_cleaned_rid(r, rid, maxcsn);
+ csn_free(&maxcsn);
+
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "Abort CleanAllRUV
Task: abort task found, "
+ "resuming abort of rid(%d).\n", rid);
+ /*
+ * Setup the data struct, and fire off the abort thread.
+ */
+ data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
+ if (data == NULL) {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "Abort
CleanAllRUV Task: failed to allocate cleanruv_data.\n");
+ } else {
+ ridstr = slapi_ch_smprintf("%d:%s", rid, repl_root);
+ payload = create_ruv_payload(ridstr);
+ slapi_ch_free_string(&ridstr);
+
+ if(payload == NULL){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort
CleanAllRUV Task: failed to create extended "
+ "op payload\n");
+ } else {
+ /* setup the data */
+ data->repl_obj = NULL;
+ data->replica = NULL;
+ data->rid = rid;
+ data->task = NULL;
+ data->payload = payload;
+ data->repl_root = slapi_ch_strdup(repl_root);
+ data->sdn = slapi_sdn_dup(r->repl_root);
+
+ thread = PR_CreateThread(PR_USER_THREAD, replica_abort_task_thread,
+ (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+ if (thread == NULL) {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "Abort
CleanAllRUV Task: unable to create abort cleanAllRUV "
+ "thread for rid(%d)\n", (int)data->rid);
+ }
+ }
+ }
+ }
+ }
+
+done:
+ slapi_ch_array_free(clean_vals);
+}
+
/* This function updates the entry to contain information generated
during replica initialization.
Returns 0 if successful and -1 otherwise */
@@ -2782,26 +2968,25 @@ static const char *root_glue =
static int
replica_create_ruv_tombstone(Replica *r)
{
- int return_value = LDAP_LOCAL_ERROR;
- char *root_entry_str;
- Slapi_Entry *e = NULL;
+ int return_value = LDAP_LOCAL_ERROR;
+ char *root_entry_str;
+ Slapi_Entry *e = NULL;
const char *purl = NULL;
RUV *ruv;
struct berval **bvals = NULL;
Slapi_PBlock *pb = NULL;
int rc;
- PR_ASSERT(NULL != r && NULL != r->repl_root);
- root_entry_str = slapi_ch_smprintf(root_glue, slapi_sdn_get_ndn(r->repl_root),
- RUV_STORAGE_ENTRY_UNIQUEID);
+ PR_ASSERT(NULL != r && NULL != r->repl_root);
- e = slapi_str2entry(root_entry_str, SLAPI_STR2ENTRY_TOMBSTONE_CHECK);
+ root_entry_str = slapi_ch_smprintf(root_glue, slapi_sdn_get_ndn(r->repl_root),
RUV_STORAGE_ENTRY_UNIQUEID);
+
+ e = slapi_str2entry(root_entry_str, SLAPI_STR2ENTRY_TOMBSTONE_CHECK);
if (e == NULL)
goto done;
/* Add ruv */
- if (r->repl_ruv == NULL)
- {
+ if (r->repl_ruv == NULL){
CSNGen *gen;
CSN *csn;
char csnstr [CSN_STRSIZE];
@@ -2810,42 +2995,34 @@ replica_create_ruv_tombstone(Replica *r)
gen = (CSNGen *)object_get_data(r->repl_csngen);
PR_ASSERT (gen);
- if (csngen_new_csn(gen, &csn, PR_FALSE /* notify */) == CSN_SUCCESS)
- {
- (void)csn_as_string(csn, PR_FALSE, csnstr);
- csn_free(&csn);
+ if (csngen_new_csn(gen, &csn, PR_FALSE /* notify */) == CSN_SUCCESS){
+ (void)csn_as_string(csn, PR_FALSE, csnstr);
+ csn_free(&csn);
- /* if this is an updateable replica - add its own
- element to the RUV so that referrals work correctly */
- if (r->repl_type == REPLICA_TYPE_UPDATABLE)
- purl = multimaster_get_local_purl();
+ /*
+ * if this is an updateable replica - add its own
+ * element to the RUV so that referrals work correctly
+ */
+ if (r->repl_type == REPLICA_TYPE_UPDATABLE)
+ purl = multimaster_get_local_purl();
- if (ruv_init_new(csnstr, r->repl_rid, purl, &ruv) == RUV_SUCCESS)
- {
- r->repl_ruv = object_new((void*)ruv, (FNFree)ruv_destroy);
- r->repl_ruv_dirty = PR_TRUE;
- return_value = LDAP_SUCCESS;
- }
- else
- {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "Cannot create new replica update vector for %s\n",
- slapi_sdn_get_dn(r->repl_root));
- ruv_destroy(&ruv);
+ if (ruv_init_new(csnstr, r->repl_rid, purl, &ruv) == RUV_SUCCESS){
+ r->repl_ruv = object_new((void*)ruv, (FNFree)ruv_destroy);
+ r->repl_ruv_dirty = PR_TRUE;
+ return_value = LDAP_SUCCESS;
+ } else {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Cannot create
new replica update vector for %s\n",
+ slapi_sdn_get_dn(r->repl_root));
+ ruv_destroy(&ruv);
goto done;
- }
- }
- else
- {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "Cannot obtain CSN for new replica update vector for %s\n",
- slapi_sdn_get_dn(r->repl_root));
- csn_free(&csn);
+ }
+ } else {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Cannot obtain CSN
for new replica update vector for %s\n",
+ slapi_sdn_get_dn(r->repl_root));
+ csn_free(&csn);
goto done;
- }
- }
- else /* failed to write the entry because DB was not initialized - retry */
- {
+ }
+ } else { /* failed to write the entry because DB was not initialized - retry */
ruv = (RUV*) object_get_data (r->repl_ruv);
PR_ASSERT (ruv);
}
@@ -2853,30 +3030,22 @@ replica_create_ruv_tombstone(Replica *r)
PR_ASSERT (r->repl_ruv);
rc = ruv_to_bervals(ruv, &bvals);
- if (rc != RUV_SUCCESS)
- {
+ if (rc != RUV_SUCCESS){
goto done;
}
/* ONREPL this is depricated function but there is currently no better API to use */
rc = slapi_entry_add_values(e, type_ruvElement, bvals);
- if (rc != 0)
- {
+ if (rc != 0){
goto done;
}
-
- pb = slapi_pblock_new();
- slapi_add_entry_internal_set_pb(
- pb,
- e,
- NULL /* controls */,
- repl_get_plugin_identity(PLUGIN_MULTIMASTER_REPLICATION),
- OP_FLAG_TOMBSTONE_ENTRY | OP_FLAG_REPLICATED | OP_FLAG_REPL_FIXUP |
- OP_FLAG_REPL_RUV);
- slapi_add_internal_pb(pb);
- e = NULL; /* add consumes e, upon success or failure */
- slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &return_value);
+ pb = slapi_pblock_new();
+ slapi_add_entry_internal_set_pb(pb, e, NULL /* controls
*/, repl_get_plugin_identity(PLUGIN_MULTIMASTER_REPLICATION),
+ OP_FLAG_TOMBSTONE_ENTRY | OP_FLAG_REPLICATED | OP_FLAG_REPL_FIXUP |
OP_FLAG_REPL_RUV);
+ slapi_add_internal_pb(pb);
+ e = NULL; /* add consumes e, upon success or failure */
+ slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &return_value);
if (return_value == LDAP_SUCCESS)
r->repl_ruv_dirty = PR_FALSE;
@@ -2889,50 +3058,50 @@ done:
if (pb)
slapi_pblock_destroy(pb);
- slapi_ch_free((void **) &root_entry_str);
+ slapi_ch_free_string(&root_entry_str);
- return return_value;
+ return return_value;
}
static void
assign_csn_callback(const CSN *csn, void *data)
{
- Replica *r = (Replica *)data;
+ Replica *r = (Replica *)data;
Object *ruv_obj;
RUV *ruv;
- PR_ASSERT(NULL != csn);
- PR_ASSERT(NULL != r);
+ PR_ASSERT(NULL != csn);
+ PR_ASSERT(NULL != r);
ruv_obj = replica_get_ruv (r);
PR_ASSERT (ruv_obj);
ruv = (RUV*)object_get_data (ruv_obj);
PR_ASSERT (ruv);
- PR_Lock(r->repl_lock);
+ PR_Lock(r->repl_lock);
- r->repl_csn_assigned = PR_TRUE;
+ r->repl_csn_assigned = PR_TRUE;
- if (NULL != r->min_csn_pl)
- {
- if (csnplInsert(r->min_csn_pl, csn) != 0)
- {
- char csn_str[CSN_STRSIZE]; /* For logging only */
- /* Ack, we can't keep track of min csn. Punt. */
- if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
- slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "assign_csn_callback: "
- "failed to insert csn %s for replica %s\n",
- csn_as_string(csn, PR_FALSE, csn_str),
- slapi_sdn_get_dn(r->repl_root));
- }
- csnplFree(&r->min_csn_pl);
- }
- }
+ if (NULL != r->min_csn_pl)
+ {
+ if (csnplInsert(r->min_csn_pl, csn) != 0)
+ {
+ char csn_str[CSN_STRSIZE]; /* For logging only */
+ /* Ack, we can't keep track of min csn. Punt. */
+ if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
+ slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name,
"assign_csn_callback: "
+ "failed to insert csn %s for replica %s\n",
+ csn_as_string(csn, PR_FALSE, csn_str),
+ slapi_sdn_get_dn(r->repl_root));
+ }
+ csnplFree(&r->min_csn_pl);
+ }
+ }
ruv_add_csn_inprogress (ruv, csn);
- PR_Unlock(r->repl_lock);
+ PR_Unlock(r->repl_lock);
object_release (ruv_obj);
}
@@ -3167,19 +3336,41 @@ replica_set_tombstone_reap_interval (Replica *r, long interval)
PR_Unlock(r->repl_lock);
}
+static void
+replica_strip_cleaned_rids(Replica *r)
+{
+ Object *RUVObj;
+ RUV *ruv = NULL;
+ ReplicaId rid[32] = {0};
+ int i = 0;
+
+ RUVObj = replica_get_ruv(r);
+ ruv = (RUV*)object_get_data (RUVObj);
+
+ ruv_get_cleaned_rids(ruv, rid);
+ while(rid[i] != 0){
+ ruv_delete_replica(ruv, rid[i]);
+ replica_set_ruv_dirty(r);
+ replica_write_ruv(r);
+ i++;
+ }
+ object_release(RUVObj);
+}
+
/* Update the tombstone entry to reflect the content of the ruv */
static void
replica_replace_ruv_tombstone(Replica *r)
{
Slapi_PBlock *pb = NULL;
- char *dn;
- int rc;
-
Slapi_Mod smod;
Slapi_Mod smod_last_modified;
LDAPMod *mods [3];
+ char *dn;
+ int rc;
+
+ PR_ASSERT(NULL != r && NULL != r->repl_root);
- PR_ASSERT(NULL != r && NULL != r->repl_root);
+ replica_strip_cleaned_rids(r);
PR_Lock(r->repl_lock);
@@ -3188,14 +3379,14 @@ replica_replace_ruv_tombstone(Replica *r)
ruv_last_modified_to_smod ((RUV*)object_get_data(r->repl_ruv),
&smod_last_modified);
dn = _replica_get_config_dn (r->repl_root);
- if (NULL == dn) {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "replica_replace_ruv_tombstone: "
- "failed to get the config dn for %s\n",
- slapi_sdn_get_dn (r->repl_root));
- PR_Unlock(r->repl_lock);
- goto bail;
- }
+ if (NULL == dn) {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "replica_replace_ruv_tombstone: "
+ "failed to get the config dn for %s\n",
+ slapi_sdn_get_dn (r->repl_root));
+ PR_Unlock(r->repl_lock);
+ goto bail;
+ }
mods[0] = (LDAPMod*)slapi_mod_get_ldapmod_byref(&smod);
mods[1] = (LDAPMod*)slapi_mod_get_ldapmod_byref(&smod_last_modified);
@@ -3219,12 +3410,12 @@ replica_replace_ruv_tombstone(Replica *r)
if (rc != LDAP_SUCCESS)
{
- if ((rc != LDAP_NO_SUCH_OBJECT) || !replica_is_state_flag_set(r, REPLICA_IN_USE))
- {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_replace_ruv_tombstone: "
- "failed to update replication update vector for replica %s: LDAP "
- "error - %d\n", (char*)slapi_sdn_get_dn (r->repl_root), rc);
- }
+ if ((rc != LDAP_NO_SUCH_OBJECT) || !replica_is_state_flag_set(r,
REPLICA_IN_USE))
+ {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_replace_ruv_tombstone: "
+ "failed to update replication update vector for replica %s: LDAP
"
+ "error - %d\n", (char*)slapi_sdn_get_dn (r->repl_root),
rc);
+ }
}
slapi_ch_free ((void**)&dn);
@@ -3240,14 +3431,18 @@ replica_update_ruv_consumer(Replica *r, RUV *supplier_ruv)
ReplicaId supplier_id = 0;
char *supplier_purl = NULL;
- if ( ruv_get_first_id_and_purl(supplier_ruv, &supplier_id, &supplier_purl) ==
RUV_SUCCESS )
+ if ( ruv_get_first_id_and_purl(supplier_ruv, &supplier_id, &supplier_purl) ==
RUV_SUCCESS)
{
RUV *local_ruv = NULL;
PR_Lock(r->repl_lock);
local_ruv = (RUV*)object_get_data (r->repl_ruv);
- PR_ASSERT (local_ruv);
+
+ if(is_cleaned_rid(supplier_id) || local_ruv == NULL){
+ PR_Unlock(r->repl_lock);
+ return;
+ }
if ( ruv_local_contains_supplier(local_ruv, supplier_id) == 0 )
{
@@ -3579,3 +3774,67 @@ replica_get_attr ( Slapi_PBlock *pb, const char* type, void *value
)
return rc;
}
+
+void
+replica_add_cleanruv_data(Replica *r, char *val)
+{
+ int i;
+
+ PR_Lock(r->repl_lock);
+
+ for (i = 0; i < CLEANRIDSIZ && r->repl_cleanruv_data[i] != NULL; i++);
/* goto the end of the list */
+ r->repl_cleanruv_data[i] = slapi_ch_strdup(val); /* append to list */
+ r->repl_cleanruv_data[i + 1] = NULL;
+
+ PR_Unlock(r->repl_lock);
+}
+
+void
+replica_remove_cleanruv_data(Replica *r, char *val)
+{
+ int i;
+
+ PR_Lock(r->repl_lock);
+
+ for(i = 0; i < CLEANRIDSIZ && r->repl_cleanruv_data[i] &&
strcmp(r->repl_cleanruv_data[i], val) != 0; i++);
+ if( i < CLEANRIDSIZ ){
+ slapi_ch_free_string(&r->repl_cleanruv_data[i]);
+ for(; i < CLEANRIDSIZ; i++){
+ /* rewrite entire array */
+ r->repl_cleanruv_data[i] = r->repl_cleanruv_data[i + 1];
+ }
+ }
+
+ PR_Unlock(r->repl_lock);
+}
+
+CSN *
+replica_get_cleanruv_maxcsn(Replica *r, ReplicaId rid)
+{
+ CSN *newcsn;
+ char *csnstr;
+ char *token;
+ char *iter;
+ int repl_rid = 0;
+ int i;
+
+ PR_Lock(r->repl_lock);
+
+ for(i = 0; i < CLEANRIDSIZ && r->repl_cleanruv_data[i]; i++){
+ token = ldap_utf8strtok_r(r->repl_cleanruv_data[i], ":",
&iter);
+ if(token){
+ repl_rid = atoi(token);
+ }
+ csnstr = ldap_utf8strtok_r(iter, ":", &iter);
+ if(repl_rid == rid){
+ newcsn = csn_new();
+ csn_init_by_string(newcsn, csnstr);
+ PR_Unlock(r->repl_lock);
+ return newcsn;
+ }
+ }
+
+ PR_Unlock(r->repl_lock);
+
+ return NULL;
+}
diff --git a/ldap/servers/plugins/replication/repl5_replica_config.c
b/ldap/servers/plugins/replication/repl5_replica_config.c
index 771af61..89651cf 100644
--- a/ldap/servers/plugins/replication/repl5_replica_config.c
+++ b/ldap/servers/plugins/replication/repl5_replica_config.c
@@ -2,15 +2,15 @@
* This Program is free software; you can redistribute it and/or modify it under
* the terms of the GNU General Public License as published by the Free Software
* Foundation; version 2 of the License.
- *
+ *
* This Program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
* FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.
- *
+ *
* You should have received a copy of the GNU General Public License along with
* this Program; if not, write to the Free Software Foundation, Inc., 59 Temple
* Place, Suite 330, Boston, MA 02111-1307 USA.
- *
+ *
* In addition, as a special exception, Red Hat, Inc. gives You the additional
* right to link the code of this Program with code not covered under the GNU
* General Public License ("Non-GPL Code") and to distribute linked
combinations
@@ -28,9 +28,9 @@
* version of the file, but you are not obligated to do so. If you do not wish to
* provide this exception without modification, you must delete this exception
* statement from your version and license this file solely under the GPL without
- * exception.
- *
- *
+ * exception.
+ *
+ *
* Copyright (C) 2001 Sun Microsystems, Inc. Used by permission.
* Copyright (C) 2005 Red Hat, Inc.
* All rights reserved.
@@ -58,13 +58,17 @@
#define CLEANRUVLEN 8
#define CLEANALLRUV "CLEANALLRUV"
#define CLEANALLRUVLEN 11
-#define RELEASERUV "RELEASERUV"
-#define RELEASERUVLEN 10
#define REPLICA_RDN "cn=replica"
+#define CLEANALLRUV_ID "CleanAllRUV Task"
+#define ABORT_CLEANALLRUV_ID "Abort CleanAllRUV Task"
int slapi_log_urp = SLAPI_LOG_REPL;
-static ReplicaId cleaned_rid = 0;
-static int released_rid = 0;
+static ReplicaId cleaned_rids[CLEANRIDSIZ + 1] = {0};
+static ReplicaId aborted_rids[CLEANRIDSIZ + 1] = {0};
+static Slapi_RWLock *rid_lock = NULL;
+static Slapi_RWLock *abort_rid_lock = NULL;
+static PRLock *notify_lock = NULL;
+static PRCondVar *notify_cvar = NULL;
/* Forward Declartions */
static int replica_config_add (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter,
int *returncode, char *returntext, void *arg);
@@ -72,7 +76,7 @@ static int replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* e,
Slapi_Entry*
static int replica_config_post_modify (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry*
entryAfter, int *returncode, char *returntext, void *arg);
static int replica_config_delete (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry*
entryAfter, int *returncode, char *returntext, void *arg);
static int replica_config_search (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry*
entryAfter, int *returncode, char *returntext, void *arg);
-
+static int replica_cleanall_ruv_task(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry
*eAfter, int *returncode, char *returntext, void *arg);
static int replica_config_change_type_and_id (Replica *r, const char *new_type, const
char *new_id, char *returntext, int apply_mods);
static int replica_config_change_updatedn (Replica *r, const LDAPMod *mod, char
*returntext, int apply_mods);
static int replica_config_change_flags (Replica *r, const char *new_flags, char
*returntext, int apply_mods);
@@ -80,9 +84,18 @@ static int replica_execute_task (Object *r, const char *task_name, char
*returnt
static int replica_execute_cl2ldif_task (Object *r, char *returntext);
static int replica_execute_ldif2cl_task (Object *r, char *returntext);
static int replica_execute_cleanruv_task (Object *r, ReplicaId rid, char *returntext);
-static int replica_execute_cleanall_ruv_task (Object *r, ReplicaId rid, char
*returntext);
-static int replica_execute_release_ruv_task(Object *r, ReplicaId rid);
-static struct berval *create_ruv_payload(char *value);
+static int replica_execute_cleanall_ruv_task (Object *r, ReplicaId rid, Slapi_Task *task,
char *returntext);
+static void replica_cleanallruv_thread(void *arg);
+static void replica_send_cleanruv_task(Repl_Agmt *agmt, ReplicaId rid, Slapi_Task
*task);
+static int check_agmts_are_alive(Replica *replica, ReplicaId rid, Slapi_Task *task);
+static int check_agmts_are_caught_up(Replica *replica, ReplicaId rid, char *maxcsn,
Slapi_Task *task);
+static int replica_cleanallruv_send_extop(Repl_Agmt *ra, ReplicaId rid, Slapi_Task *task,
struct berval *payload, int check_result);
+static int replica_cleanallruv_send_abort_extop(Repl_Agmt *ra, Slapi_Task *task, struct
berval *payload);
+static int replica_cleanallruv_check_maxcsn(Repl_Agmt *agmt, char *rid_text, char
*maxcsn, Slapi_Task *task);
+static int replica_cleanallruv_replica_alive(Repl_Agmt *agmt);
+static int replica_cleanallruv_check_ruv(Repl_Agmt *ra, char *rid_text, Slapi_Task
*task);
+static int get_cleanruv_task_count();
+static int get_abort_cleanruv_task_count();
static int replica_cleanup_task (Object *r, const char *task_name, char *returntext, int
apply_mods);
static int replica_task_done(Replica *replica);
static multimaster_mtnode_extension * _replica_config_get_mtnode_ext (const Slapi_Entry
*e);
@@ -105,29 +118,59 @@ int
replica_config_init()
{
s_configLock = PR_NewLock ();
+
if (s_configLock == NULL)
{
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_init: "
- "failed to cretate configuration lock; NSPR error -
%d\n",
+ "failed to create configuration lock; NSPR error -
%d\n",
PR_GetError ());
return -1;
}
+ rid_lock = slapi_new_rwlock();
+ if(rid_lock == NULL)
+ {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_init: "
+ "failed to create rid_lock; NSPR error - %d\n", PR_GetError ());
+ return -1;
+ }
+ abort_rid_lock = slapi_new_rwlock();
+ if(abort_rid_lock == NULL)
+ {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_init: "
+ "failed to create abort_rid_lock; NSPR error - %d\n", PR_GetError ());
+ return -1;
+ }
+ if ( ( notify_lock = PR_NewLock()) == NULL ) {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_init: "
+ "failed to create notify lock; NSPR error - %d\n", PR_GetError ());
+ return -1;
+ }
+ if ( ( notify_cvar = PR_NewCondVar( notify_lock )) == NULL ) {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_init: "
+ "failed to create notify cond var; NSPR error - %d\n", PR_GetError ());
+ return -1;
+ }
/* config DSE must be initialized before we get here */
slapi_config_register_callback(SLAPI_OPERATION_ADD, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
- CONFIG_FILTER, replica_config_add, NULL);
+ CONFIG_FILTER, replica_config_add, NULL);
slapi_config_register_callback(SLAPI_OPERATION_MODIFY, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
CONFIG_FILTER, replica_config_modify,NULL);
slapi_config_register_callback(SLAPI_OPERATION_MODRDN, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
CONFIG_FILTER, dont_allow_that, NULL);
slapi_config_register_callback(SLAPI_OPERATION_DELETE, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
- CONFIG_FILTER, replica_config_delete,NULL);
+ CONFIG_FILTER, replica_config_delete,NULL);
slapi_config_register_callback(SLAPI_OPERATION_SEARCH, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
- CONFIG_FILTER, replica_config_search,NULL);
- slapi_config_register_callback(SLAPI_OPERATION_MODIFY, DSE_FLAG_POSTOP,
+ CONFIG_FILTER, replica_config_search,NULL);
+ slapi_config_register_callback(SLAPI_OPERATION_MODIFY, DSE_FLAG_POSTOP,
CONFIG_BASE, LDAP_SCOPE_SUBTREE,
CONFIG_FILTER, replica_config_post_modify,
NULL);
+
+ /* register the CLEANALLRUV & ABORT task */
+ slapi_task_register_handler("cleanallruv", replica_cleanall_ruv_task);
+ slapi_task_register_handler("abort cleanallruv",
replica_cleanall_ruv_abort);
+
return 0;
}
@@ -142,7 +185,7 @@ replica_config_destroy ()
/* config DSE must be initialized before we get here */
slapi_config_remove_callback(SLAPI_OPERATION_ADD, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
- CONFIG_FILTER, replica_config_add);
+ CONFIG_FILTER, replica_config_add);
slapi_config_remove_callback(SLAPI_OPERATION_MODIFY, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
CONFIG_FILTER, replica_config_modify);
slapi_config_remove_callback(SLAPI_OPERATION_MODRDN, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
@@ -151,26 +194,26 @@ replica_config_destroy ()
CONFIG_FILTER, replica_config_delete);
slapi_config_remove_callback(SLAPI_OPERATION_SEARCH, DSE_FLAG_PREOP, CONFIG_BASE,
LDAP_SCOPE_SUBTREE,
CONFIG_FILTER, replica_config_search);
- slapi_config_remove_callback(SLAPI_OPERATION_MODIFY, DSE_FLAG_PREOP,
+ slapi_config_remove_callback(SLAPI_OPERATION_MODIFY, DSE_FLAG_PREOP,
CONFIG_BASE, LDAP_SCOPE_SUBTREE,
CONFIG_FILTER, replica_config_post_modify);
}
-static int
-replica_config_add (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter,
+static int
+replica_config_add (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter,
int *returncode, char *errorbuf, void *arg)
{
Replica *r = NULL;
- multimaster_mtnode_extension *mtnode_ext;
+ multimaster_mtnode_extension *mtnode_ext;
char *replica_root = (char*)slapi_entry_attr_get_charptr (e, attr_replicaRoot);
char buf [SLAPI_DSE_RETURNTEXT_SIZE];
char *errortext = errorbuf ? errorbuf : buf;
-
+
if (errorbuf)
{
errorbuf[0] = '\0';
- }
-
+ }
+
*returncode = LDAP_SUCCESS;
PR_Lock (s_configLock);
@@ -185,16 +228,16 @@ replica_config_add (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry*
entryAfter,
{
PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE, "replica already
configured for %s", replica_root);
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_add:
%s\n", errortext);
- *returncode = LDAP_UNWILLING_TO_PERFORM;
- goto done;
+ *returncode = LDAP_UNWILLING_TO_PERFORM;
+ goto done;
}
/* create replica object */
r = replica_new_from_entry (e, errortext, PR_TRUE /* is a newly added entry */);
if (r == NULL)
{
- *returncode = LDAP_OPERATIONS_ERROR;
- goto done;
+ *returncode = LDAP_OPERATIONS_ERROR;
+ goto done;
}
/* Set the mapping tree node state, and the referrals from the RUV */
@@ -220,24 +263,24 @@ done:
if (*returncode != LDAP_SUCCESS)
{
- if (mtnode_ext->replica)
+ if (mtnode_ext->replica)
object_release (mtnode_ext->replica);
return SLAPI_DSE_CALLBACK_ERROR;
}
- else
+ else
return SLAPI_DSE_CALLBACK_OK;
}
-static int
-replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry* e,
+static int
+replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore, Slapi_Entry* e,
int *returncode, char *returntext, void *arg)
{
int rc= 0;
LDAPMod **mods;
int i, apply_mods;
- multimaster_mtnode_extension *mtnode_ext;
+ multimaster_mtnode_extension *mtnode_ext;
Replica *r = NULL;
- char *replica_root = NULL;
+ char *replica_root = NULL;
char buf [SLAPI_DSE_RETURNTEXT_SIZE];
char *errortext = returntext ? returntext : buf;
char *config_attr, *config_attr_value;
@@ -252,7 +295,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
/* just let internal operations originated from replication plugin to go through */
slapi_pblock_get (pb, SLAPI_OPERATION, &op);
- slapi_pblock_get (pb, SLAPI_PLUGIN_IDENTITY, &identity);
+ slapi_pblock_get (pb, SLAPI_PLUGIN_IDENTITY, &identity);
if (operation_is_flag_set(op, OP_FLAG_INTERNAL) &&
(identity == repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION)))
@@ -264,7 +307,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
replica_root = (char*)slapi_entry_attr_get_charptr (e, attr_replicaRoot);
PR_Lock (s_configLock);
-
+
mtnode_ext = _replica_config_get_mtnode_ext (e);
PR_ASSERT (mtnode_ext);
@@ -276,7 +319,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE, "replica does not exist
for %s", replica_root);
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_modify:
%s\n",
errortext);
- *returncode = LDAP_OPERATIONS_ERROR;
+ *returncode = LDAP_OPERATIONS_ERROR;
goto done;
}
@@ -287,7 +330,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
for (apply_mods = 0; apply_mods <= 1; apply_mods++)
{
/* we only allow the replica ID and type to be modified together e.g.
- if converting a read only replica to a master or vice versa -
+ if converting a read only replica to a master or vice versa -
we will need to change both the replica ID and the type at the same
time - we must disallow changing the replica ID if the type is not
being changed and vice versa
@@ -304,7 +347,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
break;
config_attr = (char *) mods[i]->mod_type;
- PR_ASSERT (config_attr);
+ PR_ASSERT (config_attr);
/* disallow modifications or removal of replica root,
replica name and replica state attributes */
@@ -313,14 +356,14 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
strcasecmp (config_attr, attr_state) == 0)
{
*returncode = LDAP_UNWILLING_TO_PERFORM;
- PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE, "modification of
%s attribute is not allowed",
- config_attr);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_modify: %s\n",
+ PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE, "modification of
%s attribute is not allowed",
+ config_attr);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_modify: %s\n",
errortext);
}
/* this is a request to delete an attribute */
- else if (mods[i]->mod_op & LDAP_MOD_DELETE || mods[i]->mod_bvalues
== NULL
- || mods[i]->mod_bvalues[0]->bv_val == NULL)
+ else if ((mods[i]->mod_op & LDAP_MOD_DELETE) ||
mods[i]->mod_bvalues == NULL
+ || mods[i]->mod_bvalues[0]->bv_val == NULL)
{
/* currently, you can only remove referral,
legacy consumer or bind dn attribute */
@@ -345,10 +388,9 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
else
{
*returncode = LDAP_UNWILLING_TO_PERFORM;
- PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE, "deletion of
%s attribute is not allowed", config_attr);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_modify: %s\n",
- errortext);
- }
+ PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE, "deletion of
%s attribute is not allowed", config_attr);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_modify: %s\n", errortext);
+ }
}
else /* modify an attribute */
{
@@ -356,8 +398,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
if (strcasecmp (config_attr, attr_replicaBindDn) == 0)
{
- *returncode = replica_config_change_updatedn (r, mods[i],
- errortext, apply_mods);
+ *returncode = replica_config_change_updatedn (r, mods[i], errortext,
apply_mods);
}
else if (strcasecmp (config_attr, attr_replicaType) == 0)
{
@@ -367,16 +408,14 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
{
new_repl_id = slapi_ch_strdup(config_attr_value);
}
- else if (strcasecmp (config_attr, attr_flags) == 0)
+ else if (strcasecmp (config_attr, attr_flags) == 0)
{
- *returncode = replica_config_change_flags (r, config_attr_value,
- errortext, apply_mods);
+ *returncode = replica_config_change_flags (r, config_attr_value,
errortext, apply_mods);
}
- else if (strcasecmp (config_attr, TASK_ATTR) == 0)
+ else if (strcasecmp (config_attr, TASK_ATTR) == 0)
{
- *returncode = replica_execute_task (mtnode_ext->replica,
config_attr_value,
- errortext, apply_mods);
- }
+ *returncode = replica_execute_task (mtnode_ext->replica,
config_attr_value, errortext, apply_mods);
+ }
else if (strcasecmp (config_attr, attr_replicaReferral) == 0)
{
if (apply_mods)
@@ -391,14 +430,14 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
if (!replica_is_legacy_consumer (r)) {
consumer5_set_mapping_tree_state_for_replica(r, NULL);
}
- }
+ }
}
else if (strcasecmp (config_attr, type_replicaPurgeDelay) == 0)
{
- if (apply_mods && config_attr_value && config_attr_value[0])
+ if (apply_mods && config_attr_value && config_attr_value[0])
{
PRUint32 delay;
- if (isdigit (config_attr_value[0]))
+ if (isdigit (config_attr_value[0]))
{
delay = (unsigned int)atoi(config_attr_value);
replica_set_purge_delay(r, delay);
@@ -409,7 +448,7 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
}
else if (strcasecmp (config_attr, type_replicaTombstonePurgeInterval) == 0)
{
- if (apply_mods && config_attr_value && config_attr_value[0])
+ if (apply_mods && config_attr_value && config_attr_value[0])
{
long interval;
interval = atol (config_attr_value);
@@ -439,17 +478,14 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
*returncode = LDAP_UNWILLING_TO_PERFORM;
PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE,
"modification of attribute %s is not allowed in replica entry",
config_attr);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_modify: %s\n",
- errortext);
- }
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_modify: %s\n", errortext);
+ }
}
}
if (new_repl_id || new_repl_type)
{
- *returncode = replica_config_change_type_and_id(r, new_repl_type,
- new_repl_id, errortext,
- apply_mods);
+ *returncode = replica_config_change_type_and_id(r, new_repl_type, new_repl_id,
errortext, apply_mods);
slapi_ch_free_string(&new_repl_id);
slapi_ch_free_string(&new_repl_type);
}
@@ -458,9 +494,9 @@ replica_config_modify (Slapi_PBlock *pb, Slapi_Entry* entryBefore,
Slapi_Entry*
done:
if (mtnode_ext->replica)
object_release (mtnode_ext->replica);
-
+
/* slapi_ch_free accepts NULL pointer */
- slapi_ch_free ((void**)&replica_root);
+ slapi_ch_free_string(&replica_root);
PR_Unlock (s_configLock);
@@ -474,9 +510,9 @@ done:
}
}
-static int
+static int
replica_config_post_modify(Slapi_PBlock *pb,
- Slapi_Entry* entryBefore,
+ Slapi_Entry* entryBefore,
Slapi_Entry* e,
int *returncode,
char *returntext,
@@ -485,9 +521,9 @@ replica_config_post_modify(Slapi_PBlock *pb,
int rc= 0;
LDAPMod **mods;
int i, apply_mods;
- multimaster_mtnode_extension *mtnode_ext;
+ multimaster_mtnode_extension *mtnode_ext;
Replica *r = NULL;
- char *replica_root = NULL;
+ char *replica_root = NULL;
char buf [SLAPI_DSE_RETURNTEXT_SIZE];
char *errortext = returntext ? returntext : buf;
char *config_attr, *config_attr_value;
@@ -503,7 +539,7 @@ replica_config_post_modify(Slapi_PBlock *pb,
/* just let internal operations originated from replication plugin to go through */
slapi_pblock_get (pb, SLAPI_OPERATION, &op);
- slapi_pblock_get (pb, SLAPI_PLUGIN_IDENTITY, &identity);
+ slapi_pblock_get (pb, SLAPI_PLUGIN_IDENTITY, &identity);
if (operation_is_flag_set(op, OP_FLAG_INTERNAL) &&
(identity == repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION)))
@@ -515,7 +551,7 @@ replica_config_post_modify(Slapi_PBlock *pb,
replica_root = (char*)slapi_entry_attr_get_charptr (e, attr_replicaRoot);
PR_Lock (s_configLock);
-
+
mtnode_ext = _replica_config_get_mtnode_ext (e);
PR_ASSERT (mtnode_ext);
@@ -529,7 +565,7 @@ replica_config_post_modify(Slapi_PBlock *pb,
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_config_post_modify: %s\n",
errortext);
- *returncode = LDAP_OPERATIONS_ERROR;
+ *returncode = LDAP_OPERATIONS_ERROR;
goto done;
}
@@ -540,7 +576,7 @@ replica_config_post_modify(Slapi_PBlock *pb,
for (apply_mods = 0; apply_mods <= 1; apply_mods++)
{
/* we only allow the replica ID and type to be modified together e.g.
- if converting a read only replica to a master or vice versa -
+ if converting a read only replica to a master or vice versa -
we will need to change both the replica ID and the type at the same
time - we must disallow changing the replica ID if the type is not
being changed and vice versa
@@ -554,7 +590,7 @@ replica_config_post_modify(Slapi_PBlock *pb,
break;
config_attr = (char *) mods[i]->mod_type;
- PR_ASSERT (config_attr);
+ PR_ASSERT (config_attr);
/* disallow modifications or removal of replica root,
replica name and replica state attributes */
@@ -564,16 +600,16 @@ replica_config_post_modify(Slapi_PBlock *pb,
{
*returncode = LDAP_UNWILLING_TO_PERFORM;
PR_snprintf (errortext, SLAPI_DSE_RETURNTEXT_SIZE,
- "modification of %s attribute is not allowed",
- config_attr);
+ "modification of %s attribute is not allowed",
+ config_attr);
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "replica_config_post_modify: %s\n",
+ "replica_config_post_modify: %s\n",
errortext);
}
/* this is a request to delete an attribute */
- else if (mods[i]->mod_op & LDAP_MOD_DELETE ||
+ else if ((mods[i]->mod_op & LDAP_MOD_DELETE) ||
mods[i]->mod_bvalues == NULL ||
- mods[i]->mod_bvalues[0]->bv_val == NULL)
+ mods[i]->mod_bvalues[0]->bv_val == NULL)
{
;
}
@@ -581,7 +617,7 @@ replica_config_post_modify(Slapi_PBlock *pb,
{
config_attr_value = (char *) mods[i]->mod_bvalues[0]->bv_val;
- if (strcasecmp (config_attr, TASK_ATTR) == 0)
+ if (strcasecmp (config_attr, TASK_ATTR) == 0)
{
flag_need_cleanup = 1;
}
@@ -605,7 +641,7 @@ done:
if (mtnode_ext->replica)
object_release (mtnode_ext->replica);
-
+
if (*returncode != LDAP_SUCCESS)
{
return SLAPI_DSE_CALLBACK_ERROR;
@@ -616,8 +652,8 @@ done:
}
}
-static int
-replica_config_delete (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter,
+static int
+replica_config_delete (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter,
int *returncode, char *returntext, void *arg)
{
multimaster_mtnode_extension *mtnode_ext;
@@ -650,26 +686,26 @@ replica_config_delete (Slapi_PBlock *pb, Slapi_Entry* e,
Slapi_Entry* entryAfter
return SLAPI_DSE_CALLBACK_OK;
}
-static int
-replica_config_search (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter, int
*returncode,
+static int
+replica_config_search (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry* entryAfter, int
*returncode,
char *returntext, void *arg)
{
- multimaster_mtnode_extension *mtnode_ext;
+ multimaster_mtnode_extension *mtnode_ext;
int changeCount = 0;
PRBool reapActive = PR_FALSE;
char val [64];
/* add attribute that contains number of entries in the changelog for this replica */
-
+
PR_Lock (s_configLock);
-
+
mtnode_ext = _replica_config_get_mtnode_ext (e);
PR_ASSERT (mtnode_ext);
-
+
if (mtnode_ext->replica) {
Replica *replica;
object_acquire (mtnode_ext->replica);
- if (cl5GetState () == CL5_STATE_OPEN) {
+ if (cl5GetState () == CL5_STATE_OPEN) {
changeCount = cl5GetOperationCount (mtnode_ext->replica);
}
replica = (Replica*)object_get_data (mtnode_ext->replica);
@@ -688,9 +724,9 @@ replica_config_search (Slapi_PBlock *pb, Slapi_Entry* e, Slapi_Entry*
entryAfter
return SLAPI_DSE_CALLBACK_OK;
}
-static int
+static int
replica_config_change_type_and_id (Replica *r, const char *new_type,
- const char *new_id, char *returntext,
+ const char *new_id, char *returntext,
int apply_mods)
{
int type;
@@ -770,8 +806,8 @@ replica_config_change_type_and_id (Replica *r, const char *new_type,
return LDAP_SUCCESS;
}
-static int
-replica_config_change_updatedn (Replica *r, const LDAPMod *mod, char *returntext,
+static int
+replica_config_change_updatedn (Replica *r, const LDAPMod *mod, char *returntext,
int apply_mods)
{
PR_ASSERT (r);
@@ -790,7 +826,7 @@ replica_config_change_updatedn (Replica *r, const LDAPMod *mod, char
*returntext
return LDAP_SUCCESS;
}
-static int replica_config_change_flags (Replica *r, const char *new_flags,
+static int replica_config_change_flags (Replica *r, const char *new_flags,
char *returntext, int apply_mods)
{
PR_ASSERT (r);
@@ -807,10 +843,10 @@ static int replica_config_change_flags (Replica *r, const char
*new_flags,
return LDAP_SUCCESS;
}
-static int replica_execute_task (Object *r, const char *task_name, char *returntext,
+static int replica_execute_task (Object *r, const char *task_name, char *returntext,
int apply_mods)
{
-
+
if (strcasecmp (task_name, CL2LDIF_TASK) == 0)
{
if (apply_mods)
@@ -848,13 +884,14 @@ static int replica_execute_task (Object *r, const char *task_name,
char *returnt
{
int temprid = atoi(&(task_name[CLEANALLRUVLEN]));
if (temprid <= 0 || temprid >= READ_ONLY_REPLICA_ID){
- PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE, "Invalid replica id (%d) for
task - (%s)", temprid, task_name);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_execute_task:
%s\n", returntext);
- return LDAP_OPERATIONS_ERROR;
+ PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE, "Invalid replica id (%d) for
task - (%s)", temprid, task_name);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_execute_task:
%s\n", returntext);
+ return LDAP_OPERATIONS_ERROR;
}
if (apply_mods)
{
- return replica_execute_cleanall_ruv_task(r, (ReplicaId)temprid, returntext);
+ Slapi_Task *empty_task = NULL;
+ return replica_execute_cleanall_ruv_task(r, (ReplicaId)temprid, empty_task,
returntext);
}
else
return LDAP_SUCCESS;
@@ -862,22 +899,22 @@ static int replica_execute_task (Object *r, const char *task_name,
char *returnt
else
{
PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE, "unsupported replica task
- %s", task_name);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "replica_execute_task: %s\n", returntext);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "replica_execute_task: %s\n", returntext);
return LDAP_OPERATIONS_ERROR;
}
-
+
}
-static int
-replica_cleanup_task (Object *r, const char *task_name, char *returntext,
+static int
+replica_cleanup_task (Object *r, const char *task_name, char *returntext,
int apply_mods)
{
int rc = LDAP_SUCCESS;
if (apply_mods) {
Replica *replica = (Replica*)object_get_data (r);
if (NULL == replica) {
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
} else {
rc = replica_task_done(replica);
}
@@ -913,19 +950,19 @@ replica_task_done(Replica *replica)
mod.mod_type = (char *)TASK_ATTR;
mod.mod_bvalues = NULL;
- slapi_modify_internal_set_pb_ext(pb, replica_sdn, mods, NULL/* controls */,
- NULL/* uniqueid */,
+ slapi_modify_internal_set_pb_ext(pb, replica_sdn, mods, NULL/* controls */,
+ NULL/* uniqueid */,
repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION),
0/* flags */);
slapi_modify_internal_pb (pb);
slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
if ((rc != LDAP_SUCCESS) && (rc != LDAP_NO_SUCH_ATTRIBUTE)) {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_task_done: "
"failed to remove (%s) attribute from (%s) entry; "
"LDAP error - %d\n",
- TASK_ATTR, replica_dn, rc);
+ TASK_ATTR, replica_dn, rc);
}
slapi_pblock_destroy (pb);
@@ -945,9 +982,9 @@ static int replica_execute_cl2ldif_task (Object *r, char *returntext)
if (cl5GetState () != CL5_STATE_OPEN)
{
PR_snprintf (returntext, SLAPI_DSE_RETURNTEXT_SIZE, "changelog is not
open");
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "replica_execute_cl2ldif_task: %s\n", returntext);
- rc = LDAP_OPERATIONS_ERROR;
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "replica_execute_cl2ldif_task: %s\n", returntext);
+ rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
@@ -958,25 +995,25 @@ static int replica_execute_cl2ldif_task (Object *r, char
*returntext)
<replica name>.ldif */
clDir = cl5GetDir ();
if (NULL == clDir) {
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
replica = (Replica*)object_get_data (r);
if (NULL == replica) {
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
PR_snprintf (fName, MAXPATHLEN, "%s/%s.ldif", clDir, replica_get_name
(replica));
slapi_ch_free_string (&clDir);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"Beginning changelog export of replica
\"%s\"\n",
replica_get_name(replica));
rc = cl5ExportLDIF (fName, rlist);
if (rc == CL5_SUCCESS) {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"Finished changelog export of replica
\"%s\"\n",
replica_get_name(replica));
rc = LDAP_SUCCESS;
@@ -984,9 +1021,9 @@ static int replica_execute_cl2ldif_task (Object *r, char
*returntext)
PR_snprintf (returntext, SLAPI_DSE_RETURNTEXT_SIZE,
"Failed changelog export replica %s; "
"changelog error - %d", replica_get_name(replica), rc);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_execute_cl2ldif_task: %s\n", returntext);
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
}
bail:
return rc;
@@ -1004,8 +1041,8 @@ static int replica_execute_ldif2cl_task (Object *r, char
*returntext)
if (cl5GetState () != CL5_STATE_OPEN)
{
PR_snprintf (returntext, SLAPI_DSE_RETURNTEXT_SIZE, "changelog is not
open");
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
- "replica_execute_ldif2cl_task: %s\n", returntext);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ "replica_execute_ldif2cl_task: %s\n", returntext);
rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
@@ -1017,13 +1054,13 @@ static int replica_execute_ldif2cl_task (Object *r, char
*returntext)
<replica name>.ldif */
clDir = cl5GetDir ();
if (NULL == clDir) {
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
replica = (Replica*)object_get_data (r);
if (NULL == replica) {
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
@@ -1035,19 +1072,19 @@ static int replica_execute_ldif2cl_task (Object *r, char
*returntext)
PR_snprintf (returntext, SLAPI_DSE_RETURNTEXT_SIZE,
"failed to close changelog to import changelog data; "
"changelog error - %d", rc);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_execute_ldif2cl_task: %s\n", returntext);
rc = LDAP_OPERATIONS_ERROR;
goto bail;
}
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"Beginning changelog import of replica
\"%s\"\n",
replica_get_name(replica));
imprc = cl5ImportLDIF (clDir, fName, rlist);
slapi_ch_free_string (&clDir);
if (CL5_SUCCESS == imprc)
{
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"Finished changelog import of replica
\"%s\"\n",
replica_get_name(replica));
}
@@ -1056,7 +1093,7 @@ static int replica_execute_ldif2cl_task (Object *r, char
*returntext)
PR_snprintf (returntext, SLAPI_DSE_RETURNTEXT_SIZE,
"Failed changelog import replica %s; "
"changelog error - %d", replica_get_name(replica), rc);
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_execute_ldif2cl_task: %s\n", returntext);
imprc = LDAP_OPERATIONS_ERROR;
}
@@ -1069,10 +1106,10 @@ static int replica_execute_ldif2cl_task (Object *r, char
*returntext)
}
else
{
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name,
"replica_execute_ldif2cl_task: failed to start changelog at %s\n",
config.dir?config.dir:"null config dir");
- rc = LDAP_OPERATIONS_ERROR;
+ rc = LDAP_OPERATIONS_ERROR;
}
bail:
changelog5_config_done(&config);
@@ -1080,7 +1117,7 @@ bail:
return imprc?imprc:rc;
}
-static multimaster_mtnode_extension *
+static multimaster_mtnode_extension *
_replica_config_get_mtnode_ext (const Slapi_Entry *e)
{
const char *replica_root;
@@ -1095,7 +1132,7 @@ _replica_config_get_mtnode_ext (const Slapi_Entry *e)
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_add:
"
"configuration entry %s missing %s attribute\n",
slapi_entry_get_dn((Slapi_Entry *)e),
- attr_replicaRoot);
+ attr_replicaRoot);
return NULL;
}
@@ -1107,14 +1144,14 @@ _replica_config_get_mtnode_ext (const Slapi_Entry *e)
{
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "replica_config_add:
"
"failed to locate mapping tree node for dn %s\n",
- slapi_sdn_get_dn(sdn));
+ slapi_sdn_get_dn(sdn));
}
else
{
/* check if replica object already exists for the specified subtree */
- ext = (multimaster_mtnode_extension *)repl_con_get_ext (REPL_CON_EXT_MTNODE,
mtnode);
+ ext = (multimaster_mtnode_extension *)repl_con_get_ext (REPL_CON_EXT_MTNODE,
mtnode);
}
-
+
slapi_sdn_free (&sdn);
return ext;
@@ -1123,26 +1160,24 @@ _replica_config_get_mtnode_ext (const Slapi_Entry *e)
int
replica_execute_cleanruv_task_ext(Object *r, ReplicaId rid)
{
- slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "cleanruv_extop: calling
clean_ruv_ext\n");
- return replica_execute_cleanruv_task(r, rid, NULL);
+ return replica_execute_cleanruv_task(r, rid, NULL);
}
static int
replica_execute_cleanruv_task (Object *r, ReplicaId rid, char *returntext /* not used
*/)
{
- int rc = 0;
Object *RUVObj;
RUV *local_ruv = NULL;
Replica *replica = (Replica*)object_get_data (r);
-
+ int rc = 0;
PR_ASSERT (replica);
slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "cleanruv_task: cleaning rid
(%d)...\n",(int)rid);
RUVObj = replica_get_ruv(replica);
PR_ASSERT(RUVObj);
local_ruv = (RUV*)object_get_data (RUVObj);
- /* Need to check that :
- * - rid is not the local one
+ /* Need to check that :
+ * - rid is not the local one
* - rid is not the last one
*/
if ((replica_get_rid(replica) == rid) ||
@@ -1156,12 +1191,11 @@ replica_execute_cleanruv_task (Object *r, ReplicaId rid, char
*returntext /* not
/* Update Mapping Tree to reflect RUV changes */
consumer5_set_mapping_tree_state_for_replica(replica, NULL);
-
+
/*
- * Clean the changelog RUV's, and set the rids
+ * Clean the changelog RUV's
*/
cl5CleanRUV(rid);
- delete_released_rid();
if (rc != RUV_SUCCESS){
slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanruv_task: task
failed(%d)\n",rc);
@@ -1171,506 +1205,1566 @@ replica_execute_cleanruv_task (Object *r, ReplicaId rid, char
*returntext /* not
return LDAP_SUCCESS;
}
-static int
-replica_execute_cleanall_ruv_task (Object *r, ReplicaId rid, char *returntext)
+const char *
+fetch_attr(Slapi_Entry *e, const char *attrname, const char *default_val)
{
- PRThread *thread = NULL;
- Repl_Connection *conn;
- Replica *replica;
- Object *agmt_obj;
- Repl_Agmt *agmt;
- ConnResult crc;
- cleanruv_data *data = NULL;
- const Slapi_DN *dn = NULL;
- struct berval *payload = NULL;
- char *ridstr = NULL;
- int send_msgid = 0;
- int agmt_count = 0;
- int rc = 0;
+ Slapi_Attr *attr;
+ Slapi_Value *val = NULL;
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: cleaning rid
(%d)...\n",(int)rid);
+ if (slapi_entry_attr_find(e, attrname, &attr) != 0)
+ return default_val;
- /*
- * Grab the replica
- */
- if(r){
- replica = (Replica*)object_get_data (r);
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: replica is
NULL, aborting task\n");
- return -1;
- }
- /*
- * Create payload
- */
- ridstr = slapi_ch_smprintf("%d:%s", rid,
slapi_sdn_get_dn(replica_get_root(replica)));
- payload = create_ruv_payload(ridstr);
- if(payload == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
create ext op payload, aborting task\n");
- slapi_ch_free_string(&ridstr);
- return -1;
- }
+ slapi_attr_first_value(attr, &val);
+ return slapi_value_get_string(val);
+}
+static int
+replica_cleanall_ruv_task(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *eAfter,
+ int *returncode, char *returntext, void *arg)
+{
+ Slapi_Task *task = NULL;
+ const Slapi_DN *task_dn;
+ Slapi_DN *dn = NULL;
+ Object *r;
+ const char *base_dn;
+ const char *rid_str;
+ ReplicaId rid;
+ int rc = SLAPI_DSE_CALLBACK_OK;
+
+ /* allocate new task now */
+ task = slapi_new_task(slapi_entry_get_ndn(e));
+ if(task == NULL){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Failed
to create new task\n");
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
- set_cleaned_rid(rid);
+ /*
+ * Get our task settings
+ */
+ if ((base_dn = fetch_attr(e, "replica-base-dn", 0)) == NULL){
+ *returncode = LDAP_OBJECT_CLASS_VIOLATION;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ if ((rid_str = fetch_attr(e, "replica-id", 0)) == NULL){
+ *returncode = LDAP_OBJECT_CLASS_VIOLATION;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
- agmt_obj = agmtlist_get_first_agreement_for_replica (replica);
- while (agmt_obj)
- {
- agmt = (Repl_Agmt*)object_get_data (agmt_obj);
- if(!agmt_is_enabled(agmt)){
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- continue;
- }
- dn = agmt_get_dn_byref(agmt);
- conn = (Repl_Connection *)agmt_get_connection(agmt);
- if(conn == NULL){
- /* no connection for this agreement, and move on */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: the replica
(%s), is "
- "missing the connection. This replica will not be cleaned.\n",
slapi_sdn_get_dn(dn));
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- continue;
- }
- crc = conn_connect(conn);
- if (CONN_OPERATION_FAILED == crc ){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
connect "
- "to repl agreement connection (%s), error %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_TRANSIENT_ERROR);
- } else if (CONN_SSL_NOT_ENABLED == crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
acquire "
- "repl agmt connection (%s), errror %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_FATAL_ERROR);
- } else {
- conn_cancel_linger(conn);
- crc = conn_send_extended_operation(conn, REPL_CLEANRUV_OID, payload, NULL,
&send_msgid);
- if (CONN_OPERATION_SUCCESS != crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
send "
- "cleanruv extended op to repl agmt (%s), error %d\n",
slapi_sdn_get_dn(dn), crc);
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task:
successfully sent "
- "cleanruv extended op to (%s)\n",slapi_sdn_get_dn(dn));
- agmt_count++;
- }
- conn_start_linger(conn);
- }
- if(crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: replica
(%s) has not "
- "been cleaned. You will need to rerun the CLEANALLRUV task on this
replica.\n", slapi_sdn_get_dn(dn));
- rc = crc;
- }
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- }
+ task_dn = slapi_entry_get_sdn(e);
+ /*
+ * Check the rid
+ */
+ rid = atoi(rid_str);
+ if (rid <= 0 || rid >= READ_ONLY_REPLICA_ID){
+ PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE, "Invalid replica id (%d)
for task - (%s)",
+ rid, slapi_sdn_get_dn(task_dn));
+ cleanruv_log(task, CLEANALLRUV_ID, "%s", returntext);
+ rc = LDAP_OPERATIONS_ERROR;
+ goto out;
+ }
+ /*
+ * Get the replica object
+ */
+ dn = slapi_sdn_new_dn_byval(base_dn);
+ if((r = replica_get_replica_from_dn(dn)) == NULL){
+ *returncode = LDAP_OPERATIONS_ERROR ;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
- /*
- * We're done with the payload, free it.
- */
- if(payload){
- ber_bvfree(payload);
- }
- slapi_ch_free_string(&ridstr);
+ /* clean the RUV's */
+ rc = replica_execute_cleanall_ruv_task (r, rid, task, returntext);
- /*
- * Now run the cleanruv task
- */
- replica_execute_cleanruv_task (r, rid, returntext);
-
- if(rc == 0 && agmt_count > 0){ /* success, but we need to check our replicas
*/
- /*
- * Launch the cleanruv monitoring thread. Once all the replicas are cleaned it will
release the rid
- */
- data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
- if (data == NULL) {
- slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
allocate "
- "cleanruv_data. Aborting task.\n");
- return -1;
- }
- data->repl_obj = r;
- data->rid = rid;
-
- thread = PR_CreateThread(PR_USER_THREAD, replica_cleanallruv_monitor_thread,
- (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
- PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
- if (thread == NULL) {
- slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: unable to
create cleanAllRUV "
- "monitoring thread. Aborting task.\n");
- }
- } else if(rc == 0){ /* success */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Successfully
Finished.\n");
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Task failed
(%d)\n",rc);
- }
+out:
+ if(rc){
+ cleanruv_log(task, CLEANALLRUV_ID, "Task failed...(%d)", rc);
+ *returncode = rc;
+ slapi_task_finish(task, *returncode);
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ } else {
+ rc = SLAPI_DSE_CALLBACK_OK;
+ }
+ slapi_sdn_free(&dn);
- return rc;
+ return rc;
}
/*
- * After all the cleanAllRUV extended ops have been sent, we need to monitoring those
replicas'
- * RUVs. Once the rid is cleaned, then we need to release it, and push this
"release"
- * to the other replicas
+ * CLEANALLRUV task
+ *
+ * [1] Get the maxcsn from the RUV of the rid we want to clean
+ * [2] Create the payload for the "cleanallruv" extended ops
+ * [3] Create "monitor" thread to do the real work.
+ *
*/
-void
-replica_cleanallruv_monitor_thread(void *arg)
+static int
+replica_execute_cleanall_ruv_task (Object *r, ReplicaId rid, Slapi_Task *task, char
*returntext)
{
- Object *agmt_obj;
- LDAP *ld;
- Repl_Connection *conn;
- Repl_Agmt *agmt;
- Replica *replica;;
- cleanruv_data *data = arg;
- LDAPMessage *result, *entry;
- BerElement *ber;
- time_t start_time;
- struct berval **vals;
- char *rid_text;
- char *attrs[2];
- char *attr;
- int replicas_cleaned = 0;
- int found = 0;
- int crc;
- int rc = 0;
- int i;
+ PRThread *thread = NULL;
+ Slapi_Task *pre_task = NULL; /* this is supposed to be null for logging */
+ Replica *replica;
+ Object *ruv_obj;
+ cleanruv_data *data = NULL;
+ CSN *maxcsn = NULL;
+ const RUV *ruv;
+ struct berval *payload = NULL;
+ char *ridstr = NULL;
+ char csnstr[CSN_STRSIZE];
+ int rc = 0;
+
+ if(get_cleanruv_task_count() >= CLEANRIDSIZ){
+ /* we are already running the maximum number of tasks */
+ cleanruv_log(pre_task, CLEANALLRUV_ID,
+ "Exceeded maximum number of active CLEANALLRUV
tasks(%d)",CLEANRIDSIZ);
+ returntext = PR_smprintf("Exceeded maximum number of active CLEANALLRUV
tasks(%d), "
+ "you must wait for one to finish.", CLEANRIDSIZ);
+ return LDAP_UNWILLING_TO_PERFORM;
+ }
- /*
- * Initialize our settings
- */
- attrs[0] = "nsds50ruv";
- attrs[1] = NULL;
- rid_text = slapi_ch_smprintf("{replica %d ldap", data->rid);
- replica = (Replica*)object_get_data (data->repl_obj);
- start_time = current_time();
+ /*
+ * Grab the replica
+ */
+ if(r){
+ replica = (Replica*)object_get_data (r);
+ } else {
+ cleanruv_log(pre_task, CLEANALLRUV_ID, "Replica is NULL, aborting task");
+ rc = -1;
+ goto fail;
+ }
+ /*
+ * Check if this is a consumer
+ */
+ if(replica_get_type(replica) == REPLICA_TYPE_READONLY){
+ /* this is a consumer, send error */
+ cleanruv_log(pre_task, CLEANALLRUV_ID, "Failed to clean rid (%d), task can
not be run on a consumer",rid);
+ if(task){
+ rc = -1;
+ slapi_task_finish(task, rc);
+ }
+ return -1;
+ }
+ /*
+ * Grab the max csn of the deleted replica
+ */
+ ruv_obj = replica_get_ruv(replica);
+ ruv = object_get_data (ruv_obj);
+ if(ruv_get_rid_max_csn(ruv, &maxcsn, rid) == RUV_BAD_DATA){
+ /* no maxcsn, can not proceed */
+ cleanruv_log(pre_task, CLEANALLRUV_ID, "Could not find maxcsn for rid
(%d)", rid);
+ rc = -1;
+ object_release(ruv_obj);
+ goto fail;
+ } else {
+ object_release(ruv_obj);
+ if(maxcsn == NULL || csn_get_replicaid(maxcsn) == 0){
+ /*
+ * This is for consistency with extop csn creation, where
+ * we want the csn string to be "0000000000000000000" not
""
+ */
+ csn_free(&maxcsn);
+ maxcsn = csn_new();
+ csn_init_by_string(maxcsn, "");
+ }
+ csn_as_string(maxcsn, PR_FALSE, csnstr);
+ }
+ /*
+ * Create payload
+ */
+ ridstr = slapi_ch_smprintf("%d:%s:%s", rid,
slapi_sdn_get_dn(replica_get_root(replica)), csnstr);
+ payload = create_ruv_payload(ridstr);
+ slapi_ch_free_string(&ridstr);
+
+ if(payload == NULL){
+ cleanruv_log(pre_task, CLEANALLRUV_ID, "Failed to create extended op
payload, aborting task");
+ rc = -1;
+ goto fail;
+ }
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Waiting for
all the replicas to get cleaned...\n");
+ /*
+ * Launch the cleanallruv thread. Once all the replicas are cleaned it will release
the rid
+ */
+ data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
+ if (data == NULL) {
+ cleanruv_log(pre_task, CLEANALLRUV_ID, "Failed to allocate cleanruv_data.
Aborting task.");
+ rc = -1;
+ goto fail;
+ }
+ data->repl_obj = r;
+ data->replica = replica;
+ data->rid = rid;
+ data->task = task;
+ data->maxcsn = maxcsn;
+ data->payload = payload;
+ data->sdn = NULL;
+
+ thread = PR_CreateThread(PR_USER_THREAD, replica_cleanallruv_thread,
+ (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+ if (thread == NULL) {
+ rc = -1;
+ goto fail;
+ } else {
+ goto done;
+ }
- while(!slapi_is_shutting_down())
- {
- DS_Sleep(PR_SecondsToInterval(10));
- found = 0;
- agmt_obj = agmtlist_get_first_agreement_for_replica (replica);
- while (agmt_obj){
- agmt = (Repl_Agmt*)object_get_data (agmt_obj);
- if(!agmt_is_enabled(agmt)){
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- continue;
- }
- /*
- * Get the replication connection
- */
- conn = (Repl_Connection *)agmt_get_connection(agmt);
- crc = conn_connect(conn);
- if (CONN_OPERATION_FAILED == crc ){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: monitor
thread failed to connect "
- "to repl agreement connection (%s), error %d\n","",
ACQUIRE_TRANSIENT_ERROR);
- continue;
- } else if (CONN_SSL_NOT_ENABLED == crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: monitor
thread failed to acquire "
- "repl agmt connection (%s), error %d\n","",
ACQUIRE_FATAL_ERROR);
- continue;
- }
- /*
- * Get the LDAP connection handle from the conn
- */
- ld = conn_get_ldap(conn);
- if(ld == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: monitor
thread failed to get LDAP "
- "handle from the replication agmt (%s). Moving on to the next
agmt.\n",agmt_get_long_name(agmt));
- continue;
- }
- /*
- * Search this replica for its tombstone/ruv entry
- */
- conn_cancel_linger(conn);
- conn_lock(conn);
- rc = ldap_search_ext_s(ld, slapi_sdn_get_dn(agmt_get_replarea(agmt)),
LDAP_SCOPE_SUBTREE,
- "(&(nsuniqueid=ffffffff-ffffffff-ffffffff-ffffffff)(objectclass=nstombstone))",
- attrs, 0, NULL, NULL, NULL, 0, &result);
- if(rc != LDAP_SUCCESS){
- /*
- * Couldn't contact ldap server, what should we do?
- *
- * Skip it and move on! It's the admin's job to make sure all the replicas
are up and running
- */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: monitor
thread failed to contact "
- "agmt (%s), moving on to the next agmt.\n", agmt_get_long_name(agmt));
- conn_unlock(conn);
- conn_start_linger(conn);
- continue;
- }
- /*
- * There is only one entry. Check its "nsds50ruv" for our cleaned rid
- */
- entry = ldap_first_entry( ld, result );
- if ( entry != NULL ) {
- for ( attr = ldap_first_attribute( ld, entry, &ber ); attr != NULL; attr =
ldap_next_attribute( ld, entry, ber ) ){
- /* make sure the attribute is nsds50ruv */
- if(strcasecmp(attr,"nsds50ruv") != 0){
- continue;
- }
- if ((vals = ldap_get_values_len( ld, entry, attr)) != NULL ) {
- for ( i = 0; vals[i] && vals[i]->bv_val; i++ ) {
- /* look for this replica */
- if(strstr(rid_text, vals[i]->bv_val)){
- /* rid has not been cleaned yet, start over */
- found = 1;
- break;
- }
- }
- ldap_value_free_len(vals);
- }
- ldap_memfree( attr );
- }
- if ( ber != NULL ) {
- ber_free( ber, 0 );
- }
- }
- ldap_msgfree( result );
- /*
- * Unlock the connection, and start the linger timer
- */
- conn_unlock(conn);
- conn_start_linger(conn);
-
- if(found){
- /*
- * If we've been trying to clean these replicas for over an hour, just quit
- */
- if( (current_time() - start_time) > 3600){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: timed out
checking if replicas have "
- "been cleaned. The rid has not been released, you need to rerun the
task.\n");
- goto done;
- }
- /*
- * a rid has not been cleaned yet, go back to sleep and check them again
- */
- break;
- }
+fail:
+ cleanruv_log(pre_task, CLEANALLRUV_ID, "Failed to clean rid (%d)",rid);
+ if(task){
+ slapi_task_finish(task, rc);
+ }
+ if(payload){
+ ber_bvfree(payload);
+ }
+ csn_free(&maxcsn);
+ object_release (r);
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- }
- if(!found){
- /*
- * The replicas have been cleaned! Next, release the rid
- */
- replicas_cleaned = 1;
- break;
- }
- } /* while */
+done:
- /*
- * If the replicas are cleaned, release the rid
- */
- if(replicas_cleaned){
- rc = replica_execute_release_ruv_task(data->repl_obj, data->rid);
- if(rc == 0){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task:
Successfully Finished. All active "
- "replicas have been cleaned.\n");
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Failed:
Replica ID was not released (%d) "
- "You will need to rerun the task.\n", rc);
- }
- } else {
- /*
- * Shutdown
- */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: slapd
shutting down, you will need to rerun the task.\n");
- }
+ return rc;
+}
-done:
- slapi_ch_free((void **)&rid_text);
- slapi_ch_free((void **)&data);
+void
+replica_cleanallruv_thread_ext(void *arg)
+{
+ replica_cleanallruv_thread(arg);
}
/*
- * This function releases the cleaned rid so that it can be reused.
- * We send this operation to all the known active replicas.
+ * CLEANALLRUV Thread
+ *
+ * [1] Wait for the maxcsn to be covered
+ * [2] Make sure all the replicas are alive
+ * [3] Set the cleaned rid
+ * [4] Send the cleanAllRUV extop to all the replicas
+ * [5] Manually send the CLEANRUV task to replicas that do not support CLEANALLRUV
+ * [6] Wait for all the replicas to be cleaned.
+ * [7] Trigger cl trimming, release the rid, and remove all the "cleanallruv"
attributes
+ * from the config.
*/
-static int
-replica_execute_release_ruv_task(Object *r, ReplicaId rid)
+static void
+replica_cleanallruv_thread(void *arg)
{
- Repl_Connection *conn;
- Replica *replica = (Replica*)object_get_data (r);
- Object *agmt_obj;
- Repl_Agmt *agmt;
- ConnResult crc;
- const Slapi_DN *dn = NULL;
- struct berval *payload = NULL;
- char *ridstr = NULL;
- int send_msgid = 0;
- int rc = 0;
+ Object *ruv_obj = NULL;
+ Object *agmt_obj = NULL;
+ Repl_Agmt *agmt = NULL;
+ RUV *ruv = NULL;
+ cleanruv_data *data = arg;
+ char csnstr[CSN_STRSIZE];
+ char *returntext = NULL;
+ char *rid_text = NULL;
+ int found_dirty_rid = 1;
+ int agmt_not_notified = 1;
+ int interval = 10;
+ int free_obj = 0;
+ int rc = 0;
+
+ /*
+ * Initialize our settings
+ */
+ if(data->replica == NULL && data->repl_obj == NULL){
+ /*
+ * This thread was initiated at startup because the process did not finish. Due
+ * to timing issues, we need to wait to grab the replica obj until we get here.
+ */
+ data->repl_obj = replica_get_replica_from_dn(data->sdn);
+ data->replica = (Replica*)object_get_data(data->repl_obj);
+ free_obj = 1;
+ }
+ if(data->replica == NULL && data->repl_obj){
+ data->replica = (Replica*)object_get_data(data->repl_obj);
+ }
+ if( data->replica && data->repl_obj == NULL){
+ data->repl_obj = object_new(data->replica, NULL);
+ free_obj = 1;
+ }
+ if(data->task){
+ slapi_task_begin(data->task, 1);
+ }
+ rid_text = slapi_ch_smprintf("{replica %d ldap", data->rid);
+ csn_as_string(data->maxcsn, PR_FALSE, csnstr);
+
+ /*
+ * Add the cleanallruv task to the repl config - so we can handle restarts
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID, "Cleaning rid (%d)...",
data->rid);
+ add_cleaned_rid(data->rid, data->replica, csnstr); /* marks config that we
started cleaning a rid */
+ /*
+ * First, wait for the maxcsn to be covered
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID, "Waiting to process all the updates
from the deleted replica...");
+ ruv_obj = replica_get_ruv(data->replica);
+ ruv = object_get_data (ruv_obj);
+ while(data->maxcsn && !is_task_aborted(data->rid) &&
!is_cleaned_rid(data->rid) && !slapi_is_shutting_down()){
+ if(csn_get_replicaid(data->maxcsn) == 0 ||
ruv_covers_csn(ruv,data->maxcsn)){
+ /* We are caught up, now we can clean the ruv's */
+ break;
+ }
+ DS_Sleep(PR_SecondsToInterval(5));
+ }
+ object_release(ruv_obj);
+ /*
+ * Next, make sure all the replicas are up and running before sending off the clean
ruv tasks
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID,"Waiting for all the replicas to be
online...");
+ if(check_agmts_are_alive(data->replica, data->rid, data->task)){
+ /* error, aborted or shutdown */
+ goto done;
+ }
+ /*
+ * Make sure all the replicas have seen the max csn
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID,"Waiting for all the replicas to
receive all the deleted replica updates...");
+ if(check_agmts_are_caught_up(data->replica, data->rid, csnstr,
data->task)){
+ /* error, aborted or shutdown */
+ goto done;
+ }
+ /*
+ * Set the rid as notified - this blocks the changelog from sending out updates
+ * during this process, as well as prevents the db ruv from getting polluted.
+ */
+ set_cleaned_rid(data->rid);
+ /*
+ * Now send the cleanruv extended op to all the agreements
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID, "Sending cleanAllRUV task to all the
replicas...");
+ while(agmt_not_notified && !is_task_aborted(data->rid) &&
!slapi_is_shutting_down()){
+ agmt_obj = agmtlist_get_first_agreement_for_replica (data->replica);
+ if(agmt_obj == NULL){
+ /* no agmts, just clean this replica */
+ break;
+ }
+ while (agmt_obj){
+ agmt = (Repl_Agmt*)object_get_data (agmt_obj);
+ if(!agmt_is_enabled(agmt)){ /* skip disabled replicas */
+ agmt_obj = agmtlist_get_next_agreement_for_replica (data->replica,
agmt_obj);
+ continue;
+ }
+ if(replica_cleanallruv_send_extop(agmt, data->rid, data->task,
data->payload, 1) == 0){
+ agmt_not_notified = 0;
+ } else {
+ agmt_not_notified = 1;
+ break;
+ }
+ agmt_obj = agmtlist_get_next_agreement_for_replica (data->replica,
agmt_obj);
+ }
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: releasing rid
(%d)...\n", rid);
+ if(is_task_aborted(data->rid)){
+ goto done;
+ }
+ if(agmt_not_notified == 0){
+ break;
+ }
+ /*
+ * need to sleep between passes
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID, "Not all replicas have received
the "
+ "cleanallruv extended op, retrying in %d seconds",interval);
+ PR_Lock( notify_lock );
+ PR_WaitCondVar( notify_cvar, PR_SecondsToInterval(interval) );
+ PR_Unlock( notify_lock );
+
+ if(interval < 14400){ /* 4 hour max */
+ interval = interval * 2;
+ } else {
+ interval = 14400;
+ }
+ }
+ /*
+ * Run the CLEANRUV task
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID,"Cleaning local ruv's...");
+ replica_execute_cleanruv_task (data->repl_obj, data->rid, returntext);
+ /*
+ * Wait for all the replicas to be cleaned
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID,"Waiting for all the replicas to be
cleaned...");
+
+ interval = 10;
+ while(found_dirty_rid && !is_task_aborted(data->rid) &&
!slapi_is_shutting_down()){
+ agmt_obj = agmtlist_get_first_agreement_for_replica (data->replica);
+ if(agmt_obj == NULL){
+ break;
+ }
+ while (agmt_obj && !slapi_is_shutting_down()){
+ agmt = (Repl_Agmt*)object_get_data (agmt_obj);
+ if(!agmt_is_enabled(agmt)){ /* skip disabled replicas */
+ agmt_obj = agmtlist_get_next_agreement_for_replica (data->replica,
agmt_obj);
+ continue;
+ }
+ if(replica_cleanallruv_check_ruv(agmt, rid_text, data->task) == 0){
+ found_dirty_rid = 0;
+ } else {
+ found_dirty_rid = 1;
+ break;
+ }
+ agmt_obj = agmtlist_get_next_agreement_for_replica (data->replica,
agmt_obj);
+ }
+ /* If the task is abort or everyone is cleaned, break out */
+ if(is_task_aborted(data->rid)){
+ goto done;
+ }
+ if(found_dirty_rid == 0){
+ break;
+ }
+ /*
+ * need to sleep between passes
+ */
+ cleanruv_log(data->task, CLEANALLRUV_ID, "Replicas have not been cleaned
yet, "
+ "retrying in %d seconds", interval);
+ PR_Lock( notify_lock );
+ PR_WaitCondVar( notify_cvar, PR_SecondsToInterval(interval) );
+ PR_Unlock( notify_lock );
+
+ if(interval < 14400){ /* 4 hour max */
+ interval = interval * 2;
+ } else {
+ interval = 14400;
+ }
+ } /* while */
- /*
- * Set the released rid, and trigger cl trimmming
- */
- set_released_rid((int)rid);
- trigger_cl_trimming();
- /*
- * Create payload
- */
- ridstr = slapi_ch_smprintf("%d:%s", rid,
slapi_sdn_get_dn(replica_get_root(replica)));
- payload = create_ruv_payload(ridstr);
- if(payload == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_task: failed to
create ext op payload, aborting op\n");
- rc = -1;
- goto done;
- }
+done:
+ /*
+ * If the replicas are cleaned, release the rid, and trim the changelog
+ */
+ if(!found_dirty_rid){
+ trigger_cl_trimming(data->rid);
+ delete_cleaned_rid(data->replica, data->rid, data->maxcsn);
+ cleanruv_log(data->task, CLEANALLRUV_ID, "Successfully cleaned
rid(%d).", data->rid);
+ slapi_task_finish(data->task, rc);
+ } else {
+ /*
+ * Shutdown or abort
+ */
+ if(!is_task_aborted(data->rid)){
+ cleanruv_log(data->task, CLEANALLRUV_ID,"Server shutting down.
Process will resume at server startup");
+ } else {
+ cleanruv_log(data->task, CLEANALLRUV_ID,"Task aborted for
rid(%d).",data->rid);
+ }
+ if(data->task){
+ slapi_task_finish(data->task, rc);
+ }
+ }
- agmt_obj = agmtlist_get_first_agreement_for_replica (replica);
- while (agmt_obj)
- {
- agmt = (Repl_Agmt*)object_get_data (agmt_obj);
- if(!agmt_is_enabled(agmt)){
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- continue;
- }
- dn = agmt_get_dn_byref(agmt);
- conn = (Repl_Connection *)agmt_get_connection(agmt);
- if(conn == NULL){
- /* no connection for this agreement, log error, and move on */
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- continue;
- }
- crc = conn_connect(conn);
- if (CONN_OPERATION_FAILED == crc ){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_task: failed to
connect "
- "to repl agmt (%s), error %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_TRANSIENT_ERROR);
- rc = LDAP_OPERATIONS_ERROR;
- } else if (CONN_SSL_NOT_ENABLED == crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_task: failed to
acquire "
- "repl agmt (%s), error %d\n",slapi_sdn_get_dn(dn), ACQUIRE_FATAL_ERROR);
- rc = LDAP_OPERATIONS_ERROR;
- } else {
- conn_cancel_linger(conn);
- crc = conn_send_extended_operation(conn, REPL_RELEASERUV_OID, payload, NULL,
&send_msgid);
- if (CONN_OPERATION_SUCCESS != crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
send "
- "releaseRUV extended op to repl agmt (%s), error %d\n",
slapi_sdn_get_dn(dn), crc);
- rc = LDAP_OPERATIONS_ERROR;
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task:
successfully sent "
- "releaseRUV extended op to (%s)\n",slapi_sdn_get_dn(dn));
- }
- conn_start_linger(conn);
- }
- if(crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: replica
(%s) has not "
- "been released. You will need to rerun the task\n",
- slapi_sdn_get_dn(dn));
- rc = crc;
- }
- agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
- }
+ if(data->payload){
+ ber_bvfree(data->payload);
+ }
+ if(data->repl_obj && free_obj){
+ object_release(data->repl_obj);
+ }
+ slapi_sdn_free(&data->sdn);
+ slapi_ch_free_string(&rid_text);
+ csn_free(&data->maxcsn);
+ slapi_ch_free((void **)&data);
+}
-done:
- /*
- * reset the released/clean rid
- */
- if(rc == 0){
- set_released_rid(ALREADY_RELEASED);
- delete_cleaned_rid();
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Successfully
released rid (%d)\n", rid);
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Failed to
release rid (%d), error (%d)\n", rid, rc);
- }
+/*
+ * Waits for all the repl agmts to be have have the maxcsn. Returns error only on abort
or shutdown
+ */
+static int
+check_agmts_are_caught_up(Replica *replica, ReplicaId rid, char *maxcsn, Slapi_Task
*task)
+{
+ Object *agmt_obj;
+ Repl_Agmt *agmt;
+ char *rid_text;
+ int not_all_caughtup = 1;
+ int interval = 10;
+
+ rid_text = slapi_ch_smprintf("{replica %d ldap", rid);
+
+ while(not_all_caughtup && !is_task_aborted(rid) &&
!slapi_is_shutting_down()){
+ agmt_obj = agmtlist_get_first_agreement_for_replica (replica);
+ if(agmt_obj == NULL){
+ not_all_caughtup = 0;
+ break;
+ }
+ while (agmt_obj){
+ agmt = (Repl_Agmt*)object_get_data (agmt_obj);
+ if(!agmt_is_enabled(agmt)){ /* skip disabled replicas */
+ agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
+ continue;
+ }
+ if(replica_cleanallruv_check_maxcsn(agmt, rid_text, maxcsn, task) == 0){
+ not_all_caughtup = 0;
+ } else {
+ not_all_caughtup = 1;
+ break;
+ }
+ agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
+ } /* agmt while */
- if(payload)
- ber_bvfree(payload);
+ if(not_all_caughtup == 0 || is_task_aborted(rid) ){
+ break;
+ }
+ cleanruv_log(task, CLEANALLRUV_ID, "Not all replicas caught up, retrying in
%d seconds",interval);
+ PR_Lock( notify_lock );
+ PR_WaitCondVar( notify_cvar, PR_SecondsToInterval(interval) );
+ PR_Unlock( notify_lock );
+
+ if(interval < 14400){ /* 4 hour max */
+ interval = interval * 2;
+ } else {
+ interval = 14400;
+ }
+ }
+ slapi_ch_free_string(&rid_text);
- slapi_ch_free_string(&ridstr);
+ if(is_task_aborted(rid)){
+ return -1;
+ }
- return rc;
+ return not_all_caughtup;
}
-static struct berval *
-create_ruv_payload(char *value){
- struct berval *req_data = NULL;
- BerElement *tmp_bere = NULL;
+/*
+ * Waits for all the repl agmts to be online. Returns error only on abort or shutdown
+ */
+static int
+check_agmts_are_alive(Replica *replica, ReplicaId rid, Slapi_Task *task)
+{
+ Object *agmt_obj;
+ Repl_Agmt *agmt;
+ int not_all_alive = 1;
+ int interval = 10;
+
+ while(not_all_alive && is_task_aborted(rid) == 0 &&
!slapi_is_shutting_down()){
+ agmt_obj = agmtlist_get_first_agreement_for_replica (replica);
+ if(agmt_obj == NULL){
+ not_all_alive = 0;
+ break;
+ }
+ while (agmt_obj){
+ agmt = (Repl_Agmt*)object_get_data (agmt_obj);
+ if(!agmt_is_enabled(agmt)){ /* skip disabled replicas */
+ agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
+ continue;
+ }
+ if(replica_cleanallruv_replica_alive(agmt) == 0){
+ not_all_alive = 0;
+ } else {
+ not_all_alive = 1;
+ break;
+ }
+ agmt_obj = agmtlist_get_next_agreement_for_replica (replica, agmt_obj);
+ }
- if ((tmp_bere = der_alloc()) == NULL){
- goto error;
- }
- if (ber_printf(tmp_bere, "{s", value) == -1){
- goto error;
- }
+ if(not_all_alive == 0 || is_task_aborted(rid)){
+ break;
+ }
+ cleanruv_log(task, CLEANALLRUV_ID, "Not all replicas online, retrying in %d
seconds...",interval);
+ PR_Lock( notify_lock );
+ PR_WaitCondVar( notify_cvar, PR_SecondsToInterval(interval) );
+ PR_Unlock( notify_lock );
- if (ber_printf(tmp_bere, "}") == -1){
- goto error;
- }
+ if(interval < 14400){ /* 4 hour max */
+ interval = interval * 2;
+ } else {
+ interval = 14400;
+ }
+ }
+ if(is_task_aborted(rid)){
+ return -1;
+ }
- if (ber_flatten(tmp_bere, &req_data) == -1){
- goto error;
- }
+ return not_all_alive;
+}
- goto done;
+/*
+ * Create the CLEANALLRUV extended op payload
+ */
+struct berval *
+create_ruv_payload(char *value)
+{
+ struct berval *req_data = NULL;
+ BerElement *tmp_bere = NULL;
+
+ if ((tmp_bere = der_alloc()) == NULL){
+ goto error;
+ }
+ if (ber_printf(tmp_bere, "{s", value) == -1){
+ goto error;
+ }
+ if (ber_printf(tmp_bere, "}") == -1){
+ goto error;
+ }
+ if (ber_flatten(tmp_bere, &req_data) == -1){
+ goto error;
+ }
+ goto done;
error:
- if (NULL != req_data){
- ber_bvfree(req_data);
- req_data = NULL;
- }
+ if (NULL != req_data){
+ ber_bvfree(req_data);
+ req_data = NULL;
+ }
done:
- if (NULL != tmp_bere){
- ber_free(tmp_bere, 1);
- tmp_bere = NULL;
- }
+ if (NULL != tmp_bere){
+ ber_free(tmp_bere, 1);
+ tmp_bere = NULL;
+ }
+
+ return req_data;
+}
+
+/*
+ * Manually add the CLEANRUV task to replicas that do not support
+ * the CLEANALLRUV task.
+ */
+static void
+replica_send_cleanruv_task(Repl_Agmt *agmt, ReplicaId rid, Slapi_Task *task)
+{
+ Repl_Connection *conn;
+ ConnResult crc = 0;
+ LDAP *ld;
+ Slapi_DN *dn;
+ struct berval *vals[2];
+ struct berval val;
+ LDAPMod *mods[2];
+ LDAPMod mod;
+ char *repl_dn = NULL;
+ char data[15];
+ int rc;
- return req_data;
+ if((conn = conn_new(agmt)) == NULL){
+ return;
+ }
+ crc = conn_connect(conn);
+ if (CONN_OPERATION_SUCCESS != crc){
+ conn_delete_internal_ext(conn);
+ return;
+ }
+ ld = conn_get_ldap(conn);
+ if(ld == NULL){
+ conn_delete_internal_ext(conn);
+ return;
+ }
+ val.bv_len = PR_snprintf(data, sizeof(data), "CLEANRUV%d", rid);
+ dn = agmt_get_replarea(agmt);
+ mod.mod_op = LDAP_MOD_ADD|LDAP_MOD_BVALUES;
+ mod.mod_type = "nsds5task";
+ mod.mod_bvalues = vals;
+ vals [0] = &val;
+ vals [1] = NULL;
+ val.bv_val = data;
+ mods[0] = &mod;
+ mods[1] = NULL;
+ repl_dn = PR_smprintf("cn=replica,cn=\"%s\",cn=mapping
tree,cn=config", slapi_sdn_get_dn(dn));
+ /*
+ * Add task to remote replica
+ */
+ rc = ldap_modify_ext_s( ld, repl_dn, mods, NULL, NULL);
+
+ if(rc != LDAP_SUCCESS){
+ cleanruv_log(task, CLEANALLRUV_ID, "Failed to add CLEANRUV task replica "
+ "(%s). You will need to manually run the CLEANRUV task on this replica
(%s) error (%d)",
+ agmt_get_long_name(agmt), agmt_get_hostname(agmt), rc);
+ }
+ slapi_ch_free_string(&repl_dn);
+ conn_delete_internal_ext(conn);
}
+/*
+ * Check if the rid is in our list of "cleaned" rids
+ */
int
is_cleaned_rid(ReplicaId rid)
{
- if(rid == cleaned_rid){
- return 1;
- } else {
- return 0;
- }
+ int i;
+
+ slapi_rwlock_rdlock(rid_lock);
+ for(i = 0; i < CLEANRIDSIZ && cleaned_rids[i] != 0; i++){
+ if(rid == cleaned_rids[i]){
+ slapi_rwlock_unlock(rid_lock);
+ return 1;
+ }
+ }
+ slapi_rwlock_unlock(rid_lock);
+
+ return 0;
}
+int
+is_task_aborted(ReplicaId rid)
+{
+ int i;
+
+ if(rid == 0){
+ return 0;
+ }
+ slapi_rwlock_rdlock(abort_rid_lock);
+ for(i = 0; i < CLEANRIDSIZ && aborted_rids[i] != 0; i++){
+ if(rid == aborted_rids[i]){
+ slapi_rwlock_unlock(abort_rid_lock);
+ return 1;
+ }
+ }
+ slapi_rwlock_unlock(abort_rid_lock);
+ return 0;
+}
+
+/*
+ * Just add the rid to the in memory, as we don't want it to survive after a
restart,
+ * This prevent the changelog from sending updates from this rid, and the local ruv
+ * will not be updated either.
+ */
void
-set_cleaned_rid( ReplicaId rid )
+set_cleaned_rid(ReplicaId rid)
{
- cleaned_rid = rid;
+ int i;
+
+ slapi_rwlock_wrlock(rid_lock);
+ for(i = 0; i < CLEANRIDSIZ; i++){
+ if(cleaned_rids[i] == 0){
+ cleaned_rids[i] = rid;
+ cleaned_rids[i + 1] = 0;
+ }
+ }
+ slapi_rwlock_unlock(rid_lock);
}
+/*
+ * Add the rid and maxcsn to the repl config (so we can resume after a server restart)
+ */
void
-delete_cleaned_rid()
+add_cleaned_rid(ReplicaId rid, Replica *r, char *maxcsn)
{
- cleaned_rid = 0;
+ Slapi_PBlock *pb;
+ struct berval *vals[2];
+ struct berval val;
+ LDAPMod *mods[2];
+ LDAPMod mod;
+ char data[CSN_STRSIZE + 10];
+ char *dn;
+ int rc;
+
+ /*
+ * Write the rid & maxcsn to the config entry
+ */
+ val.bv_len = PR_snprintf(data, sizeof(data),"%d:%s", rid, maxcsn);
+ dn = replica_get_dn(r);
+ pb = slapi_pblock_new();
+ mod.mod_op = LDAP_MOD_ADD|LDAP_MOD_BVALUES;
+ mod.mod_type = (char *)type_replicaCleanRUV;
+ mod.mod_bvalues = vals;
+ vals [0] = &val;
+ vals [1] = NULL;
+ val.bv_val = data;
+ mods[0] = &mod;
+ mods[1] = NULL;
+
+ replica_add_cleanruv_data(r, val.bv_val);
+
+ slapi_modify_internal_set_pb (pb, dn, mods, NULL, NULL,
+ repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), 0);
+ slapi_modify_internal_pb (pb);
+ slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+ if (rc != LDAP_SUCCESS){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "CleanAllRUV Task: failed
to update replica "
+ "config (%d), rid (%d)\n", rc, rid);
+ }
+ slapi_ch_free_string(&dn);
+ slapi_pblock_destroy(pb);
}
-int
-get_released_rid()
+/*
+ * Add aborted rid and repl root to config in case of a server restart
+ */
+void
+add_aborted_rid(ReplicaId rid, Replica *r, char *repl_root)
{
- return released_rid;
+ Slapi_PBlock *pb;
+ struct berval *vals[2];
+ struct berval val;
+ LDAPMod *mods[2];
+ LDAPMod mod;
+ char *data;
+ char *dn;
+ int rc;
+ int i;
+
+ slapi_rwlock_wrlock(abort_rid_lock);
+ for(i = 0; i < CLEANRIDSIZ; i++){
+ if(aborted_rids[i] == 0){
+ aborted_rids[i] = rid;
+ aborted_rids[i + 1] = 0;
+ break;
+ }
+ }
+ slapi_rwlock_unlock(abort_rid_lock);
+ /*
+ * Write the rid to the config entry
+ */
+ dn = replica_get_dn(r);
+ pb = slapi_pblock_new();
+ data = PR_smprintf("%d:%s", rid, repl_root);
+ mod.mod_op = LDAP_MOD_ADD|LDAP_MOD_BVALUES;
+ mod.mod_type = (char *)type_replicaAbortCleanRUV;
+ mod.mod_bvalues = vals;
+ vals [0] = &val;
+ vals [1] = NULL;
+ val.bv_val = data;
+ val.bv_len = strlen (data);
+ mods[0] = &mod;
+ mods[1] = NULL;
+
+ slapi_modify_internal_set_pb (pb, dn, mods, NULL, NULL,
+ repl_get_plugin_identity (PLUGIN_MULTIMASTER_REPLICATION), 0);
+ slapi_modify_internal_pb (pb);
+ slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+ if (rc != LDAP_SUCCESS){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort CleanAllRUV Task:
failed to update "
+ "replica config (%d), rid (%d)\n", rc, rid);
+ }
+
+ slapi_ch_free_string(&dn);
+ slapi_ch_free_string(&data);
+ slapi_pblock_destroy(pb);
}
-int
-is_released_rid(int rid)
+void
+delete_aborted_rid(Replica *r, ReplicaId rid, char *repl_root){
+ Slapi_PBlock *pb;
+ LDAPMod *mods[2];
+ LDAPMod mod;
+ struct berval *vals[2];
+ struct berval val;
+ char *data;
+ char *dn;
+ int rc;
+ int i;
+
+ if(r == NULL)
+ return;
+
+ /*
+ * Remove this rid, and optimize the array
+ */
+ slapi_rwlock_wrlock(abort_rid_lock);
+ for(i = 0; i < CLEANRIDSIZ && aborted_rids[i] != rid; i++); /* found rid,
stop */
+ for(; i < CLEANRIDSIZ; i++){
+ /* rewrite entire array */
+ aborted_rids[i] = aborted_rids[i + 1];
+ }
+ slapi_rwlock_unlock(abort_rid_lock);
+ /*
+ * Prepare the mods for the config entry
+ */
+ dn = replica_get_dn(r);
+ pb = slapi_pblock_new();
+ data = PR_smprintf("%d:%s", (int)rid, repl_root);
+
+ mod.mod_op = LDAP_MOD_DELETE|LDAP_MOD_BVALUES;
+ mod.mod_type = (char *)type_replicaAbortCleanRUV;
+ mod.mod_bvalues = vals;
+ vals [0] = &val;
+ vals [1] = NULL;
+ val.bv_val = data;
+ val.bv_len = strlen (data);
+ mods[0] = &mod;
+ mods[1] = NULL;
+
+ slapi_modify_internal_set_pb(pb, dn, mods, NULL, NULL, repl_get_plugin_identity
(PLUGIN_MULTIMASTER_REPLICATION), 0);
+ slapi_modify_internal_pb (pb);
+ slapi_pblock_destroy (pb);
+ slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+ if (rc != LDAP_SUCCESS){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort CleanAllRUV Task:
failed to remove replica "
+ "config (%d), rid (%d)\n", rc, rid);
+ }
+
+ slapi_ch_free_string(&dn);
+ slapi_ch_free_string(&data);
+}
+
+/*
+ * Remove the rid from our list, and the config
+ */
+void
+delete_cleaned_rid(Replica *r, ReplicaId rid, CSN *maxcsn)
{
- if(rid == released_rid){
- return 1;
- } else {
- return 0;
- }
+ Slapi_PBlock *pb;
+ Object *agmt_obj;
+ Repl_Agmt *agmt;
+ LDAPMod *mods[2];
+ LDAPMod mod;
+ struct berval *vals[2];
+ struct berval val;
+ char *dn;
+ char data[CSN_STRSIZE + 10];
+ char csnstr[CSN_STRSIZE];
+ int rc;
+ int i;
+
+ if(r == NULL || maxcsn == NULL)
+ return;
+
+ /*
+ * Remove this rid, and optimize the array
+ */
+ slapi_rwlock_wrlock(rid_lock);
+ for(i = 0; i < CLEANRIDSIZ && cleaned_rids[i] != rid; i++); /* found rid,
stop */
+ for(; i < CLEANRIDSIZ; i++){
+ /* rewrite entire array */
+ cleaned_rids[i] = cleaned_rids[i + 1];
+ }
+ slapi_rwlock_unlock(rid_lock);
+ /*
+ * Prepare the mods for the config entry
+ */
+ csn_as_string(maxcsn, PR_FALSE, csnstr);
+ val.bv_len = PR_snprintf(data, sizeof(data), "%d:%s", (int)rid, csnstr);
+ dn = replica_get_dn(r);
+ pb = slapi_pblock_new();
+ mod.mod_op = LDAP_MOD_DELETE|LDAP_MOD_BVALUES;
+ mod.mod_type = (char *)type_replicaCleanRUV;
+ mod.mod_bvalues = vals;
+ vals [0] = &val;
+ vals [1] = NULL;
+ val.bv_val = data;
+ mods[0] = &mod;
+ mods[1] = NULL;
+
+ replica_remove_cleanruv_data(r, data);
+
+ slapi_modify_internal_set_pb(pb, dn, mods, NULL, NULL, repl_get_plugin_identity
(PLUGIN_MULTIMASTER_REPLICATION), 0);
+ slapi_modify_internal_pb (pb);
+ slapi_pblock_get(pb, SLAPI_PLUGIN_INTOP_RESULT, &rc);
+ if (rc != LDAP_SUCCESS){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "CleanAllRUV Task: failed
to remove replica config "
+ "(%d), rid (%d)\n", rc, rid);
+ }
+ slapi_pblock_destroy (pb);
+ slapi_ch_free_string(&dn);
+ /*
+ * Now release the cleaned rid from the repl agmts
+ */
+ agmt_obj = agmtlist_get_first_agreement_for_replica (r);
+ while (agmt_obj){
+ agmt = (Repl_Agmt*)object_get_data (agmt_obj);
+ if(!agmt_is_enabled(agmt)){
+ agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
+ continue;
+ }
+ agmt_set_cleanruv_data(agmt, rid, CLEANRUV_RELEASED);
+ agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
+ }
}
+/*
+ * Abort the CLEANALLRUV task
+ */
int
-is_already_released_rid()
+replica_cleanall_ruv_abort(Slapi_PBlock *pb, Slapi_Entry *e, Slapi_Entry *eAfter,
+ int *returncode, char *returntext, void *arg)
{
- if(released_rid == ALREADY_RELEASED){
- return 1;
- } else {
- return 0;
- }
+ PRThread *thread = NULL;
+ struct berval *payload = NULL;
+ Slapi_Task *task = NULL;
+ Replica *replica;
+ ReplicaId rid;
+ cleanruv_data *data = NULL;
+ const Slapi_DN *dn;
+ Object *r;
+ CSN *maxcsn;
+ const char *base_dn;
+ const char *rid_str;
+ char *ridstr;
+ int rc = SLAPI_DSE_CALLBACK_OK;
+
+ if(get_abort_cleanruv_task_count() >= CLEANRIDSIZ){
+ /* we are already running the maximum number of tasks */
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,
+ "Exceeded maximum number of active ABORT CLEANALLRUV
tasks(%d)",CLEANRIDSIZ);
+ returntext = PR_smprintf("Exceeded maximum number of active ABORT
CLEANALLRUV tasks(%d), "
+ "you must wait for one to finish.", CLEANRIDSIZ);
+ *returncode = LDAP_OPERATIONS_ERROR;
+ return SLAPI_DSE_CALLBACK_ERROR;
+ }
+
+ /* allocate new task now */
+ task = slapi_new_task(slapi_entry_get_ndn(e));
+ /*
+ * Get our task settings
+ */
+ if ((rid_str = fetch_attr(e, "replica-id", 0)) == NULL){
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,"Missing required attr
\"replica-id\"");
+ *returncode = LDAP_OBJECT_CLASS_VIOLATION;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ if ((base_dn = fetch_attr(e, "replica-base-dn", 0)) == NULL){
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,"Missing required attr
\"replica-base-dn\"");
+ *returncode = LDAP_OBJECT_CLASS_VIOLATION;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ /*
+ * Check the rid
+ */
+ rid = atoi(rid_str);
+ if (rid <= 0 || rid >= READ_ONLY_REPLICA_ID){
+ PR_snprintf(returntext, SLAPI_DSE_RETURNTEXT_SIZE, "Invalid replica id (%d)
for task - (%s)",
+ rid, slapi_sdn_get_dn(slapi_entry_get_sdn(e)));
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,"%s", returntext);
+ *returncode = LDAP_OPERATIONS_ERROR;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ /*
+ * Get the replica object
+ */
+ dn = slapi_sdn_new_dn_byval(base_dn);
+ if((r = replica_get_replica_from_dn(dn)) == NULL){
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,"Failed to find replica from
dn(%s)", base_dn);
+ *returncode = LDAP_OPERATIONS_ERROR;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ /*
+ * Create payload
+ */
+ ridstr = slapi_ch_smprintf("%d:%s", rid, base_dn);
+ payload = create_ruv_payload(ridstr);
+
+ if(payload == NULL){
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID, "Failed to create extended op
payload, aborting task");
+ *returncode = LDAP_OPERATIONS_ERROR;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ /*
+ * Stop the cleaning, and delete the rid
+ */
+ replica = (Replica*)object_get_data (r);
+ maxcsn = replica_get_cleanruv_maxcsn(replica, rid);
+ delete_cleaned_rid(replica, rid, maxcsn);
+ add_aborted_rid(rid, replica, (char *)base_dn);
+ stop_ruv_cleaning();
+ /*
+ * Prepare the abort struct and fire off the thread
+ */
+ data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
+ if (data == NULL) {
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,"Failed to allocate
abort_cleanruv_data. Aborting task.");
+ *returncode = LDAP_OPERATIONS_ERROR;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ goto out;
+ }
+ data->repl_obj = r; /* released in replica_abort_task_thread() */
+ data->replica = replica;
+ data->task = task;
+ data->payload = payload;
+ data->rid = rid;
+ data->repl_root = slapi_ch_strdup(base_dn);
+ data->sdn = NULL;
+
+ thread = PR_CreateThread(PR_USER_THREAD, replica_abort_task_thread,
+ (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+ if (thread == NULL) {
+ object_release(r);
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID,"Unable to create abort thread.
Aborting task.");
+ *returncode = LDAP_OPERATIONS_ERROR;
+ rc = SLAPI_DSE_CALLBACK_ERROR;
+ }
+
+out:
+
+ csn_free(&maxcsn);
+ slapi_ch_free_string(&ridstr);
+
+ if(rc != SLAPI_DSE_CALLBACK_OK){
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID, "Abort Task failed (%d)",
rc);
+ slapi_task_finish(task, rc);
+ }
+
+ return rc;
+}
+
+/*
+ * Abort CLEANALLRUV task thread
+ */
+void
+replica_abort_task_thread(void *arg)
+{
+ cleanruv_data *data = (cleanruv_data *)arg;
+ Repl_Agmt *agmt;
+ Object *agmt_obj;
+ int agmt_not_notified = 1;
+ int interval = 10;
+ int release_it = 0;
+
+ /*
+ * Need to build the replica from the dn
+ */
+ if(data->replica == NULL && data->repl_obj == NULL){
+ /*
+ * This thread was initiated at startup because the process did not finish. Due
+ * to timing issues, we need to wait to grab the replica obj until we get here.
+ */
+ if((data->repl_obj = replica_get_replica_from_dn(data->sdn)) == NULL){
+ cleanruv_log(data->task, ABORT_CLEANALLRUV_ID, "Failed to get replica
from dn (%s).", slapi_sdn_get_dn(data->sdn));
+ goto done;
+ }
+ if(data->replica == NULL && data->repl_obj){
+ data->replica = (Replica*)object_get_data(data->repl_obj);
+ }
+ release_it = 1;
+ }
+
+ /*
+ * Now send the cleanruv extended op to all the agreements
+ */
+ while(agmt_not_notified && !slapi_is_shutting_down()){
+ agmt_obj = agmtlist_get_first_agreement_for_replica (data->replica);
+ while (agmt_obj){
+ agmt = (Repl_Agmt*)object_get_data (agmt_obj);
+ if(!agmt_is_enabled(agmt)){ /* skip disabled replicas */
+ agmt_obj = agmtlist_get_next_agreement_for_replica (data->replica,
agmt_obj);
+ continue;
+ }
+ if(replica_cleanallruv_send_abort_extop(agmt, data->task,
data->payload)){
+ agmt_not_notified = 1;
+ break;
+ } else {
+ /* success */
+ agmt_not_notified = 0;
+ }
+ agmt_obj = agmtlist_get_next_agreement_for_replica (data->replica,
agmt_obj);
+ } /* while loop for agmts */
+
+ if(agmt_not_notified == 0){
+ /* everybody has been contacted */
+ break;
+ }
+ /*
+ * need to sleep between passes
+ */
+ cleanruv_log(data->task, ABORT_CLEANALLRUV_ID,"Retrying in %d
seconds",interval);
+ PR_Lock( notify_lock );
+ PR_WaitCondVar( notify_cvar, PR_SecondsToInterval(interval) );
+ PR_Unlock( notify_lock );
+
+ if(interval < 14400){ /* 4 hour max */
+ interval = interval * 2;
+ } else {
+ interval = 14400;
+ }
+ } /* while */
+
+done:
+ if(agmt_not_notified){
+ /* failure */
+ cleanruv_log(data->task, ABORT_CLEANALLRUV_ID,"Abort task failed, will
resume the task at the next server startup.");
+ } else {
+ /*
+ * Clean up the config
+ */
+ delete_aborted_rid(data->replica, data->rid, data->repl_root);
+ cleanruv_log(data->task, ABORT_CLEANALLRUV_ID, "Successfully aborted
cleanAllRUV task for rid(%d)", data->rid);
+ }
+ if(data->task){
+ slapi_task_finish(data->task, agmt_not_notified);
+ }
+ if(data->repl_obj && release_it)
+ object_release(data->repl_obj);
+ if(data->payload){
+ ber_bvfree(data->payload);
+ }
+ slapi_ch_free_string(&data->repl_root);
+ slapi_sdn_free(&data->sdn);
+ slapi_ch_free((void **)&data);
+}
+
+static int
+replica_cleanallruv_send_abort_extop(Repl_Agmt *ra, Slapi_Task *task, struct berval
*payload)
+{
+ Repl_Connection *conn = NULL;
+ ConnResult crc = 0;
+ int msgid = 0;
+ int rc = 0;
+
+ if((conn = conn_new(ra)) == NULL){
+ return -1;
+ }
+ if(conn_connect(conn) == CONN_OPERATION_SUCCESS){
+ crc = conn_send_extended_operation(conn, REPL_ABORT_CLEANRUV_OID, payload, NULL,
&msgid);
+ /*
+ * success or failure, just return the error code
+ */
+ rc = crc;
+ if(rc){
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID, "Failed to send extop to
replica(%s).", agmt_get_long_name(ra));
+ }
+ } else {
+ cleanruv_log(task, ABORT_CLEANALLRUV_ID, "Failed to connect to
replica(%s).", agmt_get_long_name(ra));
+ rc = -1;
+ }
+ conn_delete_internal_ext(conn);
+
+ return rc;
+}
+
+
+static int
+replica_cleanallruv_send_extop(Repl_Agmt *ra, ReplicaId rid, Slapi_Task *task, struct
berval *payload, int check_result)
+{
+ Repl_Connection *conn = NULL;
+ ConnResult crc = 0;
+ int msgid = 0;
+ int rc = 0;
+
+ if((conn = conn_new(ra)) == NULL){
+ return -1;
+ }
+ if(conn_connect(conn) == CONN_OPERATION_SUCCESS){
+ crc = conn_send_extended_operation(conn, REPL_CLEANRUV_OID, payload, NULL,
&msgid);
+ if(crc == CONN_OPERATION_SUCCESS && check_result){
+ struct berval *retsdata = NULL;
+ char *retoid = NULL;
+
+ crc = conn_read_result_ex(conn, &retoid, &retsdata, NULL, msgid,
NULL, 1);
+ if (CONN_OPERATION_SUCCESS == crc ){
+ struct berval **ruv_bervals = NULL;
+ struct berval *data = NULL;
+ char *data_guid = NULL;
+
+ decode_repl_ext_response(retsdata, &rc, &ruv_bervals,
&data_guid, &data);
+ /* just free everything, we only wanted "rc" */
+ slapi_ch_free_string(&data_guid);
+ if(data)
+ ber_bvfree(data);
+ if (ruv_bervals)
+ ber_bvecfree(ruv_bervals);
+
+ if(rc == 0 ){ /* rc == 1 is success */
+ cleanruv_log(task, CLEANALLRUV_ID,"Replica %s does not support
the CLEANALLRUV task. Sending replica CLEANRUV task...",
+ slapi_sdn_get_dn(agmt_get_dn_byref(ra)));
+ /*
+ * Ok, this replica doesn't know about CLEANALLRUV, so just
manually
+ * add the CLEANRUV task to the replica.
+ */
+ replica_send_cleanruv_task(ra, rid, task);
+ } else {
+ /* extop was accepted */
+ rc = 0;
+ }
+ if (NULL != retoid)
+ ldap_memfree(retoid);
+ if (NULL != retsdata)
+ ber_bvfree(retsdata);
+ }
+ agmt_set_cleanruv_data(ra, rid, CLEANRUV_NOTIFIED);
+ } else {
+ /*
+ * success or failure, just return the error code
+ */
+ rc = crc;
+ }
+ } else {
+ rc =-1;
+ }
+ conn_delete_internal_ext(conn);
+
+ return rc;
+}
+
+static int
+replica_cleanallruv_check_maxcsn(Repl_Agmt *agmt, char *rid_text, char *maxcsn,
Slapi_Task *task)
+{
+ Repl_Connection *conn = NULL;
+ LDAP *ld;
+ Slapi_DN *dn = agmt_get_replarea(agmt);
+ struct berval **vals;
+ LDAPMessage *result = NULL, *entry = NULL;
+ BerElement *ber;
+ char *attrs[2];
+ char *attr = NULL;
+ char *iter = NULL;
+ char *ruv_part = NULL;
+ int found_rid = 0;
+ int part_count = 0;
+ int rc = 0, i;
+
+ if((conn = conn_new(agmt)) == NULL){
+ return -1;
+ }
+
+ if(conn_connect(conn) == CONN_OPERATION_SUCCESS){
+ attrs[0] = "nsds50ruv";
+ attrs[1] = NULL;
+ ld = conn_get_ldap(conn);
+ if(ld == NULL){
+ conn_delete_internal_ext(conn);
+ return -1;
+ }
+ rc = ldap_search_ext_s(ld, slapi_sdn_get_dn(dn), LDAP_SCOPE_SUBTREE,
+
"(&(nsuniqueid=ffffffff-ffffffff-ffffffff-ffffffff)(objectclass=nstombstone))",
+ attrs, 0, NULL, NULL, NULL, 0, &result);
+ slapi_sdn_free(&dn);
+ if(rc != LDAP_SUCCESS){
+ cleanruv_log(task, CLEANALLRUV_ID,"Failed to contact "
+ "agmt (%s) error (%d), will retry later.",
agmt_get_long_name(agmt), rc);
+ conn_delete_internal_ext(conn);
+ return -1;
+ }
+ entry = ldap_first_entry( ld, result );
+ if ( entry != NULL ) {
+ for ( attr = ldap_first_attribute( ld, entry, &ber ); attr != NULL; attr
= ldap_next_attribute( ld, entry, ber ) ){
+ /* make sure the attribute is nsds50ruv */
+ if(strcasecmp(attr,"nsds50ruv") != 0){
+ ldap_memfree( attr );
+ continue;
+ }
+ found_rid = 0;
+ if ((vals = ldap_get_values_len( ld, entry, attr)) != NULL ) {
+ for ( i = 0; vals[i] && vals[i]->bv_val; i++ ) {
+ /* look for this replica */
+ if(strstr(vals[i]->bv_val, rid_text)){
+ found_rid = 1;
+ /* get the max csn compare it to our known max csn */
+ ruv_part = ldap_utf8strtok_r(vals[i]->bv_val, "
", &iter);
+ for(part_count = 1; ruv_part && part_count < 5;
part_count++){
+ ruv_part = ldap_utf8strtok_r(iter, " ",
&iter);
+ }
+ if(part_count == 5){
+ /* we have the maxcsn */
+ if(strcmp(ruv_part, maxcsn)){
+ /* we are not caught up yet, free, and return */
+ ldap_value_free_len(vals);
+ ldap_memfree( attr );
+ ldap_msgfree( result );
+ if(ber){
+ ber_free( ber, 0 );
+ }
+ conn_delete_internal_ext(conn);
+ return -1;
+ } else {
+ /* ok this replica has all the updates from the
deleted replica */
+ rc = 0;
+ }
+ } else {
+ /* there is no maxcsn for this rid - treat it as caught
up */
+ rc = 0;
+ }
+ }
+ }
+ if(!found_rid){
+ /* must have been cleaned already */
+ rc = 0;
+ }
+ ldap_value_free_len(vals);
+ }
+ ldap_memfree( attr );
+ }
+ if ( ber != NULL ) {
+ ber_free( ber, 0 );
+ }
+ }
+ if(result)
+ ldap_msgfree( result );
+ } else {
+ rc = -1;
+ }
+ conn_delete_internal_ext(conn);
+
+ return rc;
+}
+
+static int
+replica_cleanallruv_replica_alive(Repl_Agmt *agmt)
+{
+ Repl_Connection *conn = NULL;
+ LDAP *ld = NULL;
+ LDAPMessage *result = NULL;
+ int rc = 0;
+
+ if((conn = conn_new(agmt)) == NULL){
+ return -1;
+ }
+ if(conn_connect(conn) == CONN_OPERATION_SUCCESS){
+ ld = conn_get_ldap(conn);
+ if(ld == NULL){
+ slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "cleanAllRUV_task:
failed to get LDAP "
+ "handle from the replication agmt (%s). Moving on to the next
agmt.\n",agmt_get_long_name(agmt));
+ conn_delete_internal_ext(conn);
+ return -1;
+ }
+ if(ldap_search_ext_s(ld, "", LDAP_SCOPE_BASE,
"objectclass=top",
+ NULL, 0, NULL, NULL, NULL, 0, &result) == LDAP_SUCCESS)
+ {
+ rc = 0;
+ } else {
+ rc = -1;
+ }
+ if(result)
+ ldap_msgfree( result );
+ } else {
+ rc = -1;
+ }
+ conn_delete_internal_ext(conn);
+
+ return rc;
}
+static int
+replica_cleanallruv_check_ruv(Repl_Agmt *ra, char *rid_text, Slapi_Task *task)
+{
+ Repl_Connection *conn = NULL;
+ BerElement *ber = NULL;
+ struct berval **vals = NULL;
+ LDAPMessage *result = NULL, *entry = NULL;
+ LDAP *ld = NULL;
+ Slapi_DN *dn = agmt_get_replarea(ra);
+ char *attrs[2];
+ char *attr = NULL;
+ int rc = 0, i;
+
+ if((conn = conn_new(ra)) == NULL){
+ return -1;
+ }
+ if(conn_connect(conn) == CONN_OPERATION_SUCCESS){
+ attrs[0] = "nsds50ruv";
+ attrs[1] = NULL;
+ ld = conn_get_ldap(conn);
+ if(ld == NULL){
+ cleanruv_log(task, CLEANALLRUV_ID,"Failed to get LDAP handle from "
+ "the replication agmt (%s). Moving on to the next
agmt.",agmt_get_long_name(ra));
+ rc = -1;
+ goto done;
+ }
+
+ rc = ldap_search_ext_s(ld, slapi_sdn_get_dn(dn), LDAP_SCOPE_SUBTREE,
+
"(&(nsuniqueid=ffffffff-ffffffff-ffffffff-ffffffff)(objectclass=nstombstone))",
+ attrs, 0, NULL, NULL, NULL, 0, &result);
+ slapi_sdn_free(&dn);
+ if(rc != LDAP_SUCCESS){
+ cleanruv_log(task, CLEANALLRUV_ID,"Failed to contact "
+ "agmt (%s) error (%d), will retry later.",
agmt_get_long_name(ra), rc);
+ rc = -1;
+ goto done;
+ }
+ entry = ldap_first_entry( ld, result );
+ if ( entry != NULL ) {
+ for ( attr = ldap_first_attribute( ld, entry, &ber ); attr != NULL; attr
= ldap_next_attribute( ld, entry, ber ) ){
+ /* make sure the attribute is nsds50ruv */
+ if(strcasecmp(attr,"nsds50ruv") != 0){
+ ldap_memfree( attr );
+ continue;
+ }
+ if ((vals = ldap_get_values_len( ld, entry, attr)) != NULL ) {
+ for ( i = 0; vals[i] && vals[i]->bv_val; i++ ) {
+ /* look for this replica */
+ if(strstr(vals[i]->bv_val, rid_text)){
+ /* rid has not been cleaned yet, free and return */
+ rc = -1;
+ ldap_value_free_len(vals);
+ ldap_memfree( attr );
+ if ( ber != NULL ) {
+ ber_free( ber, 0 );
+ ber = NULL;
+ }
+ goto done;
+ } else {
+ rc = 0;
+ }
+ }
+ ldap_value_free_len(vals);
+ }
+ ldap_memfree( attr );
+ } /* for loop */
+ if ( ber != NULL ) {
+ ber_free( ber, 0 );
+ }
+ }
+done:
+ if(result)
+ ldap_msgfree( result );
+ } else {
+ return -1;
+ }
+ conn_delete_internal_ext(conn);
+
+ return rc;
+}
+
+static int
+get_cleanruv_task_count()
+{
+ int i, count = 0;
+
+ slapi_rwlock_wrlock(rid_lock);
+ for(i = 0; i < CLEANRIDSIZ; i++){
+ if(cleaned_rids[i] != 0){
+ count++;
+ }
+ }
+ slapi_rwlock_unlock(rid_lock);
+
+ return count;
+}
+
+static int
+get_abort_cleanruv_task_count()
+{
+ int i, count = 0;
+
+ slapi_rwlock_wrlock(rid_lock);
+ for(i = 0; i < CLEANRIDSIZ; i++){
+ if(aborted_rids[i] != 0){
+ count++;
+ }
+ }
+ slapi_rwlock_unlock(rid_lock);
+
+ return count;
+}
+
+/*
+ * Notify sleeping CLEANALLRUV threads to stop
+ */
void
-set_released_rid( int rid )
+stop_ruv_cleaning()
{
- released_rid = rid;
+ if(notify_lock){
+ PR_Lock( notify_lock );
+ PR_NotifyCondVar( notify_cvar );
+ PR_Unlock( notify_lock );
+ }
}
+/*
+ * Write our logging to the task and error log
+ */
void
-delete_released_rid()
+cleanruv_log(Slapi_Task *task, char *task_type, char *fmt, ...)
{
- released_rid = 0;
+ va_list ap1;
+ va_list ap2;
+ va_list ap3;
+ va_list ap4;
+ char *errlog_fmt;
+
+ va_start(ap1, fmt);
+ va_start(ap2, fmt);
+ va_start(ap3, fmt);
+ va_start(ap4, fmt);
+
+ if(task){
+ slapi_task_log_notice_ext(task, fmt, ap1);
+ slapi_task_log_status_ext(task, fmt, ap2);
+ slapi_task_inc_progress(task);
+ }
+ errlog_fmt = PR_smprintf("%s: %s\n",task_type, fmt);
+ slapi_log_error_ext(SLAPI_LOG_FATAL, repl_plugin_name, errlog_fmt, ap3, ap4);
+ slapi_ch_free_string(&errlog_fmt);
+
+ va_end(ap1);
+ va_end(ap2);
+ va_end(ap3);
+ va_end(ap4);
}
+
diff --git a/ldap/servers/plugins/replication/repl5_ruv.c
b/ldap/servers/plugins/replication/repl5_ruv.c
index 9ad5c5a..fb8ea09 100644
--- a/ldap/servers/plugins/replication/repl5_ruv.c
+++ b/ldap/servers/plugins/replication/repl5_ruv.c
@@ -487,9 +487,9 @@ ruv_replace_replica_purl (RUV *ruv, ReplicaId rid, const char
*replica_purl)
replica = ruvGetReplica (ruv, rid);
if (replica != NULL)
{
- if (strcmp(replica->replica_purl, replica_purl)) { /* purl updated */
+ if (replica->replica_purl == NULL || strcmp(replica->replica_purl,
replica_purl)) { /* purl updated */
/* Replace replica_purl in RUV since supplier has been updated. */
- slapi_ch_free((void **)&(replica->replica_purl));
+ slapi_ch_free_string(&replica->replica_purl);
replica->replica_purl = slapi_ch_strdup(replica_purl);
/* Also, reset csn and min_csn. */
replica->csn = replica->min_csn = NULL;
@@ -863,10 +863,6 @@ ruv_covers_csn_internal(const RUV *ruv, const CSN *csn, PRBool
strict)
{
rid = csn_get_replicaid(csn);
replica = ruvGetReplica (ruv, rid);
- if((is_released_rid(rid)) || (replica == NULL && is_already_released_rid()) ){
- /* this is a released rid, so return true */
- return PR_TRUE;
- }
if (replica == NULL)
{
/*
@@ -930,7 +926,7 @@ ruv_covers_csn_strict(const RUV *ruv, const CSN *csn)
* or max{maxcsns of all ruv elements} if get_the_max != 0.
*/
static int
-ruv_get_min_or_max_csn(const RUV *ruv, CSN **csn, int get_the_max)
+ruv_get_min_or_max_csn(const RUV *ruv, CSN **csn, int get_the_max, ReplicaId rid)
{
int return_value;
@@ -960,12 +956,18 @@ ruv_get_min_or_max_csn(const RUV *ruv, CSN **csn, int get_the_max)
{
continue;
}
-
- if (found == NULL ||
- (!get_the_max && csn_compare(found, replica->csn)>0) ||
- ( get_the_max && csn_compare(found, replica->csn)<0))
- {
- found = replica->csn;
+ if(rid){ /* we are only interested in this rid's maxcsn */
+ if(replica->rid == rid){
+ found = replica->csn;
+ break;
+ }
+ } else {
+ if (found == NULL ||
+ (!get_the_max && csn_compare(found, replica->csn)>0) ||
+ ( get_the_max && csn_compare(found, replica->csn)<0))
+ {
+ found = replica->csn;
+ }
}
}
if (found == NULL)
@@ -983,15 +985,20 @@ ruv_get_min_or_max_csn(const RUV *ruv, CSN **csn, int get_the_max)
}
int
+ruv_get_rid_max_csn(const RUV *ruv, CSN **csn, ReplicaId rid){
+ return ruv_get_min_or_max_csn(ruv, csn, 1 /* get the max */, rid);
+}
+
+int
ruv_get_max_csn(const RUV *ruv, CSN **csn)
{
- return ruv_get_min_or_max_csn(ruv, csn, 1 /* get the max */);
+ return ruv_get_min_or_max_csn(ruv, csn, 1 /* get the max */, 0 /* rid */);
}
int
ruv_get_min_csn(const RUV *ruv, CSN **csn)
{
- return ruv_get_min_or_max_csn(ruv, csn, 0 /* get the min */);
+ return ruv_get_min_or_max_csn(ruv, csn, 0 /* get the min */, 0 /* rid */);
}
int
@@ -1097,6 +1104,22 @@ ruv_to_bervals(const RUV *ruv, struct berval ***bvals)
return return_value;
}
+void
+ruv_get_cleaned_rids(RUV *ruv, ReplicaId *rids)
+{
+ RUVElement *replica;
+ int cookie;
+ int i = 0;
+
+ for (replica = dl_get_first (ruv->elements, &cookie); replica;
+ replica = dl_get_next (ruv->elements, &cookie))
+ {
+ if(is_cleaned_rid(replica->rid)){
+ rids[i++] = replica->rid;
+ }
+ }
+}
+
int
ruv_to_smod(const RUV *ruv, Slapi_Mod *smod)
{
@@ -1428,15 +1451,14 @@ int ruv_add_csn_inprogress (RUV *ruv, const CSN *csn)
slapi_rwlock_wrlock (ruv->lock);
if(is_cleaned_rid(rid)){
- slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "ruv_add_csn_inprogress:
invalid replica ID"
- "(%d), aborting update\n", rid);
/* return success because we want to consume the update, but not perform it */
+ rc = RUV_COVERS_CSN;
goto done;
}
replica = ruvGetReplica (ruv, rid);
if (replica == NULL)
{
- replica = ruvAddReplicaNoCSN (ruv, csn_get_replicaid (csn), NULL/*purl*/);
+ replica = ruvAddReplicaNoCSN (ruv, rid, NULL/*purl*/);
if (replica == NULL)
{
if (slapi_is_loglevel_set(SLAPI_LOG_REPL)) {
diff --git a/ldap/servers/plugins/replication/repl5_ruv.h
b/ldap/servers/plugins/replication/repl5_ruv.h
index d329dc3..f2d96f3 100644
--- a/ldap/servers/plugins/replication/repl5_ruv.h
+++ b/ldap/servers/plugins/replication/repl5_ruv.h
@@ -117,6 +117,7 @@ PRBool ruv_covers_csn(const RUV *ruv, const CSN *csn);
PRBool ruv_covers_csn_strict(const RUV *ruv, const CSN *csn);
int ruv_get_min_csn(const RUV *ruv, CSN **csn);
int ruv_get_max_csn(const RUV *ruv, CSN **csn);
+int ruv_get_rid_max_csn(const RUV *ruv, CSN **csn, ReplicaId rid);
int ruv_enumerate_elements (const RUV *ruv, FNEnumRUV fn, void *arg);
int ruv_to_smod(const RUV *ruv, Slapi_Mod *smod);
int ruv_last_modified_to_smod(const RUV *ruv, Slapi_Mod *smod);
diff --git a/ldap/servers/plugins/replication/repl_extop.c
b/ldap/servers/plugins/replication/repl_extop.c
index f6378d2..88bb9e3 100644
--- a/ldap/servers/plugins/replication/repl_extop.c
+++ b/ldap/servers/plugins/replication/repl_extop.c
@@ -1338,7 +1338,7 @@ replica_config_get_mtnode_by_dn(const char *dn)
/*
* Decode the ber element passed to us by the cleanAllRUV task
*/
-static int
+int
decode_cleanruv_payload(struct berval *extop_value, char **payload)
{
BerElement *tmp_bere = NULL;
@@ -1373,313 +1373,313 @@ free_and_return:
return rc;
}
-/*
- * Process the REPL_CLEANRUV_OID extended operation.
- *
- * The payload consists of the replica ID, and the repl root dn. Since this is
- * basically a replication operation, it could of originated here and bounced
- * back from another master. So check the rid against the "cleaned_rid". If
- * it's a match, then we were already here, and we can just return success.
- *
- * Otherwise, we the set the cleaned_rid from the payload, fire off extended ops
- * to all the replica agreements on this replica. Then perform the actual
- * cleanruv_task on this replica.
- */
int
-multimaster_extop_cleanruv(Slapi_PBlock *pb){
+multimaster_extop_abort_cleanruv(Slapi_PBlock *pb)
+{
multimaster_mtnode_extension *mtnode_ext;
PRThread *thread = NULL;
- Repl_Connection *conn;
- const Slapi_DN *dn;
- Replica *r = NULL;
- Object *agmt_obj;
- Repl_Agmt *agmt;
- ConnResult crc;
- cleanruv_data *data = NULL;
- struct berval *extop_value;
+ cleanruv_data *data;
+ Replica *r;
+ ReplicaId rid;
+ CSN *maxcsn;
+ struct berval *extop_payload;
char *extop_oid;
char *repl_root;
char *payload = NULL;
char *iter;
- int send_msgid = 0;
- int agmt_count = 0;
- int rid = 0;
int rc = 0;
slapi_pblock_get(pb, SLAPI_EXT_OP_REQ_OID, &extop_oid);
- slapi_pblock_get(pb, SLAPI_EXT_OP_REQ_VALUE, &extop_value);
+ slapi_pblock_get(pb, SLAPI_EXT_OP_REQ_VALUE, &extop_payload);
if (NULL == extop_oid || strcmp(extop_oid, REPL_CLEANRUV_OID) != 0 ||
- NULL == extop_value || NULL == extop_value->bv_val){
+ NULL == extop_payload || NULL == extop_payload->bv_val){
/* something is wrong, error out */
- return -1;
+ return LDAP_OPERATIONS_ERROR;
}
/*
- * Extract the rid and repl_root from the payload
+ * Decode the payload, and grab our settings
*/
- if(decode_cleanruv_payload(extop_value, &payload)){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
decode payload. Aborting ext op\n");
- return -1;
+ if(decode_cleanruv_payload(extop_payload, &payload)){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort cleanAllRUV task: failed
to decode payload. Aborting ext op\n");
+ return LDAP_OPERATIONS_ERROR;
}
rid = atoi(ldap_utf8strtok_r(payload, ":", &iter));
repl_root = ldap_utf8strtok_r(iter, ":", &iter);
- /*
- * If we already cleaned this server, just return success
- */
- if(is_cleaned_rid(rid)){
- slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "cleanAllRUV_extop: rid (%d) has
already been cleaned, skipping\n",rid);
- return rc;
+ if(!is_cleaned_rid(rid) || is_task_aborted(rid)){
+ /* This replica has already been aborted, or was never cleaned, or already finished
cleaning */
+ goto out;
} else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: cleaning
rid (%d)...\n", rid);
- set_cleaned_rid(rid);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort cleanAllRUV task:
aborting cleanallruv task for rid(%d)\n", rid);
}
-
/*
* Get the node, so we can get the replica and its agreements
*/
if((mtnode_ext = replica_config_get_mtnode_by_dn(repl_root)) == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
get replication node "
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort cleanAllRUV task: failed
to get replication node "
"from (%s), aborting operation\n", repl_root);
- return -1;
+ rc = LDAP_OPERATIONS_ERROR;
+ goto out;
}
- if (mtnode_ext->replica)
+ if (mtnode_ext->replica){
object_acquire (mtnode_ext->replica);
- if (mtnode_ext->replica == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: replica is
missing from (%s), "
+ } else {
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort cleanAllRUV task:
replica is missing from (%s), "
"aborting operation\n",repl_root);
rc = LDAP_OPERATIONS_ERROR;
- goto free_and_return;
+ goto out;
}
r = (Replica*)object_get_data (mtnode_ext->replica);
+ if(r == NULL){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "Abort cleanAllRUV task:
replica is NULL, aborting task\n");
+ rc = LDAP_OPERATIONS_ERROR;
+ goto out;
+ }
/*
- * Send out extended ops to each repl agreement
+ * Prepare the abort data
*/
- agmt_obj = agmtlist_get_first_agreement_for_replica (r);
- while (agmt_obj)
- {
- agmt = (Repl_Agmt*)object_get_data (agmt_obj);
- if(!agmt_is_enabled(agmt)){
- agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
- continue;
- }
- dn = agmt_get_dn_byref(agmt);
- conn = (Repl_Connection *)agmt_get_connection(agmt);
- if(conn == NULL){
- /* no connection for this agreement, move on to the next agmt */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: the
replica (%s), is "
- "missing the connection. This replica will not be cleaned.\n",
slapi_sdn_get_dn(dn));
- agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
- continue;
- }
- crc = conn_connect(conn);
- if (CONN_OPERATION_FAILED == crc ){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
connect "
- "to repl agreement connection (%s), error %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_TRANSIENT_ERROR);
- rc = LDAP_OPERATIONS_ERROR;
- } else if (CONN_SSL_NOT_ENABLED == crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
acquire "
- "repl agmt connection (%s), error %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_FATAL_ERROR);
- rc = LDAP_OPERATIONS_ERROR;
- } else {
- conn_cancel_linger(conn);
- crc = conn_send_extended_operation(conn, REPL_CLEANRUV_OID, extop_value, NULL,
&send_msgid);
- if (CONN_OPERATION_SUCCESS != crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
send "
- "clean_ruv extended op to repl agmt (%s), error %d\n",
slapi_sdn_get_dn(dn), crc);
- rc = LDAP_OPERATIONS_ERROR;
- } else {
- /* success */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop:
successfully sent "
- "extended op to (%s)\n",slapi_sdn_get_dn(dn) );
- agmt_count++;
- }
- conn_start_linger(conn);
- }
- if(crc != CONN_OPERATION_SUCCESS){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: replica
(%s) has not "
- "been cleaned. You will need to rerun the CLEANALLRUV task on this
replica\n",
- slapi_sdn_get_dn(dn) );
- rc = LDAP_OPERATIONS_ERROR;
- }
- agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
- }
-
- /* now clean the ruv */
- replica_execute_cleanruv_task_ext(mtnode_ext->replica, rid);
-
-free_and_return:
-
- if(rc == 0 && agmt_count > 0){
- /*
- * Launch the cleanruv monitoring thread. Once all the replicas are cleaned it will
release the rid
- */
- data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
- if (data == NULL) {
- slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
allocate "
- "cleanruv_Data\n");
- return -1;
- }
- data->repl_obj = mtnode_ext->replica;
- data->rid = rid;
-
- thread = PR_CreateThread(PR_USER_THREAD, replica_cleanallruv_monitor_thread,
- (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
- PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
- if (thread == NULL) {
- slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: unable to
create cleanAllRUV "
- "monitoring thread. Aborting task.\n");
+ data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
+ if (data == NULL) {
+ slapi_log_error( SLAPI_LOG_REPL, repl_plugin_name, "Abort cleanAllRUV task: failed
to allocate "
+ "abort_cleanruv_data. Aborting task.\n");
+ rc = LDAP_OPERATIONS_ERROR;
+ goto out;
+ }
+ data->repl_obj = mtnode_ext->replica; /* released in replica_abort_task_thread()
*/
+ data->replica = r;
+ data->task = NULL;
+ data->payload = slapi_ch_bvdup(extop_payload);
+ data->rid = rid;
+ data->repl_root = slapi_ch_strdup(repl_root);
+ /*
+ * Stop the cleaning, and delete the rid
+ */
+ maxcsn = replica_get_cleanruv_maxcsn(r, rid);
+ delete_cleaned_rid(r, rid, maxcsn);
+ csn_free(&maxcsn);
+ add_aborted_rid(rid, r, repl_root);
+ stop_ruv_cleaning();
+ /*
+ * Send out the extended ops to the replicas
+ */
+ thread = PR_CreateThread(PR_USER_THREAD, replica_abort_task_thread,
+ (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+ if (thread == NULL) {
+ if(mtnode_ext->replica){
+ object_release(mtnode_ext->replica);
}
- } else if (rc == 0){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop:
Successfully Finished.\n");
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanALLRUV_extop: failed to
clean rid (%d), error (%d)\n",rid, rc);
+ slapi_log_error( SLAPI_LOG_REPL, repl_plugin_name, "Abort cleanAllRUV task: unable
to create abort "
+ "thread. Aborting task.\n");
+ rc = LDAP_OPERATIONS_ERROR;
}
- if (mtnode_ext->replica)
- object_release (mtnode_ext->replica);
+out:
+ slapi_ch_free_string(&payload);
return rc;
}
-
/*
- * Process the REPL_RELEASERUV_OID extended operation
+ * Process the REPL_CLEANRUV_OID extended operation.
*
- * Once, all the replicas in the replication farm have been cleaned, then
- * we need to "release" the cleaned_rid, or else we will reject all updates
- * that come from that rid until we restart the server.
+ * The payload consists of the replica ID, repl root dn, and the maxcsn. Since this is
+ * basically a replication operation, it could of originated here and bounced
+ * back from another master. So check the rid against the "cleaned_rid". If
+ * it's a match, then we were already here, and we can just return success.
*
- * We set the cleaned_ruv to zero(invalid rid), and then fire off extended
- * operations to all of the replicas
+ * Otherwise, we the set the cleaned_rid from the payload, fire off extended ops
+ * to all the replica agreements on this replica. Then perform the actual
+ * cleanruv_task on this replica.
*/
int
-multimaster_extop_releaseruv(Slapi_PBlock *pb){
+multimaster_extop_cleanruv(Slapi_PBlock *pb)
+{
multimaster_mtnode_extension *mtnode_ext;
- Repl_Connection *conn;
- const Slapi_DN *dn;
+ PRThread *thread = NULL;
Replica *r = NULL;
- Object *agmt_obj;
- Repl_Agmt *agmt;
- ConnResult crc;
- struct berval *extop_value;
- char *payload = NULL;
+ cleanruv_data *data = NULL;
+ CSN *maxcsn;
+ struct berval *extop_payload;
+ struct berval *resp_bval = NULL;
+ BerElement *resp_bere = NULL;
char *extop_oid;
char *repl_root;
+ char *payload = NULL;
+ char *csnstr = NULL;
char *iter;
- int send_msgid = 0;
+ int release_it = 0;
int rid = 0;
int rc = 0;
slapi_pblock_get(pb, SLAPI_EXT_OP_REQ_OID, &extop_oid);
- slapi_pblock_get(pb, SLAPI_EXT_OP_REQ_VALUE, &extop_value);
+ slapi_pblock_get(pb, SLAPI_EXT_OP_REQ_VALUE, &extop_payload);
- if (NULL == extop_oid || strcmp(extop_oid, REPL_RELEASERUV_OID) != 0 ||
- NULL == extop_value || NULL == extop_value->bv_val){
+ if (NULL == extop_oid || strcmp(extop_oid, REPL_CLEANRUV_OID) != 0 ||
+ NULL == extop_payload || NULL == extop_payload->bv_val){
/* something is wrong, error out */
- return -1;
+ rc = -1;
+ goto free_and_return;
}
-
- if(decode_cleanruv_payload(extop_value, &payload)){
- slapi_log_error(SLAPI_LOG_FATAL,repl_plugin_name, "releaseRUV_extop: failed to
decode payload, aborting ext op.\n");
- return -1;
+ /*
+ * Decode the payload
+ */
+ if(decode_cleanruv_payload(extop_payload, &payload)){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
decode payload. Aborting ext op\n");
+ rc = -1;
+ goto free_and_return;
}
rid = atoi(ldap_utf8strtok_r(payload, ":", &iter));
repl_root = ldap_utf8strtok_r(iter, ":", &iter);
-
+ csnstr = ldap_utf8strtok_r(iter, ":", &iter);
+ maxcsn = csn_new();
+ csn_init_by_string(maxcsn, csnstr);
/*
- * If we already released this ruv, just return.
+ * If we already cleaned this server, just return success
*/
- if(is_released_rid(rid) || is_already_released_rid()){
- slapi_log_error(SLAPI_LOG_REPL, repl_plugin_name, "cleanAllRUV_extop: rid (%d) has
already been released, skipping.\n",rid);
- return 0;
+ if(is_cleaned_rid(rid)){
+ csn_free(&maxcsn);
+ rc = 1;
+ goto free_and_return;
} else {
- /* set the released rid, and trigger trimming */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: releasing
rid (%d)...\n", rid);
- set_released_rid((int)rid);
- trigger_cl_trimming();
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: cleaning rid
(%d)...\n", rid);
}
+ /*
+ * Get the node, so we can get the replica and its agreements
+ */
if((mtnode_ext = replica_config_get_mtnode_by_dn(repl_root)) == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_extop: failed to
get node "
- "from replication root dn(%s), aborting operation.\n", repl_root);
- return -1;
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
get replication node "
+ "from (%s), aborting operation\n", repl_root);
+ rc = -1;
+ goto free_and_return;
}
- if (mtnode_ext->replica)
+
+ if (mtnode_ext->replica){
object_acquire (mtnode_ext->replica);
+ release_it = 1;
+ }
if (mtnode_ext->replica == NULL){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_extop: replica is
missing from (%s), "
- "aborting operation.\n", repl_root);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: replica is
missing from (%s), "
+ "aborting operation\n",repl_root);
rc = LDAP_OPERATIONS_ERROR;
goto free_and_return;
}
+
r = (Replica*)object_get_data (mtnode_ext->replica);
- /*
- * Loop over the agreements, and send out extended ops
- */
- agmt_obj = agmtlist_get_first_agreement_for_replica (r);
- while (agmt_obj)
- {
- agmt = (Repl_Agmt*)object_get_data (agmt_obj);
- if(!agmt_is_enabled(agmt)){
- agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
- continue;
+ if(r == NULL){
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: replica is
NULL, aborting task\n");
+ rc = -1;
+ goto free_and_return;
+ }
+
+ if(replica_get_type(r) != REPLICA_TYPE_READONLY){
+ /*
+ * Launch the cleanruv monitoring thread. Once all the replicas are cleaned it will
release the rid
+ *
+ * This will also release mtnode_ext->replica
+ */
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: launching
cleanAllRUV thread...\n");
+ data = (cleanruv_data*)slapi_ch_calloc(1, sizeof(cleanruv_data));
+ if (data == NULL) {
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: failed to
allocate "
+ "cleanruv_Data\n");
+ rc = -1;
+ goto free_and_return;
}
- dn = agmt_get_dn_byref(agmt);
- conn = (Repl_Connection *)agmt_get_connection(agmt);
- if(conn == NULL){
- /* no connection for this agreement, log error, and move on */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: the
replica (%s), is "
- "missing the connection. This replica will not be released.\n",
slapi_sdn_get_dn(dn));
- agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
- continue;
+ data->repl_obj = mtnode_ext->replica;
+ data->replica = r;
+ data->rid = rid;
+ data->task = NULL;
+ data->maxcsn = maxcsn;
+ data->payload = slapi_ch_bvdup(extop_payload);
+
+ thread = PR_CreateThread(PR_USER_THREAD, replica_cleanallruv_thread_ext,
+ (void *)data, PR_PRIORITY_NORMAL, PR_GLOBAL_THREAD,
+ PR_UNJOINABLE_THREAD, SLAPD_DEFAULT_THREAD_STACKSIZE);
+ if (thread == NULL) {
+ rc = -1;
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: unable to
create cleanAllRUV "
+ "monitoring thread. Aborting task.\n");
}
- crc = conn_connect(conn);
- if (CONN_OPERATION_FAILED == crc ){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_extop: failed to
connect "
- "to repl agreement connection (%s), error %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_TRANSIENT_ERROR);
- rc = LDAP_OPERATIONS_ERROR;
- } else if (CONN_SSL_NOT_ENABLED == crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "releaseRUV_extop: failed to
acquire "
- "repl agmt connection (%s), error %d\n",slapi_sdn_get_dn(dn),
ACQUIRE_FATAL_ERROR);
- rc = LDAP_OPERATIONS_ERROR;
- } else {
- conn_cancel_linger(conn);
- crc = conn_send_extended_operation(conn, REPL_RELEASERUV_OID, extop_value, NULL,
&send_msgid);
- if (CONN_OPERATION_SUCCESS != crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: failed to
send "
- "releaseRUV extended op to repl agmt (%s), error
%d\n", slapi_sdn_get_dn(dn), crc);
- rc = LDAP_OPERATIONS_ERROR;
+ } else { /* this is a read-only consumer */
+ /*
+ * wait for the maxcsn to be covered
+ */
+ Object *ruv_obj;
+ const RUV *ruv;
+
+ ruv_obj = replica_get_ruv(r);
+ ruv = object_get_data (ruv_obj);
+
+ while(!is_task_aborted(rid) && !slapi_is_shutting_down()){
+ if(!ruv_contains_replica(ruv, rid)){
+ /* we've already been cleaned */
+ break;
+ }
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: checking if
we're caught up...\n");
+ if(ruv_covers_csn(ruv,maxcsn) || csn_get_replicaid(maxcsn) == 0){
+ /* We are caught up */
+ break;
} else {
- /* success */
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop:
successfully sent "
- "releaseRUV extended op to (%s)\n",slapi_sdn_get_dn(dn) );
- rc = 0;
+ char csnstr[CSN_STRSIZE];
+ csn_as_string(maxcsn, PR_FALSE, csnstr);
+ slapi_log_error( SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: not ruv
caught up maxcsn(%s)\n", csnstr);
}
- conn_start_linger(conn);
- }
- if(crc){
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: replica
(%s) has not "
- "been released. You will need to rerun the task.\n",
- slapi_sdn_get_dn(dn) );
+ DS_Sleep(PR_SecondsToInterval(5));
}
- agmt_obj = agmtlist_get_next_agreement_for_replica (r, agmt_obj);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: we're
caught up...\n");
+ /*
+ * Set cleaned rid in memory only - does not survive a server restart
+ */
+ set_cleaned_rid(rid);
+ /*
+ * Clean the ruv
+ */
+ replica_execute_cleanruv_task_ext(mtnode_ext->replica, rid);
+
+ /* free everything */
+ object_release(ruv_obj);
+ csn_free(&maxcsn);
+ if (mtnode_ext->replica && release_it)
+ object_release (mtnode_ext->replica);
+ /*
+ * This read-only replica has no easy way to tell when it's safe to release the
rid.
+ * So we won't release it, not until a server restart.
+ */
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: You must
restart the server if you want to reuse rid(%d).\n", rid);
+ slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_task: Successfully
cleaned rid(%d).\n", rid);
}
free_and_return:
+ if(rc && release_it){
+ if (mtnode_ext->replica)
+ object_release (mtnode_ext->replica);
+ }
+ slapi_ch_free_string(&payload);
+
/*
- * Set the rid as "ALREADY_RELEASED, and remove the cleaned ruv
+ * Craft a message so we know this replica supports the task
*/
- if(rc == 0){
- set_released_rid(ALREADY_RELEASED);
- delete_cleaned_rid();
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop:
Successfully released rid (%d)\n", rid);
- } else {
- slapi_log_error(SLAPI_LOG_FATAL, repl_plugin_name, "cleanAllRUV_extop: Failed to
release rid(%d), error (%d), "
- "please retry the task.\n",rid, rc);
- }
+ if ((resp_bere = der_alloc())){
- if(mtnode_ext->replica)
- object_release (mtnode_ext->replica);
+ ber_int_t response = 1;
+
+ ber_printf(resp_bere, "{e}", response);
+ ber_flatten(resp_bere, &resp_bval);
+ slapi_pblock_set(pb, SLAPI_EXT_OP_RET_VALUE, resp_bval);
+ slapi_send_ldap_result(pb, LDAP_SUCCESS, NULL, NULL, 0, NULL);
+ /* resp_bere */
+ if (NULL != resp_bere)
+ {
+ ber_free(resp_bere, 1);
+ }
+ /* resp_bval */
+ if (NULL != resp_bval)
+ {
+ ber_bvfree(resp_bval);
+ }
+ }
return rc;
}
diff --git a/ldap/servers/plugins/replication/repl_globals.c
b/ldap/servers/plugins/replication/repl_globals.c
index 9445394..383d562 100644
--- a/ldap/servers/plugins/replication/repl_globals.c
+++ b/ldap/servers/plugins/replication/repl_globals.c
@@ -110,6 +110,8 @@ const char *type_replicaChangeCount =
"nsds5ReplicaChangeCount";
const char *type_replicaTombstonePurgeInterval =
"nsds5ReplicaTombstonePurgeInterval";
const char *type_replicaLegacyConsumer = "nsds5ReplicaLegacyConsumer";
const char *type_ruvElementUpdatetime = "nsruvReplicaLastModified";
+const char *type_replicaCleanRUV = "nsds5ReplicaCleanRUV";
+const char *type_replicaAbortCleanRUV = "nsds5ReplicaAbortCleanRUV";
/* Attribute names for replication agreement attributes */
const char *type_nsds5ReplicaHost = "nsds5ReplicaHost";
@@ -128,6 +130,7 @@ const char *type_nsds5ReplicaBusyWaitTime =
"nsds5ReplicaBusyWaitTime";
const char *type_nsds5ReplicaSessionPauseTime =
"nsds5ReplicaSessionPauseTime";
const char *type_nsds5ReplicaEnabled = "nsds5ReplicaEnabled";
const char *type_nsds5ReplicaStripAttrs = "nsds5ReplicaStripAttrs";
+const char *type_nsds5ReplicaCleanRUVnotified =
"nsds5ReplicaCleanRUVNotified";
/* windows sync specific attributes */
const char *type_nsds7WindowsReplicaArea = "nsds7WindowsReplicaSubtree";
@@ -139,7 +142,7 @@ const char *type_nsds7DirsyncCookie = "nsds7DirsyncCookie";
const char *type_winSyncInterval = "winSyncInterval";
const char *type_oneWaySync = "oneWaySync";
-/* To Allow Consumer Initialisation when adding an agreement - */
+/* To Allow Consumer Initialization when adding an agreement - */
const char *type_nsds5BeginReplicaRefresh = "nsds5BeginReplicaRefresh";
static int repl_active_threads;
diff --git a/ldap/servers/slapd/log.c b/ldap/servers/slapd/log.c
index 2ffd1f6..e622485 100644
--- a/ldap/servers/slapd/log.c
+++ b/ldap/servers/slapd/log.c
@@ -1969,6 +1969,33 @@ slapi_log_error( int severity, char *subsystem, char *fmt, ... )
}
int
+slapi_log_error_ext(int severity, char *subsystem, char *fmt, va_list varg1, va_list
varg2)
+{
+ int rc = 0;
+
+ if ( severity < SLAPI_LOG_MIN || severity > SLAPI_LOG_MAX ) {
+ (void)slapd_log_error_proc( subsystem, "slapi_log_error: invalid severity %d
(message %s)\n",
+ severity, fmt );
+ return( -1 );
+ }
+
+ if (
+ #ifdef _WIN32
+ *module_ldap_debug
+ #else
+ slapd_ldap_debug
+ #endif
+ & slapi_log_map[ severity ] )
+ {
+ rc = slapd_log_error_proc_internal( subsystem, fmt, varg1, varg2 );
+ } else {
+ rc = 0; /* nothing to be logged --> always return success */
+ }
+
+ return( rc );
+}
+
+int
slapi_is_loglevel_set ( const int loglevel )
{
diff --git a/ldap/servers/slapd/slapi-plugin.h b/ldap/servers/slapd/slapi-plugin.h
index 3fd8aed..c16800f 100644
--- a/ldap/servers/slapd/slapi-plugin.h
+++ b/ldap/servers/slapd/slapi-plugin.h
@@ -5636,6 +5636,7 @@ int slapi_log_error( int severity, char *subsystem, char *fmt, ...
)
#else
;
#endif
+int slapi_log_error_ext( int severity, char *subsystem, char *fmt, va_list varg1, va_list
varg2);
/* allowed values for the "severity" parameter */
#define SLAPI_LOG_FATAL 0
@@ -6146,6 +6147,8 @@ void slapi_task_log_notice(Slapi_Task *task, char *format, ...)
#else
;
#endif
+void slapi_task_log_status_ext(Slapi_Task *task, char *format, va_list varg);
+void slapi_task_log_notice_ext(Slapi_Task *task, char *format, va_list varg);
/*
* slapi_new_task: create new task, fill in DN, and setup modify callback
diff --git a/ldap/servers/slapd/task.c b/ldap/servers/slapd/task.c
index 6d60775..5fc0b08 100644
--- a/ldap/servers/slapd/task.c
+++ b/ldap/servers/slapd/task.c
@@ -215,6 +215,17 @@ void slapi_task_log_status(Slapi_Task *task, char *format, ...)
slapi_task_status_changed(task);
}
+void slapi_task_log_status_ext(Slapi_Task *task, char *format, va_list ap)
+{
+ if (! task->task_status)
+ task->task_status = (char *)slapi_ch_malloc(10 * LOG_BUFFER);
+ if (! task->task_status)
+ return; /* out of memory? */
+
+ PR_vsnprintf(task->task_status, (10 * LOG_BUFFER), format, ap);
+ slapi_task_status_changed(task);
+}
+
/* this adds a line to the 'nsTaskLog' value, which is cumulative (anything
* logged here is added to the end)
*/
@@ -266,6 +277,51 @@ void slapi_task_log_notice(Slapi_Task *task, char *format, ...)
slapi_task_status_changed(task);
}
+void slapi_task_log_notice_ext(Slapi_Task *task, char *format, va_list ap)
+{
+ char buffer[LOG_BUFFER];
+ size_t len;
+
+ PR_vsnprintf(buffer, LOG_BUFFER, format, ap);
+
+ if (task->task_log_lock) {
+ PR_Lock(task->task_log_lock);
+ }
+ len = 2 + strlen(buffer) + (task->task_log ? strlen(task->task_log) : 0);
+ if ((len > MAX_SCROLLBACK_BUFFER) && task->task_log) {
+ size_t i;
+ char *newbuf;
+
+ /* start from middle of buffer, and find next linefeed */
+ i = strlen(task->task_log)/2;
+ while (task->task_log[i] && (task->task_log[i] != '\n'))
+ i++;
+ if (task->task_log[i])
+ i++;
+ len = strlen(task->task_log) - i + 2 + strlen(buffer);
+ newbuf = (char *)slapi_ch_malloc(len);
+ strcpy(newbuf, task->task_log + i);
+ slapi_ch_free((void **)&task->task_log);
+ task->task_log = newbuf;
+ } else {
+ if (! task->task_log) {
+ task->task_log = (char *)slapi_ch_malloc(len);
+ task->task_log[0] = 0;
+ } else {
+ task->task_log = (char *)slapi_ch_realloc(task->task_log, len);
+ }
+ }
+
+ if (task->task_log[0])
+ strcat(task->task_log, "\n");
+ strcat(task->task_log, buffer);
+ if (task->task_log_lock) {
+ PR_Unlock(task->task_log_lock);
+ }
+
+ slapi_task_status_changed(task);
+}
+
/* update attributes in the entry under "cn=tasks" to match the current
* status of the task. */
void slapi_task_status_changed(Slapi_Task *task)