1/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
2/*
3 *     Copyright 2010 Couchbase, Inc
4 *
5 *   Licensed under the Apache License, Version 2.0 (the "License");
6 *   you may not use this file except in compliance with the License.
7 *   You may obtain a copy of the License at
8 *
9 *       http://www.apache.org/licenses/LICENSE-2.0
10 *
11 *   Unless required by applicable law or agreed to in writing, software
12 *   distributed under the License is distributed on an "AS IS" BASIS,
13 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 *   See the License for the specific language governing permissions and
15 *   limitations under the License.
16 */
17
18#include <stdio.h>
19#include <stdlib.h>
20#include <string.h>
21#include <stdint.h>
22#include <time.h>
23#include<mutex>
24#if !defined(WIN32) && !defined(_WIN32)
25#include <unistd.h>
26#endif
27
28#include "libforestdb/forestdb.h"
29#include "test.h"
30
31#include "internal_types.h"
32#include "functional_util.h"
33
34void rollback_secondary_kvs()
35{
36    TEST_INIT();
37    memleak_start();
38
39    int r;
40    void *value_out;
41    size_t valuelen_out;
42
43    fdb_status status;
44    fdb_config fconfig = fdb_get_default_config();
45    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
46    fdb_file_handle *dbfile;
47    fdb_kvs_handle *kv1, *kv2;
48
49    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
50    (void)r;
51
52    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
53    fdb_kvs_open_default(dbfile, &kv1, &kvs_config);
54    fdb_kvs_open_default(dbfile, &kv2, &kvs_config);
55
56    // seq:2
57    status = fdb_set_kv(kv1, (void *) "a", 1, (void *)"val-a", 5);
58    TEST_CHK(status == FDB_RESULT_SUCCESS);
59    status = fdb_set_kv(kv1, (void *) "b", 1, (void *)"val-b", 5);
60    TEST_CHK(status == FDB_RESULT_SUCCESS);
61    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
62
63    // seq:3
64    status = fdb_set_kv(kv1, (void *) "b", 1, (void *)"val-v", 5);
65    TEST_CHK(status == FDB_RESULT_SUCCESS);
66    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
67
68    // seq:4
69    status = fdb_del_kv(kv1, (void *)"a", 1);
70    TEST_CHK(status == FDB_RESULT_SUCCESS);
71    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
72
73    // get 'a' via kv2
74    status = fdb_get_kv(kv2, (void *)"b", 1, &value_out, &valuelen_out);
75    free(value_out);
76    TEST_CHK(status == FDB_RESULT_SUCCESS);
77
78    // rollback seq:3 via kv2
79    status = fdb_rollback(&kv2, 3);
80    TEST_CHK(status == FDB_RESULT_SUCCESS);
81
82    fdb_close(dbfile);
83    fdb_shutdown();
84
85    memleak_end();
86    TEST_RESULT("rollback secondary kv");
87}
88
89
90void multi_version_test()
91{
92    TEST_INIT();
93
94    memleak_start();
95
96    int i, r;
97    int n = 2;
98    fdb_file_handle *dbfile, *dbfile_new;
99    fdb_kvs_handle *db;
100    fdb_kvs_handle *db_new;
101    fdb_doc **doc = alca(fdb_doc*, n);
102    fdb_doc *rdoc;
103    fdb_status status;
104
105    char keybuf[256], metabuf[256], bodybuf[256];
106
107    // remove all previous mvcc_test files
108    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
109    (void)r;
110
111    fdb_config fconfig = fdb_get_default_config();
112    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
113    fconfig.buffercache_size = 1048576;
114    fconfig.wal_threshold = 1024;
115    fconfig.flags = FDB_OPEN_FLAG_CREATE;
116    fconfig.compaction_threshold = 0;
117
118    // open db
119    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
120    fdb_kvs_open_default(dbfile, &db, &kvs_config);
121    status = fdb_set_log_callback(db, logCallbackFunc, (void *) "multi_version_test");
122    TEST_CHK(status == FDB_RESULT_SUCCESS);
123
124    // insert documents
125    for (i=0;i<n;++i){
126        sprintf(keybuf, "key%d", i);
127        sprintf(metabuf, "meta%d", i);
128        sprintf(bodybuf, "body%d", i);
129        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
130            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
131        fdb_set(db, doc[i]);
132    }
133
134    // commit
135    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
136
137    // open same db file using a new handle
138    fdb_open(&dbfile_new, "./mvcc_test1", &fconfig);
139    fdb_kvs_open_default(dbfile_new, &db_new, &kvs_config);
140    status = fdb_set_log_callback(db_new, logCallbackFunc, (void *) "multi_version_test");
141    TEST_CHK(status == FDB_RESULT_SUCCESS);
142
143    // update documents using the old handle
144    for (i=0;i<n;++i){
145        sprintf(metabuf, "meta2%d", i);
146        sprintf(bodybuf, "body2%d", i);
147        fdb_doc_update(&doc[i], (void*)metabuf, strlen(metabuf),
148            (void*)bodybuf, strlen(bodybuf));
149        fdb_set(db, doc[i]);
150    }
151
152    // manually flush WAL and commit using the old handle
153    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
154
155    // retrieve documents using the old handle
156    for (i=0;i<n;++i){
157        // search by key
158        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
159        status = fdb_get(db, rdoc);
160
161        TEST_CHK(status == FDB_RESULT_SUCCESS);
162        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
163        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
164
165        // free result document
166        fdb_doc_free(rdoc);
167    }
168
169    // retrieve documents using the new handle
170    for (i=0;i<n;++i){
171        // search by key
172        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
173        status = fdb_get(db_new, rdoc);
174
175        TEST_CHK(status == FDB_RESULT_SUCCESS);
176        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
177        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
178
179        // free result document
180        fdb_doc_free(rdoc);
181    }
182
183    // close and re-open the new handle
184    fdb_kvs_close(db_new);
185    fdb_close(dbfile_new);
186    fdb_open(&dbfile_new, "./mvcc_test1", &fconfig);
187    fdb_kvs_open_default(dbfile_new, &db_new, &kvs_config);
188    status = fdb_set_log_callback(db_new, logCallbackFunc, (void *) "multi_version_test");
189    TEST_CHK(status == FDB_RESULT_SUCCESS);
190
191    // retrieve documents using the new handle
192    for (i=0;i<n;++i){
193        // search by key
194        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
195        status = fdb_get(db_new, rdoc);
196
197        TEST_CHK(status == FDB_RESULT_SUCCESS);
198        // the new version of data should be read
199        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
200        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
201
202        // free result document
203        fdb_doc_free(rdoc);
204    }
205
206
207    // free all documents
208    for (i=0;i<n;++i){
209        fdb_doc_free(doc[i]);
210    }
211
212    // close db file
213    fdb_kvs_close(db);
214    fdb_kvs_close(db_new);
215    fdb_close(dbfile);
216    fdb_close(dbfile_new);
217
218    // free all resources
219    fdb_shutdown();
220
221    memleak_end();
222
223    TEST_RESULT("multi version test");
224}
225
226void crash_recovery_test(bool walflush)
227{
228    TEST_INIT();
229
230    memleak_start();
231
232    int i, r;
233    int n = 10;
234    fdb_file_handle *dbfile;
235    fdb_kvs_handle *db;
236    fdb_doc **doc = alca(fdb_doc*, n);
237    fdb_doc *rdoc;
238    fdb_file_info file_info;
239    uint64_t bid;
240    const char *test_file = "./mvcc_test2";
241    fdb_status status;
242
243    char keybuf[256], metabuf[256], bodybuf[256];
244
245    // remove previous mvcc_test files
246    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
247    (void)r;
248
249    fdb_config fconfig = fdb_get_default_config();
250    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
251    fconfig.buffercache_size = 0;
252    fconfig.seqtree_opt = FDB_SEQTREE_USE; // enable seqtree since get_byseq
253    fconfig.wal_threshold = 1024;
254    fconfig.flags = FDB_OPEN_FLAG_CREATE;
255    fconfig.compaction_threshold = 0;
256
257    // reopen db
258    fdb_open(&dbfile, test_file, &fconfig);
259    fdb_kvs_open_default(dbfile, &db, &kvs_config);
260    status = fdb_set_log_callback(db, logCallbackFunc,
261                                  (void *) "crash_recovery_test");
262    TEST_CHK(status == FDB_RESULT_SUCCESS);
263
264    // insert documents
265    for (i=0;i<n;++i){
266        sprintf(keybuf, "key%d", i);
267        sprintf(metabuf, "meta%d", i);
268        sprintf(bodybuf, "body%d", i);
269        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
270            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
271        fdb_set(db, doc[i]);
272    }
273
274    // commit
275    if(walflush){
276        status = fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
277        TEST_CHK(status == FDB_RESULT_SUCCESS);
278    } else {
279        status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
280        TEST_CHK(status == FDB_RESULT_SUCCESS);
281    }
282
283    status = fdb_get_file_info(dbfile, &file_info);
284    TEST_CHK(status == FDB_RESULT_SUCCESS);
285    bid = file_info.file_size / fconfig.blocksize;
286
287    // close the db
288    fdb_kvs_close(db);
289    fdb_close(dbfile);
290
291    // Shutdown forest db in the middle of the test to simulate crash
292    fdb_shutdown();
293
294    // append 9K of non-block aligned garbage at end of file
295    r = _disk_dump(test_file, bid * fconfig.blocksize,
296                   (2 * fconfig.blocksize) + (fconfig.blocksize / 4));
297    TEST_CHK(r >= 0);
298
299    // reopen the same file
300    status = fdb_open(&dbfile, test_file, &fconfig);
301    TEST_CHK(status == FDB_RESULT_SUCCESS);
302    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
303    TEST_CHK(status == FDB_RESULT_SUCCESS);
304    status = fdb_set_log_callback(db, logCallbackFunc,
305                                  (void *) "crash_recovery_test");
306    TEST_CHK(status == FDB_RESULT_SUCCESS);
307
308    // retrieve documents
309    for (i=0;i<n;++i){
310        // search by key
311        fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
312        status = fdb_get(db, rdoc);
313
314        TEST_CHK(status == FDB_RESULT_SUCCESS);
315        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
316        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
317
318        // free result document
319        fdb_doc_free(rdoc);
320    }
321
322    // retrieve documents by sequence number
323    for (i=0;i<n;++i){
324        // search by seq
325        fdb_doc_create(&rdoc, NULL, 0, NULL, 0, NULL, 0);
326        rdoc->seqnum = i + 1;
327        status = fdb_get_byseq(db, rdoc);
328
329        TEST_CHK(status == FDB_RESULT_SUCCESS);
330
331        // free result document
332        fdb_doc_free(rdoc);
333    }
334
335    // free all documents
336    for (i=0;i<n;++i){
337        fdb_doc_free(doc[i]);
338    }
339
340    // close db file
341    fdb_kvs_close(db);
342    fdb_close(dbfile);
343
344    // free all resources
345    fdb_shutdown();
346
347    memleak_end();
348
349    if(walflush){
350        TEST_RESULT("crash recovery test - walflush");
351    } else {
352        TEST_RESULT("crash recovery test - normal flush");
353    }
354}
355
356void snapshot_test()
357{
358    TEST_INIT();
359
360    memleak_start();
361
362    int i, r;
363    int n = 20;
364    int count;
365    fdb_file_handle *dbfile;
366    fdb_kvs_handle *db;
367    fdb_kvs_handle *snap_db;
368    fdb_seqnum_t snap_seq;
369    fdb_doc **doc = alca(fdb_doc*, n);
370    fdb_doc *rdoc = NULL;
371    fdb_kvs_info kvs_info;
372    fdb_status status;
373    fdb_iterator *iterator;
374
375    char keybuf[256], metabuf[256], bodybuf[256];
376
377    // remove previous mvcc_test files
378    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
379    (void)r;
380
381    fdb_config fconfig = fdb_get_default_config();
382    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
383    fconfig.buffercache_size = 0;
384    fconfig.seqtree_opt = FDB_SEQTREE_USE; // enable seqtree since get_byseq
385    fconfig.wal_threshold = 1024;
386    fconfig.flags = FDB_OPEN_FLAG_CREATE;
387    fconfig.compaction_threshold = 0;
388
389    // remove previous mvcc_test files
390    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
391    (void)r;
392
393    // open db
394    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
395    fdb_kvs_open_default(dbfile, &db, &kvs_config);
396    TEST_CHK(status == FDB_RESULT_SUCCESS);
397
398    // Create a snapshot from an empty database file
399    status = fdb_snapshot_open(db, &snap_db, 0);
400    TEST_CHK(status == FDB_RESULT_SUCCESS);
401    // check if snapshot's sequence number is zero.
402    fdb_get_kvs_info(snap_db, &kvs_info);
403    TEST_CHK(kvs_info.last_seqnum == 0);
404    // create an iterator on the snapshot for full range
405    fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
406    // Iterator should not return any items.
407    status = fdb_iterator_get(iterator, &rdoc);
408    TEST_CHK(status == FDB_RESULT_ITERATOR_FAIL);
409    fdb_iterator_close(iterator);
410    fdb_kvs_close(snap_db);
411
412    // ------- Setup test ----------------------------------
413    // insert documents of 0-4
414    for (i=0; i<n/4; i++){
415        sprintf(keybuf, "key%d", i);
416        sprintf(metabuf, "meta%d", i);
417        sprintf(bodybuf, "body%d", i);
418        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
419            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
420        fdb_set(db, doc[i]);
421    }
422
423    // commit with a manual WAL flush (these docs go into HB-trie)
424    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
425
426    // insert documents from 4 - 8
427    for (; i < n/2 - 1; i++){
428        sprintf(keybuf, "key%d", i);
429        sprintf(metabuf, "meta%d", i);
430        sprintf(bodybuf, "body%d", i);
431        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
432            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
433        fdb_set(db, doc[i]);
434    }
435
436    // We want to create:
437    // |WALFlushHDR|Key-Value1|HDR|Key-Value2|SnapshotHDR|Key-Value1|HDR|
438    // Insert doc 9 with a different value to test duplicate elimination..
439    sprintf(keybuf, "key%d", i);
440    sprintf(metabuf, "meta%d", i);
441    sprintf(bodybuf, "Body%d", i);
442    fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
443            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
444    fdb_set(db, doc[i]);
445
446    // commit again without a WAL flush (last doc goes into the AVL tree)
447    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
448
449    // Insert doc 9 now again with expected value..
450    *(char *)doc[i]->body = 'b';
451    fdb_set(db, doc[i]);
452    // commit again without a WAL flush (these documents go into the AVL trees)
453    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
454
455    // TAKE SNAPSHOT: pick up sequence number of a commit without a WAL flush
456    snap_seq = doc[i]->seqnum;
457
458    // Now re-insert doc 9 as another duplicate (only newer sequence number)
459    fdb_set(db, doc[i]);
460    // commit again without a WAL flush (last doc goes into the AVL tree)
461    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
462
463    // insert documents from 10-14 into HB-trie
464    for (++i; i < (n/2 + n/4); i++){
465        sprintf(keybuf, "key%d", i);
466        sprintf(metabuf, "meta%d", i);
467        sprintf(bodybuf, "body%d", i);
468        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
469            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
470        fdb_set(db, doc[i]);
471    }
472    // manually flush WAL & commit
473    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
474
475    status = fdb_set_log_callback(db, logCallbackFunc,
476                                  (void *) "snapshot_test");
477    TEST_CHK(status == FDB_RESULT_SUCCESS);
478    // ---------- Snapshot tests begin -----------------------
479    // Attempt to take snapshot with out-of-range marker..
480    status = fdb_snapshot_open(db, &snap_db, 999999);
481    TEST_CHK(status == FDB_RESULT_NO_DB_INSTANCE);
482
483    // Init Snapshot of open file with saved document seqnum as marker
484    status = fdb_snapshot_open(db, &snap_db, snap_seq);
485    TEST_CHK(status == FDB_RESULT_SUCCESS);
486
487    // check snapshot's sequence number
488    fdb_get_kvs_info(snap_db, &kvs_info);
489    TEST_CHK(kvs_info.last_seqnum == snap_seq);
490
491    // insert documents from 15 - 19 on file into the WAL
492    for (; i < n; i++){
493        sprintf(keybuf, "key%d", i);
494        sprintf(metabuf, "meta%d", i);
495        sprintf(bodybuf, "body%d", i);
496        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
497            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
498        fdb_set(db, doc[i]);
499    }
500    // commit without a WAL flush (This WAL must not affect snapshot)
501    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
502
503    // create an iterator on the snapshot for full range
504    fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
505
506    // repeat until fail
507    i=0;
508    count=0;
509    do {
510        status = fdb_iterator_get(iterator, &rdoc);
511        TEST_CHK(status == FDB_RESULT_SUCCESS);
512
513        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
514        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
515        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
516
517        fdb_doc_free(rdoc);
518        rdoc = NULL;
519        i ++;
520        count++;
521    } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
522
523    TEST_CHK(count==n/2); // Only unique items from the first half
524
525    fdb_iterator_close(iterator);
526
527    // create a sequence iterator on the snapshot for full range
528    fdb_iterator_sequence_init(snap_db, &iterator, 0, 0, FDB_ITR_NONE);
529
530    // repeat until fail
531    i=0;
532    count=0;
533    do {
534        status = fdb_iterator_get(iterator, &rdoc);
535        TEST_CHK(status == FDB_RESULT_SUCCESS);
536
537        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
538        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
539        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
540
541        fdb_doc_free(rdoc);
542        rdoc = NULL;
543        i ++;
544        count++;
545    } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
546
547    TEST_CHK(count==n/2); // Only unique items from the first half
548
549    fdb_iterator_close(iterator);
550
551    // close db handle
552    fdb_kvs_close(db);
553    // close snapshot handle
554    fdb_kvs_close(snap_db);
555    // close db file
556    fdb_close(dbfile);
557
558    // free all documents
559    for (i=0;i<n;++i){
560        fdb_doc_free(doc[i]);
561    }
562
563    // 1. Open Database File
564    status = fdb_open(&dbfile, "./mvcc_test2", &fconfig);
565    TEST_CHK(status == FDB_RESULT_SUCCESS);
566
567    // 2. Open KV Store
568    status = fdb_kvs_open(dbfile, &db, "kv2", &kvs_config);
569    TEST_CHK(status == FDB_RESULT_SUCCESS);
570
571    // 2. Create Key 'a' and Commit
572    status = fdb_set_kv(db, (void *) "a", 1, (void *)"val-a", 5);
573    TEST_CHK(status == FDB_RESULT_SUCCESS);
574    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
575    TEST_CHK(status == FDB_RESULT_SUCCESS);
576
577    // 3. Create Key 'b' and Commit
578    status = fdb_set_kv(db, (void *)"b", 1, (void *)"val-b", 5);
579    TEST_CHK(status == FDB_RESULT_SUCCESS);
580    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
581    TEST_CHK(status == FDB_RESULT_SUCCESS);
582
583    // 4. Create Key 'c' and Commit
584    status = fdb_set_kv(db, (void *)"c", 1, (void *)"val-c", 5);
585    TEST_CHK(status == FDB_RESULT_SUCCESS);
586    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
587    TEST_CHK(status == FDB_RESULT_SUCCESS);
588
589    // 5. Remember seqnum for opening snapshot
590    status = fdb_get_kvs_info(db, &kvs_info);
591    TEST_CHK(status == FDB_RESULT_SUCCESS);
592    snap_seq = kvs_info.last_seqnum;
593
594    // 6.  Create an iterator
595    status = fdb_iterator_init(db, &iterator, (void *)"a", 1,
596                               (void *)"c", 1, FDB_ITR_NONE);
597    TEST_CHK(status == FDB_RESULT_SUCCESS);
598
599    // 7. Do a seek
600    status = fdb_iterator_seek(iterator, (void *)"b", 1, FDB_ITR_SEEK_HIGHER);
601    TEST_CHK(status == FDB_RESULT_SUCCESS);
602
603    // 8. Close the iterator
604    status = fdb_iterator_close(iterator);
605    TEST_CHK(status == FDB_RESULT_SUCCESS);
606
607    // 9. Open a snapshot at the same point
608    status = fdb_snapshot_open(db, &snap_db, snap_seq);
609    TEST_CHK(status == FDB_RESULT_SUCCESS);
610
611    // That works, now attempt to do exact same sequence on a snapshot
612    // The snapshot is opened at the exact same place that the original
613    // database handle should be at (last seq 3)
614
615    // 10.  Create an iterator on snapshot
616    status = fdb_iterator_init(snap_db, &iterator, (void *)"a", 1,
617                               (void *)"c", 1, FDB_ITR_NONE);
618    TEST_CHK(status == FDB_RESULT_SUCCESS);
619
620    // 11. Do a seek
621    status = fdb_iterator_seek(iterator, (void *)"b", 1, FDB_ITR_SEEK_HIGHER);
622    TEST_CHK(status == FDB_RESULT_SUCCESS);
623
624    // 12. Close the iterator
625    status = fdb_iterator_close(iterator);
626    TEST_CHK(status == FDB_RESULT_SUCCESS);
627
628    // close db handle
629    fdb_kvs_close(db);
630    // close snapshot handle
631    fdb_kvs_close(snap_db);
632    // close db file
633    fdb_close(dbfile);
634
635    // free all resources
636    fdb_shutdown();
637
638    memleak_end();
639
640    TEST_RESULT("snapshot test");
641}
642
643void snapshot_stats_test()
644{
645    TEST_INIT();
646
647    memleak_start();
648
649    int i, r;
650    int n = 20;
651    fdb_file_handle *dbfile;
652    fdb_kvs_handle *db;
653    fdb_kvs_handle *snap_db;
654    fdb_seqnum_t snap_seq;
655    fdb_doc **doc = alca(fdb_doc*, n);
656    fdb_kvs_info kvs_info;
657    fdb_status status;
658
659    char keybuf[256], metabuf[256], bodybuf[256];
660
661    // remove previous mvcc_test files
662    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
663    (void)r;
664
665    fdb_config fconfig = fdb_get_default_config();
666    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
667    fconfig.buffercache_size = 0;
668    fconfig.wal_threshold = 1024;
669    fconfig.flags = FDB_OPEN_FLAG_CREATE;
670    fconfig.compaction_threshold = 0;
671
672    // remove previous mvcc_test files
673    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
674    (void)r;
675
676    // open db
677    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
678    fdb_kvs_open_default(dbfile, &db, &kvs_config);
679    TEST_CHK(status == FDB_RESULT_SUCCESS);
680
681    // ------- Setup test ----------------------------------
682    // insert documents
683    for (i=0; i<n; i++){
684        sprintf(keybuf, "key%d", i);
685        sprintf(metabuf, "meta%d", i);
686        sprintf(bodybuf, "body%d", i);
687        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
688            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
689        fdb_set(db, doc[i]);
690        fdb_doc_free(doc[i]);
691    }
692
693    // commit with a manual WAL flush (these docs go into HB-trie)
694    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
695
696    // retrieve the sequence number for a snapshot open
697    status = fdb_get_kvs_info(db, &kvs_info);
698    TEST_CHK(status == FDB_RESULT_SUCCESS);
699    snap_seq = kvs_info.last_seqnum;
700
701    status = fdb_set_log_callback(db, logCallbackFunc,
702                                  (void *) "snapshot_stats_test");
703    TEST_CHK(status == FDB_RESULT_SUCCESS);
704
705    // Init Snapshot of open file with saved document seqnum as marker
706    status = fdb_snapshot_open(db, &snap_db, snap_seq);
707    TEST_CHK(status == FDB_RESULT_SUCCESS);
708
709    // check snapshot's sequence number
710    fdb_get_kvs_info(snap_db, &kvs_info);
711    TEST_CHK(kvs_info.last_seqnum == snap_seq);
712
713    TEST_CHK(kvs_info.doc_count == (size_t)n);
714    TEST_CHK(kvs_info.file == dbfile);
715
716    // close snapshot handle
717    fdb_kvs_close(snap_db);
718
719    // Test stats by creating an in-memory snapshot
720    status = fdb_snapshot_open(db, &snap_db, FDB_SNAPSHOT_INMEM);
721    TEST_CHK(status == FDB_RESULT_SUCCESS);
722
723    // check snapshot's sequence number
724    fdb_get_kvs_info(snap_db, &kvs_info);
725    TEST_CHK(kvs_info.last_seqnum == snap_seq);
726
727    TEST_CHK(kvs_info.doc_count == (size_t)n);
728    TEST_CHK(kvs_info.file == dbfile);
729
730    // close snapshot handle
731    fdb_kvs_close(snap_db);
732
733    // close db handle
734    fdb_kvs_close(db);
735    // close db file
736    fdb_close(dbfile);
737
738    // free all resources
739    fdb_shutdown();
740
741    memleak_end();
742
743    TEST_RESULT("snapshot stats test");
744}
745
746void snapshot_with_uncomitted_data_test()
747{
748    TEST_INIT();
749
750    int n = 10, value_len=32;
751    int i, r, idx;
752    char cmd[256];
753    char key[256], *value;
754    char keystr[] = "key%06d";
755    char valuestr[] = "value%d";
756    fdb_file_handle *db_file;
757    fdb_kvs_handle *db0, *db1, *db2, *snap;
758    fdb_config config;
759    fdb_kvs_config kvs_config;
760    fdb_kvs_info info;
761    fdb_seqnum_t seqnum;
762    fdb_status s; (void)s;
763
764    sprintf(cmd, SHELL_DEL " mvcc_test* > errorlog.txt");
765    r = system(cmd);
766    (void)r;
767
768    memleak_start();
769
770    value = (char*)malloc(value_len);
771
772    srand(1234);
773    config = fdb_get_default_config();
774    config.seqtree_opt = FDB_SEQTREE_USE;
775    config.wal_flush_before_commit = true;
776    config.multi_kv_instances = true;
777    config.buffercache_size = 0*1024*1024;
778
779    kvs_config = fdb_get_default_kvs_config();
780
781    s = fdb_open(&db_file, "./mvcc_test9", &config);
782    TEST_CHK(s == FDB_RESULT_SUCCESS);
783    s = fdb_kvs_open(db_file, &db0, NULL, &kvs_config);
784    TEST_CHK(s == FDB_RESULT_SUCCESS);
785    s = fdb_kvs_open(db_file, &db1, "db1", &kvs_config);
786    TEST_CHK(s == FDB_RESULT_SUCCESS);
787    s = fdb_kvs_open(db_file, &db2, "db2", &kvs_config);
788    TEST_CHK(s == FDB_RESULT_SUCCESS);
789
790    // insert docs in all KV stores
791    for (i=0;i<n;++i){
792        idx = i;
793        sprintf(key, keystr, idx);
794        memset(value, 'x', value_len);
795        memcpy(value + value_len - 6, "<end>", 6);
796        sprintf(value, valuestr, idx);
797        s = fdb_set_kv(db0, key, strlen(key)+1, value, value_len);
798        TEST_CHK(s == FDB_RESULT_SUCCESS);
799        s = fdb_set_kv(db1, key, strlen(key)+1, value, value_len);
800        TEST_CHK(s == FDB_RESULT_SUCCESS);
801        s = fdb_set_kv(db2, key, strlen(key)+1, value, value_len);
802        TEST_CHK(s == FDB_RESULT_SUCCESS);
803    }
804    // try to open snapshot before commit
805    s = fdb_get_kvs_info(db0, &info);
806    TEST_CHK(s == FDB_RESULT_SUCCESS);
807    s = fdb_snapshot_open(db0, &snap, info.last_seqnum);
808    TEST_CHK(s != FDB_RESULT_SUCCESS);
809
810    s = fdb_get_kvs_info(db1, &info);
811    TEST_CHK(s == FDB_RESULT_SUCCESS);
812    s = fdb_snapshot_open(db1, &snap, info.last_seqnum);
813    TEST_CHK(s != FDB_RESULT_SUCCESS);
814
815    s = fdb_get_kvs_info(db2, &info);
816    TEST_CHK(s == FDB_RESULT_SUCCESS);
817    s = fdb_snapshot_open(db2, &snap, info.last_seqnum);
818    TEST_CHK(s != FDB_RESULT_SUCCESS);
819
820    s = fdb_commit(db_file, FDB_COMMIT_NORMAL);
821    TEST_CHK(s == FDB_RESULT_SUCCESS);
822
823    s = fdb_get_kvs_info(db1, &info);
824    TEST_CHK(s == FDB_RESULT_SUCCESS);
825    seqnum = info.last_seqnum;
826
827    s = fdb_get_kvs_info(db2, &info);
828    TEST_CHK(s == FDB_RESULT_SUCCESS);
829    TEST_CHK(seqnum == info.last_seqnum);
830
831    s = fdb_get_kvs_info(db0, &info);
832    TEST_CHK(s == FDB_RESULT_SUCCESS);
833    TEST_CHK(seqnum == info.last_seqnum);
834
835    // now insert docs into default and db2 only, without commit
836    for (i=0;i<n;++i){
837        idx = i;
838        sprintf(key, keystr, idx);
839        memset(value, 'x', value_len);
840        memcpy(value + value_len - 6, "<end>", 6);
841        sprintf(value, valuestr, idx);
842        s = fdb_set_kv(db0, key, strlen(key)+1, value, value_len);
843        TEST_CHK(s == FDB_RESULT_SUCCESS);
844        s = fdb_set_kv(db2, key, strlen(key)+1, value, value_len);
845        TEST_CHK(s == FDB_RESULT_SUCCESS);
846    }
847
848    // open snapshot on db1
849    s = fdb_get_kvs_info(db1, &info);
850    TEST_CHK(s == FDB_RESULT_SUCCESS);
851    // latest (comitted) seqnum
852    s = fdb_snapshot_open(db1, &snap, info.last_seqnum);
853    TEST_CHK(s == FDB_RESULT_SUCCESS);
854    s = fdb_kvs_close(snap);
855    TEST_CHK(s == FDB_RESULT_SUCCESS);
856
857    // open snapshot on db2
858    s = fdb_get_kvs_info(db2, &info);
859    TEST_CHK(s == FDB_RESULT_SUCCESS);
860
861    // latest (uncomitted) seqnum
862    s = fdb_snapshot_open(db2, &snap, info.last_seqnum);
863    TEST_CHK(s != FDB_RESULT_SUCCESS);
864    // committed seqnum
865    s = fdb_snapshot_open(db2, &snap, seqnum);
866    TEST_CHK(s == FDB_RESULT_SUCCESS);
867    s = fdb_kvs_close(snap);
868    TEST_CHK(s == FDB_RESULT_SUCCESS);
869
870    // open snapshot on default KVS
871    s = fdb_get_kvs_info(db0, &info);
872    TEST_CHK(s == FDB_RESULT_SUCCESS);
873
874    // latest (uncomitted) seqnum
875    s = fdb_snapshot_open(db0, &snap, info.last_seqnum);
876    TEST_CHK(s != FDB_RESULT_SUCCESS);
877    // committed seqnum
878    s = fdb_snapshot_open(db0, &snap, seqnum);
879    TEST_CHK(s == FDB_RESULT_SUCCESS);
880    s = fdb_kvs_close(snap);
881    TEST_CHK(s == FDB_RESULT_SUCCESS);
882
883    s = fdb_close(db_file);
884    TEST_CHK(s == FDB_RESULT_SUCCESS);
885    s = fdb_shutdown();
886    TEST_CHK(s == FDB_RESULT_SUCCESS);
887    free(value);
888    memleak_end();
889
890    TEST_RESULT("snapshot with uncomitted data in other KVS test");
891}
892
893void in_memory_snapshot_test()
894{
895    TEST_INIT();
896
897    memleak_start();
898
899    int i, r;
900    int n = 20;
901    int count;
902    fdb_file_handle *dbfile;
903    fdb_kvs_handle *db;
904    fdb_kvs_handle *snap_db;
905    fdb_seqnum_t snap_seq;
906    fdb_doc **doc = alca(fdb_doc*, n);
907    fdb_doc *rdoc;
908    fdb_kvs_info kvs_info;
909    fdb_status status;
910    fdb_iterator *iterator;
911
912    char keybuf[256], metabuf[256], bodybuf[256];
913
914    // remove previous mvcc_test files
915    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
916    (void)r;
917
918    fdb_config fconfig = fdb_get_default_config();
919    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
920    fconfig.buffercache_size = 0;
921    fconfig.wal_threshold = 1024;
922    fconfig.flags = FDB_OPEN_FLAG_CREATE;
923    fconfig.compaction_threshold = 0;
924
925    // remove previous mvcc_test files
926    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
927    (void)r;
928
929    // open db
930    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
931    TEST_CHK(status == FDB_RESULT_SUCCESS);
932    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
933    TEST_CHK(status == FDB_RESULT_SUCCESS);
934
935    // Create a snapshot from an empty database file
936    status = fdb_snapshot_open(db, &snap_db, 0);
937    TEST_CHK(status == FDB_RESULT_SUCCESS);
938    // check if snapshot's sequence number is zero.
939    fdb_get_kvs_info(snap_db, &kvs_info);
940    TEST_CHK(kvs_info.last_seqnum == 0);
941    // create an iterator on the snapshot for full range
942    fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
943    // Iterator should not return any items.
944    status = fdb_iterator_get(iterator, &rdoc);
945    TEST_CHK(status == FDB_RESULT_ITERATOR_FAIL);
946    fdb_iterator_close(iterator);
947    fdb_kvs_close(snap_db);
948
949    // ------- Setup test ----------------------------------
950    // insert documents of 0-4
951    for (i=0; i<n/4; i++){
952        sprintf(keybuf, "key%d", i);
953        sprintf(metabuf, "meta%d", i);
954        sprintf(bodybuf, "body%d", i);
955        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
956            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
957        fdb_set(db, doc[i]);
958    }
959
960    // commit with a manual WAL flush (these docs go into HB-trie)
961    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
962
963    // insert documents from 5 - 8
964    for (; i < n/2 - 1; i++){
965        sprintf(keybuf, "key%d", i);
966        sprintf(metabuf, "meta%d", i);
967        sprintf(bodybuf, "body%d", i);
968        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
969            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
970        fdb_set(db, doc[i]);
971    }
972
973    // We want to create:
974    // |WALFlushHDR|Key-Value1|HDR|Key-Value2|SnapshotHDR|Key-Value1|HDR|
975    // Insert doc 9 with a different value to test duplicate elimination..
976    sprintf(keybuf, "key%d", i);
977    sprintf(metabuf, "meta%d", i);
978    sprintf(bodybuf, "Body%d", i);
979    fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
980            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
981    fdb_set(db, doc[i]);
982
983    // commit again without a WAL flush (last doc goes into the AVL tree)
984    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
985
986    // Insert doc 9 now again with expected value..
987    *(char *)doc[i]->body = 'b';
988    fdb_set(db, doc[i]);
989
990    // TAKE SNAPSHOT: pick up sequence number of a commit without a WAL flush
991    snap_seq = doc[i]->seqnum;
992
993    // Creation of a snapshot with a sequence number that was taken WITHOUT a
994    // commit must fail even if it from the latest document that was set...
995    fdb_get_kvs_info(db, &kvs_info);
996    status = fdb_snapshot_open(db, &snap_db, kvs_info.last_seqnum);
997    TEST_CHK(status == FDB_RESULT_NO_DB_INSTANCE);
998
999    // ---------- Snapshot tests begin -----------------------
1000    // Initialize an in-memory snapshot Without a Commit...
1001    // WAL items are not flushed...
1002    status = fdb_snapshot_open(db, &snap_db, FDB_SNAPSHOT_INMEM);
1003    TEST_CHK(status == FDB_RESULT_SUCCESS);
1004
1005    // commit again without a WAL flush (these documents go into the AVL trees)
1006    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1007
1008    // Now re-insert doc 9 as another duplicate (only newer sequence number)
1009    fdb_set(db, doc[i]);
1010
1011    // commit again without a WAL flush (last doc goes into the AVL tree)
1012    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1013
1014    // insert documents from 10-14 into HB-trie
1015    for (++i; i < (n/2 + n/4); i++){
1016        sprintf(keybuf, "key%d", i);
1017        sprintf(metabuf, "meta%d", i);
1018        sprintf(bodybuf, "body%d", i);
1019        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1020                (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1021        fdb_set(db, doc[i]);
1022    }
1023
1024    status = fdb_set_log_callback(db, logCallbackFunc,
1025                                  (void *) "in_memory_snapshot_test");
1026    TEST_CHK(status == FDB_RESULT_SUCCESS);
1027
1028    // check snapshot's sequence number
1029    fdb_get_kvs_info(snap_db, &kvs_info);
1030    TEST_CHK(kvs_info.last_seqnum == snap_seq);
1031
1032    // insert documents from 15 - 19 on file into the WAL
1033    for (; i < n; i++){
1034        sprintf(keybuf, "key%d", i);
1035        sprintf(metabuf, "meta%d", i);
1036        sprintf(bodybuf, "body%d", i);
1037        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1038            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1039        fdb_set(db, doc[i]);
1040    }
1041    // commit without a WAL flush (This WAL must not affect snapshot)
1042    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1043
1044    // create an iterator on the snapshot for full range
1045    fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
1046
1047    // repeat until fail
1048    i=0;
1049    count=0;
1050    rdoc = NULL;
1051    do {
1052        status = fdb_iterator_get(iterator, &rdoc);
1053        TEST_CHK(status == FDB_RESULT_SUCCESS);
1054
1055        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
1056        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
1057        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1058
1059        fdb_doc_free(rdoc);
1060        rdoc = NULL;
1061        i++;
1062        count++;
1063    } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
1064
1065    TEST_CHK(count==n/2); // Only unique items from the first half
1066
1067    fdb_iterator_close(iterator);
1068
1069    // close db handle
1070    fdb_kvs_close(db);
1071    // close snapshot handle
1072    fdb_kvs_close(snap_db);
1073
1074    // close db file
1075    fdb_close(dbfile);
1076
1077    // free all documents
1078    for (i=0;i<n;++i){
1079        fdb_doc_free(doc[i]);
1080    }
1081
1082    // free all resources
1083    fdb_shutdown();
1084
1085    memleak_end();
1086
1087    TEST_RESULT("in-memory snapshot test");
1088}
1089
1090void in_memory_snapshot_cleanup_test()
1091{
1092    TEST_INIT();
1093
1094    memleak_start();
1095
1096    int r;
1097    fdb_file_handle *dbfile;
1098    fdb_kvs_handle *db;
1099    fdb_kvs_handle *snap_db1, *snap_db2, *snap_db3;
1100    fdb_kvs_handle *psnap_db1, *psnap_db2, *psnap_db3;
1101    fdb_doc *doc = NULL;
1102    fdb_doc *rdoc;
1103    fdb_status status;
1104
1105    char keybuf[32], metabuf[32], bodybuf[32];
1106
1107    // remove previous mvcc_test files
1108    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
1109    (void)r;
1110
1111    fdb_config fconfig = fdb_get_default_config();
1112    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
1113    fconfig.wal_threshold = 1024;
1114    fconfig.flags = FDB_OPEN_FLAG_CREATE;
1115    fconfig.compaction_threshold = 0;
1116    fconfig.block_reusing_threshold = 0;
1117
1118    // remove previous mvcc_test files
1119    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
1120    (void)r;
1121
1122    // open db
1123    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
1124    TEST_CHK(status == FDB_RESULT_SUCCESS);
1125    status = fdb_kvs_open_default(dbfile, &db, &kvs_config);
1126    TEST_CHK(status == FDB_RESULT_SUCCESS);
1127
1128    status = fdb_set_log_callback(db, logCallbackFunc,
1129                                  (void *) "in_memory_snapshot_cleanup_test");
1130    TEST_CHK(status == FDB_RESULT_SUCCESS);
1131
1132    sprintf(keybuf, "key");
1133    sprintf(metabuf, "meta");
1134    sprintf(bodybuf, "body");
1135    fdb_doc_create(&doc, (void*)keybuf, strlen(keybuf),
1136            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1137    fdb_set(db, doc);
1138
1139    // commit without a WAL flush
1140    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1141
1142    status = fdb_snapshot_open(db, &snap_db1, FDB_SNAPSHOT_INMEM);
1143    TEST_CHK(status == FDB_RESULT_SUCCESS);
1144
1145    status = fdb_set(db, doc);
1146    TEST_CHK(status == FDB_RESULT_SUCCESS);
1147
1148    // commit without a WAL flush
1149    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1150    TEST_CHK(status == FDB_RESULT_SUCCESS);
1151
1152    status = fdb_snapshot_open(db, &snap_db2, FDB_SNAPSHOT_INMEM);
1153    TEST_CHK(status == FDB_RESULT_SUCCESS);
1154
1155    status = fdb_set(db, doc);
1156    TEST_CHK(status == FDB_RESULT_SUCCESS);
1157
1158    // commit without a WAL flush
1159    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1160    TEST_CHK(status == FDB_RESULT_SUCCESS);
1161
1162    status = fdb_set(db, doc);
1163    TEST_CHK(status == FDB_RESULT_SUCCESS);
1164
1165    // commit without a WAL flush
1166    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1167    TEST_CHK(status == FDB_RESULT_SUCCESS);
1168
1169    fdb_doc_free(doc);
1170
1171    // Update doc with new Key
1172    char newKey[32];
1173    sprintf(newKey, "Key");
1174    fdb_doc_create(&doc, (void*)newKey, strlen(newKey),
1175            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1176    fdb_set(db, doc);
1177
1178    status = fdb_snapshot_open(db, &snap_db3, FDB_SNAPSHOT_INMEM);
1179    TEST_CHK(status == FDB_RESULT_SUCCESS);
1180
1181    fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
1182
1183    status = fdb_get(snap_db1, rdoc);
1184    TEST_CHK(status == FDB_RESULT_SUCCESS);
1185
1186    status = fdb_get(snap_db2, rdoc);
1187    TEST_CHK(status == FDB_RESULT_SUCCESS);
1188
1189    status = fdb_get(snap_db3, rdoc);
1190    TEST_CHK(status == FDB_RESULT_SUCCESS);
1191
1192    fdb_doc_free(rdoc);
1193
1194    status = fdb_snapshot_open(db, &psnap_db1, 1);
1195    TEST_CHK(status == FDB_RESULT_SUCCESS);
1196
1197    status = fdb_snapshot_open(db, &psnap_db2, 2);
1198    TEST_CHK(status == FDB_RESULT_SUCCESS);
1199
1200    status = fdb_snapshot_open(db, &psnap_db3, 3);
1201    TEST_CHK(status == FDB_RESULT_SUCCESS);
1202
1203    // close snapshot handle
1204    fdb_kvs_close(snap_db1);
1205    fdb_kvs_close(snap_db2);
1206    fdb_kvs_close(snap_db3);
1207
1208    // commit without a WAL flush
1209    status = fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1210    TEST_CHK(status == FDB_RESULT_SUCCESS);
1211
1212    fdb_kvs_close(psnap_db1);
1213    fdb_kvs_close(psnap_db2);
1214    fdb_kvs_close(psnap_db3);
1215
1216    // close db handle
1217    fdb_kvs_close(db);
1218
1219    // close db file
1220    fdb_close(dbfile);
1221
1222    fdb_doc_free(doc);
1223
1224    // free all resources
1225    fdb_shutdown();
1226
1227    memleak_end();
1228
1229    TEST_RESULT("in-memory snapshot cleanup test");
1230}
1231
1232void in_memory_snapshot_on_dirty_hbtrie_test()
1233{
1234    TEST_INIT();
1235
1236    int n = 300, value_len=32;
1237    int i, r, idx, c;
1238    char cmd[256];
1239    char key[256], *value;
1240    char keystr[] = "k%05d";
1241    char keystr2[] = "k%06d";
1242    char valuestr[] = "value%08d";
1243    fdb_file_handle *db_file;
1244    fdb_kvs_handle *db, *snap, *snap_clone;
1245    fdb_config config;
1246    fdb_kvs_config kvs_config;
1247    fdb_status s;
1248    fdb_iterator *fit, *fit_normal, *fit_clone;
1249    fdb_doc *doc;
1250
1251    sprintf(cmd, SHELL_DEL " mvcc_test* > errorlog.txt");
1252    r = system(cmd);
1253    (void)r;
1254
1255    memleak_start();
1256
1257    value = (char*)malloc(value_len);
1258
1259    config = fdb_get_default_config();
1260    config.durability_opt = FDB_DRB_ASYNC;
1261    config.seqtree_opt = FDB_SEQTREE_USE;
1262    config.wal_flush_before_commit = true;
1263    config.wal_threshold = n/5;
1264    config.multi_kv_instances = true;
1265    config.buffercache_size = 0;
1266
1267    kvs_config = fdb_get_default_kvs_config();
1268
1269    s = fdb_open(&db_file, "./mvcc_test", &config);
1270    TEST_CHK(s == FDB_RESULT_SUCCESS);
1271    s = fdb_kvs_open(db_file, &db, "db", &kvs_config);
1272    TEST_CHK(s == FDB_RESULT_SUCCESS);
1273
1274    // write a few documents and commit & wal flush
1275    for (i=0;i<n/10;++i){
1276        idx = i;
1277        sprintf(key, keystr, idx);
1278        memset(value, 'x', value_len);
1279        memcpy(value + value_len - 6, "<end>", 6);
1280        sprintf(value, valuestr, idx);
1281        s = fdb_set_kv(db, key, strlen(key)+1, value, value_len);
1282        TEST_CHK(s == FDB_RESULT_SUCCESS);
1283    }
1284    s = fdb_commit(db_file, FDB_COMMIT_MANUAL_WAL_FLUSH);
1285    TEST_CHK(s == FDB_RESULT_SUCCESS);
1286
1287    // create an in-memory snapshot and its clone
1288    s = fdb_snapshot_open(db, &snap, FDB_SNAPSHOT_INMEM);
1289    TEST_CHK(s == FDB_RESULT_SUCCESS);
1290    s = fdb_snapshot_open(snap, &snap_clone, FDB_SNAPSHOT_INMEM);
1291    TEST_CHK(s == FDB_RESULT_SUCCESS);
1292
1293    // write a number of documents in order to make WAL be flushed before commit
1294    for (i=n/10;i<n;++i){
1295        idx = i;
1296        sprintf(key, keystr, idx);
1297        memset(value, 'x', value_len);
1298        memcpy(value + value_len - 6, "<end>", 6);
1299        sprintf(value, valuestr, idx);
1300        s = fdb_set_kv(db, key, strlen(key)+1, value, value_len);
1301        TEST_CHK(s == FDB_RESULT_SUCCESS);
1302    }
1303
1304    void *v_out;
1305    size_t vlen_out;
1306    idx = n/10;
1307    sprintf(key, keystr, idx);
1308    s = fdb_get_kv(snap, key, strlen(key)+1, &v_out, &vlen_out);
1309    // should not be able to retrieve
1310    TEST_CHK(s != FDB_RESULT_SUCCESS);
1311
1312    s = fdb_get_kv(snap_clone, key, strlen(key)+1, &v_out, &vlen_out);
1313    // should not be able to retrieve in also clone
1314    TEST_CHK(s != FDB_RESULT_SUCCESS);
1315
1316    // close snapshot and its clone
1317    s = fdb_kvs_close(snap);
1318    TEST_CHK(s == FDB_RESULT_SUCCESS);
1319    s = fdb_kvs_close(snap_clone);
1320    TEST_CHK(s == FDB_RESULT_SUCCESS);
1321
1322    // create an in-memory snapshot
1323    s = fdb_snapshot_open(db, &snap, FDB_SNAPSHOT_INMEM);
1324    TEST_CHK(s == FDB_RESULT_SUCCESS);
1325
1326    s = fdb_iterator_init(snap, &fit, NULL, 0, NULL, 0, 0x0);
1327    TEST_CHK(s == FDB_RESULT_SUCCESS);
1328    c = 0;
1329    do {
1330        doc = NULL;
1331        s = fdb_iterator_get(fit, &doc);
1332        if (s != FDB_RESULT_SUCCESS) break;
1333
1334        idx = c;
1335        sprintf(key, keystr, idx);
1336        memset(value, 'x', value_len);
1337        memcpy(value + value_len - 6, "<end>", 6);
1338        sprintf(value, valuestr, idx);
1339        TEST_CMP(doc->key, key, doc->keylen);
1340        TEST_CMP(doc->body, value, doc->bodylen);
1341        c++;
1342        fdb_doc_free(doc);
1343    } while(fdb_iterator_next(fit) == FDB_RESULT_SUCCESS);
1344    TEST_CHK(c == n);
1345
1346    s = fdb_iterator_close(fit);
1347    TEST_CHK(s == FDB_RESULT_SUCCESS);
1348
1349    // create a clone
1350    s = fdb_snapshot_open(snap, &snap_clone, FDB_SNAPSHOT_INMEM);
1351    TEST_CHK(s == FDB_RESULT_SUCCESS);
1352
1353    // create iterators on snapshot and its clone
1354    s = fdb_iterator_init(snap, &fit, NULL, 0, NULL, 0, 0x0);
1355    TEST_CHK(s == FDB_RESULT_SUCCESS);
1356    s = fdb_iterator_init(snap_clone, &fit_clone, NULL, 0, NULL, 0, 0x0);
1357    TEST_CHK(s == FDB_RESULT_SUCCESS);
1358
1359    // also create an iterator on a normal handle
1360    s = fdb_iterator_init(db, &fit_normal, NULL, 0, NULL, 0, 0x0);
1361    TEST_CHK(s == FDB_RESULT_SUCCESS);
1362
1363    c = 0;
1364    do {
1365        doc = NULL;
1366        s = fdb_iterator_get(fit, &doc);
1367        if (s != FDB_RESULT_SUCCESS) break;
1368
1369        idx = c;
1370        sprintf(key, keystr, idx);
1371        memset(value, 'x', value_len);
1372        memcpy(value + value_len - 6, "<end>", 6);
1373        sprintf(value, valuestr, idx);
1374        TEST_CMP(doc->key, key, doc->keylen);
1375        TEST_CMP(doc->body, value, doc->bodylen);
1376        c++;
1377        fdb_doc_free(doc);
1378
1379        doc = NULL;
1380        s = fdb_iterator_get(fit, &doc);
1381        TEST_CHK(s == FDB_RESULT_SUCCESS);
1382        TEST_CMP(doc->key, key, doc->keylen);
1383        TEST_CMP(doc->body, value, doc->bodylen);
1384        fdb_doc_free(doc);
1385
1386        if (c == n/5) {
1387            // insert new docs in the middle of iteration
1388            for (i=0; i<n*10; ++i){
1389                idx = i;
1390                sprintf(key, keystr2, idx);
1391                memset(value, 'x', value_len);
1392                memcpy(value + value_len - 6, "<end>", 6);
1393                sprintf(value, valuestr, idx);
1394                s = fdb_set_kv(db, key, strlen(key)+1, value, value_len);
1395                TEST_CHK(s == FDB_RESULT_SUCCESS);
1396            }
1397        }
1398
1399        s = fdb_iterator_next(fit);
1400        // result should be same to clone
1401        if (s != FDB_RESULT_SUCCESS) {
1402            s = fdb_iterator_next(fit_clone);
1403            TEST_CHK(s != FDB_RESULT_SUCCESS);
1404        } else {
1405            s = fdb_iterator_next(fit_clone);
1406            TEST_CHK(s == FDB_RESULT_SUCCESS);
1407        }
1408    } while(s == FDB_RESULT_SUCCESS);
1409    TEST_CHK(c == n);
1410
1411    // close iterators
1412    s = fdb_iterator_close(fit);
1413    TEST_CHK(s == FDB_RESULT_SUCCESS);
1414    s = fdb_iterator_close(fit_clone);
1415    TEST_CHK(s == FDB_RESULT_SUCCESS);
1416
1417    // the results should be same in the normal iterator
1418    c = 0;
1419    do {
1420        doc = NULL;
1421        s = fdb_iterator_get(fit_normal, &doc);
1422        if (s != FDB_RESULT_SUCCESS) break;
1423
1424        idx = c;
1425        sprintf(key, keystr, idx);
1426        memset(value, 'x', value_len);
1427        memcpy(value + value_len - 6, "<end>", 6);
1428        sprintf(value, valuestr, idx);
1429        TEST_CMP(doc->key, key, doc->keylen);
1430        TEST_CMP(doc->body, value, doc->bodylen);
1431        c++;
1432        fdb_doc_free(doc);
1433    } while(fdb_iterator_next(fit_normal) == FDB_RESULT_SUCCESS);
1434    TEST_CHK(c == n);
1435
1436    s = fdb_iterator_close(fit_normal);
1437    TEST_CHK(s == FDB_RESULT_SUCCESS);
1438
1439    s = fdb_kvs_close(snap);
1440    TEST_CHK(s == FDB_RESULT_SUCCESS);
1441
1442    s = fdb_kvs_close(snap_clone);
1443    TEST_CHK(s == FDB_RESULT_SUCCESS);
1444
1445    s = fdb_close(db_file);
1446    TEST_CHK(s == FDB_RESULT_SUCCESS);
1447    s = fdb_shutdown();
1448    TEST_CHK(s == FDB_RESULT_SUCCESS);
1449    free(value);
1450
1451    memleak_end();
1452
1453    TEST_RESULT("in-memory snapshot on dirty HB+trie nodes test");
1454}
1455
1456
1457struct cb_inmem_snap_args {
1458    fdb_kvs_handle *handle;
1459    int n;
1460    int move_count;
1461    int value_len;
1462    char *keystr;
1463    char *valuestr;
1464};
1465
1466static fdb_compact_decision cb_inmem_snap(fdb_file_handle *fhandle,
1467                            fdb_compaction_status status,
1468                            const char *kv_name,
1469                            fdb_doc *doc_in, uint64_t old_offset,
1470                            uint64_t new_offset,
1471                            void *ctx)
1472{
1473    TEST_INIT();
1474    int c, idx;
1475    char key[256], value[256];
1476    void *value_out;
1477    size_t valuelen;
1478    fdb_kvs_handle *snap;
1479    fdb_kvs_info info;
1480    fdb_iterator *fit;
1481    fdb_doc *doc;
1482    fdb_status s;
1483    fdb_compact_decision ret = FDB_CS_KEEP_DOC;
1484    struct cb_inmem_snap_args *args = (struct cb_inmem_snap_args *)ctx;
1485    (void)args;
1486
1487    if (status == FDB_CS_MOVE_DOC) {
1488        TEST_CHK(kv_name);
1489        if (doc_in->deleted) {
1490            ret = FDB_CS_DROP_DOC;
1491        }
1492        args->move_count++;
1493        if (args->move_count == 2) {
1494            // open in-memory snapshot
1495            s = fdb_snapshot_open(args->handle, &snap, FDB_SNAPSHOT_INMEM);
1496            TEST_CHK(s == FDB_RESULT_SUCCESS);
1497
1498            s = fdb_get_kvs_info(snap, &info);
1499            TEST_CHK(s == FDB_RESULT_SUCCESS);
1500            TEST_CHK(info.last_seqnum == (fdb_seqnum_t)args->n);
1501
1502            s = fdb_iterator_init(snap, &fit, NULL, 0, NULL, 0, 0x0);
1503            TEST_CHK(s == FDB_RESULT_SUCCESS);
1504
1505            c = 0;
1506            do {
1507                doc = NULL;
1508                s = fdb_iterator_get(fit, &doc);
1509                if (s != FDB_RESULT_SUCCESS) {
1510                    break;
1511                }
1512                idx = c;
1513                sprintf(key, args->keystr, idx);
1514                memset(value, 'x', args->value_len);
1515                memcpy(value + args->value_len - 6, "<end>", 6);
1516                sprintf(value, args->valuestr, idx);
1517                TEST_CMP(doc->key, key, doc->keylen);
1518                TEST_CMP(doc->body, value, doc->bodylen);
1519
1520                c++;
1521                fdb_doc_free(doc);
1522            } while (fdb_iterator_next(fit) == FDB_RESULT_SUCCESS);
1523            TEST_CHK(c == args->n);
1524
1525            s = fdb_iterator_close(fit);
1526            TEST_CHK(s == FDB_RESULT_SUCCESS);
1527
1528            fdb_kvs_close(snap);
1529
1530        } else if (args->move_count == 5) {
1531            // insert new doc
1532            sprintf(key, "new_key");
1533            sprintf(value, "new_value");
1534            s = fdb_set_kv(args->handle, (void*)key, strlen(key)+1, (void*)value, strlen(value)+1);
1535            TEST_CHK(s == FDB_RESULT_SUCCESS);
1536
1537            // open in-memory snapshot
1538            s = fdb_snapshot_open(args->handle, &snap, FDB_SNAPSHOT_INMEM);
1539            TEST_CHK(s == FDB_RESULT_SUCCESS);
1540
1541            s = fdb_get_kvs_info(snap, &info);
1542            TEST_CHK(s == FDB_RESULT_SUCCESS);
1543            TEST_CHK(info.last_seqnum == (fdb_seqnum_t)args->n+1);
1544
1545            s = fdb_iterator_init(snap, &fit, NULL, 0, NULL, 0, 0x0);
1546            TEST_CHK(s == FDB_RESULT_SUCCESS);
1547
1548            c = 0;
1549            do {
1550                doc = NULL;
1551                s = fdb_iterator_get(fit, &doc);
1552                if (s != FDB_RESULT_SUCCESS) {
1553                    break;
1554                }
1555                if (c < args->n) {
1556                    idx = c;
1557                    sprintf(key, args->keystr, idx);
1558                    memset(value, 'x', args->value_len);
1559                    memcpy(value + args->value_len - 6, "<end>", 6);
1560                    sprintf(value, args->valuestr, idx);
1561                    TEST_CMP(doc->key, key, doc->keylen);
1562                    TEST_CMP(doc->body, value, doc->bodylen);
1563                } else {
1564                    // new document
1565                    sprintf(key, "new_key");
1566                    sprintf(value, "new_value");
1567                    TEST_CMP(doc->key, key, doc->keylen);
1568                    TEST_CMP(doc->body, value, doc->bodylen);
1569                }
1570
1571                c++;
1572                fdb_doc_free(doc);
1573            } while (fdb_iterator_next(fit) == FDB_RESULT_SUCCESS);
1574            TEST_CHK(c == args->n + 1);
1575
1576            s = fdb_iterator_close(fit);
1577            TEST_CHK(s == FDB_RESULT_SUCCESS);
1578
1579            s = fdb_get_kv(snap, (void*)key, strlen(key)+1, &value_out, &valuelen);
1580            TEST_CHK(s == FDB_RESULT_SUCCESS);
1581            TEST_CMP(value_out, value, valuelen);
1582            fdb_free_block(value_out);
1583
1584            fdb_kvs_close(snap);
1585        }
1586    }
1587    return ret;
1588}
1589
1590void in_memory_snapshot_compaction_test()
1591{
1592    TEST_INIT();
1593
1594    int n = 10, value_len=32;
1595    int i, r, c, idx;
1596    char cmd[256];
1597    char key[256], *value;
1598    char keystr[] = "k%05d";
1599    char valuestr[] = "value%08d";
1600    void *value_out;
1601    size_t valuelen;
1602    fdb_file_handle *db_file;
1603    fdb_kvs_handle *db, *db2, *snap;
1604    fdb_config config;
1605    fdb_kvs_config kvs_config;
1606    fdb_kvs_info info;
1607    fdb_iterator *fit;
1608    fdb_doc *doc;
1609    fdb_status s;
1610    struct cb_inmem_snap_args cargs;
1611
1612    sprintf(cmd, SHELL_DEL " mvcc_test* > errorlog.txt");
1613    r = system(cmd);
1614    (void)r;
1615
1616    memleak_start();
1617
1618    value = (char*)malloc(value_len);
1619
1620    config = fdb_get_default_config();
1621    config.durability_opt = FDB_DRB_ASYNC;
1622    config.seqtree_opt = FDB_SEQTREE_USE;
1623    config.wal_flush_before_commit = true;
1624    config.wal_threshold = n/5;
1625    config.multi_kv_instances = true;
1626    config.buffercache_size = 0;
1627    config.compaction_cb = cb_inmem_snap;
1628    config.compaction_cb_mask = FDB_CS_MOVE_DOC;
1629    config.compaction_cb_ctx = &cargs;
1630
1631    kvs_config = fdb_get_default_kvs_config();
1632
1633    s = fdb_open(&db_file, "./mvcc_test", &config);
1634    TEST_CHK(s == FDB_RESULT_SUCCESS);
1635    s = fdb_kvs_open(db_file, &db, "db", &kvs_config);
1636    TEST_CHK(s == FDB_RESULT_SUCCESS);
1637
1638    s = fdb_kvs_open(db_file, &db2, "db", &kvs_config);
1639    TEST_CHK(s == FDB_RESULT_SUCCESS);
1640
1641    cargs.handle = db2;
1642    cargs.move_count = 0;
1643    cargs.n = n;
1644    cargs.keystr = keystr;
1645    cargs.valuestr = valuestr;
1646    cargs.value_len = value_len;
1647
1648    // write
1649    for (i=0;i<n;++i){
1650        idx = i;
1651        sprintf(key, keystr, idx);
1652        memset(value, 'x', value_len);
1653        memcpy(value + value_len - 6, "<end>", 6);
1654        sprintf(value, valuestr, idx);
1655        s = fdb_set_kv(db, key, strlen(key)+1, value, value_len);
1656        TEST_CHK(s == FDB_RESULT_SUCCESS);
1657    }
1658    s = fdb_commit(db_file, FDB_COMMIT_MANUAL_WAL_FLUSH);
1659    TEST_CHK(s == FDB_RESULT_SUCCESS);
1660
1661    s = fdb_compact(db_file, "./mvcc_test2");
1662    TEST_CHK(s == FDB_RESULT_SUCCESS);
1663
1664    // open in-memory snapshot
1665    s = fdb_snapshot_open(db, &snap, FDB_SNAPSHOT_INMEM);
1666    TEST_CHK(s == FDB_RESULT_SUCCESS);
1667
1668    s = fdb_get_kvs_info(snap, &info);
1669    TEST_CHK(s == FDB_RESULT_SUCCESS);
1670    TEST_CHK(info.last_seqnum == (fdb_seqnum_t)n+1);
1671
1672    s = fdb_iterator_init(snap, &fit, NULL, 0, NULL, 0, 0x0);
1673    TEST_CHK(s == FDB_RESULT_SUCCESS);
1674
1675    c = 0;
1676    do {
1677        doc = NULL;
1678        s = fdb_iterator_get(fit, &doc);
1679        if (s != FDB_RESULT_SUCCESS) {
1680            break;
1681        }
1682        if (c < n) {
1683            idx = c;
1684            sprintf(key, keystr, idx);
1685            memset(value, 'x', value_len);
1686            memcpy(value + value_len - 6, "<end>", 6);
1687            sprintf(value, valuestr, idx);
1688            TEST_CMP(doc->key, key, doc->keylen);
1689            TEST_CMP(doc->body, value, doc->bodylen);
1690        } else {
1691            // new document
1692            sprintf(key, "new_key");
1693            sprintf(value, "new_value");
1694            TEST_CMP(doc->key, key, doc->keylen);
1695            TEST_CMP(doc->body, value, doc->bodylen);
1696        }
1697
1698        c++;
1699        fdb_doc_free(doc);
1700    } while (fdb_iterator_next(fit) == FDB_RESULT_SUCCESS);
1701    TEST_CHK(c == n + 1);
1702
1703    s = fdb_iterator_close(fit);
1704    TEST_CHK(s == FDB_RESULT_SUCCESS);
1705
1706    s = fdb_get_kv(snap, (void*)key, strlen(key)+1, &value_out, &valuelen);
1707    TEST_CHK(s == FDB_RESULT_SUCCESS);
1708    TEST_CMP(value_out, value, valuelen);
1709    fdb_free_block(value_out);
1710
1711    fdb_kvs_close(snap);
1712
1713    s = fdb_close(db_file);
1714    TEST_CHK(s == FDB_RESULT_SUCCESS);
1715    s = fdb_shutdown();
1716    TEST_CHK(s == FDB_RESULT_SUCCESS);
1717    free(value);
1718
1719    memleak_end();
1720
1721    TEST_RESULT("in-memory snapshot with concurrent compaction test");
1722}
1723
1724void snapshot_clone_test()
1725{
1726    TEST_INIT();
1727
1728    memleak_start();
1729
1730    int i, r;
1731    int n = 20;
1732    int count;
1733    fdb_file_handle *dbfile;
1734    fdb_kvs_handle *db;
1735    fdb_kvs_handle *snap_db, *snap_inmem;
1736    fdb_kvs_handle *snap_db2, *snap_inmem2; // clones from stable & inmemory
1737    fdb_seqnum_t snap_seq;
1738    fdb_doc **doc = alca(fdb_doc*, n);
1739    fdb_doc *rdoc;
1740    fdb_kvs_info kvs_info;
1741    fdb_status status;
1742    fdb_iterator *iterator;
1743
1744    char keybuf[256], metabuf[256], bodybuf[256];
1745
1746    // remove previous mvcc_test files
1747    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
1748    (void)r;
1749
1750    fdb_config fconfig = fdb_get_default_config();
1751    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
1752    fconfig.buffercache_size = 0;
1753    fconfig.wal_threshold = 1024;
1754    fconfig.seqtree_opt = FDB_SEQTREE_USE; // enable seqtree since get_byseq
1755    fconfig.flags = FDB_OPEN_FLAG_CREATE;
1756    fconfig.compaction_threshold = 0;
1757
1758    // remove previous mvcc_test files
1759    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
1760    (void)r;
1761
1762    // open db
1763    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
1764    fdb_kvs_open_default(dbfile, &db, &kvs_config);
1765    TEST_CHK(status == FDB_RESULT_SUCCESS);
1766
1767    // Create a snapshot from an empty database file
1768    status = fdb_snapshot_open(db, &snap_db, 0);
1769    TEST_CHK(status == FDB_RESULT_SUCCESS);
1770    // check if snapshot's sequence number is zero.
1771    fdb_get_kvs_info(snap_db, &kvs_info);
1772    TEST_CHK(kvs_info.last_seqnum == 0);
1773    // create an iterator on the snapshot for full range
1774    fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
1775    // Iterator should not return any items.
1776    status = fdb_iterator_next(iterator);
1777    TEST_CHK(status == FDB_RESULT_ITERATOR_FAIL);
1778    fdb_iterator_close(iterator);
1779    fdb_kvs_close(snap_db);
1780
1781    // ------- Setup test ----------------------------------
1782    // insert documents of 0-4
1783    for (i=0; i<n/4; i++){
1784        sprintf(keybuf, "key%d", i);
1785        sprintf(metabuf, "meta%d", i);
1786        sprintf(bodybuf, "body%d", i);
1787        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1788            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1789        fdb_set(db, doc[i]);
1790    }
1791
1792    // commit with a manual WAL flush (these docs go into HB-trie)
1793    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1794
1795    // insert documents from 4 - 8
1796    for (; i < n/2 - 1; i++){
1797        sprintf(keybuf, "key%d", i);
1798        sprintf(metabuf, "meta%d", i);
1799        sprintf(bodybuf, "body%d", i);
1800        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1801            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1802        fdb_set(db, doc[i]);
1803    }
1804
1805    // We want to create:
1806    // |WALFlushHDR|Key-Value1|HDR|Key-Value2|SnapshotHDR|Key-Value1|HDR|
1807    // Insert doc 9 with a different value to test duplicate elimination..
1808    sprintf(keybuf, "key%d", i);
1809    sprintf(metabuf, "meta%d", i);
1810    sprintf(bodybuf, "Body%d", i);
1811    fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1812            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1813    fdb_set(db, doc[i]);
1814
1815    // commit again without a WAL flush (last doc goes into the AVL tree)
1816    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1817
1818    // Insert doc 9 now again with expected value..
1819    *(char *)doc[i]->body = 'b';
1820    fdb_set(db, doc[i]);
1821    // commit again without a WAL flush (these documents go into the AVL trees)
1822    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1823
1824    // TAKE SNAPSHOT: pick up sequence number of a commit without a WAL flush
1825    snap_seq = doc[i]->seqnum;
1826
1827    // Initialize an in-memory snapshot Without a Commit...
1828    // WAL items are not flushed...
1829    status = fdb_snapshot_open(db, &snap_inmem, FDB_SNAPSHOT_INMEM);
1830    TEST_CHK(status == FDB_RESULT_SUCCESS);
1831
1832    // Now re-insert doc 9 as another duplicate (only newer sequence number)
1833    fdb_set(db, doc[i]);
1834    // commit again without a WAL flush (last doc goes into the AVL tree)
1835    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1836
1837    // insert documents from 10-14 into HB-trie
1838    for (++i; i < (n/2 + n/4); i++){
1839        sprintf(keybuf, "key%d", i);
1840        sprintf(metabuf, "meta%d", i);
1841        sprintf(bodybuf, "body%d", i);
1842        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1843            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1844        fdb_set(db, doc[i]);
1845    }
1846    // manually flush WAL & commit
1847    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
1848
1849    status = fdb_set_log_callback(db, logCallbackFunc,
1850                                  (void *) "snapshot_clone_test");
1851    TEST_CHK(status == FDB_RESULT_SUCCESS);
1852    // ---------- Snapshot tests begin -----------------------
1853    // Attempt to take snapshot with out-of-range marker..
1854    status = fdb_snapshot_open(db, &snap_db, 999999);
1855    TEST_CHK(status == FDB_RESULT_NO_DB_INSTANCE);
1856
1857    // Init Snapshot of open file with saved document seqnum as marker
1858    status = fdb_snapshot_open(db, &snap_db, snap_seq);
1859    TEST_CHK(status == FDB_RESULT_SUCCESS);
1860
1861    // Clone a snapshot into another snapshot...
1862    status = fdb_snapshot_open(snap_db, &snap_db2, snap_seq);
1863    TEST_CHK(status == FDB_RESULT_SUCCESS);
1864
1865    // close snapshot handle
1866    status = fdb_kvs_close(snap_db);
1867    TEST_CHK(status == FDB_RESULT_SUCCESS);
1868
1869    // check snapshot's sequence number
1870    fdb_get_kvs_info(snap_db2, &kvs_info);
1871    TEST_CHK(kvs_info.last_seqnum == snap_seq);
1872
1873    // insert documents from 15 - 19 on file into the WAL
1874    for (; i < n; i++){
1875        sprintf(keybuf, "key%d", i);
1876        sprintf(metabuf, "meta%d", i);
1877        sprintf(bodybuf, "body%d", i);
1878        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
1879            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
1880        fdb_set(db, doc[i]);
1881    }
1882    // commit without a WAL flush (This WAL must not affect snapshot)
1883    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
1884
1885    // Clone the in-memory snapshot into another snapshot
1886    status = fdb_snapshot_open(snap_inmem, &snap_inmem2, FDB_SNAPSHOT_INMEM);
1887    TEST_CHK(status == FDB_RESULT_SUCCESS);
1888
1889    // Stable Snapshot Scan Tests..........
1890    // Retrieve metaonly by key from snapshot
1891    i = snap_seq + 1;
1892    fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
1893    status = fdb_get_metaonly(snap_db2, rdoc);
1894    TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
1895    fdb_doc_free(rdoc);
1896
1897    i = 5;
1898    fdb_doc_create(&rdoc, doc[i]->key, doc[i]->keylen, NULL, 0, NULL, 0);
1899    status = fdb_get_metaonly(snap_db2, rdoc);
1900    TEST_CHK(status == FDB_RESULT_SUCCESS);
1901    TEST_CMP(rdoc->key, doc[i]->key, doc[i]->keylen);
1902    TEST_CMP(rdoc->meta, doc[i]->meta, doc[i]->metalen);
1903    fdb_doc_free(rdoc);
1904
1905    // Retrieve by seq from snapshot
1906    fdb_doc_create(&rdoc, NULL, 0, NULL, 0, NULL, 0);
1907    rdoc->seqnum = 6;
1908    status = fdb_get_byseq(snap_db2, rdoc);
1909    TEST_CHK(status == FDB_RESULT_SUCCESS);
1910    TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
1911    TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1912    fdb_doc_free(rdoc);
1913    rdoc = NULL;
1914
1915    // create an iterator on the snapshot for full range
1916    fdb_iterator_init(snap_db2, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
1917
1918    // repeat until fail
1919    i=0;
1920    count=0;
1921    do {
1922        status = fdb_iterator_get(iterator, &rdoc);
1923        TEST_CHK(status == FDB_RESULT_SUCCESS);
1924        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
1925        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
1926        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1927
1928        fdb_doc_free(rdoc);
1929        rdoc = NULL;
1930        i ++;
1931        count++;
1932    } while (fdb_iterator_next(iterator) == FDB_RESULT_SUCCESS);
1933
1934    TEST_CHK(count==n/2); // Only unique items from the first half
1935
1936    fdb_iterator_close(iterator);
1937
1938    // create an iterator on the in-memory snapshot clone for full range
1939    fdb_iterator_init(snap_inmem2, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
1940
1941    // repeat until fail
1942    i=0;
1943    count=0;
1944    do {
1945        status = fdb_iterator_get(iterator, &rdoc);
1946        TEST_CHK(status == FDB_RESULT_SUCCESS);
1947        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
1948        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
1949        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
1950
1951        fdb_doc_free(rdoc);
1952        rdoc = NULL;
1953        i ++;
1954        count++;
1955    } while(fdb_iterator_next(iterator) == FDB_RESULT_SUCCESS);
1956
1957    TEST_CHK(count==n/2); // Only unique items from the first half
1958
1959    fdb_iterator_close(iterator);
1960
1961    // close db handle
1962    fdb_kvs_close(db);
1963    // close the in-memory snapshot handle
1964    fdb_kvs_close(snap_inmem);
1965    // close the in-memory clone snapshot handle
1966    fdb_kvs_close(snap_inmem2);
1967    // close the clone snapshot handle
1968    fdb_kvs_close(snap_db2);
1969    // close db file
1970    fdb_close(dbfile);
1971
1972    // free all documents
1973    for (i=0;i<n;++i){
1974        fdb_doc_free(doc[i]);
1975    }
1976
1977    // free all resources
1978    fdb_shutdown();
1979
1980    memleak_end();
1981
1982    TEST_RESULT("snapshot clone test");
1983}
1984
1985struct parallel_clone_t {
1986    fdb_doc **doc;
1987    fdb_kvs_handle *snap_db;
1988    int num_docs;
1989};
1990
1991void *snap_clone_thread(void *args)
1992{
1993    struct parallel_clone_t *t = (struct parallel_clone_t *)args;
1994    fdb_kvs_handle *clone_db;
1995    fdb_iterator *iterator;
1996    fdb_doc *rdoc = NULL;
1997    fdb_doc **doc = t->doc;
1998    int i;
1999    int n = t->num_docs;
2000    fdb_status status;
2001    int count;
2002    TEST_INIT();
2003
2004    status = fdb_snapshot_open(t->snap_db, &clone_db, FDB_SNAPSHOT_INMEM);
2005    TEST_CHK(status == FDB_RESULT_SUCCESS);
2006
2007    // create an iterator on the in-memory snapshot clone for full range
2008    fdb_iterator_init(clone_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
2009
2010    // repeat until fail
2011    i=0;
2012    count=0;
2013    do {
2014        status = fdb_iterator_get(iterator, &rdoc);
2015        TEST_CHK(status == FDB_RESULT_SUCCESS);
2016        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
2017        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
2018
2019        fdb_doc_free(rdoc);
2020        rdoc = NULL;
2021        i ++;
2022        count++;
2023    } while(fdb_iterator_next(iterator) == FDB_RESULT_SUCCESS);
2024
2025    TEST_CHK(count==n/2); // Only unique items from the first half
2026
2027    fdb_iterator_close(iterator);
2028
2029    fdb_kvs_close(clone_db);
2030    thread_exit(0);
2031    return NULL;
2032}
2033
2034void snapshot_parallel_clone_test()
2035{
2036    TEST_INIT();
2037
2038    memleak_start();
2039
2040    int i, r;
2041    int num_cloners = 30; // parallel snapshot clone operations
2042    int n = 20480; // 10 dirty wal flushes at least
2043    fdb_file_handle *dbfile;
2044    fdb_kvs_handle *db;
2045    fdb_kvs_handle *snap_inmem;
2046    fdb_doc **doc = alca(fdb_doc*, n);
2047    thread_t *tid = alca(thread_t, num_cloners);
2048    void *thread_ret;
2049    fdb_status status;
2050    struct parallel_clone_t clone_data;
2051
2052    char keybuf[256], bodybuf[256];
2053
2054    // remove previous mvcc_test files
2055    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2056    (void)r;
2057
2058    fdb_config fconfig = fdb_get_default_config();
2059    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2060    fconfig.buffercache_size = 0;
2061    fconfig.wal_threshold = 1024;
2062    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2063    fconfig.compaction_threshold = 0;
2064
2065    // remove previous mvcc_test files
2066    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2067    (void)r;
2068
2069    // open db
2070    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2071    fdb_kvs_open_default(dbfile, &db, &kvs_config);
2072    TEST_CHK(status == FDB_RESULT_SUCCESS);
2073
2074    status = fdb_set_log_callback(db, logCallbackFunc,
2075                                  (void *) "snapshot_parallel_clone_test");
2076    TEST_CHK(status == FDB_RESULT_SUCCESS);
2077    // ------- Setup test ----------------------------------
2078    for (i=0; i<n/2; i++){
2079        sprintf(keybuf, "key%5d", i);
2080        sprintf(bodybuf, "body%d", i);
2081        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2082            NULL, 0, (void*)bodybuf, strlen(bodybuf));
2083        fdb_set(db, doc[i]);
2084    }
2085
2086    // Initialize an in-memory snapshot Without a Commit...
2087    // WAL items are not flushed...
2088    status = fdb_snapshot_open(db, &snap_inmem, FDB_SNAPSHOT_INMEM);
2089    TEST_CHK(status == FDB_RESULT_SUCCESS);
2090
2091    clone_data.doc = doc;
2092    clone_data.num_docs = n;
2093    clone_data.snap_db = snap_inmem;
2094
2095    for (int j = num_cloners - 1; j>=0; --j) {
2096        thread_create(&tid[j], snap_clone_thread, &clone_data);
2097    }
2098
2099    for (; i < n; i++){
2100        sprintf(keybuf, "key%5d", i);
2101        sprintf(bodybuf, "BODY%d", i);
2102        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2103            NULL, 0, (void*)bodybuf, strlen(bodybuf));
2104        fdb_set(db, doc[i]);
2105    }
2106    // commit without a WAL flush (This WAL must not affect snapshot)
2107    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2108
2109    for (int j = num_cloners - 1; j>=0; --j) {
2110        thread_join(tid[j], &thread_ret);
2111    }
2112
2113    // close db handle
2114    fdb_kvs_close(db);
2115    // close the in-memory snapshot handle
2116    fdb_kvs_close(snap_inmem);
2117    // close db file
2118    fdb_close(dbfile);
2119
2120    // free all documents
2121    for (i=0;i<n;++i){
2122        fdb_doc_free(doc[i]);
2123    }
2124
2125    // free all resources
2126    fdb_shutdown();
2127
2128    memleak_end();
2129
2130    TEST_RESULT("snapshot parallel clone test");
2131}
2132
2133void snapshot_markers_in_file_test(bool multi_kv)
2134{
2135    TEST_INIT();
2136
2137    memleak_start();
2138
2139    int i, r;
2140    int n = 20;
2141    int num_kvs = 4; // keep this the same as number of fdb_commit() calls
2142    fdb_file_handle *dbfile;
2143    fdb_kvs_handle **db = alca(fdb_kvs_handle *, num_kvs);
2144    fdb_doc **doc = alca(fdb_doc*, n);
2145    fdb_status status;
2146    fdb_snapshot_info_t *markers;
2147    uint64_t num_markers;
2148
2149    char keybuf[256], metabuf[256], bodybuf[256];
2150    char kv_name[8];
2151
2152    fdb_config fconfig = fdb_get_default_config();
2153    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2154    fconfig.buffercache_size = 0;
2155    fconfig.wal_threshold = 1024;
2156    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2157    fconfig.compaction_threshold = 0;
2158    fconfig.multi_kv_instances = multi_kv;
2159    // for creating the strict number of snapshots, disable block reusing
2160    fconfig.block_reusing_threshold = 0;
2161
2162    // remove previous mvcc_test files
2163    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2164    (void)r;
2165
2166    // open db
2167    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2168    if (multi_kv) {
2169        for (r = 0; r < num_kvs; ++r) {
2170            sprintf(kv_name, "kv%d", r);
2171            fdb_kvs_open(dbfile, &db[r], kv_name, &kvs_config);
2172        }
2173    } else {
2174        num_kvs = 1;
2175        fdb_kvs_open_default(dbfile, &db[0], &kvs_config);
2176    }
2177
2178   // ------- Setup test ----------------------------------
2179   // insert documents of 0-4
2180    for (i=0; i<n/4; i++){
2181        sprintf(keybuf, "key%d", i);
2182        sprintf(metabuf, "meta%d", i);
2183        sprintf(bodybuf, "body%d", i);
2184        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2185            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2186        for (r = 0; r < num_kvs; ++r) {
2187            fdb_set(db[r], doc[i]);
2188        }
2189    }
2190
2191    // commit with a manual WAL flush (these docs go into HB-trie)
2192    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2193
2194    // insert documents from 5 - 9
2195    for (; i < n/2; i++){
2196        sprintf(keybuf, "key%d", i);
2197        sprintf(metabuf, "meta%d", i);
2198        sprintf(bodybuf, "body%d", i);
2199        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2200            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2201        for (r = 0; r < num_kvs; ++r) {
2202            fdb_set(db[r], doc[i]);
2203        }
2204    }
2205
2206    // commit again without a WAL flush
2207    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2208
2209    // insert documents from 10-14 into HB-trie
2210    for (; i < (n/2 + n/4); i++){
2211        sprintf(keybuf, "key%d", i);
2212        sprintf(metabuf, "meta%d", i);
2213        sprintf(bodybuf, "body%d", i);
2214        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2215            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2216        for (r = 0; r < num_kvs; ++r) {
2217            fdb_set(db[r], doc[i]);
2218        }
2219    }
2220    // manually flush WAL & commit
2221    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2222
2223    // insert documents from 15 - 19 on file into the WAL
2224    for (; i < n; i++){
2225        sprintf(keybuf, "key%d", i);
2226        sprintf(metabuf, "meta%d", i);
2227        sprintf(bodybuf, "body%d", i);
2228        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2229            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2230        for (r = 0; r < num_kvs; ++r) {
2231            fdb_set(db[r], doc[i]);
2232        }
2233    }
2234    // commit without a WAL flush
2235    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2236
2237    for (r = 0; r < num_kvs; ++r) {
2238        status = fdb_set_log_callback(db[r], logCallbackFunc,
2239                                      (void *) "snapshot_markers_in_file");
2240        TEST_CHK(status == FDB_RESULT_SUCCESS);
2241    }
2242
2243    status = fdb_get_all_snap_markers(dbfile, &markers, &num_markers);
2244    TEST_CHK(status == FDB_RESULT_SUCCESS);
2245
2246    if (!multi_kv) {
2247        TEST_CHK(num_markers == 4);
2248        for (r = 0; r < num_kvs; ++r) {
2249            TEST_CHK(markers[r].num_kvs_markers == 1);
2250            TEST_CHK(markers[r].kvs_markers[0].seqnum
2251                     == (fdb_seqnum_t)(n - r*5));
2252        }
2253    } else {
2254        TEST_CHK(num_markers == 8);
2255        for (r = 0; r < num_kvs; ++r) {
2256            TEST_CHK(markers[r].num_kvs_markers == num_kvs);
2257            for (i = 0; i < num_kvs; ++i) {
2258                TEST_CHK(markers[r].kvs_markers[i].seqnum
2259                         == (fdb_seqnum_t)(n - r*5));
2260                sprintf(kv_name, "kv%d", i);
2261                TEST_CMP(markers[r].kvs_markers[i].kv_store_name, kv_name, 3);
2262            }
2263        }
2264    }
2265
2266    status = fdb_free_snap_markers(markers, num_markers);
2267    TEST_CHK(status == FDB_RESULT_SUCCESS);
2268
2269    // close db file
2270    fdb_close(dbfile);
2271
2272    // free all documents
2273    for (i=0;i<n;++i){
2274        fdb_doc_free(doc[i]);
2275    }
2276
2277    // free all resources
2278    fdb_shutdown();
2279
2280    memleak_end();
2281
2282    sprintf(bodybuf, "snapshot markers in file test %s", multi_kv ?
2283                                                        "multiple kv mode:"
2284                                                      : "single kv mode:");
2285    TEST_RESULT(bodybuf);
2286}
2287
2288void snapshot_without_seqtree(bool multi_kv)
2289{
2290    TEST_INIT();
2291
2292    memleak_start();
2293
2294    int i, j, r;
2295    int n = 10;
2296    int num_commits = 3, num_kvs = 3;
2297    fdb_file_handle *dbfile;
2298    fdb_kvs_handle **db = alca(fdb_kvs_handle *, num_kvs);
2299    fdb_doc **doc = alca(fdb_doc*, n);
2300    fdb_status status;
2301    fdb_snapshot_info_t *markers;
2302    uint64_t num_markers;
2303
2304    char keybuf[256], metabuf[256], bodybuf[256];
2305    char kv_name[8];
2306
2307    fdb_config fconfig = fdb_get_default_config();
2308    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2309    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2310    fconfig.multi_kv_instances = multi_kv;
2311    // for creating the strict number of snapshots, disable block reusing
2312    fconfig.block_reusing_threshold = 0;
2313    fconfig.seqtree_opt = FDB_SEQTREE_NOT_USE;
2314
2315    // remove previous mvcc_test files
2316    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2317    (void)r;
2318
2319    // open db
2320    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2321    if (multi_kv) {
2322        for (r = 0; r < num_kvs; ++r) {
2323            sprintf(kv_name, "kv%d", r);
2324            fdb_kvs_open(dbfile, &db[r], kv_name, &kvs_config);
2325        }
2326    } else {
2327        num_kvs = 1;
2328        fdb_kvs_open_default(dbfile, &db[0], &kvs_config);
2329    }
2330
2331    // Load documents mulitple times
2332    for (j=0; j < num_commits; ++j) {
2333        for (i=0; i<n; i++){
2334            sprintf(keybuf, "key%d", i);
2335            sprintf(metabuf, "meta-%d-%d", j, i);
2336            sprintf(bodybuf, "body-%d-%d", j, i);
2337            fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2338                           (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2339            for (r = 0; r < num_kvs; ++r) {
2340                fdb_set(db[r], doc[i]);
2341            }
2342            fdb_doc_free(doc[i]);
2343        }
2344        if (j % 2 == 0) {
2345            fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2346        } else {
2347            fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2348        }
2349    }
2350
2351    status = fdb_get_all_snap_markers(dbfile, &markers, &num_markers);
2352    TEST_CHK(status == FDB_RESULT_SUCCESS);
2353
2354    if (!multi_kv) {
2355        TEST_CHK(num_markers == 3);
2356        for (r = 0; r < num_kvs; ++r) {
2357            TEST_CHK(markers[r].num_kvs_markers == 1);
2358            TEST_CHK(markers[r].kvs_markers[0].seqnum
2359                     == (fdb_seqnum_t) (n * (num_commits - r)));
2360        }
2361    } else {
2362        TEST_CHK(num_markers == 6);
2363        for (r = 0; r < num_kvs; ++r) {
2364            TEST_CHK(markers[r].num_kvs_markers == num_kvs);
2365            for (i = 0; i < num_kvs; ++i) {
2366                TEST_CHK(markers[r].kvs_markers[i].seqnum
2367                         == (fdb_seqnum_t) (n * (num_commits - r)));
2368                sprintf(kv_name, "kv%d", i);
2369                TEST_CMP(markers[r].kvs_markers[i].kv_store_name, kv_name, 3);
2370            }
2371        }
2372    }
2373
2374    fdb_kvs_handle *snap_db;
2375    fdb_iterator *iterator;
2376    fdb_doc *rdoc = NULL;
2377    // Open the snapshot with the snapshot markers from the second commit
2378    for (r = 0; r < num_kvs; ++r) {
2379        status = fdb_snapshot_open(db[r], &snap_db, markers[1].kvs_markers[r].seqnum);
2380        TEST_CHK(status == FDB_RESULT_SUCCESS);
2381
2382        // Verify all the keys in the snapshot
2383        status = fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
2384        TEST_CHK(status == FDB_RESULT_SUCCESS);
2385        i=0;
2386        do {
2387            status = fdb_iterator_get(iterator, &rdoc);
2388            TEST_CHK(status == FDB_RESULT_SUCCESS);
2389            sprintf(keybuf, "key%d", i);
2390            sprintf(metabuf, "meta-1-%d", i);
2391            sprintf(bodybuf, "body-1-%d", i);
2392            TEST_CMP(rdoc->key, keybuf, rdoc->keylen);
2393            TEST_CMP(rdoc->meta, metabuf, rdoc->metalen);
2394            TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
2395            fdb_doc_free(rdoc);
2396            rdoc = NULL;
2397            i++;
2398        } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
2399        TEST_CHK(i==n);
2400        status = fdb_iterator_close(iterator);
2401        TEST_CHK(status == FDB_RESULT_SUCCESS);
2402
2403        // Verify that a sequence-based iteration is not allowed.
2404        status = fdb_iterator_sequence_init(snap_db, &iterator, 0, 0, FDB_ITR_NONE);
2405        TEST_CHK(status != FDB_RESULT_SUCCESS);
2406
2407        status = fdb_kvs_close(snap_db);
2408        TEST_CHK(status == FDB_RESULT_SUCCESS);
2409    }
2410
2411    status = fdb_free_snap_markers(markers, num_markers);
2412    TEST_CHK(status == FDB_RESULT_SUCCESS);
2413
2414    // close db file
2415    fdb_close(dbfile);
2416
2417    // free all resources
2418    fdb_shutdown();
2419
2420    memleak_end();
2421
2422    sprintf(bodybuf, "snapshot without seqtree in %s",
2423            multi_kv ? "multiple kv mode:" : "single kv mode:");
2424    TEST_RESULT(bodybuf);
2425}
2426
2427void snapshot_with_deletes_test()
2428{
2429    TEST_INIT();
2430
2431    memleak_start();
2432
2433    int i, r;
2434    int n = 10;
2435    fdb_file_handle *dbfile;
2436    fdb_kvs_handle *db;
2437    fdb_kvs_handle *snap_db;
2438    fdb_doc **doc = alca(fdb_doc*, n);
2439    fdb_doc *rdoc = NULL;
2440    fdb_kvs_info kvs_info;
2441    fdb_status status;
2442    fdb_iterator *iterator;
2443
2444    char keybuf[256], metabuf[256], bodybuf[256];
2445
2446    // remove previous mvcc_test files
2447    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2448    (void)r;
2449
2450    fdb_config fconfig = fdb_get_default_config();
2451    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2452    fconfig.buffercache_size = 0;
2453
2454    // remove previous mvcc_test files
2455    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2456    (void)r;
2457
2458    // open db
2459    status = fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2460    fdb_kvs_open_default(dbfile, &db, &kvs_config);
2461    TEST_CHK(status == FDB_RESULT_SUCCESS);
2462
2463    status = fdb_set_log_callback(db, logCallbackFunc,
2464                                  (void *) "snapshot_with_deletes_test");
2465    TEST_CHK(status == FDB_RESULT_SUCCESS);
2466
2467    // ------- Setup test ----------------------------------
2468    // insert even documents into main index
2469    for (i=0; i<n; i = i + 2){
2470        sprintf(keybuf, "key%d", i);
2471        sprintf(metabuf, "meta%d", i);
2472        sprintf(bodybuf, "body%d", i);
2473        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2474            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2475        fdb_set(db, doc[i]);
2476    }
2477
2478    i = 7;
2479    sprintf(keybuf, "key%d", i);
2480    fdb_set_kv(db, keybuf, strlen(keybuf), NULL, 0);
2481
2482    // commit with a manual WAL flush (these docs go into HB-trie)
2483    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2484
2485    // let odd documents be in WAL section
2486    for (i = 1; i < n; i = i + 2){
2487        sprintf(keybuf, "key%d", i);
2488        sprintf(metabuf, "meta%d", i);
2489        sprintf(bodybuf, "body%d", i);
2490        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2491            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2492        fdb_set(db, doc[i]);
2493    }
2494
2495    // Delete WAL doc 7
2496    i = 7;
2497    fdb_del(db, doc[i]);
2498
2499    // Create an iterator on the live DB over full range
2500    fdb_iterator_init(db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
2501
2502    // attempt to seek to the deleted key7
2503    status = fdb_iterator_seek(iterator, doc[i]->key, doc[i]->keylen,
2504                               FDB_ITR_SEEK_HIGHER);
2505    TEST_CHK(status == FDB_RESULT_SUCCESS);
2506    status = fdb_iterator_get(iterator, &rdoc);
2507    TEST_CHK(status == FDB_RESULT_SUCCESS);
2508
2509    i = 8; // The next higher document must be returned
2510    TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
2511    TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
2512    TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
2513
2514    fdb_doc_free(rdoc);
2515    rdoc = NULL;
2516
2517    fdb_iterator_close(iterator);
2518
2519    // Open an in-memory snapshot over live DB
2520    status = fdb_snapshot_open(db, &snap_db, FDB_SNAPSHOT_INMEM);
2521    TEST_CHK(status == FDB_RESULT_SUCCESS);
2522
2523    // check snapshot's sequence number
2524    fdb_get_kvs_info(snap_db, &kvs_info);
2525    TEST_CHK(kvs_info.last_seqnum == (fdb_seqnum_t)n+2);
2526    TEST_CHK(kvs_info.deleted_count == 1);
2527
2528    // re-create an iterator on the snapshot for full range
2529    fdb_iterator_init(snap_db, &iterator, NULL, 0, NULL, 0, FDB_ITR_NONE);
2530
2531    i = 7;
2532    // attempt to seek to the deleted key7
2533    status = fdb_iterator_seek(iterator, doc[i]->key, doc[i]->keylen,
2534                               FDB_ITR_SEEK_HIGHER);
2535    TEST_CHK(status == FDB_RESULT_SUCCESS);
2536    status = fdb_iterator_get(iterator, &rdoc);
2537    TEST_CHK(status == FDB_RESULT_SUCCESS);
2538
2539    i = 8; // The next higher document must be returned
2540    TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
2541    TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
2542    TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
2543
2544    fdb_doc_free(rdoc);
2545    rdoc = NULL;
2546
2547    fdb_iterator_close(iterator);
2548
2549    // close db handle
2550    fdb_kvs_close(db);
2551    // close snapshot handle
2552    fdb_kvs_close(snap_db);
2553    // close db file
2554    fdb_close(dbfile);
2555
2556    // free all documents
2557    for (i=0;i<n;++i){
2558        fdb_doc_free(doc[i]);
2559    }
2560
2561    // free all resources
2562    fdb_shutdown();
2563
2564    memleak_end();
2565
2566    TEST_RESULT("snapshot with deletes test");
2567}
2568
2569void rollback_forward_seqnum()
2570{
2571
2572    TEST_INIT();
2573    memleak_start();
2574
2575    int r;
2576    int i, n=100;
2577    int rb1_seqnum, rb2_seqnum;
2578    char keybuf[256];
2579    char setop[3];
2580    fdb_file_handle *dbfile;
2581    fdb_iterator *it;
2582    fdb_kvs_handle *kv1, *mirror_kv1;
2583    fdb_kvs_info info;
2584    fdb_config fconfig = fdb_get_default_config();
2585    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2586    fdb_doc **doc = alca(fdb_doc*, n+1);
2587    fdb_doc *rdoc = NULL;
2588    fdb_status status;
2589    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2590    (void)r;
2591
2592    fconfig.wal_threshold = n;
2593    fconfig.seqtree_opt = FDB_SEQTREE_USE; // enable seqtree since get_byseq
2594    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2595    fconfig.compaction_threshold = 0;
2596    fconfig.purging_interval = 5;
2597    fconfig.block_reusing_threshold = 0;
2598
2599    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2600    fdb_kvs_open(dbfile, &kv1, "kv1", &kvs_config);
2601    fdb_kvs_open(dbfile, &mirror_kv1, NULL, &kvs_config);
2602
2603
2604    // set n docs within both dbs
2605    for(i=0;i<=n;++i){
2606        sprintf(keybuf, "key%d", i);
2607        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2608            NULL, 0, NULL, 0);
2609        fdb_set(kv1, doc[i]);
2610        fdb_set_kv(mirror_kv1, keybuf, strlen(keybuf), setop, 3);
2611    } // last set should have caused a wal flush
2612
2613    fdb_del(kv1, doc[n]);
2614
2615    // commit and save seqnum1
2616    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2617    fdb_get_kvs_info(kv1, &info);
2618    rb1_seqnum = info.last_seqnum;
2619
2620    // delete all docs in kv1
2621    for(i=0;i<n;++i){
2622        fdb_del(kv1, doc[i]);
2623    }
2624
2625    // commit and save seqnum2
2626    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2627    fdb_get_kvs_info(kv1, &info);
2628    rb2_seqnum = info.last_seqnum;
2629
2630    // sets again
2631    for(i=0;i<n;++i){
2632        doc[i]->deleted = false;
2633        fdb_set(kv1, doc[i]);
2634    }
2635
2636    // commit
2637    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2638
2639    // rollback to first seqnum
2640    status = fdb_rollback(&kv1, rb1_seqnum);
2641    TEST_CHK(status == FDB_RESULT_SUCCESS);
2642
2643    fdb_get_kvs_info(kv1, &info);
2644    TEST_CHK(info.deleted_count == 1);
2645
2646    // rollback to second seqnum
2647    status = fdb_rollback(&kv1, rb2_seqnum);
2648    TEST_CHK(status == FDB_RESULT_NO_DB_INSTANCE);
2649
2650    status = fdb_iterator_sequence_init(mirror_kv1, &it, 0, 0, FDB_ITR_NONE);
2651    TEST_CHK(status == FDB_RESULT_SUCCESS);
2652
2653    do {
2654        status = fdb_iterator_get(it, &rdoc);
2655        TEST_CHK(status == FDB_RESULT_SUCCESS);
2656        if (rdoc->seqnum != (uint64_t)n+1) {
2657            status = fdb_get_metaonly_byseq(kv1, rdoc);
2658            TEST_CHK(status == FDB_RESULT_SUCCESS);
2659            TEST_CHK(rdoc->deleted == false);
2660        } else {
2661            status = fdb_get_metaonly(kv1, rdoc);
2662            TEST_CHK(status == FDB_RESULT_SUCCESS);
2663            TEST_CHK(rdoc->deleted == true);
2664        }
2665        fdb_doc_free(rdoc);
2666        rdoc = NULL;
2667    } while(fdb_iterator_next(it) != FDB_RESULT_ITERATOR_FAIL);
2668
2669    for (i=0;i<=n;++i){
2670        fdb_doc_free(doc[i]);
2671    }
2672    fdb_iterator_close(it);
2673    fdb_kvs_close(kv1);
2674    fdb_close(dbfile);
2675    fdb_shutdown();
2676    memleak_end();
2677
2678    TEST_RESULT("rollback forward seqnum");
2679}
2680
2681void rollback_test(bool multi_kv)
2682{
2683    TEST_INIT();
2684
2685    memleak_start();
2686
2687    int i, r;
2688    int n = 20;
2689    int count;
2690    fdb_file_handle *dbfile, *dbfile_txn;
2691    fdb_kvs_handle *db, *db_txn;
2692    fdb_seqnum_t rollback_seq;
2693    fdb_doc **doc = alca(fdb_doc*, n);
2694    fdb_doc *rdoc = NULL;
2695    fdb_kvs_info kvs_info;
2696    fdb_status status;
2697    fdb_iterator *iterator;
2698
2699    char keybuf[256], metabuf[256], bodybuf[256];
2700
2701    // remove previous mvcc_test files
2702    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2703    (void)r;
2704
2705    fdb_config fconfig = fdb_get_default_config();
2706    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2707    fconfig.buffercache_size = 0;
2708    fconfig.seqtree_opt = FDB_SEQTREE_USE; // enable seqtree since get_byseq
2709    fconfig.wal_threshold = 1024;
2710    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2711    fconfig.compaction_threshold = 0;
2712    fconfig.multi_kv_instances = multi_kv;
2713
2714    // remove previous mvcc_test files
2715    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2716    (void)r;
2717
2718    // open db
2719    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2720    if (multi_kv) {
2721        fdb_kvs_open_default(dbfile, &db, &kvs_config);
2722    } else {
2723        fdb_kvs_open(dbfile, &db, NULL, &kvs_config);
2724    }
2725
2726   // ------- Setup test ----------------------------------
2727   // insert documents of 0-4
2728    for (i=0; i<n/4; i++){
2729        sprintf(keybuf, "key%d", i);
2730        sprintf(metabuf, "meta%d", i);
2731        sprintf(bodybuf, "body%d", i);
2732        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2733            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2734        fdb_set(db, doc[i]);
2735    }
2736
2737    // commit with a manual WAL flush (these docs go into HB-trie)
2738    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2739
2740    // insert documents from 4 - 9
2741    for (; i < n/2; i++){
2742        sprintf(keybuf, "key%d", i);
2743        sprintf(metabuf, "meta%d", i);
2744        sprintf(bodybuf, "body%d", i);
2745        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2746            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2747        fdb_set(db, doc[i]);
2748    }
2749
2750    // commit again without a WAL flush
2751    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2752
2753    // ROLLBACK POINT: pick up sequence number of a commit without a WAL flush
2754    rollback_seq = doc[i-1]->seqnum;
2755
2756    // insert documents from 10-14 into HB-trie
2757    for (; i < (n/2 + n/4); i++){
2758        sprintf(keybuf, "key%d", i);
2759        sprintf(metabuf, "meta%d", i);
2760        sprintf(bodybuf, "body%d", i);
2761        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2762            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2763        fdb_set(db, doc[i]);
2764    }
2765    // manually flush WAL & commit
2766    fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
2767
2768    // insert documents from 15 - 19 on file into the WAL
2769    for (; i < n; i++){
2770        sprintf(keybuf, "key%d", i);
2771        sprintf(metabuf, "meta%d", i);
2772        sprintf(bodybuf, "body%d", i);
2773        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
2774            (void*)metabuf, strlen(metabuf), (void*)bodybuf, strlen(bodybuf));
2775        fdb_set(db, doc[i]);
2776    }
2777    // commit without a WAL flush
2778    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2779
2780    status = fdb_set_log_callback(db, logCallbackFunc,
2781                                  (void *) "rollback_test");
2782    TEST_CHK(status == FDB_RESULT_SUCCESS);
2783
2784    // ---------- Rollback tests begin -----------------------
2785    // We have DB file with 5 HB-trie docs, 5 unflushed WAL docs occuring twice
2786    // Attempt to rollback to out-of-range marker..
2787    status = fdb_rollback(&db, 999999);
2788    TEST_CHK(status == FDB_RESULT_NO_DB_INSTANCE);
2789
2790    // Open another handle & begin transaction
2791    fdb_open(&dbfile_txn, "./mvcc_test1", &fconfig);
2792    if (multi_kv) {
2793        fdb_kvs_open_default(dbfile_txn, &db_txn, &kvs_config);
2794    } else {
2795        fdb_kvs_open(dbfile_txn, &db_txn, NULL, &kvs_config);
2796    }
2797    fdb_begin_transaction(dbfile_txn, FDB_ISOLATION_READ_COMMITTED);
2798    // Attempt to rollback while the transaction is active
2799    status =  fdb_rollback(&db, rollback_seq);
2800    // Must fail
2801    TEST_CHK(status == FDB_RESULT_FAIL_BY_TRANSACTION);
2802    fdb_abort_transaction(dbfile_txn);
2803    fdb_kvs_close(db_txn);
2804    fdb_close(dbfile_txn);
2805
2806    // Rollback to saved marker from above
2807    status = fdb_rollback(&db, rollback_seq);
2808    TEST_CHK(status == FDB_RESULT_SUCCESS);
2809
2810    // check handle's sequence number
2811    fdb_get_kvs_info(db, &kvs_info);
2812    TEST_CHK(kvs_info.last_seqnum == rollback_seq);
2813
2814    // Modify an item and update into the rollbacked file..
2815    i = n/2;
2816    *(char *)doc[i]->body = 'B';
2817    fdb_set(db, doc[i]);
2818    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2819
2820    // create an iterator on the rollback for full range
2821    fdb_iterator_sequence_init(db, &iterator, 0, 0, FDB_ITR_NONE);
2822
2823    // repeat until fail
2824    i=0;
2825    count=0;
2826    do {
2827        status = fdb_iterator_get(iterator, &rdoc);
2828        TEST_CHK(status == FDB_RESULT_SUCCESS);
2829
2830        TEST_CMP(rdoc->key, doc[i]->key, rdoc->keylen);
2831        TEST_CMP(rdoc->meta, doc[i]->meta, rdoc->metalen);
2832        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
2833
2834        fdb_doc_free(rdoc);
2835        rdoc = NULL;
2836        i ++;
2837        count++;
2838    } while (fdb_iterator_next(iterator) != FDB_RESULT_ITERATOR_FAIL);
2839
2840    TEST_CHK(count==n/2 + 1); // Items from first half and newly set item
2841
2842    fdb_iterator_close(iterator);
2843
2844    // close db file
2845    fdb_kvs_close(db);
2846    fdb_close(dbfile);
2847
2848    // free all documents
2849    for (i=0;i<n;++i){
2850        fdb_doc_free(doc[i]);
2851    }
2852
2853    // free all resources
2854    fdb_shutdown();
2855
2856    memleak_end();
2857
2858    sprintf(bodybuf, "rollback test %s", multi_kv ? "multiple kv mode:"
2859                                                  : "single kv mode:");
2860    TEST_RESULT(bodybuf);
2861}
2862
2863
2864
2865void rollback_and_snapshot_test()
2866{
2867    TEST_INIT();
2868
2869    memleak_start();
2870
2871    fdb_seqnum_t seqnum, rollback_seqnum;
2872    fdb_kvs_info kvs_info;
2873    fdb_status status;
2874    fdb_config config;
2875    fdb_kvs_config kvs_config;
2876    fdb_file_handle *dbfile;
2877    fdb_kvs_handle *db,  *snapshot;
2878    int r;
2879
2880    // remove previous mvcc_test files
2881    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2882    (void)r;
2883
2884    // MB-12530 open db
2885    config = fdb_get_default_config();
2886    kvs_config = fdb_get_default_kvs_config();
2887    status = fdb_open(&dbfile, "mvcc_test", &config);
2888    TEST_CHK(status == FDB_RESULT_SUCCESS);
2889    status = fdb_kvs_open(dbfile, &db, NULL, &kvs_config);
2890    TEST_CHK(status == FDB_RESULT_SUCCESS);
2891
2892    // 2. Create Key 'a' and Commit
2893    status = fdb_set_kv(db, (void *) "a", 1, (void *)"val-a", 5);
2894    TEST_CHK(status == FDB_RESULT_SUCCESS);
2895    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2896    TEST_CHK(status == FDB_RESULT_SUCCESS);
2897
2898    status = fdb_get_kvs_info(db, &kvs_info);
2899    TEST_CHK(status == FDB_RESULT_SUCCESS);
2900
2901    // 3. Create Key 'b' and Commit
2902    status = fdb_set_kv(db, (void *)"b", 1, (void *)"val-b", 5);
2903    TEST_CHK(status == FDB_RESULT_SUCCESS);
2904    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2905    TEST_CHK(status == FDB_RESULT_SUCCESS);
2906
2907    status = fdb_get_kvs_info(db, &kvs_info);
2908    TEST_CHK(status == FDB_RESULT_SUCCESS);
2909    seqnum = kvs_info.last_seqnum;
2910
2911    // 4.  Remember this as our rollback point
2912    rollback_seqnum = seqnum;
2913
2914    // 5. Create Key 'c' and Commit
2915    status = fdb_set_kv(db, (void *)"c", 1,(void *) "val-c", 5);
2916    TEST_CHK(status == FDB_RESULT_SUCCESS);
2917    status = fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2918    TEST_CHK(status == FDB_RESULT_SUCCESS);
2919
2920    status = fdb_get_kvs_info(db, &kvs_info);
2921    TEST_CHK(status == FDB_RESULT_SUCCESS);
2922
2923    // 6. Rollback to rollback point (seq 2)
2924    status = fdb_rollback(&db, rollback_seqnum);
2925    TEST_CHK(status == FDB_RESULT_SUCCESS);
2926
2927    status = fdb_get_kvs_info(db, &kvs_info);
2928    TEST_CHK(status == FDB_RESULT_SUCCESS);
2929    seqnum = kvs_info.last_seqnum;
2930
2931    // 7. Verify that Key 'c' is not found
2932    void *val;
2933    size_t vallen;
2934    status = fdb_get_kv(db, (void *)"c", 1, &val, &vallen);
2935    TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
2936
2937    // 8. Open a snapshot at the same point
2938    status = fdb_snapshot_open(db, &snapshot, seqnum);
2939    TEST_CHK(status == FDB_RESULT_SUCCESS);
2940
2941    // 9. Verify that Key 'c' is not found
2942    status = fdb_get_kv(snapshot, (void *)"c", 1, &val, &vallen);
2943    TEST_CHK(status == FDB_RESULT_KEY_NOT_FOUND);
2944
2945    // close the snapshot db
2946    fdb_kvs_close(snapshot);
2947
2948    // close db file
2949    fdb_close(dbfile);
2950
2951    // free all resources
2952    fdb_shutdown();
2953
2954    memleak_end();
2955
2956    TEST_RESULT("rollback and snapshot test");
2957}
2958
2959void rollback_ncommits()
2960{
2961
2962    TEST_INIT();
2963    memleak_start();
2964
2965    int r;
2966    int i, j, n=100;
2967    int ncommits=10;
2968    char keybuf[256];
2969    fdb_file_handle *dbfile;
2970    fdb_kvs_handle *kv1, *kv2;
2971    fdb_kvs_info info;
2972    fdb_config fconfig = fdb_get_default_config();
2973    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
2974    fdb_status status;
2975    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
2976    (void)r;
2977
2978    fconfig.wal_threshold = 1024;
2979    fconfig.flags = FDB_OPEN_FLAG_CREATE;
2980    fconfig.compaction_threshold = 0;
2981    fconfig.block_reusing_threshold = 0;
2982
2983    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
2984    fdb_kvs_open(dbfile, &kv1, "kv1", &kvs_config);
2985    fdb_kvs_open(dbfile, &kv2, NULL, &kvs_config);
2986
2987
2988    for(j=0;j<ncommits;++j){
2989
2990        // set n docs per commit
2991        for(i=0;i<n;++i){
2992            sprintf(keybuf, "key%02d%03d", j, i);
2993            fdb_set_kv(kv1, keybuf, strlen(keybuf), NULL, 0);
2994            fdb_set_kv(kv2, keybuf, strlen(keybuf), NULL, 0);
2995        }
2996        // alternate commit pattern
2997        if((j % 2) == 0){
2998            fdb_commit(dbfile, FDB_COMMIT_NORMAL);
2999        } else {
3000            fdb_commit(dbfile, FDB_COMMIT_MANUAL_WAL_FLUSH);
3001        }
3002
3003        // doc_count should match seqnum since they are unique
3004        fdb_get_kvs_info(kv1, &info);
3005        TEST_CHK(info.doc_count == info.last_seqnum);
3006    }
3007
3008    // iteratively rollback 5 commits
3009     for(j=ncommits;j>0;--j){
3010        status = fdb_rollback(&kv1, j*n);
3011        TEST_CHK(status == FDB_RESULT_SUCCESS);
3012
3013        // check rollback doc_count
3014        fdb_get_kvs_info(kv1, &info);
3015        TEST_CHK(info.doc_count == info.last_seqnum);
3016    }
3017
3018    fdb_kvs_close(kv1);
3019    fdb_close(dbfile);
3020    fdb_shutdown();
3021    memleak_end();
3022
3023    TEST_RESULT("rollback n commits");
3024}
3025
3026void transaction_test()
3027{
3028    TEST_INIT();
3029
3030    memleak_start();
3031
3032    int i, r;
3033    int n = 10;
3034    fdb_file_handle *dbfile, *dbfile_txn1, *dbfile_txn2, *dbfile_txn3;
3035    fdb_kvs_handle *db, *db_txn1, *db_txn2, *db_txn3;
3036    fdb_doc **doc = alca(fdb_doc*, n);
3037    fdb_doc *rdoc;
3038    fdb_status status;
3039
3040    char keybuf[256], metabuf[256], bodybuf[256];
3041
3042    // remove previous mvcc_test files
3043    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
3044    (void)r;
3045
3046    fdb_config fconfig = fdb_get_default_config();
3047    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
3048    fconfig.buffercache_size = 0;
3049    fconfig.wal_threshold = 1024;
3050    fconfig.flags = FDB_OPEN_FLAG_CREATE;
3051    fconfig.purging_interval = 0;
3052    fconfig.compaction_threshold = 0;
3053
3054    // open db
3055    fdb_open(&dbfile, "mvcc_test1", &fconfig);
3056    fdb_kvs_open_default(dbfile, &db, &kvs_config);
3057    status = fdb_set_log_callback(db, logCallbackFunc,
3058                                  (void *) "transaction_test");
3059    TEST_CHK(status == FDB_RESULT_SUCCESS);
3060
3061    // open db and begin transactions
3062    fdb_open(&dbfile_txn1, "mvcc_test1", &fconfig);
3063    fdb_open(&dbfile_txn2, "mvcc_test1", &fconfig);
3064    fdb_kvs_open_default(dbfile_txn1, &db_txn1, &kvs_config);
3065    fdb_kvs_open_default(dbfile_txn2, &db_txn2, &kvs_config);
3066    fdb_begin_transaction(dbfile_txn1, FDB_ISOLATION_READ_COMMITTED);
3067    fdb_begin_transaction(dbfile_txn2, FDB_ISOLATION_READ_COMMITTED);
3068
3069    // insert half docs into txn1
3070    for (i=0;i<n/2;++i){
3071        sprintf(keybuf, "key%d", i);
3072        sprintf(metabuf, "meta%d", i);
3073        sprintf(bodybuf, "body%d", i);
3074        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
3075                                (void*)metabuf, strlen(metabuf),
3076                                (void*)bodybuf, strlen(bodybuf));
3077        fdb_set(db_txn1, doc[i]);
3078    }
3079
3080    // insert other half docs into txn2
3081    for (i=n/2;i<n;++i){
3082        sprintf(keybuf, "key%d", i);
3083        sprintf(metabuf, "meta%d", i);
3084        sprintf(bodybuf, "body%d", i);
3085        fdb_doc_create(&doc[i], (void*)keybuf, strlen(keybuf),
3086                                (void*)metabuf, strlen(metabuf),
3087                                (void*)bodybuf, strlen(bodybuf));
3088        fdb_set(db_txn2, doc[i]);
3089    }
3090
3091    // uncommitted docs should not be read by the other transaction that
3092    // doesn't allow uncommitted reads.
3093    for (i=0;i<n;++i){
3094        sprintf(keybuf, "key%d", i);
3095        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3096        status = fdb_get(db_txn1, rdoc);
3097        if (i<n/2) {
3098            TEST_CHK(status == FDB_RESULT_SUCCESS);
3099        } else {
3100            TEST_CHK(status != FDB_RESULT_SUCCESS);
3101        }
3102        fdb_doc_free(rdoc);
3103    }
3104
3105    // uncommitted docs can be read by the transaction that allows uncommitted reads.
3106    fdb_open(&dbfile_txn3, "mvcc_test1", &fconfig);
3107    fdb_kvs_open_default(dbfile_txn3, &db_txn3, &kvs_config);
3108    fdb_begin_transaction(dbfile_txn3, FDB_ISOLATION_READ_UNCOMMITTED);
3109    for (i=0;i<n;++i){
3110        sprintf(keybuf, "key%d", i);
3111        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3112        status = fdb_get(db_txn3, rdoc);
3113        TEST_CHK(status == FDB_RESULT_SUCCESS);
3114        fdb_doc_free(rdoc);
3115    }
3116    fdb_end_transaction(dbfile_txn3, FDB_COMMIT_NORMAL);
3117    fdb_close(dbfile_txn3);
3118
3119    // commit and end txn1
3120    fdb_end_transaction(dbfile_txn1, FDB_COMMIT_NORMAL);
3121
3122    // uncommitted docs should not be read generally
3123    for (i=0;i<n;++i){
3124        sprintf(keybuf, "key%d", i);
3125        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3126        status = fdb_get(db, rdoc);
3127        if (i<n/2) {
3128            TEST_CHK(status == FDB_RESULT_SUCCESS);
3129        } else {
3130            TEST_CHK(status != FDB_RESULT_SUCCESS);
3131        }
3132        fdb_doc_free(rdoc);
3133    }
3134
3135    // read uncommitted docs using the same transaction
3136    for (i=0;i<n;++i){
3137        sprintf(keybuf, "key%d", i);
3138        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3139        status = fdb_get(db_txn2, rdoc);
3140        TEST_CHK(status == FDB_RESULT_SUCCESS);
3141        fdb_doc_free(rdoc);
3142    }
3143
3144    // abort txn2
3145    fdb_abort_transaction(dbfile_txn2);
3146
3147    // close & re-open db file
3148    fdb_close(dbfile);
3149    fdb_open(&dbfile, "mvcc_test1", &fconfig);
3150    fdb_kvs_open_default(dbfile, &db, &kvs_config);
3151    status = fdb_set_log_callback(db, logCallbackFunc,
3152                                  (void *) "transaction_test");
3153    TEST_CHK(status == FDB_RESULT_SUCCESS);
3154
3155    // uncommitted docs should not be read after reopening the file either
3156    for (i=0;i<n;++i){
3157        sprintf(keybuf, "key%d", i);
3158        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3159        status = fdb_get(db, rdoc);
3160        if (i<n/2) {
3161            TEST_CHK(status == FDB_RESULT_SUCCESS);
3162        } else {
3163            TEST_CHK(status != FDB_RESULT_SUCCESS);
3164        }
3165        fdb_doc_free(rdoc);
3166    }
3167
3168    // insert other half docs & commit
3169    for (i=n/2;i<n;++i){
3170        fdb_set(db, doc[i]);
3171    }
3172    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
3173
3174    // now all docs can be read generally
3175    for (i=0;i<n;++i){
3176        sprintf(keybuf, "key%d", i);
3177        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3178        status = fdb_get(db, rdoc);
3179        TEST_CHK(status == FDB_RESULT_SUCCESS);
3180        fdb_doc_free(rdoc);
3181    }
3182
3183    // begin transactions
3184    fdb_begin_transaction(dbfile_txn1, FDB_ISOLATION_READ_COMMITTED);
3185    fdb_begin_transaction(dbfile_txn2, FDB_ISOLATION_READ_COMMITTED);
3186
3187    // concurrently update docs
3188    for (i=0;i<n;++i){
3189        sprintf(keybuf, "key%d", i);
3190        sprintf(metabuf, "meta%d", i);
3191        sprintf(bodybuf, "body%d_txn1", i);
3192        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf),
3193                                (void*)metabuf, strlen(metabuf),
3194                                (void*)bodybuf, strlen(bodybuf));
3195        fdb_set(db_txn1, rdoc);
3196        fdb_doc_free(rdoc);
3197
3198        sprintf(bodybuf, "body%d_txn2", i);
3199        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf),
3200                                (void*)metabuf, strlen(metabuf),
3201                                (void*)bodybuf, strlen(bodybuf));
3202        fdb_set(db_txn2, rdoc);
3203        fdb_doc_free(rdoc);
3204    }
3205
3206    // retrieve check
3207    for (i=0;i<n;++i){
3208        // general retrieval
3209        sprintf(keybuf, "key%d", i);
3210        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3211        status = fdb_get(db, rdoc);
3212        TEST_CHK(status == FDB_RESULT_SUCCESS);
3213        TEST_CMP(rdoc->body, doc[i]->body, rdoc->bodylen);
3214        fdb_doc_free(rdoc);
3215
3216        // get from txn1
3217        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3218        status = fdb_get(db_txn1, rdoc);
3219        TEST_CHK(status == FDB_RESULT_SUCCESS);
3220        sprintf(bodybuf, "body%d_txn1", i);
3221        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3222        fdb_doc_free(rdoc);
3223
3224        // get from txn2
3225        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3226        status = fdb_get(db_txn2, rdoc);
3227        TEST_CHK(status == FDB_RESULT_SUCCESS);
3228        sprintf(bodybuf, "body%d_txn2", i);
3229        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3230        fdb_doc_free(rdoc);
3231    }
3232
3233    // commit txn2 & retrieve check
3234    fdb_end_transaction(dbfile_txn2, FDB_COMMIT_NORMAL);
3235    for (i=0;i<n;++i){
3236        // general retrieval
3237        sprintf(keybuf, "key%d", i);
3238        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3239        status = fdb_get(db, rdoc);
3240        TEST_CHK(status == FDB_RESULT_SUCCESS);
3241        sprintf(bodybuf, "body%d_txn2", i);
3242        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3243        fdb_doc_free(rdoc);
3244
3245        // get from txn1
3246        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3247        status = fdb_get(db_txn1, rdoc);
3248        TEST_CHK(status == FDB_RESULT_SUCCESS);
3249        sprintf(bodybuf, "body%d_txn1", i);
3250        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3251        fdb_doc_free(rdoc);
3252    }
3253
3254    // commit txn1 & retrieve check
3255    fdb_end_transaction(dbfile_txn1, FDB_COMMIT_NORMAL);
3256    for (i=0;i<n;++i){
3257        // general retrieval
3258        sprintf(keybuf, "key%d", i);
3259        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3260        status = fdb_get(db, rdoc);
3261        TEST_CHK(status == FDB_RESULT_SUCCESS);
3262        sprintf(bodybuf, "body%d_txn1", i);
3263        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3264        fdb_doc_free(rdoc);
3265    }
3266
3267    // begin new transaction
3268    fdb_begin_transaction(dbfile_txn1, FDB_ISOLATION_READ_COMMITTED);
3269    // update doc#5
3270    i = 5;
3271    sprintf(keybuf, "key%d", i);
3272    sprintf(metabuf, "meta%d", i);
3273    sprintf(bodybuf, "body%d_before_compaction", i);
3274    fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf),
3275                            (void*)metabuf, strlen(metabuf),
3276                            (void*)bodybuf, strlen(bodybuf));
3277    fdb_set(db_txn1, rdoc);
3278    fdb_doc_free(rdoc);
3279
3280    // do compaction
3281    fdb_compact(dbfile, "mvcc_test2");
3282
3283    // retrieve doc#5
3284    // using txn1
3285    fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3286    status = fdb_get(db_txn1, rdoc);
3287    TEST_CHK(status == FDB_RESULT_SUCCESS);
3288    TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3289    fdb_doc_free(rdoc);
3290
3291    // general retrieval
3292    fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3293    status = fdb_get(db, rdoc);
3294    TEST_CHK(status == FDB_RESULT_SUCCESS);
3295    sprintf(bodybuf, "body%d_txn1", i);
3296    TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3297    fdb_doc_free(rdoc);
3298
3299    // commit transaction
3300    fdb_end_transaction(dbfile_txn1, FDB_COMMIT_NORMAL);
3301    // retrieve check
3302    for (i=0;i<n;++i){
3303        // general get
3304        sprintf(keybuf, "key%d", i);
3305        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3306        status = fdb_get(db, rdoc);
3307        TEST_CHK(status == FDB_RESULT_SUCCESS);
3308        if (i != 5) {
3309            sprintf(bodybuf, "body%d_txn1", i);
3310        } else {
3311            sprintf(bodybuf, "body%d_before_compaction", i);
3312        }
3313        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3314        fdb_doc_free(rdoc);
3315    }
3316
3317    // close & re-open db file
3318    fdb_close(dbfile);
3319    fdb_open(&dbfile, "mvcc_test2", &fconfig);
3320    fdb_kvs_open_default(dbfile, &db, &kvs_config);
3321    status = fdb_set_log_callback(db, logCallbackFunc,
3322                                  (void *) "transaction_test");
3323    TEST_CHK(status == FDB_RESULT_SUCCESS);
3324    // retrieve check again
3325    for (i=0;i<n;++i){
3326        // general get
3327        sprintf(keybuf, "key%d", i);
3328        fdb_doc_create(&rdoc, (void*)keybuf, strlen(keybuf), NULL, 0, NULL, 0);
3329        status = fdb_get(db, rdoc);
3330        TEST_CHK(status == FDB_RESULT_SUCCESS);
3331        if (i != 5) {
3332            sprintf(bodybuf, "body%d_txn1", i);
3333        } else {
3334            sprintf(bodybuf, "body%d_before_compaction", i);
3335        }
3336        TEST_CMP(rdoc->body, bodybuf, rdoc->bodylen);
3337        fdb_doc_free(rdoc);
3338    }
3339
3340    // close db file
3341    fdb_close(dbfile);
3342    fdb_close(dbfile_txn1);
3343    fdb_close(dbfile_txn2);
3344
3345    // free all documents
3346    for (i=0;i<n;++i){
3347        fdb_doc_free(doc[i]);
3348    }
3349
3350    // free all resources
3351    fdb_shutdown();
3352
3353    memleak_end();
3354
3355    TEST_RESULT("transaction test");
3356}
3357
3358void transaction_simple_api_test()
3359{
3360    TEST_INIT();
3361
3362    memleak_start();
3363
3364    int i, r;
3365    int n = 10;
3366    size_t valuelen;
3367    void *value;
3368    fdb_file_handle *dbfile, *dbfile_txn1, *dbfile_txn2;
3369    fdb_kvs_handle *db, *db_txn1, *db_txn2;
3370    fdb_status status;
3371
3372    char keybuf[256], bodybuf[256];
3373
3374    // remove previous mvcc_test files
3375    r = system(SHELL_DEL" mvcc_test* > errorlog.txt");
3376    (void)r;
3377
3378    fdb_config fconfig = fdb_get_default_config();
3379    fdb_kvs_config kvs_config = fdb_get_default_kvs_config();
3380    fconfig.buffercache_size = 0;
3381    fconfig.wal_threshold = 1024;
3382    fconfig.flags = FDB_OPEN_FLAG_CREATE;
3383    fconfig.purging_interval = 0;
3384    fconfig.compaction_threshold = 0;
3385
3386    // open db
3387    fdb_open(&dbfile, "./mvcc_test1", &fconfig);
3388    fdb_kvs_open_default(dbfile, &db, &kvs_config);
3389    status = fdb_set_log_callback(db, logCallbackFunc,
3390                                  (void *) "transaction_simple_api_test");
3391    TEST_CHK(status == FDB_RESULT_SUCCESS);
3392
3393    // insert key-value pairs
3394    for (i=0;i<n;++i){
3395        sprintf(keybuf, "key%d", i);
3396        sprintf(bodybuf, "body%d", i);
3397        status = fdb_set_kv(db, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3398        TEST_CHK(status == FDB_RESULT_SUCCESS);
3399    }
3400
3401    // commit
3402    fdb_commit(dbfile, FDB_COMMIT_NORMAL);
3403
3404    // open db and begin transactions
3405    fdb_open(&dbfile_txn1, "./mvcc_test1", &fconfig);
3406    fdb_open(&dbfile_txn2, "./mvcc_test1", &fconfig);
3407    fdb_kvs_open_default(dbfile_txn1, &db_txn1, &kvs_config);
3408    fdb_kvs_open_default(dbfile_txn2, &db_txn2, &kvs_config);
3409    fdb_begin_transaction(dbfile_txn1, FDB_ISOLATION_READ_COMMITTED);
3410    fdb_begin_transaction(dbfile_txn2, FDB_ISOLATION_READ_COMMITTED);
3411
3412    // concurrently update docs
3413    for (i=0;i<n;++i){
3414        sprintf(keybuf, "key%d", i);
3415        sprintf(bodybuf, "body%d_txn1", i);
3416        fdb_set_kv(db_txn1, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3417
3418        sprintf(keybuf, "key%d", i);
3419        sprintf(bodybuf, "body%d_txn2", i);
3420        fdb_set_kv(db_txn2, keybuf, strlen(keybuf), bodybuf, strlen(bodybuf));
3421    }
3422
3423    // retrieve key-value pairs
3424    for (i=0;i<n;++i){
3425        // general retrieval
3426        sprintf(keybuf, "key%d", i);
3427        sprintf(bodybuf, "body%d", i);
3428        status = fdb_get_kv(db, keybuf, strlen(keybuf), &value, &valuelen);
3429        TEST_CHK(status == FDB_RESULT_SUCCESS);
3430        TEST_CMP(value, bodybuf, valuelen);
3431        fdb_free_block(value);
3432
3433        // txn1
3434        sprintf(keybuf, "key%d", i);
3435        sprintf(bodybuf, "body%d_txn1", i);
3436        status = fdb_get_kv(db