We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support `ignore_availability` in preparer
Motivation
----------
There are two possible behaviors: 
- either requests are ignored when fetched from the database: https://github.com/rucio/rucio/blob/9a5d044d9f5760a224dba9feb14b71bfdf213252/lib/rucio/core/request.py#L420 
- or `ignore_availability` is set and they are marked NO_SOURCES: https://github.com/rucio/rucio/blob/9a5d044d9f5760a224dba9feb14b71bfdf213252/lib/rucio/core/transfer.py#L863 

Submitter has the ability to switch between these two modes using a CLI switch. Preparer is hard-coded into `ignore_availability = False`

Modification
------------
Implement this switch from the submitter:

https://github.com/rucio/rucio/blob/ac95ef3fabb8e1fcabddf1075cc19ae06db44a34/bin/rucio-conveyor-submitter#L108

in the preparer: 
https://github.com/rucio/rucio/blob/2a41a64899f41b356660fff80a7530765b5d59e0/bin/rucio-conveyor-preparer#L35

And pass it through the call stack to this function:
https://github.com/rucio/rucio/blob/e3c77683184063424fdcf9ecfda7d37cb4641c4f/lib/rucio/daemons/conveyor/preparer.py#L128

Add a test for the two behaviors. A example test (which will have to be simplified and adapted for this issue):
https://github.com/rucio/rucio/blob/2d81b39b8560f376f412fcba62a622a6db748780/lib/rucio/tests/test_conveyor.py#L611



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `lib/rucio/daemons/conveyor/preparer.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 # Copyright European Organization for Nuclear Research (CERN) since 2012
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #    http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 
16 import functools
17 import logging
18 import threading
19 from time import time
20 from typing import TYPE_CHECKING
21 
22 import rucio.db.sqla.util
23 from rucio.common import exception
24 from rucio.common.config import config_get_list
25 from rucio.common.exception import RucioException
26 from rucio.common.logging import setup_logging
27 from rucio.core import transfer as transfer_core
28 from rucio.core.request import set_requests_state_if_possible, list_transfer_requests_and_source_replicas
29 from rucio.core.transfer import prepare_transfers, list_transfer_admin_accounts, build_transfer_paths
30 from rucio.core.topology import Topology
31 from rucio.db.sqla.constants import RequestState, RequestType
32 from rucio.daemons.common import run_daemon
33 
34 if TYPE_CHECKING:
35     from typing import Optional, List
36     from sqlalchemy.orm import Session
37     from rucio.daemons.common import HeartbeatHandler
38 
39 graceful_stop = threading.Event()
40 
41 
42 def stop():
43     """
44     Graceful exit.
45     """
46 
47     graceful_stop.set()
48 
49 
50 def run(once=False, threads=1, sleep_time=10, bulk=100):
51     """
52     Running the preparer daemon either once or by default in a loop until stop is called.
53     """
54     setup_logging()
55 
56     if rucio.db.sqla.util.is_old_db():
57         raise exception.DatabaseException('Database was not updated, daemon won\'t start')
58 
59     def preparer_kwargs():
60         # not sure if this is needed for threading.Thread, but it always returns a fresh dictionary
61         return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk}
62 
63     threads = [threading.Thread(target=preparer, name=f'conveyor-preparer-{i}', kwargs=preparer_kwargs(), daemon=True) for i in range(threads)]
64     for thr in threads:
65         thr.start()
66 
67     all_running = True
68     while all_running:
69         for thr in threads:
70             thr.join(timeout=3.14)
71             if not thr.is_alive() or graceful_stop.is_set():
72                 all_running = False
73                 break
74 
75     if graceful_stop.is_set() or once:
76         logging.info('conveyor-preparer: gracefully stopping')
77     else:
78         logging.warning('conveyor-preparer: stopping out of the ordinary')
79         graceful_stop.set()
80 
81     for thr in threads:
82         thr.join(timeout=3.14)
83 
84     logging.info('conveyor-preparer: stopped')
85 
86 
87 def preparer(once, sleep_time, bulk, partition_wait_time=10):
88     # Make an initial heartbeat so that all instanced daemons have the correct worker number on the next try
89     logger_prefix = executable = 'conveyor-preparer'
90     transfertools = config_get_list('conveyor', 'transfertool', False, None)
91 
92     run_daemon(
93         once=once,
94         graceful_stop=graceful_stop,
95         executable=executable,
96         logger_prefix=logger_prefix,
97         partition_wait_time=partition_wait_time,
98         sleep_time=sleep_time,
99         run_once_fnc=functools.partial(
100             run_once,
101             transfertools=transfertools,
102             bulk=bulk,
103         ),
104         activities=None,
105     )
106 
107 
108 def run_once(
109         transfertools: "Optional[List[str]]" = None,
110         bulk: int = 100,
111         heartbeat_handler: "Optional[HeartbeatHandler]" = None,
112         session: "Optional[Session]" = None,
113         **kwargs
114 ) -> bool:
115     if heartbeat_handler:
116         worker_number, total_workers, logger = heartbeat_handler.live()
117     else:
118         # This is used in tests
119         worker_number, total_workers, logger = 0, 0, logging.log
120     if not transfertools:
121         transfertools = list(transfer_core.TRANSFERTOOL_CLASSES_BY_NAME)
122 
123     start_time = time()
124     requests_handled = 0
125     try:
126         admin_accounts = list_transfer_admin_accounts()
127         topology = Topology.create_from_config(logger=logger)
128         requests_with_sources = list_transfer_requests_and_source_replicas(
129             total_workers=total_workers,
130             worker_number=worker_number,
131             limit=bulk,
132             request_state=RequestState.PREPARING,
133             request_type=[RequestType.TRANSFER, RequestType.STAGEIN],
134             session=session
135         )
136         ret = build_transfer_paths(
137             topology=topology,
138             requests_with_sources=list(requests_with_sources.values()),
139             admin_accounts=admin_accounts,
140             preparer_mode=True,
141             logger=logger,
142             session=session,
143         )
144         requests_handled = sum(len(i) for i in ret)
145         if not requests_handled:
146             updated_msg = 'had nothing to do'
147         else:
148             candidate_paths, reqs_no_source, reqs_scheme_mismatch, reqs_only_tape_source, _ = ret
149             updated_reqs, reqs_no_transfertool = prepare_transfers(candidate_paths, transfertools=transfertools, logger=logger, session=session)
150             updated_msg = f'updated {len(updated_reqs)}/{bulk} requests'
151 
152             if reqs_no_transfertool:
153                 logger(logging.INFO, "Ignoring request because of unsupported transfertool: %s", reqs_no_transfertool)
154             reqs_no_source.update(reqs_no_transfertool)
155             if reqs_no_source:
156                 logger(logging.INFO, "Marking requests as no-sources: %s", reqs_no_source)
157                 set_requests_state_if_possible(reqs_no_source, RequestState.NO_SOURCES, logger=logger)
158             if reqs_only_tape_source:
159                 logger(logging.INFO, "Marking requests as only-tape-sources: %s", reqs_only_tape_source)
160                 set_requests_state_if_possible(reqs_only_tape_source, RequestState.ONLY_TAPE_SOURCES, logger=logger)
161             if reqs_scheme_mismatch:
162                 logger(logging.INFO, "Marking requests as scheme-mismatch: %s", reqs_scheme_mismatch)
163                 set_requests_state_if_possible(reqs_scheme_mismatch, RequestState.MISMATCH_SCHEME, logger=logger)
164     except RucioException:
165         logger(logging.ERROR, 'errored with a RucioException, retrying later', exc_info=True)
166         updated_msg = 'errored'
167     logger(logging.INFO, '%s, taking %.3f seconds' % (updated_msg, time() - start_time))
168 
169     must_sleep = False
170     if requests_handled < bulk / 2:
171         logger(logging.INFO, "Only %s transfers, which is less than half of the bulk %s", requests_handled, bulk)
172         must_sleep = True
173     return must_sleep
174
```
--- 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/lib/rucio/daemons/conveyor/preparer.py b/lib/rucio/daemons/conveyor/preparer.py
--- a/lib/rucio/daemons/conveyor/preparer.py
+++ b/lib/rucio/daemons/conveyor/preparer.py
@@ -47,7 +47,7 @@
     graceful_stop.set()
 
 
