worker.c File Reference

---

Detailed Description

CDiameterPeer Worker Procedures.

This the process pool representation that is used for processing incoming messages.

Author:

Dragos Vingarzan vingarzan -at- fokus dot fraunhofer dot de

Definition in file worker.c.

#include <time.h>
#include <stdlib.h>
#include <sys/types.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/sem.h>
#include "utils.h"
#include "globals.h"
#include "config.h"
#include "worker.h"
#include "diameter_api.h"

Go to the source code of this file.

Data Structures
union	semun
Functions
int	dp_add_pid (pid_t pid)
	Add a pid to the local process list.
void	dp_del_pid (pid_t pid)
	Delete a pid from the process list.
static void	cdp_lock_get (int sid)
	Gets the lock on a semaphore and blocks until it is available.
static void	cdp_lock_release (int sid)
	Releases the lock on a sempahore.
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.
Variables
dp_config *	config
	Configuration for this diameter peer.
task_queue_t *	tasks
	queue of tasks
cdp_cb_list_t *	callbacks
	list of callbacks for message processing
sembuf	cdp_sem_lock = { 0, -1, 0}
	sembuf structure to lock a semaphore
sembuf	cdp_sem_unlock = { 0, +1, 0}
	sembuf structure to unlock a semaphore
semun	cdp_semun_lock
semun	cdp_semun_unlock
semun	cdp_semun_init
semun	cdp_semun_destroy

---

Function Documentation

int dp_add_pid

(

pid_t

pid

)

[inline]

Add a pid to the local process list.

Parameters:

pid

newly forked pid

Returns:

1 on success or 0 on error

Definition at line 95 of file diameter_peer.c.

References pid_list_head_t::head, LOG_NO_MEM, _pid_list_t::next, _pid_list_t::pid, pid_list, pid_list_lock, _pid_list_t::prev, and pid_list_head_t::tail.

{
    pid_list_t *n;
    lock_get(pid_list_lock);
    n = shm_malloc(sizeof(pid_list_t));
    if (!n){
        LOG_NO_MEM("shm",sizeof(pid_list_t));
        lock_release(pid_list_lock);
        return 0;
    }
    n->pid = pid;
    n->next = 0;
    n->prev = pid_list->tail;
    if (!pid_list->head) pid_list->head = n;
    if (pid_list->tail) pid_list->tail->next = n;
    pid_list->tail = n;
    lock_release(pid_list_lock);
    return 1;
}

void dp_del_pid

(

pid_t

pid

)

[inline]

Delete a pid from the process list.

Parameters:

pid

- the pid to remove

Definition at line 132 of file diameter_peer.c.

References pid_list_head_t::head, _pid_list_t::next, _pid_list_t::pid, pid_list, pid_list_lock, _pid_list_t::prev, and pid_list_head_t::tail.

{   
    pid_list_t *i;
    lock_get(pid_list_lock);
    i = pid_list->head;
    if (!i) {
        lock_release(pid_list_lock);
        return;
    }
    while(i && i->pid!=pid) i = i->next;
    if (i){
        if (i->prev) i->prev->next = i->next;
        else pid_list->head = i->next;
        if (i->next) i->next->prev = i->prev;
        else pid_list->tail = i->prev;
        shm_free(i);
    }
    lock_release(pid_list_lock);
}

static void cdp_lock_get

(

int

sid

)

[inline, static]

Gets the lock on a semaphore and blocks until it is available.

This procedures does not consume CPU cycles as a busy-waiting would and it is used for blocking on the task queue without a big impact on performance.

Parameters:

sid

- semaphore id

Returns:

when the sempahore is aquired or shutdown

Definition at line 98 of file worker.c.

References cdp_sem_lock, errno, and shutdownx.

Referenced by put_task(), take_task(), and worker_init().

{
    if((semop(sid, &cdp_sem_lock, 1)) == -1)
    {
        if (shutdownx&&(*shutdownx)) return;
        LOG(L_INFO,"ERROR:cdp_lock_get(): Error on semop > %s\n",strerror(errno));
    }
}

static void cdp_lock_release

(

int

sid

)

[inline, static]

Releases the lock on a sempahore.

Parameters:

sid

- the semaphore id

Definition at line 111 of file worker.c.

References cdp_semun_unlock, task_queue_t::end, errno, task_queue_t::full, shutdownx, task_queue_t::start, and tasks.

Referenced by put_task(), take_task(), and worker_poison_queue().

{
    if( semctl(sid, 0, SETVAL, cdp_semun_unlock) == -1
    /*semop(sid, &cdp_sem_unlock, 1) == -1*/)
    {
        if (shutdownx&&(*shutdownx)) return;
        LOG(L_INFO,"ERROR:cdp_lock_release(): Error on semop %s > %d: %s Q[%2d/%2d]\n",
            sid==tasks->full?"full":"empty",errno,strerror(errno),
            tasks->start,tasks->end);
    }
}

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);
}

---

Variable Documentation

dp_config* config

Configuration for this diameter peer.

Definition at line 76 of file diameter_peer.c.

task_queue_t* tasks

queue of tasks

Definition at line 76 of file worker.c.

Referenced by cdp_lock_release(), put_task(), take_task(), worker_destroy(), worker_init(), worker_poison_queue(), and worker_process().

cdp_cb_list_t* callbacks

list of callbacks for message processing

Definition at line 78 of file worker.c.

Referenced by cb_add(), cb_remove(), worker_destroy(), worker_init(), and worker_process().

struct sembuf cdp_sem_lock = { 0, -1, 0}

sembuf structure to lock a semaphore

Definition at line 80 of file worker.c.

Referenced by cdp_lock_get().

struct sembuf cdp_sem_unlock = { 0, +1, 0}

sembuf structure to unlock a semaphore

Definition at line 81 of file worker.c.

union semun cdp_semun_lock

union semun cdp_semun_unlock

Referenced by cdp_lock_release().

union semun cdp_semun_init

Referenced by worker_init().

union semun cdp_semun_destroy

Referenced by worker_destroy().

---