Added communication with master

Makefile.am

@@ -182,6 +182,7 @@ dist_tmate_slave_SOURCES = \
 	tmate-client-decoder.c \
 	tmate-client-encoder.c \
 	tmate-msgpack.c \
+	tmate-master.c \
 	tmate-slave.c \
 	tmate-ssh-client-pty.c \
 	tmate-ssh-client.c \

tmate-client-decoder.c

@@ -1,6 +1,7 @@
 #include <ctype.h>
 #include <unistd.h>
 #include "tmate.h"
+#include "tmate-protocol.h"
 
 char *tmate_left_status, *tmate_right_status;
 
@@ -362,14 +363,14 @@ void tmate_dispatch_daemon_message(struct tmate_session *session,
 
 	int cmd = unpack_int(uk);
 	switch (cmd) {
-	dispatch(TMATE_HEADER,		tmate_header);
+	dispatch(TMATE_OUT_HEADER,		tmate_header);
-	dispatch(TMATE_SYNC_LAYOUT,	tmate_sync_layout);
+	dispatch(TMATE_OUT_SYNC_LAYOUT,		tmate_sync_layout);
-	dispatch(TMATE_PTY_DATA,	tmate_pty_data);
+	dispatch(TMATE_OUT_PTY_DATA,		tmate_pty_data);
-	dispatch(TMATE_EXEC_CMD,	tmate_exec_cmd);
+	dispatch(TMATE_OUT_EXEC_CMD,		tmate_exec_cmd);
-	dispatch(TMATE_FAILED_CMD,	tmate_failed_cmd);
+	dispatch(TMATE_OUT_FAILED_CMD,		tmate_failed_cmd);
-	dispatch(TMATE_STATUS,		tmate_status);
+	dispatch(TMATE_OUT_STATUS,		tmate_status);
-	dispatch(TMATE_SYNC_COPY_MODE,	tmate_sync_copy_mode);
+	dispatch(TMATE_OUT_SYNC_COPY_MODE,	tmate_sync_copy_mode);
-	dispatch(TMATE_WRITE_COPY_MODE,	tmate_write_copy_mode);
+	dispatch(TMATE_OUT_WRITE_COPY_MODE,	tmate_write_copy_mode);
 	default: tmate_fatal("Bad message type: %d", cmd);
 	}
 }

tmate-client-encoder.c

@@ -1,11 +1,12 @@
 #include "tmate.h"
+#include "tmate-protocol.h"
 
 #define pack(what, ...) _pack(&tmate_session->client_encoder, what, __VA_ARGS__)
 
 static void __tmate_notify(const char *msg)
 {
 	pack(array, 2);
-	pack(int, TMATE_NOTIFY);
+	pack(int, TMATE_IN_NOTIFY);
 	pack(string, msg);
 }
 
@@ -101,7 +102,7 @@ void tmate_send_client_ready(void)
 		return;
 
 	pack(array, 1);
-	pack(int, TMATE_CLIENT_READY);
+	pack(int, TMATE_IN_READY);
 }
 
 void tmate_send_env(const char *name, const char *value)
@@ -110,7 +111,7 @@ void tmate_send_env(const char *name, const char *value)
 		return;
 
 	pack(array, 3);
-	pack(int, TMATE_CLIENT_ENV);
+	pack(int, TMATE_IN_SET_ENV);
 	pack(string, name);
 	pack(string, value);
 }
@@ -118,7 +119,7 @@ void tmate_send_env(const char *name, const char *value)
 void tmate_client_resize(u_int sx, u_int sy)
 {
 	pack(array, 3);
-	pack(int, TMATE_CLIENT_RESIZE);
+	pack(int, TMATE_IN_RESIZE);
 	/* cast to signed, -1 == no clients */
 	pack(int, sx);
 	pack(int, sy);
@@ -132,7 +133,7 @@ void tmate_client_pane_key(int pane_id, int key)
 	 */
 
 	pack(array, 2);
-	pack(int, TMATE_CLIENT_PANE_KEY);
+	pack(int, TMATE_IN_PANE_KEY);
 	pack(int, key);
 }
 
