summaryrefslogtreecommitdiffstats
path: root/flow/gsl/gslopmaster.c
diff options
context:
space:
mode:
authortpearson <tpearson@283d02a7-25f6-0310-bc7c-ecb5cbfe19da>2010-01-05 00:01:18 +0000
committertpearson <tpearson@283d02a7-25f6-0310-bc7c-ecb5cbfe19da>2010-01-05 00:01:18 +0000
commit42995d7bf396933ee60c5f89c354ea89cf13df0d (patch)
treecfdcea0ac57420e7baf570bfe435e107bb842541 /flow/gsl/gslopmaster.c
downloadarts-42995d7bf396933ee60c5f89c354ea89cf13df0d.tar.gz
arts-42995d7bf396933ee60c5f89c354ea89cf13df0d.zip
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
Diffstat (limited to 'flow/gsl/gslopmaster.c')
-rw-r--r--flow/gsl/gslopmaster.c783
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: */