Changeset: 2fb5efb12e57 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=2fb5efb12e57
Modified Files:
monetdb5/modules/mal/wlc.c
monetdb5/modules/mal/wlc.mal
sql/backends/monet5/wlr.c
sql/test/wlcr/Tests/All
sql/test/wlcr/Tests/wlc80.py
sql/test/wlcr/Tests/wlcr00.py
sql/test/wlcr/Tests/wlr01.py
sql/test/wlcr/Tests/wlr10.py
sql/test/wlcr/Tests/wlr100.py
sql/test/wlcr/Tests/wlr100.stable.err
sql/test/wlcr/Tests/wlr100.stable.out
sql/test/wlcr/Tests/wlr20.py
sql/test/wlcr/Tests/wlr20.stable.err
sql/test/wlcr/Tests/wlr20.stable.out
sql/test/wlcr/Tests/wlr30.py
sql/test/wlcr/Tests/wlr30.stable.err
sql/test/wlcr/Tests/wlr30.stable.out
sql/test/wlcr/Tests/wlr40.py
sql/test/wlcr/Tests/wlr40.stable.err
sql/test/wlcr/Tests/wlr40.stable.out
sql/test/wlcr/Tests/wlr50.py
sql/test/wlcr/Tests/wlr50.stable.err
sql/test/wlcr/Tests/wlr50.stable.out
sql/test/wlcr/Tests/wlr70.py
sql/test/wlcr/Tests/wlr70.stable.err
sql/test/wlcr/Tests/wlr70.stable.out
sql/test/wlcr/Tests/wlr80.py
Branch: wlcrNov2019
Log Message:
A round of improvements. Still seeing some timeing issues, where the replicator
is started too late, despite 5 sec delay.
diffs (truncated from 1573 to 300 lines):
diff --git a/monetdb5/modules/mal/wlc.c b/monetdb5/modules/mal/wlc.c
--- a/monetdb5/modules/mal/wlc.c
+++ b/monetdb5/modules/mal/wlc.c
@@ -588,10 +588,10 @@ WLCpreparewrite(Client cntxt)
if( wlc_beat == 0 )
msg = WLCcloselogger();
+ MT_lock_unset(&wlc_lock);
trimMalVariables(cntxt->wlc, NULL);
resetMalBlk(cntxt->wlc, 0);
cntxt->wlc_kind = WLC_QUERY;
- MT_lock_unset(&wlc_lock);
} else
throw(MAL,"wlc.write","WLC log path missing ");
diff --git a/monetdb5/modules/mal/wlc.mal b/monetdb5/modules/mal/wlc.mal
--- a/monetdb5/modules/mal/wlc.mal
+++ b/monetdb5/modules/mal/wlc.mal
@@ -18,7 +18,7 @@ pattern master(path:str)
address WLCmaster
comment "Activate the workload-capture-replay process. Use a different
location for the logs.";
-pattern stopmaster()
+pattern stop()
address WLCstop
comment "Stop capturing the logs";
diff --git a/sql/backends/monet5/wlr.c b/sql/backends/monet5/wlr.c
--- a/sql/backends/monet5/wlr.c
+++ b/sql/backends/monet5/wlr.c
@@ -45,8 +45,6 @@
//#define _WLR_DEBUG_
-MT_Lock wlr_lock = MT_LOCK_INITIALIZER("wlr_lock");
-
/* The current status of the replica processing.
* It is based on the assumption that at most one replica thread is running
* importing data from a single master.
@@ -193,7 +191,7 @@ WLRgetMaster(void)
if (len == -1 || len >= FILENAME_MAX)
throw(MAL, "wlr.getMaster", "wlc.config filename path is too
large");
if((dir = GDKfilepath(0,path,"wlc.config",0)) == NULL)
- throw(MAL,"wlr.getMaster","Could not access wlc.config file\n");
+ throw(MAL,"wlr.getMaster","Could not access wlc.config file
%s/wlc.config\n", path);
fd = fopen(dir,"r");
GDKfree(dir);
@@ -203,7 +201,7 @@ WLRgetMaster(void)
throw(MAL,"wlr.getMaster","Master not identified\n");
wlc_state = WLC_CLONE; // not used as master
} else
- throw(MAL,"wlr.getMaster","Could not access wlc.config file\n");
+ throw(MAL,"wlr.getMaster","Could not get read access to
'%s'config file\n", wlr_master);
return MAL_SUCCEED;
}
@@ -535,8 +533,8 @@ WLRreplicate(Client cntxt, MalBlkPtr mb,
time_t clk;
struct tm ctm;
char clktxt[26];
- int duration = 2000;
- lng limit;
+ int duration = 5000;
+ lng limit = INT64_MAX;
str msg = MAL_SUCCEED;
(void) cntxt;
@@ -544,7 +542,7 @@ WLRreplicate(Client cntxt, MalBlkPtr mb,
throw(MAL, "sql.replicate", "No replication configuration");
if( !wlr_thread)
- throw(MAL, "sql.replicate", "Replicator not started, call
setmaster() first ");
+ throw(MAL, "sql.replicate", "Replicator not started, call
wlr.master() first ");
if( pci->argc == 0)
wlr_limit = INT64_MAX;
@@ -568,7 +566,7 @@ WLRreplicate(Client cntxt, MalBlkPtr mb,
if ( limit < 0 && timelimit[0] == 0)
throw(MAL, "sql.replicate", "Stop tag limit should be positive
or timestamp should be set");
- if ( limit < INT64_MAX && limit >= wlc_tag)
+ if (limit < INT64_MAX && limit >= wlc_tag)
throw(MAL, "sql.replicate", "Stop tag limit "LLFMT" be less
than wlc_tag "LLFMT, limit, wlc_tag);
if ( limit >= 0)
wlr_limit = limit;
@@ -577,13 +575,6 @@ WLRreplicate(Client cntxt, MalBlkPtr mb,
throw(MAL, "sql.replicate", "No replication master set");
WLRputConfig();
- if ( wlr_limit == INT64_MAX){
-#ifdef _WLR_DEBUG_
- fprintf(stderr, "#replicate running in the background \n");
-#endif
- return MAL_SUCCEED;
- }
-
// the client thread should wait for the replicator to its job
gettimeofday(&clock, NULL);
clk = clock.tv_sec;
@@ -598,6 +589,8 @@ WLRreplicate(Client cntxt, MalBlkPtr mb,
#endif
while ( (wlr_tag < wlr_limit ) || (wlr_timelimit[0] &&
strncmp(clktxt, wlr_timelimit, sizeof(wlr_timelimit)) > 0) ) {
+ if( wlr_limit == INT64_MAX && wlr_tag >= wlc_tag -1 )
+ break;
if ( wlr_error[0])
throw(MAL, "sql.replicate", "tag "LLFMT": %s", wlr_tag,
wlr_error);
if ( wlr_tag == wlc_tag)
@@ -615,8 +608,12 @@ WLRreplicate(Client cntxt, MalBlkPtr mb,
}
duration -= 200;
- if ( duration < 0)
- throw(SQL,"wlr.startreplicate",SQLSTATE(42000) "Timeout
to wait for replicator to catch up ");
+ if ( duration < 0){
+ if( wlr_limit == INT64_MAX && wlr_timelimit[0] == 0)
+ break;
+ throw(SQL,"wlr.startreplicate",SQLSTATE(42000) "Timeout
to wait for replicator to catch up."
+ "Catched up until "LLFMT", " LLFMT " pending", wlr_tag,
wlr_limit - wlr_tag);
+ }
// don't make the sleep too short.
MT_sleep_ms( 200);
}
@@ -658,6 +655,7 @@ str
WLRstart(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
int len;
+ str msg;
(void) cntxt;
(void) mb;
@@ -667,15 +665,13 @@ WLRstart(Client cntxt, MalBlkPtr mb, Mal
len = snprintf(wlr_master, IDLENGTH, "%s", *getArgReference_str(stk,
pci, 1));
if (len == -1 || len >= IDLENGTH)
throw(MAL, "wlr.replicate", SQLSTATE(42000) "Input value is too
large for wlr_master buffer");
- if( WLRgetMaster())
- throw(MAL, "wlr.replicate", SQLSTATE(42000) "Can not access the
master configuration record ");
+ if( (msg =WLRgetMaster()) != MAL_SUCCEED)
+ return msg;
- MT_lock_set(&wlr_lock); // avoid creation of multiple replicators in
parallel.
// time the consolidation process in the background
if (MT_create_thread(&wlr_thread, WLRprocessScheduler, (void*) NULL,
MT_THR_DETACHED, "WLRprocessSched") < 0) {
- MT_lock_unset(&wlr_lock);
throw(SQL,"wlr.init",SQLSTATE(42000) "Starting wlr
manager failed");
}
#ifdef _WLR_DEBUG_
@@ -683,13 +679,12 @@ WLRstart(Client cntxt, MalBlkPtr mb, Mal
#else
(void) cntxt;
#endif
- MT_lock_unset(&wlr_lock);
// Wait until the replicator is properly initialized
while( wlr_state != WLR_RUN && wlr_error[0] == 0){
#ifdef _WLR_DEBUG_
fprintf(stderr,"#WLR replicator initializing\n");
#endif
- MT_sleep_ms( 200);
+ MT_sleep_ms( 50);
}
return MAL_SUCCEED;
}
diff --git a/sql/test/wlcr/Tests/All b/sql/test/wlcr/Tests/All
--- a/sql/test/wlcr/Tests/All
+++ b/sql/test/wlcr/Tests/All
@@ -35,9 +35,6 @@ wlr50
wlc70
wlr70
-# stop the master
-wlc80
-wlr80
#
#stop the master
wlc100
diff --git a/sql/test/wlcr/Tests/wlc80.py b/sql/test/wlcr/Tests/wlc80.py
--- a/sql/test/wlcr/Tests/wlc80.py
+++ b/sql/test/wlcr/Tests/wlc80.py
@@ -15,7 +15,6 @@ if not tstdb or not dbfarm:
#clean up first
dbname = tstdb
-dbnameclone = tstdb + '-clone'
s = process.server(dbname = dbname, stdin = process.PIPE, stdout =
process.PIPE, stderr = process.PIPE)
@@ -24,8 +23,8 @@ c = process.client('sql', server = s, st
#continue logging
cout, cerr = c.communicate('''\
call wlc.stop();
-create table tmp70(i int, s string);
-insert into tmp70 values(1,'thanks'), (2,'for the fish');
+create table tmp80(i int, s string);
+insert into tmp80 values(1,'thanks'), (2,'for the fish');
select * from tmp70;
''')
diff --git a/sql/test/wlcr/Tests/wlcr00.py b/sql/test/wlcr/Tests/wlcr00.py
--- a/sql/test/wlcr/Tests/wlcr00.py
+++ b/sql/test/wlcr/Tests/wlcr00.py
@@ -14,7 +14,7 @@ if not tstdb or not dbfarm:
sys.exit(1)
dbname = tstdb
-dbnameclone = tstdb + '-clone'
+dbnameclone = tstdb + 'clone'
# clean up before we start
if os.path.exists(os.path.join(dbfarm, dbname)):
diff --git a/sql/test/wlcr/Tests/wlr01.py b/sql/test/wlcr/Tests/wlr01.py
--- a/sql/test/wlcr/Tests/wlr01.py
+++ b/sql/test/wlcr/Tests/wlr01.py
@@ -23,7 +23,7 @@ def freeport():
cloneport = freeport()
dbname = tstdb
-dbnameclone = tstdb + '-clone'
+dbnameclone = tstdb + 'clone'
#master = process.server(dbname = dbname, stdin = process.PIPE, stdout =
process.PIPE, stderr = process.PIPE)
slave = process.server(dbname = dbnameclone, mapiport = cloneport, stdin =
process.PIPE, stdout = process.PIPE, stderr = process.PIPE)
@@ -58,5 +58,5 @@ def listfiles(path):
except IOError:
sys.stderr.write('Failure to read file ' + file + '\n')
-listfiles(os.path.join(dbfarm, tstdb))
-listfiles(os.path.join(dbfarm, tstdb, 'wlc_logs'))
+# listfiles(os.path.join(dbfarm, tstdb))
+# listfiles(os.path.join(dbfarm, tstdb, 'wlc_logs'))
diff --git a/sql/test/wlcr/Tests/wlr10.py b/sql/test/wlcr/Tests/wlr10.py
--- a/sql/test/wlcr/Tests/wlr10.py
+++ b/sql/test/wlcr/Tests/wlr10.py
@@ -23,7 +23,7 @@ def freeport():
cloneport = freeport()
dbname = tstdb
-dbnameclone = tstdb + '-clone'
+dbnameclone = tstdb + 'clone'
#master = process.server(dbname = dbname, stdin = process.PIPE, stdout =
process.PIPE, stderr = process.PIPE)
slave = process.server(dbname = dbnameclone, mapiport = cloneport, stdin =
process.PIPE, stdout = process.PIPE, stderr = process.PIPE)
@@ -58,5 +58,5 @@ def listfiles(path):
except IOError:
sys.stderr.write('Failure to read file ' + file + '\n')
-listfiles(os.path.join(dbfarm, tstdb))
-listfiles(os.path.join(dbfarm, tstdb, 'wlc_logs'))
+# listfiles(os.path.join(dbfarm, tstdb))
+# listfiles(os.path.join(dbfarm, tstdb, 'wlc_logs'))
diff --git a/sql/test/wlcr/Tests/wlr100.py b/sql/test/wlcr/Tests/wlr100.py
--- a/sql/test/wlcr/Tests/wlr100.py
+++ b/sql/test/wlcr/Tests/wlr100.py
@@ -23,7 +23,7 @@ def freeport():
cloneport = freeport()
dbname = tstdb
-dbnameclone = tstdb + '-clone'
+dbnameclone = tstdb + 'clone'
#master = process.server(dbname = dbname, stdin = process.PIPE, stdout =
process.PIPE, stderr = process.PIPE)
slave = process.server(dbname = dbnameclone, mapiport = cloneport, stdin =
process.PIPE, stdout = process.PIPE, stderr = process.PIPE)
@@ -31,10 +31,11 @@ slave = process.server(dbname = dbnamecl
c = process.client('sql', server = slave, stdin = process.PIPE, stdout =
process.PIPE, stderr = process.PIPE)
cout, cerr = c.communicate('''\
+call wlr.master('%s');
call wlr.replicate();
select * from tmp;
call wlr.stop();
-''' )
+''' % dbname )
sout, serr = slave.communicate()
#mout, merr = master.communicate()
diff --git a/sql/test/wlcr/Tests/wlr100.stable.err
b/sql/test/wlcr/Tests/wlr100.stable.err
--- a/sql/test/wlcr/Tests/wlr100.stable.err
+++ b/sql/test/wlcr/Tests/wlr100.stable.err
@@ -17,11 +17,14 @@ stderr of test 'wlr100` in directory 'sq
# builtin opt sql_debug = 0
# cmdline opt gdk_nr_threads = 0
# cmdline opt mapi_open = true
-# cmdline opt monet_prompt =
-# cmdline opt mapi_port = 42547
-# cmdline opt mapi_usock = /var/tmp/mtest-30891/.s.monetdb.42547
-# cmdline opt gdk_dbpath =
/export/scratch1/mk/wlcr//Linux/var/MonetDB/mTests_sql_test_wlcr-clone
-# cmdline opt gdk_debug = 536870922
+# cmdline opt mapi_port = 34573
+# cmdline opt mapi_usock = /var/tmp/mtest-22411/.s.monetdb.34573
+# cmdline opt gdk_dbpath =
/export/scratch1/mk/branches/wlcr//Linux/var/MonetDB/mTests_sql_test_wlcrclone
+#client1:!ERROR:SQLException:wlr.startreplicate:42000!Timeout to wait for
replicator to catch up.Catched up until 8, 9223372036854775799 pending
+MAPI = (monetdb) /var/tmp/mtest-22885/.s.monetdb.44647
+QUERY = call wlr.replicate();
+ERROR = !Timeout to wait for replicator to catch up.Catched up until 8,
9223372036854775799 pending
+CODE = 42000
# 21:26:05 >
# 21:26:05 > "Done."
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
