/* * LASH * * Copyright (C) 2002 Robert Ham * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program; if not, write to the Free Software * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. */ #define _GNU_SOURCE /* addrinfo */ #include #include #include #include #include #include #include #include #include #include #include #include #include "conn_mgr.h" #include "conn.h" #include "server.h" void *conn_mgr_recv_run(void *data); void *conn_mgr_send_run(void *data); void conn_mgr_free(conn_mgr_t * conn_mgr) { conn_t *conn; lash_list_t *list; close(conn_mgr->listen_socket); for (list = conn_mgr->open_connections; list; list = lash_list_next(list)) { conn = (conn_t *) list->data; LASH_DEBUGARGS("closing open connection %ld ('%s')", conn->id, conn_get_str_id(conn)); conn_destroy(conn); } lash_list_free(conn_mgr->open_connections); conn_mgr->open_connections = NULL; for (list = conn_mgr->connections; list; list = list->next) { conn = (conn_t *) list->data; LASH_DEBUGARGS("closing connection %ld ('%s')", conn->id, conn_get_str_id(conn)); conn_destroy(conn); } lash_list_free(conn_mgr->connections); conn_mgr->connections = NULL; conn_mgr->listen_socket = 0; FD_ZERO(&conn_mgr->sockets); conn_mgr->fd_max = 0; conn_mgr->client_events = NULL; pthread_mutex_destroy(&conn_mgr->connections_lock); pthread_mutex_destroy(&conn_mgr->client_event_lock); pthread_cond_destroy(&conn_mgr->client_event_cond); } conn_mgr_t * conn_mgr_new(server_t * server) { conn_mgr_t *mgr; int err; mgr = lash_malloc0(sizeof(conn_mgr_t)); pthread_mutex_init(&mgr->connections_lock, NULL); pthread_mutex_init(&mgr->client_event_lock, NULL); pthread_cond_init(&mgr->client_event_cond, NULL); FD_ZERO(&mgr->sockets); mgr->server = server; err = conn_mgr_start(mgr); if (err) { if (mgr) conn_mgr_destroy(mgr); return NULL; } return mgr; } void conn_mgr_destroy(conn_mgr_t * conn_mgr) { LASH_PRINT_DEBUG("stopping"); conn_mgr_stop(conn_mgr); LASH_PRINT_DEBUG("stopped"); conn_mgr_free(conn_mgr); free(conn_mgr); } static int conn_mgr_set_socket_opts(int sock) { int err; int reuse; /* * reuse ports. this is so we can bind again quickly after shutting down. */ reuse = 1; err = setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)); if (err == -1) { fprintf(stderr, "%s: could not set SO_REUSEADDR on socket: %s\n", __FUNCTION__, strerror(errno)); } /* * make sure socket is closed on exec() */ err = fcntl(sock, F_SETFD, FD_CLOEXEC); if (err == -1) { fprintf(stderr, "%s: could not set close-on-exec on listen socket: %s\n", __FUNCTION__, strerror(errno)); return -1; } return 0; } int conn_mgr_bind_socket(int *sock, struct addrinfo *addr) { int err; *sock = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol); if (*sock == -1) { LASH_DEBUGARGS ("could not create socket with params domain=%d, type=%d, protocol=%d: %s", addr->ai_family, addr->ai_socktype, addr->ai_protocol, strerror(errno)); return -1; } conn_mgr_set_socket_opts(*sock); err = bind(*sock, addr->ai_addr, addr->ai_addrlen); if (err) { LASH_DEBUGARGS("could not bind socket: %s", strerror(errno)); err = close(*sock); if (err) { fprintf(stderr, "%s: error closing unconnected socket: %s", __FUNCTION__, strerror(errno)); } return -1; } return 0; } int conn_mgr_start(conn_mgr_t * conn_mgr) { struct addrinfo hints; struct addrinfo *addrs; struct addrinfo *addr; int bound = 0; int err; memset(&hints, 0, sizeof(hints)); hints.ai_socktype = SOCK_STREAM; hints.ai_flags = AI_PASSIVE; err = getaddrinfo(NULL, LASH_PORT, &hints, &addrs); if (err) { fprintf(stderr, "%s: could not look up service name: %s\n", __FUNCTION__, gai_strerror(err)); return -1; } /* try ipv6 first */ if (!no_v6) for (addr = addrs; addr; addr = addr->ai_next) { if (addr->ai_family == AF_INET6) { err = conn_mgr_bind_socket(&conn_mgr->listen_socket, addr); if (err) continue; else { bound = 1; break; } } } if (!bound) for (addr = addrs; addr; addr = addr->ai_next) { if (no_v6 && addr->ai_family == AF_INET6) continue; err = conn_mgr_bind_socket(&conn_mgr->listen_socket, addr); if (err) continue; else { bound = 1; break; } } freeaddrinfo(addrs); if (!bound) { fprintf(stderr, "%s: could not create listen socket\n", __FUNCTION__); return -1; } /* start the listening */ err = listen(conn_mgr->listen_socket, 20); if (err == -1) { fprintf(stderr, "%s: error setting socket to listen: %s\n", __FUNCTION__, strerror(errno)); return -1; } printf("Listening for connections\n"); FD_SET(conn_mgr->listen_socket, &conn_mgr->sockets); conn_mgr->fd_max = conn_mgr->listen_socket; /* start up the threads */ LASH_PRINT_DEBUG("starting recv thread"); err = pthread_create(&conn_mgr->recv_thread, NULL, conn_mgr_recv_run, conn_mgr); if (err) { fprintf(stderr, "%s: could not start recv thread\n", __FUNCTION__); abort(); } LASH_PRINT_DEBUG("recv thread started; starting send thread"); pthread_create(&conn_mgr->send_thread, NULL, conn_mgr_send_run, conn_mgr); LASH_PRINT_DEBUG("send thread started"); return 0; } void conn_mgr_stop(conn_mgr_t * conn_mgr) { conn_mgr->quit = 1; pthread_cond_signal(&conn_mgr->client_event_cond); if (conn_mgr->send_thread) pthread_join(conn_mgr->send_thread, NULL); if (conn_mgr->recv_thread) pthread_join(conn_mgr->recv_thread, NULL); } void conn_mgr_send_client_event(conn_mgr_t * conn_mgr, server_event_t * event) { pthread_mutex_lock(&conn_mgr->client_event_lock); conn_mgr->client_events = lash_list_append(conn_mgr->client_events, event); pthread_mutex_unlock(&conn_mgr->client_event_lock); pthread_cond_signal(&conn_mgr->client_event_cond); } void conn_mgr_send_client_lash_event(conn_mgr_t * conn_mgr, unsigned long conn_id, lash_event_t * lash_event) { server_event_t *server_event; if (!lash_event) return; server_event = server_event_new(); server_event_set_conn_id(server_event, conn_id); server_event_set_lash_event(server_event, lash_event); conn_mgr_send_client_event(conn_mgr, server_event); } void conn_mgr_send_client_lash_config(conn_mgr_t * conn_mgr, unsigned int conn_id, lash_config_t * lash_config) { server_event_t *server_event; if (!lash_config) return; server_event = server_event_new(); server_event_set_conn_id(server_event, conn_id); server_event_set_lash_config(server_event, lash_config); conn_mgr_send_client_event(conn_mgr, server_event); } void conn_mgr_send_client_lash_comm_event(conn_mgr_t * conn_mgr, unsigned int conn_id, lash_comm_event_t * event) { server_event_t *server_event; if (!event) return; server_event = server_event_new(); server_event_set_conn_id(server_event, conn_id); server_event_set_lash_comm_event(server_event, event); conn_mgr_send_client_event(conn_mgr, server_event); } void conn_mgr_send_server_lash_event(conn_mgr_t * conn_mgr, unsigned long conn_id, lash_event_t * lash_event) { server_event_t *server_event; if (!lash_event) return; server_event = server_event_new(); server_event_set_conn_id(server_event, conn_id); server_event_set_lash_event(server_event, lash_event); server_send_event(conn_mgr->server, server_event); } void conn_mgr_send_server_lash_config(conn_mgr_t * conn_mgr, unsigned int conn_id, lash_config_t * lash_config) { server_event_t *server_event; if (!lash_config) return; server_event = server_event_new(); server_event_set_conn_id(server_event, conn_id); server_event_set_lash_config(server_event, lash_config); server_send_event(conn_mgr->server, server_event); } void conn_mgr_send_server_lash_comm_event(conn_mgr_t * conn_mgr, unsigned int conn_id, lash_comm_event_t * event) { server_event_t *server_event; if (!event) return; server_event = server_event_new(); server_event_set_conn_id(server_event, conn_id); server_event_set_lash_comm_event(server_event, event); server_send_event(conn_mgr->server, server_event); } /*********************************** *********** recv thread *********** ***********************************/ void conn_mgr_accept_connection(conn_mgr_t * conn_mgr) { conn_t *conn; /*struct sockaddr_storage ss; size_t ss_len;*/ int err; conn = conn_new(); /* accept the connection */ conn->socket = accept(conn_mgr->listen_socket, NULL, NULL); /*(struct sockaddr *) &ss, * &ss_len); */ if (conn->socket == -1) { fprintf(stderr, "%s: error accepting socket connection from '%s': %s\n", __FUNCTION__, conn_get_str_id(conn), strerror(errno)); conn_destroy(conn); return; } LASH_DEBUGARGS("accepted connection from '%s' with id %ld", conn_get_str_id(conn), conn->id); /* * make sure socket is closed on exec() */ err = fcntl(conn_mgr->listen_socket, F_SETFD, FD_CLOEXEC); if (err == -1) { fprintf(stderr, "%s: could not set close-on-exec on connection %ld ('%s'): %s\n", __FUNCTION__, conn->id, conn_get_str_id(conn), strerror(errno)); exit(1); } /* socket stuff */ FD_SET(conn->socket, &conn_mgr->sockets); if (conn->socket > conn_mgr->fd_max) conn_mgr->fd_max = conn->socket; /* add the connection to the open connection list */ conn_mgr->open_connections = lash_list_append(conn_mgr->open_connections, conn); } void conn_mgr_connect_client(conn_mgr_t * conn_mgr, conn_t * conn) { lash_comm_event_t *event; server_event_t *server_event; int err; lash_connect_params_t *params; char id[37]; /* read the Connect event */ event = lash_comm_event_new(); LASH_PRINT_DEBUG("recieving Connect"); err = lash_comm_recv_event(conn->socket, event); if (err == -1 || err == -2 || err == -3 || lash_comm_event_get_type(event) != LASH_Comm_Event_Connect || !LASH_PROTOCOL_IS_VALID(event->event_data.connect->protocol_version)) { /* there was an error */ if (err == -1) fprintf(stderr, "%s: there was a recieve error from connection '%s': disconnecting client\n", __FUNCTION__, conn_get_str_id(conn)); else if (err == -2) fprintf(stderr, "%s: connection '%s' closed before sending Connect event\n", __FUNCTION__, conn_get_str_id(conn)); else if (err == -3) fprintf(stderr, "%s: connection '%s' is using the wrong low-level protocol version\n", __FUNCTION__, conn_get_str_id(conn)); else if (lash_comm_event_get_type(event) != LASH_Comm_Event_Connect) fprintf(stderr, "%s: connection '%s' sent an event (of type %d) that wasn't Connect before it was connected; removing\n", __FUNCTION__, conn_get_str_id(conn), lash_comm_event_get_type(event)); else if (!LASH_PROTOCOL_IS_VALID (event->event_data.connect->protocol_version)) { lash_comm_event_t mismatch_event; fprintf(stderr, "%s: connection '%s' is using protocol %s; disconnecting\n", __FUNCTION__, conn_get_str_id(conn), lash_protocol_string(event->event_data.connect-> protocol_version)); lash_comm_event_set_protocol_mismatch(&mismatch_event, LASH_PROTOCOL_VERSION); lash_comm_send_event(conn->socket, &mismatch_event); } lash_comm_event_destroy(event); FD_CLR(conn->socket, &conn_mgr->sockets); conn_mgr->open_connections = lash_list_remove(conn_mgr->open_connections, conn); conn_destroy(conn); return; } params = event->event_data.connect; uuid_unparse(params->id, id); LASH_PRINT_DEBUG("connecting new client with:"); LASH_DEBUGARGS(" connection id: %ld", conn->id); LASH_DEBUGARGS(" source address: %s", conn_get_str_id(conn)); LASH_DEBUGARGS(" protocol: %s", lash_protocol_string(event->event_data.connect-> protocol_version)); LASH_DEBUGARGS(" class: '%s'", params->class); LASH_DEBUGARGS(" id: '%s'", id); LASH_DEBUGARGS(" working dir: '%s'", params->working_dir); LASH_DEBUGARGS(" requested proj: '%s'", params->project); LASH_DEBUGARGS(" flags: %d", params->flags); LASH_DEBUGARGS(" %d args:", params->argc); for (err = 0; err < params->argc; err++) { LASH_DEBUGARGS(" arg %d: '%s'", err, params->argv[err]); } /* tell the server there's a new client connection */ server_event = server_event_new(); server_event_set_conn_id(server_event, conn->id); server_event_set_lash_connect_params(server_event, event->event_data.connect); server_send_event(conn_mgr->server, server_event); event->event_data.connect = NULL; lash_comm_event_destroy(event); /* remove the connection from the open list */ conn_mgr->open_connections = lash_list_remove(conn_mgr->open_connections, conn); /* add the connection to the connection list */ pthread_mutex_lock(&conn_mgr->connections_lock); conn_mgr->connections = lash_list_append(conn_mgr->connections, conn); pthread_mutex_unlock(&conn_mgr->connections_lock); conn_set_recv_stamp(conn); } void conn_mgr_disconnect_client(conn_mgr_t * conn_mgr, conn_t * conn, int notify_server) { unsigned long conn_id; /* clear up what we know of the connection */ conn_mgr->connections = lash_list_remove(conn_mgr->connections, conn); conn_id = conn->id; FD_CLR(conn->socket, &conn_mgr->sockets); conn_unlock(conn); conn_destroy(conn); if (notify_server) { server_event_t *server_event; server_event = server_event_new(); server_event_set_type(server_event, Client_Disconnect); server_event_set_conn_id(server_event, conn_id); server_send_event(conn_mgr->server, server_event); } LASH_DEBUGARGS("disconnected client on connection %ld", conn_id); } void conn_mgr_deal_with_event(conn_mgr_t * conn_mgr, conn_t * conn, lash_comm_event_t * event) { switch (lash_comm_event_get_type(event)) { case LASH_Comm_Event_Event: conn_mgr_send_server_lash_event(conn_mgr, conn->id, lash_comm_event_take_event(event)); lash_comm_event_destroy(event); break; case LASH_Comm_Event_Config: conn_mgr_send_server_lash_config(conn_mgr, conn->id, lash_comm_event_take_config(event)); lash_comm_event_destroy(event); break; case LASH_Comm_Event_Ping: lash_comm_event_set_type(event, LASH_Comm_Event_Pong); conn_mgr_send_client_lash_comm_event(conn_mgr, conn->id, event); break; case LASH_Comm_Event_Pong: lash_comm_event_destroy(event); break; default: break; } } void conn_mgr_recv_event(conn_mgr_t * conn_mgr, conn_t * conn) { int err; lash_comm_event_t *event; /* recieve the event */ event = lash_comm_event_new(); err = lash_comm_recv_event(conn->socket, event); if (err == -1) { fprintf(stderr, "%s: error recieving event from connection '%s'\n", __FUNCTION__, conn_get_str_id(conn)); return; } /* connection disconnect */ if (err == -2 || lash_comm_event_get_type(event) == LASH_Comm_Event_Close) { LASH_DEBUGARGS("%s: connection '%s' disconnected", conn_get_str_id(conn)); if (lash_comm_event_get_type(event) == LASH_Comm_Event_Close) lash_comm_event_destroy(event); conn_mgr_disconnect_client(conn_mgr, conn, 1); /* done what we might have done */ pthread_mutex_unlock(&conn_mgr->connections_lock); return; } /* give up the lock here, as we're definately not messing with it now */ pthread_mutex_unlock(&conn_mgr->connections_lock); /* this function will destroy the event */ conn_mgr_deal_with_event(conn_mgr, conn, event); /* reset ping */ conn_set_recv_stamp(conn); /* unlock the connection */ conn_unlock(conn); } void conn_mgr_read_socket(conn_mgr_t * conn_mgr, int socket) { lash_list_t *list; conn_t *conn; /* check if it's a new connection */ if (socket == conn_mgr->listen_socket) { conn_mgr_accept_connection(conn_mgr); return; } /* see if it's an open connection */ list = conn_mgr->open_connections; while (list) { conn = (conn_t *) list->data; if (conn->socket == socket) { conn_mgr_connect_client(conn_mgr, conn); return; } list = list->next; } /* else find it in the existing connections */ pthread_mutex_lock(&conn_mgr->connections_lock); list = conn_mgr->connections; while (list) { conn = (conn_t *) list->data; if (conn->socket == socket) { conn_lock(conn); /* must unlock connection! */ conn_mgr_recv_event(conn_mgr, conn); return; } list = list->next; } fprintf(stderr, "%s: could not find a connection with socket %d! this should, like, never happen. you shouldn't be reading this. if you are, bad things have happened. your computer will likely melt after about 20 minutes.", __FUNCTION__, socket); abort(); } void conn_mgr_check_timeouts(conn_mgr_t * conn_mgr) { lash_list_t *list, *next; conn_t *conn; time_t now; now = time(NULL); if (now == (time_t) - 1) { fprintf(stderr, "%s: could not get time, aborting!: %s\n", __FUNCTION__, strerror(errno)); abort(); } pthread_mutex_lock(&conn_mgr->connections_lock); list = conn_mgr->connections; while (list) { conn = (conn_t *) list->data; next = list->next; conn_lock(conn); if (conn_get_pinged(conn)) { if (conn_ping_timed_out(conn, now)) { fprintf(stderr, "%s: connection '%ld' has not responded to ping for %ld seconds, disconnecting it\n", __FUNCTION__, conn->id, CONN_TIMEOUT); conn_mgr_disconnect_client(conn_mgr, conn, 1); } else conn_unlock(conn); } else { if (conn_recv_timed_out(conn, now)) { /* send a ping */ lash_comm_event_t *event; /* LASH_DEBUGARGS ("pinging connection '%ld'", conn_get_id (conn)); */ event = lash_comm_event_new(); lash_comm_event_set_type(event, LASH_Comm_Event_Ping); conn_mgr_send_client_lash_comm_event(conn_mgr, conn->id, event); conn_set_ping_stamp(conn); } conn_unlock(conn); } list = next; } pthread_mutex_unlock(&conn_mgr->connections_lock); } void * conn_mgr_recv_run(void *data) { conn_mgr_t *conn_mgr; fd_set sockets; int err, i; struct timeval select_timeout; conn_mgr = (conn_mgr_t *) data; while (!conn_mgr->quit) { sockets = conn_mgr->sockets; select_timeout.tv_sec = 1; select_timeout.tv_usec = 0; err = select(conn_mgr->fd_max + 1, &sockets, NULL, NULL, &select_timeout); if (err == -1) { if (errno == EINTR) continue; fprintf(stderr, "%s: error calling select(): %s\n", __FUNCTION__, strerror(errno)); return NULL; } if (conn_mgr->quit) break; for (i = 0; i <= conn_mgr->fd_max; i++) { if (FD_ISSET(i, &sockets)) { conn_mgr_read_socket(conn_mgr, i); } } conn_mgr_check_timeouts(conn_mgr); } LASH_PRINT_DEBUG("finished"); return NULL; } /**************************************** ************ send thread *************** ****************************************/ void conn_mgr_send_lash_comm_event_to_client(conn_t * conn, lash_comm_event_t * lash_comm_event) { int err; err = lash_comm_send_event(conn->socket, lash_comm_event); if (err == -1) { fprintf(stderr, "%s: could not send event to client\n", __FUNCTION__); } lash_comm_event_destroy(lash_comm_event); } void conn_mgr_send_disconnect_client(conn_mgr_t * conn_mgr, unsigned long conn_id) { lash_list_t *list; conn_t *conn; pthread_mutex_lock(&conn_mgr->connections_lock); for (list = conn_mgr->connections; list; list = list->next) { conn = (conn_t *) list->data; if (conn->id == conn_id) { /* remove it */ conn_lock(conn); conn_mgr_disconnect_client(conn_mgr, conn, 0); pthread_mutex_unlock(&conn_mgr->connections_lock); return; } } pthread_mutex_unlock(&conn_mgr->connections_lock); fprintf(stderr, "%s: request from server to remove unknown connection %ld\n", __FUNCTION__, conn_id); } void conn_mgr_send_server_event_to_client(conn_mgr_t * conn_mgr, server_event_t * server_event) { lash_comm_event_t *lash_comm_event = NULL; unsigned long conn_id; lash_list_t *list; conn_t *conn; conn_id = server_event->conn_id; /* extract the comm event */ switch (server_event->type) { case Client_Event: lash_comm_event = lash_comm_event_new(); lash_comm_event_set_event(lash_comm_event, server_event_take_lash_event(server_event)); break; case Client_Config: lash_comm_event = lash_comm_event_new(); lash_comm_event_set_config(lash_comm_event, server_event_take_lash_config (server_event)); break; case Client_Comm_Event: lash_comm_event = server_event_take_lash_comm_event(server_event); break; case Client_Disconnect: conn_mgr_send_disconnect_client(conn_mgr, server_event->conn_id); server_event_destroy(server_event); return; case Client_Connect: default: fprintf(stderr, "%s: recieved unknown send request of type %d from server\n", __FUNCTION__, server_event->type); server_event_destroy(server_event); return; } server_event_destroy(server_event); /* attempt to find the connection */ pthread_mutex_lock(&conn_mgr->connections_lock); list = conn_mgr->connections; while (list) { conn = (conn_t *) list->data; if (conn->id == conn_id) { /* send it */ conn_lock(conn); pthread_mutex_unlock(&conn_mgr->connections_lock); conn_mgr_send_lash_comm_event_to_client(conn, lash_comm_event); conn_unlock(conn); return; } list = list->next; } pthread_mutex_unlock(&conn_mgr->connections_lock); fprintf(stderr, "%s: could not send event to unknown connection id %ld\n", __FUNCTION__, conn_id); } void * conn_mgr_send_run(void *data) { conn_mgr_t *conn_mgr; lash_list_t *list; server_event_t *server_event; LASH_PRINT_DEBUG("send thread starting"); conn_mgr = (conn_mgr_t *) data; while (!conn_mgr->quit) { pthread_mutex_lock(&conn_mgr->client_event_lock); list = conn_mgr->client_events; if (list) conn_mgr->client_events = NULL; else { pthread_cond_wait(&conn_mgr->client_event_cond, &conn_mgr->client_event_lock); list = conn_mgr->client_events; conn_mgr->client_events = NULL; } pthread_mutex_unlock(&conn_mgr->client_event_lock); if (conn_mgr->quit) { break; } while (list) { server_event = (server_event_t *) list->data; conn_mgr_send_server_event_to_client(conn_mgr, server_event); list = list->next; } } LASH_PRINT_DEBUG("finished"); return NULL; } /* EOF */