BUG: GH11786 Thread safety issue with read_csv

[jdeschenes]

closes #11786

Fixed an issue with thread safety when calling read_csv with a StringIO object.

The issue was caused by a misplaced PyGilSate_Ensure()

[jreback]

@jdeschenes gr8! can you add in the example from the issue as a smoke test. (e.g. just have it run), then read in with a single trhead and compare.

and pls add a release note when you are satisified.

[jdeschenes]

Alright, I did this quickly as I don't have time to work on this right now. How long until next release?

[jreback]

@jdeschenes oh have a while.....when you have a chance...thanks!

[jreback]

@jdeschenes if you can have a look at this again would be great.

[jdeschenes]

ping! @jreback

[jreback]

assert that the read in values match a single threaded reader. (e.g. compare frames)

[jreback]

is this restricted to only a BytesIO? I think we need a test of this using actual files. e.g. create a frame, write to out, then read it from the files.

[jdeschenes]

It only applies to BytesIO, the function is called only when you are not passing a path. It requires a 'file-like' object.

[jreback]

move this to pandas/io/tests/test_parsers.py.

[jreback]

this is a bug in read_csv when reading using multiple threads (not restricted to StringIO)

[jdeschenes]

It only applies if you pass in an object that has a 'read' attribute(BytesIO)

[mrocklin]

Thank you both for keeping up on this.

[jreback]

@jdeschenes IIRC this issue is repro with actual files. Is that not the case? is it only StringIO/BytesIO. are they not thread-safe?

[jreback]

@jdeschenes can you respond to my comments.

[jdeschenes]

Hi @jreback,

the issue is solely reproducible with StringIO. The root cause of this bug is in function buffer_rd_bytes in
pandas/src/parser/io.c. This function is only used when a StringIO/BytesIO is passed to the read_csv function.

The function was calling Py_XDECREF before ensuring that the thread had the GIL. This behavior could not be seen before since the GIL was always locked throughout the read_csv function call.

I am not aware of any issues when reading from disk and this pull request will not fix any problem related to this.

I think that the release notes should be kept as is.

Let me know what you think.

[jreback]

ok, can you add a test that validates the issue that reading from a disk with multiple threads is ok (so we don't regress).

[jreback]

use double backticks around StringIO

[jreback]

ahh I see you already added this. ok then.

[jreback]

actually, can you add a test where you read from a single file with multiple readers? similar to the examples in the original issue.

[jdeschenes]

Hmmm, is this ok? As per your comment above?

[jreback]

sort of. I think this functionailty will be used to read a single file in parts (not the same file several times)

something like this:

import pandas as pd
from pandas.util import testing as tm
import numpy as np
from multiprocessing.pool import ThreadPool
pd.options.display.width=200
fn = 'test.csv'
np.random.seed(1234)

N = 1000000
df = pd.DataFrame(np.random.randn(N,5),columns=list('abcde'))
df['foo'] = 'foo'
df['bar'] = 'bar'
df['baz'] = 'baz'
df['date'] = pd.date_range('20000101 09:00:00',periods=N,freq='s')
df['int'] = np.arange(N)

df.to_csv(fn)

def f():
    return pd.read_csv(fn, index_col=0, header=0)

def g(procs=4):

    def reader(arg):
        start, nrows = arg

        # read from the start line for nrows
        if not start:
            return pd.read_csv(fn, index_col=0, header=0, nrows=nrows)

        # skip the header entirely
        return pd.read_csv(fn, index_col=0, header=None, skiprows=int(start)+1,nrows=nrows)

    # our tasks
    tasks = [ (N*i/procs, N/procs) for i in range(procs) ]
    pool = ThreadPool(processes=procs)

    # execute
    results = pool.map(reader, tasks)

    # collate the results
    # use the header from the first piece
    header = results[0].columns
    for r in results[1:]:
        r.columns = header
    return pd.concat(results)

def compare(result, expected):
    tm.assert_frame_equal(result, expected)

[jreback]

note AFAICT this DOES work!. I just want to have this as a test for validation.

Like to test this with actual files and StringIO

[jreback]

pls add the issue number as a comment (on all tests)

[jreback]

pls run git diff master | flake8 --diff as much PEP checking has been one on these files.

[mrocklin]

FWIW using BytesIO has actual use cases in distributed computing, it isn't just a test case.

Many parallel storage systems won't give you access to the hard disk but will instead deliver a bunch of bytes. In this case the best way I've found to use pd.read_csv is to hand it a BytesIO object.

[jreback]

@mrocklin oh of course. just covering the bases. I suspect people have tried multi-threading to read files as well :)

[jdeschenes]

It would be very interesting to see if there is any benefit in using a ThreadPool for reading from a BytesIO. We are spending a lot of time into the GIL, thanks to the buffer_rd_bytes function. It should probably be benchmarked.

I have a suspicion that it doesn't help at all(It might be even a net loss).

I added the test for the file read. I didn't do it for the BytesIO. The code would effectively look a lot like what I did up top... Grabbing a list of BytesIO and processing them in a ThreadPool. I can take a look at this a bit later, if that is required.

[jreback]

@jdeschenes thanks!

certainly would take addtl benchmarks / fixes!

[kayvonr]

Hey all - any estimate of when this will be go out in a production release? Encountering this bug very very frequently with 0.17.1, and would like to get back up to a newer version of pandas again soon

Thanks

[jreback]

planning on a RC in about 2 weeks, so release should be roughly mid-feb or so