contrib/ofed/opensm/complib/cl_dispatcher.c

   1 /*
   2  * Copyright (c) 2004-2006 Voltaire, Inc. All rights reserved.
   3  * Copyright (c) 2002-2005 Mellanox Technologies LTD. All rights reserved.
   4  * Copyright (c) 1996-2003 Intel Corporation. All rights reserved.
   5  *
   6  * This software is available to you under a choice of one of two
   7  * licenses.  You may choose to be licensed under the terms of the GNU
   8  * General Public License (GPL) Version 2, available from the file
   9  * COPYING in the main directory of this source tree, or the
  10  * OpenIB.org BSD license below:
  11  *
  12  *     Redistribution and use in source and binary forms, with or
  13  *     without modification, are permitted provided that the following
  14  *     conditions are met:
  15  *
  16  *      - Redistributions of source code must retain the above
  17  *        copyright notice, this list of conditions and the following
  18  *        disclaimer.
  19  *
  20  *      - Redistributions in binary form must reproduce the above
  21  *        copyright notice, this list of conditions and the following
  22  *        disclaimer in the documentation and/or other materials
  23  *        provided with the distribution.
  24  *
  25  * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
  26  * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
  27  * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
  28  * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS
  29  * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN
  30  * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
  31  * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
  32  * SOFTWARE.
  33  *
  34  */
  35
  36 /*
  37  * Abstract:
  38  *    Implementation of Dispatcher abstraction.
  39  *
  40  */
  41
  42 #if HAVE_CONFIG_H
  43 #  include <config.h>
  44 #endif                          /* HAVE_CONFIG_H */
  45
  46 #include <stdlib.h>
  47 #include <complib/cl_dispatcher.h>
  48 #include <complib/cl_thread.h>
  49 #include <complib/cl_timer.h>
  50
  51 /* give some guidance when we build our cl_pool of messages */
  52 #define CL_DISP_INITIAL_MSG_COUNT   256
  53 #define CL_DISP_MSG_GROW_SIZE       64
  54
  55 /* give some guidance when we build our cl_pool of registration elements */
  56 #define CL_DISP_INITIAL_REG_COUNT   16
  57 #define CL_DISP_REG_GROW_SIZE       16
  58
  59 /********************************************************************
  60    __cl_disp_worker
  61
  62    Description:
  63    This function takes messages off the FIFO and calls Processmsg()
  64    This function executes as passive level.
  65
  66    Inputs:
  67    p_disp - Pointer to Dispatcher object
  68
  69    Outputs:
  70    None
  71
  72    Returns:
  73    None
  74 ********************************************************************/
  75 void __cl_disp_worker(IN void *context)
  76 {
  77         cl_disp_msg_t *p_msg;
  78         cl_dispatcher_t *p_disp = (cl_dispatcher_t *) context;
  79
  80         cl_spinlock_acquire(&p_disp->lock);
  81
  82         /* Process the FIFO until we drain it dry. */
  83         while (cl_qlist_count(&p_disp->msg_fifo)) {
  84                 /* Pop the message at the head from the FIFO. */
  85                 p_msg =
  86                     (cl_disp_msg_t *) cl_qlist_remove_head(&p_disp->msg_fifo);
  87
  88                 /* we track the tim ethe last message spent in the queue */
  89                 p_disp->last_msg_queue_time_us =
  90                     cl_get_time_stamp() - p_msg->in_time;
  91
  92                 /*
  93                  * Release the spinlock while the message is processed.
  94                  * The user's callback may reenter the dispatcher
  95                  * and cause the lock to be reaquired.
  96                  */
  97                 cl_spinlock_release(&p_disp->lock);
  98                 p_msg->p_dest_reg->pfn_rcv_callback((void *)p_msg->p_dest_reg->
  99                                                     context,
 100                                                     (void *)p_msg->p_data);
 101
 102                 cl_atomic_dec(&p_msg->p_dest_reg->ref_cnt);
 103
 104                 /* The client has seen the data.  Notify the sender as appropriate. */
 105                 if (p_msg->pfn_xmt_callback) {
 106                         p_msg->pfn_xmt_callback((void *)p_msg->context,
 107                                                 (void *)p_msg->p_data);
 108                         cl_atomic_dec(&p_msg->p_src_reg->ref_cnt);
 109                 }
 110
 111                 /* Grab the lock for the next iteration through the list. */
 112                 cl_spinlock_acquire(&p_disp->lock);
 113
 114                 /* Return this message to the pool. */
 115                 cl_qpool_put(&p_disp->msg_pool, (cl_pool_item_t *) p_msg);
 116         }
 117
 118         cl_spinlock_release(&p_disp->lock);
 119 }
 120
 121 void cl_disp_construct(IN cl_dispatcher_t * const p_disp)
 122 {
 123         CL_ASSERT(p_disp);
 124
 125         cl_qlist_init(&p_disp->reg_list);
 126         cl_ptr_vector_construct(&p_disp->reg_vec);
 127         cl_qlist_init(&p_disp->msg_fifo);
 128         cl_spinlock_construct(&p_disp->lock);
 129         cl_qpool_construct(&p_disp->msg_pool);
 130 }
 131
 132 void cl_disp_shutdown(IN cl_dispatcher_t * const p_disp)
 133 {
 134         CL_ASSERT(p_disp);
 135
 136         /* Stop the thread pool. */
 137         cl_thread_pool_destroy(&p_disp->worker_threads);
 138
 139         /* Process all outstanding callbacks. */
 140         __cl_disp_worker(p_disp);
 141
 142         /* Free all registration info. */
 143         while (!cl_is_qlist_empty(&p_disp->reg_list))
 144                 free(cl_qlist_remove_head(&p_disp->reg_list));
 145 }
 146
 147 void cl_disp_destroy(IN cl_dispatcher_t * const p_disp)
 148 {
 149         CL_ASSERT(p_disp);
 150
 151         cl_spinlock_destroy(&p_disp->lock);
 152         /* Destroy the message pool */
 153         cl_qpool_destroy(&p_disp->msg_pool);
 154         /* Destroy the pointer vector of registrants. */
 155         cl_ptr_vector_destroy(&p_disp->reg_vec);
 156 }
 157
 158 cl_status_t cl_disp_init(IN cl_dispatcher_t * const p_disp,
 159                          IN const uint32_t thread_count,
 160                          IN const char *const name)
 161 {
 162         cl_status_t status;
 163
 164         CL_ASSERT(p_disp);
 165
 166         cl_disp_construct(p_disp);
 167
 168         status = cl_spinlock_init(&p_disp->lock);
 169         if (status != CL_SUCCESS) {
 170                 cl_disp_destroy(p_disp);
 171                 return (status);
 172         }
 173
 174         /* Specify no upper limit to the number of messages in the pool */
 175         status = cl_qpool_init(&p_disp->msg_pool, CL_DISP_INITIAL_MSG_COUNT,
 176                                0, CL_DISP_MSG_GROW_SIZE, sizeof(cl_disp_msg_t),
 177                                NULL, NULL, NULL);
 178         if (status != CL_SUCCESS) {
 179                 cl_disp_destroy(p_disp);
 180                 return (status);
 181         }
 182
 183         status = cl_ptr_vector_init(&p_disp->reg_vec, CL_DISP_INITIAL_REG_COUNT,
 184                                     CL_DISP_REG_GROW_SIZE);
 185         if (status != CL_SUCCESS) {
 186                 cl_disp_destroy(p_disp);
 187                 return (status);
 188         }
 189
 190         status = cl_thread_pool_init(&p_disp->worker_threads, thread_count,
 191                                      __cl_disp_worker, p_disp, name);
 192         if (status != CL_SUCCESS)
 193                 cl_disp_destroy(p_disp);
 194
 195         return (status);
 196 }
 197
 198 cl_disp_reg_handle_t cl_disp_register(IN cl_dispatcher_t * const p_disp,
 199                                       IN const cl_disp_msgid_t msg_id,
 200                                       IN cl_pfn_msgrcv_cb_t pfn_callback
 201                                       OPTIONAL,
 202                                       IN const void *const context OPTIONAL)
 203 {
 204         cl_disp_reg_info_t *p_reg;
 205         cl_status_t status;
 206
 207         CL_ASSERT(p_disp);
 208
 209         /* Check that the requested registrant ID is available. */
 210         cl_spinlock_acquire(&p_disp->lock);
 211         if ((msg_id != CL_DISP_MSGID_NONE) &&
 212             (msg_id < cl_ptr_vector_get_size(&p_disp->reg_vec)) &&
 213             (cl_ptr_vector_get(&p_disp->reg_vec, msg_id))) {
 214                 cl_spinlock_release(&p_disp->lock);
 215                 return (NULL);
 216         }
 217
 218         /* Get a registration info from the pool. */
 219         p_reg = (cl_disp_reg_info_t *) malloc(sizeof(cl_disp_reg_info_t));
 220         if (!p_reg) {
 221                 cl_spinlock_release(&p_disp->lock);
 222                 return (NULL);
 223         } else {
 224                 memset(p_reg, 0, sizeof(cl_disp_reg_info_t));
 225         }
 226
 227         p_reg->p_disp = p_disp;
 228         p_reg->ref_cnt = 0;
 229         p_reg->pfn_rcv_callback = pfn_callback;
 230         p_reg->context = context;
 231         p_reg->msg_id = msg_id;
 232
 233         /* Insert the registration in the list. */
 234         cl_qlist_insert_tail(&p_disp->reg_list, (cl_list_item_t *) p_reg);
 235
 236         /* Set the array entry to the registrant. */
 237         /* The ptr_vector grow automatically as necessary. */
 238         if (msg_id != CL_DISP_MSGID_NONE) {
 239                 status = cl_ptr_vector_set(&p_disp->reg_vec, msg_id, p_reg);
 240                 if (status != CL_SUCCESS) {
 241                         free(p_reg);
 242                         cl_spinlock_release(&p_disp->lock);
 243                         return (NULL);
 244                 }
 245         }
 246
 247         cl_spinlock_release(&p_disp->lock);
 248
 249         return (p_reg);
 250 }
 251
 252 void cl_disp_unregister(IN const cl_disp_reg_handle_t handle)
 253 {
 254         cl_disp_reg_info_t *p_reg;
 255         cl_dispatcher_t *p_disp;
 256
 257         if (handle == CL_DISP_INVALID_HANDLE)
 258                 return;
 259
 260         p_reg = (cl_disp_reg_info_t *) handle;
 261         p_disp = p_reg->p_disp;
 262         CL_ASSERT(p_disp);
 263
 264         cl_spinlock_acquire(&p_disp->lock);
 265         /*
 266          * Clear the registrant vector entry.  This will cause any further
 267          * post calls to fail.
 268          */
 269         if (p_reg->msg_id != CL_DISP_MSGID_NONE) {
 270                 CL_ASSERT(p_reg->msg_id <
 271                           cl_ptr_vector_get_size(&p_disp->reg_vec));
 272                 cl_ptr_vector_set(&p_disp->reg_vec, p_reg->msg_id, NULL);
 273         }
 274         cl_spinlock_release(&p_disp->lock);
 275
 276         while (p_reg->ref_cnt > 0)
 277                 cl_thread_suspend(1);
 278
 279         cl_spinlock_acquire(&p_disp->lock);
 280         /* Remove the registrant from the list. */
 281         cl_qlist_remove_item(&p_disp->reg_list, (cl_list_item_t *) p_reg);
 282         /* Return the registration info to the pool */
 283         free(p_reg);
 284
 285         cl_spinlock_release(&p_disp->lock);
 286 }
 287
 288 cl_status_t cl_disp_post(IN const cl_disp_reg_handle_t handle,
 289                          IN const cl_disp_msgid_t msg_id,
 290                          IN const void *const p_data,
 291                          IN cl_pfn_msgdone_cb_t pfn_callback OPTIONAL,
 292                          IN const void *const context OPTIONAL)
 293 {
 294         cl_disp_reg_info_t *p_src_reg = (cl_disp_reg_info_t *) handle;
 295         cl_disp_reg_info_t *p_dest_reg;
 296         cl_dispatcher_t *p_disp;
 297         cl_disp_msg_t *p_msg;
 298
 299         p_disp = handle->p_disp;
 300         CL_ASSERT(p_disp);
 301         CL_ASSERT(msg_id != CL_DISP_MSGID_NONE);
 302
 303         cl_spinlock_acquire(&p_disp->lock);
 304         /* Check that the recipient exists. */
 305         if (cl_ptr_vector_get_size(&p_disp->reg_vec) <= msg_id) {
 306                 cl_spinlock_release(&p_disp->lock);
 307                 return (CL_NOT_FOUND);
 308         }
 309
 310         p_dest_reg = cl_ptr_vector_get(&p_disp->reg_vec, msg_id);
 311         if (!p_dest_reg) {
 312                 cl_spinlock_release(&p_disp->lock);
 313                 return (CL_NOT_FOUND);
 314         }
 315
 316         /* Get a free message from the pool. */
 317         p_msg = (cl_disp_msg_t *) cl_qpool_get(&p_disp->msg_pool);
 318         if (!p_msg) {
 319                 cl_spinlock_release(&p_disp->lock);
 320                 return (CL_INSUFFICIENT_MEMORY);
 321         }
 322
 323         /* Initialize the message */
 324         p_msg->p_src_reg = p_src_reg;
 325         p_msg->p_dest_reg = p_dest_reg;
 326         p_msg->p_data = p_data;
 327         p_msg->pfn_xmt_callback = pfn_callback;
 328         p_msg->context = context;
 329         p_msg->in_time = cl_get_time_stamp();
 330
 331         /*
 332          * Increment the sender's reference count if they request a completion
 333          * notification.
 334          */
 335         if (pfn_callback)
 336                 cl_atomic_inc(&p_src_reg->ref_cnt);
 337
 338         /* Increment the recipient's reference count. */
 339         cl_atomic_inc(&p_dest_reg->ref_cnt);
 340
 341         /* Queue the message in the FIFO. */
 342         cl_qlist_insert_tail(&p_disp->msg_fifo, (cl_list_item_t *) p_msg);
 343         cl_spinlock_release(&p_disp->lock);
 344
 345         /* Signal the thread pool that there is work to be done. */
 346         cl_thread_pool_signal(&p_disp->worker_threads);
 347         return (CL_SUCCESS);
 348 }
 349
 350 void cl_disp_get_queue_status(IN const cl_disp_reg_handle_t handle,
 351                               OUT uint32_t * p_num_queued_msgs,
 352                               OUT uint64_t * p_last_msg_queue_time_ms)
 353 {
 354         cl_dispatcher_t *p_disp = ((cl_disp_reg_info_t *) handle)->p_disp;
 355
 356         cl_spinlock_acquire(&p_disp->lock);
 357
 358         if (p_last_msg_queue_time_ms)
 359                 *p_last_msg_queue_time_ms =
 360                     p_disp->last_msg_queue_time_us / 1000;
 361
 362         if (p_num_queued_msgs)
 363                 *p_num_queued_msgs = cl_qlist_count(&p_disp->msg_fifo);
 364
 365         cl_spinlock_release(&p_disp->lock);
 366 }