summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/ipcpd/ipcp.h2
-rw-r--r--src/ipcpd/shim-udp/main.c195
-rw-r--r--src/lib/dev.c134
3 files changed, 159 insertions, 172 deletions
diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index c9002d4d..70da0675 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -30,8 +30,6 @@
enum ipcp_state {
IPCP_INIT = 0,
- IPCP_ENROLLING,
- IPCP_BOOTSTRAPPING,
IPCP_ENROLLED,
IPCP_DISCONNECTED,
IPCP_SHUTDOWN
diff --git a/src/ipcpd/shim-udp/main.c b/src/ipcpd/shim-udp/main.c
index 74fa0d2b..e3f7fcdd 100644
--- a/src/ipcpd/shim-udp/main.c
+++ b/src/ipcpd/shim-udp/main.c
@@ -50,6 +50,7 @@
#include <stdlib.h>
#include <pthread.h>
#include <sys/wait.h>
+#include <fcntl.h>
#define THIS_TYPE IPCP_SHIM_UDP
#define LISTEN_PORT htons(0x0D1F)
@@ -93,9 +94,6 @@ struct shim_ap_data {
pthread_t sduloop;
pthread_t handler;
pthread_t sdu_reader;
-
- rw_lock_t thread_lock;
-
} * _ap_instance;
static int shim_ap_init(char * ap_name)
@@ -152,8 +150,6 @@ static int shim_ap_init(char * ap_name)
}
rw_lock_init(&_ap_instance->flows_lock);
- rw_lock_init(&_ap_instance->thread_lock);
- rw_lock_init(&_ap_instance->data_lock);
return 0;
}
@@ -165,7 +161,10 @@ void shim_ap_fini()
if (_ap_instance == NULL)
return;
- rw_lock_wrlock(&_ap_instance->data_lock);
+ rw_lock_wrlock(&_ipcp->state_lock);
+
+ if (_ipcp->state != IPCP_SHUTDOWN)
+ LOG_WARN("Cleaning up AP while not in shutdown.");
if (_ap_instance->api != NULL)
instance_name_destroy(_ap_instance->api);
@@ -184,9 +183,9 @@ void shim_ap_fini()
rw_lock_unlock(&_ap_instance->flows_lock);
- rw_lock_unlock(&_ap_instance->data_lock);
-
free(_ap_instance);
+
+ rw_lock_unlock(&_ipcp->state_lock);
}
/* only call this under flows_lock */
@@ -209,12 +208,12 @@ static ssize_t ipcp_udp_flow_write(int fd, void * buf, size_t count)
size_t index;
struct rb_entry e;
- rw_lock_rdlock(&_ap_instance->data_lock);
+ rw_lock_rdlock(&_ipcp->state_lock);
index = shm_create_du_buff(_ap_instance->dum, count, 0, buf, count);
if (index == -1) {
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
return -1;
}
@@ -226,17 +225,13 @@ static ssize_t ipcp_udp_flow_write(int fd, void * buf, size_t count)
if (shm_ap_rbuff_write(_ap_instance->flows[fd].rb, &e) < 0) {
rw_lock_unlock(&_ap_instance->flows_lock);
-
shm_release_du_buff(_ap_instance->dum, index);
-
- rw_lock_unlock(&_ap_instance->data_lock);
-
+ rw_lock_unlock(&_ipcp->state_lock);
return -EPIPE;
}
rw_lock_unlock(&_ap_instance->flows_lock);
-
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
return 0;
}
@@ -256,8 +251,8 @@ struct ipcp_udp_data {
struct sockaddr_in s_saddr;
int s_fd;
+ /* only modify under _ap_instance->flows_lock */
fd_set flow_fd_s;
- rw_lock_t fd_lock;
};
struct ipcp_udp_data * ipcp_udp_data_create()
@@ -279,8 +274,6 @@ struct ipcp_udp_data * ipcp_udp_data_create()
return NULL;
}
- rw_lock_init(&udp_data->fd_lock);
-
FD_ZERO(&udp_data->flow_fd_s);
return udp_data;
@@ -309,13 +302,7 @@ void ipcp_sig_handler(int sig, siginfo_t * info, void * c)
clean_threads = true;
}
- _ipcp->state = IPCP_SHUTDOWN;
-
- rw_lock_unlock(&_ipcp->state_lock);
-
if (clean_threads) {
- rw_lock_wrlock(&_ap_instance->thread_lock);
-
pthread_cancel(_ap_instance->handler);
pthread_cancel(_ap_instance->sdu_reader);
pthread_cancel(_ap_instance->sduloop);
@@ -323,12 +310,14 @@ void ipcp_sig_handler(int sig, siginfo_t * info, void * c)
pthread_join(_ap_instance->sduloop, NULL);
pthread_join(_ap_instance->handler, NULL);
pthread_join(_ap_instance->sdu_reader, NULL);
-
- rw_lock_unlock(&_ap_instance->thread_lock);
}
pthread_cancel(_ap_instance->mainloop);
+ _ipcp->state = IPCP_SHUTDOWN;
+
+ rw_lock_unlock(&_ipcp->state_lock);
+
pthread_sigmask(SIG_UNBLOCK, &sigset, NULL);
}
@@ -349,18 +338,25 @@ static void * ipcp_udp_listener()
while (true) {
int fd;
int port_id;
+
+ rw_lock_rdlock(&_ipcp->state_lock);
+
memset(&buf, 0, SHIM_UDP_BUF_SIZE);
n = sizeof c_saddr;
n = recvfrom(sfd, buf, SHIM_UDP_BUF_SIZE, 0,
(struct sockaddr *) &c_saddr, (unsigned *) &n);
- if (n < 0)
+ if (n < 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
continue;
+ }
/* flow alloc request from other host */
if (gethostbyaddr((const char *) &c_saddr.sin_addr.s_addr,
sizeof(c_saddr.sin_addr.s_addr), AF_INET)
- == NULL)
+ == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
continue;
+ }
fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
@@ -383,12 +379,14 @@ static void * ipcp_udp_listener()
if (connect(fd,
(struct sockaddr *) &c_saddr, sizeof c_saddr) < 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
close(fd);
continue;
}
/* echo back the packet */
if (send(fd, buf, strlen(buf), 0) < 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Failed to echo back the packet.");
close(fd);
continue;
@@ -403,9 +401,10 @@ static void * ipcp_udp_listener()
UNKNOWN_AE);
if (port_id < 0) {
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Could not get port id from IRMd");
close(fd);
- rw_lock_unlock(&_ap_instance->flows_lock);
continue;
}
@@ -414,6 +413,7 @@ static void * ipcp_udp_listener()
_ap_instance->flows[fd].state = FLOW_PENDING;
rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBG("Pending allocation request, port_id %u, UDP fd %d.",
port_id, fd);
@@ -427,9 +427,10 @@ static void * ipcp_udp_sdu_reader()
int n;
int fd;
char buf[SHIM_UDP_MAX_SDU_SIZE];
- struct timeval tv = {0, 1000};
+ struct timeval tv = {0, 100};
struct sockaddr_in r_saddr;
fd_set read_fds;
+ int flags;
while (true) {
rw_lock_rdlock(&_ipcp->state_lock);
@@ -439,20 +440,21 @@ static void * ipcp_udp_sdu_reader()
return (void *) 0;
}
- rw_lock_unlock(&_ipcp->state_lock);
-
- rw_lock_rdlock(&shim_data(_ipcp)->fd_lock);
+ rw_lock_rdlock(&_ap_instance->flows_lock);
read_fds = shim_data(_ipcp)->flow_fd_s;
- rw_lock_unlock(&shim_data(_ipcp)->fd_lock);
-
- if (select(FD_SETSIZE, &read_fds, NULL, NULL, &tv) <= 0)
+ if (select(FD_SETSIZE, &read_fds, NULL, NULL, &tv) <= 0) {
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
continue;
+ }
for (fd = 0; fd < FD_SETSIZE; ++fd) {
if (!FD_ISSET(fd, &read_fds))
continue;
+ flags = fcntl(fd, F_GETFL, 0);
+ fcntl(fd, F_SETFL, flags | O_NONBLOCK);
n = sizeof r_saddr;
if ((n = recvfrom(fd,
@@ -467,6 +469,9 @@ static void * ipcp_udp_sdu_reader()
if (ipcp_udp_flow_write(fd, buf, n) < 0)
LOG_ERR("Failed to write SDU.");
}
+
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
}
return (void *) 0;
@@ -488,14 +493,10 @@ static void * ipcp_udp_sdu_loop(void * o)
return (void *) 0;
}
- rw_lock_unlock(&_ipcp->state_lock);
-
- rw_lock_rdlock(&_ap_instance->data_lock);
-
e = shm_ap_rbuff_read(_ap_instance->rb);
if (e == NULL) {
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
continue;
}
@@ -503,7 +504,7 @@ static void * ipcp_udp_sdu_loop(void * o)
_ap_instance->dum,
e->index);
if (len == -1) {
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
free(e);
continue;
}
@@ -512,30 +513,27 @@ static void * ipcp_udp_sdu_loop(void * o)
fd = port_id_to_fd(e->port_id);
- rw_lock_unlock(&_ap_instance->flows_lock);
-
if (fd == -1) {
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
free(e);
continue;
}
if (len == 0) {
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
free(e);
continue;
}
- rw_lock_unlock(&_ap_instance->data_lock);
-
if (send(fd, buf, len, 0) < 0)
LOG_ERR("Failed to send SDU.");
- rw_lock_rdlock(&_ap_instance->data_lock);
-
shm_release_du_buff(_ap_instance->dum, e->index);
- rw_lock_unlock(&_ap_instance->data_lock);
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
free(e);
}
@@ -563,14 +561,11 @@ static int ipcp_udp_bootstrap(struct dif_config * conf)
return -1;
}
- _ipcp->state = IPCP_BOOTSTRAPPING;
-
- rw_lock_unlock(&_ipcp->state_lock);
-
if (inet_ntop(AF_INET,
&conf->ip_addr,
ipstr,
INET_ADDRSTRLEN) == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Failed to convert IP address");
return -1;
}
@@ -580,6 +575,7 @@ static int ipcp_udp_bootstrap(struct dif_config * conf)
&conf->dns_addr,
dnsstr,
INET_ADDRSTRLEN) == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Failed to convert DNS address");
return -1;
}
@@ -592,6 +588,7 @@ static int ipcp_udp_bootstrap(struct dif_config * conf)
/* UDP listen server */
if ((fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Can't create socket.");
return -1;
}
@@ -615,11 +612,12 @@ static int ipcp_udp_bootstrap(struct dif_config * conf)
if (bind(fd,
(struct sockaddr *) &shim_data(_ipcp)->s_saddr,
sizeof shim_data(_ipcp)->s_saddr ) < 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Couldn't bind to %s.", ipstr);
return -1;
}
- rw_lock_wrlock(&_ap_instance->thread_lock);
+ FD_CLR(shim_data(_ipcp)->s_fd, &shim_data(_ipcp)->flow_fd_s);
pthread_create(&_ap_instance->handler,
NULL,
@@ -635,16 +633,6 @@ static int ipcp_udp_bootstrap(struct dif_config * conf)
ipcp_udp_sdu_loop,
NULL);
- rw_lock_unlock(&_ap_instance->thread_lock);
-
- rw_lock_wrlock(&shim_data(_ipcp)->fd_lock);
-
- FD_CLR(shim_data(_ipcp)->s_fd, &shim_data(_ipcp)->flow_fd_s);
-
- rw_lock_unlock(&shim_data(_ipcp)->fd_lock);
-
- rw_lock_wrlock(&_ipcp->state_lock);
-
_ipcp->state = IPCP_ENROLLED;
rw_lock_unlock(&_ipcp->state_lock);
@@ -806,9 +794,8 @@ static int ipcp_udp_name_reg(char * name)
return -1; /* -ENOTENROLLED */
}
- rw_lock_unlock(&_ipcp->state_lock);
-
if (ipcp_data_add_reg_entry(_ipcp->data, name)) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Failed to add %s to local registry.", name);
return -1;
}
@@ -822,11 +809,13 @@ static int ipcp_udp_name_reg(char * name)
if (inet_ntop(AF_INET, &ip_addr,
ipstr, INET_ADDRSTRLEN) == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
return -1;
}
if (inet_ntop(AF_INET, &dns_addr,
dnsstr, INET_ADDRSTRLEN) == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
return -1;
}
@@ -835,11 +824,14 @@ static int ipcp_udp_name_reg(char * name)
if (ddns_send(cmd)) {
ipcp_data_del_reg_entry(_ipcp->data, name);
+ rw_lock_unlock(&_ipcp->state_lock);
return -1;
}
}
#endif
+ rw_lock_unlock(&_ipcp->state_lock);
+
LOG_DBG("Registered %s.", name);
return 0;
@@ -870,12 +862,11 @@ static int ipcp_udp_name_unreg(char * name)
return -1; /* -ENOTENROLLED */
}
- rw_lock_unlock(&_ipcp->state_lock);
-
dns_addr = shim_data(_ipcp)->dns_addr;
if (dns_addr != 0) {
if (inet_ntop(AF_INET, &dns_addr, dnsstr, INET_ADDRSTRLEN)
== NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
return -1;
}
sprintf(cmd, "server %s\nupdate delete %s A\nsend\nquit\n",
@@ -887,6 +878,8 @@ static int ipcp_udp_name_unreg(char * name)
ipcp_data_del_reg_entry(_ipcp->data, name);
+ rw_lock_unlock(&_ipcp->state_lock);
+
LOG_DBG("Unregistered %s.", name);
return 0;
@@ -923,11 +916,10 @@ static int ipcp_udp_flow_alloc(int port_id,
return -1; /* -ENOTENROLLED */
}
- rw_lock_unlock(&_ipcp->state_lock);
-
if (strlen(dst_name) > 255
|| strlen(src_ap_name) > 255
|| strlen(src_ae_name) > 255) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Name too long for this shim.");
return -1;
}
@@ -944,6 +936,7 @@ static int ipcp_udp_flow_alloc(int port_id,
l_saddr.sin_port = 0;
if (bind(fd, (struct sockaddr *) &l_saddr, sizeof l_saddr) < 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
close(fd);
return -1;
}
@@ -954,6 +947,7 @@ static int ipcp_udp_flow_alloc(int port_id,
if (dns_addr != 0) {
ip_addr = ddns_resolve(dst_name, dns_addr);
if (ip_addr == 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBGF("Could not resolve %s.", dst_name);
close(fd);
return -1;
@@ -962,6 +956,7 @@ static int ipcp_udp_flow_alloc(int port_id,
#endif
h = gethostbyname(dst_name);
if (h == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBGF("Could not resolve %s.", dst_name);
close(fd);
return -1;
@@ -979,6 +974,7 @@ static int ipcp_udp_flow_alloc(int port_id,
if (sendto(fd, dst_name, strlen(dst_name), 0,
(struct sockaddr *) &r_saddr, sizeof r_saddr) < 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Failed to send packet");
close(fd);
return -1;
@@ -988,6 +984,7 @@ static int ipcp_udp_flow_alloc(int port_id,
recv_buf = malloc(strlen(dst_name) + 1);
if (recv_buf == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Failed to malloc recv_buff.");
close(fd);
return -1;
@@ -1004,6 +1001,7 @@ static int ipcp_udp_flow_alloc(int port_id,
(struct sockaddr *) &rf_saddr,
sizeof rf_saddr)
< 0) {
+ rw_lock_unlock(&_ipcp->state_lock);
close(fd);
free(recv_buf);
return -1;
@@ -1016,6 +1014,7 @@ static int ipcp_udp_flow_alloc(int port_id,
rb = shm_ap_rbuff_open(n_pid);
if (rb == NULL) {
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_ERR("Could not open N + 1 ringbuffer.");
close(fd);
return -1; /* -ENORBUFF */
@@ -1026,22 +1025,21 @@ static int ipcp_udp_flow_alloc(int port_id,
_ap_instance->flows[fd].state = FLOW_ALLOCATED;
_ap_instance->flows[fd].rb = rb;
+ FD_SET(fd, &shim_data(_ipcp)->flow_fd_s);
+
rw_lock_unlock(&_ap_instance->flows_lock);
/* tell IRMd that flow allocation "worked" */
if (ipcp_flow_alloc_reply(getpid(), port_id, 0)) {
+ rw_lock_unlock(&_ipcp->state_lock);
shm_ap_rbuff_close(rb);
LOG_ERR("Failed to notify IRMd about flow allocation reply");
close(fd);
return -1;
}
- rw_lock_wrlock(&shim_data(_ipcp)->fd_lock);
-
- FD_SET(fd, &shim_data(_ipcp)->flow_fd_s);
-
- rw_lock_unlock(&shim_data(_ipcp)->fd_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBG("Allocated flow with port_id %d on UDP fd %d.", port_id, fd);
@@ -1053,58 +1051,52 @@ static int ipcp_udp_flow_alloc_resp(int port_id,
int response)
{
struct shm_ap_rbuff * rb;
+ struct timespec wait = {0, 1000000};
int fd = -1;
if (response)
return 0;
+ rw_lock_unlock(&_ipcp->state_lock);
+
/* awaken pending flow */
- rw_lock_rdlock(&_ap_instance->flows_lock);
+ rw_lock_wrlock(&_ap_instance->flows_lock);
fd = port_id_to_fd(port_id);
if (fd < 0) {
rw_lock_unlock(&_ap_instance->flows_lock);
-
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBGF("Could not find flow with port_id %d.", port_id);
return 0;
}
if (_ap_instance->flows[fd].state != FLOW_PENDING) {
rw_lock_unlock(&_ap_instance->flows_lock);
-
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBGF("Flow was not pending.");
return -1;
}
- rw_lock_unlock(&_ap_instance->flows_lock);
-
rb = shm_ap_rbuff_open(n_pid);
if (rb == NULL) {
LOG_ERR("Could not open N + 1 ringbuffer.");
-
- rw_lock_wrlock(&_ap_instance->flows_lock);
-
_ap_instance->flows[fd].state = FLOW_NULL;
_ap_instance->flows[fd].port_id = 0;
-
rw_lock_unlock(&_ap_instance->flows_lock);
-
+ rw_lock_unlock(&_ipcp->state_lock);
return 0;
}
- rw_lock_wrlock(&_ap_instance->flows_lock);
-
_ap_instance->flows[fd].state = FLOW_ALLOCATED;
_ap_instance->flows[fd].rb = rb;
- rw_lock_unlock(&_ap_instance->flows_lock);
-
- rw_lock_wrlock(&shim_data(_ipcp)->fd_lock);
-
FD_SET(fd, &shim_data(_ipcp)->flow_fd_s);
- rw_lock_unlock(&shim_data(_ipcp)->fd_lock);
+ nanosleep(&wait, NULL);
+
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBG("Accepted flow, port_id %d on UDP fd %d.", port_id, fd);
@@ -1115,13 +1107,15 @@ static int ipcp_udp_flow_dealloc(int port_id)
{
int fd = -1;
struct shm_ap_rbuff * rb;
+ struct timespec wait = {0, 1000000};
+ rw_lock_rdlock(&_ipcp->state_lock);
rw_lock_wrlock(&_ap_instance->flows_lock);
fd = port_id_to_fd(port_id);
if (fd < 0) {
rw_lock_unlock(&_ap_instance->flows_lock);
-
+ rw_lock_unlock(&_ipcp->state_lock);
LOG_DBGF("Could not find flow with port_id %d.", port_id);
return 0;
}
@@ -1131,19 +1125,18 @@ static int ipcp_udp_flow_dealloc(int port_id)
rb = _ap_instance->flows[fd].rb;
_ap_instance->flows[fd].rb = NULL;
- rw_lock_unlock(&_ap_instance->flows_lock);
-
if (rb != NULL)
shm_ap_rbuff_close(rb);
- rw_lock_wrlock(&shim_data(_ipcp)->fd_lock);
-
FD_CLR(fd, &shim_data(_ipcp)->flow_fd_s);
- rw_lock_unlock(&shim_data(_ipcp)->fd_lock);
+ nanosleep(&wait, NULL);
close(fd);
+ rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ipcp->state_lock);
+
return 0;
}
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 67c58702..03559a1b 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -133,9 +133,9 @@ void ap_fini(void)
if (_ap_instance->flows[i].rb != NULL)
shm_ap_rbuff_close(_ap_instance->flows[i].rb);
- rw_lock_unlock(&_ap_instance->data_lock);
-
free(_ap_instance);
+
+ rw_lock_unlock(&_ap_instance->data_lock);
}
#if 0
@@ -163,26 +163,29 @@ int ap_reg(char ** difs,
return -EINVAL;
}
- if (_ap_instance == NULL)
+ rw_lock_rdlock(&_ap_instance->data_lock);
+
+ if (_ap_instance == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1; /* -ENOTINIT */
+ }
msg.code = IRM_MSG_CODE__IRM_AP_REG;
msg.has_pid = true;
msg.dif_name = difs;
msg.n_dif_name = len;
- rw_lock_rdlock(&_ap_instance->data_lock);
-
msg.pid = _ap_instance->api->id;
msg.ap_name = _ap_instance->api->name;
- rw_lock_unlock(&_ap_instance->data_lock);
-
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_result) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -192,10 +195,11 @@ int ap_reg(char ** difs,
irm_msg__free_unpacked(recv_msg, NULL);
- rw_lock_wrlock(&_ap_instance->data_lock);
+ rw_lock_wrlock(&_ap_instance->flows_lock);
fd = bmp_allocate(_ap_instance->fds);
+ rw_lock_unlock(&_ap_instance->flows_lock);
rw_lock_unlock(&_ap_instance->data_lock);
return fd;
@@ -214,23 +218,24 @@ int ap_unreg(char ** difs,
return -EINVAL;
}
+ rw_lock_rdlock(&_ap_instance->data_lock);
+
msg.code = IRM_MSG_CODE__IRM_AP_UNREG;
msg.has_pid = true;
msg.dif_name = difs;
msg.n_dif_name = len;
- rw_lock_rdlock(&_ap_instance->data_lock);
-
msg.pid = _ap_instance->api->id;
msg.ap_name = _ap_instance->api->name;
- rw_lock_unlock(&_ap_instance->data_lock);
-
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_result) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -238,6 +243,8 @@ int ap_unreg(char ** difs,
ret = recv_msg->result;
irm_msg__free_unpacked(recv_msg, NULL);
+ rw_lock_unlock(&_ap_instance->data_lock);
+
return ret;
}
@@ -249,20 +256,21 @@ int flow_accept(int fd,
irm_msg_t * recv_msg = NULL;
int cfd = -1;
+ rw_lock_rdlock(&_ap_instance->data_lock);
+
msg.code = IRM_MSG_CODE__IRM_FLOW_ACCEPT;
msg.has_pid = true;
- rw_lock_rdlock(&_ap_instance->data_lock);
-
msg.pid = _ap_instance->api->id;
- rw_lock_unlock(&_ap_instance->data_lock);
-
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_pid || !recv_msg->has_port_id) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -270,6 +278,7 @@ int flow_accept(int fd,
if (ap_name != NULL) {
*ap_name = strdup(recv_msg->ap_name);
if (*ap_name == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -279,30 +288,22 @@ int flow_accept(int fd,
if (ae_name != NULL) {
*ae_name = strdup(recv_msg->ae_name);
if (*ae_name == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
}
- rw_lock_wrlock(&_ap_instance->data_lock);
+ rw_lock_wrlock(&_ap_instance->flows_lock);
cfd = bmp_allocate(_ap_instance->fds);
- rw_lock_unlock(&_ap_instance->data_lock);
-
- rw_lock_wrlock(&_ap_instance->flows_lock);
-
_ap_instance->flows[cfd].rb = shm_ap_rbuff_open(recv_msg->pid);
if (_ap_instance->flows[cfd].rb == NULL) {
- rw_lock_wrlock(&_ap_instance->data_lock);
-
bmp_release(_ap_instance->fds, cfd);
-
+ rw_lock_unlock(&_ap_instance->flows_lock);
rw_lock_unlock(&_ap_instance->data_lock);
-
irm_msg__free_unpacked(recv_msg, NULL);
-
- rw_lock_unlock(&_ap_instance->flows_lock);
return -1;
}
@@ -313,10 +314,6 @@ int flow_accept(int fd,
irm_msg__free_unpacked(recv_msg, NULL);
- rw_lock_wrlock(&_ap_instance->data_lock);
-
- bmp_release(_ap_instance->fds, fd);
-
rw_lock_unlock(&_ap_instance->data_lock);
return cfd;
@@ -329,15 +326,11 @@ int flow_alloc_resp(int fd,
irm_msg_t * recv_msg = NULL;
int ret = -1;
- msg.code = IRM_MSG_CODE__IRM_FLOW_ALLOC_RESP;
- msg.has_pid = true;
-
rw_lock_rdlock(&_ap_instance->data_lock);
+ msg.code = IRM_MSG_CODE__IRM_FLOW_ALLOC_RESP;
+ msg.has_pid = true;
msg.pid = _ap_instance->api->id;
-
- rw_lock_unlock(&_ap_instance->data_lock);
-
msg.has_port_id = true;
rw_lock_rdlock(&_ap_instance->flows_lock);
@@ -350,10 +343,13 @@ int flow_alloc_resp(int fd,
msg.response = response;
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_result) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -361,6 +357,8 @@ int flow_alloc_resp(int fd,
ret = recv_msg->result;
irm_msg__free_unpacked(recv_msg, NULL);
+ rw_lock_unlock(&_ap_instance->data_lock);
+
return ret;
}
@@ -375,6 +373,8 @@ int flow_alloc(char * dst_name,
if (dst_name == NULL)
return -EINVAL;
+ rw_lock_rdlock(&_ap_instance->data_lock);
+
if (src_ae_name == NULL)
src_ae_name = UNKNOWN_AE;
@@ -383,39 +383,30 @@ int flow_alloc(char * dst_name,
msg.ae_name = src_ae_name;
msg.has_pid = true;
- rw_lock_rdlock(&_ap_instance->data_lock);
-
msg.pid = _ap_instance->api->id;
msg.ap_name = _ap_instance->api->name;
- rw_lock_unlock(&_ap_instance->data_lock);
-
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_pid || !recv_msg->has_port_id) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
- rw_lock_wrlock(&_ap_instance->data_lock);
+ rw_lock_wrlock(&_ap_instance->flows_lock);
fd = bmp_allocate(_ap_instance->fds);
- rw_lock_unlock(&_ap_instance->data_lock);
-
- rw_lock_wrlock(&_ap_instance->flows_lock);
-
_ap_instance->flows[fd].rb = shm_ap_rbuff_open(recv_msg->pid);
if (_ap_instance->flows[fd].rb == NULL) {
- rw_lock_wrlock(&_ap_instance->data_lock);
-
bmp_release(_ap_instance->fds, fd);
-
- rw_lock_unlock(&_ap_instance->data_lock);
-
rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -424,6 +415,7 @@ int flow_alloc(char * dst_name,
_ap_instance->flows[fd].oflags = FLOW_O_DEFAULT;
rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
@@ -436,6 +428,8 @@ int flow_alloc_res(int fd)
irm_msg_t * recv_msg = NULL;
int result = 0;
+ rw_lock_rdlock(&_ap_instance->data_lock);
+
msg.code = IRM_MSG_CODE__IRM_FLOW_ALLOC_RES;
msg.has_port_id = true;
@@ -446,10 +440,13 @@ int flow_alloc_res(int fd)
rw_lock_unlock(&_ap_instance->flows_lock);
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_result) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -457,6 +454,8 @@ int flow_alloc_res(int fd)
result = recv_msg->result;
irm_msg__free_unpacked(recv_msg, NULL);
+ rw_lock_unlock(&_ap_instance->data_lock);
+
return result;
}
@@ -466,6 +465,8 @@ int flow_dealloc(int fd)
irm_msg_t * recv_msg = NULL;
int ret = -1;
+ rw_lock_rdlock(&_ap_instance->data_lock);
+
msg.code = IRM_MSG_CODE__IRM_FLOW_DEALLOC;
msg.has_port_id = true;
@@ -477,19 +478,18 @@ int flow_dealloc(int fd)
shm_ap_rbuff_close(_ap_instance->flows[fd].rb);
_ap_instance->flows[fd].rb = NULL;
- rw_lock_wrlock(&_ap_instance->data_lock);
-
bmp_release(_ap_instance->fds, fd);
- rw_lock_unlock(&_ap_instance->data_lock);
-
rw_lock_unlock(&_ap_instance->flows_lock);
recv_msg = send_recv_irm_msg(&msg);
- if (recv_msg == NULL)
+ if (recv_msg == NULL) {
+ rw_lock_unlock(&_ap_instance->data_lock);
return -1;
+ }
if (!recv_msg->has_result) {
+ rw_lock_unlock(&_ap_instance->data_lock);
irm_msg__free_unpacked(recv_msg, NULL);
return -1;
}
@@ -497,6 +497,8 @@ int flow_dealloc(int fd)
ret = recv_msg->result;
irm_msg__free_unpacked(recv_msg, NULL);
+ rw_lock_unlock(&_ap_instance->data_lock);
+
return ret;
}
@@ -551,16 +553,12 @@ ssize_t flow_write(int fd, void * buf, size_t count)
if (_ap_instance->flows[fd].oflags & FLOW_O_NONBLOCK) {
if (shm_ap_rbuff_write(_ap_instance->flows[fd].rb, &e) < 0) {
shm_release_du_buff(_ap_instance->dum, index);
-
rw_lock_unlock(&_ap_instance->flows_lock);
-
rw_lock_unlock(&_ap_instance->data_lock);
-
return -EPIPE;
}
rw_lock_unlock(&_ap_instance->flows_lock);
-
rw_lock_unlock(&_ap_instance->data_lock);
return 0;
@@ -569,9 +567,8 @@ ssize_t flow_write(int fd, void * buf, size_t count)
;
}
- rw_lock_unlock(&_ap_instance->data_lock);
-
rw_lock_unlock(&_ap_instance->flows_lock);
+ rw_lock_unlock(&_ap_instance->data_lock);
return 0;
}
@@ -583,7 +580,6 @@ ssize_t flow_read(int fd, void * buf, size_t count)
uint8_t * sdu;
rw_lock_rdlock(&_ap_instance->data_lock);
-
rw_lock_rdlock(&_ap_instance->flows_lock);
if (_ap_instance->flows[fd].oflags & FLOW_O_NONBLOCK) {
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-31 03:52:01 +0000