worker.h File Reference

---

Detailed Description

CDiameterPeer Worker Procedures.

Author:

Dragos Vingarzan vingarzan -at- fokus dot fraunhofer dot de

Definition in file worker.h.

#include "peer.h"
#include "diameter.h"
#include "utils.h"

Go to the source code of this file.

Data Structures
struct	_task_t
	task element More...
struct	task_queue_t
	task queue More...
struct	_cdp_cb_t
	callback element for message processing More...
struct	cdp_cb_list_t
	list of callback elements for message processing More...
Typedefs
typedef int(*)	worker_init_function (int rank)
	function to be called on worker initialization
typedef _task_t	task_t
	task element
typedef int(*)	cdp_cb_f (peer *p, AAAMessage *msg, void *ptr)
	callback function to be called on message processing
typedef _cdp_cb_t	cdp_cb_t
	callback element for message processing
Functions
void	worker_init ()
	Initializes the worker structures, like the task queue.
void	worker_destroy ()
	Destroys the worker structures.
int	cb_add (cdp_cb_f cb, void *ptr)
void	cb_remove (cdp_cb_t *cb)
int	put_task (peer *p, AAAMessage *msg)
	Adds a message as a task to the task queue.
task_t	take_task ()
	Remove and return the first task from the queue (FIFO).
void	worker_poison_queue ()
	Poisons the worker queue.
void	worker_process (int id)
	This is the main worker process.

---

Typedef Documentation

typedef int(*) worker_init_function(int rank)

function to be called on worker initialization

Definition at line 64 of file worker.h.

typedef struct _task_t task_t

task element

typedef int(*) cdp_cb_f(peer *p, AAAMessage *msg, void *ptr)

callback function to be called on message processing

Definition at line 84 of file worker.h.

typedef struct _cdp_cb_t cdp_cb_t

callback element for message processing

---

Function Documentation

void worker_init

(

)

Initializes the worker structures, like the task queue.

Definition at line 126 of file worker.c.

References callbacks, cdp_lock_get(), cdp_semun_init, config, task_queue_t::empty, task_queue_t::end, errno, task_queue_t::full, cdp_cb_list_t::head, task_queue_t::lock, LOG_NO_MEM, task_queue_t::max, task_queue_t::queue, dp_config::queue_length, task_queue_t::start, cdp_cb_list_t::tail, and tasks.

Referenced by diameter_peer_init().

{
    tasks = shm_malloc(sizeof(task_queue_t));
    
    tasks->lock = lock_alloc();
    tasks->lock = lock_init(tasks->lock);
    
    tasks->empty = semget(IPC_PRIVATE,1,0666 | IPC_CREAT );
    if (tasks->empty==-1){
        LOG(L_ERR,"ERROR:worker_init(): Error creating semaphore for empty queue > %s\n",strerror(errno));
    }else
        semctl(tasks->empty, 0, SETVAL, cdp_semun_init );
    tasks->full = semget(IPC_PRIVATE,1, 0666 | IPC_CREAT );
    if (tasks->full==-1){
        LOG(L_ERR,"ERROR:worker_init(): Error creating semaphore for full queue > %s\n",strerror(errno));
    }else
        semctl(tasks->full, 0, SETVAL, cdp_semun_init);
    
    tasks->start = 0;
    tasks->end = 0;
    tasks->max = config->queue_length;
    tasks->queue = shm_malloc(tasks->max*sizeof(task_t));
    if (!tasks->queue) LOG_NO_MEM("shm",tasks->max*sizeof(task_t));
    memset(tasks->queue,0,tasks->max*sizeof(task_t));
        
    callbacks = shm_malloc(sizeof(cdp_cb_list_t));
    callbacks->head = 0; 
    callbacks->tail = 0;
    
    cdp_lock_get(tasks->empty);
//  lock_release(tasks->full);  
}

void worker_destroy

(

)

Destroys the worker structures.

Definition at line 162 of file worker.c.

References AAAFreeMessage(), callbacks, cb_remove(), cdp_semun_destroy, task_queue_t::empty, task_queue_t::full, cdp_cb_list_t::head, task_queue_t::lock, task_queue_t::max, _task_t::msg, task_queue_t::queue, and tasks.

