استفاده از کلاس Monitor در کنترل Thread

[amin_sltny]

سلام.

من یه برنامه تحت شبکه ساخته ام که روال کلیش به صورت زیره:

کاربران وقتی به سرور متصل میشوند و کانکت میشوند برایشان یک Thread ساخته می شود یهنی به ازای هر کاربر یه Thread ساخته میشه که در این Thread حلقه دریافت اطلاعات از کاربر قرار داره.

یه کلاس دیگه دارم که یه صفه (Queue) ودر این صف اطلاعات از Thread ها به داخل این صف وارد می شوند و مورد پردازش قرار گرفته و جواب به کاربر ارسال می گردد.

حالا من برای مدیریت این Thread ها می خوام از کلاس Monitor استفاده کنم. ولی نمیدونم چه طوری باید استفاده و کنم و کجا ازش استفاده بشه!!

لطفا کمکم کنید.

-------------------------------------------------------------------------------------------------------------------------------------------------------
وب (http://www.kanonportal.ir/)

[tooraj_azizi_1035]

Thread های شما Producer و Thread ای که از صف نتایج رو بر میگردونه Consumer هست. شما سناریوی Producer/Consumer رو در برنامتون دارید.
How to: Implement a Producer-Consumer Dataflow Pattern (http://msdn.microsoft.com/en-us/library/hh228601.aspx)

[amin_sltny]

Thread های شما Producer و Thread ای که از صف نتایج رو بر میگردونه Consumer هست. شما سناریوی Producer/Consumer رو در برنامتون دارید.
How to: Implement a Producer-Consumer Dataflow Pattern (http://msdn.microsoft.com/en-us/library/hh228601.aspx)

میشه بیشتر توضیح بدین من از این صفحه چیزی نفهمیدم!!:ناراحت::متفکر:

[amin_sltny]

کسی نیست به من کمک کنه؟؟؟؟؟؟؟؟؟؟؟؟؟؟؟؟؟؟:گری� �:

[tooraj_azizi_1035]

ساختمان داده ConcurrentQueue رو پیشنهاد می کنم. یک صف هست که چندین ترد می تونن همزمان به اون آیتم اضافه می کنن.
اضافه کردن با متد Enqueue هست و برداشتن از صف با متد TryDequeue.
تمام هماهنگی ها به طور داخلی در طراحی این کلاس در نظر گرفته شده و شما مسئول Sync ترد ها نیستید و استفاده از Monitor منتفی هست.

البته اگر اصرار بر FIFO بودن دارید باید از صف استفاده کنید چون آیتمی که آخر وارد می شود حتماً اول برداشته می شود.


ConcurrentQueue<T> (http://msdn.microsoft.com/en-us/library/dd267265.aspx) handles all synchronization internally. If two threads call TryDequeue at precisely the same moment, neither operation is blocked. When a conflict is detected between two threads, one thread has to try again to retrieve the next element, and the synchronization is handled internally.
TryDequeue tries to remove an element from the queue. If the method is successful, the item is removed and the method returns true; otherwise, it returns false. That happens atomically with respect to other operations on the queue. If the queue was populated with code such as q.Enqueue("a"); q.Enqueue("b"); q.Enqueue("c"); and two threads concurrently try to dequeue an element, one thread will dequeue a and the other thread will dequeue b. Both calls to TryDequeue will return true, because they were both able to dequeue an element. If each thread goes back to dequeue an additional element, one of the threads will dequeue c and return true, whereas the other thread will find the queue empty and will return false.

[amin_sltny]

ساختمان داده ConcurrentQueue رو پیشنهاد می کنم. یک صف هست که چندین ترد می تونن همزمان به اون آیتم اضافه می کنن.
اضافه کردن با متد Enqueue هست و برداشتن از صف با متد TryDequeue.
تمام هماهنگی ها به طور داخلی در طراحی این کلاس در نظر گرفته شده و شما مسئول Sync ترد ها نیستید و استفاده از Monitor منتفی هست.

البته اگر اصرار بر FIFO بودن دارید باید از صف استفاده کنید چون آیتمی که آخر وارد می شود حتماً اول برداشته می شود.


ConcurrentQueue<T> (http://msdn.microsoft.com/en-us/library/dd267265.aspx) handles all synchronization internally. If two threads call TryDequeue at precisely the same moment, neither operation is blocked. When a conflict is detected between two threads, one thread has to try again to retrieve the next element, and the synchronization is handled internally.
TryDequeue tries to remove an element from the queue. If the method is successful, the item is removed and the method returns true; otherwise, it returns false. That happens atomically with respect to other operations on the queue. If the queue was populated with code such as q.Enqueue("a"); q.Enqueue("b"); q.Enqueue("c"); and two threads concurrently try to dequeue an element, one thread will dequeue a and the other thread will dequeue b. Both calls to TryDequeue will return true, because they were both able to dequeue an element. If each thread goes back to dequeue an additional element, one of the threads will dequeue c and return true, whereas the other thread will find the queue empty and will return false.



از این اینکه این ساختمان را به من معرفی کردید سپاسگذارم ولی اگه میشه طرز استفاده از کلاس monitor را هم برایم بگید

[tooraj_azizi_1035]

کلاس Monitor دارای متدی به نام Enter هست که ورود به ناحیه بحرانی رو اعلام می کنه و متد Exit خروج رو اعلام میکنه.
همون صف با هماهنگی Monitor:
http://msdn.microsoft.com/en-us/library/de0542zz.aspx


using System;
using System.Threading;
using System.Collections.Generic;
using System.Text;

class SafeQueue<T>
{
// A queue that is protected by Monitor.
private Queue<T> m_inputQueue = new Queue<T>();

// Lock the queue and add an element.
public void Enqueue(T qValue)
{
// Request the lock, and block until it is obtained.
Monitor.Enter(m_inputQueue);
try
{
// When the lock is obtained, add an element.
m_inputQueue.Enqueue(qValue);
}
finally
{
// Ensure that the lock is released.
Monitor.Exit(m_inputQueue);
}
}

// Try to add an element to the queue: Add the element to the queue
// only if the lock is immediately available.
public bool TryEnqueue(T qValue)
{
// Request the lock.
if (Monitor.TryEnter(m_inputQueue))
{
try
{
m_inputQueue.Enqueue(qValue);
}
finally
{
// Ensure that the lock is released.
Monitor.Exit(m_inputQueue);
}
return true;
}
else
{
return false;
}
}

// Try to add an element to the queue: Add the element to the queue
// only if the lock becomes available during the specified time
// interval.
public bool TryEnqueue(T qValue, int waitTime)
{
// Request the lock.
if (Monitor.TryEnter(m_inputQueue, waitTime))
{
try
{
m_inputQueue.Enqueue(qValue);
}
finally
{
// Ensure that the lock is released.
Monitor.Exit(m_inputQueue);
}
return true;
}
else
{
return false;
}
}

// Lock the queue and dequeue an element.
public T Dequeue()
{
T retval;

// Request the lock, and block until it is obtained.
Monitor.Enter(m_inputQueue);
try
{
// When the lock is obtained, dequeue an element.
retval = m_inputQueue.Dequeue();
}
finally
{
// Ensure that the lock is released.
Monitor.Exit(m_inputQueue);
}

return retval;
}

// Delete all elements that equal the given object.
public int Remove(T qValue)
{
int removedCt = 0;

// Wait until the lock is available and lock the queue.
Monitor.Enter(m_inputQueue);
try
{
int counter = m_inputQueue.Count;
while (counter > 0)
// Check each element.
{
T elem = m_inputQueue.Dequeue();
if (!elem.Equals(qValue))
{
m_inputQueue.Enqueue(elem);
}
else
{
// Keep a count of items removed.
removedCt += 1;
}
counter = counter - 1;
}
}
finally
{
// Ensure that the lock is released.
Monitor.Exit(m_inputQueue);
}

return removedCt;
}

// Print all queue elements.
public string PrintAllElements()
{
StringBuilder output = new StringBuilder();

// Lock the queue.
Monitor.Enter(m_inputQueue);
try
{
foreach( T elem in m_inputQueue )
{
// Print the next element.
output.AppendLine(elem.ToString());
}
}
finally
{
// Ensure that the lock is released.
Monitor.Exit(m_inputQueue);
}

return output.ToString();
}
}

public class Example
{
private static SafeQueue<int> q = new SafeQueue<int>();
private static int threadsRunning = 0;
private static int[][] results = new int[3][];

static void Main()
{
Console.WriteLine("Working...");

for(int i = 0; i < 3; i++)
{
Thread t = new Thread(ThreadProc);
t.Start(i);
Interlocked.Increment(ref threadsRunning);
}
}

private static void ThreadProc(object state)
{
DateTime finish = DateTime.Now.AddSeconds(10);
Random rand = new Random();
int[] result = { 0, 0, 0, 0, 0, 0, 0, 0, 0 };
int threadNum = (int) state;

while (DateTime.Now < finish)

{
int what = rand.Next(250);
int how = rand.Next(100);

if (how < 16)
{
q.Enqueue(what);
result[(int)ThreadResultIndex.EnqueueCt] += 1;
}
else if (how < 32)
{
if (q.TryEnqueue(what))
{
result[(int)ThreadResultIndex.TryEnqueueSucceedCt] += 1;
}
else
{
result[(int)ThreadResultIndex.TryEnqueueFailCt] += 1;
}
}
else if (how < 48)
{
// Even a very small wait significantly increases the success
// rate of the conditional enqueue operation.
if (q.TryEnqueue(what, 10))
{
result[(int)ThreadResultIndex.TryEnqueueWaitSucceedCt] += 1;
}
else
{
result[(int)ThreadResultIndex.TryEnqueueWaitFailCt] += 1;
}
}
else if (how < 96)
{
result[(int)ThreadResultIndex.DequeueCt] += 1;
try
{
q.Dequeue();
}
catch
{
result[(int)ThreadResultIndex.DequeueExCt] += 1;
}
}
else
{
result[(int)ThreadResultIndex.RemoveCt] += 1;
result[(int)ThreadResultIndex.RemovedCt] += q.Remove(what);
}
}

results[threadNum] = result;

if (0 == Interlocked.Decrement(ref threadsRunning))
{
StringBuilder sb = new StringBuilder(
" Thread 1 Thread 2 Thread 3 Total\n");

for(int row = 0; row < 9; row++)
{
int total = 0;
sb.Append(titles[row]);

for(int col = 0; col < 3; col++)
{
sb.Append(String.Format("{0,9}", results[col][row]));
total += results[col][row];
}

sb.AppendLine(String.Format("{0,9}", total));
}

Console.WriteLine(sb.ToString());
}
}

private static string[] titles = {
"Enqueue ",
"TryEnqueue succeeded ",
"TryEnqueue failed ",
"TryEnqueue(T, wait) succeeded ",
"TryEnqueue(T, wait) failed ",
"Dequeue attempts ",
"Dequeue exceptions ",
"Remove operations ",
"Queue elements removed "};

private enum ThreadResultIndex
{
EnqueueCt,
TryEnqueueSucceedCt,
TryEnqueueFailCt,
TryEnqueueWaitSucceedCt,
TryEnqueueWaitFailCt,
DequeueCt,
DequeueExCt,
RemoveCt,
RemovedCt
};
}

/* This example produces output similar to the following:

Working...
Thread 1 Thread 2 Thread 3 Total
Enqueue 277382 515209 308464 1101055
TryEnqueue succeeded 276873 514621 308099 1099593
TryEnqueue failed 109 181 134 424
TryEnqueue(T, wait) succeeded 276913 514434 307607 1098954
TryEnqueue(T, wait) failed 2 0 0 2
Dequeue attempts 830980 1544081 924164 3299225
Dequeue exceptions 12102 21589 13539 47230
Remove operations 69550 129479 77351 276380
Queue elements removed 11957 22572 13043 47572
*/