Commit 5473d025 authored by James Clark's avatar James Clark

overhauled daemon reg

parent dfc7b081
Pipeline #83621 passed with stages
in 1 minute and 56 seconds
......@@ -395,23 +395,19 @@ def daemon(aparser):
for rset in rsets:
minimum_gps = rsets[rset]['minimum-gps']
cache_try_interval = aparser.daemon_sleep
for ntry in range(MAX_CACHE_TRIES+1):
try:
LOGGER.debug("Finding %s in DiskCache", rset)
rsets[rset]['diskcache'] = DiskCacheFile(
aparser.cachefile,
minimum_gps=minimum_gps,
minimum_gps=rsets[rset]['minimum-gps'],
maximum_gps=rsets[rset]['maximum-gps'],
regexp=rsets[rset]['regexp'],
prune=True,
update_file_count=True)
last_check = time.time()
# update the minimum time to scan from to the last-modified
# time of the cache file
minimum_gps = rsets[rset]['diskcache'].mtime()
# break out of the retry-loop
break
except (IOError, OSError, StopIteration) as cache_error:
# StopIteration if cachefile is incomplete
......@@ -427,28 +423,28 @@ def daemon(aparser):
time.sleep(cache_try_interval)
cache_try_interval *= 2
# If check enabled or if diskcache has updated, heck for new data
if force_check_cache or \
rsets[rset]['diskcache'].mtime() > last_check:
# Register any new files found
if [entry for entry in rsets[rset]['diskcache'].expand()]:
LOGGER.info('Found new files, registering')
LOGGER.info("%s: looking for new data", rset)
# Inject each rset (ignores pre-registered files)
inject_data(rset, rsets[rset], rse_info,
allow_uploads=aparser.allow_uploads,
dry_run=aparser.dry_run)
# Disable force check-cache after initial read
force_check_cache = False
# Update the minimum time to register from next round using
# segments from the diskcache
segment_ends = []
for entry in rsets[rset]['diskcache']:
for seg in entry['segmentlist']:
segment_ends.append(max(seg))
rsets[rset]['minimum-gps'] = max(segment_ends)
else:
LOGGER.info("%s: diskcache not modified", rset)
LOGGER.info('No new files found')
if aparser.run_once:
# break out of the daemon while-loop
break
# Update last-modified time to that of the cache
last_check = time.time()
# Snooze to allow cache updates
LOGGER.info("Going to sleep for %d s...", aparser.daemon_sleep)
......
......@@ -94,9 +94,7 @@ class DatasetInjector(object):
3) Create Rucio dataset
4) Register Rucio dataset
data is a dictionary with either a list of files to register or another
dictionary including a diskcache object. If the latter, DatasetInjector
will parse out the file list and metadata.
data is a dictionary with a list of files to register
"""
# pylint: disable=too-many-instance-attributes,too-many-arguments
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment