/*
* Copyright (c) 2004-2006 Voltaire, Inc. All rights reserved.
* Copyright (c) 2002-2005 Mellanox Technologies LTD. All rights reserved.
* Copyright (c) 1996-2003 Intel Corporation. All rights reserved.
*
* This software is available to you under a choice of one of two
* licenses. You may choose to be licensed under the terms of the GNU
* General Public License (GPL) Version 2, available from the file
* COPYING in the main directory of this source tree, or the
* OpenIB.org BSD license below:
*
* Redistribution and use in source and binary forms, with or
* without modification, are permitted provided that the following
* conditions are met:
*
* - Redistributions of source code must retain the above
* copyright notice, this list of conditions and the following
* disclaimer.
*
* - Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following
* disclaimer in the documentation and/or other materials
* provided with the distribution.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
* BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
* ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
*/
/*
* Abstract:
* Implementation of Dispatcher abstraction.
*
*/
#if HAVE_CONFIG_H
# include <config.h>
#endif /* HAVE_CONFIG_H */
#include <stdlib.h>
#include <complib/cl_dispatcher.h>
#include <complib/cl_thread.h>
#include <complib/cl_timer.h>
/* give some guidance when we build our cl_pool of messages */
#define CL_DISP_INITIAL_MSG_COUNT 256
#define CL_DISP_MSG_GROW_SIZE 64
/* give some guidance when we build our cl_pool of registration elements */
#define CL_DISP_INITIAL_REG_COUNT 16
#define CL_DISP_REG_GROW_SIZE 16
/********************************************************************
__cl_disp_worker
Description:
This function takes messages off the FIFO and calls Processmsg()
This function executes as passive level.
Inputs:
p_disp - Pointer to Dispatcher object
Outputs:
None
Returns:
None
********************************************************************/
void __cl_disp_worker(IN void *context)
{
cl_disp_msg_t *p_msg;
cl_dispatcher_t *p_disp = (cl_dispatcher_t *) context;
cl_spinlock_acquire(&p_disp->lock);
/* Process the FIFO until we drain it dry. */
while (cl_qlist_count(&p_disp->msg_fifo)) {
/* Pop the message at the head from the FIFO. */
p_msg =
(cl_disp_msg_t *) cl_qlist_remove_head(&p_disp->msg_fifo);
/* we track the time the last message spent in the queue */
p_disp->last_msg_queue_time_us =
cl_get_time_stamp() - p_msg->in_time;
/*
* Release the spinlock while the message is processed.
* The user's callback may reenter the dispatcher
* and cause the lock to be reaquired.
*/
cl_spinlock_release(&p_disp->lock);
p_msg->p_dest_reg->pfn_rcv_callback((void *)p_msg->p_dest_reg->
context,
(void *)p_msg->p_data);
cl_atomic_dec(&p_msg->p_dest_reg->ref_cnt);
/* The client has seen the data. Notify the sender as appropriate. */
if (p_msg->pfn_xmt_callback) {
p_msg->pfn_xmt_callback((void *)p_msg->context,
(void *)p_msg->p_data);
cl_atomic_dec(&p_msg->p_src_reg->ref_cnt);
}
/* Grab the lock for the next iteration through the list. */
cl_spinlock_acquire(&p_disp->lock);
/* Return this message to the pool. */
cl_qpool_put(&p_disp->msg_pool, (cl_pool_item_t *) p_msg);
}
cl_spinlock_release(&p_disp->lock);
}
void cl_disp_construct(IN cl_dispatcher_t * const p_disp)
{
CL_ASSERT(p_disp);
cl_qlist_init(&p_disp->reg_list);
cl_ptr_vector_construct(&p_disp->reg_vec);
cl_qlist_init(&p_disp->msg_fifo);
cl_spinlock_construct(&p_disp->lock);
cl_qpool_construct(&p_disp->msg_pool);
}
void cl_disp_shutdown(IN cl_dispatcher_t * const p_disp)
{
CL_ASSERT(p_disp);
/* Stop the thread pool. */
cl_thread_pool_destroy(&p_disp->worker_threads);
/* Process all outstanding callbacks. */
__cl_disp_worker(p_disp);
/* Free all registration info. */
while (!cl_is_qlist_empty(&p_disp->reg_list))
free(cl_qlist_remove_head(&p_disp->reg_list));
}
void cl_disp_destroy(IN cl_dispatcher_t * const p_disp)
{
CL_ASSERT(p_disp);
cl_spinlock_destroy(&p_disp->lock);
/* Destroy the message pool */
cl_qpool_destroy(&p_disp->msg_pool);
/* Destroy the pointer vector of registrants. */
cl_ptr_vector_destroy(&p_disp->reg_vec);
}
cl_status_t cl_disp_init(IN cl_dispatcher_t * const p_disp,
IN const uint32_t thread_count,
IN const char *const name)
{
cl_status_t status;
CL_ASSERT(p_disp);
cl_disp_construct(p_disp);
status = cl_spinlock_init(&p_disp->lock);
if (status != CL_SUCCESS) {
cl_disp_destroy(p_disp);
return (status);
}
/* Specify no upper limit to the number of messages in the pool */
status = cl_qpool_init(&p_disp->msg_pool, CL_DISP_INITIAL_MSG_COUNT,
0, CL_DISP_MSG_GROW_SIZE, sizeof(cl_disp_msg_t),
NULL, NULL, NULL);
if (status != CL_SUCCESS) {
cl_disp_destroy(p_disp);
return (status);
}
status = cl_ptr_vector_init(&p_disp->reg_vec, CL_DISP_INITIAL_REG_COUNT,
CL_DISP_REG_GROW_SIZE);
if (status != CL_SUCCESS) {
cl_disp_destroy(p_disp);
return (status);
}
status = cl_thread_pool_init(&p_disp->worker_threads, thread_count,
__cl_disp_worker, p_disp, name);
if (status != CL_SUCCESS)
cl_disp_destroy(p_disp);
return (status);
}
cl_disp_reg_handle_t cl_disp_register(IN cl_dispatcher_t * const p_disp,
IN const cl_disp_msgid_t msg_id,
IN cl_pfn_msgrcv_cb_t pfn_callback
OPTIONAL,
IN const void *const context OPTIONAL)
{
cl_disp_reg_info_t *p_reg;
cl_status_t status;
CL_ASSERT(p_disp);
/* Check that the requested registrant ID is available. */
cl_spinlock_acquire(&p_disp->lock);
if ((msg_id != CL_DISP_MSGID_NONE) &&
(msg_id < cl_ptr_vector_get_size(&p_disp->reg_vec)) &&
(cl_ptr_vector_get(&p_disp->reg_vec, msg_id))) {
cl_spinlock_release(&p_disp->lock);
return (NULL);
}
/* Get a registration info from the pool. */
p_reg = (cl_disp_reg_info_t *) malloc(sizeof(cl_disp_reg_info_t));
if (!p_reg) {
cl_spinlock_release(&p_disp->lock);
return (NULL);
} else {
memset(p_reg, 0, sizeof(cl_disp_reg_info_t));
}
p_reg->p_disp = p_disp;
p_reg->ref_cnt = 0;
p_reg->pfn_rcv_callback = pfn_callback;
p_reg->context = context;
p_reg->msg_id = msg_id;
/* Insert the registration in the list. */
cl_qlist_insert_tail(&p_disp->reg_list, (cl_list_item_t *) p_reg);
/* Set the array entry to the registrant. */
/* The ptr_vector grow automatically as necessary. */
if (msg_id != CL_DISP_MSGID_NONE) {
status = cl_ptr_vector_set(&p_disp->reg_vec, msg_id, p_reg);
if (status != CL_SUCCESS) {
free(p_reg);
cl_spinlock_release(&p_disp->lock);
return (NULL);
}
}
cl_spinlock_release(&p_disp->lock);
return (p_reg);
}
void cl_disp_unregister(IN const cl_disp_reg_handle_t handle)
{
cl_disp_reg_info_t *p_reg;
cl_dispatcher_t *p_disp;
if (handle == CL_DISP_INVALID_HANDLE)
return;
p_reg = (cl_disp_reg_info_t *) handle;
p_disp = p_reg->p_disp;
CL_ASSERT(p_disp);
cl_spinlock_acquire(&p_disp->lock);
/*
* Clear the registrant vector entry. This will cause any further
* post calls to fail.
*/
if (p_reg->msg_id != CL_DISP_MSGID_NONE) {
CL_ASSERT(p_reg->msg_id <
cl_ptr_vector_get_size(&p_disp->reg_vec));
cl_ptr_vector_set(&p_disp->reg_vec, p_reg->msg_id, NULL);
}
cl_spinlock_release(&p_disp->lock);
while (p_reg->ref_cnt > 0)
cl_thread_suspend(1);
cl_spinlock_acquire(&p_disp->lock);
/* Remove the registrant from the list. */
cl_qlist_remove_item(&p_disp->reg_list, (cl_list_item_t *) p_reg);
free(p_reg);
cl_spinlock_release(&p_disp->lock);
}
cl_status_t cl_disp_post(IN const cl_disp_reg_handle_t handle,
IN const cl_disp_msgid_t msg_id,
IN const void *const p_data,
IN cl_pfn_msgdone_cb_t pfn_callback OPTIONAL,
IN const void *const context OPTIONAL)
{
cl_disp_reg_info_t *p_src_reg = (cl_disp_reg_info_t *) handle;
cl_disp_reg_info_t *p_dest_reg;
cl_dispatcher_t *p_disp;
cl_disp_msg_t *p_msg;
p_disp = handle->p_disp;
CL_ASSERT(p_disp);
CL_ASSERT(msg_id != CL_DISP_MSGID_NONE);
cl_spinlock_acquire(&p_disp->lock);
/* Check that the recipient exists. */
if (cl_ptr_vector_get_size(&p_disp->reg_vec) <= msg_id) {
cl_spinlock_release(&p_disp->lock);
return (CL_NOT_FOUND);
}
p_dest_reg = cl_ptr_vector_get(&p_disp->reg_vec, msg_id);
if (!p_dest_reg) {
cl_spinlock_release(&p_disp->lock);
return (CL_NOT_FOUND);
}
/* Get a free message from the pool. */
p_msg = (cl_disp_msg_t *) cl_qpool_get(&p_disp->msg_pool);
if (!p_msg) {
cl_spinlock_release(&p_disp->lock);
return (CL_INSUFFICIENT_MEMORY);
}
/* Initialize the message */
p_msg->p_src_reg = p_src_reg;
p_msg->p_dest_reg = p_dest_reg;
p_msg->p_data = p_data;
p_msg->pfn_xmt_callback = pfn_callback;
p_msg->context = context;
p_msg->in_time = cl_get_time_stamp();
/*
* Increment the sender's reference count if they request a completion
* notification.
*/
if (pfn_callback)
cl_atomic_inc(&p_src_reg->ref_cnt);
/* Increment the recipient's reference count. */
cl_atomic_inc(&p_dest_reg->ref_cnt);
/* Queue the message in the FIFO. */
cl_qlist_insert_tail(&p_disp->msg_fifo, (cl_list_item_t *) p_msg);
cl_spinlock_release(&p_disp->lock);
/* Signal the thread pool that there is work to be done. */
cl_thread_pool_signal(&p_disp->worker_threads);
return (CL_SUCCESS);
}
void cl_disp_get_queue_status(IN const cl_disp_reg_handle_t handle,
OUT uint32_t * p_num_queued_msgs,
OUT uint64_t * p_last_msg_queue_time_ms)
{
cl_dispatcher_t *p_disp = ((cl_disp_reg_info_t *) handle)->p_disp;
cl_spinlock_acquire(&p_disp->lock);
if (p_last_msg_queue_time_ms)
*p_last_msg_queue_time_ms =
p_disp->last_msg_queue_time_us / 1000;
if (p_num_queued_msgs)
*p_num_queued_msgs = cl_qlist_count(&p_disp->msg_fifo);
cl_spinlock_release(&p_disp->lock);
}