mars-project / mars-project/mars

[BUG] Mars client can't fetch large data

Open
#3,122 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.