Compare commits

...

7 commits

Author SHA1 Message Date
Romain Vimont b4ee797401 TODO 2021-10-30 23:10:05 +02:00
Romain Vimont d7106d7009 wip 2021-10-30 22:46:19 +02:00
Romain Vimont 9e6687e698 server_thread 2021-10-30 22:37:57 +02:00
Romain Vimont dc05033441 server_cbs 2021-10-30 22:37:57 +02:00
Romain Vimont dc652f67ce server_info 2021-10-30 22:37:57 +02:00
Romain Vimont 55641ceb64 Copy server params
This is a preliminary step necessary to move the server to a separate
thread.
2021-10-30 22:37:57 +02:00
Romain Vimont fb330cc603 Reorder server and server_params
This will allow to define a server_params field in server.
2021-10-30 22:37:57 +02:00
4 changed files with 265 additions and 111 deletions

View file

@ -1,2 +1,5 @@
#define EVENT_NEW_FRAME SDL_USEREVENT
#define EVENT_STREAM_STOPPED (SDL_USEREVENT + 1)
#define EVENT_NEW_FRAME SDL_USEREVENT
#define EVENT_STREAM_STOPPED (SDL_USEREVENT + 1)
#define EVENT_SERVER_CONNECTION_FAILED (SDL_USEREVENT + 2)
#define EVENT_SERVER_CONNECTED (SDL_USEREVENT + 3)
#define EVENT_SERVER_DISCONNECTED (SDL_USEREVENT + 4)

View file

