libcephfs_proxy: add the number of supported operations to negotiation

The v0 negotiation structure has been modified to hold the total number
of operations and callbacks supported by the peer. The changes are done in
a way that it's completely transparent and harmless for a peer expecting
the previous definition.

This will be useful to quickly check if the daemon supports some
operation, or the client supports some callback before sending them.

Signed-off-by: Xavi Hernandez <xhernandez@gmail.com>

diff --git a/src/libcephfs_proxy/libcephfs_proxy.c b/src/libcephfs_proxy/libcephfs_proxy.c
index c9ff2bff22617a3d03ed8312b3e2e16646ce6655..ae4355273911bc8cf48ef582d29e1bacf6f125c5 100644 (file)
--- a/src/libcephfs_proxy/libcephfs_proxy.c
+++ b/src/libcephfs_proxy/libcephfs_proxy.c
@@ -89,7 +89,7 @@ static int32_t proxy_global_connect(void)
                proxy_link_negotiate_init(&global_cmount.neg, 0, PROXY_FEAT_ALL,
                                          0,
                                          PROXY_FEAT_EMBEDDED_PERMS,
-                                         PROXY_LINK_PROTOCOL_VERSION);
+                                         PROXY_LINK_PROTOCOL_VERSION, 0, 0);
 
                err = proxy_link_handshake_client(&global_cmount.link, err,
                                                  &global_cmount.neg,
@@ -245,7 +245,9 @@ __public int ceph_create(struct ceph_mount_info **cmount, const char *const id)
        proxy_link_negotiate_init(&ceph_mount->neg, 0, PROXY_FEAT_ALL, 0,
                                  PROXY_FEAT_ASYNC_IO |
                                  PROXY_FEAT_EMBEDDED_PERMS,
-                                 PROXY_LINK_PROTOCOL_VERSION);
+                                 PROXY_LINK_PROTOCOL_VERSION, 0,
+                                 LIBCEPHFSD_CBK_TOTAL_OPS);
+
 
        err = proxy_link_handshake_client(&ceph_mount->link, sd,
                                          &ceph_mount->neg,

diff --git a/src/libcephfs_proxy/libcephfsd.c b/src/libcephfs_proxy/libcephfsd.c
index 04e731059b52d02296c37e1a2c74be19a17fe2f1..3ba6811db42b937b5a41ec6a67bec1ed305ec4ed 100644 (file)
--- a/src/libcephfs_proxy/libcephfsd.c
+++ b/src/libcephfs_proxy/libcephfsd.c
@@ -2306,7 +2306,8 @@ static void serve_connection(proxy_worker_t *worker)
        client = container_of(worker, proxy_client_t, worker);
 
        proxy_link_negotiate_init(&client->neg, 0, PROXY_FEAT_ALL, 0, 0,
-                                 PROXY_LINK_PROTOCOL_VERSION);
+                                 PROXY_LINK_PROTOCOL_VERSION,
+                                 LIBCEPHFSD_OP_TOTAL_OPS, 0);
 
        err = proxy_link_handshake_server(client->link, client->sd,
                                          &client->neg,

diff --git a/src/libcephfs_proxy/proxy_link.c b/src/libcephfs_proxy/proxy_link.c
index 2dbb24229600fb52ab704ba30bd39765f932808e..7e5ebd250c5adf0c6a6626f0d3374ab1e0676415 100644 (file)
--- a/src/libcephfs_proxy/proxy_link.c
+++ b/src/libcephfs_proxy/proxy_link.c
@@ -5,6 +5,7 @@
 
 #include "proxy_link.h"
 #include "proxy_manager.h"
+#include "proxy_requests.h"
 #include "proxy_helpers.h"
 #include "proxy_log.h"
 
@@ -236,8 +237,49 @@ int32_t proxy_link_ctrl_recv(int32_t sd, void *data, int32_t size, int32_t type,
        return len;
 }
 
+static void proxy_link_negotiate_v0_prepare(proxy_link_negotiate_v0_t *v0,
+                                           bool legacy)
+{
+       uint16_t version, min_version;
+
+       if (legacy) {
+               version = v0->legacy.version;
+               min_version = v0->legacy.min_version;
+               v0->version = version;
+               v0->min_version = min_version;
+
+               /* This comes from the client. It doesn't support any operation,
+                * just callbacks. */
+               v0->num_ops = 0;
+               if ((v0->flags & PROXY_NEG_V0_OPS) == 0) {
+                       v0->num_cbks = LIBCEPHFSD_CBK_LL_NONBLOCKING_FSYNC + 1;
+                       v0->flags = 0;
+               }
+       } else if ((v0->flags & PROXY_NEG_V0_OPS) == 0) {
+               version = v0->legacy.version;
+               min_version = v0->legacy.min_version;
+               v0->version = version;
+               v0->min_version = min_version;
+               v0->num_ops = LIBCEPHFSD_OP_LL_NONBLOCKING_FSYNC + 1;
+               /* This comes from the daemon. It doesn't support any callback,
+                * just operations. */
+               v0->num_cbks = 0;
+               v0->flags = 0;
+       }
+}
+
+static void proxy_link_negotiate_v0_legacy(proxy_link_negotiate_t *src,
+                                          proxy_link_negotiate_t *dst)
+{
+       memcpy(dst, src, sizeof(proxy_link_negotiate_t));
+
+       dst->v0.legacy.version = src->v0.version;
+       dst->v0.legacy.min_version = src->v0.min_version;
+}
+
 static int32_t proxy_link_negotiate_read(proxy_link_t *link, int32_t sd,
-                                        proxy_link_negotiate_t *neg)
+                                        proxy_link_negotiate_t *neg,
+                                        bool legacy)
 {
        char buffer[128];
        char *ptr;
@@ -253,6 +295,8 @@ static int32_t proxy_link_negotiate_read(proxy_link_t *link, int32_t sd,
                return err;
        }
 
+       proxy_link_negotiate_v0_prepare(&neg->v0, legacy);
+
        ptr += sizeof(neg->v0);
        size = neg->v0.size;
 
@@ -305,7 +349,8 @@ static int32_t proxy_link_negotiate_read(proxy_link_t *link, int32_t sd,
 
 static int32_t proxy_link_negotiate_check(proxy_link_negotiate_t *local,
                                          proxy_link_negotiate_t *remote,
-                                         proxy_link_negotiate_cbk_t cbk)
+                                         proxy_link_negotiate_cbk_t cbk,
+                                         bool client)
 {
        uint32_t supported, enabled;
        int32_t err;
@@ -315,6 +360,14 @@ static int32_t proxy_link_negotiate_check(proxy_link_negotiate_t *local,
                local->v0.size = remote->v0.size;
        }
 
+       if (client) {
+               local->v0.num_ops = remote->v0.num_ops;
+       } else {
+               local->v0.num_cbks = remote->v0.num_cbks;
+       }
+
+       local->v0.flags &= remote->v0.flags;
+
        if (remote->v0.version == 0) {
                /* Legacy peer. If we require any feature, the peer won't
                 * support it, so we can't continue */
@@ -390,20 +443,24 @@ static int32_t proxy_link_negotiate_client(proxy_link_t *link, int32_t sd,
                                           proxy_link_negotiate_t *neg,
                                           proxy_link_negotiate_cbk_t cbk)
 {
-       proxy_link_negotiate_t remote;
+       proxy_link_negotiate_t legacy, remote;
        int32_t err;
 
-       err = proxy_link_write(link, sd, neg, neg->v0.size);
+       /* Convert the negotiation structure to legacy for the first
+        * nogotiation packet. */
+       proxy_link_negotiate_v0_legacy(neg, &legacy);
+
+       err = proxy_link_write(link, sd, &legacy, legacy.v0.size);
        if (err < 0) {
                return err;
        }
 
-       err = proxy_link_negotiate_read(link, sd, &remote);
+       err = proxy_link_negotiate_read(link, sd, &remote, false);
        if (err < 0) {
                return err;
        }
 
-       err = proxy_link_negotiate_check(neg, &remote, cbk);
+       err = proxy_link_negotiate_check(neg, &remote, cbk, true);
        if (err < 0) {
                return err;
        }
@@ -440,20 +497,25 @@ static int32_t proxy_link_negotiate_server(proxy_link_t *link, int32_t sd,
                                           proxy_link_negotiate_t *neg,
                                           proxy_link_negotiate_cbk_t cbk)
 {
-       proxy_link_negotiate_t remote;
+       proxy_link_negotiate_t remote, legacy;
        int32_t err;
 
-       err = proxy_link_negotiate_read(link, sd, &remote);
+       err = proxy_link_negotiate_read(link, sd, &remote, true);
        if (err < 0) {
                return err;
        }
 
-       err = proxy_link_negotiate_check(neg, &remote, cbk);
+       err = proxy_link_negotiate_check(neg, &remote, cbk, false);
        if (err < 0) {
                return err;
        }
 
-       err = proxy_link_write(link, sd, neg, neg->v0.size);
+       if ((remote.v0.flags & PROXY_NEG_V0_OPS) == 0) {
+               proxy_link_negotiate_v0_legacy(neg, &legacy);
+               err = proxy_link_write(link, sd, &legacy, legacy.v0.size);
+       } else {
+               err = proxy_link_write(link, sd, neg, neg->v0.size);
+       }
        if (err < 0) {
                return err;
        }
@@ -557,9 +619,10 @@ int32_t proxy_link_handshake_client(proxy_link_t *link, int32_t sd,
 
        if (version.minor == LIBCEPHFSD_MINOR) {
                /* The server doesn't support negotiation. */
-               proxy_link_negotiate_init_v0(&legacy, 0, 0);
+               proxy_link_negotiate_init_v0(&legacy, 0, 0,
+                       LIBCEPHFSD_OP_LL_NONBLOCKING_FSYNC + 1, 0);
 
-               return proxy_link_negotiate_check(neg, &legacy, cbk);
+               return proxy_link_negotiate_check(neg, &legacy, cbk, true);
        }
 
        if (version.minor != LIBCEPHFSD_MINOR_NEG) {
@@ -612,9 +675,10 @@ int32_t proxy_link_handshake_server(proxy_link_t *link, int32_t sd,
        }
 
        if (size == 0) {
-               proxy_link_negotiate_init_v0(&legacy, 0, 0);
+               proxy_link_negotiate_init_v0(&legacy, 0, 0, 0,
+                       LIBCEPHFSD_CBK_LL_NONBLOCKING_FSYNC + 1);
 
-               err = proxy_link_negotiate_check(neg, &legacy, cbk);
+               err = proxy_link_negotiate_check(neg, &legacy, cbk, false);
                if (err < 0) {
                        return err;
                }

diff --git a/src/libcephfs_proxy/proxy_link.h b/src/libcephfs_proxy/proxy_link.h
index 2dd851c7f1577e38a9776896bb2949b7434c452d..c570c07c68159976636ec6cd95d92a8e0fd5af10 100644 (file)
--- a/src/libcephfs_proxy/proxy_link.h
+++ b/src/libcephfs_proxy/proxy_link.h
@@ -4,6 +4,7 @@
 
 #include <sys/socket.h>
 #include <sys/un.h>
+#include <endian.h>
 
 #include "proxy.h"
 
@@ -22,6 +23,12 @@
 /* The maximum supported protocol version. */
 #define PROXY_LINK_PROTOCOL_VERSION PROXY_PROTOCOL_V1
 
+/* Flags for the proxy_link_negotiate_v0_t structure. */
+enum {
+       /* If set, the v0 structure is using the new layout. */
+       PROXY_NEG_V0_OPS = 1
+};
+
 /* Version 0 structure will be used to handle legacy clients that don't support
  * negotiation. */
 typedef struct _proxy_link_negotiate_v0 {
@@ -29,14 +36,38 @@ typedef struct _proxy_link_negotiate_v0 {
         * NEG_VERSION_SIZE() to avoid alignement issues. */
        uint16_t size;
 
-       /* Version of the negotiation structure. */
-       uint16_t version;
-
-       /* Minimum version that the peer needs to support to proceed. */
-       uint16_t min_version;
-
-       /* Reserved. Must be 0. */
-       uint16_t flags;
+       /* We keep the previous structure definition as 'legacy' for backward
+        * compatibility. It will be used in the initial negotiation request
+        * sent by the client. The flag PROXY_NEG_V0_OPS indicates if the
+        * new layout is supported. More details in proxy_link_negotiate_*
+        * functions. */
+       union {
+               struct {
+                       /* Version of the negotiation structure. */
+                       uint16_t version;
+
+                       /* Minimum version that the peer needs to support to
+                        * proceed. */
+                       uint16_t min_version;
+               } legacy;
+               struct {
+                       /* Version of the negotiation structure. */
+                       uint8_t version;
+
+                       /* Minimum version that the peer needs to support to
+                        * proceed. */
+                       uint8_t min_version;
+
+                       /* Total number of operations supported. */
+                       uint16_t num_ops;
+               };
+       };
+
+       /* Total number of callbacks supported. */
+       uint8_t num_cbks;
+
+       /* Flags of the v0 structure. */
+       uint8_t flags;
 } proxy_link_negotiate_v0_t;
 
 typedef struct _proxy_link_negotiate_v1 {
@@ -84,24 +115,31 @@ typedef struct _proxy_link_ans {
         sizeof(proxy_link_negotiate_v##_ver##_t))
 #define NEG_VERSION_SIZE(_ver) NEG_VERSION_SIZE_1(_ver)
 
-#define proxy_link_negotiate_init_v0(_neg, _ver, _min) \
+#define proxy_link_negotiate_init_v0(_neg, _ver, _min, _ops, _cbks) \
        do { \
                (_neg)->v0.size = NEG_VERSION_SIZE(_ver); \
                (_neg)->v0.version = (_ver); \
                (_neg)->v0.min_version = (_min); \
-               (_neg)->v0.flags = 0; \
+               (_neg)->v0.num_ops = (_ops); \
+               (_neg)->v0.num_cbks = (_cbks); \
+               (_neg)->v0.flags = PROXY_NEG_V0_OPS; \
        } while (0)
 
+#define proxy_op_supported(_neg, _op) ((_neg)->v0.num_ops > (_op))
+#define proxy_cbk_supported(_neg, _cbk) ((_neg)->v0.num_cbks > (_cbk))
+
 /* NEG_VERSION: Add new arguments and initialize the link->neg.vX with them. */
 static inline void proxy_link_negotiate_init(proxy_link_negotiate_t *neg,
                                             uint32_t min_version,
                                             uint32_t supported,
                                             uint32_t required,
                                             uint32_t enabled,
-                                            uint32_t protocol)
+                                            uint32_t protocol,
+                                            uint32_t num_ops,
+                                            uint32_t num_cbks)
 {
        proxy_link_negotiate_init_v0(neg, PROXY_LINK_NEGOTIATE_VERSION,
-                                    min_version);
+                                    min_version, num_ops, num_cbks);
 
        neg->v1.supported = supported;
        neg->v1.required = required;