[PATCH 4/5] session: introduce session_owner

[Vladislav Shpilevoy <v.shpilevoy@tarantool.org>]

Session owner stores a session type specific data. For example,
IProto session has authentication salt, console session has
file descriptor.

For #2677 session owner of IProto and console will have push()
virtual function to do box.session.push, which implementation
depends on a session type.

Needed for #2677

Signed-off-by: Vladislav Shpilevoy <v.shpilevoy@tarantool.org>
---
 src/box/applier.cc        |  4 ++-
 src/box/authentication.cc |  4 +--
 src/box/authentication.h  |  3 +-
 src/box/box.cc            |  4 +--
 src/box/box.h             |  2 +-
 src/box/iproto.cc         | 86 +++++++++++++++++++++++++++++++++++++++------
 src/box/lua/session.c     | 73 +++++++++++++++++++++++++++++++++-----
 src/box/session.cc        | 72 +++++++++++++++++++++++++++++++++-----
 src/box/session.h         | 89 ++++++++++++++++++++++++++++++++++++++++-------
 src/box/vinyl.c           |  3 +-
 10 files changed, 293 insertions(+), 47 deletions(-)

diff --git a/src/box/applier.cc b/src/box/applier.cc
index 6bfe5a99a..581139509 100644
--- a/src/box/applier.cc
+++ b/src/box/applier.cc
@@ -533,7 +533,9 @@ applier_f(va_list ap)
 	 * Set correct session type for use in on_replace()
 	 * triggers.
 	 */