@@ -159,7 +160,7 @@ int tmate_should_exec_cmd_locally(const struct cmd_entry *cmd)
 void tmate_client_cmd(int client_id, const char *cmd)
 {
 	pack(array, 3);
-	pack(int, TMATE_CLIENT_EXEC_CMD);
+	pack(int, TMATE_IN_EXEC_CMD);
 	pack(int, client_id);
 	pack(string, cmd);
 }

tmate-master.c

@@ -0,0 +1,138 @@
+#include <sys/socket.h>
+#include <netinet/tcp.h>
+#include <fcntl.h>
+#include <time.h>
+
+#include "tmate.h"
+#include "tmate-protocol.h"
+
+#define pack(what, ...) _pack(&tmate_session->master_encoder, what, __VA_ARGS__)
+
+void tmate_send_master_keyframe(struct tmate_session *session)
+{
+	struct timespec time_diff;
+
+	if (!tmate_has_master())
+		return;
+
+	/* Eventually the deamon will send the timestamps and keyframe number */
+
+	if (clock_gettime(CLOCK_MONOTONIC, &session->keyframe_start_time) < 0)
+		tmate_fatal("Cannot get time");
+
+	timespec_subtract(&time_diff, &session->keyframe_start_time,
+			  &session->session_start_time);
+	session->keyframe_size = 0;
+
+	pack(array, 3);
+	pack(int, TMATE_CTL_KEYFRAME);
+	pack(unsigned_int, session->keyframe_cnt++);
+	pack(unsigned_long_long, timespec_to_millisec(&time_diff));
+}
+
+void tmate_send_master_daemon_msg(struct tmate_session *session,
+				  struct tmate_unpacker *uk)
+{
+	struct timespec time_diff, current_time;
+	int i;
+
+	if (!tmate_has_master())
+		return;
+
+	if (clock_gettime(CLOCK_MONOTONIC, &current_time) < 0)
+		tmate_fatal("Cannot get time");
+
+	timespec_subtract(&time_diff, &current_time,
+			  &session->keyframe_start_time);
+
+	pack(array, 3);
+	pack(int, TMATE_CTL_DEAMON_OUT_MSG);
+	pack(unsigned_int, timespec_to_millisec(&time_diff));
+
+	pack(array, uk->argc);
+	for (i = 0; i < uk->argc; i++)
+		pack(object, uk->argv[i]);
+}
+
+void tmate_send_master_header(struct tmate_session *session)
+{
+	if (!tmate_has_master())
+		return;
+
+	pack(array, 6);
+	pack(int, TMATE_CTL_AUTH);
+	pack(int, CONTROL_PROTOCOL_VERSION);
+	pack(string, session->ssh_client.ip_address);
+	pack(string, session->ssh_client.pubkey);
+	pack(string, session->session_token);
+	pack(string, session->session_token_ro);
+}
+
+void tmate_init_master_session(struct tmate_session *session)
+{
+	if (!tmate_has_master())
+		return;
+
+	if (clock_gettime(CLOCK_MONOTONIC, &session->session_start_time) < 0)
+		tmate_fatal("Cannot get time");
+
+	session->keyframe_cnt = 0;
+	session->keyframe_size = 0;
+}
+
+static int _tmate_connect_to_master(const char *hostname, int port)
+{
+	int sockfd = -1;
+	struct sockaddr_in servaddr;
+	struct hostent *host;
+
+	sockfd = socket(AF_INET, SOCK_STREAM, 0);
+	if (sockfd < 0)
+		tmate_fatal("Cannot create socket");
+
+	host = gethostbyname(hostname);
+	if (!host)
+		tmate_fatal("Cannot resolve %s", hostname);
+
+	memset(&servaddr, 0, sizeof(servaddr));
+	servaddr.sin_family = host->h_addrtype;
+	memcpy(&servaddr.sin_addr, host->h_addr, host->h_length);
+	servaddr.sin_port = htons(port);
+
+	if (connect(sockfd, &servaddr, sizeof(servaddr)) < 0)
+		tmate_fatal("Cannot connect to master at %s:%d", hostname, port);
+
+	int flag = 1;
+	if (setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)) < 0)
+		tmate_fatal("Can't set master socket to TCP_NODELAY");
+
+	if (fcntl(sockfd, F_SETFL, O_NONBLOCK) < 0)
+		tmate_fatal("Can't set master socket to non-blocking");
+
+	tmate_notice("Connected to master at %s:%d", hostname, port);
+
+	return sockfd;
+}
+
+int tmate_connect_to_master(void)
+{
+	return _tmate_connect_to_master(tmate_settings->master_hostname,
+					tmate_settings->master_port);
+}
+
+void timespec_subtract(struct timespec *result,
+		       struct timespec *x, struct timespec *y)
+{
+	if (x->tv_nsec < y->tv_nsec) {
+		result->tv_sec = x->tv_sec - y->tv_sec - 1;
+		result->tv_nsec = x->tv_nsec - y->tv_nsec + 1000000000;
+	} else {
+		result->tv_sec = x->tv_sec - y->tv_sec;
+		result->tv_nsec = x->tv_nsec - y->tv_nsec;
+	}
+}
+
+unsigned long long timespec_to_millisec(struct timespec *ts)
+{
+	return ts->tv_sec * 1000ULL + ts->tv_nsec / 1000000ULL;
+}

