OutOfMemoryException to write to a MemoryStream

Question

I have a small sample application that I was working on trying to get some new .Net 4.0 parallel extensions (they are very nice). I am facing a (possibly very stupid) problem with an OutOfMemoryException. My main application, which I am looking to connect this sample to reading some data and a large number of files, does some processing on them and then writes them somewhere. I ran into some problems when the files got bigger (maybe GB) and were concerned about memory, so I wanted to parallelize what led me along this path.

Now below the code gets OOME on small files, and I think I just missed something. He will read in 10-15 files and write them in parellel beautifully, but then he suffocates on the next. It looks like he read and wrote about 650 MB. A second set of eyes will be appreciated.

I read in MemorySteam from FileStream because this is what is needed for the main application, and I'm just trying to reproduce it to some extent. It reads data and files from all types of places and works with them like with MemoryStreams.

It uses .Net 4.0 Beta 2, VS 2010.

namespace ParellelJob
{
class Program
{
    BlockingCollection<FileHolder> serviceToSolutionShare;
    static void Main(string[] args)
    {
        Program p = new Program();
        p.serviceToSolutionShare = new BlockingCollection<FileHolder>();
        ServiceStage svc = new ServiceStage(ref p.serviceToSolutionShare);
        SolutionStage sol = new SolutionStage(ref p.serviceToSolutionShare);

        var svcTask = Task.Factory.StartNew(() => svc.Execute());
        var solTask = Task.Factory.StartNew(() => sol.Execute());

        while (!solTask.IsCompleted)
        {

        }

    }
}

class ServiceStage
{
    BlockingCollection<FileHolder> outputCollection;
    public ServiceStage(ref BlockingCollection<FileHolder> output)
    {
        outputCollection = output;
    }

    public void Execute()
    {
        var di = new DirectoryInfo(@"C:\temp\testfiles");
        var files = di.GetFiles();
        foreach (FileInfo fi in files)
        {
            using (var fs = new FileStream(fi.FullName, FileMode.Open, FileAccess.Read))
            {
                int b;
                var ms = new MemoryStream();
                while ((b = fs.ReadByte()) != -1)
                {
                    ms.WriteByte((byte)b); //OutOfMemoryException Occurs Here
                }
                var f = new FileHolder();
                f.filename = fi.Name;
                f.contents = ms;

                outputCollection.TryAdd(f);
            }
        }
        outputCollection.CompleteAdding();

    }
}

class SolutionStage
{
    BlockingCollection<FileHolder> inputCollection;
    public SolutionStage(ref BlockingCollection<FileHolder> input)
    {
        inputCollection = input;
    }
    public void Execute()
    {
        FileHolder current;
        while (!inputCollection.IsCompleted)
        {
            if (inputCollection.TryTake(out current))
            {
                using (var fs = new FileStream(String.Format(@"c:\temp\parellel\{0}", current.filename), FileMode.OpenOrCreate, FileAccess.Write))
                {
                    using (MemoryStream ms = (MemoryStream)current.contents)
                    {
                        ms.WriteTo(fs);
                        current.contents.Close();
                    }
                }
            }
        }
    }
}

class FileHolder
{
    public string filename { get; set; }
    public Stream contents { get; set; }
}
}

Answer 1

, while main , CPU. solTask.Wait().

, 1 , 2 (1 , 1 /.

PS1: , MemStream. , , . - :

var ms = new MemoryStream(fs.Length);

, , (LOH). , ​​ ?

PS2: ref , .

Answer 2

, ServiceStage.Execute

var ms = new MemoryStream();

, ms . . , .