tdb2: allow transaction to nest.
[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                 tdb->stats.frees++;
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 tdb->last_error = 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->last_error = 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 tdb->last_error = 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 tdb->last_error = 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 tdb->last_error = 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 tdb->last_error = 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 tdb->last_error = 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                 tdb->last_error = off;
257                 return false;
258         }
259         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_RDLCK);
260
261         tdb->last_error = TDB_SUCCESS;
262         return off ? true : false;
263 }
264
265 enum TDB_ERROR tdb_delete(struct tdb_context *tdb, struct tdb_data key)
266 {
267         tdb_off_t off;
268         struct tdb_used_record rec;
269         struct hash_info h;
270         enum TDB_ERROR ecode;
271
272         off = find_and_lock(tdb, key, F_WRLCK, &h, &rec, NULL);
273         if (TDB_OFF_IS_ERR(off)) {
274                 return tdb->last_error = off;
275         }
276
277         if (!off) {
278                 ecode = TDB_ERR_NOEXIST;
279                 goto unlock;
280         }
281
282         ecode = delete_from_hash(tdb, &h);
283         if (ecode != TDB_SUCCESS) {
284                 goto unlock;
285         }
286
287         /* Free the deleted entry. */
288         tdb->stats.frees++;
289         ecode = add_free_record(tdb, off,
290                                 sizeof(struct tdb_used_record)
291                                 + rec_key_length(&rec)
292                                 + rec_data_length(&rec)
293                                 + rec_extra_padding(&rec));
294
295         if (tdb->flags & TDB_SEQNUM)
296                 tdb_inc_seqnum(tdb);
297
298 unlock:
299         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_WRLCK);
300         return tdb->last_error = ecode;
301 }
302
303 unsigned int tdb_get_flags(struct tdb_context *tdb)
304 {
305         return tdb->flags;
306 }
307
308 void tdb_add_flag(struct tdb_context *tdb, unsigned flag)
309 {
310         if (tdb->flags & TDB_INTERNAL) {
311                 tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL,
312                                              TDB_LOG_USE_ERROR,
313                                              "tdb_add_flag: internal db");
314                 return;
315         }
316         switch (flag) {
317         case TDB_NOLOCK:
318                 tdb->flags |= TDB_NOLOCK;
319                 break;
320         case TDB_NOMMAP:
321                 tdb->flags |= TDB_NOMMAP;
322                 tdb_munmap(tdb->file);
323                 break;
324         case TDB_NOSYNC:
325                 tdb->flags |= TDB_NOSYNC;
326                 break;
327         case TDB_SEQNUM:
328                 tdb->flags |= TDB_SEQNUM;
329                 break;
330         case TDB_ALLOW_NESTING:
331                 tdb->flags |= TDB_ALLOW_NESTING;
332                 break;
333         default:
334                 tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL,
335                                              TDB_LOG_USE_ERROR,
336                                              "tdb_add_flag: Unknown flag %u",
337                                              flag);
338         }
339 }
340
341 void tdb_remove_flag(struct tdb_context *tdb, unsigned flag)
342 {
343         if (tdb->flags & TDB_INTERNAL) {
344                 tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL,
345                                              TDB_LOG_USE_ERROR,
346                                              "tdb_remove_flag: internal db");
347                 return;
348         }
349         switch (flag) {
350         case TDB_NOLOCK:
351                 tdb->flags &= ~TDB_NOLOCK;
352                 break;
353         case TDB_NOMMAP:
354                 tdb->flags &= ~TDB_NOMMAP;
355                 tdb_mmap(tdb);
356                 break;
357         case TDB_NOSYNC:
358                 tdb->flags &= ~TDB_NOSYNC;
359                 break;
360         case TDB_SEQNUM:
361                 tdb->flags &= ~TDB_SEQNUM;
362                 break;
363         case TDB_ALLOW_NESTING:
364                 tdb->flags &= ~TDB_ALLOW_NESTING;
365                 break;
366         default:
367                 tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL,
368                                              TDB_LOG_USE_ERROR,
369                                              "tdb_remove_flag: Unknown flag %u",
370                                              flag);
371         }
372 }
373
374 const char *tdb_errorstr(enum TDB_ERROR ecode)
375 {
376         /* Gcc warns if you miss a case in the switch, so use that. */
377         switch (ecode) {
378         case TDB_SUCCESS: return "Success";
379         case TDB_ERR_CORRUPT: return "Corrupt database";
380         case TDB_ERR_IO: return "IO Error";
381         case TDB_ERR_LOCK: return "Locking error";
382         case TDB_ERR_OOM: return "Out of memory";
383         case TDB_ERR_EXISTS: return "Record exists";
384         case TDB_ERR_EINVAL: return "Invalid parameter";
385         case TDB_ERR_NOEXIST: return "Record does not exist";
386         case TDB_ERR_RDONLY: return "write not permitted";
387         }
388         return "Invalid error code";
389 }
390
391 enum TDB_ERROR tdb_error(struct tdb_context *tdb)
392 {
393         return tdb->last_error;
394 }
395
396 enum TDB_ERROR COLD tdb_logerr(struct tdb_context *tdb,
397                                enum TDB_ERROR ecode,
398                                enum tdb_log_level level,
399                                const char *fmt, ...)
400 {
401         char *message;
402         va_list ap;
403         size_t len;
404         /* tdb_open paths care about errno, so save it. */
405         int saved_errno = errno;
406
407         if (!tdb->log_fn)
408                 return ecode;
409
410         va_start(ap, fmt);
411         len = vasprintf(&message, fmt, ap);
412         va_end(ap);
413
414         if (len < 0) {
415                 tdb->log_fn(tdb, TDB_LOG_ERROR,
416                             "out of memory formatting message:", tdb->log_data);
417                 tdb->log_fn(tdb, level, fmt, tdb->log_data);
418         } else {
419                 tdb->log_fn(tdb, level, message, tdb->log_data);
420                 free(message);
421         }
422         errno = saved_errno;
423         return ecode;
424 }
425
426 enum TDB_ERROR tdb_parse_record_(struct tdb_context *tdb,
427                                  TDB_DATA key,
428                                  enum TDB_ERROR (*parse)(TDB_DATA k,
429                                                          TDB_DATA d,
430                                                          void *data),
431                                  void *data)
432 {
433         tdb_off_t off;
434         struct tdb_used_record rec;
435         struct hash_info h;
436         enum TDB_ERROR ecode;
437
438         off = find_and_lock(tdb, key, F_RDLCK, &h, &rec, NULL);
439         if (TDB_OFF_IS_ERR(off)) {
440                 return tdb->last_error = off;
441         }
442
443         if (!off) {
444                 ecode = TDB_ERR_NOEXIST;
445         } else {
446                 const void *dptr;
447                 dptr = tdb_access_read(tdb, off + sizeof(rec) + key.dsize,
448                                        rec_data_length(&rec), false);
449                 if (TDB_PTR_IS_ERR(dptr)) {
450                         ecode = TDB_PTR_ERR(dptr);
451                 } else {
452                         TDB_DATA d = tdb_mkdata(dptr, rec_data_length(&rec));
453
454                         ecode = parse(key, d, data);
455                         tdb_access_release(tdb, dptr);
456                 }
457         }
458
459         tdb_unlock_hashes(tdb, h.hlock_start, h.hlock_range, F_RDLCK);
460         return tdb->last_error = ecode;
461 }
462
463 const char *tdb_name(const struct tdb_context *tdb)
464 {
465         return tdb->name;
466 }
467
468 int64_t tdb_get_seqnum(struct tdb_context *tdb)
469 {
470         tdb_off_t off = tdb_read_off(tdb, offsetof(struct tdb_header, seqnum));
471         if (TDB_OFF_IS_ERR(off))
472                 tdb->last_error = off;
473         else
474                 tdb->last_error = TDB_SUCCESS;
475         return off;
476 }
477         
478
479 int tdb_fd(const struct tdb_context *tdb)
480 {
481         return tdb->file->fd;
482 }