tmate-msgpack.c

@@ -1,4 +1,5 @@
 #include "tmate.h"
+#include "tmate-protocol.h"
 
 static void on_encoder_buffer_ready(evutil_socket_t fd, short what, void *arg)
 {

tmate-protocol.h

@@ -0,0 +1,36 @@
+#ifndef TMATE_PROTOCOL_H
+#define TMATE_PROTOCOL_H
+
+ /* 17 and not 16 because the sender does not takes into account envelope size */
+#define TMATE_MAX_MESSAGE_SIZE (17*1024)
+
+#define CONTROL_PROTOCOL_VERSION 1
+
+/* TODO document each msg */
+
+enum tmate_control_msg_types {
+	TMATE_CTL_AUTH,
+	TMATE_CTL_DEAMON_OUT_MSG,
+	TMATE_CTL_KEYFRAME,
+};
+
+enum tmate_daemon_out_msg_types {
+	TMATE_OUT_HEADER,
+	TMATE_OUT_SYNC_LAYOUT,
+	TMATE_OUT_PTY_DATA,
+	TMATE_OUT_EXEC_CMD,
+	TMATE_OUT_FAILED_CMD,
+	TMATE_OUT_STATUS,
+	TMATE_OUT_SYNC_COPY_MODE,
+	TMATE_OUT_WRITE_COPY_MODE,
+};
+enum tmate_daemon_in_msg_types {
+	TMATE_IN_NOTIFY,
+	TMATE_IN_PANE_KEY,
+	TMATE_IN_RESIZE,
+	TMATE_IN_EXEC_CMD,
+	TMATE_IN_SET_ENV,
+	TMATE_IN_READY,
+};
+
+#endif

tmate-slave.c

@@ -33,7 +33,7 @@ extern int client_connect(char *path, int start_server);
 struct tmate_settings _tmate_settings = {
 	.keys_dir        = TMATE_SSH_DEFAULT_KEYS_DIR,
 	.ssh_port        = TMATE_SSH_DEFAULT_PORT,
-	.master_hostname = TMATE_DEFAULT_MASTER_HOST,
+	.master_hostname = NULL,
 	.master_port     = TMATE_DEFAULT_MASTER_PORT,
 	.tmate_host      = NULL,
 	.log_level       = LOG_NOTICE,
@@ -328,16 +328,16 @@ static void tmate_spawn_slave_server(struct tmate_session *session)
 	 */
 	setup_ncurse(STDOUT_FILENO, "screen-256color");
 
+	tmate_daemon_init(session);
+
 	close_fds_except((int[]){session->tmux_socket_fd,
 				 ssh_get_fd(session->ssh_client.session),
-				 log_file ? fileno(log_file) : -1}, 3);
+				 log_file ? fileno(log_file) : -1,
+				 session->master_fd}, 4);
 
 	jail();
