1 // L2TPNS Clustering Stuff
10 #include <sys/socket.h>
11 #include <netinet/in.h>
12 #include <arpa/inet.h>
13 #include <sys/ioctl.h>
19 #include <linux/rtnetlink.h>
32 * All cluster packets have the same format.
34 * One or more instances of
36 * a 32 bit 'extra' data dependant on the 'type'.
37 * zero or more bytes of structure data, dependant on the type.
42 extern int cluster_sockfd
; // The filedescriptor for the cluster communications port.
44 in_addr_t my_address
= 0; // The network address of my ethernet port.
45 static int walk_session_number
= 0; // The next session to send when doing the slow table walk.
46 static int walk_bundle_number
= 0; // The next bundle to send when doing the slow table walk.
47 static int walk_tunnel_number
= 0; // The next tunnel to send when doing the slow table walk.
48 static int walk_groupe_number
= 0; // The next groupe to send when doing the slow table walk.
49 int forked
= 0; // Sanity check: CLI must not diddle with heartbeat table
51 #define MAX_HEART_SIZE (8192) // Maximum size of heartbeat packet. Must be less than max IP packet size :)
52 #define MAX_CHANGES (MAX_HEART_SIZE/(sizeof(sessiont) + sizeof(int) ) - 2) // Assumes a session is the biggest type!
57 } cluster_changes
[MAX_CHANGES
]; // Queue of changed structures that need to go out when next heartbeat.
62 uint8_t data
[MAX_HEART_SIZE
];
63 } past_hearts
[HB_HISTORY_SIZE
]; // Ring buffer of heartbeats that we've recently sent out. Needed so
64 // we can re-transmit if needed.
71 } peers
[CLUSTER_MAX_SIZE
]; // List of all the peers we've heard from.
72 static int num_peers
; // Number of peers in list.
74 static int rle_decompress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
);
75 static int rle_compress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
);
78 // Create a listening socket
80 // This joins the cluster multi-cast group.
84 struct sockaddr_in addr
;
85 struct sockaddr_in interface_addr
;
90 config
->cluster_undefined_sessions
= MAXSESSION
-1;
91 config
->cluster_undefined_bundles
= MAXBUNDLE
-1;
92 config
->cluster_undefined_tunnels
= MAXTUNNEL
-1;
93 config
->cluster_undefined_groupes
= MAXGROUPE
-1;
95 if (!config
->cluster_address
)
97 if (!*config
->cluster_interface
)
100 cluster_sockfd
= socket(AF_INET
, SOCK_DGRAM
, IPPROTO_UDP
);
102 memset(&addr
, 0, sizeof(addr
));
103 addr
.sin_family
= AF_INET
;
104 addr
.sin_port
= htons(CLUSTERPORT
);
105 addr
.sin_addr
.s_addr
= INADDR_ANY
;
106 setsockopt(cluster_sockfd
, SOL_SOCKET
, SO_REUSEADDR
, &addr
, sizeof(addr
));
108 opt
= fcntl(cluster_sockfd
, F_GETFL
, 0);
109 fcntl(cluster_sockfd
, F_SETFL
, opt
| O_NONBLOCK
);
111 if (bind(cluster_sockfd
, (void *) &addr
, sizeof(addr
)) < 0)
113 LOG(0, 0, 0, "Failed to bind cluster socket: %s\n", strerror(errno
));
117 strcpy(ifr
.ifr_name
, config
->cluster_interface
);
118 if (ioctl(cluster_sockfd
, SIOCGIFADDR
, &ifr
) < 0)
120 LOG(0, 0, 0, "Failed to get interface address for (%s): %s\n", config
->cluster_interface
, strerror(errno
));
124 memcpy(&interface_addr
, &ifr
.ifr_addr
, sizeof(interface_addr
));
125 my_address
= interface_addr
.sin_addr
.s_addr
;
127 // Join multicast group.
128 mreq
.imr_multiaddr
.s_addr
= config
->cluster_address
;
129 mreq
.imr_interface
= interface_addr
.sin_addr
;
132 opt
= 0; // Turn off multicast loopback.
133 setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_LOOP
, &opt
, sizeof(opt
));
135 if (config
->cluster_mcast_ttl
!= 1)
138 if (config
->cluster_mcast_ttl
> 0)
139 ttl
= config
->cluster_mcast_ttl
< 256 ? config
->cluster_mcast_ttl
: 255;
141 setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_TTL
, &ttl
, sizeof(ttl
));
144 if (setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_ADD_MEMBERSHIP
, &mreq
, sizeof(mreq
)) < 0)
146 LOG(0, 0, 0, "Failed to setsockopt (join mcast group): %s\n", strerror(errno
));
150 if (setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_IF
, &interface_addr
, sizeof(interface_addr
)) < 0)
152 LOG(0, 0, 0, "Failed to setsockopt (set mcast interface): %s\n", strerror(errno
));
156 config
->cluster_last_hb
= TIME
;
157 config
->cluster_seq_number
= -1;
159 return cluster_sockfd
;
164 // Send a chunk of data to the entire cluster (usually via the multicast
168 static int cluster_send_data(void *data
, int datalen
)
170 struct sockaddr_in addr
= {0};
172 if (!cluster_sockfd
) return -1;
173 if (!config
->cluster_address
) return 0;
175 addr
.sin_addr
.s_addr
= config
->cluster_address
;
176 addr
.sin_port
= htons(CLUSTERPORT
);
177 addr
.sin_family
= AF_INET
;
179 LOG(5, 0, 0, "Cluster send data: %d bytes\n", datalen
);
181 if (sendto(cluster_sockfd
, data
, datalen
, MSG_NOSIGNAL
, (void *) &addr
, sizeof(addr
)) < 0)
183 LOG(0, 0, 0, "sendto: %s\n", strerror(errno
));
191 // Add a chunk of data to a heartbeat packet.
192 // Maintains the format. Assumes that the caller
193 // has passed in a big enough buffer!
195 static void add_type(uint8_t **p
, int type
, int more
, uint8_t *data
, int size
)
197 *((uint32_t *) (*p
)) = type
;
198 *p
+= sizeof(uint32_t);
200 *((uint32_t *)(*p
)) = more
;
201 *p
+= sizeof(uint32_t);
203 if (data
&& size
> 0) {
204 memcpy(*p
, data
, size
);
209 // advertise our presence via BGP or gratuitous ARP
210 static void advertise_routes(void)
214 bgp_enable_routing(1);
217 if (config
->send_garp
)
218 send_garp(config
->bind_address
); // Start taking traffic.
221 // withdraw our routes (BGP only)
222 static void withdraw_routes(void)
226 bgp_enable_routing(0);
230 static void cluster_uptodate(void)
232 if (config
->cluster_iam_uptodate
)
235 if (config
->cluster_undefined_sessions
|| config
->cluster_undefined_tunnels
||
236 config
->cluster_undefined_bundles
|| config
->cluster_undefined_groupes
)
239 config
->cluster_iam_uptodate
= 1;
241 LOG(0, 0, 0, "Now uptodate with master.\n");
246 // Send a unicast UDP packet to a peer with 'data' as the
249 static int peer_send_data(in_addr_t peer
, uint8_t *data
, int size
)
251 struct sockaddr_in addr
= {0};
253 if (!cluster_sockfd
) return -1;
254 if (!config
->cluster_address
) return 0;
259 addr
.sin_addr
.s_addr
= peer
;
260 addr
.sin_port
= htons(CLUSTERPORT
);
261 addr
.sin_family
= AF_INET
;
263 LOG_HEX(5, "Peer send", data
, size
);
265 if (sendto(cluster_sockfd
, data
, size
, MSG_NOSIGNAL
, (void *) &addr
, sizeof(addr
)) < 0)
267 LOG(0, 0, 0, "sendto: %s\n", strerror(errno
));
275 // Send a structured message to a peer with a single element of type 'type'.
277 static int peer_send_message(in_addr_t peer
, int type
, int more
, uint8_t *data
, int size
)
279 uint8_t buf
[65536]; // Vast overkill.
282 LOG(4, 0, 0, "Sending message to peer (type %d, more %d, size %d)\n", type
, more
, size
);
283 add_type(&p
, type
, more
, data
, size
);
285 return peer_send_data(peer
, buf
, (p
-buf
) );
288 // send a packet to the master
289 static int _forward_packet(uint8_t *data
, int size
, in_addr_t addr
, int port
, int type
)
291 uint8_t buf
[65536]; // Vast overkill.
294 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
297 LOG(4, 0, 0, "Forwarding packet from %s to master (size %d)\n", fmtaddr(addr
, 0), size
);
300 add_type(&p
, type
, addr
, (uint8_t *) &port
, sizeof(port
)); // ick. should be uint16_t
301 memcpy(p
, data
, size
);
304 return peer_send_data(config
->cluster_master_address
, buf
, (p
- buf
));
308 // Forward a state changing packet to the master.
310 // The master just processes the payload as if it had
311 // received it off the tun device.
312 //(note: THIS ROUTINE WRITES TO pack[-6]).
313 int master_forward_packet(uint8_t *data
, int size
, in_addr_t addr
, uint16_t port
, uint16_t indexudp
)
315 uint8_t *p
= data
- (3 * sizeof(uint32_t));
317 uint32_t indexandport
= port
| ((indexudp
<< 16) & 0xFFFF0000);
319 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
322 LOG(4, 0, 0, "Forwarding packet from %s to master (size %d)\n", fmtaddr(addr
, 0), size
);
325 add_type(&p
, C_FORWARD
, addr
, (uint8_t *) &indexandport
, sizeof(indexandport
));
327 return peer_send_data(config
->cluster_master_address
, psave
, size
+ (3 * sizeof(uint32_t)));
330 // Forward PPPOE packet to the master.
331 //(note: THIS ROUTINE WRITES TO pack[-4]).
332 int master_forward_pppoe_packet(uint8_t *data
, int size
, uint8_t codepad
)
334 uint8_t *p
= data
- (2 * sizeof(uint32_t));
337 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
340 LOG(4, 0, 0, "Forward PPPOE packet to master, code %s (size %d)\n", get_string_codepad(codepad
), size
);
343 add_type(&p
, C_PPPOE_FORWARD
, codepad
, NULL
, 0);
345 return peer_send_data(config
->cluster_master_address
, psave
, size
+ (2 * sizeof(uint32_t)));
348 // Forward a DAE RADIUS packet to the master.
349 int master_forward_dae_packet(uint8_t *data
, int size
, in_addr_t addr
, int port
)
351 return _forward_packet(data
, size
, addr
, port
, C_FORWARD_DAE
);
355 // Forward a throttled packet to the master for handling.
357 // The master just drops the packet into the appropriate
358 // token bucket queue, and lets normal processing take care
361 int master_throttle_packet(int tbfid
, uint8_t *data
, int size
)
363 uint8_t buf
[65536]; // Vast overkill.
366 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
369 LOG(4, 0, 0, "Throttling packet master (size %d, tbfid %d)\n", size
, tbfid
);
371 add_type(&p
, C_THROTTLE
, tbfid
, data
, size
);
373 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
) );
378 // Forward a walled garden packet to the master for handling.
380 // The master just writes the packet straight to the tun
381 // device (where is will normally loop through the
382 // firewall rules, and come back in on the tun device)
384 // (Note that this must be called with the tun header
385 // as the start of the data).
386 int master_garden_packet(sessionidt s
, uint8_t *data
, int size
)
388 uint8_t buf
[65536]; // Vast overkill.
391 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
394 LOG(4, 0, 0, "Walled garden packet to master (size %d)\n", size
);
396 add_type(&p
, C_GARDEN
, s
, data
, size
);
398 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
));
403 // Forward a MPPP packet to the master for handling.
405 // (Note that this must be called with the tun header
406 // as the start of the data).
407 // (i.e. this routine writes to data[-8]).
408 int master_forward_mppp_packet(sessionidt s
, uint8_t *data
, int size
)
410 uint8_t *p
= data
- (2 * sizeof(uint32_t));
413 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
416 LOG(4, 0, 0, "Forward MPPP packet to master (size %d)\n", size
);
418 add_type(&p
, C_MPPP_FORWARD
, s
, NULL
, 0);
420 return peer_send_data(config
->cluster_master_address
, psave
, size
+ (2 * sizeof(uint32_t)));
425 // Send a chunk of data as a heartbeat..
426 // We save it in the history buffer as we do so.
428 static void send_heartbeat(int seq
, uint8_t *data
, int size
)
432 if (size
> sizeof(past_hearts
[0].data
))
434 LOG(0, 0, 0, "Tried to heartbeat something larger than the maximum packet!\n");
438 i
= seq
% HB_HISTORY_SIZE
;
439 past_hearts
[i
].seq
= seq
;
440 past_hearts
[i
].size
= size
;
441 memcpy(&past_hearts
[i
].data
, data
, size
); // Save it.
442 cluster_send_data(data
, size
);
446 // Send an 'i am alive' message to every machine in the cluster.
448 void cluster_send_ping(time_t basetime
)
450 uint8_t buff
[100 + sizeof(pingt
)];
454 if (config
->cluster_iam_master
&& basetime
) // We're heartbeating so no need to ping.
457 LOG(5, 0, 0, "Sending cluster ping...\n");
460 x
.addr
= config
->bind_address
;
461 x
.undef
= config
->cluster_undefined_sessions
+ config
->cluster_undefined_tunnels
+
462 config
->cluster_undefined_groupes
+ config
->cluster_undefined_bundles
;
463 x
.basetime
= basetime
;
465 add_type(&p
, C_PING
, basetime
, (uint8_t *) &x
, sizeof(x
));
466 cluster_send_data(buff
, (p
-buff
) );
470 // Walk the session counters looking for non-zero ones to send
471 // to the master. We send up to 600 of them at one time.
472 // We examine a maximum of 3000 sessions.
473 // (50k max session should mean that we normally
474 // examine the entire session table every 25 seconds).
476 #define MAX_B_RECS (600)
477 void master_update_counts(void)
480 bytest b
[MAX_B_RECS
+1];
482 if (config
->cluster_iam_master
) // Only happens on the slaves.
485 if (!config
->cluster_master_address
) // If we don't have a master, skip it for a while.
488 i
= MAX_B_RECS
* 5; // Examine max 3000 sessions;
489 if (config
->cluster_highest_sessionid
> i
)
490 i
= config
->cluster_highest_sessionid
;
492 for ( c
= 0; i
> 0 ; --i
) {
493 // Next session to look at.
494 walk_session_number
++;
495 if ( walk_session_number
> config
->cluster_highest_sessionid
)
496 walk_session_number
= 1;
498 if (!sess_local
[walk_session_number
].cin
&& !sess_local
[walk_session_number
].cout
)
499 continue; // Unchanged. Skip it.
501 b
[c
].sid
= walk_session_number
;
502 b
[c
].pin
= sess_local
[walk_session_number
].pin
;
503 b
[c
].pout
= sess_local
[walk_session_number
].pout
;
504 b
[c
].cin
= sess_local
[walk_session_number
].cin
;
505 b
[c
].cout
= sess_local
[walk_session_number
].cout
;
508 sess_local
[walk_session_number
].pin
= sess_local
[walk_session_number
].pout
= 0;
509 sess_local
[walk_session_number
].cin
= sess_local
[walk_session_number
].cout
= 0;
511 if (++c
> MAX_B_RECS
) // Send a max of 600 elements in a packet.
515 if (!c
) // Didn't find any that changes. Get out of here!
519 // Forward the data to the master.
520 LOG(4, 0, 0, "Sending byte counters to master (%d elements)\n", c
);
521 peer_send_message(config
->cluster_master_address
, C_BYTES
, c
, (uint8_t *) &b
, sizeof(b
[0]) * c
);
526 // On the master, check how our slaves are going. If
527 // one of them's not up-to-date we'll heartbeat faster.
528 // If we don't have any of them, then we need to turn
529 // on our own packet handling!
531 void cluster_check_slaves(void)
534 static int have_peers
= 0;
535 int had_peers
= have_peers
;
538 if (!config
->cluster_iam_master
)
539 return; // Only runs on the master...
541 config
->cluster_iam_uptodate
= 1; // cleared in loop below
543 for (i
= have_peers
= 0; i
< num_peers
; i
++)
545 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
) < t
)
546 continue; // Stale peer! Skip them.
548 if (!peers
[i
].basetime
)
549 continue; // Shutdown peer! Skip them.
551 if (peers
[i
].uptodate
)
554 config
->cluster_iam_uptodate
= 0; // Start fast heartbeats
557 // in a cluster, withdraw/add routes when we get a peer/lose peers
558 if (have_peers
!= had_peers
)
560 if (had_peers
< config
->cluster_master_min_adv
&&
561 have_peers
>= config
->cluster_master_min_adv
)
564 else if (had_peers
>= config
->cluster_master_min_adv
&&
565 have_peers
< config
->cluster_master_min_adv
)
571 // Check that we have a master. If it's been too
572 // long since we heard from a master then hold an election.
574 void cluster_check_master(void)
576 int i
, count
, high_unique_id
= 0;
579 static int probed
= 0;
582 if (config
->cluster_iam_master
)
583 return; // Only runs on the slaves...
585 // If the master is late (missed 2 hearbeats by a second and a
586 // hair) it may be that the switch has dropped us from the
587 // multicast group, try unicasting probes to the master
588 // which will hopefully respond with a unicast heartbeat that
589 // will allow us to limp along until the querier next runs.
590 if (config
->cluster_master_address
591 && TIME
> (config
->cluster_last_hb
+ 2 * config
->cluster_hb_interval
+ 11))
593 if (!probed
|| (TIME
> (probed
+ 2 * config
->cluster_hb_interval
)))
596 LOG(1, 0, 0, "Heartbeat from master %.1fs late, probing...\n",
597 0.1 * (TIME
- (config
->cluster_last_hb
+ config
->cluster_hb_interval
)));
599 peer_send_message(config
->cluster_master_address
,
600 C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
602 } else { // We got a recent heartbeat; reset the probe flag.
606 if (TIME
< (config
->cluster_last_hb
+ config
->cluster_hb_timeout
))
607 return; // Everything's ok!
609 config
->cluster_last_hb
= TIME
+ 1; // Just the one election thanks.
610 config
->cluster_master_address
= 0;
612 LOG(0, 0, 0, "Master timed out! Holding election...\n");
614 // In the process of shutting down, can't be master
618 for (i
= have_peers
= 0; i
< num_peers
; i
++)
620 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
) < t
)
621 continue; // Stale peer! Skip them.
623 if (!peers
[i
].basetime
)
624 continue; // Shutdown peer! Skip them.
626 if (peers
[i
].basetime
< basetime
) {
627 LOG(1, 0, 0, "Expecting %s to become master\n", fmtaddr(peers
[i
].peer
, 0));
628 return; // They'll win the election. Get out of here.
631 if (peers
[i
].basetime
== basetime
&&
632 peers
[i
].peer
> my_address
) {
633 LOG(1, 0, 0, "Expecting %s to become master\n", fmtaddr(peers
[i
].peer
, 0));
634 return; // They'll win the election. Wait for them to come up.
637 if (peers
[i
].uptodate
)
641 // Wow. it's been ages since I last heard a heartbeat
642 // and I'm better than an of my peers so it's time
643 // to become a master!!!
645 config
->cluster_iam_master
= 1;
646 pppoe_send_garp(); // gratuitous arp of the pppoe interface
648 LOG(0, 0, 0, "I am declaring myself the master!\n");
650 if (have_peers
< config
->cluster_master_min_adv
)
655 if (config
->cluster_seq_number
== -1)
656 config
->cluster_seq_number
= 0;
659 // Go through and mark all the tunnels as defined.
660 // Count the highest used tunnel number as well.
662 config
->cluster_highest_tunnelid
= 0;
663 for (i
= 0; i
< MAXTUNNEL
; ++i
) {
664 if (tunnel
[i
].state
== TUNNELUNDEF
)
665 tunnel
[i
].state
= TUNNELFREE
;
667 if (tunnel
[i
].state
!= TUNNELFREE
&& i
> config
->cluster_highest_tunnelid
)
668 config
->cluster_highest_tunnelid
= i
;
672 // Go through and mark all the bundles as defined.
673 // Count the highest used bundle number as well.
675 config
->cluster_highest_bundleid
= 0;
676 for (i
= 0; i
< MAXBUNDLE
; ++i
) {
677 if (bundle
[i
].state
== BUNDLEUNDEF
)
678 bundle
[i
].state
= BUNDLEFREE
;
680 if (bundle
[i
].state
!= BUNDLEFREE
&& i
> config
->cluster_highest_bundleid
)
681 config
->cluster_highest_bundleid
= i
;
685 // Go through and mark all the groupes as defined.
686 // Count the highest used groupe number as well.
688 config
->cluster_highest_groupeid
= 0;
689 for (i
= 0; i
< MAXGROUPE
; ++i
)
691 if (grpsession
[i
].state
== GROUPEUNDEF
)
692 grpsession
[i
].state
= GROUPEFREE
;
694 if (grpsession
[i
].state
!= GROUPEFREE
&& i
> config
->cluster_highest_groupeid
)
695 config
->cluster_highest_groupeid
= i
;
699 // Go through and mark all the sessions as being defined.
700 // reset the idle timeouts.
701 // add temporary byte counters to permanent ones.
702 // Re-string the free list.
703 // Find the ID of the highest session.
706 config
->cluster_highest_sessionid
= 0;
707 for (i
= 0, count
= 0; i
< MAXSESSION
; ++i
) {
708 if (session
[i
].tunnel
== T_UNDEF
) {
709 session
[i
].tunnel
= T_FREE
;
713 if (!session
[i
].opened
) { // Unused session. Add to free list.
714 memset(&session
[i
], 0, sizeof(session
[i
]));
715 session
[i
].tunnel
= T_FREE
;
716 session
[last_free
].next
= i
;
722 // Reset idle timeouts..
723 session
[i
].last_packet
= session
[i
].last_data
= time_now
;
725 // Reset die relative to our uptime rather than the old master's
726 if (session
[i
].die
) session
[i
].die
= TIME
;
728 // Accumulate un-sent byte/packet counters.
729 increment_counter(&session
[i
].cin
, &session
[i
].cin_wrap
, sess_local
[i
].cin
);
730 increment_counter(&session
[i
].cout
, &session
[i
].cout_wrap
, sess_local
[i
].cout
);
731 session
[i
].cin_delta
+= sess_local
[i
].cin
;
732 session
[i
].cout_delta
+= sess_local
[i
].cout
;
734 session
[i
].pin
+= sess_local
[i
].pin
;
735 session
[i
].pout
+= sess_local
[i
].pout
;
737 sess_local
[i
].cin
= sess_local
[i
].cout
= 0;
738 sess_local
[i
].pin
= sess_local
[i
].pout
= 0;
740 sess_local
[i
].radius
= 0; // Reset authentication as the radius blocks aren't up to date.
742 if (session
[i
].unique_id
>= high_unique_id
) // This is different to the index into the session table!!!
743 high_unique_id
= session
[i
].unique_id
+1;
745 session
[i
].tbf_in
= session
[i
].tbf_out
= 0; // Remove stale pointers from old master.
746 throttle_session(i
, session
[i
].throttle_in
, session
[i
].throttle_out
);
748 config
->cluster_highest_sessionid
= i
;
751 session
[last_free
].next
= 0; // End of chain.
752 last_id
= high_unique_id
; // Keep track of the highest used session ID.
756 rebuild_address_pool();
758 // If we're not the very first master, this is a big issue!
760 LOG(0, 0, 0, "Warning: Fixed %d uninitialized sessions in becoming master!\n", count
);
762 config
->cluster_undefined_sessions
= 0;
763 config
->cluster_undefined_bundles
= 0;
764 config
->cluster_undefined_tunnels
= 0;
765 config
->cluster_undefined_groupes
= 0;
766 config
->cluster_iam_uptodate
= 1; // assume all peers are up-to-date
768 // FIXME. We need to fix up the tunnel control message
769 // queue here! There's a number of other variables we
770 // should also update.
775 // Check that our session table is validly matching what the
776 // master has in mind.
778 // In particular, if we have too many sessions marked 'undefined'
779 // we fix it up here, and we ensure that the 'first free session'
782 static void cluster_check_sessions(int highsession
, int freesession_ptr
, int highbundle
, int hightunnel
, int highgroupe
)
786 sessionfree
= freesession_ptr
; // Keep the freesession ptr valid.
788 if (config
->cluster_iam_uptodate
)
791 if (highsession
> config
->cluster_undefined_sessions
&& highbundle
> config
->cluster_undefined_bundles
&&
792 highgroupe
> config
->cluster_undefined_groupes
&& hightunnel
> config
->cluster_undefined_tunnels
)
795 // Clear out defined sessions, counting the number of
797 config
->cluster_undefined_sessions
= 0;
798 for (i
= 1 ; i
< MAXSESSION
; ++i
) {
799 if (i
> highsession
) {
800 if (session
[i
].tunnel
== T_UNDEF
) session
[i
].tunnel
= T_FREE
; // Defined.
804 if (session
[i
].tunnel
== T_UNDEF
)
805 ++config
->cluster_undefined_sessions
;
808 // Clear out defined bundles, counting the number of
810 config
->cluster_undefined_bundles
= 0;
811 for (i
= 1 ; i
< MAXBUNDLE
; ++i
) {
812 if (i
> highbundle
) {
813 if (bundle
[i
].state
== BUNDLEUNDEF
) bundle
[i
].state
= BUNDLEFREE
; // Defined.
817 if (bundle
[i
].state
== BUNDLEUNDEF
)
818 ++config
->cluster_undefined_bundles
;
821 // Clear out defined tunnels, counting the number of
823 config
->cluster_undefined_tunnels
= 0;
824 for (i
= 1 ; i
< MAXTUNNEL
; ++i
) {
825 if (i
> hightunnel
) {
826 if (tunnel
[i
].state
== TUNNELUNDEF
) tunnel
[i
].state
= TUNNELFREE
; // Defined.
830 if (tunnel
[i
].state
== TUNNELUNDEF
)
831 ++config
->cluster_undefined_tunnels
;
834 // Clear out defined groupe, counting the number of
836 config
->cluster_undefined_groupes
= 0;
837 for (i
= 1 ; i
< MAXGROUPE
; ++i
) {
838 if (i
> highgroupe
) {
839 if (grpsession
[i
].state
== GROUPEUNDEF
) grpsession
[i
].state
= GROUPEFREE
; // Defined.
843 if (grpsession
[i
].state
== GROUPEUNDEF
)
844 ++config
->cluster_undefined_groupes
;
847 if (config
->cluster_undefined_sessions
|| config
->cluster_undefined_tunnels
|| config
->cluster_undefined_bundles
|| config
->cluster_undefined_groupes
) {
848 LOG(2, 0, 0, "Cleared undefined sessions/bundles/tunnels. %d sess (high %d), %d bund (high %d), %d grp (high %d), %d tunn (high %d)\n",
849 config
->cluster_undefined_sessions
, highsession
, config
->cluster_undefined_bundles
, highbundle
,
850 config
->cluster_undefined_groupes
, highgroupe
, config
->cluster_undefined_tunnels
, hightunnel
);
854 // Are we up to date?
856 if (!config
->cluster_iam_uptodate
)
860 static int hb_add_type(uint8_t **p
, int type
, int id
)
863 case C_CSESSION
: { // Compressed C_SESSION.
864 uint8_t c
[sizeof(sessiont
) * 2]; // Bigger than worst case.
865 uint8_t *d
= (uint8_t *) &session
[id
];
869 size
= rle_compress( &d
, sizeof(sessiont
), c
, sizeof(c
) );
871 // Did we compress the full structure, and is the size actually
873 if ( (d
- orig
) == sizeof(sessiont
) && size
< sizeof(sessiont
) ) {
874 add_type(p
, C_CSESSION
, id
, c
, size
);
877 // Failed to compress : Fall through.
880 add_type(p
, C_SESSION
, id
, (uint8_t *) &session
[id
], sizeof(sessiont
));
883 case C_CBUNDLE
: { // Compressed C_BUNDLE
884 uint8_t c
[sizeof(bundlet
) * 2]; // Bigger than worst case.
885 uint8_t *d
= (uint8_t *) &bundle
[id
];
889 size
= rle_compress( &d
, sizeof(bundlet
), c
, sizeof(c
) );
891 // Did we compress the full structure, and is the size actually
893 if ( (d
- orig
) == sizeof(bundlet
) && size
< sizeof(bundlet
) ) {
894 add_type(p
, C_CBUNDLE
, id
, c
, size
);
897 // Failed to compress : Fall through.
901 add_type(p
, C_BUNDLE
, id
, (uint8_t *) &bundle
[id
], sizeof(bundlet
));
904 case C_CGROUPE
: { // Compressed C_GROUPE
905 uint8_t c
[sizeof(groupsesst
) * 2]; // Bigger than worst case.
906 uint8_t *d
= (uint8_t *) &grpsession
[id
];
910 size
= rle_compress( &d
, sizeof(groupsesst
), c
, sizeof(c
) );
912 // Did we compress the full structure, and is the size actually
914 if ( (d
- orig
) == sizeof(groupsesst
) && size
< sizeof(groupsesst
) )
916 add_type(p
, C_CGROUPE
, id
, c
, size
);
919 // Failed to compress : Fall through.
922 add_type(p
, C_GROUPE
, id
, (uint8_t *) &grpsession
[id
], sizeof(groupsesst
));
925 case C_CTUNNEL
: { // Compressed C_TUNNEL
926 uint8_t c
[sizeof(tunnelt
) * 2]; // Bigger than worst case.
927 uint8_t *d
= (uint8_t *) &tunnel
[id
];
931 size
= rle_compress( &d
, sizeof(tunnelt
), c
, sizeof(c
) );
933 // Did we compress the full structure, and is the size actually
935 if ( (d
- orig
) == sizeof(tunnelt
) && size
< sizeof(tunnelt
) ) {
936 add_type(p
, C_CTUNNEL
, id
, c
, size
);
939 // Failed to compress : Fall through.
942 add_type(p
, C_TUNNEL
, id
, (uint8_t *) &tunnel
[id
], sizeof(tunnelt
));
945 LOG(0, 0, 0, "Found an invalid type in heart queue! (%d)\n", type
);
953 // Send a heartbeat, incidently sending out any queued changes..
955 void cluster_heartbeat()
957 int i
, count
= 0, tcount
= 0, bcount
= 0, gcount
= 0;
958 uint8_t buff
[MAX_HEART_SIZE
+ sizeof(heartt
) + sizeof(int) ];
962 if (!config
->cluster_iam_master
) // Only the master does this.
965 config
->cluster_table_version
+= config
->cluster_num_changes
;
967 // Fill out the heartbeat header.
968 memset(&h
, 0, sizeof(h
));
970 h
.version
= HB_VERSION
;
971 h
.seq
= config
->cluster_seq_number
;
972 h
.basetime
= basetime
;
973 h
.clusterid
= config
->bind_address
; // Will this do??
974 h
.basetime
= basetime
;
975 h
.highsession
= config
->cluster_highest_sessionid
;
976 h
.freesession
= sessionfree
;
977 h
.hightunnel
= config
->cluster_highest_tunnelid
;
978 h
.highbundle
= config
->cluster_highest_bundleid
;
979 h
.highgroupe
= config
->cluster_highest_groupeid
;
980 h
.size_sess
= sizeof(sessiont
); // Just in case.
981 h
.size_bund
= sizeof(bundlet
);
982 h
.size_tunn
= sizeof(tunnelt
);
983 h
.nextgrpid
= gnextgrpid
;
984 h
.interval
= config
->cluster_hb_interval
;
985 h
.timeout
= config
->cluster_hb_timeout
;
986 h
.table_version
= config
->cluster_table_version
;
988 add_type(&p
, C_HEARTBEAT
, HB_VERSION
, (uint8_t *) &h
, sizeof(h
));
990 for (i
= 0; i
< config
->cluster_num_changes
; ++i
) {
991 hb_add_type(&p
, cluster_changes
[i
].type
, cluster_changes
[i
].id
);
994 if (p
> (buff
+ sizeof(buff
))) { // Did we somehow manage to overun the buffer?
995 LOG(0, 0, 0, "FATAL: Overran the heartbeat buffer! This is fatal. Exiting. (size %d)\n", (int) (p
- buff
));
1001 // Fill out the packet with sessions from the session table...
1002 // (not forgetting to leave space so we can get some tunnels,bundle,groupe in too )
1003 while ( (p
+ sizeof(uint32_t) * 2 + sizeof(sessiont
) * 2 ) < (buff
+ MAX_HEART_SIZE
) ) {
1005 if (!walk_session_number
) // session #0 isn't valid.
1006 ++walk_session_number
;
1008 if (count
>= config
->cluster_highest_sessionid
) // If we're a small cluster, don't go wild.
1011 hb_add_type(&p
, C_CSESSION
, walk_session_number
);
1012 walk_session_number
= (1+walk_session_number
)%(config
->cluster_highest_sessionid
+1); // +1 avoids divide by zero.
1014 ++count
; // Count the number of extra sessions we're sending.
1018 // Fill out the packet with tunnels from the tunnel table...
1019 // This effectively means we walk the tunnel table more quickly
1020 // than the session table. This is good because stuffing up a
1021 // tunnel is a much bigger deal than stuffing up a session.
1023 int maxsize
= (sizeof(tunnelt
) < sizeof(bundlet
)) ? sizeof(bundlet
):sizeof(tunnelt
);
1024 maxsize
= (sizeof(groupsesst
) < maxsize
) ? maxsize
:sizeof(groupsesst
);
1025 maxsize
+= (sizeof(uint32_t) * 2);
1027 // Fill out the packet with tunnels,bundlets, groupes from the tables...
1028 while ( (p
+ maxsize
) < (buff
+ MAX_HEART_SIZE
) )
1030 if ((tcount
>= config
->cluster_highest_tunnelid
) &&
1031 (bcount
>= config
->cluster_highest_bundleid
) &&
1032 (gcount
>= config
->cluster_highest_groupeid
))
1035 if ( ((p
+ sizeof(uint32_t) * 2 + sizeof(tunnelt
) ) < (buff
+ MAX_HEART_SIZE
)) &&
1036 (tcount
< config
->cluster_highest_tunnelid
))
1038 if (!walk_tunnel_number
) // tunnel #0 isn't valid.
1039 ++walk_tunnel_number
;
1041 hb_add_type(&p
, C_CTUNNEL
, walk_tunnel_number
);
1042 walk_tunnel_number
= (1+walk_tunnel_number
)%(config
->cluster_highest_tunnelid
+1); // +1 avoids divide by zero.
1047 if ( ((p
+ sizeof(uint32_t) * 2 + sizeof(bundlet
) ) < (buff
+ MAX_HEART_SIZE
)) &&
1048 (bcount
< config
->cluster_highest_bundleid
))
1050 if (!walk_bundle_number
) // bundle #0 isn't valid.
1051 ++walk_bundle_number
;
1053 hb_add_type(&p
, C_CBUNDLE
, walk_bundle_number
);
1054 walk_bundle_number
= (1+walk_bundle_number
)%(config
->cluster_highest_bundleid
+1); // +1 avoids divide by zero.
1059 if ( ((p
+ sizeof(uint32_t) * 2 + sizeof(groupsesst
) ) < (buff
+ MAX_HEART_SIZE
)) &&
1060 (gcount
< config
->cluster_highest_groupeid
))
1062 if (!walk_groupe_number
) // groupe #0 isn't valid.
1063 ++walk_groupe_number
;
1065 hb_add_type(&p
, C_CGROUPE
, walk_groupe_number
);
1066 walk_groupe_number
= (1+walk_groupe_number
)%(config
->cluster_highest_groupeid
+1); // +1 avoids divide by zero.
1072 // Did we do something wrong?
1073 if (p
> (buff
+ sizeof(buff
))) { // Did we somehow manage to overun the buffer?
1074 LOG(0, 0, 0, "Overran the heartbeat buffer now! This is fatal. Exiting. (size %d)\n", (int) (p
- buff
));
1079 LOG(4, 0, 0, "Sending v%d heartbeat #%d, change #%" PRIu64
" with %d changes "
1080 "(%d x-sess, %d x-bundles, %d x-tunnels, %d x-groupes, %d highsess, %d highbund, %d hightun, %d highgrp, size %d)\n",
1081 HB_VERSION
, h
.seq
, h
.table_version
, config
->cluster_num_changes
,
1082 count
, bcount
, tcount
, gcount
, config
->cluster_highest_sessionid
, config
->cluster_highest_bundleid
,
1083 config
->cluster_highest_tunnelid
, config
->cluster_highest_groupeid
, (int) (p
- buff
));
1085 config
->cluster_num_changes
= 0;
1087 send_heartbeat(h
.seq
, buff
, (p
-buff
) ); // Send out the heartbeat to the cluster, keeping a copy of it.
1089 config
->cluster_seq_number
= (config
->cluster_seq_number
+1)%HB_MAX_SEQ
; // Next seq number to use.
1093 // A structure of type 'type' has changed; Add it to the queue to send.
1095 static int type_changed(int type
, int id
)
1099 for (i
= 0 ; i
< config
->cluster_num_changes
; ++i
)
1101 if ( cluster_changes
[i
].id
== id
&& cluster_changes
[i
].type
== type
)
1103 // Already marked for change, remove it
1104 --config
->cluster_num_changes
;
1105 memmove(&cluster_changes
[i
],
1106 &cluster_changes
[i
+1],
1107 (config
->cluster_num_changes
- i
) * sizeof(cluster_changes
[i
]));
1112 cluster_changes
[config
->cluster_num_changes
].type
= type
;
1113 cluster_changes
[config
->cluster_num_changes
].id
= id
;
1114 ++config
->cluster_num_changes
;
1116 if (config
->cluster_num_changes
> MAX_CHANGES
)
1117 cluster_heartbeat(); // flush now
1122 // A particular session has been changed!
1123 int cluster_send_session(int sid
)
1125 if (!config
->cluster_iam_master
) {
1126 LOG(0, sid
, 0, "I'm not a master, but I just tried to change a session!\n");
1131 LOG(0, sid
, 0, "cluster_send_session called from child process!\n");
1135 return type_changed(C_CSESSION
, sid
);
1138 // A particular bundle has been changed!
1139 int cluster_send_bundle(int bid
)
1141 if (!config
->cluster_iam_master
) {
1142 LOG(0, 0, bid
, "I'm not a master, but I just tried to change a bundle!\n");
1146 return type_changed(C_CBUNDLE
, bid
);
1149 // A particular groupe has been changed!
1150 int cluster_send_groupe(int gid
)
1152 if (!config
->cluster_iam_master
)
1154 LOG(0, 0, gid
, "I'm not a master, but I just tried to change a groupe!\n");
1158 return type_changed(C_CGROUPE
, gid
);
1161 // A particular tunnel has been changed!
1162 int cluster_send_tunnel(int tid
)
1164 if (!config
->cluster_iam_master
) {
1165 LOG(0, 0, tid
, "I'm not a master, but I just tried to change a tunnel!\n");
1169 return type_changed(C_CTUNNEL
, tid
);
1174 // We're a master, and a slave has just told us that it's
1175 // missed a packet. We'll resend it every packet since
1176 // the last one it's seen.
1178 static int cluster_catchup_slave(int seq
, in_addr_t slave
)
1183 LOG(1, 0, 0, "Slave %s sent LASTSEEN with seq %d\n", fmtaddr(slave
, 0), seq
);
1184 if (!config
->cluster_iam_master
) {
1185 LOG(1, 0, 0, "Got LASTSEEN but I'm not a master! Redirecting it to %s.\n",
1186 fmtaddr(config
->cluster_master_address
, 0));
1188 peer_send_message(slave
, C_MASTER
, config
->cluster_master_address
, NULL
, 0);
1192 diff
= config
->cluster_seq_number
- seq
; // How many packet do we need to send?
1196 if (diff
>= HB_HISTORY_SIZE
) { // Ouch. We don't have the packet to send it!
1197 LOG(0, 0, 0, "A slave asked for message %d when our seq number is %d. Killing it.\n",
1198 seq
, config
->cluster_seq_number
);
1199 return peer_send_message(slave
, C_KILL
, seq
, NULL
, 0);// Kill the slave. Nothing else to do.
1202 LOG(1, 0, 0, "Sending %d catchup packets to slave %s\n", diff
, fmtaddr(slave
, 0) );
1204 // Now resend every packet that it missed, in order.
1205 while (seq
!= config
->cluster_seq_number
) {
1206 s
= seq
% HB_HISTORY_SIZE
;
1207 if (seq
!= past_hearts
[s
].seq
) {
1208 LOG(0, 0, 0, "Tried to re-send heartbeat for %s but %d doesn't match %d! (%d,%d)\n",
1209 fmtaddr(slave
, 0), seq
, past_hearts
[s
].seq
, s
, config
->cluster_seq_number
);
1210 return -1; // What to do here!?
1212 peer_send_data(slave
, past_hearts
[s
].data
, past_hearts
[s
].size
);
1213 seq
= (seq
+1)%HB_MAX_SEQ
; // Increment to next seq number.
1215 return 0; // All good!
1219 // We've heard from another peer! Add it to the list
1220 // that we select from at election time.
1222 static int cluster_add_peer(in_addr_t peer
, time_t basetime
, pingt
*pp
, int size
)
1225 in_addr_t clusterid
;
1228 // Allow for backward compatability.
1229 // Just the ping packet into a new structure to allow
1230 // for the possibility that we might have received
1231 // more or fewer elements than we were expecting.
1232 if (size
> sizeof(p
))
1235 memset( (void *) &p
, 0, sizeof(p
) );
1236 memcpy( (void *) &p
, (void *) pp
, size
);
1239 if (clusterid
!= config
->bind_address
)
1242 LOG(4, 0, 0, "Skipping ping from %s (different cluster)\n", fmtaddr(peer
, 0));
1246 for (i
= 0; i
< num_peers
; ++i
)
1248 if (peers
[i
].peer
!= peer
)
1251 // This peer already exists. Just update the timestamp.
1252 peers
[i
].basetime
= basetime
;
1253 peers
[i
].timestamp
= TIME
;
1254 peers
[i
].uptodate
= !p
.undef
;
1258 // Is this the master shutting down??
1259 if (peer
== config
->cluster_master_address
) {
1260 LOG(3, 0, 0, "Master %s %s\n", fmtaddr(config
->cluster_master_address
, 0),
1261 basetime
? "has restarted!" : "shutting down...");
1263 config
->cluster_master_address
= 0;
1264 config
->cluster_last_hb
= 0; // Force an election.
1265 cluster_check_master();
1270 LOG(4, 0, 0, "Adding %s as a peer\n", fmtaddr(peer
, 0));
1272 // Not found. Is there a stale slot to re-use?
1273 for (i
= 0; i
< num_peers
; ++i
)
1275 if (!peers
[i
].basetime
) // Shutdown
1278 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
* 10) < TIME
) // Stale.
1282 if (i
>= CLUSTER_MAX_SIZE
)
1285 LOG(0, 0, 0, "Tried to add %s as a peer, but I already have %d of them!\n", fmtaddr(peer
, 0), i
);
1289 peers
[i
].peer
= peer
;
1290 peers
[i
].basetime
= basetime
;
1291 peers
[i
].timestamp
= TIME
;
1292 peers
[i
].uptodate
= !p
.undef
;
1296 LOG(1, 0, 0, "Added %s as a new peer. Now %d peers\n", fmtaddr(peer
, 0), num_peers
);
1302 // A slave responds with C_MASTER when it gets a message which should have gone to a master.
1303 static int cluster_set_master(in_addr_t peer
, in_addr_t master
)
1305 if (config
->cluster_iam_master
) // Sanity...
1308 LOG(3, 0, 0, "Peer %s set the master to %s...\n", fmtaddr(peer
, 0),
1309 fmtaddr(master
, 1));
1311 config
->cluster_master_address
= master
;
1314 // catchup with new master
1315 peer_send_message(master
, C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
1317 // delay next election
1318 config
->cluster_last_hb
= TIME
;
1321 // run election (or reset "probed" if master was set)
1322 cluster_check_master();
1326 /* Handle the slave updating the byte counters for the master. */
1328 // Note that we don't mark the session as dirty; We rely on
1329 // the slow table walk to propogate this back out to the slaves.
1331 static int cluster_handle_bytes(uint8_t *data
, int size
)
1335 b
= (bytest
*) data
;
1337 LOG(3, 0, 0, "Got byte counter update (size %d)\n", size
);
1339 /* Loop around, adding the byte
1340 counts to each of the sessions. */
1342 while (size
>= sizeof(*b
) ) {
1343 if (b
->sid
> MAXSESSION
) {
1344 LOG(0, 0, 0, "Got C_BYTES with session #%d!\n", b
->sid
);
1345 return -1; /* Abort processing */
1348 session
[b
->sid
].pin
+= b
->pin
;
1349 session
[b
->sid
].pout
+= b
->pout
;
1351 increment_counter(&session
[b
->sid
].cin
, &session
[b
->sid
].cin_wrap
, b
->cin
);
1352 increment_counter(&session
[b
->sid
].cout
, &session
[b
->sid
].cout_wrap
, b
->cout
);
1354 session
[b
->sid
].cin_delta
+= b
->cin
;
1355 session
[b
->sid
].cout_delta
+= b
->cout
;
1358 session
[b
->sid
].last_packet
= session
[b
->sid
].last_data
= time_now
;
1360 session
[b
->sid
].last_data
= time_now
;
1367 LOG(0, 0, 0, "Got C_BYTES with %d bytes of trailing junk!\n", size
);
1373 // Handle receiving a session structure in a heartbeat packet.
1375 static int cluster_recv_session(int more
, uint8_t *p
)
1377 if (more
>= MAXSESSION
) {
1378 LOG(0, 0, 0, "DANGER: Received a heartbeat session id > MAXSESSION!\n");
1382 if (session
[more
].tunnel
== T_UNDEF
) {
1383 if (config
->cluster_iam_uptodate
) { // Sanity.
1384 LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined session!\n");
1386 --config
->cluster_undefined_sessions
;
1390 load_session(more
, (sessiont
*) p
); // Copy session into session table..
1392 LOG(5, more
, 0, "Received session update (%d undef)\n", config
->cluster_undefined_sessions
);
1394 if (!config
->cluster_iam_uptodate
)
1395 cluster_uptodate(); // Check to see if we're up to date.
1400 static int cluster_recv_bundle(int more
, uint8_t *p
)
1402 if (more
>= MAXBUNDLE
) {
1403 LOG(0, 0, 0, "DANGER: Received a bundle id > MAXBUNDLE!\n");
1407 if (bundle
[more
].state
== BUNDLEUNDEF
) {
1408 if (config
->cluster_iam_uptodate
) { // Sanity.
1409 LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined bundle!\n");
1411 --config
->cluster_undefined_bundles
;
1415 memcpy(&bundle
[more
], p
, sizeof(bundle
[more
]) );
1417 LOG(5, 0, more
, "Received bundle update\n");
1419 if (!config
->cluster_iam_uptodate
)
1420 cluster_uptodate(); // Check to see if we're up to date.
1425 static int cluster_recv_groupe(int more
, uint8_t *p
)
1427 if (more
>= MAXGROUPE
) {
1428 LOG(0, 0, 0, "DANGER: Received a group id > MAXGROUPE!\n");
1432 if (grpsession
[more
].state
== GROUPEUNDEF
) {
1433 if (config
->cluster_iam_uptodate
) { // Sanity.
1434 LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined group!\n");
1436 --config
->cluster_undefined_groupes
;
1440 grp_cluster_load_groupe(more
, (groupsesst
*) p
); // Copy groupe into groupe table..
1442 LOG(5, 0, more
, "Received group update (%d undef)\n", config
->cluster_undefined_groupes
);
1444 if (!config
->cluster_iam_uptodate
)
1445 cluster_uptodate(); // Check to see if we're up to date.
1450 static int cluster_recv_tunnel(int more
, uint8_t *p
)
1452 if (more
>= MAXTUNNEL
) {
1453 LOG(0, 0, 0, "DANGER: Received a tunnel session id > MAXTUNNEL!\n");
1457 if (tunnel
[more
].state
== TUNNELUNDEF
) {
1458 if (config
->cluster_iam_uptodate
) { // Sanity.
1459 LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined tunnel!\n");
1461 --config
->cluster_undefined_tunnels
;
1465 memcpy(&tunnel
[more
], p
, sizeof(tunnel
[more
]) );
1468 // Clear tunnel control messages. These are dynamically allocated.
1469 // If we get unlucky, this may cause the tunnel to drop!
1471 tunnel
[more
].controls
= tunnel
[more
].controle
= NULL
;
1472 tunnel
[more
].controlc
= 0;
1474 LOG(5, 0, more
, "Received tunnel update\n");
1476 if (!config
->cluster_iam_uptodate
)
1477 cluster_uptodate(); // Check to see if we're up to date.
1483 // pre v6 heartbeat session structure
1504 uint32_t cin_wrap
, cout_wrap
;
1505 uint32_t cin_delta
, cout_delta
;
1506 uint16_t throttle_in
;
1507 uint16_t throttle_out
;
1513 uint32_t session_timeout
;
1514 uint32_t idle_timeout
;
1517 in_addr_t dns1
, dns2
;
1518 routet route
[MAXROUTE
];
1521 int random_vector_length
;
1522 uint8_t random_vector
[MAXTEL
];
1524 char called
[MAXTEL
];
1525 char calling
[MAXTEL
];
1526 uint32_t tx_connect_speed
;
1527 uint32_t rx_connect_speed
;
1534 uint16_t snoop_port
;
1535 uint8_t walled_garden
;
1536 uint8_t ipv6prefixlen
;
1537 struct in6_addr ipv6route
;
1538 char reserved_3
[11];
1541 static uint8_t *convert_session(struct oldsession
*old
)
1543 static sessiont
new;
1546 memset(&new, 0, sizeof(new));
1548 new.next
= old
->next
;
1550 new.tunnel
= old
->tunnel
;
1551 new.flags
= old
->flags
;
1552 new.ppp
.phase
= old
->ppp
.phase
;
1553 new.ppp
.lcp
= old
->ppp
.lcp
;
1554 new.ppp
.ipcp
= old
->ppp
.ipcp
;
1555 new.ppp
.ipv6cp
= old
->ppp
.ipv6cp
;
1556 new.ppp
.ccp
= old
->ppp
.ccp
;
1558 new.ip_pool_index
= old
->ip_pool_index
;
1559 new.unique_id
= old
->unique_id
;
1560 new.magic
= old
->magic
;
1562 new.pout
= old
->pout
;
1564 new.cout
= old
->cout
;
1565 new.cin_wrap
= old
->cin_wrap
;
1566 new.cout_wrap
= old
->cout_wrap
;
1567 new.cin_delta
= old
->cin_delta
;
1568 new.cout_delta
= old
->cout_delta
;
1569 new.throttle_in
= old
->throttle_in
;
1570 new.throttle_out
= old
->throttle_out
;
1571 new.filter_in
= old
->filter_in
;
1572 new.filter_out
= old
->filter_out
;
1574 new.opened
= old
->opened
;
1576 new.session_timeout
= old
->session_timeout
;
1577 new.idle_timeout
= old
->idle_timeout
;
1578 new.last_packet
= old
->last_packet
;
1579 new.last_data
= old
->last_data
;
1580 new.dns1
= old
->dns1
;
1581 new.dns2
= old
->dns2
;
1582 new.tbf_in
= old
->tbf_in
;
1583 new.tbf_out
= old
->tbf_out
;
1584 new.random_vector_length
= old
->random_vector_length
;
1585 new.tx_connect_speed
= old
->tx_connect_speed
;
1586 new.rx_connect_speed
= old
->rx_connect_speed
;
1587 new.timeout
= old
->timeout
;
1588 new.mrru
= old
->mrru
;
1589 new.mssf
= old
->mssf
;
1590 new.epdis
= old
->epdis
;
1591 new.bundle
= old
->bundle
;
1592 new.snoop_ip
= old
->snoop_ip
;
1593 new.snoop_port
= old
->snoop_port
;
1594 new.walled_garden
= old
->walled_garden
;
1595 new.ipv6prefixlen
= old
->ipv6prefixlen
;
1596 new.ipv6route
= old
->ipv6route
;
1598 memcpy(new.random_vector
, old
->random_vector
, sizeof(new.random_vector
));
1599 memcpy(new.user
, old
->user
, sizeof(new.user
));
1600 memcpy(new.called
, old
->called
, sizeof(new.called
));
1601 memcpy(new.calling
, old
->calling
, sizeof(new.calling
));
1603 for (i
= 0; i
< MAXROUTE
; i
++)
1604 memcpy(&new.route
[i
], &old
->route
[i
], sizeof(new.route
[i
]));
1606 return (uint8_t *) &new;
1610 // Process a heartbeat..
1612 // v6: added RADIUS class attribute, re-ordered session structure
1613 // v7: added tunnelt attribute at the end of struct (tunnelt size change)
1614 static int cluster_process_heartbeat(uint8_t *data
, int size
, int more
, uint8_t *p
, in_addr_t addr
)
1617 int s
= size
- (p
-data
);
1622 # error "need to update cluster_process_heartbeat()"
1625 // we handle versions 5 through 8
1626 if (hb_ver
< 5 || hb_ver
> HB_VERSION
) {
1627 LOG(0, 0, 0, "Received a heartbeat version that I don't support (%d)!\n", hb_ver
);
1628 return -1; // Ignore it??
1631 if (size
> sizeof(past_hearts
[0].data
)) {
1632 LOG(0, 0, 0, "Received an oversize heartbeat from %s (%d)!\n", fmtaddr(addr
, 0), size
);
1643 if (h
->clusterid
!= config
->bind_address
)
1644 return -1; // It's not part of our cluster.
1646 if (config
->cluster_iam_master
) { // Sanity...
1647 // Note that this MUST match the election process above!
1649 LOG(0, 0, 0, "I just got a heartbeat from master %s, but _I_ am the master!\n", fmtaddr(addr
, 0));
1651 LOG(0, 0, 0, "Heartbeat with zero basetime! Ignoring\n");
1652 return -1; // Skip it.
1655 if (h
->table_version
> config
->cluster_table_version
) {
1656 LOG(0, 0, 0, "They've seen more state changes (%" PRIu64
" vs my %" PRIu64
") so I'm gone!\n",
1657 h
->table_version
, config
->cluster_table_version
);
1663 if (h
->table_version
< config
->cluster_table_version
)
1666 if (basetime
> h
->basetime
) {
1667 LOG(0, 0, 0, "They're an older master than me so I'm gone!\n");
1672 if (basetime
< h
->basetime
)
1675 if (my_address
< addr
) { // Tie breaker.
1676 LOG(0, 0, 0, "They're a higher IP address than me, so I'm gone!\n");
1682 // Send it a unicast heartbeat to see give it a chance to die.
1683 // NOTE: It's actually safe to do seq-number - 1 without checking
1686 cluster_catchup_slave(config
->cluster_seq_number
- 1, addr
);
1688 return -1; // Skip it.
1692 // Try and guard against a stray master appearing.
1694 // Ignore heartbeats received from another master before the
1695 // timeout (less a smidgen) for the old master has elapsed.
1697 // Note that after a clean failover, the cluster_master_address
1698 // is cleared, so this doesn't run.
1700 if (config
->cluster_master_address
&& addr
!= config
->cluster_master_address
) {
1701 LOG(0, 0, 0, "Ignoring stray heartbeat from %s, current master %s has not yet timed out (last heartbeat %.1f seconds ago).\n",
1702 fmtaddr(addr
, 0), fmtaddr(config
->cluster_master_address
, 1),
1703 0.1 * (TIME
- config
->cluster_last_hb
));
1704 return -1; // ignore
1707 if (config
->cluster_seq_number
== -1) // Don't have one. Just align to the master...
1708 config
->cluster_seq_number
= h
->seq
;
1710 config
->cluster_last_hb
= TIME
; // Reset to ensure that we don't become master!!
1711 config
->cluster_last_hb_ver
= hb_ver
; // remember what cluster version the master is using
1713 if (config
->cluster_seq_number
!= h
->seq
) { // Out of sequence heartbeat!
1714 static int lastseen_seq
= 0;
1715 static time_t lastseen_time
= 0;
1717 // limit to once per second for a particular seq#
1718 int ask
= (config
->cluster_seq_number
!= lastseen_seq
|| time_now
!= lastseen_time
);
1720 LOG(1, 0, 0, "HB: Got seq# %d but was expecting %d. %s.\n",
1721 h
->seq
, config
->cluster_seq_number
,
1722 ask
? "Asking for resend" : "Ignoring");
1726 lastseen_seq
= config
->cluster_seq_number
;
1727 lastseen_time
= time_now
;
1728 peer_send_message(addr
, C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
1731 config
->cluster_last_hb
= TIME
; // Reset to ensure that we don't become master!!
1733 // Just drop the packet. The master will resend it as part of the catchup.
1737 // Save the packet in our buffer.
1738 // This is needed in case we become the master.
1739 config
->cluster_seq_number
= (h
->seq
+1)%HB_MAX_SEQ
;
1740 i
= h
->seq
% HB_HISTORY_SIZE
;
1741 past_hearts
[i
].seq
= h
->seq
;
1742 past_hearts
[i
].size
= size
;
1743 memcpy(&past_hearts
[i
].data
, data
, size
); // Save it.
1746 // Check that we don't have too many undefined sessions, and
1747 // that the free session pointer is correct.
1748 gnextgrpid
= h
->nextgrpid
;
1749 cluster_check_sessions(h
->highsession
, h
->freesession
, h
->highbundle
, h
->hightunnel
, h
->highgroupe
);
1751 if (h
->interval
!= config
->cluster_hb_interval
)
1753 LOG(2, 0, 0, "Master set ping/heartbeat interval to %u (was %u)\n",
1754 h
->interval
, config
->cluster_hb_interval
);
1756 config
->cluster_hb_interval
= h
->interval
;
1759 if (h
->timeout
!= config
->cluster_hb_timeout
)
1761 LOG(2, 0, 0, "Master set heartbeat timeout to %u (was %u)\n",
1762 h
->timeout
, config
->cluster_hb_timeout
);
1764 config
->cluster_hb_timeout
= h
->timeout
;
1767 // Ok. process the packet...
1770 type
= *((uint32_t *) p
);
1771 p
+= sizeof(uint32_t);
1772 s
-= sizeof(uint32_t);
1774 more
= *((uint32_t *) p
);
1775 p
+= sizeof(uint32_t);
1776 s
-= sizeof(uint32_t);
1779 case C_CSESSION
: { // Compressed session structure.
1780 uint8_t c
[ sizeof(sessiont
) + 2];
1782 uint8_t *orig_p
= p
;
1784 size
= rle_decompress((uint8_t **) &p
, s
, c
, sizeof(c
) );
1787 // session struct changed with v5
1790 if (size
!= sizeof(struct oldsession
)) {
1791 LOG(0, 0, 0, "DANGER: Received a v%d CSESSION that didn't decompress correctly!\n", hb_ver
);
1792 // Now what? Should exit! No-longer up to date!
1795 cluster_recv_session(more
, convert_session((struct oldsession
*) c
));
1799 if (size
!= sizeof(sessiont
)) { // Ouch! Very very bad!
1800 if ((hb_ver
< HB_VERSION
) && (size
< sizeof(sessiont
)))
1802 // set to 0 the unused variables
1803 memset(&c
[size
], 0, (sizeof(sessiont
) - size
));
1804 LOG(3, 0, 0, "WARNING: Received a CSESSION from %s hb_version %d != %d current version !\n", fmtaddr(addr
, 2), hb_ver
, HB_VERSION
);
1805 // New feature not activated until the master has not been upgraded.
1809 LOG(0, 0, 0, "DANGER: Received a CSESSION that didn't decompress correctly!\n");
1810 // Now what? Should exit! No-longer up to date!
1815 cluster_recv_session(more
, c
);
1821 if (s
< sizeof(struct oldsession
))
1824 cluster_recv_session(more
, convert_session((struct oldsession
*) p
));
1826 p
+= sizeof(struct oldsession
);
1827 s
-= sizeof(struct oldsession
);
1831 if ( s
< sizeof(session
[more
]))
1834 cluster_recv_session(more
, p
);
1836 p
+= sizeof(session
[more
]);
1837 s
-= sizeof(session
[more
]);
1840 case C_CTUNNEL
: { // Compressed tunnel structure.
1841 uint8_t c
[ sizeof(tunnelt
) + 2];
1843 uint8_t *orig_p
= p
;
1845 size
= rle_decompress((uint8_t **) &p
, s
, c
, sizeof(c
));
1848 if ( ((hb_ver
>= HB_VERSION
) && (size
!= sizeof(tunnelt
))) ||
1849 ((hb_ver
< HB_VERSION
) && (size
> sizeof(tunnelt
))) )
1850 { // Ouch! Very very bad!
1851 LOG(0, 0, 0, "DANGER: Received a CTUNNEL that didn't decompress correctly!\n");
1852 // Now what? Should exit! No-longer up to date!
1856 cluster_recv_tunnel(more
, c
);
1861 if ( s
< sizeof(tunnel
[more
]))
1864 cluster_recv_tunnel(more
, p
);
1866 p
+= sizeof(tunnel
[more
]);
1867 s
-= sizeof(tunnel
[more
]);
1870 case C_CBUNDLE
: { // Compressed bundle structure.
1871 uint8_t c
[ sizeof(bundlet
) + 2];
1873 uint8_t *orig_p
= p
;
1875 size
= rle_decompress((uint8_t **) &p
, s
, c
, sizeof(c
));
1878 if (size
!= sizeof(bundlet
) ) { // Ouch! Very very bad!
1879 LOG(0, 0, 0, "DANGER: Received a CBUNDLE that didn't decompress correctly!\n");
1880 // Now what? Should exit! No-longer up to date!
1884 cluster_recv_bundle(more
, c
);
1889 if ( s
< sizeof(bundle
[more
]))
1892 cluster_recv_bundle(more
, p
);
1894 p
+= sizeof(bundle
[more
]);
1895 s
-= sizeof(bundle
[more
]);
1899 { // Compressed Groupe structure.
1900 uint8_t c
[ sizeof(groupsesst
) + 2];
1902 uint8_t *orig_p
= p
;
1904 size
= rle_decompress((uint8_t **) &p
, s
, c
, sizeof(c
));
1907 if (size
!= sizeof(groupsesst
) )
1908 { // Ouch! Very very bad!
1909 LOG(0, 0, 0, "DANGER: Received a C_CGROUPE that didn't decompress correctly!\n");
1910 // Now what? Should exit! No-longer up to date!
1914 cluster_recv_groupe(more
, c
);
1918 if ( s
< sizeof(grpsession
[more
]))
1921 cluster_recv_groupe(more
, p
);
1923 p
+= sizeof(grpsession
[more
]);
1924 s
-= sizeof(grpsession
[more
]);
1928 LOG(0, 0, 0, "DANGER: I received a heartbeat element where I didn't understand the type! (%d)\n", type
);
1929 return -1; // can't process any more of the packet!!
1933 if (config
->cluster_master_address
!= addr
)
1935 LOG(0, 0, 0, "My master just changed from %s to %s!\n",
1936 fmtaddr(config
->cluster_master_address
, 0), fmtaddr(addr
, 1));
1938 config
->cluster_master_address
= addr
;
1941 config
->cluster_last_hb
= TIME
; // Successfully received a heartbeat!
1942 config
->cluster_table_version
= h
->table_version
;
1946 LOG(0, 0, 0, "I got an incomplete heartbeat packet! This means I'm probably out of sync!!\n");
1951 // We got a packet on the cluster port!
1952 // Handle pings, lastseens, and heartbeats!
1954 int processcluster(uint8_t *data
, int size
, in_addr_t addr
)
1960 if (addr
== my_address
)
1961 return -1; // Ignore it. Something looped back the multicast!
1963 LOG(5, 0, 0, "Process cluster: %d bytes from %s\n", size
, fmtaddr(addr
, 0));
1965 if (s
<= 0) // Any data there??
1971 type
= *((uint32_t *) p
);
1972 p
+= sizeof(uint32_t);
1973 s
-= sizeof(uint32_t);
1975 more
= *((uint32_t *) p
);
1976 p
+= sizeof(uint32_t);
1977 s
-= sizeof(uint32_t);
1981 case C_PING
: // Update the peers table.
1982 return cluster_add_peer(addr
, more
, (pingt
*) p
, s
);
1984 case C_MASTER
: // Our master is wrong
1985 return cluster_set_master(addr
, more
);
1987 case C_LASTSEEN
: // Catch up a slave (slave missed a packet).
1988 return cluster_catchup_slave(more
, addr
);
1990 case C_FORWARD
: // Forwarded control packet. pass off to processudp.
1991 case C_FORWARD_DAE
: // Forwarded DAE packet. pass off to processdae.
1992 if (!config
->cluster_iam_master
)
1994 LOG(0, 0, 0, "I'm not the master, but I got a C_FORWARD%s from %s?\n",
1995 type
== C_FORWARD_DAE
? "_DAE" : "", fmtaddr(addr
, 0));
2001 struct sockaddr_in a
;
2003 a
.sin_addr
.s_addr
= more
;
2005 a
.sin_port
= (*(int *) p
) & 0xFFFF;
2006 indexudp
= ((*(int *) p
) >> 16) & 0xFFFF;
2010 LOG(4, 0, 0, "Got a forwarded %spacket... (%s:%d)\n",
2011 type
== C_FORWARD_DAE
? "DAE " : "", fmtaddr(more
, 0), a
.sin_port
);
2014 if (type
== C_FORWARD_DAE
)
2016 struct in_addr local
;
2017 local
.s_addr
= config
->bind_address
? config
->bind_address
: my_address
;
2018 processdae(p
, s
, &a
, sizeof(a
), &local
);
2021 processudp(p
, s
, &a
, indexudp
);
2025 case C_PPPOE_FORWARD
:
2026 if (!config
->cluster_iam_master
)
2028 LOG(0, 0, 0, "I'm not the master, but I got a C_PPPOE_FORWARD from %s?\n", fmtaddr(addr
, 0));
2033 pppoe_process_forward(p
, s
, addr
);
2037 case C_MPPP_FORWARD
:
2038 // Receive a MPPP packet from a slave.
2039 if (!config
->cluster_iam_master
) {
2040 LOG(0, 0, 0, "I'm not the master, but I got a C_MPPP_FORWARD from %s?\n", fmtaddr(addr
, 0));
2047 case C_THROTTLE
: { // Receive a forwarded packet from a slave.
2048 if (!config
->cluster_iam_master
) {
2049 LOG(0, 0, 0, "I'm not the master, but I got a C_THROTTLE from %s?\n", fmtaddr(addr
, 0));
2053 tbf_queue_packet(more
, p
, s
); // The TBF id tells wether it goes in or out.
2057 // Receive a walled garden packet from a slave.
2058 if (!config
->cluster_iam_master
) {
2059 LOG(0, 0, 0, "I'm not the master, but I got a C_GARDEN from %s?\n", fmtaddr(addr
, 0));
2067 if (!config
->cluster_iam_master
) {
2068 LOG(0, 0, 0, "I'm not the master, but I got a C_BYTES from %s?\n", fmtaddr(addr
, 0));
2072 return cluster_handle_bytes(p
, s
);
2074 case C_KILL
: // The master asked us to die!? (usually because we're too out of date).
2075 if (config
->cluster_iam_master
) {
2076 LOG(0, 0, 0, "_I_ am master, but I received a C_KILL from %s! (Seq# %d)\n", fmtaddr(addr
, 0), more
);
2079 if (more
!= config
->cluster_seq_number
) {
2080 LOG(0, 0, 0, "The master asked us to die but the seq number didn't match!?\n");
2084 if (addr
!= config
->cluster_master_address
) {
2085 LOG(0, 0, 0, "Received a C_KILL from %s which doesn't match config->cluster_master_address (%s)\n",
2086 fmtaddr(addr
, 0), fmtaddr(config
->cluster_master_address
, 1));
2087 // We can only warn about it. The master might really have switched!
2090 LOG(0, 0, 0, "Received a valid C_KILL: I'm going to die now.\n");
2092 exit(0); // Lets be paranoid;
2093 return -1; // Just signalling the compiler.
2096 LOG(4, 0, 0, "Got a heartbeat from %s\n", fmtaddr(addr
, 0));
2097 return cluster_process_heartbeat(data
, size
, more
, p
, addr
);
2100 LOG(0, 0, 0, "Strange type packet received on cluster socket (%d)\n", type
);
2106 LOG(0, 0, 0, "I got a _short_ cluster heartbeat packet! This means I'm probably out of sync!!\n");
2110 //====================================================================================================
2112 int cmd_show_cluster(struct cli_def
*cli
, const char *command
, char **argv
, int argc
)
2116 if (CLI_HELP_REQUESTED
)
2117 return CLI_HELP_NO_ARGS
;
2119 cli_print(cli
, "Cluster status : %s", config
->cluster_iam_master
? "Master" : "Slave" );
2120 cli_print(cli
, "My address : %s", fmtaddr(my_address
, 0));
2121 cli_print(cli
, "VIP address : %s", fmtaddr(config
->bind_address
, 0));
2122 cli_print(cli
, "Multicast address: %s", fmtaddr(config
->cluster_address
, 0));
2123 cli_print(cli
, "Multicast i'face : %s", config
->cluster_interface
);
2125 if (!config
->cluster_iam_master
) {
2126 cli_print(cli
, "My master : %s (last heartbeat %.1f seconds old)",
2127 config
->cluster_master_address
2128 ? fmtaddr(config
->cluster_master_address
, 0)
2130 0.1 * (TIME
- config
->cluster_last_hb
));
2131 cli_print(cli
, "Uptodate : %s", config
->cluster_iam_uptodate
? "Yes" : "No");
2132 cli_print(cli
, "Table version # : %" PRIu64
, config
->cluster_table_version
);
2133 cli_print(cli
, "Next sequence number expected: %d", config
->cluster_seq_number
);
2134 cli_print(cli
, "%d sessions undefined of %d", config
->cluster_undefined_sessions
, config
->cluster_highest_sessionid
);
2135 cli_print(cli
, "%d bundles undefined of %d", config
->cluster_undefined_bundles
, config
->cluster_highest_bundleid
);
2136 cli_print(cli
, "%d groupes undefined of %d", config
->cluster_undefined_groupes
, config
->cluster_highest_groupeid
);
2137 cli_print(cli
, "%d tunnels undefined of %d", config
->cluster_undefined_tunnels
, config
->cluster_highest_tunnelid
);
2139 cli_print(cli
, "Table version # : %" PRIu64
, config
->cluster_table_version
);
2140 cli_print(cli
, "Next heartbeat # : %d", config
->cluster_seq_number
);
2141 cli_print(cli
, "Highest session : %d", config
->cluster_highest_sessionid
);
2142 cli_print(cli
, "Highest bundle : %d", config
->cluster_highest_bundleid
);
2143 cli_print(cli
, "Highest groupe : %d", config
->cluster_highest_groupeid
);
2144 cli_print(cli
, "Highest tunnel : %d", config
->cluster_highest_tunnelid
);
2145 cli_print(cli
, "%d changes queued for sending", config
->cluster_num_changes
);
2147 cli_print(cli
, "%d peers.", num_peers
);
2150 cli_print(cli
, "%20s %10s %8s", "Address", "Basetime", "Age");
2151 for (i
= 0; i
< num_peers
; ++i
) {
2152 cli_print(cli
, "%20s %10u %8d", fmtaddr(peers
[i
].peer
, 0),
2153 peers
[i
].basetime
, TIME
- peers
[i
].timestamp
);
2159 // Simple run-length-encoding compression.
2161 // 1 byte < 128 = count of non-zero bytes following. // Not legal to be zero.
2162 // n non-zero bytes;
2164 // 1 byte > 128 = (count - 128) run of zero bytes. //
2166 // count == 0 indicates end of compressed stream.
2168 // Compress from 'src' into 'dst'. return number of bytes
2170 // Updates *src_p to indicate 1 past last bytes used.
2172 // We could get an extra byte in the zero runs by storing (count-1)
2173 // but I'm playing it safe.
2175 // Worst case is a 50% expansion in space required (trying to
2176 // compress { 0x00, 0x01 } * N )
2177 static int rle_compress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
)
2180 int orig_dsize
= dsize
;
2184 while (ssize
> 0 && dsize
> 2) {
2186 x
= dst
++; --dsize
; // Reserve space for count byte..
2188 if (*src
) { // Copy a run of non-zero bytes.
2189 while (*src
&& count
< 127 && ssize
> 0 && dsize
> 1) { // Count number of non-zero bytes.
2194 *x
= count
; // Store number of non-zero bytes. Guarenteed to be non-zero!
2196 } else { // Compress a run of zero bytes.
2197 while (*src
== 0 && count
< 127 && ssize
> 0) {
2206 *dst
++ = 0x0; // Add Stop byte.
2210 return (orig_dsize
- dsize
);
2214 // Decompress the buffer into **p.
2215 // 'psize' is the size of the decompression buffer available.
2217 // Returns the number of bytes decompressed.
2219 // Decompresses from '*src_p' into 'dst'.
2220 // Return the number of dst bytes used.
2221 // Updates the 'src_p' pointer to point to the
2222 // first un-used byte.
2223 static int rle_decompress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
)
2226 int orig_dsize
= dsize
;
2227 uint8_t *src
= *src_p
;
2229 while (ssize
>0 && dsize
> 0) { // While there's more to decompress, and there's room in the decompress buffer...
2230 count
= *src
++; --ssize
; // get the count byte from the source.
2231 if (count
== 0x0) // End marker reached? If so, finish.
2234 if (count
& 0x80) { // Decompress a run of zeros
2235 for (count
&= 0x7f ; count
> 0 && dsize
> 0; --count
) {
2239 } else { // Copy run of non-zero bytes.
2240 for ( ; count
> 0 && ssize
&& dsize
; --count
) { // Copy non-zero bytes across.
2247 return (orig_dsize
- dsize
);