Repository: rucio/rucio
Issue: Disabling source_rse_id filling on queue_requests

Motivation
----------
The `source_rse_id` filling when queuing requests takes an unreasonable amount of time. 

Modification
------------
Disable the functionality until a better way can be found to do that.


Changed files:
lib/rucio/core/request.py, lib/rucio/tests/test_request.py, lib/rucio/client/uploadclient.py

diff --git a/lib/rucio/client/uploadclient.py b/lib/rucio/client/uploadclient.py
index a44d531e5..d2f4f9a33 100644
--- a/lib/rucio/client/uploadclient.py
+++ b/lib/rucio/client/uploadclient.py
@@ -37,7 +37,7 @@ from rucio.client.client import Client
 from rucio.common.exception import (RucioException, RSEBlacklisted, DataIdentifierAlreadyExists,
                                     DataIdentifierNotFound, NoFilesUploaded, NotAllFilesUploaded,
                                     ResourceTemporaryUnavailable, ServiceUnavailable, InputValidationError)
-from rucio.common.utils import adler32, execute, generate_uuid, md5, send_trace, GLOBALLY_SUPPORTED_CHECKSUMS
+from rucio.common.utils import adler32, detect_client_location, execute, generate_uuid, md5, send_trace, GLOBALLY_SUPPORTED_CHECKSUMS
 from rucio.rse import rsemanager as rsemgr
 from rucio import version
 
@@ -57,6 +57,7 @@ class UploadClient:
 
         self.logger = logger
         self.client = _client if _client else Client()
+        self.client_location = detect_client_location()
         self.tracing = tracing
         if not self.tracing:
             logger.debug('Tracing is turned off.')
@@ -196,7 +197,20 @@ class UploadClient:
                     logger.info('File already exists on RSE. Skipping upload')
                     trace['stateReason'] = 'File already exists'
                     continue
-            protocols = rsemgr.get_protocols_ordered(rse_settings=rse_settings, operation='write', scheme=force_scheme)
+
+            # resolving local area networks
+            domain = 'wan'
+            rse_attributes = {}
+            try:
+                rse_attributes = self.client.list_rse_attributes(rse)
+            except:
+                logger.warning('Attributes of the RSE: %s not available.' % rse)
+            if (self.client_location and 'lan' in rse_settings['domain'] and 'site' in rse_attributes):
+                if self.client_location['site'] == rse_attributes['site']:
+                    domain = 'lan'
+
+            # protocol handling and upload
+            protocols = rsemgr.get_protocols_ordered(rse_settings=rse_settings, operation='write', scheme=force_scheme, domain=domain)
             protocols.reverse()
             success = False
             state_reason = ''
diff --git a/lib/rucio/core/request.py b/lib/rucio/core/request.py
index 02ef5f52d..2ddbfc191 100644
--- a/lib/rucio/core/request.py
+++ b/lib/rucio/core/request.py
@@ -28,7 +28,7 @@ from six import string_types
 
 from sqlalchemy import and_, or_, func, update
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.sql import tuple_
+# from sqlalchemy.sql import tuple_
 from sqlalchemy.sql.expression import asc, bindparam, text, false, true
 
 from rucio.common.exception import RequestNotFound, RucioException, UnsupportedOperation, ConfigNotFound
@@ -156,27 +156,28 @@ def queue_requests(requests, session=None):
             for request in query_existing_requests:
                 existing_requests.append(request)
 
+    # Temporary disabled
     source_rses = {}
-    request_scopes_names = [(request['scope'], request['name']) for request in requests]
-    for chunked_requests in chunks(request_scopes_names, 50):
-        results = session.query(models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.rse_id, models.Distance.dest_rse_id, models.Distance.ranking)\\
-                         .filter(tuple_(models.RSEFileAssociation.scope, models.RSEFileAssociation.name).in_(chunked_requests))\\
-                         .join(models.Distance, models.Distance.src_rse_id == models.RSEFileAssociation.rse_id)\\
-                         .all()
-        for result in results:
-            scope = result[0]
-            name = result[1]
-            src_rse_id = result[2]
-            dest_rse_id = result[3]
-            distance = result[4]
-            if scope not in source_rses:
-                source_rses[scope] = {}
-            if name not in source_rses[scope]:
-                source_rses[scope][name] = {}
-            if dest_rse_id not in source_rses[scope][name]:
-                source_rses[scope][name][dest_rse_id] = {}
-            if src_rse_id not in source_rses[scope][name][dest_rse_id]:
-                source_rses[scope][name][dest_rse_id][src_rse_id] = distance
+    # request_scopes_names = [(request['scope'], request['name']) for request in requests]
+    # for chunked_requests in chunks(request_scopes_names, 50):
+    #    results = session.query(models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.rse_id, models.Distance.dest_rse_id, models.Distance.ranking)\\
+    #                     .filter(tuple_(models.RSEFileAssociation.scope, models.RSEFileAssociation.name).in_(chunked_requests))\\
+    #                     .join(models.Distance, models.Distance.src_rse_id == models.RSEFileAssociation.rse_id)\\
+    #                     .all()
+    #    for result in results:
+    #        scope = result[0]
+    #        name = result[1]
+    #        src_rse_id = result[2]
+    #        dest_rse_id = result[3]
+    #        distance = result[4]
+    #        if scope not in source_rses:
+    #            source_rses[scope] = {}
+    #        if name not in source_rses[scope]:
+    #            source_rses[scope][name] = {}
+    #        if dest_rse_id not in source_rses[scope][name]:
+    #            source_rses[scope][name][dest_rse_id] = {}
+    #        if src_rse_id not in source_rses[scope][name][dest_rse_id]:
+    #            source_rses[scope][name][dest_rse_id][src_rse_id] = distance
 
     try:
         throttler_mode = get('throttler', 'mode', default=None, use_cache=False, session=session)
