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