further work done for simultaneous rekeying/delete
still some cases which cause trouble
This commit is contained in:
@@ -29,6 +29,7 @@
|
||||
#include <sys/socket.h>
|
||||
#include <linux/netlink.h>
|
||||
#include <linux/rtnetlink.h>
|
||||
#include <linux/udp.h>
|
||||
#include <pthread.h>
|
||||
#include <unistd.h>
|
||||
#include <fcntl.h>
|
||||
@@ -383,10 +384,7 @@ static status_t add_sa(private_kernel_interface_t *this,
|
||||
}
|
||||
|
||||
struct xfrm_encap_tmpl* encap = (struct xfrm_encap_tmpl*)RTA_DATA(rthdr);
|
||||
/* UDP_ENCAP_ESPINUDP, see /usr/src/linux/include/linux/udp.h
|
||||
* we could probably use 3 here (as pluto does) although the
|
||||
* result is eventually the same. */
|
||||
encap->encap_type = 2;
|
||||
encap->encap_type = UDP_ENCAP_ESPINUDP;
|
||||
encap->encap_sport = ntohs(natt->sport);
|
||||
encap->encap_dport = ntohs(natt->dport);
|
||||
memset(&encap->encap_oa, 0, sizeof (xfrm_address_t));
|
||||
@@ -440,9 +438,9 @@ static status_t update_sa_hosts(
|
||||
this->logger->log(this->logger, CONTROL|LEVEL2, "getting SA");
|
||||
|
||||
struct nlmsghdr *hdr = (struct nlmsghdr*)request;
|
||||
hdr->nlmsg_flags = NLM_F_REQUEST ;
|
||||
hdr->nlmsg_flags = NLM_F_REQUEST;
|
||||
hdr->nlmsg_type = XFRM_MSG_GETSA;
|
||||
hdr->nlmsg_len = NLMSG_LENGTH(sizeof(struct xfrm_usersa_info));
|
||||
hdr->nlmsg_len = NLMSG_LENGTH(sizeof(struct xfrm_usersa_id));
|
||||
|
||||
struct xfrm_usersa_id *sa_id = (struct xfrm_usersa_id*)NLMSG_DATA(hdr);
|
||||
sa_id->daddr = dst->get_xfrm_addr(dst);
|
||||
@@ -996,12 +994,11 @@ kernel_interface_t *kernel_interface_create()
|
||||
/* public functions */
|
||||
this->public.get_spi = (status_t(*)(kernel_interface_t*,host_t*,host_t*,protocol_id_t,u_int32_t,u_int32_t*))get_spi;
|
||||
this->public.add_sa = (status_t(*)(kernel_interface_t *,host_t*,host_t*,u_int32_t,protocol_id_t,u_int32_t,u_int64_t,u_int64_t,algorithm_t*,algorithm_t*,prf_plus_t*,natt_conf_t*,bool))add_sa;
|
||||
this->public.add_policy = (status_t(*)(kernel_interface_t*,host_t*, host_t*,host_t*,host_t*,u_int8_t,u_int8_t,int,int,protocol_id_t,u_int32_t))add_policy;
|
||||
this->public.update_sa_hosts = (status_t(*)(kernel_interface_t*,host_t*,host_t*,host_t*,host_t*,int,int,u_int32_t,protocol_id_t))update_sa_hosts;
|
||||
this->public.del_sa = (status_t(*)(kernel_interface_t*,host_t*,u_int32_t,protocol_id_t))del_sa;
|
||||
this->public.add_policy = (status_t(*)(kernel_interface_t*,host_t*, host_t*,host_t*,host_t*,u_int8_t,u_int8_t,int,int,protocol_id_t,u_int32_t))add_policy;
|
||||
this->public.query_policy = (status_t(*)(kernel_interface_t*,host_t*,host_t*,host_t*,host_t*,u_int8_t,u_int8_t,int,int,time_t*))query_policy;
|
||||
this->public.del_policy = (status_t(*)(kernel_interface_t*,host_t*,host_t*,host_t*,host_t*,u_int8_t,u_int8_t,int,int))del_policy;
|
||||
|
||||
this->public.destroy = (void(*)(kernel_interface_t*)) destroy;
|
||||
|
||||
/* private members */
|
||||
|
||||
@@ -45,13 +45,6 @@ struct private_receiver_t {
|
||||
* Public part of a receiver_t object.
|
||||
*/
|
||||
receiver_t public;
|
||||
|
||||
/**
|
||||
* @brief Thread function started at creation of the receiver object.
|
||||
*
|
||||
* @param this calling object
|
||||
*/
|
||||
void (*receive_packets) (private_receiver_t *this);
|
||||
|
||||
/**
|
||||
* Assigned thread.
|
||||
@@ -114,11 +107,10 @@ receiver_t * receiver_create()
|
||||
private_receiver_t *this = malloc_thing(private_receiver_t);
|
||||
|
||||
this->public.destroy = (void(*)(receiver_t*)) destroy;
|
||||
this->receive_packets = receive_packets;
|
||||
|
||||
this->logger = logger_manager->get_logger(logger_manager, RECEIVER);
|
||||
|
||||
if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->receive_packets, this) != 0)
|
||||
if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))receive_packets, this) != 0)
|
||||
{
|
||||
this->logger->log(this->logger, ERROR, "Receiver thread could not be started");
|
||||
free(this);
|
||||
|
||||
@@ -42,15 +42,6 @@ struct private_scheduler_t {
|
||||
* Public part of a scheduler_t object.
|
||||
*/
|
||||
scheduler_t public;
|
||||
|
||||
/**
|
||||
* @brief Get events from the event queue and add them to to job queue.
|
||||
*
|
||||
* Thread function started at creation of the scheduler object.
|
||||
*
|
||||
* @param this calling object
|
||||
*/
|
||||
void (*get_events) (private_scheduler_t *this);
|
||||
|
||||
/**
|
||||
* Assigned thread.
|
||||
@@ -75,7 +66,7 @@ static void get_events(private_scheduler_t * this)
|
||||
|
||||
this->logger->log(this->logger, CONTROL, "scheduler thread running, thread_ID: %06u", (int)pthread_self());
|
||||
|
||||
for (;;)
|
||||
while (TRUE)
|
||||
{
|
||||
this->logger->log(this->logger, CONTROL|LEVEL2, "Waiting for next event...");
|
||||
/* get a job, this block until one is available */
|
||||
@@ -109,11 +100,10 @@ scheduler_t * scheduler_create()
|
||||
private_scheduler_t *this = malloc_thing(private_scheduler_t);
|
||||
|
||||
this->public.destroy = (void(*)(scheduler_t*)) destroy;
|
||||
this->get_events = get_events;
|
||||
|
||||
this->logger = logger_manager->get_logger(logger_manager, SCHEDULER);
|
||||
|
||||
if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->get_events, this) != 0)
|
||||
if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))get_events, this) != 0)
|
||||
{
|
||||
/* thread could not be created */
|
||||
this->logger->log(this->logger, ERROR, "Scheduler thread could not be created!");
|
||||
|
||||
@@ -49,13 +49,6 @@ struct private_sender_t {
|
||||
*/
|
||||
pthread_t assigned_thread;
|
||||
|
||||
/**
|
||||
* @brief The thread function, sends out packets.
|
||||
*
|
||||
* @param this calling object
|
||||
*/
|
||||
void (*send_packets) (private_sender_t * this);
|
||||
|
||||
/**
|
||||
* A logger for this sender_t object.
|
||||
*/
|
||||
@@ -68,7 +61,7 @@ struct private_sender_t {
|
||||
*/
|
||||
static void send_packets(private_sender_t * this)
|
||||
{
|
||||
packet_t * current_packet;
|
||||
packet_t *current_packet;
|
||||
status_t status;
|
||||
|
||||
/* cancellation disabled by default */
|
||||
@@ -76,11 +69,11 @@ static void send_packets(private_sender_t * this)
|
||||
|
||||
this->logger->log(this->logger, CONTROL, "sender thread running, thread_ID: %06u", (int)pthread_self());
|
||||
|
||||
while (1)
|
||||
while (TRUE)
|
||||
{
|
||||
current_packet = charon->send_queue->get(charon->send_queue);
|
||||
this->logger->log(this->logger, CONTROL|LEVEL1, "Got a packet, sending it");
|
||||
status = charon->socket->send(charon->socket,current_packet);
|
||||
status = charon->socket->send(charon->socket, current_packet);
|
||||
if (status != SUCCESS)
|
||||
{
|
||||
this->logger->log(this->logger, ERROR, "Sending failed, socket returned %s",
|
||||
@@ -97,10 +90,10 @@ static void destroy(private_sender_t *this)
|
||||
{
|
||||
this->logger->log(this->logger, CONTROL | LEVEL1, "Going to terminate sender thread");
|
||||
pthread_cancel(this->assigned_thread);
|
||||
|
||||
|
||||
pthread_join(this->assigned_thread, NULL);
|
||||
this->logger->log(this->logger, CONTROL | LEVEL1, "Sender thread terminated");
|
||||
|
||||
|
||||
free(this);
|
||||
}
|
||||
|
||||
@@ -111,12 +104,11 @@ sender_t * sender_create()
|
||||
{
|
||||
private_sender_t *this = malloc_thing(private_sender_t);
|
||||
|
||||
this->send_packets = send_packets;
|
||||
this->public.destroy = (void(*)(sender_t*)) destroy;
|
||||
|
||||
this->logger = logger_manager->get_logger(logger_manager, SENDER);
|
||||
|
||||
if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))this->send_packets, this) != 0)
|
||||
if (pthread_create(&(this->assigned_thread), NULL, (void*(*)(void*))send_packets, this) != 0)
|
||||
{
|
||||
this->logger->log(this->logger, ERROR, "Sender thread could not be created");
|
||||
free(this);
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
/**
|
||||
* @file thread_pool.c
|
||||
*
|
||||
*
|
||||
* @brief Implementation of thread_pool_t.
|
||||
*
|
||||
*
|
||||
*/
|
||||
|
||||
/*
|
||||
@@ -57,12 +57,7 @@ struct private_thread_pool_t {
|
||||
/**
|
||||
* Logger of the thread pool.
|
||||
*/
|
||||
logger_t *pool_logger;
|
||||
|
||||
/**
|
||||
* Logger of the worker threads.
|
||||
*/
|
||||
logger_t *worker_logger;
|
||||
logger_t *logger;
|
||||
} ;
|
||||
|
||||
/**
|
||||
@@ -76,7 +71,9 @@ static void process_jobs(private_thread_pool_t *this)
|
||||
/* cancellation disabled by default */
|
||||
pthread_setcancelstate(PTHREAD_CANCEL_DISABLE, NULL);
|
||||
|
||||
this->worker_logger->log(this->worker_logger, CONTROL, "worker thread running, thread_ID: %06u", (int)pthread_self());
|
||||
this->logger->log(this->logger, CONTROL,
|
||||
"worker thread running, thread_ID: %06u",
|
||||
(int)pthread_self());
|
||||
|
||||
while (TRUE)
|
||||
{
|
||||
@@ -107,7 +104,8 @@ static void destroy(private_thread_pool_t *this)
|
||||
int current;
|
||||
/* flag thread for termination */
|
||||
for (current = 0; current < this->pool_size; current++) {
|
||||
this->pool_logger->log(this->pool_logger, CONTROL, "cancelling worker thread #%d", current+1);
|
||||
this->logger->log(this->logger, CONTROL,
|
||||
"cancelling worker thread #%d", current+1);
|
||||
pthread_cancel(this->threads[current]);
|
||||
}
|
||||
|
||||
@@ -115,11 +113,13 @@ static void destroy(private_thread_pool_t *this)
|
||||
for (current = 0; current < this->pool_size; current++) {
|
||||
if (pthread_join(this->threads[current], NULL) == 0)
|
||||
{
|
||||
this->pool_logger->log(this->pool_logger, CONTROL, "worker thread #%d terminated", current+1);
|
||||
this->logger->log(this->logger, CONTROL,
|
||||
"worker thread #%d terminated", current+1);
|
||||
}
|
||||
else
|
||||
{
|
||||
this->pool_logger->log(this->pool_logger, ERROR, "could not terminate worker thread #%d", current+1);
|
||||
this->logger->log(this->logger, ERROR,
|
||||
"could not terminate worker thread #%d", current+1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -143,32 +143,35 @@ thread_pool_t *thread_pool_create(size_t pool_size)
|
||||
/* initialize member */
|
||||
this->pool_size = pool_size;
|
||||
this->threads = malloc(sizeof(pthread_t) * pool_size);
|
||||
this->pool_logger = logger_manager->get_logger(logger_manager, THREAD_POOL);
|
||||
this->worker_logger = logger_manager->get_logger(logger_manager, WORKER);
|
||||
this->logger = logger_manager->get_logger(logger_manager, THREAD_POOL);
|
||||
|
||||
/* try to create as many threads as possible, up to pool_size */
|
||||
for (current = 0; current < pool_size; current++)
|
||||
{
|
||||
if (pthread_create(&(this->threads[current]), NULL, (void*(*)(void*))process_jobs, this) == 0)
|
||||
if (pthread_create(&(this->threads[current]), NULL,
|
||||
(void*(*)(void*))process_jobs, this) == 0)
|
||||
{
|
||||
this->pool_logger->log(this->pool_logger, CONTROL, "created worker thread #%d", current+1);
|
||||
this->logger->log(this->logger, CONTROL,
|
||||
"created worker thread #%d", current+1);
|
||||
}
|
||||
else
|
||||
{
|
||||
/* creation failed, is it the first one? */
|
||||
if (current == 0)
|
||||
{
|
||||
this->pool_logger->log(this->pool_logger, ERROR, "Could not create any thread");
|
||||
this->logger->log(this->logger, ERROR, "Could not create any thread");
|
||||
free(this->threads);
|
||||
free(this);
|
||||
return NULL;
|
||||
}
|
||||
/* not all threads could be created, but at least one :-/ */
|
||||
this->pool_logger->log(this->pool_logger, ERROR, "Could only create %d from requested %d threads!", current, pool_size);
|
||||
this->logger->log(this->logger, ERROR,
|
||||
"Could only create %d from requested %d threads!",
|
||||
current, pool_size);
|
||||
|
||||
this->pool_size = current;
|
||||
return (thread_pool_t*)this;
|
||||
}
|
||||
}
|
||||
}
|
||||
return (thread_pool_t*)this;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user