lib: Implementation of flow related ops

This adds the messages that are sent to the IPCPs related to
flows. Some messages are also sent to the IRMd (e.g. when a new flow
arrives).

6 files changed, 376 insertions, 47 deletions

diff --git a/include/ouroboros/ipcp.h b/include/ouroboros/ipcp.h
index b39a6b0d..fdaf5c4a 100644
--- a/include/ouroboros/ipcp.h
+++ b/include/ouroboros/ipcp.h
@@ -23,10 +23,10 @@
 #ifndef OUROBOROS_IPCP_H
 #define OUROBOROS_IPCP_H
 
-#include <sys/types.h>
+#include <ouroboros/common.h>
+#include <ouroboros/instance_name.h>
 
-#include "common.h"
-#include "instance_name.h"
+#include <sys/types.h>
 
 struct ipcp;
 
@@ -44,19 +44,17 @@ int   ipcp_unreg(pid_t   pid,
 
 int   ipcp_bootstrap(pid_t               pid,
                      struct dif_config * conf);
-int   ipcp_enroll(pid_t   pid,
-                  char *  dif_name,
-                  char *  member_name,
-                  char ** n_1_difs,
-                  ssize_t n_1_difs_size);
+int   ipcp_enroll(pid_t  pid,
+                  char * member_name,
+                  char * n_1_dif);
 
 /* Flow related ops, these go from IRMd to IPCP */
 
 int   ipcp_ap_reg(pid_t    pid,
                   uint32_t reg_api_id,
                   char *   ap_name);
-int   ipcp_ap_unreg(pid_t  pid,
-                    char * ap_name);
+int   ipcp_ap_unreg(pid_t    pid,
+                    uint32_t reg_api_id);
 
 int   ipcp_flow_alloc(pid_t             pid,
                       uint32_t          port_id,
diff --git a/include/ouroboros/sockets.h b/include/ouroboros/sockets.h
index 9904eecd..0c517bd4 100644
--- a/include/ouroboros/sockets.h
+++ b/include/ouroboros/sockets.h
@@ -37,7 +37,7 @@ typedef IpcpMsg ipcp_msg_t;
 #define IRM_MSG_BUF_SIZE 256
 
 #define IPCP_SOCK_PATH_PREFIX "/tmp/ipcp_sock"
-#define IPCP_MSG_BUFS_SIZE IRM_MSG_BUF_SIZE
+#define IPCP_MSG_BUF_SIZE IRM_MSG_BUF_SIZE
 
 /* Returns the full socket path of an IPCP */
 char *      ipcp_sock_path(pid_t pid);
diff --git a/src/irmd/main.c b/src/irmd/main.c
index af0f85e2..0256248b 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -165,8 +165,7 @@ static int enroll_ipcp(instance_name_t  * api,
 
         n_1_difs_size = da_resolve_dap(member, n_1_difs);
         if (n_1_difs_size != 0)
-                if (ipcp_enroll(pid, dif_name, member,
-                                n_1_difs, n_1_difs_size)) {
+                if (ipcp_enroll(pid, member, n_1_difs[0])) {
                         LOG_ERR("Could not enroll IPCP");
                         return -1;
                 }
@@ -268,6 +267,24 @@ static int flow_cntl(int fd,
         return -1;
 }
 
+static int flow_req_arr(uint32_t reg_api_id,
+                        char *   ap_name,
+                        char *   ae_name)
+{
+        return -1;
+}
+
+static int flow_alloc_reply(uint32_t port_id,
+                            int      result)
+{
+        return -1;
+}
+
+static int flow_dealloc_ipcp(uint32_t port_id)
+{
+        return -1;
+}
+
 /* FIXME: Close sockfd on closing and release irm */
 int main()
 {
@@ -391,6 +408,21 @@ int main()
                         ret_msg.result = flow_cntl(msg->fd,
                                                    msg->oflags);
                         break;
+                case IRM_MSG_CODE__IPCP_FLOW_REQ_ARR:
+                        ret_msg.has_fd = true;
+                        ret_msg.fd = flow_req_arr(msg->port_id,
+                                                  msg->ap_name,
+                                                  msg->ae_name);
+                        break;
+                case IRM_MSG_CODE__IPCP_FLOW_ALLOC_REPLY:
+                        ret_msg.has_result = true;
+                        ret_msg.result = flow_alloc_reply(msg->port_id,
+                                                          msg->result);
+                        break;
+                case IRM_MSG_CODE__IPCP_FLOW_DEALLOC:
+                        ret_msg.has_result = true;
+                        ret_msg.result = flow_dealloc_ipcp(msg->port_id);
+                        break;
                 default:
                         LOG_ERR("Don't know that message code");
                         break;
diff --git a/src/lib/ipcp.c b/src/lib/ipcp.c
index 2caeaad3..d61fcb50 100644
--- a/src/lib/ipcp.c
+++ b/src/lib/ipcp.c
@@ -35,35 +35,37 @@
 #include <sys/types.h>
 #include <sys/wait.h>
 
-static int send_ipcp_msg(pid_t pid,
-                         ipcp_msg_t * msg)
+static ipcp_msg_t * send_recv_ipcp_msg(pid_t pid,
+                                       ipcp_msg_t * msg)
 {
        int sockfd = 0;
        buffer_t buf;
-       char * sock_path;
+       char * sock_path = NULL;
+       ssize_t count = 0;
+       ipcp_msg_t * recv_msg = NULL;
 
        sock_path = ipcp_sock_path(pid);
        if (sock_path == NULL)
-               return -1;
+               return NULL;
 
        sockfd = client_socket_open(sock_path);
        if (sockfd < 0) {
                free(sock_path);
-               return -1;
+               return NULL;
        }
 
        buf.size = ipcp_msg__get_packed_size(msg);
        if (buf.size == 0) {
                close(sockfd);
                free(sock_path);
-               return -1;
+               return NULL;
        }
 
        buf.data = malloc(buf.size);
        if (buf.data == NULL) {
                close(sockfd);
                free(sock_path);
-               return -ENOMEM;
+               return NULL;
        }
 
        ipcp_msg__pack(msg, buf.data);
@@ -72,13 +74,29 @@ static int send_ipcp_msg(pid_t pid,
                free(sock_path);
                free(buf.data);
                close(sockfd);
-               return -1;
+               return NULL;
+       }
+
+       count = read(sockfd, buf.data, IPCP_MSG_BUF_SIZE);
+       if (count <= 0) {
+               free(sock_path);
+               free(buf.data);
+               close(sockfd);
+               return NULL;
+       }
+
+       recv_msg = ipcp_msg__unpack(NULL, count, buf.data);
+       if (recv_msg == NULL) {
+               free(sock_path);
+               free(buf.data);
+               close(sockfd);
+               return NULL;
        }
 
        free(buf.data);
        free(sock_path);
        close(sockfd);
-       return 0;
+       return recv_msg;
 }
 
 pid_t ipcp_create(instance_name_t * api,
@@ -164,6 +182,8 @@ int ipcp_reg(pid_t pid,
              size_t difs_size)
 {
         ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
 
         if (difs == NULL ||
             difs_size == 0 ||
@@ -174,12 +194,19 @@ int ipcp_reg(pid_t pid,
         msg.dif_name = difs;
         msg.n_dif_name = difs_size;
 
-        if (send_ipcp_msg(pid, &msg)) {
-                LOG_ERR("Failed to send message to daemon");
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
                 return -1;
         }
 
-        return 0;
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
 }
 
 int ipcp_unreg(pid_t pid,
@@ -187,6 +214,8 @@ int ipcp_unreg(pid_t pid,
                size_t difs_size)
 {
         ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
 
         if (difs == NULL ||
             difs_size == 0 ||
@@ -197,42 +226,54 @@ int ipcp_unreg(pid_t pid,
         msg.dif_name = difs;
         msg.n_dif_name = difs_size;
 
-        if (send_ipcp_msg(pid, &msg)) {
-                LOG_ERR("Failed to send message to daemon");
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
                 return -1;
         }
 
-        return 0;
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
 }
 
 int ipcp_bootstrap(pid_t pid,
                    struct dif_config * conf)
 {
         ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
 
         msg.code = IPCP_MSG_CODE__IPCP_BOOTSTRAP;
 
-        if (send_ipcp_msg(pid, &msg)) {
-                LOG_ERR("Failed to send message to daemon");
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
                 return -1;
         }
 
-        return 0;
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
 }
 
 int ipcp_enroll(pid_t pid,
-                char * dif_name,
                 char * member_name,
-                char ** n_1_difs,
-                ssize_t n_1_difs_size)
+                char * n_1_dif)
 {
         ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
 
-        if (n_1_difs == NULL ||
-            n_1_difs_size == 0 ||
-            n_1_difs[0] == NULL ||
-            dif_name == NULL ||
-            member_name == NULL)
+        if (n_1_dif == NULL || member_name == NULL)
                 return -EINVAL;
 
         msg.code = IPCP_MSG_CODE__IPCP_ENROLL;
@@ -241,17 +282,261 @@ int ipcp_enroll(pid_t pid,
                 LOG_ERR("Failed to malloc");
                 return -1;
         }
-        msg.dif_name[0] = dif_name;
+        msg.dif_name[0] = n_1_dif;
         msg.ap_name = member_name;
-        msg.n_1_dif_name = n_1_difs;
-        msg.n_n_1_dif_name = n_1_difs_size;
 
-        if (send_ipcp_msg(pid, &msg)) {
-                LOG_ERR("Failed to send message to daemon");
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL) {
                 free(msg.dif_name);
                 return -1;
         }
 
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                free(msg.dif_name);
+                return -1;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
         free(msg.dif_name);
-        return 0;
+
+        return ret;
+}
+
+int ipcp_ap_reg(pid_t    pid,
+                uint32_t reg_api_id,
+                char *   ap_name)
+{
+        ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
+
+        if (ap_name == NULL)
+                return -1;
+
+        msg.code = IPCP_MSG_CODE__IPCP_AP_REG;
+        msg.ap_name = ap_name;
+        msg.has_port_id = true;
+        msg.port_id = reg_api_id;
+
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                return -1;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
+int ipcp_ap_unreg(pid_t    pid,
+                  uint32_t reg_api_id)
+{
+        ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
+
+        msg.code = IPCP_MSG_CODE__IPCP_AP_UNREG;
+        msg.has_port_id = true;
+        msg.port_id = reg_api_id;
+
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                return -1;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
+int ipcp_flow_alloc(pid_t             pid,
+                    uint32_t          port_id,
+                    char *            dst_ap_name,
+                    char *            src_ap_name,
+                    char *            src_ae_name,
+                    struct qos_spec * qos)
+{
+        ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
+
+        if (dst_ap_name == NULL || src_ap_name == NULL || src_ae_name == NULL)
+                return -EINVAL;
+
+        msg.code = IPCP_MSG_CODE__IPCP_FLOW_ALLOC;
+        msg.ap_name = src_ap_name;
+        msg.ae_name = src_ae_name;
+        msg.dst_ap_name = dst_ap_name;
+        msg.has_port_id = true;
+        msg.port_id = port_id;
+
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                return -1;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
+int ipcp_flow_alloc_resp(pid_t    pid,
+                         uint32_t port_id,
+                         int      result)
+{
+        ipcp_msg_t msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg = NULL;
+        int ret = -1;
+
+        msg.code = IPCP_MSG_CODE__IPCP_FLOW_ALLOC_RESP;
+        msg.has_port_id = true;
+        msg.port_id = port_id;
+        msg.has_result = true;
+        msg.result = result;
+
+        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                return -1;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
+int ipcp_flow_req_arr(pid_t    pid,
+                      uint32_t reg_api_id,
+                      char *   ap_name,
+                      char *   ae_name)
+{
+        irm_msg_t msg = IRM_MSG__INIT;
+        irm_msg_t * recv_msg = NULL;
+        int fd = -1;
+
+        if (ap_name == NULL || ae_name == NULL)
+                return -EINVAL;
+
+        msg.code = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR;
+        msg.ap_name = ap_name;
+        msg.ae_name = ae_name;
+        msg.port_id = reg_api_id;
+        msg.has_port_id = true;
+        msg.pid = pid;
+        msg.has_pid = true;
+
+        recv_msg = send_recv_irm_msg(&msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_fd == false) {
+                irm_msg__free_unpacked(recv_msg, NULL);
+                return -1;
+        }
+
+        fd = recv_msg->fd;
+        irm_msg__free_unpacked(recv_msg, NULL);
+
+        return fd;
+}
+
+int ipcp_flow_alloc_reply(pid_t    pid,
+                          uint32_t port_id,
+                          int      result)
+{
+        irm_msg_t msg = IRM_MSG__INIT;
+        irm_msg_t * recv_msg = NULL;
+        int ret = -1;
+
+        msg.code = IRM_MSG_CODE__IPCP_FLOW_ALLOC_REPLY;
+        msg.port_id = port_id;
+        msg.has_port_id = true;
+        msg.result = result;
+        msg.has_result = true;
+
+        recv_msg = send_recv_irm_msg(&msg);
+        if (recv_msg == NULL)
+                return -1;
+
+        if (recv_msg->has_result == false) {
+                irm_msg__free_unpacked(recv_msg, NULL);
+                return -1;
+        }
+
+        ret = recv_msg->result;
+        irm_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
+}
+
+
+int ipcp_flow_dealloc(pid_t    pid,
+                      uint32_t port_id)
+{
+        if (pid != 0) {
+                ipcp_msg_t msg = IPCP_MSG__INIT;
+                ipcp_msg_t * recv_msg = NULL;
+                int ret = -1;
+
+                msg.code = IPCP_MSG_CODE__IPCP_FLOW_DEALLOC;
+                msg.has_port_id = true;
+                msg.port_id = port_id;
+
+                recv_msg = send_recv_ipcp_msg(pid, &msg);
+                if (recv_msg == NULL)
+                        return -1;
+
+                if (recv_msg->has_result == false) {
+                        ipcp_msg__free_unpacked(recv_msg, NULL);
+                        return -1;
+                }
+
+                ret = recv_msg->result;
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+
+                return ret;
+        } else {
+                irm_msg_t msg = IRM_MSG__INIT;
+                irm_msg_t * recv_msg = NULL;
+                int ret = -1;
+
+                msg.code = IRM_MSG_CODE__IPCP_FLOW_DEALLOC;
+                msg.has_port_id = true;
+                msg.port_id = port_id;
+
+                recv_msg = send_recv_irm_msg(&msg);
+                if (recv_msg == NULL)
+                        return -1;
+
+                if (recv_msg->has_result == false) {
+                        irm_msg__free_unpacked(recv_msg, NULL);
+                        return -1;
+                }
+
+                ret = recv_msg->result;
+                irm_msg__free_unpacked(recv_msg, NULL);
+
+                return ret;
+        }
 }
diff --git a/src/lib/ipcpd_messages.proto b/src/lib/ipcpd_messages.proto
index 0715fbe0..bcdd54ae 100644
--- a/src/lib/ipcpd_messages.proto
+++ b/src/lib/ipcpd_messages.proto
@@ -3,6 +3,12 @@ enum ipcp_msg_code {
         IPCP_ENROLL = 2;
         IPCP_REG = 3;
         IPCP_UNREG = 4;
+        IPCP_AP_REG = 5;
+        IPCP_AP_UNREG = 6;
+        IPCP_FLOW_ALLOC = 7;
+        IPCP_FLOW_ALLOC_RESP = 8;
+        IPCP_FLOW_DEALLOC = 9;
+        IPCP_REPLY = 10;
 };
 
 message ipcp_msg {
@@ -10,5 +16,8 @@ message ipcp_msg {
         optional string ap_name = 2;
         // Missing dif_config field here
         repeated string dif_name = 4;
-        repeated string n_1_dif_name = 5;
+        optional int32 result = 5;
+        optional uint32 port_id = 6;
+        optional string ae_name = 7;
+        optional string dst_ap_name = 8;
 };
diff --git a/src/lib/irmd_messages.proto b/src/lib/irmd_messages.proto
index a524a7fb..44070755 100644
--- a/src/lib/irmd_messages.proto
+++ b/src/lib/irmd_messages.proto
@@ -15,7 +15,10 @@ enum irm_msg_code {
         IRM_FLOW_CONTROL = 14;
         IRM_FLOW_WRITE = 15;
         IRM_FLOW_READ = 16;
-        IRM_REPLY = 17;
+        IPCP_FLOW_REQ_ARR = 17;
+        IPCP_FLOW_ALLOC_REPLY = 18;
+        IPCP_FLOW_DEALLOC = 19;
+        IRM_REPLY = 20;
 };
 
 message irm_msg {
@@ -31,4 +34,6 @@ message irm_msg {
         // Missing qos_spec here
         optional int32 oflags = 10;
         optional string dst_ap_name = 11;
+        optional uint32 port_id = 12;
+        optional int32 pid = 13;
 };