We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Optimize the update of the final states in the necromancer
Motivation
----------
Currently the necromancer update up-to 10000000 bad replicas states in one transaction


Modification
------------
Update by chunks



--- 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/rse/protocols/__init__.py`
Content:
```
1 # Copyright European Organization for Nuclear Research (CERN)
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # You may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
6 
7 # These are all protocols which are considered in good health
8 # based on generally available support libraries, native dependencies,
9 # and implementations in Rucio
10 supported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https', 'storm']
11
```

Path: `lib/rucio/daemons/badreplicas/necromancer.py`
Content:
```
1 # Copyright 2014-2018 CERN for the benefit of the ATLAS collaboration.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 #
15 # Authors:
16 # - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2017
17 # - Vincent Garonne <vgaronne@gmail.com>, 2015-2018
18 # - Mario Lassnig <mario.lassnig@cern.ch>, 2015
19 # - Wen Guan <wguan.icedew@gmail.com>, 2015
20 # - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019
21 #
22 # PY3K COMPATIBLE
23 
24 import logging
25 import os
26 import socket
27 import threading
28 import time
29 
30 from sys import exc_info, stdout, argv
31 from traceback import format_exception
32 
33 from rucio.db.sqla.constants import ReplicaState
34 from rucio.common.config import config_get
35 from rucio.common.exception import DatabaseException
36 from rucio.core import monitor, heartbeat
37 from rucio.core.replica import list_bad_replicas, get_replicas_state, list_bad_replicas_history, update_bad_replicas_history
38 from rucio.core.rule import update_rules_for_lost_replica, update_rules_for_bad_replica
39 
40 
41 logging.basicConfig(stream=stdout,
42                     level=getattr(logging,
43                                   config_get('common', 'loglevel',
44                                              raise_exception=False,
45                                              default='DEBUG').upper()),
46                     format='%(asctime)s\t%(process)d\t%(levelname)s\t%(message)s')
47 
48 graceful_stop = threading.Event()
49 
50 
51 def necromancer(thread=0, bulk=5, once=False):
52     """
53     Creates a Necromancer Worker that gets a list of bad replicas for a given hash,
54     identify lost DIDs and for non-lost ones, set the locks and rules for reevaluation.
55 
56     :param thread: Thread number at startup.
57     :param bulk: The number of requests to process.
58     :param once: Run only once.
59     """
60 
61     sleep_time = 60
62     update_history_threshold = 3600
63     update_history_time = time.time()
64 
65     executable = ' '.join(argv)
66     hostname = socket.getfqdn()
67     pid = os.getpid()
68     hb_thread = threading.current_thread()
69     heartbeat.sanity_check(executable=executable, hostname=hostname)
70 
71     while not graceful_stop.is_set():
72 
73         heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)
74         prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'] + 1, heart_beat['nr_threads'])
75 
76         stime = time.time()
77         replicas = []
78         try:
79             replicas = list_bad_replicas(limit=bulk, thread=heart_beat['assign_thread'], total_threads=heart_beat['nr_threads'])
80 
81             for replica in replicas:
82                 scope, name, rse_id, rse = replica['scope'], replica['name'], replica['rse_id'], replica['rse']
83                 logging.info(prepend_str + 'Working on %s:%s on %s' % (scope, name, rse))
84 
85                 list_replicas = get_replicas_state(scope=scope, name=name)
86                 if ReplicaState.AVAILABLE not in list_replicas and ReplicaState.TEMPORARY_UNAVAILABLE not in list_replicas:
87                     logging.info(prepend_str + 'File %s:%s has no other available or temporary available replicas, it will be marked as lost' % (scope, name))
88                     try:
89                         update_rules_for_lost_replica(scope=scope, name=name, rse_id=rse_id, nowait=True)
90                         monitor.record_counter(counters='necromancer.badfiles.lostfile', delta=1)
91                     except DatabaseException as error:
92                         logging.info(prepend_str + '%s' % (str(error)))
93 
94                 else:
95                     rep = list_replicas.get(ReplicaState.AVAILABLE, [])
96                     unavailable_rep = list_replicas.get(ReplicaState.TEMPORARY_UNAVAILABLE, [])
97                     logging.info(prepend_str + 'File %s:%s can be recovered. Available sources : %s + Unavailable sources : %s' % (scope, name, str(rep), str(unavailable_rep)))
98                     try:
99                         update_rules_for_bad_replica(scope=scope, name=name, rse_id=rse_id, nowait=True)
100                         monitor.record_counter(counters='necromancer.badfiles.recovering', delta=1)
101                     except DatabaseException as error:
102                         logging.info(prepend_str + '%s' % (str(error)))
103 
104             logging.info(prepend_str + 'It took %s seconds to process %s replicas' % (str(time.time() - stime), str(len(replicas))))
105         except Exception:
106             exc_type, exc_value, exc_traceback = exc_info()
107             logging.critical(prepend_str + ''.join(format_exception(exc_type, exc_value, exc_traceback)).strip())
108 
109         if once:
110             break
111         else:
112             now = time.time()
113             if (now - update_history_time) > update_history_threshold:
114                 logging.info(prepend_str + 'Last update of history table %s seconds ago. Running update.' % (now - update_history_time))
115                 bad_replicas = list_bad_replicas_history(limit=10000000,
116                                                          thread=heart_beat['assign_thread'],
117                                                          total_threads=heart_beat['nr_threads'])
118                 for rse_id in bad_replicas:
119                     update_bad_replicas_history(bad_replicas[rse_id], rse_id)
120                 logging.info(prepend_str + 'History table updated in %s seconds' % (time.time() - now))
121                 update_history_time = time.time()
122 
123             tottime = time.time() - stime
124             if len(replicas) == bulk:
125                 logging.info(prepend_str + 'Processed maximum number of replicas according to the bulk size. Restart immediately next cycle')
126             elif tottime < sleep_time:
127                 logging.info(prepend_str + 'Will sleep for %s seconds' % (str(sleep_time - tottime)))
128                 time.sleep(sleep_time - tottime)
129                 continue
130 
131     logging.info(prepend_str + 'Graceful stop requested')
132     heartbeat.die(executable, hostname, pid, hb_thread)
133     logging.info(prepend_str + 'Graceful stop done')
134 
135 
136 def run(threads=1, bulk=100, once=False):
137     """
138     Starts up the necromancer threads.
139     """
140 
141     if once:
142         logging.info('Will run only one iteration in a single threaded mode')
143         necromancer(bulk=bulk, once=once)
144     else:
145         logging.info('starting necromancer threads')
146         thread_list = [threading.Thread(target=necromancer, kwargs={'once': once,
147                                                                     'thread': i,
148                                                                     'bulk': bulk}) for i in range(0, threads)]
149         [t.start() for t in thread_list]
150 
151         logging.info('waiting for interrupts')
152 
153         # Interruptible joins require a timeout.
154         while thread_list:
155             thread_list = [thread.join(timeout=3.14) for thread in thread_list if thread and thread.isAlive()]
156 
157 
158 def stop(signum=None, frame=None):
159     """
160     Graceful exit.
161     """
162     graceful_stop.set()
163
```
--- 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/badreplicas/necromancer.py b/lib/rucio/daemons/badreplicas/necromancer.py
--- a/lib/rucio/daemons/badreplicas/necromancer.py
+++ b/lib/rucio/daemons/badreplicas/necromancer.py
@@ -13,7 +13,7 @@
 # limitations under the License.
 #
 # Authors:
-# - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2017
+# - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2019
 # - Vincent Garonne <vgaronne@gmail.com>, 2015-2018
 # - Mario Lassnig <mario.lassnig@cern.ch>, 2015
 # - Wen Guan <wguan.icedew@gmail.com>, 2015
@@ -21,17 +21,21 @@
 #
 # PY3K COMPATIBLE
 
+from __future__ import division
+
 import logging
 import os
 import socket
 import threading
 import time
 
+from math import ceil
 from sys import exc_info, stdout, argv
 from traceback import format_exception
 
 from rucio.db.sqla.constants import ReplicaState
 from rucio.common.config import config_get
+from rucio.common.utils import chunks
 from rucio.common.exception import DatabaseException
 from rucio.core import monitor, heartbeat
 from rucio.core.replica import list_bad_replicas, get_replicas_state, list_bad_replicas_history, update_bad_replicas_history
@@ -112,11 +116,18 @@
             now = time.time()
             if (now - update_history_time) > update_history_threshold:
                 logging.info(prepend_str + 'Last update of history table %s seconds ago. Running update.' % (now - update_history_time))
-                bad_replicas = list_bad_replicas_history(limit=10000000,
+                bad_replicas = list_bad_replicas_history(limit=1000000,
                                                          thread=heart_beat['assign_thread'],
                                                          total_threads=heart_beat['nr_threads'])
                 for rse_id in bad_replicas:
-                    update_bad_replicas_history(bad_replicas[rse_id], rse_id)
+                    chunk_size = 1000
+                    nchunk = int(ceil(len(bad_replicas[rse_id]) / chunk_size))
+                    logging.debug(prepend_str + 'Update history for rse_id %s' % (rse_id))
+                    cnt = 0
+                    for chunk in chunks(bad_replicas[rse_id], chunk_size):
+                        logging.debug(prepend_str + ' History for rse_id %s : chunk %i/%i' % (rse_id, cnt, nchunk))
+                        cnt += 1
+                        update_bad_replicas_history(chunk, rse_id)
                 logging.info(prepend_str + 'History table updated in %s seconds' % (time.time() - now))
                 update_history_time = time.time()
 
diff --git a/lib/rucio/rse/protocols/__init__.py b/lib/rucio/rse/protocols/__init__.py
--- a/lib/rucio/rse/protocols/__init__.py
+++ b/lib/rucio/rse/protocols/__init__.py
@@ -7,4 +7,4 @@
 # These are all protocols which are considered in good health
 # based on generally available support libraries, native dependencies,
 # and implementations in Rucio
-supported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https']
+supported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https', 'storm']

{"golden_diff": "diff --git a/lib/rucio/daemons/badreplicas/necromancer.py b/lib/rucio/daemons/badreplicas/necromancer.py\n--- a/lib/rucio/daemons/badreplicas/necromancer.py\n+++ b/lib/rucio/daemons/badreplicas/necromancer.py\n@@ -13,7 +13,7 @@\n # limitations under the License.\n #\n # Authors:\n-# - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2017\n+# - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2019\n # - Vincent Garonne <vgaronne@gmail.com>, 2015-2018\n # - Mario Lassnig <mario.lassnig@cern.ch>, 2015\n # - Wen Guan <wguan.icedew@gmail.com>, 2015\n@@ -21,17 +21,21 @@\n #\n # PY3K COMPATIBLE\n \n+from __future__ import division\n+\n import logging\n import os\n import socket\n import threading\n import time\n \n+from math import ceil\n from sys import exc_info, stdout, argv\n from traceback import format_exception\n \n from rucio.db.sqla.constants import ReplicaState\n from rucio.common.config import config_get\n+from rucio.common.utils import chunks\n from rucio.common.exception import DatabaseException\n from rucio.core import monitor, heartbeat\n from rucio.core.replica import list_bad_replicas, get_replicas_state, list_bad_replicas_history, update_bad_replicas_history\n@@ -112,11 +116,18 @@\n             now = time.time()\n             if (now - update_history_time) > update_history_threshold:\n                 logging.info(prepend_str + 'Last update of history table %s seconds ago. Running update.' % (now - update_history_time))\n-                bad_replicas = list_bad_replicas_history(limit=10000000,\n+                bad_replicas = list_bad_replicas_history(limit=1000000,\n                                                          thread=heart_beat['assign_thread'],\n                                                          total_threads=heart_beat['nr_threads'])\n                 for rse_id in bad_replicas:\n-                    update_bad_replicas_history(bad_replicas[rse_id], rse_id)\n+                    chunk_size = 1000\n+                    nchunk = int(ceil(len(bad_replicas[rse_id]) / chunk_size))\n+                    logging.debug(prepend_str + 'Update history for rse_id %s' % (rse_id))\n+                    cnt = 0\n+                    for chunk in chunks(bad_replicas[rse_id], chunk_size):\n+                        logging.debug(prepend_str + ' History for rse_id %s : chunk %i/%i' % (rse_id, cnt, nchunk))\n+                        cnt += 1\n+                        update_bad_replicas_history(chunk, rse_id)\n                 logging.info(prepend_str + 'History table updated in %s seconds' % (time.time() - now))\n                 update_history_time = time.time()\n \ndiff --git a/lib/rucio/rse/protocols/__init__.py b/lib/rucio/rse/protocols/__init__.py\n--- a/lib/rucio/rse/protocols/__init__.py\n+++ b/lib/rucio/rse/protocols/__init__.py\n@@ -7,4 +7,4 @@\n # These are all protocols which are considered in good health\n # based on generally available support libraries, native dependencies,\n # and implementations in Rucio\n-supported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https']\n+supported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https', 'storm']\n", "issue": "Optimize the update of the final states in the necromancer\nMotivation\r\n----------\r\nCurrently the necromancer update up-to 10000000 bad replicas states in one transaction\r\n\r\n\r\nModification\r\n------------\r\nUpdate by chunks\r\n\r\n\n", "before_files": [{"content": "# Copyright European Organization for Nuclear Research (CERN)\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 http://www.apache.org/licenses/LICENSE-2.0\n\n# These are all protocols which are considered in good health\n# based on generally available support libraries, native dependencies,\n# and implementations in Rucio\nsupported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https', 'storm']\n", "path": "lib/rucio/rse/protocols/__init__.py"}, {"content": "# Copyright 2014-2018 CERN for the benefit of the ATLAS collaboration.\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#\n# Authors:\n# - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2017\n# - Vincent Garonne <vgaronne@gmail.com>, 2015-2018\n# - Mario Lassnig <mario.lassnig@cern.ch>, 2015\n# - Wen Guan <wguan.icedew@gmail.com>, 2015\n# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019\n#\n# PY3K COMPATIBLE\n\nimport logging\nimport os\nimport socket\nimport threading\nimport time\n\nfrom sys import exc_info, stdout, argv\nfrom traceback import format_exception\n\nfrom rucio.db.sqla.constants import ReplicaState\nfrom rucio.common.config import config_get\nfrom rucio.common.exception import DatabaseException\nfrom rucio.core import monitor, heartbeat\nfrom rucio.core.replica import list_bad_replicas, get_replicas_state, list_bad_replicas_history, update_bad_replicas_history\nfrom rucio.core.rule import update_rules_for_lost_replica, update_rules_for_bad_replica\n\n\nlogging.basicConfig(stream=stdout,\n                    level=getattr(logging,\n                                  config_get('common', 'loglevel',\n                                             raise_exception=False,\n                                             default='DEBUG').upper()),\n                    format='%(asctime)s\\t%(process)d\\t%(levelname)s\\t%(message)s')\n\ngraceful_stop = threading.Event()\n\n\ndef necromancer(thread=0, bulk=5, once=False):\n    \"\"\"\n    Creates a Necromancer Worker that gets a list of bad replicas for a given hash,\n    identify lost DIDs and for non-lost ones, set the locks and rules for reevaluation.\n\n    :param thread: Thread number at startup.\n    :param bulk: The number of requests to process.\n    :param once: Run only once.\n    \"\"\"\n\n    sleep_time = 60\n    update_history_threshold = 3600\n    update_history_time = time.time()\n\n    executable = ' '.join(argv)\n    hostname = socket.getfqdn()\n    pid = os.getpid()\n    hb_thread = threading.current_thread()\n    heartbeat.sanity_check(executable=executable, hostname=hostname)\n\n    while not graceful_stop.is_set():\n\n        heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)\n        prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'] + 1, heart_beat['nr_threads'])\n\n        stime = time.time()\n        replicas = []\n        try:\n            replicas = list_bad_replicas(limit=bulk, thread=heart_beat['assign_thread'], total_threads=heart_beat['nr_threads'])\n\n            for replica in replicas:\n                scope, name, rse_id, rse = replica['scope'], replica['name'], replica['rse_id'], replica['rse']\n                logging.info(prepend_str + 'Working on %s:%s on %s' % (scope, name, rse))\n\n                list_replicas = get_replicas_state(scope=scope, name=name)\n                if ReplicaState.AVAILABLE not in list_replicas and ReplicaState.TEMPORARY_UNAVAILABLE not in list_replicas:\n                    logging.info(prepend_str + 'File %s:%s has no other available or temporary available replicas, it will be marked as lost' % (scope, name))\n                    try:\n                        update_rules_for_lost_replica(scope=scope, name=name, rse_id=rse_id, nowait=True)\n                        monitor.record_counter(counters='necromancer.badfiles.lostfile', delta=1)\n                    except DatabaseException as error:\n                        logging.info(prepend_str + '%s' % (str(error)))\n\n                else:\n                    rep = list_replicas.get(ReplicaState.AVAILABLE, [])\n                    unavailable_rep = list_replicas.get(ReplicaState.TEMPORARY_UNAVAILABLE, [])\n                    logging.info(prepend_str + 'File %s:%s can be recovered. Available sources : %s + Unavailable sources : %s' % (scope, name, str(rep), str(unavailable_rep)))\n                    try:\n                        update_rules_for_bad_replica(scope=scope, name=name, rse_id=rse_id, nowait=True)\n                        monitor.record_counter(counters='necromancer.badfiles.recovering', delta=1)\n                    except DatabaseException as error:\n                        logging.info(prepend_str + '%s' % (str(error)))\n\n            logging.info(prepend_str + 'It took %s seconds to process %s replicas' % (str(time.time() - stime), str(len(replicas))))\n        except Exception:\n            exc_type, exc_value, exc_traceback = exc_info()\n            logging.critical(prepend_str + ''.join(format_exception(exc_type, exc_value, exc_traceback)).strip())\n\n        if once:\n            break\n        else:\n            now = time.time()\n            if (now - update_history_time) > update_history_threshold:\n                logging.info(prepend_str + 'Last update of history table %s seconds ago. Running update.' % (now - update_history_time))\n                bad_replicas = list_bad_replicas_history(limit=10000000,\n                                                         thread=heart_beat['assign_thread'],\n                                                         total_threads=heart_beat['nr_threads'])\n                for rse_id in bad_replicas:\n                    update_bad_replicas_history(bad_replicas[rse_id], rse_id)\n                logging.info(prepend_str + 'History table updated in %s seconds' % (time.time() - now))\n                update_history_time = time.time()\n\n            tottime = time.time() - stime\n            if len(replicas) == bulk:\n                logging.info(prepend_str + 'Processed maximum number of replicas according to the bulk size. Restart immediately next cycle')\n            elif tottime < sleep_time:\n                logging.info(prepend_str + 'Will sleep for %s seconds' % (str(sleep_time - tottime)))\n                time.sleep(sleep_time - tottime)\n                continue\n\n    logging.info(prepend_str + 'Graceful stop requested')\n    heartbeat.die(executable, hostname, pid, hb_thread)\n    logging.info(prepend_str + 'Graceful stop done')\n\n\ndef run(threads=1, bulk=100, once=False):\n    \"\"\"\n    Starts up the necromancer threads.\n    \"\"\"\n\n    if once:\n        logging.info('Will run only one iteration in a single threaded mode')\n        necromancer(bulk=bulk, once=once)\n    else:\n        logging.info('starting necromancer threads')\n        thread_list = [threading.Thread(target=necromancer, kwargs={'once': once,\n                                                                    'thread': i,\n                                                                    'bulk': bulk}) for i in range(0, threads)]\n        [t.start() for t in thread_list]\n\n        logging.info('waiting for interrupts')\n\n        # Interruptible joins require a timeout.\n        while thread_list:\n            thread_list = [thread.join(timeout=3.14) for thread in thread_list if thread and thread.isAlive()]\n\n\ndef stop(signum=None, frame=None):\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n    graceful_stop.set()\n", "path": "lib/rucio/daemons/badreplicas/necromancer.py"}], "after_files": [{"content": "# Copyright European Organization for Nuclear Research (CERN)\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 http://www.apache.org/licenses/LICENSE-2.0\n\n# These are all protocols which are considered in good health\n# based on generally available support libraries, native dependencies,\n# and implementations in Rucio\nsupported_protocols = ['gsiftp', 'srm', 'root', 'davs', 'http', 'https', 'file', 's3', 's3+rucio', 's3+https', 'storm']\n", "path": "lib/rucio/rse/protocols/__init__.py"}, {"content": "# Copyright 2014-2018 CERN for the benefit of the ATLAS collaboration.\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#\n# Authors:\n# - Cedric Serfon <cedric.serfon@cern.ch>, 2014-2019\n# - Vincent Garonne <vgaronne@gmail.com>, 2015-2018\n# - Mario Lassnig <mario.lassnig@cern.ch>, 2015\n# - Wen Guan <wguan.icedew@gmail.com>, 2015\n# - Hannes Hansen <hannes.jakob.hansen@cern.ch>, 2018-2019\n#\n# PY3K COMPATIBLE\n\nfrom __future__ import division\n\nimport logging\nimport os\nimport socket\nimport threading\nimport time\n\nfrom math import ceil\nfrom sys import exc_info, stdout, argv\nfrom traceback import format_exception\n\nfrom rucio.db.sqla.constants import ReplicaState\nfrom rucio.common.config import config_get\nfrom rucio.common.utils import chunks\nfrom rucio.common.exception import DatabaseException\nfrom rucio.core import monitor, heartbeat\nfrom rucio.core.replica import list_bad_replicas, get_replicas_state, list_bad_replicas_history, update_bad_replicas_history\nfrom rucio.core.rule import update_rules_for_lost_replica, update_rules_for_bad_replica\n\n\nlogging.basicConfig(stream=stdout,\n                    level=getattr(logging,\n                                  config_get('common', 'loglevel',\n                                             raise_exception=False,\n                                             default='DEBUG').upper()),\n                    format='%(asctime)s\\t%(process)d\\t%(levelname)s\\t%(message)s')\n\ngraceful_stop = threading.Event()\n\n\ndef necromancer(thread=0, bulk=5, once=False):\n    \"\"\"\n    Creates a Necromancer Worker that gets a list of bad replicas for a given hash,\n    identify lost DIDs and for non-lost ones, set the locks and rules for reevaluation.\n\n    :param thread: Thread number at startup.\n    :param bulk: The number of requests to process.\n    :param once: Run only once.\n    \"\"\"\n\n    sleep_time = 60\n    update_history_threshold = 3600\n    update_history_time = time.time()\n\n    executable = ' '.join(argv)\n    hostname = socket.getfqdn()\n    pid = os.getpid()\n    hb_thread = threading.current_thread()\n    heartbeat.sanity_check(executable=executable, hostname=hostname)\n\n    while not graceful_stop.is_set():\n\n        heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)\n        prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'] + 1, heart_beat['nr_threads'])\n\n        stime = time.time()\n        replicas = []\n        try:\n            replicas = list_bad_replicas(limit=bulk, thread=heart_beat['assign_thread'], total_threads=heart_beat['nr_threads'])\n\n            for replica in replicas:\n                scope, name, rse_id, rse = replica['scope'], replica['name'], replica['rse_id'], replica['rse']\n                logging.info(prepend_str + 'Working on %s:%s on %s' % (scope, name, rse))\n\n                list_replicas = get_replicas_state(scope=scope, name=name)\n                if ReplicaState.AVAILABLE not in list_replicas and ReplicaState.TEMPORARY_UNAVAILABLE not in list_replicas:\n                    logging.info(prepend_str + 'File %s:%s has no other available or temporary available replicas, it will be marked as lost' % (scope, name))\n                    try:\n                        update_rules_for_lost_replica(scope=scope, name=name, rse_id=rse_id, nowait=True)\n                        monitor.record_counter(counters='necromancer.badfiles.lostfile', delta=1)\n                    except DatabaseException as error:\n                        logging.info(prepend_str + '%s' % (str(error)))\n\n                else:\n                    rep = list_replicas.get(ReplicaState.AVAILABLE, [])\n                    unavailable_rep = list_replicas.get(ReplicaState.TEMPORARY_UNAVAILABLE, [])\n                    logging.info(prepend_str + 'File %s:%s can be recovered. Available sources : %s + Unavailable sources : %s' % (scope, name, str(rep), str(unavailable_rep)))\n                    try:\n                        update_rules_for_bad_replica(scope=scope, name=name, rse_id=rse_id, nowait=True)\n                        monitor.record_counter(counters='necromancer.badfiles.recovering', delta=1)\n                    except DatabaseException as error:\n                        logging.info(prepend_str + '%s' % (str(error)))\n\n            logging.info(prepend_str + 'It took %s seconds to process %s replicas' % (str(time.time() - stime), str(len(replicas))))\n        except Exception:\n            exc_type, exc_value, exc_traceback = exc_info()\n            logging.critical(prepend_str + ''.join(format_exception(exc_type, exc_value, exc_traceback)).strip())\n\n        if once:\n            break\n        else:\n            now = time.time()\n            if (now - update_history_time) > update_history_threshold:\n                logging.info(prepend_str + 'Last update of history table %s seconds ago. Running update.' % (now - update_history_time))\n                bad_replicas = list_bad_replicas_history(limit=1000000,\n                                                         thread=heart_beat['assign_thread'],\n                                                         total_threads=heart_beat['nr_threads'])\n                for rse_id in bad_replicas:\n                    chunk_size = 1000\n                    nchunk = int(ceil(len(bad_replicas[rse_id]) / chunk_size))\n                    logging.debug(prepend_str + 'Update history for rse_id %s' % (rse_id))\n                    cnt = 0\n                    for chunk in chunks(bad_replicas[rse_id], chunk_size):\n                        logging.debug(prepend_str + ' History for rse_id %s : chunk %i/%i' % (rse_id, cnt, nchunk))\n                        cnt += 1\n                        update_bad_replicas_history(chunk, rse_id)\n                logging.info(prepend_str + 'History table updated in %s seconds' % (time.time() - now))\n                update_history_time = time.time()\n\n            tottime = time.time() - stime\n            if len(replicas) == bulk:\n                logging.info(prepend_str + 'Processed maximum number of replicas according to the bulk size. Restart immediately next cycle')\n            elif tottime < sleep_time:\n                logging.info(prepend_str + 'Will sleep for %s seconds' % (str(sleep_time - tottime)))\n                time.sleep(sleep_time - tottime)\n                continue\n\n    logging.info(prepend_str + 'Graceful stop requested')\n    heartbeat.die(executable, hostname, pid, hb_thread)\n    logging.info(prepend_str + 'Graceful stop done')\n\n\ndef run(threads=1, bulk=100, once=False):\n    \"\"\"\n    Starts up the necromancer threads.\n    \"\"\"\n\n    if once:\n        logging.info('Will run only one iteration in a single threaded mode')\n        necromancer(bulk=bulk, once=once)\n    else:\n        logging.info('starting necromancer threads')\n        thread_list = [threading.Thread(target=necromancer, kwargs={'once': once,\n                                                                    'thread': i,\n                                                                    'bulk': bulk}) for i in range(0, threads)]\n        [t.start() for t in thread_list]\n\n        logging.info('waiting for interrupts')\n\n        # Interruptible joins require a timeout.\n        while thread_list:\n            thread_list = [thread.join(timeout=3.14) for thread in thread_list if thread and thread.isAlive()]\n\n\ndef stop(signum=None, frame=None):\n    \"\"\"\n    Graceful exit.\n    \"\"\"\n    graceful_stop.set()\n", "path": "lib/rucio/daemons/badreplicas/necromancer.py"}]}