diff --git a/lib/rucio/tests/test_request.py b/lib/rucio/tests/test_request.py
index ca4e1babe..32c141475 100644
--- a/lib/rucio/tests/test_request.py
+++ b/lib/rucio/tests/test_request.py
@@ -25,7 +25,7 @@ from rucio.common.types import InternalAccount, InternalScope
 from rucio.common.utils import generate_uuid, parse_response
 from rucio.core.config import set as config_set
 from rucio.core.did import attach_dids, add_did
-from rucio.core.distance import add_distance
+# from rucio.core.distance import add_distance
 from rucio.core.replica import add_replica
 from rucio.core.request import release_all_waiting_requests, queue_requests, get_request_by_did, release_waiting_requests_per_free_volume,\\
     release_waiting_requests_grouped_fifo, release_waiting_requests_fifo, list_requests, release_waiting_requests_per_deadline
@@ -136,38 +136,38 @@ class TestRequestCoreQueue(object):
         request = get_request_by_did(self.scope, name3, self.dest_rse_id2, session=self.db_session)
         assert_equal(request['state'], constants.RequestState.QUEUED)
 
-    def test_queue_requests_source_rse(self):
-        """ REQUEST (CORE): queue requests and select correct source RSE. """
-        # test correct selection of source RSE
-        name = generate_uuid()
-        size = 8
-        add_replica(self.source_rse_id, self.scope, name, size, self.account, session=self.db_session)
-        add_replica(self.source_rse_id2, self.scope, name, size, self.account, session=self.db_session)
-        add_distance(self.source_rse_id, self.dest_rse_id, 1, session=self.db_session)
-        add_distance(self.source_rse_id2, self.dest_rse_id, 2, session=self.db_session)
-        requests = [{
-            'dest_rse_id': self.dest_rse_id,
-            'request_type': constants.RequestType.TRANSFER,
-            'request_id': generate_uuid(),
-            'name': name,
-            'scope': self.scope,
-            'rule_id': generate_uuid(),
-            'retry_count': 1,
-            'requested_at': datetime.now().replace(year=2015),
-            'attributes': {
-                'activity': self.user_activity,
-                'bytes': size,
-                'md5': '',
-                'adler32': ''
-            }
-        }]
-        queue_requests(requests, session=self.db_session)
-        request = get_request_by_did(self.scope, name, self.dest_rse_id, session=self.db_session)
-        # select source RSE with smallest distance
-        assert_equal(request['source_rse_id'], self.source_rse_id)
-        assert_equal(request['name'], name)
-        assert_equal(request['scope'], self.scope)
-        assert_equal(request['state'], constants.RequestState.QUEUED)
+    # def test_queue_requests_source_rse(self):
+    #     """ REQUEST (CORE): queue requests and select correct source RSE. """
+    #     # test correct selection of source RSE
+    #     name = generate_uuid()
+    #     size = 8
+    #     add_replica(self.source_rse_id, self.scope, name, size, self.account, session=self.db_session)
+    #     add_replica(self.source_rse_id2, self.scope, name, size, self.account, session=self.db_session)
+    #     add_distance(self.source_rse_id, self.dest_rse_id, 1, session=self.db_session)
+    #     add_distance(self.source_rse_id2, self.dest_rse_id, 2, session=self.db_session)
+    #     requests = [{
+    #         'dest_rse_id': self.dest_rse_id,
+    #         'request_type': constants.RequestType.TRANSFER,
+    #         'request_id': generate_uuid(),
+    #         'name': name,
+    #         'scope': self.scope,
+    #         'rule_id': generate_uuid(),
+    #         'retry_count': 1,
+    #         'requested_at': datetime.now().replace(year=2015),
+    #         'attributes': {
+    #             'activity': self.user_activity,
+    #             'bytes': size,
+    #             'md5': '',
+    #             'adler32': ''
+    #         }
+    #     }]
+    #     queue_requests(requests, session=self.db_session)
+    #     request = get_request_by_did(self.scope, name, self.dest_rse_id, session=self.db_session)
+    #     # select source RSE with smallest distance
+    #     assert_equal(request['source_rse_id'], self.source_rse_id)
+    #     assert_equal(request['name'], name)
+    #     assert_equal(request['scope'], self.scope)
+    #     assert_equal(request['state'], constants.RequestState.QUEUED)
 
     def test_queue_requests_state_1(self):
         """ REQUEST (CORE): queue requests and set correct request state. """