sock/posix: Add flush

Initiator drivers (e.g nvme/tcp) don't use poll groups but rather directly
poll the qpair. In this case we want to allow the polling function (e.g
_qpair_process_completions()) to flush async writes pending on the socket.

Signed-off-by: Or Gerlitz <ogerlitz@mellanox.com>
Change-Id: Ibd8c73691213d58e287b7110d0f5a381a89a64d0
Reviewed-on: https://review.gerrithub.io/c/spdk/spdk/+/475419
Reviewed-by: Ben Walker <benjamin.walker@intel.com>
Reviewed-by: Jim Harris <james.r.harris@intel.com>
Reviewed-by: Shuhei Matsumoto <shuhei.matsumoto.xt@hitachi.com>
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>

include/spdk/sock.h

@@ -146,6 +146,15 @@ struct spdk_sock *spdk_sock_accept(struct spdk_sock *sock);
  */
 int spdk_sock_close(struct spdk_sock **sock);
 
+/**
+ * Flush a socket from data gathered in previous writev_async calls.
+ *
+ * \param sock Socket to flush.
+ *
+ * \return 0 on success, -1 on failure.
+ */
+int spdk_sock_flush(struct spdk_sock *sock);
+
 /**
  * Receive a message from the given socket.
  *

include/spdk_internal/sock.h

@@ -94,6 +94,7 @@ struct spdk_net_impl {
 	ssize_t (*writev)(struct spdk_sock *sock, struct iovec *iov, int iovcnt);
 
 	void (*writev_async)(struct spdk_sock *sock, struct spdk_sock_request *req);
+	int (*flush)(struct spdk_sock *sock);
 
 	int (*set_recvlowat)(struct spdk_sock *sock, int nbytes);
 	int (*set_recvbuf)(struct spdk_sock *sock, int sz);

lib/sock/sock.c

@@ -329,6 +329,12 @@ spdk_sock_writev_async(struct spdk_sock *sock, struct spdk_sock_request *req)
 	sock->net_impl->writev_async(sock, req);
 }
 
+int
+spdk_sock_flush(struct spdk_sock *sock)
+{
+	return sock->net_impl->flush(sock);
+}
+
 int
 spdk_sock_set_recvlowat(struct spdk_sock *sock, int nbytes)
 {

module/sock/posix/posix.c

@@ -545,6 +545,12 @@ _sock_flush(struct spdk_sock *sock)
 	return 0;
 }
 
+static int
+spdk_posix_sock_flush(struct spdk_sock *_sock)
+{
+	return _sock_flush(_sock);
+}
+
 static ssize_t
 spdk_posix_sock_recv(struct spdk_sock *_sock, void *buf, size_t len)
 {
@@ -870,6 +876,7 @@ static struct spdk_net_impl g_posix_net_impl = {
 	.readv		= spdk_posix_sock_readv,
 	.writev		= spdk_posix_sock_writev,
 	.writev_async	= spdk_posix_sock_writev_async,
+	.flush		= spdk_posix_sock_flush,
 	.set_recvlowat	= spdk_posix_sock_set_recvlowat,
 	.set_recvbuf	= spdk_posix_sock_set_recvbuf,
 	.set_sendbuf	= spdk_posix_sock_set_sendbuf,

test/common/lib/test_sock.c

@@ -51,6 +51,7 @@ DEFINE_STUB(spdk_sock_set_recvlowat, int, (struct spdk_sock *sock, int nbytes),
 DEFINE_STUB(spdk_sock_set_recvbuf, int, (struct spdk_sock *sock, int sz), 0);
 DEFINE_STUB(spdk_sock_set_sendbuf, int, (struct spdk_sock *sock, int sz), 0);
 DEFINE_STUB_V(spdk_sock_writev_async, (struct spdk_sock *sock, struct spdk_sock_request *req));
+DEFINE_STUB(spdk_sock_flush, int, (struct spdk_sock *sock), 0);
 DEFINE_STUB(spdk_sock_is_ipv6, bool, (struct spdk_sock *sock), false);
 DEFINE_STUB(spdk_sock_is_ipv4, bool, (struct spdk_sock *sock), true);
 DEFINE_STUB(spdk_sock_is_connected, bool, (struct spdk_sock *sock), true);