C#•2y ago

Iterating over AsyncEnumerable and writing to MemoryStream

Mainly a question on how you all would go about this, I am thinking of just initializing a SemaphoreSlim(1,1) and waiting while inside the iteration then writing to the memory stream. Any opposition to this? The compiled code inside an AsyncEnumerable seems to be what I'd expect so I would need some locking while writing to the stream regardless.

13 Replies

NacDanOP•2y ago

For instance

        SemaphoreSlim semaphoreSlim = new(1, 1);
        await foreach (var batch in batches)
        {
            try
            {
                await semaphoreSlim.WaitAsync();
                var records = batch.AsCollection();
                totalBatches++;
                totalBatches += records.Count;
                
                await WriteToParquet(memoryStream, records);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        }

        SemaphoreSlim semaphoreSlim = new(1, 1);
        await foreach (var batch in batches)
        {
            try
            {
                await semaphoreSlim.WaitAsync();
                var records = batch.AsCollection();
                totalBatches++;
                totalBatches += records.Count;
                
                await WriteToParquet(memoryStream, records);
            }
            finally
            {
                semaphoreSlim.Release();
            }
        }

Mayor McCheese•2y ago

Why not block outside the foreach? Or are you concerned about the order going into parquet?

Aaron•2y ago

what's the semaphore for? it seems to be doing nothing here

NacDanOP•2y ago

Correct, WriteToParquet takes in the existing memorystream

Mayor McCheese•2y ago

My comment on this somehow truncated 😦 Are you rationally worried?

NacDanOP•2y ago

Force synchonization while writing to the MemoryStream

Aaron•2y ago

await foreach (var batch in batches)
        {
            var records = batch.AsCollection();
            totalBatches++;
            totalBatches += records.Count;
                
            await WriteToParquet(memoryStream, records);
        }

await foreach (var batch in batches)
        {
            var records = batch.AsCollection();
            totalBatches++;
            totalBatches += records.Count;
                
            await WriteToParquet(memoryStream, records);
        }

that's the exact same code it will behave the same way

NacDanOP•2y ago

Is that because it is disposed before the next iteration?

Aaron•2y ago

what is disposed

NacDanOP•2y ago

the current enumerator My main concern was unintentionally writing to the stream concurrently. I'm not all too familiar with AsyncEnumerables

Mayor McCheese•2y ago

It's possible I suppose that batches could come in the wrong order, we don't know the implementation of batches, but you won't solve that with a semaphore

Aaron•2y ago

AsyncEnumerable is essentially just a normal enumerable, but made of tasks instead of normal items it doesn't inherently make you process items in parallel this code will still process items one at a time if you wanted to process items in parallel, you'd have to use Parallel.ForEachAsync

NacDanOP•2y ago

I am seeing the compiler transformations right now yes I see now Of what it looks like under the hood Gotcha that makes sense. Thanks

Gaming

Programming

Iterating over AsyncEnumerable and writing to MemoryStream