HAMMER 56A/Many: Performance tuning - MEDIA STRUCTURES CHANGED!
[dragonfly.git] / sys / vfs / hammer / hammer_object.c
1 /*
2  * Copyright (c) 2007-2008 The DragonFly Project.  All rights reserved.
3  * 
4  * This code is derived from software contributed to The DragonFly Project
5  * by Matthew Dillon <dillon@backplane.com>
6  * 
7  * Redistribution and use in source and binary forms, with or without
8  * modification, are permitted provided that the following conditions
9  * are met:
10  * 
11  * 1. Redistributions of source code must retain the above copyright
12  *    notice, this list of conditions and the following disclaimer.
13  * 2. Redistributions in binary form must reproduce the above copyright
14  *    notice, this list of conditions and the following disclaimer in
15  *    the documentation and/or other materials provided with the
16  *    distribution.
17  * 3. Neither the name of The DragonFly Project nor the names of its
18  *    contributors may be used to endorse or promote products derived
19  *    from this software without specific, prior written permission.
20  * 
21  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
22  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
23  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
24  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
25  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
26  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
27  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
28  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
29  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
31  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
32  * SUCH DAMAGE.
33  * 
34  * $DragonFly: src/sys/vfs/hammer/hammer_object.c,v 1.69 2008/06/17 04:02:38 dillon Exp $
35  */
36
37 #include "hammer.h"
38
39 static int hammer_mem_add(hammer_record_t record);
40 static int hammer_mem_lookup(hammer_cursor_t cursor);
41 static int hammer_mem_first(hammer_cursor_t cursor);
42 static int hammer_rec_trunc_callback(hammer_record_t record,
43                                 void *data __unused);
44 static int hammer_record_needs_overwrite_delete(hammer_record_t record);
45
46 struct rec_trunc_info {
47         u_int16_t       rec_type;
48         int64_t         trunc_off;
49 };
50
51 /*
52  * Red-black tree support.  Comparison code for insertion.
53  */
54 static int
55 hammer_rec_rb_compare(hammer_record_t rec1, hammer_record_t rec2)
56 {
57         if (rec1->leaf.base.rec_type < rec2->leaf.base.rec_type)
58                 return(-1);
59         if (rec1->leaf.base.rec_type > rec2->leaf.base.rec_type)
60                 return(1);
61
62         if (rec1->leaf.base.key < rec2->leaf.base.key)
63                 return(-1);
64         if (rec1->leaf.base.key > rec2->leaf.base.key)
65                 return(1);
66
67         /*
68          * Never match against an item deleted by the front-end.
69          *
70          * rec1 is greater then rec2 if rec1 is marked deleted.
71          * rec1 is less then rec2 if rec2 is marked deleted.
72          *
73          * Multiple deleted records may be present, do not return 0
74          * if both are marked deleted.
75          */
76         if (rec1->flags & HAMMER_RECF_DELETED_FE)
77                 return(1);
78         if (rec2->flags & HAMMER_RECF_DELETED_FE)
79                 return(-1);
80
81         return(0);
82 }
83
84 /*
85  * Basic record comparison code similar to hammer_btree_cmp().
86  */
87 static int
88 hammer_rec_cmp(hammer_base_elm_t elm, hammer_record_t rec)
89 {
90         if (elm->rec_type < rec->leaf.base.rec_type)
91                 return(-3);
92         if (elm->rec_type > rec->leaf.base.rec_type)
93                 return(3);
94
95         if (elm->key < rec->leaf.base.key)
96                 return(-2);
97         if (elm->key > rec->leaf.base.key)
98                 return(2);
99
100         /*
101          * Never match against an item deleted by the front-end.
102          * elm is less then rec if rec is marked deleted.
103          */
104         if (rec->flags & HAMMER_RECF_DELETED_FE)
105                 return(-1);
106         return(0);
107 }
108
109 /*
110  * Special LOOKUP_INFO to locate an overlapping record.  This used by
111  * the reservation code to implement small-block records (whos keys will
112  * be different depending on data_len, when representing the same base
113  * offset).
114  *
115  * NOTE: The base file offset of a data record is (key - data_len), not (key).
116  */
117 static int
118 hammer_rec_overlap_compare(hammer_btree_leaf_elm_t leaf, hammer_record_t rec)
119 {
120         if (leaf->base.rec_type < rec->leaf.base.rec_type)
121                 return(-3);
122         if (leaf->base.rec_type > rec->leaf.base.rec_type)
123                 return(3);
124
125         if (leaf->base.rec_type == HAMMER_RECTYPE_DATA) {
126                 /* leaf_end <= rec_beg */
127                 if (leaf->base.key <= rec->leaf.base.key - rec->leaf.data_len)
128                         return(-2);
129                 /* leaf_beg >= rec_end */
130                 if (leaf->base.key - leaf->data_len >= rec->leaf.base.key)
131                         return(2);
132         } else {
133                 if (leaf->base.key < rec->leaf.base.key)
134                         return(-2);
135                 if (leaf->base.key > rec->leaf.base.key)
136                         return(2);
137         }
138
139         /*
140          * Never match against an item deleted by the front-end.
141          * leaf is less then rec if rec is marked deleted.
142          */
143         if (rec->flags & HAMMER_RECF_DELETED_FE)
144                 return(-1);
145         return(0);
146 }
147
148 /*
149  * RB_SCAN comparison code for hammer_mem_first().  The argument order
150  * is reversed so the comparison result has to be negated.  key_beg and
151  * key_end are both range-inclusive.
152  *
153  * Localized deletions are not cached in-memory.
154  */
155 static
156 int
157 hammer_rec_scan_cmp(hammer_record_t rec, void *data)
158 {
159         hammer_cursor_t cursor = data;
160         int r;
161
162         r = hammer_rec_cmp(&cursor->key_beg, rec);
163         if (r > 1)
164                 return(-1);
165         r = hammer_rec_cmp(&cursor->key_end, rec);
166         if (r < -1)
167                 return(1);
168         return(0);
169 }
170
171 /*
172  * This compare function is used when simply looking up key_beg.
173  */
174 static
175 int
176 hammer_rec_find_cmp(hammer_record_t rec, void *data)
177 {
178         hammer_cursor_t cursor = data;
179         int r;
180
181         r = hammer_rec_cmp(&cursor->key_beg, rec);
182         if (r > 1)
183                 return(-1);
184         if (r < -1)
185                 return(1);
186         return(0);
187 }
188
189 /*
190  * Locate blocks within the truncation range.  Partial blocks do not count.
191  */
192 static
193 int
194 hammer_rec_trunc_cmp(hammer_record_t rec, void *data)
195 {
196         struct rec_trunc_info *info = data;
197
198         if (rec->leaf.base.rec_type < info->rec_type)
199                 return(-1);
200         if (rec->leaf.base.rec_type > info->rec_type)
201                 return(1);
202
203         switch(rec->leaf.base.rec_type) {
204         case HAMMER_RECTYPE_DB:
205                 /*
206                  * DB record key is not beyond the truncation point, retain.
207                  */
208                 if (rec->leaf.base.key < info->trunc_off)
209                         return(-1);
210                 break;
211         case HAMMER_RECTYPE_DATA:
212                 /*
213                  * DATA record offset start is not beyond the truncation point,
214                  * retain.
215                  */
216                 if (rec->leaf.base.key - rec->leaf.data_len < info->trunc_off)
217                         return(-1);
218                 break;
219         default:
220                 panic("hammer_rec_trunc_cmp: unexpected record type");
221         }
222
223         /*
224          * The record start is >= the truncation point, return match,
225          * the record should be destroyed.
226          */
227         return(0);
228 }
229
230 RB_GENERATE(hammer_rec_rb_tree, hammer_record, rb_node, hammer_rec_rb_compare);
231 RB_GENERATE_XLOOKUP(hammer_rec_rb_tree, INFO, hammer_record, rb_node,
232                     hammer_rec_overlap_compare, hammer_btree_leaf_elm_t);
233
234 /*
235  * Allocate a record for the caller to finish filling in.  The record is
236  * returned referenced.
237  */
238 hammer_record_t
239 hammer_alloc_mem_record(hammer_inode_t ip, int data_len)
240 {
241         hammer_record_t record;
242
243         ++hammer_count_records;
244         record = kmalloc(sizeof(*record), M_HAMMER, M_WAITOK | M_ZERO);
245         record->flush_state = HAMMER_FST_IDLE;
246         record->ip = ip;
247         record->leaf.base.btype = HAMMER_BTREE_TYPE_RECORD;
248         record->leaf.data_len = data_len;
249         hammer_ref(&record->lock);
250
251         if (data_len) {
252                 record->data = kmalloc(data_len, M_HAMMER, M_WAITOK | M_ZERO);
253                 record->flags |= HAMMER_RECF_ALLOCDATA;
254                 ++hammer_count_record_datas;
255         }
256
257         return (record);
258 }
259
260 void
261 hammer_wait_mem_record_ident(hammer_record_t record, const char *ident)
262 {
263         while (record->flush_state == HAMMER_FST_FLUSH) {
264                 record->flags |= HAMMER_RECF_WANTED;
265                 tsleep(record, 0, ident, 0);
266         }
267 }
268
269 /*
270  * Called from the backend, hammer_inode.c, after a record has been
271  * flushed to disk.  The record has been exclusively locked by the
272  * caller and interlocked with BE.
273  *
274  * We clean up the state, unlock, and release the record (the record
275  * was referenced by the fact that it was in the HAMMER_FST_FLUSH state).
276  */
277 void
278 hammer_flush_record_done(hammer_record_t record, int error)
279 {
280         hammer_inode_t target_ip;
281
282         KKASSERT(record->flush_state == HAMMER_FST_FLUSH);
283         KKASSERT(record->flags & HAMMER_RECF_INTERLOCK_BE);
284
285         if (error) {
286                 /*
287                  * An error occured, the backend was unable to sync the
288                  * record to its media.  Leave the record intact.
289                  */
290                 Debugger("flush_record_done error");
291         }
292
293         if (record->flags & HAMMER_RECF_DELETED_BE) {
294                 if ((target_ip = record->target_ip) != NULL) {
295                         TAILQ_REMOVE(&target_ip->target_list, record,
296                                      target_entry);
297                         record->target_ip = NULL;
298                         hammer_test_inode(target_ip);
299                 }
300                 record->flush_state = HAMMER_FST_IDLE;
301         } else {
302                 if (record->target_ip) {
303                         record->flush_state = HAMMER_FST_SETUP;
304                         hammer_test_inode(record->ip);
305                         hammer_test_inode(record->target_ip);
306                 } else {
307                         record->flush_state = HAMMER_FST_IDLE;
308                 }
309         }
310         record->flags &= ~HAMMER_RECF_INTERLOCK_BE;
311         if (record->flags & HAMMER_RECF_WANTED) {
312                 record->flags &= ~HAMMER_RECF_WANTED;
313                 wakeup(record);
314         }
315         hammer_rel_mem_record(record);
316 }
317
318 /*
319  * Release a memory record.  Records marked for deletion are immediately
320  * removed from the RB-Tree but otherwise left intact until the last ref
321  * goes away.
322  */
323 void
324 hammer_rel_mem_record(struct hammer_record *record)
325 {
326         hammer_inode_t ip, target_ip;
327
328         hammer_unref(&record->lock);
329
330         if (record->lock.refs == 0) {
331                 /*
332                  * Upon release of the last reference wakeup any waiters.
333                  * The record structure may get destroyed so callers will
334                  * loop up and do a relookup.
335                  *
336                  * WARNING!  Record must be removed from RB-TREE before we
337                  * might possibly block.  hammer_test_inode() can block!
338                  */
339                 ip = record->ip;
340
341                 /*
342                  * Upon release of the last reference a record marked deleted
343                  * is destroyed.
344                  */
345                 if (record->flags & HAMMER_RECF_DELETED_FE) {
346                         KKASSERT(ip->lock.refs > 0);
347                         KKASSERT(record->flush_state != HAMMER_FST_FLUSH);
348
349                         /*
350                          * target_ip may have zero refs, we have to ref it
351                          * to prevent it from being ripped out from under
352                          * us.
353                          */
354                         if ((target_ip = record->target_ip) != NULL) {
355                                 TAILQ_REMOVE(&target_ip->target_list,
356                                              record, target_entry);
357                                 record->target_ip = NULL;
358                                 hammer_ref(&target_ip->lock);
359                         }
360
361                         if (record->flags & HAMMER_RECF_ONRBTREE) {
362                                 RB_REMOVE(hammer_rec_rb_tree,
363                                           &record->ip->rec_tree,
364                                           record);
365                                 KKASSERT(ip->rsv_recs > 0);
366                                 --ip->hmp->rsv_recs;
367                                 --ip->rsv_recs;
368                                 ip->hmp->rsv_databytes -= record->leaf.data_len;
369                                 record->flags &= ~HAMMER_RECF_ONRBTREE;
370
371                                 if ((ip->flags & HAMMER_INODE_PARTIALW) &&
372                                     ip->rsv_recs <= hammer_limit_irecs) {
373                                         ip->flags &= ~HAMMER_INODE_PARTIALW;
374                                         wakeup(&ip->flags);
375                                 }
376                                 if (RB_EMPTY(&record->ip->rec_tree)) {
377                                         record->ip->flags &= ~HAMMER_INODE_XDIRTY;
378                                         record->ip->sync_flags &= ~HAMMER_INODE_XDIRTY;
379                                         hammer_test_inode(record->ip);
380                                 }
381                         }
382
383                         /*
384                          * Do this test after removing record from the B-Tree.
385                          */
386                         if (target_ip) {
387                                 hammer_test_inode(target_ip);
388                                 hammer_rel_inode(target_ip, 0);
389                         }
390
391                         if (record->flags & HAMMER_RECF_ALLOCDATA) {
392                                 --hammer_count_record_datas;
393                                 kfree(record->data, M_HAMMER);
394                                 record->flags &= ~HAMMER_RECF_ALLOCDATA;
395                         }
396                         if (record->resv) {
397                                 hammer_blockmap_reserve_complete(ip->hmp,
398                                                                  record->resv);
399                                 record->resv = NULL;
400                         }
401                         record->data = NULL;
402                         --hammer_count_records;
403                         kfree(record, M_HAMMER);
404                 }
405         }
406 }
407
408 /*
409  * Record visibility depends on whether the record is being accessed by
410  * the backend or the frontend.
411  *
412  * Return non-zero if the record is visible, zero if it isn't or if it is
413  * deleted.
414  */
415 static __inline
416 int
417 hammer_ip_iterate_mem_good(hammer_cursor_t cursor, hammer_record_t record)
418 {
419         if (cursor->flags & HAMMER_CURSOR_BACKEND) {
420                 if (record->flags & HAMMER_RECF_DELETED_BE)
421                         return(0);
422         } else {
423                 if (record->flags & HAMMER_RECF_DELETED_FE)
424                         return(0);
425         }
426         return(1);
427 }
428
429 /*
430  * This callback is used as part of the RB_SCAN function for in-memory
431  * records.  We terminate it (return -1) as soon as we get a match.
432  *
433  * This routine is used by frontend code.
434  *
435  * The primary compare code does not account for ASOF lookups.  This
436  * code handles that case as well as a few others.
437  */
438 static
439 int
440 hammer_rec_scan_callback(hammer_record_t rec, void *data)
441 {
442         hammer_cursor_t cursor = data;
443
444         /*
445          * We terminate on success, so this should be NULL on entry.
446          */
447         KKASSERT(cursor->iprec == NULL);
448
449         /*
450          * Skip if the record was marked deleted.
451          */
452         if (hammer_ip_iterate_mem_good(cursor, rec) == 0)
453                 return(0);
454
455         /*
456          * Skip if not visible due to our as-of TID
457          */
458         if (cursor->flags & HAMMER_CURSOR_ASOF) {
459                 if (cursor->asof < rec->leaf.base.create_tid)
460                         return(0);
461                 if (rec->leaf.base.delete_tid &&
462                     cursor->asof >= rec->leaf.base.delete_tid) {
463                         return(0);
464                 }
465         }
466
467         /*
468          * If the record is queued to the flusher we have to block until
469          * it isn't.  Otherwise we may see duplication between our memory
470          * cache and the media.
471          */
472         hammer_ref(&rec->lock);
473
474 #warning "This deadlocks"
475 #if 0
476         if (rec->flush_state == HAMMER_FST_FLUSH)
477                 hammer_wait_mem_record(rec);
478 #endif
479
480         /*
481          * The record may have been deleted while we were blocked.
482          */
483         if (hammer_ip_iterate_mem_good(cursor, rec) == 0) {
484                 hammer_rel_mem_record(rec);
485                 return(0);
486         }
487
488         /*
489          * Set the matching record and stop the scan.
490          */
491         cursor->iprec = rec;
492         return(-1);
493 }
494
495
496 /*
497  * Lookup an in-memory record given the key specified in the cursor.  Works
498  * just like hammer_btree_lookup() but operates on an inode's in-memory
499  * record list.
500  *
501  * The lookup must fail if the record is marked for deferred deletion.
502  */
503 static
504 int
505 hammer_mem_lookup(hammer_cursor_t cursor)
506 {
507         int error;
508
509         KKASSERT(cursor->ip);
510         if (cursor->iprec) {
511                 hammer_rel_mem_record(cursor->iprec);
512                 cursor->iprec = NULL;
513         }
514         hammer_rec_rb_tree_RB_SCAN(&cursor->ip->rec_tree, hammer_rec_find_cmp,
515                                    hammer_rec_scan_callback, cursor);
516
517         if (cursor->iprec == NULL)
518                 error = ENOENT;
519         else
520                 error = 0;
521         return(error);
522 }
523
524 /*
525  * hammer_mem_first() - locate the first in-memory record matching the
526  * cursor within the bounds of the key range.
527  */
528 static
529 int
530 hammer_mem_first(hammer_cursor_t cursor)
531 {
532         hammer_inode_t ip;
533
534         ip = cursor->ip;
535         KKASSERT(ip != NULL);
536
537         if (cursor->iprec) {
538                 hammer_rel_mem_record(cursor->iprec);
539                 cursor->iprec = NULL;
540         }
541
542         hammer_rec_rb_tree_RB_SCAN(&ip->rec_tree, hammer_rec_scan_cmp,
543                                    hammer_rec_scan_callback, cursor);
544
545         /*
546          * Adjust scan.node and keep it linked into the RB-tree so we can
547          * hold the cursor through third party modifications of the RB-tree.
548          */
549         if (cursor->iprec)
550                 return(0);
551         return(ENOENT);
552 }
553
554 void
555 hammer_mem_done(hammer_cursor_t cursor)
556 {
557         if (cursor->iprec) {
558                 hammer_rel_mem_record(cursor->iprec);
559                 cursor->iprec = NULL;
560         }
561 }
562
563 /************************************************************************
564  *                   HAMMER IN-MEMORY RECORD FUNCTIONS                  *
565  ************************************************************************
566  *
567  * These functions manipulate in-memory records.  Such records typically
568  * exist prior to being committed to disk or indexed via the on-disk B-Tree.
569  */
570
571 /*
572  * Add a directory entry (dip,ncp) which references inode (ip).
573  *
574  * Note that the low 32 bits of the namekey are set temporarily to create
575  * a unique in-memory record, and may be modified a second time when the
576  * record is synchronized to disk.  In particular, the low 32 bits cannot be
577  * all 0's when synching to disk, which is not handled here.
578  */
579 int
580 hammer_ip_add_directory(struct hammer_transaction *trans,
581                      struct hammer_inode *dip, struct namecache *ncp,
582                      struct hammer_inode *ip)
583 {
584         hammer_record_t record;
585         int error;
586         int bytes;
587
588         bytes = ncp->nc_nlen;   /* NOTE: terminating \0 is NOT included */
589         record = hammer_alloc_mem_record(dip, HAMMER_ENTRY_SIZE(bytes));
590         if (++trans->hmp->namekey_iterator == 0)
591                 ++trans->hmp->namekey_iterator;
592
593         record->type = HAMMER_MEM_RECORD_ADD;
594         record->leaf.base.localization = HAMMER_LOCALIZE_MISC;
595         record->leaf.base.obj_id = dip->obj_id;
596         record->leaf.base.key = hammer_directory_namekey(ncp->nc_name, bytes);
597         record->leaf.base.key += trans->hmp->namekey_iterator;
598         record->leaf.base.rec_type = HAMMER_RECTYPE_DIRENTRY;
599         record->leaf.base.obj_type = ip->ino_leaf.base.obj_type;
600         record->data->entry.obj_id = ip->obj_id;
601         bcopy(ncp->nc_name, record->data->entry.name, bytes);
602
603         ++ip->ino_data.nlinks;
604         hammer_modify_inode(ip, HAMMER_INODE_DDIRTY);
605
606         /*
607          * The target inode and the directory entry are bound together.
608          */
609         record->target_ip = ip;
610         record->flush_state = HAMMER_FST_SETUP;
611         TAILQ_INSERT_TAIL(&ip->target_list, record, target_entry);
612
613         /*
614          * The inode now has a dependancy and must be taken out of the idle
615          * state.  An inode not in an idle state is given an extra reference.
616          */
617         if (ip->flush_state == HAMMER_FST_IDLE) {
618                 hammer_ref(&ip->lock);
619                 ip->flush_state = HAMMER_FST_SETUP;
620         }
621         error = hammer_mem_add(record);
622         return(error);
623 }
624
625 /*
626  * Delete the directory entry and update the inode link count.  The
627  * cursor must be seeked to the directory entry record being deleted.
628  *
629  * The related inode should be share-locked by the caller.  The caller is
630  * on the frontend.
631  *
632  * This function can return EDEADLK requiring the caller to terminate
633  * the cursor, any locks, wait on the returned record, and retry.
634  */
635 int
636 hammer_ip_del_directory(struct hammer_transaction *trans,
637                      hammer_cursor_t cursor, struct hammer_inode *dip,
638                      struct hammer_inode *ip)
639 {
640         hammer_record_t record;
641         int error;
642
643         if (hammer_cursor_inmem(cursor)) {
644                 /*
645                  * In-memory (unsynchronized) records can simply be freed.
646                  * Even though the HAMMER_RECF_DELETED_FE flag is ignored
647                  * by the backend, we must still avoid races against the
648                  * backend potentially syncing the record to the media. 
649                  *
650                  * We cannot call hammer_ip_delete_record(), that routine may
651                  * only be called from the backend.
652                  */
653                 record = cursor->iprec;
654                 if (record->flags & HAMMER_RECF_INTERLOCK_BE) {
655                         KKASSERT(cursor->deadlk_rec == NULL);
656                         hammer_ref(&record->lock);
657                         cursor->deadlk_rec = record;
658                         error = EDEADLK;
659                 } else {
660                         KKASSERT(record->type == HAMMER_MEM_RECORD_ADD);
661                         record->flags |= HAMMER_RECF_DELETED_FE;
662                         error = 0;
663                 }
664         } else {
665                 /*
666                  * If the record is on-disk we have to queue the deletion by
667                  * the record's key.  This also causes lookups to skip the
668                  * record.
669                  */
670                 KKASSERT(dip->flags &
671                          (HAMMER_INODE_ONDISK | HAMMER_INODE_DONDISK));
672                 record = hammer_alloc_mem_record(dip, 0);
673                 record->type = HAMMER_MEM_RECORD_DEL;
674                 record->leaf.base = cursor->leaf->base;
675
676                 record->target_ip = ip;
677                 record->flush_state = HAMMER_FST_SETUP;
678                 TAILQ_INSERT_TAIL(&ip->target_list, record, target_entry);
679
680                 /*
681                  * The inode now has a dependancy and must be taken out of
682                  * the idle state.  An inode not in an idle state is given
683                  * an extra reference.
684                  */
685                 if (ip->flush_state == HAMMER_FST_IDLE) {
686                         hammer_ref(&ip->lock);
687                         ip->flush_state = HAMMER_FST_SETUP;
688                 }
689
690                 error = hammer_mem_add(record);
691         }
692
693         /*
694          * One less link.  The file may still be open in the OS even after
695          * all links have gone away.
696          *
697          * We have to terminate the cursor before syncing the inode to
698          * avoid deadlocking against ourselves.  XXX this may no longer
699          * be true.
700          *
701          * If nlinks drops to zero and the vnode is inactive (or there is
702          * no vnode), call hammer_inode_unloadable_check() to zonk the
703          * inode.  If we don't do this here the inode will not be destroyed
704          * on-media until we unmount.
705          */
706         if (error == 0) {
707                 --ip->ino_data.nlinks;
708                 hammer_modify_inode(ip, HAMMER_INODE_DDIRTY);
709                 if (ip->ino_data.nlinks == 0 &&
710                     (ip->vp == NULL || (ip->vp->v_flag & VINACTIVE))) {
711                         hammer_done_cursor(cursor);
712                         hammer_inode_unloadable_check(ip, 1);
713                         hammer_flush_inode(ip, 0);
714                 }
715
716         }
717         return(error);
718 }
719
720 /*
721  * Add a record to an inode.
722  *
723  * The caller must allocate the record with hammer_alloc_mem_record(ip) and
724  * initialize the following additional fields:
725  *
726  * The related inode should be share-locked by the caller.  The caller is
727  * on the frontend.
728  *
729  * record->rec.entry.base.base.key
730  * record->rec.entry.base.base.rec_type
731  * record->rec.entry.base.base.data_len
732  * record->data         (a copy will be kmalloc'd if it cannot be embedded)
733  */
734 int
735 hammer_ip_add_record(struct hammer_transaction *trans, hammer_record_t record)
736 {
737         hammer_inode_t ip = record->ip;
738         int error;
739
740         KKASSERT(record->leaf.base.localization != 0);
741         record->leaf.base.obj_id = ip->obj_id;
742         record->leaf.base.obj_type = ip->ino_leaf.base.obj_type;
743         error = hammer_mem_add(record);
744         return(error);
745 }
746
747 /*
748  * Locate a bulk record in-memory.  Bulk records allow disk space to be
749  * reserved so the front-end can flush large data writes without having
750  * to queue the BIO to the flusher.  Only the related record gets queued
751  * to the flusher.
752  */
753 static hammer_record_t
754 hammer_ip_get_bulk(hammer_inode_t ip, off_t file_offset, int bytes)
755 {
756         hammer_record_t record;
757         struct hammer_btree_leaf_elm leaf;
758
759         bzero(&leaf, sizeof(leaf));
760         leaf.base.obj_id = ip->obj_id;
761         leaf.base.key = file_offset + bytes;
762         leaf.base.create_tid = 0;
763         leaf.base.delete_tid = 0;
764         leaf.base.rec_type = HAMMER_RECTYPE_DATA;
765         leaf.base.obj_type = 0;                 /* unused */
766         leaf.base.btype = HAMMER_BTREE_TYPE_RECORD;     /* unused */
767         leaf.base.localization = HAMMER_LOCALIZE_MISC;
768         leaf.data_len = bytes;
769
770         record = hammer_rec_rb_tree_RB_LOOKUP_INFO(&ip->rec_tree, &leaf);
771         if (record)
772                 hammer_ref(&record->lock);
773         return(record);
774 }
775
776 /*
777  * Reserve blockmap space placemarked with an in-memory record.  
778  *
779  * This routine is called by the front-end in order to be able to directly
780  * flush a buffer cache buffer.
781  */
782 hammer_record_t
783 hammer_ip_add_bulk(hammer_inode_t ip, off_t file_offset, void *data, int bytes,
784                    int *errorp)
785 {
786         hammer_record_t record;
787         hammer_record_t conflict;
788         int zone;
789         int flags;
790
791         /*
792          * Deal with conflicting in-memory records.  We cannot have multiple
793          * in-memory records for the same offset without seriously confusing
794          * the backend, including but not limited to the backend issuing
795          * delete-create-delete sequences and asserting on the delete_tid
796          * being the same as the create_tid.
797          *
798          * If we encounter a record with the backend interlock set we cannot
799          * immediately delete it without confusing the backend.
800          */
801         while ((conflict = hammer_ip_get_bulk(ip, file_offset, bytes)) !=NULL) {
802                 if (conflict->flags & HAMMER_RECF_INTERLOCK_BE) {
803                         conflict->flags |= HAMMER_RECF_WANTED;
804                         tsleep(conflict, 0, "hmrrc3", 0);
805                 } else {
806                         conflict->flags |= HAMMER_RECF_DELETED_FE;
807                 }
808                 hammer_rel_mem_record(conflict);
809         }
810
811         /*
812          * Create a record to cover the direct write.  This is called with
813          * the related BIO locked so there should be no possible conflict.
814          *
815          * The backend is responsible for finalizing the space reserved in
816          * this record.
817          *
818          * XXX bytes not aligned, depend on the reservation code to
819          * align the reservation.
820          */
821         record = hammer_alloc_mem_record(ip, 0);
822         zone = (bytes >= HAMMER_BUFSIZE) ? HAMMER_ZONE_LARGE_DATA_INDEX :
823                                            HAMMER_ZONE_SMALL_DATA_INDEX;
824         record->resv = hammer_blockmap_reserve(ip->hmp, zone, bytes,
825                                                &record->leaf.data_offset,
826                                                errorp);
827         if (record->resv == NULL) {
828                 kprintf("hammer_ip_add_bulk: reservation failed\n");
829                 hammer_rel_mem_record(record);
830                 return(NULL);
831         }
832         record->type = HAMMER_MEM_RECORD_DATA;
833         record->leaf.base.rec_type = HAMMER_RECTYPE_DATA;
834         record->leaf.base.obj_type = ip->ino_leaf.base.obj_type;
835         record->leaf.base.obj_id = ip->obj_id;
836         record->leaf.base.key = file_offset + bytes;
837         record->leaf.base.localization = HAMMER_LOCALIZE_MISC;
838         record->leaf.data_len = bytes;
839         record->leaf.data_crc = crc32(data, bytes);
840         flags = record->flags;
841
842         hammer_ref(&record->lock);      /* mem_add eats a reference */
843         *errorp = hammer_mem_add(record);
844         KKASSERT(*errorp == 0);
845
846         return (record);
847 }
848
849 /*
850  * Frontend truncation code.  Scan in-memory records only.  On-disk records
851  * and records in a flushing state are handled by the backend.  The vnops
852  * setattr code will handle the block containing the truncation point.
853  *
854  * Partial blocks are not deleted.
855  */
856 int
857 hammer_ip_frontend_trunc(struct hammer_inode *ip, off_t file_size)
858 {
859         struct rec_trunc_info info;
860
861         switch(ip->ino_data.obj_type) {
862         case HAMMER_OBJTYPE_REGFILE:
863                 info.rec_type = HAMMER_RECTYPE_DATA;
864                 break;
865         case HAMMER_OBJTYPE_DBFILE:
866                 info.rec_type = HAMMER_RECTYPE_DB;
867                 break;
868         default:
869                 return(EINVAL);
870         }
871         info.trunc_off = file_size;
872         hammer_rec_rb_tree_RB_SCAN(&ip->rec_tree, hammer_rec_trunc_cmp,
873                                    hammer_rec_trunc_callback, &info);
874         return(0);
875 }
876
877 static int
878 hammer_rec_trunc_callback(hammer_record_t record, void *data __unused)
879 {
880         if (record->flags & HAMMER_RECF_DELETED_FE)
881                 return(0);
882         if (record->flush_state == HAMMER_FST_FLUSH)
883                 return(0);
884         KKASSERT((record->flags & HAMMER_RECF_INTERLOCK_BE) == 0);
885         hammer_ref(&record->lock);
886         record->flags |= HAMMER_RECF_DELETED_FE;
887         hammer_rel_mem_record(record);
888         return(0);
889 }
890
891 /*
892  * Return 1 if the caller must check for and delete existing records
893  * before writing out a new data record.
894  *
895  * Return 0 if the caller can just insert the record into the B-Tree without
896  * checking.
897  */
898 static int
899 hammer_record_needs_overwrite_delete(hammer_record_t record)
900 {
901         hammer_inode_t ip = record->ip;
902         int64_t file_offset;
903         int r;
904
905         if (ip->ino_data.obj_type == HAMMER_OBJTYPE_DBFILE)
906                 file_offset = record->leaf.base.key;
907         else
908                 file_offset = record->leaf.base.key - record->leaf.data_len;
909         r = (file_offset < ip->sync_trunc_off);
910         if (ip->ino_data.obj_type == HAMMER_OBJTYPE_DBFILE) {
911                 if (ip->sync_trunc_off <= record->leaf.base.key)
912                         ip->sync_trunc_off = record->leaf.base.key + 1;
913         } else {
914                 if (ip->sync_trunc_off < record->leaf.base.key)
915                         ip->sync_trunc_off = record->leaf.base.key;
916         }
917         return(r);
918 }
919
920 /*
921  * Backend code.  Sync a record to the media.
922  */
923 int
924 hammer_ip_sync_record_cursor(hammer_cursor_t cursor, hammer_record_t record)
925 {
926         hammer_transaction_t trans = cursor->trans;
927         int64_t file_offset;
928         void *bdata;
929         int error;
930
931         KKASSERT(record->flush_state == HAMMER_FST_FLUSH);
932         KKASSERT(record->flags & HAMMER_RECF_INTERLOCK_BE);
933         KKASSERT(record->leaf.base.localization != 0);
934
935         /*
936          * If this is a bulk-data record placemarker there may be an existing
937          * record on-disk, indicating a data overwrite.  If there is the
938          * on-disk record must be deleted before we can insert our new record.
939          *
940          * We've synthesized this record and do not know what the create_tid
941          * on-disk is, nor how much data it represents.
942          *
943          * Keep in mind that (key) for data records is (base_offset + len),
944          * not (base_offset).  Also, we only want to get rid of on-disk
945          * records since we are trying to sync our in-memory record, call
946          * hammer_ip_delete_range() with truncating set to 1 to make sure
947          * it skips in-memory records.
948          *
949          * It is ok for the lookup to return ENOENT.
950          *
951          * NOTE OPTIMIZATION: sync_trunc_off is used to determine if we have
952          * to call hammer_ip_delete_range() or not.  This also means we must
953          * update sync_trunc_off() as we write.
954          */
955         if (record->type == HAMMER_MEM_RECORD_DATA &&
956             hammer_record_needs_overwrite_delete(record)) {
957                 file_offset = record->leaf.base.key - record->leaf.data_len;
958                 KKASSERT((file_offset & HAMMER_BUFMASK) == 0);
959                 error = hammer_ip_delete_range(
960                                 cursor, record->ip,
961                                 file_offset, file_offset + HAMMER_BUFSIZE - 1,
962                                 1);
963                 if (error && error != ENOENT)
964                         goto done;
965         }
966
967         /*
968          * Setup the cursor.
969          */
970         hammer_normalize_cursor(cursor);
971         cursor->key_beg = record->leaf.base;
972         cursor->flags &= ~HAMMER_CURSOR_INITMASK;
973         cursor->flags |= HAMMER_CURSOR_BACKEND;
974         cursor->flags &= ~HAMMER_CURSOR_INSERT;
975
976         /*
977          * Records can wind up on-media before the inode itself is on-media.
978          * Flag the case.
979          */
980         record->ip->flags |= HAMMER_INODE_DONDISK;
981
982         /*
983          * If we are deleting a directory entry an exact match must be
984          * found on-disk.
985          */
986         if (record->type == HAMMER_MEM_RECORD_DEL) {
987                 error = hammer_btree_lookup(cursor);
988                 if (error == 0) {
989                         error = hammer_ip_delete_record(cursor, record->ip,
990                                                         trans->tid);
991                         if (error == 0) {
992                                 record->flags |= HAMMER_RECF_DELETED_FE;
993                                 record->flags |= HAMMER_RECF_DELETED_BE;
994                         }
995                 }
996                 goto done;
997         }
998
999         /*
1000          * We are inserting.
1001          *
1002          * Issue a lookup to position the cursor and locate the cluster.  The
1003          * target key should not exist.  If we are creating a directory entry
1004          * we may have to iterate the low 32 bits of the key to find an unused
1005          * key.
1006          */
1007         cursor->flags |= HAMMER_CURSOR_INSERT;
1008
1009         for (;;) {
1010                 error = hammer_btree_lookup(cursor);
1011                 if (hammer_debug_inode)
1012                         kprintf("DOINSERT LOOKUP %d\n", error);
1013                 if (error)
1014                         break;
1015                 if (record->leaf.base.rec_type != HAMMER_RECTYPE_DIRENTRY) {
1016                         kprintf("hammer_ip_sync_record: duplicate rec "
1017                                 "at (%016llx)\n", record->leaf.base.key);
1018                         Debugger("duplicate record1");
1019                         error = EIO;
1020                         break;
1021                 }
1022                 if (++trans->hmp->namekey_iterator == 0)
1023                         ++trans->hmp->namekey_iterator;
1024                 record->leaf.base.key &= ~(0xFFFFFFFFLL);
1025                 record->leaf.base.key |= trans->hmp->namekey_iterator;
1026                 cursor->key_beg.key = record->leaf.base.key;
1027         }
1028 #if 0
1029         if (record->type == HAMMER_MEM_RECORD_DATA)
1030                 kprintf("sync_record  %016llx ---------------- %016llx %d\n",
1031                         record->leaf.base.key - record->leaf.data_len,
1032                         record->leaf.data_offset, error);
1033 #endif
1034                         
1035
1036         if (error != ENOENT)
1037                 goto done;
1038
1039         /*
1040          * Allocate the record and data.  The result buffers will be
1041          * marked as being modified and further calls to
1042          * hammer_modify_buffer() will result in unneeded UNDO records.
1043          *
1044          * Support zero-fill records (data == NULL and data_len != 0)
1045          */
1046         if (record->type == HAMMER_MEM_RECORD_DATA) {
1047                 /*
1048                  * The data portion of a bulk-data record has already been
1049                  * committed to disk, we need only adjust the layer2
1050                  * statistics in the same transaction as our B-Tree insert.
1051                  */
1052                 KKASSERT(record->leaf.data_offset != 0);
1053                 hammer_blockmap_free(trans, record->leaf.data_offset,
1054                                      -record->leaf.data_len);
1055                 error = 0;
1056         } else if (record->data && record->leaf.data_len) {
1057                 /*
1058                  * Wholely cached record, with data.  Allocate the data.
1059                  */
1060                 bdata = hammer_alloc_data(trans, record->leaf.data_len,
1061                                           record->leaf.base.rec_type,
1062                                           &record->leaf.data_offset,
1063                                           &cursor->data_buffer, &error);
1064                 if (bdata == NULL)
1065                         goto done;
1066                 record->leaf.data_crc = crc32(record->data,
1067                                               record->leaf.data_len);
1068                 hammer_modify_buffer(trans, cursor->data_buffer, NULL, 0);
1069                 bcopy(record->data, bdata, record->leaf.data_len);
1070                 hammer_modify_buffer_done(cursor->data_buffer);
1071         } else {
1072                 /*
1073                  * Wholely cached record, without data.
1074                  */
1075                 record->leaf.data_offset = 0;
1076                 record->leaf.data_crc = 0;
1077         }
1078
1079         error = hammer_btree_insert(cursor, &record->leaf);
1080         if (hammer_debug_inode && error)
1081                 kprintf("BTREE INSERT error %d @ %016llx:%d key %016llx\n", error, cursor->node->node_offset, cursor->index, record->leaf.base.key);
1082
1083         /*
1084          * Our record is on-disk, normally mark the in-memory version as
1085          * deleted.  If the record represented a directory deletion but
1086          * we had to sync a valid directory entry to disk we must convert
1087          * the record to a covering delete so the frontend does not have
1088          * visibility on the synced entry.
1089          */
1090         if (error == 0) {
1091                 if (record->flags & HAMMER_RECF_CONVERT_DELETE) {
1092                         KKASSERT(record->type == HAMMER_MEM_RECORD_ADD);
1093                         record->flags &= ~HAMMER_RECF_DELETED_FE;
1094                         record->type = HAMMER_MEM_RECORD_DEL;
1095                         KKASSERT(record->flush_state == HAMMER_FST_FLUSH);
1096                         record->flags &= ~HAMMER_RECF_CONVERT_DELETE;
1097                         /* hammer_flush_record_done takes care of the rest */
1098                 } else {
1099                         record->flags |= HAMMER_RECF_DELETED_FE;
1100                         record->flags |= HAMMER_RECF_DELETED_BE;
1101                 }
1102         } else {
1103                 if (record->leaf.data_offset) {
1104                         hammer_blockmap_free(trans, record->leaf.data_offset,
1105                                              record->leaf.data_len);
1106                 }
1107         }
1108
1109 done:
1110         return(error);
1111 }
1112
1113 /*
1114  * Add the record to the inode's rec_tree.  The low 32 bits of a directory
1115  * entry's key is used to deal with hash collisions in the upper 32 bits.
1116  * A unique 64 bit key is generated in-memory and may be regenerated a
1117  * second time when the directory record is flushed to the on-disk B-Tree.
1118  *
1119  * A referenced record is passed to this function.  This function
1120  * eats the reference.  If an error occurs the record will be deleted.
1121  *
1122  * A copy of the temporary record->data pointer provided by the caller
1123  * will be made.
1124  */
1125 static
1126 int
1127 hammer_mem_add(hammer_record_t record)
1128 {
1129         hammer_mount_t hmp = record->ip->hmp;
1130
1131         /*
1132          * Make a private copy of record->data
1133          */
1134         if (record->data)
1135                 KKASSERT(record->flags & HAMMER_RECF_ALLOCDATA);
1136
1137         /*
1138          * Insert into the RB tree, find an unused iterator if this is
1139          * a directory entry.
1140          */
1141         while (RB_INSERT(hammer_rec_rb_tree, &record->ip->rec_tree, record)) {
1142                 if (record->leaf.base.rec_type != HAMMER_RECTYPE_DIRENTRY){
1143                         record->flags |= HAMMER_RECF_DELETED_FE;
1144                         hammer_rel_mem_record(record);
1145                         return (EEXIST);
1146                 }
1147                 if (++hmp->namekey_iterator == 0)
1148                         ++hmp->namekey_iterator;
1149                 record->leaf.base.key &= ~(0xFFFFFFFFLL);
1150                 record->leaf.base.key |= hmp->namekey_iterator;
1151         }
1152         ++hmp->rsv_recs;
1153         ++record->ip->rsv_recs;
1154         record->ip->hmp->rsv_databytes += record->leaf.data_len;
1155         record->flags |= HAMMER_RECF_ONRBTREE;
1156         hammer_modify_inode(record->ip, HAMMER_INODE_XDIRTY);
1157         hammer_rel_mem_record(record);
1158         return(0);
1159 }
1160
1161 /************************************************************************
1162  *                   HAMMER INODE MERGED-RECORD FUNCTIONS               *
1163  ************************************************************************
1164  *
1165  * These functions augment the B-Tree scanning functions in hammer_btree.c
1166  * by merging in-memory records with on-disk records.
1167  */
1168
1169 /*
1170  * Locate a particular record either in-memory or on-disk.
1171  *
1172  * NOTE: This is basically a standalone routine, hammer_ip_next() may
1173  * NOT be called to iterate results.
1174  */
1175 int
1176 hammer_ip_lookup(hammer_cursor_t cursor)
1177 {
1178         int error;
1179
1180         /*
1181          * If the element is in-memory return it without searching the
1182          * on-disk B-Tree
1183          */
1184         KKASSERT(cursor->ip);
1185         error = hammer_mem_lookup(cursor);
1186         if (error == 0) {
1187                 cursor->leaf = &cursor->iprec->leaf;
1188                 return(error);
1189         }
1190         if (error != ENOENT)
1191                 return(error);
1192
1193         /*
1194          * If the inode has on-disk components search the on-disk B-Tree.
1195          */
1196         if ((cursor->ip->flags & (HAMMER_INODE_ONDISK|HAMMER_INODE_DONDISK)) == 0)
1197                 return(error);
1198         error = hammer_btree_lookup(cursor);
1199         if (error == 0)
1200                 error = hammer_btree_extract(cursor, HAMMER_CURSOR_GET_LEAF);
1201         return(error);
1202 }
1203
1204 /*
1205  * Locate the first record within the cursor's key_beg/key_end range,
1206  * restricted to a particular inode.  0 is returned on success, ENOENT
1207  * if no records matched the requested range, or some other error.
1208  *
1209  * When 0 is returned hammer_ip_next() may be used to iterate additional
1210  * records within the requested range.
1211  *
1212  * This function can return EDEADLK, requiring the caller to terminate
1213  * the cursor and try again.
1214  */
1215 int
1216 hammer_ip_first(hammer_cursor_t cursor)
1217 {
1218         hammer_inode_t ip = cursor->ip;
1219         int error;
1220
1221         KKASSERT(ip != NULL);
1222
1223         /*
1224          * Clean up fields and setup for merged scan
1225          */
1226         cursor->flags &= ~HAMMER_CURSOR_DELBTREE;
1227         cursor->flags |= HAMMER_CURSOR_ATEDISK | HAMMER_CURSOR_ATEMEM;
1228         cursor->flags |= HAMMER_CURSOR_DISKEOF | HAMMER_CURSOR_MEMEOF;
1229         if (cursor->iprec) {
1230                 hammer_rel_mem_record(cursor->iprec);
1231                 cursor->iprec = NULL;
1232         }
1233
1234         /*
1235          * Search the on-disk B-Tree.  hammer_btree_lookup() only does an
1236          * exact lookup so if we get ENOENT we have to call the iterate
1237          * function to validate the first record after the begin key.
1238          *
1239          * The ATEDISK flag is used by hammer_btree_iterate to determine
1240          * whether it must index forwards or not.  It is also used here
1241          * to select the next record from in-memory or on-disk.
1242          *
1243          * EDEADLK can only occur if the lookup hit an empty internal
1244          * element and couldn't delete it.  Since this could only occur
1245          * in-range, we can just iterate from the failure point.
1246          */
1247         if (ip->flags & (HAMMER_INODE_ONDISK|HAMMER_INODE_DONDISK)) {
1248                 error = hammer_btree_lookup(cursor);
1249                 if (error == ENOENT || error == EDEADLK) {
1250                         cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
1251                         if (hammer_debug_general & 0x2000)
1252                                 kprintf("error %d node %p %016llx index %d\n", error, cursor->node, cursor->node->node_offset, cursor->index);
1253                         error = hammer_btree_iterate(cursor);
1254                 }
1255                 if (error && error != ENOENT) 
1256                         return(error);
1257                 if (error == 0) {
1258                         cursor->flags &= ~HAMMER_CURSOR_DISKEOF;
1259                         cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
1260                 } else {
1261                         cursor->flags |= HAMMER_CURSOR_ATEDISK;
1262                 }
1263         }
1264
1265         /*
1266          * Search the in-memory record list (Red-Black tree).  Unlike the
1267          * B-Tree search, mem_first checks for records in the range.
1268          */
1269         error = hammer_mem_first(cursor);
1270         if (error && error != ENOENT)
1271                 return(error);
1272         if (error == 0) {
1273                 cursor->flags &= ~HAMMER_CURSOR_MEMEOF;
1274                 cursor->flags &= ~HAMMER_CURSOR_ATEMEM;
1275                 if (hammer_ip_iterate_mem_good(cursor, cursor->iprec) == 0)
1276                         cursor->flags |= HAMMER_CURSOR_ATEMEM;
1277         }
1278
1279         /*
1280          * This will return the first matching record.
1281          */
1282         return(hammer_ip_next(cursor));
1283 }
1284
1285 /*
1286  * Retrieve the next record in a merged iteration within the bounds of the
1287  * cursor.  This call may be made multiple times after the cursor has been
1288  * initially searched with hammer_ip_first().
1289  *
1290  * 0 is returned on success, ENOENT if no further records match the
1291  * requested range, or some other error code is returned.
1292  */
1293 int
1294 hammer_ip_next(hammer_cursor_t cursor)
1295 {
1296         hammer_btree_elm_t elm;
1297         hammer_record_t rec, save;
1298         int error;
1299         int r;
1300
1301 next_btree:
1302         /*
1303          * Load the current on-disk and in-memory record.  If we ate any
1304          * records we have to get the next one. 
1305          *
1306          * If we deleted the last on-disk record we had scanned ATEDISK will
1307          * be clear and DELBTREE will be set, forcing a call to iterate. The
1308          * fact that ATEDISK is clear causes iterate to re-test the 'current'
1309          * element.  If ATEDISK is set, iterate will skip the 'current'
1310          * element.
1311          *
1312          * Get the next on-disk record
1313          */
1314         if (cursor->flags & (HAMMER_CURSOR_ATEDISK|HAMMER_CURSOR_DELBTREE)) {
1315                 if ((cursor->flags & HAMMER_CURSOR_DISKEOF) == 0) {
1316                         error = hammer_btree_iterate(cursor);
1317                         cursor->flags &= ~HAMMER_CURSOR_DELBTREE;
1318                         if (error == 0) {
1319                                 cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
1320                                 hammer_cache_node(cursor->node,
1321                                                   &cursor->ip->cache[1]);
1322                         } else {
1323                                 cursor->flags |= HAMMER_CURSOR_DISKEOF |
1324                                                  HAMMER_CURSOR_ATEDISK;
1325                         }
1326                 }
1327         }
1328
1329 next_memory:
1330         /*
1331          * Get the next in-memory record.  The record can be ripped out
1332          * of the RB tree so we maintain a scan_info structure to track
1333          * the next node.
1334          *
1335          * hammer_rec_scan_cmp:  Is the record still in our general range,
1336          *                       (non-inclusive of snapshot exclusions)?
1337          * hammer_rec_scan_callback: Is the record in our snapshot?
1338          */
1339         if (cursor->flags & HAMMER_CURSOR_ATEMEM) {
1340                 if ((cursor->flags & HAMMER_CURSOR_MEMEOF) == 0) {
1341                         save = cursor->iprec;
1342                         cursor->iprec = NULL;
1343                         rec = save ? hammer_rec_rb_tree_RB_NEXT(save) : NULL;
1344                         while (rec) {
1345                                 if (hammer_rec_scan_cmp(rec, cursor) != 0)
1346                                         break;
1347                                 if (hammer_rec_scan_callback(rec, cursor) != 0)
1348                                         break;
1349                                 rec = hammer_rec_rb_tree_RB_NEXT(rec);
1350                         }
1351                         if (save)
1352                                 hammer_rel_mem_record(save);
1353                         if (cursor->iprec) {
1354                                 KKASSERT(cursor->iprec == rec);
1355                                 cursor->flags &= ~HAMMER_CURSOR_ATEMEM;
1356                         } else {
1357                                 cursor->flags |= HAMMER_CURSOR_MEMEOF;
1358                         }
1359                 }
1360         }
1361
1362         /*
1363          * The memory record may have become stale while being held in
1364          * cursor->iprec.  We are interlocked against the backend on 
1365          * with regards to B-Tree entries.
1366          */
1367         if ((cursor->flags & HAMMER_CURSOR_ATEMEM) == 0) {
1368                 if (hammer_ip_iterate_mem_good(cursor, cursor->iprec) == 0) {
1369                         cursor->flags |= HAMMER_CURSOR_ATEMEM;
1370                         goto next_memory;
1371                 }
1372         }
1373
1374         /*
1375          * Extract either the disk or memory record depending on their
1376          * relative position.
1377          */
1378         error = 0;
1379         switch(cursor->flags & (HAMMER_CURSOR_ATEDISK | HAMMER_CURSOR_ATEMEM)) {
1380         case 0:
1381                 /*
1382                  * Both entries valid.   Compare the entries and nominally
1383                  * return the first one in the sort order.  Numerous cases
1384                  * require special attention, however.
1385                  */
1386                 elm = &cursor->node->ondisk->elms[cursor->index];
1387                 r = hammer_btree_cmp(&elm->base, &cursor->iprec->leaf.base);
1388
1389                 /*
1390                  * If the two entries differ only by their key (-2/2) or
1391                  * create_tid (-1/1), and are DATA records, we may have a
1392                  * nominal match.  We have to calculate the base file
1393                  * offset of the data.
1394                  */
1395                 if (r <= 2 && r >= -2 && r != 0 &&
1396                     cursor->ip->ino_data.obj_type == HAMMER_OBJTYPE_REGFILE &&
1397                     cursor->iprec->type == HAMMER_MEM_RECORD_DATA) {
1398                         int64_t base1 = elm->leaf.base.key - elm->leaf.data_len;
1399                         int64_t base2 = cursor->iprec->leaf.base.key -
1400                                         cursor->iprec->leaf.data_len;
1401                         if (base1 == base2)
1402                                 r = 0;
1403                 }
1404
1405                 if (r < 0) {
1406                         error = hammer_btree_extract(cursor,
1407                                                      HAMMER_CURSOR_GET_LEAF);
1408                         cursor->flags |= HAMMER_CURSOR_ATEDISK;
1409                         break;
1410                 }
1411
1412                 /*
1413                  * If the entries match exactly the memory entry is either
1414                  * an on-disk directory entry deletion or a bulk data
1415                  * overwrite.  If it is a directory entry deletion we eat
1416                  * both entries.
1417                  *
1418                  * For the bulk-data overwrite case it is possible to have
1419                  * visibility into both, which simply means the syncer
1420                  * hasn't gotten around to doing the delete+insert sequence
1421                  * on the B-Tree.  Use the memory entry and throw away the
1422                  * on-disk entry.
1423                  *
1424                  * If the in-memory record is not either of these we
1425                  * probably caught the syncer while it was syncing it to
1426                  * the media.  Since we hold a shared lock on the cursor,
1427                  * the in-memory record had better be marked deleted at
1428                  * this point.
1429                  */
1430                 if (r == 0) {
1431                         if (cursor->iprec->type == HAMMER_MEM_RECORD_DEL) {
1432                                 if ((cursor->flags & HAMMER_CURSOR_DELETE_VISIBILITY) == 0) {
1433                                         cursor->flags |= HAMMER_CURSOR_ATEDISK;
1434                                         cursor->flags |= HAMMER_CURSOR_ATEMEM;
1435                                         goto next_btree;
1436                                 }
1437                         } else if (cursor->iprec->type == HAMMER_MEM_RECORD_DATA) {
1438                                 if ((cursor->flags & HAMMER_CURSOR_DELETE_VISIBILITY) == 0) {
1439                                         cursor->flags |= HAMMER_CURSOR_ATEDISK;
1440                                 }
1441                                 /* fall through to memory entry */
1442                         } else {
1443                                 panic("hammer_ip_next: duplicate mem/b-tree entry");
1444                                 cursor->flags |= HAMMER_CURSOR_ATEMEM;
1445                                 goto next_memory;
1446                         }
1447                 }
1448                 /* fall through to the memory entry */
1449         case HAMMER_CURSOR_ATEDISK:
1450                 /*
1451                  * Only the memory entry is valid.
1452                  */
1453                 cursor->leaf = &cursor->iprec->leaf;
1454                 cursor->flags |= HAMMER_CURSOR_ATEMEM;
1455
1456                 /*
1457                  * If the memory entry is an on-disk deletion we should have
1458                  * also had found a B-Tree record.  If the backend beat us
1459                  * to it it would have interlocked the cursor and we should
1460                  * have seen the in-memory record marked DELETED_FE.
1461                  */
1462                 if (cursor->iprec->type == HAMMER_MEM_RECORD_DEL &&
1463                     (cursor->flags & HAMMER_CURSOR_DELETE_VISIBILITY) == 0) {
1464                         panic("hammer_ip_next: del-on-disk with no b-tree entry");
1465                 }
1466                 break;
1467         case HAMMER_CURSOR_ATEMEM:
1468                 /*
1469                  * Only the disk entry is valid
1470                  */
1471                 error = hammer_btree_extract(cursor, HAMMER_CURSOR_GET_LEAF);
1472                 cursor->flags |= HAMMER_CURSOR_ATEDISK;
1473                 break;
1474         default:
1475                 /*
1476                  * Neither entry is valid
1477                  *
1478                  * XXX error not set properly
1479                  */
1480                 cursor->leaf = NULL;
1481                 error = ENOENT;
1482                 break;
1483         }
1484         return(error);
1485 }
1486
1487 /*
1488  * Resolve the cursor->data pointer for the current cursor position in
1489  * a merged iteration.
1490  */
1491 int
1492 hammer_ip_resolve_data(hammer_cursor_t cursor)
1493 {
1494         hammer_record_t record;
1495         int error;
1496
1497         if (hammer_cursor_inmem(cursor)) {
1498                 /*
1499                  * The data associated with an in-memory record is usually
1500                  * kmalloced, but reserve-ahead data records will have an
1501                  * on-disk reference.
1502                  *
1503                  * NOTE: Reserve-ahead data records must be handled in the
1504                  * context of the related high level buffer cache buffer
1505                  * to interlock against async writes.
1506                  */
1507                 record = cursor->iprec;
1508                 cursor->data = record->data;
1509                 error = 0;
1510                 if (cursor->data == NULL) {
1511                         KKASSERT(record->leaf.base.rec_type ==
1512                                  HAMMER_RECTYPE_DATA);
1513                         cursor->data = hammer_bread(cursor->trans->hmp,
1514                                                     record->leaf.data_offset,
1515                                                     &error,
1516                                                     &cursor->data_buffer);
1517                 }
1518         } else {
1519                 cursor->leaf = &cursor->node->ondisk->elms[cursor->index].leaf;
1520                 error = hammer_btree_extract(cursor, HAMMER_CURSOR_GET_DATA);
1521         }
1522         return(error);
1523 }
1524
1525 /*
1526  * Backend truncation / record replacement - delete records in range.
1527  *
1528  * Delete all records within the specified range for inode ip.  In-memory
1529  * records still associated with the frontend are ignored.
1530  *
1531  * NOTE: An unaligned range will cause new records to be added to cover
1532  * the edge cases. (XXX not implemented yet).
1533  *
1534  * NOTE: Replacement via reservations (see hammer_ip_sync_record_cursor())
1535  * also do not deal with unaligned ranges.
1536  *
1537  * NOTE: ran_end is inclusive (e.g. 0,1023 instead of 0,1024).
1538  *
1539  * NOTE: Record keys for regular file data have to be special-cased since
1540  * they indicate the end of the range (key = base + bytes).
1541  */
1542 int
1543 hammer_ip_delete_range(hammer_cursor_t cursor, hammer_inode_t ip,
1544                        int64_t ran_beg, int64_t ran_end, int truncating)
1545 {
1546         hammer_transaction_t trans = cursor->trans;
1547         hammer_btree_leaf_elm_t leaf;
1548         int error;
1549         int64_t off;
1550
1551 #if 0
1552         kprintf("delete_range %p %016llx-%016llx\n", ip, ran_beg, ran_end);
1553 #endif
1554
1555         KKASSERT(trans->type == HAMMER_TRANS_FLS);
1556 retry:
1557         hammer_normalize_cursor(cursor);
1558         cursor->key_beg.localization = HAMMER_LOCALIZE_MISC;
1559         cursor->key_beg.obj_id = ip->obj_id;
1560         cursor->key_beg.create_tid = 0;
1561         cursor->key_beg.delete_tid = 0;
1562         cursor->key_beg.obj_type = 0;
1563         cursor->asof = ip->obj_asof;
1564         cursor->flags &= ~HAMMER_CURSOR_INITMASK;
1565         cursor->flags |= HAMMER_CURSOR_ASOF;
1566         cursor->flags |= HAMMER_CURSOR_DELETE_VISIBILITY;
1567         cursor->flags |= HAMMER_CURSOR_BACKEND;
1568
1569         cursor->key_end = cursor->key_beg;
1570         if (ip->ino_data.obj_type == HAMMER_OBJTYPE_DBFILE) {
1571                 cursor->key_beg.key = ran_beg;
1572                 cursor->key_beg.rec_type = HAMMER_RECTYPE_DB;
1573                 cursor->key_end.rec_type = HAMMER_RECTYPE_DB;
1574                 cursor->key_end.key = ran_end;
1575         } else {
1576                 /*
1577                  * The key in the B-Tree is (base+bytes), so the first possible
1578                  * matching key is ran_beg + 1.
1579                  */
1580                 int64_t tmp64;
1581
1582                 cursor->key_beg.key = ran_beg + 1;
1583                 cursor->key_beg.rec_type = HAMMER_RECTYPE_DATA;
1584                 cursor->key_end.rec_type = HAMMER_RECTYPE_DATA;
1585
1586                 tmp64 = ran_end + MAXPHYS + 1;  /* work around GCC-4 bug */
1587                 if (tmp64 < ran_end)
1588                         cursor->key_end.key = 0x7FFFFFFFFFFFFFFFLL;
1589                 else
1590                         cursor->key_end.key = ran_end + MAXPHYS + 1;
1591         }
1592         cursor->flags |= HAMMER_CURSOR_END_INCLUSIVE;
1593
1594         error = hammer_ip_first(cursor);
1595
1596         /*
1597          * Iterate through matching records and mark them as deleted.
1598          */
1599         while (error == 0) {
1600                 leaf = cursor->leaf;
1601
1602                 KKASSERT(leaf->base.delete_tid == 0);
1603
1604                 /*
1605                  * There may be overlap cases for regular file data.  Also
1606                  * remember the key for a regular file record is (base + len),
1607                  * NOT (base).
1608                  */
1609                 if (leaf->base.rec_type == HAMMER_RECTYPE_DATA) {
1610                         off = leaf->base.key - leaf->data_len;
1611                         /*
1612                          * Check the left edge case.  We currently do not
1613                          * split existing records.
1614                          */
1615                         if (off < ran_beg) {
1616                                 panic("hammer left edge case %016llx %d\n",
1617                                         leaf->base.key, leaf->data_len);
1618                         }
1619
1620                         /*
1621                          * Check the right edge case.  Note that the
1622                          * record can be completely out of bounds, which
1623                          * terminates the search.
1624                          *
1625                          * base->key is exclusive of the right edge while
1626                          * ran_end is inclusive of the right edge.  The
1627                          * (key - data_len) left boundary is inclusive.
1628                          *
1629                          * XXX theory-check this test at some point, are
1630                          * we missing a + 1 somewhere?  Note that ran_end
1631                          * could overflow.
1632                          */
1633                         if (leaf->base.key - 1 > ran_end) {
1634                                 if (leaf->base.key - leaf->data_len > ran_end)
1635                                         break;
1636                                 panic("hammer right edge case\n");
1637                         }
1638                 }
1639
1640                 /*
1641                  * Delete the record.  When truncating we do not delete
1642                  * in-memory (data) records because they represent data
1643                  * written after the truncation.
1644                  *
1645                  * This will also physically destroy the B-Tree entry and
1646                  * data if the retention policy dictates.  The function
1647                  * will set HAMMER_CURSOR_DELBTREE which hammer_ip_next()
1648                  * uses to perform a fixup.
1649                  */
1650                 if (truncating == 0 || hammer_cursor_ondisk(cursor))
1651                         error = hammer_ip_delete_record(cursor, ip, trans->tid);
1652                 if (error)
1653                         break;
1654                 error = hammer_ip_next(cursor);
1655         }
1656         if (cursor->node)
1657                 hammer_cache_node(cursor->node, &ip->cache[1]);
1658
1659         if (error == EDEADLK) {
1660                 hammer_done_cursor(cursor);
1661                 error = hammer_init_cursor(trans, cursor, &ip->cache[0], ip);
1662                 if (error == 0)
1663                         goto retry;
1664         }
1665         if (error == ENOENT)
1666                 error = 0;
1667         return(error);
1668 }
1669
1670 /*
1671  * Backend truncation - delete all records.
1672  *
1673  * Delete all user records associated with an inode except the inode record
1674  * itself.  Directory entries are not deleted (they must be properly disposed
1675  * of or nlinks would get upset).
1676  */
1677 int
1678 hammer_ip_delete_range_all(hammer_cursor_t cursor, hammer_inode_t ip,
1679                            int *countp)
1680 {
1681         hammer_transaction_t trans = cursor->trans;
1682         hammer_btree_leaf_elm_t leaf;
1683         int error;
1684
1685         KKASSERT(trans->type == HAMMER_TRANS_FLS);
1686 retry:
1687         hammer_normalize_cursor(cursor);
1688         cursor->key_beg.localization = HAMMER_LOCALIZE_MISC;
1689         cursor->key_beg.obj_id = ip->obj_id;
1690         cursor->key_beg.create_tid = 0;
1691         cursor->key_beg.delete_tid = 0;
1692         cursor->key_beg.obj_type = 0;
1693         cursor->key_beg.rec_type = HAMMER_RECTYPE_INODE + 1;
1694         cursor->key_beg.key = HAMMER_MIN_KEY;
1695
1696         cursor->key_end = cursor->key_beg;
1697         cursor->key_end.rec_type = 0xFFFF;
1698         cursor->key_end.key = HAMMER_MAX_KEY;
1699
1700         cursor->asof = ip->obj_asof;
1701         cursor->flags &= ~HAMMER_CURSOR_INITMASK;
1702         cursor->flags |= HAMMER_CURSOR_END_INCLUSIVE | HAMMER_CURSOR_ASOF;
1703         cursor->flags |= HAMMER_CURSOR_DELETE_VISIBILITY;
1704         cursor->flags |= HAMMER_CURSOR_BACKEND;
1705
1706         error = hammer_ip_first(cursor);
1707
1708         /*
1709          * Iterate through matching records and mark them as deleted.
1710          */
1711         while (error == 0) {
1712                 leaf = cursor->leaf;
1713
1714                 KKASSERT(leaf->base.delete_tid == 0);
1715
1716                 /*
1717                  * Mark the record and B-Tree entry as deleted.  This will
1718                  * also physically delete the B-Tree entry, record, and
1719                  * data if the retention policy dictates.  The function
1720                  * will set HAMMER_CURSOR_DELBTREE which hammer_ip_next()
1721                  * uses to perform a fixup.
1722                  *
1723                  * Directory entries (and delete-on-disk directory entries)
1724                  * must be synced and cannot be deleted.
1725                  */
1726                 if (leaf->base.rec_type != HAMMER_RECTYPE_DIRENTRY) {
1727                         error = hammer_ip_delete_record(cursor, ip, trans->tid);
1728                         ++*countp;
1729                 }
1730                 if (error)
1731                         break;
1732                 error = hammer_ip_next(cursor);
1733         }
1734         if (cursor->node)
1735                 hammer_cache_node(cursor->node, &ip->cache[1]);
1736         if (error == EDEADLK) {
1737                 hammer_done_cursor(cursor);
1738                 error = hammer_init_cursor(trans, cursor, &ip->cache[0], ip);
1739                 if (error == 0)
1740                         goto retry;
1741         }
1742         if (error == ENOENT)
1743                 error = 0;
1744         return(error);
1745 }
1746
1747 /*
1748  * Delete the record at the current cursor.  On success the cursor will
1749  * be positioned appropriately for an iteration but may no longer be at
1750  * a leaf node.
1751  *
1752  * This routine is only called from the backend.
1753  *
1754  * NOTE: This can return EDEADLK, requiring the caller to terminate the
1755  * cursor and retry.
1756  */
1757 int
1758 hammer_ip_delete_record(hammer_cursor_t cursor, hammer_inode_t ip,
1759                         hammer_tid_t tid)
1760 {
1761         hammer_btree_elm_t elm;
1762         hammer_mount_t hmp;
1763         int error;
1764         int dodelete;
1765
1766         KKASSERT(cursor->flags & HAMMER_CURSOR_BACKEND);
1767         KKASSERT(tid != 0);
1768
1769         /*
1770          * In-memory (unsynchronized) records can simply be freed.  This
1771          * only occurs in range iterations since all other records are
1772          * individually synchronized.  Thus there should be no confusion with
1773          * the interlock.
1774          */
1775         if (hammer_cursor_inmem(cursor)) {
1776                 KKASSERT((cursor->iprec->flags & HAMMER_RECF_INTERLOCK_BE) ==0);
1777                 cursor->iprec->flags |= HAMMER_RECF_DELETED_FE;
1778                 cursor->iprec->flags |= HAMMER_RECF_DELETED_BE;
1779                 return(0);
1780         }
1781
1782         /*
1783          * On-disk records are marked as deleted by updating their delete_tid.
1784          * This does not effect their position in the B-Tree (which is based
1785          * on their create_tid).
1786          */
1787         error = hammer_btree_extract(cursor, HAMMER_CURSOR_GET_LEAF);
1788         elm = NULL;
1789         hmp = cursor->node->hmp;
1790
1791         /*
1792          * If we were mounted with the nohistory option, we physically
1793          * delete the record.
1794          */
1795         dodelete = hammer_nohistory(ip);
1796
1797         if (error == 0) {
1798                 error = hammer_cursor_upgrade(cursor);
1799                 if (error == 0) {
1800                         elm = &cursor->node->ondisk->elms[cursor->index];
1801                         hammer_modify_node(cursor->trans, cursor->node,
1802                                            &elm->leaf.base.delete_tid,
1803                                            sizeof(elm->leaf.base.delete_tid));
1804                         elm->leaf.base.delete_tid = tid;
1805                         hammer_modify_node_done(cursor->node);
1806
1807                         /*
1808                          * An on-disk record cannot have the same delete_tid
1809                          * as its create_tid.  In a chain of record updates
1810                          * this could result in a duplicate record.
1811                          */
1812                         KKASSERT(elm->leaf.base.delete_tid != elm->leaf.base.create_tid);
1813                 }
1814         }
1815
1816         if (error == 0 && dodelete) {
1817                 error = hammer_delete_at_cursor(cursor, NULL);
1818                 if (error) {
1819                         panic("hammer_ip_delete_record: unable to physically delete the record!\n");
1820                         error = 0;
1821                 }
1822         }
1823         return(error);
1824 }
1825
1826 int
1827 hammer_delete_at_cursor(hammer_cursor_t cursor, int64_t *stat_bytes)
1828 {
1829         hammer_btree_elm_t elm;
1830         hammer_off_t data_offset;
1831         int32_t data_len;
1832         u_int16_t rec_type;
1833         int error;
1834
1835         elm = &cursor->node->ondisk->elms[cursor->index];
1836         KKASSERT(elm->base.btype == HAMMER_BTREE_TYPE_RECORD);
1837
1838         data_offset = elm->leaf.data_offset;
1839         data_len = elm->leaf.data_len;
1840         rec_type = elm->leaf.base.rec_type;
1841
1842         error = hammer_btree_delete(cursor);
1843         if (error == 0) {
1844                 /*
1845                  * This forces a fixup for the iteration because
1846                  * the cursor is now either sitting at the 'next'
1847                  * element or sitting at the end of a leaf.
1848                  */
1849                 if ((cursor->flags & HAMMER_CURSOR_DISKEOF) == 0) {
1850                         cursor->flags |= HAMMER_CURSOR_DELBTREE;
1851                         cursor->flags &= ~HAMMER_CURSOR_ATEDISK;
1852                 }
1853         }
1854         if (error == 0) {
1855                 switch(data_offset & HAMMER_OFF_ZONE_MASK) {
1856                 case HAMMER_ZONE_LARGE_DATA:
1857                 case HAMMER_ZONE_SMALL_DATA:
1858                 case HAMMER_ZONE_META:
1859                         hammer_blockmap_free(cursor->trans,
1860                                              data_offset, data_len);
1861                         break;
1862                 default:
1863                         break;
1864                 }
1865         }
1866         return (error);
1867 }
1868
1869 /*
1870  * Determine whether we can remove a directory.  This routine checks whether
1871  * a directory is empty or not and enforces flush connectivity.
1872  *
1873  * Flush connectivity requires that we block if the target directory is
1874  * currently flushing, otherwise it may not end up in the same flush group.
1875  *
1876  * Returns 0 on success, ENOTEMPTY or EDEADLK (or other errors) on failure.
1877  */
1878 int
1879 hammer_ip_check_directory_empty(hammer_transaction_t trans, hammer_inode_t ip)
1880 {
1881         struct hammer_cursor cursor;
1882         int error;
1883
1884         /*
1885          * Check directory empty
1886          */
1887         hammer_init_cursor(trans, &cursor, &ip->cache[0], ip);
1888
1889         cursor.key_beg.localization = HAMMER_LOCALIZE_MISC;
1890         cursor.key_beg.obj_id = ip->obj_id;
1891         cursor.key_beg.create_tid = 0;
1892         cursor.key_beg.delete_tid = 0;
1893         cursor.key_beg.obj_type = 0;
1894         cursor.key_beg.rec_type = HAMMER_RECTYPE_INODE + 1;
1895         cursor.key_beg.key = HAMMER_MIN_KEY;
1896
1897         cursor.key_end = cursor.key_beg;
1898         cursor.key_end.rec_type = 0xFFFF;
1899         cursor.key_end.key = HAMMER_MAX_KEY;
1900
1901         cursor.asof = ip->obj_asof;
1902         cursor.flags |= HAMMER_CURSOR_END_INCLUSIVE | HAMMER_CURSOR_ASOF;
1903
1904         error = hammer_ip_first(&cursor);
1905         if (error == ENOENT)
1906                 error = 0;
1907         else if (error == 0)
1908                 error = ENOTEMPTY;
1909         hammer_done_cursor(&cursor);
1910         return(error);
1911 }
1912