-
 	event_reinit(ev_base);
 
-	tmate_daemon_init(session);
-
 	tmux_server_init(IDENTIFY_UTF8 | IDENTIFY_256COLOURS);
 	/* never reached */
 }
@@ -399,6 +399,11 @@ static void tmate_spawn_slave_client(struct tmate_session *session)
 	dup2(slave_pty, STDERR_FILENO);
 
 	setup_ncurse(slave_pty, "screen-256color");
+
+	tmate_ssh_client_pty_init(session);
+
+	/* the unused session->master_fd will get closed automatically */
+
 	close_fds_except((int[]){STDIN_FILENO, STDOUT_FILENO, STDERR_FILENO,
 				 session->tmux_socket_fd,
 				 ssh_get_fd(session->ssh_client.session),
@@ -406,8 +411,6 @@ static void tmate_spawn_slave_client(struct tmate_session *session)
 	jail();
 	event_reinit(ev_base);
 
-	tmate_ssh_client_pty_init(session);
-
 	ret = client_main(argc, argv, IDENTIFY_UTF8 | IDENTIFY_256COLOURS);
 	tmate_flush_pty(session);
 	exit(ret);

tmate-ssh-client.c

@@ -1,8 +1,84 @@
 #include "tmate.h"
+#include <errno.h>
 
-static void on_decoder_read(void *userdata, struct tmate_unpacker *uk)
+static void on_master_decoder_read(void *userdata, struct tmate_unpacker *uk)
+{
+	/* struct tmate_session *session = userdata; */
+	tmate_info("Received master data!");
+}
+
+static void on_master_read(struct bufferevent *bev, void *_session)
+{
+	struct tmate_session *session = _session;
+	struct evbuffer *master_in;
+	ssize_t written;
+	char *buf;
+	size_t len;
+
+	master_in = bufferevent_get_input(session->bev_master);
+
+	while (evbuffer_get_length(master_in)) {
+		tmate_decoder_get_buffer(&session->client_decoder, &buf, &len);
+
+		if (len == 0)
+			tmate_fatal("No more room in client decoder. Message too big?");
+
+		written = evbuffer_remove(master_in, buf, len);
+		if (written < 0)
+			tmate_fatal("Cannot read master buffer");
+	}
+}
+
+static void on_master_encoder_write(void *userdata, struct evbuffer *buffer)
 {
 	struct tmate_session *session = userdata;
+	struct evbuffer *master_out;
+	size_t len;
+
+	master_out = bufferevent_get_output(session->bev_master);
+
+	len = evbuffer_get_length(buffer);
+
+	if (session->keyframe_size + len > TMATE_KEYFRAME_MAX_SIZE) {
+		if (session->keyframe_size == 0)
+			tmate_fatal("keyframe max size too small");
+		tmate_send_master_keyframe(session);
+	}
+
+	session->keyframe_size += len;
+
+	if (evbuffer_add_buffer(master_out, buffer) < 0)
+		tmate_fatal("Cannot write to master buffer");
+
+}
+
+static void on_master_event(struct bufferevent *bev, short events, void *_session)
+{
+	if (events & BEV_EVENT_EOF)
+		tmate_fatal("Connection to master closed");
+
+	if (events & BEV_EVENT_ERROR)
+		tmate_fatal("Connection to master error: %s",
+			    evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()));
+}
+
+static void on_daemon_decoder_read(void *userdata, struct tmate_unpacker *uk)
+{
+	struct tmate_session *session = userdata;
+	struct timespec time_diff, current_time;
+
+	if (tmate_has_master()) {
+		if (clock_gettime(CLOCK_MONOTONIC, &current_time) < 0)
+			tmate_fatal("Cannot get time");
+
+		timespec_subtract(&time_diff, &current_time,
+				  &session->keyframe_start_time);
+		if (time_diff.tv_sec > TMATE_KEYFRAME_INTERVAL_SEC - 1)
+			tmate_send_master_keyframe(session);
+
+		tmate_send_master_daemon_msg(session, uk);
+	}
+
 	tmate_dispatch_daemon_message(session, uk);
 }
 
@@ -37,7 +113,7 @@ static int on_ssh_channel_read(ssh_session _session, ssh_channel channel,
 	return written;
 }
 
-static void on_encoder_write(void *userdata, struct evbuffer *buffer)
+static void on_daemon_encoder_write(void *userdata, struct evbuffer *buffer)
 {
 	struct tmate_session *session = userdata;
 	ssize_t len, written;
@@ -62,6 +138,27 @@ static void on_encoder_write(void *userdata, struct evbuffer *buffer)
 	}
 }
 
