include: Add a flow_join operation for broadcast

This adds a new flow_join operaiton for broadcast, which is a much
safer solution than overloading destination name semantics.  The
internal API now also has a different IPCP_FLOW_JOIN operation. The
IRMd doesn't need to query broadcasts IPCPs for the name, it can just
check if an IPCP with the layer name exists. The broadcast IPCP
doesn't need to implement the query proxy call anymore.

Signed-off-by: Dimitri Staessens <dimitri.staessens@ugent.be>
Signed-off-by: Sander Vrijders <sander.vrijders@ugent.be>

3 files changed, 85 insertions, 16 deletions

diff --git a/src/irmd/ipcp.c b/src/irmd/ipcp.c
index 7f3f4807..08547d01 100644
--- a/src/irmd/ipcp.c
+++ b/src/irmd/ipcp.c
@@ -435,12 +435,13 @@ int ipcp_query(pid_t           pid,
         return ret;
 }
 
-int ipcp_flow_alloc(pid_t           pid,
-                    int             flow_id,
-                    pid_t           n_pid,
-                    const uint8_t * dst,
-                    size_t          len,
-                    qosspec_t       qs)
+static int __ipcp_flow_alloc(pid_t           pid,
+                             int             flow_id,
+                             pid_t           n_pid,
+                             const uint8_t * dst,
+                             size_t          len,
+                             qosspec_t       qs,
+                             bool            join)
 {
         ipcp_msg_t    msg      = IPCP_MSG__INIT;
         qosspec_msg_t qs_msg;
@@ -449,7 +450,10 @@ int ipcp_flow_alloc(pid_t           pid,
 
         assert(dst);
 
-        msg.code         = IPCP_MSG_CODE__IPCP_FLOW_ALLOC;
+        if (join)
+                msg.code         = IPCP_MSG_CODE__IPCP_FLOW_JOIN;
+        else
+                msg.code         = IPCP_MSG_CODE__IPCP_FLOW_ALLOC;
         msg.has_flow_id  = true;
         msg.flow_id      = flow_id;
         msg.has_pid      = true;
@@ -475,6 +479,26 @@ int ipcp_flow_alloc(pid_t           pid,
         return ret;
 }
 
+int ipcp_flow_alloc(pid_t           pid,
+                    int             flow_id,
+                    pid_t           n_pid,
+                    const uint8_t * dst,
+                    size_t          len,
+                    qosspec_t       qs)
+{
+        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst, len, qs, false);
+}
+
+int ipcp_flow_join(pid_t           pid,
+                   int             flow_id,
+                   pid_t           n_pid,
+                   const uint8_t * dst,
+                   size_t          len,
+                   qosspec_t       qs)
+{
+        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst, len, qs, true);
+}
+
 int ipcp_flow_alloc_resp(pid_t pid,
                          int   flow_id,
                          pid_t n_pid,
diff --git a/src/irmd/ipcp.h b/src/irmd/ipcp.h
index 07b9c44a..611bada2 100644
--- a/src/irmd/ipcp.h
+++ b/src/irmd/ipcp.h
@@ -69,6 +69,13 @@ int   ipcp_flow_alloc(pid_t           pid,
                       size_t          len,
                       qosspec_t       qs);
 
+int   ipcp_flow_join(pid_t           pid,
+                     int             flow_id,
+                     pid_t           n_pid,
+                     const uint8_t * dst,
+                     size_t          len,
+                     qosspec_t       qs);
+
 int   ipcp_flow_alloc_resp(pid_t pid,
                            int   flow_id,
                            pid_t n_pid,
diff --git a/src/irmd/main.c b/src/irmd/main.c
index 67e16de0..802b01f0 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -330,6 +330,19 @@ static struct ipcp_entry * get_ipcp_entry_by_name(const char * name)
         return NULL;
 }
 
+static struct ipcp_entry * get_ipcp_entry_by_layer(const char * layer)
+{
+        struct list_head * p;
+
+        list_for_each(p, &irmd.ipcps) {
+                struct ipcp_entry * e = list_entry(p, struct ipcp_entry, next);
+                if (strcmp(layer, e->layer) == 0)
+                        return e;
+        }
+
+        return NULL;
+}
+
 static struct ipcp_entry * get_ipcp_by_dst_name(const char * name,
                                                 pid_t        src)
 {
@@ -1267,7 +1280,8 @@ static int flow_alloc(pid_t              pid,
                       const char *       dst,
                       qosspec_t          qs,
                       struct timespec *  timeo,
-                      struct irm_flow ** e)
+                      struct irm_flow ** e,
+                      bool               join)
 {
         struct irm_flow *   f;
         struct ipcp_entry * ipcp;
@@ -1275,7 +1289,10 @@ static int flow_alloc(pid_t              pid,
         int                 state;
         uint8_t *           hash;
 
-        ipcp = get_ipcp_by_dst_name(dst, pid);
+        if (join)
+                ipcp = get_ipcp_entry_by_layer(dst);
+        else
+                ipcp = get_ipcp_by_dst_name(dst, pid);
         if (ipcp == NULL) {
                 log_info("Destination %s unreachable.", dst);
                 return -1;
@@ -1310,12 +1327,22 @@ static int flow_alloc(pid_t              pid,
 
         str_hash(ipcp->dir_hash_algo, hash, dst);
 
-        if (ipcp_flow_alloc(ipcp->pid, flow_id, pid, hash,
-                            IPCP_HASH_LEN(ipcp), qs)) {
-                /* sanitizer cleans this */
-                log_info("Flow_allocation failed.");
-                free(hash);
-                return -EAGAIN;
+        if (join) {
+                if (ipcp_flow_join(ipcp->pid, flow_id, pid, hash,
+                                   IPCP_HASH_LEN(ipcp), qs)) {
+                        /* sanitizer cleans this */
+                        log_info("Flow_join failed.");
+                        free(hash);
+                        return -EAGAIN;
+                }
+        } else {
+                if (ipcp_flow_alloc(ipcp->pid, flow_id, pid, hash,
+                                    IPCP_HASH_LEN(ipcp), qs)) {
+                        /* sanitizer cleans this */
+                        log_info("Flow_allocation failed.");
+                        free(hash);
+                        return -EAGAIN;
+                }
         }
 
         free(hash);
@@ -1978,7 +2005,18 @@ static void * mainloop(void * o)
                 case IRM_MSG_CODE__IRM_FLOW_ALLOC:
                         result = flow_alloc(msg->pid, msg->dst,
                                             msg_to_spec(msg->qosspec),
-                                            timeo, &e);
+                                            timeo, &e, false);
+                        if (result == 0) {
+                                ret_msg->has_flow_id = true;
+                                ret_msg->flow_id     = e->flow_id;
+                                ret_msg->has_pid     = true;
+                                ret_msg->pid         = e->n_1_pid;
+                        }
+                        break;
+                case IRM_MSG_CODE__IRM_FLOW_JOIN:
+                        result = flow_alloc(msg->pid, msg->dst,
+                                            msg_to_spec(msg->qosspec),
+                                            timeo, &e, true);
                         if (result == 0) {
                                 ret_msg->has_flow_id = true;
                                 ret_msg->flow_id     = e->flow_id;