mars-project / mars-project/mars
[BUG] Mars client can't fetch large data
- Dominant language
- Python
- Stars
- 2.7k
- Forks
- 325
- PR merge metrics
- No merged PRs in 30d
Description
**Describe the bug**
Mars client can't fetch large data, otherwise tornado will throw `Content-Length too long` exception
**To Reproduce**
To help us reproducing this bug, please provide information below:
1. Your Python version: 3.7.9
2. The version of Mars you use: 0.9.0
3. Versions of crucial packages, such as numpy, scipy and pandas
4. Full stack of the error.
```
2022-06-07 20:39:59,643 - tornado.general - INFO - Malformed HTTP message from None: Content-Length too long
INFO:tornado.general:Malformed HTTP message from None: Content-Length too long
---------------------------------------------------------------------------
HTTPStreamClosedError Traceback (most recent call last)
/tmp/ipykernel_1930/3440895109.py in
2 r=mars.core.ExecutableTuple([
3 mt.random.RandomState(0).rand(1000_0000, 5, chunk_size=100_0000),
----> 4 mt.random.RandomState(0).rand(1000_0000, 5, chunk_size=100_0000),
5 ]).execute()._fetch()
~/miniconda3/lib/python3.7/site-packages/mars/core/entity/executable.py in _fetch(self, session, **kw)
278 session = _get_session(self, session)
279 self._check_session(session, "fetch")
--> 280 return fetch(*self, session=session, **kw)
281
282 def _fetch_infos(self, fields=None, session=None, **kw):
~/miniconda3/lib/python3.7/site-packages/mars/deploy/oscar/session.py in fetch(tileable, session, *tileables, **kwargs)
1907
1908 session = _ensure_sync(session)
-> 1909 return session.fetch(tileable, *tileables, **kwargs)
1910
1911
~/miniconda3/lib/python3.7/site-packages/mars/deploy/oscar/session.py in fetch(self, *tileables, **kwargs)
1686 def fetch(self, *tileables, **kwargs) -> list:
1687 coro = _fetch(*tileables, session=self._isolated_session, **kwargs)
-> 1688 return asyncio.run_coroutine_threadsafe(coro, self._loop).result()
1689
1690 @implements(AbstractSyncSession.fetch_infos)
~/miniconda3/lib/python3.7/concurrent/futures/_base.py in result(self, timeout)
433 raise CancelledError()
434 elif self._state == FINISHED:
--> 435 return self.__get_result()
436 else:
437 raise TimeoutError()
~/miniconda3/lib/python3.7/concurrent/futures/_base.py in __get_result(self)
382 def __get_result(self):
383 if self._exception:
--> 384 raise self._exception
385 else:
386 return self._result
~/miniconda3/lib/python3.7/site-packages/mars/deploy/oscar/session.py in _fetch(tileable, session, *tileables, **kwargs)
1875 tileable, tileables = tileable[0], tileable[1:]
1876 session = _get_isolated_session(session)
-> 1877 data = await session.fetch(tileable, *tileables, **kwargs)
1878 return data[0] if len(tileables) == 0 else data
1879
~/miniconda3/lib/python3.7/site-packages/mars/deploy/oscar/session.py in fetch(self, *tileables, **kwargs)
1132 ):
1133 await fetcher.append(chunk.key, meta, fetch_info.indexes)
-> 1134 fetched_data = await fetcher.get()
1135 for fetch_info, data in zip(
1136 itertools.chain(*fetch_infos_list), fetched_data
~/miniconda3/lib/python3.7/site-packages/mars/services/task/execution/mars/fetcher.py in get(self)
51 gets = self._storage_api_to_gets[storage_api]
52 fetched_data = await storage_api.get.batch(
---> 53 *map(operator.itemgetter(0), gets)
54 )
55 for get, data in zip(gets, fetched_data):
~/miniconda3/lib/python3.7/site-packages/mars/oscar/batch.py in _async_batch(self, *delays)
151 elif self.batch_func:
152 args_list, kwargs_list = self._gen_args_kwargs_list(delays)
--> 153 return await self.batch_func(args_list, kwargs_list)
154 else:
155 # this function has no batch implementation
~/miniconda3/lib/python3.7/site-packages/mars/services/storage/api/web.py in get_batch(self, args_list, kwargs_list)
141 path=path,
142 method="POST",
--> 143 data=serialize_serializable(get_chunks),
144 )
145 return deserialize_serializable(res.body)
~/miniconda3/lib/python3.7/site-packages/mars/services/web/core.py in _request_url(self, method, path, wrap_timeout_exception, **kwargs)
233 if self.request_rewriter:
234 request = self.request_rewriter(request)
--> 235 res = await self._client.fetch(request, raise_error=False)
236 except HTTPTimeoutError as ex:
237 if wrap_timeout_exception:
~/miniconda3/lib/python3.7/site-packages/tornado/simple_httpclient.py in on_connection_close(self)
582 raise self.stream.error
583 try:
--> 584 raise HTTPStreamClosedError(message)
585 except HTTPStreamClosedError:
586 self._handle_exception(*sys.exc_info())
HTTPStreamClosedError: Connection closed
```
5. Minimized code to reproduce the error.
```python
mt.random.RandomState(0).rand(2000_0000, 5, chunk_size=100_0000).execute().fetch()
```
**Expected behavior**
A clear and concise description of what you expected to happen.
**Additional context**
Add any other context about the problem here.
Contributor guide
Research direction
Reproduce the failure with the minimized Python example, then trace the fetch path through mars/deploy/oscar/session.py, mars/services/task/execution/mars/fetcher.py, mars/services/storage/api/web.py, and mars/services/web/core.py. Check where the large request triggers Tornado's “Content-Length too long” error; done means large data can be fetched without the connection closing or raising that exception.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100