Merge pull request #457 from leapmotion/feature-produce-consume-buffer

Adds thread safe queue

Author: Sloftin891

Assets/LeapMotion/Scripts/DataStructures/Editor/Tests/ProduceConsumeBufferTest.cs

@@ -0,0 +1,66 @@
+using System;
+using System.Threading;
+using NUnit.Framework;
+
+namespace Leap.Unity.Tests {
+
+  public class ProduceConsumeBufferTest {
+
+    private ProduceConsumeBuffer<TestStruct> buffer;
+
+    [SetUp]
+    public void Setup() {
+      buffer = new ProduceConsumeBuffer<TestStruct>(16);
+    }
+
+    [TearDown]
+    public void Teardown() {
+      buffer = null;
+    }
+
+    [Test]
+    [Timeout(1000)]
+    public void Test() {
+      Thread consumer = new Thread(new ThreadStart(consumerThread));
+      Thread producer = new Thread(new ThreadStart(producerThread));
+
+      consumer.Start();
+      producer.Start();
+
+      consumer.Join();
+      producer.Join();
+    }
+
+    private void consumerThread() {
+      try {
+        for (int i = 0; i < buffer.Capacity; i++) {
+          TestStruct s;
+          s.index = i;
+          s.name = i.ToString();
+          while (!buffer.TryEnqueue(ref s)) { }
+        }
+      } catch (Exception e) {
+        Assert.Fail(e.Message);
+      }
+    }
+
+    private void producerThread() {
+      try {
+        for (int i = 0; i < buffer.Capacity; i++) {
+          TestStruct s;
+          while (!buffer.TryDequeue(out s)) { }
+
+          Assert.That(s.index, Is.EqualTo(i));
+          Assert.That(s.name, Is.EqualTo(i.ToString()));
+        }
+      } catch (Exception e) {
+        Assert.Fail(e.Message);
+      }
+    }
+
+    private struct TestStruct {
+      public int index;
+      public string name;
+    }
+  }
+}

Assets/LeapMotion/Scripts/DataStructures/Editor/Tests/ProduceConsumeBufferTest.cs.meta

@@ -0,0 +1,83 @@
+using UnityEngine;
+using System;
+
+namespace Leap.Unity {
+
+  public class ProduceConsumeBuffer<T> {
+    private T[] _buffer;
+    private uint _bufferMask;
+    private uint _head, _tail;
+
+    /// <summary>
+    /// Constructs a new produce consumer buffer of at least a certain capacity.  Once the
+    /// buffer is created, the capacity cannot be modified.
+    /// 
+    /// If the minimum capacity is a power of two, it will be used as the actual capacity.
+    /// If the minimum capacity is not a power of two, the next highest power of two will
+    /// be used as the capacity.  This behavior is an optimization, Internally this class 
+    /// uses a bitwise AND operation instead of a slower modulus operation for indexing, 
+    /// which only is possible if the array length is a power of two.
+    /// </summary>
+    public ProduceConsumeBuffer(int minCapacity) {
+      if (minCapacity <= 0) {
+        throw new ArgumentOutOfRangeException("The capacity of the ProduceConsumeBuffer must be positive and non-zero.");
+      }
+
+      int capacity;
+      int closestPowerOfTwo = Mathf.ClosestPowerOfTwo(minCapacity);
+      if (closestPowerOfTwo == minCapacity) {
+        capacity = minCapacity;
+      } else {
+        if (closestPowerOfTwo < minCapacity) {
+          capacity = closestPowerOfTwo * 2;
+        } else {
+          capacity = closestPowerOfTwo;
+        }
+      }
+
+      _buffer = new T[capacity];
+      _bufferMask = (uint)(capacity - 1);
+      _head = 0;
+      _tail = 0;
+    }
+
+    /// <summary>
+    /// Returns the maximum number of elements that the buffer can hold.
+    /// </summary>
+    public int Capacity {
+      get {
+        return _buffer.Length;
+      }
+    }
+
+    /// <summary>
+    /// Tries to enqueue a value into the buffer.  If the buffer is already full, this
+    /// method will perform no action and return false.  This method is only safe to
+    /// be called from a single producer thread.
+    /// </summary>
+    public bool TryEnqueue(ref T t) {
+      uint nextTail = (_tail + 1) & _bufferMask;
+      if (nextTail == _head) return false;
+
+      _buffer[_tail] = t;
+      _tail = nextTail;
+      return true;
+    }
+
+    /// <summary>
+    /// Tries to dequeue a value off of the buffer.  If the buffer is empty this method
+    /// will perform no action and return false.  This method is only safe to be
+    /// called from a single consumer thread.
+    /// </summary>
+    public bool TryDequeue(out T t) {
+      if (_tail == _head) {
+        t = default(T);
+        return false;
+      }
+
+      t = _buffer[_head];
+      _head = (_head + 1) & _bufferMask;
+      return true;
+    }
+  }
+}