-	current_session()->type = SESSION_TYPE_APPLIER;
+	struct session_owner applier_owner;
+	session_owner_create(&applier_owner, SESSION_TYPE_APPLIER);
+	session_set_owner(current_session(), &applier_owner);
 
 	/* Re-connect loop */
 	while (!fiber_is_cancelled()) {
diff --git a/src/box/authentication.cc b/src/box/authentication.cc
index fef549c55..811974cb9 100644
--- a/src/box/authentication.cc
+++ b/src/box/authentication.cc
@@ -37,7 +37,7 @@
 static char zero_hash[SCRAMBLE_SIZE];
 
 void
-authenticate(const char *user_name, uint32_t len,
+authenticate(const char *user_name, uint32_t len, const char *salt,
 	     const char *tuple)
 {
 	struct user *user = user_find_by_name_xc(user_name, len);
@@ -84,7 +84,7 @@ authenticate(const char *user_name, uint32_t len,
 			   "invalid scramble size");
 	}
 
-	if (scramble_check(scramble, session->salt, user->def->hash2)) {
+	if (scramble_check(scramble, salt, user->def->hash2)) {
 		auth_res.is_authenticated = false;
 		if (session_run_on_auth_triggers(&auth_res) != 0)
 			diag_raise();
diff --git a/src/box/authentication.h b/src/box/authentication.h
index e91fe0a0e..9935e3548 100644
--- a/src/box/authentication.h
+++ b/src/box/authentication.h
@@ -45,6 +45,7 @@ struct on_auth_trigger_ctx {
 
 
 void
-authenticate(const char *user_name, uint32_t len, const char *tuple);
+authenticate(const char *user_name, uint32_t len, const char *salt,
+	     const char *tuple);
 
 #endif /* INCLUDES_TARANTOOL_BOX_AUTHENTICATION_H */
diff --git a/src/box/box.cc b/src/box/box.cc
index cb3199624..9fb85c04f 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -1233,7 +1233,7 @@ box_on_join(const tt_uuid *instance_uuid)
 }
 
 void
-box_process_auth(struct auth_request *request)
+box_process_auth(struct auth_request *request, const char *salt)
 {
 	rmean_collect(rmean_box, IPROTO_AUTH, 1);
 
@@ -1243,7 +1243,7 @@ box_process_auth(struct auth_request *request)
 
 	const char *user = request->user_name;
 	uint32_t len = mp_decode_strl(&user);
-	authenticate(user, len, request->scramble);
+	authenticate(user, len, salt, request->scramble);
 }
 
 void
diff --git a/src/box/box.h b/src/box/box.h
index c9b5aad01..84899cc13 100644
--- a/src/box/box.h
+++ b/src/box/box.h
@@ -150,7 +150,7 @@ box_reset_stat(void);
 } /* extern "C" */
 
 void
-box_process_auth(struct auth_request *request);
+box_process_auth(struct auth_request *request, const char *salt);
 
 void
 box_process_join(struct ev_io *io, struct xrow_header *header);
diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index a75127a90..ad4b1a757 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -59,6 +59,63 @@
 #include "replication.h" /* instance_uuid */
 #include "iproto_constants.h"
 #include "rmean.h"
+#include "random.h"
+
+enum { IPROTO_SALT_SIZE = 32 };
+
+/** Owner of binary IProto sessions. */
+struct iproto_session_owner {
+	struct session_owner base;
+	/** Authentication salt. */
+	char salt[IPROTO_SALT_SIZE];
+	/** IProto connection. */
+	struct iproto_connection *connection;
+};
+
+static struct session_owner *
+iproto_session_owner_dup(struct session_owner *owner);
+
+static int
+iproto_session_owner_fd(const struct session_owner *owner);
+
+static const struct session_owner_vtab iproto_session_owner_vtab = {
+	/* .dup = */ iproto_session_owner_dup,
+	/* .delete = */ (void (*)(struct session_owner *)) free,
+	/* .fd = */ iproto_session_owner_fd,
+};
+
+static struct session_owner *
+iproto_session_owner_dup(struct session_owner *owner)
+{
+	assert(owner->vtab == &iproto_session_owner_vtab);
+	size_t size = sizeof(struct iproto_session_owner);
+	struct session_owner *dup = (struct session_owner *) malloc(size);
+	if (dup == NULL) {
+		diag_set(OutOfMemory, size, "malloc", "iproto_session_owner");
+		return NULL;
+	}
+	memcpy(dup, owner, size);
+	return dup;
+}
+
+static inline void
+iproto_session_owner_create(struct iproto_session_owner *owner,
+			    struct iproto_connection *connection)
+{
+	owner->base.type = SESSION_TYPE_BINARY;
+	owner->base.vtab = &iproto_session_owner_vtab;
+	owner->connection = connection;
+	random_bytes(owner->salt, IPROTO_SALT_SIZE);
+}
+
+static inline char *
+iproto_session_salt(struct session *session)
+{
+	struct iproto_session_owner *session_owner =
+		(struct iproto_session_owner *) session->owner;
+	assert(session_owner->base.vtab == &iproto_session_owner_vtab);
+	return session_owner->salt;
+}
 
 /* The number of iproto messages in flight */
 enum { IPROTO_MSG_MAX = 768 };
@@ -368,6 +425,15 @@ struct iproto_connection
 static struct mempool iproto_connection_pool;
 static RLIST_HEAD(stopped_connections);
 
+static int
+iproto_session_owner_fd(const struct session_owner *owner)
+{
+	assert(owner->vtab == &iproto_session_owner_vtab);
+	const struct iproto_session_owner *session_owner =
+		((const struct iproto_session_owner *) owner);
+	return session_owner->connection->input.fd;
+}
+
 /**
  * Return true if we have not enough spare messages
  * in the message pool. Disconnect messages are
@@ -1033,11 +1099,6 @@ tx_process_disconnect(struct cmsg *m)
 	struct iproto_connection *con = msg->connection;
 	if (con->session) {
 		tx_fiber_init(con->session, 0);
-		/*
-		 * The socket is already closed in iproto thread,
-		 * prevent box.session.peer() from using it.
-		 */
-		con->session->fd = -1;
 		if (! rlist_empty(&session_on_disconnect))
 			session_run_on_disconnect_triggers(con->session);
 		session_destroy(con->session);
@@ -1328,8 +1389,9 @@ tx_process_misc(struct cmsg *m)
 {
 	struct iproto_msg *msg = tx_accept_msg(m);
 	struct obuf *out = msg->connection->tx.p_obuf;
+	struct session *session = msg->connection->session;
 
-	tx_fiber_init(msg->connection->session, msg->header.sync);
+	tx_fiber_init(session, msg->header.sync);
 
 	if (tx_check_schema(msg->header.schema_version))
 		goto error;
@@ -1337,7 +1399,8 @@ tx_process_misc(struct cmsg *m)
 	try {
 		switch (msg->header.type) {
 		case IPROTO_AUTH:
-			box_process_auth(&msg->auth);
+			box_process_auth(&msg->auth,
+					 iproto_session_salt(session));
 			iproto_reply_ok_xc(out, msg->header.sync,
 					   ::schema_version);
 			break;
@@ -1481,15 +1544,18 @@ tx_process_connect(struct cmsg *m)
 	struct iproto_connection *con = msg->connection;
 	struct obuf *out = msg->connection->tx.p_obuf;
 	try {              /* connect. */
-		con->session = session_create(con->input.fd, SESSION_TYPE_BINARY);
+		struct iproto_session_owner owner;
+		iproto_session_owner_create(&owner, con);
+		con->session = session_create((struct session_owner *) &owner);
 		if (con->session == NULL)
 			diag_raise();
 		tx_fiber_init(con->session, 0);
 		static __thread char greeting[IPROTO_GREETING_SIZE];
 		/* TODO: dirty read from tx thread */
 		struct tt_uuid uuid = INSTANCE_UUID;
-		greeting_encode(greeting, tarantool_version_id(),
-				&uuid, con->session->salt, SESSION_SEED_SIZE);
+		greeting_encode(greeting, tarantool_version_id(), &uuid,
+				iproto_session_salt(con->session),
+				IPROTO_SALT_SIZE);
 		obuf_dup_xc(out, greeting, IPROTO_GREETING_SIZE);
 		if (! rlist_empty(&session_on_connect)) {
 			if (session_run_on_connect_triggers(con->session) != 0)
diff --git a/src/box/lua/session.c b/src/box/lua/session.c
index d8e91bf1f..af8411068 100644
--- a/src/box/lua/session.c
+++ b/src/box/lua/session.c
@@ -42,6 +42,54 @@
 #include "box/user.h"
 #include "box/schema.h"
 
+/** Owner of a console session. */
+struct console_session_owner {
+	struct session_owner base;
+	/** Console socket descriptor. Expects text data. */
+	int fd;
+};
+
+static struct session_owner *
+console_session_owner_dup(struct session_owner *owner);
+
+static int
+console_session_owner_fd(const struct session_owner *owner);
+
+static const struct session_owner_vtab console_session_owner_vtab = {
+	/* .dup = */ console_session_owner_dup,
+	/* .delete = */ (void (*)(struct session_owner *)) free,
+	/* .fd = */ console_session_owner_fd,
+};
+
+static struct session_owner *
+console_session_owner_dup(struct session_owner *owner)
+{
+	assert(owner->vtab == &console_session_owner_vtab);
+	size_t size = sizeof(struct console_session_owner);
+	struct session_owner *dup = (struct session_owner *) malloc(size);
+	if (dup == NULL) {
+		diag_set(OutOfMemory, size, "malloc", "console_session_owner");
+		return NULL;
+	}
+	memcpy(dup, owner, size);
+	return dup;
+}
+
+static int
+console_session_owner_fd(const struct session_owner *owner)
+{
+	assert(owner->vtab == &console_session_owner_vtab);
+	return ((const struct console_session_owner *) owner)->fd;
+}
+
+static inline void
+console_session_owner_create(struct console_session_owner *owner, int fd)
+{
+	owner->base.type = SESSION_TYPE_CONSOLE;
+	owner->base.vtab = &console_session_owner_vtab;
+	owner->fd = fd;
+}
+
 static const char *sessionlib_name = "box.session";
 
 /* Create session and pin it to fiber */
@@ -56,14 +104,24 @@ lbox_session_create(struct lua_State *L)
 				     "session from Lua");
 	}
 	struct session *session = fiber_get_session(fiber());
+	int fd = luaL_optinteger(L, 1, -1);
 	if (session == NULL) {
-		int fd = luaL_optinteger(L, 1, -1);
-		session = session_create_on_demand(fd);
+		struct session_owner owner;
+		session_owner_create(&owner, type);
+		session = session_create_on_demand(&owner);
 		if (session == NULL)
 			return luaT_error(L);
 	}
-	/* If a session already exists, simply reset its type */
-	session->type = type;
+	/* If a session already exists, simply reset its owner. */
+	if (type == SESSION_TYPE_CONSOLE) {
+		struct console_session_owner owner;
+		console_session_owner_create(&owner, fd);
+		session_set_owner(session, (struct session_owner *) &owner);
+	} else {
+		struct session_owner owner;
+		session_owner_create(&owner, type);
+		session_set_owner(session, &owner);
+	}
 	lua_pushnumber(L, session->id);
 	return 1;
 }
@@ -90,7 +148,7 @@ lbox_session_id(struct lua_State *L)
 static int
 lbox_session_type(struct lua_State *L)
 {
-	lua_pushstring(L, session_type_strs[current_session()->type]);
+	lua_pushstring(L, session_type_strs[session_type(current_session())]);
 	return 1;
 }
 
@@ -237,7 +295,7 @@ lbox_session_fd(struct lua_State *L)
 	struct session *session = session_find(sid);
 	if (session == NULL)
 		luaL_error(L, "session.fd(): session does not exist");
-	lua_pushinteger(L, session->fd);
+	lua_pushinteger(L, session_fd(session));
 	return 1;
 }
 
@@ -251,7 +309,6 @@ lbox_session_peer(struct lua_State *L)
 	if (lua_gettop(L) > 1)
 		luaL_error(L, "session.peer(sid): bad arguments");
 
-	int fd;
 	struct session *session;
 	if (lua_gettop(L) == 1)
 		session = session_find(luaL_checkint(L, 1));
@@ -259,7 +316,7 @@ lbox_session_peer(struct lua_State *L)
 		session = current_session();
 	if (session == NULL)
 		luaL_error(L, "session.peer(): session does not exist");
-	fd = session->fd;
+	int fd = session_fd(session);
 	if (fd < 0) {
 		lua_pushnil(L); /* no associated peer */
 		return 1;
diff --git a/src/box/session.cc b/src/box/session.cc
index 0b0c5ae44..908ec9c4e 100644
--- a/src/box/session.cc
+++ b/src/box/session.cc
@@ -33,7 +33,6 @@
 #include "memory.h"
 #include "assoc.h"
 #include "trigger.h"
-#include "random.h"
 #include "user.h"
 #include "error.h"
 
@@ -54,6 +53,48 @@ RLIST_HEAD(session_on_connect);
 RLIST_HEAD(session_on_disconnect);
 RLIST_HEAD(session_on_auth);
 
+static struct session_owner *
+generic_session_owner_dup(struct session_owner *owner);
+
+static int
+generic_session_owner_fd(const struct session_owner *owner);
+
+static const struct session_owner_vtab generic_session_owner_vtab = {
+	/* .dup = */ generic_session_owner_dup,
+	/* .delete = */ (void (*)(struct session_owner *)) free,
+	/* .fd = */ generic_session_owner_fd,
+};
+
+static struct session_owner *
+generic_session_owner_dup(struct session_owner *owner)
+{
+	assert(owner->vtab == &generic_session_owner_vtab);
+	struct session_owner *dup =
+		(struct session_owner *) malloc(sizeof(*dup));
+	if (dup == NULL) {
+		diag_set(OutOfMemory, sizeof(*dup), "malloc",
+			 "default_session_owner");
+		return NULL;
+	}
+	memcpy(dup, owner, sizeof(*dup));
+	return dup;
+}
+
+static int
+generic_session_owner_fd(const struct session_owner *owner)
+{
+	assert(owner->vtab == &generic_session_owner_vtab);
+	(void) owner;
+	return -1;
+}
+
+void
+session_owner_create(struct session_owner *owner, enum session_type type)
+{
+	owner->type = type;
+	owner->vtab = &generic_session_owner_vtab;
+}
+
 static inline uint64_t
 sid_max()
 {
@@ -80,7 +121,7 @@ session_on_stop(struct trigger *trigger, void * /* event */)
 }
 
 struct session *
-session_create(int fd, enum session_type type)
+session_create(struct session_owner *owner)
 {
 	struct session *session =
 		(struct session *) mempool_alloc(&session_pool);
@@ -90,14 +131,15 @@ session_create(int fd, enum session_type type)
 		return NULL;
 	}
 	session->id = sid_max();
-	session->fd =  fd;
+	session->owner = session_owner_dup(owner);
+	if (session->owner == NULL) {
+		mempool_free(&session_pool, session);
+		return NULL;
+	}
 	session->sync = 0;
-	session->type = type;
 	/* For on_connect triggers. */
 	credentials_init(&session->credentials, guest_user->auth_token,
 			 guest_user->def->uid);
-	if (fd >= 0)
-		random_bytes(session->salt, SESSION_SEED_SIZE);
 	struct mh_i64ptr_node_t node;
 	node.key = session->id;
 	node.val = session;
@@ -105,6 +147,7 @@ session_create(int fd, enum session_type type)
 	mh_int_t k = mh_i64ptr_put(session_registry, &node, NULL, NULL);
 
 	if (k == mh_end(session_registry)) {
+		session_owner_delete(owner);
 		mempool_free(&session_pool, session);
 		diag_set(OutOfMemory, 0, "session hash", "new session");
 		return NULL;
@@ -112,13 +155,25 @@ session_create(int fd, enum session_type type)
 	return session;
 }
 
+int
+session_set_owner(struct session *session, struct session_owner *new_owner)
+{
+	struct session_owner *dup = session_owner_dup(new_owner);
+	if (dup == NULL)
+		return -1;
+	if (session->owner != NULL)
+		session_owner_delete(session->owner);
+	session->owner = dup;
+	return 0;
+}
+
 struct session *
-session_create_on_demand(int fd)
+session_create_on_demand(struct session_owner *owner)
 {
 	assert(fiber_get_session(fiber()) == NULL);
 
 	/* Create session on demand */
-	struct session *s = session_create(fd, SESSION_TYPE_BACKGROUND);
+	struct session *s = session_create(owner);
 	if (s == NULL)
 		return NULL;
 	s->fiber_on_stop = {
@@ -186,6 +241,7 @@ session_destroy(struct session *session)
 {
 	struct mh_i64ptr_node_t node = { session->id, NULL };
 	mh_i64ptr_remove(session_registry, &node, NULL);
+	session_owner_delete(session->owner);
 	mempool_free(&session_pool, session);
 }
 
diff --git a/src/box/session.h b/src/box/session.h
index 4f9235ea8..105dcab17 100644
--- a/src/box/session.h
+++ b/src/box/session.h
@@ -47,8 +47,6 @@ session_init();
 void
 session_free();
 
-enum {	SESSION_SEED_SIZE = 32, SESSION_DELIM_SIZE = 16 };
-
 enum session_type {
 	SESSION_TYPE_BACKGROUND = 0,
 	SESSION_TYPE_BINARY,
@@ -60,6 +58,49 @@ enum session_type {
 
 extern const char *session_type_strs[];
 
+struct session_owner_vtab;
+
+/**
+ * Object to store session type specific data. For example, IProto
+ * stores iproto_connection, console stores file descriptor.
+ */
+struct session_owner {
+	/** Session type. */
+	enum session_type type;
+	/** Virtual session owner methods. */
+	const struct session_owner_vtab *vtab;
+};
+
+struct session_owner_vtab {
+	/** Allocate a duplicate of an owner. */
+	struct session_owner *(*dup)(struct session_owner *);
+	/** Destroy an owner, and free its memory. */
+	void (*free)(struct session_owner *);
+	/** Get the descriptor of an owner, if has. Else -1. */
+	int (*fd)(const struct session_owner *);
+};
+
+static inline struct session_owner *
+session_owner_dup(struct session_owner *owner)
+{
+	return owner->vtab->dup(owner);
+}
+
+static inline void
+session_owner_delete(struct session_owner *owner)
+{
+	owner->vtab->free(owner);
+}
+
+/**
+ * Initialize a session owner with @a type and with default
+ * virtual methods.
+ * @param owner Session owner to initialize. Is copied inside.
+ * @param type Session type.
+ */
+void
+session_owner_create(struct session_owner *owner, enum session_type type);
+
 /**
  * Abstraction of a single user session:
  * for now, only provides accounting of established
@@ -72,10 +113,8 @@ extern const char *session_type_strs[];
 struct session {
 	/** Session id. */
 	uint64_t id;
-	/** File descriptor - socket of the connected peer.
-	 * Only if the session has a peer.
-	 */
-	int fd;
+	/** Session owner with type specific data. */
+	struct session_owner *owner;
 	/**
 	 * For iproto requests, we set this field
 	 * to the value of packet sync. Since the
@@ -85,15 +124,24 @@ struct session {
 	 * the first yield.
 	 */
 	uint64_t sync;
-	enum session_type type;
-	/** Authentication salt. */
-	char salt[SESSION_SEED_SIZE];
 	/** Session user id and global grants */
 	struct credentials credentials;
 	/** Trigger for fiber on_stop to cleanup created on-demand session */
 	struct trigger fiber_on_stop;
 };
 
+static inline enum session_type
+session_type(const struct session *session)
+{
+	return session->owner->type;
+}
+
+static inline int
+session_fd(const struct session *session)
+{
+	return session->owner->vtab->fd(session->owner);
+}
+
 /**
  * Find a session by id.
  */
@@ -154,7 +202,7 @@ extern struct credentials admin_credentials;
  * trigger to destroy it when this fiber ends.
  */
 struct session *
-session_create_on_demand(int fd);
+session_create_on_demand(struct session_owner *owner);
 
 /*
  * When creating a new fiber, the database (box)
@@ -171,7 +219,9 @@ current_session()
 {
 	struct session *session = fiber_get_session(fiber());
 	if (session == NULL) {
-		session = session_create_on_demand(-1);
+		struct session_owner owner;
+		session_owner_create(&owner, SESSION_TYPE_BACKGROUND);
+		session = session_create_on_demand(&owner);
 		if (session == NULL)
 			diag_raise();
 	}
@@ -191,7 +241,9 @@ effective_user()
 		(struct credentials *) fiber_get_key(fiber(),
 						     FIBER_KEY_USER);
 	if (u == NULL) {
-		session_create_on_demand(-1);
+		struct session_owner owner;
+		session_owner_create(&owner, SESSION_TYPE_BACKGROUND);
+		session_create_on_demand(&owner);
 		u = (struct credentials *) fiber_get_key(fiber(),
 							 FIBER_KEY_USER);
 	}
@@ -216,7 +268,18 @@ session_storage_cleanup(int sid);
  * trigger fails or runs out of resources.
  */
 struct session *
-session_create(int fd, enum session_type type);
+session_create(struct session_owner *owner);
+
+/**
+ * Set new owner of a session.
+ * @param session Session to change owner.
+ * @param new_owner New session owner. Is duplicated inside.
+ *
+ * @retval -1 Memory error.
+ * @retval  0 Success.
+ */
+int
+session_set_owner(struct session *session, struct session_owner *new_owner);
 
 /**
  * Destroy a session.
diff --git a/src/box/vinyl.c b/src/box/vinyl.c
index e0c30757c..7ae12f597 100644
--- a/src/box/vinyl.c
+++ b/src/box/vinyl.c
@@ -2453,7 +2453,8 @@ vinyl_engine_prepare(struct engine *engine, struct txn *txn)
 	 * available for the admin to track the lag so let the applier
 	 * wait as long as necessary for memory dump to complete.
 	 */
-	double timeout = (current_session()->type != SESSION_TYPE_APPLIER ?
+	double timeout = (session_type(current_session()) !=
+			  SESSION_TYPE_APPLIER ?
 			  env->timeout : TIMEOUT_INFINITY);
 	/*
 	 * Reserve quota needed by the transaction before allocating
-- 
2.14.3 (Apple Git-98)