Fix multi-threading logger

Author: MicaelCarvalho

bootstrap/lib/logger.py

@@ -15,6 +15,7 @@
 import os
 import sqlite3
 import sys
+from contextlib import closing
 
 
 class Logger(object):
@@ -71,15 +72,22 @@ def code(value):
 
     log_level = None  # log level
     dir_logs = None
-    sqlite_cur = None
     sqlite_file = None
-    sqlite_conn = None
+    connection = None
     path_txt = None
     file_txt = None
     name = None
     max_lineno_width = 3
 
-    def __new__(cls, dir_logs=None, name='logs'):
+    def __new__(cls, dir_logs=None, name=None):
+        return Logger._get_instance(dir_logs, name)
+
+    def __call__(self, *args, **kwargs):
+        return self.log_message(*args, **kwargs, stack_displacement=2)
+
+    @staticmethod
+    def _get_instance(dir_logs=None, name=None):
+        name = name or 'logs'
         if Logger._instance is None:
             Logger._instance = object.__new__(Logger)
             Logger._instance.set_level(Logger._instance.INFO)
@@ -90,16 +98,12 @@ def __new__(cls, dir_logs=None, name='logs'):
                 Logger._instance.path_txt = os.path.join(dir_logs, '{}.txt'.format(name))
                 Logger._instance.file_txt = open(os.path.join(dir_logs, '{}.txt'.format(name)), 'a+')
                 Logger._instance.sqlite_file = os.path.join(dir_logs, '{}.sqlite'.format(name))
-                Logger._instance.init_sqlite()
             else:
                 Logger._instance.log_message('No logs files will be created (dir_logs attribute is empty)',
                                              log_level=Logger.WARNING)
 
         return Logger._instance
 
-    def __call__(self, *args, **kwargs):
-        return self.log_message(*args, **kwargs, stack_displacement=2)
-
     def set_level(self, log_level):
         self.log_level = log_level
 
@@ -173,24 +177,24 @@ def print_subitem(prefix, subdictionary, stack_displacement=3):
         self.log_message('{}: {}'.format(group, description), log_level=log_level, stack_displacement=stack_displacement)
         print_subitem('  ', dictionary, stack_displacement=stack_displacement + 1)
 
-    def _execute(self, statement, parameters=None, commit=True):
+    def _execute(self, statement, parameters=None, commit=True, cursor=None):
         assert parameters is None or isinstance(parameters, tuple)
         parameters = parameters or ()
-        return_value = self.sqlite_cur.execute(statement, parameters)
+        return_value = cursor.execute(statement, parameters)
         if commit:
-            self.sqlite_conn.commit()
+            self.get_conn().commit()
         return return_value
 
-    def _run_query(self, query, parameters=None):
-        return self._execute(query, parameters, commit=False)
+    def _run_query(self, query, parameters=None, cursor=None):
+        return self._execute(query, parameters, commit=False, cursor=cursor)
 
     def _get_internal_table_name(self, table_name):
         return f'_{table_name}'
 
-    def _check_table_exists(self, table_name):
+    def _check_table_exists(self, table_name, cursor=None):
         table_name = self._get_internal_table_name(table_name)
         query = "SELECT name FROM sqlite_master WHERE type='table' AND name=?"
-        return self._run_query(query, (table_name,))
+        return self._run_query(query, (table_name,), cursor=cursor)
 
     def _create_table(self, table_name):
         table_name = self._get_internal_table_name(table_name)
