We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Improve Dask performance by tuning `Client` parameters

### Describe the problem
<!-- Describe the problem clearly here. -->
Dask generally performs well, but it could be improved. I played around with changing `n_workers` as a parameter to `Client` and I found that the default does not take full advantage of the hardware in many cases.

### Source code / logs

To update here: 
https://github.com/modin-project/modin/blob/898889c8ac8804b7eed4cb3d733ee0d1e29686a8/modin/pandas/__init__.py#L227
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `modin/pandas/__init__.py`
Content:
```
1 import pandas
2 
3 __pandas_version__ = "0.25.3"
4 
5 if pandas.__version__ != __pandas_version__:
6     raise ImportError(
7         "The pandas version installed does not match the required pandas "
8         "version in Modin. Please install pandas {} to use "
9         "Modin.".format(__pandas_version__)
10     )
11 
12 from pandas import (
13     eval,
14     unique,
15     value_counts,
16     cut,
17     to_numeric,
18     factorize,
19     test,
20     qcut,
21     date_range,
22     period_range,
23     Index,
24     MultiIndex,
25     CategoricalIndex,
26     bdate_range,
27     DatetimeIndex,
28     Timedelta,
29     Timestamp,
30     to_timedelta,
31     set_eng_float_format,
32     options,
33     set_option,
34     NaT,
35     PeriodIndex,
36     Categorical,
37     Interval,
38     UInt8Dtype,
39     UInt16Dtype,
40     UInt32Dtype,
41     UInt64Dtype,
42     SparseDtype,
43     Int8Dtype,
44     Int16Dtype,
45     Int32Dtype,
46     Int64Dtype,
47     CategoricalDtype,
48     DatetimeTZDtype,
49     IntervalDtype,
50     PeriodDtype,
51     RangeIndex,
52     Int64Index,
53     UInt64Index,
54     Float64Index,
55     TimedeltaIndex,
56     IntervalIndex,
57     IndexSlice,
58     Grouper,
59     array,
60     Period,
61     show_versions,
62     DateOffset,
63     timedelta_range,
64     infer_freq,
65     interval_range,
66     ExcelWriter,
67     SparseArray,
68     SparseSeries,
69     SparseDataFrame,
70     datetime,
71     NamedAgg,
72 )
73 import threading
74 import os
75 import types
76 import sys
77 
78 from .. import __version__
79 from .concat import concat
80 from .dataframe import DataFrame
81 from .datetimes import to_datetime
82 from .io import (
83     read_csv,
84     read_parquet,
85     read_json,
86     read_html,
87     read_clipboard,
88     read_excel,
89     read_hdf,
90     read_feather,
91     read_msgpack,
92     read_stata,
93     read_sas,
94     read_pickle,
95     read_sql,
96     read_gbq,
97     read_table,
98     read_fwf,
99     read_sql_table,
100     read_sql_query,
101     read_spss,
102     ExcelFile,
103     to_pickle,
104     HDFStore,
105 )
106 from .reshape import get_dummies, melt, crosstab, lreshape, wide_to_long
107 from .series import Series
108 from .general import (
109     isna,
110     isnull,
111     merge,
112     merge_asof,
113     merge_ordered,
114     pivot_table,
115     notnull,
116     notna,
117     pivot,
118 )
119 from .plotting import Plotting as plotting
120 from .. import __execution_engine__ as execution_engine
121 
122 # Set this so that Pandas doesn't try to multithread by itself
123 os.environ["OMP_NUM_THREADS"] = "1"
124 num_cpus = 1
125 
126 
127 def initialize_ray():
128     import ray
129 
130     """Initializes ray based on environment variables and internal defaults."""
131     if threading.current_thread().name == "MainThread":
132         plasma_directory = None
133         cluster = os.environ.get("MODIN_RAY_CLUSTER", None)
134         redis_address = os.environ.get("MODIN_REDIS_ADDRESS", None)
135         if cluster == "True" and redis_address is not None:
136             # We only start ray in a cluster setting for the head node.
137             ray.init(
138                 include_webui=False,
139                 ignore_reinit_error=True,
140                 redis_address=redis_address,
141                 logging_level=100,
142             )
143         elif cluster is None:
144             object_store_memory = os.environ.get("MODIN_MEMORY", None)
145             if os.environ.get("MODIN_OUT_OF_CORE", "False").title() == "True":
146                 from tempfile import gettempdir
147 
148                 plasma_directory = gettempdir()
149                 # We may have already set the memory from the environment variable, we don't
150                 # want to overwrite that value if we have.
151                 if object_store_memory is None:
152                     # Round down to the nearest Gigabyte.
153                     mem_bytes = ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9
154                     # Default to 8x memory for out of core
155                     object_store_memory = 8 * mem_bytes
156             # In case anything failed above, we can still improve the memory for Modin.
157             if object_store_memory is None:
158                 # Round down to the nearest Gigabyte.
159                 object_store_memory = int(
160                     0.6 * ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9
161                 )
162                 # If the memory pool is smaller than 2GB, just use the default in ray.
163                 if object_store_memory == 0:
164                     object_store_memory = None
165             else:
166                 object_store_memory = int(object_store_memory)
167             ray.init(
168                 include_webui=False,
169                 ignore_reinit_error=True,
170                 plasma_directory=plasma_directory,
171                 object_store_memory=object_store_memory,
172                 redis_address=redis_address,
173                 logging_level=100,
174                 memory=object_store_memory,
175             )
176         # Register custom serializer for method objects to avoid warning message.
177         # We serialize `MethodType` objects when we use AxisPartition operations.
178         ray.register_custom_serializer(types.MethodType, use_pickle=True)
179 
180         # Register a fix import function to run on all_workers including the driver.
181         # This is a hack solution to fix #647, #746
182         def move_stdlib_ahead_of_site_packages(*args):
183             site_packages_path = None
184             site_packages_path_index = -1
185             for i, path in enumerate(sys.path):
186                 if sys.exec_prefix in path and path.endswith("site-packages"):
187                     site_packages_path = path
188                     site_packages_path_index = i
189                     # break on first found
190                     break
191 
192             if site_packages_path is not None:
193                 # stdlib packages layout as follows:
194                 # - python3.x
195                 #   - typing.py
196                 #   - site-packages/
197                 #     - pandas
198                 # So extracting the dirname of the site_packages can point us
199                 # to the directory containing standard libraries.
200                 sys.path.insert(
201                     site_packages_path_index, os.path.dirname(site_packages_path)
202                 )
203 
204         move_stdlib_ahead_of_site_packages()
205         ray.worker.global_worker.run_function_on_all_workers(
206             move_stdlib_ahead_of_site_packages
207         )
208 
209 
210 if execution_engine == "Ray":
211     import ray
212 
213     initialize_ray()
214     num_cpus = ray.cluster_resources()["CPU"]
215 elif execution_engine == "Dask":  # pragma: no cover
216     from distributed.client import _get_global_client
217     import warnings
218 
219     warnings.warn("The Dask Engine for Modin is experimental.")
220 
221     if threading.current_thread().name == "MainThread":
222         # initialize the dask client
223         client = _get_global_client()
224         if client is None:
225             from distributed import Client
226 
227             client = Client()
228         num_cpus = sum(client.ncores().values())
229 elif execution_engine != "Python":
230     raise ImportError("Unrecognized execution engine: {}.".format(execution_engine))
231 
232 DEFAULT_NPARTITIONS = max(4, int(num_cpus))
233 
234 __all__ = [
235     "DataFrame",
236     "Series",
237     "read_csv",
238     "read_parquet",
239     "read_json",
240     "read_html",
241     "read_clipboard",
242     "read_excel",
243     "read_hdf",
244     "read_feather",
245     "read_msgpack",
246     "read_stata",
247     "read_sas",
248     "read_pickle",
249     "read_sql",
250     "read_gbq",
251     "read_table",
252     "read_spss",
253     "concat",
254     "eval",
255     "unique",
256     "value_counts",
257     "cut",
258     "to_numeric",
259     "factorize",
260     "test",
261     "qcut",
262     "to_datetime",
263     "get_dummies",
264     "isna",
265     "isnull",
266     "merge",
267     "pivot_table",
268     "date_range",
269     "Index",
270     "MultiIndex",
271     "Series",
272     "bdate_range",
273     "period_range",
274     "DatetimeIndex",
275     "to_timedelta",
276     "set_eng_float_format",
277     "options",
278     "set_option",
279     "CategoricalIndex",
280     "Timedelta",
281     "Timestamp",
282     "NaT",
283     "PeriodIndex",
284     "Categorical",
285     "__version__",
286     "melt",
287     "crosstab",
288     "plotting",
289     "Interval",
290     "UInt8Dtype",
291     "UInt16Dtype",
292     "UInt32Dtype",
293     "UInt64Dtype",
294     "SparseDtype",
295     "Int8Dtype",
296     "Int16Dtype",
297     "Int32Dtype",
298     "Int64Dtype",
299     "CategoricalDtype",
300     "DatetimeTZDtype",
301     "IntervalDtype",
302     "PeriodDtype",
303     "RangeIndex",
304     "Int64Index",
305     "UInt64Index",
306     "Float64Index",
307     "TimedeltaIndex",
308     "IntervalIndex",
309     "IndexSlice",
310     "Grouper",
311     "array",
312     "Period",
313     "show_versions",
314     "DateOffset",
315     "timedelta_range",
316     "infer_freq",
317     "interval_range",
318     "ExcelWriter",
319     "read_fwf",
320     "read_sql_table",
321     "read_sql_query",
322     "ExcelFile",
323     "to_pickle",
324     "HDFStore",
325     "lreshape",
326     "wide_to_long",
327     "merge_asof",
328     "merge_ordered",
329     "notnull",
330     "notna",
331     "pivot",
332     "SparseArray",
333     "SparseSeries",
334     "SparseDataFrame",
335     "datetime",
336     "NamedAgg",
337     "DEFAULT_NPARTITIONS",
338 ]
339 
340 del pandas
341
```
--- 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/modin/pandas/__init__.py b/modin/pandas/__init__.py
--- a/modin/pandas/__init__.py
+++ b/modin/pandas/__init__.py
@@ -223,9 +223,10 @@
         client = _get_global_client()
         if client is None:
             from distributed import Client
