diff options
Diffstat (limited to 'qpid/extras/dispatch/src/container.c')
| -rw-r--r-- | qpid/extras/dispatch/src/container.c | 868 |
1 files changed, 0 insertions, 868 deletions
diff --git a/qpid/extras/dispatch/src/container.c b/qpid/extras/dispatch/src/container.c deleted file mode 100644 index 3ae24d81b0..0000000000 --- a/qpid/extras/dispatch/src/container.c +++ /dev/null @@ -1,868 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -#include <stdio.h> -#include <string.h> -#include "dispatch_private.h" -#include <qpid/dispatch/container.h> -#include <qpid/dispatch/server.h> -#include <qpid/dispatch/message.h> -#include <proton/engine.h> -#include <proton/message.h> -#include <qpid/dispatch/ctools.h> -#include <qpid/dispatch/hash.h> -#include <qpid/dispatch/threading.h> -#include <qpid/dispatch/iterator.h> -#include <qpid/dispatch/log.h> -#include <qpid/dispatch/agent.h> - -static char *module="CONTAINER"; - -struct dx_node_t { - dx_container_t *container; - const dx_node_type_t *ntype; - char *name; - void *context; - dx_dist_mode_t supported_dist; - dx_lifetime_policy_t life_policy; -}; - -ALLOC_DECLARE(dx_node_t); -ALLOC_DEFINE(dx_node_t); -ALLOC_DEFINE(dx_link_item_t); - - -struct dx_link_t { - pn_link_t *pn_link; - void *context; - dx_node_t *node; -}; - -ALLOC_DECLARE(dx_link_t); -ALLOC_DEFINE(dx_link_t); - - -struct dx_delivery_t { - pn_delivery_t *pn_delivery; - dx_delivery_t *peer; - void *context; - uint64_t disposition; - dx_link_t *link; -}; - -ALLOC_DECLARE(dx_delivery_t); -ALLOC_DEFINE(dx_delivery_t); - - -typedef struct dxc_node_type_t { - DEQ_LINKS(struct dxc_node_type_t); - const dx_node_type_t *ntype; -} dxc_node_type_t; -DEQ_DECLARE(dxc_node_type_t, dxc_node_type_list_t); - -static int DX_CONTAINER_CLASS_CONTAINER = 1; -static int DX_CONTAINER_CLASS_NODE_TYPE = 2; -static int DX_CONTAINER_CLASS_NODE = 3; - -typedef struct container_class_t { - dx_container_t *container; - int class_id; -} container_class_t; - -struct dx_container_t { - dx_dispatch_t *dx; - dx_server_t *server; - dx_hash_t *node_type_map; - dx_hash_t *node_map; - sys_mutex_t *lock; - dx_node_t *default_node; - dxc_node_type_list_t node_type_list; - dx_agent_class_t *class_container; - dx_agent_class_t *class_node_type; - dx_agent_class_t *class_node; -}; - -static void setup_outgoing_link(dx_container_t *container, pn_link_t *pn_link) -{ - sys_mutex_lock(container->lock); - dx_node_t *node = 0; - const char *source = pn_terminus_get_address(pn_link_remote_source(pn_link)); - dx_field_iterator_t *iter; - // TODO - Extract the name from the structured source - - if (source) { - iter = dx_field_iterator_string(source, ITER_VIEW_NODE_ID); - dx_hash_retrieve(container->node_map, iter, (void*) &node); - dx_field_iterator_free(iter); - } - sys_mutex_unlock(container->lock); - - if (node == 0) { - if (container->default_node) - node = container->default_node; - else { - // Reject the link - // TODO - When the API allows, add an error message for "no available node" - pn_link_close(pn_link); - return; - } - } - - dx_link_t *link = new_dx_link_t(); - if (!link) { - pn_link_close(pn_link); - return; - } - - link->pn_link = pn_link; - link->context = 0; - link->node = node; - - pn_link_set_context(pn_link, link); - node->ntype->outgoing_handler(node->context, link); -} - - -static void setup_incoming_link(dx_container_t *container, pn_link_t *pn_link) -{ - sys_mutex_lock(container->lock); - dx_node_t *node = 0; - const char *target = pn_terminus_get_address(pn_link_remote_target(pn_link)); - dx_field_iterator_t *iter; - // TODO - Extract the name from the structured target - - if (target) { - iter = dx_field_iterator_string(target, ITER_VIEW_NODE_ID); - dx_hash_retrieve(container->node_map, iter, (void*) &node); - dx_field_iterator_free(iter); - } - sys_mutex_unlock(container->lock); - - if (node == 0) { - if (container->default_node) - node = container->default_node; - else { - // Reject the link - // TODO - When the API allows, add an error message for "no available node" - pn_link_close(pn_link); - return; - } - } - - dx_link_t *link = new_dx_link_t(); - if (!link) { - pn_link_close(pn_link); - return; - } - - link->pn_link = pn_link; - link->context = 0; - link->node = node; - - pn_link_set_context(pn_link, link); - node->ntype->incoming_handler(node->context, link); -} - - -static int do_writable(pn_link_t *pn_link) -{ - dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); - if (!link) - return 0; - - dx_node_t *node = link->node; - if (!node) - return 0; - - return node->ntype->writable_handler(node->context, link); -} - - -static void do_receive(pn_delivery_t *pnd) -{ - pn_link_t *pn_link = pn_delivery_link(pnd); - dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); - dx_delivery_t *delivery = (dx_delivery_t*) pn_delivery_get_context(pnd); - - if (link) { - dx_node_t *node = link->node; - if (node) { - if (!delivery) { - delivery = new_dx_delivery_t(); - delivery->pn_delivery = pnd; - delivery->peer = 0; - delivery->context = 0; - delivery->disposition = 0; - delivery->link = link; - pn_delivery_set_context(pnd, delivery); - } - - node->ntype->rx_handler(node->context, link, delivery); - return; - } - } - - // - // Reject the delivery if we couldn't find a node to handle it - // - pn_link_advance(pn_link); - pn_link_flow(pn_link, 1); - pn_delivery_update(pnd, PN_REJECTED); - pn_delivery_settle(pnd); -} - - -static void do_updated(pn_delivery_t *pnd) -{ - pn_link_t *pn_link = pn_delivery_link(pnd); - dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); - dx_delivery_t *delivery = (dx_delivery_t*) pn_delivery_get_context(pnd); - - if (link && delivery) { - dx_node_t *node = link->node; - if (node) - node->ntype->disp_handler(node->context, link, delivery); - } -} - - -static int close_handler(void* unused, pn_connection_t *conn) -{ - // - // Close all links, passing False as the 'closed' argument. These links are not - // being properly 'detached'. They are being orphaned. - // - pn_link_t *pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE); - while (pn_link) { - dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); - dx_node_t *node = link->node; - if (node && link) - node->ntype->link_detach_handler(node->context, link, 0); - pn_link_close(pn_link); - free_dx_link_t(link); - pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE); - } - - // teardown all sessions - pn_session_t *ssn = pn_session_head(conn, 0); - while (ssn) { - pn_session_close(ssn); - ssn = pn_session_next(ssn, 0); - } - - // teardown the connection - pn_connection_close(conn); - return 0; -} - - -static int process_handler(dx_container_t *container, void* unused, pn_connection_t *conn) -{ - pn_session_t *ssn; - pn_link_t *pn_link; - pn_delivery_t *delivery; - int event_count = 0; - - // Step 1: setup the engine's connection, and any sessions and links - // that may be pending. - - // initialize the connection if it's new - if (pn_connection_state(conn) & PN_LOCAL_UNINIT) { - pn_connection_open(conn); - event_count++; - } - - // open all pending sessions - ssn = pn_session_head(conn, PN_LOCAL_UNINIT); - while (ssn) { - pn_session_open(ssn); - ssn = pn_session_next(ssn, PN_LOCAL_UNINIT); - event_count++; - } - - // configure and open any pending links - pn_link = pn_link_head(conn, PN_LOCAL_UNINIT); - while (pn_link) { - if (pn_link_is_sender(pn_link)) - setup_outgoing_link(container, pn_link); - else - setup_incoming_link(container, pn_link); - pn_link = pn_link_next(pn_link, PN_LOCAL_UNINIT); - event_count++; - } - - - // Step 2: Now drain all the pending deliveries from the connection's - // work queue and process them - - delivery = pn_work_head(conn); - while (delivery) { - if (pn_delivery_readable(delivery)) - do_receive(delivery); - - if (pn_delivery_updated(delivery)) { - do_updated(delivery); - pn_delivery_clear(delivery); - } - delivery = pn_work_next(delivery); - event_count++; - } - - // - // Step 2.5: Call the attached node's writable handler for all active links - // on the connection. Note that in Dispatch, links are considered - // bidirectional. Incoming and outgoing only pertains to deliveries and - // deliveries are a subset of the traffic that flows both directions on links. - // - pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE); - while (pn_link) { - assert(pn_session_connection(pn_link_session(pn_link)) == conn); - event_count += do_writable(pn_link); - pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_ACTIVE); - } - - // Step 3: Clean up any links or sessions that have been closed by the - // remote. If the connection has been closed remotely, clean that up - // also. - - // teardown any terminating links - pn_link = pn_link_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); - while (pn_link) { - dx_link_t *link = (dx_link_t*) pn_link_get_context(pn_link); - dx_node_t *node = link->node; - if (node) - node->ntype->link_detach_handler(node->context, link, 1); // TODO - get 'closed' from detach message - pn_link_close(pn_link); - pn_link = pn_link_next(pn_link, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); - event_count++; - } - - // teardown any terminating sessions - ssn = pn_session_head(conn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); - while (ssn) { - pn_session_close(ssn); - ssn = pn_session_next(ssn, PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED); - event_count++; - } - - // teardown the connection if it's terminating - if (pn_connection_state(conn) == (PN_LOCAL_ACTIVE | PN_REMOTE_CLOSED)) { - pn_connection_close(conn); - event_count++; - } - - return event_count; -} - - -static void open_handler(dx_container_t *container, dx_connection_t *conn, dx_direction_t dir) -{ - const dx_node_type_t *nt; - - // - // Note the locking structure in this function. Generally this would be unsafe, but since - // this particular list is only ever appended to and never has items inserted or deleted, - // this usage is safe in this case. - // - sys_mutex_lock(container->lock); - dxc_node_type_t *nt_item = DEQ_HEAD(container->node_type_list); - sys_mutex_unlock(container->lock); - - pn_connection_open(dx_connection_pn(conn)); - - while (nt_item) { - nt = nt_item->ntype; - if (dir == DX_INCOMING) { - if (nt->inbound_conn_open_handler) - nt->inbound_conn_open_handler(nt->type_context, conn); - } else { - if (nt->outbound_conn_open_handler) - nt->outbound_conn_open_handler(nt->type_context, conn); - } - - sys_mutex_lock(container->lock); - nt_item = DEQ_NEXT(nt_item); - sys_mutex_unlock(container->lock); - } -} - - -static int handler(void *handler_context, void *conn_context, dx_conn_event_t event, dx_connection_t *dx_conn) -{ - dx_container_t *container = (dx_container_t*) handler_context; - pn_connection_t *conn = dx_connection_pn(dx_conn); - - switch (event) { - case DX_CONN_EVENT_LISTENER_OPEN: open_handler(container, dx_conn, DX_INCOMING); break; - case DX_CONN_EVENT_CONNECTOR_OPEN: open_handler(container, dx_conn, DX_OUTGOING); break; - case DX_CONN_EVENT_CLOSE: return close_handler(conn_context, conn); - case DX_CONN_EVENT_PROCESS: return process_handler(container, conn_context, conn); - } - - return 0; -} - - -static void container_schema_handler(void *context, void *correlator) -{ -} - - -static void container_query_handler(void* context, const char *id, void *correlator) -{ - container_class_t *cls = (container_class_t*) context; - - if (cls->class_id == DX_CONTAINER_CLASS_CONTAINER) { - dx_agent_value_uint(correlator, "node_type_count", dx_hash_size(cls->container->node_type_map)); - dx_agent_value_uint(correlator, "node_count", dx_hash_size(cls->container->node_map)); - if (cls->container->default_node) - dx_agent_value_string(correlator, "default_node_type", cls->container->default_node->ntype->type_name); - else - dx_agent_value_null(correlator, "default_node_type"); - dx_agent_value_complete(correlator, false); - - } else if (cls->class_id == DX_CONTAINER_CLASS_NODE_TYPE) { - - } else if (cls->class_id == DX_CONTAINER_CLASS_NODE) { - - } -} - - -dx_agent_class_t *setup_class(dx_container_t *container, const char *fqname, int id) -{ - container_class_t *cls = NEW(container_class_t); - cls->container = container; - cls->class_id = id; - - return dx_agent_register_class(container->dx, fqname, cls, - container_schema_handler, - container_query_handler); -} - - -dx_container_t *dx_container(dx_dispatch_t *dx) -{ - dx_container_t *container = NEW(dx_container_t); - - container->dx = dx; - container->server = dx->server; - container->node_type_map = dx_hash(6, 4, 1); // 64 buckets, item batches of 4 - container->node_map = dx_hash(10, 32, 0); // 1K buckets, item batches of 32 - container->lock = sys_mutex(); - container->default_node = 0; - DEQ_INIT(container->node_type_list); - - dx_log(module, LOG_TRACE, "Container Initializing"); - dx_server_set_conn_handler(dx, handler, container); - - return container; -} - - -void dx_container_setup_agent(dx_dispatch_t *dx) -{ - dx->container->class_container = - setup_class(dx->container, "org.apache.qpid.dispatch.container", DX_CONTAINER_CLASS_CONTAINER); - dx->container->class_node_type = - setup_class(dx->container, "org.apache.qpid.dispatch.container.node_type", DX_CONTAINER_CLASS_NODE_TYPE); - dx->container->class_node = - setup_class(dx->container, "org.apache.qpid.dispatch.container.node", DX_CONTAINER_CLASS_NODE); -} - - -void dx_container_free(dx_container_t *container) -{ - // TODO - Free the nodes - // TODO - Free the node types - sys_mutex_free(container->lock); - free(container); -} - - -int dx_container_register_node_type(dx_dispatch_t *dx, const dx_node_type_t *nt) -{ - dx_container_t *container = dx->container; - - int result; - dx_field_iterator_t *iter = dx_field_iterator_string(nt->type_name, ITER_VIEW_ALL); - dxc_node_type_t *nt_item = NEW(dxc_node_type_t); - DEQ_ITEM_INIT(nt_item); - nt_item->ntype = nt; - - sys_mutex_lock(container->lock); - result = dx_hash_insert_const(container->node_type_map, iter, nt, 0); - DEQ_INSERT_TAIL(container->node_type_list, nt_item); - sys_mutex_unlock(container->lock); - - dx_field_iterator_free(iter); - if (result < 0) - return result; - dx_log(module, LOG_TRACE, "Node Type Registered - %s", nt->type_name); - - return 0; -} - - -dx_node_t *dx_container_set_default_node_type(dx_dispatch_t *dx, - const dx_node_type_t *nt, - void *context, - dx_dist_mode_t supported_dist) -{ - dx_container_t *container = dx->container; - - if (container->default_node) - dx_container_destroy_node(container->default_node); - - if (nt) { - container->default_node = dx_container_create_node(dx, nt, 0, context, supported_dist, DX_LIFE_PERMANENT); - dx_log(module, LOG_TRACE, "Node of type '%s' installed as default node", nt->type_name); - } else { - container->default_node = 0; - dx_log(module, LOG_TRACE, "Default node removed"); - } - - return container->default_node; -} - - -dx_node_t *dx_container_create_node(dx_dispatch_t *dx, - const dx_node_type_t *nt, - const char *name, - void *context, - dx_dist_mode_t supported_dist, - dx_lifetime_policy_t life_policy) -{ - dx_container_t *container = dx->container; - int result; - dx_node_t *node = new_dx_node_t(); - if (!node) - return 0; - - node->container = container; - node->ntype = nt; - node->name = 0; - node->context = context; - node->supported_dist = supported_dist; - node->life_policy = life_policy; - - if (name) { - dx_field_iterator_t *iter = dx_field_iterator_string(name, ITER_VIEW_ALL); - sys_mutex_lock(container->lock); - result = dx_hash_insert(container->node_map, iter, node, 0); - sys_mutex_unlock(container->lock); - dx_field_iterator_free(iter); - if (result < 0) { - free_dx_node_t(node); - return 0; - } - - node->name = (char*) malloc(strlen(name) + 1); - strcpy(node->name, name); - } - - if (name) - dx_log(module, LOG_TRACE, "Node of type '%s' created with name '%s'", nt->type_name, name); - - return node; -} - - -void dx_container_destroy_node(dx_node_t *node) -{ - dx_container_t *container = node->container; - - if (node->name) { - dx_field_iterator_t *iter = dx_field_iterator_string(node->name, ITER_VIEW_ALL); - sys_mutex_lock(container->lock); - dx_hash_remove(container->node_map, iter); - sys_mutex_unlock(container->lock); - dx_field_iterator_free(iter); - free(node->name); - } - - free_dx_node_t(node); -} - - -void dx_container_node_set_context(dx_node_t *node, void *node_context) -{ - node->context = node_context; -} - - -dx_dist_mode_t dx_container_node_get_dist_modes(const dx_node_t *node) -{ - return node->supported_dist; -} - - -dx_lifetime_policy_t dx_container_node_get_life_policy(const dx_node_t *node) -{ - return node->life_policy; -} - - -dx_link_t *dx_link(dx_node_t *node, dx_connection_t *conn, dx_direction_t dir, const char* name) -{ - pn_session_t *sess = pn_session(dx_connection_pn(conn)); - dx_link_t *link = new_dx_link_t(); - - if (dir == DX_OUTGOING) - link->pn_link = pn_sender(sess, name); - else - link->pn_link = pn_receiver(sess, name); - link->context = node->context; - link->node = node; - - pn_link_set_context(link->pn_link, link); - - pn_session_open(sess); - - return link; -} - - -void dx_link_set_context(dx_link_t *link, void *context) -{ - link->context = context; -} - - -void *dx_link_get_context(dx_link_t *link) -{ - return link->context; -} - - -void dx_link_set_conn_context(dx_link_t *link, void *context) -{ - pn_session_t *pn_sess = pn_link_session(link->pn_link); - if (!pn_sess) - return; - pn_connection_t *pn_conn = pn_session_connection(pn_sess); - if (!pn_conn) - return; - dx_connection_t *conn = (dx_connection_t*) pn_connection_get_context(pn_conn); - if (!conn) - return; - dx_connection_set_link_context(conn, context); -} - - -void *dx_link_get_conn_context(dx_link_t *link) -{ - pn_session_t *pn_sess = pn_link_session(link->pn_link); - if (!pn_sess) - return 0; - pn_connection_t *pn_conn = pn_session_connection(pn_sess); - if (!pn_conn) - return 0; - dx_connection_t *conn = (dx_connection_t*) pn_connection_get_context(pn_conn); - if (!conn) - return 0; - return dx_connection_get_link_context(conn); -} - - -pn_link_t *dx_link_pn(dx_link_t *link) -{ - return link->pn_link; -} - - -dx_connection_t *dx_link_connection(dx_link_t *link) -{ - if (!link || !link->pn_link) - return 0; - - pn_session_t *sess = pn_link_session(link->pn_link); - if (!sess) - return 0; - - pn_connection_t *conn = pn_session_connection(sess); - if (!conn) - return 0; - - dx_connection_t *ctx = pn_connection_get_context(conn); - if (!ctx) - return 0; - - return ctx; -} - - -pn_terminus_t *dx_link_source(dx_link_t *link) -{ - return pn_link_source(link->pn_link); -} - - -pn_terminus_t *dx_link_target(dx_link_t *link) -{ - return pn_link_target(link->pn_link); -} - - -pn_terminus_t *dx_link_remote_source(dx_link_t *link) -{ - return pn_link_remote_source(link->pn_link); -} - - -pn_terminus_t *dx_link_remote_target(dx_link_t *link) -{ - return pn_link_remote_target(link->pn_link); -} - - -void dx_link_activate(dx_link_t *link) -{ - if (!link || !link->pn_link) - return; - - pn_session_t *sess = pn_link_session(link->pn_link); - if (!sess) - return; - - pn_connection_t *conn = pn_session_connection(sess); - if (!conn) - return; - - dx_connection_t *ctx = pn_connection_get_context(conn); - if (!ctx) - return; - - dx_server_activate(ctx); -} - - -void dx_link_close(dx_link_t *link) -{ - pn_link_close(link->pn_link); -} - - -dx_delivery_t *dx_delivery(dx_link_t *link, pn_delivery_tag_t tag) -{ - pn_link_t *pnl = dx_link_pn(link); - - // - // If there is a current delivery on this outgoing link, something - // is wrong with the delivey algorithm. We assume that the current - // delivery ('pnd' below) is the one created by pn_delivery. If it is - // not, then my understanding of how proton works is incorrect. - // - assert(!pn_link_current(pnl)); - - pn_delivery(pnl, tag); - pn_delivery_t *pnd = pn_link_current(pnl); - - if (!pnd) - return 0; - - dx_delivery_t *delivery = new_dx_delivery_t(); - delivery->pn_delivery = pnd; - delivery->peer = 0; - delivery->context = 0; - delivery->disposition = 0; - delivery->link = link; - pn_delivery_set_context(pnd, delivery); - - return delivery; -} - - -void dx_delivery_free(dx_delivery_t *delivery, uint64_t final_disposition) -{ - if (delivery->pn_delivery) { - if (final_disposition > 0) - pn_delivery_update(delivery->pn_delivery, final_disposition); - pn_delivery_set_context(delivery->pn_delivery, 0); - pn_delivery_settle(delivery->pn_delivery); - } - if (delivery->peer) - delivery->peer->peer = 0; - free_dx_delivery_t(delivery); -} - - -void dx_delivery_set_peer(dx_delivery_t *delivery, dx_delivery_t *peer) -{ - delivery->peer = peer; -} - - -void dx_delivery_set_context(dx_delivery_t *delivery, void *context) -{ - delivery->context = context; -} - - -void *dx_delivery_context(dx_delivery_t *delivery) -{ - return delivery->context; -} - - -dx_delivery_t *dx_delivery_peer(dx_delivery_t *delivery) -{ - return delivery->peer; -} - - -pn_delivery_t *dx_delivery_pn(dx_delivery_t *delivery) -{ - return delivery->pn_delivery; -} - - -void dx_delivery_settle(dx_delivery_t *delivery) -{ - if (delivery->pn_delivery) { - pn_delivery_settle(delivery->pn_delivery); - delivery->pn_delivery = 0; - } -} - - -bool dx_delivery_settled(dx_delivery_t *delivery) -{ - return pn_delivery_settled(delivery->pn_delivery); -} - - -bool dx_delivery_disp_changed(dx_delivery_t *delivery) -{ - return delivery->disposition != pn_delivery_remote_state(delivery->pn_delivery); -} - - -uint64_t dx_delivery_disp(dx_delivery_t *delivery) -{ - delivery->disposition = pn_delivery_remote_state(delivery->pn_delivery); - return delivery->disposition; -} - - -dx_link_t *dx_delivery_link(dx_delivery_t *delivery) -{ - return delivery->link; -} - |