[Core][multiprocessing] Make Pool.imap() and Pool.imap_unordered() throw TypeError on non-iterable/iterator

[simontindemans]

Description

Currently, the Ray implementations of Pool.imap() and Pool.imap_unordered() accept arguments of the type (func, iterable), where iterable can be an actual iterable, an iterator, or a single input. In In the letter case, the IMapIterator constructor treats iterable as [iterable]. The code implementing this (since this PR, which is functionally identical to previous versions in this regard) is as follows:

        try:
            self._iterator = iter(iterable)
        except TypeError:
            # for compatibility with prior releases, encapsulate non-iterable in a list
            iterable = [iterable]
            self._iterator = iter(iterable)

This behaviour deviates from the core Python implementations of map and multiprocessing.imap and multiprocessing.imap_unordered, which all throw a TypeError when given a single value as iterable.

As this potentially breaks existing code, this was proposed as an issue to be addressed in a future release here:
Originally posted by @edoakes in #24117 (comment)

Use case

The aim is to have a consistent interface for:

• Python map
• multiprocessing: imap and imap_unordered
• ray.util.multiprocessing: imap and imap_unordered

[simontindemans]

Minimal reproducing code:

import ray.util.multiprocessing as raymp
import multiprocessing as mp

def f(x):
    print(x*x)

if __name__ == "__main__":

    # standard python map
    # works
    [value for value in map(f, [3])]
    # TypeError
    try:
        [value for value in map(f, 3)]
    except TypeError:
        print("TypeError as expected")

    # multiprocessing imap
    pool = mp.Pool()
    # works
    [value for value in pool.imap(f, [3])]
    # TypeError
    try:
        [value for value in pool.imap(f, 3)]
    except TypeError:
        print("TypeError as expected")
    pool.terminate()

    # ray multiprocessing imap
    pool = raymp.Pool()
    # works
    [value for value in pool.imap(f, [3])]
    # No TypeError
    [value for value in pool.imap(f, 3)]
    print("WARNING: should throw TypeError") 
    pool.terminate() 

[jjyao]

We can do it as part of 2.0 if we decide to do it.

[edoakes]

@jjyao do we have a label tracking items for 2.0?

[jjyao]

@edoakes Not yet. Going to create one.

[simontindemans]

Perfect; you can ping me if/when you'd like to make the change. It's fairly trivial: removing a four-line workaround I had to insert to maintain compatibility.

[hora-anyscale]

@simontindemans - if you could share a PR that fixes this it would be very much appreciated.

[rkooo567]

We will send a warning in 2.3, raise an exception at 2.4 cc @cadedaniel