Discussion :

[Sp4ce]

Bonjour,

je cherche un expert du C# qui pourrait répondre à des questions précises :

J'ai deux threads et un MemoryStream nommé _buffer et un Stream de sortie nommé outStream ou _outStream.

Un des threads remplie le MemoryStream :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
 void _speaker_WroteOuputDataEvent(Stream outStream, VozEventStreamArgs args)
{
    byte[] buf = args.Data;
    //ZvLogManager.InfoFormat("writing {0} on the stream {1}", buf.Length, outStream);
    outStream.Write(buf, 0, buf.Length);
    _buffer.Write(buf, 0, buf.Length);
}

l'autre thread lit le MemoryStream et écrit dans outStream

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
 public override void Start()
{
    int _chunckLen = 1000;
    int bytesRead;
    byte[] buf = new byte[_chunckLen];
    while ((bytesRead = _buffer.Read(buf, 0, _chunckLen)) != 0)
        _outStream.Write(buf, 0, bytesRead);
}

Je voudrais savoir ce que vous pensez de ce design ?

On est dans un cas Producteur/Consommateur. Le premier thread écrit dans le MemoryStream et le deuxième lit.

Dernière précision, il va y avoir un seul producteur et plusieurs consomateurs. Est-ce que je devrais utiliser la fonction BeginRead ou lieu de Read parce qu'on a des lectures et écriture en simultanée sur le même stream ?

[tomlev]

S'il y a plusieurs consommateurs, ce design ne me semble pas très adapté : en effet, le premier consommateur qui va lire le stream va faire avancer la position dans le stream, si bien qui le consommateur suivant ne pourra plus les lire à moins de revenir en arrière...

Ou alors il faudrait faire un "dispatcher" qui écrit les données dans les stream des différents consommateurs. Chaque consommateur aurait son stream, et le producteur écrirait dans tous.

Par exemple tu pourrais écrire une classe comme ça :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
    public class StreamDispatcher
    {
        private List<Stream> outputs = new List<Stream>();
 
        public List<Stream> Outputs
        {
            get { return outputs; }
        }
 
        public void Write(byte[] buffer, int offset, int count)
        {
            foreach (Stream s in Outputs)
            {
                s.Write(buffer, offset, count);
            }
        }
    }

Chaque consommateur "s'abonne" en ajoutant son propre MemoryStream (ou tout autre Stream d'ailleurs), et le producteur écrit dans le dispatcher.

[Sp4ce]

Ok, je pensais un peu que ce problème allait survenir... j'aime bien ta solution.

Maintenant, nouvelles contraintes :

Le producteur commence la production de son MemoryStream à t=0
Un consommateur arrive au temps t = 1
Un consommateur arrive au temps t = 2

Les deux consommateurs doivent pouvoir lire l'intégralité des données produites depuis le début.

Voila ce que je propose :

on garde le stream dispatcher, sauf que celui ci garde une copie intact de ce qu'on écrit. Quand on consomateur s'inscrit, il initialise le stream avec la copie des données.

StreamDispatcher :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
public class StreamDispatcher
{
    private List<Stream> _outputs = new List<Stream>();
    private MemoryStream _buffer;
 
    public StreamDispatcher(MemoryStream buffer)
    {
        _buffer = buffer;
    }
 
    public void Write(byte[] buffer, int offset, int count)
    {
       foreach (Stream s in _outputs)
       {
           s.Write(buffer, offset, count);
       }
    }
 
    public Stream Subscribe()
    {
        byte[] bytes = new byte[_buffer.length];
        _buffer.read(bytes, 0, _buffer.length);
        Stream newBuffer = new stream(bytes);
        _outputs.Add(newBuffer);
        return newBuffer;
    }
}

Thread producteur :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
 
void init(MemoryStream buffer)
{
     _bufer = buffer;
     _streamDispatcher = new StreamDispatcher(_buffer); 
}
 
void _speaker_WroteOuputDataEvent(Stream outStream, VozEventStreamArgs args)
{
    byte[] buf = args.Data;
    _buffer.Write(buf, 0, buf.Length);
    _streamDispatcher.Write(buf, 0, buf.Length);
}

Thread consommateurs :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
public override void Start()
{
    int _chunckLen = 1000;
    int bytesRead;
    byte[] buf = new byte[_chunckLen];
    MemoryStream buffer = _streamDispatcher.Subsribe();
    while ((bytesRead = buffer.Read(buf, 0, _chunckLen)) != 0)
        _outStream.Write(buf, 0, bytesRead);
}

Je pense que je dois mettre deux trois lock quelque part, mais l'idée est pas mal ou pas ?

Je dois peut être réinitialiser le _buffer du streamdispatcher après un subscribe (avec un

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

Seek(0,SeekOrigin.Begin);

)?

[tomlev]

Dans le principe ça me semble pas mal !

[SaumonAgile]

ça manque sérieusement de synchronisation tout ça
Pourquoi ne pas utiliser simplement un ReaderWriterLock ?

[Sp4ce]

Ok, Merci pour vos aides, ça m'a permit de réfléchir un peu plus à mon design.

De toute façon, je dois bosser sur autre chose en ce moment (écriture d'un lib svn pour C#) et je dois revenir sur ce projet de streaming plus tard.

A+