+            import multiprocessing
 
-            client = Client()
-        num_cpus = sum(client.ncores().values())
+            num_cpus = multiprocessing.cpu_count()
+            client = Client(n_workers=num_cpus)
 elif execution_engine != "Python":
     raise ImportError("Unrecognized execution engine: {}.".format(execution_engine))

{"golden_diff": "diff --git a/modin/pandas/__init__.py b/modin/pandas/__init__.py\n--- a/modin/pandas/__init__.py\n+++ b/modin/pandas/__init__.py\n@@ -223,9 +223,10 @@\n         client = _get_global_client()\n         if client is None:\n             from distributed import Client\n+            import multiprocessing\n \n-            client = Client()\n-        num_cpus = sum(client.ncores().values())\n+            num_cpus = multiprocessing.cpu_count()\n+            client = Client(n_workers=num_cpus)\n elif execution_engine != \"Python\":\n     raise ImportError(\"Unrecognized execution engine: {}.\".format(execution_engine))\n", "issue": "Improve Dask performance by tuning `Client` parameters\n\r\n### Describe the problem\r\n<!-- Describe the problem clearly here. -->\r\nDask generally performs well, but it could be improved. I played around with changing `n_workers` as a parameter to `Client` and I found that the default does not take full advantage of the hardware in many cases.\r\n\r\n### Source code / logs\r\n\r\nTo update here: \r\nhttps://github.com/modin-project/modin/blob/898889c8ac8804b7eed4cb3d733ee0d1e29686a8/modin/pandas/__init__.py#L227\r\n<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->\r\n\n", "before_files": [{"content": "import pandas\n\n__pandas_version__ = \"0.25.3\"\n\nif pandas.__version__ != __pandas_version__:\n    raise ImportError(\n        \"The pandas version installed does not match the required pandas \"\n        \"version in Modin. Please install pandas {} to use \"\n        \"Modin.\".format(__pandas_version__)\n    )\n\nfrom pandas import (\n    eval,\n    unique,\n    value_counts,\n    cut,\n    to_numeric,\n    factorize,\n    test,\n    qcut,\n    date_range,\n    period_range,\n    Index,\n    MultiIndex,\n    CategoricalIndex,\n    bdate_range,\n    DatetimeIndex,\n    Timedelta,\n    Timestamp,\n    to_timedelta,\n    set_eng_float_format,\n    options,\n    set_option,\n    NaT,\n    PeriodIndex,\n    Categorical,\n    Interval,\n    UInt8Dtype,\n    UInt16Dtype,\n    UInt32Dtype,\n    UInt64Dtype,\n    SparseDtype,\n    Int8Dtype,\n    Int16Dtype,\n    Int32Dtype,\n    Int64Dtype,\n    CategoricalDtype,\n    DatetimeTZDtype,\n    IntervalDtype,\n    PeriodDtype,\n    RangeIndex,\n    Int64Index,\n    UInt64Index,\n    Float64Index,\n    TimedeltaIndex,\n    IntervalIndex,\n    IndexSlice,\n    Grouper,\n    array,\n    Period,\n    show_versions,\n    DateOffset,\n    timedelta_range,\n    infer_freq,\n    interval_range,\n    ExcelWriter,\n    SparseArray,\n    SparseSeries,\n    SparseDataFrame,\n    datetime,\n    NamedAgg,\n)\nimport threading\nimport os\nimport types\nimport sys\n\nfrom .. import __version__\nfrom .concat import concat\nfrom .dataframe import DataFrame\nfrom .datetimes import to_datetime\nfrom .io import (\n    read_csv,\n    read_parquet,\n    read_json,\n    read_html,\n    read_clipboard,\n    read_excel,\n    read_hdf,\n    read_feather,\n    read_msgpack,\n    read_stata,\n    read_sas,\n    read_pickle,\n    read_sql,\n    read_gbq,\n    read_table,\n    read_fwf,\n    read_sql_table,\n    read_sql_query,\n    read_spss,\n    ExcelFile,\n    to_pickle,\n    HDFStore,\n)\nfrom .reshape import get_dummies, melt, crosstab, lreshape, wide_to_long\nfrom .series import Series\nfrom .general import (\n    isna,\n    isnull,\n    merge,\n    merge_asof,\n    merge_ordered,\n    pivot_table,\n    notnull,\n    notna,\n    pivot,\n)\nfrom .plotting import Plotting as plotting\nfrom .. import __execution_engine__ as execution_engine\n\n# Set this so that Pandas doesn't try to multithread by itself\nos.environ[\"OMP_NUM_THREADS\"] = \"1\"\nnum_cpus = 1\n\n\ndef initialize_ray():\n    import ray\n\n    \"\"\"Initializes ray based on environment variables and internal defaults.\"\"\"\n    if threading.current_thread().name == \"MainThread\":\n        plasma_directory = None\n        cluster = os.environ.get(\"MODIN_RAY_CLUSTER\", None)\n        redis_address = os.environ.get(\"MODIN_REDIS_ADDRESS\", None)\n        if cluster == \"True\" and redis_address is not None:\n            # We only start ray in a cluster setting for the head node.\n            ray.init(\n                include_webui=False,\n                ignore_reinit_error=True,\n                redis_address=redis_address,\n                logging_level=100,\n            )\n        elif cluster is None:\n            object_store_memory = os.environ.get(\"MODIN_MEMORY\", None)\n            if os.environ.get(\"MODIN_OUT_OF_CORE\", \"False\").title() == \"True\":\n                from tempfile import gettempdir\n\n                plasma_directory = gettempdir()\n                # We may have already set the memory from the environment variable, we don't\n                # want to overwrite that value if we have.\n                if object_store_memory is None:\n                    # Round down to the nearest Gigabyte.\n                    mem_bytes = ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9\n                    # Default to 8x memory for out of core\n                    object_store_memory = 8 * mem_bytes\n            # In case anything failed above, we can still improve the memory for Modin.\n            if object_store_memory is None:\n                # Round down to the nearest Gigabyte.\n                object_store_memory = int(\n                    0.6 * ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9\n                )\n                # If the memory pool is smaller than 2GB, just use the default in ray.\n                if object_store_memory == 0:\n                    object_store_memory = None\n            else:\n                object_store_memory = int(object_store_memory)\n            ray.init(\n                include_webui=False,\n                ignore_reinit_error=True,\n                plasma_directory=plasma_directory,\n                object_store_memory=object_store_memory,\n                redis_address=redis_address,\n                logging_level=100,\n                memory=object_store_memory,\n            )\n        # Register custom serializer for method objects to avoid warning message.\n        # We serialize `MethodType` objects when we use AxisPartition operations.\n        ray.register_custom_serializer(types.MethodType, use_pickle=True)\n\n        # Register a fix import function to run on all_workers including the driver.\n        # This is a hack solution to fix #647, #746\n        def move_stdlib_ahead_of_site_packages(*args):\n            site_packages_path = None\n            site_packages_path_index = -1\n            for i, path in enumerate(sys.path):\n                if sys.exec_prefix in path and path.endswith(\"site-packages\"):\n                    site_packages_path = path\n                    site_packages_path_index = i\n                    # break on first found\n                    break\n\n            if site_packages_path is not None:\n                # stdlib packages layout as follows:\n                # - python3.x\n                #   - typing.py\n                #   - site-packages/\n                #     - pandas\n                # So extracting the dirname of the site_packages can point us\n                # to the directory containing standard libraries.\n                sys.path.insert(\n                    site_packages_path_index, os.path.dirname(site_packages_path)\n                )\n\n        move_stdlib_ahead_of_site_packages()\n        ray.worker.global_worker.run_function_on_all_workers(\n            move_stdlib_ahead_of_site_packages\n        )\n\n\nif execution_engine == \"Ray\":\n    import ray\n\n    initialize_ray()\n    num_cpus = ray.cluster_resources()[\"CPU\"]\nelif execution_engine == \"Dask\":  # pragma: no cover\n    from distributed.client import _get_global_client\n    import warnings\n\n    warnings.warn(\"The Dask Engine for Modin is experimental.\")\n\n    if threading.current_thread().name == \"MainThread\":\n        # initialize the dask client\n        client = _get_global_client()\n        if client is None:\n            from distributed import Client\n\n            client = Client()\n        num_cpus = sum(client.ncores().values())\nelif execution_engine != \"Python\":\n    raise ImportError(\"Unrecognized execution engine: {}.\".format(execution_engine))\n\nDEFAULT_NPARTITIONS = max(4, int(num_cpus))\n\n__all__ = [\n    \"DataFrame\",\n    \"Series\",\n    \"read_csv\",\n    \"read_parquet\",\n    \"read_json\",\n    \"read_html\",\n    \"read_clipboard\",\n    \"read_excel\",\n    \"read_hdf\",\n    \"read_feather\",\n    \"read_msgpack\",\n    \"read_stata\",\n    \"read_sas\",\n    \"read_pickle\",\n    \"read_sql\",\n    \"read_gbq\",\n    \"read_table\",\n    \"read_spss\",\n    \"concat\",\n    \"eval\",\n    \"unique\",\n    \"value_counts\",\n    \"cut\",\n    \"to_numeric\",\n    \"factorize\",\n    \"test\",\n    \"qcut\",\n    \"to_datetime\",\n    \"get_dummies\",\n    \"isna\",\n    \"isnull\",\n    \"merge\",\n    \"pivot_table\",\n    \"date_range\",\n    \"Index\",\n    \"MultiIndex\",\n    \"Series\",\n    \"bdate_range\",\n    \"period_range\",\n    \"DatetimeIndex\",\n    \"to_timedelta\",\n    \"set_eng_float_format\",\n    \"options\",\n    \"set_option\",\n    \"CategoricalIndex\",\n    \"Timedelta\",\n    \"Timestamp\",\n    \"NaT\",\n    \"PeriodIndex\",\n    \"Categorical\",\n    \"__version__\",\n    \"melt\",\n    \"crosstab\",\n    \"plotting\",\n    \"Interval\",\n    \"UInt8Dtype\",\n    \"UInt16Dtype\",\n    \"UInt32Dtype\",\n    \"UInt64Dtype\",\n    \"SparseDtype\",\n    \"Int8Dtype\",\n    \"Int16Dtype\",\n    \"Int32Dtype\",\n    \"Int64Dtype\",\n    \"CategoricalDtype\",\n    \"DatetimeTZDtype\",\n    \"IntervalDtype\",\n    \"PeriodDtype\",\n    \"RangeIndex\",\n    \"Int64Index\",\n    \"UInt64Index\",\n    \"Float64Index\",\n    \"TimedeltaIndex\",\n    \"IntervalIndex\",\n    \"IndexSlice\",\n    \"Grouper\",\n    \"array\",\n    \"Period\",\n    \"show_versions\",\n    \"DateOffset\",\n    \"timedelta_range\",\n    \"infer_freq\",\n    \"interval_range\",\n    \"ExcelWriter\",\n    \"read_fwf\",\n    \"read_sql_table\",\n    \"read_sql_query\",\n    \"ExcelFile\",\n    \"to_pickle\",\n    \"HDFStore\",\n    \"lreshape\",\n    \"wide_to_long\",\n    \"merge_asof\",\n    \"merge_ordered\",\n    \"notnull\",\n    \"notna\",\n    \"pivot\",\n    \"SparseArray\",\n    \"SparseSeries\",\n    \"SparseDataFrame\",\n    \"datetime\",\n    \"NamedAgg\",\n    \"DEFAULT_NPARTITIONS\",\n]\n\ndel pandas\n", "path": "modin/pandas/__init__.py"}], "after_files": [{"content": "import pandas\n\n__pandas_version__ = \"0.25.3\"\n\nif pandas.__version__ != __pandas_version__:\n    raise ImportError(\n        \"The pandas version installed does not match the required pandas \"\n        \"version in Modin. Please install pandas {} to use \"\n        \"Modin.\".format(__pandas_version__)\n    )\n\nfrom pandas import (\n    eval,\n    unique,\n    value_counts,\n    cut,\n    to_numeric,\n    factorize,\n    test,\n    qcut,\n    date_range,\n    period_range,\n    Index,\n    MultiIndex,\n    CategoricalIndex,\n    bdate_range,\n    DatetimeIndex,\n    Timedelta,\n    Timestamp,\n    to_timedelta,\n    set_eng_float_format,\n    options,\n    set_option,\n    NaT,\n    PeriodIndex,\n    Categorical,\n    Interval,\n    UInt8Dtype,\n    UInt16Dtype,\n    UInt32Dtype,\n    UInt64Dtype,\n    SparseDtype,\n    Int8Dtype,\n    Int16Dtype,\n    Int32Dtype,\n    Int64Dtype,\n    CategoricalDtype,\n    DatetimeTZDtype,\n    IntervalDtype,\n    PeriodDtype,\n    RangeIndex,\n    Int64Index,\n    UInt64Index,\n    Float64Index,\n    TimedeltaIndex,\n    IntervalIndex,\n    IndexSlice,\n    Grouper,\n    array,\n    Period,\n    show_versions,\n    DateOffset,\n    timedelta_range,\n    infer_freq,\n    interval_range,\n    ExcelWriter,\n    SparseArray,\n    SparseSeries,\n    SparseDataFrame,\n    datetime,\n    NamedAgg,\n)\nimport threading\nimport os\nimport types\nimport sys\n\nfrom .. import __version__\nfrom .concat import concat\nfrom .dataframe import DataFrame\nfrom .datetimes import to_datetime\nfrom .io import (\n    read_csv,\n    read_parquet,\n    read_json,\n    read_html,\n    read_clipboard,\n    read_excel,\n    read_hdf,\n    read_feather,\n    read_msgpack,\n    read_stata,\n    read_sas,\n    read_pickle,\n    read_sql,\n    read_gbq,\n    read_table,\n    read_fwf,\n    read_sql_table,\n    read_sql_query,\n    read_spss,\n    ExcelFile,\n    to_pickle,\n    HDFStore,\n)\nfrom .reshape import get_dummies, melt, crosstab, lreshape, wide_to_long\nfrom .series import Series\nfrom .general import (\n    isna,\n    isnull,\n    merge,\n    merge_asof,\n    merge_ordered,\n    pivot_table,\n    notnull,\n    notna,\n    pivot,\n)\nfrom .plotting import Plotting as plotting\nfrom .. import __execution_engine__ as execution_engine\n\n# Set this so that Pandas doesn't try to multithread by itself\nos.environ[\"OMP_NUM_THREADS\"] = \"1\"\nnum_cpus = 1\n\n\ndef initialize_ray():\n    import ray\n\n    \"\"\"Initializes ray based on environment variables and internal defaults.\"\"\"\n    if threading.current_thread().name == \"MainThread\":\n        plasma_directory = None\n        cluster = os.environ.get(\"MODIN_RAY_CLUSTER\", None)\n        redis_address = os.environ.get(\"MODIN_REDIS_ADDRESS\", None)\n        if cluster == \"True\" and redis_address is not None:\n            # We only start ray in a cluster setting for the head node.\n            ray.init(\n                include_webui=False,\n                ignore_reinit_error=True,\n                redis_address=redis_address,\n                logging_level=100,\n            )\n        elif cluster is None:\n            object_store_memory = os.environ.get(\"MODIN_MEMORY\", None)\n            if os.environ.get(\"MODIN_OUT_OF_CORE\", \"False\").title() == \"True\":\n                from tempfile import gettempdir\n\n                plasma_directory = gettempdir()\n                # We may have already set the memory from the environment variable, we don't\n                # want to overwrite that value if we have.\n                if object_store_memory is None:\n                    # Round down to the nearest Gigabyte.\n                    mem_bytes = ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9\n                    # Default to 8x memory for out of core\n                    object_store_memory = 8 * mem_bytes\n            # In case anything failed above, we can still improve the memory for Modin.\n            if object_store_memory is None:\n                # Round down to the nearest Gigabyte.\n                object_store_memory = int(\n                    0.6 * ray.utils.get_system_memory() // 10 ** 9 * 10 ** 9\n                )\n                # If the memory pool is smaller than 2GB, just use the default in ray.\n                if object_store_memory == 0:\n                    object_store_memory = None\n            else:\n                object_store_memory = int(object_store_memory)\n            ray.init(\n                include_webui=False,\n                ignore_reinit_error=True,\n                plasma_directory=plasma_directory,\n                object_store_memory=object_store_memory,\n                redis_address=redis_address,\n                logging_level=100,\n                memory=object_store_memory,\n            )\n        # Register custom serializer for method objects to avoid warning message.\n        # We serialize `MethodType` objects when we use AxisPartition operations.\n        ray.register_custom_serializer(types.MethodType, use_pickle=True)\n\n        # Register a fix import function to run on all_workers including the driver.\n        # This is a hack solution to fix #647, #746\n        def move_stdlib_ahead_of_site_packages(*args):\n            site_packages_path = None\n            site_packages_path_index = -1\n            for i, path in enumerate(sys.path):\n                if sys.exec_prefix in path and path.endswith(\"site-packages\"):\n                    site_packages_path = path\n                    site_packages_path_index = i\n                    # break on first found\n                    break\n\n            if site_packages_path is not None:\n                # stdlib packages layout as follows:\n                # - python3.x\n                #   - typing.py\n                #   - site-packages/\n                #     - pandas\n                # So extracting the dirname of the site_packages can point us\n                # to the directory containing standard libraries.\n                sys.path.insert(\n                    site_packages_path_index, os.path.dirname(site_packages_path)\n                )\n\n        move_stdlib_ahead_of_site_packages()\n        ray.worker.global_worker.run_function_on_all_workers(\n            move_stdlib_ahead_of_site_packages\n        )\n\n\nif execution_engine == \"Ray\":\n    import ray\n\n    initialize_ray()\n    num_cpus = ray.cluster_resources()[\"CPU\"]\nelif execution_engine == \"Dask\":  # pragma: no cover\n    from distributed.client import _get_global_client\n    import warnings\n\n    warnings.warn(\"The Dask Engine for Modin is experimental.\")\n\n    if threading.current_thread().name == \"MainThread\":\n        # initialize the dask client\n        client = _get_global_client()\n        if client is None:\n            from distributed import Client\n            import multiprocessing\n\n            num_cpus = multiprocessing.cpu_count()\n            client = Client(n_workers=num_cpus)\nelif execution_engine != \"Python\":\n    raise ImportError(\"Unrecognized execution engine: {}.\".format(execution_engine))\n\nDEFAULT_NPARTITIONS = max(4, int(num_cpus))\n\n__all__ = [\n    \"DataFrame\",\n    \"Series\",\n    \"read_csv\",\n    \"read_parquet\",\n    \"read_json\",\n    \"read_html\",\n    \"read_clipboard\",\n    \"read_excel\",\n    \"read_hdf\",\n    \"read_feather\",\n    \"read_msgpack\",\n    \"read_stata\",\n    \"read_sas\",\n    \"read_pickle\",\n    \"read_sql\",\n    \"read_gbq\",\n    \"read_table\",\n    \"read_spss\",\n    \"concat\",\n    \"eval\",\n    \"unique\",\n    \"value_counts\",\n    \"cut\",\n    \"to_numeric\",\n    \"factorize\",\n    \"test\",\n    \"qcut\",\n    \"to_datetime\",\n    \"get_dummies\",\n    \"isna\",\n    \"isnull\",\n    \"merge\",\n    \"pivot_table\",\n    \"date_range\",\n    \"Index\",\n    \"MultiIndex\",\n    \"Series\",\n    \"bdate_range\",\n    \"period_range\",\n    \"DatetimeIndex\",\n    \"to_timedelta\",\n    \"set_eng_float_format\",\n    \"options\",\n    \"set_option\",\n    \"CategoricalIndex\",\n    \"Timedelta\",\n    \"Timestamp\",\n    \"NaT\",\n    \"PeriodIndex\",\n    \"Categorical\",\n    \"__version__\",\n    \"melt\",\n    \"crosstab\",\n    \"plotting\",\n    \"Interval\",\n    \"UInt8Dtype\",\n    \"UInt16Dtype\",\n    \"UInt32Dtype\",\n    \"UInt64Dtype\",\n    \"SparseDtype\",\n    \"Int8Dtype\",\n    \"Int16Dtype\",\n    \"Int32Dtype\",\n    \"Int64Dtype\",\n    \"CategoricalDtype\",\n    \"DatetimeTZDtype\",\n    \"IntervalDtype\",\n    \"PeriodDtype\",\n    \"RangeIndex\",\n    \"Int64Index\",\n    \"UInt64Index\",\n    \"Float64Index\",\n    \"TimedeltaIndex\",\n    \"IntervalIndex\",\n    \"IndexSlice\",\n    \"Grouper\",\n    \"array\",\n    \"Period\",\n    \"show_versions\",\n    \"DateOffset\",\n    \"timedelta_range\",\n    \"infer_freq\",\n    \"interval_range\",\n    \"ExcelWriter\",\n    \"read_fwf\",\n    \"read_sql_table\",\n    \"read_sql_query\",\n    \"ExcelFile\",\n    \"to_pickle\",\n    \"HDFStore\",\n    \"lreshape\",\n    \"wide_to_long\",\n    \"merge_asof\",\n    \"merge_ordered\",\n    \"notnull\",\n    \"notna\",\n    \"pivot\",\n    \"SparseArray\",\n    \"SparseSeries\",\n    \"SparseDataFrame\",\n    \"datetime\",\n    \"NamedAgg\",\n    \"DEFAULT_NPARTITIONS\",\n]\n\ndel pandas\n", "path": "modin/pandas/__init__.py"}]}