Changeset: 2b68815ae1ba for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/2b68815ae1ba
Modified Files:
clients/Tests/exports.stable.out
gdk/gdk.h
gdk/gdk_utils.c
monetdb5/mal/mal_pipelines.c
monetdb5/modules/mal/pipeline.c
monetdb5/modules/mal/pp_hash.c
monetdb5/modules/mal/pp_slicer.c
sql/backends/monet5/rel_orderby.c
Branch: default
Log Message:
Implemented gdk_nr_pipeline_threads option (GDKnr_pipeline_threads variable).
diffs (141 lines):
diff --git a/clients/Tests/exports.stable.out b/clients/Tests/exports.stable.out
--- a/clients/Tests/exports.stable.out
+++ b/clients/Tests/exports.stable.out
@@ -314,6 +314,7 @@ gdk_return GDKmergeidx(BAT *b, BAT **a,
void *GDKmmap(const char *path, int mode, size_t len);
int GDKms(void);
gdk_return GDKmunmap(void *addr, size_t len);
+int GDKnr_pipeline_threads;
int GDKnr_threads;
void GDKprepareExit(void);
void GDKprintinfo(void);
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -552,6 +552,7 @@ gdk_export void GDKsetdebug(unsigned deb
gdk_export unsigned GDKgetdebug(void);
gdk_export int GDKnr_threads;
+gdk_export int GDKnr_pipeline_threads;
/* API */
diff --git a/gdk/gdk_utils.c b/gdk/gdk_utils.c
--- a/gdk/gdk_utils.c
+++ b/gdk/gdk_utils.c
@@ -453,6 +453,7 @@ size_t _MT_npages = 0; /* variable hold
static lng programepoch;
int GDKnr_threads = 0;
+int GDKnr_pipeline_threads = 0;
void
MT_init(void)
@@ -1190,6 +1191,9 @@ GDKinit(opt *set, int setlen, bool embed
}
if (GDKnr_threads > THREADS)
GDKnr_threads = THREADS;
+ GDKnr_pipeline_threads = GDKgetenv_int("gdk_nr_pipeline_threads", 0);
+ if (GDKnr_pipeline_threads == 0)
+ GDKnr_pipeline_threads = GDKnr_threads;
#if defined(HAVE_SYSCONF) && defined(_SC_LEVEL3_CACHE_SIZE)
GDKL3_size = (size_t)sysconf(_SC_LEVEL3_CACHE_SIZE); /* error (-1),
will lead too huge L3_size */
#else
diff --git a/monetdb5/mal/mal_pipelines.c b/monetdb5/mal/mal_pipelines.c
--- a/monetdb5/mal/mal_pipelines.c
+++ b/monetdb5/mal/mal_pipelines.c
@@ -251,7 +251,7 @@ PIPELINESinitialize(void)
MT_lock_unset(&mal_contextLock);
return 0;
}
- for (i = 0; i < GDKnr_threads; i++) {
+ for (i = 0; i < GDKnr_pipeline_threads; i++) {
char name[MT_NAME_LEN];
snprintf(name, sizeof(name), "PIPELINEsema%d", i);
workers[i].flag = IDLE;
@@ -300,13 +300,13 @@ runMALpipelines(Client cntxt, MalBlkPtr
.maxparts = maxparts,
.sink = sink,
.master_counter = 0,
- .nr_workers = GDKnr_threads,
+ .nr_workers = GDKnr_pipeline_threads,
.status = 0,
.error = NULL,
.errbuf = GDKgetbuf(),
};
if (maxparts > 0)
- s->nr_workers = MIN(maxparts, GDKnr_threads);
+ s->nr_workers = MIN(maxparts, GDKnr_pipeline_threads);
if (s->nr_workers > 1)
cntxt->sqlprofiler = false;
/* initialize with direct increment of all threads at once */
@@ -322,13 +322,13 @@ runMALpipelines(Client cntxt, MalBlkPtr
/* somehow get number of workers from statement/barrier */
int j = s->nr_workers;
while (j) {
- for (int i = 0; i < GDKnr_threads && j > 0; i++)
+ for (int i = 0; i < GDKnr_pipeline_threads && j > 0; i++)
if (!workers[i].q->last)
j--;
if (j < s->nr_workers) {
/* run with less threads ?? */
s->nr_workers -= j;
- for (int i = 0, k = s->nr_workers; i < GDKnr_threads &&
k > 0; i++) {
+ for (int i = 0, k = s->nr_workers; i <
GDKnr_pipeline_threads && k > 0; i++) {
if (!workers[i].q->last) {
q_enqueue(workers[i].q, s);
k--;
diff --git a/monetdb5/modules/mal/pipeline.c b/monetdb5/modules/mal/pipeline.c
--- a/monetdb5/modules/mal/pipeline.c
+++ b/monetdb5/modules/mal/pipeline.c
@@ -250,8 +250,8 @@ PPcounter(Client cntxt, MalBlkPtr mb, Ma
else
nr = (int)((cnt+SLICE_SIZE-1)/SLICE_SIZE);
FORCEMITODEBUG
- if (nr < GDKnr_threads)
- nr = MIN((int)GDKnr_threads,(int)cnt);
+ if (nr < GDKnr_pipeline_threads)
+ nr = MIN((int)GDKnr_pipeline_threads,(int)cnt);
} else { /* get nparts */
nr = *getArgReference_int(stk, pci, 1);
}
diff --git a/monetdb5/modules/mal/pp_hash.c b/monetdb5/modules/mal/pp_hash.c
--- a/monetdb5/modules/mal/pp_hash.c
+++ b/monetdb5/modules/mal/pp_hash.c
@@ -3874,8 +3874,8 @@ OAHASHno_slices(Client ctx, int *no_slic
else
*no_slices = (int)((h->size+SLICE_SIZE-1)/SLICE_SIZE);
FORCEMITODEBUG
- if (*no_slices < GDKnr_threads)
- *no_slices = GDKnr_threads;
+ if (*no_slices < GDKnr_pipeline_threads)
+ *no_slices = GDKnr_pipeline_threads;
BBPunfix(b->batCacheid);
return MAL_SUCCEED;
}
diff --git a/monetdb5/modules/mal/pp_slicer.c b/monetdb5/modules/mal/pp_slicer.c
--- a/monetdb5/modules/mal/pp_slicer.c
+++ b/monetdb5/modules/mal/pp_slicer.c
@@ -211,8 +211,8 @@ SLICERno_slices(Client cntxt, MalBlkPtr
else
*res = (int)((cnt+SLICE_SIZE-1)/SLICE_SIZE);
FORCEMITODEBUG
- if (*res < GDKnr_threads)
- *res = MIN((int)GDKnr_threads,(int)cnt);
+ if (*res < GDKnr_pipeline_threads)
+ *res = MIN(GDKnr_pipeline_threads, (int) cnt);
BBPunfix(b->batCacheid);
return MAL_SUCCEED;
}
diff --git a/sql/backends/monet5/rel_orderby.c
b/sql/backends/monet5/rel_orderby.c
--- a/sql/backends/monet5/rel_orderby.c
+++ b/sql/backends/monet5/rel_orderby.c
@@ -137,7 +137,7 @@ stmt *
rel2bin_orderby(backend *be, sql_rel *rel, list *refs)
{
assert(rel->r);
- InstrPtr sop = stmt_sop_new(be, GDKnr_threads); /* keep sorted parts in
a set */
+ InstrPtr sop = stmt_sop_new(be, GDKnr_pipeline_threads); /* keep sorted
parts in a set */
//list *shared = rel2bin_project_prepare(be, rel);
stmt *pp = NULL;
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
