Add locks for input and output caches.

IPython/core/history.py

@@ -22,6 +22,7 b' import sqlite3'
 import threading
 
 from collections import defaultdict
+from contextlib import nested
 
 # Our own packages
 from IPython.config.configurable import Configurable
@@ -127,6 +128,8 b' class HistoryManager(Configurable):'
                 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()
 
@@ -383,6 +386,7 b' class HistoryManager(Configurable):'
         self.input_hist_parsed.append(source)
         self.input_hist_raw.append(source_raw)
         
+        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:
@@ -416,6 +420,7 b' class HistoryManager(Configurable):'
             return
         output = json.dumps(self.output_hist_reprs[line_num])
         
+        with self.db_output_cache_lock:
         self.db_output_cache.append((line_num, output))
         if self.db_cache_size <= 1:
             self.save_flag.set()
@@ -436,6 +441,8 b' class HistoryManager(Configurable):'
         """Write any entries in the cache to the database."""
         if conn is None:
             conn = self.db
+            
+        with self.db_input_cache_lock:
         try:
             self._writeout_input_cache(conn)
         except sqlite3.IntegrityError:
@@ -450,6 +457,7 b' class HistoryManager(Configurable):'
         finally:
             self.db_input_cache = []
             
+        with self.db_output_cache_lock:
         try:
             self._writeout_output_cache(conn)
         except sqlite3.IntegrityError: