Copy of aRts for Trinity modifications

git-svn-id: svn://anonsvn.kde.org/home/kde/branches/trinity/dependencies/arts@1070145 283d02a7-25f6-0310-bc7c-ecb5cbfe19da

1 files changed, 783 insertions, 0 deletions

diff --git a/flow/gsl/gslopmaster.c b/flow/gsl/gslopmaster.c
new file mode 100644
index 0000000..f71ec3d
--- /dev/null
+++ b/flow/gsl/gslopmaster.c
@@ -0,0 +1,783 @@
+/* GSL Engine - Flow module operation engine
+ * Copyright (C) 2001 Tim Janik
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General
+ * Public License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+#include <string.h>
+#include <sys/poll.h>
+#include <sys/time.h>
+#include <errno.h>
+
+#include "gslopmaster.h"
+
+#include "gslcommon.h"
+#include "gslopnode.h"
+#include "gsloputil.h"
+#include "gslopschedule.h"
+#include "gslieee754.h"
+
+
+
+#define	NODE_FLAG_RECONNECT(node)	G_STMT_START { (node)->reconnected = (node)->module.klass->reconnect != NULL; } G_STMT_END
+
+
+/* --- time stamping (debugging) --- */
+#define	ToyprofStamp		struct timeval
+#define	toyprof_clock_name()	("Glibc gettimeofday(2)")
+#define toyprof_stampinit()	/* nothing */
+#define	toyprof_stamp(st)	gettimeofday (&(st), 0)
+#define	toyprof_stamp_ticks()	(1000000)
+static inline guint64
+toyprof_elapsed (ToyprofStamp fstamp,
+		 ToyprofStamp lstamp)
+{
+  guint64 first = fstamp.tv_sec * toyprof_stamp_ticks () + fstamp.tv_usec;
+  guint64 last  = lstamp.tv_sec * toyprof_stamp_ticks () + lstamp.tv_usec;
+  return last - first;
+}
+
+
+/* --- typedefs & structures --- */
+typedef struct _Poll Poll;
+struct _Poll
+{
+  Poll	     *next;
+  GslPollFunc poll_func;
+  gpointer    data;
+  guint       n_fds;
+  GPollFD    *fds;
+  GslFreeFunc free_func;
+};
+
+
+/* --- prototypes --- */
+static void	master_schedule_discard	(void);
+
+
+/* --- variables --- */
+static gboolean	       master_need_reflow = FALSE;
+static gboolean	       master_need_process = FALSE;
+static EngineNode     *master_consumer_list = NULL;
+const gfloat           gsl_engine_master_zero_block[GSL_STREAM_MAX_VALUES] = { 0, }; /* FIXME */
+static Poll	      *master_poll_list = NULL;
+static guint           master_n_pollfds = 0;
+static guint           master_pollfds_changed = FALSE;
+static GPollFD         master_pollfds[GSL_ENGINE_MAX_POLLFDS];
+static EngineSchedule *master_schedule = NULL;
+
+
+/* --- functions --- */
+static void
+add_consumer (EngineNode *node)
+{
+  g_return_if_fail (ENGINE_NODE_IS_CONSUMER (node) && node->toplevel_next == NULL && node->integrated);
+  
+  node->toplevel_next = master_consumer_list;
+  master_consumer_list = node;
+}
+
+static void
+remove_consumer (EngineNode *node)
+{
+  EngineNode *tmp, *last = NULL;
+  
+  g_return_if_fail (!ENGINE_NODE_IS_CONSUMER (node) || !node->integrated);
+  
+  for (tmp = master_consumer_list; tmp; last = tmp, tmp = last->toplevel_next)
+    if (tmp == node)
+      break;
+  g_return_if_fail (tmp != NULL);
+  if (last)
+    last->toplevel_next = node->toplevel_next;
+  else
+    master_consumer_list = node->toplevel_next;
+  node->toplevel_next = NULL;
+}
+
+static void
+master_idisconnect_node (EngineNode *node,
+			 guint       istream)
+{
+  EngineNode *src_node = node->inputs[istream].src_node;
+  guint ostream = node->inputs[istream].src_stream;
+  gboolean was_consumer;
+  
+  g_assert (ostream < ENGINE_NODE_N_OSTREAMS (src_node) &&
+	    src_node->outputs[ostream].n_outputs > 0);	/* these checks better pass */
+  
+  node->inputs[istream].src_node = NULL;
+  node->inputs[istream].src_stream = ~0;
+  node->module.istreams[istream].connected = FALSE;
+  was_consumer = ENGINE_NODE_IS_CONSUMER (src_node);
+  src_node->outputs[ostream].n_outputs -= 1;
+  src_node->module.ostreams[ostream].connected = src_node->outputs[ostream].n_outputs > 0;
+  src_node->output_nodes = gsl_ring_remove (src_node->output_nodes, node);
+  NODE_FLAG_RECONNECT (node);
+  NODE_FLAG_RECONNECT (src_node);
+  /* add to consumer list */
+  if (!was_consumer && ENGINE_NODE_IS_CONSUMER (src_node))
+    add_consumer (src_node);
+}
+
+static void
+master_jdisconnect_node (EngineNode *node,
+			 guint       jstream,
+			 guint       con)
+{
+  EngineNode *src_node = node->jinputs[jstream][con].src_node;
+  guint i, ostream = node->jinputs[jstream][con].src_stream;
+  gboolean was_consumer;
+  
+  g_assert (ostream < ENGINE_NODE_N_OSTREAMS (src_node) &&
+	    node->module.jstreams[jstream].n_connections > 0 &&
+	    src_node->outputs[ostream].n_outputs > 0);	/* these checks better pass */
+  
+  i = --node->module.jstreams[jstream].n_connections;
+  node->jinputs[jstream][con] = node->jinputs[jstream][i];
+  node->module.jstreams[jstream].values[i] = NULL; /* float**values 0-termination */
+  was_consumer = ENGINE_NODE_IS_CONSUMER (src_node);
+  src_node->outputs[ostream].n_outputs -= 1;
+  src_node->module.ostreams[ostream].connected = src_node->outputs[ostream].n_outputs > 0;
+  src_node->output_nodes = gsl_ring_remove (src_node->output_nodes, node);
+  NODE_FLAG_RECONNECT (node);
+  NODE_FLAG_RECONNECT (src_node);
+  /* add to consumer list */
+  if (!was_consumer && ENGINE_NODE_IS_CONSUMER (src_node))
+    add_consumer (src_node);
+}
+
+static void
+master_disconnect_node_outputs (EngineNode *src_node,
+				EngineNode *dest_node)
+{
+  gint i, j;
+  
+  for (i = 0; i < ENGINE_NODE_N_ISTREAMS (dest_node); i++)
+    if (dest_node->inputs[i].src_node == src_node)
+      master_idisconnect_node (dest_node, i);
+  for (j = 0; j < ENGINE_NODE_N_JSTREAMS (dest_node); j++)
+    for (i = 0; i < dest_node->module.jstreams[j].n_connections; i++)
+      if (dest_node->jinputs[j][i].src_node == src_node)
+	master_jdisconnect_node (dest_node, j, i--);
+}
+
+static void
+master_process_job (GslJob *job)
+{
+  switch (job->job_id)
+    {
+      EngineNode *node, *src_node;
+      Poll *poll, *poll_last;
+      guint istream, jstream, ostream, con;
+      EngineFlowJob *fjob;
+      gboolean was_consumer;
+    case ENGINE_JOB_INTEGRATE:
+      node = job->data.node;
+      JOB_DEBUG ("integrate(%p)", node);
+      g_return_if_fail (node->integrated == FALSE);
+      g_return_if_fail (node->sched_tag == FALSE);
+      _engine_mnl_integrate (node);
+      if (ENGINE_NODE_IS_CONSUMER (node))
+	add_consumer (node);
+      node->counter = 0;
+      NODE_FLAG_RECONNECT (node);
+      master_need_reflow |= TRUE;
+      break;
+    case ENGINE_JOB_DISCARD:
+      /* FIXME: free pending flow jobs */
+      node = job->data.node;
+      JOB_DEBUG ("discard(%p)", node);
+      g_return_if_fail (node->integrated == TRUE);
+      /* disconnect inputs */
+      for (istream = 0; istream < ENGINE_NODE_N_ISTREAMS (node); istream++)
+	if (node->inputs[istream].src_node)
+	  master_idisconnect_node (node, istream);
+      for (jstream = 0; jstream < ENGINE_NODE_N_JSTREAMS (node); jstream++)
+	while (node->module.jstreams[jstream].n_connections)
+	  master_jdisconnect_node (node, jstream, node->module.jstreams[jstream].n_connections - 1);
+      /* disconnect outputs */
+      while (node->output_nodes)
+	master_disconnect_node_outputs (node, node->output_nodes->data);
+      /* remove from consumer list */
+      if (ENGINE_NODE_IS_CONSUMER (node))
+	{
+	  _engine_mnl_remove (node);
+	  remove_consumer (node);
+	}
+      else
+	_engine_mnl_remove (node);
+      node->counter = 0;
+      master_need_reflow |= TRUE;
+      master_schedule_discard ();	/* discard schedule so node may be freed */
+      break;
+    case ENGINE_JOB_SET_CONSUMER:
+    case ENGINE_JOB_UNSET_CONSUMER:
+      node = job->data.node;
+      JOB_DEBUG ("toggle_consumer(%p)", node);
+      was_consumer = ENGINE_NODE_IS_CONSUMER (node);
+      node->is_consumer = job->job_id == ENGINE_JOB_SET_CONSUMER;
+      if (was_consumer != ENGINE_NODE_IS_CONSUMER (node))
+	{
+	  if (ENGINE_NODE_IS_CONSUMER (node))
+	    add_consumer (node);
+	  else
+	    remove_consumer (node);
+	  master_need_reflow |= TRUE;
+	}
+      break;
+    case ENGINE_JOB_ICONNECT:
+      node = job->data.connection.dest_node;
+      src_node = job->data.connection.src_node;
+      istream = job->data.connection.dest_ijstream;
+      ostream = job->data.connection.src_ostream;
+      JOB_DEBUG ("connect(%p,%u,%p,%u)", node, istream, src_node, ostream);
+      g_return_if_fail (node->integrated == TRUE);
+      g_return_if_fail (src_node->integrated == TRUE);
+      g_return_if_fail (node->inputs[istream].src_node == NULL);
+      node->inputs[istream].src_node = src_node;
+      node->inputs[istream].src_stream = ostream;
+      node->module.istreams[istream].connected = TRUE;
+      /* remove from consumer list */
+      was_consumer = ENGINE_NODE_IS_CONSUMER (src_node);
+      src_node->outputs[ostream].n_outputs += 1;
+      src_node->module.ostreams[ostream].connected = TRUE;
+      src_node->output_nodes = gsl_ring_append (src_node->output_nodes, node);
+      NODE_FLAG_RECONNECT (node);
+      NODE_FLAG_RECONNECT (src_node);
+      src_node->counter = 0;	/* FIXME: counter reset? */
+      if (was_consumer && !ENGINE_NODE_IS_CONSUMER (src_node))
+	remove_consumer (src_node);
+      master_need_reflow |= TRUE;
+      break;
+    case ENGINE_JOB_JCONNECT:
+      node = job->data.connection.dest_node;
+      src_node = job->data.connection.src_node;
+      jstream = job->data.connection.dest_ijstream;
+      ostream = job->data.connection.src_ostream;
+      JOB_DEBUG ("jconnect(%p,%u,%p,%u)", node, jstream, src_node, ostream);
+      g_return_if_fail (node->integrated == TRUE);
+      g_return_if_fail (src_node->integrated == TRUE);
+      con = node->module.jstreams[jstream].n_connections++;
+      node->jinputs[jstream] = g_renew (EngineJInput, node->jinputs[jstream], node->module.jstreams[jstream].n_connections);
+      node->module.jstreams[jstream].values = g_renew (const gfloat*, node->module.jstreams[jstream].values, node->module.jstreams[jstream].n_connections + 1);
+      node->module.jstreams[jstream].values[node->module.jstreams[jstream].n_connections] = NULL; /* float**values 0-termination */
+      node->jinputs[jstream][con].src_node = src_node;
+      node->jinputs[jstream][con].src_stream = ostream;
+      /* remove from consumer list */
+      was_consumer = ENGINE_NODE_IS_CONSUMER (src_node);
+      src_node->outputs[ostream].n_outputs += 1;
+      src_node->module.ostreams[ostream].connected = TRUE;
+      src_node->output_nodes = gsl_ring_append (src_node->output_nodes, node);
+      NODE_FLAG_RECONNECT (node);
+      NODE_FLAG_RECONNECT (src_node);
+      src_node->counter = 0;	/* FIXME: counter reset? */
+      if (was_consumer && !ENGINE_NODE_IS_CONSUMER (src_node))
+	remove_consumer (src_node);
+      master_need_reflow |= TRUE;
+      break;
+    case ENGINE_JOB_IDISCONNECT:
+      node = job->data.connection.dest_node;
+      JOB_DEBUG ("idisconnect(%p,%u)", node, job->data.connection.dest_ijstream);
+      g_return_if_fail (node->integrated == TRUE);
+      g_return_if_fail (node->inputs[job->data.connection.dest_ijstream].src_node != NULL);
+      master_idisconnect_node (node, job->data.connection.dest_ijstream);
+      master_need_reflow |= TRUE;
+      break;
+    case ENGINE_JOB_JDISCONNECT:
+      node = job->data.connection.dest_node;
+      jstream = job->data.connection.dest_ijstream;
+      src_node = job->data.connection.src_node;
+      ostream = job->data.connection.src_ostream;
+      JOB_DEBUG ("jdisconnect(%p,%u,%p,%u)", node, jstream, src_node, ostream);
+      g_return_if_fail (node->integrated == TRUE);
+      g_return_if_fail (node->module.jstreams[jstream].n_connections > 0);
+      for (con = 0; con < node->module.jstreams[jstream].n_connections; con++)
+	if (node->jinputs[jstream][con].src_node == src_node &&
+	    node->jinputs[jstream][con].src_stream == ostream)
+	  break;
+      if (con < node->module.jstreams[jstream].n_connections)
+	{
+	  master_jdisconnect_node (node, jstream, con);
+	  master_need_reflow |= TRUE;
+	}
+      else
+	g_warning ("jdisconnect(dest:%p,%u,src:%p,%u): no such connection", node, jstream, src_node, ostream);
+      break;
+    case ENGINE_JOB_ACCESS:
+      node = job->data.access.node;
+      JOB_DEBUG ("access node(%p): %p(%p)", node, job->data.access.access_func, job->data.access.data);
+      g_return_if_fail (node->integrated == TRUE);
+      job->data.access.access_func (&node->module, job->data.access.data);
+      break;
+    case ENGINE_JOB_FLOW_JOB:
+      node = job->data.flow_job.node;
+      fjob = job->data.flow_job.fjob;
+      JOB_DEBUG ("add flow_job(%p,%p)", node, fjob);
+      g_return_if_fail (node->integrated == TRUE);
+      job->data.flow_job.fjob = NULL;	/* ownership taken over */
+      _engine_node_insert_flow_job (node, fjob);
+      _engine_mnl_reorder (node);
+      break;
+    case ENGINE_JOB_DEBUG:
+      JOB_DEBUG ("debug");
+      g_printerr ("JOB-DEBUG: %s\n", job->data.debug);
+      break;
+    case ENGINE_JOB_ADD_POLL:
+      JOB_DEBUG ("add poll %p(%p,%u)", job->data.poll.poll_func, job->data.poll.data, job->data.poll.n_fds);
+      if (job->data.poll.n_fds + master_n_pollfds > GSL_ENGINE_MAX_POLLFDS)
+	g_error ("adding poll job exceeds maximum number of poll-fds (%u > %u)",
+		 job->data.poll.n_fds + master_n_pollfds, GSL_ENGINE_MAX_POLLFDS);
+      poll = gsl_new_struct0 (Poll, 1);
+      poll->poll_func = job->data.poll.poll_func;
+      poll->data = job->data.poll.data;
+      poll->free_func = job->data.poll.free_func;
+      job->data.poll.free_func = NULL;		/* don't free data this round */
+      poll->n_fds = job->data.poll.n_fds;
+      poll->fds = poll->n_fds ? master_pollfds + master_n_pollfds : master_pollfds;
+      master_n_pollfds += poll->n_fds;
+      if (poll->n_fds)
+	master_pollfds_changed = TRUE;
+      memcpy (poll->fds, job->data.poll.fds, sizeof (poll->fds[0]) * poll->n_fds);
+      poll->next = master_poll_list;
+      master_poll_list = poll;
+      break;
+    case ENGINE_JOB_REMOVE_POLL:
+      JOB_DEBUG ("remove poll %p(%p)", job->data.poll.poll_func, job->data.poll.data);
+      for (poll = master_poll_list, poll_last = NULL; poll; poll_last = poll, poll = poll_last->next)
+	if (poll->poll_func == job->data.poll.poll_func && poll->data == job->data.poll.data)
+	  {
+	    if (poll_last)
+	      poll_last->next = poll->next;
+	    else
+	      master_poll_list = poll->next;
+	    break;
+	  }
+      if (poll)
+	{
+	  job->data.poll.free_func = poll->free_func;	/* free data with job */
+	  poll_last = poll;
+	  if (poll_last->n_fds)
+	    {
+	      for (poll = master_poll_list; poll; poll = poll->next)
+		if (poll->fds > poll_last->fds)
+		  poll->fds -= poll_last->n_fds;
+	      g_memmove (poll_last->fds, poll_last->fds + poll_last->n_fds,
+			 ((guint8*) (master_pollfds + master_n_pollfds)) -
+			 ((guint8*) (poll_last->fds + poll_last->n_fds)));
+	      master_n_pollfds -= poll_last->n_fds;
+	      master_pollfds_changed = TRUE;
+	    }
+	  gsl_delete_struct (Poll, poll_last);
+	}
+      else
+	g_warning (G_STRLOC ": failed to remove unknown poll function %p(%p)",
+		   job->data.poll.poll_func, job->data.poll.data);
+      break;
+    default:
+      g_assert_not_reached ();
+    }
+  JOB_DEBUG ("done");
+}
+
+static void
+master_poll_check (glong   *timeout_p,
+		   gboolean check_with_revents)
+{
+  gboolean need_processing = FALSE;
+  Poll *poll;
+  
+  if (master_need_process || *timeout_p == 0)
+    {
+      master_need_process = TRUE;
+      return;
+    }
+  for (poll = master_poll_list; poll; poll = poll->next)
+    {
+      glong timeout = -1;
+      
+      if (poll->poll_func (poll->data, gsl_engine_block_size (), &timeout,
+			   poll->n_fds, poll->n_fds ? poll->fds : NULL, check_with_revents)
+	  || timeout == 0)
+	{
+	  need_processing |= TRUE;
+	  *timeout_p = 0;
+	  break;
+	}
+      else if (timeout > 0)
+	*timeout_p = *timeout_p < 0 ? timeout : MIN (*timeout_p, timeout);
+    }
+  master_need_process = need_processing;
+}
+
+static inline guint64
+master_handle_flow_jobs (EngineNode *node,
+			 guint64     max_tick)
+{
+  EngineFlowJob *fjob = _engine_node_pop_flow_job (node, max_tick);
+  
+  if_reject (fjob)
+    do
+      {
+	g_printerr ("FJob: at:%lld from:%lld \n", node->counter, fjob->any.tick_stamp);
+	switch (fjob->fjob_id)
+	  {
+	  case ENGINE_FLOW_JOB_ACCESS:
+	    fjob->access.access_func (&node->module, fjob->access.data);
+	    break;
+	  default:
+	    g_assert_not_reached (); /* FIXME */
+	  }
+	fjob = _engine_node_pop_flow_job (node, max_tick);
+      }
+    while (fjob);
+  
+  return _engine_node_peek_flow_job_stamp (node);
+}
+
+static void
+master_process_locked_node (EngineNode *node,
+			    guint       n_values)
+{
+  guint64 final_counter = GSL_TICK_STAMP + n_values;
+  
+  while (node->counter < final_counter)
+    {
+      guint64 next_counter = master_handle_flow_jobs (node, node->counter);
+      guint64 new_counter = MIN (next_counter, final_counter);
+      guint i, j, diff = node->counter - GSL_TICK_STAMP;
+      
+      for (i = 0; i < ENGINE_NODE_N_ISTREAMS (node); i++)
+	{
+	  EngineNode *inode = node->inputs[i].src_node;
+	  
+	  if (inode)
+	    {
+	      ENGINE_NODE_LOCK (inode);
+	      if (inode->counter < final_counter)
+		master_process_locked_node (inode, final_counter - node->counter);
+	      node->module.istreams[i].values = inode->outputs[node->inputs[i].src_stream].buffer;
+	      node->module.istreams[i].values += diff;
+	      ENGINE_NODE_UNLOCK (inode);
+	    }
+	  else
+	    node->module.istreams[i].values = gsl_engine_master_zero_block;
+	}
+      for (j = 0; j < ENGINE_NODE_N_JSTREAMS (node); j++)
+	for (i = 0; i < node->module.jstreams[j].n_connections; i++)
+	  {
+	    EngineNode *inode = node->jinputs[j][i].src_node;
+	    
+	    ENGINE_NODE_LOCK (inode);
+	    if (inode->counter < final_counter)
+	      master_process_locked_node (inode, final_counter - node->counter);
+	    node->module.jstreams[j].values[i] = inode->outputs[node->jinputs[j][i].src_stream].buffer;
+	    node->module.jstreams[j].values[i] += diff;
+	    ENGINE_NODE_UNLOCK (inode);
+	  }
+      for (i = 0; i < ENGINE_NODE_N_OSTREAMS (node); i++)
+	node->module.ostreams[i].values = node->outputs[i].buffer + diff;
+      if_reject (node->reconnected)
+	{
+	  node->module.klass->reconnect (&node->module);
+	  node->reconnected = FALSE;
+	}
+      node->module.klass->process (&node->module, new_counter - node->counter);
+      for (i = 0; i < ENGINE_NODE_N_OSTREAMS (node); i++)
+	{
+	  /* FIXME: this takes the worst possible performance hit to support virtualization */
+	  if (node->module.ostreams[i].values != node->outputs[i].buffer + diff)
+	    memcpy (node->outputs[i].buffer + diff, node->module.ostreams[i].values,
+		    (new_counter - node->counter) * sizeof (gfloat));
+	}
+      node->counter = new_counter;
+    }
+}
+
+static GslLong gsl_profile_modules = 0;	/* set to 1 in gdb to get profile output */
+
+static void
+master_process_flow (void)
+{
+  guint64 new_counter = GSL_TICK_STAMP + gsl_engine_block_size ();
+  GslLong profile_maxtime = 0;
+  GslLong profile_modules = gsl_profile_modules;
+  EngineNode *profile_node = NULL;
+  
+  g_return_if_fail (master_need_process == TRUE);
+  
+  g_assert (gsl_fpu_okround () == TRUE);
+  
+  MAS_DEBUG ("process_flow");
+  if (master_schedule)
+    {
+      EngineNode *node;
+      
+      _engine_schedule_restart (master_schedule);
+      _engine_set_schedule (master_schedule);
+      
+      node = _engine_pop_unprocessed_node ();
+      while (node)
+	{
+	  ToyprofStamp profile_stamp1, profile_stamp2;
+	  
+	  if_reject (profile_modules)
+	    toyprof_stamp (profile_stamp1);
+	  
+	  master_process_locked_node (node, gsl_engine_block_size ());
+	  
+	  if_reject (profile_modules)
+	    {
+	      GslLong duration;
+	      
+	      toyprof_stamp (profile_stamp2);
+	      duration = toyprof_elapsed (profile_stamp1, profile_stamp2);
+	      if (duration > profile_maxtime)
+		{
+		  profile_maxtime = duration;
+		  profile_node = node;
+		}
+	    }
+	  
+	  _engine_push_processed_node (node);
+	  node = _engine_pop_unprocessed_node ();
+	}
+      
+      if_reject (profile_modules)
+	{
+	  if (profile_node)
+	    {
+	      if (profile_maxtime > profile_modules)
+		g_print ("Excess Node: %p  Duration: %lu usecs     ((void(*)())%p)         \n",
+			 profile_node, profile_maxtime, profile_node->module.klass->process);
+	      else
+		g_print ("Slowest Node: %p  Duration: %lu usecs     ((void(*)())%p)         \r",
+			 profile_node, profile_maxtime, profile_node->module.klass->process);
+	    }
+	}
+      
+      /* walk unscheduled nodes which have flow jobs */
+      node = _engine_mnl_head ();
+      while (node && GSL_MNL_HEAD_NODE (node))
+	{
+	  EngineNode *tmp = node->mnl_next;
+	  EngineFlowJob *fjob = _engine_node_pop_flow_job (node, new_counter);
+	  
+	  if (fjob)
+	    {
+	      while (fjob)
+		{
+		  g_printerr ("ignoring flow_job %p\n", fjob);
+		  fjob = _engine_node_pop_flow_job (node, new_counter);
+		}
+	      _engine_mnl_reorder (node);
+	    }
+	  node = tmp;
+	}
+      
+      /* nothing new to process, wait on slaves */
+      _engine_wait_on_unprocessed ();
+      
+      _engine_unset_schedule (master_schedule);
+      _gsl_tick_stamp_inc ();
+      _engine_recycle_const_values ();
+    }
+  master_need_process = FALSE;
+}
+
+static void
+master_reschedule_flow (void)
+{
+  EngineNode *node;
+  
+  g_return_if_fail (master_need_reflow == TRUE);
+  
+  MAS_DEBUG ("flow_reschedule");
+  if (!master_schedule)
+    master_schedule = _engine_schedule_new ();
+  else
+    {
+      _engine_schedule_unsecure (master_schedule);
+      _engine_schedule_clear (master_schedule);
+    }
+  for (node = master_consumer_list; node; node = node->toplevel_next)
+    _engine_schedule_consumer_node (master_schedule, node);
+  _engine_schedule_secure (master_schedule);
+  master_need_reflow = FALSE;
+}
+
+static void
+master_schedule_discard (void)
+{
+  g_return_if_fail (master_need_reflow == TRUE);
+  
+  if (master_schedule)
+    {
+      _engine_schedule_unsecure (master_schedule);
+      _engine_schedule_destroy (master_schedule);
+      master_schedule = NULL;
+    }
+}
+
+
+/* --- MasterThread main loop --- */
+gboolean
+_engine_master_prepare (GslEngineLoop *loop)
+{
+  gboolean need_dispatch;
+  guint i;
+  
+  g_return_val_if_fail (loop != NULL, FALSE);
+  
+  /* setup and clear pollfds here already, so master_poll_check() gets no junk (and IRIX can't handle non-0 revents) */
+  loop->fds_changed = master_pollfds_changed;
+  master_pollfds_changed = FALSE;
+  loop->n_fds = master_n_pollfds;
+  loop->fds = master_pollfds;
+  for (i = 0; i < loop->n_fds; i++)
+    loop->fds[i].revents = 0;
+  loop->revents_filled = FALSE;
+  
+  loop->timeout = -1;
+  /* cached checks first */
+  need_dispatch = master_need_reflow || master_need_process;
+  /* lengthy query */
+  if (!need_dispatch)
+    need_dispatch = _engine_job_pending ();
+  /* invoke custom poll checks */
+  if (!need_dispatch)
+    {
+      master_poll_check (&loop->timeout, FALSE);
+      need_dispatch = master_need_process;
+    }
+  if (need_dispatch)
+    loop->timeout = 0;
+  
+  MAS_DEBUG ("PREPARE: need_dispatch=%u timeout=%6ld n_fds=%u",
+	     need_dispatch,
+	     loop->timeout, loop->n_fds);
+  
+  return need_dispatch;
+}
+
+gboolean
+_engine_master_check (const GslEngineLoop *loop)
+{
+  gboolean need_dispatch;
+  
+  g_return_val_if_fail (loop != NULL, FALSE);
+  g_return_val_if_fail (loop->n_fds == master_n_pollfds, FALSE);
+  g_return_val_if_fail (loop->fds == master_pollfds, FALSE);
+  if (loop->n_fds)
+    g_return_val_if_fail (loop->revents_filled == TRUE, FALSE);
+  
+  /* cached checks first */
+  need_dispatch = master_need_reflow || master_need_process;
+  /* lengthy query */
+  if (!need_dispatch)
+    need_dispatch = _engine_job_pending ();
+  /* invoke custom poll checks */
+  if (!need_dispatch)
+    {
+      glong dummy = -1;
+      
+      master_poll_check (&dummy, TRUE);
+      need_dispatch = master_need_process;
+    }
+  
+  MAS_DEBUG ("CHECK: need_dispatch=%u", need_dispatch);
+  
+  return need_dispatch;
+}
+
+void
+_engine_master_dispatch_jobs (void)
+{
+  GslJob *job;
+  
+  job = _engine_pop_job ();
+  while (job)
+    {
+      master_process_job (job);
+      job = _engine_pop_job ();	/* have to process _all_ jobs */
+    }
+}
+
+void
+_engine_master_dispatch (void)
+{
+  /* processing has prime priority, but we can't process the
+   * network, until all jobs have been handled and if necessary
+   * rescheduled the network.
+   * that's why we have to handle everything at once and can't
+   * preliminarily return after just handling jobs or rescheduling.
+   */
+  _engine_master_dispatch_jobs ();
+  if (master_need_reflow)
+    master_reschedule_flow ();
+  if (master_need_process)
+    master_process_flow ();
+}
+
+void
+_engine_master_thread (gpointer data)
+{
+  gboolean run = TRUE;
+  
+  /* assert sane configuration checks, since we're simply casting structures */
+  g_assert (sizeof (struct pollfd) == sizeof (GPollFD) &&
+	    G_STRUCT_OFFSET (GPollFD, fd) == G_STRUCT_OFFSET (struct pollfd, fd) &&
+	    G_STRUCT_OFFSET (GPollFD, events) == G_STRUCT_OFFSET (struct pollfd, events) &&
+	    G_STRUCT_OFFSET (GPollFD, revents) == G_STRUCT_OFFSET (struct pollfd, revents));
+  
+  /* add the thread wakeup pipe to master pollfds, so we get woken
+   * up in time (even though we evaluate the pipe contents later)
+   */
+  gsl_thread_get_pollfd (master_pollfds);
+  master_n_pollfds += 1;
+  master_pollfds_changed = TRUE;
+  
+  toyprof_stampinit ();
+  
+  while (run)
+    {
+      GslEngineLoop loop;
+      gboolean need_dispatch;
+      
+      need_dispatch = _engine_master_prepare (&loop);
+      
+      if (!need_dispatch)
+	{
+	  gint err;
+	  
+	  err = poll ((struct pollfd*) loop.fds, loop.n_fds, loop.timeout);
+	  
+	  if (err >= 0)
+	    loop.revents_filled = TRUE;
+	  else
+	    g_printerr (G_STRLOC ": poll() error: %s\n", g_strerror (errno));
+	  
+	  if (loop.revents_filled)
+	    need_dispatch = _engine_master_check (&loop);
+	}
+      
+      if (need_dispatch)
+	_engine_master_dispatch ();
+      
+      /* handle thread pollfd messages */
+      run = gsl_thread_sleep (0);
+    }
+}
+/* vim:set ts=8 sts=2 sw=2: */