Referenced by diameter_peer_destroy().

{
    int i;
//  LOG(L_CRIT,"-1-\n");
/*  lock_get(tasks->lock);*/
    for(i=0;i<tasks->max;i++){
        if (tasks->queue[i].msg) AAAFreeMessage(&(tasks->queue[i].msg));
    }
//  LOG(L_CRIT,"-2-\n");    
    shm_free(tasks->queue);
    lock_destroy(tasks->lock);
    lock_dealloc((void*)tasks->lock);
//  LOG(L_CRIT,"-3-\n");    
    
    //lock_release(tasks->empty);
    semctl(tasks->empty, 0, IPC_RMID, cdp_semun_destroy);
//  LOG(L_CRIT,"-4-\n");    
    
    semctl(tasks->full, 0, IPC_RMID, cdp_semun_destroy);
    
    shm_free(tasks);
    
    while(callbacks->head)
        cb_remove(callbacks->head);
    shm_free(callbacks);
}

int cb_add	(	cdp_cb_f	cb,
		void *	ptr
	)

Definition at line 190 of file worker.c.

References callbacks, _cdp_cb_t::cb, cdp_cb_list_t::head, LOG_NO_MEM, _cdp_cb_t::next, _cdp_cb_t::prev, _cdp_cb_t::ptr, and cdp_cb_list_t::tail.

Referenced by diameter_peer_init().

{
    cdp_cb_t *x;
    x = shm_malloc(sizeof(cdp_cb_t));
    if (!x){
        LOG_NO_MEM("shm",sizeof(cdp_cb_t));
        return 0;
    }
    x->cb = cb;
    x->ptr = shm_malloc(sizeof(void*));
    if (!x->ptr){
        LOG_NO_MEM("shm",sizeof(void*));
        return 0;
    }
    *(x->ptr) = ptr;
    x->next = 0;
    x->prev = callbacks->tail;
    if (callbacks->tail) callbacks->tail->next = x;
    callbacks->tail = x;
    if (!callbacks->head) callbacks->head = x;
    return 1;   
}

void cb_remove

(

cdp_cb_t *

cb

)

Definition at line 214 of file worker.c.

References callbacks, cdp_cb_list_t::head, _cdp_cb_t::next, _cdp_cb_t::prev, _cdp_cb_t::ptr, and cdp_cb_list_t::tail.

Referenced by worker_destroy().

{
    cdp_cb_t *x;
    x = callbacks->head;
    while(x && x!=cb) x = x->next;
    if (!x) return;
    if (x->prev) x->prev->next = x->next;
    else callbacks->head = x->next;
    if (x->next) x->next->prev = x->prev;
    else callbacks->tail = x->prev;
    
    if (x->ptr) shm_free(x->ptr);
    shm_free(x);
}

int put_task	(	peer *	p,
		AAAMessage *	msg
	)

Adds a message as a task to the task queue.

This blocks if the task queue is full, untill there is space.

Parameters:

	p	- the peer that the message was received from
	msg	- the message

Returns:

1 on success, 0 on failure (eg. shutdown in progress)

Definition at line 236 of file worker.c.

References cdp_lock_get(), cdp_lock_release(), task_queue_t::empty, task_queue_t::end, task_queue_t::full, task_queue_t::lock, task_queue_t::max, _task_t::msg, _task_t::p, task_queue_t::queue, shutdownx, task_queue_t::start, and tasks.

Referenced by Rcv_Process().

{
//  LOG(L_CRIT,"+1+\n");
    lock_get(tasks->lock);
//  LOG(L_CRIT,"+2+\n");
    while ((tasks->end+1)%tasks->max == tasks->start){
//      LOG(L_CRIT,"+3+\n");
        lock_release(tasks->lock);
//      LOG(L_CRIT,"+4+\n");
        if (*shutdownx) {
            cdp_lock_release(tasks->full);
            return 0;
        }
//      LOG(L_ERR,"+");
        cdp_lock_get(tasks->full);
//      LOG(L_CRIT,"+5+\n");
        lock_get(tasks->lock);
    }
    tasks->queue[tasks->end].p = p;
    tasks->queue[tasks->end].msg = msg;
    tasks->end = (tasks->end+1) % tasks->max;
    cdp_lock_release(tasks->empty);
    lock_release(tasks->lock);
    return 1;
}

