4 Copyright (C) Andrew Tridgell 2006
5 Copyright (C) Ronnie sahlberg 2011
7 This program is free software; you can redistribute it and/or modify
8 it under the terms of the GNU General Public License as published by
9 the Free Software Foundation; either version 3 of the License, or
10 (at your option) any later version.
12 This program is distributed in the hope that it will be useful,
13 but WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 GNU General Public License for more details.
17 You should have received a copy of the GNU General Public License
18 along with this program; if not, see <http://www.gnu.org/licenses/>.
22 #include "system/network.h"
23 #include "system/filesys.h"
27 #include "lib/tdb_wrap/tdb_wrap.h"
28 #include "lib/util/dlinklist.h"
29 #include "lib/util/debug.h"
31 #include "ctdb_private.h"
33 #include "common/common.h"
34 #include "common/logging.h"
37 find an attached ctdb_db handle given a name
39 struct ctdb_db_context
*ctdb_db_handle(struct ctdb_context
*ctdb
, const char *name
)
41 struct ctdb_db_context
*tmp_db
;
42 for (tmp_db
=ctdb
->db_list
;tmp_db
;tmp_db
=tmp_db
->next
) {
43 if (strcmp(name
, tmp_db
->db_name
) == 0) {
52 return the lmaster given a key
54 uint32_t ctdb_lmaster(struct ctdb_context
*ctdb
, const TDB_DATA
*key
)
56 uint32_t idx
, lmaster
;
58 idx
= ctdb_hash(key
) % ctdb
->vnn_map
->size
;
59 lmaster
= ctdb
->vnn_map
->map
[idx
];
66 construct an initial header for a record with no ltdb header yet
68 static void ltdb_initial_header(struct ctdb_db_context
*ctdb_db
,
70 struct ctdb_ltdb_header
*header
)
73 /* initial dmaster is the lmaster */
74 header
->dmaster
= ctdb_lmaster(ctdb_db
->ctdb
, &key
);
75 header
->flags
= CTDB_REC_FLAG_AUTOMATIC
;
80 fetch a record from the ltdb, separating out the header information
81 and returning the body of the record. A valid (initial) header is
82 returned if the record is not present
84 int ctdb_ltdb_fetch(struct ctdb_db_context
*ctdb_db
,
85 TDB_DATA key
, struct ctdb_ltdb_header
*header
,
86 TALLOC_CTX
*mem_ctx
, TDB_DATA
*data
)
89 struct ctdb_context
*ctdb
= ctdb_db
->ctdb
;
91 rec
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
92 if (rec
.dsize
< sizeof(*header
)) {
93 /* return an initial header */
94 if (rec
.dptr
) free(rec
.dptr
);
95 if (ctdb
->vnn_map
== NULL
) {
96 /* called from the client */
98 header
->dmaster
= (uint32_t)-1;
101 ltdb_initial_header(ctdb_db
, key
, header
);
105 if (ctdb_db
->persistent
|| header
->dmaster
== ctdb_db
->ctdb
->pnn
) {
106 if (ctdb_ltdb_store(ctdb_db
, key
, header
, tdb_null
) != 0) {
108 (__location__
"failed to store initial header\n"));
114 *header
= *(struct ctdb_ltdb_header
*)rec
.dptr
;
117 data
->dsize
= rec
.dsize
- sizeof(struct ctdb_ltdb_header
);
118 data
->dptr
= talloc_memdup(mem_ctx
,
119 sizeof(struct ctdb_ltdb_header
)+rec
.dptr
,
125 CTDB_NO_MEMORY(ctdb
, data
->dptr
);
132 fetch a record from the ltdb, separating out the header information
133 and returning the body of the record.
134 if the record does not exist, *header will be NULL
137 int ctdb_ltdb_fetch_with_header(struct ctdb_db_context
*ctdb_db
,
138 TDB_DATA key
, struct ctdb_ltdb_header
*header
,
139 TALLOC_CTX
*mem_ctx
, TDB_DATA
*data
)
143 rec
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
144 if (rec
.dsize
< sizeof(*header
)) {
152 *header
= *(struct ctdb_ltdb_header
*)rec
.dptr
;
154 data
->dsize
= rec
.dsize
- sizeof(struct ctdb_ltdb_header
);
155 data
->dptr
= talloc_memdup(mem_ctx
,
156 sizeof(struct ctdb_ltdb_header
)+rec
.dptr
,
167 write a record to a normal database
169 int ctdb_ltdb_store(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
,
170 struct ctdb_ltdb_header
*header
, TDB_DATA data
)
172 struct ctdb_context
*ctdb
= ctdb_db
->ctdb
;
175 bool seqnum_suppressed
= false;
177 if (ctdb_db
->ctdb_ltdb_store_fn
) {
178 return ctdb_db
->ctdb_ltdb_store_fn(ctdb_db
, key
, header
, data
);
181 if (ctdb
->flags
& CTDB_FLAG_TORTURE
) {
182 struct ctdb_ltdb_header
*h2
;
183 rec
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
184 h2
= (struct ctdb_ltdb_header
*)rec
.dptr
;
185 if (rec
.dptr
&& rec
.dsize
>= sizeof(h2
) && h2
->rsn
> header
->rsn
) {
186 DEBUG(DEBUG_CRIT
,("RSN regression! %llu %llu\n",
187 (unsigned long long)h2
->rsn
, (unsigned long long)header
->rsn
));
189 if (rec
.dptr
) free(rec
.dptr
);
192 rec
.dsize
= sizeof(*header
) + data
.dsize
;
193 rec
.dptr
= talloc_size(ctdb
, rec
.dsize
);
194 CTDB_NO_MEMORY(ctdb
, rec
.dptr
);
196 memcpy(rec
.dptr
, header
, sizeof(*header
));
197 memcpy(rec
.dptr
+ sizeof(*header
), data
.dptr
, data
.dsize
);
199 /* Databases with seqnum updates enabled only get their seqnum
200 changes when/if we modify the data */
201 if (ctdb_db
->seqnum_update
!= NULL
) {
203 old
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
205 if ( (old
.dsize
== rec
.dsize
)
206 && !memcmp(old
.dptr
+sizeof(struct ctdb_ltdb_header
),
207 rec
.dptr
+sizeof(struct ctdb_ltdb_header
),
208 rec
.dsize
-sizeof(struct ctdb_ltdb_header
)) ) {
209 tdb_remove_flags(ctdb_db
->ltdb
->tdb
, TDB_SEQNUM
);
210 seqnum_suppressed
= true;
212 if (old
.dptr
) free(old
.dptr
);
214 ret
= tdb_store(ctdb_db
->ltdb
->tdb
, key
, rec
, TDB_REPLACE
);
216 DEBUG(DEBUG_ERR
, (__location__
" Failed to store dynamic data\n"));
218 if (seqnum_suppressed
) {
219 tdb_add_flags(ctdb_db
->ltdb
->tdb
, TDB_SEQNUM
);
222 talloc_free(rec
.dptr
);
228 lock a record in the ltdb, given a key
230 int ctdb_ltdb_lock(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
232 return tdb_chainlock(ctdb_db
->ltdb
->tdb
, key
);
236 unlock a record in the ltdb, given a key
238 int ctdb_ltdb_unlock(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
240 int ret
= tdb_chainunlock(ctdb_db
->ltdb
->tdb
, key
);
242 DEBUG(DEBUG_ERR
,("tdb_chainunlock failed on db %s [%s]\n", ctdb_db
->db_name
, tdb_errorstr(ctdb_db
->ltdb
->tdb
)));
249 delete a record from a normal database
251 int ctdb_ltdb_delete(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
253 if (ctdb_db
->persistent
!= 0) {
254 DEBUG(DEBUG_ERR
,("Trying to delete emty record in persistent database\n"));
257 if (tdb_delete(ctdb_db
->ltdb
->tdb
, key
) != 0) {
258 DEBUG(DEBUG_ERR
,("Failed to delete empty record."));
264 int ctdb_trackingdb_add_pnn(struct ctdb_context
*ctdb
, TDB_DATA
*data
, uint32_t pnn
)
266 int byte_pos
= pnn
/ 8;
267 int bit_mask
= 1 << (pnn
% 8);
269 if (byte_pos
+ 1 > data
->dsize
) {
272 buf
= malloc(byte_pos
+ 1);
273 memset(buf
, 0, byte_pos
+ 1);
275 DEBUG(DEBUG_ERR
, ("Out of memory when allocating buffer of %d bytes for trackingdb\n", byte_pos
+ 1));
278 if (data
->dptr
!= NULL
) {
279 memcpy(buf
, data
->dptr
, data
->dsize
);
282 data
->dptr
= (uint8_t *)buf
;
283 data
->dsize
= byte_pos
+ 1;
286 data
->dptr
[byte_pos
] |= bit_mask
;
290 void ctdb_trackingdb_traverse(struct ctdb_context
*ctdb
, TDB_DATA data
, ctdb_trackingdb_cb cb
, void *private_data
)
294 for(i
= 0; i
< data
.dsize
; i
++) {
297 for (j
=0; j
<8; j
++) {
300 if (data
.dptr
[i
] & mask
) {
301 cb(ctdb
, i
* 8 + j
, private_data
);
308 this is the dummy null procedure that all databases support
310 int ctdb_null_func(struct ctdb_call_info
*call
)
316 this is a plain fetch procedure that all databases support
318 int ctdb_fetch_func(struct ctdb_call_info
*call
)
320 call
->reply_data
= &call
->record_data
;
325 this is a plain fetch procedure that all databases support
326 this returns the full record including the ltdb header
328 int ctdb_fetch_with_header_func(struct ctdb_call_info
*call
)
330 call
->reply_data
= talloc(call
, TDB_DATA
);
331 if (call
->reply_data
== NULL
) {
334 call
->reply_data
->dsize
= sizeof(struct ctdb_ltdb_header
) + call
->record_data
.dsize
;
335 call
->reply_data
->dptr
= talloc_size(call
->reply_data
, call
->reply_data
->dsize
);
336 if (call
->reply_data
->dptr
== NULL
) {
339 memcpy(call
->reply_data
->dptr
, call
->header
, sizeof(struct ctdb_ltdb_header
));
340 memcpy(&call
->reply_data
->dptr
[sizeof(struct ctdb_ltdb_header
)], call
->record_data
.dptr
, call
->record_data
.dsize
);