opensm/complib/cl_dispatcher.c

67    p_disp - Pointer to Dispatcher object
78 	cl_dispatcher_t *p_disp = (cl_dispatcher_t *) context;
80 	cl_spinlock_acquire(&p_disp->lock);
83 	while (cl_qlist_count(&p_disp->msg_fifo)) {
86 		    (cl_disp_msg_t *) cl_qlist_remove_head(&p_disp->msg_fifo);
89 		p_disp->last_msg_queue_time_us =
97 		cl_spinlock_release(&p_disp->lock);
112 		cl_spinlock_acquire(&p_disp->lock);
115 		cl_qpool_put(&p_disp->msg_pool, (cl_pool_item_t *) p_msg);
118 	cl_spinlock_release(&p_disp->lock);
123 void cl_disp_construct(IN cl_dispatcher_t * const p_disp)
125 	CL_ASSERT(p_disp);
127 	cl_qlist_init(&p_disp->reg_list);
128 	cl_ptr_vector_construct(&p_disp->reg_vec);
129 	cl_qlist_init(&p_disp->msg_fifo);
130 	cl_spinlock_construct(&p_disp->lock);
131 	cl_qpool_construct(&p_disp->msg_pool);
136 void cl_disp_shutdown(IN cl_dispatcher_t * const p_disp)
138 	CL_ASSERT(p_disp);
141 	cl_thread_pool_destroy(&p_disp->worker_threads);
144 	__cl_disp_worker(p_disp);
147 	while (!cl_is_qlist_empty(&p_disp->reg_list))
148 		free(cl_qlist_remove_head(&p_disp->reg_list));
153 void cl_disp_destroy(IN cl_dispatcher_t * const p_disp)
155 	CL_ASSERT(p_disp);
157 	cl_spinlock_destroy(&p_disp->lock);
159 	cl_qpool_destroy(&p_disp->msg_pool);
161 	cl_ptr_vector_destroy(&p_disp->reg_vec);
167 cl_disp_init(IN cl_dispatcher_t * const p_disp,
172 	CL_ASSERT(p_disp);
174 	cl_disp_construct(p_disp);
176 	status = cl_spinlock_init(&p_disp->lock);
178 		cl_disp_destroy(p_disp);
183 	status = cl_qpool_init(&p_disp->msg_pool, CL_DISP_INITIAL_MSG_COUNT,
187 		cl_disp_destroy(p_disp);
191 	status = cl_ptr_vector_init(&p_disp->reg_vec, CL_DISP_INITIAL_REG_COUNT,
194 		cl_disp_destroy(p_disp);
198 	status = cl_thread_pool_init(&p_disp->worker_threads, thread_count,
199 				     __cl_disp_worker, p_disp, name);
201 		cl_disp_destroy(p_disp);
209 cl_disp_register(IN cl_dispatcher_t * const p_disp,
217 	CL_ASSERT(p_disp);
220 	cl_spinlock_acquire(&p_disp->lock);
222 	    (msg_id < cl_ptr_vector_get_size(&p_disp->reg_vec)) &&
223 	    (cl_ptr_vector_get(&p_disp->reg_vec, msg_id))) {
224 		cl_spinlock_release(&p_disp->lock);
231 		cl_spinlock_release(&p_disp->lock);
237 	p_reg->p_disp = p_disp;
244 	cl_qlist_insert_tail(&p_disp->reg_list, (cl_list_item_t *) p_reg);
249 		status = cl_ptr_vector_set(&p_disp->reg_vec, msg_id, p_reg);
252 			cl_spinlock_release(&p_disp->lock);
257 	cl_spinlock_release(&p_disp->lock);
267 	cl_dispatcher_t *p_disp;
273 	p_disp = p_reg->p_disp;
274 	CL_ASSERT(p_disp);
276 	cl_spinlock_acquire(&p_disp->lock);
283 			  cl_ptr_vector_get_size(&p_disp->reg_vec));
284 		cl_ptr_vector_set(&p_disp->reg_vec, p_reg->msg_id, NULL);
286 	cl_spinlock_release(&p_disp->lock);
291 	cl_spinlock_acquire(&p_disp->lock);
293 	cl_qlist_remove_item(&p_disp->reg_list, (cl_list_item_t *) p_reg);
297 	cl_spinlock_release(&p_disp->lock);
311 	cl_dispatcher_t *p_disp;
314 	p_disp = handle->p_disp;
315 	CL_ASSERT(p_disp);
318 	cl_spinlock_acquire(&p_disp->lock);
320 	p_dest_reg = cl_ptr_vector_get(&p_disp->reg_vec, msg_id);
322 		cl_spinlock_release(&p_disp->lock);
327 	p_msg = (cl_disp_msg_t *) cl_qpool_get(&p_disp->msg_pool);
329 		cl_spinlock_release(&p_disp->lock);
352 	cl_qlist_insert_tail(&p_disp->msg_fifo, (cl_list_item_t *) p_msg);
353 	cl_spinlock_release(&p_disp->lock);
356 	cl_thread_pool_signal(&p_disp->worker_threads);
365 	cl_dispatcher_t *p_disp = ((cl_disp_reg_info_t *) handle)->p_disp;
367 	cl_spinlock_acquire(&p_disp->lock);
371 		    p_disp->last_msg_queue_time_us / 1000;
374 		*p_num_queued_msgs = cl_qlist_count(&p_disp->msg_fifo);
376 	cl_spinlock_release(&p_disp->lock);