Repository: rucio/rucio
Issue: Preparer sleeping even if the backlog is not empty

Motivation
----------
When a request has no candidate sources, the preparer will skip them. It will then compare the number of handled (non-skipped) requests to the bulk size to decide if it must sleep or not. This effectively leads to preparer sleeping each time at least one of the requests has no sources even if the backlog is not empty. 


Modification
------------
Be more intelligent when deciding if we have to sleep.


Changed files:
lib/rucio/core/request.py, lib/rucio/daemons/conveyor/preparer.py, lib/rucio/tests/test_preparer.py

diff --git a/lib/rucio/core/request.py b/lib/rucio/core/request.py
index 411bcab69..2f2c3bdc7 100644
--- a/lib/rucio/core/request.py
+++ b/lib/rucio/core/request.py
@@ -19,7 +19,6 @@ import logging
 import traceback
 from collections import namedtuple
 from configparser import NoOptionError, NoSectionError
-from itertools import filterfalse
 from typing import TYPE_CHECKING
 
 from six import string_types
@@ -43,13 +42,13 @@ from rucio.db.sqla.util import create_id_temp_table
 RequestAndState = namedtuple('RequestAndState', ['request_id', 'request_state'])
 
 if TYPE_CHECKING:
-    from typing import Any, Dict, Iterable, Iterator, List, Optional, Callable, Set, Union
+    from typing import Any, Dict, Iterator, List, Optional, Callable, Set, Union, Sequence
     from sqlalchemy.orm import Session
 
     RequestResult = Dict[str, Any]
     RequestResultOrState = Union[RequestResult, RequestAndState]
     RowIterator = Iterator[RequestResult]
