diff --git a/smiley/db.py b/smiley/db.py index bb69488..8dd638e 100644 --- a/smiley/db.py +++ b/smiley/db.py @@ -119,9 +119,19 @@ class DB(processor.EventProcessor): """Database connection and API. """ + # Getting a new cursor and committing for every insert is very slow. + # Instead, commits will only occur after this many statements have run. + # If you need to ensure that all transactions have been committed, + # call flush() + SQL_STATEMENT_BUFFER_SIZE = 20 + def __init__(self, name): + self._uncommitted_statement_count = 0 self._name = name self.conn = self._open_db(name) + # Cursor used by _insert for all insert/update/delete segments. + self._cursor = self.conn.cursor() + # TODO force a flush on sys.atexit()? Is that too dirty? return @staticmethod @@ -142,45 +152,68 @@ def _open_db(filename): cursor.executescript(schema) return conn + def flush(self): + """Ensure that all changes to the db have been committed""" + if self._uncommitted_statement_count > 0: + self.conn.commit() + self._uncommitted_statement_count = 0 + self._cursor = self.conn.cursor() + + def _insert(self, *args, **kwargs): + """Execute a sql query that modifies the database + + Should be used for all insert, update, and delete stmts + All arguments are passed to self._cursor.execute()""" + self._cursor.execute(*args, **kwargs) + self._uncommitted_statement_count += 1 + if self._uncommitted_statement_count >= self.SQL_STATEMENT_BUFFER_SIZE: + self.flush() + + def get_cursor(self): + """Get a sqlite cursor to run select statements on""" + # Flush before querying to ensure that we get the latest results + self.flush() + return self.conn.cursor() + def start_run(self, run_id, cwd, description, start_time): "Record the beginning of a run." # LOG.debug('start_run(%s)', run_id) - with transaction(self.conn) as c: - try: - c.execute( - u""" - INSERT INTO run (id, cwd, description, start_time) - VALUES (:id, :cwd, :description, :start_time) - """, - {'id': run_id, - 'cwd': cwd, - 'description': jsonutil.dumps(description), - 'start_time': start_time} - ) - except sqlite3.IntegrityError: - raise ValueError('There is already a run with id %s in %s' % ( - run_id, self._name)) - - def end_run(self, run_id, end_time, message, traceback, stats): - "Record the end of a run." - # LOG.debug('end_run(%s)', run_id) - with transaction(self.conn) as c: - c.execute( + try: + self._insert( u""" - UPDATE run - SET - end_time = :end_time, - error_message = :message, - traceback = :traceback, - stats = :stats - WHERE id = :id + INSERT INTO run (id, cwd, description, start_time) + VALUES (:id, :cwd, :description, :start_time) """, {'id': run_id, - 'end_time': end_time, - 'message': message, - 'traceback': jsonutil.dumps(traceback), - 'stats': stats or None} + 'cwd': cwd, + 'description': jsonutil.dumps(description), + 'start_time': start_time} ) + except sqlite3.IntegrityError: + raise ValueError('There is already a run with id %s in %s' % ( + run_id, self._name)) + + def end_run(self, run_id, end_time, message, traceback, stats): + "Record the end of a run." + # LOG.debug('end_run(%s)', run_id) + self._insert( + u""" + UPDATE run + SET + end_time = :end_time, + error_message = :message, + traceback = :traceback, + stats = :stats + WHERE id = :id + """, + {'id': run_id, + 'end_time': end_time, + 'message': message, + 'traceback': jsonutil.dumps(traceback), + 'stats': stats or None} + ) + # Always flush when a run is completed; the program's likely exiting. + self.flush() def get_runs(self, only_errors=False, sort_order='ASC'): "Return the run data." @@ -188,42 +221,42 @@ def get_runs(self, only_errors=False, sort_order='ASC'): if only_errors: query.append(u"WHERE error_message is not null") query.append(u"ORDER BY start_time %s" % sort_order) - with transaction(self.conn) as c: - c.execute(u' '.join(query)) - return (_make_run(r) for r in c.fetchall()) + c = self.get_cursor() + c.execute(u' '.join(query)) + return (_make_run(r) for r in c.fetchall()) def get_run(self, run_id): "Return the run data." - with transaction(self.conn) as c: - c.execute( - u"SELECT * FROM run WHERE id = :run_id", - {'run_id': run_id}, - ) - row = c.fetchone() - if row is None: - raise NoSuchRun(run_id) - return _make_run(row) + c = self.get_cursor() + c.execute( + u"SELECT * FROM run WHERE id = :run_id", + {'run_id': run_id}, + ) + row = c.fetchone() + if row is None: + raise NoSuchRun(run_id) + return _make_run(row) def get_thread_details(self, run_id): "Return the names of the threads used in the run." - with transaction(self.conn) as c: - c.execute( - u""" - SELECT trace.run_id, trace.thread_id, - MIN(timestamp) AS start_time, - MAX(timestamp) AS end_time, - COUNT(trace.id) AS num_events, - location_counts.num_locations AS num_locations - FROM trace - JOIN location_counts - ON trace.run_id = location_counts.run_id - AND trace.thread_id = location_counts.thread_id - WHERE trace.run_id = :run_id - GROUP BY trace.thread_id - """, - {'run_id': run_id}, - ) - return (_make_thread(r) for r in c.fetchall()) + c = self.get_cursor() + c.execute( + u""" + SELECT trace.run_id, trace.thread_id, + MIN(timestamp) AS start_time, + MAX(timestamp) AS end_time, + COUNT(trace.id) AS num_events, + location_counts.num_locations AS num_locations + FROM trace + JOIN location_counts + ON trace.run_id = location_counts.run_id + AND trace.thread_id = location_counts.thread_id + WHERE trace.run_id = :run_id + GROUP BY trace.thread_id + """, + {'run_id': run_id}, + ) + return (_make_thread(r) for r in c.fetchall()) def trace(self, run_id, thread_id, call_id, event, func_name, line_no, filename, @@ -231,68 +264,66 @@ def trace(self, run_id, thread_id, call_id, event, timestamp): "Record an event during a run." # LOG.debug('trace(filename=%s)', filename) - with transaction(self.conn) as c: + self._insert( + u""" + INSERT INTO trace + (run_id, thread_id, call_id, event, + func_name, line_no, filename, + trace_arg, local_vars, + timestamp) + VALUES + (:run_id, :thread_id, :call_id, :event, + :func_name, :line_no, :filename, + :trace_arg, :local_vars, + :timestamp) + """, + {'run_id': run_id, + 'thread_id': thread_id, + 'call_id': call_id, + 'event': event, + 'func_name': func_name, + 'line_no': line_no, + 'filename': filename, + 'trace_arg': jsonutil.dumps(trace_arg), + 'local_vars': jsonutil.dumps(local_vars), + 'timestamp': timestamp, + } + ) + + def get_trace(self, run_id, thread_id=None): + "Return the run data." + c = self.get_cursor() + if thread_id: c.execute( u""" - INSERT INTO trace - (run_id, thread_id, call_id, event, - func_name, line_no, filename, - trace_arg, local_vars, - timestamp) - VALUES - (:run_id, :thread_id, :call_id, :event, - :func_name, :line_no, :filename, - :trace_arg, :local_vars, - :timestamp) + SELECT * + FROM trace + WHERE run_id = :run_id + AND thread_id = :thread_id + ORDER BY id """, - {'run_id': run_id, - 'thread_id': thread_id, - 'call_id': call_id, - 'event': event, - 'func_name': func_name, - 'line_no': line_no, - 'filename': filename, - 'trace_arg': jsonutil.dumps(trace_arg), - 'local_vars': jsonutil.dumps(local_vars), - 'timestamp': timestamp, - } + {'run_id': run_id, 'thread_id': thread_id}, ) - - def get_trace(self, run_id, thread_id=None): - "Return the run data." - with transaction(self.conn) as c: - if thread_id: - c.execute( - u""" - SELECT * - FROM trace - WHERE run_id = :run_id - AND thread_id = :thread_id - ORDER BY id - """, - {'run_id': run_id, 'thread_id': thread_id}, - ) - else: - c.execute( - u"SELECT * FROM trace WHERE run_id = :run_id ORDER BY id", - {'run_id': run_id}, - ) - return (_make_trace(t) - for t in c.fetchall()) + else: + c.execute( + u"SELECT * FROM trace WHERE run_id = :run_id ORDER BY id", + {'run_id': run_id}, + ) + return (_make_trace(t) + for t in c.fetchall()) def delete_run(self, run_id): """Remove a run and all of its trace events from the database""" # Ensure that the run exists. This will raise NoSuchRun if it doesn't. self.get_run(run_id) - with transaction(self.conn) as c: - c.execute( - u""" DELETE FROM trace WHERE run_id = :run_id""", - {"run_id": run_id} - ) - c.execute( - u"""DELETE FROM run WHERE id = :run_id""", - {"run_id": run_id} - ) + self._insert( + u""" DELETE FROM trace WHERE run_id = :run_id""", + {"run_id": run_id} + ) + self._insert( + u"""DELETE FROM run WHERE id = :run_id""", + {"run_id": run_id} + ) def cache_file_for_run(self, run_id, filename, body): signature_maker = hashlib.sha1() @@ -305,104 +336,104 @@ def cache_file_for_run(self, run_id, filename, body): else: signature_maker.update(body) signature = signature_maker.hexdigest() - with transaction(self.conn) as c: - try: - c.execute( - u""" - INSERT INTO file (signature, name, body) - VALUES (:signature, :filename, :body) - """, - {'signature': signature, - 'filename': filename, - 'body': body, - }, - ) - except sqlite3.IntegrityError: - pass - try: - c.execute( - u""" - INSERT INTO run_file - (run_id, signature) - VALUES (:run_id, :signature) - """, - {'run_id': run_id, - 'signature': signature, - }, - ) - except sqlite3.IntegrityError: - pass - return signature - - def get_file_signature(self, run_id, filename): - """Return the file signature for the named file within the run. - """ - # LOG.debug('get_file_signature(%s)', filename) - with transaction(self.conn) as c: - c.execute( + try: + self._insert( u""" - SELECT signature - FROM file JOIN run_file USING (signature) - WHERE - name = :filename - AND - run_id = :run_id + INSERT INTO file (signature, name, body) + VALUES (:signature, :filename, :body) """, - {'filename': filename, - 'run_id': run_id, + {'signature': signature, + 'filename': filename, + 'body': body, }, ) - row = c.fetchone() - # LOG.debug(' -> %s', row) - return row['signature'] if row else '' - - def get_files_for_run(self, run_id): - with transaction(self.conn) as c: - c.execute( + except sqlite3.IntegrityError: + pass + try: + # TODO only run this insert if the above one succeeds? + self._insert( u""" - SELECT name, signature, run_id - FROM file JOIN run_file USING (signature) - WHERE - run_id = :run_id - ORDER BY name ASC + INSERT INTO run_file + (run_id, signature) + VALUES (:run_id, :signature) """, {'run_id': run_id, + 'signature': signature, }, ) - return (_make_file(row) for row in c.fetchall()) + except sqlite3.IntegrityError: + pass + return signature + + def get_file_signature(self, run_id, filename): + """Return the file signature for the named file within the run. + """ + # LOG.debug('get_file_signature(%s)', filename) + c = self.get_cursor() + c.execute( + u""" + SELECT signature + FROM file JOIN run_file USING (signature) + WHERE + name = :filename + AND + run_id = :run_id + """, + {'filename': filename, + 'run_id': run_id, + }, + ) + row = c.fetchone() + # LOG.debug(' -> %s', row) + return row['signature'] if row else '' + + def get_files_for_run(self, run_id): + c = self.get_cursor() + c.execute( + u""" + SELECT name, signature, run_id + FROM file JOIN run_file USING (signature) + WHERE + run_id = :run_id + ORDER BY name ASC + """, + {'run_id': run_id, + }, + ) + return (_make_file(row) for row in c.fetchall()) def get_cached_file(self, run_id, filename): - with transaction(self.conn) as c: - c.execute( - u""" - SELECT body - FROM file JOIN run_file USING (signature) - WHERE - name = :filename - AND - run_id = :run_id - """, - {'filename': filename, - 'run_id': run_id, - }, - ) - row = c.fetchone() - return row['body'] if row else '' + c = self.get_cursor() + c.execute( + u""" + SELECT body + FROM file JOIN run_file USING (signature) + WHERE + name = :filename + AND + run_id = :run_id + """, + {'filename': filename, + 'run_id': run_id, + }, + ) + row = c.fetchone() + return row['body'] if row else '' def get_cached_file_by_id(self, run_id, file_id): - with transaction(self.conn) as c: - c.execute( - u""" - SELECT name, body - FROM file JOIN run_file USING (signature) - WHERE - signature = :signature - AND - run_id = :run_id - """, - {'signature': file_id, - 'run_id': run_id, - }, - ) - row = c.fetchone() - return (row['name'], row['body']) if row else ('', '') + c = self.get_cursor() + c.execute( + u""" + SELECT name, body + FROM file JOIN run_file USING (signature) + WHERE + signature = :signature + AND + run_id = :run_id + """, + {'signature': file_id, + 'run_id': run_id, + }, + ) + row = c.fetchone() + return (row['name'], row['body']) if row else ('', '') diff --git a/test_app/test.py b/test_app/test.py index 1d36e7c..6308517 100644 --- a/test_app/test.py +++ b/test_app/test.py @@ -17,6 +17,8 @@ for t in threads: print('Waiting for', t.name) t.join() + elif '-m' in sys.argv: + test_funcs.run_many_functions(10, 4) else: test_funcs.a() if '-e' in sys.argv: diff --git a/test_app/test_funcs.py b/test_app/test_funcs.py index 27f7339..9640577 100644 --- a/test_app/test_funcs.py +++ b/test_app/test_funcs.py @@ -50,6 +50,13 @@ def large_data_structure(): } big_data['key2'] = 'key' +def run_many_functions(branch_factor, depth): + """A function that takes a long time to run (O(branch_factor^depth)) + + Useful for testing performance""" + if depth > 1: + for _ in range(branch_factor): + run_many_functions(branch_factor, depth - 1) def a(): print('args:', sys.argv)