Iterator data source

[syleshfb]

What does this PR do? Please describe:
Introduces a read_iterator data pipeline factory that uses an Iterator as a data source.

Check list:

• Was the content of this PR discussed and approved via a GitHub issue? (no need for typos or documentation improvements)
• Did you read the contributor guideline?
• Did you make sure that your PR does only one thing instead of bundling different changes together?
• Did you make sure to update the documentation with your changes? (if necessary)
• Did you write any new necessary tests?
• Did you verify new and existing tests pass locally with your changes?
• Did you update the CHANGELOG? (no need for typos, documentation, or minor internal changes)

[cbalioglu]

Thanks again for this PR! Overall looks nice. I think the way how we handle resetting a Python iterator needs a different approach though. I left my feedback in iterator_data_source::reset. Let me know what you think.

[cbalioglu]

Nice, I liked how you deal with GIL here.

[cbalioglu]

Please remove these two blank lines.

[cbalioglu]

Nit: For readability I would prefer to have the py:module::import()::attr() as a separate statement. Same in reload_position.

[cbalioglu]

I suggest adding a brief inline comment here about when reloaded_ becomes true and why we need it. It only becomes clear once you read reload_position(). Mentioning it here briefly would give the reader a clue beforehand.

[cbalioglu]

Can you think of a way to move this statement to next() (e.g. by checking a "first-time-read" flag)? Since calling a Python iterator might be expensive or might raise an exception, it is unsafe to have this in reset(). Resetting should ideally not raise any exception and should be lightweight and finish in constant time (which we cannot guarantee here).

[cbalioglu]

I think the whole reset approach to the Python iterator needs a bit rethinking. There are many cases where the user has no way to reset a Python iterator. Think for instance that they pass a range(1234) or enumerate(other_iter) to read_iterator(), there is no formal API in Python to "reset" such iterators. I think the safer approach would be for reset_fn (or whatever we call it) to return a new Python iterator. It will give the user the flexibility to return a new iterator (like in range, enumerate, or itertools iterators), or if they have a custom reset API, just call it, and return the same instance.

[cbalioglu]

Really solid, thanks a lot again! The reset functionality looks much nicer now. LGTM!

[artemru]

I'll need to read it in more details but I think this PR closes #353 !

[artemru]

I think we need to add into the doc string explicitly that iterator should be picklable (and correctly) since the pipeline state will be serialized using pickle (which can make weird thing sometimes).
I wonder if we do want to add some custom function for state_load/state_save along with reset_fn.
One different approach that could make sense in some case (where the elements production is fast but serialization can be complicated) to restore the state is to do reset + skip the number of element that were already produced.

[syleshfb]

I think we need to add into the doc string explicitly that iterator should be picklable (and correctly) since the pipeline state will be serialized using pickle (which can make weird thing sometimes).

#705

I wonder if we do want to add some custom function for state_load/state_save along with reset_fn.
One different approach that could make sense in some case (where the elements production is fast but serialization can be complicated) to restore the state is to do reset + skip the number of element that were already produced.

I believe the user can already do this through __getstate__ and __setstate__ methods in the iterator class. For example:

from fairseq2.data import read_iterator
from typing import Iterator

class A(Iterator):
    def __init__(self):
        self.counter = 0
        self.generator = self.make_generator()
    def make_generator(self):
        for i in range(100):
            yield i
    def __iter__(self):
        return self
    def __next__(self):
        self.counter += 1
        return next(self.generator)
    def __getstate__(self):
        return self.counter
    def __setstate__(self, counter):
        self.generator = self.make_generator()
        for i in range(counter):
            next(self.generator)
        self.counter = counter

pipeline = read_iterator(A(), reset_fn = lambda x : A(), infinite=False).and_return()
it = iter(pipeline)
print(next(it), next(it), next(it)) # 0 1 2 
d = pipeline.state_dict()
print(next(it), next(it), next(it)) # 3 4 5
pipeline.load_state_dict(d)
print(next(it), next(it), next(it)) # 3 4 5

unless you'd prefer this built-in.