@@ -200,15 +204,17 @@ def _create_table(self, table_name):
                 "__timestamp" DATETIME DEFAULT CURRENT_TIMESTAMP
             );
         """
-        self._execute(statement)
+        with closing(self.get_conn().cursor()) as cursor:
+            self._execute(statement, cursor=cursor)
 
     def _list_columns(self, table_name):
         table_name = self._get_internal_table_name(table_name)
         query = "SELECT name FROM PRAGMA_TABLE_INFO(?)"
-        qry_cur = self._run_query(query, (table_name,))
-        columns = (res[0] for res in qry_cur)
-        # remove __id and __timestamp columns
-        columns = [c for c in columns if not c.startswith('__')]
+        with closing(self.get_conn().cursor()) as cursor:
+            qry_cur = self._run_query(query, (table_name,), cursor=cursor)
+            columns = (res[0] for res in qry_cur)
+            # remove __id and __timestamp columns
+            columns = [c for c in columns if not c.startswith('__')]
         return columns
 
     @staticmethod
@@ -223,7 +229,8 @@ def _add_column(self, table_name, column_name, value_sample=None):
         table_name = self._get_internal_table_name(table_name)
         value_type = self._get_data_type(value_sample)
         statement = f'ALTER TABLE {table_name} ADD COLUMN "{column_name}" {value_type}'
-        return self._execute(statement)
+        with closing(self.get_conn().cursor()) as cursor:
+            return self._execute(statement, cursor=cursor)
 
     def _flatten_dict(self, dictionary, flatten_dict=None, prefix=''):
         flatten_dict = flatten_dict if flatten_dict is not None else {}
@@ -247,51 +254,58 @@ def _insert_row(self, table_name, flat_dictionary):
         value_placeholder = ', '.join(['?'] * len(columns))
         statement = f'INSERT INTO {table_name} ({column_string}) VALUES({value_placeholder})'
         parameters = tuple(val for val in flat_dictionary.values())
-        return self._execute(statement, parameters)
+        with closing(self.get_conn().cursor()) as cursor:
+            return self._execute(statement, parameters, cursor=cursor)
 
     def log_dict(self, group, dictionary, description='', stack_displacement=2, should_print=False, log_level=SUMMARY):
         if log_level < self.log_level:
             return -1
 
         flat_dictionary = self._flatten_dict(dictionary)
-        if self._check_table_exists(group).fetchone():
-            columns = self._list_columns(group)
-            for key in flat_dictionary:
-                if key not in columns:
-                    self.log_message(f'Key "{key}" is unknown. New keys are not allowed', log_level=self.ERROR)
-            for column_name in columns:
-                if column_name not in flat_dictionary:
-                    self.log_message(f'Key "{column_name}" not in the dictionary to be logged', log_level=self.ERROR)
-        else:
-            self._create_table(group)
-            for key, value in flat_dictionary.items():
-                self._add_column(group, key, value)
+        with closing(self.get_conn().cursor()) as cursor:
+            if self._check_table_exists(group, cursor=cursor).fetchone():
+                columns = self._list_columns(group)
+                for key in flat_dictionary:
+                    if key not in columns:
+                        self.log_message(f'Key "{key}" is unknown. New keys are not allowed', log_level=self.ERROR)
+                for column_name in columns:
+                    if column_name not in flat_dictionary:
+                        self.log_message(f'Key "{column_name}" not in the dictionary to be logged', log_level=self.ERROR)
+            else:
+                self._create_table(group)
+                for key, value in flat_dictionary.items():
+                    self._add_column(group, key, value)
 
         self._insert_row(group, flat_dictionary)
 
         if should_print:
             self.log_dict_message(group, dictionary, description, stack_displacement + 1, log_level)
 
-    def select(self, group, columns=None):
-        table_name = self._get_internal_table_name(group)
-        table_columns = self._list_columns(group)
+    @staticmethod
+    def select(group, columns=None):
+        logger = Logger._get_instance(dir_logs=None, name=None)
+        table_name = logger._get_internal_table_name(group)
+        table_columns = logger._list_columns(group)
         if columns is None:
             column_string = '*'
         else:
             for c in columns:
                 if c not in table_columns:
-                    self.log_message(f'Unknown column "{c}"', log_level=self.ERROR)
+                    logger.log_message(f'Unknown column "{c}"', log_level=Logger.ERROR)
             column_string = ', '.join([f'"{c}"' for c in columns])
         statement = f'SELECT {column_string} FROM {table_name}'
-        return self._execute(statement)
-
-    def init_sqlite(self):
-        pre_existing = os.path.isfile(self.sqlite_file)
-        self.sqlite_conn = sqlite3.connect(self.sqlite_file)
-        self.sqlite_cur = self.sqlite_conn.cursor()
-        if not pre_existing:
-            self._create_table('bootstrap')
+        with closing(logger.get_conn().cursor()) as cursor:
+            return logger._execute(statement, cursor=cursor, commit=False).fetchall()
+
+    def get_conn(self):
+        if self.connection is None:
+            pre_existing = os.path.isfile(self.sqlite_file)
+            connection = sqlite3.connect(self.sqlite_file, check_same_thread=False, isolation_level='IMMEDIATE')
+            self.connection = connection
+            if not pre_existing:
+                self._create_table('bootstrap')
+        return self.connection
 
     def flush(self):
         if self.dir_logs:
-            self.sqlite_conn.commit()
+            self.get_conn().commit()

bootstrap/views/factory.py

@@ -5,8 +5,6 @@
 
 
 def factory(engine=None):
-    Logger()('Creating views...')
-
     # if views does not exist, pick view
     # to support backward compatibility
     if 'views' in Options():
@@ -24,6 +22,8 @@ def factory(engine=None):
 
     exp_dir = Options()['exp.dir']
 
+    Logger(exp_dir)('Creating views...')
+
     if 'names' in opt:
         view = make_multi_views(opt, exp_dir)
         return view

bootstrap/views/plotly.py

@@ -1,5 +1,4 @@
 import os
-import json
 import math
 import plotly.graph_objects as go
 from plotly.subplots import make_subplots
@@ -49,7 +48,6 @@ def generate(self):
         for log_name in log_names:
             path_sqlite = os.path.join(self.exp_dir, '{}.sqlite'.format(log_name))
             if os.path.isfile(path_sqlite):
-                Logger._instance = None
                 data_sqlite = Logger(dir_logs=self.exp_dir, name=log_name)
                 data_dict[log_name] = data_sqlite
             else:
@@ -93,16 +91,19 @@ def generate(self):
                 group = view['view_name'].split('.')[0]
                 columns = [view['view_name'].split('.')[1]]
 
-                y = [x[0] for x in data_dict[view['log_name']].select(group, columns)]
-                x = list(range(len(y)))
-
-                scatter = go.Scatter(
-                    x=x,
-                    y=y,
-                    name=view['view_interim'],
-                    line={'color': color}
-                )
-                figure.append_trace(scatter, figure_pos_y, figure_pos_x)
+                try:
+                    y = [x[0] for x in data_dict[view['log_name']].select(group, columns)]
+                    x = list(range(len(y)))
+
+                    scatter = go.Scatter(
+                        x=x,
+                        y=y,
+                        name=view['view_interim'],
+                        line={'color': color}
+                    )
+                    figure.append_trace(scatter, figure_pos_y, figure_pos_x)
+                except Exception:
+                    pass
 
         figure['layout'].update(
             autosize=True,