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
@contextmanager
def get_connection(db):
try:
conn = sqlite3.connect(db)
yield conn
conn.commit() # if no execption
except Exception:
conn.rollback() # if we have an exception
raise
finally:
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:
- Your code as published would not run because in function
write_result variable results was undefined.
- My preference is to name constants such as
db in upper case (DB).
- 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.
- 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'
@contextmanager
def get_connection():
try:
conn = sqlite3.connect(DB)
yield conn
conn.commit() # if no execption
except Exception:
conn.rollback() # if we have an exception
raise
finally:
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:
try:
con.execute("INSERT INTO results VALUES (?, ?)", (param, result)) # error
except sqlite3.IntegrityError as e:
# Row already exists:
pass
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:
reset()
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),
start=range_start
):
write_result(con, param, result)
# For cross-platform compatibility:
if __name__ == '__main__':
main()
