Update of /cvsroot/monetdb/pathfinder/runtime
In directory 23jxhf1.ch3.sourceforge.com:/tmp/cvs-serv5166
Modified Files:
Tag: Aug2009_NFI
pathfinder.mx
Log Message:
some finishing touches on the subexpression caching mechanism
implemented the more useful semantics of pf:session-use that waits
for a concurrent pf:session already in progress (computing the same expression)
rather than blindly recomputing it.
- ws of pf:session-use must have access to the parent cache
- semaphores needed to block the pf:session-use queries and wake them up
fixed issues in concurrency:
- temp doc container must be copied in pf:session-use sessions
- structer check in updates (pf:session-use is forbidden)
adapted caching garbage collection policy against ever-growing temp
document containers.
- hard limit of 1M nodes in it (copying the tdc then costs ~100ms)
- terminate session if full (we cannot garbage collect it)
U pathfinder.mx
Index: pathfinder.mx
===================================================================
RCS file: /cvsroot/monetdb/pathfinder/runtime/pathfinder.mx,v
retrieving revision 1.462.6.8
retrieving revision 1.462.6.9
diff -u -d -r1.462.6.8 -r1.462.6.9
--- pathfinder.mx 8 Oct 2009 16:44:00 -0000 1.462.6.8
+++ pathfinder.mx 10 Oct 2009 03:08:58 -0000 1.462.6.9
@@ -280,6 +280,9 @@
# - CACHE_VAL[void,bat] bat-of-bats (iter,pos,..)
# - CACHE_LRU[void,lng] last time used (usec)
# - CACHE_SIZE[void,lng] size in bytes of VAL
+# - _CACHE_ID[void,str] name in writable ws
+# - _CACHE_VAL[void,bat] val in writable ws
+# - _CACHE_LRU[void,lng] last time used (usec) in writable ws
@- ws definition
@@ -355,29 +358,32 @@
@:w...@1(CACHE_VAL, 46, void, bat, void, void, oid_nil)@
@:w...@1(CACHE_LRU, 47, void, lng, void, void, oid_nil)@
@:w...@1(CACHE_SIZE, 48, void, lng, void, void, oid_nil)@
+@:w...@1(_CACHE_ID, 49, void, str, void, void, oid_nil)@
+@:w...@1(_CACHE_VAL, 50, void, bat, void, void, oid_nil)@
+@:w...@1(_CACHE_LRU, 51, void, lng, void, void, oid_nil)@
-@:w...@1(_MAP_PID, 49, void, bat, void, oid, PRE_BASE)@
-@:w...@1(_RID_SIZE, 50, void, bat, void, int, PRE_BASE)@
-@:w...@1(_RID_LEVEL, 51, void, bat, void, chr, PRE_BASE)@
-@:w...@1(_RID_PROP, 52, void, bat, void, oid, PRE_BASE)@
-@:w...@1(_RID_KIND, 53, void, bat, void, chr, PRE_BASE)@
-@:w...@1(_RID_NID, 54, void, bat, void, oid, PRE_BASE)@
-@:w...@1(_NID_RID, 55, void, bat, void, oid, PRE_BASE)@
-@:w...@1(_FRAG_ROOT, 56, void, bat, oid, oid, oid_nil)@
-@:w...@1(_QN_HISTOGRAM, 57, void, bat, void, str, PRE_BASE)@
-@:w...@1(_QN_PREFIX_URI_LOC,58, void, bat, void, str, PRE_BASE)@
-@:w...@1(_QN_URI_LOC, 59, void, bat, void, str, PRE_BASE)@
-@:w...@1(_QN_PREFIX, 60, void, bat, void, str, PRE_BASE)@
-@:w...@1(_QN_URI, 61, void, bat, void, str, PRE_BASE)@
-@:w...@1(_QN_LOC, 62, void, bat, void, str, PRE_BASE)@
-@:w...@1(_PROP_TEXT, 63, void, bat, void, str, PRE_BASE)@
-@:w...@1(_PROP_COM, 64, void, bat, void, str, PRE_BASE)@
-@:w...@1(_PROP_INS, 65, void, bat, void, str, PRE_BASE)@
-@:w...@1(_PROP_TGT, 66, void, bat, void, str, PRE_BASE)@
-@:w...@1(_PROP_VAL, 67, void, bat, void, str, PRE_BASE)@
-@:w...@1(_ATTR_OWN, 68, void, bat, void, oid, PRE_BASE)@
-@:w...@1(_ATTR_QN, 69, void, bat, void, oid, PRE_BASE)@
-@:w...@1(_ATTR_PROP, 70, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_MAP_PID, 52, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_RID_SIZE, 53, void, bat, void, int, PRE_BASE)@
+@:w...@1(_RID_LEVEL, 54, void, bat, void, chr, PRE_BASE)@
+@:w...@1(_RID_PROP, 55, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_RID_KIND, 56, void, bat, void, chr, PRE_BASE)@
+@:w...@1(_RID_NID, 57, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_NID_RID, 58, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_FRAG_ROOT, 59, void, bat, oid, oid, oid_nil)@
+@:w...@1(_QN_HISTOGRAM, 60, void, bat, void, str, PRE_BASE)@
+@:w...@1(_QN_PREFIX_URI_LOC,61, void, bat, void, str, PRE_BASE)@
+@:w...@1(_QN_URI_LOC, 62, void, bat, void, str, PRE_BASE)@
+@:w...@1(_QN_PREFIX, 63, void, bat, void, str, PRE_BASE)@
+@:w...@1(_QN_URI, 64, void, bat, void, str, PRE_BASE)@
+@:w...@1(_QN_LOC, 65, void, bat, void, str, PRE_BASE)@
+@:w...@1(_PROP_TEXT, 66, void, bat, void, str, PRE_BASE)@
+@:w...@1(_PROP_COM, 67, void, bat, void, str, PRE_BASE)@
+@:w...@1(_PROP_INS, 68, void, bat, void, str, PRE_BASE)@
+@:w...@1(_PROP_TGT, 69, void, bat, void, str, PRE_BASE)@
+@:w...@1(_PROP_VAL, 70, void, bat, void, str, PRE_BASE)@
+@:w...@1(_ATTR_OWN, 71, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_ATTR_QN, 72, void, bat, void, oid, PRE_BASE)@
+@:w...@1(_ATTR_PROP, 73, void, bat, void, oid, PRE_BASE)@
@-
The bottom segment are the master copies of the top segment.
The middle segment is private to the query.
@@ -387,7 +393,7 @@
BEWARE: the logger version number below needs to be incremented whenever the
working set schema is changed.
@= ws_decl
-@:w...@1_decl(WS_SIZE, 71)@
+@:w...@1_decl(WS_SIZE, 74)@
@:w...@1_decl(QNAME, 2)@
@:w...@1_decl(BOOL, 3)@
@:w...@1_decl(INT, 4)@
@@ -529,6 +535,7 @@
var xrpc_lock := pflock_get(6); # master XRPC lock, to protect above
xrpc_* BATs
var xrpc_hostport := my_hostname() + ":" + str(get_xrpc_port());
var xrpc_querynr := (lng(get_xrpc_port()) * 100000LL) - 1LL;
+var xrpc_cache := bat(sema,str); # [sema,qid:id] for letting
pf:session-use wait for pf:session to produce results
# dummy variables to make MIL test scripts work
# (otherwise always overridden by a declaration in the pf-generated MIL plan)
@@ -557,6 +564,7 @@
xrpc_statuses.inplace(idx, "abort");
}
lock_unset(xrpc_locks.find(idx));
+ _ws_cache_wakeup(xrpc_qid + ":", xrpc_qid + ";");
}
# background check to time-out waiting 2PC transactions, and prune them fully
after an hour
@@ -882,7 +890,7 @@
var ws_rid := ws_mem.slice(PRE_SIZE, PRE_NID).rename("ws_rid");
# of those, all master rid_* bats, and their (remapped) pre_*
views
ws_mem := ws_mem.slice(NID_RID, count(ws_mem)).rename("ws_mem");
-
+var ws_new_map :=
[oid](mirror(ws_nme)).inplace(ws_nme.slice(CACHE_ID,_CACHE_LRU).project(oid(_CACHE_ID)),
true);
# scans for empty collections, and produces a list of kill-bats (fast - done
inside the short lock)
PROC _collection_cleanup() : BAT[str,str]
@@ -1454,26 +1462,80 @@
var b := bat(htp,ttp,10000);
if (not(isnil(seq))) b.seqbase(seq);
return bat(void,bat).seqbase(PRE_BASE).append(b);
- }
+ }
return bat(void,tpe).seqbase(PRE_BASE); # a constant bat
}
+PROC _ws_cache_expr(bat ws, str id, bit use) : bit {
+ var res := false;
+ if (use) # check master first
+ res := ws.fetch(_CACHE_ID).texist(id);
+ if (not(res))
+ res := ws.fetch(CACHE_ID).texist(id);
+ if (not(res)) {
+ id := xrpc_qid + ":" + id;
+ if (not(use)) {
+ xrpc_cache.insert(sem, id);
+ } else if (xrpc_cache.texist(id)) {
+ sem := sema_create(0);
+ xrpc_cache.insert(sem, id);
+ }
+ }
+ return res;
+}
+
PROC ws_cache_expr(bat ws, str id) : bit {
- return not(ws.fetch(CACHE_ID).texist(id));
+ var sem := sema_nil, use := (xrpc_mode.search("use") >= 0);
+ lock_set(xrpc_lock);
+ var res, err := CATCH(res := _ws_cache_expr(ws, id, use));
+ lock_unset(xrpc_lock);
+ if (not(isnil(err))) ERROR(err);
+ if (res) return false;
+ if (not(isnil(sem))) {
+ sema_down(sem);
+ sema_destroy(sem);
+ }
+ return true;
+}
+
+PROC _ws_cache_get(bat ws, str id) : bit {
+ var delta := 0; # get from master, else from private cache
+ if (texist(ws.fetch(_CACHE_ID), id)) delta := 4;
+ var idx := reverse(ws.fetch(delta+CACHE_ID)).find(id);
+ inplace(ws.fetch(delta+CACHE_LRU), idx, usec(), true);
+ return ws.fetch(delta+CACHE_VAL).fetch(idx);
}
PROC ws_cache_get(bat ws, str id) : bit {
- var idx := reverse(ws.fetch(CACHE_ID)).find(id);
- inplace(ws.fetch(CACHE_LRU), idx, usec(), true);
- return ws.fetch(CACHE_VAL).fetch(idx);
+ lock_set(xrpc_lock);
+ var res, err := CATCH(res := _ws_cache_get(ws, id));
+ lock_unset(xrpc_lock);
+ if (not(isnil(err))) ERROR(err);
+ return res;
}
-PROC ws_cache_put(bat ws, str id, bat val) : void {
+PROC _ws_cache_wakeup(str lo, str hi) : void {
+ var wakeup := xrpc_cache.ord_select(lo, hi);
+ [sema_up](wakeup.hmark(0...@0).ord_select(sema_nil,sema_nil));
+ xrpc_cache.delete(wakeup);
+}
+
+PROC _ws_cache_put(bat ws, str id, bat val) : void {
ws.fetch(CACHE_ID).append(id);
ws.fetch(CACHE_VAL).append(val);
ws.fetch(CACHE_SIZE).append(sum([batsize](val)));
ws.fetch(CACHE_LRU).append(0LL);
-#print(ws.fetch(CACHE_ID), ws.fetch(CACHE_VAL), ws.fetch(CACHE_SIZE),
ws.fetch(CACHE_LRU));
+ var cache_id := xrpc_qid + ":" + id;
+ _ws_cache_wakeup(cache_id, cache_id);
+}
+
+PROC ws_cache_put(bat ws, str id, bat val) : void {
+ lock_set(xrpc_lock);
+ var err := CATCH(_ws_cache_put(ws, id, val));
+ lock_unset(xrpc_lock);
+ if (not(isnil(err))) ERROR(err);
+print(ws.fetch(CACHE_ID), ws.fetch(CACHE_VAL), ws.fetch(CACHE_SIZE),
ws.fetch(CACHE_LRU));
+print(ws.fetch(_CACHE_ID), ws.fetch(_CACHE_VAL), ws.fetch(_CACHE_LRU));
}
PROC ws_cache_end(bat ws) : void {
@@ -1493,7 +1555,8 @@
ws.inplace(oid(CACHE_LRU),
pivot.leftfetchjoin(ws.fetch(CACHE_LRU)).access(BAT_APPEND), true);
ws.inplace(oid(CACHE_SIZE),
pivot.leftfetchjoin(ws.fetch(CACHE_SIZE)).access(BAT_APPEND), true);
}
-#print(ws.fetch(CACHE_ID), ws.fetch(CACHE_VAL), ws.fetch(CACHE_SIZE),
ws.fetch(CACHE_LRU));
+print(ws.fetch(CACHE_ID), ws.fetch(CACHE_VAL), ws.fetch(CACHE_SIZE),
ws.fetch(CACHE_LRU));
+print(ws.fetch(_CACHE_ID), ws.fetch(_CACHE_VAL), ws.fetch(_CACHE_LRU));
}
@@ -1532,7 +1595,27 @@
xrpc_statuses.inplace(idx, "reserved");
if (xrpc_mode.search("use") < 0) return ws;
}
- ws := [access]([copy](ws),BAT_WRITE); # use a copy, no locking required
+ # use a copy of the ws so it is not affected by concurrent queries;
give it a private cache
+ var cache_id := ws.fetch(CACHE_ID);
+ var cache_val := ws.fetch(CACHE_VAL);
+ var cache_lru := ws.fetch(CACHE_LRU);
+ ws := [access]([copy](ws_new_map.leftfetchjoin(ws)), BAT_WRITE);
+ ws.inplace(oid(CACHE_ID), bat(void,str).seqbase(0...@0), true);
+ ws.inplace(oid(CACHE_VAL), bat(void,bat).seqbase(0...@0), true);
+ ws.inplace(oid(CACHE_LRU), bat(void,lng).seqbase(0...@0), true);
+ ws.inplace(oid(CACHE_SIZE), bat(void,lng).seqbase(0...@0), true);
+ ws.inplace(oid(_CACHE_ID), cache_id, true);
+ ws.inplace(oid(_CACHE_VAL), cache_val, true);
+ ws.inplace(oid(_CACHE_LRU), cache_lru, true);
+
+ # the tempdoc container must be deep copied, as read-only queries
change it
+ #var temp_doc_cont := [access]([copy]([fetch](ws.slice(0,ATTR_PROP),
WS)), BAT_WRITE);
+ #temp_doc_cont.hmark(0...@0).leftfetchjoin(ws).[inplace](WS,
temp_doc_cont.tmark(0...@0));
+ #temp_doc_cont.hmark(0...@0).leftfetchjoin(ws).[inplace](WS,
temp_doc_cont.tmark(0...@0));
+ #temp_doc_cont := [access]([copy]([fetch](ws.slice(QN_NID,QN_NID_UNQ),
WS)), BAT_WRITE);
+ #temp_doc_cont.hmark(0...@0).leftfetchjoin(ws).[inplace](WS,
temp_doc_cont.tmark(0...@0));
+ #temp_doc_cont :=
[access]([copy]([fetch](ws.slice(PRE_CONT,ATTR_CONT), WS)), BAT_WRITE);
+ #temp_doc_cont.hmark(0...@0).leftfetchjoin(ws).[inplace](WS,
temp_doc_cont.tmark(0...@0));
} else {
ws := [ws_new](mirror(ws_tpe), ws_tpe, ws_htp, ws_ttp, ws_seq);
if (xrpc < 0) {
@@ -1548,6 +1631,8 @@
var id := and(lng(newoid(1)), 2147483647LL);
var wsid := <<(id, 32) + (lng(update) << 30) + lng(ws);
ws.access(BAT_READ).rename(str(wsid));
+print(ws.fetch(CACHE_ID), ws.fetch(CACHE_VAL), ws.fetch(CACHE_SIZE),
ws.fetch(CACHE_LRU));
+print(ws.fetch(_CACHE_ID), ws.fetch(_CACHE_VAL), ws.fetch(_CACHE_LRU));
return ws;
}
@@ -1558,10 +1643,15 @@
if ((xrpc_qid = "") or (xrpc_mode = "use-cache-repeatable")) {
ws_destroy(ws);
} else {
- ws_cache_end(ws);
+ if (count(ws.fetch(PRE_SIZE).fetch(WS)) > 1000000) { # limit max
tempdoc copy time to ~100ms
+ xrpc_timeout := 0LL; # if tempdoc container gets too big, we have
to kill the ws
+ } else {
+ ws_cache_end(ws); # apply LRU to the cached subsequences
+ }
lock_set(xrpc_lock);
- CATCH(_ws_xrpc_end(xrpc_qid, err));
+ err := CATCH(_ws_xrpc_end(xrpc_qid, err));
lock_unset(xrpc_lock);
+ if (not(isnil(err))) ERROR(err);
}
# do not auto-restart XRPC updates (deadlock danger)
if (xrpc_method != "") return 2;
@@ -1571,9 +1661,8 @@
PROC ws_create(int update) : BAT[void,bat]
{
if (bit(and(update,2)) and (xrpc_mode.search("cache") >= 0)) {
- # updates in a session terminate the session
- xrpc_timeout := 0;
- xrpc_mode := "cache-repeatable";
+ if (xrpc_mode.search("use") >= 0) ERROR("ws_create(): cannot update in
a pf:session-use session");
+ xrpc_timeout := 0; # updates in a session terminate the session
}
# NOTE: use pre-query MIL variables xrpc_qid/xrpc_timeout to possibly
re-use an existing ws
------------------------------------------------------------------------------
Come build with us! The BlackBerry(R) Developer Conference in SF, CA
is the only developer event you need to attend this year. Jumpstart your
developing skills, take BlackBerry mobile applications to market and stay
ahead of the curve. Join us from November 9 - 12, 2009. Register now!
http://p.sf.net/sfu/devconference
_______________________________________________
Monetdb-pf-checkins mailing list
[email protected]
https://lists.sourceforge.net/lists/listinfo/monetdb-pf-checkins
