Update of /cvsroot/monetdb/MonetDB/src/gdk
In directory sc8-pr-cvs16.sourceforge.net:/tmp/cvs-serv6359/src/gdk
Modified Files:
gdk_logger.mx
Log Message:
The logger basically has 2 layers of transactions
(the log users (sql or xquery) level (1) and the logger level transactions (2))
Within a single (2) transaction snapshots can come and go. The administration
via the catalog bat (ie a persistent bat with delta management) wasn't
enough. Therefor a new 'freed' bat is kept which stores snapshots, created
in the current level (2) transaction, and freed in the current level (1)
transaction.
This solves the constent growing dbfarm for the sql sequence of
insert into/delete all.
Index: gdk_logger.mx
===================================================================
RCS file: /cvsroot/monetdb/MonetDB/src/gdk/gdk_logger.mx,v
retrieving revision 1.11
retrieving revision 1.12
diff -u -d -r1.11 -r1.12
--- gdk_logger.mx 3 Dec 2007 08:32:18 -0000 1.11
+++ gdk_logger.mx 23 Dec 2007 20:59:30 -0000 1.12
@@ -107,6 +107,10 @@
BAT *catalog; /* int, str */
BAT *seqs; /* int, lng */
BAT *snapshots; /* int, int the bid and tid of snapshot bat */
+ BAT *freed; /* snapshots can be created and destroyed,
+ in a single logger transaction. Thse snapshot
+ bats should be freed directly (on
+ transaction commit). */
} logger;
#define BATSIZE 0
@@ -894,13 +898,14 @@
int res;
BATseqbase(n, 0);
+
/* first loop over deleted then over current and new */
for (p = b->batDeleted; p < b->batFirst; p++) {
bat col = *(log_bid *) BUNhead(bi, p);
str name = BBPname(col);
if (debug & 1)
- fprintf(stderr, "commit deleted %s (%d)\n", name, col);
+ fprintf(stderr, "commit deleted %s (%d) %s\n", name,
col, (b==catalog)?BUNtail(bi,p):"snapshot");
BUNappend(n, name, FALSE);
}
BATloop(b, p, q) {
@@ -908,7 +913,7 @@
str name = BBPname(col);
if (debug & 1)
- fprintf(stderr, "commit new %s (%d)\n", name, col);
+ fprintf(stderr, "commit new %s (%d) %s\n", name, col,
(b==catalog)?BUNtail(bi,p):"snapshot");
BUNappend(n, name, FALSE);
}
BUNappend(n, BBPname(catalog->batCacheid), FALSE);
@@ -1007,6 +1012,7 @@
}
lg->snapshots = BATdescriptor(snapshots);
}
+ lg->freed = BATnew(TYPE_int, TYPE_void, 1);
snprintf(filename, BUFSIZ, "%s%s", lg->dir, LOGFILE);
@@ -1097,6 +1103,7 @@
BBPdecref(lg->catalog->batCacheid, TRUE);
logbat_destroy(lg->catalog);
+ logbat_destroy(lg->freed);
}
GDKfree(lg->dir);
logger_close(lg);
@@ -1134,6 +1141,8 @@
if (GDKmove(lg->dir, LOGFILE, "bak", lg->dir, LOGFILE, ext) < 0)
return LOG_ERR;
+
+ lg->changes = 0;
} else {
GDKerror("logger could not open %s\n", filename);
return LOG_ERR;
@@ -1231,12 +1240,13 @@
log_write_string(lg, name) == LOG_ERR)
return LOG_ERR;
+ if (lg->debug & 1)
+ fprintf(stderr, "persists bat %s (%d) %s\n",
+ name, b->batCacheid, (flag==LOG_USE)?"use":"create");
+
if (flag == LOG_USE) {
- if (BUNfnd(lg->snapshots, &b->batCacheid) != BUN_NONE) {
+ if (BUNfnd(lg->snapshots, &b->batCacheid) != BUN_NONE)
BUNdelHead(lg->snapshots, &b->batCacheid, FALSE);
- } else {
- BBPincref(b->batCacheid, TRUE);
- }
BUNins(lg->snapshots, &b->batCacheid, &lg->tid, FALSE);
return LOG_OK;
}
@@ -1257,7 +1267,7 @@
return LOG_ERR;
if (lg->debug & 1)
- fprintf(stderr, "Logged new bat [%s,%s] %s " SZFMT "\n", ha,
ta, name, BATcount(b));
+ fprintf(stderr, "Logged new bat [%s,%s] %s " SZFMT " (%d)\n",
ha, ta, name, BATcount(b), b->batCacheid);
return log_bat(lg, b, name);
}
@@ -1480,7 +1490,27 @@
if (lg->debug)
fprintf(stderr, "log_tend %d\n", lg->tid);
- /* first subcommit the snapshots */
+ /* first subcommit the freed snapshots */
+ if (BATcount(lg->freed)) {
+ BUN p, q;
+ BAT *n = logbat_new(TYPE_void, TYPE_str, BATcount(lg->freed));
+
+ BATseqbase(n, 0);
+
+ BATloop(lg->freed, p, q) {
+ bat col = *(log_bid *) Hloc(lg->freed, p);
+ str name = BBPname(col);
+
+ if (lg->debug & 1)
+ fprintf(stderr, "commit deleted (snapshot) %s
(%d)\n", name, col);
+ BUNappend(n, name, FALSE);
+ }
+ res = TMsubcommit(n);
+ BBPreclaim(n);
+ BATclear(lg->freed);
+ BATcommit(lg->freed);
+ }
+ /* then subcommit the new snapshots */
if (DELTAdirty(lg->snapshots)) {
/* sub commit all new snapshots */
BAT *b = BATselect(lg->snapshots, &lg->tid, &lg->tid);
@@ -1561,6 +1591,10 @@
BATmode(lb, TRANSIENT);
BBPdecref(bid, TRUE);
logbat_destroy(lb);
+
+ if (lg->debug & 1)
+ fprintf(stderr, "bm_commit: delete %d (%d)\n",
+ bid, BBP[bid].lrefs);
}
for (p = b->batInserted; p < BUNlast(b); p++) {
@@ -1571,6 +1605,10 @@
if (BATcount(lb) > (size_t) REMAP_PAGE_MAXSIZE)
BATmmap(lb, STORE_MMAP, STORE_MMAP, STORE_MMAP);
logbat_destroy(lb);
+
+ if (lg->debug & 1)
+ fprintf(stderr, "bm_commit: create %d (%d)\n",
+ bid, BBP[bid].lrefs);
}
if (bm_subcommit(lg->catalog, lg->catalog, lg->debug) != 0)
return LOG_ERR;
@@ -1583,10 +1621,27 @@
log_bid bid = logger_find_bat(lg, name);
if (bid) {
- if (bid != b->batCacheid)
+ if (bid != b->batCacheid) {
+ /* if this is a not logger commited snapshot bat,
+ make it transient */
+ BUN p = BUNfnd(lg->catalog, &bid);
+
+ if (p >= lg->catalog->batInserted &&
+ BUNfnd(lg->snapshots, &bid) != BUN_NONE) {
+ BAT *lb = BATdescriptor(bid);
+
+ BUNdelHead(lg->snapshots, &bid, FALSE);
+ BATmode(lb, TRANSIENT);
+ BBPdecref(bid, TRUE); /* from catalog */
+ if (lg->debug & 1)
+ fprintf(stderr, "logger_add_bat release
snapshot %d (%d)\n", bid, BBP[bid].lrefs);
+ logbat_destroy(lb);
+ BUNins(lg->freed, &bid, NULL, FALSE);
+ }
BUNdelHead(lg->catalog, &bid, FALSE);
- else
+ } else {
return bid;
+ }
}
bid = b->batCacheid;
if (lg->debug)
-------------------------------------------------------------------------
This SF.net email is sponsored by: Microsoft
Defy all challenges. Microsoft(R) Visual Studio 2005.
http://clk.atdmt.com/MRT/go/vse0120000070mrt/direct/01/
_______________________________________________
Monetdb-checkins mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/monetdb-checkins
