Line data Source code
1 : /*
2 : * SPDX-License-Identifier: MPL-2.0
3 : *
4 : * This Source Code Form is subject to the terms of the Mozilla Public
5 : * License, v. 2.0. If a copy of the MPL was not distributed with this
6 : * file, You can obtain one at http://mozilla.org/MPL/2.0/.
7 : *
8 : * Copyright 2024 MonetDB Foundation;
9 : * Copyright August 2008 - 2023 MonetDB B.V.;
10 : * Copyright 1997 - July 2008 CWI.
11 : */
12 :
13 : #include "monetdb_config.h"
14 : #include "gdk.h"
15 : #include "gdk_private.h"
16 : #include "gdk_logger.h"
17 : #include "gdk_logger_internals.h"
18 : #include "mutils.h"
19 : #include <string.h>
20 :
21 : static gdk_return log_add_bat(logger *lg, BAT *b, log_id id, int tid);
22 : static gdk_return log_del_bat(logger *lg, log_bid bid);
23 : /*
24 : * The logger uses a directory to store its log files. One master log
25 : * file stores information about the version of the logger and the
26 : * type mapping it uses. This file is a simple ascii file with the
27 : * following format:
28 : * {6DIGIT-VERSION\n[id,type_name\n]*}
29 : * The transaction log files have a binary format.
30 : */
31 :
32 : #define LOG_START 0
33 : #define LOG_END 1
34 : #define LOG_UPDATE_CONST 2
35 : #define LOG_UPDATE_BULK 3
36 : #define LOG_UPDATE 4
37 : #define LOG_CREATE 5
38 : #define LOG_DESTROY 6
39 : #define LOG_SEQ 7
40 : #define LOG_CLEAR 8 /* DEPRECATED */
41 : #define LOG_BAT_GROUP 9
42 :
43 : #ifdef NATIVE_WIN32
44 : #define getfilepos _ftelli64
45 : #else
46 : #ifdef HAVE_FSEEKO
47 : #define getfilepos ftello
48 : #else
49 : #define getfilepos ftell
50 : #endif
51 : #endif
52 :
53 : #define BATSIZE 0
54 :
55 : #define LOG_DISABLED(lg) ((lg)->debug&128 || (lg)->inmemory || (lg)->flushnow)
56 :
57 : static const char *log_commands[] = {
58 : "LOG_START",
59 : "LOG_END",
60 : "LOG_UPDATE_CONST",
61 : "LOG_UPDATE_BULK",
62 : "LOG_UPDATE",
63 : "LOG_CREATE",
64 : "LOG_DESTROY",
65 : "LOG_SEQ",
66 : "", /* LOG_CLEAR IS DEPRECATED */
67 : "LOG_BAT_GROUP",
68 : };
69 :
70 : typedef struct logaction {
71 : int type; /* type of change */
72 : lng nr;
73 : int tt;
74 : lng id;
75 : lng offset;
76 : log_id cid; /* id of object */
77 : BAT *b; /* temporary bat with changes */
78 : BAT *uid; /* temporary bat with bun positions to update */
79 : } logaction;
80 :
81 : /* during the recover process a number of transactions could be active */
82 : typedef struct trans {
83 : int tid; /* transaction id */
84 : int sz; /* sz of the changes array */
85 : int nr; /* nr of changes */
86 :
87 : logaction *changes;
88 :
89 : struct trans *tr;
90 : } trans;
91 :
92 : typedef struct logformat_t {
93 : bte flag;
94 : int id;
95 : } logformat;
96 :
97 : typedef enum { LOG_OK, LOG_EOF, LOG_ERR } log_return;
98 :
99 : static gdk_return bm_commit(logger *lg, logged_range *pending, uint32_t *updated, BUN maxupdated);
100 : static gdk_return tr_grow(trans *tr);
101 :
102 : #define log_lock(lg) MT_lock_set(&(lg)->lock)
103 : #define log_unlock(lg) MT_lock_unset(&(lg)->lock)
104 :
105 : static inline bte
106 933685 : find_type(logger *lg, int tpe)
107 : {
108 933685 : assert(tpe >= 0 && tpe < MAXATOMS);
109 933685 : return lg->type_id[tpe];
110 : }
111 :
112 : static inline int
113 545705 : find_type_nr(logger *lg, bte tpe)
114 : {
115 545705 : int nr = lg->type_nr[tpe < 0 ? 256 + tpe : tpe];
116 545705 : if (nr == 255)
117 0 : return -1;
118 : return nr;
119 : }
120 :
121 : static BUN
122 420680 : log_find(BAT *b, BAT *d, int val)
123 : {
124 420680 : BUN p;
125 :
126 420680 : assert(b->ttype == TYPE_int);
127 420680 : assert(d->ttype == TYPE_oid);
128 420680 : BATiter bi = bat_iterator(b);
129 420680 : if (BAThash(b) == GDK_SUCCEED) {
130 420680 : MT_rwlock_rdlock(&b->thashlock);
131 701678 : HASHloop_int(bi, b->thash, p, &val) {
132 184184 : oid pos = p;
133 184184 : if (BUNfnd(d, &pos) == BUN_NONE) {
134 184184 : MT_rwlock_rdunlock(&b->thashlock);
135 184184 : bat_iterator_end(&bi);
136 184184 : return p;
137 : }
138 : }
139 236496 : MT_rwlock_rdunlock(&b->thashlock);
140 : } else { /* unlikely: BAThash failed */
141 0 : int *t = (int *) bi.base;
142 :
143 0 : for (p = 0; p < bi.count; p++) {
144 0 : if (t[p] == val) {
145 0 : oid pos = p;
146 0 : if (BUNfnd(d, &pos) == BUN_NONE) {
147 0 : bat_iterator_end(&bi);
148 0 : return p;
149 : }
150 : }
151 : }
152 : }
153 236496 : bat_iterator_end(&bi);
154 236496 : return BUN_NONE;
155 : }
156 :
157 : static log_bid
158 646804 : internal_find_bat(logger *lg, log_id id, int tid)
159 : {
160 646804 : BUN p;
161 :
162 646804 : if (BAThash(lg->catalog_id) == GDK_SUCCEED) {
163 646804 : BATiter cni = bat_iterator(lg->catalog_id);
164 646804 : MT_rwlock_rdlock(&cni.b->thashlock);
165 646804 : if (tid < 0) {
166 390466 : HASHloop_int(cni, cni.b->thash, p, &id) {
167 204851 : oid pos = p;
168 204851 : if (BUNfnd(lg->dcatalog, &pos) == BUN_NONE) {
169 204851 : MT_rwlock_rdunlock(&cni.b->thashlock);
170 204851 : bat_iterator_end(&cni);
171 204851 : return *(log_bid *) Tloc(lg->catalog_bid, p);
172 : }
173 : }
174 : } else {
175 361256 : BUN cp = BUN_NONE;
176 1469508 : HASHloop_int(cni, cni.b->thash, p, &id) {
177 1073174 : lng lid = *(lng *) Tloc(lg->catalog_lid, p);
178 1073174 : if (lid != lng_nil && lid <= tid) {
179 : break;
180 : }
181 : cp = p;
182 : }
183 361256 : if (cp != BUN_NONE) {
184 361008 : MT_rwlock_rdunlock(&cni.b->thashlock);
185 361008 : bat_iterator_end(&cni);
186 361008 : return *(log_bid *) Tloc(lg->catalog_bid, cp);
187 : }
188 : }
189 80945 : MT_rwlock_rdunlock(&cni.b->thashlock);
190 80945 : bat_iterator_end(&cni);
191 80945 : return 0; /* not found */
192 : }
193 : return -1; /* error creating hash */
194 : }
195 :
196 : static inline void
197 14321 : logbat_destroy(BAT *b)
198 : {
199 17882 : BBPreclaim(b);
200 3427 : }
201 :
202 : static BAT *
203 15365 : logbat_new(int tt, BUN size, role_t role)
204 : {
205 15365 : BAT *nb = COLnew(0, tt, size, role);
206 :
207 15365 : if (nb) {
208 15365 : BBP_pid(nb->batCacheid) = 0;
209 15365 : if (role == PERSISTENT) {
210 9645 : BATmode(nb, false);
211 9645 : nb = BATsetaccess(nb, BAT_READ);
212 : }
213 : } else {
214 0 : TRC_CRITICAL(GDK, "creating new BAT[%s]#" BUNFMT " failed\n", ATOMname(tt), size);
215 : }
216 15365 : return nb;
217 : }
218 :
219 : static bool
220 757596 : log_read_format(logger *lg, logformat *data)
221 : {
222 757596 : assert(!lg->inmemory);
223 757596 : if (mnstr_read(lg->input_log, &data->flag, 1, 1) == 1) {
224 745181 : if (mnstr_readInt(lg->input_log, &data->id) == 1)
225 : return true;
226 : /* could only read part, so complain */
227 0 : TRC_CRITICAL(GDK, "read failed\n");
228 : }
229 : return false;
230 : }
231 :
232 : static gdk_return
233 751958 : log_write_format(logger *lg, logformat *data)
234 : {
235 751958 : assert(data->id || data->flag);
236 751958 : assert(!lg->inmemory);
237 751958 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
238 1503916 : if (mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR &&
239 1503916 : mnstr_write(lg->current->output_log, &data->flag, 1, 1) == 1 &&
240 751958 : mnstr_writeInt(lg->current->output_log, data->id))
241 : return GDK_SUCCEED;
242 0 : TRC_CRITICAL(GDK, "write failed\n");
243 0 : return GDK_FAIL;
244 : }
245 :
246 : static log_return
247 2764 : log_read_seq(logger *lg, logformat *l)
248 : {
249 2764 : int seq = l->id;
250 2764 : lng val;
251 2764 : BUN p;
252 :
253 2764 : assert(!lg->inmemory);
254 2764 : if (mnstr_readLng(lg->input_log, &val) != 1) {
255 0 : TRC_CRITICAL(GDK, "read failed\n");
256 0 : return LOG_EOF;
257 : }
258 2764 : if (lg->flushing)
259 : return LOG_OK;
260 :
261 60 : if ((p = log_find(lg->seqs_id, lg->dseqs, seq)) != BUN_NONE &&
262 59 : p >= lg->seqs_id->batInserted) {
263 39 : assert(lg->seqs_val->hseqbase == 0);
264 39 : if (BUNreplace(lg->seqs_val, p, &val, true) != GDK_SUCCEED) {
265 0 : TRC_CRITICAL(GDK, "replace of %s_seqs_val failed\n", lg->fn);
266 0 : return LOG_ERR;
267 : }
268 : } else {
269 20 : if (p != BUN_NONE) {
270 20 : oid pos = p;
271 20 : if (BUNappend(lg->dseqs, &pos, true) != GDK_SUCCEED) {
272 0 : TRC_CRITICAL(GDK, "append to %s_dseqs failed\n", lg->fn);
273 0 : return LOG_ERR;
274 : }
275 : }
276 42 : if (BUNappend(lg->seqs_id, &seq, true) != GDK_SUCCEED ||
277 21 : BUNappend(lg->seqs_val, &val, true) != GDK_SUCCEED) {
278 0 : TRC_CRITICAL(GDK, "append to %s_seqs_val/id failed\n", lg->fn);
279 0 : return LOG_ERR;
280 : }
281 : }
282 : return LOG_OK;
283 : }
284 :
285 : #if 0
286 : static gdk_return
287 : log_write_id(logger *lg, int id)
288 : {
289 : assert(!lg->inmemory);
290 : assert(id >= 0);
291 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
292 : if (mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR &&
293 : mnstr_writeInt(lg->current->output_log, id))
294 : return GDK_SUCCEED;
295 : TRC_CRITICAL(GDK, "write failed\n");
296 : return GDK_FAIL;
297 : }
298 :
299 : static log_return
300 : log_read_id(logger *lg, log_id *id)
301 : {
302 : assert(!lg->inmemory);
303 : if (mnstr_readInt(lg->input_log, id) != 1) {
304 : TRC_CRITICAL(GDK, "read failed\n");
305 : return LOG_EOF;
306 : }
307 : return LOG_OK;
308 : }
309 : #endif
310 :
311 : static log_return
312 19080 : string_reader(logger *lg, BAT *b, lng nr)
313 : {
314 19080 : size_t sz = 0;
315 19080 : lng SZ = 0;
316 19080 : log_return res = LOG_OK;
317 :
318 38243 : while (nr && res == LOG_OK) {
319 19163 : if (mnstr_readLng(lg->input_log, &SZ) != 1) {
320 0 : TRC_CRITICAL(GDK, "read failed\n");
321 0 : return LOG_EOF;
322 : }
323 19163 : sz = (size_t) SZ;
324 19163 : char *buf = lg->rbuf;
325 19163 : if (lg->rbufsize < sz) {
326 2 : if (!(buf = GDKrealloc(lg->rbuf, sz))) {
327 0 : TRC_CRITICAL(GDK, "couldn't grow string buffer\n");
328 0 : return LOG_ERR;
329 : }
330 2 : lg->rbuf = buf;
331 2 : lg->rbufsize = sz;
332 : }
333 :
334 19163 : if (mnstr_read(lg->input_log, buf, sz, 1) != 1) {
335 0 : TRC_CRITICAL(GDK, "read failed\n");
336 0 : return LOG_EOF;
337 : }
338 : /* handle strings */
339 : char *t = buf;
340 : /* chunked */
341 : #define CHUNK_SIZE 1024
342 : char *strings[CHUNK_SIZE];
343 : int cur = 0;
344 :
345 92725 : for (; nr > 0 && res == LOG_OK && t < (buf + sz); nr--) {
346 73562 : strings[cur++] = t;
347 73562 : if (cur == CHUNK_SIZE &&
348 4 : b &&
349 4 : BUNappendmulti(b, strings, cur, true) != GDK_SUCCEED) {
350 0 : TRC_CRITICAL(GDK, "append to string bat failed\n");
351 0 : res = LOG_ERR;
352 : }
353 73562 : if (cur == CHUNK_SIZE)
354 33 : cur = 0;
355 : /* find next */
356 5577346 : while (*t)
357 5503784 : t++;
358 73562 : t++;
359 : }
360 19163 : if (cur &&
361 294 : b &&
362 294 : BUNappendmulti(b, strings, cur, true) != GDK_SUCCEED) {
363 0 : TRC_CRITICAL(GDK, "append to string bat failed\n");
364 0 : res = LOG_ERR;
365 : }
366 : }
367 : return res;
368 : }
369 :
370 :
371 : struct offset {
372 : lng os; /* offset within source BAT in logfile */
373 : lng nr; /* number of values to be copied */
374 : lng od; /* offset within destination BAT in database */
375 : };
376 :
377 : static log_return
378 389229 : log_read_updates(logger *lg, trans *tr, logformat *l, log_id id, BAT **cands)
379 : {
380 389229 : log_return res = LOG_OK;
381 389229 : lng nr, pnr;
382 389229 : bte type_id = -1;
383 389229 : int tpe;
384 :
385 389229 : assert(!lg->inmemory);
386 389229 : TRC_DEBUG(WAL, "found %d %s", id, l->flag == LOG_UPDATE ? "update" : "update_buld");
387 :
388 778458 : if (mnstr_readLng(lg->input_log, &nr) != 1 ||
389 389229 : mnstr_read(lg->input_log, &type_id, 1, 1) != 1) {
390 0 : TRC_CRITICAL(GDK, "read failed\n");
391 0 : return LOG_EOF;
392 : }
393 :
394 389229 : pnr = nr;
395 389229 : tpe = find_type_nr(lg, type_id);
396 389229 : if (tpe >= 0) {
397 389229 : BAT *uid = NULL;
398 389229 : BAT *r = NULL;
399 389229 : void *(*rt)(ptr, size_t *, stream *, size_t) = BATatoms[tpe].atomRead;
400 389229 : lng offset;
401 :
402 389229 : assert(nr <= (lng) BUN_MAX);
403 389229 : if (!lg->flushing && l->flag == LOG_UPDATE) {
404 0 : uid = COLnew(0, TYPE_oid, (BUN) nr, PERSISTENT);
405 0 : if (uid == NULL) {
406 0 : TRC_CRITICAL(GDK, "creating bat failed\n");
407 28266 : return LOG_ERR;
408 : }
409 : }
410 :
411 389229 : if (l->flag == LOG_UPDATE_CONST) {
412 121511 : if (mnstr_readLng(lg->input_log, &offset) != 1) {
413 0 : TRC_CRITICAL(GDK, "read failed\n");
414 0 : return LOG_EOF;
415 : }
416 121511 : if (cands) {
417 : /* This const range actually represents a segment of candidates corresponding to updated bat entries */
418 :
419 28266 : if (BATcount(*cands) == 0 || lg->flushing) {
420 : /* when flushing, we only need the offset and count of the last segment of inserts. */
421 28266 : assert((*cands)->ttype == TYPE_void);
422 28266 : BATtseqbase(*cands, (oid) offset);
423 28266 : BATsetcount(*cands, (BUN) nr);
424 0 : } else if (!lg->flushing) {
425 0 : assert(BATcount(*cands) > 0);
426 0 : BAT *dense = BATdense(0, (oid) offset, (BUN) nr);
427 0 : BAT *newcands = NULL;
428 0 : if (!dense) {
429 0 : TRC_CRITICAL(GDK, "creating bat failed\n");
430 0 : res = LOG_ERR;
431 0 : } else if ((*cands)->ttype == TYPE_void) {
432 0 : if ((newcands = BATmergecand(*cands, dense))) {
433 0 : BBPreclaim(*cands);
434 0 : *cands = newcands;
435 : } else {
436 0 : TRC_CRITICAL(GDK, "creating bat failed\n");
437 0 : res = LOG_ERR;
438 : }
439 : } else {
440 0 : assert((*cands)->ttype == TYPE_oid);
441 0 : assert(BATcount(*cands) > 0);
442 0 : if (BATappend(*cands, dense, NULL, true) != GDK_SUCCEED) {
443 0 : TRC_CRITICAL(GDK, "appending to bat failed\n");
444 0 : res = LOG_ERR;
445 : }
446 : }
447 0 : BBPreclaim(dense);
448 : }
449 :
450 : /* We have to read the value to update the read cursor */
451 28266 : size_t tlen = lg->rbufsize;
452 28266 : void *t = rt(lg->rbuf, &tlen, lg->input_log, 1);
453 28266 : if (t == NULL) {
454 0 : TRC_CRITICAL(GDK, "read failed\n");
455 0 : res = LOG_EOF;
456 : }
457 28266 : return res;
458 : }
459 : }
460 :
461 360963 : if (!lg->flushing) {
462 1258 : r = COLnew(0, tpe, (BUN) nr, PERSISTENT);
463 1258 : if (r == NULL) {
464 0 : if (uid)
465 0 : BBPreclaim(uid);
466 0 : return LOG_ERR;
467 : }
468 : }
469 :
470 360963 : if (l->flag == LOG_UPDATE_CONST) {
471 93245 : size_t tlen = lg->rbufsize;
472 93245 : void *t = rt(lg->rbuf, &tlen, lg->input_log, 1);
473 93245 : if (t == NULL) {
474 0 : TRC_CRITICAL(GDK, "read failed\n");
475 0 : res = LOG_EOF;
476 : } else {
477 93245 : lg->rbuf = t;
478 93245 : lg->rbufsize = tlen;
479 93245 : if (r) {
480 8130 : for (BUN p = 0; p < (BUN) nr; p++) {
481 7886 : if (BUNappend(r, t, true) != GDK_SUCCEED) {
482 0 : TRC_CRITICAL(GDK, "append to bat failed\n");
483 0 : res = LOG_ERR;
484 : }
485 : }
486 : }
487 : }
488 267718 : } else if (l->flag == LOG_UPDATE_BULK) {
489 267700 : if (mnstr_readLng(lg->input_log, &offset) != 1) {
490 0 : if (r)
491 0 : BBPreclaim(r);
492 0 : TRC_CRITICAL(GDK, "read failed\n");
493 0 : return LOG_EOF;
494 : }
495 267700 : if (tpe == TYPE_msk) {
496 0 : if (r) {
497 0 : if (mnstr_readIntArray(lg->input_log, Tloc(r, 0), (size_t) ((nr + 31) / 32)))
498 0 : BATsetcount(r, (BUN) nr);
499 : else {
500 0 : TRC_CRITICAL(GDK, "read failed\n");
501 0 : res = LOG_EOF;
502 : }
503 : } else {
504 0 : size_t tlen = lg->rbufsize / sizeof(int);
505 0 : size_t cnt = 0, snr = (size_t) nr;
506 0 : snr = (snr + 31) / 32;
507 0 : assert(tlen);
508 0 : for (; res == LOG_OK && snr > 0; snr -= cnt) {
509 0 : cnt = snr > tlen ? tlen : snr;
510 0 : if (!mnstr_readIntArray(lg->input_log, lg->rbuf, cnt)) {
511 0 : TRC_CRITICAL(GDK, "read failed\n");
512 0 : res = LOG_EOF;
513 : }
514 : }
515 : }
516 : } else {
517 267700 : if (!ATOMvarsized(tpe)) {
518 248044 : size_t cnt = 0, snr = (size_t) nr;
519 248044 : size_t tlen = lg->rbufsize / ATOMsize(tpe), ntlen = lg->rbufsize;
520 248044 : assert(tlen);
521 : /* read in chunks of max
522 : * BUFSIZE/width rows */
523 508757 : for (; res == LOG_OK && snr > 0; snr -= cnt) {
524 260713 : cnt = snr > tlen ? tlen : snr;
525 260713 : void *t = rt(lg->rbuf, &ntlen, lg->input_log, cnt);
526 :
527 260713 : if (t == NULL) {
528 : res = LOG_EOF;
529 : break;
530 : }
531 260713 : assert(t == lg->rbuf);
532 260713 : if (r && BUNappendmulti(r, t, cnt, true) != GDK_SUCCEED) {
533 0 : TRC_CRITICAL(GDK, "append to bat failed\n");
534 0 : res = LOG_ERR;
535 : }
536 : }
537 19656 : } else if (tpe == TYPE_str) {
538 : /* efficient string */
539 19074 : res = string_reader(lg, r, nr);
540 : } else {
541 1206 : for (; res == LOG_OK && nr > 0; nr--) {
542 624 : size_t tlen = lg->rbufsize;
543 624 : void *t = rt(lg->rbuf, &tlen, lg->input_log, 1);
544 :
545 624 : if (t == NULL) {
546 : /* see if failure was due to
547 : * malloc or something less
548 : * serious (in the current
549 : * context) */
550 0 : if (strstr(GDKerrbuf, "alloc") == NULL)
551 : res = LOG_EOF;
552 : else
553 0 : res = LOG_ERR;
554 0 : TRC_CRITICAL(GDK, "read failed\n");
555 : } else {
556 624 : lg->rbuf = t;
557 624 : lg->rbufsize = tlen;
558 624 : if (r && BUNappend(r, t, true) != GDK_SUCCEED) {
559 0 : TRC_CRITICAL(GDK, "append to bat failed\n");
560 0 : res = LOG_ERR;
561 : }
562 : }
563 : }
564 : }
565 : }
566 : } else {
567 18 : void *(*rh)(ptr, size_t *, stream *, size_t) = BATatoms[TYPE_oid].atomRead;
568 18 : void *hv = ATOMnil(TYPE_oid);
569 18 : offset = 0;
570 :
571 18 : if (hv == NULL) {
572 0 : TRC_CRITICAL(GDK, "read failed\n");
573 0 : res = LOG_EOF;
574 : }
575 1071 : for (; res == LOG_OK && nr > 0; nr--) {
576 1053 : size_t hlen = sizeof(oid);
577 1053 : void *h = rh(hv, &hlen, lg->input_log, 1);
578 1053 : assert(hlen == sizeof(oid));
579 1053 : assert(h == hv);
580 1053 : if ((uid && BUNappend(uid, h, true) != GDK_SUCCEED)) {
581 0 : TRC_CRITICAL(GDK, "append to bat failed\n");
582 0 : res = LOG_ERR;
583 : }
584 : }
585 18 : nr = pnr;
586 18 : if (tpe == TYPE_msk) {
587 0 : if (r) {
588 0 : if (mnstr_readIntArray(lg->input_log, Tloc(r, 0), (size_t) ((nr + 31) / 32)))
589 0 : BATsetcount(r, (BUN) nr);
590 : else {
591 0 : TRC_CRITICAL(GDK, "read failed\n");
592 0 : res = LOG_EOF;
593 : }
594 : } else {
595 0 : for (lng i = 0; i < nr; i += 32) {
596 0 : int v;
597 0 : switch (mnstr_readInt(lg->input_log, &v)) {
598 0 : case 1:
599 0 : continue;
600 : case 0:
601 : res = LOG_EOF;
602 : break;
603 : default:
604 : res = LOG_ERR;
605 : break;
606 : }
607 0 : TRC_CRITICAL(GDK, "read failed\n");
608 0 : break;
609 : }
610 : }
611 18 : } else if (tpe == TYPE_str) {
612 : /* efficient string */
613 6 : res = string_reader(lg, r, nr);
614 : } else {
615 58 : for (; res == LOG_OK && nr > 0; nr--) {
616 46 : size_t tlen = lg->rbufsize;
617 46 : void *t = rt(lg->rbuf, &tlen, lg->input_log, 1);
618 :
619 46 : if (t == NULL) {
620 0 : if (strstr(GDKerrbuf, "malloc") == NULL)
621 : res = LOG_EOF;
622 : else
623 0 : res = LOG_ERR;
624 0 : TRC_CRITICAL(GDK, "read failed\n");
625 : } else {
626 46 : lg->rbuf = t;
627 46 : lg->rbufsize = tlen;
628 46 : if ((r && BUNappend(r, t, true) != GDK_SUCCEED)) {
629 0 : TRC_CRITICAL(GDK, "append to bat failed\n");
630 0 : res = LOG_ERR;
631 : }
632 : }
633 : }
634 : }
635 18 : GDKfree(hv);
636 : }
637 :
638 360963 : if (res == LOG_OK) {
639 360963 : if (tr_grow(tr) == GDK_SUCCEED) {
640 360963 : tr->changes[tr->nr].type = l->flag;
641 360963 : if (l->flag == LOG_UPDATE_BULK && offset == -1) {
642 139182 : assert(cands); /* bat r is part of a group of bats logged together. */
643 139182 : struct canditer ci;
644 139182 : canditer_init(&ci, NULL, *cands);
645 139182 : const oid first = canditer_peek(&ci);
646 139182 : const oid last = canditer_last(&ci);
647 139182 : offset = (lng) first;
648 139182 : pnr = (lng) (last - first) + 1;
649 139182 : if (!lg->flushing) {
650 911 : assert(uid == NULL);
651 911 : uid = *cands;
652 911 : BBPfix((*cands)->batCacheid);
653 911 : tr->changes[tr->nr].type = LOG_UPDATE;
654 : }
655 : }
656 360963 : if (l->flag == LOG_UPDATE_CONST) {
657 93245 : assert(!cands); /* TODO: This might change in the future. */
658 93245 : tr->changes[tr->nr].type = LOG_UPDATE_BULK;
659 : }
660 360963 : tr->changes[tr->nr].nr = pnr;
661 360963 : tr->changes[tr->nr].tt = tpe;
662 360963 : tr->changes[tr->nr].cid = id;
663 360963 : tr->changes[tr->nr].offset = offset;
664 360963 : tr->changes[tr->nr].b = r;
665 360963 : tr->changes[tr->nr].uid = uid;
666 360963 : tr->nr++;
667 : } else {
668 0 : TRC_CRITICAL(GDK, "memory allocation failed\n");
669 0 : res = LOG_ERR;
670 : }
671 : }
672 360963 : if (res != LOG_OK) {
673 0 : if (r)
674 0 : BBPreclaim(r);
675 0 : if (cands && uid)
676 0 : BBPunfix((*cands)->batCacheid);
677 0 : else if (uid)
678 0 : BBPreclaim(uid);
679 : }
680 : } else {
681 : /* bat missing ERROR or ignore ? currently error. */
682 0 : TRC_CRITICAL(GDK, "unknown type\n");
683 0 : res = LOG_ERR;
684 : }
685 : return res;
686 : }
687 :
688 :
689 : static gdk_return
690 579412 : la_bat_update_count(logger *lg, log_id id, lng cnt, int tid)
691 : {
692 579412 : BATiter cni = bat_iterator_nolock(lg->catalog_id);
693 :
694 579412 : if (BAThash(lg->catalog_id) == GDK_SUCCEED) {
695 579412 : MT_rwlock_rdlock(&cni.b->thashlock);
696 579412 : BUN p, cp = BUN_NONE;
697 :
698 2079645 : HASHloop_int(cni, cni.b->thash, p, &id) {
699 1293756 : lng lid = *(lng *) Tloc(lg->catalog_lid, p);
700 :
701 1293756 : if (lid != lng_nil && lid <= tid)
702 : break;
703 : cp = p;
704 : }
705 579412 : if (cp != BUN_NONE) {
706 579212 : lng ocnt = *(lng *) Tloc(lg->catalog_cnt, cp);
707 579212 : assert(lg->catalog_cnt->hseqbase == 0);
708 579212 : if (ocnt < cnt && BUNreplace(lg->catalog_cnt, cp, &cnt, false) != GDK_SUCCEED) {
709 0 : MT_rwlock_rdunlock(&cni.b->thashlock);
710 0 : return GDK_FAIL;
711 : }
712 : }
713 579412 : MT_rwlock_rdunlock(&cni.b->thashlock);
714 579412 : return GDK_SUCCEED;
715 : }
716 : return GDK_FAIL;
717 : }
718 :
719 : static gdk_return
720 360963 : la_bat_updates(logger *lg, logaction *la, int tid)
721 : {
722 360963 : log_bid bid = internal_find_bat(lg, la->cid, tid);
723 360963 : BAT *b = NULL;
724 :
725 360963 : if (bid < 0)
726 : return GDK_FAIL;
727 360963 : if (!bid) {
728 : /* object already gone, nothing needed */
729 : return GDK_SUCCEED;
730 : }
731 :
732 360960 : if (!lg->flushing) {
733 1258 : b = BATdescriptor(bid);
734 1258 : if (b == NULL)
735 : return GDK_FAIL;
736 : }
737 360960 : BUN cnt = 0;
738 360960 : if (la->type == LOG_UPDATE_BULK) {
739 360031 : if (!lg->flushing) {
740 347 : cnt = BATcount(b);
741 347 : int is_msk = (b->ttype == TYPE_msk);
742 : /* handle offset 0 ie clear */
743 347 : if ( /* DISABLES CODE */ (0) && la->offset == 0 && cnt)
744 : BATclear(b, true);
745 : /* handle offset */
746 347 : if (cnt <= (BUN) la->offset) {
747 222 : msk t = 1;
748 222 : if (cnt < (BUN) la->offset) { /* insert nils */
749 0 : const void *tv = (is_msk) ? &t : ATOMnilptr(b->ttype);
750 0 : lng i, d = la->offset - BATcount(b);
751 0 : for (i = 0; i < d; i++) {
752 0 : if (BUNappend(b, tv, true) != GDK_SUCCEED) {
753 0 : logbat_destroy(b);
754 0 : return GDK_FAIL;
755 : }
756 : }
757 : }
758 222 : if (BATcount(b) == (BUN) la->offset && BATappend(b, la->b, NULL, true) != GDK_SUCCEED) {
759 0 : logbat_destroy(b);
760 0 : return GDK_FAIL;
761 : }
762 : } else {
763 125 : BATiter vi = bat_iterator(la->b);
764 125 : BUN p, q;
765 :
766 2167 : for (p = 0, q = (BUN) la->offset; p < (BUN) la->nr; p++, q++) {
767 2042 : const void *t = BUNtail(vi, p);
768 :
769 2042 : if (q < cnt) {
770 2041 : if (b->tnosorted == q)
771 7 : b->tnosorted = 0;
772 2041 : if (b->tnorevsorted == q)
773 7 : b->tnorevsorted = 0;
774 2041 : if (b->tnokey[0] == q ||
775 2034 : b->tnokey[1] == q) {
776 7 : b->tnokey[0] = 0;
777 7 : b->tnokey[1] = 0;
778 : }
779 2041 : b->tkey = false;
780 2041 : b->tsorted = false;
781 2041 : b->tkey = false;
782 2041 : if (BUNreplace(b, q, t, true) != GDK_SUCCEED) {
783 0 : logbat_destroy(b);
784 0 : bat_iterator_end(&vi);
785 0 : return GDK_FAIL;
786 : }
787 : } else {
788 1 : if (BUNappend(b, t, true) != GDK_SUCCEED) {
789 0 : logbat_destroy(b);
790 0 : bat_iterator_end(&vi);
791 0 : return GDK_FAIL;
792 : }
793 : }
794 : }
795 125 : bat_iterator_end(&vi);
796 : }
797 : }
798 929 : } else if (la->type == LOG_UPDATE) {
799 929 : if (!lg->flushing && BATupdate(b, la->uid, la->b, true) != GDK_SUCCEED) {
800 : return GDK_FAIL;
801 : }
802 : }
803 360960 : cnt = (BUN) (la->offset + la->nr);
804 360960 : if (la_bat_update_count(lg, la->cid, cnt, tid) != GDK_SUCCEED) {
805 0 : if (b)
806 0 : logbat_destroy(b);
807 0 : return GDK_FAIL;
808 : }
809 360960 : if (b)
810 1258 : logbat_destroy(b);
811 : return GDK_SUCCEED;
812 : }
813 :
814 : static log_return
815 8740 : log_read_destroy(logger *lg, trans *tr, log_id id)
816 : {
817 8740 : (void) lg;
818 8740 : assert(!lg->inmemory);
819 8740 : if (tr_grow(tr) == GDK_SUCCEED) {
820 8740 : tr->changes[tr->nr].type = LOG_DESTROY;
821 8740 : tr->changes[tr->nr].cid = id;
822 8740 : tr->nr++;
823 8740 : return LOG_OK;
824 : }
825 0 : TRC_CRITICAL(GDK, "memory allocation failed\n");
826 0 : return LOG_ERR;
827 : }
828 :
829 : static gdk_return
830 40 : la_bat_destroy(logger *lg, logaction *la, int tid)
831 : {
832 40 : log_bid bid = internal_find_bat(lg, la->cid, tid);
833 :
834 40 : if (bid < 0)
835 : return GDK_FAIL;
836 40 : if (!bid) {
837 : #ifndef NDEBUG
838 0 : GDKwarning("failed to find bid for object %d\n", la->cid);
839 : #endif
840 0 : return GDK_SUCCEED;
841 : }
842 40 : if (bid && log_del_bat(lg, bid) != GDK_SUCCEED)
843 : return GDK_FAIL;
844 : return GDK_SUCCEED;
845 : }
846 :
847 : static log_return
848 156476 : log_read_create(logger *lg, trans *tr, log_id id)
849 : {
850 156476 : bte tt;
851 156476 : int tpe;
852 :
853 156476 : assert(!lg->inmemory);
854 156476 : TRC_DEBUG(WAL, "create %d", id);
855 :
856 156476 : if (mnstr_read(lg->input_log, &tt, 1, 1) != 1) {
857 0 : TRC_CRITICAL(GDK, "read failed\n");
858 0 : return LOG_EOF;
859 : }
860 :
861 156476 : tpe = find_type_nr(lg, tt);
862 : /* read create */
863 156476 : if (tr_grow(tr) == GDK_SUCCEED) {
864 156476 : tr->changes[tr->nr].type = LOG_CREATE;
865 156476 : tr->changes[tr->nr].tt = tpe;
866 156476 : tr->changes[tr->nr].cid = id;
867 156476 : tr->nr++;
868 156476 : return LOG_OK;
869 : }
870 0 : TRC_CRITICAL(GDK, "memory allocation failed\n");
871 0 : return LOG_ERR;
872 : }
873 :
874 : static gdk_return
875 253 : la_bat_create(logger *lg, logaction *la, int tid)
876 : {
877 253 : BAT *b;
878 :
879 : /* formerly head column type, should be void */
880 253 : if ((b = COLnew(0, la->tt, BATSIZE, PERSISTENT)) == NULL)
881 : return GDK_FAIL;
882 :
883 253 : if (la->tt < 0)
884 0 : BATtseqbase(b, 0);
885 :
886 506 : if ((b = BATsetaccess(b, BAT_READ)) == NULL ||
887 253 : log_add_bat(lg, b, la->cid, tid) != GDK_SUCCEED) {
888 0 : logbat_destroy(b);
889 0 : return GDK_FAIL;
890 : }
891 253 : logbat_destroy(b);
892 253 : return GDK_SUCCEED;
893 : }
894 :
895 : static gdk_return
896 233 : log_write_new_types(logger *lg, FILE *fp)
897 : {
898 233 : bte id = 0;
899 :
900 : /* write types and insert into bats */
901 233 : memset(lg->type_id, -1, sizeof(lg->type_id));
902 233 : memset(lg->type_nr, 255, sizeof(lg->type_nr));
903 : /* first the fixed sized types */
904 7220 : for (int i = 0; i < GDKatomcnt; i++) {
905 6987 : if (ATOMvarsized(i))
906 1862 : continue;
907 5125 : lg->type_id[i] = id;
908 5125 : lg->type_nr[id] = i;
909 5125 : if (fprintf(fp, "%d,%s\n", id, BATatoms[i].name) < 0)
910 : return GDK_FAIL;
911 5125 : id++;
912 : }
913 : /* second the var sized types */
914 : id = -127; /* start after nil */
915 7220 : for (int i = 0; i < GDKatomcnt; i++) {
916 6987 : if (!ATOMvarsized(i))
917 5125 : continue;
918 1862 : lg->type_id[i] = id;
919 1862 : lg->type_nr[256 + id] = i;
920 1862 : if (fprintf(fp, "%d,%s\n", id, BATatoms[i].name) < 0)
921 : return GDK_FAIL;
922 1862 : id++;
923 : }
924 : return GDK_SUCCEED;
925 : }
926 :
927 : #define TR_SIZE 1024
928 :
929 : static trans *
930 56458 : tr_create(trans *tr, int tid)
931 : {
932 56458 : trans *ntr = GDKmalloc(sizeof(trans));
933 :
934 56458 : if (ntr == NULL)
935 : return NULL;
936 56458 : ntr->tid = tid;
937 56458 : ntr->sz = TR_SIZE;
938 56458 : ntr->nr = 0;
939 56458 : ntr->changes = GDKmalloc(sizeof(logaction) * TR_SIZE);
940 56458 : if (ntr->changes == NULL) {
941 0 : GDKfree(ntr);
942 0 : return NULL;
943 : }
944 56458 : ntr->tr = tr;
945 56458 : return ntr;
946 : }
947 :
948 : static gdk_return
949 526179 : la_apply(logger *lg, logaction *c, int tid)
950 : {
951 526179 : gdk_return ret = GDK_SUCCEED;
952 :
953 526179 : switch (c->type) {
954 360963 : case LOG_UPDATE_BULK:
955 : case LOG_UPDATE:
956 360963 : ret = la_bat_updates(lg, c, tid);
957 360963 : break;
958 156476 : case LOG_CREATE:
959 156476 : if (!lg->flushing)
960 253 : ret = la_bat_create(lg, c, tid);
961 : break;
962 8740 : case LOG_DESTROY:
963 8740 : if (!lg->flushing)
964 40 : ret = la_bat_destroy(lg, c, tid);
965 : break;
966 : default:
967 0 : MT_UNREACHABLE();
968 : }
969 526179 : return ret;
970 : }
971 :
972 : static void
973 526179 : la_destroy(logaction *c)
974 : {
975 526179 : if ((c->type == LOG_UPDATE || c->type == LOG_UPDATE_BULK) && c->b)
976 1258 : logbat_destroy(c->b);
977 526179 : if (c->type == LOG_UPDATE && c->uid)
978 911 : logbat_destroy(c->uid);
979 526179 : }
980 :
981 : static gdk_return
982 526179 : tr_grow(trans *tr)
983 : {
984 526179 : if (tr->nr == tr->sz) {
985 7 : logaction *changes;
986 7 : tr->sz <<= 1;
987 7 : changes = GDKrealloc(tr->changes, tr->sz * sizeof(logaction));
988 7 : if (changes == NULL)
989 : return GDK_FAIL;
990 7 : tr->changes = changes;
991 : }
992 : /* cleanup the next */
993 526179 : tr->changes[tr->nr].b = NULL;
994 526179 : return GDK_SUCCEED;
995 : }
996 :
997 : static trans *
998 56458 : tr_destroy(trans *tr)
999 : {
1000 56458 : trans *r = tr->tr;
1001 :
1002 56458 : GDKfree(tr->changes);
1003 56458 : GDKfree(tr);
1004 56458 : return r;
1005 : }
1006 :
1007 : static trans *
1008 0 : tr_abort_(logger *lg, trans *tr, int s)
1009 : {
1010 0 : int i;
1011 :
1012 0 : (void) lg;
1013 :
1014 0 : TRC_DEBUG(WAL, "abort");
1015 :
1016 0 : for (i = s; i < tr->nr; i++)
1017 0 : la_destroy(&tr->changes[i]);
1018 0 : return tr_destroy(tr);
1019 : }
1020 :
1021 : static trans *
1022 0 : tr_abort(logger *lg, trans *tr)
1023 : {
1024 0 : return tr_abort_(lg, tr, 0);
1025 : }
1026 :
1027 : static trans *
1028 56458 : tr_commit(logger *lg, trans *tr)
1029 : {
1030 56458 : int i;
1031 :
1032 56458 : TRC_DEBUG(WAL, "commit");
1033 :
1034 582637 : for (i = 0; i < tr->nr; i++) {
1035 526179 : if (la_apply(lg, &tr->changes[i], tr->tid) != GDK_SUCCEED) {
1036 0 : TRC_CRITICAL(GDK, "aborting transaction\n");
1037 0 : do {
1038 0 : tr = tr_abort_(lg, tr, i);
1039 0 : } while (tr != NULL);
1040 : return (trans *) -1;
1041 : }
1042 526179 : la_destroy(&tr->changes[i]);
1043 : }
1044 56458 : lg->saved_tid = tr->tid;
1045 56458 : return tr_destroy(tr);
1046 : }
1047 :
1048 : static gdk_return
1049 105 : log_read_types_file(logger *lg, FILE *fp, int version)
1050 : {
1051 105 : int id = 0;
1052 105 : char atom_name[IDLENGTH];
1053 105 : bool seen_geom = false;
1054 :
1055 : /* scanf should use IDLENGTH somehow */
1056 3221 : while (fscanf(fp, "%d,%63s\n", &id, atom_name) == 2) {
1057 3116 : if (version < 52303 && strcmp(atom_name, "BAT") == 0)
1058 8 : continue;
1059 3108 : int i = ATOMindex(atom_name);
1060 :
1061 3108 : if (id < -127 || id > 127 || i < 0) {
1062 0 : GDKerror("unknown type in log file '%s'\n", atom_name);
1063 0 : return GDK_FAIL;
1064 : }
1065 3108 : seen_geom |= strcmp(atom_name, "mbr") == 0 || strcmp(atom_name, "wkb") == 0;
1066 3108 : lg->type_id[i] = (int8_t) id;
1067 3108 : lg->type_nr[id < 0 ? 256 + id : id] = i;
1068 : }
1069 : #ifdef HAVE_GEOM
1070 105 : if (!seen_geom && ATOMindex("mbr") > 0) {
1071 0 : GDKerror("incompatible database: server supports GEOM, but database does not\n");
1072 0 : return GDK_FAIL;
1073 : }
1074 : #endif
1075 : (void) seen_geom;
1076 : return GDK_SUCCEED;
1077 : }
1078 :
1079 :
1080 : gdk_return
1081 233 : log_create_types_file(logger *lg, const char *filename)
1082 : {
1083 233 : FILE *fp;
1084 :
1085 233 : if ((fp = MT_fopen(filename, "w")) == NULL) {
1086 0 : GDKerror("cannot create log file %s\n", filename);
1087 0 : return GDK_FAIL;
1088 : }
1089 233 : if (fprintf(fp, "%06d\n\n", lg->version) < 0) {
1090 0 : fclose(fp);
1091 0 : GDKerror("writing log file %s failed", filename);
1092 0 : if (MT_remove(filename) < 0)
1093 0 : GDKsyserror("remove %s failed\n", filename);
1094 0 : return GDK_FAIL;
1095 : }
1096 :
1097 233 : if (log_write_new_types(lg, fp) != GDK_SUCCEED) {
1098 0 : fclose(fp);
1099 0 : GDKerror("writing log file %s failed", filename);
1100 0 : if (MT_remove(filename) < 0)
1101 0 : GDKsyserror("remove %s failed\n", filename);
1102 0 : return GDK_FAIL;
1103 : }
1104 233 : if (fflush(fp) < 0 || (!(ATOMIC_GET(&GDKdebug) & NOSYNCMASK)
1105 : #if defined(_MSC_VER)
1106 : && _commit(_fileno(fp)) < 0
1107 : #elif defined(HAVE_FDATASYNC)
1108 0 : && fdatasync(fileno(fp)) < 0
1109 : #elif defined(HAVE_FSYNC)
1110 : && fsync(fileno(fp)) < 0
1111 : #endif
1112 : )) {
1113 0 : GDKsyserror("flushing log file %s failed", filename);
1114 0 : fclose(fp);
1115 0 : if (MT_remove(filename) < 0)
1116 0 : GDKsyserror("remove %s failed\n", filename);
1117 0 : return GDK_FAIL;
1118 : }
1119 233 : if (fclose(fp) < 0) {
1120 0 : GDKsyserror("closing log file %s failed", filename);
1121 0 : if (MT_remove(filename) < 0)
1122 0 : GDKsyserror("remove %s failed\n", filename);
1123 0 : return GDK_FAIL;
1124 : }
1125 : return GDK_SUCCEED;
1126 : }
1127 :
1128 : #define rotation_lock(lg) MT_lock_set(&(lg)->rotation_lock)
1129 : #define rotation_unlock(lg) MT_lock_unset(&(lg)->rotation_lock)
1130 : #define rotation_trylock(lg, ms) MT_lock_trytime(&(lg)->rotation_lock, ms)
1131 :
1132 : static gdk_return
1133 12661 : log_open_output(logger *lg)
1134 : {
1135 12661 : logged_range *new_range = (logged_range *) GDKmalloc(sizeof(logged_range));
1136 :
1137 12661 : if (!new_range) {
1138 0 : TRC_CRITICAL(GDK, "allocation failure\n");
1139 0 : return GDK_FAIL;
1140 : }
1141 25321 : if (!LOG_DISABLED(lg)) {
1142 12660 : char id[32];
1143 12660 : char *filename;
1144 :
1145 12660 : if (snprintf(id, sizeof(id), LLFMT, lg->id) >= (int) sizeof(id)) {
1146 0 : TRC_CRITICAL(GDK, "filename is too large\n");
1147 0 : GDKfree(new_range);
1148 0 : return GDK_FAIL;
1149 : }
1150 12660 : if ((filename = GDKfilepath(BBPselectfarm(PERSISTENT, 0, offheap), lg->dir, LOGFILE, id)) == NULL) {
1151 0 : TRC_CRITICAL(GDK, "allocation failure\n");
1152 0 : GDKfree(new_range);
1153 0 : return GDK_FAIL;
1154 : }
1155 :
1156 12660 : TRC_INFO(WAL, "opening %s.%s", LOGFILE, id);
1157 12660 : new_range->output_log = open_wstream(filename);
1158 12660 : if (new_range->output_log) {
1159 12660 : short byteorder = 1234;
1160 12660 : mnstr_write(new_range->output_log, &byteorder, sizeof(byteorder), 1);
1161 : }
1162 :
1163 12660 : if (new_range->output_log == NULL || mnstr_errnr(new_range->output_log) != MNSTR_NO__ERROR) {
1164 0 : TRC_CRITICAL(GDK, "creating %s failed: %s\n", filename, mnstr_peek_error(NULL));
1165 0 : close_stream(new_range->output_log);
1166 0 : GDKfree(new_range);
1167 0 : GDKfree(filename);
1168 0 : return GDK_FAIL;
1169 : }
1170 12660 : GDKfree(filename);
1171 : } else {
1172 1 : new_range->output_log = NULL;
1173 : }
1174 12661 : ATOMIC_INIT(&new_range->refcount, 1);
1175 12661 : ATOMIC_INIT(&new_range->last_ts, 0);
1176 12661 : ATOMIC_INIT(&new_range->flushed_ts, 0);
1177 12661 : ATOMIC_INIT(&new_range->drops, 0);
1178 12661 : new_range->id = lg->id;
1179 12661 : new_range->next = NULL;
1180 12661 : logged_range *current = lg->current;
1181 12661 : assert(current && current->next == NULL);
1182 12661 : new_range->cnt = current->cnt;
1183 12661 : current->next = new_range;
1184 12661 : lg->file_age = GDKusec();
1185 12661 : return GDK_SUCCEED;
1186 : }
1187 :
1188 : static inline void
1189 12855 : log_close_input(logger *lg)
1190 : {
1191 12855 : if (!lg->inmemory && lg->input_log) {
1192 12420 : TRC_INFO(WAL, "closing input log %s", mnstr_name(lg->input_log));
1193 12420 : close_stream(lg->input_log);
1194 : }
1195 12855 : lg->input_log = NULL;
1196 12855 : }
1197 :
1198 : static inline void
1199 330 : log_close_output(logger *lg)
1200 : {
1201 330 : if (!LOG_DISABLED(lg) && lg->current->output_log) {
1202 329 : TRC_INFO(WAL, "closing output log %s", mnstr_name(lg->current->output_log));
1203 329 : close_stream(lg->current->output_log);
1204 : }
1205 330 : lg->current->output_log = NULL;
1206 330 : }
1207 :
1208 : static gdk_return
1209 12525 : log_open_input(logger *lg, const char *filename, bool *filemissing)
1210 : {
1211 12525 : TRC_INFO(WAL, "opening input log %s", filename);
1212 12525 : lg->input_log = open_rstream(filename);
1213 :
1214 : /* if the file doesn't exist, there is nothing to be read back */
1215 12525 : if (lg->input_log == NULL || mnstr_errnr(lg->input_log) != MNSTR_NO__ERROR) {
1216 105 : log_close_input(lg);
1217 105 : *filemissing = true;
1218 105 : return GDK_SUCCEED;
1219 : }
1220 12420 : short byteorder;
1221 12420 : switch (mnstr_read(lg->input_log, &byteorder, sizeof(byteorder), 1)) {
1222 0 : case -1:
1223 0 : log_close_input(lg);
1224 0 : TRC_CRITICAL(GDK, "read failed\n");
1225 0 : return GDK_FAIL;
1226 5 : case 0:
1227 : /* empty file is ok */
1228 5 : log_close_input(lg);
1229 5 : return GDK_SUCCEED;
1230 12415 : case 1:
1231 : /* if not empty, must start with correct byte order mark */
1232 12415 : if (byteorder != 1234) {
1233 0 : TRC_CRITICAL(GDK, "incorrect byte order word in file %s\n", filename);
1234 0 : log_close_input(lg);
1235 0 : return GDK_FAIL;
1236 : }
1237 : break;
1238 : }
1239 : return GDK_SUCCEED;
1240 : }
1241 :
1242 : static log_return
1243 12415 : log_read_transaction(logger *lg, uint32_t *updated, BUN maxupdated)
1244 : {
1245 12415 : logformat l;
1246 12415 : trans *tr = NULL;
1247 12415 : log_return err = LOG_OK;
1248 12415 : bool ok = true;
1249 12415 : ATOMIC_BASE_TYPE dbg = ATOMIC_GET(&GDKdebug);
1250 :
1251 12415 : (void) maxupdated; /* only used inside assert() */
1252 :
1253 12415 : if (!lg->flushing)
1254 169 : ATOMIC_AND(&GDKdebug, ~CHECKMASK);
1255 :
1256 12415 : BAT *cands = NULL; /* used in case of LOG_BAT_GROUP */
1257 :
1258 757596 : while (err == LOG_OK && (ok = log_read_format(lg, &l))) {
1259 745181 : if (l.flag == 0 && l.id == 0) {
1260 12415 : err = LOG_EOF;
1261 : break;
1262 : }
1263 :
1264 745181 : TRC_DEBUG_IF(WAL) {
1265 0 : if (l.flag > 0 && l.flag != LOG_CLEAR &&
1266 : l.flag < (bte) (sizeof(log_commands) / sizeof(log_commands[0])))
1267 0 : TRC_DEBUG_ENDIF(WAL, "%s %d", log_commands[(int) l.flag], l.id);
1268 : else
1269 0 : TRC_DEBUG_ENDIF(WAL, "%d %d", l.flag, l.id);
1270 : }
1271 745181 : switch (l.flag) {
1272 554445 : case LOG_UPDATE_CONST:
1273 : case LOG_UPDATE_BULK:
1274 : case LOG_UPDATE:
1275 : case LOG_CREATE:
1276 : case LOG_DESTROY:
1277 554445 : if (tr != NULL && updated && BAThash(lg->catalog_id) == GDK_SUCCEED) {
1278 552737 : BATiter cni = bat_iterator(lg->catalog_id);
1279 552737 : BUN p;
1280 552737 : BUN posnew = BUN_NONE;
1281 552737 : BUN posold = BUN_NONE;
1282 552737 : MT_rwlock_rdlock(&cni.b->thashlock);
1283 3971809 : HASHloop_int(cni, cni.b->thash, p, &l.id) {
1284 3061031 : lng lid = *(lng *) Tloc(lg->catalog_lid, p);
1285 3061031 : if (lid == lng_nil || lid > tr->tid)
1286 : posnew = p;
1287 1397523 : else if (lid == tr->tid)
1288 3347293 : posold = p;
1289 : }
1290 552737 : MT_rwlock_rdunlock(&cni.b->thashlock);
1291 552737 : bat_iterator_end(&cni);
1292 : /* Normally at this point, posnew is the
1293 : * location of the bat that this
1294 : * transaction is working on, and posold
1295 : * is the location of the previous
1296 : * version of the bat. If LOG_CREATE,
1297 : * both are relevant, since the latter
1298 : * is the new bat, and the former is the
1299 : * to-be-destroyed bat. For
1300 : * LOG_DESTROY, only posnew should be
1301 : * relevant, but for the other types, if
1302 : * the table is destroyed later in the
1303 : * same transaction, we need posold, and
1304 : * else (the normal case) we need
1305 : * posnew. */
1306 552737 : if (posnew != BUN_NONE) {
1307 544033 : assert(posnew < maxupdated);
1308 544033 : updated[posnew / 32] |= 1U << (posnew % 32);
1309 : }
1310 552737 : if ((l.flag == LOG_CREATE || posnew == BUN_NONE) && posold != BUN_NONE) {
1311 161144 : assert(posold < maxupdated);
1312 161144 : updated[posold / 32] |= 1U << (posold % 32);
1313 : }
1314 : }
1315 : break;
1316 : default:
1317 : /* do nothing */
1318 : break;
1319 : }
1320 : /* the functions we call here can succeed (LOG_OK),
1321 : * but they can also fail for two different reasons:
1322 : * they can run out of input (LOG_EOF -- this is not
1323 : * serious, we just abort the remaining transactions),
1324 : * or some malloc or BAT update fails (LOG_ERR -- this
1325 : * is serious, we must abort the complete process);
1326 : * the latter failure causes the current function to
1327 : * return GDK_FAIL */
1328 745181 : switch (l.flag) {
1329 56458 : case LOG_START:
1330 56458 : assert(!lg->flushing || l.id <= lg->tid);
1331 56458 : if (!lg->flushing && l.id > lg->tid)
1332 109 : lg->tid = l.id; /* should only happen during initialization */
1333 56458 : if ((tr = tr_create(tr, l.id)) == NULL) {
1334 0 : TRC_CRITICAL(GDK, "memory allocation failed\n");
1335 0 : err = LOG_ERR;
1336 0 : break;
1337 : }
1338 56458 : TRC_DEBUG(WAL, "tstart %d\n", tr->tid);
1339 : break;
1340 56458 : case LOG_END:
1341 56458 : if (tr == NULL)
1342 : err = LOG_EOF;
1343 56458 : else if (tr->tid != l.id) /* abort record */
1344 0 : tr = tr_abort(lg, tr);
1345 : else
1346 56458 : tr = tr_commit(lg, tr);
1347 : break;
1348 2764 : case LOG_SEQ:
1349 2764 : err = log_read_seq(lg, &l);
1350 2764 : break;
1351 389229 : case LOG_UPDATE_CONST:
1352 : case LOG_UPDATE_BULK:
1353 : case LOG_UPDATE:
1354 389229 : if (tr == NULL)
1355 : err = LOG_EOF;
1356 : else {
1357 610995 : err = log_read_updates(lg, tr, &l, l.id, cands ? &cands : NULL);
1358 : }
1359 : break;
1360 156476 : case LOG_CREATE:
1361 156476 : if (tr == NULL)
1362 : err = LOG_EOF;
1363 : else
1364 156476 : err = log_read_create(lg, tr, l.id);
1365 : break;
1366 8740 : case LOG_DESTROY:
1367 8740 : if (tr == NULL)
1368 : err = LOG_EOF;
1369 : else
1370 8740 : err = log_read_destroy(lg, tr, l.id);
1371 : break;
1372 75056 : case LOG_BAT_GROUP:
1373 75056 : if (tr == NULL)
1374 : err = LOG_EOF;
1375 : else {
1376 75056 : if (l.id > 0) {
1377 : /* START OF LOG_BAT_GROUP */
1378 37528 : cands = COLnew(0, TYPE_void, 0, SYSTRANS);
1379 37528 : if (!cands) {
1380 0 : TRC_CRITICAL(GDK, "creating bat failed\n");
1381 0 : err = LOG_ERR;
1382 : }
1383 37528 : } else if (cands == NULL) {
1384 : /* should have gone through the
1385 : * above option earlier */
1386 0 : TRC_CRITICAL(GDK, "unexpected error\n");
1387 0 : err = LOG_ERR;
1388 : } else {
1389 : /* END OF LOG_BAT_GROUP */
1390 37528 : BBPunfix(cands->batCacheid);
1391 37528 : cands = NULL;
1392 : }
1393 : }
1394 : break;
1395 0 : default:
1396 0 : TRC_CRITICAL(GDK, "unrecognized log entry %d", l.flag);
1397 0 : err = LOG_ERR;
1398 : }
1399 745181 : if (tr == (trans *) -1) {
1400 : /* message already generated by tr_commit */
1401 : err = LOG_ERR;
1402 12415 : tr = NULL;
1403 : break;
1404 : }
1405 : }
1406 12415 : while (tr) {
1407 0 : TRC_WARNING(GDK, "aborting transaction\n");
1408 0 : tr = tr_abort(lg, tr);
1409 : }
1410 12415 : if (!lg->flushing)
1411 169 : ATOMIC_SET(&GDKdebug, dbg);
1412 :
1413 12415 : BBPreclaim(cands);
1414 12415 : if (!ok)
1415 12415 : return LOG_EOF;
1416 : return err;
1417 : }
1418 :
1419 : static gdk_return
1420 279 : log_readlog(logger *lg, const char *filename, bool *filemissing)
1421 : {
1422 279 : log_return err = LOG_OK;
1423 279 : time_t t0, t1;
1424 279 : struct stat sb;
1425 :
1426 279 : assert(!lg->inmemory);
1427 :
1428 279 : TRC_INFO(WAL, "opening %s\n", filename);
1429 :
1430 279 : gdk_return res = log_open_input(lg, filename, filemissing);
1431 279 : if (!lg->input_log || res != GDK_SUCCEED)
1432 : return res;
1433 169 : int fd;
1434 169 : if ((fd = getFileNo(lg->input_log)) < 0 || fstat(fd, &sb) < 0) {
1435 0 : GDKsyserror("fstat on opened file %s failed\n", filename);
1436 0 : log_close_input(lg);
1437 : /* If the file could be opened, but fstat fails,
1438 : * something weird is going on */
1439 0 : return GDK_FAIL;
1440 : }
1441 169 : t0 = time(NULL);
1442 169 : TRC_INFO_IF(WAL) {
1443 0 : TRC_INFO_ENDIF(WAL, "Start reading the write-ahead log '%s'\n", filename);
1444 0 : GDKtracer_flush_buffer();
1445 : }
1446 338 : while (err != LOG_EOF && err != LOG_ERR) {
1447 169 : t1 = time(NULL);
1448 169 : if (t1 - t0 > 10) {
1449 0 : lng fpos;
1450 0 : t0 = t1;
1451 : /* not more than once every 10 seconds */
1452 0 : fpos = (lng) getfilepos(getFile(lg->input_log));
1453 0 : TRC_INFO_IF(WAL) {
1454 0 : if (fpos >= 0) {
1455 0 : TRC_INFO_ENDIF(WAL, "still reading write-ahead log \"%s\" (%d%% done)\n",
1456 : filename, (int) ((fpos * 100 + 50) / sb.st_size));
1457 0 : GDKtracer_flush_buffer();
1458 : }
1459 : }
1460 : }
1461 169 : err = log_read_transaction(lg, NULL, 0);
1462 : }
1463 169 : log_close_input(lg);
1464 169 : lg->input_log = NULL;
1465 :
1466 : /* remaining transactions are not committed, ie abort */
1467 169 : TRC_INFO_IF(WAL) {
1468 0 : TRC_INFO_ENDIF(WAL, "Finished reading the write-ahead log '%s'\n", filename);
1469 0 : GDKtracer_flush_buffer();
1470 : }
1471 : /* we cannot distinguish errors from incomplete transactions
1472 : * (even if we would log aborts in the logs). So we simply
1473 : * abort and move to the next log file */
1474 169 : return err == LOG_ERR ? GDK_FAIL : GDK_SUCCEED;
1475 : }
1476 :
1477 : /*
1478 : * The log files are incrementally numbered, starting from 2. They are
1479 : * processed in the same sequence.
1480 : */
1481 : static gdk_return
1482 105 : log_readlogs(logger *lg, const char *filename)
1483 : {
1484 105 : gdk_return res = GDK_SUCCEED;
1485 :
1486 105 : assert(!lg->inmemory);
1487 105 : TRC_DEBUG(WAL, "logger id is " LLFMT " last logger id is " LLFMT "\n", lg->id, lg->saved_id);
1488 :
1489 105 : char log_filename[FILENAME_MAX];
1490 105 : if (lg->saved_id >= lg->id) {
1491 105 : bool filemissing = false;
1492 :
1493 105 : lg->id = lg->saved_id + 1;
1494 384 : while (res == GDK_SUCCEED && !filemissing) {
1495 279 : if (snprintf(log_filename, sizeof(log_filename), "%s." LLFMT, filename, lg->id) >= FILENAME_MAX) {
1496 0 : GDKerror("Logger filename path is too large\n");
1497 0 : return GDK_FAIL;
1498 : }
1499 279 : res = log_readlog(lg, log_filename, &filemissing);
1500 279 : if (!filemissing) {
1501 174 : lg->saved_id++;
1502 174 : lg->id++;
1503 : }
1504 : }
1505 : }
1506 : return res;
1507 : }
1508 :
1509 : static gdk_return
1510 12372 : log_commit(logger *lg, logged_range *pending, uint32_t *updated, BUN maxupdated)
1511 : {
1512 12372 : TRC_DEBUG(WAL, "commit");
1513 :
1514 12372 : return bm_commit(lg, pending, updated, maxupdated);
1515 : }
1516 :
1517 : static gdk_return
1518 105 : check_version(logger *lg, FILE *fp, bool *needsnew)
1519 : {
1520 105 : int version = 0;
1521 :
1522 105 : assert(!lg->inmemory);
1523 105 : if (fscanf(fp, "%6d", &version) != 1) {
1524 0 : GDKerror("Could not read the version number from the file '%s/log'.\n", lg->dir);
1525 0 : fclose(fp);
1526 0 : return GDK_FAIL;
1527 : }
1528 105 : if (version != lg->version) {
1529 16 : if (lg->prefuncp == NULL ||
1530 8 : (*lg->prefuncp) (lg->funcdata, version, lg->version) != GDK_SUCCEED) {
1531 0 : GDKerror("Incompatible database version %06d, "
1532 : "this server supports version %06d.\n%s",
1533 : version, lg->version,
1534 : version < lg->version ? "Maybe you need to upgrade to an intermediate release first.\n" : "");
1535 0 : fclose(fp);
1536 0 : return GDK_FAIL;
1537 : }
1538 8 : *needsnew = true; /* we need to write a new log file */
1539 : } else {
1540 97 : lg->postfuncp = NULL; /* don't call */
1541 97 : *needsnew = false; /* log file already up-to-date */
1542 : }
1543 210 : if (fgetc(fp) != '\n' || /* skip \n */
1544 105 : fgetc(fp) != '\n') { /* skip \n */
1545 0 : GDKerror("Badly formatted log file");
1546 0 : fclose(fp);
1547 0 : return GDK_FAIL;
1548 : }
1549 105 : if (log_read_types_file(lg, fp, version) != GDK_SUCCEED) {
1550 0 : fclose(fp);
1551 0 : return GDK_FAIL;
1552 : }
1553 105 : fclose(fp);
1554 105 : return GDK_SUCCEED;
1555 : }
1556 :
1557 : static BAT *
1558 351 : bm_tids(BAT *b, BAT *d)
1559 : {
1560 351 : BUN sz = BATcount(b);
1561 351 : BAT *tids = BATdense(0, 0, sz);
1562 :
1563 351 : if (tids == NULL)
1564 : return NULL;
1565 :
1566 351 : if (BATcount(d)) {
1567 351 : BAT *diff = BATdiff(tids, d, NULL, NULL, false, false, BUN_NONE);
1568 351 : logbat_destroy(tids);
1569 351 : tids = diff;
1570 : }
1571 : return tids;
1572 : }
1573 :
1574 :
1575 : static gdk_return
1576 8289 : log_switch_bat(BAT *old, BAT *new, const char *fn, const char *name)
1577 : {
1578 8289 : char bak[IDLENGTH];
1579 :
1580 8289 : if (BATmode(old, true) != GDK_SUCCEED) {
1581 0 : GDKerror("cannot convert old %s to transient", name);
1582 0 : return GDK_FAIL;
1583 : }
1584 8289 : if (strconcat_len(bak, sizeof(bak), fn, "_", name, NULL) >= sizeof(bak)) {
1585 0 : GDKerror("name %s_%s too long\n", fn, name);
1586 0 : return GDK_FAIL;
1587 : }
1588 8289 : if (BBPrename(old, NULL) != 0 || BBPrename(new, bak) != 0) {
1589 0 : GDKerror("rename (%s) failed\n", bak);
1590 0 : return GDK_FAIL;
1591 : }
1592 8289 : BBPretain(new->batCacheid);
1593 8289 : return GDK_SUCCEED;
1594 : }
1595 :
1596 : static gdk_return
1597 331 : bm_get_counts(logger *lg)
1598 : {
1599 331 : BUN p, q;
1600 331 : const log_bid *bids = (const log_bid *) Tloc(lg->catalog_bid, 0);
1601 :
1602 27602 : BATloop(lg->catalog_bid, p, q) {
1603 27271 : oid pos = p;
1604 27271 : lng cnt = 0;
1605 27271 : lng lid = lng_nil;
1606 :
1607 27271 : if (BUNfnd(lg->dcatalog, &pos) == BUN_NONE) {
1608 27271 : BAT *b = BBPquickdesc(bids[p]);
1609 27271 : assert(b);
1610 27271 : cnt = BATcount(b);
1611 : } else {
1612 0 : lid = BBP_desc(bids[p])->batCacheid != 0 && log_find(lg->catalog_bid, lg->dcatalog, bids[p]) == BUN_NONE ? 1 : -1;
1613 : }
1614 27271 : if (BUNappend(lg->catalog_cnt, &cnt, false) != GDK_SUCCEED)
1615 0 : return GDK_FAIL;
1616 27271 : if (BUNappend(lg->catalog_lid, &lid, false) != GDK_SUCCEED)
1617 : return GDK_FAIL;
1618 : }
1619 : return GDK_SUCCEED;
1620 : }
1621 :
1622 : static int
1623 8289 : subcommit_list_add(int next, bat *n, BUN *sizes, bat bid, BUN sz)
1624 : {
1625 8289 : assert(sz <= BBP_desc(bid)->batCount || sz == BUN_NONE);
1626 1875620 : for (int i = 0; i < next; i++) {
1627 1867331 : if (n[i] == bid) {
1628 0 : sizes[i] = sz;
1629 0 : return next;
1630 : }
1631 : }
1632 8289 : n[next] = bid;
1633 8289 : sizes[next++] = sz;
1634 8289 : return next;
1635 : }
1636 :
1637 : static int
1638 2529 : cleanup_and_swap(logger *lg, int *r, const log_bid *bids, lng *lids, lng *cnts,
1639 : BAT *catalog_bid, BAT *catalog_id, BAT *dcatalog, BUN cleanup)
1640 : {
1641 2529 : BAT *nbids, *noids, *ncnts, *nlids, *ndels;
1642 2529 : BUN p, q;
1643 2529 : int err = 0, rcnt = 0;
1644 :
1645 2529 : BUN ocnt = BATcount(catalog_bid);
1646 2529 : nbids = logbat_new(TYPE_int, ocnt - cleanup, PERSISTENT);
1647 2529 : noids = logbat_new(TYPE_int, ocnt - cleanup, PERSISTENT);
1648 2529 : ncnts = logbat_new(TYPE_lng, ocnt - cleanup, SYSTRANS);
1649 2529 : nlids = logbat_new(TYPE_lng, ocnt - cleanup, SYSTRANS);
1650 2529 : ndels = logbat_new(TYPE_oid, BATcount(dcatalog) - cleanup, PERSISTENT);
1651 :
1652 2529 : if (nbids == NULL || noids == NULL || ncnts == NULL || nlids == NULL || ndels == NULL) {
1653 0 : logbat_destroy(nbids);
1654 0 : logbat_destroy(noids);
1655 0 : logbat_destroy(ncnts);
1656 0 : logbat_destroy(nlids);
1657 0 : logbat_destroy(ndels);
1658 0 : return 0;
1659 : }
1660 :
1661 2529 : oid *poss = Tloc(dcatalog, 0);
1662 178825 : BATloop(dcatalog, p, q) {
1663 176296 : oid pos = poss[p];
1664 :
1665 176296 : if (lids[pos] == lng_nil || lids[pos] > lg->saved_tid)
1666 0 : continue;
1667 :
1668 176296 : if (lids[pos] >= 0) {
1669 176296 : bat bid = bids[pos];
1670 176296 : BAT *lb = BBP_desc(bid);
1671 :
1672 176296 : if (lb->batCacheid == 0 || BATmode(lb, true /*transient */ ) != GDK_SUCCEED) {
1673 0 : GDKwarning("Failed to set bat(%d) transient\n", bid);
1674 : } else {
1675 176296 : lids[pos] = -1; /* mark as transient */
1676 176296 : r[rcnt++] = bid;
1677 : }
1678 : }
1679 : }
1680 :
1681 2529 : int *oids = (int *) Tloc(catalog_id, 0);
1682 2529 : q = BATcount(catalog_bid);
1683 893455 : for (p = 0; p < q && !err; p++) {
1684 890926 : bat col = bids[p];
1685 890926 : int nid = oids[p];
1686 890926 : lng lid = lids[p];
1687 890926 : lng cnt = cnts[p];
1688 890926 : oid pos = p;
1689 :
1690 : /* only project out the deleted with lid == -1
1691 : * update dcatalog */
1692 890926 : if (lid == -1)
1693 176296 : continue; /* remove */
1694 :
1695 1429260 : if (BUNappend(nbids, &col, true) != GDK_SUCCEED ||
1696 1429260 : BUNappend(noids, &nid, true) != GDK_SUCCEED ||
1697 1429260 : BUNappend(nlids, &lid, false) != GDK_SUCCEED ||
1698 714630 : BUNappend(ncnts, &cnt, false) != GDK_SUCCEED)
1699 : err = 1;
1700 714630 : if (BUNfnd(lg->dcatalog, &pos) != BUN_NONE) {
1701 0 : pos = (oid) (BATcount(nbids) - 1);
1702 0 : if (BUNappend(ndels, &pos, true) != GDK_SUCCEED)
1703 714630 : err = 1;
1704 : }
1705 : }
1706 :
1707 2529 : if (err) {
1708 0 : logbat_destroy(nbids);
1709 0 : logbat_destroy(noids);
1710 0 : logbat_destroy(ndels);
1711 0 : logbat_destroy(ncnts);
1712 0 : logbat_destroy(nlids);
1713 0 : return 0;
1714 : }
1715 : /* point of no return */
1716 5058 : if (log_switch_bat(catalog_bid, nbids, lg->fn, "catalog_bid") != GDK_SUCCEED ||
1717 5058 : log_switch_bat(catalog_id, noids, lg->fn, "catalog_id") != GDK_SUCCEED ||
1718 2529 : log_switch_bat(dcatalog, ndels, lg->fn, "dcatalog") != GDK_SUCCEED) {
1719 0 : logbat_destroy(nbids);
1720 0 : logbat_destroy(noids);
1721 0 : logbat_destroy(ndels);
1722 0 : logbat_destroy(ncnts);
1723 0 : logbat_destroy(nlids);
1724 0 : return -1;
1725 : }
1726 2529 : r[rcnt++] = lg->catalog_bid->batCacheid;
1727 2529 : r[rcnt++] = lg->catalog_id->batCacheid;
1728 2529 : r[rcnt++] = lg->dcatalog->batCacheid;
1729 :
1730 2529 : assert(BATcount(lg->dcatalog) - cleanup == BATcount(ndels));
1731 :
1732 2529 : logbat_destroy(lg->catalog_bid);
1733 2529 : logbat_destroy(lg->catalog_id);
1734 2529 : logbat_destroy(lg->dcatalog);
1735 :
1736 2529 : lg->catalog_bid = nbids;
1737 2529 : lg->catalog_id = noids;
1738 2529 : lg->dcatalog = ndels;
1739 :
1740 : /* failing to rename these two bats is not fatal */
1741 2529 : if (BBPrename(lg->catalog_cnt, NULL) != GDK_SUCCEED)
1742 2529 : GDKclrerr();
1743 2529 : if (BBPrename(lg->catalog_lid, NULL) != GDK_SUCCEED)
1744 2529 : GDKclrerr();
1745 2529 : BBPunfix(lg->catalog_cnt->batCacheid);
1746 2529 : BBPunfix(lg->catalog_lid->batCacheid);
1747 :
1748 2529 : lg->catalog_cnt = ncnts;
1749 2529 : lg->catalog_lid = nlids;
1750 2529 : char bak[FILENAME_MAX];
1751 2529 : strconcat_len(bak, sizeof(bak), lg->fn, "_catalog_cnt", NULL);
1752 2529 : if (BBPrename(lg->catalog_cnt, bak) < 0)
1753 0 : GDKclrerr();
1754 2529 : strconcat_len(bak, sizeof(bak), lg->fn, "_catalog_lid", NULL);
1755 2529 : if (BBPrename(lg->catalog_lid, bak) < 0)
1756 0 : GDKclrerr();
1757 2529 : rotation_lock(lg);
1758 10567 : for (logged_range *p = lg->pending; p; p = p->next) {
1759 8038 : p->cnt -= cleanup;
1760 : }
1761 2529 : rotation_unlock(lg);
1762 2529 : return rcnt;
1763 : }
1764 :
1765 : /* this function is called with log_lock() held; it releases the lock
1766 : * before returning */
1767 : static gdk_return
1768 12824 : bm_subcommit(logger *lg, logged_range *pending, uint32_t *updated, BUN maxupdated)
1769 : {
1770 12824 : BUN cnt = pending ? pending->cnt : BATcount(lg->catalog_bid);
1771 12824 : BUN dcnt = BATcount(lg->dcatalog);
1772 12824 : BUN p, q;
1773 12824 : BAT *catalog_bid = lg->catalog_bid;
1774 12824 : BAT *catalog_id = lg->catalog_id;
1775 12824 : BAT *dcatalog = lg->dcatalog;
1776 12824 : BUN nn = 13 + cnt;
1777 12824 : bat *n = GDKmalloc(sizeof(bat) * nn);
1778 12824 : bat *r = GDKmalloc(sizeof(bat) * nn);
1779 12824 : BUN *sizes = GDKmalloc(sizeof(BUN) * nn);
1780 12824 : int i = 0, rcnt = 0;
1781 12824 : gdk_return res;
1782 12824 : const log_bid *bids;
1783 12824 : lng *cnts = NULL, *lids = NULL;
1784 12824 : BUN cleanup = 0;
1785 12824 : lng t0 = 0;
1786 :
1787 12824 : if (n == NULL || r == NULL || sizes == NULL) {
1788 0 : GDKfree(n);
1789 0 : GDKfree(r);
1790 0 : GDKfree(sizes);
1791 0 : log_unlock(lg);
1792 0 : return GDK_FAIL;
1793 : }
1794 :
1795 12824 : sizes[i] = 0;
1796 12824 : n[i++] = 0; /* n[0] is not used */
1797 12824 : bids = (const log_bid *) Tloc(catalog_bid, 0);
1798 12824 : if (lg->catalog_cnt)
1799 12598 : cnts = (lng *) Tloc(lg->catalog_cnt, 0);
1800 12824 : if (lg->catalog_lid)
1801 12598 : lids = (lng *) Tloc(lg->catalog_lid, 0);
1802 3747982 : BATloop(catalog_bid, p, q) {
1803 3735158 : if (lids && lids[p] != lng_nil && lids[p] <= lg->saved_tid) {
1804 176296 : cleanup++;
1805 176296 : if (lids[p] == -1)
1806 0 : continue;
1807 352592 : if (BUNfnd(dcatalog, &(oid){p}) == BUN_NONE &&
1808 176296 : BUNappend(dcatalog, &(oid){p}, true) != GDK_SUCCEED) {
1809 0 : while (BATcount(dcatalog) > dcnt) {
1810 0 : if (BUNdelete(dcatalog, BATcount(dcatalog) - 1) != GDK_SUCCEED) {
1811 0 : TRC_CRITICAL(WAL, "delete after failed append failed\n");
1812 0 : break;
1813 : }
1814 : }
1815 0 : GDKfree(n);
1816 0 : GDKfree(r);
1817 0 : GDKfree(sizes);
1818 0 : log_unlock(lg);
1819 0 : return GDK_FAIL;
1820 : }
1821 : }
1822 3735158 : if (updated && p < maxupdated && (updated[p / 32] & (1U << (p % 32))) == 0) {
1823 1849571 : continue;
1824 : }
1825 1885587 : bat col = bids[p];
1826 :
1827 1885587 : TRC_DEBUG(WAL, "new %s (%d)\n", BBP_logical(col), col);
1828 1885587 : assert(col);
1829 1885587 : sizes[i] = cnts ? (BUN) cnts[p] : 0;
1830 1885587 : n[i++] = col;
1831 : }
1832 : /* now commit catalog, so it's also up to date on disk */
1833 12824 : sizes[i] = cnt;
1834 12824 : n[i++] = catalog_bid->batCacheid;
1835 12824 : sizes[i] = cnt;
1836 12824 : n[i++] = catalog_id->batCacheid;
1837 12824 : sizes[i] = BATcount(dcatalog);
1838 12824 : n[i++] = dcatalog->batCacheid;
1839 :
1840 12824 : if (cleanup) {
1841 2529 : if ((rcnt = cleanup_and_swap(lg, r, bids, lids, cnts,
1842 : catalog_bid, catalog_id, dcatalog,
1843 : cleanup)) < 0) {
1844 0 : GDKfree(n);
1845 0 : GDKfree(r);
1846 0 : GDKfree(sizes);
1847 0 : log_unlock(lg);
1848 0 : return GDK_FAIL;
1849 : }
1850 2529 : cnt -= cleanup;
1851 : }
1852 12824 : if (dcatalog != lg->dcatalog) {
1853 2529 : i = subcommit_list_add(i, n, sizes, lg->catalog_bid->batCacheid, cnt);
1854 2529 : i = subcommit_list_add(i, n, sizes, lg->catalog_id->batCacheid, cnt);
1855 2529 : i = subcommit_list_add(i, n, sizes, lg->dcatalog->batCacheid, BATcount(lg->dcatalog));
1856 : }
1857 12824 : if (lg->seqs_id) {
1858 12598 : sizes[i] = BATcount(lg->seqs_id);
1859 12598 : n[i++] = lg->seqs_id->batCacheid;
1860 12598 : sizes[i] = BATcount(lg->seqs_id);
1861 12598 : n[i++] = lg->seqs_val->batCacheid;
1862 : }
1863 12824 : if (!cleanup && lg->seqs_id && BATcount(lg->dseqs) > (BATcount(lg->seqs_id) / 2) && BATcount(lg->dseqs) > 10) {
1864 351 : BAT *tids, *ids, *vals;
1865 :
1866 351 : tids = bm_tids(lg->seqs_id, lg->dseqs);
1867 351 : if (tids == NULL) {
1868 0 : GDKfree(n);
1869 0 : GDKfree(r);
1870 0 : GDKfree(sizes);
1871 0 : log_unlock(lg);
1872 0 : return GDK_FAIL;
1873 : }
1874 351 : ids = logbat_new(TYPE_int, BATcount(tids), PERSISTENT);
1875 351 : vals = logbat_new(TYPE_lng, BATcount(tids), PERSISTENT);
1876 :
1877 351 : if (ids == NULL || vals == NULL) {
1878 0 : logbat_destroy(tids);
1879 0 : logbat_destroy(ids);
1880 0 : logbat_destroy(vals);
1881 0 : GDKfree(n);
1882 0 : GDKfree(r);
1883 0 : GDKfree(sizes);
1884 0 : log_unlock(lg);
1885 0 : return GDK_FAIL;
1886 : }
1887 :
1888 702 : if (BATappend(ids, lg->seqs_id, tids, true) != GDK_SUCCEED ||
1889 351 : BATappend(vals, lg->seqs_val, tids, true) != GDK_SUCCEED) {
1890 0 : logbat_destroy(tids);
1891 0 : logbat_destroy(ids);
1892 0 : logbat_destroy(vals);
1893 0 : GDKfree(n);
1894 0 : GDKfree(r);
1895 0 : GDKfree(sizes);
1896 0 : log_unlock(lg);
1897 0 : return GDK_FAIL;
1898 : }
1899 351 : logbat_destroy(tids);
1900 351 : BATclear(lg->dseqs, true);
1901 :
1902 702 : if (log_switch_bat(lg->seqs_id, ids, lg->fn, "seqs_id") != GDK_SUCCEED ||
1903 351 : log_switch_bat(lg->seqs_val, vals, lg->fn, "seqs_val") != GDK_SUCCEED) {
1904 0 : logbat_destroy(ids);
1905 0 : logbat_destroy(vals);
1906 0 : GDKfree(n);
1907 0 : GDKfree(r);
1908 0 : GDKfree(sizes);
1909 0 : log_unlock(lg);
1910 0 : return GDK_FAIL;
1911 : }
1912 351 : i = subcommit_list_add(i, n, sizes, ids->batCacheid, BATcount(ids));
1913 351 : i = subcommit_list_add(i, n, sizes, vals->batCacheid, BATcount(ids));
1914 :
1915 351 : if (BBP_lrefs(lg->seqs_id->batCacheid) > 0)
1916 280 : r[rcnt++] = lg->seqs_id->batCacheid;
1917 351 : if (BBP_lrefs(lg->seqs_val->batCacheid) > 0)
1918 280 : r[rcnt++] = lg->seqs_val->batCacheid;
1919 :
1920 351 : logbat_destroy(lg->seqs_id);
1921 351 : logbat_destroy(lg->seqs_val);
1922 :
1923 351 : lg->seqs_id = ids;
1924 351 : lg->seqs_val = vals;
1925 : }
1926 12824 : if (lg->seqs_id) {
1927 12598 : sizes[i] = BATcount(lg->dseqs);
1928 12598 : n[i++] = lg->dseqs->batCacheid;
1929 : }
1930 :
1931 12824 : assert((BUN) i <= nn);
1932 12824 : log_unlock(lg);
1933 12824 : TRC_DEBUG_IF(WAL)
1934 0 : t0 = GDKusec();
1935 13050 : res = TMsubcommit_list(n, cnts ? sizes : NULL, i, lg->saved_id);
1936 12824 : TRC_DEBUG(WAL, "subcommit " LLFMT "usec\n", GDKusec() - t0);
1937 12824 : if (res == GDK_SUCCEED) { /* now cleanup */
1938 197267 : for (i = 0; i < rcnt; i++) {
1939 184443 : TRC_DEBUG_IF(WAL) {
1940 0 : TRC_DEBUG_ENDIF(WAL, "release %d\n", r[i]);
1941 0 : if (BBP_lrefs(r[i]) != 2)
1942 0 : TRC_DEBUG_ENDIF(WAL, "release %d %d\n", r[i], BBP_lrefs(r[i]));
1943 : }
1944 184443 : BBPrelease(r[i]);
1945 : }
1946 : }
1947 12824 : GDKfree(n);
1948 12824 : GDKfree(r);
1949 12824 : GDKfree(sizes);
1950 12824 : if (res != GDK_SUCCEED)
1951 0 : TRC_CRITICAL(GDK, "commit failed\n");
1952 : return res;
1953 : }
1954 :
1955 : static gdk_return
1956 330 : log_filename(logger *lg, char bak[FILENAME_MAX], char filename[FILENAME_MAX])
1957 : {
1958 330 : str filenamestr = NULL;
1959 :
1960 330 : if ((filenamestr = GDKfilepath(0, lg->dir, LOGFILE, NULL)) == NULL)
1961 : return GDK_FAIL;
1962 330 : size_t len = strcpy_len(filename, filenamestr, FILENAME_MAX);
1963 330 : GDKfree(filenamestr);
1964 330 : if (len >= FILENAME_MAX) {
1965 0 : GDKerror("Logger filename path is too large\n");
1966 0 : return GDK_FAIL;
1967 : }
1968 330 : if (bak) {
1969 330 : len = strconcat_len(bak, FILENAME_MAX, filename, ".bak", NULL);
1970 330 : if (len >= FILENAME_MAX) {
1971 0 : GDKerror("Logger filename path is too large\n");
1972 0 : return GDK_FAIL;
1973 : }
1974 : }
1975 : return GDK_SUCCEED;
1976 : }
1977 :
1978 : static gdk_return
1979 12420 : log_cleanup(logger *lg, lng id)
1980 : {
1981 12420 : char log_id[FILENAME_MAX];
1982 :
1983 12420 : if (snprintf(log_id, sizeof(log_id), LLFMT, id) >= FILENAME_MAX) {
1984 0 : GDKerror("log_id filename is too large\n");
1985 0 : return GDK_FAIL;
1986 : }
1987 12420 : if (GDKunlink(0, lg->dir, LOGFILE, log_id) != GDK_SUCCEED) {
1988 0 : GDKwarning("failed to remove old WAL %s.%s\n", LOGFILE, log_id);
1989 0 : GDKclrerr(); /* clear error from unlink */
1990 : }
1991 : return GDK_SUCCEED;
1992 : }
1993 :
1994 : #ifdef GDKLIBRARY_JSON
1995 : static gdk_return
1996 331 : log_json_upgrade_finalize(void)
1997 : {
1998 331 : int json_tpe = ATOMindex("json");
1999 661 : if (!GDKinmemory(0) &&
2000 330 : GDKunlink(0, BATDIR, "jsonupgradeneeded", NULL) == GDK_FAIL) {
2001 0 : TRC_CRITICAL(GDK, "Failed to remove json upgrade signal file");
2002 0 : return GDK_FAIL;
2003 : }
2004 331 : BATatoms[json_tpe].atomRead = (void *(*)(void *, size_t *, stream *, size_t))strRead;
2005 :
2006 331 : return GDK_SUCCEED;
2007 : }
2008 : #endif
2009 :
2010 : /* Load data from the logger logdir
2011 : * Initialize new directories and catalog files if none are present,
2012 : * unless running in read-only mode
2013 : * Load data and persist it in the BATs */
2014 : static gdk_return
2015 331 : log_load(const char *fn, logger *lg, char filename[FILENAME_MAX])
2016 : {
2017 331 : FILE *fp = NULL;
2018 331 : char bak[FILENAME_MAX];
2019 331 : bat catalog_bid, catalog_id, dcatalog;
2020 331 : bool needcommit = false;
2021 331 : ATOMIC_BASE_TYPE dbg = ATOMIC_GET(&GDKdebug);
2022 331 : bool readlogs = false;
2023 331 : bool needsnew = false; /* need to write new log file? */
2024 :
2025 : /* refactor */
2026 331 : if (!LOG_DISABLED(lg)) {
2027 330 : if (log_filename(lg, bak, filename) != GDK_SUCCEED)
2028 0 : goto error;
2029 : }
2030 :
2031 331 : lg->catalog_bid = NULL;
2032 331 : lg->catalog_id = NULL;
2033 331 : lg->catalog_cnt = NULL;
2034 331 : lg->catalog_lid = NULL;
2035 331 : lg->dcatalog = NULL;
2036 :
2037 331 : lg->seqs_id = NULL;
2038 331 : lg->seqs_val = NULL;
2039 331 : lg->dseqs = NULL;
2040 :
2041 331 : if (!LOG_DISABLED(lg)) {
2042 : /* try to open logfile backup, or failing that, the file
2043 : * itself. we need to know whether this file exists when
2044 : * checking the database consistency later on */
2045 330 : if ((fp = MT_fopen(bak, "r")) != NULL) {
2046 0 : fclose(fp);
2047 0 : fp = NULL;
2048 0 : if (GDKunlink(0, lg->dir, LOGFILE, NULL) != GDK_SUCCEED ||
2049 0 : GDKmove(0, lg->dir, LOGFILE, "bak", lg->dir, LOGFILE, NULL, true) != GDK_SUCCEED)
2050 0 : goto error;
2051 330 : } else if (errno != ENOENT) {
2052 0 : GDKsyserror("open %s failed", bak);
2053 0 : goto error;
2054 : }
2055 330 : fp = MT_fopen(filename, "r");
2056 330 : if (fp == NULL && errno != ENOENT) {
2057 0 : GDKsyserror("open %s failed", filename);
2058 0 : goto error;
2059 : }
2060 : }
2061 :
2062 331 : strconcat_len(bak, sizeof(bak), fn, "_catalog_bid", NULL);
2063 331 : catalog_bid = BBPindex(bak);
2064 :
2065 : /* initialize arrays for type mapping, to be read from disk */
2066 331 : memset(lg->type_id, -1, sizeof(lg->type_id));
2067 331 : memset(lg->type_nr, 255, sizeof(lg->type_nr));
2068 :
2069 : /* this is intentional - if catalog_bid is 0, force it to find
2070 : * the persistent catalog */
2071 331 : if (catalog_bid == 0) {
2072 : /* catalog does not exist, so the log file also
2073 : * shouldn't exist */
2074 226 : if (fp != NULL) {
2075 0 : GDKerror("there is no logger catalog, "
2076 : "but there is a log file.\n");
2077 0 : goto error;
2078 : }
2079 :
2080 226 : lg->catalog_bid = logbat_new(TYPE_int, BATSIZE, PERSISTENT);
2081 226 : lg->catalog_id = logbat_new(TYPE_int, BATSIZE, PERSISTENT);
2082 226 : lg->dcatalog = logbat_new(TYPE_oid, BATSIZE, PERSISTENT);
2083 :
2084 226 : if (lg->catalog_bid == NULL || lg->catalog_id == NULL || lg->dcatalog == NULL) {
2085 0 : GDKerror("cannot create catalog bats");
2086 0 : goto error;
2087 : }
2088 226 : TRC_INFO(WAL, "create %s catalog\n", fn);
2089 :
2090 : /* give the catalog bats names so we can find them
2091 : * next time */
2092 226 : strconcat_len(bak, sizeof(bak), fn, "_catalog_bid", NULL);
2093 226 : if (BBPrename(lg->catalog_bid, bak) < 0) {
2094 0 : goto error;
2095 : }
2096 :
2097 226 : strconcat_len(bak, sizeof(bak), fn, "_catalog_id", NULL);
2098 226 : if (BBPrename(lg->catalog_id, bak) < 0) {
2099 0 : goto error;
2100 : }
2101 :
2102 226 : strconcat_len(bak, sizeof(bak), fn, "_dcatalog", NULL);
2103 226 : if (BBPrename(lg->dcatalog, bak) < 0) {
2104 0 : goto error;
2105 : }
2106 :
2107 226 : if (!LOG_DISABLED(lg)) {
2108 225 : if (GDKcreatedir(filename) != GDK_SUCCEED) {
2109 0 : GDKerror("cannot create directory for log file %s\n", filename);
2110 0 : goto error;
2111 : }
2112 225 : if (log_create_types_file(lg, filename) != GDK_SUCCEED)
2113 0 : goto error;
2114 : }
2115 :
2116 226 : BBPretain(lg->catalog_bid->batCacheid);
2117 226 : BBPretain(lg->catalog_id->batCacheid);
2118 226 : BBPretain(lg->dcatalog->batCacheid);
2119 :
2120 226 : log_lock(lg);
2121 : /* bm_subcommit releases the lock */
2122 226 : if (bm_subcommit(lg, NULL, NULL, 0) != GDK_SUCCEED) {
2123 : /* cannot commit catalog, so remove log */
2124 0 : if (MT_remove(filename) < 0)
2125 0 : GDKsyserror("remove %s failed\n", filename);
2126 0 : BBPrelease(lg->catalog_bid->batCacheid);
2127 0 : BBPrelease(lg->catalog_id->batCacheid);
2128 0 : BBPrelease(lg->dcatalog->batCacheid);
2129 0 : goto error;
2130 : }
2131 : } else {
2132 : /* find the persistent catalog. As non persistent bats
2133 : * require a logical reference we also add a logical
2134 : * reference for the persistent bats */
2135 105 : BUN p, q;
2136 105 : BAT *b, *o, *d;
2137 :
2138 105 : assert(!lg->inmemory);
2139 :
2140 : /* the catalog exists, and so should the log file */
2141 105 : if (fp == NULL && !LOG_DISABLED(lg)) {
2142 0 : GDKerror("There is a logger catalog, but no log file.\n");
2143 0 : goto error;
2144 : }
2145 : if (fp != NULL) {
2146 : /* check_version always closes fp */
2147 105 : if (check_version(lg, fp, &needsnew) != GDK_SUCCEED) {
2148 0 : fp = NULL;
2149 0 : goto error;
2150 : }
2151 : readlogs = true;
2152 : fp = NULL;
2153 : }
2154 :
2155 105 : if (lg->catalog_bid == NULL && lg->catalog_id == NULL && lg->dcatalog == NULL) {
2156 105 : b = BATdescriptor(catalog_bid);
2157 105 : if (b == NULL) {
2158 0 : GDKerror("inconsistent database, catalog does not exist");
2159 0 : goto error;
2160 : }
2161 :
2162 105 : strconcat_len(bak, sizeof(bak), fn, "_catalog_id", NULL);
2163 105 : catalog_id = BBPindex(bak);
2164 105 : o = BATdescriptor(catalog_id);
2165 105 : if (o == NULL) {
2166 0 : BBPunfix(b->batCacheid);
2167 0 : GDKerror("inconsistent database, catalog_id does not exist");
2168 0 : goto error;
2169 : }
2170 :
2171 105 : strconcat_len(bak, sizeof(bak), fn, "_dcatalog", NULL);
2172 105 : dcatalog = BBPindex(bak);
2173 105 : d = BATdescriptor(dcatalog);
2174 105 : if (d == NULL) {
2175 0 : GDKerror("cannot create dcatalog bat");
2176 0 : BBPunfix(b->batCacheid);
2177 0 : BBPunfix(o->batCacheid);
2178 0 : goto error;
2179 : }
2180 :
2181 105 : lg->catalog_bid = b;
2182 105 : lg->catalog_id = o;
2183 105 : lg->dcatalog = d;
2184 105 : const log_bid *bids = (const log_bid *) Tloc(lg->catalog_bid, 0);
2185 27376 : BATloop(lg->catalog_bid, p, q) {
2186 27271 : bat bid = bids[p];
2187 27271 : oid pos = p;
2188 :
2189 27271 : if (BBPretain(bid) == 0 && /* any bid in the catalog_bid, needs one logical ref */
2190 0 : BUNfnd(lg->dcatalog, &pos) == BUN_NONE &&
2191 0 : BUNappend(lg->dcatalog, &pos, true) != GDK_SUCCEED)
2192 0 : goto error;
2193 : }
2194 : }
2195 105 : if ((lg->catalog_bid = BATsetaccess(lg->catalog_bid, BAT_READ)) == NULL ||
2196 105 : (lg->catalog_id = BATsetaccess(lg->catalog_id, BAT_READ)) == NULL ||
2197 105 : (lg->dcatalog = BATsetaccess(lg->dcatalog, BAT_READ)) == NULL) {
2198 0 : goto error;
2199 : }
2200 105 : BBPretain(lg->catalog_bid->batCacheid);
2201 105 : BBPretain(lg->catalog_id->batCacheid);
2202 105 : BBPretain(lg->dcatalog->batCacheid);
2203 : }
2204 : /* failing to rename the catalog_cnt and catalog_lid bats is not
2205 : * fatal */
2206 331 : lg->catalog_cnt = logbat_new(TYPE_lng, 1, SYSTRANS);
2207 331 : if (lg->catalog_cnt == NULL) {
2208 0 : GDKerror("failed to create catalog_cnt bat");
2209 0 : goto error;
2210 : }
2211 331 : strconcat_len(bak, sizeof(bak), fn, "_catalog_cnt", NULL);
2212 331 : if (BBPrename(lg->catalog_cnt, bak) < 0)
2213 0 : GDKclrerr();
2214 331 : lg->catalog_lid = logbat_new(TYPE_lng, 1, SYSTRANS);
2215 331 : if (lg->catalog_lid == NULL) {
2216 0 : GDKerror("failed to create catalog_lid bat");
2217 0 : goto error;
2218 : }
2219 331 : strconcat_len(bak, sizeof(bak), fn, "_catalog_lid", NULL);
2220 331 : if (BBPrename(lg->catalog_lid, bak) < 0)
2221 0 : GDKclrerr();
2222 331 : if (bm_get_counts(lg) != GDK_SUCCEED)
2223 0 : goto error;
2224 :
2225 331 : strconcat_len(bak, sizeof(bak), fn, "_seqs_id", NULL);
2226 331 : if (BBPindex(bak)) {
2227 105 : lg->seqs_id = BATdescriptor(BBPindex(bak));
2228 105 : strconcat_len(bak, sizeof(bak), fn, "_seqs_val", NULL);
2229 105 : lg->seqs_val = BATdescriptor(BBPindex(bak));
2230 105 : strconcat_len(bak, sizeof(bak), fn, "_dseqs", NULL);
2231 105 : lg->dseqs = BATdescriptor(BBPindex(bak));
2232 105 : if (lg->seqs_id == NULL ||
2233 105 : lg->seqs_val == NULL ||
2234 : lg->dseqs == NULL) {
2235 0 : GDKerror("Logger_new: cannot load seqs bats");
2236 0 : goto error;
2237 : }
2238 105 : if ((lg->seqs_val = BATsetaccess(lg->seqs_val, BAT_READ)) == NULL ||
2239 105 : (lg->seqs_id = BATsetaccess(lg->seqs_id, BAT_READ)) == NULL ||
2240 105 : (lg->dseqs = BATsetaccess(lg->dseqs, BAT_READ)) == NULL) {
2241 0 : goto error;
2242 : }
2243 : } else {
2244 226 : lg->seqs_id = logbat_new(TYPE_int, 1, PERSISTENT);
2245 226 : lg->seqs_val = logbat_new(TYPE_lng, 1, PERSISTENT);
2246 226 : lg->dseqs = logbat_new(TYPE_oid, 1, PERSISTENT);
2247 226 : if (lg->seqs_id == NULL ||
2248 226 : lg->seqs_val == NULL ||
2249 : lg->dseqs == NULL) {
2250 0 : GDKerror("Logger_new: cannot create seqs bats");
2251 0 : goto error;
2252 : }
2253 :
2254 226 : strconcat_len(bak, sizeof(bak), fn, "_seqs_id", NULL);
2255 226 : if (BBPrename(lg->seqs_id, bak) < 0) {
2256 0 : goto error;
2257 : }
2258 :
2259 226 : strconcat_len(bak, sizeof(bak), fn, "_seqs_val", NULL);
2260 226 : if (BBPrename(lg->seqs_val, bak) < 0) {
2261 0 : goto error;
2262 : }
2263 :
2264 226 : strconcat_len(bak, sizeof(bak), fn, "_dseqs", NULL);
2265 226 : if (BBPrename(lg->dseqs, bak) < 0) {
2266 0 : goto error;
2267 : }
2268 : needcommit = true;
2269 : }
2270 331 : dbg = ATOMIC_GET(&GDKdebug);
2271 331 : ATOMIC_AND(&GDKdebug, ~CHECKMASK);
2272 331 : if (needcommit && bm_commit(lg, NULL, NULL, 0) != GDK_SUCCEED) {
2273 0 : GDKerror("Logger_new: commit failed");
2274 0 : goto error;
2275 : }
2276 331 : ATOMIC_SET(&GDKdebug, dbg);
2277 :
2278 331 : if (readlogs) {
2279 105 : ulng log_id = lg->saved_id + 1;
2280 105 : if (log_readlogs(lg, filename) != GDK_SUCCEED) {
2281 0 : goto error;
2282 : }
2283 105 : if (GDKgetenv_isyes("process-wal-and-exit")) {
2284 0 : printf("# mserver5 exiting\n");
2285 0 : exit(0);
2286 : }
2287 105 : if (lg->postfuncp && (*lg->postfuncp) (lg->funcdata, lg) != GDK_SUCCEED)
2288 0 : goto error;
2289 105 : if (needsnew) {
2290 8 : if (GDKmove(0, lg->dir, LOGFILE, NULL, lg->dir, LOGFILE, "bak", true) != GDK_SUCCEED) {
2291 0 : TRC_CRITICAL(GDK, "couldn't move log to log.bak\n");
2292 0 : return GDK_FAIL;
2293 : }
2294 8 : if (log_create_types_file(lg, filename) != GDK_SUCCEED) {
2295 0 : TRC_CRITICAL(GDK, "couldn't write new log\n");
2296 0 : return GDK_FAIL;
2297 : }
2298 : }
2299 105 : dbg = ATOMIC_GET(&GDKdebug);
2300 105 : ATOMIC_AND(&GDKdebug, ~CHECKMASK);
2301 105 : if (log_commit(lg, NULL, NULL, 0) != GDK_SUCCEED) {
2302 0 : goto error;
2303 : }
2304 105 : ATOMIC_SET(&GDKdebug, dbg);
2305 279 : for (; log_id <= lg->saved_id; log_id++)
2306 174 : (void) log_cleanup(lg, log_id); /* ignore error of removing file */
2307 113 : if (needsnew &&
2308 8 : GDKunlink(0, lg->dir, LOGFILE, "bak") != GDK_SUCCEED) {
2309 0 : TRC_CRITICAL(GDK, "couldn't remove old log.bak file\n");
2310 0 : return GDK_FAIL;
2311 : }
2312 : } else {
2313 226 : lg->id = lg->saved_id + 1;
2314 226 : if (GDKgetenv_isyes("process-wal-and-exit")) {
2315 0 : printf("# mserver5 exiting\n");
2316 0 : exit(0);
2317 : }
2318 : }
2319 : #ifdef GDKLIBRARY_JSON
2320 331 : if (log_json_upgrade_finalize() == GDK_FAIL)
2321 0 : goto error;
2322 : #endif
2323 : return GDK_SUCCEED;
2324 0 : error:
2325 0 : if (fp)
2326 0 : fclose(fp);
2327 0 : logbat_destroy(lg->catalog_bid);
2328 0 : logbat_destroy(lg->catalog_id);
2329 0 : logbat_destroy(lg->dcatalog);
2330 0 : logbat_destroy(lg->seqs_id);
2331 0 : logbat_destroy(lg->seqs_val);
2332 0 : logbat_destroy(lg->dseqs);
2333 0 : MT_lock_destroy(&lg->lock);
2334 0 : MT_lock_destroy(&lg->rotation_lock);
2335 0 : GDKfree(lg->fn);
2336 0 : GDKfree(lg->dir);
2337 0 : GDKfree(lg->rbuf);
2338 0 : GDKfree(lg->wbuf);
2339 0 : GDKfree(lg);
2340 0 : ATOMIC_SET(&GDKdebug, dbg);
2341 : /* We do not call log_json_upgrade_finalize here because we want
2342 : * the upgrade to run again next time we try, so we do not want
2343 : * to remove the signal file just yet.
2344 : */
2345 0 : return GDK_FAIL;
2346 : }
2347 :
2348 : /* Initialize a new logger
2349 : * It will load any data in the logdir and persist it in the BATs*/
2350 : static logger *
2351 331 : log_new(int debug, const char *fn, const char *logdir, int version, preversionfix_fptr prefuncp,
2352 : postversionfix_fptr postfuncp, void *funcdata)
2353 : {
2354 331 : logger *lg;
2355 331 : char filename[FILENAME_MAX];
2356 :
2357 331 : lng max_dropped = GDKgetenv_int("wal_max_dropped", 100000);
2358 331 : lng max_file_age = GDKgetenv_int("wal_max_file_age", 600);
2359 331 : lng max_file_size = 0;
2360 :
2361 331 : if (GDKdebug & TESTINGMASK) {
2362 : max_file_size = 2048; /* 2 KiB */
2363 : } else {
2364 12 : const char *max_file_size_str = GDKgetenv("wal_max_file_size");
2365 12 : max_file_size = max_file_size_str ? strtoul(max_file_size_str, NULL, 10) : 2147483648;
2366 : }
2367 :
2368 331 : if (!GDKinmemory(0) && MT_path_absolute(logdir)) {
2369 0 : TRC_CRITICAL(GDK, "logdir must be relative path\n");
2370 0 : return NULL;
2371 : }
2372 :
2373 331 : if (snprintf(filename, sizeof(filename), "%s%c%s%c", logdir, DIR_SEP, fn, DIR_SEP) >= FILENAME_MAX) {
2374 0 : TRC_CRITICAL(GDK, "filename is too large\n");
2375 0 : return NULL;
2376 : }
2377 :
2378 331 : lg = GDKmalloc(sizeof(struct logger));
2379 331 : if (lg == NULL) {
2380 0 : TRC_CRITICAL(GDK, "allocating logger structure failed\n");
2381 0 : return NULL;
2382 : }
2383 :
2384 662 : *lg = (logger) {
2385 331 : .inmemory = GDKinmemory(0),
2386 : .debug = debug,
2387 : .version = version,
2388 : .prefuncp = prefuncp,
2389 : .postfuncp = postfuncp,
2390 : .funcdata = funcdata,
2391 :
2392 331 : .max_dropped = max_dropped >= 0 ? max_dropped : 100000,
2393 : .file_age = 0,
2394 331 : .max_file_age = max_file_age >= 0 ? max_file_age * 1000000 : 600000000,
2395 331 : .max_file_size = max_file_size >= 0 ? max_file_size : 2147483648,
2396 :
2397 : .id = 0,
2398 331 : .saved_id = getBBPlogno(), /* get saved log number from bbp */
2399 : .nr_flushers = ATOMIC_VAR_INIT(0),
2400 331 : .fn = GDKstrdup(fn),
2401 331 : .dir = GDKstrdup(filename),
2402 : .rbufsize = 64 * 1024,
2403 331 : .rbuf = GDKmalloc(64 * 1024),
2404 : .wbufsize = 64 * 1024,
2405 331 : .wbuf = GDKmalloc(64 * 1024),
2406 : };
2407 :
2408 : /* probably open file and check version first, then call call old logger code */
2409 331 : if (lg->fn == NULL ||
2410 331 : lg->dir == NULL ||
2411 331 : lg->rbuf == NULL ||
2412 : lg->wbuf == NULL) {
2413 0 : TRC_CRITICAL(GDK, "allocating for logger structure failed\n");
2414 0 : GDKfree(lg->fn);
2415 0 : GDKfree(lg->dir);
2416 0 : GDKfree(lg->rbuf);
2417 0 : GDKfree(lg->wbuf);
2418 0 : GDKfree(lg);
2419 0 : return NULL;
2420 : }
2421 331 : TRC_DEBUG(WAL, "dir set to %s\n", lg->dir);
2422 :
2423 331 : MT_lock_init(&lg->lock, fn);
2424 331 : MT_lock_init(&lg->rotation_lock, "rotation_lock");
2425 331 : MT_lock_init(&lg->flush_lock, "flush_lock");
2426 331 : MT_cond_init(&lg->excl_flush_cv);
2427 :
2428 331 : if (log_load(fn, lg, filename) == GDK_SUCCEED) {
2429 : return lg;
2430 : }
2431 : return NULL;
2432 : }
2433 :
2434 : static logged_range *
2435 68439 : do_flush_range_cleanup(logger *lg)
2436 : {
2437 68439 : logged_range *frange = lg->flush_ranges;
2438 68439 : logged_range *first = frange;
2439 :
2440 68439 : if (frange == NULL)
2441 : return NULL;
2442 80769 : while (frange->next) {
2443 12872 : if (ATOMIC_GET(&frange->refcount) > 1)
2444 : break;
2445 12330 : frange = frange->next;
2446 : }
2447 68439 : if (first == frange) {
2448 : return first;
2449 : }
2450 :
2451 12330 : logged_range *flast = frange;
2452 :
2453 12330 : lg->flush_ranges = flast;
2454 :
2455 24660 : for (frange = first; frange && frange != flast; frange = frange->next) {
2456 12330 : ATOMIC_DEC(&frange->refcount);
2457 12330 : if (!LOG_DISABLED(lg) && frange->output_log) {
2458 0 : TRC_INFO(WAL, "closing output log %s", mnstr_name(frange->output_log));
2459 0 : close_stream(frange->output_log);
2460 0 : frange->output_log = NULL;
2461 : }
2462 : }
2463 : return flast;
2464 : }
2465 :
2466 : void
2467 330 : log_destroy(logger *lg)
2468 : {
2469 330 : log_close_input(lg);
2470 330 : logged_range *last = do_flush_range_cleanup(lg);
2471 330 : (void) last;
2472 330 : assert(last == lg->current && last == lg->flush_ranges);
2473 330 : log_close_output(lg);
2474 744 : for (logged_range * p = lg->pending; p; p = lg->pending) {
2475 414 : lg->pending = p->next;
2476 414 : GDKfree(p);
2477 : }
2478 330 : if (LOG_DISABLED(lg)) {
2479 1 : lg->saved_id = lg->id;
2480 1 : lg->saved_tid = lg->tid;
2481 1 : log_commit(lg, NULL, NULL, 0);
2482 : }
2483 330 : if (lg->catalog_bid) {
2484 330 : log_lock(lg);
2485 330 : BUN p, q;
2486 330 : BAT *b = lg->catalog_bid;
2487 :
2488 : /* free resources */
2489 330 : const log_bid *bids = (const log_bid *) Tloc(b, 0);
2490 87113 : BATloop(b, p, q) {
2491 86783 : bat bid = bids[p];
2492 :
2493 86783 : BBPrelease(bid);
2494 : }
2495 :
2496 330 : BBPrelease(lg->catalog_bid->batCacheid);
2497 330 : BBPrelease(lg->catalog_id->batCacheid);
2498 330 : BBPrelease(lg->dcatalog->batCacheid);
2499 330 : logbat_destroy(lg->catalog_bid);
2500 330 : logbat_destroy(lg->catalog_id);
2501 330 : logbat_destroy(lg->dcatalog);
2502 :
2503 330 : logbat_destroy(lg->catalog_cnt);
2504 330 : logbat_destroy(lg->catalog_lid);
2505 330 : log_unlock(lg);
2506 : }
2507 330 : MT_lock_destroy(&lg->lock);
2508 330 : MT_lock_destroy(&lg->rotation_lock);
2509 330 : MT_lock_destroy(&lg->flush_lock);
2510 330 : GDKfree(lg->fn);
2511 330 : GDKfree(lg->dir);
2512 330 : GDKfree(lg->rbuf);
2513 330 : GDKfree(lg->wbuf);
2514 330 : GDKfree(lg);
2515 330 : }
2516 :
2517 : /* Create a new logger */
2518 : logger *
2519 331 : log_create(int debug, const char *fn, const char *logdir, int version,
2520 : preversionfix_fptr prefuncp, postversionfix_fptr postfuncp,
2521 : void *funcdata)
2522 : {
2523 331 : logger *lg;
2524 331 : TRC_INFO_IF(WAL) {
2525 0 : TRC_INFO_ENDIF(WAL, "Started processing logs %s/%s version %d\n", fn, logdir, version);
2526 0 : GDKtracer_flush_buffer();
2527 : }
2528 331 : lg = log_new(debug, fn, logdir, version, prefuncp, postfuncp, funcdata);
2529 331 : if (lg == NULL)
2530 : return NULL;
2531 331 : TRC_INFO_IF(WAL) {
2532 0 : TRC_INFO_ENDIF(WAL, "Finished processing logs %s/%s\n", fn, logdir);
2533 0 : GDKtracer_flush_buffer();
2534 : }
2535 331 : if (GDKsetenv("recovery", "finished") != GDK_SUCCEED) {
2536 0 : log_destroy(lg);
2537 0 : return NULL;
2538 : }
2539 331 : assert(lg->current == NULL);
2540 331 : logged_range dummy = {
2541 331 : .cnt = BATcount(lg->catalog_bid),
2542 : };
2543 331 : lg->current = &dummy;
2544 331 : if (log_open_output(lg) != GDK_SUCCEED) {
2545 0 : lg->current = NULL;
2546 0 : log_destroy(lg);
2547 0 : return NULL;
2548 : }
2549 331 : lg->current = lg->current->next;
2550 331 : assert(lg->pending == NULL && lg->flush_ranges == NULL);
2551 331 : lg->pending = lg->current;
2552 331 : lg->flush_ranges = lg->current;
2553 331 : return lg;
2554 : }
2555 :
2556 : static logged_range *
2557 7915 : log_next_logfile(logger *lg, ulng ts)
2558 : {
2559 7915 : int m = (ATOMIC_GET(&GDKdebug) & TESTINGMASK) ? 1000 : 100;
2560 7915 : if (!lg->pending || !lg->pending->next)
2561 : return NULL;
2562 7915 : rotation_lock(lg);
2563 7915 : if (ATOMIC_GET(&lg->pending->refcount) == 0 && lg->pending != lg->current && lg->pending != lg->flush_ranges &&
2564 7915 : (ulng) ATOMIC_GET(&lg->pending->last_ts) == (ulng) ATOMIC_GET(&lg->pending->flushed_ts) &&
2565 7915 : (ulng) ATOMIC_GET(&lg->pending->flushed_ts) <= ts) {
2566 6661 : rotation_unlock(lg);
2567 6661 : logged_range *p = lg->pending;
2568 6661 : for (int i = 1;
2569 12246 : i < m && ATOMIC_GET(&p->refcount) == 0 && p->next && p->next != lg->current &&
2570 5616 : p->next != lg->flush_ranges && (ulng) ATOMIC_GET(&p->last_ts) == (ulng) ATOMIC_GET(&p->flushed_ts)
2571 17861 : && (ulng) ATOMIC_GET(&p->flushed_ts) <= ts; i++)
2572 5585 : p = p->next;
2573 6661 : return p;
2574 : }
2575 1254 : rotation_unlock(lg);
2576 1254 : return NULL;
2577 : }
2578 :
2579 : static void
2580 6661 : log_cleanup_range(logger *lg, ulng id)
2581 : {
2582 6661 : rotation_lock(lg);
2583 18907 : while (lg->pending && lg->pending->id <= id) {
2584 12246 : logged_range *p;
2585 12246 : p = lg->pending;
2586 12246 : if (p)
2587 12246 : lg->pending = p->next;
2588 12246 : GDKfree(p);
2589 : }
2590 6661 : rotation_unlock(lg);
2591 6661 : }
2592 :
2593 : static void
2594 68112 : do_rotate(logger *lg)
2595 : {
2596 68112 : logged_range *cur = lg->current;
2597 68112 : logged_range *next = cur->next;
2598 68112 : if (next) {
2599 12330 : assert(ATOMIC_GET(&next->refcount) == 1);
2600 12330 : lg->current = next;
2601 12330 : if (!LOG_DISABLED(lg) && ATOMIC_GET(&cur->refcount) == 1 && cur->output_log) {
2602 12122 : close_stream(cur->output_log);
2603 12122 : cur->output_log = NULL;
2604 : }
2605 : }
2606 68112 : }
2607 :
2608 : gdk_return
2609 10363 : log_activate(logger *lg)
2610 : {
2611 10363 : bool flush_cleanup = false;
2612 10363 : gdk_return res = GDK_SUCCEED;
2613 :
2614 10363 : rotation_lock(lg);
2615 10363 : const lng current_file_size = LOG_DISABLED(lg) ? 0 : (lng) getfilepos(getFile(lg->current->output_log));
2616 :
2617 10363 : if (current_file_size == -1) {
2618 0 : rotation_unlock(lg);
2619 0 : return GDK_FAIL;
2620 : }
2621 : /* file size of 2 means only endian indicator present
2622 : * (i.e. effectively empty) */
2623 10363 : if (current_file_size <= 2) {
2624 7025 : rotation_unlock(lg);
2625 7025 : return GDK_SUCCEED;
2626 : }
2627 :
2628 3338 : if (!lg->flushnow &&
2629 3338 : !lg->current->next &&
2630 3338 : current_file_size > 2 &&
2631 3338 : (ATOMIC_GET(&lg->current->drops) > (ulng)lg->max_dropped ||
2632 3331 : current_file_size > lg->max_file_size ||
2633 3292 : (GDKusec() - lg->file_age) > lg->max_file_age) &&
2634 46 : (ulng) ATOMIC_GET(&lg->current->last_ts) > 0 &&
2635 46 : lg->saved_id + 1 == lg->id &&
2636 46 : ATOMIC_GET(&lg->current->refcount) == 1 /* no pending work on this file */ ) {
2637 45 : lg->id++;
2638 : /* start new file */
2639 45 : res = log_open_output(lg);
2640 45 : flush_cleanup = true;
2641 45 : do_rotate(lg);
2642 : }
2643 45 : if (flush_cleanup)
2644 45 : (void) do_flush_range_cleanup(lg);
2645 3338 : rotation_unlock(lg);
2646 3338 : return res;
2647 : }
2648 :
2649 : gdk_return
2650 7915 : log_flush(logger *lg, ulng ts)
2651 : {
2652 7915 : logged_range *pending = log_next_logfile(lg, ts);
2653 7915 : ulng lid = pending ? pending->id : 0, olid = lg->saved_id;
2654 7915 : if (LOG_DISABLED(lg)) {
2655 0 : lg->saved_id = lid;
2656 0 : lg->saved_tid = lg->tid;
2657 0 : if (lid)
2658 0 : log_cleanup_range(lg, lg->saved_id);
2659 0 : if (log_commit(lg, NULL, NULL, 0) != GDK_SUCCEED)
2660 0 : TRC_ERROR(GDK, "failed to commit");
2661 0 : return GDK_SUCCEED;
2662 : }
2663 7915 : if (lg->saved_id >= lid)
2664 : return GDK_SUCCEED;
2665 6661 : rotation_lock(lg);
2666 6661 : ulng lgid = lg->id;
2667 6661 : rotation_unlock(lg);
2668 6661 : if (lg->saved_id + 1 >= lgid) /* logger should first release the file */
2669 : return GDK_SUCCEED;
2670 6661 : log_return res = LOG_OK;
2671 6661 : ulng cid = olid;
2672 6661 : assert(lid <= lgid);
2673 : uint32_t *updated = NULL;
2674 : BUN nupdated = 0;
2675 : size_t allocated = 0;
2676 18907 : while (cid < lid && res == LOG_OK) {
2677 12246 : if (!lg->input_log) {
2678 12246 : char *filename;
2679 12246 : char id[32];
2680 12246 : if (snprintf(id, sizeof(id), LLFMT, cid + 1) >= (int) sizeof(id)) {
2681 0 : GDKfree(updated);
2682 0 : TRC_CRITICAL(GDK, "log_id filename is too large\n");
2683 0 : return GDK_FAIL;
2684 : }
2685 12246 : if ((filename = GDKfilepath(BBPselectfarm(PERSISTENT, 0, offheap), lg->dir, LOGFILE, id)) == NULL) {
2686 0 : GDKfree(updated);
2687 0 : return GDK_FAIL;
2688 : }
2689 12246 : if (strlen(filename) >= FILENAME_MAX) {
2690 0 : GDKfree(updated);
2691 0 : TRC_CRITICAL(GDK, "Logger filename path is too large\n");
2692 0 : GDKfree(filename);
2693 0 : return GDK_FAIL;
2694 : }
2695 :
2696 12246 : bool filemissing = false;
2697 12246 : if (log_open_input(lg, filename, &filemissing) != GDK_SUCCEED) {
2698 0 : GDKfree(updated);
2699 0 : GDKfree(filename);
2700 0 : return GDK_FAIL;
2701 : }
2702 12246 : GDKfree(filename);
2703 : }
2704 : /* we read the full file because skipping is impossible with current log format */
2705 12246 : log_lock(lg);
2706 12246 : if (updated == NULL) {
2707 6661 : nupdated = BATcount(lg->catalog_id);
2708 6661 : allocated = ((nupdated + 31) & ~31) / 8;
2709 6661 : if (allocated == 0)
2710 0 : allocated = 4;
2711 6661 : updated = GDKzalloc(allocated);
2712 6661 : if (updated == NULL) {
2713 0 : log_unlock(lg);
2714 0 : return GDK_FAIL;
2715 : }
2716 5585 : } else if (nupdated < BATcount(lg->catalog_id)) {
2717 115 : BUN n = BATcount(lg->catalog_id);
2718 115 : size_t a = ((n + 31) & ~31) / 8;
2719 115 : if (a > allocated) {
2720 28 : uint32_t *p = GDKrealloc(updated, a);
2721 28 : if (p == NULL) {
2722 0 : GDKfree(updated);
2723 0 : log_unlock(lg);
2724 0 : return GDK_FAIL;
2725 : }
2726 28 : updated = p;
2727 28 : memset(updated + allocated / 4, 0, a - allocated);
2728 28 : allocated = a;
2729 : }
2730 : nupdated = n;
2731 : }
2732 12246 : lg->flushing = true;
2733 12246 : res = log_read_transaction(lg, updated, nupdated);
2734 12246 : lg->flushing = false;
2735 12246 : log_unlock(lg);
2736 12246 : if (res == LOG_EOF) {
2737 12246 : log_close_input(lg);
2738 12246 : res = LOG_OK;
2739 : }
2740 12246 : cid++;
2741 : }
2742 6661 : if (lid > olid && res == LOG_OK) {
2743 6661 : rotation_lock(lg); /* protect against concurrent log_tflush rotate check */
2744 6661 : lg->saved_id = lid;
2745 6661 : rotation_unlock(lg);
2746 6661 : if (log_commit(lg, pending, updated, nupdated) != GDK_SUCCEED) {
2747 0 : TRC_ERROR(GDK, "failed to commit");
2748 0 : res = LOG_ERR;
2749 0 : rotation_lock(lg);
2750 0 : lg->saved_id = olid; /* reset !! */
2751 0 : rotation_unlock(lg);
2752 : }
2753 0 : if (res != LOG_ERR) {
2754 18907 : while (olid < lid) {
2755 : /* Try to cleanup, remove old log file, continue on failure! */
2756 12246 : olid++;
2757 12246 : (void) log_cleanup(lg, olid);
2758 : }
2759 : }
2760 6661 : if (res == LOG_OK)
2761 6661 : log_cleanup_range(lg, lg->saved_id);
2762 : }
2763 6661 : GDKfree(updated);
2764 6661 : return res == LOG_ERR ? GDK_FAIL : GDK_SUCCEED;
2765 : }
2766 :
2767 : /* Clean-up write-ahead log files already persisted in the BATs, leaving only the most recent one.
2768 : * Only the bak- files are deleted for the preserved WAL files.
2769 : */
2770 : lng
2771 28682 : log_changes(logger *lg)
2772 : {
2773 28682 : if (LOG_DISABLED(lg))
2774 : return 0;
2775 28682 : rotation_lock(lg);
2776 28682 : lng changes = lg->id - lg->saved_id - 1;
2777 28682 : rotation_unlock(lg);
2778 28682 : return changes;
2779 : }
2780 :
2781 : int
2782 462 : log_sequence(logger *lg, int seq, lng *id)
2783 : {
2784 462 : log_lock(lg);
2785 462 : BUN p = log_find(lg->seqs_id, lg->dseqs, seq);
2786 :
2787 462 : if (p != BUN_NONE) {
2788 343 : *id = *(lng *) Tloc(lg->seqs_val, p);
2789 :
2790 343 : log_unlock(lg);
2791 343 : return 1;
2792 : }
2793 119 : log_unlock(lg);
2794 119 : return 0;
2795 : }
2796 :
2797 : gdk_return
2798 192099 : log_constant(logger *lg, int type, const void *val, log_id id, lng offset, lng cnt)
2799 : {
2800 192099 : bte tpe = find_type(lg, type);
2801 192099 : gdk_return ok = GDK_SUCCEED;
2802 192099 : logformat l;
2803 192099 : lng nr;
2804 192099 : l.flag = LOG_UPDATE_CONST;
2805 192099 : l.id = id;
2806 192099 : nr = cnt;
2807 :
2808 192099 : if (LOG_DISABLED(lg) || !nr) {
2809 : /* logging is switched off */
2810 68071 : if (nr) {
2811 68071 : log_lock(lg);
2812 68071 : ok = la_bat_update_count(lg, id, offset + cnt, lg->tid);
2813 68071 : log_unlock(lg);
2814 : }
2815 68071 : return ok;
2816 : }
2817 :
2818 124028 : gdk_return(*wt) (const void *, stream *, size_t) = BATatoms[type].atomWrite;
2819 :
2820 124028 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
2821 248056 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR ||
2822 248056 : log_write_format(lg, &l) != GDK_SUCCEED ||
2823 248056 : !mnstr_writeLng(lg->current->output_log, nr) ||
2824 248056 : mnstr_write(lg->current->output_log, &tpe, 1, 1) != 1 ||
2825 124028 : !mnstr_writeLng(lg->current->output_log, offset)) {
2826 0 : ATOMIC_DEC(&lg->current->refcount);
2827 0 : ok = GDK_FAIL;
2828 0 : goto bailout;
2829 : }
2830 :
2831 124028 : ok = wt(val, lg->current->output_log, 1);
2832 :
2833 124028 : TRC_DEBUG(WAL, "Logged %d " LLFMT " inserts\n", id, nr);
2834 :
2835 124028 : bailout:
2836 124028 : if (ok != GDK_SUCCEED) {
2837 0 : const char *err = mnstr_peek_error(lg->current->output_log);
2838 0 : TRC_CRITICAL(GDK, "write failed%s%s\n", err ? ": " : "", err ? err : "");
2839 : }
2840 : return ok;
2841 : }
2842 :
2843 : static gdk_return
2844 19476 : string_writer(logger *lg, BAT *b, lng offset, lng nr)
2845 : {
2846 19476 : size_t bufsz = lg->wbufsize, resize = 0;
2847 19476 : BUN end = (BUN) (offset + nr);
2848 19476 : char *buf = lg->wbuf;
2849 19476 : gdk_return res = GDK_SUCCEED;
2850 :
2851 19476 : if (!buf)
2852 : return GDK_FAIL;
2853 19476 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
2854 19476 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR)
2855 : return GDK_FAIL;
2856 19476 : BATiter bi = bat_iterator(b);
2857 19476 : BUN p = (BUN) offset;
2858 39004 : for (; p < end;) {
2859 19528 : size_t sz = 0;
2860 19528 : if (resize) {
2861 2 : if ((buf = GDKrealloc(lg->wbuf, resize)) == NULL) {
2862 : res = GDK_FAIL;
2863 : break;
2864 : }
2865 2 : lg->wbuf = buf;
2866 2 : lg->wbufsize = bufsz = resize;
2867 2 : resize = 0;
2868 : }
2869 19528 : char *dst = buf;
2870 93499 : for (; p < end && sz < bufsz; p++) {
2871 74023 : const char *s = BUNtvar(bi, p);
2872 74023 : size_t len = strlen(s) + 1;
2873 74023 : if ((sz + len) > bufsz) {
2874 52 : if (len > bufsz)
2875 2 : resize = len + bufsz;
2876 : break;
2877 : } else {
2878 73971 : memcpy(dst, s, len);
2879 73971 : dst += len;
2880 73971 : sz += len;
2881 : }
2882 : }
2883 39054 : if (sz &&
2884 39052 : (!mnstr_writeLng(lg->current->output_log, (lng) sz) ||
2885 19526 : mnstr_write(lg->current->output_log, buf, sz, 1) != 1)) {
2886 : res = GDK_FAIL;
2887 : break;
2888 : }
2889 : }
2890 19476 : bat_iterator_end(&bi);
2891 19476 : return res;
2892 : }
2893 :
2894 : static gdk_return
2895 503051 : internal_log_bat(logger *lg, BAT *b, log_id id, lng offset, lng cnt, int sliced, lng total_cnt)
2896 : {
2897 503051 : bte tpe = find_type(lg, b->ttype);
2898 503051 : gdk_return ok = GDK_SUCCEED;
2899 503051 : logformat l;
2900 503051 : BUN p;
2901 503051 : lng nr;
2902 503051 : l.flag = LOG_UPDATE_BULK;
2903 503051 : l.id = id;
2904 503051 : nr = cnt;
2905 :
2906 503051 : if (LOG_DISABLED(lg) || !nr) {
2907 : /* logging is switched off */
2908 204642 : if (nr)
2909 150381 : return la_bat_update_count(lg, id, offset + cnt, lg->tid);
2910 : return GDK_SUCCEED;
2911 : }
2912 :
2913 268921 : gdk_return(*wt) (const void *, stream *, size_t) = BATatoms[b->ttype].atomWrite;
2914 :
2915 268921 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
2916 268921 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR) {
2917 0 : ok = GDK_FAIL;
2918 0 : goto bailout;
2919 : }
2920 :
2921 268921 : if (lg->total_cnt == 0) /* signals single bulk message or first part of bat logged in parts */
2922 537634 : if (log_write_format(lg, &l) != GDK_SUCCEED ||
2923 666207 : !mnstr_writeLng(lg->current->output_log, total_cnt ? total_cnt : cnt) ||
2924 537634 : mnstr_write(lg->current->output_log, &tpe, 1, 1) != 1 ||
2925 409061 : !mnstr_writeLng(lg->current->output_log, total_cnt ? -1 : offset)) { /* offset = -1 indicates bat was logged in parts */
2926 0 : ok = GDK_FAIL;
2927 0 : goto bailout;
2928 : }
2929 268921 : if (!total_cnt)
2930 128573 : total_cnt = cnt;
2931 268921 : lg->total_cnt += cnt;
2932 :
2933 268921 : if (lg->total_cnt == total_cnt) /* This is the last to be logged part of this bat, we can already reset the total_cnt */
2934 268817 : lg->total_cnt = 0;
2935 :
2936 : /* if offset is just for the log, but BAT is already sliced, reset offset */
2937 268921 : if (sliced)
2938 1512 : offset = 0;
2939 268921 : if (b->ttype == TYPE_msk) {
2940 0 : BATiter bi = bat_iterator(b);
2941 0 : if (offset % 32 == 0) {
2942 0 : if (!mnstr_writeIntArray(lg->current->output_log, (int *) ((char *) bi.base + offset / 32),
2943 0 : (size_t) ((nr + 31) / 32)))
2944 0 : ok = GDK_FAIL;
2945 : } else {
2946 0 : for (lng i = 0; i < nr; i += 32) {
2947 : uint32_t v = 0;
2948 0 : for (int j = 0; j < 32 && i + j < nr; j++)
2949 0 : v |= (uint32_t) Tmskval(&bi, (BUN) (offset + i + j)) << j;
2950 0 : if (!mnstr_writeInt(lg->current->output_log, (int) v)) {
2951 : ok = GDK_FAIL;
2952 : break;
2953 : }
2954 : }
2955 : }
2956 0 : bat_iterator_end(&bi);
2957 517730 : } else if (b->ttype < TYPE_str && !isVIEW(b)) {
2958 248809 : BATiter bi = bat_iterator(b);
2959 248809 : const void *t = BUNtail(bi, (BUN) offset);
2960 :
2961 248809 : ok = wt(t, lg->current->output_log, (size_t) nr);
2962 248809 : bat_iterator_end(&bi);
2963 20112 : } else if (b->ttype == TYPE_str) {
2964 : /* efficient string writes */
2965 19470 : ok = string_writer(lg, b, offset, nr);
2966 : } else {
2967 642 : BATiter bi = bat_iterator(b);
2968 642 : BUN end = (BUN) (offset + nr);
2969 1642 : for (p = (BUN) offset; p < end && ok == GDK_SUCCEED; p++) {
2970 1000 : const void *t = BUNtail(bi, p);
2971 :
2972 1000 : ok = wt(t, lg->current->output_log, 1);
2973 : }
2974 642 : bat_iterator_end(&bi);
2975 : }
2976 :
2977 268921 : TRC_DEBUG(WAL, "Logged %d " LLFMT " inserts\n", id, nr);
2978 :
2979 268921 : bailout:
2980 268921 : if (ok != GDK_SUCCEED) {
2981 0 : ATOMIC_DEC(&lg->current->refcount);
2982 0 : const char *err = mnstr_peek_error(lg->current->output_log);
2983 0 : TRC_CRITICAL(GDK, "write failed%s%s\n", err ? ": " : "", err ? err : "");
2984 : }
2985 : return ok;
2986 : }
2987 :
2988 : /*
2989 : * Changes made to the BAT descriptor should be stored in the log
2990 : * files. Actually, we need to save the descriptor file, perhaps we
2991 : * should simply introduce a versioning scheme.
2992 : */
2993 : gdk_return
2994 235764 : log_bat_persists(logger *lg, BAT *b, log_id id)
2995 : {
2996 235764 : log_lock(lg);
2997 235764 : bte ta = find_type(lg, b->ttype);
2998 235764 : logformat l;
2999 :
3000 235764 : if (log_add_bat(lg, b, id, -1) != GDK_SUCCEED) {
3001 0 : log_unlock(lg);
3002 0 : if (!LOG_DISABLED(lg))
3003 0 : ATOMIC_DEC(&lg->current->refcount);
3004 0 : return GDK_FAIL;
3005 : }
3006 :
3007 235764 : l.flag = LOG_CREATE;
3008 235764 : l.id = id;
3009 235764 : if (!LOG_DISABLED(lg)) {
3010 156537 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
3011 313074 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR ||
3012 313074 : log_write_format(lg, &l) != GDK_SUCCEED ||
3013 156537 : mnstr_write(lg->current->output_log, &ta, 1, 1) != 1) {
3014 0 : log_unlock(lg);
3015 0 : ATOMIC_DEC(&lg->current->refcount);
3016 0 : return GDK_FAIL;
3017 : }
3018 : }
3019 235764 : TRC_DEBUG(WAL, "id (%d) bat (%d)\n", id, b->batCacheid);
3020 235764 : gdk_return r = internal_log_bat(lg, b, id, 0, BATcount(b), 0, 0);
3021 235764 : log_unlock(lg);
3022 235764 : if (r != GDK_SUCCEED)
3023 0 : ATOMIC_DEC(&lg->current->refcount);
3024 : return r;
3025 : }
3026 :
3027 : gdk_return
3028 22132 : log_bat_transient(logger *lg, log_id id)
3029 : {
3030 22132 : log_lock(lg);
3031 22132 : log_bid bid = internal_find_bat(lg, id, -1);
3032 22132 : logformat l;
3033 :
3034 22132 : if (bid < 0) {
3035 0 : log_unlock(lg);
3036 0 : return GDK_FAIL;
3037 : }
3038 22132 : if (!bid) {
3039 0 : GDKerror("log_bat_transient failed to find bid for object %d\n", id);
3040 0 : log_unlock(lg);
3041 0 : return GDK_FAIL;
3042 : }
3043 22132 : l.flag = LOG_DESTROY;
3044 22132 : l.id = id;
3045 :
3046 22132 : if (!LOG_DISABLED(lg)) {
3047 9637 : if (log_write_format(lg, &l) != GDK_SUCCEED) {
3048 0 : TRC_CRITICAL(GDK, "write failed\n");
3049 0 : log_unlock(lg);
3050 0 : ATOMIC_DEC(&lg->current->refcount);
3051 0 : return GDK_FAIL;
3052 : }
3053 : }
3054 22132 : TRC_DEBUG(WAL, "Logged destroyed bat (%d) %d\n", id, bid);
3055 22132 : BAT *b = BBPquickdesc(bid);
3056 22132 : assert(b);
3057 22132 : BUN cnt = BATcount(b);
3058 22132 : ATOMIC_ADD(&lg->current->drops, cnt);
3059 22132 : gdk_return r = log_del_bat(lg, bid);
3060 22132 : log_unlock(lg);
3061 22132 : if (r != GDK_SUCCEED)
3062 0 : ATOMIC_DEC(&lg->current->refcount);
3063 : return r;
3064 : }
3065 :
3066 : static gdk_return
3067 117206 : log_bat_group(logger *lg, log_id id)
3068 : {
3069 117206 : if (LOG_DISABLED(lg))
3070 : return GDK_SUCCEED;
3071 :
3072 76380 : logformat l;
3073 76380 : l.flag = LOG_BAT_GROUP;
3074 76380 : l.id = id;
3075 76380 : gdk_return r = log_write_format(lg, &l);
3076 76380 : return r;
3077 : }
3078 :
3079 : gdk_return
3080 58603 : log_bat_group_start(logger *lg, log_id id)
3081 : {
3082 : /*positive table id represent start of logged table */
3083 58603 : return log_bat_group(lg, id);
3084 : }
3085 :
3086 : gdk_return
3087 58603 : log_bat_group_end(logger *lg, log_id id)
3088 : {
3089 : /*negative table id represent end of logged table */
3090 58603 : return log_bat_group(lg, -id);
3091 : }
3092 :
3093 : gdk_return
3094 264576 : log_bat(logger *lg, BAT *b, log_id id, lng offset, lng cnt, lng total_cnt)
3095 : {
3096 264576 : log_lock(lg);
3097 264576 : gdk_return r = internal_log_bat(lg, b, id, offset, cnt, 0, total_cnt);
3098 264576 : log_unlock(lg);
3099 264576 : return r;
3100 : }
3101 :
3102 : gdk_return
3103 2771 : log_delta(logger *lg, BAT *uid, BAT *uval, log_id id)
3104 : {
3105 2771 : log_lock(lg);
3106 2771 : bte tpe = find_type(lg, uval->ttype);
3107 2771 : gdk_return ok = GDK_SUCCEED;
3108 2771 : logformat l;
3109 2771 : BUN p;
3110 2771 : lng nr;
3111 :
3112 2771 : if (BATtdense(uid)) {
3113 2711 : ok = internal_log_bat(lg, uval, id, uid->tseqbase, BATcount(uval), 1, 0);
3114 2711 : log_unlock(lg);
3115 2711 : if (!LOG_DISABLED(lg) && ok != GDK_SUCCEED)
3116 0 : ATOMIC_DEC(&lg->current->refcount);
3117 2711 : return ok;
3118 : }
3119 :
3120 60 : assert(uid->ttype == TYPE_oid || uid->ttype == TYPE_void);
3121 :
3122 60 : l.flag = LOG_UPDATE;
3123 60 : l.id = id;
3124 60 : nr = (BATcount(uval));
3125 60 : assert(nr);
3126 :
3127 60 : if (LOG_DISABLED(lg)) {
3128 : /* logging is switched off */
3129 41 : log_unlock(lg);
3130 41 : return GDK_SUCCEED;
3131 : }
3132 :
3133 19 : BATiter vi = bat_iterator(uval);
3134 19 : gdk_return(*wh) (const void *, stream *, size_t) = BATatoms[TYPE_oid].atomWrite;
3135 19 : gdk_return(*wt) (const void *, stream *, size_t) = BATatoms[uval->ttype].atomWrite;
3136 :
3137 19 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
3138 38 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR ||
3139 38 : log_write_format(lg, &l) != GDK_SUCCEED ||
3140 38 : !mnstr_writeLng(lg->current->output_log, nr) ||
3141 19 : mnstr_write(lg->current->output_log, &tpe, 1, 1) != 1) {
3142 0 : ok = GDK_FAIL;
3143 0 : goto bailout;
3144 : }
3145 1074 : for (p = 0; p < BATcount(uid) && ok == GDK_SUCCEED; p++) {
3146 1055 : const oid id = BUNtoid(uid, p);
3147 :
3148 1055 : ok = wh(&id, lg->current->output_log, 1);
3149 : }
3150 19 : if (uval->ttype == TYPE_msk) {
3151 0 : if (!mnstr_writeIntArray(lg->current->output_log, vi.base,
3152 0 : (BATcount(uval) + 31) / 32))
3153 0 : ok = GDK_FAIL;
3154 32 : } else if (uval->ttype < TYPE_str && !isVIEW(uval)) {
3155 13 : const void *t = BUNtail(vi, 0);
3156 :
3157 13 : ok = wt(t, lg->current->output_log, (size_t) nr);
3158 6 : } else if (uval->ttype == TYPE_str) {
3159 : /* efficient string writes */
3160 6 : ok = string_writer(lg, uval, 0, nr);
3161 : } else {
3162 0 : for (p = 0; p < BATcount(uid) && ok == GDK_SUCCEED; p++) {
3163 0 : const void *val = BUNtail(vi, p);
3164 :
3165 0 : ok = wt(val, lg->current->output_log, 1);
3166 : }
3167 : }
3168 :
3169 19 : TRC_DEBUG(WAL, "Logged %d " LLFMT " inserts\n", id, nr);
3170 :
3171 19 : bailout:
3172 19 : bat_iterator_end(&vi);
3173 19 : if (ok != GDK_SUCCEED) {
3174 0 : const char *err = mnstr_peek_error(lg->current->output_log);
3175 0 : TRC_CRITICAL(GDK, "write failed%s%s\n", err ? ": " : "", err ? err : "");
3176 0 : ATOMIC_DEC(&lg->current->refcount);
3177 : }
3178 19 : log_unlock(lg);
3179 19 : return ok;
3180 : }
3181 :
3182 : static inline bool
3183 56857 : check_rotation_conditions(logger *lg)
3184 : {
3185 56857 : if (LOG_DISABLED(lg))
3186 : return false;
3187 :
3188 56854 : if (lg->current->next)
3189 : return false; /* do not rotate if there is already a prepared next current */
3190 56854 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR)
3191 : return true;
3192 56854 : const lng current_file_size = (lng) getfilepos(getFile(lg->current->output_log));
3193 :
3194 56854 : if (current_file_size == -1)
3195 : return false;
3196 :
3197 56854 : assert(current_file_size >= 0);
3198 :
3199 56854 : if (current_file_size == 2)
3200 : return false;
3201 :
3202 2749 : bool res = (lg->saved_id + 1 >= lg->id && ATOMIC_GET(&lg->current->drops) > (ulng)lg->max_dropped) ||
3203 57211 : current_file_size > lg->max_file_size ||
3204 49980 : (GDKusec() - lg->file_age) > lg->max_file_age;
3205 :
3206 54469 : return res;
3207 : }
3208 :
3209 : gdk_return
3210 62462 : log_tend(logger *lg)
3211 : {
3212 62462 : TRC_DEBUG(WAL, "tend %d\n", lg->tid);
3213 :
3214 62462 : if (LOG_DISABLED(lg))
3215 : return GDK_SUCCEED;
3216 :
3217 56854 : gdk_return result;
3218 56854 : logformat l;
3219 56854 : l.flag = LOG_END;
3220 56854 : l.id = lg->tid;
3221 :
3222 56854 : if ((result = log_write_format(lg, &l)) == GDK_SUCCEED)
3223 56854 : ATOMIC_INC(&lg->nr_flushers);
3224 : return result;
3225 : }
3226 :
3227 : #define flush_lock(lg) MT_lock_set(&(lg)->flush_lock)
3228 : #define flush_unlock(lg) MT_lock_unset(&(lg)->flush_lock)
3229 :
3230 : static inline gdk_return
3231 56126 : do_flush(logged_range *range)
3232 : {
3233 : /* assumes flush lock */
3234 56126 : stream *output_log = range->output_log;
3235 56126 : ulng ts = ATOMIC_GET(&range->last_ts);
3236 :
3237 56126 : if (mnstr_flush(output_log, MNSTR_FLUSH_DATA) ||
3238 56126 : (!(ATOMIC_GET(&GDKdebug) & NOSYNCMASK) && mnstr_fsync(output_log)))
3239 0 : return GDK_FAIL;
3240 56126 : ATOMIC_SET(&range->flushed_ts, ts);
3241 56126 : return GDK_SUCCEED;
3242 : }
3243 :
3244 : static inline void
3245 62459 : log_tdone(logger *lg, logged_range *range, ulng commit_ts)
3246 : {
3247 62459 : (void) lg;
3248 62459 : TRC_DEBUG(WAL, "tdone " LLFMT "\n", commit_ts);
3249 :
3250 62459 : if ((ulng) ATOMIC_GET(&range->last_ts) < commit_ts)
3251 61731 : ATOMIC_SET(&range->last_ts, commit_ts);
3252 62459 : }
3253 :
3254 : gdk_return
3255 62462 : log_tflush(logger *lg, ulng file_id, ulng commit_ts)
3256 : {
3257 62462 : rotation_lock(lg);
3258 62462 : if (lg->flushnow) {
3259 5605 : logged_range *p = lg->current;
3260 5605 : assert(lg->flush_ranges == lg->current);
3261 5605 : assert(ATOMIC_GET(&lg->current->flushed_ts) == ATOMIC_GET(&lg->current->last_ts));
3262 5605 : log_tdone(lg, lg->current, commit_ts);
3263 5605 : ATOMIC_SET(&lg->current->flushed_ts, commit_ts);
3264 5605 : lg->id++;
3265 5605 : lg->flushnow = false;
3266 5605 : if (log_open_output(lg) != GDK_SUCCEED)
3267 0 : GDKfatal("Could not create new log file\n"); /* TODO: does not have to be fatal (yet) */
3268 5605 : do_rotate(lg);
3269 5605 : (void) do_flush_range_cleanup(lg);
3270 5605 : assert(lg->flush_ranges == lg->current);
3271 5605 : rotation_unlock(lg);
3272 5605 : return log_commit(lg, p, NULL, 0);
3273 : }
3274 :
3275 56857 : if (LOG_DISABLED(lg)) {
3276 3 : rotation_unlock(lg);
3277 3 : return GDK_SUCCEED;
3278 : }
3279 :
3280 56854 : logged_range *frange = do_flush_range_cleanup(lg);
3281 :
3282 57178 : while (frange->next && frange->id < file_id) {
3283 : assert(frange->next);
3284 : frange = frange->next;
3285 : }
3286 :
3287 56854 : log_tdone(lg, frange, commit_ts);
3288 :
3289 56854 : if ((ulng) ATOMIC_GET(&frange->flushed_ts) < commit_ts) {
3290 : /* delay needed ? */
3291 :
3292 56126 : flush_lock(lg);
3293 : /* check it one more time */
3294 56126 : if ((ulng) ATOMIC_GET(&frange->flushed_ts) < commit_ts)
3295 56126 : do_flush(frange);
3296 56126 : flush_unlock(lg);
3297 : }
3298 : /* else somebody else has flushed our log file */
3299 :
3300 56854 : if (ATOMIC_DEC(&frange->refcount) == 1 && !LOG_DISABLED(lg)) {
3301 54910 : if (frange != lg->current && frange->output_log) {
3302 208 : close_stream(frange->output_log);
3303 208 : frange->output_log = NULL;
3304 : }
3305 : }
3306 :
3307 56854 : if (ATOMIC_DEC(&lg->nr_flushers) == 0) {
3308 : /* I am the last flusher
3309 : * if present,
3310 : * wake up the exclusive flusher in log_tstart */
3311 : /* rotation_lock is still being held */
3312 55159 : MT_cond_signal(&lg->excl_flush_cv);
3313 : }
3314 56854 : rotation_unlock(lg);
3315 :
3316 56854 : return GDK_SUCCEED;
3317 : }
3318 :
3319 : static gdk_return
3320 6900 : log_tsequence_(logger *lg, int seq, lng val)
3321 : {
3322 6900 : logformat l;
3323 :
3324 6900 : if (LOG_DISABLED(lg))
3325 : return GDK_SUCCEED;
3326 2832 : l.flag = LOG_SEQ;
3327 2832 : l.id = seq;
3328 :
3329 2832 : TRC_DEBUG(WAL, "tsequence(%d," LLFMT ")\n", seq, val);
3330 :
3331 2832 : assert(mnstr_errnr(lg->current->output_log) == MNSTR_NO__ERROR);
3332 5664 : if (mnstr_errnr(lg->current->output_log) != MNSTR_NO__ERROR ||
3333 5664 : log_write_format(lg, &l) != GDK_SUCCEED ||
3334 2832 : !mnstr_writeLng(lg->current->output_log, val)) {
3335 0 : TRC_CRITICAL(GDK, "write failed\n");
3336 0 : ATOMIC_DEC(&lg->current->refcount);
3337 0 : return GDK_FAIL;
3338 : }
3339 : return GDK_SUCCEED;
3340 : }
3341 :
3342 : /* a transaction in it self */
3343 : gdk_return
3344 6900 : log_tsequence(logger *lg, int seq, lng val)
3345 : {
3346 6900 : BUN p;
3347 :
3348 6900 : TRC_DEBUG(WAL, "tsequence(%d," LLFMT ")\n", seq, val);
3349 :
3350 6900 : log_lock(lg);
3351 6900 : MT_lock_set(&lg->seqs_id->theaplock);
3352 6900 : BUN inserted = lg->seqs_id->batInserted;
3353 6900 : MT_lock_unset(&lg->seqs_id->theaplock);
3354 6900 : if ((p = log_find(lg->seqs_id, lg->dseqs, seq)) != BUN_NONE && p >= inserted) {
3355 1629 : assert(lg->seqs_val->hseqbase == 0);
3356 1629 : if (BUNreplace(lg->seqs_val, p, &val, true) != GDK_SUCCEED) {
3357 0 : log_unlock(lg);
3358 0 : return GDK_FAIL;
3359 : }
3360 : } else {
3361 4909 : if (p != BUN_NONE) {
3362 4909 : oid pos = p;
3363 4909 : if (BUNappend(lg->dseqs, &pos, true) != GDK_SUCCEED) {
3364 0 : log_unlock(lg);
3365 0 : return GDK_FAIL;
3366 : }
3367 : }
3368 10542 : if (BUNappend(lg->seqs_id, &seq, true) != GDK_SUCCEED ||
3369 5271 : BUNappend(lg->seqs_val, &val, true) != GDK_SUCCEED) {
3370 0 : log_unlock(lg);
3371 0 : return GDK_FAIL;
3372 : }
3373 : }
3374 6900 : gdk_return r = log_tsequence_(lg, seq, val);
3375 6900 : log_unlock(lg);
3376 6900 : return r;
3377 : }
3378 :
3379 : static gdk_return
3380 12598 : bm_commit(logger *lg, logged_range *pending, uint32_t *updated, BUN maxupdated)
3381 : {
3382 12598 : log_lock(lg);
3383 12598 : BAT *b = lg->catalog_bid;
3384 12598 : const log_bid *bids;
3385 :
3386 12598 : bids = (log_bid *) Tloc(b, 0);
3387 248307 : for (BUN p = b->batInserted, cnt = pending ? pending->cnt : BATcount(b); p < cnt; p++) {
3388 235709 : log_bid bid = bids[p];
3389 235709 : BAT *lb = BBP_desc(bid);
3390 :
3391 235709 : assert(bid);
3392 235709 : if (lb->batCacheid == 0 || BATmode(lb, false) != GDK_SUCCEED) {
3393 0 : GDKwarning("Failed to set bat (%d%s) persistent\n", bid, !lb ? " gone" : "");
3394 0 : log_unlock(lg);
3395 0 : return GDK_FAIL;
3396 : }
3397 :
3398 235709 : assert(lb->batRestricted != BAT_WRITE);
3399 :
3400 235709 : TRC_DEBUG(WAL, "create %d (%d)\n", bid, BBP_lrefs(bid));
3401 : }
3402 : /* bm_subcommit releases the lock */
3403 12598 : return bm_subcommit(lg, pending, updated, maxupdated);
3404 : }
3405 :
3406 : static gdk_return
3407 236017 : log_add_bat(logger *lg, BAT *b, log_id id, int tid)
3408 : {
3409 236017 : log_bid bid = internal_find_bat(lg, id, tid);
3410 236017 : lng cnt = 0;
3411 236017 : lng lid = lng_nil;
3412 :
3413 236017 : assert(b->batRestricted != BAT_WRITE);
3414 236017 : assert(b->batRole == PERSISTENT);
3415 236017 : if (bid < 0)
3416 : return GDK_FAIL;
3417 236017 : if (bid) {
3418 155075 : if (bid != b->batCacheid) {
3419 155072 : if (log_del_bat(lg, bid) != GDK_SUCCEED)
3420 : return GDK_FAIL;
3421 : } else {
3422 : return GDK_SUCCEED;
3423 : }
3424 : }
3425 236014 : bid = b->batCacheid;
3426 236014 : TRC_DEBUG(WAL, "create %d\n", id);
3427 236014 : assert(log_find(lg->catalog_bid, lg->dcatalog, bid) == BUN_NONE);
3428 472028 : if (BUNappend(lg->catalog_bid, &bid, true) != GDK_SUCCEED ||
3429 472028 : BUNappend(lg->catalog_id, &id, true) != GDK_SUCCEED ||
3430 472028 : BUNappend(lg->catalog_cnt, &cnt, false) != GDK_SUCCEED ||
3431 236014 : BUNappend(lg->catalog_lid, &lid, false) != GDK_SUCCEED)
3432 0 : return GDK_FAIL;
3433 236014 : if (lg->current)
3434 235761 : lg->current->cnt++;
3435 236014 : BBPretain(bid);
3436 236014 : return GDK_SUCCEED;
3437 : }
3438 :
3439 : static gdk_return
3440 177244 : log_del_bat(logger *lg, log_bid bid)
3441 : {
3442 177244 : BUN p = log_find(lg->catalog_bid, lg->dcatalog, bid);
3443 177244 : lng lid = lg->tid;
3444 :
3445 177244 : assert(p != BUN_NONE);
3446 177244 : if (p == BUN_NONE) {
3447 : GDKerror("cannot find BAT\n");
3448 : return GDK_FAIL;
3449 : }
3450 :
3451 177244 : assert(lg->catalog_lid->hseqbase == 0);
3452 177244 : return BUNreplace(lg->catalog_lid, p, &lid, false);
3453 : }
3454 :
3455 : /* returns -1 on failure, 0 when not found, > 0 when found */
3456 : log_bid
3457 27652 : log_find_bat(logger *lg, log_id id)
3458 : {
3459 27652 : log_lock(lg);
3460 27652 : log_bid bid = internal_find_bat(lg, id, -1);
3461 27652 : log_unlock(lg);
3462 27652 : if (!bid) {
3463 0 : GDKerror("logger_find_bat failed to find bid for object %d\n", id);
3464 0 : return GDK_FAIL;
3465 : }
3466 : return bid;
3467 : }
3468 :
3469 :
3470 :
3471 : gdk_return
3472 62463 : log_tstart(logger *lg, bool flushnow, ulng *file_id)
3473 : {
3474 62463 : rotation_lock(lg);
3475 62463 : if (flushnow) {
3476 5606 : if (file_id == NULL) {
3477 : /* special case: ask store_manager to rotate log file */
3478 1 : lg->file_age = 0;
3479 1 : rotation_unlock(lg);
3480 1 : return GDK_SUCCEED;
3481 : }
3482 : /* I am now the exclusive flusher */
3483 5605 : while (ATOMIC_GET(&lg->nr_flushers)) {
3484 : /* I am waiting until all existing flushers are done */
3485 0 : MT_cond_wait(&lg->excl_flush_cv, &lg->rotation_lock);
3486 : }
3487 5605 : assert(ATOMIC_GET(&lg->nr_flushers) == 0);
3488 :
3489 5605 : if (ATOMIC_GET(&lg->current->last_ts)) {
3490 2190 : lg->id++;
3491 2190 : if (log_open_output(lg) != GDK_SUCCEED)
3492 0 : GDKfatal("Could not create new log file\n"); /* TODO: does not have to be fatal (yet) */
3493 : }
3494 5605 : do_rotate(lg);
3495 5605 : (void) do_flush_range_cleanup(lg);
3496 5605 : rotation_unlock(lg);
3497 :
3498 5605 : if (lg->saved_id + 1 < lg->id)
3499 5120 : log_flush(lg, (1ULL << 63));
3500 5605 : lg->flushnow = flushnow;
3501 : } else {
3502 56857 : if (check_rotation_conditions(lg)) {
3503 4490 : lg->id++;
3504 4490 : if (log_open_output(lg) != GDK_SUCCEED)
3505 0 : GDKfatal("Could not create new log file\n"); /* TODO: does not have to be fatal (yet) */
3506 : }
3507 56857 : do_rotate(lg);
3508 56857 : rotation_unlock(lg);
3509 : }
3510 :
3511 62462 : if (LOG_DISABLED(lg))
3512 : return GDK_SUCCEED;
3513 :
3514 56854 : ATOMIC_INC(&lg->current->refcount);
3515 56854 : *file_id = lg->current->id;
3516 56854 : logformat l;
3517 56854 : l.flag = LOG_START;
3518 56854 : l.id = ++lg->tid;
3519 :
3520 56854 : TRC_DEBUG(WAL, "tstart %d\n", lg->tid);
3521 56854 : if (log_write_format(lg, &l) != GDK_SUCCEED) {
3522 0 : ATOMIC_DEC(&lg->current->refcount);
3523 0 : return GDK_FAIL;
3524 : }
3525 :
3526 : return GDK_SUCCEED;
3527 : }
3528 :
3529 : void
3530 116 : log_printinfo(logger *lg)
3531 : {
3532 116 : if (!rotation_trylock(lg, 1000)) {
3533 0 : printf("Logger is currently locked, so no logger information\n");
3534 0 : return;
3535 : }
3536 116 : printf("logger %s:\n", lg->fn);
3537 116 : printf("current log file "ULLFMT", last handled log file "ULLFMT"\n",
3538 : lg->id, lg->saved_id);
3539 116 : printf("current transaction id %d, saved transaction id %d\n",
3540 : lg->tid, lg->saved_tid);
3541 116 : printf("number of flushers: %d\n", (int) ATOMIC_GET(&lg->nr_flushers));
3542 116 : printf("number of catalog entries "BUNFMT", of which "BUNFMT" deleted\n",
3543 116 : lg->catalog_bid->batCount, lg->dcatalog->batCount);
3544 300 : for (logged_range *p = lg->pending; p; p = p->next) {
3545 184 : char buf[32];
3546 184 : if ((lg->debug & 128 || lg->inmemory) ||
3547 184 : p->output_log == NULL ||
3548 116 : snprintf(buf, sizeof(buf), ", file size %"PRIu64, (uint64_t) getfilepos(getFile(lg->current->output_log))) >= (int) sizeof(buf))
3549 68 : buf[0] = 0;
3550 252 : printf("pending range "ULLFMT": drops %"PRIu64", last_ts %"PRIu64", flushed_ts %"PRIu64", refcount %"PRIu64"%s%s\n", p->id, (uint64_t) ATOMIC_GET(&p->drops), (uint64_t) ATOMIC_GET(&p->last_ts), (uint64_t) ATOMIC_GET(&p->flushed_ts), (uint64_t) ATOMIC_GET(&p->refcount), buf, p == lg->current ? " (current)" : "");
3551 : }
3552 116 : rotation_unlock(lg);
3553 : }
|
