Merge PR ceph#61877 into main

* refs/pull/61877/head:
	libcephfs_proxy: implement client side async rw operation
	libcephfs_proxy: implement server side async rw operation
	libcephfs_proxy: implement the async rw callback
	libcephfs_proxy: define async rw opcode and structures
	libcephfs_proxy: define some helper macros for callbacks
	libcephfs_proxy: move pointer obfuscation functions to proxy_helpers.h
	libcephfs_proxy: negotiate and use the async cbk feature
	libcephfs_proxy: implement asynchronous callbacks
	libcephfs_proxy: gracefully handle connection close
	libcephfs_proxy: add negotiation documentation
	libcephfs_proxy: replace legacy handshake by negotation in server side
	libcephfs_proxy: replace legacy handshake by negotation in client side
	libcephfs_proxy: implement server side negotiation
	libcephfs_proxy: implement client side negotiation
	libcephfs_proxy: implement receiving of negotiate structure
	libcephfs_proxy: add negotiation structures
	libcephfs_proxy: add support for control messages
	libcephfs_proxy: reuse proxy_link_{read|write}

Reviewed-by: Sachin Prabhu <[email protected]>
Reviewed-by: Venky Shankar <[email protected]>

Author: vshankar

doc/dev/libcephfs_proxy.rst

@@ -70,6 +70,30 @@ overhead for encoding and decoding, and would provide an easy way to support
 backward compatibility if the network protocol needs to be modified in the
 future.
 
+**Feature negotiation**
+
+During the initial connection through the UNIX socket, the client will
+initiate a negotiation of the features that it want to enable. Currently it
+supports a bitmap of features supported/required/desired, but it allows the
+structure containing the negotiated data to be extended by adding more fields
+without breaking backward compatibility.
+
+The structure itself contains the version and size of the structure, as well as
+the minimum required version supported, allowing the other peer to read the
+transmitted data even if it doesn't fully understand its contents, and adjust
+to the highest possible version known by both peers.
+
+The client will send the bitmap of supported features, the bitmap of features
+that must be available (otherwise the connection cannot proceed), and the
+bitmap of features the client would like to enable. When the server receives
+it, it will create a bitmap of the supported features on both sides, and verify
+that all required features by the server are supported by the client. It will
+also add its desired features to the features desired by the client. This data
+will be sent back to the client, who will do a final check and decide the
+final set of enabled features. Once this feature set is sent to the server,
+both peers can start using the enabled features.
+
+
 Design of the *libcephfs_proxy.so* library
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

src/libcephfs_proxy/CMakeLists.txt

@@ -1,4 +1,4 @@
-set(proxy_common_srcs proxy_link.c proxy_log.c)
+set(proxy_common_srcs proxy_link.c proxy_async.c proxy_log.c)
 set(libcephfsd_srcs libcephfsd.c proxy_manager.c proxy_mount.c proxy_helpers.c ${proxy_common_srcs})
 set(libcephfs_proxy_srcs libcephfs_proxy.c ${proxy_common_srcs})
 

src/libcephfs_proxy/libcephfs_proxy.c

@@ -6,29 +6,42 @@
 #include "proxy_log.h"
 #include "proxy_helpers.h"
 #include "proxy_requests.h"
+#include "proxy_async.h"
 
 /* We override the definition of the ceph_mount_info structure to contain
  * internal proxy information. This is already a black box for libcephfs users,
  * so this won't be noticed. */
 struct ceph_mount_info {
 	proxy_link_t link;
+	proxy_link_negotiate_t neg;
+	proxy_async_t async;
 	uint64_t cmount;
 };
 
 /* The global_cmount is used to stablish an initial connection to serve requests
  * not related to a real cmount, like ceph_version or ceph_userperm_new. */
-static struct ceph_mount_info global_cmount = { PROXY_LINK_DISCONNECTED, 0 };
+static struct ceph_mount_info global_cmount = {
+	.link = PROXY_LINK_DISCONNECTED,
+	.neg = {},
+	.cmount = 0
+};
 
 static bool client_stop(proxy_link_t *link)
 {
 	return false;
 }
 
+static int32_t proxy_negotiation_check(proxy_link_negotiate_t *neg)
+{
+	proxy_log(LOG_INFO, 0, "Features enabled: %08x", neg->v1.enabled);
+
+	return 0;
+}
+
 static int32_t proxy_connect(proxy_link_t *link)
 {
-	CEPH_REQ(hello, req, 0, ans, 0);
 	char *path, *env;
-	int32_t sd, err;
+	int32_t sd;
 
 	path = PROXY_SOCKET;
 	env = getenv(PROXY_SOCKET_ENV);
@@ -41,31 +54,7 @@ static int32_t proxy_connect(proxy_link_t *link)
 		return sd;
 	}
 
