Skip to content

Commit 30a0e53

Browse files
JairikJairik
committed
Initial QPEOMP implementation
1 parent 39c86be commit 30a0e53

File tree

4 files changed

+134
-17
lines changed

4 files changed

+134
-17
lines changed

‎.gitignore‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,3 +9,4 @@ build/tests/duplicate-test
99
build/tests/executeEngine-serial-test
1010
build/tests/serial-SELECT-test
1111
tokenizer/src/tokenizer.o
12+
QPEOMP

‎QPEOMP‎

10.4 KB
Binary file not shown.

‎QPEOMP.c‎

Lines changed: 132 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -4,23 +4,35 @@
44
#include <stdio.h>
55
#include <stdlib.h>
66
#include <string.h>
7+
#include <strings.h>
78
#include <time.h>
89
#include <omp.h>
910
#include "../include/executeEngine-serial.h"
1011
#include "../include/connectEngine.h"
12+
#include "../include/printHelper.h"
13+
1114
// ANSI color codes for pretty printing
1215
#define CYAN "\x1b[36m"
1316
#define YELLOW "\x1b[33m"
1417
#define BOLD "\x1b[1m"
1518
#define RESET "\x1b[0m"
1619

20+
// Helper to safely copy strings with truncation (local version)
21+
static inline void safe_copy(char *dst, size_t n, const char *src) {
22+
snprintf(dst, n, "%.*s", (int)n - 1, src);
23+
}
24+
1725
int main(int argc, char *argv[]) {
18-
19-
// NOTE: defaulting to sample queries file for ease of testing. Can implement CLI arg later.
20-
(void)argc; (void)argv; // Unused for now. Prevent warnings.
26+
27+
// Pull out number of defined threads from CLI args (default to 8)
28+
int num_threads = 8;
29+
if (argc >= 2) {
30+
num_threads = atoi(argv[1]);
31+
}
32+
omp_set_num_threads(num_threads);
2133

2234
// Start a timer for total runtime statistics
23-
clock_t totalStart = clock();
35+
double totalStart = omp_get_wtime();
2436

2537
// Instantiate an engine object to handle the execution of the query
2638
struct engineS *engine = initializeEngineSerial(
@@ -32,7 +44,7 @@ int main(int argc, char *argv[]) {
3244
);
3345

3446
// End timer for engine initialization
35-
double initTimeTaken = ((double)clock() - totalStart) / CLOCKS_PER_SEC;
47+
double initTimeTaken = ((double)omp_get_wtime() - totalStart) / CLOCKS_PER_SEC;
3648

3749
// Load the COMMANDS into memory (from COMMAND text file)
3850
const char *query_file = "sample-queries.txt";
@@ -66,24 +78,128 @@ int main(int argc, char *argv[]) {
6678
fclose(fp);
6779

6880
// End timer for loading queries
69-
double loadTimeTaken = ((double)clock() - totalStart) / CLOCKS_PER_SEC;
70-
71-
// Run each command from the command input file
72-
char *query = strtok(buffer, ";");
73-
while (query) {
74-
// Trim whitespace
75-
query = trim(query);
76-
if (*query) {
77-
run_test_query(engine, query, ROW_LIMIT); // Limit output to 10 rows for testing
81+
double loadTimeTaken = ((double)omp_get_wtime() - totalStart) / CLOCKS_PER_SEC;
82+
83+
// Split queries into an array to avoid strtok race conditions
84+
#define MAX_QUERIES 1000
85+
char *queries[MAX_QUERIES];
86+
int query_count = 0;
87+
88+
char *token = strtok(buffer, ";");
89+
while (token && query_count < MAX_QUERIES) {
90+
queries[query_count++] = token;
91+
token = strtok(NULL, ";");
92+
}
93+
94+
// Parallel Execution with Ordered Output
95+
#pragma omp parallel for ordered schedule(dynamic)
96+
for (int i = 0; i < query_count; i++) {
97+
char *query = trim(queries[i]);
98+
if (!*query) continue;
99+
100+
// Tokenize each query
101+
Token tokens[MAX_TOKENS];
102+
int num_tokens = tokenize(query, tokens, MAX_TOKENS);
103+
104+
// Parse tokens and instantiate benchmarking variables
105+
ParsedSQL parsed;
106+
struct resultSetS *result = NULL;
107+
bool success = false;
108+
double execTime = 0;
109+
int rowsAffected = 0;
110+
bool parseFailed = false;
111+
if (num_tokens > 0) {
112+
parsed = parse_tokens(tokens);
113+
114+
// Prepare Select Items
115+
const char *selectItems[parsed.num_columns > 0 ? parsed.num_columns : 1];
116+
int numSelectItems = 0;
117+
if (!parsed.select_all) {
118+
numSelectItems = parsed.num_columns;
119+
for (int k = 0; k < numSelectItems; k++) selectItems[k] = parsed.columns[k];
120+
}
121+
122+
double start = omp_get_wtime(); // Start timing for benchmarking
123+
124+
// Execute based on command type
125+
if (parsed.command == CMD_INSERT) {
126+
if (parsed.num_values == 12) {
127+
record r;
128+
r.command_id = strtoull(parsed.insert_values[0], NULL, 10);
129+
safe_copy(r.raw_command, sizeof(r.raw_command), parsed.insert_values[1]);
130+
safe_copy(r.base_command, sizeof(r.base_command), parsed.insert_values[2]);
131+
safe_copy(r.shell_type, sizeof(r.shell_type), parsed.insert_values[3]);
132+
r.exit_code = atoi(parsed.insert_values[4]);
133+
safe_copy(r.timestamp, sizeof(r.timestamp), parsed.insert_values[5]);
134+
r.sudo_used = (strcasecmp(parsed.insert_values[6], "true") == 0 || strcmp(parsed.insert_values[6], "1") == 0);
135+
safe_copy(r.working_directory, sizeof(r.working_directory), parsed.insert_values[7]);
136+
r.user_id = atoi(parsed.insert_values[8]);
137+
safe_copy(r.user_name, sizeof(r.user_name), parsed.insert_values[9]);
138+
safe_copy(r.host_name, sizeof(r.host_name), parsed.insert_values[10]);
139+
r.risk_level = atoi(parsed.insert_values[11]);
140+
141+
success = executeQueryInsertSerial(engine, parsed.table, &r);
142+
}
143+
}
144+
else if (parsed.command == CMD_DELETE) {
145+
struct whereClauseS *whereClause = convert_conditions(&parsed);
146+
result = executeQueryDeleteSerial(engine, parsed.table, whereClause);
147+
if (result) rowsAffected = result->numRecords;
148+
free_where_clause_list(whereClause);
149+
}
150+
else if (parsed.command == CMD_SELECT) {
151+
struct whereClauseS *whereClause = convert_conditions(&parsed);
152+
result = executeQuerySelectSerial(engine, selectItems, numSelectItems, parsed.table, whereClause);
153+
free_where_clause_list(whereClause);
154+
}
155+
156+
execTime = (double)(omp_get_wtime() - start) / CLOCKS_PER_SEC;
157+
} else {
158+
parseFailed = true;
78159
}
79-
query = strtok(NULL, ";");
160+
161+
// Print all results in order
162+
#pragma omp ordered
163+
{
164+
printf("Executing Query: %s\n", query);
165+
166+
if (parseFailed) {
167+
printf("Tokenization failed.\n");
168+
} else {
169+
if (parsed.command == CMD_INSERT) {
170+
if (parsed.num_values != 12) {
171+
printf("Error: INSERT requires exactly 12 values.\n");
172+
} else if (success) {
173+
printf("Insert successful. Execution Time: %ld\n\n", (long)execTime);
174+
} else {
175+
printf("Insert failed. Execution Time: %ld\n\n", (long)execTime);
176+
}
177+
} else if (parsed.command == CMD_DELETE) {
178+
if (result) {
179+
printf("Delete successful. Rows affected: %d. Execution Time: %ld\n\n", rowsAffected, (long)execTime);
180+
} else {
181+
printf("Delete failed. Execution Time: %ld\n\n", (long)execTime);
182+
}
183+
} else if (parsed.command == CMD_SELECT) {
184+
printTable(NULL, result, ROW_LIMIT);
185+
printf("\n");
186+
} else if (parsed.command == CMD_NONE) {
187+
printf("No command detected.\n");
188+
} else {
189+
fprintf(stderr, "Unsupported command.\n");
190+
}
191+
}
192+
}
193+
194+
// Cleanup (Local)
195+
if (result) freeResultSet(result);
80196
}
81197

82198
free(buffer);
83199
destroyEngineSerial(engine);
84200

85201
// Print total runtime statistics in pretty colors
86-
double totalTimeTaken = ((double)clock() - totalStart) / CLOCKS_PER_SEC;
202+
double totalTimeTaken = ((double)omp_get_wtime() - totalStart) / CLOCKS_PER_SEC;
87203
printf(CYAN "======= Execution Summary =======" RESET "\n");
88204
printf(CYAN "Engine Initialization Time: " RESET YELLOW "%.4f seconds\n" RESET, initTimeTaken);
89205
printf(CYAN "Query Loading Time: " RESET YELLOW "%.4f seconds\n" RESET, loadTimeTaken - initTimeTaken);

‎connectEngine.c‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
/* Helper functions for QPE files. Used for testing with query text files. Cannot be reliably parralelized */
1+
/* Helper functions for some QPE files. Used for testing with query text files. Cannot be reliably parralelized */
22

33
#define _POSIX_C_SOURCE 200809L // For strdup
44
#include <stdio.h>

0 commit comments

Comments
 (0)