Added input and output threads

This commit is contained in:
Michael Jumper 2011-03-16 01:24:17 -07:00
parent a9f702ed27
commit 3c878e1d59
5 changed files with 108 additions and 51 deletions

View File

@ -45,6 +45,7 @@ AC_PROG_LIBTOOL
# Checks for libraries. # Checks for libraries.
AC_CHECK_LIB([dl], [dlopen],, AC_MSG_ERROR("libdl is required for loading client plugins")) AC_CHECK_LIB([dl], [dlopen],, AC_MSG_ERROR("libdl is required for loading client plugins"))
AC_CHECK_LIB([png], [png_write_png],, AC_MSG_ERROR("libpng is required for writing png messages")) AC_CHECK_LIB([png], [png_write_png],, AC_MSG_ERROR("libpng is required for writing png messages"))
AC_CHECK_LIB([pthread], [pthread_create])
AC_CHECK_LIB([wsock32], [main]) AC_CHECK_LIB([wsock32], [main])
# Checks for header files. # Checks for header files.

View File

@ -67,19 +67,6 @@
*/ */
#define GUAC_SYNC_FREQUENCY 5000 #define GUAC_SYNC_FREQUENCY 5000
/**
* The number of milliseconds to wait for messages in any phase before
* timing out and closing the connection with an error.
*/
#define GUAC_TIMEOUT 15000
/**
* The number of microseconds to wait for messages in any phase before
* timing out and closing the conncetion with an error. This is always
* equal to GUAC_TIMEOUT * 1000.
*/
#define GUAC_USEC_TIMEOUT (GUAC_TIMEOUT*1000)
/** /**
* The amount of time to wait after handling server messages. If a client * The amount of time to wait after handling server messages. If a client
* plugin has a message handler, and sends instructions when server messages * plugin has a message handler, and sends instructions when server messages

View File

@ -51,6 +51,19 @@
*/ */
/**
* The number of milliseconds to wait for messages in any phase before
* timing out and closing the connection with an error.
*/
#define GUAC_TIMEOUT 15000
/**
* The number of microseconds to wait for messages in any phase before
* timing out and closing the conncetion with an error. This is always
* equal to GUAC_TIMEOUT * 1000.
*/
#define GUAC_USEC_TIMEOUT (GUAC_TIMEOUT*1000)
/** /**
* Represents a single instruction within the Guacamole protocol. * Represents a single instruction within the Guacamole protocol.
*/ */

View File