-	req.id = LIBCEPHFS_LIB_CLIENT;
-	err = proxy_link_send(sd, req_iov, 1);
-	if (err < 0) {
-		goto failed;
-	}
-	err = proxy_link_recv(sd, ans_iov, 1);
-	if (err < 0) {
-		goto failed;
-	}
-
-	proxy_log(LOG_INFO, 0, "Connected to libcephfsd version %d.%d",
-		  ans.major, ans.minor);
-
-	if ((ans.major != LIBCEPHFSD_MAJOR) ||
-	    (ans.minor != LIBCEPHFSD_MINOR)) {
-		err = proxy_log(LOG_ERR, ENOTSUP, "Version not supported");
-		goto failed;
-	}
-
 	return sd;
-
-failed:
-	proxy_link_close(link);
-
-	return err;
 }
 
 static void proxy_disconnect(proxy_link_t *link)
@@ -81,6 +70,19 @@ static int32_t proxy_global_connect(void)
 
 	if (!proxy_link_is_connected(&global_cmount.link)) {
 		err = proxy_connect(&global_cmount.link);
+		if (err < 0) {
+			return err;
+		}
+
+		proxy_link_negotiate_init(&global_cmount.neg, 0, PROXY_FEAT_ALL,
+					  0, 0);
+
+		err = proxy_link_handshake_client(&global_cmount.link, err,
+						  &global_cmount.neg,
+						  proxy_negotiation_check);
+		if (err < 0) {
+			proxy_disconnect(&global_cmount.link);
+		}
 	}
 
 	return err;
@@ -178,6 +180,24 @@ __public int ceph_create(struct ceph_mount_info **cmount, const char *const id)
 	}
 	sd = err;
 
+	proxy_link_negotiate_init(&ceph_mount->neg, 0, PROXY_FEAT_ALL, 0,
+				  PROXY_FEAT_ASYNC_IO);
+
+	err = proxy_link_handshake_client(&ceph_mount->link, sd,
+					  &ceph_mount->neg,
+					  proxy_negotiation_check);
+	if (err < 0) {
+		goto failed_link;
+	}
+
+	if ((ceph_mount->neg.v1.enabled & PROXY_FEAT_ASYNC_CBK) != 0) {
+		err = proxy_async_client(&ceph_mount->async, &ceph_mount->link,
+					 sd);
+		if (err < 0) {
+			goto failed_link;
+		}
+	}
+
 	CEPH_STR_ADD(req, id, id);
 
 	err = CEPH_CALL(sd, LIBCEPHFSD_OP_CREATE, req, ans);
@@ -891,3 +911,38 @@ __public UserPerm *ceph_mount_perms(struct ceph_mount_info *cmount)
 
 	return value_ptr(ans.userperm);
 }
+
+__public int64_t ceph_ll_nonblocking_readv_writev(
+	struct ceph_mount_info *cmount, struct ceph_ll_io_info *io_info)
+{
+	CEPH_REQ(ceph_ll_nonblocking_readv_writev, req,
+		 io_info->write ? io_info->iovcnt : 0, ans, 0);
+	int32_t i, err;
+
+	if ((cmount->neg.v1.enabled & PROXY_FEAT_ASYNC_IO) == 0) {
+		return -EOPNOTSUPP;
+	}
+
+	req.info = ptr_checksum(&cmount->async.random, io_info);
+	req.fh = (uintptr_t)io_info->fh;
+	req.off = io_info->off;
+	req.size = 0;
+	req.write = io_info->write;
+	req.fsync = io_info->fsync;
+	req.syncdataonly = io_info->syncdataonly;
+
+	for (i = 0; i < io_info->iovcnt; i++) {
+		if (io_info->write) {
+			CEPH_BUFF_ADD(req, io_info->iov[i].iov_base,
+				      io_info->iov[i].iov_len);
+		}
+		req.size += io_info->iov[i].iov_len;
+	}
+
+	err = CEPH_PROCESS(cmount, LIBCEPHFSD_OP_LL_NONBLOCKING_RW, req, ans);
+	if (err < 0) {
+		return err;
+	}
+
+	return ans.res;
+}