We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[DataPipe] Mapper DataPipe should not deepcopy when index specified
## 🐛 Bug

I was adding the following line to prevent in-place modification of the data from source DataPipe. 
https://github.com/pytorch/pytorch/blob/d37c02be08dfc022daf2ee1ddeda2a37b4551cac/torch/utils/data/datapipes/iter/callable.py#L102-L103

But, in fact, this would break when input includes file handle, because file handle can not be serialized.

So, in order to support file handle, we need to remove deepcopy. But, for the sake of preventing in-place modification, we need to add documentation to wiki about remove data attached to DataPipe instance. We prefer using iterator to generate data.

Then, we need to also change the `IterableWrapper` to do a `deepcopy` if possible. https://github.com/pytorch/pytorch/blob/a49907f984670781a718ef6aa0046709886eae5a/torch/utils/data/datapipes/iter/utils.py#L12-L17

cc @VitalyFedyunin @ejguan

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `torch/utils/data/datapipes/iter/utils.py`
Content:
```
1 from torch.utils.data import IterDataPipe
2 
3 
4 class IterableWrapperIterDataPipe(IterDataPipe):
5     r""":class:`IterableWrapperIterDataPipe`.
6 
7     Iterable datapipe that wraps an iterable object.
8 
9     Args:
10         iterable: Iterable object to be wrapped into an IterDataPipe
11     """
12     def __init__(self, iterable):
13         self.iterable = iterable
14 
15     def __iter__(self):
16         for data in self.iterable:
17             yield data
18 
19     def __len__(self):
20         return len(self.iterable)
21
```

