c# - Producer/Consumer, Stream buffer problem -

i'm trying write buffermanager manages 3 streams. typical usage slow producer , fast consumer. idea behind 3 buffers producer has buffer write in , consumer gets latest data produced.

now have this, , sort-off works.

namespace yariifstream {      /// <summary>     /// class manages 3 buffers used if data streams     /// </summary>     public class yariifstream     {         private stream writebuf; ///<value>the stream used writing</value>         private stream readbuf; ///<value>the stream used reading</value>         private stream swapbuf; ///<value>the stream used swapping</value>         private bool firsttime; ///<value>boolean used checking if first time writebuffers asked</value>         private object sync; ///<value>object used syncing</value>          /// <summary>         /// initializes new instance of yari.yariifstream class expandable buffers         /// </summary>         public yariifstream()         {             sync = new object();             eerste = true;              writebuf = new memorystream();             readbuf = new memorystream();             swapbuf = new memorystream();         }          /// <summary>         /// returns stream buffer new data ready read         /// </summary>         /// <returns>stream</returns>         public stream getreadbuffer()         {             lock (sync)             {                 monitor.wait(sync);                 stream tempbuf = swapbuf;                 swapbuf = readbuf;                 readbuf = tempbuf;             }             return readbuf;         }          /// <summary>         /// returns stream buffer ready written data         /// </summary>         /// <returns>stream</returns>         public stream getwritebuffer()         {             lock (sync)             {                 stream tempbuf = swapbuf;                 swapbuf = writebuf;                 writebuf = tempbuf;                 if (!firsttime)                 {                     monitor.pulse(sync);                 }                 else                 {                     firsttime = false;                  }             }             //thread.sleep(1);             return writebuf;         }      } } 

the firsttime check used because first time writebuffer asked, can not pulse consumer because buffer still has written data. when writebuffer asked second time, can sure previous buffer contains data.

i have 2 threads, 1 producer , 1 consumer. output:

prod: uv_hjd`alv   cons: n/<]g[)8fv prod: n/<]g[)8fv   cons: 5ud*tj-qkv prod: 5ud*tj-qkv   cons: 4lx&z7qqja prod: 4lx&z7qqja   cons: kjuuvyca.b prod: kjuuvyca.b 

now it's ok consumer lags 1 behind, supposed that. can see lose first string of data wich main problem.

the other problems this:

• if remove firsttime check, works. shouldn't in opinion...
• if add thread.sleep(1); in getwritebuffer() works. don't understand.

thanks in advance enlightenment.

i've fixed problem. replaced stream instances byte[]. works fine. don't know why stream not work, not want spent more time figuring out.

here's new code running same problem.

/// <summary> /// namespace provides crossthread-, concurrentproof buffer manager.  /// </summary> namespace yariifstream {      /// <summary>     /// class manages 3 buffers used if data streams     /// </summary>     public class yariifstream     {         private byte[] writebuf; ///<value>the buffer used writing</value>         private byte[] readbuf; ///<value>the buffer used reading</value>         private byte[] swapbuf; ///<value>the buffer used swapping</value>         private bool firsttime; ///<value>boolean used checking if first time writebuffers asked</value>         private object sync; ///<value>object used syncing</value>          /// <summary>         /// initializes new instance of yari.yariifstream class expandable buffers initial capacity specified         /// </summary>         /// <param name="capacity">initial capacity of buffers</param>         public yariifstream(int capacity)         {             sync = new object();             firsttime = true;              writebuf = new byte[capacity];             readbuf = new byte[capacity];             swapbuf = new byte[capacity];         }          /// <summary>         /// returns buffer new data ready read         /// </summary>         /// <returns>byte[]</returns>         public byte[] getreadbuffer()         {             byte[] tempbuf;             lock (sync)             {                 monitor.wait(sync);                 tempbuf = swapbuf;                 swapbuf = readbuf;             }             readbuf = tempbuf;              return readbuf;         }          /// <summary>         /// returns buffer ready written data         /// </summary>         /// <returns>byte[]</returns>         public byte[] getwritebuffer()         {             byte[] tempbuf;             lock (sync)             {                 tempbuf = swapbuf;                 swapbuf = writebuf;                  writebuf = tempbuf;                  if (!firsttime)                 {                     monitor.pulse(sync);                 }                 else                 {                     firsttime = false;                 }             }             return writebuf;         }     } }