Threading e SqlFileStream. Il processo non può accedere al file specificato perché è stato aperto in un'altra transazione

Question

Sto estraendo il contenuto dei file nella tabella dei file SQL. Il seguente codice funziona se non utilizzo Parallel.

Ricevo la seguente eccezione, durante la lettura simultanea del flusso di file sql (Parallela).

Il processo non può accedere al file specificato perché è stato aperto in un'altra transazione.

TL; DR:

Durante la lettura di un file da FileTable (utilizzando GET_FILESTREAM_TRANSACTION_CONTEXT) in una Parallel.ForEach ottengo l'eccezione di cui sopra.

Codice di esempio per voi di provare:

https://gist.github.com/NerdPad/6d9b399f2f5f5e5c6519

Longer Version:

Fetch allegati, ed estrarre il contenuto:

var documents = new List<ExtractedContent>(); 
using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) 
{ 
    var attachments = await dao.GetAttachmentsAsync(); 

    // Extract the content simultaneously 
    // documents = attachments.ToDbDocuments().ToList(); // This works 
    Parallel.ForEach(attachments, a => documents.Add(a.ToDbDocument())); // this doesn't 

    ts.Complete(); 
} 

DAO Leggi File Table:

public async Task<IEnumerable<SearchAttachment>> GetAttachmentsAsync() 
{ 
    try 
    { 
     var commandStr = "...."; 

     IEnumerable<SearchAttachment> attachments = null; 
     using (var connection = new SqlConnection(this.DatabaseContext.Database.Connection.ConnectionString)) 
     using (var command = new SqlCommand(commandStr, connection)) 
     { 
      connection.Open(); 

      using (var reader = await command.ExecuteReaderAsync()) 
      { 
       attachments = reader.ToSearchAttachments().ToList(); 
      } 
     } 

     return attachments; 
    } 
    catch (System.Exception) 
    { 
     throw; 
    } 
} 

creare oggetti per ogni file: L'oggetto contiene un riferimento al GET_FILESTREAM_TRANSACTION_CONTEXT

public static IEnumerable<SearchAttachment> ToSearchAttachments(this SqlDataReader reader) 
{ 
    if (!reader.HasRows) 
    { 
     yield break; 
    } 

    // Convert each row to SearchAttachment 
    while (reader.Read()) 
    { 
     yield return new SearchAttachment 
     { 
      ... 
      ... 
      UNCPath = reader.To<string>(Constants.UNCPath), 
      ContentStream = reader.To<byte[]>(Constants.Stream) // GET_FILESTREAM_TRANSACTION_CONTEXT() 
      ... 
      ... 
     }; 
    } 
} 

Leggere il file utilizzando SqlFileStream: eccezione è buttato qui

public static ExtractedContent ToDbDocument(this SearchAttachment attachment) 
{ 
    // Read the file 
    // Exception is thrown here 
    using (var stream = new SqlFileStream(attachment.UNCPath, attachment.ContentStream, FileAccess.Read, FileOptions.SequentialScan, 4096)) 
    { 
     ... 
     // extract content from the file 
    } 

    .... 
} 

Update 1:

Secondo this articolo sembra che potrebbe essere un problema livello di isolamento. Qualcuno ha mai affrontato un problema simile?

Answer 1

L'operazione non scorre dentro al Parallel.ForEach, è necessario portare manualmente l'operazione in.

//Switched to a thread safe collection. 
var documents = new ConcurrentQueue<ExtractedContent>(); 
using (var ts = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled)) 
{ 
    var attachments = await dao.GetAttachmentsAsync(); 
    //Grab a reference to the current transaction. 
    var transaction = Transaction.Current; 
    Parallel.ForEach(attachments, a => 
    { 
     //Spawn a dependant clone of the transaction 
     using (var depTs = transaction.DependentClone(DependentCloneOption.RollbackIfNotComplete)) 
     { 
      documents.Enqueue(a.ToDbDocument()); 
      depTs.Complete(); 
     } 
    }); 

    ts.Complete(); 
} 

Ho anche passato da List<ExtractedContent> a ConcurrentQueue<ExtractedContent> perché non ti è permesso chiamata .Add( in un elenco da più thread a lo stesso tempo.