rm long-term connection support

Author: littleEast7

alluxiofs/client/config.py

@@ -19,7 +19,6 @@ def __init__(
         worker_data_port: int = ALLUXIO_WORKER_S3_SERVER_PORT_DEFAULT_VALUE,
         fallback_to_ufs_enabled=True,
         ufs_info_refresh_interval_minutes=2,
-        concurrency=64,
         log_level="INFO",
         log_dir=None,
         log_tag_allowlist=None,
@@ -73,10 +72,6 @@ def __init__(
             fallback_to_ufs_enabled, bool
         ), "'fallback_to_ufs_enabled' should be a boolean"
 
-        assert (
-            isinstance(concurrency, int) and concurrency > 0
-        ), "'concurrency' should be a positive integer"
-
         assert isinstance(log_level, str) and log_level in [
             "DEBUG",
             "INFO",
@@ -212,7 +207,6 @@ def __init__(
         self.worker_hosts = worker_hosts
         self.worker_http_port = worker_http_port
         self.worker_data_port = worker_data_port
-        self.concurrency = concurrency
         self.log_level = log_level
         self.log_dir = log_dir
         self.log_tag_allowlist = log_tag_allowlist

alluxiofs/client/core.py

@@ -165,7 +165,6 @@ def __init__(
         )
         self.logger = TagAdapter(base_logger, {"tag": "[ALLUXIO_CLIENT]"})
 
-        self.session = self._create_session(self.config.concurrency)
         self.data_manager = None
         if self.config.load_balance_domain:
             self.loadbalancer = None
@@ -298,7 +297,7 @@ def listdir(self, path):
         )
         params = {"path": path}
 
-        response = self.session.get(
+        response = requests.get(
             LIST_URL_FORMAT.format(
                 worker_host=worker_host, http_port=worker_http_port
             ),
@@ -364,7 +363,7 @@ def get_file_status(self, path):
         )
         params = {"path": path}
 
-        response = self.session.get(
+        response = requests.get(
             GET_FILE_STATUS_URL_FORMAT.format(
                 worker_host=worker_host,
                 http_port=worker_http_port,
@@ -441,7 +440,7 @@ def submit_load(
             "opType": OpType.SUBMIT.value,
             "verbose": json.dumps(verbose),
         }
-        response = self.session.get(
+        response = requests.get(
             LOAD_URL_FORMAT.format(
                 worker_host=worker_host,
                 http_port=worker_http_port,
@@ -470,7 +469,7 @@ def stop_load(
             path
         )
         params = {"path": path, "opType": OpType.STOP.value}
-        response = self.session.get(
+        response = requests.get(
             LOAD_URL_FORMAT.format(
                 worker_host=worker_host,
                 http_port=worker_http_port,
@@ -1534,7 +1533,7 @@ def _load_file(
             "opType": OpType.SUBMIT.value,
             "verbose": json.dumps(verbose),
         }
-        response = self.session.get(
+        response = requests.get(
             LOAD_URL_FORMAT.format(
                 worker_host=worker_host,
                 http_port=worker_http_port,
@@ -1583,7 +1582,7 @@ def _load_file(
     def _load_progress_internal(
         self, load_url: str, params: Dict
     ) -> (LoadState, str):
-        response = self.session.get(load_url, params=params)
+        response = requests.get(load_url, params=params)
         self._check_response(response)
         content = json.loads(response.content.decode("utf-8"))
         if "jobState" not in content:
@@ -1630,7 +1629,7 @@ def _read_page(
                 page_length=length,
             )
             self.logger.debug(f"Reading page request {page_url}")
-        response = self.session.get(page_url)
+        response = requests.get(page_url)
         self._check_response(response)
         return response.content
 
@@ -1701,7 +1700,7 @@ def _get_preferred_worker_address(self, full_ufs_path):
                 file_path=full_ufs_path,
             )
         try:
-            response = self.session.get(url)
+            response = requests.get(url)
             response.raise_for_status()
             data = json.loads(response.content)
             ip = data["Host"]
@@ -2202,7 +2201,7 @@ def _get_preferred_worker_host(self, full_ufs_path):
                 file_path=full_ufs_path,
             )
         try:
-            response = self.session.get(url)
+            response = requests.get(url)
             response.raise_for_status()
             data = json.loads(response.content)
             ip = data["Host"]
@@ -2230,7 +2229,7 @@ async def _request(
         data=None,
     ) -> Tuple[int, bytes]:
         await self._set_session()
-        async with self.session.request(
+        async with requests.request(
             method=method.value,
             url=url,
             params=params,

config_template.yaml

@@ -90,18 +90,6 @@ log_tag_allowlist: null
 # Concurrency Control Configuration
 # ----------------------------------------------------------------------------
 
-# concurrency: Maximum number of concurrent operations
-#   Purpose: Controls the maximum number of concurrent operations the client can execute simultaneously (e.g., concurrent reads, writes)
-#   Affected Areas:
-#     - HTTP request connection pool size (pool_connections and pool_maxsize)
-#     - Thread pool size for concurrent file operations
-#   Default Value: 64
-#   Tuning Recommendations:
-#     - High concurrency scenarios: Consider increasing (e.g., 128 or 256)
-#     - Resource-constrained scenarios: Consider decreasing (e.g., 32 or 16)
-#     - Should be adjusted based on actual network bandwidth and Worker node performance
-concurrency: 64
-
 # ----------------------------------------------------------------------------
 # Local Cache Configuration
 # ----------------------------------------------------------------------------