@transactional_session
def list_and_mark_unlocked_replicas(limit, bytes=None, rse_id=None, delay_seconds=600, only_delete_obsolete=False, session=None):
    '\n    List RSE File replicas with no locks.\n\n    :param limit:                    Number of replicas returned.\n    :param bytes:                    The amount of needed bytes.\n    :param rse_id:                   The rse_id.\n    :param delay_seconds:            The delay to query replicas in BEING_DELETED state\n    :param only_delete_obsolete      If set to True, will only return the replicas with EPOCH tombstone\n    :param session:                  The database session in use.\n\n    :returns: a list of dictionary replica.\n    '
    none_value = None
    query = session.query(models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.path, models.RSEFileAssociation.bytes, models.RSEFileAssociation.tombstone, models.RSEFileAssociation.state).with_hint(models.RSEFileAssociation, 'INDEX_RS_ASC(replicas REPLICAS_TOMBSTONE_IDX)  NO_INDEX_FFS(replicas REPLICAS_TOMBSTONE_IDX)', 'oracle').filter((models.RSEFileAssociation.tombstone < datetime.utcnow())).filter((models.RSEFileAssociation.lock_cnt == 0)).filter((case([((models.RSEFileAssociation.tombstone != none_value), models.RSEFileAssociation.rse_id)]) == rse_id)).filter(or_(models.RSEFileAssociation.state.in_((ReplicaState.AVAILABLE, ReplicaState.UNAVAILABLE, ReplicaState.BAD)), and_((models.RSEFileAssociation.state == ReplicaState.BEING_DELETED), (models.RSEFileAssociation.updated_at < (datetime.utcnow() - timedelta(seconds=delay_seconds)))))).filter((~ exists(select([1]).prefix_with('/*+ INDEX(SOURCES SOURCES_SC_NM_DST_IDX) */', dialect='oracle').where(and_((models.RSEFileAssociation.scope == models.Source.scope), (models.RSEFileAssociation.name == models.Source.name), (models.RSEFileAssociation.rse_id == models.Source.rse_id)))))).with_for_update(skip_locked=True).order_by(models.RSEFileAssociation.tombstone)
    needed_space = bytes
    (total_bytes, total_files) = (0, 0)
    rows = []
    replica_clause = []
    for (scope, name, path, bytes, tombstone, state) in query.yield_per(1000):
        replica_cnt = session.query(func.count(models.RSEFileAssociation.scope)).with_hint(models.RSEFileAssociation, 'index(REPLICAS REPLICAS_PK)', 'oracle').filter(and_((models.RSEFileAssociation.scope == scope), (models.RSEFileAssociation.name == name), (models.RSEFileAssociation.rse_id != rse_id))).one()
        if (replica_cnt[0] > 1):
            if (state != ReplicaState.UNAVAILABLE):
                if (tombstone != OBSOLETE):
                    if only_delete_obsolete:
                        break
                    if ((needed_space is not None) and (total_bytes > needed_space)):
                        break
                total_bytes += bytes
                total_files += 1
                if (total_files > limit):
                    break
            rows.append({'scope': scope, 'name': name, 'path': path, 'bytes': bytes, 'tombstone': tombstone, 'state': state})
            replica_clause.append(and_((models.RSEFileAssociation.scope == scope), (models.RSEFileAssociation.name == name), (models.RSEFileAssociation.rse_id == rse_id)))
        else:
            request_cnt = session.query(func.count()).with_hint(models.Request, 'INDEX(requests REQUESTS_SCOPE_NAME_RSE_IDX)', 'oracle').filter(and_((models.Request.scope == scope), (models.Request.name == name))).one()
            if (request_cnt[0] == 0):
                if (tombstone != OBSOLETE):
                    if only_delete_obsolete:
                        break
                    if ((needed_space is not None) and (total_bytes > needed_space)):
                        break
                total_bytes += bytes
                total_files += 1
                if (total_files > limit):
                    break
                rows.append({'scope': scope, 'name': name, 'path': path, 'bytes': bytes, 'tombstone': tombstone, 'state': state})
                replica_clause.append(and_((models.RSEFileAssociation.scope == scope), (models.RSEFileAssociation.name == name), (models.RSEFileAssociation.rse_id == rse_id)))
    for chunk in chunks(replica_clause, 100):
        session.query(models.RSEFileAssociation).filter(or_(*chunk)).with_hint(models.RSEFileAssociation, text='INDEX(REPLICAS REPLICAS_PK)', dialect_name='oracle').update({'updated_at': datetime.utcnow(), 'state': ReplicaState.BEING_DELETED, 'tombstone': datetime(1970, 1, 1)}, synchronize_session=False)
    return rows
List RSE File replicas with no locks.

:param limit:                    Number of replicas returned.
:param bytes:                    The amount of needed bytes.
:param rse_id:                   The rse_id.
:param delay_seconds:            The delay to query replicas in BEING_DELETED state
:param only_delete_obsolete      If set to True, will only return the replicas with EPOCH tombstone
:param session:                  The database session in use.