Path: `torch/utils/data/datapipes/iter/callable.py`
Content:
```
1 import copy
2 import warnings
3 from torch.utils.data import IterDataPipe, _utils, functional_datapipe, DataChunk
4 from typing import Callable, Dict, Iterator, Optional, Sized, Tuple, TypeVar
5 
6 try:
7     import dill
8 
9     # XXX: By default, dill writes the Pickler dispatch table to inject its
10     # own logic there. This globally affects the behavior of the standard library
11     # pickler for any user who transitively depends on this module!
12     # Undo this extension to avoid altering the behavior of the pickler globally.
13     dill.extend(use_dill=False)
14     DILL_AVAILABLE = True
15 except ImportError:
16     DILL_AVAILABLE = False
17 
18 T_co = TypeVar("T_co", covariant=True)
19 
20 
21 @functional_datapipe("map")
22 class MapperIterDataPipe(IterDataPipe[T_co]):
23     r""":class:`MapperIterDataPipe`.
24 
25     Iterable DataPipe to run a function over each item from the source DataPipe.
26     The function can be any regular python function or partial object. Lambda
27     function is not recommended as it is not supported by pickle.
28 
29     Args:
30         datapipe: Source Iterable DataPipe
31         fn: Function called over each item
32         input_col: Index or indices of data which `fn` is applied
33             - None as default to apply `fn` to the data directly.
34             - Integer(s) is used for list/tuple.
35             - Key(s) is used for dict.
36         output_col: Index of data where result of `fn` is placed. `output_col` can be specified only when `input_col` is not None
37             - None as default to replace the index that `input_col` specified;
38               For `input_col` with multiple indices, the left-most one is used, and other indices will be removed.
39             - Integer is used for list/tuple. -1 represents to append result at the end.
40             - Key is used for dict. New key is acceptable.
41         fn_args: Positional arguments for `fn`
42         fn_kwargs: Keyword arguments for `fn`
43         nesting_level: Determines which level the fn gets applied to, by default it applies to the top level (= 0).
44             This also accepts -1 as input to apply the function to the lowest nesting level. It currently doesn't support
45             argument < -1.
46     """
47     datapipe: IterDataPipe
48     fn: Callable
49 
50     def __init__(
51         self,
52         datapipe: IterDataPipe,
53         fn: Callable,
54         input_col=None,
55         output_col=None,
56         *,
57         fn_args: Optional[Tuple] = None,
58         fn_kwargs: Optional[Dict] = None,
59         nesting_level: int = 0,
60     ) -> None:
61         super().__init__()
62         self.datapipe = datapipe
63         # Partial object has no attribute '__name__', but can be pickled
64         if hasattr(fn, "__name__") and fn.__name__ == "<lambda>" and not DILL_AVAILABLE:
65             warnings.warn(
66                 "Lambda function is not supported for pickle, please use "
67                 "regular python function or functools.partial instead."
68             )
69         self.fn = fn  # type: ignore[assignment]
70         self.input_col = input_col
71         if input_col is None and output_col is not None:
72             raise ValueError("`output_col` must be None when `input_col` is None.")
73         if isinstance(output_col, (list, tuple)):
74             if len(output_col) > 1:
75                 raise ValueError("`output_col` must be a single-element list or tuple")
76             output_col = output_col[0]
77         self.output_col = output_col
78         self.args = () if fn_args is None else fn_args
79         self.kwargs = {} if fn_kwargs is None else fn_kwargs
80         if nesting_level < -1:
81             raise ValueError("nesting_level must be -1 or >= 0")
82         self.nesting_level = nesting_level
83 
84     def _apply_fn(self, data):
85         if self.input_col is None and self.output_col is None:
86             return self.fn(data, *self.args, **self.kwargs)
87 
88         if self.input_col is None:
89             res = self.fn(data, *self.args, **self.kwargs)
90         elif isinstance(self.input_col, (list, tuple)):
91             args = tuple(data[col] for col in self.input_col)
92             res = self.fn(*args, *self.args, **self.kwargs)
93         else:
94             res = self.fn(data[self.input_col], *self.args, **self.kwargs)
95 
96         # Copy tuple to list and run in-place modification because tuple is immutable.
97         if isinstance(data, tuple):
98             t_flag = True
99             data = list(data)
100         else:
101             t_flag = False
102             # Deepcopy data to prevent the original data modified. E.g. list, dict
103             data = copy.deepcopy(data)
104 
105         if self.output_col is None:
106             if isinstance(self.input_col, (list, tuple)):
107                 data[self.input_col[0]] = res
108                 for idx in sorted(self.input_col[1:], reverse=True):
109                     del data[idx]
110             else:
111                 data[self.input_col] = res
112         else:
113             if self.output_col == -1:
114                 data.append(res)
115             else:
116                 data[self.output_col] = res
117 
118         # Convert list back to tuple
119         return tuple(data) if t_flag else data
120 
121     def _apply(self, data, nesting_level):
122         if nesting_level == 0:
123             return self._apply_fn(data)
124         elif nesting_level > 0:
125             if isinstance(data, DataChunk):
126                 return type(data)(
127                     [self._apply(i, nesting_level - 1) for i in data.raw_iterator()]
128                 )
129             elif isinstance(data, list):
130                 return [self._apply(i, nesting_level - 1) for i in data]
131             else:
132                 raise IndexError(
133                     f"nesting_level {self.nesting_level} out of range (exceeds data pipe depth)"
134                 )
135         else:
136             if isinstance(data, DataChunk):
137                 return type(data)(
138                     [self._apply(i, nesting_level) for i in data.raw_iterator()]
139                 )
140             elif isinstance(data, list):
141                 return [self._apply(i, nesting_level) for i in data]
142             else:
143                 return self._apply_fn(data)
144 
145     def __iter__(self) -> Iterator[T_co]:
146         for data in self.datapipe:
147             yield self._apply(data, self.nesting_level)
148 
149     def __len__(self) -> int:
150         if isinstance(self.datapipe, Sized):
151             return len(self.datapipe)
152         raise TypeError(
153             "{} instance doesn't have valid length".format(type(self).__name__)
154         )
155 
156     def __getstate__(self):
157         if DILL_AVAILABLE:
158             dill_function = dill.dumps(self.fn)
159         else:
160             dill_function = self.fn
161         state = (
162             self.datapipe,
163             dill_function,
164             self.input_col,
165             self.output_col,
166             self.args,
167             self.kwargs,
168             self.nesting_level,
169         )
170         return state
171 
172     def __setstate__(self, state):
173         (
174             self.datapipe,
175             dill_function,
176             self.input_col,
177             self.output_col,
178             self.args,
179             self.kwargs,
180             self.nesting_level,
181         ) = state
182         if DILL_AVAILABLE:
183             self.fn = dill.loads(dill_function)  # type: ignore[assignment]
184         else:
185             self.fn = dill_function  # type: ignore[assignment]
186 
187 
188 @functional_datapipe("collate")
189 class CollatorIterDataPipe(MapperIterDataPipe):
190     r""":class:`CollatorIterDataPipe`.
191 
192     Iterable DataPipe to collate samples from datapipe to Tensor(s) by `util_.collate.default_collate`,
193     or customized Data Structure by collate_fn.
194 
195     Args:
196         datapipe: Iterable DataPipe being collated
197         collate_fn: Customized collate function to collect and combine data or a batch of data.
198             Default function collates to Tensor(s) based on data type.
199         fn_args: Positional arguments for `collate_fn`
200         fn_kwargs: Keyword arguments for `collate_fn`
201 
202     Example: Convert integer data to float Tensor
203         >>> class MyIterDataPipe(torch.utils.data.IterDataPipe):
204         ...     def __init__(self, start, end):
205         ...         super(MyIterDataPipe).__init__()
206         ...         assert end > start, "this example code only works with end >= start"
207         ...         self.start = start
208         ...         self.end = end
209         ...
210         ...     def __iter__(self):
211         ...         return iter(range(self.start, self.end))
212         ...
213         ...     def __len__(self):
214         ...         return self.end - self.start
215         ...
216         >>> ds = MyIterDataPipe(start=3, end=7)
217         >>> print(list(ds))
218         [3, 4, 5, 6]
219 
220         >>> def collate_fn(batch):
221         ...     return torch.tensor(batch, dtype=torch.float)
222         ...
223         >>> collated_ds = CollateIterDataPipe(ds, collate_fn=collate_fn)
224         >>> print(list(collated_ds))
225         [tensor(3.), tensor(4.), tensor(5.), tensor(6.)]
226     """
227 
228     def __init__(
229         self,
230         datapipe: IterDataPipe,
231         collate_fn: Callable = _utils.collate.default_collate,
232         fn_args: Optional[Tuple] = None,
233         fn_kwargs: Optional[Dict] = None,
234     ) -> None:
235         super().__init__(datapipe, fn=collate_fn, fn_args=fn_args, fn_kwargs=fn_kwargs)
236
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/torch/utils/data/datapipes/iter/callable.py b/torch/utils/data/datapipes/iter/callable.py
--- a/torch/utils/data/datapipes/iter/callable.py
+++ b/torch/utils/data/datapipes/iter/callable.py
@@ -1,4 +1,3 @@
-import copy
 import warnings
 from torch.utils.data import IterDataPipe, _utils, functional_datapipe, DataChunk
 from typing import Callable, Dict, Iterator, Optional, Sized, Tuple, TypeVar
@@ -99,8 +98,6 @@
             data = list(data)
         else:
             t_flag = False
-            # Deepcopy data to prevent the original data modified. E.g. list, dict
-            data = copy.deepcopy(data)
 
         if self.output_col is None:
             if isinstance(self.input_col, (list, tuple)):
diff --git a/torch/utils/data/datapipes/iter/utils.py b/torch/utils/data/datapipes/iter/utils.py
--- a/torch/utils/data/datapipes/iter/utils.py
+++ b/torch/utils/data/datapipes/iter/utils.py
@@ -1,3 +1,5 @@
+import copy
+import warnings
 from torch.utils.data import IterDataPipe
 
 
@@ -8,12 +10,34 @@
 
     Args:
         iterable: Iterable object to be wrapped into an IterDataPipe
+        deepcopy: Option to deepcopy input iterable object for each
+            iteration.
+
+    .. note::
+      If `deepcopy` is set to False explicitly, users should ensure
+      that data pipeline doesn't contain any in-place operations over
+      the iterable instance, in order to prevent data inconsistency
+      across iterations.
     """
-    def __init__(self, iterable):
+    def __init__(self, iterable, deepcopy=True):
         self.iterable = iterable
+        self.deepcopy = deepcopy
 
     def __iter__(self):
-        for data in self.iterable:
+        source_data = self.iterable
+        if self.deepcopy:
+            try:
+                source_data = copy.deepcopy(self.iterable)
+            # For the case that data cannot be deep-copied,
+            # all in-place operations will affect iterable variable.
+            # When this DataPipe is iterated second time, it will
+            # yield modified items.
+            except TypeError:
+                warnings.warn(
+                    "The input iterable can not be deepcopied, "
+                    "please be aware of in-place modification would affect source data"
+                )
+        for data in source_data:
             yield data
 
     def __len__(self):

{"golden_diff": "diff --git a/torch/utils/data/datapipes/iter/callable.py b/torch/utils/data/datapipes/iter/callable.py\n--- a/torch/utils/data/datapipes/iter/callable.py\n+++ b/torch/utils/data/datapipes/iter/callable.py\n@@ -1,4 +1,3 @@\n-import copy\n import warnings\n from torch.utils.data import IterDataPipe, _utils, functional_datapipe, DataChunk\n from typing import Callable, Dict, Iterator, Optional, Sized, Tuple, TypeVar\n@@ -99,8 +98,6 @@\n             data = list(data)\n         else:\n             t_flag = False\n-            # Deepcopy data to prevent the original data modified. E.g. list, dict\n-            data = copy.deepcopy(data)\n \n         if self.output_col is None:\n             if isinstance(self.input_col, (list, tuple)):\ndiff --git a/torch/utils/data/datapipes/iter/utils.py b/torch/utils/data/datapipes/iter/utils.py\n--- a/torch/utils/data/datapipes/iter/utils.py\n+++ b/torch/utils/data/datapipes/iter/utils.py\n@@ -1,3 +1,5 @@\n+import copy\n+import warnings\n from torch.utils.data import IterDataPipe\n \n \n@@ -8,12 +10,34 @@\n \n     Args:\n         iterable: Iterable object to be wrapped into an IterDataPipe\n+        deepcopy: Option to deepcopy input iterable object for each\n+            iteration.\n+\n+    .. note::\n+      If `deepcopy` is set to False explicitly, users should ensure\n+      that data pipeline doesn't contain any in-place operations over\n+      the iterable instance, in order to prevent data inconsistency\n+      across iterations.\n     \"\"\"\n-    def __init__(self, iterable):\n+    def __init__(self, iterable, deepcopy=True):\n         self.iterable = iterable\n+        self.deepcopy = deepcopy\n \n     def __iter__(self):\n-        for data in self.iterable:\n+        source_data = self.iterable\n+        if self.deepcopy:\n+            try:\n+                source_data = copy.deepcopy(self.iterable)\n+            # For the case that data cannot be deep-copied,\n+            # all in-place operations will affect iterable variable.\n+            # When this DataPipe is iterated second time, it will\n+            # yield modified items.\n+            except TypeError:\n+                warnings.warn(\n+                    \"The input iterable can not be deepcopied, \"\n+                    \"please be aware of in-place modification would affect source data\"\n+                )\n+        for data in source_data:\n             yield data\n \n     def __len__(self):\n", "issue": "[DataPipe] Mapper DataPipe should not deepcopy when index specified\n## \ud83d\udc1b Bug\r\n\r\nI was adding the following line to prevent in-place modification of the data from source DataPipe. \r\nhttps://github.com/pytorch/pytorch/blob/d37c02be08dfc022daf2ee1ddeda2a37b4551cac/torch/utils/data/datapipes/iter/callable.py#L102-L103\r\n\r\nBut, in fact, this would break when input includes file handle, because file handle can not be serialized.\r\n\r\nSo, in order to support file handle, we need to remove deepcopy. But, for the sake of preventing in-place modification, we need to add documentation to wiki about remove data attached to DataPipe instance. We prefer using iterator to generate data.\r\n\r\nThen, we need to also change the `IterableWrapper` to do a `deepcopy` if possible. https://github.com/pytorch/pytorch/blob/a49907f984670781a718ef6aa0046709886eae5a/torch/utils/data/datapipes/iter/utils.py#L12-L17\n\ncc @VitalyFedyunin @ejguan\n", "before_files": [{"content": "from torch.utils.data import IterDataPipe\n\n\nclass IterableWrapperIterDataPipe(IterDataPipe):\n    r\"\"\":class:`IterableWrapperIterDataPipe`.\n\n    Iterable datapipe that wraps an iterable object.\n\n    Args:\n        iterable: Iterable object to be wrapped into an IterDataPipe\n    \"\"\"\n    def __init__(self, iterable):\n        self.iterable = iterable\n\n    def __iter__(self):\n        for data in self.iterable:\n            yield data\n\n    def __len__(self):\n        return len(self.iterable)\n", "path": "torch/utils/data/datapipes/iter/utils.py"}, {"content": "import copy\nimport warnings\nfrom torch.utils.data import IterDataPipe, _utils, functional_datapipe, DataChunk\nfrom typing import Callable, Dict, Iterator, Optional, Sized, Tuple, TypeVar\n\ntry:\n    import dill\n\n    # XXX: By default, dill writes the Pickler dispatch table to inject its\n    # own logic there. This globally affects the behavior of the standard library\n    # pickler for any user who transitively depends on this module!\n    # Undo this extension to avoid altering the behavior of the pickler globally.\n    dill.extend(use_dill=False)\n    DILL_AVAILABLE = True\nexcept ImportError:\n    DILL_AVAILABLE = False\n\nT_co = TypeVar(\"T_co\", covariant=True)\n\n\n@functional_datapipe(\"map\")\nclass MapperIterDataPipe(IterDataPipe[T_co]):\n    r\"\"\":class:`MapperIterDataPipe`.\n\n    Iterable DataPipe to run a function over each item from the source DataPipe.\n    The function can be any regular python function or partial object. Lambda\n    function is not recommended as it is not supported by pickle.\n\n    Args:\n        datapipe: Source Iterable DataPipe\n        fn: Function called over each item\n        input_col: Index or indices of data which `fn` is applied\n            - None as default to apply `fn` to the data directly.\n            - Integer(s) is used for list/tuple.\n            - Key(s) is used for dict.\n        output_col: Index of data where result of `fn` is placed. `output_col` can be specified only when `input_col` is not None\n            - None as default to replace the index that `input_col` specified;\n              For `input_col` with multiple indices, the left-most one is used, and other indices will be removed.\n            - Integer is used for list/tuple. -1 represents to append result at the end.\n            - Key is used for dict. New key is acceptable.\n        fn_args: Positional arguments for `fn`\n        fn_kwargs: Keyword arguments for `fn`\n        nesting_level: Determines which level the fn gets applied to, by default it applies to the top level (= 0).\n            This also accepts -1 as input to apply the function to the lowest nesting level. It currently doesn't support\n            argument < -1.\n    \"\"\"\n    datapipe: IterDataPipe\n    fn: Callable\n\n    def __init__(\n        self,\n        datapipe: IterDataPipe,\n        fn: Callable,\n        input_col=None,\n        output_col=None,\n        *,\n        fn_args: Optional[Tuple] = None,\n        fn_kwargs: Optional[Dict] = None,\n        nesting_level: int = 0,\n    ) -> None:\n        super().__init__()\n        self.datapipe = datapipe\n        # Partial object has no attribute '__name__', but can be pickled\n        if hasattr(fn, \"__name__\") and fn.__name__ == \"<lambda>\" and not DILL_AVAILABLE:\n            warnings.warn(\n                \"Lambda function is not supported for pickle, please use \"\n                \"regular python function or functools.partial instead.\"\n            )\n        self.fn = fn  # type: ignore[assignment]\n        self.input_col = input_col\n        if input_col is None and output_col is not None:\n            raise ValueError(\"`output_col` must be None when `input_col` is None.\")\n        if isinstance(output_col, (list, tuple)):\n            if len(output_col) > 1:\n                raise ValueError(\"`output_col` must be a single-element list or tuple\")\n            output_col = output_col[0]\n        self.output_col = output_col\n        self.args = () if fn_args is None else fn_args\n        self.kwargs = {} if fn_kwargs is None else fn_kwargs\n        if nesting_level < -1:\n            raise ValueError(\"nesting_level must be -1 or >= 0\")\n        self.nesting_level = nesting_level\n\n    def _apply_fn(self, data):\n        if self.input_col is None and self.output_col is None:\n            return self.fn(data, *self.args, **self.kwargs)\n\n        if self.input_col is None:\n            res = self.fn(data, *self.args, **self.kwargs)\n        elif isinstance(self.input_col, (list, tuple)):\n            args = tuple(data[col] for col in self.input_col)\n            res = self.fn(*args, *self.args, **self.kwargs)\n        else:\n            res = self.fn(data[self.input_col], *self.args, **self.kwargs)\n\n        # Copy tuple to list and run in-place modification because tuple is immutable.\n        if isinstance(data, tuple):\n            t_flag = True\n            data = list(data)\n        else:\n            t_flag = False\n            # Deepcopy data to prevent the original data modified. E.g. list, dict\n            data = copy.deepcopy(data)\n\n        if self.output_col is None:\n            if isinstance(self.input_col, (list, tuple)):\n                data[self.input_col[0]] = res\n                for idx in sorted(self.input_col[1:], reverse=True):\n                    del data[idx]\n            else:\n                data[self.input_col] = res\n        else:\n            if self.output_col == -1:\n                data.append(res)\n            else:\n                data[self.output_col] = res\n\n        # Convert list back to tuple\n        return tuple(data) if t_flag else data\n\n    def _apply(self, data, nesting_level):\n        if nesting_level == 0:\n            return self._apply_fn(data)\n        elif nesting_level > 0:\n            if isinstance(data, DataChunk):\n                return type(data)(\n                    [self._apply(i, nesting_level - 1) for i in data.raw_iterator()]\n                )\n            elif isinstance(data, list):\n                return [self._apply(i, nesting_level - 1) for i in data]\n            else:\n                raise IndexError(\n                    f\"nesting_level {self.nesting_level} out of range (exceeds data pipe depth)\"\n                )\n        else:\n            if isinstance(data, DataChunk):\n                return type(data)(\n                    [self._apply(i, nesting_level) for i in data.raw_iterator()]\n                )\n            elif isinstance(data, list):\n                return [self._apply(i, nesting_level) for i in data]\n            else:\n                return self._apply_fn(data)\n\n    def __iter__(self) -> Iterator[T_co]:\n        for data in self.datapipe:\n            yield self._apply(data, self.nesting_level)\n\n    def __len__(self) -> int:\n        if isinstance(self.datapipe, Sized):\n            return len(self.datapipe)\n        raise TypeError(\n            \"{} instance doesn't have valid length\".format(type(self).__name__)\n        )\n\n    def __getstate__(self):\n        if DILL_AVAILABLE:\n            dill_function = dill.dumps(self.fn)\n        else:\n            dill_function = self.fn\n        state = (\n            self.datapipe,\n            dill_function,\n            self.input_col,\n            self.output_col,\n            self.args,\n            self.kwargs,\n            self.nesting_level,\n        )\n        return state\n\n    def __setstate__(self, state):\n        (\n            self.datapipe,\n            dill_function,\n            self.input_col,\n            self.output_col,\n            self.args,\n            self.kwargs,\n            self.nesting_level,\n        ) = state\n        if DILL_AVAILABLE:\n            self.fn = dill.loads(dill_function)  # type: ignore[assignment]\n        else:\n            self.fn = dill_function  # type: ignore[assignment]\n\n\n@functional_datapipe(\"collate\")\nclass CollatorIterDataPipe(MapperIterDataPipe):\n    r\"\"\":class:`CollatorIterDataPipe`.\n\n    Iterable DataPipe to collate samples from datapipe to Tensor(s) by `util_.collate.default_collate`,\n    or customized Data Structure by collate_fn.\n\n    Args:\n        datapipe: Iterable DataPipe being collated\n        collate_fn: Customized collate function to collect and combine data or a batch of data.\n            Default function collates to Tensor(s) based on data type.\n        fn_args: Positional arguments for `collate_fn`\n        fn_kwargs: Keyword arguments for `collate_fn`\n\n    Example: Convert integer data to float Tensor\n        >>> class MyIterDataPipe(torch.utils.data.IterDataPipe):\n        ...     def __init__(self, start, end):\n        ...         super(MyIterDataPipe).__init__()\n        ...         assert end > start, \"this example code only works with end >= start\"\n        ...         self.start = start\n        ...         self.end = end\n        ...\n        ...     def __iter__(self):\n        ...         return iter(range(self.start, self.end))\n        ...\n        ...     def __len__(self):\n        ...         return self.end - self.start\n        ...\n        >>> ds = MyIterDataPipe(start=3, end=7)\n        >>> print(list(ds))\n        [3, 4, 5, 6]\n\n        >>> def collate_fn(batch):\n        ...     return torch.tensor(batch, dtype=torch.float)\n        ...\n        >>> collated_ds = CollateIterDataPipe(ds, collate_fn=collate_fn)\n        >>> print(list(collated_ds))\n        [tensor(3.), tensor(4.), tensor(5.), tensor(6.)]\n    \"\"\"\n\n    def __init__(\n        self,\n        datapipe: IterDataPipe,\n        collate_fn: Callable = _utils.collate.default_collate,\n        fn_args: Optional[Tuple] = None,\n        fn_kwargs: Optional[Dict] = None,\n    ) -> None:\n        super().__init__(datapipe, fn=collate_fn, fn_args=fn_args, fn_kwargs=fn_kwargs)\n", "path": "torch/utils/data/datapipes/iter/callable.py"}], "after_files": [{"content": "import copy\nimport warnings\nfrom torch.utils.data import IterDataPipe\n\n\nclass IterableWrapperIterDataPipe(IterDataPipe):\n    r\"\"\":class:`IterableWrapperIterDataPipe`.\n\n    Iterable datapipe that wraps an iterable object.\n\n    Args:\n        iterable: Iterable object to be wrapped into an IterDataPipe\n        deepcopy: Option to deepcopy input iterable object for each\n            iteration.\n\n    .. note::\n      If `deepcopy` is set to False explicitly, users should ensure\n      that data pipeline doesn't contain any in-place operations over\n      the iterable instance, in order to prevent data inconsistency\n      across iterations.\n    \"\"\"\n    def __init__(self, iterable, deepcopy=True):\n        self.iterable = iterable\n        self.deepcopy = deepcopy\n\n    def __iter__(self):\n        source_data = self.iterable\n        if self.deepcopy:\n            try:\n                source_data = copy.deepcopy(self.iterable)\n            # For the case that data cannot be deep-copied,\n            # all in-place operations will affect iterable variable.\n            # When this DataPipe is iterated second time, it will\n            # yield modified items.\n            except TypeError:\n                warnings.warn(\n                    \"The input iterable can not be deepcopied, \"\n                    \"please be aware of in-place modification would affect source data\"\n                )\n        for data in source_data:\n            yield data\n\n    def __len__(self):\n        return len(self.iterable)\n", "path": "torch/utils/data/datapipes/iter/utils.py"}, {"content": "import warnings\nfrom torch.utils.data import IterDataPipe, _utils, functional_datapipe, DataChunk\nfrom typing import Callable, Dict, Iterator, Optional, Sized, Tuple, TypeVar\n\ntry:\n    import dill\n\n    # XXX: By default, dill writes the Pickler dispatch table to inject its\n    # own logic there. This globally affects the behavior of the standard library\n    # pickler for any user who transitively depends on this module!\n    # Undo this extension to avoid altering the behavior of the pickler globally.\n    dill.extend(use_dill=False)\n    DILL_AVAILABLE = True\nexcept ImportError:\n    DILL_AVAILABLE = False\n\nT_co = TypeVar(\"T_co\", covariant=True)\n\n\n@functional_datapipe(\"map\")\nclass MapperIterDataPipe(IterDataPipe[T_co]):\n    r\"\"\":class:`MapperIterDataPipe`.\n\n    Iterable DataPipe to run a function over each item from the source DataPipe.\n    The function can be any regular python function or partial object. Lambda\n    function is not recommended as it is not supported by pickle.\n\n    Args:\n        datapipe: Source Iterable DataPipe\n        fn: Function called over each item\n        input_col: Index or indices of data which `fn` is applied\n            - None as default to apply `fn` to the data directly.\n            - Integer(s) is used for list/tuple.\n            - Key(s) is used for dict.\n        output_col: Index of data where result of `fn` is placed. `output_col` can be specified only when `input_col` is not None\n            - None as default to replace the index that `input_col` specified;\n              For `input_col` with multiple indices, the left-most one is used, and other indices will be removed.\n            - Integer is used for list/tuple. -1 represents to append result at the end.\n            - Key is used for dict. New key is acceptable.\n        fn_args: Positional arguments for `fn`\n        fn_kwargs: Keyword arguments for `fn`\n        nesting_level: Determines which level the fn gets applied to, by default it applies to the top level (= 0).\n            This also accepts -1 as input to apply the function to the lowest nesting level. It currently doesn't support\n            argument < -1.\n    \"\"\"\n    datapipe: IterDataPipe\n    fn: Callable\n\n    def __init__(\n        self,\n        datapipe: IterDataPipe,\n        fn: Callable,\n        input_col=None,\n        output_col=None,\n        *,\n        fn_args: Optional[Tuple] = None,\n        fn_kwargs: Optional[Dict] = None,\n        nesting_level: int = 0,\n    ) -> None:\n        super().__init__()\n        self.datapipe = datapipe\n        # Partial object has no attribute '__name__', but can be pickled\n        if hasattr(fn, \"__name__\") and fn.__name__ == \"<lambda>\" and not DILL_AVAILABLE:\n            warnings.warn(\n                \"Lambda function is not supported for pickle, please use \"\n                \"regular python function or functools.partial instead.\"\n            )\n        self.fn = fn  # type: ignore[assignment]\n        self.input_col = input_col\n        if input_col is None and output_col is not None:\n            raise ValueError(\"`output_col` must be None when `input_col` is None.\")\n        if isinstance(output_col, (list, tuple)):\n            if len(output_col) > 1:\n                raise ValueError(\"`output_col` must be a single-element list or tuple\")\n            output_col = output_col[0]\n        self.output_col = output_col\n        self.args = () if fn_args is None else fn_args\n        self.kwargs = {} if fn_kwargs is None else fn_kwargs\n        if nesting_level < -1:\n            raise ValueError(\"nesting_level must be -1 or >= 0\")\n        self.nesting_level = nesting_level\n\n    def _apply_fn(self, data):\n        if self.input_col is None and self.output_col is None:\n            return self.fn(data, *self.args, **self.kwargs)\n\n        if self.input_col is None:\n            res = self.fn(data, *self.args, **self.kwargs)\n        elif isinstance(self.input_col, (list, tuple)):\n            args = tuple(data[col] for col in self.input_col)\n            res = self.fn(*args, *self.args, **self.kwargs)\n        else:\n            res = self.fn(data[self.input_col], *self.args, **self.kwargs)\n\n        # Copy tuple to list and run in-place modification because tuple is immutable.\n        if isinstance(data, tuple):\n            t_flag = True\n            data = list(data)\n        else:\n            t_flag = False\n\n        if self.output_col is None:\n            if isinstance(self.input_col, (list, tuple)):\n                data[self.input_col[0]] = res\n                for idx in sorted(self.input_col[1:], reverse=True):\n                    del data[idx]\n            else:\n                data[self.input_col] = res\n        else:\n            if self.output_col == -1:\n                data.append(res)\n            else:\n                data[self.output_col] = res\n\n        # Convert list back to tuple\n        return tuple(data) if t_flag else data\n\n    def _apply(self, data, nesting_level):\n        if nesting_level == 0:\n            return self._apply_fn(data)\n        elif nesting_level > 0:\n            if isinstance(data, DataChunk):\n                return type(data)(\n                    [self._apply(i, nesting_level - 1) for i in data.raw_iterator()]\n                )\n            elif isinstance(data, list):\n                return [self._apply(i, nesting_level - 1) for i in data]\n            else:\n                raise IndexError(\n                    f\"nesting_level {self.nesting_level} out of range (exceeds data pipe depth)\"\n                )\n        else:\n            if isinstance(data, DataChunk):\n                return type(data)(\n                    [self._apply(i, nesting_level) for i in data.raw_iterator()]\n                )\n            elif isinstance(data, list):\n                return [self._apply(i, nesting_level) for i in data]\n            else:\n                return self._apply_fn(data)\n\n    def __iter__(self) -> Iterator[T_co]:\n        for data in self.datapipe:\n            yield self._apply(data, self.nesting_level)\n\n    def __len__(self) -> int:\n        if isinstance(self.datapipe, Sized):\n            return len(self.datapipe)\n        raise TypeError(\n            \"{} instance doesn't have valid length\".format(type(self).__name__)\n        )\n\n    def __getstate__(self):\n        if DILL_AVAILABLE:\n            dill_function = dill.dumps(self.fn)\n        else:\n            dill_function = self.fn\n        state = (\n            self.datapipe,\n            dill_function,\n            self.input_col,\n            self.output_col,\n            self.args,\n            self.kwargs,\n            self.nesting_level,\n        )\n        return state\n\n    def __setstate__(self, state):\n        (\n            self.datapipe,\n            dill_function,\n            self.input_col,\n            self.output_col,\n            self.args,\n            self.kwargs,\n            self.nesting_level,\n        ) = state\n        if DILL_AVAILABLE:\n            self.fn = dill.loads(dill_function)  # type: ignore[assignment]\n        else:\n            self.fn = dill_function  # type: ignore[assignment]\n\n\n@functional_datapipe(\"collate\")\nclass CollatorIterDataPipe(MapperIterDataPipe):\n    r\"\"\":class:`CollatorIterDataPipe`.\n\n    Iterable DataPipe to collate samples from datapipe to Tensor(s) by `util_.collate.default_collate`,\n    or customized Data Structure by collate_fn.\n\n    Args:\n        datapipe: Iterable DataPipe being collated\n        collate_fn: Customized collate function to collect and combine data or a batch of data.\n            Default function collates to Tensor(s) based on data type.\n        fn_args: Positional arguments for `collate_fn`\n        fn_kwargs: Keyword arguments for `collate_fn`\n\n    Example: Convert integer data to float Tensor\n        >>> class MyIterDataPipe(torch.utils.data.IterDataPipe):\n        ...     def __init__(self, start, end):\n        ...         super(MyIterDataPipe).__init__()\n        ...         assert end > start, \"this example code only works with end >= start\"\n        ...         self.start = start\n        ...         self.end = end\n        ...\n        ...     def __iter__(self):\n        ...         return iter(range(self.start, self.end))\n        ...\n        ...     def __len__(self):\n        ...         return self.end - self.start\n        ...\n        >>> ds = MyIterDataPipe(start=3, end=7)\n        >>> print(list(ds))\n        [3, 4, 5, 6]\n\n        >>> def collate_fn(batch):\n        ...     return torch.tensor(batch, dtype=torch.float)\n        ...\n        >>> collated_ds = CollateIterDataPipe(ds, collate_fn=collate_fn)\n        >>> print(list(collated_ds))\n        [tensor(3.), tensor(4.), tensor(5.), tensor(6.)]\n    \"\"\"\n\n    def __init__(\n        self,\n        datapipe: IterDataPipe,\n        collate_fn: Callable = _utils.collate.default_collate,\n        fn_args: Optional[Tuple] = None,\n        fn_kwargs: Optional[Dict] = None,\n    ) -> None:\n        super().__init__(datapipe, fn=collate_fn, fn_args=fn_args, fn_kwargs=fn_kwargs)\n", "path": "torch/utils/data/datapipes/iter/callable.py"}]}