@ -44,6 +44,7 @@
#endif #endif
#include <string.h> #include <string.h>
#include <dlfcn.h> #include <dlfcn.h>
#include <pthread.h>
#include "log.h" #include "log.h"
#include "guacio.h" #include "guacio.h"
@ -122,7 +123,7 @@ guac_client* guac_get_client(int client_fd) {
int result; int result;
/* Wait for data until timeout */ /* Wait for data until timeout */
result = guac_select(io, GUAC_USEC_TIMEOUT); result = guac_instructions_waiting(io);
if (result == 0) { if (result == 0) {
guac_send_error(io, "Select timeout."); guac_send_error(io, "Select timeout.");
guac_close(io); guac_close(io);
@ -212,7 +213,7 @@ guac_client* guac_get_client(int client_fd) {
int result; int result;
/* Wait for data until timeout */ /* Wait for data until timeout */
result = guac_select(io, GUAC_USEC_TIMEOUT); result = guac_instructions_waiting(io);
if (result == 0) { if (result == 0) {
guac_send_error(io, "Connect timeout."); guac_send_error(io, "Connect timeout.");
guac_close(io); guac_close(io);
@ -324,33 +325,31 @@ void __guac_sleep(int millis) {
} }
typedef struct __guac_client_thread_common {
void guac_start_client(guac_client* client) { guac_client* client;
GUACIO* io = client->io;
guac_instruction instruction;
int wait_result;
long last_received_timestamp; long last_received_timestamp;
long last_sent_timestamp; long last_sent_timestamp;
/* Init timestamps */ int client_active;
last_received_timestamp = last_sent_timestamp = __guac_current_timestamp();
/* VNC Client Loop */ } __guac_client_thread_common;
for (;;) {
/* Get current time and check timeout */
void* __guac_client_output_thread(void* data) {
__guac_client_thread_common* common = (__guac_client_thread_common*) data;
guac_client* client = common->client;
GUACIO* io = client->io;
/* Guacamole client output loop */
while (common->client_active) {
/* Occasionally ping client with sync */
long timestamp = __guac_current_timestamp(); long timestamp = __guac_current_timestamp();
if (timestamp - last_received_timestamp > GUAC_TIMEOUT) { if (timestamp - common->last_sent_timestamp > GUAC_SYNC_FREQUENCY) {
guac_send_error(io, "Sync timeout."); guac_send_sync(io, timestamp);
guac_flush(io); common->last_sent_timestamp = timestamp;
return;
}
/* If not timed out, ping client with sync */
if (timestamp - last_sent_timestamp > GUAC_SYNC_FREQUENCY) {
last_sent_timestamp = timestamp;
guac_send_sync(io, last_sent_timestamp);
guac_flush(io); guac_flush(io);
} }
@ -361,13 +360,13 @@ void guac_start_client(guac_client* client) {
int last_total_written = io->total_written; int last_total_written = io->total_written;
/* Only handle messages if synced within threshold */ /* Only handle messages if synced within threshold */
if (last_sent_timestamp - last_received_timestamp if (common->last_sent_timestamp - common->last_received_timestamp
< GUAC_SYNC_THRESHOLD) { < GUAC_SYNC_THRESHOLD) {
int retval = client->handle_messages(client); int retval = client->handle_messages(client);
if (retval) { if (retval) {
GUAC_LOG_ERROR("Error handling server messages"); GUAC_LOG_ERROR("Error handling server messages");
return; break;
} }
/* If data was written during message handling */ /* If data was written during message handling */
@ -377,8 +376,8 @@ void guac_start_client(guac_client* client) {
__guac_sleep(GUAC_SERVER_MESSAGE_HANDLE_FREQUENCY); __guac_sleep(GUAC_SERVER_MESSAGE_HANDLE_FREQUENCY);
/* Update sync timestamp and send sync instruction */ /* Update sync timestamp and send sync instruction */
last_sent_timestamp = __guac_current_timestamp(); common->last_sent_timestamp = __guac_current_timestamp();
guac_send_sync(io, last_sent_timestamp); guac_send_sync(io, common->last_sent_timestamp);
} }
@ -387,7 +386,29 @@ void guac_start_client(guac_client* client) {
} }
wait_result = guac_instructions_waiting(io); /* If no message handler, just sleep until next sync ping */
else
__guac_sleep(GUAC_SYNC_FREQUENCY);
} /* End of output loop */
common->client_active = 0;
return NULL;
}
void* __guac_client_input_thread(void* data) {
__guac_client_thread_common* common = (__guac_client_thread_common*) data;
guac_client* client = common->client;
GUACIO* io = client->io;
guac_instruction instruction;
/* Guacamole client input loop */
while (common->client_active) {
int wait_result = guac_instructions_waiting(io);
if (wait_result > 0) { if (wait_result > 0) {
int retval; int retval;
@ -398,11 +419,11 @@ void guac_start_client(guac_client* client) {
do { do {
if (strcmp(instruction.opcode, "sync") == 0) { if (strcmp(instruction.opcode, "sync") == 0) {
last_received_timestamp = atol(instruction.argv[0]); common->last_received_timestamp = atol(instruction.argv[0]);
if (last_received_timestamp > last_sent_timestamp) { if (common->last_received_timestamp > common->last_sent_timestamp) {
guac_send_error(io, "Received sync from future."); guac_send_error(io, "Received sync from future.");
guac_free_instruction_data(&instruction); guac_free_instruction_data(&instruction);
return; break;
} }
} }
else if (strcmp(instruction.opcode, "mouse") == 0) { else if (strcmp(instruction.opcode, "mouse") == 0) {
@ -418,7 +439,7 @@ void guac_start_client(guac_client* client) {
GUAC_LOG_ERROR("Error handling mouse instruction"); GUAC_LOG_ERROR("Error handling mouse instruction");
guac_free_instruction_data(&instruction); guac_free_instruction_data(&instruction);
return; break;
} }
} }
@ -435,7 +456,7 @@ void guac_start_client(guac_client* client) {
GUAC_LOG_ERROR("Error handling key instruction"); GUAC_LOG_ERROR("Error handling key instruction");
guac_free_instruction_data(&instruction); guac_free_instruction_data(&instruction);
return; break;
} }
} }
@ -451,7 +472,7 @@ void guac_start_client(guac_client* client) {
GUAC_LOG_ERROR("Error handling clipboard instruction"); GUAC_LOG_ERROR("Error handling clipboard instruction");
guac_free_instruction_data(&instruction); guac_free_instruction_data(&instruction);
return; break;
} }
} }
@ -459,7 +480,7 @@ void guac_start_client(guac_client* client) {
else if (strcmp(instruction.opcode, "disconnect") == 0) { else if (strcmp(instruction.opcode, "disconnect") == 0) {
GUAC_LOG_INFO("Client requested disconnect"); GUAC_LOG_INFO("Client requested disconnect");
guac_free_instruction_data(&instruction); guac_free_instruction_data(&instruction);
return; break;
} }
guac_free_instruction_data(&instruction); guac_free_instruction_data(&instruction);
@ -468,13 +489,13 @@ void guac_start_client(guac_client* client) {
if (retval < 0) { if (retval < 0) {
GUAC_LOG_ERROR("Error reading instruction from stream"); GUAC_LOG_ERROR("Error reading instruction from stream");
return; break;
} }
} }
if (retval < 0) { if (retval < 0) {
GUAC_LOG_ERROR("Error or end of stream"); GUAC_LOG_ERROR("Error or end of stream");
return; /* EOF or error */ break; /* EOF or error */
} }
/* Otherwise, retval == 0 implies unfinished instruction */ /* Otherwise, retval == 0 implies unfinished instruction */
@ -482,10 +503,45 @@ void guac_start_client(guac_client* client) {
} }
else if (wait_result < 0) { else if (wait_result < 0) {
GUAC_LOG_ERROR("Error waiting for next instruction"); GUAC_LOG_ERROR("Error waiting for next instruction");
break;
}
else { /* wait_result == 0 */
GUAC_LOG_ERROR("Timeout");
break;
}
}
common->client_active = 0;
return NULL;
}
void guac_start_client(guac_client* client) {
pthread_t input_thread, output_thread;
__guac_client_thread_common common;
/* Init thread data */
common.client = client;
common.last_received_timestamp = common.last_sent_timestamp = __guac_current_timestamp();
common.client_active = 1;
if (pthread_create(&output_thread, NULL, __guac_client_output_thread, (void*) &common)) {
/* THIS FUNCTION SHOULD RETURN A VALUE! */
return; return;
} }
if (pthread_create(&input_thread, NULL, __guac_client_input_thread, (void*) &common)) {
/* THIS FUNCTION SHOULD RETURN A VALUE! */
return;
} }
/* Wait for I/O threads */
pthread_join(input_thread, NULL);
pthread_join(output_thread, NULL);
/* Done */
} }

View File

@ -529,6 +529,6 @@ int guac_instructions_waiting(GUACIO* io) {
if (io->instructionbuf_used_length > 0) if (io->instructionbuf_used_length > 0)
return 1; return 1;
return guac_select(io, 1000); return guac_select(io, GUAC_USEC_TIMEOUT);
} }