:returns: a list of dictionary replica.
@transactional_session
def list_and_mark_unlocked_replicas(limit, bytes=None, rse_id=None, delay_seconds=600, only_delete_obsolete=False, session=None):
    '\n    List RSE File replicas with no locks.\n\n    :param limit:                    Number of replicas returned.\n    :param bytes:                    The amount of needed bytes.\n    :param rse_id:                   The rse_id.\n    :param delay_seconds:            The delay to query replicas in BEING_DELETED state\n    :param only_delete_obsolete      If set to True, will only return the replicas with EPOCH tombstone\n    :param session:                  The database session in use.\n\n    :returns: a list of dictionary replica.\n    '
    none_value = None
    query = session.query(models.RSEFileAssociation.scope, models.RSEFileAssociation.name, models.RSEFileAssociation.path, models.RSEFileAssociation.bytes, models.RSEFileAssociation.tombstone, models.RSEFileAssociation.state).with_hint(models.RSEFileAssociation, 'INDEX_RS_ASC(replicas REPLICAS_TOMBSTONE_IDX)  NO_INDEX_FFS(replicas REPLICAS_TOMBSTONE_IDX)', 'oracle').filter((models.RSEFileAssociation.tombstone < datetime.utcnow())).filter((models.RSEFileAssociation.lock_cnt == 0)).filter((case([((models.RSEFileAssociation.tombstone != none_value), models.RSEFileAssociation.rse_id)]) == rse_id)).filter(or_(models.RSEFileAssociation.state.in_((ReplicaState.AVAILABLE, ReplicaState.UNAVAILABLE, ReplicaState.BAD)), and_((models.RSEFileAssociation.state == ReplicaState.BEING_DELETED), (models.RSEFileAssociation.updated_at < (datetime.utcnow() - timedelta(seconds=delay_seconds)))))).filter((~ exists(select([1]).prefix_with('/*+ INDEX(SOURCES SOURCES_SC_NM_DST_IDX) */', dialect='oracle').where(and_((models.RSEFileAssociation.scope == models.Source.scope), (models.RSEFileAssociation.name == models.Source.name), (models.RSEFileAssociation.rse_id == models.Source.rse_id)))))).with_for_update(skip_locked=True).order_by(models.RSEFileAssociation.tombstone)
    needed_space = bytes
    (total_bytes, total_files) = (0, 0)
    rows = []
    replica_clause = []
    for (scope, name, path, bytes, tombstone, state) in query.yield_per(1000):
        replica_cnt = session.query(func.count(models.RSEFileAssociation.scope)).with_hint(models.RSEFileAssociation, 'index(REPLICAS REPLICAS_PK)', 'oracle').filter(and_((models.RSEFileAssociation.scope == scope), (models.RSEFileAssociation.name == name), (models.RSEFileAssociation.rse_id != rse_id))).one()
        if (replica_cnt[0] > 1):
            if (state != ReplicaState.UNAVAILABLE):
                if (tombstone != OBSOLETE):
                    if only_delete_obsolete:
                        break
                    if ((needed_space is not None) and (total_bytes > needed_space)):
                        break
                total_bytes += bytes
                total_files += 1
                if (total_files > limit):
                    break
            rows.append({'scope': scope, 'name': name, 'path': path, 'bytes': bytes, 'tombstone': tombstone, 'state': state})
            replica_clause.append(and_((models.RSEFileAssociation.scope == scope), (models.RSEFileAssociation.name == name), (models.RSEFileAssociation.rse_id == rse_id)))
        else:
            request_cnt = session.query(func.count()).with_hint(models.Request, 'INDEX(requests REQUESTS_SCOPE_NAME_RSE_IDX)', 'oracle').filter(and_((models.Request.scope == scope), (models.Request.name == name))).one()
            if (request_cnt[0] == 0):
                if (tombstone != OBSOLETE):
                    if only_delete_obsolete:
                        break
                    if ((needed_space is not None) and (total_bytes > needed_space)):
                        break
                total_bytes += bytes
                total_files += 1
                if (total_files > limit):
                    break
                rows.append({'scope': scope, 'name': name, 'path': path, 'bytes': bytes, 'tombstone': tombstone, 'state': state})
                replica_clause.append(and_((models.RSEFileAssociation.scope == scope), (models.RSEFileAssociation.name == name), (models.RSEFileAssociation.rse_id == rse_id)))
    for chunk in chunks(replica_clause, 100):
        session.query(models.RSEFileAssociation).filter(or_(*chunk)).with_hint(models.RSEFileAssociation, text='INDEX(REPLICAS REPLICAS_PK)', dialect_name='oracle').update({'updated_at': datetime.utcnow(), 'state': ReplicaState.BEING_DELETED, 'tombstone': datetime(1970, 1, 1)}, synchronize_session=False)
    return rows