+static void init_master(struct tmate_session *session)
+{
+	/* session->master_fd is already connected */
+	session->bev_master = bufferevent_socket_new(ev_base, session->master_fd,
+						     BEV_OPT_CLOSE_ON_FREE);
+	if (!session->bev_master)
+		tmate_fatal("Cannot setup socket bufferevent");
+
+	bufferevent_setcb(session->bev_master,
+			  on_master_read, NULL, on_master_event, session);
+	bufferevent_enable(session->bev_master, EV_READ | EV_WRITE);
+
+	tmate_encoder_init(&session->master_encoder, on_master_encoder_write, session);
+	tmate_decoder_init(&session->master_decoder, on_master_decoder_read, session);
+
+	tmate_init_master_session(session);
+	tmate_send_master_header(session);
+
+	tmate_send_master_keyframe(session);
+}
+
 void tmate_daemon_init(struct tmate_session *session)
 {
 	struct tmate_ssh_client *client = &session->ssh_client;
@@ -72,6 +169,9 @@ void tmate_daemon_init(struct tmate_session *session)
 	client->channel_cb.channel_data_function = on_ssh_channel_read,
 	ssh_set_channel_callbacks(client->channel, &client->channel_cb);
 
-	tmate_encoder_init(&session->client_encoder, on_encoder_write, session);
+	tmate_encoder_init(&session->client_encoder, on_daemon_encoder_write, session);
-	tmate_decoder_init(&session->client_decoder, on_decoder_read, session);
+	tmate_decoder_init(&session->client_decoder, on_daemon_decoder_read, session);
+
+	if (tmate_has_master())
+		init_master(session);
 }

tmate-ssh-server.c

