Транзакции и многопоточный доступ к базе данных

Недавно мне понадобилось выполнить следующий код (представлен в максимально упрощенном виде):

public void Start()
{
    using (var transactionScope = new TransactionScope())
    {
        ...
        GetOrCreateCompany(someValue);
        ...
        transactionScope.Complete();
    }
}

private Company GetOrCreateCompany(string companyName)
{
    var company = _companiesRepository.GetCompany(companyName); //простая выборка из таблицы по названию; если компания не найдена - возвращается null
    if (company == null)
        company = _companiesRepository.Add(companyName);
    return company;
}

Код этот выполнялся в многопоточной среде, где каждый поток на вход получал метод Start (а значит у каждого потока была своя транзакция).

У этого, казалось бы, простого кода есть несколько нюансов, о которых и пойдет речь под катом.



Для поставленной задачи есть общее решение: выставить необходимые constraint'ы, заключить транзакцию в цикл, и если возникает исключение — пробовать заново (если превышено некоторое число попыток — выкидываем исключение наверх). Однако в случае большого количества потоков в результате конфликтующих блокировок этот подход работает очень медленно (проще сказать, что не работает). Поэтому я начал реализовывать более специфичное решение.

Ниже код моего первого решения (в дальнейшем специфика транзакций и локов будет рассматривать на примере MySQL):

public void Start()
{
    using (var transactionScope = new TransactionScope(TransactionScopeOption.Requires, new TransactionOptions() { IsolationLevel = IsolationLevel.Serializable }))
    {
        ...
        GetOrCreateCompany(someValue);
        ...
        transactionScope.Complete();
    }
}

private Company GetOrCreateCompany(string companyName)
{
    var company = _companiesRepository.GetCompanyWithWriteLock(companyName); //делаем выборку в стиле SELECT ... FOR UPDATE
    if (company == null)
        company = _companiesRepository.Add(companyName);
    return company;
}

В приведенном коде блокируется выборка компании вместе с соответствующим range (за счет использования Serializable) и разблокируется только после коммита. Следующая транзакция, которая попытается считать ту же компанию, будет ждать коммита заблокировавшей транзакции.

В общем-то это решение работает. Но смотрим на закон Амдала, потом на наш код, потом еще раз на Закон Амдала — и становится грустно: мало того, что мы вынуждены блокировать записи вплоть до коммита транзакции (а ведь объем кода ниже блокировки/блокировок не фиксирован), так еще и ставим write-lock даже если выборка вернет искомую компанию (а значит добавлять ее не нужно). Если первый пункт является основой этого решения, и от него никуда не деться, то со вторым мы еще поборемся.

Выборка компании не вернет нам null в двух случаях:

1. Компания была добавлена в рамках предыдущей транзакции, которая уже закоммичена
2. Компания была добавлена в рамках текущей транзакции, и станет видна другим транзакциям только после коммита текущей транзакции

Давайте попробуем отдельно обрабатывать первый случай, и прежде чем блокировать, будем выполнять запрос вне текущей транзакции, который будет проверять, нет ли искомых данных среди уже закоммиченных транзакций:

private Company GetOrCreateCompany(string companyName)
{
    Company company;
    //сделаем запрос вне текущей транзакции, который проверит, нет ли искомой компании среди уже закоммиченных. Если есть - мы просто вернем ее, не воспользовавшись блокировками уровня кода
    using (var independentTransactionScope = new TransactionScope(TransactionScopeOption.Suppress))
    {
         company = _companiesRepository.GetCompany(companyName);
    }
    //если компания видна вне нашей текущей транзакции, значит она была добавлена уже закоммиченной транзакцией - возвращаем ее без блокировки
    if (company != null)
       return;

    //company может быть null в двух случаях: 1. она была добавлена в текущей транзакции, и другие транзакции ее еще не видят 2. компания еще не была добавлена
    var company = _companiesRepository.GetCompanyWithWriteLock(companyName);
    if (company == null) //компания не видна даже в текущей транзакции - значит она еще не была создана
        company = _companiesRepository.Add(companyName);
    }
}

Небольшой минус этой оптимизации в том, что приходится дублировать выборку, если значения не видно вне текущей транзакции (иными словами, если внутри независимого TransactionScope вернулся null для выборки). Но на фоне выигрыша в производительности на эту лишнюю выборку можно закрыть глаза.

Зачем я написал эту статью? Меня не покидает ощущение, что я пошел не в том направлении: проблема тривиальна, а решение… не очень. Кроме того, мне не удалось найти примеров решения этой задачи, а это еще больше меня озадачивает. Надеюсь, комментаторы прояснят ситуацию, предложив альтернативные решения, либо согласившись с моим.

Update
В личной переписке хабрачеловек Shaddix (по совместительству, мой коллега) предложил интересное развитие идеи использования независимой транзакции — поместить в независимую транзакцию не только считывание, но и добавление. Эта, на первый взгляд, небольшая модифиция очень сильно все меняет.

Сначала реализация:

public class TransactionCode
{
    private static readonly object _lockObject = new object();

    public void Start()
    {
        using (var transactionScope = new TransactionScope())
        {
            ...
            GetOrCreateCompany(someValue);
            ...     
            transactionScope.Complete();
         }
    }

    private Company GetOrCreateCompany(string companyName)
    {
         Company company;
         //проверяем, нет ли компании среду уже закоммиченных транзакций
         using (var indepdentTransactionScope = new TransactionScope(TransactionScopeOption.RequiresNew))
         {
              var company = _companiesRepository.GetCompany(companyName);
         }
         if (company != null)
             return;
         //если среди закоммиченных нет, значит ее нужно добавить

         //добавляем в независимой транзакции
         using (var independentTransactionScope = new TransactionScope(TransactionScopeOption.RequiresNew))
         {
             var company = _companiesRepository.GetCompanyWithWriteLock(companyName);
             if (company == null)
                company = _companiesRepository.Add(companyName);
             independentTransactionScope.Complete();
         }           
     }
}

Теперь мы не добавляем компанию в основной транзакции, а используем для этих целей независимую транзакцию. Основная фишка здесь в том, что так как добавление происходит в независимой транзакции, которая сразу коммитится, то добавленное значение сразу (а не после коммита основной транзакции, как это было в предыдущем решении) станет видно другим транзакциям. Ну и в том, что этот подход эффективнее, нет никаких сомнений: в моем случае прирост производительности составил 300% (причем чем динамичнее данные, тем больше прирост за счет меньшего времени жизни блокировок).
Но у этого решения есть и недостатки:
1. Самый главный недостаток — если основная транзакция откатится, то добавленные в независимой транзакции данные не откатятся (мы их закоммитили). В общем случае недостаток довольно критичный… но не для меня: в методах типа GetOrCreate у меня добавляются относительно независимые данные, которые все равно рано или поздно добавятся: не в этой транзакции, так в следующей.
2. У меня есть личные предубеждения против использования not-readonly транзакций с TransactionScopeOption.RequiresNew и TransactionScopeOption.Suppress (раскрою их в следующем посте).

Таким образом, несмотря на то, что последнее решение мне нравится больше, в общем случае нельзя сказать, что одно из них лучше другого — они разные.