libucore/include/ucore/threadqueue.h

#ifndef UC_THREADQUEUE_H_
#define UC_THREADQUEUE_H_ 

#include <pthread.h>

#ifdef __cplusplus
extern "C" {
#endif
/**
 * @defgroup ThreadQueue ThreadQueue
 * @{
 *
 * Little API for waitable queues, typically used for passing messages
 * between threads.
 * 
 * @author Nils O. Selåsdal <NOS@Utel.no>
 */

 
/**
 * A thread message used to add and retrieve messages
 * from in a queue.
 * An application must not touch this struct 
 * when a message is present (i.e. it has been added
 * to a queue, but not retrieved yet) in a queue.
 *
 */
struct uc_threadmsg{
    /**
     * A message type the application can use to 
     * discriminate different messages.
     * A negative value will cause a uc_threadmsg to
     * be added to the head of a queue.
     */
    long msgtype;

    /**
     * Internal pointer used by the uc_thread_queue.
     * Applications must not use this member.
     */
    struct uc_threadmsg *next;
};

/**
 * Callback function for the walk and destroy functions.
 */
typedef void (*uc_thread_queue_walk_func) (struct uc_threadmsg *msg, void *cookie);


/**
 * A ThreadQueue
 * 
 * 
 * You should threat this struct as opaque never ever access any of 
 * the variables in this struct.
 * You have been warned.
 *
 * The uc_thread_queue_ functions only deal with struct uc_threadmsg
 * structs.
 * In order for the message passing to be useful, more data needs to be
 * associated with a message, it's up to the application to manage this.
 * One way is to e.g. add the struct uc_threadmsg as the first member
 * of a larger struct, and recover the larger struct after getting a 
 * struct uc_threadmsg out of the queue. e.g.
 *
 * @code
 * struct my_msg {
 *   struct uc_threadmsg tmsg;
 *   int foo;
 *   char bar[32];
 * };
 *
 * struct my_msg *msg = malloc(sizeof *msg);
 * msg->tmsg.msgtype = MSGTYP1;
 * ...
 * uc_thread_queue_add(queue, &msg->tmsg);
 *
 * The receiver end does e.g.
 *
 * struct uc_threadmsg *tmsg;
 * uc_thread_queue_get(queue, NULL, &tmsg);
 * switch(tmsg->msgtype) {
 *   case MYMSG1; {
 *      struct my_msg *msg = (struct my_msg*)tmsg;
 *      ...
 *      free(msg);
 *      break
 *   ...
 *   }
 * }
 *
 * @endcode
 *
 *
 *
 *
 */
struct uc_threadqueue {
/**
 * Number of elements in the queue.
 * Use #threadqueue_length to read it.
 */
    unsigned int num_elements;

/** Max number of elements this queue will hold */
    unsigned int max_elements;
/**
 * Mutex for the queue.
 */
    pthread_mutex_t mutex;
/**
 * Condition variable for readers on the queue.
 */
    pthread_cond_t read_cond;
/**
 * Condition variable for writers on the queue (if the queue is full).
 */
    pthread_cond_t write_cond;
/** 
 * Number of threads blocking on writing to the queue
 */
    unsigned short num_read_waiters;
/** 
 * Number of threads blocking on reading from  the queue
 */
    unsigned short num_write_waiters;

/**
 * Internal pointers for the messages in the queue.
 */
    struct uc_threadmsg *first,**last;
};

/**
 * Initializes a queue.
 * 
 * thread_queue_init initializes a new threadqueue. A new queue must always
 * be initialized before it is used.
 * A max number of elements the queue will hold must be given.
 * Adding more elements than a queue will hold will e.g. cause
 * uc_thread_queue_add to block until space becomes available.
 *
 * @param queue Pointer to the queue that should be initialized
 * @param max_elements Max number of elements this queue can hold.
 *
 * @return 0 on success EINVAL if queue is NULL or max_elements <= 0, or
 * another errno value if pthread_ functions fails
 */
int uc_thread_queue_init(struct uc_threadqueue *queue, unsigned int max_elements);

/**
 * Adds a message to a queue
 *
 * thread_queue_add adds a "message" to the specified queue. 
 * It is up to the application to manage the data and memory indicated by the
 * struct uc_threadmsg. 
 * The struct uc_threadmsg is assumed to be an intrusive pointer,
 * e.g. it can be the first member of a larger struct actually containing the
 * data to be passed over.
 *
 * Nothing is copied so the application must keep track on (de)allocation of the pointers.
 * A message type can also be also specified, to e.g. help discriminate
 * messages when messages are received.
 *
 * If the message_type member of @msg is negative, the message is added to
 * the front of the queue, i.e. it can be considered a "priority" message.
 *
 * @param queue Pointer to the queue on where the message should be added.
 * @param timeout timeout on how long to wait on a message, or NULL for no timeout
 * @param data the "message".
 * @return 0 on succes ENOMEM if out of memory EINVAL if queue is NULL, ETIMEDOUT if timeout occured or other
 * errno values if pthread functions failed.
 */
int uc_thread_queue_tryadd(struct uc_threadqueue *queue, const struct timespec *timeout, struct uc_threadmsg *msg);

/** 
 * Like uc_thread_queue_add but will wait indefintly.
 * @see uc_thread_queue_add
 */
static inline int uc_thread_queue_add(struct uc_threadqueue *queue, struct uc_threadmsg *msg)
{
    return uc_thread_queue_tryadd(queue, NULL, msg);
}

/**
 * Gets a message from a queue
 * 
 * thread_queue_tryget gets a message from the specified queue, it will block
 * the caling thread untill a message arrives, or the (optional) timeout occurs.
 * If timeout is NULL, there will be no timeout, and thread_queue_get will wait
 * untill a message arrives.
 *
 * struct timespec is defined as:
 * @code
 *  struct timespec {
 *                 long    tv_sec;         // seconds 
 *                 long    tv_nsec;        // nanoseconds 
 *             };
 * @endcode
 * 
 * @param queue Pointer to the queue to wait on for a message.
 * @param timeout timeout on how long to wait on a message, or NULL for no timeout
 * @param msg pointer where the uc_threadmsg* is stored
 *
 * @return 0 on success EINVAL if queue is NULL ETIMEDOUT if timeout occurs
 */
int uc_thread_queue_tryget(struct uc_threadqueue *queue, const struct timespec *timeout, struct uc_threadmsg **msg);

/**
 * Like uc_thread_queue_tryget, but will wait indefintly for an item
 * to become available
 * @see uc_thread_queue_tryget
 */
static inline int uc_thread_queue_get(struct uc_threadqueue *queue, struct uc_threadmsg **msg)
{
    return uc_thread_queue_tryget(queue, NULL, msg);
}


/**
 * Gets the length of a queue
 * 
 * threadqueue_length returns the number of messages waiting in the queue
 * 
 * @param queue Pointer to the queue for which to get the length
 * @return the length(number of pending messages) in the queue
 */
unsigned int uc_thread_queue_length( struct uc_threadqueue *queue );

/**
 * Destroy the queue.
 *
 * If free_func is != NULL, free_func will be called for every item, allowing you to free
 * the item.
 * You cannot call this if there are someone currently adding or getting messages
 * from the queue.
 * After a queue have been cleaned, it cannot be used again untill #thread_queue_init
 * has been called on the queue.
 * 
 * @param queue Pointer to the queue that should be cleaned
 * @param free_func pointer to function that will be called for each item. 
 *  The function must not in anyway interact with the queue.
 * @param cookie pointer passed back to the uc_thread_queue_walk_func
 * @return 0 on success EINVAL if queue is NULL EBUSY if someone is holding any locks on the queue
 */
int uc_thread_queue_destroy(struct uc_threadqueue *queue, uc_thread_queue_walk_func free_func, void *cookie);

/**
 * Walk the elements of the queue, intended for debugging
 * 
 * The supplied function will be called for each item in the queue, internal queue
 * locks are held while the function is called, so the supplied function must not 
 * interact with the queue, else deadlock occurs.
 * 
 * @param queue Pointer to the queue to walk
 * @param free_func pointer to function that will be called for each item. 
 * @param cookie pointer passed back to the uc_thread_queue_walk_func
 *  The function must not in anyway interact with the queue.
 * @return 0 on success 
 */
int uc_thread_queue_walk(struct uc_threadqueue *queue, uc_thread_queue_walk_func walk_func, void *cookie);


/**
 * Clear the queue.
 * The free_func unless NULL, will be called for each element 
 * locks are held while the function is called, so the supplied function must not 
 * interact with the queue, else deadlock occurs.
 *
 * @param queue Pointer to the queue to clear
 * @param free_func pointer to function that will be called for each item, intended to e.g. release memory
 * @param cookie pointer passed back to the uc_thread_queue_walk_func
 * @return 0 on success 
*/
int uc_thread_queue_clear(struct uc_threadqueue *queue, uc_thread_queue_walk_func free_func, void *cookie);


/**
 * @}*/

#ifdef __cplusplus
}
#endif

#endif