task_t take_task

(

)

Remove and return the first task from the queue (FIFO).

This blocks until there is something in the queue.

Returns:

the first task from the queue or an empty task on error (eg. shutdown in progress)

Definition at line 267 of file worker.c.

References cdp_lock_get(), cdp_lock_release(), task_queue_t::empty, task_queue_t::end, task_queue_t::full, task_queue_t::lock, task_queue_t::max, _task_t::msg, task_queue_t::queue, shutdownx, task_queue_t::start, and tasks.

Referenced by worker_process().

{
    task_t t={0,0};
//  LOG(L_CRIT,"-1-\n");
    lock_get(tasks->lock);
//  LOG(L_CRIT,"-2-\n");
    while(tasks->start == tasks->end){
//      LOG(L_CRIT,"-3-\n");
        lock_release(tasks->lock);
//      LOG(L_CRIT,"-4-\n");
        if (*shutdownx) {
            cdp_lock_release(tasks->empty);
            return t;
        }
//      LOG(L_ERR,"-");
        cdp_lock_get(tasks->empty);
//      LOG(L_CRIT,"-5-\n");
        
        lock_get(tasks->lock);
//      LOG(L_CRIT,"-6-\n");
    }
//  LOG(L_CRIT,"-7-\n");
    
    t = tasks->queue[tasks->start];
    tasks->queue[tasks->start].msg = 0;
    tasks->start = (tasks->start+1) % tasks->max;
    cdp_lock_release(tasks->full);
    lock_release(tasks->lock);
    return t;
}

void worker_poison_queue

(

)

Poisons the worker queue.

Actually it just releases the task queue locks so that the workers get to evaluate if a shutdown is in process and exit.

Definition at line 303 of file worker.c.

References cdp_lock_release(), task_queue_t::empty, and tasks.

Referenced by diameter_peer_destroy(), and worker_process().

{
//  int i;
//  for(i=0;i<config->workers;i++)
    cdp_lock_release(tasks->empty);
}

void worker_process

(

int

id

)

This is the main worker process.

Takes tasks from the queue in a loop and processes them by calling the registered callbacks.

Parameters:

id

- id of the worker

Returns:

never, exits on shutdown.

Definition at line 316 of file worker.c.

References AAAFreeMessage(), callbacks, _cdp_cb_t::cb, dp_del_pid(), task_queue_t::end, cdp_cb_list_t::head, is_req, memlog, _task_t::msg, _cdp_cb_t::next, _task_t::p, _cdp_cb_t::ptr, shutdownx, task_queue_t::start, take_task(), tasks, and worker_poison_queue().

Referenced by diameter_peer_start().

{
    task_t t;
    cdp_cb_t *cb;
    int r;
    LOG(L_INFO,"INFO:[%d] Worker process started...\n",id); 
    /* init the application level for this child */
    while(1){
        if (shutdownx&&(*shutdownx)) break;
        t = take_task();
        if (!t.msg) {
            if (shutdownx&&(*shutdownx)) break;
            LOG(L_INFO,"INFO:worker_process(): [%d] got empty task Q(%d/%d)\n",id,tasks->start,tasks->end);
            continue;
        }       
        LOG(L_DBG,"DBG:worker_process(): [%d] got task Q(%d/%d)\n",id,tasks->start,tasks->end);
        r = is_req(t.msg);
        for(cb = callbacks->head;cb;cb = cb->next)
            (*(cb->cb))(t.p,t.msg,*(cb->ptr));
        
        if (r){
            AAAFreeMessage(&(t.msg));
        }else{
            /* will be freed by the user in upper api */
            /*AAAFreeMessage(&(t.msg));*/
        }
    }
    worker_poison_queue();
    LOG(L_INFO,"INFO:[%d]... Worker process finished\n",id);    
#ifdef CDP_FOR_SER
#else
    #ifdef PKG_MALLOC
        LOG(memlog, "Worker[%d] Memory status (pkg):\n",id);
        //pkg_status();
        #ifdef pkg_sums
            pkg_sums();
        #endif 
    #endif
    dp_del_pid(getpid());   
#endif
    exit(0);
}

---