@@ -300,6 +300,10 @@ void tmate_ssh_server_main(struct tmate_session *session,
 	pid_t pid;
 
 	setup_signals();
+
+	if (tmate_has_master())
+		close(tmate_connect_to_master());
+
 	bind = prepare_ssh(keys_dir, port);
 
 	for (;;) {
@@ -314,6 +318,14 @@ void tmate_ssh_server_main(struct tmate_session *session,
 		if (ssh_bind_accept(bind, client->session) < 0)
 			tmate_fatal("Error accepting connection: %s", ssh_get_error(bind));
 
+		/*
+		 * We should die if we can't connect to master. This way the
+		 * tmate daemon will pick another server to work on.
+		 */
+		session->master_fd = -1;
+		if (tmate_has_master())
+			session->master_fd = tmate_connect_to_master();
+
 		if (get_ip(ssh_get_fd(client->session),
 			   client->ip_address, sizeof(client->ip_address)) < 0)
 			tmate_fatal("Error getting IP address from connection");
@@ -325,6 +337,7 @@ void tmate_ssh_server_main(struct tmate_session *session,
 			tmate_info("Child spawned pid=%d, ip=%s",
 				    pid, client->ip_address);
 			ssh_free(client->session);
+			close(session->master_fd);
 		} else {
 			ssh_bind_free(bind);
 			session->session_token = "init";

tmate.h

@@ -7,6 +7,7 @@
 #include <libssh/libssh.h>
 #include <libssh/callbacks.h>
 #include <event.h>
+#include <time.h>
 
 #include "tmux.h"
 struct tmate_session;
@@ -106,15 +107,6 @@ extern void unpack_array(struct tmate_unpacker *uk, struct tmate_unpacker *neste
 
 #define TMATE_LATEST_VERSION "1.8.10"
 
-enum tmate_client_commands {
-	TMATE_NOTIFY,
-	TMATE_CLIENT_PANE_KEY,
-	TMATE_CLIENT_RESIZE,
-	TMATE_CLIENT_EXEC_CMD,
-	TMATE_CLIENT_ENV,
-	TMATE_CLIENT_READY,
-};
-
 extern void tmate_client_encoder_init(struct tmate_encoder *encoder);
 
 extern void printflike1 tmate_notify(const char *fmt, ...);
@@ -133,29 +125,17 @@ extern void tmate_send_client_ready(void);
 /* tmate-client-decoder.c */
 
 #define TMATE_HLIMIT 2000
- /* 17 and not 16 because the sender does not takes into account envelope size */
-#define TMATE_MAX_MESSAGE_SIZE (17*1024)
-
-extern char *tmate_left_status, *tmate_right_status;
-
-enum tmate_commands {
-	TMATE_HEADER,
-	TMATE_SYNC_LAYOUT,
-	TMATE_PTY_DATA,
-	TMATE_EXEC_CMD,
-	TMATE_FAILED_CMD,
-	TMATE_STATUS,
-	TMATE_SYNC_COPY_MODE,
-	TMATE_WRITE_COPY_MODE,
-};
-
 #define TMATE_PANE_ACTIVE 1
 
+extern char *tmate_left_status, *tmate_right_status;
 extern void tmate_dispatch_daemon_message(struct tmate_session *session,
 					  struct tmate_unpacker *uk);
 
 /* tmate-ssh-client.c */
 
+#define TMATE_KEYFRAME_INTERVAL_SEC 10
+#define TMATE_KEYFRAME_MAX_SIZE 1024*1024
+
 extern void tmate_daemon_init(struct tmate_session *session);
 
 /* tmate-ssh-client-pty.c */
@@ -208,7 +188,6 @@ extern void tmate_ssh_server_main(struct tmate_session *session,
 
 #define TMATE_SSH_DEFAULT_KEYS_DIR "keys"
 
-#define TMATE_DEFAULT_MASTER_HOST NULL
 #define TMATE_DEFAULT_MASTER_PORT 7000
 
 #define TMATE_TOKEN_LEN 25
@@ -231,18 +210,23 @@ struct tmate_session {
 	int tmux_socket_fd;
 
 	/* only for deamon */
+	const char *session_token;
+	const char *session_token_ro;
 
 	struct tmate_encoder client_encoder;
 	struct tmate_decoder client_decoder;
 	int client_protocol_version;
+	struct event ev_notify_timer;
 
+	int master_fd;
+	struct bufferevent *bev_master;
 	struct tmate_encoder master_encoder;
 	struct tmate_decoder master_decoder;
 
-	const char *session_token;
+	struct timespec session_start_time;
-	const char *session_token_ro;
+	struct timespec keyframe_start_time;
-
+	unsigned int keyframe_cnt;
-	struct event ev_notify_timer;
+	size_t keyframe_size;
 
 	/* only for client-pty */
 	int pty;
@@ -256,6 +240,25 @@ extern long tmate_get_random_long(void);
 extern void request_server_termination(void);
 extern void tmate_spawn_slave(struct tmate_session *session);
 
+/* tmate-master.c */
+
+extern void tmate_send_master_keyframe(struct tmate_session *session);
+extern void tmate_send_master_daemon_msg(struct tmate_session *session,
+					 struct tmate_unpacker *uk);
+extern void tmate_send_master_header(struct tmate_session *session);
+extern void tmate_init_master_session(struct tmate_session *session);
+
+extern int tmate_connect_to_master(void);
+static inline bool tmate_has_master(void)
+{
+	return !!tmate_settings->master_hostname;
+}
+
+
+extern void timespec_subtract(struct timespec *result,
+			      struct timespec *x, struct timespec *y);
+extern unsigned long long timespec_to_millisec(struct timespec *ts);
+
 /* tmate-debug.c */
 
 extern void tmate_preload_trace_lib(void);