forked from Azure/haproxy
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpeers.c
More file actions
3315 lines (2848 loc) · 93.5 KB
/
Copy pathpeers.c
File metadata and controls
3315 lines (2848 loc) · 93.5 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
* Peer synchro management.
*
* Copyright 2010 EXCELIANCE, Emeric Brun <ebrun@exceliance.fr>
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version
* 2 of the License, or (at your option) any later version.
*
*/
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <haproxy/api.h>
#include <haproxy/applet.h>
#include <haproxy/channel.h>
#include <haproxy/cli.h>
#include <haproxy/dict.h>
#include <haproxy/errors.h>
#include <haproxy/fd.h>
#include <haproxy/frontend.h>
#include <haproxy/net_helper.h>
#include <haproxy/obj_type-t.h>
#include <haproxy/peers.h>
#include <haproxy/proxy.h>
#include <haproxy/session-t.h>
#include <haproxy/signal.h>
#include <haproxy/stats-t.h>
#include <haproxy/stick_table.h>
#include <haproxy/stream.h>
#include <haproxy/stream_interface.h>
#include <haproxy/task.h>
#include <haproxy/thread.h>
#include <haproxy/time.h>
#include <haproxy/tools.h>
/*******************************/
/* Current peer learning state */
/*******************************/
/******************************/
/* Current peers section resync state */
/******************************/
#define PEERS_F_RESYNC_LOCAL 0x00000001 /* Learn from local finished or no more needed */
#define PEERS_F_RESYNC_REMOTE 0x00000002 /* Learn from remote finished or no more needed */
#define PEERS_F_RESYNC_ASSIGN 0x00000004 /* A peer was assigned to learn our lesson */
#define PEERS_F_RESYNC_PROCESS 0x00000008 /* The assigned peer was requested for resync */
#define PEERS_F_DONOTSTOP 0x00010000 /* Main table sync task block process during soft stop
to push data to new process */
#define PEERS_RESYNC_STATEMASK (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
#define PEERS_RESYNC_FROMLOCAL 0x00000000
#define PEERS_RESYNC_FROMREMOTE PEERS_F_RESYNC_LOCAL
#define PEERS_RESYNC_FINISHED (PEERS_F_RESYNC_LOCAL|PEERS_F_RESYNC_REMOTE)
/***********************************/
/* Current shared table sync state */
/***********************************/
#define SHTABLE_F_TEACH_STAGE1 0x00000001 /* Teach state 1 complete */
#define SHTABLE_F_TEACH_STAGE2 0x00000002 /* Teach state 2 complete */
/******************************/
/* Remote peer teaching state */
/******************************/
#define PEER_F_TEACH_PROCESS 0x00000001 /* Teach a lesson to current peer */
#define PEER_F_TEACH_FINISHED 0x00000008 /* Teach conclude, (wait for confirm) */
#define PEER_F_TEACH_COMPLETE 0x00000010 /* All that we know already taught to current peer, used only for a local peer */
#define PEER_F_LEARN_ASSIGN 0x00000100 /* Current peer was assigned for a lesson */
#define PEER_F_LEARN_NOTUP2DATE 0x00000200 /* Learn from peer finished but peer is not up to date */
#define PEER_F_ALIVE 0x20000000 /* Used to flag a peer a alive. */
#define PEER_F_HEARTBEAT 0x40000000 /* Heartbeat message to send. */
#define PEER_F_DWNGRD 0x80000000 /* When this flag is enabled, we must downgrade the supported version announced during peer sessions. */
#define PEER_TEACH_RESET ~(PEER_F_TEACH_PROCESS|PEER_F_TEACH_FINISHED) /* PEER_F_TEACH_COMPLETE should never be reset */
#define PEER_LEARN_RESET ~(PEER_F_LEARN_ASSIGN|PEER_F_LEARN_NOTUP2DATE)
#define PEER_RESYNC_TIMEOUT 5000 /* 5 seconds */
#define PEER_RECONNECT_TIMEOUT 5000 /* 5 seconds */
#define PEER_HEARTBEAT_TIMEOUT 3000 /* 3 seconds */
/*****************************/
/* Sync message class */
/*****************************/
enum {
PEER_MSG_CLASS_CONTROL = 0,
PEER_MSG_CLASS_ERROR,
PEER_MSG_CLASS_STICKTABLE = 10,
PEER_MSG_CLASS_RESERVED = 255,
};
/*****************************/
/* control message types */
/*****************************/
enum {
PEER_MSG_CTRL_RESYNCREQ = 0,
PEER_MSG_CTRL_RESYNCFINISHED,
PEER_MSG_CTRL_RESYNCPARTIAL,
PEER_MSG_CTRL_RESYNCCONFIRM,
PEER_MSG_CTRL_HEARTBEAT,
};
/*****************************/
/* error message types */
/*****************************/
enum {
PEER_MSG_ERR_PROTOCOL = 0,
PEER_MSG_ERR_SIZELIMIT,
};
/* network key types;
* network types were directly and mistakenly
* mapped on sample types, to keep backward
* compatiblitiy we keep those values but
* we now use a internal/network mapping
* to avoid further mistakes adding or
* modifying internals types
*/
enum {
PEER_KT_ANY = 0, /* any type */
PEER_KT_RESV1, /* UNUSED */
PEER_KT_SINT, /* signed 64bits integer type */
PEER_KT_RESV3, /* UNUSED */
PEER_KT_IPV4, /* ipv4 type */
PEER_KT_IPV6, /* ipv6 type */
PEER_KT_STR, /* char string type */
PEER_KT_BIN, /* buffer type */
PEER_KT_TYPES /* number of types, must always be last */
};
/* Map used to retrieve network type from internal type
* Note: Undeclared mapping maps entry to PEER_KT_ANY == 0
*/
static int peer_net_key_type[SMP_TYPES] = {
[SMP_T_SINT] = PEER_KT_SINT,
[SMP_T_IPV4] = PEER_KT_IPV4,
[SMP_T_IPV6] = PEER_KT_IPV6,
[SMP_T_STR] = PEER_KT_STR,
[SMP_T_BIN] = PEER_KT_BIN,
};
/* Map used to retrieve internal type from external type
* Note: Undeclared mapping maps entry to SMP_T_ANY == 0
*/
static int peer_int_key_type[PEER_KT_TYPES] = {
[PEER_KT_SINT] = SMP_T_SINT,
[PEER_KT_IPV4] = SMP_T_IPV4,
[PEER_KT_IPV6] = SMP_T_IPV6,
[PEER_KT_STR] = SMP_T_STR,
[PEER_KT_BIN] = SMP_T_BIN,
};
/*
* Parameters used by functions to build peer protocol messages. */
struct peer_prep_params {
struct {
struct peer *peer;
} hello;
struct {
unsigned int st1;
} error_status;
struct {
struct stksess *stksess;
struct shared_table *shared_table;
unsigned int updateid;
int use_identifier;
int use_timed;
struct peer *peer;
} updt;
struct {
struct shared_table *shared_table;
} swtch;
struct {
struct shared_table *shared_table;
} ack;
struct {
unsigned char head[2];
} control;
struct {
unsigned char head[2];
} error;
};
/*******************************/
/* stick table sync mesg types */
/* Note: ids >= 128 contains */
/* id message contains data */
/*******************************/
#define PEER_MSG_STKT_UPDATE 0x80
#define PEER_MSG_STKT_INCUPDATE 0x81
#define PEER_MSG_STKT_DEFINE 0x82
#define PEER_MSG_STKT_SWITCH 0x83
#define PEER_MSG_STKT_ACK 0x84
#define PEER_MSG_STKT_UPDATE_TIMED 0x85
#define PEER_MSG_STKT_INCUPDATE_TIMED 0x86
/* All the stick-table message identifiers abova have the #7 bit set */
#define PEER_MSG_STKT_BIT 7
#define PEER_MSG_STKT_BIT_MASK (1 << PEER_MSG_STKT_BIT)
/* The maximum length of an encoded data length. */
#define PEER_MSG_ENC_LENGTH_MAXLEN 5
/* Minimum 64-bits value encoded with 2 bytes */
#define PEER_ENC_2BYTES_MIN 0xf0 /* 0xf0 (or 240) */
/* 3 bytes */
#define PEER_ENC_3BYTES_MIN ((1ULL << 11) | PEER_ENC_2BYTES_MIN) /* 0x8f0 (or 2288) */
/* 4 bytes */
#define PEER_ENC_4BYTES_MIN ((1ULL << 18) | PEER_ENC_3BYTES_MIN) /* 0x408f0 (or 264432) */
/* 5 bytes */
#define PEER_ENC_5BYTES_MIN ((1ULL << 25) | PEER_ENC_4BYTES_MIN) /* 0x20408f0 (or 33818864) */
/* 6 bytes */
#define PEER_ENC_6BYTES_MIN ((1ULL << 32) | PEER_ENC_5BYTES_MIN) /* 0x1020408f0 (or 4328786160) */
/* 7 bytes */
#define PEER_ENC_7BYTES_MIN ((1ULL << 39) | PEER_ENC_6BYTES_MIN) /* 0x81020408f0 (or 554084600048) */
/* 8 bytes */
#define PEER_ENC_8BYTES_MIN ((1ULL << 46) | PEER_ENC_7BYTES_MIN) /* 0x4081020408f0 (or 70922828777712) */
/* 9 bytes */
#define PEER_ENC_9BYTES_MIN ((1ULL << 53) | PEER_ENC_8BYTES_MIN) /* 0x204081020408f0 (or 9078122083518704) */
/* 10 bytes */
#define PEER_ENC_10BYTES_MIN ((1ULL << 60) | PEER_ENC_9BYTES_MIN) /* 0x10204081020408f0 (or 1161999626690365680) */
/* #7 bit used to detect the last byte to be encoded */
#define PEER_ENC_STOP_BIT 7
/* The byte minimum value with #7 bit set */
#define PEER_ENC_STOP_BYTE (1 << PEER_ENC_STOP_BIT)
/* The left most number of bits set for PEER_ENC_2BYTES_MIN */
#define PEER_ENC_2BYTES_MIN_BITS 4
#define PEER_MSG_HEADER_LEN 2
#define PEER_STKT_CACHE_MAX_ENTRIES 128
/**********************************/
/* Peer Session IO handler states */
/**********************************/
enum {
PEER_SESS_ST_ACCEPT = 0, /* Initial state for session create by an accept, must be zero! */
PEER_SESS_ST_GETVERSION, /* Validate supported protocol version */
PEER_SESS_ST_GETHOST, /* Validate host ID correspond to local host id */
PEER_SESS_ST_GETPEER, /* Validate peer ID correspond to a known remote peer id */
/* after this point, data were possibly exchanged */
PEER_SESS_ST_SENDSUCCESS, /* Send ret code 200 (success) and wait for message */
PEER_SESS_ST_CONNECT, /* Initial state for session create on a connect, push presentation into buffer */
PEER_SESS_ST_GETSTATUS, /* Wait for the welcome message */
PEER_SESS_ST_WAITMSG, /* Wait for data messages */
PEER_SESS_ST_EXIT, /* Exit with status code */
PEER_SESS_ST_ERRPROTO, /* Send error proto message before exit */
PEER_SESS_ST_ERRSIZE, /* Send error size message before exit */
PEER_SESS_ST_END, /* Killed session */
};
/***************************************************/
/* Peer Session status code - part of the protocol */
/***************************************************/
#define PEER_SESS_SC_CONNECTCODE 100 /* connect in progress */
#define PEER_SESS_SC_CONNECTEDCODE 110 /* tcp connect success */
#define PEER_SESS_SC_SUCCESSCODE 200 /* accept or connect successful */
#define PEER_SESS_SC_TRYAGAIN 300 /* try again later */
#define PEER_SESS_SC_ERRPROTO 501 /* error protocol */
#define PEER_SESS_SC_ERRVERSION 502 /* unknown protocol version */
#define PEER_SESS_SC_ERRHOST 503 /* bad host name */
#define PEER_SESS_SC_ERRPEER 504 /* unknown peer */
#define PEER_SESSION_PROTO_NAME "HAProxyS"
#define PEER_MAJOR_VER 2
#define PEER_MINOR_VER 1
#define PEER_DWNGRD_MINOR_VER 0
static size_t proto_len = sizeof(PEER_SESSION_PROTO_NAME) - 1;
struct peers *cfg_peers = NULL;
static void peer_session_forceshutdown(struct peer *peer);
static struct ebpt_node *dcache_tx_insert(struct dcache *dc,
struct dcache_tx_entry *i);
static inline void flush_dcache(struct peer *peer);
static const char *statuscode_str(int statuscode)
{
switch (statuscode) {
case PEER_SESS_SC_CONNECTCODE:
return "CONN";
case PEER_SESS_SC_CONNECTEDCODE:
return "HSHK";
case PEER_SESS_SC_SUCCESSCODE:
return "ESTA";
case PEER_SESS_SC_TRYAGAIN:
return "RETR";
case PEER_SESS_SC_ERRPROTO:
return "PROT";
case PEER_SESS_SC_ERRVERSION:
return "VERS";
case PEER_SESS_SC_ERRHOST:
return "NAME";
case PEER_SESS_SC_ERRPEER:
return "UNKN";
default:
return "NONE";
}
}
/* This function encode an uint64 to 'dynamic' length format.
The encoded value is written at address *str, and the
caller must assure that size after *str is large enough.
At return, the *str is set at the next Byte after then
encoded integer. The function returns then length of the
encoded integer in Bytes */
int intencode(uint64_t i, char **str) {
int idx = 0;
unsigned char *msg;
msg = (unsigned char *)*str;
if (i < PEER_ENC_2BYTES_MIN) {
msg[0] = (unsigned char)i;
*str = (char *)&msg[idx+1];
return (idx+1);
}
msg[idx] =(unsigned char)i | PEER_ENC_2BYTES_MIN;
i = (i - PEER_ENC_2BYTES_MIN) >> PEER_ENC_2BYTES_MIN_BITS;
while (i >= PEER_ENC_STOP_BYTE) {
msg[++idx] = (unsigned char)i | PEER_ENC_STOP_BYTE;
i = (i - PEER_ENC_STOP_BYTE) >> PEER_ENC_STOP_BIT;
}
msg[++idx] = (unsigned char)i;
*str = (char *)&msg[idx+1];
return (idx+1);
}
/* This function returns the decoded integer or 0
if decode failed
*str point on the beginning of the integer to decode
at the end of decoding *str point on the end of the
encoded integer or to null if end is reached */
uint64_t intdecode(char **str, char *end)
{
unsigned char *msg;
uint64_t i;
int shift;
if (!*str)
return 0;
msg = (unsigned char *)*str;
if (msg >= (unsigned char *)end)
goto fail;
i = *(msg++);
if (i >= PEER_ENC_2BYTES_MIN) {
shift = PEER_ENC_2BYTES_MIN_BITS;
do {
if (msg >= (unsigned char *)end)
goto fail;
i += (uint64_t)*msg << shift;
shift += PEER_ENC_STOP_BIT;
} while (*(msg++) >= PEER_ENC_STOP_BYTE);
}
*str = (char *)msg;
return i;
fail:
*str = NULL;
return 0;
}
/*
* Build a "hello" peer protocol message.
* Return the number of written bytes written to build this messages if succeeded,
* 0 if not.
*/
static int peer_prepare_hellomsg(char *msg, size_t size, struct peer_prep_params *p)
{
int min_ver, ret;
struct peer *peer;
peer = p->hello.peer;
min_ver = (peer->flags & PEER_F_DWNGRD) ? PEER_DWNGRD_MINOR_VER : PEER_MINOR_VER;
/* Prepare headers */
ret = snprintf(msg, size, PEER_SESSION_PROTO_NAME " %u.%u\n%s\n%s %d %d\n",
PEER_MAJOR_VER, min_ver, peer->id, localpeer, (int)getpid(), relative_pid);
if (ret >= size)
return 0;
return ret;
}
/*
* Build a "handshake succeeded" status message.
* Return the number of written bytes written to build this messages if succeeded,
* 0 if not.
*/
static int peer_prepare_status_successmsg(char *msg, size_t size, struct peer_prep_params *p)
{
int ret;
ret = snprintf(msg, size, "%d\n", PEER_SESS_SC_SUCCESSCODE);
if (ret >= size)
return 0;
return ret;
}
/*
* Build an error status message.
* Return the number of written bytes written to build this messages if succeeded,
* 0 if not.
*/
static int peer_prepare_status_errormsg(char *msg, size_t size, struct peer_prep_params *p)
{
int ret;
unsigned int st1;
st1 = p->error_status.st1;
ret = snprintf(msg, size, "%d\n", st1);
if (ret >= size)
return 0;
return ret;
}
/* Set the stick-table UPDATE message type byte at <msg_type> address,
* depending on <use_identifier> and <use_timed> boolean parameters.
* Always successful.
*/
static inline void peer_set_update_msg_type(char *msg_type, int use_identifier, int use_timed)
{
if (use_timed) {
if (use_identifier)
*msg_type = PEER_MSG_STKT_UPDATE_TIMED;
else
*msg_type = PEER_MSG_STKT_INCUPDATE_TIMED;
}
else {
if (use_identifier)
*msg_type = PEER_MSG_STKT_UPDATE;
else
*msg_type = PEER_MSG_STKT_INCUPDATE;
}
}
/*
* This prepare the data update message on the stick session <ts>, <st> is the considered
* stick table.
* <msg> is a buffer of <size> to receive data message content
* If function returns 0, the caller should consider we were unable to encode this message (TODO:
* check size)
*/
static int peer_prepare_updatemsg(char *msg, size_t size, struct peer_prep_params *p)
{
uint32_t netinteger;
unsigned short datalen;
char *cursor, *datamsg;
unsigned int data_type;
void *data_ptr;
struct stksess *ts;
struct shared_table *st;
unsigned int updateid;
int use_identifier;
int use_timed;
struct peer *peer;
ts = p->updt.stksess;
st = p->updt.shared_table;
updateid = p->updt.updateid;
use_identifier = p->updt.use_identifier;
use_timed = p->updt.use_timed;
peer = p->updt.peer;
cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;
/* construct message */
/* check if we need to send the update identifier */
if (!st->last_pushed || updateid < st->last_pushed || ((updateid - st->last_pushed) != 1)) {
use_identifier = 1;
}
/* encode update identifier if needed */
if (use_identifier) {
netinteger = htonl(updateid);
memcpy(cursor, &netinteger, sizeof(netinteger));
cursor += sizeof(netinteger);
}
if (use_timed) {
netinteger = htonl(tick_remain(now_ms, ts->expire));
memcpy(cursor, &netinteger, sizeof(netinteger));
cursor += sizeof(netinteger);
}
/* encode the key */
if (st->table->type == SMP_T_STR) {
int stlen = strlen((char *)ts->key.key);
intencode(stlen, &cursor);
memcpy(cursor, ts->key.key, stlen);
cursor += stlen;
}
else if (st->table->type == SMP_T_SINT) {
netinteger = htonl(read_u32(ts->key.key));
memcpy(cursor, &netinteger, sizeof(netinteger));
cursor += sizeof(netinteger);
}
else {
memcpy(cursor, ts->key.key, st->table->key_size);
cursor += st->table->key_size;
}
HA_RWLOCK_RDLOCK(STK_SESS_LOCK, &ts->lock);
/* encode values */
for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
data_ptr = stktable_data_ptr(st->table, ts, data_type);
if (data_ptr) {
switch (stktable_data_types[data_type].std_type) {
case STD_T_SINT: {
int data;
data = stktable_data_cast(data_ptr, std_t_sint);
intencode(data, &cursor);
break;
}
case STD_T_UINT: {
unsigned int data;
data = stktable_data_cast(data_ptr, std_t_uint);
intencode(data, &cursor);
break;
}
case STD_T_ULL: {
unsigned long long data;
data = stktable_data_cast(data_ptr, std_t_ull);
intencode(data, &cursor);
break;
}
case STD_T_FRQP: {
struct freq_ctr_period *frqp;
frqp = &stktable_data_cast(data_ptr, std_t_frqp);
intencode((unsigned int)(now_ms - frqp->curr_tick), &cursor);
intencode(frqp->curr_ctr, &cursor);
intencode(frqp->prev_ctr, &cursor);
break;
}
case STD_T_DICT: {
struct dict_entry *de;
struct ebpt_node *cached_de;
struct dcache_tx_entry cde = { };
char *beg, *end;
size_t value_len, data_len;
struct dcache *dc;
de = stktable_data_cast(data_ptr, std_t_dict);
if (!de) {
/* No entry */
intencode(0, &cursor);
break;
}
dc = peer->dcache;
cde.entry.key = de;
cached_de = dcache_tx_insert(dc, &cde);
if (cached_de == &cde.entry) {
if (cde.id + 1 >= PEER_ENC_2BYTES_MIN)
break;
/* Encode the length of the remaining data -> 1 */
intencode(1, &cursor);
/* Encode the cache entry ID */
intencode(cde.id + 1, &cursor);
}
else {
/* Leave enough room to encode the remaining data length. */
end = beg = cursor + PEER_MSG_ENC_LENGTH_MAXLEN;
/* Encode the dictionary entry key */
intencode(cde.id + 1, &end);
/* Encode the length of the dictionary entry data */
value_len = de->len;
intencode(value_len, &end);
/* Copy the data */
memcpy(end, de->value.key, value_len);
end += value_len;
/* Encode the length of the data */
data_len = end - beg;
intencode(data_len, &cursor);
memmove(cursor, beg, data_len);
cursor += data_len;
}
break;
}
}
}
}
HA_RWLOCK_RDUNLOCK(STK_SESS_LOCK, &ts->lock);
/* Compute datalen */
datalen = (cursor - datamsg);
/* prepare message header */
msg[0] = PEER_MSG_CLASS_STICKTABLE;
peer_set_update_msg_type(&msg[1], use_identifier, use_timed);
cursor = &msg[2];
intencode(datalen, &cursor);
/* move data after header */
memmove(cursor, datamsg, datalen);
/* return header size + data_len */
return (cursor - msg) + datalen;
}
/*
* This prepare the switch table message to targeted share table <st>.
* <msg> is a buffer of <size> to receive data message content
* If function returns 0, the caller should consider we were unable to encode this message (TODO:
* check size)
*/
static int peer_prepare_switchmsg(char *msg, size_t size, struct peer_prep_params *params)
{
int len;
unsigned short datalen;
struct buffer *chunk;
char *cursor, *datamsg, *chunkp, *chunkq;
uint64_t data = 0;
unsigned int data_type;
struct shared_table *st;
st = params->swtch.shared_table;
cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;
/* Encode data */
/* encode local id */
intencode(st->local_id, &cursor);
/* encode table name */
len = strlen(st->table->nid);
intencode(len, &cursor);
memcpy(cursor, st->table->nid, len);
cursor += len;
/* encode table type */
intencode(peer_net_key_type[st->table->type], &cursor);
/* encode table key size */
intencode(st->table->key_size, &cursor);
chunk = get_trash_chunk();
chunkp = chunkq = chunk->area;
/* encode available known data types in table */
for (data_type = 0 ; data_type < STKTABLE_DATA_TYPES ; data_type++) {
if (st->table->data_ofs[data_type]) {
switch (stktable_data_types[data_type].std_type) {
case STD_T_SINT:
case STD_T_UINT:
case STD_T_ULL:
case STD_T_DICT:
data |= 1 << data_type;
break;
case STD_T_FRQP:
data |= 1 << data_type;
intencode(data_type, &chunkq);
intencode(st->table->data_arg[data_type].u, &chunkq);
break;
}
}
}
intencode(data, &cursor);
/* Encode stick-table entries duration. */
intencode(st->table->expire, &cursor);
if (chunkq > chunkp) {
chunk->data = chunkq - chunkp;
memcpy(cursor, chunk->area, chunk->data);
cursor += chunk->data;
}
/* Compute datalen */
datalen = (cursor - datamsg);
/* prepare message header */
msg[0] = PEER_MSG_CLASS_STICKTABLE;
msg[1] = PEER_MSG_STKT_DEFINE;
cursor = &msg[2];
intencode(datalen, &cursor);
/* move data after header */
memmove(cursor, datamsg, datalen);
/* return header size + data_len */
return (cursor - msg) + datalen;
}
/*
* This prepare the acknowledge message on the stick session <ts>, <st> is the considered
* stick table.
* <msg> is a buffer of <size> to receive data message content
* If function returns 0, the caller should consider we were unable to encode this message (TODO:
* check size)
*/
static int peer_prepare_ackmsg(char *msg, size_t size, struct peer_prep_params *p)
{
unsigned short datalen;
char *cursor, *datamsg;
uint32_t netinteger;
struct shared_table *st;
cursor = datamsg = msg + PEER_MSG_HEADER_LEN + PEER_MSG_ENC_LENGTH_MAXLEN;
st = p->ack.shared_table;
intencode(st->remote_id, &cursor);
netinteger = htonl(st->last_get);
memcpy(cursor, &netinteger, sizeof(netinteger));
cursor += sizeof(netinteger);
/* Compute datalen */
datalen = (cursor - datamsg);
/* prepare message header */
msg[0] = PEER_MSG_CLASS_STICKTABLE;
msg[1] = PEER_MSG_STKT_ACK;
cursor = &msg[2];
intencode(datalen, &cursor);
/* move data after header */
memmove(cursor, datamsg, datalen);
/* return header size + data_len */
return (cursor - msg) + datalen;
}
/*
* Function to deinit connected peer
*/
void __peer_session_deinit(struct peer *peer)
{
struct stream_interface *si;
struct stream *s;
struct peers *peers;
if (!peer->appctx)
return;
si = peer->appctx->owner;
if (!si)
return;
s = si_strm(si);
if (!s)
return;
peers = strm_fe(s)->parent;
if (!peers)
return;
if (peer->appctx->st0 == PEER_SESS_ST_WAITMSG)
HA_ATOMIC_SUB(&connected_peers, 1);
HA_ATOMIC_SUB(&active_peers, 1);
flush_dcache(peer);
/* Re-init current table pointers to force announcement on re-connect */
peer->remote_table = peer->last_local_table = NULL;
peer->appctx = NULL;
if (peer->flags & PEER_F_LEARN_ASSIGN) {
/* unassign current peer for learning */
peer->flags &= ~(PEER_F_LEARN_ASSIGN);
peers->flags &= ~(PEERS_F_RESYNC_ASSIGN|PEERS_F_RESYNC_PROCESS);
/* reschedule a resync */
peers->resync_timeout = tick_add(now_ms, MS_TO_TICKS(5000));
}
/* reset teaching and learning flags to 0 */
peer->flags &= PEER_TEACH_RESET;
peer->flags &= PEER_LEARN_RESET;
/* set this peer as dead from heartbeat point of view */
peer->flags &= ~PEER_F_ALIVE;
task_wakeup(peers->sync_task, TASK_WOKEN_MSG);
}
/*
* Callback to release a session with a peer
*/
static void peer_session_release(struct appctx *appctx)
{
struct peer *peer = appctx->ctx.peers.ptr;
/* appctx->ctx.peers.ptr is not a peer session */
if (appctx->st0 < PEER_SESS_ST_SENDSUCCESS)
return;
/* peer session identified */
if (peer) {
HA_SPIN_LOCK(PEER_LOCK, &peer->lock);
if (peer->appctx == appctx)
__peer_session_deinit(peer);
HA_SPIN_UNLOCK(PEER_LOCK, &peer->lock);
}
}
/* Retrieve the major and minor versions of peers protocol
* announced by a remote peer. <str> is a null-terminated
* string with the following format: "<maj_ver>.<min_ver>".
*/
static int peer_get_version(const char *str,
unsigned int *maj_ver, unsigned int *min_ver)
{
unsigned int majv, minv;
const char *pos, *saved;
const char *end;
saved = pos = str;
end = str + strlen(str);
majv = read_uint(&pos, end);
if (saved == pos || *pos++ != '.')
return -1;
saved = pos;
minv = read_uint(&pos, end);
if (saved == pos || pos != end)
return -1;
*maj_ver = majv;
*min_ver = minv;
return 0;
}
/*
* Parse a line terminated by an optional '\r' character, followed by a mandatory
* '\n' character.
* Returns 1 if succeeded or 0 if a '\n' character could not be found, and -1 if
* a line could not be read because the communication channel is closed.
*/
static inline int peer_getline(struct appctx *appctx)
{
int n;
struct stream_interface *si = appctx->owner;
n = co_getline(si_oc(si), trash.area, trash.size);
if (!n)
return 0;
if (n < 0 || trash.area[n - 1] != '\n') {
appctx->st0 = PEER_SESS_ST_END;
return -1;
}
if (n > 1 && (trash.area[n - 2] == '\r'))
trash.area[n - 2] = 0;
else
trash.area[n - 1] = 0;
co_skip(si_oc(si), n);
return n;
}
/*
* Send a message after having called <peer_prepare_msg> to build it.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_msg(struct appctx *appctx,
int (*peer_prepare_msg)(char *, size_t, struct peer_prep_params *),
struct peer_prep_params *params)
{
int ret, msglen;
struct stream_interface *si = appctx->owner;
msglen = peer_prepare_msg(trash.area, trash.size, params);
if (!msglen) {
/* internal error: message does not fit in trash */
appctx->st0 = PEER_SESS_ST_END;
return 0;
}
/* message to buffer */
ret = ci_putblk(si_ic(si), trash.area, msglen);
if (ret <= 0) {
if (ret == -1) {
/* No more write possible */
si_rx_room_blk(si);
return -1;
}
appctx->st0 = PEER_SESS_ST_END;
}
return ret;
}
/*
* Send a hello message.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_hellomsg(struct appctx *appctx, struct peer *peer)
{
struct peer_prep_params p = {
.hello.peer = peer,
};
return peer_send_msg(appctx, peer_prepare_hellomsg, &p);
}
/*
* Send a success peer handshake status message.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_status_successmsg(struct appctx *appctx)
{
return peer_send_msg(appctx, peer_prepare_status_successmsg, NULL);
}
/*
* Send a peer handshake status error message.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_status_errormsg(struct appctx *appctx)
{
struct peer_prep_params p = {
.error_status.st1 = appctx->st1,
};
return peer_send_msg(appctx, peer_prepare_status_errormsg, &p);
}
/*
* Send a stick-table switch message.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_switchmsg(struct shared_table *st, struct appctx *appctx)
{
struct peer_prep_params p = {
.swtch.shared_table = st,
};
return peer_send_msg(appctx, peer_prepare_switchmsg, &p);
}
/*
* Send a stick-table update acknowledgement message.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_ackmsg(struct shared_table *st, struct appctx *appctx)
{
struct peer_prep_params p = {
.ack.shared_table = st,
};
return peer_send_msg(appctx, peer_prepare_ackmsg, &p);
}
/*
* Send a stick-table update message.
* Return 0 if the message could not be built modifying the appcxt st0 to PEER_SESS_ST_END value.
* Returns -1 if there was not enough room left to send the message,
* any other negative returned value must be considered as an error with an appcxt st0
* returned value equal to PEER_SESS_ST_END.
*/
static inline int peer_send_updatemsg(struct shared_table *st, struct appctx *appctx, struct stksess *ts,
unsigned int updateid, int use_identifier, int use_timed)
{
struct peer_prep_params p = {
.updt.stksess = ts,
.updt.shared_table = st,
.updt.updateid = updateid,