-def run(once=False, threads=1, sleep_time=10, bulk=100):
+def run(once=False, threads=1, sleep_time=10, bulk=100, ignore_availability: bool = False):
     """
     Running the preparer daemon either once or by default in a loop until stop is called.
     """
@@ -58,7 +58,7 @@
 
     def preparer_kwargs():
         # not sure if this is needed for threading.Thread, but it always returns a fresh dictionary
-        return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk}
+        return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk, 'ignore_availability': ignore_availability}
 
     threads = [threading.Thread(target=preparer, name=f'conveyor-preparer-{i}', kwargs=preparer_kwargs(), daemon=True) for i in range(threads)]
     for thr in threads:
@@ -84,7 +84,7 @@
     logging.info('conveyor-preparer: stopped')
 
 
-def preparer(once, sleep_time, bulk, partition_wait_time=10):
+def preparer(once, sleep_time, bulk, ignore_availability: bool, partition_wait_time: int = 10):
     # Make an initial heartbeat so that all instanced daemons have the correct worker number on the next try
     logger_prefix = executable = 'conveyor-preparer'
     transfertools = config_get_list('conveyor', 'transfertool', False, None)
@@ -100,6 +100,7 @@
             run_once,
             transfertools=transfertools,
             bulk=bulk,
+            ignore_availability=ignore_availability,
         ),
         activities=None,
     )
