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
;
174 uint32_t hsize
= sizeof(struct ctdb_ltdb_header
);
176 bool seqnum_suppressed
= false;
178 if (ctdb_db
->ctdb_ltdb_store_fn
) {
179 return ctdb_db
->ctdb_ltdb_store_fn(ctdb_db
, key
, header
, data
);
182 if (ctdb
->flags
& CTDB_FLAG_TORTURE
) {
184 struct ctdb_ltdb_header
*h2
;
186 old
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
187 h2
= (struct ctdb_ltdb_header
*)old
.dptr
;
188 if (old
.dptr
!= NULL
&& old
.dsize
>= hsize
&&
189 h2
->rsn
> header
->rsn
) {
191 ("RSN regression! %"PRIu64
" %"PRIu64
"\n",
192 h2
->rsn
, header
->rsn
));
194 if (old
.dptr
!= NULL
) {
199 rec
[0].dsize
= hsize
;
200 rec
[0].dptr
= (uint8_t *)header
;
202 rec
[1].dsize
= data
.dsize
;
203 rec
[1].dptr
= data
.dptr
;
205 /* Databases with seqnum updates enabled only get their seqnum
206 changes when/if we modify the data */
207 if (ctdb_db
->seqnum_update
!= NULL
) {
209 old
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
211 if ((old
.dsize
== hsize
+ data
.dsize
) &&
212 memcmp(old
.dptr
+hsize
, data
.dptr
, data
.dsize
) == 0) {
213 tdb_remove_flags(ctdb_db
->ltdb
->tdb
, TDB_SEQNUM
);
214 seqnum_suppressed
= true;
216 if (old
.dptr
!= NULL
) {
220 ret
= tdb_storev(ctdb_db
->ltdb
->tdb
, key
, rec
, 2, TDB_REPLACE
);
222 DEBUG(DEBUG_ERR
, (__location__
" Failed to store dynamic data\n"));
224 if (seqnum_suppressed
) {
225 tdb_add_flags(ctdb_db
->ltdb
->tdb
, TDB_SEQNUM
);
232 lock a record in the ltdb, given a key
234 int ctdb_ltdb_lock(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
236 return tdb_chainlock(ctdb_db
->ltdb
->tdb
, key
);
240 unlock a record in the ltdb, given a key
242 int ctdb_ltdb_unlock(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
244 int ret
= tdb_chainunlock(ctdb_db
->ltdb
->tdb
, key
);
246 DEBUG(DEBUG_ERR
,("tdb_chainunlock failed on db %s [%s]\n", ctdb_db
->db_name
, tdb_errorstr(ctdb_db
->ltdb
->tdb
)));
253 delete a record from a normal database
255 int ctdb_ltdb_delete(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
257 if (ctdb_db
->persistent
!= 0) {
258 DEBUG(DEBUG_ERR
,("Trying to delete empty record in persistent database\n"));
261 if (tdb_delete(ctdb_db
->ltdb
->tdb
, key
) != 0) {
262 DEBUG(DEBUG_ERR
,("Failed to delete empty record."));
268 int ctdb_trackingdb_add_pnn(struct ctdb_context
*ctdb
, TDB_DATA
*data
, uint32_t pnn
)
270 int byte_pos
= pnn
/ 8;
271 int bit_mask
= 1 << (pnn
% 8);
273 if (byte_pos
+ 1 > data
->dsize
) {
276 buf
= malloc(byte_pos
+ 1);
277 memset(buf
, 0, byte_pos
+ 1);
279 DEBUG(DEBUG_ERR
, ("Out of memory when allocating buffer of %d bytes for trackingdb\n", byte_pos
+ 1));
282 if (data
->dptr
!= NULL
) {
283 memcpy(buf
, data
->dptr
, data
->dsize
);
286 data
->dptr
= (uint8_t *)buf
;
287 data
->dsize
= byte_pos
+ 1;
290 data
->dptr
[byte_pos
] |= bit_mask
;
294 void ctdb_trackingdb_traverse(struct ctdb_context
*ctdb
, TDB_DATA data
, ctdb_trackingdb_cb cb
, void *private_data
)
298 for(i
= 0; i
< data
.dsize
; i
++) {
301 for (j
=0; j
<8; j
++) {
304 if (data
.dptr
[i
] & mask
) {
305 cb(ctdb
, i
* 8 + j
, private_data
);
312 this is the dummy null procedure that all databases support
314 int ctdb_null_func(struct ctdb_call_info
*call
)
320 this is a plain fetch procedure that all databases support
322 int ctdb_fetch_func(struct ctdb_call_info
*call
)
324 call
->reply_data
= &call
->record_data
;
329 this is a plain fetch procedure that all databases support
330 this returns the full record including the ltdb header
332 int ctdb_fetch_with_header_func(struct ctdb_call_info
*call
)
334 call
->reply_data
= talloc(call
, TDB_DATA
);
335 if (call
->reply_data
== NULL
) {
338 call
->reply_data
->dsize
= sizeof(struct ctdb_ltdb_header
) + call
->record_data
.dsize
;
339 call
->reply_data
->dptr
= talloc_size(call
->reply_data
, call
->reply_data
->dsize
);
340 if (call
->reply_data
->dptr
== NULL
) {
343 memcpy(call
->reply_data
->dptr
, call
->header
, sizeof(struct ctdb_ltdb_header
));
344 memcpy(&call
->reply_data
->dptr
[sizeof(struct ctdb_ltdb_header
)], call
->record_data
.dptr
, call
->record_data
.dsize
);