Line data Source code
1 : /*
2 : * SPDX-License-Identifier: MPL-2.0
3 : *
4 : * This Source Code Form is subject to the terms of the Mozilla Public
5 : * License, v. 2.0. If a copy of the MPL was not distributed with this
6 : * file, You can obtain one at http://mozilla.org/MPL/2.0/.
7 : *
8 : * Copyright 2024 MonetDB Foundation;
9 : * Copyright August 2008 - 2023 MonetDB B.V.;
10 : * Copyright 1997 - July 2008 CWI.
11 : */
12 :
13 : /*
14 : * The statemens are all checked for being eligible for dataflow.
15 : */
16 : #include "monetdb_config.h"
17 : #include "opt_dataflow.h"
18 : #include "mal_instruction.h"
19 : #include "mal_interpreter.h"
20 : #include "manifold.h"
21 :
22 : /*
23 : * Dataflow processing incurs overhead and is only
24 : * relevant if multiple tasks kan be handled at the same time.
25 : * Also simple expressions dont have to be executed in parallel.
26 : *
27 : * The dataflow analysis centers around the read/write use patterns of
28 : * the variables and the occurrence of side-effect bearing functions.
29 : * Any such function should break the dataflow block as it may rely
30 : * on the sequential order in the plan.
31 : *
32 : * The following state properties can be distinguished for all variables:
33 : * VARWRITE - variable assigned a value in the dataflow block
34 : * VARREAD - variable is used in an argument
35 : * VAR2READ - variable is read in concurrent mode
36 : * VARBLOCK - variable next use terminate the // block, set after encountering an update
37 : *
38 : * Only some combinations are allowed.
39 : */
40 :
41 : #define VARFREE 0
42 : #define VARWRITE 1
43 : #define VARREAD 2
44 : #define VARBLOCK 4
45 : #define VAR2READ 8
46 :
47 : typedef char *States;
48 :
49 : #define setState(S,P,K,F) ( assert(getArg(P,K) < vlimit), (S)[getArg(P,K)] |= F)
50 : #define getState(S,P,K) ((S)[getArg(P,K)])
51 :
52 : typedef enum {
53 : no_region,
54 : singleton_region, // always a single statement
55 : dataflow_region, // statements without or with controlled side effects, in parallel
56 : existing_region, // existing barrier..exit region, copied as-is
57 : sql_region, // region of nonconflicting sql.append/sql.updates only
58 : } region_type;
59 :
60 : typedef struct {
61 : region_type type;
62 : union {
63 : struct {
64 : int level; // level of nesting
65 : } existing_region;
66 : } st;
67 : } region_state;
68 :
69 : static bool
70 1567810 : simpleFlow(InstrPtr *old, int start, int last, region_state *state)
71 : {
72 1567810 : int i, j, k;
73 1567810 : bool simple = true;
74 1567810 : InstrPtr p = NULL, q;
75 :
76 : /* ignore trivial blocks */
77 1567810 : if (last - start == 1)
78 : return true;
79 161814 : if (state->type == existing_region) {
80 : /* don't add additional barriers and garbage collection around
81 : * existing region. */
82 : return true;
83 : }
84 : /* skip sequence of simple arithmetic first */
85 323513 : for (; simple && start < last; start++)
86 162452 : if (old[start]) {
87 162452 : p = old[start];
88 322881 : simple = getModuleId(p) == calcRef || getModuleId(p) == mtimeRef
89 322881 : || getModuleId(p) == strRef || getModuleId(p) == mmathRef;
90 : }
91 278985 : for (i = start; i < last; i++)
92 260939 : if (old[i]) {
93 260939 : q = old[i];
94 249646 : simple = getModuleId(q) == calcRef || getModuleId(q) == mtimeRef
95 510585 : || getModuleId(q) == strRef || getModuleId(q) == mmathRef;
96 260939 : if (!simple) {
97 : /* if not arithmetic than we should consume the previous result directly */
98 1457362 : for (j = q->retc; j < q->argc; j++)
99 2926274 : for (k = 0; k < p->retc; k++)
100 1718535 : if (getArg(p, k) == getArg(q, j))
101 106862 : simple = true;
102 249623 : if (!simple)
103 : return false;
104 : }
105 : p = q;
106 : }
107 : return simple;
108 : }
109 :
110 : /* Updates are permitted if it is a unique update on
111 : * a BAT created in the context of this block
112 : * As far as we know, no SQL nor MAL test re-uses the
113 : * target BAT to insert again and subsequently calls dataflow.
114 : * In MAL scripts, they still can occur.
115 : */
116 :
117 : /* a limited set of MAL instructions may appear in the dataflow block*/
118 : static int
119 6059121 : dataflowBreakpoint(Client cntxt, MalBlkPtr mb, InstrPtr p, States states)
120 : {
121 6059121 : int j;
122 :
123 6059121 : if (p->token == ENDsymbol || p->barrier || isUnsafeFunction(p)
124 5921484 : || (isMultiplex(p) && MANIFOLDtypecheck(cntxt, mb, p, 0) == NULL)) {
125 137914 : return TRUE;
126 : }
127 :
128 : /* flow blocks should be closed when we reach a point
129 : where a variable is assigned more then once or already
130 : being read.
131 : */
132 12542614 : for (j = 0; j < p->retc; j++)
133 6621577 : if (getState(states, p, j) & (VARWRITE | VARREAD | VARBLOCK)) {
134 : return 1;
135 : }
136 :
137 : /* update instructions can be updated if the target variable
138 : * has not been read in the block so far */
139 5921037 : if (isUpdateInstruction(p)) {
140 : /* the SQL update functions change BATs that are not
141 : * explicitly mentioned as arguments (and certainly not as the
142 : * first argument), but that can still be available to the MAL
143 : * program (see bugs.monetdb.org/6641) */
144 189254 : if (getModuleId(p) == sqlRef)
145 : return 1;
146 186271 : return getState(states, p, p->retc) & (VARREAD | VARBLOCK);
147 : }
148 :
149 29388508 : for (j = p->retc; j < p->argc; j++) {
150 23656741 : if (getState(states, p, j) & VARBLOCK) {
151 : return 1;
152 : }
153 : }
154 5731767 : return hasSideEffects(mb, p, FALSE);
155 : }
156 :
157 : static str
158 15667540 : get_str_arg(MalBlkPtr mb, InstrPtr p, int argno)
159 : {
160 15667540 : int var = getArg(p, argno);
161 15667540 : return getVarConstant(mb, var).val.sval;
162 : }
163 :
164 : static str
165 410113 : get_sql_sname(MalBlkPtr mb, InstrPtr p)
166 : {
167 410113 : return get_str_arg(mb, p, 2);
168 : }
169 :
170 : static str
171 410113 : get_sql_tname(MalBlkPtr mb, InstrPtr p)
172 : {
173 410113 : return get_str_arg(mb, p, 3);
174 : }
175 :
176 : static str
177 15667538 : get_sql_cname(MalBlkPtr mb, InstrPtr p)
178 : {
179 15667538 : return get_str_arg(mb, p, 4);
180 : }
181 :
182 :
183 : static bool
184 1583692 : isSqlAppendUpdate(MalBlkPtr mb, InstrPtr p)
185 : {
186 1583692 : if (p->modname != sqlRef)
187 : return false;
188 1103592 : if (p->fcnname != appendRef && p->fcnname != updateRef)
189 : return false;
190 :
191 : // pattern("sql", "append", mvc_append_wrap, false, "...", args(1,8, arg("",int),
192 : // arg("mvc",int),
193 : // arg("sname",str),
194 : // arg("tname",str),
195 : // arg("cname",str),
196 : // arg("offset",lng),
197 : // batarg("pos",oid),
198 : // argany("ins",0))),
199 :
200 : // pattern("sql", "update", mvc_update_wrap, false, "...", args(1,7, arg("",int),
201 : // arg("mvc",int),
202 : // arg("sname",str),
203 : // arg("tname",str),
204 : // arg("cname",str),
205 : // argany("rids",0),
206 : // argany("upd",0)))
207 :
208 482537 : if ((p->fcnname == appendRef && p->argc != 8)
209 482537 : || (p->fcnname == updateRef && p->argc != 7))
210 : return false;
211 :
212 482537 : int mvc_var = getArg(p, 1);
213 482537 : if (getVarType(mb, mvc_var) != TYPE_int)
214 : return false;
215 :
216 482537 : int sname_var = getArg(p, 2);
217 482537 : if (getVarType(mb, sname_var) != TYPE_str || !isVarConstant(mb, sname_var))
218 : return false;
219 :
220 482537 : int tname_var = getArg(p, 3);
221 482537 : if (getVarType(mb, tname_var) != TYPE_str || !isVarConstant(mb, tname_var))
222 : return false;
223 :
224 482538 : int cname_var = getArg(p, 4);
225 482538 : if (getVarType(mb, cname_var) != TYPE_str || !isVarConstant(mb, cname_var))
226 : return false;
227 :
228 : return true;
229 : }
230 :
231 : static bool
232 482540 : sqlBreakpoint(MalBlkPtr mb, InstrPtr *first, InstrPtr *p)
233 : {
234 482540 : InstrPtr instr = *p;
235 482540 : if (!isSqlAppendUpdate(mb, instr))
236 : return true;
237 :
238 410112 : str my_sname = get_sql_sname(mb, instr);
239 410112 : str my_tname = get_sql_tname(mb, instr);
240 410112 : str my_cname = get_sql_cname(mb, instr);
241 15667537 : for (InstrPtr *q = first; q < p; q++) {
242 15257426 : str cname = get_sql_cname(mb, *q);
243 15257426 : if (strcmp(my_cname, cname) != 0) {
244 : // different cname, no conflict
245 15257425 : continue;
246 : }
247 1 : str tname = get_sql_tname(mb, *q);
248 1 : if (strcmp(my_tname, tname) != 0) {
249 : // different tname, no conflict
250 0 : continue;
251 : }
252 1 : str sname = get_sql_sname(mb, *q);
253 1 : if (strcmp(my_sname, sname) != 0) {
254 : // different sname, no conflict
255 0 : continue;
256 : }
257 : // Found a statement in the region that works on the same column so this is a breakpoint
258 : return true;
259 : }
260 :
261 : // None of the statements in the region works on this column so no breakpoint necessary
262 : return false;
263 : }
264 :
265 : static bool
266 7870960 : checkBreakpoint(Client cntxt, MalBlkPtr mb, InstrPtr *first, InstrPtr *p,
267 : States states, region_state *state)
268 : {
269 7870960 : InstrPtr instr = *p;
270 7870960 : switch (state->type) {
271 : case singleton_region:
272 : // by definition
273 : return true;
274 6059118 : case dataflow_region:
275 6059118 : return dataflowBreakpoint(cntxt, mb, instr, states);
276 7385 : case existing_region:
277 7385 : if (state->st.existing_region.level == 0) {
278 : // previous statement ended the region so we break here
279 : return true;
280 : }
281 6632 : if (blockStart(instr)) {
282 38 : state->st.existing_region.level += 1;
283 6594 : } else if (blockExit(instr)) {
284 791 : state->st.existing_region.level -= 1;
285 : }
286 : return false;
287 482532 : case sql_region:
288 482532 : return sqlBreakpoint(mb, first, p);
289 : default:
290 : // serious corruption has occurred.
291 0 : assert(0); /* corrupted region_type */
292 : abort();
293 : }
294 : assert(0); /* unreachable */
295 : return true;
296 : }
297 :
298 : static void
299 1101940 : decideRegionType(Client cntxt, MalBlkPtr mb, InstrPtr p, States states,
300 : region_state *state)
301 : {
302 1101940 : (void) cntxt;
303 :
304 1101940 : state->type = no_region;
305 1101940 : if (blockStart(p)) {
306 753 : state->type = existing_region;
307 753 : state->st.existing_region.level = 1;
308 1101187 : } else if (p->token == ENDsymbol) {
309 0 : state->type = existing_region;
310 1101187 : } else if (isSqlAppendUpdate(mb, p)) {
311 72429 : state->type = sql_region;
312 1028758 : } else if (p->barrier) {
313 673 : state->type = singleton_region;
314 1028085 : } else if (isUnsafeFunction(p)) {
315 373773 : state->type = singleton_region;
316 654323 : } else if (isUpdateInstruction(p)
317 36 : && getModuleId(p) != sqlRef
318 20 : && (getState(states, p, p->retc) & (VARREAD | VARBLOCK)) == 0) {
319 : // Special case. Unless they're from the sql module, instructions with
320 : // names like 'append', 'update', 'delete', 'grow', etc., are expected
321 : // to express their side effects as data dependencies, for example,
322 : // X5 := bat.append(X_5, ...)
323 20 : state->type = dataflow_region;
324 654304 : } else if (hasSideEffects(mb, p, false)) {
325 481065 : state->type = singleton_region;
326 173235 : } else if (isMultiplex(p)) {
327 544 : state->type = singleton_region;
328 : } else {
329 172692 : state->type = dataflow_region;
330 : }
331 1101949 : assert(state->type != no_region);
332 1101949 : }
333 :
334 :
335 : /* dataflow blocks are transparent, because they are always
336 : executed, either sequentially or in parallel */
337 :
338 : str
339 465895 : OPTdataflowImplementation(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
340 : InstrPtr pci)
341 : {
342 465895 : int i, j, k, start, slimit, breakpoint, actions = 0;
343 465895 : bool simple = true;
344 465895 : int flowblock = 0;
345 465895 : InstrPtr p, *old = NULL, q;
346 465895 : int limit, vlimit;
347 465895 : States states = NULL;
348 465895 : region_state state = { singleton_region };
349 465895 : str msg = MAL_SUCCEED;
350 :
351 : /* don't use dataflow on single processor systems */
352 465895 : if (GDKnr_threads <= 1 || cntxt->workerlimit == 1)
353 0 : goto wrapup;
354 :
355 465895 : if (optimizerIsApplied(mb, dataflowRef))
356 0 : goto wrapup;
357 465894 : (void) stk;
358 : /* inlined functions will get their dataflow control later */
359 465894 : if (mb->inlineProp)
360 0 : goto wrapup;
361 :
362 465894 : vlimit = mb->vsize;
363 465894 : states = (States) GDKzalloc(vlimit * sizeof(char));
364 465896 : if (states == NULL) {
365 0 : throw(MAL, "optimizer.dataflow", SQLSTATE(HY013) MAL_MALLOC_FAIL);
366 : }
367 :
368 465896 : setVariableScope(mb);
369 :
370 465893 : limit = mb->stop;
371 465893 : slimit = mb->ssize;
372 465893 : old = mb->stmt;
373 465893 : if (newMalBlkStmt(mb, mb->ssize) < 0) {
374 0 : GDKfree(states);
375 0 : throw(MAL, "optimizer.dataflow", SQLSTATE(HY013) MAL_MALLOC_FAIL);
376 : }
377 :
378 : /* inject new dataflow barriers using a single pass through the program */
379 465895 : start = 0;
380 465895 : state.type = singleton_region;
381 7870480 : for (i = 1; mb->errors == NULL && i < limit; i++) {
382 7870480 : p = old[i];
383 7870480 : assert(p);
384 7870480 : breakpoint = checkBreakpoint(cntxt, mb, &old[start], &old[i], states, &state);
385 7870670 : if (breakpoint) {
386 : /* close previous flow block */
387 1567773 : simple = simpleFlow(old, start, i, &state);
388 :
389 1567773 : if (!simple) {
390 143022 : if ((flowblock = newTmpVariable(mb, TYPE_bit)) < 0
391 143024 : || (q = newFcnCall(mb, languageRef, dataflowRef)) == NULL) {
392 0 : msg = createException(MAL, "optimizer.dataflow",
393 : SQLSTATE(HY013) MAL_MALLOC_FAIL);
394 0 : break;
395 : }
396 143025 : q->barrier = BARRIERsymbol;
397 143025 : getArg(q, 0) = flowblock;
398 143025 : pushInstruction(mb, q);
399 143025 : actions++;
400 : }
401 : // copyblock the collected statements
402 9439283 : for (j = start; j < i; j++) {
403 7871465 : q = old[j];
404 7871465 : pushInstruction(mb, q);
405 7871542 : old[j] = NULL;
406 : // collect BAT variables garbage collected within the block
407 7871542 : if (!simple)
408 33906686 : for (k = q->retc; k < q->argc; k++) {
409 27503828 : if (getState(states, q, k) & VAR2READ
410 7328093 : && getEndScope(mb, getArg(q, k)) == j
411 1408469 : && isaBatType(getVarType(mb, getArg(q, k)))) {
412 1343465 : InstrPtr r;
413 1343465 : r = newInstruction(NULL, languageRef, passRef);
414 1343468 : if (r == NULL) {
415 0 : msg = createException(MAL, "optimizer.dataflow",
416 : SQLSTATE(HY013)
417 : MAL_MALLOC_FAIL);
418 0 : break;
419 : }
420 1343468 : getArg(r, 0) = newTmpVariable(mb, TYPE_void);
421 1343466 : if (getArg(r, 0) < 0) {
422 0 : freeInstruction(r);
423 0 : msg = createException(MAL, "optimizer.dataflow",
424 : SQLSTATE(HY013)
425 : MAL_MALLOC_FAIL);
426 0 : break;
427 : }
428 1343466 : r = pushArgument(mb, r, getArg(q, k));
429 1343466 : pushInstruction(mb, r);
430 : }
431 : }
432 7871507 : if (msg)
433 : break;
434 : }
435 1567818 : if (msg)
436 : break;
437 : /* exit parallel block */
438 1567818 : if (!simple) {
439 143020 : q = newAssignment(mb);
440 143024 : if (q == NULL) {
441 0 : msg = createException(MAL, "optimizer.dataflow",
442 : SQLSTATE(HY013) MAL_MALLOC_FAIL);
443 0 : break;
444 : }
445 143024 : q->barrier = EXITsymbol;
446 143024 : getArg(q, 0) = flowblock;
447 143024 : pushInstruction(mb, q);
448 : }
449 1567822 : if (p->token == ENDsymbol) {
450 : break;
451 : }
452 : // Start a new region
453 1101933 : memset((char *) states, 0, vlimit * sizeof(char));
454 1101933 : start = i;
455 1101933 : decideRegionType(cntxt, mb, p, states, &state);
456 : }
457 : // remember you assigned/read variables
458 15597927 : for (k = 0; k < p->retc; k++)
459 8193109 : setState(states, p, k, VARWRITE);
460 7404818 : if (isUpdateInstruction(p)
461 823015 : && (getState(states, p, 1) == 0
462 663227 : || getState(states, p, 1) & VARWRITE))
463 420124 : setState(states, p, 1, VARBLOCK);
464 38743478 : for (k = p->retc; k < p->argc; k++)
465 31338893 : if (!isVarConstant(mb, getArg(p, k))) {
466 14197510 : if (getState(states, p, k) & VARREAD)
467 5928735 : setState(states, p, k, VAR2READ);
468 8268775 : else if (getState(states, p, k) & VARWRITE)
469 5952274 : setState(states, p, k, VARREAD);
470 : }
471 : }
472 :
473 : /* take the remainder as is */
474 114460018 : for (; i < slimit; i++)
475 113994122 : if (old[i])
476 13513188 : pushInstruction(mb, old[i]);
477 : /* Defense line against incorrect plans */
478 465896 : if (msg == MAL_SUCCEED && actions > 0) {
479 130859 : msg = chkTypes(cntxt->usermodule, mb, FALSE);
480 130858 : if (msg == MAL_SUCCEED) {
481 130858 : msg = chkFlow(mb);
482 130859 : if (msg == MAL_SUCCEED)
483 130859 : msg = chkDeclarations(mb);
484 : }
485 : }
486 335037 : wrapup:
487 : /* keep actions taken as a fake argument */
488 465892 : (void) pushInt(mb, pci, actions);
489 :
490 465891 : if (states)
491 465891 : GDKfree(states);
492 465896 : if (old)
493 465896 : GDKfree(old);
494 : return msg;
495 : }
|