@@ -110,6 +111,7 @@
         bulk: int = 100,
         heartbeat_handler: "Optional[HeartbeatHandler]" = None,
         session: "Optional[Session]" = None,
+        ignore_availability: bool = False,
         **kwargs
 ) -> bool:
     if heartbeat_handler:
@@ -124,14 +126,15 @@
     requests_handled = 0
     try:
         admin_accounts = list_transfer_admin_accounts()
-        topology = Topology.create_from_config(logger=logger)
+        topology = Topology.create_from_config(logger=logger, ignore_availability=ignore_availability)
         requests_with_sources = list_transfer_requests_and_source_replicas(
             total_workers=total_workers,
             worker_number=worker_number,
             limit=bulk,
             request_state=RequestState.PREPARING,
             request_type=[RequestType.TRANSFER, RequestType.STAGEIN],
-            session=session
+            session=session,
+            ignore_availability=ignore_availability
         )
         ret = build_transfer_paths(
             topology=topology,

{"golden_diff": "diff --git a/lib/rucio/daemons/conveyor/preparer.py b/lib/rucio/daemons/conveyor/preparer.py\n--- a/lib/rucio/daemons/conveyor/preparer.py\n+++ b/lib/rucio/daemons/conveyor/preparer.py\n@@ -47,7 +47,7 @@\n     graceful_stop.set()\n \n \n-def run(once=False, threads=1, sleep_time=10, bulk=100):\n+def run(once=False, threads=1, sleep_time=10, bulk=100, ignore_availability: bool = False):\n     \"\"\"\n     Running the preparer daemon either once or by default in a loop until stop is called.\n     \"\"\"\n@@ -58,7 +58,7 @@\n \n     def preparer_kwargs():\n         # not sure if this is needed for threading.Thread, but it always returns a fresh dictionary\n-        return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk}\n+        return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk, 'ignore_availability': ignore_availability}\n \n     threads = [threading.Thread(target=preparer, name=f'conveyor-preparer-{i}', kwargs=preparer_kwargs(), daemon=True) for i in range(threads)]\n     for thr in threads:\n@@ -84,7 +84,7 @@\n     logging.info('conveyor-preparer: stopped')\n \n \n-def preparer(once, sleep_time, bulk, partition_wait_time=10):\n+def preparer(once, sleep_time, bulk, ignore_availability: bool, partition_wait_time: int = 10):\n     # Make an initial heartbeat so that all instanced daemons have the correct worker number on the next try\n     logger_prefix = executable = 'conveyor-preparer'\n     transfertools = config_get_list('conveyor', 'transfertool', False, None)\n@@ -100,6 +100,7 @@\n             run_once,\n             transfertools=transfertools,\n             bulk=bulk,\n+            ignore_availability=ignore_availability,\n         ),\n         activities=None,\n     )\n@@ -110,6 +111,7 @@\n         bulk: int = 100,\n         heartbeat_handler: \"Optional[HeartbeatHandler]\" = None,\n         session: \"Optional[Session]\" = None,\n+        ignore_availability: bool = False,\n         **kwargs\n ) -> bool:\n     if heartbeat_handler:\n@@ -124,14 +126,15 @@\n     requests_handled = 0\n     try:\n         admin_accounts = list_transfer_admin_accounts()\n-        topology = Topology.create_from_config(logger=logger)\n+        topology = Topology.create_from_config(logger=logger, ignore_availability=ignore_availability)\n         requests_with_sources = list_transfer_requests_and_source_replicas(\n             total_workers=total_workers,\n             worker_number=worker_number,\n             limit=bulk,\n             request_state=RequestState.PREPARING,\n             request_type=[RequestType.TRANSFER, RequestType.STAGEIN],\n-            session=session\n+            session=session,\n+            ignore_availability=ignore_availability\n         )\n         ret = build_transfer_paths(\n             topology=topology,\n", "issue": "Support `ignore_availability` in preparer\nMotivation\r\n----------\r\nThere are two possible behaviors: \r\n- either requests are ignored when fetched from the database: https://github.com/rucio/rucio/blob/9a5d044d9f5760a224dba9feb14b71bfdf213252/lib/rucio/core/request.py#L420 \r\n- or `ignore_availability` is set and they are marked NO_SOURCES: https://github.com/rucio/rucio/blob/9a5d044d9f5760a224dba9feb14b71bfdf213252/lib/rucio/core/transfer.py#L863 \r\n\r\nSubmitter has the ability to switch between these two modes using a CLI switch. Preparer is hard-coded into `ignore_availability = False`\r\n\r\nModification\r\n------------\r\nImplement this switch from the submitter:\r\n\r\nhttps://github.com/rucio/rucio/blob/ac95ef3fabb8e1fcabddf1075cc19ae06db44a34/bin/rucio-conveyor-submitter#L108\r\n\r\nin the preparer: \r\nhttps://github.com/rucio/rucio/blob/2a41a64899f41b356660fff80a7530765b5d59e0/bin/rucio-conveyor-preparer#L35\r\n\r\nAnd pass it through the call stack to this function:\r\nhttps://github.com/rucio/rucio/blob/e3c77683184063424fdcf9ecfda7d37cb4641c4f/lib/rucio/daemons/conveyor/preparer.py#L128\r\n\r\nAdd a test for the two behaviors. A example test (which will have to be simplified and adapted for this issue):\r\nhttps://github.com/rucio/rucio/blob/2d81b39b8560f376f412fcba62a622a6db748780/lib/rucio/tests/test_conveyor.py#L611\r\n\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright European Organization for Nuclear Research (CERN) since 2012\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport functools\nimport logging\nimport threading\nfrom time import time\nfrom typing import TYPE_CHECKING\n\nimport rucio.db.sqla.util\nfrom rucio.common import exception\nfrom rucio.common.config import config_get_list\nfrom rucio.common.exception import RucioException\nfrom rucio.common.logging import setup_logging\nfrom rucio.core import transfer as transfer_core\nfrom rucio.core.request import set_requests_state_if_possible, list_transfer_requests_and_source_replicas\nfrom rucio.core.transfer import prepare_transfers, list_transfer_admin_accounts, build_transfer_paths\nfrom rucio.core.topology import Topology\nfrom rucio.db.sqla.constants import RequestState, RequestType\nfrom rucio.daemons.common import run_daemon\n\nif TYPE_CHECKING:\n    from typing import Optional, List\n    from sqlalchemy.orm import Session\n    from rucio.daemons.common import HeartbeatHandler\n\ngraceful_stop = threading.Event()\n\n\ndef stop():\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n\n    graceful_stop.set()\n\n\ndef run(once=False, threads=1, sleep_time=10, bulk=100):\n    \"\"\"\n    Running the preparer daemon either once or by default in a loop until stop is called.\n    \"\"\"\n    setup_logging()\n\n    if rucio.db.sqla.util.is_old_db():\n        raise exception.DatabaseException('Database was not updated, daemon won\\'t start')\n\n    def preparer_kwargs():\n        # not sure if this is needed for threading.Thread, but it always returns a fresh dictionary\n        return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk}\n\n    threads = [threading.Thread(target=preparer, name=f'conveyor-preparer-{i}', kwargs=preparer_kwargs(), daemon=True) for i in range(threads)]\n    for thr in threads:\n        thr.start()\n\n    all_running = True\n    while all_running:\n        for thr in threads:\n            thr.join(timeout=3.14)\n            if not thr.is_alive() or graceful_stop.is_set():\n                all_running = False\n                break\n\n    if graceful_stop.is_set() or once:\n        logging.info('conveyor-preparer: gracefully stopping')\n    else:\n        logging.warning('conveyor-preparer: stopping out of the ordinary')\n        graceful_stop.set()\n\n    for thr in threads:\n        thr.join(timeout=3.14)\n\n    logging.info('conveyor-preparer: stopped')\n\n\ndef preparer(once, sleep_time, bulk, partition_wait_time=10):\n    # Make an initial heartbeat so that all instanced daemons have the correct worker number on the next try\n    logger_prefix = executable = 'conveyor-preparer'\n    transfertools = config_get_list('conveyor', 'transfertool', False, None)\n\n    run_daemon(\n        once=once,\n        graceful_stop=graceful_stop,\n        executable=executable,\n        logger_prefix=logger_prefix,\n        partition_wait_time=partition_wait_time,\n        sleep_time=sleep_time,\n        run_once_fnc=functools.partial(\n            run_once,\n            transfertools=transfertools,\n            bulk=bulk,\n        ),\n        activities=None,\n    )\n\n\ndef run_once(\n        transfertools: \"Optional[List[str]]\" = None,\n        bulk: int = 100,\n        heartbeat_handler: \"Optional[HeartbeatHandler]\" = None,\n        session: \"Optional[Session]\" = None,\n        **kwargs\n) -> bool:\n    if heartbeat_handler:\n        worker_number, total_workers, logger = heartbeat_handler.live()\n    else:\n        # This is used in tests\n        worker_number, total_workers, logger = 0, 0, logging.log\n    if not transfertools:\n        transfertools = list(transfer_core.TRANSFERTOOL_CLASSES_BY_NAME)\n\n    start_time = time()\n    requests_handled = 0\n    try:\n        admin_accounts = list_transfer_admin_accounts()\n        topology = Topology.create_from_config(logger=logger)\n        requests_with_sources = list_transfer_requests_and_source_replicas(\n            total_workers=total_workers,\n            worker_number=worker_number,\n            limit=bulk,\n            request_state=RequestState.PREPARING,\n            request_type=[RequestType.TRANSFER, RequestType.STAGEIN],\n            session=session\n        )\n        ret = build_transfer_paths(\n            topology=topology,\n            requests_with_sources=list(requests_with_sources.values()),\n            admin_accounts=admin_accounts,\n            preparer_mode=True,\n            logger=logger,\n            session=session,\n        )\n        requests_handled = sum(len(i) for i in ret)\n        if not requests_handled:\n            updated_msg = 'had nothing to do'\n        else:\n            candidate_paths, reqs_no_source, reqs_scheme_mismatch, reqs_only_tape_source, _ = ret\n            updated_reqs, reqs_no_transfertool = prepare_transfers(candidate_paths, transfertools=transfertools, logger=logger, session=session)\n            updated_msg = f'updated {len(updated_reqs)}/{bulk} requests'\n\n            if reqs_no_transfertool:\n                logger(logging.INFO, \"Ignoring request because of unsupported transfertool: %s\", reqs_no_transfertool)\n            reqs_no_source.update(reqs_no_transfertool)\n            if reqs_no_source:\n                logger(logging.INFO, \"Marking requests as no-sources: %s\", reqs_no_source)\n                set_requests_state_if_possible(reqs_no_source, RequestState.NO_SOURCES, logger=logger)\n            if reqs_only_tape_source:\n                logger(logging.INFO, \"Marking requests as only-tape-sources: %s\", reqs_only_tape_source)\n                set_requests_state_if_possible(reqs_only_tape_source, RequestState.ONLY_TAPE_SOURCES, logger=logger)\n            if reqs_scheme_mismatch:\n                logger(logging.INFO, \"Marking requests as scheme-mismatch: %s\", reqs_scheme_mismatch)\n                set_requests_state_if_possible(reqs_scheme_mismatch, RequestState.MISMATCH_SCHEME, logger=logger)\n    except RucioException:\n        logger(logging.ERROR, 'errored with a RucioException, retrying later', exc_info=True)\n        updated_msg = 'errored'\n    logger(logging.INFO, '%s, taking %.3f seconds' % (updated_msg, time() - start_time))\n\n    must_sleep = False\n    if requests_handled < bulk / 2:\n        logger(logging.INFO, \"Only %s transfers, which is less than half of the bulk %s\", requests_handled, bulk)\n        must_sleep = True\n    return must_sleep\n", "path": "lib/rucio/daemons/conveyor/preparer.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n# Copyright European Organization for Nuclear Research (CERN) since 2012\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport functools\nimport logging\nimport threading\nfrom time import time\nfrom typing import TYPE_CHECKING\n\nimport rucio.db.sqla.util\nfrom rucio.common import exception\nfrom rucio.common.config import config_get_list\nfrom rucio.common.exception import RucioException\nfrom rucio.common.logging import setup_logging\nfrom rucio.core import transfer as transfer_core\nfrom rucio.core.request import set_requests_state_if_possible, list_transfer_requests_and_source_replicas\nfrom rucio.core.transfer import prepare_transfers, list_transfer_admin_accounts, build_transfer_paths\nfrom rucio.core.topology import Topology\nfrom rucio.db.sqla.constants import RequestState, RequestType\nfrom rucio.daemons.common import run_daemon\n\nif TYPE_CHECKING:\n    from typing import Optional, List\n    from sqlalchemy.orm import Session\n    from rucio.daemons.common import HeartbeatHandler\n\ngraceful_stop = threading.Event()\n\n\ndef stop():\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n\n    graceful_stop.set()\n\n\ndef run(once=False, threads=1, sleep_time=10, bulk=100, ignore_availability: bool = False):\n    \"\"\"\n    Running the preparer daemon either once or by default in a loop until stop is called.\n    \"\"\"\n    setup_logging()\n\n    if rucio.db.sqla.util.is_old_db():\n        raise exception.DatabaseException('Database was not updated, daemon won\\'t start')\n\n    def preparer_kwargs():\n        # not sure if this is needed for threading.Thread, but it always returns a fresh dictionary\n        return {'once': once, 'sleep_time': sleep_time, 'bulk': bulk, 'ignore_availability': ignore_availability}\n\n    threads = [threading.Thread(target=preparer, name=f'conveyor-preparer-{i}', kwargs=preparer_kwargs(), daemon=True) for i in range(threads)]\n    for thr in threads:\n        thr.start()\n\n    all_running = True\n    while all_running:\n        for thr in threads:\n            thr.join(timeout=3.14)\n            if not thr.is_alive() or graceful_stop.is_set():\n                all_running = False\n                break\n\n    if graceful_stop.is_set() or once:\n        logging.info('conveyor-preparer: gracefully stopping')\n    else:\n        logging.warning('conveyor-preparer: stopping out of the ordinary')\n        graceful_stop.set()\n\n    for thr in threads:\n        thr.join(timeout=3.14)\n\n    logging.info('conveyor-preparer: stopped')\n\n\ndef preparer(once, sleep_time, bulk, ignore_availability: bool, partition_wait_time: int = 10):\n    # Make an initial heartbeat so that all instanced daemons have the correct worker number on the next try\n    logger_prefix = executable = 'conveyor-preparer'\n    transfertools = config_get_list('conveyor', 'transfertool', False, None)\n\n    run_daemon(\n        once=once,\n        graceful_stop=graceful_stop,\n        executable=executable,\n        logger_prefix=logger_prefix,\n        partition_wait_time=partition_wait_time,\n        sleep_time=sleep_time,\n        run_once_fnc=functools.partial(\n            run_once,\n            transfertools=transfertools,\n            bulk=bulk,\n            ignore_availability=ignore_availability,\n        ),\n        activities=None,\n    )\n\n\ndef run_once(\n        transfertools: \"Optional[List[str]]\" = None,\n        bulk: int = 100,\n        heartbeat_handler: \"Optional[HeartbeatHandler]\" = None,\n        session: \"Optional[Session]\" = None,\n        ignore_availability: bool = False,\n        **kwargs\n) -> bool:\n    if heartbeat_handler:\n        worker_number, total_workers, logger = heartbeat_handler.live()\n    else:\n        # This is used in tests\n        worker_number, total_workers, logger = 0, 0, logging.log\n    if not transfertools:\n        transfertools = list(transfer_core.TRANSFERTOOL_CLASSES_BY_NAME)\n\n    start_time = time()\n    requests_handled = 0\n    try:\n        admin_accounts = list_transfer_admin_accounts()\n        topology = Topology.create_from_config(logger=logger, ignore_availability=ignore_availability)\n        requests_with_sources = list_transfer_requests_and_source_replicas(\n            total_workers=total_workers,\n            worker_number=worker_number,\n            limit=bulk,\n            request_state=RequestState.PREPARING,\n            request_type=[RequestType.TRANSFER, RequestType.STAGEIN],\n            session=session,\n            ignore_availability=ignore_availability\n        )\n        ret = build_transfer_paths(\n            topology=topology,\n            requests_with_sources=list(requests_with_sources.values()),\n            admin_accounts=admin_accounts,\n            preparer_mode=True,\n            logger=logger,\n            session=session,\n        )\n        requests_handled = sum(len(i) for i in ret)\n        if not requests_handled:\n            updated_msg = 'had nothing to do'\n        else:\n            candidate_paths, reqs_no_source, reqs_scheme_mismatch, reqs_only_tape_source, _ = ret\n            updated_reqs, reqs_no_transfertool = prepare_transfers(candidate_paths, transfertools=transfertools, logger=logger, session=session)\n            updated_msg = f'updated {len(updated_reqs)}/{bulk} requests'\n\n            if reqs_no_transfertool:\n                logger(logging.INFO, \"Ignoring request because of unsupported transfertool: %s\", reqs_no_transfertool)\n            reqs_no_source.update(reqs_no_transfertool)\n            if reqs_no_source:\n                logger(logging.INFO, \"Marking requests as no-sources: %s\", reqs_no_source)\n                set_requests_state_if_possible(reqs_no_source, RequestState.NO_SOURCES, logger=logger)\n            if reqs_only_tape_source:\n                logger(logging.INFO, \"Marking requests as only-tape-sources: %s\", reqs_only_tape_source)\n                set_requests_state_if_possible(reqs_only_tape_source, RequestState.ONLY_TAPE_SOURCES, logger=logger)\n            if reqs_scheme_mismatch:\n                logger(logging.INFO, \"Marking requests as scheme-mismatch: %s\", reqs_scheme_mismatch)\n                set_requests_state_if_possible(reqs_scheme_mismatch, RequestState.MISMATCH_SCHEME, logger=logger)\n    except RucioException:\n        logger(logging.ERROR, 'errored with a RucioException, retrying later', exc_info=True)\n        updated_msg = 'errored'\n    logger(logging.INFO, '%s, taking %.3f seconds' % (updated_msg, time() - start_time))\n\n    must_sleep = False\n    if requests_handled < bulk / 2:\n        logger(logging.INFO, \"Only %s transfers, which is less than half of the bulk %s\", requests_handled, bulk)\n        must_sleep = True\n    return must_sleep\n", "path": "lib/rucio/daemons/conveyor/preparer.py"}]}