Merge branch 'history-threaded' of https://github.com/takluyver/ipython into takluyver-history-threaded

IPython/core/history.py

@@ -13,11 +13,13 b''
 from __future__ import print_function
 
 # Stdlib imports
+import atexit
 import datetime
 import json
 import os
 import re
 import sqlite3
+import threading
 
 from collections import defaultdict
 
@@ -76,6 +78,11 b' class HistoryManager(Configurable):'
     db_input_cache = List()
     db_output_cache = List()
     
+    # History saving in separate thread
+    save_thread = Instance('IPython.core.history.HistorySavingThread')
+    # N.B. Event is a function returning an instance of _Event.
+    save_flag = Instance(threading._Event)
+    
     # Private interface
     # Variables used to store the three last inputs from the user.  On each new
     # history update, we populate the user's namespace with these, shifted as
@@ -119,6 +126,12 b' class HistoryManager(Configurable):'
             else:
                 # The hist_file is probably :memory: or something else.
                 raise
+                
+        self.save_flag = threading.Event()
+        self.db_input_cache_lock = threading.Lock()
+        self.db_output_cache_lock = threading.Lock()
+        self.save_thread = HistorySavingThread(self)
+        self.save_thread.start()
 
         self.new_session()
 
@@ -139,10 +152,13 b' class HistoryManager(Configurable):'
                         PRIMARY KEY (session, line))""")
         self.db.commit()
     
-    def new_session(self):
+    def new_session(self, conn=None):
         """Get a new session number."""
-        with self.db:
-            cur = self.db.execute("""INSERT INTO sessions VALUES (NULL, ?, NULL,
+        if conn is None:
+            conn = self.db
+        
+        with conn:
+            cur = conn.execute("""INSERT INTO sessions VALUES (NULL, ?, NULL,
                             NULL, "") """, (datetime.datetime.now(),))
             self.session_number = cur.lastrowid
             
@@ -373,10 +389,11 b' class HistoryManager(Configurable):'
         self.input_hist_parsed.append(source)
         self.input_hist_raw.append(source_raw)
         
-        self.db_input_cache.append((line_num, source, source_raw))
-        # Trigger to flush cache and write to DB.
-        if len(self.db_input_cache) >= self.db_cache_size:
-            self.writeout_cache()
+        with self.db_input_cache_lock:
+            self.db_input_cache.append((line_num, source, source_raw))
+            # Trigger to flush cache and write to DB.
+            if len(self.db_input_cache) >= self.db_cache_size:
+                self.save_flag.set()
 
         # update the auto _i variables
         self._iii = self._ii
@@ -406,45 +423,90 b' class HistoryManager(Configurable):'
             return
         output = json.dumps(self.output_hist_reprs[line_num])
         
-        self.db_output_cache.append((line_num, output))
+        with self.db_output_cache_lock:
+            self.db_output_cache.append((line_num, output))
         if self.db_cache_size <= 1:
-            self.writeout_cache()
-        
-    def _writeout_input_cache(self):
-        with self.db:
+            self.save_flag.set()
+    
+    def _writeout_input_cache(self, conn):
+        with conn:
             for line in self.db_input_cache:
-                self.db.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
+                conn.execute("INSERT INTO history VALUES (?, ?, ?, ?)",
                                 (self.session_number,)+line)
     
-    def _writeout_output_cache(self):
-        with self.db:
+    def _writeout_output_cache(self, conn):
+        with conn:
             for line in self.db_output_cache:
-                self.db.execute("INSERT INTO output_history VALUES (?, ?, ?)",
+                conn.execute("INSERT INTO output_history VALUES (?, ?, ?)",
                                 (self.session_number,)+line)
     
-    def writeout_cache(self):
+    def writeout_cache(self, conn=None):
         """Write any entries in the cache to the database."""
-        try:
-            self._writeout_input_cache()
-        except sqlite3.IntegrityError:
-            self.new_session()
-            print("ERROR! Session/line number was not unique in",
-                  "database. History logging moved to new session",
-                                            self.session_number)
-            try: # Try writing to the new session. If this fails, don't recurse
-                self._writeout_input_cache()
-            except sqlite3.IntegrityError:
-                pass
-        finally:
-            self.db_input_cache = []
+        if conn is None:
+            conn = self.db
             
+        with self.db_input_cache_lock:
+            try:
+                self._writeout_input_cache(conn)
+            except sqlite3.IntegrityError:
+                self.new_session(conn)
+                print("ERROR! Session/line number was not unique in",
+                      "database. History logging moved to new session",
+                                                self.session_number)
+                try: # Try writing to the new session. If this fails, don't recurse
+                    self._writeout_input_cache(conn)
+                except sqlite3.IntegrityError:
+                    pass
+            finally:
+                self.db_input_cache = []
+
+        with self.db_output_cache_lock:
+            try:
+                self._writeout_output_cache(conn)
+            except sqlite3.IntegrityError:
+                print("!! Session/line number for output was not unique",
+                      "in database. Output will not be stored.")
+            finally:
+                self.db_output_cache = []
+
+
+class HistorySavingThread(threading.Thread):
+    """This thread takes care of writing history to the database, so that
+    the UI isn't held up while that happens.
+    
+    It waits for the HistoryManager's save_flag to be set, then writes out
+    the history cache. The main thread is responsible for setting the flag when
+    the cache size reaches a defined threshold."""
+    daemon = True
+    stop_now = False
+    def __init__(self, history_manager):
+        super(HistorySavingThread, self).__init__()
+        self.history_manager = history_manager
+        atexit.register(self.stop)
+        
+    def run(self):
+        # We need a separate db connection per thread:
         try:
-            self._writeout_output_cache()
-        except sqlite3.IntegrityError:
-            print("!! Session/line number for output was not unique",
-                  "in database. Output will not be stored.")
-        finally:
-            self.db_output_cache = []
+            self.db = sqlite3.connect(self.history_manager.hist_file)
+            while True:
+                self.history_manager.save_flag.wait()
+                if self.stop_now:
+                    return
+                self.history_manager.save_flag.clear()
+                self.history_manager.writeout_cache(self.db)
+        except Exception as e:
+            print(("The history saving thread hit an unexpected error (%s)."
+                   "History will not be written to the database.") % repr(e))
+        
+    def stop(self):
+        """This can be called from the main thread to safely stop this thread.
+        
+        Note that it does not attempt to write out remaining history before
+        exiting. That should be done by calling the HistoryManager's
+        end_session method."""
+        self.stop_now = True
+        self.history_manager.save_flag.set()
+        self.join()
 
         
 # To match, e.g. ~5/8-~2/3

IPython/core/tests/test_history.py

@@ -23,10 +23,10 b' def setUp():'
 def test_history():
     ip = get_ipython()
     with TemporaryDirectory() as tmpdir:
-        # Make a new :memory: DB.
         hist_manager_ori = ip.history_manager
+        hist_file = os.path.join(tmpdir, 'history.sqlite')
         try:
-            ip.history_manager = HistoryManager(shell=ip, hist_file=':memory:')
+            ip.history_manager = HistoryManager(shell=ip, hist_file=hist_file)
             hist = ['a=1', 'def f():\n    test = 1\n    return test', u"b='€Æ¾÷ß'"]
             for i, h in enumerate(hist, start=1):
                 ip.history_manager.store_inputs(i, h)
@@ -38,10 +38,7 b' def test_history():'
             
             nt.assert_equal(ip.history_manager.input_hist_raw, [''] + hist)
             
-            # Check lines were written to DB
-            c = ip.history_manager.db.execute("SELECT source_raw FROM history")
-            nt.assert_equal([x for x, in c], hist)
-              
+            
             # New session
             ip.history_manager.reset()
             newcmds = ["z=5","class X(object):\n    pass", "k='p'"]
@@ -83,6 +80,7 b' def test_history():'
             # Duplicate line numbers - check that it doesn't crash, and
             # gets a new session
             ip.history_manager.store_inputs(1, "rogue")
+            ip.history_manager.writeout_cache()
             nt.assert_equal(ip.history_manager.session_number, 3)
         finally:
             # Restore history manager

IPython/testing/tools.py

@@ -167,7 +167,8 b' def default_config():'
     config.TerminalInteractiveShell.colors = 'NoColor'
     config.TerminalTerminalInteractiveShell.term_title = False,
     config.TerminalInteractiveShell.autocall = 0
-    config.HistoryManager.hist_file = u':memory:'
+    config.HistoryManager.hist_file = u'test_hist.sqlite'
+    config.HistoryManager.db_cache_size = 10000
     return config