Loading...
Searching...
No Matches
It_q

Inter-Thread Message Queue. More...

## Functions

static int | eventfd_increment (int fd, uint64_t inc)
static | LLIST_HEAD (it_queues)
static struct osmo_it_q * | _osmo_it_q_by_name (const char *name)
struct osmo_it_q * | osmo_it_q_by_name (const char *name)
| resolve it-queue by its [globally unique] name
static int | osmo_it_q_fd_cb (struct osmo_fd *ofd, unsigned int what)
struct osmo_it_q * | osmo_it_q_alloc (void *ctx, const char *name, unsigned int max_length, void(*read_cb)(struct osmo_it_q *q, struct llist_head *item), void *data)
| Allocate a new inter-thread message queue.
static void * | item_dequeue (struct llist_head *queue)
static void | _osmo_it_q_flush (struct osmo_it_q *q)
| Flush all messages currently present in queue.
void | osmo_it_q_flush (struct osmo_it_q *q)
| Flush all messages currently present in queue.
void | osmo_it_q_destroy (struct osmo_it_q *q)
| Destroy a message queue.
int | _osmo_it_q_enqueue (struct osmo_it_q *queue, struct llist_head *item)
| Thread-safe en-queue to an inter-thread message queue.
struct llist_head * | _osmo_it_q_dequeue (struct osmo_it_q *queue)
| Thread-safe de-queue from an inter-thread message queue.

## Variables

static pthread_rwlock_t | it_queues_rwlock = PTHREAD_RWLOCK_INITIALIZER

## Detailed Description

Inter-Thread Message Queue.

This implements a general-purpose queue between threads. It uses user-provided data types (containing a llist_head as initial member) as elements in the queue and an eventfd-based notification mechanism. Hence, it can be used for pretty much anything, including but not limited to msgbs, including msgb-wrapped osmo_prim.

The idea is that the sending thread simply calls osmo_it_q_enqueue(). The receiving thread is woken up from its osmo_select_main() loop by eventfd, and a general osmo_fd callback function for the eventfd will dequeue each item and call a queue-specific callback function.

## Function Documentation

## _osmo_it_q_by_name()

static struct osmo_it_q * _osmo_it_q_by_name | ( | const char * | name | ) |

static

References osmo_it_q::entry, llist_for_each_entry, name, and osmo_it_q::name.

Referenced by osmo_it_q_alloc(), and osmo_it_q_by_name().

## _osmo_it_q_dequeue()

struct llist_head * _osmo_it_q_dequeue | ( | struct osmo_it_q * | queue | ) |

Thread-safe de-queue from an inter-thread message queue.

Parameters

  [in] | queue | Inter-thread queue from which to dequeue

Returns
  dequeued message buffer; NULL if none available

References osmo_it_q::current_length, l, osmo_it_q::list, llist_del(), llist_empty(), osmo_it_q::mutex, llist_head::next, and OSMO_ASSERT.

Referenced by osmo_it_q_fd_cb().

## _osmo_it_q_enqueue()

int _osmo_it_q_enqueue | ( | struct osmo_it_q * | queue,
|  | struct llist_head * | item
| ) |  |

Thread-safe en-queue to an inter-thread message queue.

Parameters

  [in] | queue | Inter-thread queue on which to enqueue
  [in] | item | Item to enqueue. Must have llist_head as first member!

Returns
  0 on success; negative on error

References osmo_it_q::current_length, osmo_it_q::event_ofd, eventfd_increment(), osmo_fd::fd, osmo_it_q::list, llist_add_tail(), osmo_it_q::max_length, osmo_it_q::mutex, and OSMO_ASSERT.

## _osmo_it_q_flush()

static void _osmo_it_q_flush | ( | struct osmo_it_q * | q | ) |

static

Flush all messages currently present in queue.

References osmo_it_q::current_length, item_dequeue(), and osmo_it_q::list.

Referenced by osmo_it_q_flush().

## eventfd_increment()

static int eventfd_increment | ( | int | fd,
|  | uint64_t | inc
| ) |  |

static

Referenced by _osmo_it_q_enqueue().

## item_dequeue()

static void * item_dequeue | ( | struct llist_head * | queue | ) |

static

References llist_del(), llist_empty(), and llist_head::next.

Referenced by _osmo_it_q_flush().

## LLIST_HEAD()

static LLIST_HEAD | ( | it_queues |  | ) |

static

## osmo_it_q_alloc()

struct osmo_it_q * osmo_it_q_alloc | ( | void * | ctx,
|  | const char * | name,
|  | unsigned int | max_length,
|  | void(*)(struct osmo_it_q *q, struct llist_head *item) | read_cb,
|  | void * | data
| ) |  |

Allocate a new inter-thread message queue.

Parameters

  [in] | ctx | talloc context from which to allocate the queue
  [in] | name | human-readable string name of the queue; function creates a copy.
  [in] | read_cb | call-back function to be called for each de-queued message; may be NULL in case you don't want eventfd/osmo_select integration and will manually take care of noticing if and when to dequeue.

Returns
  a newly-allocated inter-thread message queue; NULL in case of error

References _osmo_it_q_by_name(), osmo_it_q::current_length, osmo_it_q::data, osmo_it_q::entry, osmo_it_q::event_ofd, osmo_fd::fd, INIT_LLIST_HEAD, it_queues_rwlock, osmo_it_q::list, llist_add_tail(), osmo_it_q::max_length, osmo_it_q::mutex, name, osmo_it_q::name, osmo_fd_close(), OSMO_FD_READ, osmo_fd_setup(), osmo_it_q_fd_cb(), and osmo_it_q::read_cb.

## osmo_it_q_by_name()

struct osmo_it_q * osmo_it_q_by_name | ( | const char * | name | ) |

resolve it-queue by its [globally unique] name

References _osmo_it_q_by_name(), it_queues_rwlock, and name.

## osmo_it_q_destroy()

void osmo_it_q_destroy | ( | struct osmo_it_q * | q | ) |

Destroy a message queue.

References osmo_it_q::entry, osmo_it_q::event_ofd, osmo_fd::fd, it_queues_rwlock, llist_del(), osmo_it_q::mutex, OSMO_ASSERT, osmo_fd_close(), and osmo_it_q_flush().

## osmo_it_q_fd_cb()

static int osmo_it_q_fd_cb | ( | struct osmo_fd * | ofd,
|  | unsigned int | what
| ) |  |

static

References _osmo_it_q_dequeue(), osmo_fd::data, osmo_fd::fd, OSMO_FD_READ, and osmo_it_q::read_cb.

Referenced by osmo_it_q_alloc().

## osmo_it_q_flush()

void osmo_it_q_flush | ( | struct osmo_it_q * | q | ) |

Flush all messages currently present in queue.

References _osmo_it_q_flush(), osmo_it_q::mutex, and OSMO_ASSERT.

Referenced by osmo_it_q_destroy().

## Variable Documentation

## it_queues_rwlock

pthread_rwlock_t it_queues_rwlock = PTHREAD_RWLOCK_INITIALIZER

static

Referenced by osmo_it_q_alloc(), osmo_it_q_by_name(), and osmo_it_q_destroy().