4 Copyright (C) Andrew Tridgell 2006
6 This program is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, see <http://www.gnu.org/licenses/>.
21 #include "lib/events/events.h"
22 #include "lib/tdb/include/tdb.h"
23 #include "system/network.h"
24 #include "system/filesys.h"
25 #include "../include/ctdb_private.h"
27 #include "lib/util/dlinklist.h"
30 find an attached ctdb_db handle given a name
32 struct ctdb_db_context
*ctdb_db_handle(struct ctdb_context
*ctdb
, const char *name
)
34 struct ctdb_db_context
*tmp_db
;
35 for (tmp_db
=ctdb
->db_list
;tmp_db
;tmp_db
=tmp_db
->next
) {
36 if (strcmp(name
, tmp_db
->db_name
) == 0) {
45 return the lmaster given a key
47 uint32_t ctdb_lmaster(struct ctdb_context
*ctdb
, const TDB_DATA
*key
)
49 uint32_t idx
, lmaster
;
51 idx
= ctdb_hash(key
) % ctdb
->vnn_map
->size
;
52 lmaster
= ctdb
->vnn_map
->map
[idx
];
59 construct an initial header for a record with no ltdb header yet
61 static void ltdb_initial_header(struct ctdb_db_context
*ctdb_db
,
63 struct ctdb_ltdb_header
*header
)
66 /* initial dmaster is the lmaster */
67 header
->dmaster
= ctdb_lmaster(ctdb_db
->ctdb
, &key
);
68 header
->laccessor
= header
->dmaster
;
73 fetch a record from the ltdb, separating out the header information
74 and returning the body of the record. A valid (initial) header is
75 returned if the record is not present
77 int ctdb_ltdb_fetch(struct ctdb_db_context
*ctdb_db
,
78 TDB_DATA key
, struct ctdb_ltdb_header
*header
,
79 TALLOC_CTX
*mem_ctx
, TDB_DATA
*data
)
82 struct ctdb_context
*ctdb
= ctdb_db
->ctdb
;
84 rec
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
85 if (rec
.dsize
< sizeof(*header
)) {
87 /* return an initial header */
88 if (rec
.dptr
) free(rec
.dptr
);
89 if (ctdb
->vnn_map
== NULL
) {
90 /* called from the client */
92 header
->dmaster
= (uint32_t)-1;
95 ltdb_initial_header(ctdb_db
, key
, header
);
100 ctdb_ltdb_store(ctdb_db
, key
, header
, d2
);
104 *header
= *(struct ctdb_ltdb_header
*)rec
.dptr
;
107 data
->dsize
= rec
.dsize
- sizeof(struct ctdb_ltdb_header
);
108 data
->dptr
= talloc_memdup(mem_ctx
,
109 sizeof(struct ctdb_ltdb_header
)+rec
.dptr
,
115 CTDB_NO_MEMORY(ctdb
, data
->dptr
);
123 write a record to a normal database
125 int ctdb_ltdb_store(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
,
126 struct ctdb_ltdb_header
*header
, TDB_DATA data
)
128 struct ctdb_context
*ctdb
= ctdb_db
->ctdb
;
132 if (ctdb
->flags
& CTDB_FLAG_TORTURE
) {
133 struct ctdb_ltdb_header
*h2
;
134 rec
= tdb_fetch(ctdb_db
->ltdb
->tdb
, key
);
135 h2
= (struct ctdb_ltdb_header
*)rec
.dptr
;
136 if (rec
.dptr
&& rec
.dsize
>= sizeof(h2
) && h2
->rsn
> header
->rsn
) {
137 DEBUG(DEBUG_CRIT
,("RSN regression! %llu %llu\n",
138 (unsigned long long)h2
->rsn
, (unsigned long long)header
->rsn
));
140 if (rec
.dptr
) free(rec
.dptr
);
143 rec
.dsize
= sizeof(*header
) + data
.dsize
;
144 rec
.dptr
= talloc_size(ctdb
, rec
.dsize
);
145 CTDB_NO_MEMORY(ctdb
, rec
.dptr
);
147 memcpy(rec
.dptr
, header
, sizeof(*header
));
148 memcpy(rec
.dptr
+ sizeof(*header
), data
.dptr
, data
.dsize
);
150 ret
= tdb_store(ctdb_db
->ltdb
->tdb
, key
, rec
, TDB_REPLACE
);
152 DEBUG(DEBUG_ERR
, (__location__
" Failed to store dynamic data\n"));
155 talloc_free(rec
.dptr
);
161 lock a record in the ltdb, given a key
163 int ctdb_ltdb_lock(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
165 return tdb_chainlock(ctdb_db
->ltdb
->tdb
, key
);
169 unlock a record in the ltdb, given a key
171 int ctdb_ltdb_unlock(struct ctdb_db_context
*ctdb_db
, TDB_DATA key
)
173 int ret
= tdb_chainunlock(ctdb_db
->ltdb
->tdb
, key
);
175 DEBUG(DEBUG_ERR
,("tdb_chainunlock failed\n"));