[core] Flush blocks asynchronously when hashing from memory

If we're reading data from the in-memory cache, we should not
block on the flush-to-disk aspect. As we are holding the
incremental hash lock for that piece we need to avoid lengthy
delays from operations like flushing to disk.

src/MonoTorrent/MonoTorrent.Client.PieceWriters/MemoryCache.cs

@@ -133,7 +133,7 @@ public async ReusableTask<bool> ReadAsync (ITorrentData torrent, BlockInfo block
             return await ReadFromFilesAsync (torrent, block, buffer).ConfigureAwait (false) == block.RequestLength;
         }
 
-        public async ReusableTask<bool> ReadFromCacheAsync (ITorrentData torrent, BlockInfo block, byte[] buffer)
+        public ReusableTask<bool> ReadFromCacheAsync (ITorrentData torrent, BlockInfo block, byte[] buffer)
         {
             if (torrent == null)
                 throw new ArgumentNullException (nameof (torrent));
@@ -146,24 +146,27 @@ public async ReusableTask<bool> ReadFromCacheAsync (ITorrentData torrent, BlockI
                     if (cached.Block != block)
                         continue;
 
-                    if (cached.Flushing) {
-                        Buffer.BlockCopy (cached.Buffer, 0, buffer, 0, block.RequestLength);
-                    } else {
+                    Buffer.BlockCopy (cached.Buffer, 0, buffer, 0, block.RequestLength);
+                    if (!cached.Flushing) {
                         blocks[i] = cached.SetFlushing ();
-                        using (cached.BufferReleaser) {
-                            var asyncWrite = WriteToFilesAsync (torrent, block, cached.Buffer);
-                            Buffer.BlockCopy (cached.Buffer, 0, buffer, 0, block.RequestLength);
-                            Interlocked.Add (ref cacheUsed, -block.RequestLength);
-                            await asyncWrite;
-                            blocks.Remove (cached);
-                        }
+                        FlushBlockAsync (torrent, blocks, cached);
                     }
                     Interlocked.Add (ref cacheHits, block.RequestLength);
-                    return true;
+                    return ReusableTask.FromResult(true);
                 }
             }
 
-            return false;
+            return ReusableTask.FromResult (false);
+        }
+
+        async void FlushBlockAsync (ITorrentData torrent, List<CachedBlock> blocks, CachedBlock cached)
+        {
+            // FIXME: How do we handle failures from this?
+            using (cached.BufferReleaser) {
+                await WriteToFilesAsync (torrent, cached.Block, cached.Buffer);
+                Interlocked.Add (ref cacheUsed, -cached.Block.RequestLength);
+                blocks.Remove (cached);
+            }
         }
 
         public async ReusableTask WriteAsync (ITorrentData torrent, BlockInfo block, byte[] buffer, bool preferSkipCache)