dtstream.h File Reference
#include "util/locks.h"

Data Structures

struct  dt_msg_queue
 A message buffer with dnstap messages queued up. More...
 
struct  dt_msg_entry
 An entry in the dt_msg_queue. More...
 
struct  dt_frame_read_buf
 Containing buffer and counter for reading DNSTAP frames. More...
 
struct  dt_io_thread
 IO thread that reads from the queues and writes them. More...
 
struct  dt_io_list_item
 IO thread list of queues list item lists a worker queue that should be looked at and sent to the log server. More...
 

Functions

struct dt_msg_queuedt_msg_queue_create (struct comm_base *base)
 Create new (empty) worker message queue. More...
 
void dt_msg_queue_delete (struct dt_msg_queue *mq)
 Delete a worker message queue. More...
 
void dt_msg_queue_submit (struct dt_msg_queue *mq, void *buf, size_t len)
 Submit a message to the queue. More...
 
void mq_wakeup_cb (void *arg)
 timer callback to wakeup dtio thread to process messages
 
struct dt_io_threaddt_io_thread_create (void)
 Create IO thread. More...
 
void dt_io_thread_delete (struct dt_io_thread *dtio)
 Delete the IO thread structure. More...
 
int dt_io_thread_apply_cfg (struct dt_io_thread *dtio, struct config_file *cfg)
 Apply config to the dtio thread. More...
 
int dt_io_thread_register_queue (struct dt_io_thread *dtio, struct dt_msg_queue *mq)
 Register a msg queue to the io thread. More...
 
void dt_io_thread_unregister_queue (struct dt_io_thread *dtio, struct dt_msg_queue *mq)
 Unregister queue from io thread. More...
 
int dt_io_thread_start (struct dt_io_thread *dtio, void *event_base_nothr, int numworkers)
 Start the io thread. More...
 
void dt_io_thread_stop (struct dt_io_thread *dtio)
 Stop the io thread. More...
 
void dtio_reconnect_timeout_cb (int fd, short bits, void *arg)
 callback for the dnstap reconnect, to start reconnecting to output
 
void dtio_output_cb (int fd, short bits, void *arg)
 callback for the dnstap events, to write to the output
 
void dtio_cmd_cb (int fd, short bits, void *arg)
 callback for the dnstap commandpipe, to stop the dnstap IO
 
void dtio_stop_timer_cb (int fd, short bits, void *arg)
 callback for the timer when the thread stops and wants to finish up
 
void dtio_stop_ev_cb (int fd, short bits, void *arg)
 callback for the output when the thread stops and wants to finish up
 
void dtio_tap_callback (int fd, short bits, void *arg)
 callback for unbound-dnstap-socket
 
void dtio_mainfdcallback (int fd, short bits, void *arg)
 callback for unbound-dnstap-socket
 

Detailed Description

An implementation of the Frame Streams data transport protocol for the Unbound DNSTAP message logging facility.

Function Documentation

◆ dt_msg_queue_create()

struct dt_msg_queue* dt_msg_queue_create ( struct comm_base base)

Create new (empty) worker message queue.

Limit set to default on max.

Parameters
baseevent base for wakeup timer.
Returns
NULL on malloc failure or a new queue (not locked).

References comm_timer_create(), dt_msg_queue::lock, dt_msg_queue::maxsize, mq_wakeup_cb(), and dt_msg_queue::wakeup_timer.

◆ dt_msg_queue_delete()

void dt_msg_queue_delete ( struct dt_msg_queue mq)

Delete a worker message queue.

It has to be unlinked from access, so it can be deleted without lock worries. The queue is emptied (deleted).

Parameters
mqmessage queue.

References comm_timer_delete(), dt_msg_queue_clear(), dt_msg_queue::lock, and dt_msg_queue::wakeup_timer.

◆ dt_msg_queue_submit()

void dt_msg_queue_submit ( struct dt_msg_queue mq,
void *  buf,
size_t  len 
)

Submit a message to the queue.

The queue is locked by the routine, the message is inserted, and then the queue is unlocked so the message can be picked up by the writer thread.

Parameters
mqmessage queue.
bufbuffer with message (dnstap contents). The buffer must have been malloced by caller. It is linked in the queue, and is free()d after use. If the routine fails the buffer is freed as well (and nothing happens, the item could not be logged).
lenlength of buffer.

References dt_msg_entry::buf, dt_msg_queue::cursize, dt_msg_queue_start_timer(), DTIO_MSG_FOR_WAKEUP, dt_msg_queue::first, dt_msg_entry::len, dt_msg_queue::lock, log_err(), dt_msg_queue::maxsize, dt_msg_queue::msgcount, dt_msg_entry::next, and entry::next.

◆ dt_io_thread_create()

struct dt_io_thread* dt_io_thread_create ( void  )

Create IO thread.

Returns
new io thread object. not yet started. or NULL malloc failure.

References dt_io_thread::wakeup_timer_enabled, and dt_io_thread::wakeup_timer_lock.

◆ dt_io_thread_delete()

void dt_io_thread_delete ( struct dt_io_thread dtio)

◆ dt_io_thread_apply_cfg()

◆ dt_io_thread_register_queue()

int dt_io_thread_register_queue ( struct dt_io_thread dtio,
struct dt_msg_queue mq 
)

Register a msg queue to the io thread.

It will be polled to see if there are messages and those then get removed and sent, when the thread is running.

Parameters
dtiothe io thread.
mqmessage queue to register.
Returns
false on failure (malloc failure).

References dt_msg_queue::dtio, dt_io_thread::io_list, dt_io_thread::io_list_iter, dt_msg_queue::lock, dt_io_list_item::next, and dt_io_list_item::queue.

◆ dt_io_thread_unregister_queue()

void dt_io_thread_unregister_queue ( struct dt_io_thread dtio,
struct dt_msg_queue mq 
)

Unregister queue from io thread.

Parameters
dtiothe io thread.
mqmessage queue.

References dt_msg_queue::dtio, dt_io_thread::io_list, dt_io_thread::io_list_iter, dt_msg_queue::lock, dt_io_list_item::next, and dt_io_list_item::queue.

◆ dt_io_thread_start()

int dt_io_thread_start ( struct dt_io_thread dtio,
void *  event_base_nothr,
int  numworkers 
)

Start the io thread.

Parameters
dtiothe io thread.
event_base_nothrthe event base to attach the events to, in case we are running without threads. With threads, this is ignored and a thread is started to process the dnstap log messages.
numworkersnumber of worker threads. The dnstap io thread is that number +1 as the threadnumber (in logs).
Returns
false on failure.

References dt_io_thread::commandpipe, dnstap_io(), dtio_setup_on_base(), dt_io_thread::event_base, log_err(), dt_io_thread::started, dt_io_thread::threadnum, and dt_io_thread::tid.

◆ dt_io_thread_stop()

void dt_io_thread_stop ( struct dt_io_thread dtio)