Re-work process-graph to use lock-free queues

This removes the trigger_mutex which was used for exclusive access
of the work-queue for concurrent processing.

1 files changed, 50 insertions, 49 deletions

diff --git a/libs/ardour/ardour/graph.h b/libs/ardour/ardour/graph.h
index 27238e5897..25153a19f1 100644
--- a/libs/ardour/ardour/graph.h
+++ b/libs/ardour/ardour/graph.h
@@ -1,46 +1,43 @@
 /*
-    Copyright (C) 2010 Paul Davis
-    Copyright (C) 2017 Robin Gareus <robin@gareus.org>
-    Author: Torben Hohn
-
-    This program is free software; you can redistribute it and/or modify
-    it under the terms of the GNU General Public License as published by
-    the Free Software Foundation; either version 2 of the License, or
-    (at your option) any later version.
-
-    This program is distributed in the hope that it will be useful,
-    but WITHOUT ANY WARRANTY; without even the implied warranty of
-    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-    GNU General Public License for more details.
-
-    You should have received a copy of the GNU General Public License
-    along with this program; if not, write to the Free Software
-    Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
-*/
-
+ * Copyright (C) 2010 Paul Davis
+ * Copyright (C) 2017-2019 Robin Gareus <robin@gareus.org>
+ * incl. some work from Torben Hohn
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ */
 
 #ifndef __ardour_graph_h__
 #define __ardour_graph_h__
 
 #include <list>
 #include <set>
-#include <vector>
 #include <string>
+#include <vector>
 
 #include <boost/shared_ptr.hpp>
 
-#include <glib.h>
-
+#include "pbd/mpmc_queue.h"
 #include "pbd/semutils.h"
 
-#include "ardour/libardour_visibility.h"
-#include "ardour/types.h"
 #include "ardour/audio_backend.h"
+#include "ardour/libardour_visibility.h"
 #include "ardour/session_handle.h"
+#include "ardour/types.h"
 
 namespace ARDOUR
 {
-
 class GraphNode;
 class Graph;
 
@@ -50,27 +47,27 @@ class GraphEdges;
 
 typedef boost::shared_ptr<GraphNode> node_ptr_t;
 
-typedef std::list< node_ptr_t > node_list_t;
-typedef std::set< node_ptr_t > node_set_t;
+typedef std::list<node_ptr_t> node_list_t;
+typedef std::set<node_ptr_t>  node_set_t;
 
 class LIBARDOUR_API Graph : public SessionHandleRef
 {
 public:
-	Graph (Session & session);
+	Graph (Session& session);
 
-	void trigger (GraphNode * n);
-	void rechain (boost::shared_ptr<RouteList>, GraphEdges const &);
+	void trigger (GraphNode* n);
+	void rechain (boost::shared_ptr<RouteList>, GraphEdges const&);
 
 	void dump (int chain);
-	void dec_ref();
+	void reached_terminal_node ();
 
-	void helper_thread();
+	void helper_thread ();
 
 	int process_routes (pframes_t nframes, samplepos_t start_sample, samplepos_t end_sample, bool& need_butler);
 
-	int routes_no_roll (pframes_t nframes, samplepos_t start_sample, samplepos_t end_sample, bool non_rt_pending );
+	int routes_no_roll (pframes_t nframes, samplepos_t start_sample, samplepos_t end_sample, bool non_rt_pending);
 
-	void process_one_route (Route * route);
+	void process_one_route (Route* route);
 
 	void clear_other_chain ();
 
@@ -80,36 +77,40 @@ protected:
 	virtual void session_going_away ();
 
 private:
-	volatile bool _threads_active;
-
 	void reset_thread_list ();
 	void drop_threads ();
-	void restart_cycle();
-	bool run_one();
-	void main_thread();
-	void prep();
+	void run_one ();
+	void main_thread ();
+	void prep ();
 
 	node_list_t _nodes_rt[2];
-
 	node_list_t _init_trigger_list[2];
 
-	std::vector<GraphNode *> _trigger_queue;
-	pthread_mutex_t          _trigger_mutex;
+	PBD::MPMCQueue<GraphNode*> _trigger_queue;      ///< nodes that can be processed
+	volatile guint             _trigger_queue_size; ///< number of entries in trigger-queue
 
+	/** Start worker threads */
 	PBD::Semaphore _execution_sem;
 
+	/** The number of processing threads that are asleep */
+	volatile guint _idle_thread_cnt;
+
 	/** Signalled to start a run of the graph for a process callback */
 	PBD::Semaphore _callback_start_sem;
 	PBD::Semaphore _callback_done_sem;
 
-	/** The number of processing threads that are asleep */
-	volatile gint _execution_tokens;
 	/** The number of unprocessed nodes that do not feed any other node; updated during processing */
-	volatile gint _finished_refcount;
+	volatile guint _terminal_refcnt;
+
 	/** The initial number of nodes that do not feed any other node (for each chain) */
-	volatile gint _init_finished_refcount[2];
+	guint _n_terminal_nodes[2];
+	bool  _graph_empty;
+
+	/* number of background worker threads >= 0 */
+	volatile guint _n_workers;
 
-	bool _graph_empty;
+	/* flag to terminate background threads */
+	volatile gint _terminate;
 
 	/* chain swapping */
 	Glib::Threads::Mutex _swap_mutex;
@@ -132,7 +133,7 @@ private:
 
 	/* engine / thread connection */
 	PBD::ScopedConnectionList engine_connections;
-	void engine_stopped ();
+	void                      engine_stopped ();
 };
 
 } // namespace