diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/opentracker.c | 2 | ||||
| -rw-r--r-- | src/ot_config.c | 6 | ||||
| -rw-r--r-- | src/ot_livesync.c | 46 | ||||
| -rw-r--r-- | src/ot_livesync.h | 7 | ||||
| -rw-r--r-- | src/ot_worker.c | 2 | ||||
| -rw-r--r-- | src/trackerlogic.c | 6 | ||||
| -rw-r--r-- | src/trackerlogic.h | 2 |
7 files changed, 55 insertions, 16 deletions
diff --git a/src/opentracker.c b/src/opentracker.c index 14b817e..60a6d8c 100644 --- a/src/opentracker.c +++ b/src/opentracker.c | |||
| @@ -109,7 +109,7 @@ static void *server_mainloop(void *args) { | |||
| 109 | 109 | ||
| 110 | (void)args; | 110 | (void)args; |
| 111 | 111 | ||
| 112 | if (workstruct_init(&ws) == -1) | 112 | if (workstruct_init(&ws, G_INBUF_SIZE) == -1) |
| 113 | panic("Initializing worker failed"); | 113 | panic("Initializing worker failed"); |
| 114 | 114 | ||
| 115 | while (g_opentracker_running) { | 115 | while (g_opentracker_running) { |
diff --git a/src/ot_config.c b/src/ot_config.c index 95136b5..4064d1b 100644 --- a/src/ot_config.c +++ b/src/ot_config.c | |||
| @@ -208,6 +208,7 @@ typedef enum { | |||
| 208 | CONFIG_TRACKER_REDIRECT_URL, | 208 | CONFIG_TRACKER_REDIRECT_URL, |
| 209 | CONFIG_LIVESYNC_CLUSTER_NODE_IP, | 209 | CONFIG_LIVESYNC_CLUSTER_NODE_IP, |
| 210 | CONFIG_LIVESYNC_CLUSTER_LISTEN, | 210 | CONFIG_LIVESYNC_CLUSTER_LISTEN, |
| 211 | CONFIG_LIVESYNC_CLUSTER_GROUP, | ||
| 211 | } config_kind; | 212 | } config_kind; |
| 212 | 213 | ||
| 213 | typedef struct { | 214 | typedef struct { |
| @@ -247,6 +248,7 @@ static const config_key config_keys[] = { | |||
| 247 | #ifdef WANT_SYNC_LIVE | 248 | #ifdef WANT_SYNC_LIVE |
| 248 | CONFIG_KEY("livesync.cluster.node_ip", CONFIG_LIVESYNC_CLUSTER_NODE_IP), | 249 | CONFIG_KEY("livesync.cluster.node_ip", CONFIG_LIVESYNC_CLUSTER_NODE_IP), |
| 249 | CONFIG_KEY("livesync.cluster.listen", CONFIG_LIVESYNC_CLUSTER_LISTEN), | 250 | CONFIG_KEY("livesync.cluster.listen", CONFIG_LIVESYNC_CLUSTER_LISTEN), |
| 251 | CONFIG_KEY("livesync.cluster.group", CONFIG_LIVESYNC_CLUSTER_GROUP), | ||
| 250 | #endif | 252 | #endif |
| 251 | }; | 253 | }; |
| 252 | static const size_t config_entry_count = sizeof(config_keys) / sizeof(config_keys[0]); | 254 | static const size_t config_entry_count = sizeof(config_keys) / sizeof(config_keys[0]); |
| @@ -378,6 +380,10 @@ static int parse_configfile(char *config_filename) { | |||
| 378 | set_config_option(&g_redirecturl, p); | 380 | set_config_option(&g_redirecturl, p); |
| 379 | break; | 381 | break; |
| 380 | #ifdef WANT_SYNC_LIVE | 382 | #ifdef WANT_SYNC_LIVE |
| 383 | case CONFIG_LIVESYNC_CLUSTER_GROUP: | ||
| 384 | if (!ip_parse(p, tmpip) || livesync_set_group(tmpip) == -1) | ||
| 385 | goto parse_error; | ||
| 386 | break; | ||
| 381 | case CONFIG_LIVESYNC_CLUSTER_NODE_IP: | 387 | case CONFIG_LIVESYNC_CLUSTER_NODE_IP: |
| 382 | if (!scan_ip6_net(p, &tmpnet)) | 388 | if (!scan_ip6_net(p, &tmpnet)) |
| 383 | goto parse_error; | 389 | goto parse_error; |
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 | } |
diff --git a/src/ot_livesync.h b/src/ot_livesync.h index 4bd3af0..cf3c64c 100644 --- a/src/ot_livesync.h +++ b/src/ot_livesync.h | |||
| @@ -7,7 +7,8 @@ | |||
| 7 | #include "trackerlogic.h" | 7 | #include "trackerlogic.h" |
| 8 | 8 | ||
| 9 | /* | 9 | /* |
| 10 | Syncing is done as udp packets in the multicast domain 224.0.42.5 port 9696 | 10 | Syncing is done as udp packets in the multicast group 224.0.23.5, on the |
| 11 | port livesync.cluster.listen names, 9696 by default | ||
| 11 | 12 | ||
| 12 | Each tracker should join the multicast group and send its live sync packets | 13 | Each tracker should join the multicast group and send its live sync packets |
| 13 | to that group, using a ttl of 1 | 14 | to that group, using a ttl of 1 |
| @@ -47,6 +48,10 @@ | |||
| 47 | void livesync_init(void); | 48 | void livesync_init(void); |
| 48 | void livesync_deinit(void); | 49 | void livesync_deinit(void); |
| 49 | 50 | ||
| 51 | /* The group to join and to send to, (defaults to 224.0.23.5). | ||
| 52 | Returns -1 if group is not a v4 multicast address. */ | ||
| 53 | int livesync_set_group(const ot_ip6 group); | ||
| 54 | |||
| 50 | /* Join multicast group for listening and create sending socket */ | 55 | /* Join multicast group for listening and create sending socket */ |
| 51 | void livesync_bind_mcast(ot_ip6 ip, uint16_t port); | 56 | void livesync_bind_mcast(ot_ip6 ip, uint16_t port); |
| 52 | 57 | ||
diff --git a/src/ot_worker.c b/src/ot_worker.c index 432d272..ef50945 100644 --- a/src/ot_worker.c +++ b/src/ot_worker.c | |||
| @@ -20,7 +20,7 @@ static size_t g_worker_count; | |||
| 20 | static void *worker_trampoline(void *args) { | 20 | static void *worker_trampoline(void *args) { |
| 21 | struct ot_worker *me = args; | 21 | struct ot_worker *me = args; |
| 22 | 22 | ||
| 23 | if (workstruct_init(&me->ws) == -1) | 23 | if (workstruct_init(&me->ws, G_INBUF_SIZE) == -1) |
| 24 | fprintf(stderr, "Warning: a worker could not get its buffers and will not run.\n"); | 24 | fprintf(stderr, "Warning: a worker could not get its buffers and will not run.\n"); |
| 25 | else { | 25 | else { |
| 26 | me->run(me); | 26 | me->run(me); |
diff --git a/src/trackerlogic.c b/src/trackerlogic.c index 9565090..8e30a48 100644 --- a/src/trackerlogic.c +++ b/src/trackerlogic.c | |||
| @@ -567,9 +567,11 @@ void trackerlogic_init(void) { | |||
| 567 | stats_init(); | 567 | stats_init(); |
| 568 | } | 568 | } |
| 569 | 569 | ||
| 570 | int workstruct_init(struct ot_workstruct *ws) { | 570 | /* Live sync reads whole packets rather than requests and wants a larger |
| 571 | inbuf than the request path does, hence the size. */ | ||
| 572 | int workstruct_init(struct ot_workstruct *ws, size_t inbuf_size) { | ||
| 571 | memset(ws, 0, sizeof(*ws)); | 573 | memset(ws, 0, sizeof(*ws)); |
| 572 | ws->inbuf = malloc(G_INBUF_SIZE); | 574 | ws->inbuf = malloc(inbuf_size); |
| 573 | ws->outbuf = malloc(G_OUTBUF_SIZE); | 575 | ws->outbuf = malloc(G_OUTBUF_SIZE); |
| 574 | #ifdef _DEBUG_HTTPERROR | 576 | #ifdef _DEBUG_HTTPERROR |
| 575 | ws->debugbuf = malloc(G_DEBUGBUF_SIZE); | 577 | ws->debugbuf = malloc(G_DEBUGBUF_SIZE); |
diff --git a/src/trackerlogic.h b/src/trackerlogic.h index 7292d36..692b6af 100644 --- a/src/trackerlogic.h +++ b/src/trackerlogic.h | |||
| @@ -242,7 +242,7 @@ struct ot_workstruct { | |||
| 242 | /* Everything a thread that serves requests needs for itself. Set up and torn | 242 | /* Everything a thread that serves requests needs for itself. Set up and torn |
| 243 | down in one place so that a new kind of worker cannot miss a piece. | 243 | down in one place so that a new kind of worker cannot miss a piece. |
| 244 | Returns -1 if the buffers could not be had. */ | 244 | Returns -1 if the buffers could not be had. */ |
| 245 | int workstruct_init(struct ot_workstruct *ws); | 245 | int workstruct_init(struct ot_workstruct *ws, size_t inbuf_size); |
| 246 | void workstruct_deinit(struct ot_workstruct *ws); | 246 | void workstruct_deinit(struct ot_workstruct *ws); |
| 247 | 247 | ||
| 248 | /* | 248 | /* |