-    ReduceFunction = Callable[[RowIterator], RowIterator]
+    LoggerFunction = Callable[..., Any]
 
 """
 The core request.py is specifically for handling requests.
@@ -209,9 +208,9 @@ def queue_requests(requests, session=None, logger=logging.log):
     for req in requests:
 
         if isinstance(req['attributes'], string_types):
-            req['attributes'] = json.loads(req['attributes'])
+            req['attributes'] = json.loads(req['attributes'] or '{}')
             if isinstance(req['attributes'], string_types):
-                req['attributes'] = json.loads(req['attributes'])
+                req['attributes'] = json.loads(req['attributes'] or '{}')
 
         if req['request_type'] == RequestType.TRANSFER:
             request_clause.append(and_(models.Request.scope == req['scope'],
@@ -617,7 +616,7 @@ def get_next(request_type, state, limit=100, older_than=None, rse_id=None, activ
                     res_dict = dict(res)
                     res_dict.pop('_sa_instance_state')
                     res_dict['request_id'] = res_dict['id']
-                    res_dict['attributes'] = json.loads(str(res_dict['attributes']))
+                    res_dict['attributes'] = json.loads(str(res_dict['attributes'] or '{}'))
 
                     dst_id = res_dict['dest_rse_id']
                     src_id = res_dict['source_rse_id']
@@ -747,7 +746,7 @@ def get_request(request_id, session=None):
         else:
             tmp = dict(tmp)
             tmp.pop('_sa_instance_state')
-            tmp['attributes'] = json.loads(str(tmp['attributes']))
+            tmp['attributes'] = json.loads(str(tmp['attributes'] or '{}'))
             return tmp
     except IntegrityError as error:
         raise RucioException(error.args)
@@ -773,7 +772,7 @@ def get_requests_by_transfer(external_host, transfer_id, session=None):
                 t2 = dict(t)
                 t2.pop('_sa_instance_state')
                 t2['request_id'] = t2['id']
-                t2['attributes'] = json.loads(str(t2['attributes']))
+                t2['attributes'] = json.loads(str(t2['attributes'] or '{}'))
                 result.append(t2)
             return result
         return
@@ -813,7 +812,7 @@ def get_request_by_did(scope, name, rse_id, request_type=None, session=None):
 
             tmp['source_rse'] = get_rse_name(rse_id=tmp['source_rse_id'], session=session) if tmp['source_rse_id'] is not None else None
             tmp['dest_rse'] = get_rse_name(rse_id=tmp['dest_rse_id'], session=session) if tmp['dest_rse_id'] is not None else None
-            tmp['attributes'] = json.loads(str(tmp['attributes']))
+            tmp['attributes'] = json.loads(str(tmp['attributes'] or '{}'))
 
             return tmp
     except IntegrityError as error:
@@ -1688,33 +1687,61 @@ def list_requests_history(src_rse_ids, dst_rse_ids, states=[RequestState.WAITING
 
 
 @transactional_session
-def preparer_update_requests(source_iter: "Iterable[RequestResultOrState]", session: "Optional[Session]" = None) -> int:
+def prepare_requests(
+        requests_with_sources: "Sequence[RequestWithSources]",
+        logger: "LoggerFunction" = logging.log,
+        session: "Optional[Session]" = None,
+) -> int:
     """
     Update transfer requests according to preparer settings.
     """
     count = 0
-    for rws in source_iter:
-        update_dict = dict()
-        if isinstance(rws, RequestAndState):
-            # special case where the first entry is the request id and the second is the new state
-            # (see handling of RequestState.NO_SOURCES in reduce_requests)
-            request_id = rws.request_id
-            update_dict[models.Request.state] = rws.request_state
-        else:
-            request_id = rws['request_id']
-            update_dict[models.Request.state] = __throttler_request_state(
-                activity=rws['activity'],
-                source_rse_id=rws['src_rse_id'],
-                dest_rse_id=rws['dest_rse_id'],
-                session=session,
-            )
-            update_dict[models.Request.source_rse_id] = rws['src_rse_id']
+    reqs_no_source = set()
+    for rws in requests_with_sources:
+        # Assume request doesn't have any sources. Will be removed later if sources are found.
+        reqs_no_source.add(rws.request_id)
+
+        if not rws.sources:
+            logger(logging.INFO, '%s: has no sources. Skipping.', rws)
+            continue
+
+        sources = sorted(rws.sources, key=lambda s: s.distance_ranking)
+
+        selected_source = None
+        transfertool = None
+        dest_rse_transfertools = get_supported_transfertools(rws.dest_rse.id, session=session)
+        for source in sources:
+            src_rse_transfertools = get_supported_transfertools(source.rse.id, session=session)
+            common_transfertools = dest_rse_transfertools.intersection(src_rse_transfertools)
+            if not common_transfertools:
+                continue
+            transfertool = 'fts3' if 'fts3' in common_transfertools else common_transfertools.pop()
+            selected_source = source
+
+        if not selected_source:
+            logger(logging.WARNING, '%s: all available sources were filtered', rws)
+            continue
+
+        reqs_no_source.remove(rws.request_id)
 
-            if 'transfertool' in rws:
-                update_dict[models.Request.transfertool] = rws['transfertool']
+        update_dict = {
+            models.Request.state: __throttler_request_state(
+                activity=rws.activity,
+                source_rse_id=selected_source.rse.id,
+                dest_rse_id=rws.dest_rse.id,
+                session=session,
+            ),
+            models.Request.source_rse_id: selected_source.rse.id,
+        }
+        if transfertool:
+            update_dict[models.Request.transfertool] = transfertool
 
-        session.query(models.Request).filter_by(id=request_id).update(update_dict, synchronize_session=False)
+        session.query(models.Request).filter_by(id=rws.request_id).update(update_dict, synchronize_session=False)
         count += 1
+
+    if reqs_no_source:
+        logger(logging.INFO, "Marking requests as no-sources: %s", reqs_no_source)
+        set_requests_state_if_possible(reqs_no_source, RequestState.NO_SOURCES, logger=logger, session=session)
     return count
 
 
@@ -1753,53 +1780,6 @@ def __throttler_request_state(activity, source_rse_id, dest_rse_id, session: "Op
     return RequestState.WAITING if limit_found else RequestState.QUEUED
 
 
-def reduce_requests(
-    req_sources: "List[RequestWithSources]",
-    sort_reduce_funcs: "List[ReduceFunction]",
-    logger: "Callable",
-) -> "Iterator[RequestResultOrState]":
-    """
-    Reduces the passed requests & sources objects by using the sort-reduce
-    functions, yielding the best RequestResult object or a RequestAndState
-    object. If all sources were filtered, a RequestAndState object with
-    RequestState.NO_SOURCES is yielded.
-    """
-    assert len(req_sources) != 0, 'parameter request sources must be non-empty'
-
-    def pick_result(rws: "RequestWithSources") -> "Optional[RequestResult]":
-        result = [
-            {
-                'request_id': rws.request_id,
-                'dest_rse_id': rws.dest_rse.id,
-                'activity': rws.activity,
-                'src_rse_id': source.rse.id,
-                'distance_ranking': source.distance_ranking
-            }
-            for source in rws.sources
-        ]
-        debug_log = []
-        for sort_reduce in sort_reduce_funcs:
-            newresult = list(sort_reduce(result))
-            debug_log.append('filter %s removed %s' % (sort_reduce.__name__, list(filterfalse(newresult.__contains__, result))))
-            result = newresult
-
-        if len(result) == 0:
-            logger(logging.WARNING, 'all available sources were filtered for requests with id %s', rws.request_id)
-            logger(logging.DEBUG, 'the following filters ran:\\n' + '\\n'.join(debug_log))
-        else:
-            return result[0]
-
-    def result_or_no_sources(result: "Optional[RequestResult]") -> "RequestResultOrState":
-        if result is None:
-            return RequestAndState(request_id=cur_request_id, request_state=RequestState.NO_SOURCES)
-        else:
-            return result
-
-    for rws in req_sources:
-        cur_request_id = rws.request_id
-        yield result_or_no_sources(pick_result(rws))
-
-
 def get_supported_transfertools(rse_id: str, session=None) -> "Set[str]":
     transfertool_attr = get_rse_attribute('transfertool', rse_id=rse_id, session=session)
     if transfertool_attr:
@@ -1813,41 +1793,3 @@ def get_supported_transfertools(rse_id: str, session=None) -> "Set[str]":
         if result:
             return result
     return {'fts3', 'globus'}
-
-
-def get_transfertool_filter(
-    get_transfertools: "Callable[[str], Set[str]]" = get_supported_transfertools,
-) -> "ReduceFunction":
-    def filter_requests_for_transfertools(items: "RowIterator") -> "RowIterator":
-        first = True
-        first_request_id, first_dest_rse_id, dest_rse_transfertools = None, None, None
-        for rws_dict in items:
-            if first:
-                first = False
-                first_request_id = rws_dict['request_id']
-                first_dest_rse_id = rws_dict['dest_rse_id']
-                dest_rse_transfertools = get_transfertools(first_dest_rse_id)
-            else:
-                # same request id, same request destination rse in items per call
-                assert first_request_id == rws_dict['request_id']
-                assert first_dest_rse_id == rws_dict['dest_rse_id']
-
-            src_rse_transfertools = get_transfertools(rws_dict['src_rse_id'])
-            common_transfertools = dest_rse_transfertools.intersection(src_rse_transfertools)
-            if common_transfertools:
-                if 'fts3' in common_transfertools and 'globus' in common_transfertools:
-                    rws_dict['transfertool'] = 'fts3'
-                else:
-                    rws_dict['transfertool'] = common_transfertools.pop()
-                yield rws_dict
-
-    return filter_requests_for_transfertools
-
-
-def sort_requests_minimum_distance(items: "RowIterator") -> "RowIterator":
-    yield from sorted(items, key=lambda rws_dict: rws_dict['distance_ranking'])
-
-
-def rse_lookup_filter(items: "RowIterator") -> "RowIterator":
-    yield from filter(lambda rws_dict: (rws_dict['src_rse_id'] is not None
-                                        and rws_dict['dest_rse_id'] is not None), items)
diff --git a/lib/rucio/daemons/conveyor/preparer.py b/lib/rucio/daemons/conveyor/preparer.py
index b73c093c1..edc6a0d08 100644
--- a/lib/rucio/daemons/conveyor/preparer.py
+++ b/lib/rucio/daemons/conveyor/preparer.py
@@ -23,8 +23,7 @@ import rucio.db.sqla.util
 from rucio.common import exception
 from rucio.common.exception import RucioException
 from rucio.common.logging import setup_logging
-from rucio.core.request import preparer_update_requests, reduce_requests, sort_requests_minimum_distance, \\
-    get_transfertool_filter, get_supported_transfertools, rse_lookup_filter, list_transfer_requests_and_source_replicas
+from rucio.core.request import prepare_requests, list_transfer_requests_and_source_replicas
 from rucio.db.sqla.constants import RequestState
 from rucio.daemons.common import run_daemon
 
@@ -108,29 +107,27 @@ def run_once(bulk: int = 100, heartbeat_handler: "Optional[HeartbeatHandler]" =
         worker_number, total_workers, logger = 0, 0, logging.log
 
     start_time = time()
+    requests_with_sources = []
     try:
-        req_sources = list_transfer_requests_and_source_replicas(
+        requests_with_sources = list_transfer_requests_and_source_replicas(
             total_workers=total_workers,
             worker_number=worker_number,
             limit=bulk,
             request_state=RequestState.PREPARING,
             session=session
         )
-        if not req_sources:
-            count = 0
+        if not requests_with_sources:
             updated_msg = 'had nothing to do'
         else:
-            transfertool_filter = get_transfertool_filter(lambda rse_id: get_supported_transfertools(rse_id=rse_id, session=session))
-            requests = reduce_requests(req_sources, [rse_lookup_filter, sort_requests_minimum_distance, transfertool_filter], logger=logger)
-            count = preparer_update_requests(requests, session=session)
+            count = prepare_requests(requests_with_sources=requests_with_sources, logger=logger, session=session)
             updated_msg = f'updated {count}/{bulk} requests'
     except RucioException:
         logger(logging.ERROR, 'errored with a RucioException, retrying later', exc_info=True)
-        count = 0
         updated_msg = 'errored'
     logger(logging.INFO, '%s, taking %.3f seconds' % (updated_msg, time() - start_time))
 
-    queue_empty = False
-    if count < bulk:
-        queue_empty = True
-    return queue_empty
+    must_sleep = False
+    if len(requests_with_sources) < bulk / 2:
+        logger(logging.INFO, "Only %s transfers, which is less than half of the bulk %s", len(requests_with_sources), bulk)
+        must_sleep = True
+    return must_sleep
diff --git a/lib/rucio/tests/test_preparer.py b/lib/rucio/tests/test_preparer.py
index 7c65fe277..44b2570a4 100644
--- a/lib/rucio/tests/test_preparer.py
+++ b/lib/rucio/tests/test_preparer.py
@@ -23,7 +23,7 @@ from rucio.core import config as rucio_config
 from rucio.core.did import add_did, delete_dids
 from rucio.core.distance import get_distances, add_distance
 from rucio.core.replica import add_replicas, delete_replicas
-from rucio.core.request import sort_requests_minimum_distance, get_transfertool_filter, get_supported_transfertools, list_transfer_requests_and_source_replicas
+from rucio.core.request import get_supported_transfertools, list_transfer_requests_and_source_replicas
 from rucio.core.rse import set_rse_transfer_limits, add_rse, del_rse, add_rse_attribute
 from rucio.daemons.conveyor import preparer
 from rucio.db.sqla import models
@@ -32,7 +32,7 @@ from rucio.db.sqla.session import get_session
 from rucio.tests.common import rse_name_generator
 
 if TYPE_CHECKING:
-    from typing import Set, Optional, Callable
+    from typing import Optional, Callable
     from sqlalchemy.orm import Session
 
 
@@ -296,57 +296,6 @@ def test_two_sources_one_destination(db_session, vo, file, mock_request):
         delete_replicas(rse_id=source2_rse.rse_id, files=[file], session=db_session)
 
 
-def test_sort_requests_minimum_distance():
-    request_dicts = [{}, {}, {}]
-    for i in range(len(request_dicts)):
-        request_dicts[i]['request_id'] = i
-        request_dicts[i]['distance_ranking'] = 3 - i
-
-    result = sort_requests_minimum_distance(request_dicts)
-    assert next(result)['request_id'] == 2
-    assert next(result)['request_id'] == 1
-    assert next(result)['request_id'] == 0
-    pytest.raises(StopIteration, next, result)
-
-
-def test_filter_requests_for_transfertools():
-    request_dicts = [{}, {}, {}]
-    for i in range(len(request_dicts)):
-        request_dicts[i]['request_id'] = 0  # same request for all
-        request_dicts[i]['dest_rse_id'] = 'rse1'
-        request_dicts[i]['src_rse_id'] = f'rse{2 + i}'
-
-    def get_transfertools(rse_id: str) -> "Set[str]":
-        assert rse_id
-        if rse_id == 'rse1':
-            return {'globus'}
-        elif rse_id == 'rse2':
-            return {'fts3'}
-        elif rse_id == 'rse3':
-            return {'globus'}
-        elif rse_id == 'rse4':
-            return {'fts3', 'globus'}
-        else:
-            raise AssertionError('rse_id out of range')
-
-    transfertool_filter = get_transfertool_filter(get_transfertools=get_transfertools)
-    result = list(transfertool_filter(request_dicts))
-    print(result)
-
-    assert len(result) == 2
-    result.sort(key=lambda rws_dict: rws_dict['src_rse_id'])
-    assert result[0]['request_id'] == 0
-    assert result[0]['src_rse_id'] == 'rse3'
-    assert result[0]['dest_rse_id'] == 'rse1'
-    assert 'transfertool' in result[0]
-    assert result[0]['transfertool'] == 'globus'
-    assert result[1]['request_id'] == 0
-    assert result[1]['src_rse_id'] == 'rse4'
-    assert result[1]['dest_rse_id'] == 'rse1'
-    assert 'transfertool' in result[1]
-    assert result[1]['transfertool'] == 'globus'
-
-
 def test_get_supported_transfertools_default(vo, db_session):
     with GeneratedRSE(vo=vo, db_session=db_session) as generated_rse:
         transfertools = get_supported_transfertools(rse_id=generated_rse.rse_id, session=db_session)