chiark / gitweb /
fix up compilation errors
[chiark-tcl.git] / cdb / writeable.c
1 /*
2  * cdb, cdb-wr - Tcl bindings for tinycdb and a journalling write extension
3  * Copyright 2006 Ian Jackson
4  *
5  * This program is free software; you can redistribute it and/or
6  * modify it under the terms of the GNU General Public License as
7  * published by the Free Software Foundation; either version 2 of the
8  * License, or (at your option) any later version.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License for more details.
14  *
15  * You should have received a copy of the GNU General Public License
16  * along with this library; if not, write to the Free Software
17  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
18  * 02110-1301, USA.
19  */
20
21 #include "chiark_tcl_cdb.h"
22
23 #define ftello ftell
24 #define fseeko fseek
25
26 /*---------- Forward declarations ----------*/
27
28 struct ht_forall_ctx;
29
30 /*---------- Useful routines ----------*/
31
32 static void maybe_close(int fd) {
33   if (fd>=0) close(fd);
34 }
35
36 /*==================== Subsystems and subtypes ====================*/
37
38 /*---------- Pathbuf ----------*/
39
40 typedef struct Pathbuf {
41   char *buf, *sfx;
42 } Pathbuf;
43
44 #define MAX_SUFFIX 5
45
46 static void pathbuf_init(Pathbuf *pb, const char *pathb) {
47   int l= strlen(pathb);
48   pb->buf= TALLOC(l + MAX_SUFFIX + 1);
49   memcpy(pb->buf, pathb, l);
50   pb->sfx= pb->buf + l;
51 }
52 static const char *pathbuf_sfx(Pathbuf *pb, const char *suffix) {
53   assert(strlen(suffix) <= MAX_SUFFIX);
54   strcpy(pb->sfx, suffix);
55   return pb->buf;
56 }
57 static void pathbuf_free(Pathbuf *pb) {
58   TFREE(pb->buf);
59   pb->buf= 0;
60 }
61
62 /*---------- Our hash table ----------*/
63
64 typedef struct HashTable {
65   Tcl_HashTable t;
66   Byte padding[128]; /* allow for expansion by Tcl, urgh */
67   Byte confound[16];
68 } HashTable;
69
70 typedef struct HashValue {
71   int len;
72   Byte data[1];
73 } HashValue;
74
75 static HashValue *htv_prep(int len) {
76   HashValue *hd;
77   hd= TALLOC((hd->data - (Byte*)hd) + len);
78   hd->len= len;
79   return hd;
80 }  
81 static Byte *htv_fillptr(HashValue *hd) {
82   return hd->data;
83 }
84
85 static void ht_setup(HashTable *ht) {
86   Tcl_InitHashTable(&ht->t, TCL_STRING_KEYS);
87 }
88 static void ht_update(HashTable *ht, const char *key, HashValue *val_eat) {
89   Tcl_HashEntry *he;
90   int new;
91
92   he= Tcl_CreateHashEntry(&ht->t, (char*)key, &new);
93   if (!new) TFREE(Tcl_GetHashValue(he));
94   Tcl_SetHashValue(he, val_eat);
95     /* eats the value since the data structure owns the memory */
96 }
97 static void ht_maybeupdate(HashTable *ht, const char *key,
98                            HashValue *val_eat) {
99   /* like ht_update except does not overwrite existing values */
100   Tcl_HashEntry *he;
101   int new;
102
103   he= Tcl_CreateHashEntry(&ht->t, (char*)key, &new);
104   if (!new) { TFREE(val_eat); return; }
105   Tcl_SetHashValue(he, val_eat);
106 }
107
108 static const HashValue *ht_lookup(HashTable *ht, const char *key) {
109   Tcl_HashEntry *he;
110   
111   he= Tcl_FindHashEntry(&ht->t, key);
112   if (!he) return 0;
113   
114   return Tcl_GetHashValue(he);
115 }
116
117 static int ht_forall(HashTable *ht,
118                      int (*fn)(const char *key, HashValue *val,
119                                struct ht_forall_ctx *ctx),
120                      struct ht_forall_ctx *ctx) {
121   /* Returns first positive value returned by any call to fn, or 0. */
122   Tcl_HashSearch sp;
123   Tcl_HashEntry *he;
124   const char *key;
125   HashValue *val;
126   int r;
127   
128   for (he= Tcl_FirstHashEntry(&ht->t, &sp);
129        he;
130        he= Tcl_NextHashEntry(&sp)) {
131     val= Tcl_GetHashValue(he);
132     if (!val->len) continue;
133
134     key= Tcl_GetHashKey(&ht->t, he);
135     
136     r= fn(key, val, ctx);
137     if (r) return r;
138   }
139   return 0;
140 }
141
142 static void ht_destroy(HashTable *ht) {
143   Tcl_HashSearch sp;
144   Tcl_HashEntry *he;
145   
146   for (he= Tcl_FirstHashEntry(&ht->t, &sp);
147        he;
148        he= Tcl_NextHashEntry(&sp)) {
149     /* ht_forall skips empty (deleted) entries so is no good for this */
150     TFREE(Tcl_GetHashValue(he));
151   }
152   Tcl_DeleteHashTable(&ht->t);
153 }
154
155 /*==================== Existential ====================*/
156
157 /*---------- Rw data structure ----------*/
158
159 typedef struct Rw {
160   int ix, autocompact;
161   int cdb_fd, lock_fd;
162   struct cdb cdb; /* valid iff cdb_fd >= 0 */
163   FILE *logfile;
164   HashTable logincore;
165   Pathbuf pbsome, pbother;
166   off_t mainsz;
167   ScriptToInvoke on_info, on_lexminval;
168 } Rw;
169
170 static int rw_close(Tcl_Interp *ip, Rw *rw) {
171   int rc, r;
172
173   rc= TCL_OK;
174   ht_destroy(&rw->logincore);
175   if (rw->cdb_fd >= 0) cdb_free(&rw->cdb);
176   maybe_close(rw->cdb_fd);
177   maybe_close(rw->lock_fd);
178
179   if (rw->logfile) {
180     r= fclose(rw->logfile);
181     if (r && ip) { rc= cht_posixerr(ip, errno, "probable data loss! failed to"
182                                     " fclose logfile during untidy close"); }
183   }
184
185   pathbuf_free(&rw->pbsome); pathbuf_free(&rw->pbother);
186   return rc;
187 }
188
189 static void destroy_cdbrw_idtabcb(Tcl_Interp *ip, void *rw_v) {
190   rw_close(0,rw_v);
191   TFREE(rw_v);
192 }
193 const IdDataSpec cdbtcl_rwdatabases= {
194   "cdb-rwdb", "cdb-openrwdatabases-table", destroy_cdbrw_idtabcb
195 };
196
197 /*---------- File handling ----------*/
198
199 static int acquire_lock(Tcl_Interp *ip, Pathbuf *pb, int *lockfd_r) {
200   /* *lockfd_r must be -1 on entry.  If may be set to >=0 even
201    * on error, and must be closed by the caller. */
202   mode_t um, lockmode;
203   struct flock fl;
204   int r;
205
206   um= umask(~(mode_t)0);
207   umask(um);
208
209   lockmode= 0666 & ~((um & 0444)>>1);
210   /* Remove r where umask would remove w;
211    * eg umask intending 0664 here gives 0660 */
212   
213   *lockfd_r= open(pathbuf_sfx(pb,".lock"), O_RDWR|O_CREAT, lockmode);
214   if (*lockfd_r < 0)
215     return cht_posixerr(ip, errno, "could not open/create lockfile");
216
217   fl.l_type= F_WRLCK;
218   fl.l_whence= SEEK_SET;
219   fl.l_start= 0;
220   fl.l_len= 0;
221   fl.l_pid= getpid();
222
223   r= fcntl(*lockfd_r, F_SETLK, &fl);
224   if (r == -1) {
225     if (errno == EACCES || errno == EAGAIN)
226       return cht_staticerr(ip, "lock held by another process", "CDB LOCKED");
227     else return cht_posixerr(ip, errno, "unexpected error from fcntl while"
228                              " acquiring lock");
229   }
230
231   return TCL_OK;
232 }
233
234 /*---------- Log reading and writing ----------*/
235
236 static int readlognum(FILE *f, int delim, int *num_r) {
237   int c;
238   char numbuf[20], *p, *ep;
239   unsigned long ul;
240
241   p= numbuf;
242   for (;;) {
243     c= getc(f);  if (c==EOF) return -2;
244     if (c == delim) break;
245     if (!isdigit((unsigned char)c)) return -2;
246     *p++= c;
247     if (p == numbuf+sizeof(numbuf)) return -2;
248   }
249   if (p == numbuf) return -2;
250   *p= 0;
251
252   errno=0; ul= strtoul(numbuf, &ep, 10);
253   if (*ep || errno || ul >= INT_MAX/2) return -2;
254   *num_r= ul;
255   return 0;
256 }
257
258 static int readstorelogrecord(FILE *f, HashTable *ht,
259                               int (*omitfn)(const HashValue*,
260                                             struct ht_forall_ctx *ctx),
261                               struct ht_forall_ctx *ctx,
262                               void (*updatefn)(HashTable*, const char*,
263                                                HashValue*)) {
264   /* returns:
265    *      0     for OK
266    *     -1     eof
267    *     -2     corrupt or error
268    *     -3     got newline indicating end
269    *     >0     value from omitfn
270    */
271   int keylen, vallen;
272   char *key;
273   HashValue *val;
274   int c, rc, r;
275
276   c= getc(f);
277   if (c==EOF) { return feof(f) ? -1 : -2; }
278   if (c=='\n') return -3;
279   if (c!='+') return -2;
280
281   rc= readlognum(f, ',', &keylen);  if (rc) return rc;
282   rc= readlognum(f, ':', &vallen);  if (rc) return rc;
283
284   key= TALLOC(keylen+1);
285   val= htv_prep(vallen);
286
287   r= fread(key, 1,keylen, f);
288   if (r!=keylen) goto x2_free_keyval;
289   if (memchr(key,0,keylen)) goto x2_free_keyval;
290   key[keylen]= 0;
291
292   c= getc(f);  if (c!='-') goto x2_free_keyval;
293   c= getc(f);  if (c!='>') goto x2_free_keyval;
294   
295   r= fread(htv_fillptr(val), 1,vallen, f);
296   if (r!=vallen) goto x2_free_keyval;
297
298   c= getc(f);  if (c!='\n') goto x2_free_keyval;
299
300   rc= omitfn ? omitfn(val, ctx) : TCL_OK;
301   if (rc) { assert(rc>0); TFREE(val); }
302   else updatefn(ht, key, val);
303   
304   TFREE(key);
305   return rc;
306
307  x2_free_keyval:
308   TFREE(val);
309   TFREE(key);
310   return -2;
311 }
312
313 static int writerecord(FILE *f, const char *key, const HashValue *val) {
314   int r;
315
316   r= fprintf(f, "+%d,%d:%s->", strlen(key), val->len, key);
317   if (r<0) return -1;
318   
319   r= fwrite(val->data, 1, val->len, f);
320   if (r != val->len) return -1;
321
322   r= putc('\n', f);
323   if (r==EOF) return -1;
324
325   return 0;
326 }
327
328 /*---------- Creating ----------*/
329
330 int cht_do_cdbwr_create_empty(ClientData cd, Tcl_Interp *ip,
331                               const char *pathb) {
332   static const char *const toremoves[]= { ".cdb", ".jrn", ".tmp", 0 };
333
334   Pathbuf pb, pbmain;
335   int lock_fd=-1, rc, r;
336   FILE *f= 0;
337   const char *const *toremove;
338   struct stat stab;
339
340   pathbuf_init(&pb, pathb);
341   pathbuf_init(&pbmain, pathb);
342
343   rc= acquire_lock(ip, &pb, &lock_fd);  if (rc) goto x_rc;
344
345   r= lstat(pathbuf_sfx(&pbmain, ".main"), &stab);
346   if (!r) { rc= cht_staticerr(ip, "database already exists during creation",
347                               "CDB ALREADY-EXISTS");  goto x_rc; }
348   if (errno != ENOENT) PE("check for existing database .main during creation");
349
350   for (toremove=toremoves; *toremove; toremove++) {
351     r= remove(pathbuf_sfx(&pb, *toremove));
352     if (r && errno != ENOENT)
353       PE("delete possible spurious file during creation");
354   }
355   
356   f= fopen(pathbuf_sfx(&pb, ".tmp"), "w");
357   if (!f) PE("create new database .tmp");  
358   r= putc('\n', f);  if (r==EOF) PE("write sentinel to new database .tmp");
359   r= fclose(f);  f=0;  if (r) PE("close new database .tmp during creation");
360
361   r= rename(pb.buf, pbmain.buf);
362   if (r) PE("install new database .tmp as .main (finalising creation)");
363   
364   rc= TCL_OK;
365
366  x_rc:
367   if (f) fclose(f);
368   maybe_close(lock_fd);
369   pathbuf_free(&pb);
370   pathbuf_free(&pbmain);
371   return rc;
372 }
373
374 /*---------- Info callbacks ----------*/
375
376 static int infocbv3(Tcl_Interp *ip, Rw *rw, const char *arg1,
377                     const char *arg2fmt, const char *arg3, va_list al) {
378   Tcl_Obj *aa[3];
379   int na;
380   char buf[200];
381   vsnprintf(buf, sizeof(buf), arg2fmt, al);
382
383   na= 0;
384   aa[na++]= cht_ret_string(ip, arg1);
385   aa[na++]= cht_ret_string(ip, buf);
386   if (arg3) aa[na++]= cht_ret_string(ip, arg3);
387   
388   return cht_scriptinv_invoke_fg(&rw->on_info, na, aa);
389 }
390   
391 static int infocb3(Tcl_Interp *ip, Rw *rw, const char *arg1,
392                    const char *arg2fmt, const char *arg3, ...) {
393   int rc;
394   va_list al;
395   va_start(al, arg3);
396   rc= infocbv3(ip,rw,arg1,arg2fmt,arg3,al);
397   va_end(al);
398   return rc;
399 }
400   
401 static int infocb(Tcl_Interp *ip, Rw *rw, const char *arg1,
402                   const char *arg2fmt, ...) {
403   int rc;
404   va_list al;
405   va_start(al, arg2fmt);
406   rc= infocbv3(ip,rw,arg1,arg2fmt,0,al);
407   va_end(al);
408   return rc;
409 }
410   
411 /*---------- Opening ----------*/
412
413 static int cdbinit(Tcl_Interp *ip, Rw *rw) {
414   /* On entry, cdb_fd >=0 but cdb is _undefined_/
415    * On exit, either cdb_fd<0 or cdb is initialised */
416   int r, rc;
417   
418   r= cdb_init(&rw->cdb, rw->cdb_fd);
419   if (r) {
420     rc= cht_posixerr(ip, errno, "failed to initialise cdb reader");
421     close(rw->cdb_fd);  rw->cdb_fd= -1;  return rc;
422   }
423   return TCL_OK;
424 }
425
426 int cht_do_cdbwr_open(ClientData cd, Tcl_Interp *ip, const char *pathb,
427                       Tcl_Obj *on_info, Tcl_Obj *on_lexminval,
428                       void **result) {
429   const Cdbwr_SubCommand *subcmd= cd;
430   int r, rc, mainfd=-1;
431   Rw *rw;
432   struct stat stab;
433   off_t logrecstart, logjunkpos;
434
435   rw= TALLOC(sizeof(*rw));
436   rw->ix= -1;
437   ht_setup(&rw->logincore);
438   cht_scriptinv_init(&rw->on_info);
439   cht_scriptinv_init(&rw->on_lexminval);
440   rw->cdb_fd= rw->lock_fd= -1;  rw->logfile= 0;
441   pathbuf_init(&rw->pbsome, pathb);
442   pathbuf_init(&rw->pbother, pathb);
443   rw->autocompact= 1;
444
445   rc= cht_scriptinv_set(&rw->on_info, ip, on_info, 0);
446   if (rc) goto x_rc;
447
448   rc= cht_scriptinv_set(&rw->on_lexminval, ip, on_lexminval, 0);
449   if (rc) goto x_rc;
450
451   mainfd= open(pathbuf_sfx(&rw->pbsome,".main"), O_RDONLY);
452   if (mainfd<0) PE("open existing database file .main");
453   rc= acquire_lock(ip, &rw->pbsome, &rw->lock_fd);  if (rc) goto x_rc;
454
455   r= fstat(mainfd, &stab);  if (r) PE("fstat .main");
456   rw->mainsz= stab.st_size;
457
458   rw->cdb_fd= open(pathbuf_sfx(&rw->pbsome,".cdb"), O_RDONLY);
459   if (rw->cdb_fd >=0) {
460     rc= cdbinit(ip, rw);  if (rc) goto x_rc;
461   } else if (errno == ENOENT) {
462     if (rw->mainsz > 1) {
463       rc= cht_staticerr(ip, ".cdb does not exist but .main is >1byte -"
464                         " .cdb must have been accidentally deleted!",
465                         "CDB CDBMISSING");
466       goto x_rc;
467     }
468     /* fine */
469   } else {
470     PE("open .cdb");
471   }
472
473   rw->logfile= fopen(pathbuf_sfx(&rw->pbsome,".jrn"), "r+");
474   if (!rw->logfile) {
475     if (errno != ENOENT) PE("failed to open .jrn during open");
476     rw->logfile= fopen(rw->pbsome.buf, "w");
477     if (!rw->logfile) PE("create .jrn during (clean) open");
478   } else { /* rw->logfile */
479     r= fstat(fileno(rw->logfile), &stab);
480     if (r==-1) PE("fstat .jrn during open");
481     rc= infocb(ip, rw, "open-dirty-start", "log=%luby",
482                (unsigned long)stab.st_size);
483     if (rc) goto x_rc;
484
485     for (;;) {
486       logrecstart= ftello(rw->logfile);
487       if (logrecstart < 0) PE("ftello .jrn during (dirty) open");
488       r= readstorelogrecord(rw->logfile, &rw->logincore, 0,0, ht_update);
489       if (ferror(rw->logfile)) {
490         rc= cht_posixerr(ip, errno, "error reading .jrn during (dirty) open");
491         goto x_rc;
492       }
493       if (r==-1) {
494         break;
495       } else if (r==-2 || r==-3) {
496         char buf[100];
497         logjunkpos= ftello(rw->logfile);
498         if(logjunkpos<0) PE("ftello .jrn during report of junk in dirty open");
499
500         snprintf(buf,sizeof(buf), "CDB SYNTAX LOG %lu %lu",
501                  (unsigned long)logjunkpos, (unsigned long)logrecstart);
502
503         if (!(subcmd->flags & RWSCF_OKJUNK)) {
504           Tcl_SetObjErrorCode(ip, Tcl_NewStringObj(buf,-1));
505           snprintf(buf,sizeof(buf),"%lu",(unsigned long)logjunkpos);
506           Tcl_ResetResult(ip);
507           Tcl_AppendResult(ip, "syntax error (junk) in .jrn during"
508                            " (dirty) open, at file position ", buf, (char*)0);
509           rc= TCL_ERROR;
510           goto x_rc;
511         }
512         rc= infocb3(ip, rw, "open-dirty-junk", "errorfpos=%luby", buf,
513                     (unsigned long)logjunkpos);
514         if (rc) goto x_rc;
515
516         r= fseeko(rw->logfile, logrecstart, SEEK_SET);
517         if (r) PE("failed to fseeko .jrn before junk during dirty open");
518
519         r= ftruncate(fileno(rw->logfile), logrecstart);
520         if (r) PE("ftruncate .jrn to chop junk during dirty open");
521       } else {
522         assert(!r);
523       }
524     }
525   }
526   /* now log is positioned for appending and everything is read */
527
528   *result= rw;
529   maybe_close(mainfd);
530   return TCL_OK;
531
532  x_rc:
533   rw_close(0,rw);
534   TFREE(rw);
535   maybe_close(mainfd);
536   return rc;
537 }
538
539 int cht_do_cdbwr_open_okjunk(ClientData cd, Tcl_Interp *ip, const char *pathb,
540                       Tcl_Obj *on_info, Tcl_Obj *on_lexminval,
541                       void **result) {
542   return cht_do_cdbwr_open(cd,ip,pathb,on_info,on_lexminval,result);
543 }
544
545 /*==================== COMPACTION ====================*/
546
547 struct ht_forall_ctx {
548   struct cdb_make cdbm;
549   FILE *mainfile;
550   long *reccount;
551   int lexminvall;
552   const char *lexminval; /* may be invalid if lexminvall <= 0 */
553 };
554
555 /*---------- helper functions ----------*/
556
557 static int expiredp(const HashValue *val, struct ht_forall_ctx *a) {
558   int r, l;
559   if (!val->len || a->lexminvall<=0) return 0;
560   l= val->len < a->lexminvall ? val->len : a->lexminvall;
561   r= memcmp(val->data, a->lexminval, l);
562   if (r>0) return 0;
563   if (r<0) return 1;
564   return val->len < a->lexminvall;
565 }
566
567 static int delete_ifexpired(const char *key, HashValue *val,
568                             struct ht_forall_ctx *a) {
569   if (!expiredp(val, a)) return 0;
570   val->len= 0;
571   /* we don't actually need to realloc it to free the memory because
572    * this will shortly all be deleted as part of the compaction */
573   return 0;
574 }
575
576 static int addto_cdb(const char *key, HashValue *val,
577                      struct ht_forall_ctx *a) {
578   return cdb_make_add(&a->cdbm, key, strlen(key), val->data, val->len);
579 }
580
581 static int addto_main(const char *key, HashValue *val,
582                       struct ht_forall_ctx *a) {
583   (*a->reccount)++;
584   return writerecord(a->mainfile, key, val);
585 }
586
587 /*---------- compact main entrypoint ----------*/
588
589 static int compact_core(Tcl_Interp *ip, Rw *rw, unsigned long logsz,
590                         long *reccount_r) {
591   /* creates new .cdb and .main
592    * closes logfile
593    * leaves .jrn with old data
594    * leaves cdb fd open onto old db
595    * leaves logincore full of crap
596    */
597   int r, rc;
598   int cdbfd, cdbmaking;
599   off_t errpos, newmainsz;
600   char buf[100];
601   Tcl_Obj *res;
602   struct ht_forall_ctx a;
603
604   a.mainfile= 0;
605   cdbfd= -1;
606   cdbmaking= 0;
607   *reccount_r= 0;
608   a.reccount= reccount_r;
609
610   r= fclose(rw->logfile);
611   if (r) { rc= cht_posixerr(ip, errno, "probable data loss!  failed to fclose"
612                             " logfile during compact");  goto x_rc; }
613   rw->logfile= 0;
614   
615   rc= infocb(ip, rw, "compact-start", "log=%luby main=%luby",
616              logsz, (unsigned long)rw->mainsz);
617   if (rc) goto x_rc;
618
619   if (cht_scriptinv_interp(&rw->on_lexminval)) {
620     rc= cht_scriptinv_invoke_fg(&rw->on_lexminval, 0,0);
621     if (rc) goto x_rc;
622
623     res= Tcl_GetObjResult(ip);  assert(res);
624     a.lexminval= Tcl_GetStringFromObj(res, &a.lexminvall);
625     assert(a.lexminval);
626
627     /* we rely not calling Tcl_Eval during the actual compaction;
628      * if we did Tcl_Eval then the interp result would be trashed.
629      */
630     rc= ht_forall(&rw->logincore, delete_ifexpired, &a);
631
632   } else {
633     a.lexminvall= 0;
634   }
635
636   /* merge unsuperseded records from main into hash table */
637
638   a.mainfile= fopen(pathbuf_sfx(&rw->pbsome,".main"), "r");
639   if (!a.mainfile) PE("failed to open .main for reading during compact");
640
641   for (;;) {
642     r= readstorelogrecord(a.mainfile, &rw->logincore,
643                           expiredp, &a,
644                           ht_maybeupdate);
645     if (ferror(a.mainfile)) { rc= cht_posixerr(ip, errno, "error reading"
646                          " .main during compact"); goto x_rc; }
647     if (r==-3) {
648       break;
649     } else if (r==-1 || r==-2) {
650       errpos= ftello(a.mainfile);
651       if (errpos<0) PE("ftello .main during report of syntax error");
652       snprintf(buf,sizeof(buf), "CDB %s MAIN %lu",
653                r==-1 ? "TRUNCATED" : "SYNTAX", (unsigned long)errpos);
654       Tcl_SetObjErrorCode(ip, Tcl_NewStringObj(buf,-1));
655       snprintf(buf,sizeof(buf), "%lu", (unsigned long)errpos);
656       Tcl_ResetResult(ip);
657       Tcl_AppendResult(ip,
658                        r==-1 ? "unexpected eof (truncated file)"
659                        " in .main during compact, at file position "
660                        : "syntax error"
661                        " in .main during compact, at file position ",
662                        buf, (char*)0);
663       rc= TCL_ERROR;
664       goto x_rc;
665     } else {
666       assert(!rc);
667     }
668   }
669   fclose(a.mainfile);
670   a.mainfile= 0;
671
672   /* create new cdb */
673
674   cdbfd= open(pathbuf_sfx(&rw->pbsome,".tmp"), O_WRONLY|O_CREAT|O_TRUNC, 0666);
675   if (cdbfd<0) PE("create .tmp for new cdb during compact");
676
677   r= cdb_make_start(&a.cdbm, cdbfd);
678   if (r) PE("cdb_make_start during compact");
679   cdbmaking= 1;
680
681   r= ht_forall(&rw->logincore, addto_cdb, &a);
682   if (r) PE("cdb_make_add during compact");
683
684   r= cdb_make_finish(&a.cdbm);
685   if(r) PE("cdb_make_finish during compact");
686   cdbmaking= 0;
687
688   r= fdatasync(cdbfd);  if (r) PE("fdatasync new cdb during compact");
689   r= close(cdbfd);  if (r) PE("close new cdb during compact");
690   cdbfd= -1;
691
692   r= rename(rw->pbsome.buf, pathbuf_sfx(&rw->pbother,".cdb"));
693   if (r) PE("install new .cdb during compact");
694
695   /* create new main */
696
697   a.mainfile= fopen(pathbuf_sfx(&rw->pbsome,".tmp"), "w");
698   if (!a.mainfile) PE("create .tmp for new main during compact");
699
700   r= ht_forall(&rw->logincore, addto_main, &a);
701   if (r) { rc= cht_posixerr(ip, r, "error writing to new .main"
702                             " during compact");  goto x_rc; }
703
704   r= putc('\n', a.mainfile);
705   if (r==EOF) PE("write trailing \n to main during compact");
706   
707   r= fflush(a.mainfile);  if (r) PE("fflush new main during compact");
708   r= fdatasync(fileno(a.mainfile));
709   if (r) PE("fdatasync new main during compact");
710
711   newmainsz= ftello(a.mainfile);
712   if (newmainsz<0) PE("ftello new main during compact");
713   
714   r= fclose(a.mainfile);  if (r) PE("fclose new main during compact");
715   a.mainfile= 0;
716
717   r= rename(rw->pbsome.buf, pathbuf_sfx(&rw->pbother,".main"));
718   if (r) PE("install new .main during compact");
719
720   rw->mainsz= newmainsz;
721
722   /* done! */
723   
724   rc= infocb(ip, rw, "compact-end", "main=%luby nrecs=%ld",
725              (unsigned long)rw->mainsz, *a.reccount);
726   if (rc) goto x_rc;
727
728   return rc;
729
730 x_rc:
731   if (a.mainfile) fclose(a.mainfile);
732   if (cdbmaking) cdb_make_finish(&a.cdbm);
733   maybe_close(cdbfd);
734   remove(pathbuf_sfx(&rw->pbsome,".tmp")); /* for tidyness */
735   return rc;
736 }
737
738 /*---------- Closing ----------*/
739
740 static int compact_forclose(Tcl_Interp *ip, Rw *rw, long *reccount_r) {
741   off_t logsz;
742   int r, rc;
743
744   logsz= ftello(rw->logfile);
745   if (logsz < 0) PE("ftello logfile (during tidy close)");
746
747   rc= compact_core(ip, rw, logsz, reccount_r);  if (rc) goto x_rc;
748
749   r= remove(pathbuf_sfx(&rw->pbsome,".jrn"));
750   if (r) PE("remove .jrn (during tidy close)");
751
752   return TCL_OK;
753
754 x_rc: return rc;
755 }
756   
757 int cht_do_cdbwr_close(ClientData cd, Tcl_Interp *ip, void *rw_v) {
758   Rw *rw= rw_v;
759   int rc, rc_close;
760   long reccount= -1;
761   off_t logsz;
762
763   if (rw->autocompact) rc= compact_forclose(ip, rw, &reccount);
764   else rc= TCL_OK;
765
766   if (!rc) {
767     if (rw->logfile) {
768       logsz= ftello(rw->logfile);
769       if (logsz < 0)
770         rc= cht_posixerr(ip, errno, "ftell logfile during close info");
771       else
772         rc= infocb(ip, rw, "close", "main=%luby log=%luby",
773                    rw->mainsz, logsz);
774     } else if (reccount>=0) {
775       rc= infocb(ip, rw, "close", "main=%luby nrecs=%ld",
776                  rw->mainsz, reccount);
777     } else {
778       rc= infocb(ip, rw, "close", "main=%luby", rw->mainsz);
779     }
780   }
781   rc_close= rw_close(ip,rw);
782   if (rc_close) rc= rc_close;
783   
784   cht_tabledataid_disposing(ip, rw_v, &cdbtcl_rwdatabases);
785   TFREE(rw);
786   return rc;
787 }
788
789 /*---------- Other compaction-related entrypoints ----------*/
790
791 static int compact_keepopen(Tcl_Interp *ip, Rw *rw, int force) {
792   off_t logsz;
793   long reccount;
794   int rc, r;
795
796   logsz= ftello(rw->logfile);
797   if (logsz < 0) return cht_posixerr(ip, errno, "ftell .jrn"
798                                        " during compact check or force");
799
800   if (!force && logsz < rw->mainsz / 3 + 1000) return TCL_OK;
801   /* Test case:                    ^^^ testing best value for this
802    *   main=9690434by nrecs=122803  read all in one go
803    *  no autocompact, :    6.96user 0.68system 0:08.93elapsed
804    *  auto, mulitplier 2:  7.10user 0.79system 0:09.54elapsed
805    *  auto, unity:         7.80user 0.98system 0:11.84elapsed
806    *  auto, divisor  2:    8.23user 1.05system 0:13.30elapsed
807    *  auto, divisor  3:    8.55user 1.12system 0:12.88elapsed
808    *  auto, divisor  5:    9.95user 1.43system 0:15.72elapsed
809    */
810
811   rc= compact_core(ip, rw, logsz, &reccount);  if (rc) goto x_rc;
812
813   maybe_close(rw->cdb_fd);
814   rw->cdb_fd= -1;
815   ht_destroy(&rw->logincore);
816   ht_setup(&rw->logincore);
817
818   rw->cdb_fd= open(pathbuf_sfx(&rw->pbsome,".cdb"), O_RDONLY);
819   if (rw->cdb_fd < 0) PE("reopen .cdb after compact");
820
821   rc= cdbinit(ip, rw);  if (rc) goto x_rc;
822
823   rw->logfile= fopen(pathbuf_sfx(&rw->pbsome,".jrn"), "w");
824   if (!rw->logfile) PE("reopen .jrn after compact");
825
826   r= fsync(fileno(rw->logfile));  if (r) PE("fsync .jrn after compact reopen");
827
828   return TCL_OK;
829
830 x_rc:
831   /* doom! all updates fail after this (because rw->logfile is 0), and
832    * we may be using a lot more RAM than would be ideal.  Program will
833    * have to reopen if it really wants sanity. */
834   return rc;
835 }
836
837 int cht_do_cdbwr_compact_force(ClientData cd, Tcl_Interp *ip, void *rw_v) {
838   return compact_keepopen(ip, rw_v, 1);
839 }
840 int cht_do_cdbwr_compact_check(ClientData cd, Tcl_Interp *ip, void *rw_v) {
841   return compact_keepopen(ip, rw_v, 0);
842 }
843
844 int cht_do_cdbwr_compact_explicit(ClientData cd, Tcl_Interp *ip, void *rw_v) {
845   Rw *rw= rw_v;
846   rw->autocompact= 0;
847   return TCL_OK;
848 }
849 int cht_do_cdbwr_compact_auto(ClientData cd, Tcl_Interp *ip, void *rw_v) {
850   Rw *rw= rw_v;
851   rw->autocompact= 1;
852   return TCL_OK;
853 }
854
855 /*---------- Updateing ----------*/
856
857 static int update(Tcl_Interp *ip, Rw *rw, const char *key,
858                   const Byte *data, int dlen) {
859   HashValue *val;
860   int rc, r;
861
862   if (!rw->logfile) return cht_staticerr
863     (ip, "previous compact failed; cdbwr must be closed and reopened "
864      "before any further updates", "CDB BROKEN");
865   
866   val= htv_prep(dlen);  assert(val);
867   memcpy(htv_fillptr(val), data, dlen);
868
869   r= writerecord(rw->logfile, key, val);
870   if (!r) r= fflush(rw->logfile);
871   if (r) PE("write update to logfile");
872
873   ht_update(&rw->logincore, key, val);
874
875   if (!rw->autocompact) return TCL_OK;
876   return compact_keepopen(ip, rw, 0);
877
878  x_rc:
879   TFREE(val);
880   return rc;
881 }  
882
883 int cht_do_cdbwr_update(ClientData cd, Tcl_Interp *ip,
884                         void *rw_v, const char *key, Tcl_Obj *value) {
885   int dlen;
886   const char *data;
887   data= Tcl_GetStringFromObj(value, &dlen);  assert(data);
888   return update(ip, rw_v, key, data, dlen);
889 }
890
891 int cht_do_cdbwr_update_hb(ClientData cd, Tcl_Interp *ip,
892                            void *rw_v, const char *key, HBytes_Value value) {
893   return update(ip, rw_v, key, cht_hb_data(&value), cht_hb_len(&value));
894 }
895
896 int cht_do_cdbwr_delete(ClientData cd, Tcl_Interp *ip, void *rw_v,
897                         const char *key) {
898   return update(ip, rw_v, key, 0, 0);
899 }
900
901 /*---------- Lookups ----------*/
902
903 static int lookup_rw(Tcl_Interp *ip, void *rw_v, const char *key,
904                     const Byte **data_r, int *len_r /* -1 => notfound */) {
905   Rw *rw= rw_v;
906   const HashValue *val;
907
908   val= ht_lookup(&rw->logincore, key);
909   if (val) {
910     if (val->len) { *data_r= val->data; *len_r= val->len; return TCL_OK; }
911     else { *data_r= 0; *len_r= -1; return TCL_OK; }
912   }
913
914   return cht_cdb_lookup_cdb(ip, &rw->cdb, key, strlen(key), data_r, len_r);
915
916
917 int cht_do_cdbwr_lookup(ClientData cd, Tcl_Interp *ip, void *rw_v,
918                         const char *key, Tcl_Obj *def,
919                         Tcl_Obj **result) {
920   const Byte *data;
921   int dlen, r;
922   
923   r= lookup_rw(ip, rw_v, key, &data, &dlen);  if (r) return r;
924   return cht_cdb_donesomelookup(ip, rw_v, def, result, data, dlen,
925                                 cht_cdb_storeanswer_string);
926 }
927   
928 int cht_do_cdbwr_lookup_hb(ClientData cd, Tcl_Interp *ip, void *rw_v,
929                            const char *key, Tcl_Obj *def,
930                            Tcl_Obj **result) {
931   const Byte *data;
932   int dlen, r;
933   
934   r= lookup_rw(ip, rw_v, key, &data, &dlen);  if (r) return r;
935   return cht_cdb_donesomelookup(ip, rw_v, def, result, data, dlen,
936                                 cht_cdb_storeanswer_hb);
937 }