i have subroutine processes large blocks of information. in order make use of entire cpu, divides work separate threads. after threads have completed, finishes. read creating , destroying threads uses lots of overhead, tried using threadpool, runs slower creating own threads. how can create own threads when program runs , keep reusing them? i've seen people can't done, threadpool must possible, right?
here part of code launches new threads / uses threadpool:
//initialization threads thread[] altthread = null; if (numthreads > 1) altthread = new thread[pub.numthreads - 1]; { if (numthreads > 1) { //split matrix numthreads number of even-sized blocks , execute on separate threads int threadwidth = datawidth / numthreads; if (usethreadpool) //use threadpool threads { (int = 0; < numthreads - 1; i++) { threadpool.queueuserworkitem(computepartialdataonthread, new object[] { altengine[i], threadwidth * (i + 1), threadwidth * (i + 2) }); } //get number of threads available after queue system.threading.thread.sleep(0); int startthreads, empty, endthreads; threadpool.getavailablethreads(out startthreads, out empty); computepartialdata(thisengine, 0, threadwidth); //wait threads finish { threadpool.getavailablethreads(out endthreads, out empty); system.threading.thread.sleep(1); } while (startthreads - endthreads > 0); } else //create new threads each time (can reuse these?) { (int = 0; < numthreads - 1; i++) { altthread[i] = new thread(computepartialdataonthread); altthread[i].start(new object[] { altengine[i], threadwidth * (i + 1), threadwidth * (i + 2) }); } computepartialdata(thisengine, 0, threadwidth); //wait threads finish foreach (thread t in altthread) t.join(1000); foreach (thread t in altthread) if (t.isalive) t.abort(); } } } computepartialdataonthread unpackages information , calls computepartialdata. data processed shared among threads (they don't try read/write same locations). altengine[] separate computation engine each thread.
the operation runs 10-20% using threadpool.
this sounds common requirement can solved multi-threaded producer-consumer queue. threads kept 'alive' , signaled work when new work added queue. work represented delegate (in case computepartialdataonthread) , data passed delegate queued (in case params computepartialdataonthread). useful feature implementation of managing worker threads , actual algorithms separate. here p-c queue:
public class superqueue<t> : idisposable t : class { readonly object _locker = new object(); readonly list<thread> _workers; readonly queue<t> _taskqueue = new queue<t>(); readonly action<t> _dequeueaction; /// <summary> /// initializes new instance of <see cref="superqueue{t}"/> class. /// </summary> /// <param name="workercount">the worker count.</param> /// <param name="dequeueaction">the dequeue action.</param> public superqueue(int workercount, action<t> dequeueaction) { _dequeueaction = dequeueaction; _workers = new list<thread>(workercount); // create , start separate thread each worker (int = 0; < workercount; i++) { thread t = new thread(consume) { isbackground = true, name = string.format("superqueue worker {0}",i )}; _workers.add(t); t.start(); } } /// <summary> /// enqueues task. /// </summary> /// <param name="task">the task.</param> public void enqueuetask(t task) { lock (_locker) { _taskqueue.enqueue(task); monitor.pulseall(_locker); } } /// <summary> /// consumes instance. /// </summary> void consume() { while (true) { t item; lock (_locker) { while (_taskqueue.count == 0) monitor.wait(_locker); item = _taskqueue.dequeue(); } if (item == null) return; // run actual method _dequeueaction(item); } } /// <summary> /// performs application-defined tasks associated freeing, releasing, or resetting unmanaged resources. /// </summary> public void dispose() { // enqueue 1 null task per worker make each exit. _workers.foreach(thread => enqueuetask(null)); _workers.foreach(thread => thread.join()); } } as previous posters have said, there many built in structures (look @ tpl ), use threadpool, may want @ before implementing own queue.
Comments
Post a Comment