From: Aaron Krebs Date: Sat, 9 Aug 2014 15:54:56 +0000 (+0200) Subject: Replaced node_mutex with mesh_mutex. X-Git-Url: http://git.meshlink.io/?p=meshlink;a=commitdiff_plain;h=a387ef3cd807b90bbdf39581b6aea1bce9573f19 Replaced node_mutex with mesh_mutex. This reverses the mutex use; instead of locking when modifying nodes, we unlock when not modifying the mesh. Any API functions that wish to use the mesh must aquire the lock first. TODO: Currently there is a deadlock problem, and the discovery thread is not cleanly handled yet. --- diff --git a/src/event.c b/src/event.c index b6def622..62daea6f 100644 --- a/src/event.c +++ b/src/event.c @@ -177,7 +177,7 @@ void signal_del(event_loop_t *loop, signal_t *sig) { sig->cb = NULL; } -bool event_loop_run(event_loop_t *loop) { +bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex) { loop->running = true; fd_set readable; @@ -211,7 +211,12 @@ bool event_loop_run(event_loop_t *loop) { fds = last->fd + 1; } + // release mesh mutex during select + if(mutex) + pthread_mutex_unlock(mutex); int n = select(fds, &readable, &writable, NULL, tv); + if(mutex) + pthread_mutex_lock(mutex); if(n < 0) { if(sockwouldblock(errno)) diff --git a/src/event.h b/src/event.h index 6e77fbdf..4eb18251 100644 --- a/src/event.h +++ b/src/event.h @@ -86,7 +86,7 @@ extern void signal_del(event_loop_t *loop, signal_t *sig); extern void event_loop_init(event_loop_t *loop); extern void event_loop_exit(event_loop_t *loop); -extern bool event_loop_run(event_loop_t *loop); +extern bool event_loop_run(event_loop_t *loop, pthread_mutex_t *mutex); extern void event_loop_flush_output(event_loop_t *loop); extern void event_loop_stop(event_loop_t *loop); diff --git a/src/meshlink.c b/src/meshlink.c index 7f009d94..d028d428 100644 --- a/src/meshlink.c +++ b/src/meshlink.c @@ -750,6 +750,8 @@ meshlink_handle_t *meshlink_open_with_size(const char *confbase, const char *nam // Validate arguments provided by the application bool usingname = false; + + logger(NULL, MESHLINK_DEBUG, "meshlink_open called\n"); if(!confbase || !*confbase) { logger(NULL, MESHLINK_ERROR, "No confbase given!\n"); @@ -781,7 +783,13 @@ meshlink_handle_t *meshlink_open_with_size(const char *confbase, const char *nam mesh->appname = xstrdup(appname); mesh->dclass = dclass; if (usingname) mesh->name = xstrdup(name); - pthread_mutex_init ( &(mesh->nodes_mutex), NULL); + + // initialize mutex + pthread_mutexattr_t attr; + pthread_mutexattr_init(&attr); + pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE); + pthread_mutex_init(&(mesh->mesh_mutex), &attr); + mesh->threadstarted = false; event_loop_init(&mesh->loop); mesh->loop.data = mesh; @@ -830,29 +838,33 @@ meshlink_handle_t *meshlink_open_with_size(const char *confbase, const char *nam return NULL; } + logger(NULL, MESHLINK_DEBUG, "meshlink_open returning\n"); return mesh; } static void *meshlink_main_loop(void *arg) { meshlink_handle_t *mesh = arg; + pthread_mutex_lock(&(mesh->mesh_mutex)); + try_outgoing_connections(mesh); logger(mesh, MESHLINK_DEBUG, "Starting main_loop...\n"); main_loop(mesh); logger(mesh, MESHLINK_DEBUG, "main_loop returned.\n"); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } bool meshlink_start(meshlink_handle_t *mesh) { - - logger(mesh, MESHLINK_DEBUG, "meshlink_start called\n"); - if(!mesh) { meshlink_errno = MESHLINK_EINVAL; return false; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + + logger(mesh, MESHLINK_DEBUG, "meshlink_start called\n"); // TODO: open listening sockets first @@ -860,6 +872,7 @@ bool meshlink_start(meshlink_handle_t *mesh) { if(!mesh->name ) { logger(mesh, MESHLINK_DEBUG, "No name given!\n"); meshlink_errno = MESHLINK_EINVAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -869,6 +882,7 @@ bool meshlink_start(meshlink_handle_t *mesh) { logger(mesh, MESHLINK_DEBUG, "Could not start thread: %s\n", strerror(errno)); memset(&mesh->thread, 0, sizeof mesh->thread); meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -876,18 +890,19 @@ bool meshlink_start(meshlink_handle_t *mesh) { discovery_start(mesh); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return true; } void meshlink_stop(meshlink_handle_t *mesh) { - - logger(mesh, MESHLINK_DEBUG, "meshlink_stop called\n"); - if(!mesh) { meshlink_errno = MESHLINK_EINVAL; return; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + logger(mesh, MESHLINK_DEBUG, "meshlink_stop called\n"); + // Stop discovery discovery_stop(mesh); @@ -897,8 +912,10 @@ void meshlink_stop(meshlink_handle_t *mesh) { shutdown(s->tcp.fd, SHUT_RDWR); // Wait for the main thread to finish - + pthread_mutex_unlock(&(mesh->mesh_mutex)); pthread_join(mesh->thread, NULL); + pthread_mutex_lock(&(mesh->mesh_mutex)); + mesh->threadstarted = false; // Fix the socket @@ -910,6 +927,8 @@ void meshlink_stop(meshlink_handle_t *mesh) { logger(mesh, MESHLINK_ERROR, "Could not repair listenen socket!"); else io_add(&mesh->loop, &s->tcp, handle_new_meta_connection, s, s->tcp.fd, IO_READ); + + pthread_mutex_unlock(&(mesh->mesh_mutex)); } void meshlink_close(meshlink_handle_t *mesh) { @@ -918,6 +937,9 @@ void meshlink_close(meshlink_handle_t *mesh) { return; } + // lock is not released after this + pthread_mutex_lock(&(mesh->mesh_mutex)); + // Close and free all resources used. close_network_connections(mesh); @@ -937,6 +959,7 @@ void meshlink_close(meshlink_handle_t *mesh) { free(mesh->name); free(mesh->appname); free(mesh->confbase); + pthread_mutex_destroy(&(mesh->mesh_mutex)); memset(mesh, 0, sizeof *mesh); @@ -949,7 +972,9 @@ void meshlink_set_receive_cb(meshlink_handle_t *mesh, meshlink_receive_cb_t cb) return; } + pthread_mutex_lock(&(mesh->mesh_mutex)); mesh->receive_cb = cb; + pthread_mutex_unlock(&(mesh->mesh_mutex)); } void meshlink_set_node_status_cb(meshlink_handle_t *mesh, meshlink_node_status_cb_t cb) { @@ -958,13 +983,17 @@ void meshlink_set_node_status_cb(meshlink_handle_t *mesh, meshlink_node_status_c return; } + pthread_mutex_lock(&(mesh->mesh_mutex)); mesh->node_status_cb = cb; + pthread_mutex_unlock(&(mesh->mesh_mutex)); } void meshlink_set_log_cb(meshlink_handle_t *mesh, meshlink_log_level_t level, meshlink_log_cb_t cb) { if(mesh) { + pthread_mutex_lock(&(mesh->mesh_mutex)); mesh->log_cb = cb; mesh->log_level = cb ? level : 0; + pthread_mutex_unlock(&(mesh->mesh_mutex)); } else { global_log_cb = cb; global_log_level = cb ? level : 0; @@ -985,6 +1014,8 @@ bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const return false; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + //add packet to the queue outpacketqueue_t *packet_in_queue = xzalloc(sizeof *packet_in_queue); packet_in_queue->destination=destination; @@ -992,25 +1023,32 @@ bool meshlink_send(meshlink_handle_t *mesh, meshlink_node_t *destination, const packet_in_queue->len=len; if(!meshlink_queue_push(&mesh->outpacketqueue, packet_in_queue)) { free(packet_in_queue); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } //notify event loop signal_trigger(&(mesh->loop),&(mesh->datafromapp)); + + pthread_mutex_unlock(&(mesh->mesh_mutex)); return true; } void meshlink_send_from_queue(event_loop_t* el,meshlink_handle_t *mesh) { + pthread_mutex_lock(&(mesh->mesh_mutex)); + vpn_packet_t packet; meshlink_packethdr_t *hdr = (meshlink_packethdr_t *)packet.data; outpacketqueue_t* p = meshlink_queue_pop(&mesh->outpacketqueue); if(!p) + pthread_mutex_unlock(&(mesh->mesh_mutex)); return; if (sizeof(meshlink_packethdr_t) + p->len > MAXSIZE) { + pthread_mutex_unlock(&(mesh->mesh_mutex)); //log something - return ; + return; } packet.probe = false; @@ -1024,6 +1062,8 @@ void meshlink_send_from_queue(event_loop_t* el,meshlink_handle_t *mesh) { mesh->self->in_packets++; mesh->self->in_bytes += packet.len; route(mesh, mesh->self, &packet); + + pthread_mutex_unlock(&(mesh->mesh_mutex)); return ; } @@ -1032,14 +1072,22 @@ ssize_t meshlink_get_pmtu(meshlink_handle_t *mesh, meshlink_node_t *destination) meshlink_errno = MESHLINK_EINVAL; return -1; } + pthread_mutex_lock(&(mesh->mesh_mutex)); node_t *n = (node_t *)destination; - if(!n->status.reachable) + if(!n->status.reachable) { + pthread_mutex_unlock(&(mesh->mesh_mutex)); return 0; - else if(n->mtuprobes > 30 && n->minmtu) + + } + else if(n->mtuprobes > 30 && n->minmtu) { + pthread_mutex_unlock(&(mesh->mesh_mutex)); return n->minmtu; - else + } + else { + pthread_mutex_unlock(&(mesh->mesh_mutex)); return MTU; + } } char *meshlink_get_fingerprint(meshlink_handle_t *mesh, meshlink_node_t *node) { @@ -1047,11 +1095,13 @@ char *meshlink_get_fingerprint(meshlink_handle_t *mesh, meshlink_node_t *node) { meshlink_errno = MESHLINK_EINVAL; return NULL; } + pthread_mutex_lock(&(mesh->mesh_mutex)); node_t *n = (node_t *)node; if(!node_read_ecdsa_public_key(mesh, n) || !n->ecdsa) { meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1060,6 +1110,7 @@ char *meshlink_get_fingerprint(meshlink_handle_t *mesh, meshlink_node_t *node) { if(!fingerprint) meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return fingerprint; } @@ -1069,7 +1120,12 @@ meshlink_node_t *meshlink_get_node(meshlink_handle_t *mesh, const char *name) { return NULL; } - return (meshlink_node_t *)lookup_node(mesh, (char *)name); // TODO: make lookup_node() use const + meshlink_node_t *node = NULL; + + pthread_mutex_lock(&(mesh->mesh_mutex)); + node = (meshlink_node_t *)lookup_node(mesh, (char *)name); // TODO: make lookup_node() use const + pthread_mutex_unlock(&(mesh->mesh_mutex)); + return node; } meshlink_node_t **meshlink_get_all_nodes(meshlink_handle_t *mesh, meshlink_node_t **nodes, size_t *nmemb) { @@ -1081,7 +1137,7 @@ meshlink_node_t **meshlink_get_all_nodes(meshlink_handle_t *mesh, meshlink_node_ meshlink_node_t **result; //lock mesh->nodes - pthread_mutex_lock(&(mesh->nodes_mutex)); + pthread_mutex_lock(&(mesh->mesh_mutex)); *nmemb = mesh->nodes->count; result = realloc(nodes, *nmemb * sizeof *nodes); @@ -1096,7 +1152,7 @@ meshlink_node_t **meshlink_get_all_nodes(meshlink_handle_t *mesh, meshlink_node_ meshlink_errno = MESHLINK_ENOMEM; } - pthread_mutex_unlock(&(mesh->nodes_mutex)); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return result; } @@ -1112,12 +1168,16 @@ bool meshlink_sign(meshlink_handle_t *mesh, const void *data, size_t len, void * return false; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + if(!ecdsa_sign(mesh->self->connection->ecdsa, data, len, signature)) { meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } *siglen = MESHLINK_SIGLEN; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return true; } @@ -1132,23 +1192,32 @@ bool meshlink_verify(meshlink_handle_t *mesh, meshlink_node_t *source, const voi return false; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + + bool rval = false; + struct node_t *n = (struct node_t *)source; node_read_ecdsa_public_key(mesh, n); if(!n->ecdsa) { meshlink_errno = MESHLINK_EINTERNAL; - return false; + rval = false; + } else { + rval = ecdsa_verify(((struct node_t *)source)->ecdsa, data, len, signature); } - - return ecdsa_verify(((struct node_t *)source)->ecdsa, data, len, signature); + pthread_mutex_unlock(&(mesh->mesh_mutex)); + return rval; } static bool refresh_invitation_key(meshlink_handle_t *mesh) { char filename[PATH_MAX]; + + pthread_mutex_lock(&(mesh->mesh_mutex)); snprintf(filename, sizeof filename, "%s" SLASH "invitations", mesh->confbase); if(mkdir(filename, 0700) && errno != EEXIST) { logger(mesh, MESHLINK_DEBUG, "Could not create directory %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1157,6 +1226,7 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { if(!dir) { logger(mesh, MESHLINK_DEBUG, "Could not read directory %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1186,6 +1256,7 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { logger(mesh, MESHLINK_DEBUG, "Error while reading directory %s: %s\n", filename, strerror(errno)); closedir(dir); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1202,8 +1273,10 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { } } - if(mesh->invitation_key) + if(mesh->invitation_key) { + pthread_mutex_unlock(&(mesh->mesh_mutex)); return true; + } // Create a new key if necessary. FILE *f = fopen(filename, "r"); @@ -1211,6 +1284,7 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { if(errno != ENOENT) { logger(mesh, MESHLINK_DEBUG, "Could not read %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1218,12 +1292,14 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { if(!mesh->invitation_key) { logger(mesh, MESHLINK_DEBUG, "Could not generate a new key!\n"); meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } f = fopen(filename, "w"); if(!f) { logger(mesh, MESHLINK_DEBUG, "Could not write %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } chmod(filename, 0600); @@ -1238,6 +1314,7 @@ static bool refresh_invitation_key(meshlink_handle_t *mesh) { } } + pthread_mutex_unlock(&(mesh->mesh_mutex)); return mesh->invitation_key; } @@ -1246,16 +1323,23 @@ bool meshlink_add_address(meshlink_handle_t *mesh, const char *address) { meshlink_errno = MESHLINK_EINVAL; return false; } + + bool rval = false; + + pthread_mutex_lock(&(mesh->mesh_mutex)); for(const char *p = address; *p; p++) { if(isalnum(*p) || *p == '-' || *p == '.' || *p == ':') continue; logger(mesh, MESHLINK_DEBUG, "Invalid character in address: %s\n", address); meshlink_errno = MESHLINK_EINVAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } - return append_config_file(mesh, mesh->self->name, "Address", address); + rval = append_config_file(mesh, mesh->self->name, "Address", address); + pthread_mutex_unlock(&(mesh->mesh_mutex)); + return rval; } char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { @@ -1263,11 +1347,14 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { meshlink_errno = MESHLINK_EINVAL; return NULL; } + + pthread_mutex_lock(&(mesh->mesh_mutex)); // Check validity of the new node's name if(!check_id(name)) { logger(mesh, MESHLINK_DEBUG, "Invalid name for node.\n"); meshlink_errno = MESHLINK_EINVAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1277,6 +1364,7 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { if(!access(filename, F_OK)) { logger(mesh, MESHLINK_DEBUG, "A host config file for %s already exists!\n", name); meshlink_errno = MESHLINK_EEXIST; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1284,6 +1372,7 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { if(meshlink_get_node(mesh, name)) { logger(mesh, MESHLINK_DEBUG, "A node with name %s is already known!\n", name); meshlink_errno = MESHLINK_EEXIST; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1292,11 +1381,13 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { if(!address) { logger(mesh, MESHLINK_DEBUG, "No Address known for ourselves!\n"); meshlink_errno = MESHLINK_ERESOLV; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } if(!refresh_invitation_key(mesh)) { meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1329,6 +1420,7 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { if(!ifd) { logger(mesh, MESHLINK_DEBUG, "Could not create invitation file %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } FILE *f = fdopen(ifd, "w"); @@ -1359,6 +1451,7 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { } else { logger(mesh, MESHLINK_DEBUG, "Could not create %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1374,6 +1467,7 @@ char *meshlink_invite(meshlink_handle_t *mesh, const char *name) { xasprintf(&url, "%s/%s%s", address, hash, cookie); free(address); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return url; } @@ -1382,6 +1476,8 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { meshlink_errno = MESHLINK_EINVAL; return false; } + + pthread_mutex_lock(&(mesh->mesh_mutex)); //TODO: think of a better name for this variable, or of a different way to tokenize the invitation URL. char copy[strlen(invitation) + 1]; @@ -1424,6 +1520,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { ecdsa_t *key = ecdsa_generate(); if(!key) { meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1438,6 +1535,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { struct addrinfo *ai = str2addrinfo(address, port, SOCK_STREAM); if(!ai) { meshlink_errno = MESHLINK_ERESOLV; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1446,6 +1544,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { logger(mesh, MESHLINK_DEBUG, "Could not open socket: %s\n", strerror(errno)); freeaddrinfo(ai); meshlink_errno = MESHLINK_ENETWORK; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1454,6 +1553,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { closesocket(mesh->sock); freeaddrinfo(ai); meshlink_errno = MESHLINK_ENETWORK; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1469,6 +1569,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { logger(mesh, MESHLINK_DEBUG, "Error sending request to %s port %s: %s\n", address, port, strerror(errno)); closesocket(mesh->sock); meshlink_errno = MESHLINK_ENETWORK; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1481,6 +1582,7 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { logger(mesh, MESHLINK_DEBUG, "Cannot read greeting from peer\n"); closesocket(mesh->sock); meshlink_errno = MESHLINK_ENETWORK; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1490,11 +1592,13 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(sha512(fingerprint, strlen(fingerprint), hishash)) { logger(mesh, MESHLINK_DEBUG, "Could not create hash\n%s\n", mesh->line + 2); meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } if(memcmp(hishash, mesh->hash, 18)) { logger(mesh, MESHLINK_DEBUG, "Peer has an invalid key!\n%s\n", mesh->line + 2); meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1502,18 +1606,21 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { ecdsa_t *hiskey = ecdsa_set_base64_public_key(fingerprint); if(!hiskey) { meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } // Start an SPTPS session if(!sptps_start(&mesh->sptps, mesh, true, false, key, hiskey, "meshlink invitation", 15, invitation_send, invitation_receive)) { meshlink_errno = MESHLINK_EINTERNAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } // Feed rest of input buffer to SPTPS if(!sptps_receive_data(&mesh->sptps, mesh->buffer, mesh->blen)) { meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1525,11 +1632,13 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { continue; logger(mesh, MESHLINK_DEBUG, "Error reading data from %s port %s: %s\n", address, port, strerror(errno)); meshlink_errno = MESHLINK_ENETWORK; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } if(!sptps_receive_data(&mesh->sptps, mesh->line, len)) { meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } } @@ -1542,14 +1651,17 @@ bool meshlink_join(meshlink_handle_t *mesh, const char *invitation) { if(!mesh->success) { logger(mesh, MESHLINK_DEBUG, "Connection closed by peer, invitation cancelled.\n"); meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } + pthread_mutex_unlock(&(mesh->mesh_mutex)); return true; invalid: logger(mesh, MESHLINK_DEBUG, "Invalid invitation URL or you are already connected to a Mesh ?\n"); meshlink_errno = MESHLINK_EINVAL; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1559,12 +1671,15 @@ char *meshlink_export(meshlink_handle_t *mesh) { return NULL; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + char filename[PATH_MAX]; snprintf(filename, sizeof filename, "%s" SLASH "hosts" SLASH "%s", mesh->confbase, mesh->self->name); FILE *f = fopen(filename, "r"); if(!f) { logger(mesh, MESHLINK_DEBUG, "Could not open %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } @@ -1579,11 +1694,14 @@ char *meshlink_export(meshlink_handle_t *mesh) { logger(mesh, MESHLINK_DEBUG, "Error reading from %s: %s\n", filename, strerror(errno)); fclose(f); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return NULL; } fclose(f); buf[len - 1] = 0; + + pthread_mutex_lock(&(mesh->mesh_mutex)); return buf; } @@ -1592,10 +1710,13 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { meshlink_errno = MESHLINK_EINVAL; return false; } + + pthread_mutex_lock(&(mesh->mesh_mutex)); if(strncmp(data, "Name = ", 7)) { logger(mesh, MESHLINK_DEBUG, "Invalid data\n"); meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1603,6 +1724,7 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { if(!end) { logger(mesh, MESHLINK_DEBUG, "Invalid data\n"); meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1613,6 +1735,7 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { if(!check_id(name)) { logger(mesh, MESHLINK_DEBUG, "Invalid Name\n"); meshlink_errno = MESHLINK_EPEER; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1621,12 +1744,14 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { if(!access(filename, F_OK)) { logger(mesh, MESHLINK_DEBUG, "File %s already exists, not importing\n", filename); meshlink_errno = MESHLINK_EEXIST; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } if(errno != ENOENT) { logger(mesh, MESHLINK_DEBUG, "Error accessing %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1634,6 +1759,7 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { if(!f) { logger(mesh, MESHLINK_DEBUG, "Could not create %s: %s\n", filename, strerror(errno)); meshlink_errno = MESHLINK_ESTORAGE; + pthread_mutex_unlock(&(mesh->mesh_mutex)); return false; } @@ -1642,6 +1768,7 @@ bool meshlink_import(meshlink_handle_t *mesh, const char *data) { load_all_nodes(mesh); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return true; } @@ -1651,6 +1778,8 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) { return; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + node_t *n; n = (node_t*)node; n->status.blacklisted=true; @@ -1659,6 +1788,7 @@ void meshlink_blacklist(meshlink_handle_t *mesh, meshlink_node_t *node) { //Make blacklisting persistent in the config file append_config_file(mesh, n->name, "blacklisted", "yes"); + pthread_mutex_unlock(&(mesh->mesh_mutex)); return; } @@ -1668,11 +1798,14 @@ void meshlink_whitelist(meshlink_handle_t *mesh, meshlink_node_t *node) { return; } + pthread_mutex_lock(&(mesh->mesh_mutex)); + node_t *n = (node_t *)node; n->status.blacklisted = false; //TODO: remove blacklisted = yes from the config file + pthread_mutex_unlock(&(mesh->mesh_mutex)); return; } @@ -1683,6 +1816,8 @@ extern void meshlink_hint_address(meshlink_handle_t *mesh, meshlink_node_t *node if(!mesh || !node || !addr) return; + pthread_mutex_lock(&(mesh->mesh_mutex)); + char *host = NULL, *port = NULL, *str = NULL; sockaddr2str((const sockaddr_t *)addr, &host, &port); @@ -1695,6 +1830,7 @@ extern void meshlink_hint_address(meshlink_handle_t *mesh, meshlink_node_t *node free(host); free(port); + pthread_mutex_unlock(&(mesh->mesh_mutex)); // @TODO do we want to fire off a connection attempt right away? } diff --git a/src/meshlink_internal.h b/src/meshlink_internal.h index 63a71dc0..42e1da53 100644 --- a/src/meshlink_internal.h +++ b/src/meshlink_internal.h @@ -77,7 +77,7 @@ struct meshlink_handle { pthread_t thread; bool threadstarted; pthread_mutex_t outpacketqueue_mutex; - pthread_mutex_t nodes_mutex; + pthread_mutex_t mesh_mutex; event_loop_t loop; listen_socket_t listen_socket[MAXSOCKETS]; int listen_sockets; diff --git a/src/net.c b/src/net.c index 4045c136..c615b893 100644 --- a/src/net.c +++ b/src/net.c @@ -355,7 +355,7 @@ int main_loop(meshlink_handle_t *mesh) { mesh->datafromapp.signum = 0; signal_add(&(mesh->loop),&(mesh->datafromapp), (signal_cb_t)meshlink_send_from_queue,mesh, mesh->datafromapp.signum); - if(!event_loop_run(&mesh->loop)) { + if(!event_loop_run(&(mesh->loop), &(mesh->mesh_mutex))) { logger(mesh, MESHLINK_ERROR, "Error while waiting for input: %s", strerror(errno)); return 1; } diff --git a/src/node.c b/src/node.c index e6b9f018..f43619a2 100644 --- a/src/node.c +++ b/src/node.c @@ -34,21 +34,17 @@ static int node_compare(const node_t *a, const node_t *b) { } void init_nodes(meshlink_handle_t *mesh) { - pthread_mutex_lock(&(mesh->nodes_mutex)); mesh->nodes = splay_alloc_tree((splay_compare_t) node_compare, (splay_action_t) free_node); mesh->node_udp_cache = hash_alloc(0x100, sizeof(sockaddr_t)); - pthread_mutex_unlock(&(mesh->nodes_mutex)); } void exit_nodes(meshlink_handle_t *mesh) { - pthread_mutex_lock(&(mesh->nodes_mutex)); if(mesh->node_udp_cache) hash_free(mesh->node_udp_cache); if(mesh->nodes) splay_delete_tree(mesh->nodes); mesh->node_udp_cache = NULL; mesh->nodes = NULL; - pthread_mutex_unlock(&(mesh->nodes_mutex)); } node_t *new_node(void) { @@ -87,30 +83,24 @@ void free_node(node_t *n) { } void node_add(meshlink_handle_t *mesh, node_t *n) { - pthread_mutex_lock(&(mesh->nodes_mutex)); n->mesh = mesh; splay_insert(mesh->nodes, n); - pthread_mutex_unlock(&(mesh->nodes_mutex)); } void node_del(meshlink_handle_t *mesh, node_t *n) { - pthread_mutex_lock(&(mesh->nodes_mutex)); timeout_del(&mesh->loop, &n->mtutimeout); for splay_each(edge_t, e, n->edge_tree) edge_del(mesh, e); splay_delete(mesh->nodes, n); - pthread_mutex_unlock(&(mesh->nodes_mutex)); } node_t *lookup_node(meshlink_handle_t *mesh, const char *name) { const node_t n = {.name = (char *)name}; node_t *result; - pthread_mutex_lock(&(mesh->nodes_mutex)); result = splay_search(mesh->nodes, &n); - pthread_mutex_unlock(&(mesh->nodes_mutex)); return result; }