c# - Access Queue from two delegate.BeginInvoke Async Methods Simultaneously -

c# - Access Queue<T> from two delegate.BeginInvoke Async Methods Simultaneously -

i'm inexperienced threading , asynchronous processing i'm not sure if i'm doing safe.

update: restricted .net 3.5 unfortunately no concurrentqueue (unless someone's implemented .net 3.5 or knows alternative implementation?)

update 2: found this approach using source mono implementation of concurrentqueue (which compatible code .net 3.5).

i have 1 object who's job manage queue<myobject> internally. class needs operations asynchronously (it running continuously between onstart , onstop events of windows service). anyway, i'll allow code talking:

public class myobjectqueuemanager : imyobjectqueuemanager { private bool shouldcontinueenqueuing; private readonly queue<myobject> queue; private readonly imyobjectidentifier myobjectidentifier; public myobjectqueuemanager( imyobjectidentifier myobjectidentifier ) { this.myobjectidentifier = myobjectidentifier; queue = new queue<myobject>(); } public void startqueue() { shouldcontinueenqueuing = true; var enqueuer = new myobjectenqueuer( enqueuemyobjects ); enqueuer.begininvoke( myobjectidentifier, queue, stopenqueuingmyobjects, new object() ); var dequeuer = new myobjectdequeuer( dequeuemyobjects ); dequeuer.begininvoke( queue, stopdequeuingmyobjects, new object() ); } public void stopqueue() { shouldcontinueenqueuing = false; } public event eventhandler<nextmyobjecteventargs> onnextmyobjectavailable; private void enqueuemyobjects( imyobjectidentifier identifier, queue<myobject> queue ) { while ( shouldcontinueenqueuing ) { var myobjects = identifier.getmyobjects(); foreach ( var myobject in myobjects ) { queue.enqueue( myobject ); } waitforqueuetoshrink( queue, 1000 /* arbitrary queue limiter - come settings. */ ); } } private void dequeuemyobjects( queue<myobject> queue ) { while ( queue.count > 0 || shouldcontinueenqueuing ) { if ( queue.count > 0 ) { onnextmyobjectavailable( this, new nextmyobjecteventargs( queue.dequeue() ) ); } thread.sleep( 1 ); } } private static void waitforqueuetoshrink( icollection queue, int queuelimit ) { while ( queue.count > queuelimit ) { thread.sleep( 10 ); } } private static void stopenqueuingmyobjects( iasyncresult result ) { var asyncresult = ( asyncresult ) result; var x = ( myobjectenqueuer ) asyncresult.asyncdelegate; x.endinvoke( asyncresult ); } private static void stopdequeuingmyobjects( iasyncresult result ) { var asyncresult = ( asyncresult ) result; var x = ( myobjectdequeuer ) asyncresult.asyncdelegate; x.endinvoke( asyncresult ); } private delegate void myobjectenqueuer( imyobjectidentifier myobjectidentifier, queue<myobject> queue ); private delegate void myobjectdequeuer( queue<myobject> queue ); }

my notes/thoughts/concerns:

if understand delegate.begininvoke correctly, i'm spawning 2 threads , both accessing queue (through invocation methods parameters). one thread enqueuing items. one thread dequeuing items, , checks items in queue first. i'm unsure if thread safe - queue locked 1 thread , not accessible other thread? if so, alternative approach/technique utilize allow me independently enqueue/dequeue items without holding main worker thread? if approach somehow sound (you never know) scale handle big volume? i.e. if imyobjectidentifier returning thousands of items per sec scenario work (of course, doing volume testing confirm).

as bonus, i've noticed there similar code beingness used phone call endinvoke stopenqueuingmyobjects , stopdequeuingmyobjects methods - i'd refactor utilize single method if can when tried using generic method this:

private static void stopprocessingmyobjects<t>( iasyncresult result ) { var asyncresult = ( asyncresult ) result; var x = ( t ) asyncresult.asyncdelegate; x.endinvoke( asyncresult ); }

it doesn't because doesn't know t delgate , endinvoke can't called. possible this?

try concurrentqueue<t>. thread safe implementation of queue. api different might require modification of current code implement.

c# asynchronous thread-safety queue begininvoke