implab/ImplabNet Files · Implab/Parallels/BlockingQueue.cs

Working on runnable component

cin - - Load All Authors

File last commit:

r233:d6fe09f5592c v2


                r250:9f63dade3a40

Download file

             BlockingQueue.cs
        
                    101 lines
            
             | 3.1 KiB
            
                | text/x-csharp
            
             |
                CSharpLexer
            
             / Implab / Parallels / BlockingQueue.cs
          
                    History
                
                 |
                  Source
                 | Raw
                 |Copy content
                 |Copy permalink

        cin
    
added the blocking queue

              r137
            
      using System;

      using System.Threading;

      namespace Implab.Parallels {

          public class BlockingQueue<T> : AsyncQueue<T> {

              readonly object m_lock = new object();

        cin
    
Improved AsyncQueue...

              r233
            
              public void EnqueuePulse(T value) {

        cin
    
added the blocking queue

              r137
            
                  base.Enqueue(value);

                  lock (m_lock)

                      Monitor.Pulse(m_lock);

              }

        cin
    
Improved AsyncQueue...

              r233
            
              public void EnqueueRangePulse(T[] data, int offset, int length) {

        cin
    
added the blocking queue

              r137
            
                  base.EnqueueRange(data, offset, length);

                  if (length > 1)

                      lock (m_lock)

                          Monitor.PulseAll(m_lock);

                  else

                      lock (m_lock)

                          Monitor.Pulse(m_lock);

              }

              public T GetItem(int timeout) {

                  T item;

        cin
    
fixed blocking queue

              r139
            
                  if (!TryDequeue(out item)) {

                      var t1 = Environment.TickCount;

                      var dt = timeout;

                      lock (m_lock) {

                          while (!TryDequeue(out item)) {

                              if (!Monitor.Wait(m_lock, dt))

                                  throw new TimeoutException();

                              if (timeout >= 0) {

                                  dt = timeout - Environment.TickCount + t1;

                                  if (dt < 0)

                                      throw new TimeoutException();

                              }

                          }

        cin
    
added the blocking queue

              r137
            
                      }

                  }

                  return item;

              }

              public T GetItem() {

                  T item;

        cin
    
fixed blocking queue

              r139
            
                  if (!TryDequeue(out item))

                      lock (m_lock) {

                          while (!TryDequeue(out item))

                              Monitor.Wait(m_lock);

                      }

        cin
    
added the blocking queue

              r137
            
                  return item;

              }

              public T[] GetRange(int max, int timeout) {

                  Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");

                  var buffer = new T[max];

                  int actual;

        cin
    
fixed blocking queue

              r139
            
                  if (!TryDequeueRange(buffer, 0, max, out actual)) {

                      var t1 = Environment.TickCount;

                      var dt = timeout;

                      lock (m_lock) {

                          while (!TryDequeueRange(buffer, 0, max, out actual)) {

                              if (!Monitor.Wait(m_lock, dt))

                                  throw new TimeoutException();

                              if (timeout >= 0) {

                                  dt = timeout - Environment.TickCount + t1;

                                  if (dt < 0)

                                      throw new TimeoutException();

                              }

                          }

        cin
    
added the blocking queue

              r137
            
                      }

                  }

                  var data = new T[actual];

                  Array.Copy(buffer, data, actual);

                  return data;

              }

              public T[] GetRange(int max) {

                  Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");

                  var buffer = new T[max];

                  int actual;

        cin
    
fixed blocking queue

              r139
            
                  if (!TryDequeueRange(buffer, 0, max, out actual))

        cin
    
added the blocking queue

              r137
            
                      lock (m_lock)

        cin
    
fixed blocking queue

              r139
            
                          while (!TryDequeueRange(buffer, 0, max, out actual))

                              Monitor.Wait(m_lock);

        cin
    
added the blocking queue

              r137
            
                  var data = new T[actual];

                  Array.Copy(buffer, data, actual);

                  return data;

              }

          }

      }

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages

cin added the blocking queue	r137	using System;
		using System.Threading;

		namespace Implab.Parallels {
		public class BlockingQueue<T> : AsyncQueue<T> {
		readonly object m_lock = new object();

cin Improved AsyncQueue...	r233	public void EnqueuePulse(T value) {
cin added the blocking queue	r137	base.Enqueue(value);
		lock (m_lock)
		Monitor.Pulse(m_lock);
		}

cin Improved AsyncQueue...	r233	public void EnqueueRangePulse(T[] data, int offset, int length) {
cin added the blocking queue	r137	base.EnqueueRange(data, offset, length);
		if (length > 1)
		lock (m_lock)
		Monitor.PulseAll(m_lock);
		else
		lock (m_lock)
		Monitor.Pulse(m_lock);
		}

		public T GetItem(int timeout) {
		T item;
cin fixed blocking queue	r139
		if (!TryDequeue(out item)) {
		var t1 = Environment.TickCount;
		var dt = timeout;

		lock (m_lock) {
		while (!TryDequeue(out item)) {
		if (!Monitor.Wait(m_lock, dt))
		throw new TimeoutException();
		if (timeout >= 0) {
		dt = timeout - Environment.TickCount + t1;
		if (dt < 0)
		throw new TimeoutException();
		}
		}
cin added the blocking queue	r137	}
		}
		return item;
		}

		public T GetItem() {
		T item;
cin fixed blocking queue	r139	if (!TryDequeue(out item))
		lock (m_lock) {
		while (!TryDequeue(out item))
		Monitor.Wait(m_lock);
		}
cin added the blocking queue	r137	return item;
		}

		public T[] GetRange(int max, int timeout) {
		Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");

		var buffer = new T[max];
		int actual;
cin fixed blocking queue	r139	if (!TryDequeueRange(buffer, 0, max, out actual)) {
		var t1 = Environment.TickCount;
		var dt = timeout;

		lock (m_lock) {
		while (!TryDequeueRange(buffer, 0, max, out actual)) {

		if (!Monitor.Wait(m_lock, dt))
		throw new TimeoutException();

		if (timeout >= 0) {
		dt = timeout - Environment.TickCount + t1;
		if (dt < 0)
		throw new TimeoutException();
		}
		}
cin added the blocking queue	r137	}
		}

		var data = new T[actual];
		Array.Copy(buffer, data, actual);
		return data;
		}

		public T[] GetRange(int max) {
		Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");

		var buffer = new T[max];
		int actual;
cin fixed blocking queue	r139	if (!TryDequeueRange(buffer, 0, max, out actual))
cin added the blocking queue	r137	lock (m_lock)
cin fixed blocking queue	r139	while (!TryDequeueRange(buffer, 0, max, out actual))
		Monitor.Wait(m_lock);
cin added the blocking queue	r137
		var data = new T[actual];
		Array.Copy(buffer, data, actual);
		return data;
		}
		}
		}