tdb2: TDB_SEQNUM and tdb_get_seqnum support.
[ccan] / ccan / tdb2 / tdb.c
1 #include "private.h"
2 #include <ccan/asprintf/asprintf.h>
3 #include <stdarg.h>
4
5 static enum TDB_ERROR update_rec_hdr(struct tdb_context *tdb,
6                                      tdb_off_t off,
7                                      tdb_len_t keylen,
8                                      tdb_len_t datalen,
9                                      struct tdb_used_record *rec,
10                                      uint64_t h)
11 {
12         uint64_t dataroom = rec_data_length(rec) + rec_extra_padding(rec);
13         enum TDB_ERROR ecode;
14
15         ecode = set_header(tdb, rec, TDB_USED_MAGIC, keylen, datalen,
16                            keylen + dataroom, h);
17         if (ecode == TDB_SUCCESS) {
18                 ecode = tdb_write_convert(tdb, off, rec, sizeof(*rec));
19         }
20         return ecode;
21 }
22
23 static enum TDB_ERROR replace_data(struct tdb_context *tdb,
24                                    struct hash_info *h,
25                                    struct tdb_data key, struct tdb_data dbuf,
26                                    tdb_off_t old_off, tdb_len_t old_room,
27                                    bool growing)
28 {
29         tdb_off_t new_off;
30         enum TDB_ERROR ecode;
31
32         /* Allocate a new record. */
33         new_off = alloc(tdb, key.dsize, dbuf.dsize, h->h, TDB_USED_MAGIC,
34                         growing);
35         if (TDB_OFF_IS_ERR(new_off)) {
36                 return new_off;
37         }
38
39         /* We didn't like the existing one: remove it. */
40         if (old_off) {
41                 add_stat(tdb, frees, 1);
42                 ecode = add_free_record(tdb, old_off,
43                                         sizeof(struct tdb_used_record)
44                                         + key.dsize + old_room);
45                 if (ecode == TDB_SUCCESS)
46                         ecode = replace_in_hash(tdb, h, new_off);
47         } else {
48                 ecode = add_to_hash(tdb, h, new_off);
49         }
50         if (ecode != TDB_SUCCESS) {
51                 return ecode;
52         }
53
54         new_off += sizeof(struct tdb_used_record);
55         ecode = tdb->methods->twrite(tdb, new_off, key.dptr, key.dsize);
56         if (ecode != TDB_SUCCESS) {
57                 return ecode;
58         }
59
60         new_off += key.dsize;
61         ecode = tdb->methods->twrite(tdb, new_off, dbuf.dptr, dbuf.dsize);
62         if (ecode != TDB_SUCCESS) {
63                 return ecode;
64         }
65
66         if (tdb->flags & TDB_SEQNUM)
67                 tdb_inc_seqnum(tdb);
68
69         return TDB_SUCCESS;
70 }
71
72 static enum TDB_ERROR update_data(struct tdb_context *tdb,
73                                   tdb_off_t off,
74                                   struct tdb_data dbuf,
75                                   tdb_len_t extra)
76 {
77         enum TDB_ERROR ecode;
78
79         ecode = tdb->methods->twrite(tdb, off, dbuf.dptr, dbuf.dsize);
80         if (ecode == TDB_SUCCESS && extra) {
81                 /* Put a zero in; future versions may append other data. */
82                 ecode = tdb->methods->twrite(tdb, off + dbuf.dsize, "", 1);
83         }
84         if (tdb->flags & TDB_SEQNUM)
85                 tdb_inc_seqnum(tdb);
86
87         return ecode;
88 }
89
90 enum TDB_ERROR tdb_store(struct tdb_context *tdb,
91                          struct tdb_data key, struct tdb_data dbuf, int flag)
92 {
93         struct hash_info h;
94         tdb_off_t off;
95         tdb_len_t old_room = 0;
96         struct tdb_used_record rec;
97         enum TDB_ERROR ecode;
98
99         off = find_and_lock(tdb, key, F_WRLCK, &h, &rec, NULL);
100         if (TDB_OFF_IS_ERR(off)) {
101                 return off;
102         }
103
104         /* Now we have lock on this hash bucket. */
105         if (flag == TDB_INSERT) {
106                 if (off) {
107                         ecode = TDB_ERR_EXISTS;
108                         goto out;
109                 }
110         } else {
111                 if (off) {
112                         old_room = rec_data_length(&rec)
113                                 + rec_extra_padding(&rec);
114                         if (old_room >= dbuf.dsize) {
115                                 /* Can modify in-place.  Easy! */
116                                 ecode = update_rec_hdr(tdb, off,
117                                                        key.dsize, dbuf.dsize,
118                                                        &rec, h.h);
119                                 if (ecode != TDB_SUCCESS) {
120                                         goto out;
121                                 }
122                                 ecode = update_data(tdb,
123                                                     off + sizeof(rec)
124                                                     + key.dsize, dbuf,
125                                                     old_room - dbuf.dsize);
126                                 if (ecode != TDB_SUCCESS) {
127                                         goto out;
128                                 }
129                                 tdb_unlock_hashes(tdb, h.hlock_start,
130                                                   h.hlock_range, F_WRLCK);
131                                 return TDB_SUCCESS;
132                         }
133                 } else {
134                         if (flag == TDB_MODIFY) {
135                                 /* if the record doesn't exist and we
136                                    are in TDB_MODIFY mode then we should fail
137                                    the store */
138                                 ecode = TDB_ERR_NOEXIST;
139                                 goto out;
140                         }
141                 }
142         }
143
144         /* If we didn't use the old record, this implies we're growing. */
145         ecode = replace_data(tdb, &h, key, dbuf, off, old_room, off);
146 out:
147         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_WRLCK);
148         return ecode;
149 }
150
151 enum TDB_ERROR tdb_append(struct tdb_context *tdb,
152                           struct tdb_data key, struct tdb_data dbuf)
153 {
154         struct hash_info h;
155         tdb_off_t off;
156         struct tdb_used_record rec;
157         tdb_len_t old_room = 0, old_dlen;
158         unsigned char *newdata;
159         struct tdb_data new_dbuf;
160         enum TDB_ERROR ecode;
161
162         off = find_and_lock(tdb, key, F_WRLCK, &h, &rec, NULL);
163         if (TDB_OFF_IS_ERR(off)) {
164                 return off;
165         }
166
167         if (off) {
168                 old_dlen = rec_data_length(&rec);
169                 old_room = old_dlen + rec_extra_padding(&rec);
170
171                 /* Fast path: can append in place. */
172                 if (rec_extra_padding(&rec) >= dbuf.dsize) {
173                         ecode = update_rec_hdr(tdb, off, key.dsize,
174                                                old_dlen + dbuf.dsize, &rec,
175                                                h.h);
176                         if (ecode != TDB_SUCCESS) {
177                                 goto out;
178                         }
179
180                         off += sizeof(rec) + key.dsize + old_dlen;
181                         ecode = update_data(tdb, off, dbuf,
182                                             rec_extra_padding(&rec));
183                         goto out;
184                 }
185
186                 /* Slow path. */
187                 newdata = malloc(key.dsize + old_dlen + dbuf.dsize);
188                 if (!newdata) {
189                         ecode = tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR,
190                                            "tdb_append:"
191                                            " failed to allocate %zu bytes",
192                                            (size_t)(key.dsize + old_dlen
193                                                     + dbuf.dsize));
194                         goto out;
195                 }
196                 ecode = tdb->methods->tread(tdb, off + sizeof(rec) + key.dsize,
197                                             newdata, old_dlen);
198                 if (ecode != TDB_SUCCESS) {
199                         goto out_free_newdata;
200                 }
201                 memcpy(newdata + old_dlen, dbuf.dptr, dbuf.dsize);
202                 new_dbuf.dptr = newdata;
203                 new_dbuf.dsize = old_dlen + dbuf.dsize;
204         } else {
205                 newdata = NULL;
206                 new_dbuf = dbuf;
207         }
208
209         /* If they're using tdb_append(), it implies they're growing record. */
210         ecode = replace_data(tdb, &h, key, new_dbuf, off, old_room, true);
211
212 out_free_newdata:
213         free(newdata);
214 out:
215         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_WRLCK);
216         return ecode;
217 }
218
219 enum TDB_ERROR tdb_fetch(struct tdb_context *tdb, struct tdb_data key,
220                          struct tdb_data *data)
221 {
222         tdb_off_t off;
223         struct tdb_used_record rec;
224         struct hash_info h;
225         enum TDB_ERROR ecode;
226
227         off = find_and_lock(tdb, key, F_RDLCK, &h, &rec, NULL);
228         if (TDB_OFF_IS_ERR(off)) {
229                 return off;
230         }
231
232         if (!off) {
233                 ecode = TDB_ERR_NOEXIST;
234         } else {
235                 data->dsize = rec_data_length(&rec);
236                 data->dptr = tdb_alloc_read(tdb, off + sizeof(rec) + key.dsize,
237                                             data->dsize);
238                 if (TDB_PTR_IS_ERR(data->dptr)) {
239                         ecode = TDB_PTR_ERR(data->dptr);
240                 } else
241                         ecode = TDB_SUCCESS;
242         }
243
244         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_RDLCK);
245         return ecode;
246 }
247
248 bool tdb_exists(struct tdb_context *tdb, TDB_DATA key)
249 {
250         tdb_off_t off;
251         struct tdb_used_record rec;
252         struct hash_info h;
253
254         off = find_and_lock(tdb, key, F_RDLCK, &h, &rec, NULL);
255         if (TDB_OFF_IS_ERR(off)) {
256                 return false;
257         }
258         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_RDLCK);
259
260         return off ? true : false;
261 }
262
263 enum TDB_ERROR tdb_delete(struct tdb_context *tdb, struct tdb_data key)
264 {
265         tdb_off_t off;
266         struct tdb_used_record rec;
267         struct hash_info h;
268         enum TDB_ERROR ecode;
269
270         off = find_and_lock(tdb, key, F_WRLCK, &h, &rec, NULL);
271         if (TDB_OFF_IS_ERR(off)) {
272                 return off;
273         }
274
275         if (!off) {
276                 ecode = TDB_ERR_NOEXIST;
277                 goto unlock;
278         }
279
280         ecode = delete_from_hash(tdb, &h);
281         if (ecode != TDB_SUCCESS) {
282                 goto unlock;
283         }
284
285         /* Free the deleted entry. */
286         add_stat(tdb, frees, 1);
287         ecode = add_free_record(tdb, off,
288                                 sizeof(struct tdb_used_record)
289                                 + rec_key_length(&rec)
290                                 + rec_data_length(&rec)
291                                 + rec_extra_padding(&rec));
292
293         if (tdb->flags & TDB_SEQNUM)
294                 tdb_inc_seqnum(tdb);
295
296 unlock:
297         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_WRLCK);
298         return ecode;
299 }
300
301 unsigned int tdb_get_flags(struct tdb_context *tdb)
302 {
303         return tdb->flags;
304 }
305
306 void tdb_add_flag(struct tdb_context *tdb, unsigned flag)
307 {
308         if (tdb->flags & TDB_INTERNAL) {
309                 tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
310                            "tdb_add_flag: internal db");
311                 return;
312         }
313         switch (flag) {
314         case TDB_NOLOCK:
315                 tdb->flags |= TDB_NOLOCK;
316                 break;
317         case TDB_NOMMAP:
318                 tdb->flags |= TDB_NOMMAP;
319                 tdb_munmap(tdb->file);
320                 break;
321         case TDB_NOSYNC:
322                 tdb->flags |= TDB_NOSYNC;
323                 break;
324         case TDB_SEQNUM:
325                 tdb->flags |= TDB_SEQNUM;
326                 break;
327         default:
328                 tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
329                            "tdb_add_flag: Unknown flag %u", flag);
330         }
331 }
332
333 void tdb_remove_flag(struct tdb_context *tdb, unsigned flag)
334 {
335         if (tdb->flags & TDB_INTERNAL) {
336                 tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
337                            "tdb_remove_flag: internal db");
338                 return;
339         }
340         switch (flag) {
341         case TDB_NOLOCK:
342                 tdb->flags &= ~TDB_NOLOCK;
343                 break;
344         case TDB_NOMMAP:
345                 tdb->flags &= ~TDB_NOMMAP;
346                 tdb_mmap(tdb);
347                 break;
348         case TDB_NOSYNC:
349                 tdb->flags &= ~TDB_NOSYNC;
350                 break;
351         case TDB_SEQNUM:
352                 tdb->flags &= ~TDB_SEQNUM;
353                 break;
354         default:
355                 tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
356                            "tdb_remove_flag: Unknown flag %u", flag);
357         }
358 }
359
360 const char *tdb_errorstr(enum TDB_ERROR ecode)
361 {
362         /* Gcc warns if you miss a case in the switch, so use that. */
363         switch (ecode) {
364         case TDB_SUCCESS: return "Success";
365         case TDB_ERR_CORRUPT: return "Corrupt database";
366         case TDB_ERR_IO: return "IO Error";
367         case TDB_ERR_LOCK: return "Locking error";
368         case TDB_ERR_OOM: return "Out of memory";
369         case TDB_ERR_EXISTS: return "Record exists";
370         case TDB_ERR_EINVAL: return "Invalid parameter";
371         case TDB_ERR_NOEXIST: return "Record does not exist";
372         case TDB_ERR_RDONLY: return "write not permitted";
373         }
374         return "Invalid error code";
375 }
376
377 enum TDB_ERROR COLD tdb_logerr(struct tdb_context *tdb,
378                                enum TDB_ERROR ecode,
379                                enum tdb_log_level level,
380                                const char *fmt, ...)
381 {
382         char *message;
383         va_list ap;
384         size_t len;
385         /* tdb_open paths care about errno, so save it. */
386         int saved_errno = errno;
387
388         if (!tdb->logfn)
389                 return ecode;
390
391         va_start(ap, fmt);
392         len = vasprintf(&message, fmt, ap);
393         va_end(ap);
394
395         if (len < 0) {
396                 tdb->logfn(tdb, TDB_LOG_ERROR, tdb->log_private,
397                            "out of memory formatting message:");
398                 tdb->logfn(tdb, level, tdb->log_private, fmt);
399         } else {
400                 tdb->logfn(tdb, level, tdb->log_private, message);
401                 free(message);
402         }
403         errno = saved_errno;
404         return ecode;
405 }
406
407 enum TDB_ERROR tdb_parse_record_(struct tdb_context *tdb,
408                                  TDB_DATA key,
409                                  enum TDB_ERROR (*parse)(TDB_DATA key,
410                                                          TDB_DATA data,
411                                                          void *p),
412                                  void *p)
413 {
414         tdb_off_t off;
415         struct tdb_used_record rec;
416         struct hash_info h;
417         TDB_DATA data;
418         enum TDB_ERROR ecode;
419
420         off = find_and_lock(tdb, key, F_RDLCK, &h, &rec, NULL);
421         if (TDB_OFF_IS_ERR(off)) {
422                 return off;
423         }
424
425         if (!off) {
426                 ecode = TDB_ERR_NOEXIST;
427         } else {
428                 data.dsize = rec_data_length(&rec);
429                 data.dptr = (void *)tdb_access_read(tdb,
430                                                     off + sizeof(rec)
431                                                     + key.dsize,
432                                                     data.dsize, false);
433                 if (TDB_PTR_IS_ERR(data.dptr)) {
434                         ecode = TDB_PTR_ERR(data.dptr);
435                 } else {
436                         ecode = parse(key, data, p);
437                         tdb_access_release(tdb, data.dptr);
438                 }
439         }
440
441         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_RDLCK);
442         return ecode;
443 }
444
445 const char *tdb_name(const struct tdb_context *tdb)
446 {
447         return tdb->name;
448 }
449
450 int64_t tdb_get_seqnum(struct tdb_context *tdb)
451 {
452         return tdb_read_off(tdb, offsetof(struct tdb_header, seqnum));
453 }
454         
455
456 int tdb_fd(const struct tdb_context *tdb)
457 {
458         return tdb->file->fd;
459 }