<commit_before>package org.ovirt.engine.core.bll.eventqueue;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.locks.ReentrantLock;

import javax.ejb.ConcurrencyManagement;
import javax.ejb.ConcurrencyManagementType;
import javax.ejb.Local;
import javax.ejb.Singleton;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;

import org.ovirt.engine.core.common.eventqueue.Event;
import org.ovirt.engine.core.common.eventqueue.EventQueue;
import org.ovirt.engine.core.common.eventqueue.EventResult;
import org.ovirt.engine.core.common.eventqueue.EventType;
import org.ovirt.engine.core.common.utils.Pair;
import org.ovirt.engine.core.compat.Guid;
import org.ovirt.engine.core.utils.log.Log;
import org.ovirt.engine.core.utils.log.LogFactory;
import org.ovirt.engine.core.utils.threadpool.ThreadPoolUtil;

@Singleton(name = "EventQueue")
@ConcurrencyManagement(ConcurrencyManagementType.BEAN)
@TransactionAttribute(TransactionAttributeType.SUPPORTS)
@Local(EventQueue.class)
public class EventQueueMonitor implements EventQueue {

    private static Log log = LogFactory.getLog(EventQueueMonitor.class);

    private static final ConcurrentMap<Guid, ReentrantLock> poolsLockMap = new ConcurrentHashMap<Guid, ReentrantLock>();
    private static final Map<Guid, LinkedList<Pair<Event, FutureTask<EventResult>>>> poolsEventsMap =
            new HashMap<Guid, LinkedList<Pair<Event, FutureTask<EventResult>>>>();
    private static final Map<Guid, Event> poolCurrentEventMap = new HashMap<Guid, Event>();

    @Override
    public void submitEventAsync(Event event, Callable<EventResult> callable) {
        submitTaskInternal(event, callable);
    }

    @Override
    public EventResult submitEventSync(Event event, Callable<EventResult> callable) {
        FutureTask<EventResult> task = submitTaskInternal(event, callable);
        if (task != null) {
            try {
                return task.get();
            } catch (Exception e) {
                log.errorFormat("Failed at submitEventSync, for pool {0} with exception {1}",
                        event.getStoragePoolId(), e);
            }
        }
        return null;
    }

    private FutureTask<EventResult> submitTaskInternal(Event event,
            Callable<EventResult> callable) {
        FutureTask<EventResult> task = null;
        Guid storagePoolId = event.getStoragePoolId();
        ReentrantLock lock = getPoolLock(storagePoolId);
        lock.lock();
        try {
            Event currentEvent = poolCurrentEventMap.get(storagePoolId);<commit_msg>engine: Fixing possible ConcurrentModificationException  at EventQueueMonitor<commit_after>package org.ovirt.engine.core.bll.eventqueue;

import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.locks.ReentrantLock;

import javax.ejb.ConcurrencyManagement;
import javax.ejb.ConcurrencyManagementType;
import javax.ejb.Local;
import javax.ejb.Singleton;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;

import org.ovirt.engine.core.common.eventqueue.Event;
import org.ovirt.engine.core.common.eventqueue.EventQueue;
import org.ovirt.engine.core.common.eventqueue.EventResult;
import org.ovirt.engine.core.common.eventqueue.EventType;
import org.ovirt.engine.core.common.utils.Pair;
import org.ovirt.engine.core.compat.Guid;
import org.ovirt.engine.core.utils.log.Log;
import org.ovirt.engine.core.utils.log.LogFactory;
import org.ovirt.engine.core.utils.threadpool.ThreadPoolUtil;

@Singleton(name = "EventQueue")
@ConcurrencyManagement(ConcurrencyManagementType.BEAN)
@TransactionAttribute(TransactionAttributeType.SUPPORTS)
@Local(EventQueue.class)
public class EventQueueMonitor implements EventQueue {

    private static Log log = LogFactory.getLog(EventQueueMonitor.class);

    private static final ConcurrentMap<Guid, ReentrantLock> poolsLockMap = new ConcurrentHashMap<Guid, ReentrantLock>();
    private static final Map<Guid, LinkedList<Pair<Event, FutureTask<EventResult>>>> poolsEventsMap =
            new ConcurrentHashMap<Guid, LinkedList<Pair<Event, FutureTask<EventResult>>>>();
    private static final Map<Guid, Event> poolCurrentEventMap = new ConcurrentHashMap<Guid, Event>();

    @Override
    public void submitEventAsync(Event event, Callable<EventResult> callable) {
        submitTaskInternal(event, callable);
    }

    @Override
    public EventResult submitEventSync(Event event, Callable<EventResult> callable) {
        FutureTask<EventResult> task = submitTaskInternal(event, callable);
        if (task != null) {
            try {
                return task.get();
            } catch (Exception e) {
                log.errorFormat("Failed at submitEventSync, for pool {0} with exception {1}",
                        event.getStoragePoolId(), e);
            }
        }
        return null;
    }

    private FutureTask<EventResult> submitTaskInternal(Event event,
            Callable<EventResult> callable) {
        FutureTask<EventResult> task = null;
        Guid storagePoolId = event.getStoragePoolId();
        ReentrantLock lock = getPoolLock(storagePoolId);
        lock.lock();
        try {
            Event currentEvent = poolCurrentEventMap.get(storagePoolId);
{'fraction_non_alphanumeric': 0.09905163329820864, 'fraction_numerical': 0.0007024938531787847, 'mean_word_length': 4.200913242009133, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3756662', 'n_tokens_mistral': 1607, 'n_tokens_neox': 1589, 'n_words': 314}