Changeset: c0dc081eb0a9 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=c0dc081eb0a9
Added Files:
sql/backends/monet5/bam/bam_globals.h
Modified Files:
sql/backends/monet5/bam/85_bam.sql
sql/backends/monet5/bam/bam.mal
sql/backends/monet5/bam/bam_loader.c
sql/backends/monet5/bam/bam_loader.h
sql/backends/monet5/bam/bam_sql.h
Branch: DVframework_bam
Log Message:
Make new version of bam_loader use sleeping threads from thread pool to write
stuff to binary files
diffs (truncated from 1240 to 300 lines):
diff --git a/sql/backends/monet5/bam/85_bam.sql
b/sql/backends/monet5/bam/85_bam.sql
--- a/sql/backends/monet5/bam/85_bam.sql
+++ b/sql/backends/monet5/bam/85_bam.sql
@@ -1,7 +1,7 @@
-CREATE PROCEDURE bam_loader_repos(bam_repos STRING, dbschema INT, num_threads
INT)
+CREATE PROCEDURE bam_loader_repos(bam_repos STRING, dbschema INT, nr_threads
INT)
EXTERNAL NAME bam.bam_loader_repos;
-CREATE PROCEDURE bam_loader_file(bam_file STRING, dbschema INT)
+CREATE PROCEDURE bam_loader_file(bam_file STRING, dbschema INT, nr_threads INT)
EXTERNAL NAME bam.bam_loader_file;
CREATE PROCEDURE drop_alignment_tables(dbschema INT)
diff --git a/sql/backends/monet5/bam/bam.mal b/sql/backends/monet5/bam/bam.mal
--- a/sql/backends/monet5/bam/bam.mal
+++ b/sql/backends/monet5/bam/bam.mal
@@ -1,10 +1,10 @@
module bam;
-pattern bam.bam_loader_repos(bam_repos:str, dbschema:int, num_threads:int):void
+pattern bam.bam_loader_repos(bam_repos:str, dbschema:int, nr_threads:int):void
address bam_loader_repos
comment "Read the files in the given BAM repository; fill and return a
temp_container accordingly.";
-pattern bam.bam_loader_file(bam_file:str, dbschema:int):void
+pattern bam.bam_loader_file(bam_file:str, dbschema:int, nr_threads:int):void
address bam_loader_file
comment "Read the bam file given as the first parameter; fill and return a
temp_container accordingly.";
diff --git a/sql/backends/monet5/bam/bam_globals.h
b/sql/backends/monet5/bam/bam_globals.h
new file mode 100644
--- /dev/null
+++ b/sql/backends/monet5/bam/bam_globals.h
@@ -0,0 +1,8 @@
+#ifndef _BAM_GLOBALS_H
+#define _BAM_GLOBALS_H
+
+#define ROOT "/export/scratch2/robin"
+#define MONETDB_HOME ROOT"/Monet/SOURCE/MonetDB"
+#define BAM_HOME MONETDB_HOME"/sql/backends/monet5/bam"
+
+#endif
diff --git a/sql/backends/monet5/bam/bam_loader.c
b/sql/backends/monet5/bam/bam_loader.c
--- a/sql/backends/monet5/bam/bam_loader.c
+++ b/sql/backends/monet5/bam/bam_loader.c
@@ -5,60 +5,6 @@
#include "bam_sql.h"
#include "bam_loader.h"
-#define _append_option_cond_str(elt, opt, cmp, flag) \
- if(strcmp((opt).tag, (cmp)) == 0) \
- { \
- (elt) = (opt).value; \
- (flag) = TRUE; \
- continue; \
- }
-
-#define _append_option_cond_lng(elt, opt, cmp, flag, l, s) \
- if(strcmp((opt).tag, (cmp)) == 0) \
- { \
- (l) = strtol((opt).value, &s, 10); \
- if((s) == ((opt).value) || (l) == LONG_MIN || (l) == LONG_MAX) \
- l = -1; \
- (elt) = l; \
- (flag) = TRUE; \
- continue; \
- }
-
-#define _flush_strs(str_arr, nr_str_arr, file, fp, i, strl) \
- for((i)=0; (i)<(nr_str_arr); ++(i)) \
- { \
- (strl) = strlen((str_arr)[i]); \
- if(fwrite((void *)(str_arr)[i], sizeof(char), (strl), (file)) !=
(strl)) \
- throw(MAL, "_flush_bam_storage", "Error on writing %d strings to
file '%s'", nr_str_arr, fp); \
- fputc('\n', file); \
- }
-
-/*
-static str
-_flush_strs(str *str_arr, unsigned int nr_str_arr, FILE *file, str fp,
unsigned int i, unsigned int strl)
-{
- for((i)=0; (i)<(nr_str_arr); ++(i))
- {
- (strl) = strlen((str_arr)[i]);
- if(fwrite((void *)(str_arr)[i], sizeof(char), (strl), (file)) !=
(strl))
- throw(MAL, "_flush_bam_storage", "Error on writing %d strings to
file '%s'", nr_str_arr, fp);
- fputc('\n', file);
- }
- return MAL_SUCCEED;
-}*/
-
-#define _flush_shts(sht_arr, nr_sht_arr, file, fp) \
- if(fwrite((void *)sht_arr, sizeof(sht), nr_sht_arr, file) != nr_sht_arr) \
- throw(MAL, "_flush_bam_storage", "Error on writing %d shts to file
'%s'", nr_sht_arr, fp); \
-
-#define _flush_ints(int_arr, nr_int_arr, file, fp) \
- if(fwrite((void *)int_arr, sizeof(int), nr_int_arr, file) != nr_int_arr) \
- throw(MAL, "_flush_bam_storage", "Error on writing %d ints to file
'%s'", nr_int_arr, fp); \
-
-#define _flush_lngs(lng_arr, nr_lng_arr, file, fp) \
- if(fwrite((void *)lng_arr, sizeof(lng), nr_lng_arr, file) != nr_lng_arr) \
- throw(MAL, "_flush_bam_storage", "Error on writing %d lngs to file
'%s'", nr_lng_arr, fp); \
-
/*
* File format specific structures
*/
@@ -168,10 +114,17 @@ typedef struct {
FILE *alignments_extra[4];
} _bam_storage;
-
+typedef struct {
+ int max_nr_threads;
+ bit *thread_available;
+ pthread_mutex_t threads_data_lock;
+} _threads_data;
typedef struct {
- sht thread_id;
+ _threads_data *d;
+ bit concurrent_readers;
+ pthread_t *threads; /* All readers should have a thread pool separate from
the writers */
+ int thread_id;
sht dbschema;
Client cntxt;
MalBlkPtr mb;
@@ -179,20 +132,45 @@ typedef struct {
sht nr_file_paths;
sht *current_file;
sht start_file_id;
-} _thread_data;
+ pthread_mutex_t *current_file_lock;
+ pthread_mutex_t *client_lock;
+} _reader_thread_data;
+typedef struct {
+ void *arr;
+ unsigned int el_size;
+ unsigned int arr_size;
+ FILE *file;
+ str fp;
+} _write_action_data;
-pthread_mutex_t current_file_lock;
-pthread_mutex_t client_lock;
+typedef struct {
+ _threads_data *d;
+ _write_action_data *write_actions;
+ sht nr_write_actions;
+ sht *cur_write_action;
+ pthread_t *threads;
+ int thread_id;
+ pthread_mutex_t *current_write_action_lock;
+} _writer_thread_data;
+
+
+FILE *_log;
static str
-_init_bam_storage(_bam_storage *bs, sht file_id, str file_location, int
cap_header, int cap_alignments, int cap_alignments_extra)
+_init_bam_storage(_bam_storage *bs, sht file_id, str file_location, int
max_nr_reader_threads)
{
- str flush_dir_ptn = "/export/scratch2/robin/bam_loader_flush/%d";
+ str flush_dir_ptn = DIR_FLUSH"/%d";
char flush_dir[64];
unsigned int i;
+ int cap_header = HEADER_STORAGE_LIMIT / max_nr_reader_threads;
+ int cap_alignments = ALIGNMENTS_STORAGE_LIMIT / max_nr_reader_threads;
+ int cap_alignments_extra = ALIGNMENTS_EXTRA_STORAGE_LIMIT /
max_nr_reader_threads;
+
+ mkdir(DIR_FLUSH, 0777);
+
bs->files_file_id = file_id;
bs->files_file_location = file_location;
bs->files_format_version = NULL;
@@ -377,14 +355,35 @@ static void
/* Helper functions */
-
-/*
- * Uses printf to write to merovingian.log
- */
static void
_append_to_log(str mssg)
-{
+{
printf("bam_loader log: %s\n", mssg);
+ if(_log != NULL)
+ {
+ fprintf(_log, "> %s\n\n", mssg);
+ fflush(_log);
+ }
+}
+
+static void
+_open_log(void)
+{
+ _log = fopen(LOGPATH, "a");
+ if(_log == NULL)
+ printf("Error opening logfile '%s' for writing.\n", LOGPATH);
+ else
+ fprintf(_log,
"=============================================================\n\n");
+}
+
+static void
+_close_log(void)
+{
+ if(_log != NULL)
+ {
+ fprintf(_log,
"\n=============================================================\n\n\n\n\n\n\n\n\n\n");
+ fclose(_log);
+ }
}
@@ -430,7 +429,8 @@ static int
}
static str
-_run_sql_from_file(Client cntxt, str filename, char
search[][MAX_SQL_SEARCH_REPLACE_CHARS], char
replace[][MAX_SQL_SEARCH_REPLACE_CHARS], int nr_replacement_strings)
+_run_sql_from_file(Client cntxt, pthread_mutex_t *client_lock, str filename,
+ char search[][MAX_SQL_SEARCH_REPLACE_CHARS], char
replace[][MAX_SQL_SEARCH_REPLACE_CHARS], int nr_replacement_strings)
{
char err[256];
char filepath[512];
@@ -441,9 +441,11 @@ static str
if(file_contents == NULL)
throw(MAL, "_run_sql_from_file", "%s", err);
_append_to_log(file_contents);
- pthread_mutex_lock(&client_lock);
+
+ if(client_lock != NULL) pthread_mutex_lock(client_lock);
result = SQLstatementIntern(cntxt, &file_contents,
"bam.create_alignment_tables", TRUE, FALSE);
- pthread_mutex_unlock(&client_lock);
+ if(client_lock != NULL) pthread_mutex_unlock(client_lock);
+
GDKfree(file_contents);
return result;
}
@@ -512,7 +514,6 @@ static lng
FILE *file;
nr_file_paths = _get_line_nr(repo_path);
- printf("nr_file_paths: "LLFMT"\n", nr_file_paths);
*ret_file_paths = file_paths = (str*) GDKmalloc(nr_file_paths *
sizeof(str));
if(file_paths == NULL)
@@ -590,18 +591,176 @@ static str
return MAL_SUCCEED;
}
+
+
+static str
+_write_arr(void *arr, unsigned int el_size, unsigned int arr_size, FILE *file,
str fp)
+{
+ if(fwrite(arr, el_size, arr_size, file) != arr_size)
+ throw(MAL, "_flush_bam_storage", "Error on writing %d items of size %d
to file '%s'", arr_size, el_size, fp);
+ return MAL_SUCCEED;
+}
+
+static str
+_write_strs(str *arr, unsigned int arr_size, FILE *file, str fp)
+{
+ unsigned int i;
+ str err;
+ for(i=0; i<arr_size; ++i)
+ {
+ if((err = _write_arr((void *)arr[i], sizeof(char), strlen(arr[i]),
file, fp)) != MAL_SUCCEED)
+ return err;
+ fputc('\n', file);
+ }
+ return MAL_SUCCEED;
+}
+
+static void *
+_writer(void *d);
+
+static str
+_write(_writer_thread_data *data)
+{
+ int *children = NULL;
+ _writer_thread_data *data_children = NULL;
+ int nr_children = 0;
+ int i;
+ str result;
+ /* Every write action with possible children needs its own thread pool.
Otherwise, availability of a thread can
+ only be set after it has been joined with its parent. If we however would
ignore this and just use a thread for another
+ purpose before it is joined with its parent, things would go wrong because
as soon as the parent would then call the
+ join function, the child would already be busy doing something else. */
+ pthread_t *threads = NULL;
+
+ if(data->d != NULL)
+ {
+ children = (int *)GDKmalloc(data->d->max_nr_threads * sizeof(int));
+ data_children = (_writer_thread_data
*)GDKmalloc(data->d->max_nr_threads * sizeof(_writer_thread_data));
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
