This will add the infrastructure to actively react to flow up, down
and deallocated events.
Signed-off-by: Sander Vrijders <sander.vrijders@xxxxxxxx>
---
src/ipcpd/normal/connmgr.c | 7 ++++++-
src/ipcpd/normal/connmgr.h | 21 +++++++++++----------
src/ipcpd/normal/sdu_sched.c | 25 +++++++++++++++++++++----
src/lib/dev.c | 20 +++++++++++++++++---
4 files changed, 55 insertions(+), 18 deletions(-)
diff --git a/src/ipcpd/normal/connmgr.c b/src/ipcpd/normal/connmgr.c
index 9530633..bf07ebc 100644
--- a/src/ipcpd/normal/connmgr.c
+++ b/src/ipcpd/normal/connmgr.c
@@ -193,7 +193,9 @@ static void handle_event(void * self,
(void) self;
- if (!(event == NOTIFY_DT_FLOW_UP || event == NOTIFY_DT_FLOW_DOWN))
+ if (!(event == NOTIFY_DT_FLOW_UP ||
+ event == NOTIFY_DT_FLOW_DOWN ||
+ event == NOTIFY_DT_FLOW_DEALLOC))
return;
if (get_conn_by_fd(*((int *) o), COMPID_DT, &conn))
@@ -206,6 +208,9 @@ static void handle_event(void * self,
case NOTIFY_DT_FLOW_DOWN:
notifier_event(NOTIFY_DT_CONN_DOWN, &conn);
break;
+ case NOTIFY_DT_FLOW_DEALLOC:
+ notifier_event(NOTIFY_DT_CONN_DEL, &conn);
+ break;
default:
break;
}
diff --git a/src/ipcpd/normal/connmgr.h b/src/ipcpd/normal/connmgr.h
index 510b8e4..a7e8a6e 100644
--- a/src/ipcpd/normal/connmgr.h
+++ b/src/ipcpd/normal/connmgr.h
@@ -28,16 +28,17 @@
#include "comp.h"
-#define NOTIFY_DT_CONN_ADD 0x00D0
-#define NOTIFY_DT_CONN_DEL 0x00D1
-#define NOTIFY_DT_CONN_QOS 0x00D2
-#define NOTIFY_DT_CONN_UP 0x00D3
-#define NOTIFY_DT_CONN_DOWN 0x00D4
-#define NOTIFY_DT_FLOW_UP 0x00D5
-#define NOTIFY_DT_FLOW_DOWN 0x00D6
-
-#define NOTIFY_MGMT_CONN_ADD 0x00F0
-#define NOTIFY_MGMT_CONN_DEL 0x00F1
+#define NOTIFY_DT_CONN_ADD 0x00D0
+#define NOTIFY_DT_CONN_DEL 0x00D1
+#define NOTIFY_DT_CONN_QOS 0x00D2
+#define NOTIFY_DT_CONN_UP 0x00D3
+#define NOTIFY_DT_CONN_DOWN 0x00D4
+#define NOTIFY_DT_FLOW_UP 0x00D5
+#define NOTIFY_DT_FLOW_DOWN 0x00D6
+#define NOTIFY_DT_FLOW_DEALLOC 0x00D7
+
+#define NOTIFY_MGMT_CONN_ADD 0x00F0
+#define NOTIFY_MGMT_CONN_DEL 0x00F1
int connmgr_init(void);
diff --git a/src/ipcpd/normal/sdu_sched.c b/src/ipcpd/normal/sdu_sched.c
index 0ae2289..6ce18ed 100644
--- a/src/ipcpd/normal/sdu_sched.c
+++ b/src/ipcpd/normal/sdu_sched.c
@@ -25,9 +25,11 @@
#include "config.h"
#include <ouroboros/errno.h>
+#include <ouroboros/notifier.h>
#include "ipcp.h"
#include "sdu_sched.h"
+#include "connmgr.h"
#include <assert.h>
#include <sched.h>
@@ -86,10 +88,25 @@ static void * sdu_reader(void * o)
continue;
while ((fd = fqueue_next(fq)) >= 0) {
- if (ipcp_flow_read(fd, &sdb))
- continue;
-
- sched->callback(fd, qc, sdb);
+ switch (fqueue_type(fq)) {
+ case FLOW_DEALLOC:
+ notifier_event(NOTIFY_DT_FLOW_DEALLOC, &fd);
+ break;
+ case FLOW_DOWN:
+ notifier_event(NOTIFY_DT_FLOW_DOWN, &fd);
+ break;
+ case FLOW_UP:
+ notifier_event(NOTIFY_DT_FLOW_UP, &fd);
+ break;
+ case FLOW_PKT:
+ if (ipcp_flow_read(fd, &sdb))
+ continue;
+
+ sched->callback(fd, qc, sdb);
+ break;
+ default:
+ break;
+ }
}
}
diff --git a/src/lib/dev.c b/src/lib/dev.c
index 9eade79..dd908f7 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -252,8 +252,12 @@ static void flow_fini(int fd)
shm_rbuff_close(ai.flows[fd].tx_rb);
}
- if (ai.flows[fd].set != NULL)
+ if (ai.flows[fd].set != NULL) {
+ shm_flow_set_notify(ai.flows[fd].set,
+ ai.flows[fd].port_id,
+ FLOW_DEALLOC);
shm_flow_set_close(ai.flows[fd].set);
+ }
if (ai.flows[fd].frcti != NULL)
frcti_destroy(ai.flows[fd].frcti);
@@ -435,8 +439,6 @@ static void fini(void)
frct_fini();
- shm_flow_set_destroy(ai.fqset);
-
if (ai.prog != NULL)
free(ai.prog);
@@ -452,6 +454,8 @@ static void fini(void)
}
}
+ shm_flow_set_destroy(ai.fqset);
+
for (i = 0; i < SYS_MAX_FLOWS; ++i) {
pthread_mutex_destroy(&ai.ports[i].state_lock);
pthread_cond_destroy(&ai.ports[i].state_cond);
@@ -764,9 +768,15 @@ int fccntl(int fd,
if (flow->oflags & FLOWFDOWN) {
rx_acl |= ACL_FLOWDOWN;
tx_acl |= ACL_FLOWDOWN;
+ shm_flow_set_notify(flow->set,
+ flow->port_id,
+ FLOW_DOWN);
} else {
rx_acl &= ~ACL_FLOWDOWN;
tx_acl &= ~ACL_FLOWDOWN;
+ shm_flow_set_notify(flow->set,
+ flow->port_id,
+ FLOW_UP);
}
shm_rbuff_set_acl(flow->rx_rb, rx_acl);
@@ -1425,6 +1435,10 @@ void ipcp_flow_fini(int fd)
shm_rbuff_set_acl(ai.flows[fd].rx_rb, ACL_FLOWDOWN);
shm_rbuff_set_acl(ai.flows[fd].tx_rb, ACL_FLOWDOWN);
+ shm_flow_set_notify(ai.flows[fd].set,
+ ai.flows[fd].port_id,
+ FLOW_DEALLOC);
+
rx_rb = ai.flows[fd].rx_rb;
pthread_rwlock_unlock(&ai.lock);
--
2.18.0
