The goal is to run a function compute in parallel on many inputs (10**6 in total, say) and store the results.
Each call to compute takes a few seconds and is independent of other computes.

Since running it on all inputs takes too long (even on many cores), I want to be able to start computing, interrupt it, and continue later without redoing any computations.

I've used this for the WAL setting and this to give each process a db connection.

I don't have a great understanding of what's going on under the hood in multiprocessing or sqlite3 (let alone their combination with WAL) - I'd be curious to get some intuition what might become a problem at different scales. E.g. when the number of parallel processes grows, do they have to wait longer and longer for a lock to become available, or can they submit their db transaction and move to the next computation?

EDIT: there seems to be a problem with the approach below: when pool.map finishes, it leaves behind a results.db-wal file with results that are not written into results.db. I had to insert a manual WAL checkpoint (connection.execute(f"PRAGMA wal_checkpoint(PASSIVE);") after pool.map to get all results written in the db.

import sqlite3
import multiprocessing

def compute(x):
    return 2*x

db = 'results.db'
reset_db = True  # set to False to continue earlier run

def reset(file):
    con = sqlite3.connect(file)
        con.execute("DROP TABLE results")
    except sqlite3.OperationalError:
    con.execute("CREATE TABLE results(input UNIQUE, output)")

def init_worker(function):
    function.con = sqlite3.connect(db)
    function.con.execute("PRAGMA journal_mode=WAL")

def already_present(con, param):
    return bool(con.execute("SELECT 1 FROM results WHERE input=?", (param, )).fetchone())

def write_result(con, param, result):
    with con:
        con.execute("INSERT INTO results VALUES (?, ?)", (param, results,))

def check_compute_and_write(param):
    if already_present(check_compute_and_write.con, param):
    result = compute(param)
    write_result(check_compute_and_write.con, param, result)

if reset_db:

with multiprocessing.Pool(initializer=init_worker, initargs=(check_compute_and_write,)) as pool:
    pool.map(check_compute_and_write, range(10**6))
  • \$\begingroup\$ Consider having your N workers append results to N .csv text files, or to a unified lockf()-controlled text file. And then periodically a single writer will gather and INSERT the results. A giant advantage of per-worker log files is it lets you scale out to multiple compute hosts. On mysql / postgres databases I have certainly seen long running SELECT transactions hold reader locks which starve writers, e.g. ALTER TABLE attempts. A simple COMMIT releases the lock. With sqlite, I think you need to disconnect / reconnect to release? Else the WAL can become much bigger than 4 MiB. cf: reader gaps \$\endgroup\$
    – J_H
    Feb 25 at 18:10

1 Answer 1


General Approach

From what I have personally experienced and from what I have read on the Internet, doing concurrent writes to a SQLite database either from multiple threads or multiple processes is problematic and can result in "Database is locked" errors. It seems that your bottleneck is the compute function that takes a few seconds to run. So I would compute all the values using multiprocessing in such a fashion that I can do my insert operations as results are computed and returned. The multiprocessing.pool.map method will not return until all results have been computed and only then can you start doing your insert operations, which are rather slow using SQLite. Therefore, I suggest using the multiprocessing.pool.imap method that will return results as they become available. Unlike the map method, imap will use a chunksize value of 1 if no value for this is passed. For large iterables, this is not efficient. I have included code below that can be used to compute a "suitable" chunksize. Read the comments in the code.

Use a Context Manager Consistently

In function write_result you are using the connection as a context manager, which is a good practice as it will result in a commit being performed automatically if no exception is raised in the block or a rollback being performed otherwise. But note that the connection context manager as implemented in SQLite does not close the connection upon termination of the block. That is as it should be because you might have subsequent CRUD operations you wish to perform and it does not make sense to close the connection only to have to recreate one.

Still, there are times where it would be desirable for the connection to be automatically closed when the context manager block exits, either normally or with an exception. Such a case is your reset function, which is currently not using any context manager. We can implement our own context manager that creates and closes a connection automatically:

from contextlib import contextmanager

def get_connection(db):
        conn = sqlite3.connect(db)
        yield conn
        conn.commit() # if no execption
    except Exception:
        conn.rollback() # if we have an exception
        conn.close() # always close the connection

Now function reset becomes:

def reset(file):
    with get_connection(file) as con:
        con.execute("DROP TABLE IF EXISTS results")
        con.execute("CREATE TABLE results(input UNIQUE, output)")

Note that I have added the IF EXISTS clause to the DROP TABLE command so that it will not raise an exception if the table does not exist.

Command Line Arguments

I would suggest that to make running this script more user friendly (so it does not have to be modified for each run) you modify it to take command line arguments that will determine its behavior. For example:

python my_program.py --create

This would result in initializing an empty database by calling the reset function. Or:

python my_program.py --run 1000 1999

This would call the compute function for the values in the range 1000 - 1999.

Other Observations

If you start with an empty database and from that point do runs specifying ranges that do not overlap, there is no reason why there should be attempts to insert duplicate rows. Therefore, the test you make in already_present is just a waste of time. I would prefer to modify write_result to catch a potential sqlite3.IntegrityError exception that would occur if you ever did try to insert a row that already existed. But better yet might be to do an initial query to the count of rows that already exist where the input column contains a value in the range of the passed arguments. If that count is not 0, clearly there is a user error in specifying a non-overlapping range. You might then wish to support the following command:

python my_program.py --query

This would print the max(input) value from the database. Also:

  1. Your code as published would not run because in function write_result variable results was undefined.
  2. My preference is to name constants such as db in upper case (DB).
  3. You have db defined at global scope and you use this value to pass to function reset. But in other functions the global db is accessed directly. This is inconsistent. Decide whether all functions will use a global definition of db or you will pass db to the functions that need the database name. If you go the latter route, then remove db from global scope.
  4. You wanted each process to have its own connection so using the initializer argument to the Pool initializer is appropriate. But instead of just assigning the passed connection to a global con variable, you made it an attribute of a function (which is also at global scope so it does work). I find this a confusing choice.

Example of How I would Do the Multiprocessing

In the code below I have not incorporated the command line processing I have suggested; I leave that as an exercise for you if you are interested in doing this:

import sqlite3
import multiprocessing
from contextlib import contextmanager

# Use upper case for "constants":
RESET_DB = True  # set to False to continue earlier run
DB = 'results.db'

def get_connection():
        conn = sqlite3.connect(DB)
        yield conn
        conn.commit() # if no execption
    except Exception:
        conn.rollback() # if we have an exception
        conn.close() # always close the connecti

def compute(x):
    return 2 * x

def reset():
    with get_connection() as con:
        con.execute("DROP TABLE IF EXISTS results")
        con.execute("CREATE TABLE results(input UNIQUE, output)")

def write_result(con, param, result):
    # This is using variable con as the standard SQLite context manager,
    # so a commit will be done (but no close) on the connection:
    with con:
            con.execute("INSERT INTO results VALUES (?, ?)", (param, result)) #  error
        except sqlite3.IntegrityError as e:
            # Row already exists:

def main():
    def compute_chunksize(iterable_size, pool_size):
        """This calculates "suitable" chunksize value much in the same way
        that the map function calculates it if you specify chunksize=None."""

        chunksize, remainder = divmod(iterable_size, 4 * pool_size)
        if remainder:
            chunksize += 1
        return chunksize

    if RESET_DB:

    range_start = 0
    range_end = 999
    iterable = range(range_start, range_end + 1)
    # Number of tasks to be submitted:
    iterable_size = range_end - range_start + 1

    pool_size = multiprocessing.cpu_count()
    # Compute a chunksize
    chunksize = compute_chunksize(iterable_size, pool_size)

    with get_connection() as con, \
    multiprocessing.Pool(pool_size) as pool:
        for param, result in enumerate(
            pool.imap(compute, iterable, chunksize=chunksize),
            write_result(con, param, result)

# For cross-platform compatibility:
if __name__ == '__main__':
  • \$\begingroup\$ @ga325, this seems like helpful and useful advice. I recommend you adopt it. Personally I tend to access sqlite via sqlalchemy, but meh, whatever, there are many paths to success. On items (3) & (4), I will often find that sufficient motivation to put my functions within a class, just so self.db and self.con are readily available to all. You might prefer to grab / close() the con within the for loop so you hold a lock just briefly and immediately release it. Compare and see! Let us know how it goes. \$\endgroup\$
    – J_H
    Jul 2 at 0:58
  • \$\begingroup\$ This sounds good, but I would like to see a context manager for the SQLite connection too. Alternatively, a finally clause could be used since we already have try blocks. \$\endgroup\$
    – Kate
    Jul 2 at 13:21
  • \$\begingroup\$ @Kate I have updated the answer to show how we can create a custom context manager that will do what the standard connection context manager does but also automatically close the connection when the block exists. See the section Use a Context Manager Consistently. \$\endgroup\$
    – Booboo
    Jul 2 at 14:45

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.