guard improvments

Author: matiii

src/Dijkstra.NET/Dijkstra.NET/ShortestPath/BfsParallel.cs

@@ -1,6 +1,7 @@
 namespace Dijkstra.NET.ShortestPath
 {
     using System;
+    using System.Diagnostics;
     using System.Threading;
     using Contract;
     using Model;
@@ -22,7 +23,7 @@ public BfsParallel(IConcurrentGraph<T, TEdgeCustom> graph) : base(graph)
             _table = new ProducerConsumer<T, TEdgeCustom>();
         }
 
-        public BfsParallel(IConcurrentGraph<T, TEdgeCustom> graph, double guardInterval): base(graph)
+        public BfsParallel(IConcurrentGraph<T, TEdgeCustom> graph, double guardInterval) : base(graph)
         {
             _graph = graph;
             _table = new ProducerConsumer<T, TEdgeCustom>(guardInterval);
@@ -34,7 +35,8 @@ public override IShortestPathResult Process(uint @from, uint to)
 
             IConcurrentNode<T, TEdgeCustom> nodeFrom = _graph.GetConccurentNode(from);
             nodeFrom.Distance = 0;
-            _table.Produce(nodeFrom);
+
+            _table.Initialise = () => _table.Produce(nodeFrom);
 
             _table.Producing = node =>
             {
@@ -50,9 +52,14 @@ public override IShortestPathResult Process(uint @from, uint to)
 
             _table.Consuming = job =>
             {
-                if (Reduce(_graph.GetConccurentNode(job.To), job.Distance))
+                if (Reduce(job.From,_graph.GetConccurentNode(job.To), job.Distance))
                 {
-                    _result.P.AddOrUpdate(job.To, job.From, (u, u1) => job.From);
+                    Debug.WriteLineIf(job.To == 3, $"Update ({job.From})->({job.To}) [{job.Distance}]");
+                    //_result.P.AddOrUpdate(job.To, job.From, (u, u1) =>
+                    //{
+                    //    return job.From;
+                    //});
+
                     _table.Produce(_graph.GetConccurentNode(job.To));
                 }
             };
@@ -63,29 +70,26 @@ public override IShortestPathResult Process(uint @from, uint to)
 
             return _result;
         }
-
+        private object _locker = new object();
+        private object _locker2 = new object();
         public void SetInsurance(int insurance)
         {
             _table.Insurance = insurance;
         }
 
-        private bool Reduce(IConcurrentNode<T, TEdgeCustom> to, int distance)
+        private bool Reduce(uint from, IConcurrentNode<T, TEdgeCustom> to, int distance)
         {
-            var spin = new SpinWait();
-
-            while (true)
+            lock (to)
             {
-                int initialDistance = to.Distance;
-
-                if (initialDistance > distance)
+                if (to.Distance > distance)
                 {
-                    if (to.TrySetDistance(distance, initialDistance))
-                        return true;
+                    to.Distance = distance;
+                    _result.Path[to.Key] = from;
 
-                    spin.SpinOnce();
+                    return true;
                 }
-                else
-                    return false;
+
+                return false;
             }
         }
 

src/Dijkstra.NET/Dijkstra.NET/ShortestPath/Utility/ProducerConsumer.cs

@@ -20,7 +20,7 @@ internal sealed class ProducerConsumer<T, TEdgeCustom> : IDisposable where TEdge
 
         private int _counter;
 
-        public ProducerConsumer(double guardInterval = 20)
+        public ProducerConsumer(double guardInterval = 5)
         {
             _guardTimer = new Timer(guardInterval);
 
@@ -42,6 +42,7 @@ public ProducerConsumer(double guardInterval = 20)
 
         public Action<IConcurrentNode<T, TEdgeCustom>> Producing { get; set; }
         public Action<MapReduceJob> Consuming { get; set; }
+        public Action Initialise { get; set; }
 
         public void Produce(IConcurrentNode<T, TEdgeCustom> product)
         {
@@ -61,6 +62,8 @@ private void Emit(Emitter emitter)
 
         public void Work()
         {
+            Initialise();
+
             foreach (var emitter in _table.GetConsumingEnumerable())
             {
                 Task.Factory.StartNew(() =>
@@ -81,10 +84,15 @@ public void Work()
 
         private void NotifyGuard()
         {
-            if (Interlocked.Exchange(ref _guardIsWorking, 1) == 0)
-                _guardTimer.Start();
-            else
-                Interlocked.Exchange(ref _counter, 0);
+            //if (Interlocked.Exchange(ref _guardIsWorking, 1) == 0)
+            //    _guardTimer.Start();
+            //else
+            //    Interlocked.Exchange(ref _counter, 0);
+
+            if (IsNotWorking)
+            {
+                Complete();
+            }
         }
 
         private bool IsNotWorking => _table.Count == 0 && Interlocked.CompareExchange(ref _currentJobs, 0, 0) == 0;