Changeset: c4591b3a8631 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/c4591b3a8631
Modified Files:
gdk/gdk.h
gdk/gdk_bbp.c
sql/backends/monet5/sql.c
Branch: default
Log Message:
Merged with Jul2021
diffs (256 lines):
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -1528,7 +1528,7 @@ typedef struct {
int refs; /* in-memory references on which the loaded
status of a BAT relies */
int lrefs; /* logical references on which the existence of
a BAT relies */
ATOMIC_TYPE status; /* status mask used for spin locking */
- /* MT_Id pid; non-zero thread-id if this BAT is private */
+ MT_Id pid; /* creator of this bat while "private" */
} BBPrec;
gdk_export bat BBPlimit;
@@ -1919,6 +1919,7 @@ BBPcheck(bat x)
if (x < 0 || x >= getBBPsize() || BBP_logical(x) == NULL) {
TRC_DEBUG(CHECK_, "range error %d\n", (int) x);
} else {
+ assert(BBP_pid(x) == 0 || BBP_pid(x) == MT_getpid());
return x;
}
}
diff --git a/gdk/gdk_bbp.c b/gdk/gdk_bbp.c
--- a/gdk/gdk_bbp.c
+++ b/gdk/gdk_bbp.c
@@ -365,8 +365,10 @@ BBPextend(int idx, bool buildhash)
GDKerror("failed to extend BAT pool\n");
return GDK_FAIL;
}
- for (BUN i = 0; i < BBPINIT; i++)
+ for (BUN i = 0; i < BBPINIT; i++) {
ATOMIC_INIT(&BBP[limit][i].status, 0);
+ BBP[limit][i].pid = ~(MT_Id)0;
+ }
BBPlimit += BBPINIT;
}
@@ -753,6 +755,7 @@ BBPreadEntries(FILE *fp, unsigned bbpver
BBP_refs(bid) = 0;
BBP_lrefs(bid) = 1; /* any BAT we encounter here is
persistent, so has a logical reference */
BBP_desc(bid) = bn;
+ BBP_pid(bid) = 0;
BBP_status_set(bid, BBPEXISTING); /* do we need other
status bits? */
}
#ifdef GDKLIBRARY_HASHASH
@@ -1593,6 +1596,7 @@ BBPexit(void)
BATfree(b);
}
}
+ BBP_pid(i) = 0;
BBPuncacheit(i, true);
if (BBP_logical(i) != BBP_bak(i))
GDKfree(BBP_logical(i));
@@ -2200,6 +2204,7 @@ BBPinsert(BAT *bn)
BBP_desc(i) = NULL;
BBP_refs(i) = 1; /* new bats have 1 pin */
BBP_lrefs(i) = 0; /* ie. no logical refs */
+ BBP_pid(i) = MT_getpid();
#ifdef HAVE_HGE
if (bn->ttype == TYPE_hge)
@@ -2330,6 +2335,7 @@ bbpclear(bat i, int idx, bool lock)
BBP_logical(i) = NULL;
BBP_next(i) = BBP_free(idx);
BBP_free(idx) = i;
+ BBP_pid(i) = ~(MT_Id)0; /* not zero, not a valid thread id */
if (lock)
MT_lock_unset(&GDKcacheLock(idx));
}
@@ -2547,6 +2553,7 @@ incref(bat i, bool logical, bool lock)
/* parent BATs are not relevant for logical refs */
tp = tvp = 0;
refs = ++BBP_lrefs(i);
+ BBP_pid(i) = 0;
} else {
tp = b->theap == NULL || b->theap->parentid == i ? 0 :
b->theap->parentid;
assert(tp >= 0);
@@ -2632,7 +2639,12 @@ decref(bat i, bool logical, bool release
bat tp = 0, tvp = 0;
BAT *b;
+ if (is_bat_nil(i))
+ return -1;
assert(i > 0);
+ if (BBPcheck(i) == 0)
+ return -1;
+
if (lock)
MT_lock_set(&GDKswapLock(i));
if (releaseShare) {
@@ -2741,18 +2753,12 @@ decref(bat i, bool logical, bool release
int
BBPunfix(bat i)
{
- if (BBPcheck(i) == 0) {
- return -1;
- }
return decref(i, false, false, true, "BBPunfix");
}
int
BBPrelease(bat i)
{
- if (BBPcheck(i) == 0) {
- return -1;
- }
return decref(i, true, false, true, "BBPrelease");
}
diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -211,6 +211,7 @@ logbat_new(int tt, BUN size, role_t role
BAT *nb = COLnew(0, tt, size, role);
if (nb) {
+ BBP_pid(nb->batCacheid) = 0;
if (role == PERSISTENT)
BATmode(nb, false);
} else {
@@ -1531,6 +1532,10 @@ cleanup_and_swap(logger *lg, const log_b
logbat_destroy(nlids);
return -1;
}
+ BBPrelease(lg->catalog_bid->batCacheid);
+ BBPrelease(lg->catalog_id->batCacheid);
+ BBPrelease(lg->dcatalog->batCacheid);
+
logbat_destroy(lg->catalog_bid);
logbat_destroy(lg->catalog_id);
logbat_destroy(lg->dcatalog);
diff --git a/gdk/gdk_utils.c b/gdk/gdk_utils.c
--- a/gdk/gdk_utils.c
+++ b/gdk/gdk_utils.c
@@ -1055,6 +1055,8 @@ GDKinit(opt *set, int setlen, bool embed
TRC_CRITICAL(GDK, "BBPrename of environment BATs failed");
return GDK_FAIL;
}
+ BBP_pid(GDKkey->batCacheid) = 0;
+ BBP_pid(GDKval->batCacheid) = 0;
/* store options into environment BATs */
for (i = 0; i < nlen; i++)
diff --git a/monetdb5/modules/atoms/str.c b/monetdb5/modules/atoms/str.c
--- a/monetdb5/modules/atoms/str.c
+++ b/monetdb5/modules/atoms/str.c
@@ -2937,6 +2937,10 @@ STRprelude(void *ret)
BBPrename(UTF8_toLowerTo->batCacheid,
"monet_unicode_lower_to") != 0) {
goto bailout;
}
+ BBP_pid(UTF8_toUpperFrom->batCacheid) = 0;
+ BBP_pid(UTF8_toUpperTo->batCacheid) = 0;
+ BBP_pid(UTF8_toLowerFrom->batCacheid) = 0;
+ BBP_pid(UTF8_toLowerTo->batCacheid) = 0;
}
return MAL_SUCCEED;
diff --git a/sql/backends/monet5/generator/generator.c
b/sql/backends/monet5/generator/generator.c
--- a/sql/backends/monet5/generator/generator.c
+++ b/sql/backends/monet5/generator/generator.c
@@ -1159,6 +1159,8 @@ str VLTgenerator_rangejoin(Client cntxt,
brn->trevsorted = !incr || c <= 1;
BBPkeepref(*getArgReference_bat(stk,pci,0)= bln->batCacheid);
BBPkeepref(*getArgReference_bat(stk,pci,1)= brn->batCacheid);
+ if(blow) BBPunfix(blow->batCacheid);
+ if(bhgh) BBPunfix(bhgh->batCacheid);
return msg;
}
diff --git a/sql/backends/monet5/sql.c b/sql/backends/monet5/sql.c
--- a/sql/backends/monet5/sql.c
+++ b/sql/backends/monet5/sql.c
@@ -4943,9 +4943,9 @@ SQLunionfunc(Client cntxt, MalBlkPtr mb,
if (!ret) {
/* insert into result */
- if (!(fres =
BATdescriptor(env->stk[q->argv[0]].val.bval)))
+ if (!(fres =
BBPquickdesc(env->stk[q->argv[0]].val.bval))) {
ret =
createException(MAL, "sql.unionfunc", SQLSTATE(HY005) "Cannot access column
descriptor");
- else {
+ } else {
BAT *p =
BATconstant(fres->hseqbase, res[0]->ttype, (ptr)BUNtail(bi[0], cur),
BATcount(fres), TRANSIENT);
if (p) {
@@ -4955,7 +4955,6 @@ SQLunionfunc(Client cntxt, MalBlkPtr mb,
} else {
ret =
createException(MAL, "sql.unionfunc", GDK_EXCEPTION);
}
-
BBPunfix(fres->batCacheid);
}
i=1;
for (ii = 0; i < pci->retc &&
!ret; i++) {
diff --git a/sql/backends/monet5/sql_time.c b/sql/backends/monet5/sql_time.c
--- a/sql/backends/monet5/sql_time.c
+++ b/sql/backends/monet5/sql_time.c
@@ -259,10 +259,9 @@ nil_2time_daytime(Client cntxt, MalBlkPt
(void) cntxt;
if (isaBatType(getArgType(mb, pci, 1))) {
daytime d = daytime_nil;
- if (!(b = BATdescriptor(*getArgReference_bat(stk, pci, 1))))
+ if (!(b = BBPquickdesc(*getArgReference_bat(stk, pci, 1))))
throw(SQL, "batcalc.nil_2time_daytime", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
res = BATconstant(b->hseqbase, TYPE_daytime, &d, BATcount(b),
TRANSIENT);
- BBPunfix(b->batCacheid);
if (!res)
throw(SQL, "batcalc.nil_2time_daytime", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
r = getArgReference_bat(stk, pci, 0);
@@ -750,10 +749,9 @@ nil_2time_timestamp(Client cntxt, MalBlk
(void) cntxt;
if (isaBatType(getArgType(mb, pci, 1))) {
timestamp d = timestamp_nil;
- if (!(b = BATdescriptor(*getArgReference_bat(stk, pci, 1))))
+ if (!(b = BBPquickdesc(*getArgReference_bat(stk, pci, 1))))
throw(SQL, "batcalc.nil_2time_timestamp",
SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
res = BATconstant(b->hseqbase, TYPE_timestamp, &d, BATcount(b),
TRANSIENT);
- BBPunfix(b->batCacheid);
if (!res)
throw(SQL, "batcalc.nil_2time_timestamp",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
r = getArgReference_bat(stk, pci, 0);
@@ -1540,10 +1538,9 @@ nil_2_date(Client cntxt, MalBlkPtr mb, M
(void) cntxt;
if (isaBatType(getArgType(mb, pci, 1))) {
date d = date_nil;
- if (!(b = BATdescriptor(*getArgReference_bat(stk, pci, 1))))
+ if (!(b = BBPquickdesc(*getArgReference_bat(stk, pci, 1))))
throw(SQL, "batcalc.nil_2_date", SQLSTATE(HY002)
RUNTIME_OBJECT_MISSING);
res = BATconstant(b->hseqbase, TYPE_date, &d, BATcount(b),
TRANSIENT);
- BBPunfix(b->batCacheid);
if (!res)
throw(SQL, "batcalc.nil_2_date", SQLSTATE(HY013)
MAL_MALLOC_FAIL);
r = getArgReference_bat(stk, pci, 0);
diff --git a/sql/storage/bat/bat_utils.c b/sql/storage/bat/bat_utils.c
--- a/sql/storage/bat/bat_utils.c
+++ b/sql/storage/bat/bat_utils.c
@@ -19,7 +19,10 @@ bat_destroy(BAT *b)
BAT *
bat_new(int tt, BUN size, role_t role)
{
- return COLnew(0, tt, size, role);
+ BAT *bn = COLnew(0, tt, size, role);
+ if (bn)
+ BBP_pid(bn->batCacheid) = 0;
+ return bn;
}
void
diff --git a/testing/Mtest.py.in b/testing/Mtest.py.in
--- a/testing/Mtest.py.in
+++ b/testing/Mtest.py.in
@@ -1633,6 +1633,8 @@ def GetBitsAndModsAndThreads(env) :
setpgrp = True
proc = process.Popen(cmd, stdin=process.PIPE, stdout=process.PIPE,
stderr=process.PIPE, text=True)
+ proc.stdout = process._BufferedPipe(proc.stdout)
+ proc.stderr = process._BufferedPipe(proc.stderr)
proc.killed = False
proc.onechild = True
t = Timer(float(par['TIMEOUT']), killProc, args = [proc, proc.stderr, cmd])
_______________________________________________
checkin-list mailing list
checkin-list@monetdb.org
https://www.monetdb.org/mailman/listinfo/checkin-list
