diff options
Diffstat (limited to 'src/ot_livesync.c')
| -rw-r--r-- | src/ot_livesync.c | 46 |
1 files changed, 36 insertions, 10 deletions
diff --git a/src/ot_livesync.c b/src/ot_livesync.c index 083fe1a..85577f7 100644 --- a/src/ot_livesync.c +++ b/src/ot_livesync.c | |||
| @@ -4,6 +4,7 @@ | |||
| 4 | /* System */ | 4 | /* System */ |
| 5 | #include <arpa/inet.h> | 5 | #include <arpa/inet.h> |
| 6 | #include <pthread.h> | 6 | #include <pthread.h> |
| 7 | #include <stdio.h> | ||
| 7 | #include <stdlib.h> | 8 | #include <stdlib.h> |
| 8 | #include <string.h> | 9 | #include <string.h> |
| 9 | #include <sys/socket.h> | 10 | #include <sys/socket.h> |
| @@ -21,7 +22,10 @@ | |||
| 21 | 22 | ||
| 22 | #ifdef WANT_SYNC_LIVE | 23 | #ifdef WANT_SYNC_LIVE |
| 23 | 24 | ||
| 24 | ot_ip6 groupip_1 = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5}; | 25 | /* The multicast group, legacy opentrackers joined by default 224.0.23.5, so it |
| 26 | stays the default */ | ||
| 27 | static ot_ip6 g_group = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, 224, 0, 23, 5}; | ||
| 28 | static uint16_t g_port = LIVESYNC_PORT; | ||
| 25 | 29 | ||
| 26 | #define LIVESYNC_INCOMING_BUFFSIZE (256 * 256) | 30 | #define LIVESYNC_INCOMING_BUFFSIZE (256 * 256) |
| 27 | 31 | ||
| @@ -59,7 +63,7 @@ void livesync_init(void) { | |||
| 59 | uint32_t packet_type4 = htonl(OT_SYNC_PEER4); | 63 | uint32_t packet_type4 = htonl(OT_SYNC_PEER4); |
| 60 | 64 | ||
| 61 | if (g_socket_in == -1) | 65 | if (g_socket_in == -1) |
| 62 | exerr("No socket address for live sync specified."); | 66 | return; |
| 63 | 67 | ||
| 64 | /* Prepare outgoing peers buffer */ | 68 | /* Prepare outgoing peers buffer */ |
| 65 | memcpy(g_v6_buf.data, &g_tracker_id, sizeof(g_tracker_id)); | 69 | memcpy(g_v6_buf.data, &g_tracker_id, sizeof(g_tracker_id)); |
| @@ -94,6 +98,14 @@ void livesync_deinit(void) { | |||
| 94 | close(g_socket_out); | 98 | close(g_socket_out); |
| 95 | } | 99 | } |
| 96 | 100 | ||
| 101 | int livesync_set_group(const ot_ip6 group) { | ||
| 102 | /* v4 multicast is 224.0.0.0/4, and the socket layer has nothing else */ | ||
| 103 | if (!ip_is_v4mapped(group) || (group[12] & 0xf0) != 0xe0) | ||
| 104 | return -1; | ||
| 105 | memcpy(g_group, group, sizeof(ot_ip6)); | ||
| 106 | return 0; | ||
| 107 | } | ||
| 108 | |||
| 97 | void livesync_bind_mcast(ot_ip6 ip, uint16_t port) { | 109 | void livesync_bind_mcast(ot_ip6 ip, uint16_t port) { |
| 98 | if (!ip_is_v4mapped(ip)) | 110 | if (!ip_is_v4mapped(ip)) |
| 99 | exerr("v6 mcast support not yet available."); | 111 | exerr("v6 mcast support not yet available."); |
| @@ -101,7 +113,9 @@ void livesync_bind_mcast(ot_ip6 ip, uint16_t port) { | |||
| 101 | if (g_socket_in != -1) | 113 | if (g_socket_in != -1) |
| 102 | exerr("Error: Livesync listen ip specified twice."); | 114 | exerr("Error: Livesync listen ip specified twice."); |
| 103 | 115 | ||
| 104 | if ((g_socket_in = socket_mcast_join(groupip_1, ip, port)) == -1) | 116 | g_port = port; |
| 117 | |||
| 118 | if ((g_socket_in = socket_mcast_join(g_group, ip, port)) == -1) | ||
| 105 | exerr("Error: Cant bind live sync incoming socket."); | 119 | exerr("Error: Cant bind live sync incoming socket."); |
| 106 | 120 | ||
| 107 | if ((g_socket_out = socket_mcast_bind(ip, port, 0)) == -1) | 121 | if ((g_socket_out = socket_mcast_bind(ip, port, 0)) == -1) |
| @@ -121,7 +135,7 @@ static void livesync_issue_peersync(sync_buffer *buf) { | |||
| 121 | has modified the protected element */ | 135 | has modified the protected element */ |
| 122 | pthread_mutex_unlock(&g_outbuf_mutex); | 136 | pthread_mutex_unlock(&g_outbuf_mutex); |
| 123 | 137 | ||
| 124 | socket_mcast_send(g_socket_out, mycopy, fill, groupip_1, LIVESYNC_PORT); | 138 | socket_mcast_send(g_socket_out, mycopy, fill, g_group, g_port); |
| 125 | } | 139 | } |
| 126 | 140 | ||
| 127 | static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size) { | 141 | static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size) { |
| @@ -159,6 +173,9 @@ static void livesync_handle_peersync(struct ot_workstruct *ws, size_t peer_size) | |||
| 159 | stuck when there's not enough traffic to fill udp packets fast | 173 | stuck when there's not enough traffic to fill udp packets fast |
| 160 | enough */ | 174 | enough */ |
| 161 | void livesync_ticker(void) { | 175 | void livesync_ticker(void) { |
| 176 | if (g_socket_out == -1) | ||
| 177 | return; | ||
| 178 | |||
| 162 | /* livesync_issue_peersync sets g_next_packet_time */ | 179 | /* livesync_issue_peersync sets g_next_packet_time */ |
| 163 | pthread_mutex_lock(&g_outbuf_mutex); | 180 | pthread_mutex_lock(&g_outbuf_mutex); |
| 164 | if (time_now_seconds() > g_v6_buf.next_packet_time && g_v6_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t)) | 181 | if (time_now_seconds() > g_v6_buf.next_packet_time && g_v6_buf.fill > sizeof(g_tracker_id) + sizeof(uint32_t)) |
| @@ -176,8 +193,14 @@ void livesync_ticker(void) { | |||
| 176 | /* Inform live sync about whats going on. */ | 193 | /* Inform live sync about whats going on. */ |
| 177 | void livesync_tell(struct ot_workstruct *ws) { | 194 | void livesync_tell(struct ot_workstruct *ws) { |
| 178 | size_t peer_size; /* initialized in next line */ | 195 | size_t peer_size; /* initialized in next line */ |
| 179 | ot_peer *peer_src = peer_from_peer6(&ws->peer, &peer_size); | 196 | ot_peer *peer_src; |
| 180 | sync_buffer *dest_buf = peer_size == OT_PEER_SIZE6 ? &g_v6_buf : &g_v4_buf; | 197 | sync_buffer *dest_buf; |
| 198 | |||
| 199 | if (g_socket_out == -1) | ||
| 200 | return; | ||
| 201 | |||
| 202 | peer_src = peer_from_peer6(&ws->peer, &peer_size); | ||
| 203 | dest_buf = peer_size == OT_PEER_SIZE6 ? &g_v6_buf : &g_v4_buf; | ||
| 181 | 204 | ||
| 182 | pthread_mutex_lock(&g_outbuf_mutex); | 205 | pthread_mutex_lock(&g_outbuf_mutex); |
| 183 | 206 | ||
| @@ -200,9 +223,12 @@ static void *livesync_worker(void *args) { | |||
| 200 | 223 | ||
| 201 | (void)args; | 224 | (void)args; |
| 202 | 225 | ||
| 203 | /* Initialize our "thread local storage" */ | 226 | if (workstruct_init(&ws, LIVESYNC_INCOMING_BUFFSIZE) == -1) { |
| 204 | ws.inbuf = ws.request = malloc(LIVESYNC_INCOMING_BUFFSIZE); | 227 | fprintf(stderr, "Warning: live sync could not get its buffers and will not run.\n"); |
| 205 | ws.outbuf = ws.reply = 0; | 228 | atomic_store(&thread_left, 1); |
| 229 | return NULL; | ||
| 230 | } | ||
| 231 | ws.request = ws.inbuf; | ||
| 206 | 232 | ||
| 207 | while (g_opentracker_running) { | 233 | while (g_opentracker_running) { |
| 208 | uint32_t packet_type; | 234 | uint32_t packet_type; |
| @@ -233,7 +259,7 @@ static void *livesync_worker(void *args) { | |||
| 233 | } | 259 | } |
| 234 | } | 260 | } |
| 235 | 261 | ||
| 236 | free(ws.inbuf); | 262 | workstruct_deinit(&ws); |
| 237 | atomic_store(&thread_left, 1); | 263 | atomic_store(&thread_left, 1); |
| 238 | return NULL; | 264 | return NULL; |
| 239 | } | 265 | } |
