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