}
extern bt_self_component_port_input_message_iterator *
-bt_self_component_port_input_message_iterator_create(
+bt_self_component_port_input_message_iterator_create_from_message_iterator(
+ bt_self_message_iterator *self_msg_iter,
+ bt_self_component_port_input *input_port);
+
+extern bt_self_component_port_input_message_iterator *
+bt_self_component_port_input_message_iterator_create_from_sink_component(
+ bt_self_component_sink *self_comp,
bt_self_component_port_input *input_port);
extern bt_component *
)
assert self_port_ptr
return bt2.port._UserComponentInputPort._create_from_ptr(self_port_ptr)
+
+ def _create_input_port_message_iterator(self, input_port):
+ utils._check_type(input_port, bt2.port._UserComponentInputPort)
+
+ msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create_from_sink_component(
+ self._bt_ptr, input_port._ptr
+ )
+
+ if msg_iter_ptr is None:
+ raise bt2.CreationError('cannot create message iterator object')
+
+ return bt2.message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)
import bt2.message
import collections.abc
import bt2.component
+import bt2.port
import bt2
def _bt_seek_beginning_from_native(self):
self._seek_beginning()
+ def _create_input_port_message_iterator(self, input_port):
+ utils._check_type(input_port, bt2.port._UserComponentInputPort)
+
+ msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create_from_message_iterator(
+ self._bt_ptr, input_port._ptr
+ )
+
+ if msg_iter_ptr is None:
+ raise bt2.CreationError('cannot create message iterator object')
+
+ return _UserComponentInputPortMessageIterator(msg_iter_ptr)
+
def _create_event_message(
self, event_class, parent=None, default_clock_snapshot=None
):
native_bt.self_component_port_input_as_self_component_port
)
- def create_message_iterator(self):
- msg_iter_ptr = native_bt.self_component_port_input_message_iterator_create(
- self._ptr
- )
- if msg_iter_ptr is None:
- raise bt2._MemoryError('cannot create message iterator object')
-
- return bt2.message_iterator._UserComponentInputPortMessageIterator(msg_iter_ptr)
-
class _UserComponentOutputPort(_UserComponentPort, _OutputPort):
_as_self_port_ptr = staticmethod(
BT_ASSERT(colander_data);
BT_OBJECT_PUT_REF_AND_RESET(colander_data->msg_iter);
colander_data->msg_iter =
- bt_self_component_port_input_message_iterator_create(
- self_port);
+ bt_self_component_port_input_message_iterator_create_from_sink_component(
+ self_comp, self_port);
if (!colander_data->msg_iter) {
BT_LIB_LOGE_APPEND_CAUSE("Cannot create message iterator on "
"self component input port: %![port-]+p",
iterator->auto_seek.msgs = NULL;
}
+ if (iterator->upstream_msg_iters) {
+ /*
+ * At this point the message iterator is finalized, so
+ * it's detached from any upstream message iterator.
+ */
+ BT_ASSERT(iterator->upstream_msg_iters->len == 0);
+ g_ptr_array_free(iterator->upstream_msg_iters, TRUE);
+ iterator->upstream_msg_iters = NULL;
+ }
+
destroy_base_message_iterator(obj);
}
void bt_self_component_port_input_message_iterator_try_finalize(
struct bt_self_component_port_input_message_iterator *iterator)
{
+ uint64_t i;
typedef void (*method_t)(void *);
struct bt_component_class *comp_class = NULL;
method(iterator);
}
+ /* Detach upstream message iterators */
+ for (i = 0; i < iterator->upstream_msg_iters->len; i++) {
+ struct bt_self_component_port_input_message_iterator *upstream_msg_iter =
+ iterator->upstream_msg_iters->pdata[i];
+
+ upstream_msg_iter->downstream_msg_iter = NULL;
+ }
+
+ g_ptr_array_set_size(iterator->upstream_msg_iters, 0);
+
+ /* Detach downstream message iterator */
+ if (iterator->downstream_msg_iter) {
+ gboolean existed;
+
+ BT_ASSERT(iterator->downstream_msg_iter->upstream_msg_iters);
+ existed = g_ptr_array_remove_fast(
+ iterator->downstream_msg_iter->upstream_msg_iters,
+ iterator);
+ BT_ASSERT(existed);
+ }
+
iterator->upstream_component = NULL;
iterator->upstream_port = NULL;
set_self_comp_port_input_msg_iterator_state(iterator,
static
struct bt_self_component_port_input_message_iterator *
-bt_self_component_port_input_message_iterator_create_initial(
- struct bt_component *upstream_comp,
- struct bt_port *upstream_port)
+create_self_component_input_port_message_iterator(
+ struct bt_self_message_iterator *self_downstream_msg_iter,
+ struct bt_self_component_port_input *self_port)
{
+ typedef enum bt_component_class_message_iterator_init_method_status (*init_method_t)(
+ void *, void *, void *);
+
int ret;
- struct bt_self_component_port_input_message_iterator *iterator = NULL;
+ init_method_t init_method = NULL;
+ struct bt_self_component_port_input_message_iterator *iterator =
+ NULL;
+ struct bt_self_component_port_input_message_iterator *downstream_msg_iter =
+ (void *) self_downstream_msg_iter;
+ struct bt_port *port = (void *) self_port;
+ struct bt_port *upstream_port;
+ struct bt_component *comp;
+ struct bt_component *upstream_comp;
+ struct bt_component_class *upstream_comp_cls;
- BT_ASSERT(upstream_comp);
+ BT_ASSERT_PRE_NON_NULL(port, "Input port");
+ comp = bt_port_borrow_component_inline(port);
+ BT_ASSERT_PRE(bt_port_is_connected(port),
+ "Input port is not connected: %![port-]+p", port);
+ BT_ASSERT_PRE(comp, "Input port is not part of a component: %![port-]+p",
+ port);
+ BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
+ "Input port's component's graph is canceled: "
+ "%![port-]+p, %![comp-]+c", port, comp);
+ BT_ASSERT(port->connection);
+ upstream_port = port->connection->upstream_port;
BT_ASSERT(upstream_port);
- BT_ASSERT(bt_port_is_connected(upstream_port));
- BT_LIB_LOGI("Creating initial message iterator on self component input port: "
- "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
- BT_ASSERT(bt_component_get_class_type(upstream_comp) ==
+ upstream_comp = bt_port_borrow_component_inline(upstream_port);
+ BT_ASSERT(upstream_comp);
+ BT_ASSERT_PRE(
+ bt_component_borrow_graph(upstream_comp)->config_state !=
+ BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
+ "Graph is not configured: %!+g",
+ bt_component_borrow_graph(upstream_comp));
+ upstream_comp_cls = upstream_comp->class;
+ BT_ASSERT(upstream_comp->class->type ==
BT_COMPONENT_CLASS_TYPE_SOURCE ||
- bt_component_get_class_type(upstream_comp) ==
+ upstream_comp->class->type ==
BT_COMPONENT_CLASS_TYPE_FILTER);
+ BT_LIB_LOGI("Creating message iterator on self component input port: "
+ "%![up-comp-]+c, %![up-port-]+p", upstream_comp, upstream_port);
iterator = g_new0(
struct bt_self_component_port_input_message_iterator, 1);
if (!iterator) {
}
iterator->last_ns_from_origin = INT64_MIN;
-
iterator->auto_seek.msgs = g_queue_new();
if (!iterator->auto_seek.msgs) {
BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GQueue.");
- ret = -1;
- goto end;
+ goto error;
+ }
+
+ iterator->upstream_msg_iters = g_ptr_array_new();
+ if (!iterator->upstream_msg_iters) {
+ BT_LIB_LOGE_APPEND_CAUSE("Failed to allocate a GPtrArray.");
+ goto error;
}
iterator->upstream_component = upstream_comp;
can_seek_beginning_true;
}
- BT_LIB_LOGI("Created initial message iterator on self component input port: "
- "%![up-port-]+p, %![up-comp-]+c, %![iter-]+i",
- upstream_port, upstream_comp, iterator);
- goto end;
-
-error:
- BT_OBJECT_PUT_REF_AND_RESET(iterator);
-
-end:
- return iterator;
-}
-
-struct bt_self_component_port_input_message_iterator *
-bt_self_component_port_input_message_iterator_create(
- struct bt_self_component_port_input *self_port)
-{
- typedef enum bt_component_class_message_iterator_init_method_status (*init_method_t)(
- void *, void *, void *);
-
- init_method_t init_method = NULL;
- struct bt_self_component_port_input_message_iterator *iterator =
- NULL;
- struct bt_port *port = (void *) self_port;
- struct bt_port *upstream_port;
- struct bt_component *comp;
- struct bt_component *upstream_comp;
- struct bt_component_class *upstream_comp_cls;
-
- BT_ASSERT_PRE_NON_NULL(port, "Port");
- comp = bt_port_borrow_component_inline(port);
- BT_ASSERT_PRE(bt_port_is_connected(port),
- "Port is not connected: %![port-]+p", port);
- BT_ASSERT_PRE(comp, "Port is not part of a component: %![port-]+p",
- port);
- BT_ASSERT_PRE(!bt_component_graph_is_canceled(comp),
- "Port's component's graph is canceled: "
- "%![port-]+p, %![comp-]+c", port, comp);
- BT_ASSERT(port->connection);
- upstream_port = port->connection->upstream_port;
- BT_ASSERT(upstream_port);
- upstream_comp = bt_port_borrow_component_inline(upstream_port);
- BT_ASSERT(upstream_comp);
- BT_ASSERT_PRE(
- bt_component_borrow_graph(upstream_comp)->config_state !=
- BT_GRAPH_CONFIGURATION_STATE_CONFIGURING,
- "Graph is not configured: %!+g",
- bt_component_borrow_graph(upstream_comp));
- upstream_comp_cls = upstream_comp->class;
- BT_ASSERT(upstream_comp->class->type ==
- BT_COMPONENT_CLASS_TYPE_SOURCE ||
- upstream_comp->class->type ==
- BT_COMPONENT_CLASS_TYPE_FILTER);
- iterator = bt_self_component_port_input_message_iterator_create_initial(
- upstream_comp, upstream_port);
- if (!iterator) {
- BT_LIB_LOGE_APPEND_CAUSE(
- "Cannot create self component input port message iterator.");
- goto error;
- }
-
switch (upstream_comp_cls->type) {
case BT_COMPONENT_CLASS_TYPE_SOURCE:
{
}
}
+ if (downstream_msg_iter) {
+ /* Set this message iterator's downstream message iterator */
+ iterator->downstream_msg_iter = downstream_msg_iter;
+
+ /*
+ * Add this message iterator to the downstream message
+ * iterator's array of upstream message iterators.
+ */
+ g_ptr_array_add(downstream_msg_iter->upstream_msg_iters,
+ iterator);
+ }
+
set_self_comp_port_input_msg_iterator_state(iterator,
BT_SELF_COMPONENT_PORT_INPUT_MESSAGE_ITERATOR_STATE_ACTIVE);
g_ptr_array_add(port->connection->iterators, iterator);
return iterator;
}
+struct bt_self_component_port_input_message_iterator *
+bt_self_component_port_input_message_iterator_create_from_message_iterator(
+ struct bt_self_message_iterator *self_msg_iter,
+ struct bt_self_component_port_input *input_port)
+{
+ BT_ASSERT_PRE_NON_NULL(self_msg_iter, "Message iterator");
+ return create_self_component_input_port_message_iterator(self_msg_iter,
+ input_port);
+}
+
+struct bt_self_component_port_input_message_iterator *
+bt_self_component_port_input_message_iterator_create_from_sink_component(
+ struct bt_self_component_sink *self_comp,
+ struct bt_self_component_port_input *input_port)
+{
+ BT_ASSERT_PRE_NON_NULL(self_comp, "Sink component");
+ return create_self_component_input_port_message_iterator(NULL,
+ input_port);
+}
+
void *bt_self_message_iterator_get_data(
const struct bt_self_message_iterator *self_iterator)
{
struct bt_connection *connection; /* Weak */
struct bt_graph *graph; /* Weak */
+ /*
+ * Array of
+ * `struct bt_self_component_port_input_message_iterator *`
+ * (weak).
+ *
+ * This is an array of upstream message iterators on which this
+ * iterator depends. The references are weak: an upstream
+ * message iterator is responsible for removing its entry within
+ * this array on finalization/destruction.
+ */
+ GPtrArray *upstream_msg_iters;
+
+ /*
+ * Downstream message iterator which depends on this message
+ * iterator (weak).
+ *
+ * This can be `NULL` if this message iterator's owner is a sink
+ * component.
+ */
+ struct bt_self_component_port_input_message_iterator *downstream_msg_iter;
+
struct {
bt_self_component_port_input_message_iterator_next_method next;
bt_self_component_port_input_message_iterator_seek_ns_from_origin_method seek_ns_from_origin;
bt_self_component_sink_as_self_component(self_comp));
fs_sink->upstream_iter =
- bt_self_component_port_input_message_iterator_create(
+ bt_self_component_port_input_message_iterator_create_from_sink_component(
+ self_comp,
bt_self_component_sink_borrow_input_port_by_name(
self_comp, in_port_name));
if (!fs_sink->upstream_iter) {
debug_info_msg_iter->self_comp = self_comp;
/* Create an iterator on the upstream component. */
- upstream_iterator = bt_self_component_port_input_message_iterator_create(
- input_port);
+ upstream_iterator = bt_self_component_port_input_message_iterator_create_from_message_iterator(
+ self_msg_iter, input_port);
if (!upstream_iterator) {
status = BT_COMPONENT_CLASS_MESSAGE_ITERATOR_INIT_METHOD_STATUS_MEMORY_ERROR;
goto error;
goto end;
}
- iterator = bt_self_component_port_input_message_iterator_create(
- bt_self_component_sink_borrow_input_port_by_name(comp,
+ iterator = bt_self_component_port_input_message_iterator_create_from_sink_component(
+ comp, bt_self_component_sink_borrow_input_port_by_name(comp,
in_port_name));
if (!iterator) {
status = BT_COMPONENT_CLASS_SINK_GRAPH_IS_CONFIGURED_METHOD_STATUS_MEMORY_ERROR;
bt_self_component_sink_as_self_component(comp));
BT_ASSERT(pretty);
BT_ASSERT(!pretty->iterator);
- pretty->iterator = bt_self_component_port_input_message_iterator_create(
- bt_self_component_sink_borrow_input_port_by_name(comp,
+ pretty->iterator = bt_self_component_port_input_message_iterator_create_from_sink_component(
+ comp, bt_self_component_sink_borrow_input_port_by_name(comp,
in_port_name));
if (!pretty->iterator) {
status = BT_COMPONENT_CLASS_SINK_GRAPH_IS_CONFIGURED_METHOD_STATUS_ERROR;
counter = bt_self_component_get_data(
bt_self_component_sink_as_self_component(comp));
BT_ASSERT(counter);
- iterator = bt_self_component_port_input_message_iterator_create(
- bt_self_component_sink_borrow_input_port_by_name(comp,
+ iterator = bt_self_component_port_input_message_iterator_create_from_sink_component(
+ comp, bt_self_component_sink_borrow_input_port_by_name(comp,
in_port_name));
if (!iterator) {
status = BT_COMPONENT_CLASS_SINK_GRAPH_IS_CONFIGURED_METHOD_STATUS_MEMORY_ERROR;
dummy = bt_self_component_get_data(
bt_self_component_sink_as_self_component(comp));
BT_ASSERT(dummy);
- iterator = bt_self_component_port_input_message_iterator_create(
- bt_self_component_sink_borrow_input_port_by_name(comp,
+ iterator = bt_self_component_port_input_message_iterator_create_from_sink_component(
+ comp, bt_self_component_sink_borrow_input_port_by_name(comp,
in_port_name));
if (!iterator) {
status = BT_COMPONENT_CLASS_SINK_GRAPH_IS_CONFIGURED_METHOD_STATUS_MEMORY_ERROR;
struct muxer_msg_iter {
struct muxer_comp *muxer_comp;
+ /* Weak */
+ bt_self_message_iterator *self_msg_iter;
+
/*
* Array of struct muxer_upstream_msg_iter * (owned by this).
*
static
bt_self_component_port_input_message_iterator *
create_msg_iter_on_input_port(struct muxer_comp *muxer_comp,
+ struct muxer_msg_iter *muxer_msg_iter,
bt_self_component_port_input *self_port)
{
const bt_port *port = bt_self_component_port_as_port(
// TODO: Advance the iterator to >= the time of the latest
// returned message by the muxer message
// iterator which creates it.
- msg_iter = bt_self_component_port_input_message_iterator_create(
- self_port);
+ msg_iter = bt_self_component_port_input_message_iterator_create_from_message_iterator(
+ muxer_msg_iter->self_msg_iter, self_port);
if (!msg_iter) {
BT_COMP_LOGE("Cannot create upstream message iterator on input port: "
"port-addr=%p, port-name=\"%s\"",
}
upstream_msg_iter = create_msg_iter_on_input_port(muxer_comp,
- self_port);
+ muxer_msg_iter, self_port);
if (!upstream_msg_iter) {
/* create_msg_iter_on_input_port() logs errors */
BT_ASSERT(!upstream_msg_iter);
}
muxer_msg_iter->muxer_comp = muxer_comp;
+ muxer_msg_iter->self_msg_iter = self_msg_iter;
muxer_msg_iter->last_returned_ts_ns = INT64_MIN;
muxer_msg_iter->active_muxer_upstream_msg_iters =
g_ptr_array_new_with_free_func(
trimmer_it->begin = trimmer_it->trimmer_comp->begin;
trimmer_it->end = trimmer_it->trimmer_comp->end;
trimmer_it->upstream_iter =
- bt_self_component_port_input_message_iterator_create(
+ bt_self_component_port_input_message_iterator_create_from_message_iterator(
+ self_msg_iter,
bt_self_component_filter_borrow_input_port_by_name(
self_comp, in_port_name));
if (!trimmer_it->upstream_iter) {
self._in = self._add_input_port('in')
def _graph_is_configured(self):
- self._iter = self._in.create_message_iterator()
+ self._iter = self._create_input_port_message_iterator(self._in)
def _consume(self):
next(self._iter)
self._in = self._add_input_port('in')
def _graph_is_configured(self):
- self._iter = self._in.create_message_iterator()
+ self._iter = self._create_input_port_message_iterator(self._in)
def _consume(self):
try:
return next(self._msg_iter)
def _graph_is_configured(self):
- self._msg_iter = self._input_ports['in'].create_message_iterator()
+ self._msg_iter = self._create_input_port_message_iterator(
+ self._input_ports['in']
+ )
graph = bt2.Graph()
up = graph.add_component(MySource, 'down')
comp_self._at += 1
def _graph_is_configured(self):
- self._msg_iter = self._input_port.create_message_iterator()
+ self._msg_iter = self._create_input_port_message_iterator(
+ self._input_port
+ )
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
comp_self._at += 1
def _graph_is_configured(self):
- self._msg_iter = self._input_port.create_message_iterator()
+ self._msg_iter = self._create_input_port_message_iterator(
+ self._input_port
+ )
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
comp_self._at += 1
def _graph_is_configured(self):
- self._msg_iter = self._input_port.create_message_iterator()
+ self._msg_iter = self._create_input_port_message_iterator(
+ self._input_port
+ )
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
class UserMessageIteratorTestCase(unittest.TestCase):
@staticmethod
- def _create_graph(src_comp_cls):
+ def _create_graph(src_comp_cls, flt_comp_cls=None):
class MySink(bt2._UserSinkComponent):
def __init__(self, params):
self._add_input_port('in')
next(self._msg_iter)
def _graph_is_configured(self):
- self._msg_iter = self._input_ports['in'].create_message_iterator()
+ self._msg_iter = self._create_input_port_message_iterator(
+ self._input_ports['in']
+ )
graph = bt2.Graph()
src_comp = graph.add_component(src_comp_cls, 'src')
+
+ if flt_comp_cls is not None:
+ flt_comp = graph.add_component(flt_comp_cls, 'flt')
+
sink_comp = graph.add_component(MySink, 'sink')
- graph.connect_ports(src_comp.output_ports['out'], sink_comp.input_ports['in'])
+
+ if flt_comp_cls is not None:
+ assert flt_comp is not None
+ graph.connect_ports(
+ src_comp.output_ports['out'], flt_comp.input_ports['in']
+ )
+ out_port = flt_comp.output_ports['out']
+ else:
+ out_port = src_comp.output_ports['out']
+
+ graph.connect_ports(out_port, sink_comp.input_ports['in'])
return graph
def test_init(self):
)
self.assertEqual(the_output_port_from_iter.user_data, 'user data')
+ def test_create_from_message_iterator(self):
+ class MySourceIter(bt2._UserMessageIterator):
+ def __init__(self, self_port_output):
+ nonlocal src_iter_initialized
+ src_iter_initialized = True
+
+ class MySource(bt2._UserSourceComponent, message_iterator_class=MySourceIter):
+ def __init__(self, params):
+ self._add_output_port('out')
+
+ class MyFilterIter(bt2._UserMessageIterator):
+ def __init__(self, self_port_output):
+ nonlocal flt_iter_initialized
+ flt_iter_initialized = True
+ self._up_iter = self._create_input_port_message_iterator(
+ self._component._input_ports['in']
+ )
+
+ def __next__(self):
+ return next(self._up_iter)
+
+ class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
+ def __init__(self, params):
+ self._add_input_port('in')
+ self._add_output_port('out')
+
+ src_iter_initialized = False
+ flt_iter_initialized = False
+ graph = self._create_graph(MySource, MyFilter)
+ graph.run()
+ self.assertTrue(src_iter_initialized)
+ self.assertTrue(flt_iter_initialized)
+
def test_finalize(self):
class MyIter(bt2._UserMessageIterator):
def _finalize(self):
class MyFilterIter(bt2._UserMessageIterator):
def __init__(self, port):
input_port = port.user_data
- self._upstream_iter = input_port.create_message_iterator()
+ self._upstream_iter = self._create_input_port_message_iterator(
+ input_port
+ )
def __next__(self):
return next(self._upstream_iter)
projects / babeltrace.git / commitdiff