@ -158,6 +158,10 @@ static enum event_result
handle_event(struct scrcpy *s, const struct scrcpy_options *options,
SDL_Event *event) {
switch (event->type) {
case EVENT_SERVER_DISCONNECTED:
LOGD("Server disconnected");
// Do nothing, will be managed by the "stream stopped" event
break;
case EVENT_STREAM_STOPPED:
LOGD("Video stream stopped");
return EVENT_RESULT_STOPPED_BY_EOS;
@ -216,6 +220,32 @@ event_loop(struct scrcpy *s, const struct scrcpy_options *options) {
return false;
}
static bool
await_for_server(void) {
SDL_Event event;
while (SDL_WaitEvent(&event)) {
// Should never receive disconnected event before connected
assert(event.type != EVENT_SERVER_DISCONNECTED);
switch (event.type) {
case SDL_QUIT:
LOGD("User requested to quit");
return false;
case EVENT_SERVER_CONNECTION_FAILED:
LOGE("Server connection failed");
return false;
case EVENT_SERVER_CONNECTED:
LOGD("Server connected");
return true;
default:
break;
}
}
LOGE("SDL_WaitEvent() error: %s", SDL_GetError());
return false;
}
static SDL_LogPriority
sdl_priority_from_av_level(int level) {
switch (level) {
@ -261,15 +291,35 @@ stream_on_eos(struct stream *stream, void *userdata) {
PUSH_EVENT(EVENT_STREAM_STOPPED);
}
static void
server_on_connection_failed(struct server *server, void *userdata) {
(void) server;
(void) userdata;
PUSH_EVENT(EVENT_SERVER_CONNECTION_FAILED);
}
static void
server_on_connected(struct server *server, void *userdata) {
(void) server;
(void) userdata;
PUSH_EVENT(EVENT_SERVER_CONNECTED);
}
static void
server_on_disconnected(struct server *server, void *userdata) {
(void) server;
(void) userdata;
PUSH_EVENT(EVENT_SERVER_DISCONNECTED);
}
bool
scrcpy(struct scrcpy_options *options) {
static struct scrcpy scrcpy;
struct scrcpy *s = &scrcpy;
if (!server_init(&s->server)) {
return false;
}
bool ret = false;
bool server_started = false;
@ -305,7 +355,18 @@ scrcpy(struct scrcpy_options *options) {
.force_adb_forward = options->force_adb_forward,
.power_off_on_close = options->power_off_on_close,
};
if (!server_start(&s->server, &params)) {
static const struct server_callbacks cbs = {
.on_connection_failed = server_on_connection_failed,
.on_connected = server_on_connected,
.on_disconnected = server_on_disconnected,
};
if (!server_init(&s->server, &params, &cbs, NULL)) {
return false;
}
// TODO SDL_Init(SDL_INIT_EVENTS) before starting server
if (!server_start(&s->server)) {
goto end;
}
@ -316,15 +377,15 @@ scrcpy(struct scrcpy_options *options) {
goto end;
}
char device_name[DEVICE_NAME_FIELD_LENGTH];
struct sc_size frame_size;
if (!server_connect_to(&s->server, device_name, &frame_size)) {
// Await for server without blocking Ctrl+C handling
if (!await_for_server()) {
goto end;
}
struct server_info *info = &s->server.info;
if (options->display && options->control) {
if (!file_handler_init(&s->file_handler, s->server.serial,
if (!file_handler_init(&s->file_handler, options->serial,
options->push_target)) {
goto end;
}
@ -346,7 +407,7 @@ scrcpy(struct scrcpy_options *options) {
if (!recorder_init(&s->recorder,
options->record_filename,
options->record_format,
frame_size)) {
info->frame_size)) {
goto end;
}
rec = &s->recorder;
@ -392,11 +453,11 @@ scrcpy(struct scrcpy_options *options) {
if (options->display) {
const char *window_title =
options->window_title ? options->window_title : device_name;
options->window_title ? options->window_title : info->device_name;
struct screen_params screen_params = {
.window_title = window_title,
.frame_size = frame_size,
.frame_size = info->frame_size,
.always_on_top = options->always_on_top,
.window_x = options->window_x,
.window_y = options->window_y,
@ -419,8 +480,8 @@ scrcpy(struct scrcpy_options *options) {
#ifdef HAVE_V4L2
if (options->v4l2_device) {
if (!sc_v4l2_sink_init(&s->v4l2_sink, options->v4l2_device, frame_size,
options->v4l2_buffer)) {
if (!sc_v4l2_sink_init(&s->v4l2_sink, options->v4l2_device,
info->frame_size, options->v4l2_buffer)) {
goto end;
}

View file

@ -61,6 +61,44 @@ get_server_path(void) {
return server_path;
}
static void
server_params_destroy(struct server_params *params) {
// The server stores a copy of the params provided by the user
free((char *) params->serial);
free((char *) params->crop);
free((char *) params->codec_options);
free((char *) params->encoder_name);
}
static bool
server_params_copy(struct server_params *dst, const struct server_params *src) {
*dst = *src;
// The params reference user-allocated memory, so we must copy them to
// handle them from another thread
#define COPY(FIELD) \
dst->FIELD = NULL; \
if (src->FIELD) { \
dst->FIELD = strdup(src->FIELD); \
if (!dst->FIELD) { \
goto error; \
} \
}
COPY(serial);
COPY(crop);
COPY(codec_options);
COPY(encoder_name);
#undef COPY
return true;
error:
server_params_destroy(dst);
return false;
}
static bool
push_server(const char *serial) {
char *server_path = get_server_path();
@ -103,10 +141,11 @@ disable_tunnel_forward(const char *serial, uint16_t local_port) {
static bool
disable_tunnel(struct server *server) {
const char *serial = server->params.serial;
if (server->tunnel_forward) {
return disable_tunnel_forward(server->serial, server->local_port);
return disable_tunnel_forward(serial, server->local_port);
}
return disable_tunnel_reverse(server->serial);
return disable_tunnel_reverse(serial);
}
static sc_socket
@ -118,9 +157,10 @@ listen_on_port(uint16_t port) {
static bool
enable_tunnel_reverse_any_port(struct server *server,
struct sc_port_range port_range) {
const char *serial = server->params.serial;
uint16_t port = port_range.first;
for (;;) {
if (!enable_tunnel_reverse(server->serial, port)) {
if (!enable_tunnel_reverse(serial, port)) {
// the command itself failed, it will fail on any port
return false;
}
@ -139,7 +179,7 @@ enable_tunnel_reverse_any_port(struct server *server,
}
// failure, disable tunnel and try another port
if (!disable_tunnel_reverse(server->serial)) {
if (!disable_tunnel_reverse(serial)) {
LOGW("Could not remove reverse tunnel on port %" PRIu16, port);
}
@ -165,9 +205,11 @@ static bool
enable_tunnel_forward_any_port(struct server *server,
struct sc_port_range port_range) {
server->tunnel_forward = true;
const char *serial = server->params.serial;
uint16_t port = port_range.first;
for (;;) {
if (enable_tunnel_forward(server->serial, port)) {
if (enable_tunnel_forward(serial, port)) {
// success
server->local_port = port;
return true;
@ -229,6 +271,8 @@ log_level_to_server_string(enum sc_log_level level) {
static process_t
execute_server(struct server *server, const struct server_params *params) {
const char *serial = server->params.serial;
char max_size_string[6];
char bit_rate_string[11];
char max_fps_string[6];
@ -286,7 +330,7 @@ execute_server(struct server *server, const struct server_params *params) {
// Port: 5005
// Then click on "Debug"
#endif
return adb_execute(server->serial, cmd, ARRAY_LEN(cmd));
return adb_execute(serial, cmd, ARRAY_LEN(cmd));
}
static sc_socket
@ -316,6 +360,7 @@ connect_to_server(uint16_t port, uint32_t attempts, uint32_t delay) {
// it worked!
return socket;
}
// TODO use mutex + condvar + bool stopped
if (attempts) {
SDL_Delay(delay);
}
@ -324,22 +369,31 @@ connect_to_server(uint16_t port, uint32_t attempts, uint32_t delay) {
}
bool
server_init(struct server *server) {
server->serial = NULL;
server_init(struct server *server, const struct server_params *params,
const struct server_callbacks *cbs, void *cbs_userdata) {
bool ok = server_params_copy(&server->params, params);
if (!ok) {
LOGE("Could not copy server params");
return false;
}
server->process = PROCESS_NONE;
bool ok = sc_mutex_init(&server->mutex);
ok = sc_mutex_init(&server->mutex);
if (!ok) {
server_params_destroy(&server->params);
return false;
}
ok = sc_cond_init(&server->process_terminated_cond);
if (!ok) {
sc_mutex_destroy(&server->mutex);
server_params_destroy(&server->params);
return false;
}
server->process_terminated = false;
server->connected = false;
server->server_socket = SC_INVALID_SOCKET;
server->video_socket = SC_INVALID_SOCKET;
@ -350,13 +404,69 @@ server_init(struct server *server) {
server->tunnel_enabled = false;
server->tunnel_forward = false;
assert(cbs);
assert(cbs->on_connection_failed);
assert(cbs->on_connected);
assert(cbs->on_disconnected);
server->cbs = cbs;
server->cbs_userdata = cbs_userdata;
return true;
}
static int
run_wait_server(void *data) {
run_server_connect(void *data) {
struct server *server = data;
if (!server_connect_to(server, &server->info)) {
server->cbs->on_connection_failed(server, server->cbs_userdata);
goto end;
}
server->connected = true;
server->cbs->on_connected(server, server->cbs_userdata);
end:
return 0;
}
static int
run_server(void *data) {
struct server *server = data;
const struct server_params *params = &server->params;
if (!push_server(params->serial)) {
server->cbs->on_connection_failed(server, server->cbs_userdata);
goto end;
}
if (!enable_tunnel_any_port(server, params->port_range,
params->force_adb_forward)) {
server->cbs->on_connection_failed(server, server->cbs_userdata);
goto end;
}
server->process = execute_server(server, params);
if (server->process == PROCESS_NONE) {
server->cbs->on_connection_failed(server, server->cbs_userdata);
goto end;
}
sc_thread connect_thread;
bool ok = sc_thread_create(&connect_thread, run_server_connect,
"server-connect", server);
if (!ok) {
LOGW("Could not create thread, killing the server...");
process_terminate(server->process);
server->cbs->on_connection_failed(server, server->cbs_userdata);
process_wait(server->process, false); // ignore exit code
goto end;
}
process_wait(server->process, false); // ignore exit code
LOGD("Server terminated");
sc_mutex_lock(&server->mutex);
server->process_terminated = true;
@ -370,64 +480,26 @@ run_wait_server(void *data) {
net_interrupt(server->server_socket);
}
LOGD("Server terminated");
sc_thread_join(&connect_thread, NULL);
// Written by connect_thread, sc_thread_join() provides the necessary
// memory barrier
if (server->connected) {
server->cbs->on_disconnected(server, server->cbs_userdata);
}
// Otherwise, ->on_connection_failed() is already called
end:
return 0;
}
bool
server_start(struct server *server, const struct server_params *params) {
if (params->serial) {
server->serial = strdup(params->serial);
if (!server->serial) {
return false;
}
}
if (!push_server(params->serial)) {
/* server->serial will be freed on server_destroy() */
return false;
}
if (!enable_tunnel_any_port(server, params->port_range,
params->force_adb_forward)) {
return false;
}
// server will connect to our server socket
server->process = execute_server(server, params);
if (server->process == PROCESS_NONE) {
goto error;
}
// If the server process dies before connecting to the server socket, then
// the client will be stuck forever on accept(). To avoid the problem, we
// must be able to wake up the accept() call when the server dies. To keep
// things simple and multiplatform, just spawn a new thread waiting for the
// server process and calling shutdown()/close() on the server socket if
// necessary to wake up any accept() blocking call.
bool ok = sc_thread_create(&server->wait_server_thread, run_wait_server,
"wait-server", server);
if (!ok) {
process_terminate(server->process);
process_wait(server->process, true); // ignore exit code
goto error;
}
server->tunnel_enabled = true;
return true;
error:
// The server socket (if any) will be closed on server_destroy()
disable_tunnel(server);
return false;
server_start(struct server *server) {
return sc_thread_create(&server->thread, run_server, "server", server);
}
static bool
device_read_info(sc_socket device_socket, char *device_name,
struct sc_size *size) {
device_read_info(sc_socket device_socket, struct server_info *info) {
unsigned char buf[DEVICE_NAME_FIELD_LENGTH + 4];
ssize_t r = net_recv_all(device_socket, buf, sizeof(buf));
if (r < DEVICE_NAME_FIELD_LENGTH + 4) {
@ -436,19 +508,17 @@ device_read_info(sc_socket device_socket, char *device_name,
}
// in case the client sends garbage
buf[DEVICE_NAME_FIELD_LENGTH - 1] = '\0';
// strcpy is safe here, since name contains at least
// DEVICE_NAME_FIELD_LENGTH bytes and strlen(buf) < DEVICE_NAME_FIELD_LENGTH
strcpy(device_name, (char *) buf);
size->width = (buf[DEVICE_NAME_FIELD_LENGTH] << 8)
| buf[DEVICE_NAME_FIELD_LENGTH + 1];
size->height = (buf[DEVICE_NAME_FIELD_LENGTH + 2] << 8)
| buf[DEVICE_NAME_FIELD_LENGTH + 3];
memcpy(info->device_name, (char *) buf, sizeof(info->device_name));
info->frame_size.width = (buf[DEVICE_NAME_FIELD_LENGTH] << 8)
| buf[DEVICE_NAME_FIELD_LENGTH + 1];
info->frame_size.height = (buf[DEVICE_NAME_FIELD_LENGTH + 2] << 8)
| buf[DEVICE_NAME_FIELD_LENGTH + 3];
return true;
}
bool
server_connect_to(struct server *server, char *device_name,
struct sc_size *size) {
server_connect_to(struct server *server, struct server_info *info) {
if (!server->tunnel_forward) {
server->video_socket = net_accept(server->server_socket);
if (server->video_socket == SC_INVALID_SOCKET) {
@ -489,7 +559,7 @@ server_connect_to(struct server *server, char *device_name,
server->tunnel_enabled = false;
// The sockets will be closed on stop if device_read_info() fails
return device_read_info(server->video_socket, device_name, size);
return device_read_info(server->video_socket, info);
}
void
@ -538,7 +608,7 @@ server_stop(struct server *server) {
process_terminate(server->process);
}
sc_thread_join(&server->wait_server_thread, NULL);
sc_thread_join(&server->thread, NULL);
process_close(server->process);
}
@ -559,7 +629,8 @@ server_destroy(struct server *server) {
LOGW("Could not close control socket");
}
}
free(server->serial);
server_params_destroy(&server->params);
sc_cond_destroy(&server->process_terminated_cond);
sc_mutex_destroy(&server->mutex);
}

View file

@ -14,21 +14,10 @@
#include "util/net.h"
#include "util/thread.h"
struct server {
char *serial;
process_t process;
sc_thread wait_server_thread;
sc_mutex mutex;
sc_cond process_terminated_cond;
bool process_terminated;
sc_socket server_socket; // only used if !tunnel_forward
sc_socket video_socket;
sc_socket control_socket;
uint16_t local_port; // selected from port_range
bool tunnel_enabled;
bool tunnel_forward; // use "adb forward" instead of "adb reverse"
#define DEVICE_NAME_FIELD_LENGTH 64
struct server_info {
char device_name[DEVICE_NAME_FIELD_LENGTH];
struct sc_size frame_size;
};
struct server_params {
@ -50,20 +39,50 @@ struct server_params {
bool power_off_on_close;
};
// init default values
struct server {
// The internal allocated strings are copies owned by the server
struct server_params params;
process_t process;
sc_thread thread;
sc_mutex mutex;
sc_cond process_terminated_cond;
bool process_terminated;
bool connected; // written by connect_thread
struct server_info info; // initialized once connected
sc_socket server_socket; // only used if !tunnel_forward
sc_socket video_socket;
sc_socket control_socket;
uint16_t local_port; // selected from port_range
bool tunnel_enabled;
bool tunnel_forward; // use "adb forward" instead of "adb reverse"
const struct server_callbacks *cbs;
void *cbs_userdata;
};
struct server_callbacks {
void (*on_connection_failed)(struct server *server, void *userdata);
void (*on_connected)(struct server *server, void *userdata);
void (*on_disconnected)(struct server *server, void *userdata);
};
// init the server with the given params
bool
server_init(struct server *server);
server_init(struct server *server, const struct server_params *params,
const struct server_callbacks *cbs, void *cbs_userdata);
// push, enable tunnel et start the server
bool
server_start(struct server *server, const struct server_params *params);
server_start(struct server *server);
#define DEVICE_NAME_FIELD_LENGTH 64
// block until the communication with the server is established
// device_name must point to a buffer of at least DEVICE_NAME_FIELD_LENGTH bytes
bool
server_connect_to(struct server *server, char *device_name,
struct sc_size *size);
server_connect_to(struct server *server, struct server_info *info);
// disconnect and kill the server process
void