1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
|
/** BEGIN COPYRIGHT BLOCK
* Copyright (C) 2001 Sun Microsystems, Inc. Used by permission.
* Copyright (C) 2005 Red Hat, Inc.
* All rights reserved.
*
* License: GPL (version 3 or any later version).
* See LICENSE for details.
* END COPYRIGHT BLOCK **/
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif
/* repl_connext.c - replication extension to the Connection object
*/
#include "repl5.h"
/* ***** Supplier side ***** */
/* NOT NEEDED YET */
/* ***** Consumer side ***** */
/* consumer connection extension constructor */
void *
consumer_connection_extension_constructor(void *object __attribute__((unused)), void *parent __attribute__((unused)))
{
consumer_connection_extension *ext = (consumer_connection_extension *)slapi_ch_malloc(sizeof(consumer_connection_extension));
if (ext == NULL) {
slapi_log_err(SLAPI_LOG_PLUGIN, repl_plugin_name, "consumer_connection_extension_constructor - "
"Unable to create replication consumer connection extension - out of memory\n");
} else {
ext->repl_protocol_version = REPL_PROTOCOL_UNKNOWN;
ext->replica_acquired = NULL;
ext->isreplicationsession = 0;
ext->supplier_ruv = NULL;
ext->connection = NULL;
ext->in_use_opid = -1;
ext->lock = PR_NewLock();
if (NULL == ext->lock) {
slapi_log_err(SLAPI_LOG_PLUGIN, repl_plugin_name, "consumer_connection_extension_constructor - "
"Unable to create replication consumer connection extension lock - out of memory\n");
/* no need to go through the full destructor, but still need to free up this memory */
slapi_ch_free((void **)&ext);
ext = NULL;
}
}
return ext;
}
/* consumer connection extension destructor */
void
consumer_connection_extension_destructor(void *ext, void *object __attribute__((unused)), void *parent __attribute__((unused)))
{
PRUint64 connid = 0;
if (ext) {
/* Check to see if this replication session has acquired
* a replica. If so, release it here.
*/
consumer_connection_extension *connext = (consumer_connection_extension *)ext;
if (NULL != connext->replica_acquired) {
Replica *r = connext->replica_acquired;
/* If a total update was in progress, abort it */
if (REPL_PROTOCOL_50_TOTALUPDATE == connext->repl_protocol_version) {
Slapi_PBlock *pb = slapi_pblock_new();
const Slapi_DN *repl_root_sdn = replica_get_root(r);
PR_ASSERT(NULL != repl_root_sdn);
if (NULL != repl_root_sdn) {
slapi_pblock_set(pb, SLAPI_CONNECTION, connext->connection);
slapi_pblock_set(pb, SLAPI_TARGET_SDN, (void *)repl_root_sdn);
slapi_pblock_get(pb, SLAPI_CONN_ID, &connid);
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_destructor - "
"Aborting total update in progress for replicated "
"area %s connid=%" PRIu64 "\n",
slapi_sdn_get_dn(repl_root_sdn), connid);
slapi_stop_bulk_import(pb);
} else {
slapi_log_err(SLAPI_LOG_ERR, repl_plugin_name,
"consumer_connection_extension_destructor - Can't determine root "
"of replicated area.\n");
}
slapi_pblock_destroy(pb);
/* allow reaping again */
replica_set_tombstone_reap_stop(r, PR_FALSE);
}
replica_relinquish_exclusive_access(r, connid, -1);
connext->replica_acquired = NULL;
}
if (connext->supplier_ruv) {
ruv_destroy((RUV **)&connext->supplier_ruv);
}
if (connext->lock) {
PR_DestroyLock(connext->lock);
connext->lock = NULL;
}
connext->in_use_opid = -1;
connext->connection = NULL;
slapi_ch_free((void **)&ext);
}
}
/* Obtain exclusive access to this connection extension.
* Returns the consumer_connection_extension* if successful, else NULL.
*
* This is similar to obtaining exclusive access to the replica, but not identical.
* For the connection extension, you only want to hold on to exclusive access as
* long as it is being actively used to process an operation. Mainly that means
* while processing either a 'start' or an 'end' extended operation. This makes
* certain that if another 'start' or 'end' operation is received on the connection,
* the ops will not trample on each other's state. As soon as you are done with
* that single operation, it is time to relinquish the connection extension.
* That differs from acquiring exclusive access to the replica, which is held over
* after the 'start' operation and relinquished during the 'end' operation.
*/
consumer_connection_extension *
consumer_connection_extension_acquire_exclusive_access(void *conn, PRUint64 connid, int opid)
{
consumer_connection_extension *ret = NULL;
/* step 1, grab the connext */
consumer_connection_extension *connext = (consumer_connection_extension *)
repl_con_get_ext(REPL_CON_EXT_CONN, conn);
if (NULL != connext) {
/* step 2, acquire its lock */
PR_Lock(connext->lock);
/* step 3, see if it is not in use, or in use by us */
if (0 > connext->in_use_opid) {
/* step 4, take it! */
connext->in_use_opid = opid;
ret = connext;
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_acquire_exclusive_access - "
"conn=%" PRIu64 " op=%d Acquired consumer connection extension\n",
connid, opid);
} else if (opid == connext->in_use_opid) {
ret = connext;
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_acquire_exclusive_access - "
"conn=%" PRIu64 " op=%d Reacquired consumer connection extension\n",
connid, opid);
} else {
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_acquire_exclusive_access - "
"conn=%" PRIu64 " op=%d Could not acquire consumer connection extension; it is in use by op=%d\n",
connid, opid, connext->in_use_opid);
}
/* step 5, drop the lock */
PR_Unlock(connext->lock);
} else {
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_acquire_exclusive_access - "
"conn=%" PRIu64 " op=%d Could not acquire consumer extension, it is NULL!\n",
connid, opid);
}
return ret;
}
/* Relinquish exclusive access to this connection extension.
* Returns 0 if exclusive access could NOT be relinquished, and non-zero if it was.
* Specifically:
* 1 if the extension was in use and was relinquished.
* 2 if the extension was not in use to begin with.
*
* The extension will only be relinquished if it was first acquired by this op,
* or if 'force' is TRUE. Do not use 'force' without a legitimate reason, such
* as when destroying the parent connection.
*
* cf. consumer_connection_extension_acquire_exclusive_access() for details on how,
* when, and why you would want to acquire and relinquish exclusive access.
*/
int
consumer_connection_extension_relinquish_exclusive_access(void *conn, PRUint64 connid, int opid, PRBool force)
{
int ret = 0;
/* step 1, grab the connext */
consumer_connection_extension *connext = (consumer_connection_extension *)
repl_con_get_ext(REPL_CON_EXT_CONN, conn);
if (NULL != connext) {
/* step 2, acquire its lock */
PR_Lock(connext->lock);
/* step 3, see if it is in use */
if (0 > connext->in_use_opid) {
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_relinquish_exclusive_access - "
"conn=%" PRIu64 " op=%d Consumer connection extension is not in use\n",
connid, opid);
ret = 2;
} else if (opid == connext->in_use_opid) {
/* step 4, relinquish it (normal) */
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_relinquish_exclusive_access - "
"conn=%" PRIu64 " op=%d Relinquishing consumer connection extension\n",
connid, opid);
connext->in_use_opid = -1;
ret = 1;
} else if (force) {
/* step 4, relinquish it (forced) */
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_relinquish_exclusive_access - "
"conn=%" PRIu64 " op=%d Forced to relinquish consumer connection extension held by op=%d\n",
connid, opid, connext->in_use_opid);
connext->in_use_opid = -1;
ret = 1;
} else {
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_relinquish_exclusive_access - "
"conn=%" PRIu64 " op=%d Not relinquishing consumer connection extension, it is held by op=%d!\n",
connid, opid, connext->in_use_opid);
}
/* step 5, drop the lock */
PR_Unlock(connext->lock);
} else {
slapi_log_err(SLAPI_LOG_REPL, repl_plugin_name,
"consumer_connection_extension_relinquish_exclusive_access - "
"conn=%" PRIu64 " op=%d Could not relinquish consumer extension, it is NULL!\n",
connid, opid);
}
return ret;
}
|
