cae0aea0d92c6c16da6312c08b81cc80093074b1
[kai/samba-autobuild/.git] / source4 / cluster / ctdb / common / ctdb_ltdb.c
1 /* 
2    ctdb ltdb code
3
4    Copyright (C) Andrew Tridgell  2006
5
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.
10    
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.
15    
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/>.
18 */
19
20 #include "includes.h"
21 #include "lib/events/events.h"
22 #include "../tdb/include/tdb.h"
23 #include "system/network.h"
24 #include "system/filesys.h"
25 #include "../include/ctdb_private.h"
26 #include "tdb_wrap.h"
27 #include "lib/util/dlinklist.h"
28
29 /*
30   find an attached ctdb_db handle given a name
31  */
32 struct ctdb_db_context *ctdb_db_handle(struct ctdb_context *ctdb, const char *name)
33 {
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) {
37                         return tmp_db;
38                 }
39         }
40         return NULL;
41 }
42
43
44 /*
45   return the lmaster given a key
46 */
47 uint32_t ctdb_lmaster(struct ctdb_context *ctdb, const TDB_DATA *key)
48 {
49         uint32_t idx, lmaster;
50
51         idx = ctdb_hash(key) % ctdb->vnn_map->size;
52         lmaster = ctdb->vnn_map->map[idx];
53
54         return lmaster;
55 }
56
57
58 /*
59   construct an initial header for a record with no ltdb header yet
60 */
61 static void ltdb_initial_header(struct ctdb_db_context *ctdb_db, 
62                                 TDB_DATA key,
63                                 struct ctdb_ltdb_header *header)
64 {
65         header->rsn = 0;
66         /* initial dmaster is the lmaster */
67         header->dmaster = ctdb_lmaster(ctdb_db->ctdb, &key);
68         header->laccessor = header->dmaster;
69         header->lacount = 0;
70 }
71
72
73 /*
74   fetch a record from the ltdb, separating out the header information
75   and returning the body of the record. A valid (initial) header is
76   returned if the record is not present
77 */
78 int ctdb_ltdb_fetch(struct ctdb_db_context *ctdb_db, 
79                     TDB_DATA key, struct ctdb_ltdb_header *header, 
80                     TALLOC_CTX *mem_ctx, TDB_DATA *data)
81 {
82         TDB_DATA rec;
83         struct ctdb_context *ctdb = ctdb_db->ctdb;
84
85         rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
86         if (rec.dsize < sizeof(*header)) {
87                 TDB_DATA d2;
88                 /* return an initial header */
89                 if (rec.dptr) free(rec.dptr);
90                 if (ctdb->vnn_map == NULL) {
91                         /* called from the client */
92                         ZERO_STRUCTP(data);
93                         header->dmaster = (uint32_t)-1;
94                         return -1;
95                 }
96                 ltdb_initial_header(ctdb_db, key, header);
97                 ZERO_STRUCT(d2);
98                 if (data) {
99                         *data = d2;
100                 }
101                 ctdb_ltdb_store(ctdb_db, key, header, d2);
102                 return 0;
103         }
104
105         *header = *(struct ctdb_ltdb_header *)rec.dptr;
106
107         if (data) {
108                 data->dsize = rec.dsize - sizeof(struct ctdb_ltdb_header);
109                 data->dptr = (unsigned char *)talloc_memdup(mem_ctx, 
110                                            sizeof(struct ctdb_ltdb_header)+rec.dptr,
111                                            data->dsize);
112         }
113
114         free(rec.dptr);
115         if (data) {
116                 CTDB_NO_MEMORY(ctdb, data->dptr);
117         }
118
119         return 0;
120 }
121
122
123 /*
124   fetch a record from the ltdb, separating out the header information
125   and returning the body of the record. A valid (initial) header is
126   returned if the record is not present
127 */
128 int ctdb_ltdb_store(struct ctdb_db_context *ctdb_db, TDB_DATA key, 
129                     struct ctdb_ltdb_header *header, TDB_DATA data)
130 {
131         struct ctdb_context *ctdb = ctdb_db->ctdb;
132         TDB_DATA rec;
133         int ret;
134
135         if (ctdb->flags & CTDB_FLAG_TORTURE) {
136                 struct ctdb_ltdb_header *h2;
137                 rec = tdb_fetch(ctdb_db->ltdb->tdb, key);
138                 h2 = (struct ctdb_ltdb_header *)rec.dptr;
139                 if (rec.dptr && rec.dsize >= sizeof(h2) && h2->rsn > header->rsn) {
140                         DEBUG(0,("RSN regression! %llu %llu\n",
141                                  (unsigned long long)h2->rsn, (unsigned long long)header->rsn));
142                 }
143                 if (rec.dptr) free(rec.dptr);
144         }
145
146         rec.dsize = sizeof(*header) + data.dsize;
147         rec.dptr = (unsigned char *)talloc_size(ctdb, rec.dsize);
148         CTDB_NO_MEMORY(ctdb, rec.dptr);
149
150         memcpy(rec.dptr, header, sizeof(*header));
151         memcpy(rec.dptr + sizeof(*header), data.dptr, data.dsize);
152
153         ret = tdb_store(ctdb_db->ltdb->tdb, key, rec, TDB_REPLACE);
154         talloc_free(rec.dptr);
155
156         return ret;
157 }
158
159
160 /*
161   lock a record in the ltdb, given a key
162  */
163 int ctdb_ltdb_lock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
164 {
165         return tdb_chainlock(ctdb_db->ltdb->tdb, key);
166 }
167
168 /*
169   unlock a record in the ltdb, given a key
170  */
171 int ctdb_ltdb_unlock(struct ctdb_db_context *ctdb_db, TDB_DATA key)
172 {
173         int ret = tdb_chainunlock(ctdb_db->ltdb->tdb, key);
174         if (ret != 0) {
175                 DEBUG(0,("tdb_chainunlock failed\n"));
176         }
177         return ret;
178 }