-
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
-
1001
-
1002
-
1003
-
1004
-
1005
-
1006
-
1007
-
1008
-
1009
-
1010
-
1011
-
1012
-
1013
-
1014
-
1015
-
1016
-
1017
-
1018
-
1019
-
1020
-
1021
-
1022
-
1023
-
1024
-
1025
-
1026
-
1027
-
1028
-
1029
-
1030
-
1031
-
1032
-
1033
-
1034
-
1035
-
1036
-
1037
-
1038
-
1039
-
1040
-
1041
-
1042
-
1043
-
1044
-
1045
-
1046
-
1047
-
1048
-
1049
-
1050
-
1051
-
1052
-
1053
-
1054
-
1055
-
1056
-
1057
-
1058
-
1059
-
1060
-
1061
-
1062
-
1063
-
1064
-
1065
-
1066
-
1067
-
1068
-
1069
-
1070
-
1071
-
1072
-
1073
-
1074
-
1075
-
1076
-
1077
-
1078
-
1079
-
1080
-
1081
-
1082
-
1083
-
1084
-
1085
-
1086
-
1087
-
1088
-
1089
-
1090
-
1091
-
1092
-
1093
-
1094
-
1095
-
1096
-
1097
-
1098
-
1099
-
1100
-
1101
-
1102
-
1103
-
1104
-
1105
-
1106
-
1107
-
1108
-
1109
-
1110
-
1111
-
1112
-
1113
-
1114
-
1115
-
1116
-
1117
-
1118
-
1119
-
1120
-
1121
-
1122
-
1123
-
1124
-
1125
-
1126
-
1127
-
1128
-
1129
-
1130
-
1131
-
1132
-
1133
-
1134
-
1135
-
1136
-
1137
-
1138
-
1139
-
1140
-
1141
-
1142
-
1143
-
1144
-
1145
-
1146
-
1147
-
1148
-
1149
-
1150
-
1151
-
1152
-
1153
-
1154
-
1155
-
1156
-
1157
-
1158
-
1159
-
1160
-
1161
-
1162
-
1163
-
1164
-
1165
-
1166
-
1167
-
1168
-
1169
-
1170
-
1171
-
1172
-
1173
-
1174
-
1175
-
1176
-
1177
-
1178
-
1179
-
1180
-
1181
-
1182
-
1183
-
1184
-
1185
-
1186
-
1187
-
1188
-
1189
-
1190
-
1191
-
1192
-
1193
-
1194
-
1195
-
1196
-
1197
-
1198
-
1199
-
1200
-
1201
-
1202
-
1203
-
1204
-
1205
-
1206
-
1207
-
1208
-
1209
-
1210
-
1211
-
1212
-
1213
-
1214
-
1215
-
1216
-
1217
-
1218
-
1219
-
1220
-
1221
-
1222
-
1223
-
1224
-
1225
-
1226
-
1227
-
1228
-
1229
-
1230
-
1231
-
1232
-
1233
-
1234
-
1235
-
1236
-
1237
-
1238
-
1239
-
1240
-
1241
-
1242
-
1243
-
1244
-
1245
-
1246
-
1247
-
1248
-
1249
-
1250
-
1251
-
1252
-
1253
-
1254
-
1255
-
1256
-
1257
-
1258
-
1259
-
1260
-
1261
-
1262
-
1263
-
1264
-
1265
-
1266
-
1267
-
1268
-
1269
-
1270
-
1271
-
1272
-
1273
-
1274
-
1275
-
1276
-
1277
-
1278
-
1279
-
1280
-
1281
-
1282
-
1283
-
1284
-
1285
-
1286
-
1287
-
1288
-
1289
-
1290
-
1291
-
1292
-
1293
-
1294
-
1295
-
1296
-
1297
-
1298
-
1299
-
1300
-
1301
-
1302
-
1303
-
1304
-
1305
-
1306
-
1307
-
1308
-
1309
-
1310
-
1311
-
1312
-
1313
-
1314
-
1315
-
1316
-
1317
-
1318
-
1319
-
1320
-
1321
-
1322
-
1323
-
1324
-
1325
-
1326
-
1327
-
1328
-
1329
-
1330
-
1331
-
1332
-
1333
-
1334
-
1335
-
1336
-
1337
-
1338
-
1339
-
1340
-
1341
-
1342
-
1343
-
1344
-
1345
-
1346
-
1347
-
1348
-
1349
-
1350
-
1351
-
1352
-
1353
-
1354
-
1355
-
1356
-
1357
-
1358
-
1359
-
1360
-
1361
-
1362
-
1363
-
1364
-
1365
-
1366
-
1367
-
1368
-
1369
-
1370
-
1371
-
1372
-
1373
-
1374
-
1375
-
1376
-
1377
-
1378
-
1379
-
1380
-
1381
-
1382
-
1383
-
1384
-
1385
-
1386
-
1387
-
1388
-
1389
-
1390
-
1391
-
1392
-
1393
-
1394
-
1395
-
1396
-
1397
-
1398
-
1399
-
1400
-
1401
-
1402
-
1403
-
1404
-
1405
-
1406
-
1407
-
1408
-
1409
-
1410
-
1411
-
1412
-
1413
-
1414
-
1415
-
1416
-
1417
-
1418
-
1419
-
1420
-
1421
-
1422
-
1423
-
1424
-
1425
-
1426
-
1427
-
1428
-
1429
-
1430
-
1431
-
1432
-
1433
-
1434
-
1435
-
1436
-
1437
-
1438
-
1439
-
1440
-
1441
-
1442
-
1443
-
1444
-
1445
-
1446
-
1447
-
1448
-
1449
-
1450
-
1451
-
1452
-
1453
-
1454
-
1455
-
1456
-
1457
-
1458
-
1459
-
1460
-
1461
-
1462
-
1463
-
1464
-
1465
-
1466
-
1467
-
1468
-
1469
-
1470
-
1471
-
1472
-
1473
-
1474
-
1475
-
1476
-
1477
-
1478
-
1479
-
1480
-
1481
-
1482
-
1483
-
1484
-
1485
-
1486
-
1487
-
1488
-
1489
-
1490
-
1491
-
1492
-
1493
-
1494
-
1495
-
1496
-
1497
-
1498
-
1499
-
1500
-
1501
-
1502
-
1503
-
1504
-
1505
-
1506
-
1507
-
1508
-
1509
-
1510
-
1511
-
1512
-
1513
-
1514
-
1515
-
1516
-
1517
-
1518
-
1519
-
1520
-
1521
-
1522
-
1523
-
1524
-
1525
-
1526
-
1527
-
1528
-
1529
-
1530
-
1531
-
1532
-
1533
-
1534
-
1535
-
1536
-
1537
-
1538
-
1539
-
1540
-
1541
-
1542
-
1543
-
1544
-
1545
-
1546
-
1547
-
1548
-
1549
-
1550
-
1551
-
1552
-
1553
-
1554
-
1555
-
1556
-
1557
-
1558
-
1559
-
1560
-
1561
-
1562
-
1563
-
1564
-
1565
-
1566
-
1567
-
1568
-
1569
-
1570
-
1571
-
1572
-
1573
-
1574
-
1575
-
1576
-
1577
-
1578
-
1579
-
1580
-
1581
-
1582
-
1583
-
1584
-
1585
-
1586
-
1587
-
1588
-
1589
-
1590
-
1591
-
1592
-
1593
-
1594
-
1595
-
1596
-
1597
-
1598
-
1599
-
1600
-
1601
-
1602
-
1603
-
1604
-
1605
-
1606
-
1607
-
1608
-
1609
-
1610
-
1611
-
1612
-
1613
-
1614
-
1615
-
1616
-
1617
-
1618
-
1619
-
1620
-
1621
-
1622
-
1623
-
1624
-
1625
-
1626
-
1627
-
1628
-
1629
-
1630
-
1631
-
1632
-
1633
-
1634
-
1635
-
1636
-
1637
-
1638
-
1639
-
1640
-
1641
-
1642
-
1643
-
1644
-
1645
-
1646
-
1647
-
1648
-
1649
-
1650
-
1651
-
1652
-
1653
-
1654
-
1655
-
1656
-
1657
-
1658
-
1659
-
1660
-
1661
-
1662
-
1663
-
1664
-
1665
-
1666
-
1667
-
1668
-
1669
-
1670
-
1671
-
1672
-
1673
-
1674
-
1675
-
1676
-
1677
-
1678
-
1679
-
1680
-
1681
-
1682
-
1683
-
1684
-
1685
-
1686
-
1687
-
1688
-
1689
-
1690
-
1691
-
1692
-
1693
-
1694
-
1695
-
1696
-
1697
-
1698
-
1699
-
1700
-
1701
-
1702
-
1703
-
1704
-
1705
-
1706
-
1707
-
1708
-
1709
-
1710
-
1711
-
1712
-
1713
-
1714
-
1715
-
1716
-
1717
-
1718
-
1719
-
1720
-
1721
-
1722
-
1723
-
1724
-
1725
-
1726
-
1727
-
1728
-
1729
-
1730
-
1731
-
1732
-
1733
-
1734
-
1735
-
1736
package redis
import (
"context"
"crypto/tls"
"fmt"
"math"
"net"
"runtime"
"sort"
"sync"
"sync/atomic"
"time"
"github.com/go-redis/redis/v8/internal"
"github.com/go-redis/redis/v8/internal/hashtag"
"github.com/go-redis/redis/v8/internal/pool"
"github.com/go-redis/redis/v8/internal/proto"
"github.com/go-redis/redis/v8/internal/rand"
)
var errClusterNoNodes = fmt.Errorf("redis: cluster has no nodes")
// ClusterOptions are used to configure a cluster client and should be
// passed to NewClusterClient.
type ClusterOptions struct {
// A seed list of host:port addresses of cluster nodes.
Addrs []string
// NewClient creates a cluster node client with provided name and options.
NewClient func(opt *Options) *Client
// The maximum number of retries before giving up. Command is retried
// on network errors and MOVED/ASK redirects.
// Default is 3 retries.
MaxRedirects int
// Enables read-only commands on slave nodes.
ReadOnly bool
// Allows routing read-only commands to the closest master or slave node.
// It automatically enables ReadOnly.
RouteByLatency bool
// Allows routing read-only commands to the random master or slave node.
// It automatically enables ReadOnly.
RouteRandomly bool
// Optional function that returns cluster slots information.
// It is useful to manually create cluster of standalone Redis servers
// and load-balance read/write operations between master and slaves.
// It can use service like ZooKeeper to maintain configuration information
// and Cluster.ReloadState to manually trigger state reloading.
ClusterSlots func(context.Context) ([]ClusterSlot, error)
// Following options are copied from Options struct.
Dialer func(ctx context.Context, network, addr string) (net.Conn, error)
OnConnect func(ctx context.Context, cn *Conn) error
Username string
Password string
MaxRetries int
MinRetryBackoff time.Duration
MaxRetryBackoff time.Duration
DialTimeout time.Duration
ReadTimeout time.Duration
WriteTimeout time.Duration
// PoolSize applies per cluster node and not for the whole cluster.
PoolSize int
MinIdleConns int
MaxConnAge time.Duration
PoolTimeout time.Duration
IdleTimeout time.Duration
IdleCheckFrequency time.Duration
TLSConfig *tls.Config
}
func (opt *ClusterOptions) init() {
if opt.MaxRedirects == -1 {
opt.MaxRedirects = 0
} else if opt.MaxRedirects == 0 {
opt.MaxRedirects = 3
}
if opt.RouteByLatency || opt.RouteRandomly {
opt.ReadOnly = true
}
if opt.PoolSize == 0 {
opt.PoolSize = 5 * runtime.GOMAXPROCS(0)
}
switch opt.ReadTimeout {
case -1:
opt.ReadTimeout = 0
case 0:
opt.ReadTimeout = 3 * time.Second
}
switch opt.WriteTimeout {
case -1:
opt.WriteTimeout = 0
case 0:
opt.WriteTimeout = opt.ReadTimeout
}
if opt.MaxRetries == 0 {
opt.MaxRetries = -1
}
switch opt.MinRetryBackoff {
case -1:
opt.MinRetryBackoff = 0
case 0:
opt.MinRetryBackoff = 8 * time.Millisecond
}
switch opt.MaxRetryBackoff {
case -1:
opt.MaxRetryBackoff = 0
case 0:
opt.MaxRetryBackoff = 512 * time.Millisecond
}
if opt.NewClient == nil {
opt.NewClient = NewClient
}
}
func (opt *ClusterOptions) clientOptions() *Options {
const disableIdleCheck = -1
return &Options{
Dialer: opt.Dialer,
OnConnect: opt.OnConnect,
Username: opt.Username,
Password: opt.Password,
MaxRetries: opt.MaxRetries,
MinRetryBackoff: opt.MinRetryBackoff,
MaxRetryBackoff: opt.MaxRetryBackoff,
DialTimeout: opt.DialTimeout,
ReadTimeout: opt.ReadTimeout,
WriteTimeout: opt.WriteTimeout,
PoolSize: opt.PoolSize,
MinIdleConns: opt.MinIdleConns,
MaxConnAge: opt.MaxConnAge,
PoolTimeout: opt.PoolTimeout,
IdleTimeout: opt.IdleTimeout,
IdleCheckFrequency: disableIdleCheck,
TLSConfig: opt.TLSConfig,
// If ClusterSlots is populated, then we probably have an artificial
// cluster whose nodes are not in clustering mode (otherwise there isn't
// much use for ClusterSlots config). This means we cannot execute the
// READONLY command against that node -- setting readOnly to false in such
// situations in the options below will prevent that from happening.
readOnly: opt.ReadOnly && opt.ClusterSlots == nil,
}
}
//------------------------------------------------------------------------------
type clusterNode struct {
Client *Client
latency uint32 // atomic
generation uint32 // atomic
failing uint32 // atomic
}
func newClusterNode(clOpt *ClusterOptions, addr string) *clusterNode {
opt := clOpt.clientOptions()
opt.Addr = addr
node := clusterNode{
Client: clOpt.NewClient(opt),
}
node.latency = math.MaxUint32
if clOpt.RouteByLatency {
go node.updateLatency()
}
return &node
}
func (n *clusterNode) String() string {
return n.Client.String()
}
func (n *clusterNode) Close() error {
return n.Client.Close()
}
func (n *clusterNode) updateLatency() {
const numProbe = 10
var dur uint64
for i := 0; i < numProbe; i++ {
time.Sleep(time.Duration(10+rand.Intn(10)) * time.Millisecond)
start := time.Now()
n.Client.Ping(context.TODO())
dur += uint64(time.Since(start) / time.Microsecond)
}
latency := float64(dur) / float64(numProbe)
atomic.StoreUint32(&n.latency, uint32(latency+0.5))
}
func (n *clusterNode) Latency() time.Duration {
latency := atomic.LoadUint32(&n.latency)
return time.Duration(latency) * time.Microsecond
}
func (n *clusterNode) MarkAsFailing() {
atomic.StoreUint32(&n.failing, uint32(time.Now().Unix()))
}
func (n *clusterNode) Failing() bool {
const timeout = 15 // 15 seconds
failing := atomic.LoadUint32(&n.failing)
if failing == 0 {
return false
}
if time.Now().Unix()-int64(failing) < timeout {
return true
}
atomic.StoreUint32(&n.failing, 0)
return false
}
func (n *clusterNode) Generation() uint32 {
return atomic.LoadUint32(&n.generation)
}
func (n *clusterNode) SetGeneration(gen uint32) {
for {
v := atomic.LoadUint32(&n.generation)
if gen < v || atomic.CompareAndSwapUint32(&n.generation, v, gen) {
break
}
}
}
//------------------------------------------------------------------------------
type clusterNodes struct {
opt *ClusterOptions
mu sync.RWMutex
addrs []string
nodes map[string]*clusterNode
activeAddrs []string
closed bool
_generation uint32 // atomic
}
func newClusterNodes(opt *ClusterOptions) *clusterNodes {
return &clusterNodes{
opt: opt,
addrs: opt.Addrs,
nodes: make(map[string]*clusterNode),
}
}
func (c *clusterNodes) Close() error {
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return nil
}
c.closed = true
var firstErr error
for _, node := range c.nodes {
if err := node.Client.Close(); err != nil && firstErr == nil {
firstErr = err
}
}
c.nodes = nil
c.activeAddrs = nil
return firstErr
}
func (c *clusterNodes) Addrs() ([]string, error) {
var addrs []string
c.mu.RLock()
closed := c.closed //nolint:ifshort
if !closed {
if len(c.activeAddrs) > 0 {
addrs = c.activeAddrs
} else {
addrs = c.addrs
}
}
c.mu.RUnlock()
if closed {
return nil, pool.ErrClosed
}
if len(addrs) == 0 {
return nil, errClusterNoNodes
}
return addrs, nil
}
func (c *clusterNodes) NextGeneration() uint32 {
return atomic.AddUint32(&c._generation, 1)
}
// GC removes unused nodes.
func (c *clusterNodes) GC(generation uint32) {
//nolint:prealloc
var collected []*clusterNode
c.mu.Lock()
c.activeAddrs = c.activeAddrs[:0]
for addr, node := range c.nodes {
if node.Generation() >= generation {
c.activeAddrs = append(c.activeAddrs, addr)
if c.opt.RouteByLatency {
go node.updateLatency()
}
continue
}
delete(c.nodes, addr)
collected = append(collected, node)
}
c.mu.Unlock()
for _, node := range collected {
_ = node.Client.Close()
}
}
func (c *clusterNodes) Get(addr string) (*clusterNode, error) {
node, err := c.get(addr)
if err != nil {
return nil, err
}
if node != nil {
return node, nil
}
c.mu.Lock()
defer c.mu.Unlock()
if c.closed {
return nil, pool.ErrClosed
}
node, ok := c.nodes[addr]
if ok {
return node, nil
}
node = newClusterNode(c.opt, addr)
c.addrs = appendIfNotExists(c.addrs, addr)
c.nodes[addr] = node
return node, nil
}
func (c *clusterNodes) get(addr string) (*clusterNode, error) {
var node *clusterNode
var err error
c.mu.RLock()
if c.closed {
err = pool.ErrClosed
} else {
node = c.nodes[addr]
}
c.mu.RUnlock()
return node, err
}
func (c *clusterNodes) All() ([]*clusterNode, error) {
c.mu.RLock()
defer c.mu.RUnlock()
if c.closed {
return nil, pool.ErrClosed
}
cp := make([]*clusterNode, 0, len(c.nodes))
for _, node := range c.nodes {
cp = append(cp, node)
}
return cp, nil
}
func (c *clusterNodes) Random() (*clusterNode, error) {
addrs, err := c.Addrs()
if err != nil {
return nil, err
}
n := rand.Intn(len(addrs))
return c.Get(addrs[n])
}
//------------------------------------------------------------------------------
type clusterSlot struct {
start, end int
nodes []*clusterNode
}
type clusterSlotSlice []*clusterSlot
func (p clusterSlotSlice) Len() int {
return len(p)
}
func (p clusterSlotSlice) Less(i, j int) bool {
return p[i].start < p[j].start
}
func (p clusterSlotSlice) Swap(i, j int) {
p[i], p[j] = p[j], p[i]
}
type clusterState struct {
nodes *clusterNodes
Masters []*clusterNode
Slaves []*clusterNode
slots []*clusterSlot
generation uint32
createdAt time.Time
}
func newClusterState(
nodes *clusterNodes, slots []ClusterSlot, origin string,
) (*clusterState, error) {
c := clusterState{
nodes: nodes,
slots: make([]*clusterSlot, 0, len(slots)),
generation: nodes.NextGeneration(),
createdAt: time.Now(),
}
originHost, _, _ := net.SplitHostPort(origin)
isLoopbackOrigin := isLoopback(originHost)
for _, slot := range slots {
var nodes []*clusterNode
for i, slotNode := range slot.Nodes {
addr := slotNode.Addr
if !isLoopbackOrigin {
addr = replaceLoopbackHost(addr, originHost)
}
node, err := c.nodes.Get(addr)
if err != nil {
return nil, err
}
node.SetGeneration(c.generation)
nodes = append(nodes, node)
if i == 0 {
c.Masters = appendUniqueNode(c.Masters, node)
} else {
c.Slaves = appendUniqueNode(c.Slaves, node)
}
}
c.slots = append(c.slots, &clusterSlot{
start: slot.Start,
end: slot.End,
nodes: nodes,
})
}
sort.Sort(clusterSlotSlice(c.slots))
time.AfterFunc(time.Minute, func() {
nodes.GC(c.generation)
})
return &c, nil
}
func replaceLoopbackHost(nodeAddr, originHost string) string {
nodeHost, nodePort, err := net.SplitHostPort(nodeAddr)
if err != nil {
return nodeAddr
}
nodeIP := net.ParseIP(nodeHost)
if nodeIP == nil {
return nodeAddr
}
if !nodeIP.IsLoopback() {
return nodeAddr
}
// Use origin host which is not loopback and node port.
return net.JoinHostPort(originHost, nodePort)
}
func isLoopback(host string) bool {
ip := net.ParseIP(host)
if ip == nil {
return true
}
return ip.IsLoopback()
}
func (c *clusterState) slotMasterNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) > 0 {
return nodes[0], nil
}
return c.nodes.Random()
}
func (c *clusterState) slotSlaveNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
switch len(nodes) {
case 0:
return c.nodes.Random()
case 1:
return nodes[0], nil
case 2:
if slave := nodes[1]; !slave.Failing() {
return slave, nil
}
return nodes[0], nil
default:
var slave *clusterNode
for i := 0; i < 10; i++ {
n := rand.Intn(len(nodes)-1) + 1
slave = nodes[n]
if !slave.Failing() {
return slave, nil
}
}
// All slaves are loading - use master.
return nodes[0], nil
}
}
func (c *clusterState) slotClosestNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) == 0 {
return c.nodes.Random()
}
var node *clusterNode
for _, n := range nodes {
if n.Failing() {
continue
}
if node == nil || n.Latency() < node.Latency() {
node = n
}
}
if node != nil {
return node, nil
}
// If all nodes are failing - return random node
return c.nodes.Random()
}
func (c *clusterState) slotRandomNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)
if len(nodes) == 0 {
return c.nodes.Random()
}
n := rand.Intn(len(nodes))
return nodes[n], nil
}
func (c *clusterState) slotNodes(slot int) []*clusterNode {
i := sort.Search(len(c.slots), func(i int) bool {
return c.slots[i].end >= slot
})
if i >= len(c.slots) {
return nil
}
x := c.slots[i]
if slot >= x.start && slot <= x.end {
return x.nodes
}
return nil
}
//------------------------------------------------------------------------------
type clusterStateHolder struct {
load func(ctx context.Context) (*clusterState, error)
state atomic.Value
reloading uint32 // atomic
}
func newClusterStateHolder(fn func(ctx context.Context) (*clusterState, error)) *clusterStateHolder {
return &clusterStateHolder{
load: fn,
}
}
func (c *clusterStateHolder) Reload(ctx context.Context) (*clusterState, error) {
state, err := c.load(ctx)
if err != nil {
return nil, err
}
c.state.Store(state)
return state, nil
}
func (c *clusterStateHolder) LazyReload() {
if !atomic.CompareAndSwapUint32(&c.reloading, 0, 1) {
return
}
go func() {
defer atomic.StoreUint32(&c.reloading, 0)
_, err := c.Reload(context.Background())
if err != nil {
return
}
time.Sleep(200 * time.Millisecond)
}()
}
func (c *clusterStateHolder) Get(ctx context.Context) (*clusterState, error) {
v := c.state.Load()
if v == nil {
return c.Reload(ctx)
}
state := v.(*clusterState)
if time.Since(state.createdAt) > 10*time.Second {
c.LazyReload()
}
return state, nil
}
func (c *clusterStateHolder) ReloadOrGet(ctx context.Context) (*clusterState, error) {
state, err := c.Reload(ctx)
if err == nil {
return state, nil
}
return c.Get(ctx)
}
//------------------------------------------------------------------------------
type clusterClient struct {
opt *ClusterOptions
nodes *clusterNodes
state *clusterStateHolder //nolint:structcheck
cmdsInfoCache *cmdsInfoCache //nolint:structcheck
}
// ClusterClient is a Redis Cluster client representing a pool of zero
// or more underlying connections. It's safe for concurrent use by
// multiple goroutines.
type ClusterClient struct {
*clusterClient
cmdable
hooks
ctx context.Context
}
// NewClusterClient returns a Redis Cluster client as described in
// http://redis.io/topics/cluster-spec.
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
opt.init()
c := &ClusterClient{
clusterClient: &clusterClient{
opt: opt,
nodes: newClusterNodes(opt),
},
ctx: context.Background(),
}
c.state = newClusterStateHolder(c.loadState)
c.cmdsInfoCache = newCmdsInfoCache(c.cmdsInfo)
c.cmdable = c.Process
if opt.IdleCheckFrequency > 0 {
go c.reaper(opt.IdleCheckFrequency)
}
return c
}
func (c *ClusterClient) Context() context.Context {
return c.ctx
}
func (c *ClusterClient) WithContext(ctx context.Context) *ClusterClient {
if ctx == nil {
panic("nil context")
}
clone := *c
clone.cmdable = clone.Process
clone.hooks.lock()
clone.ctx = ctx
return &clone
}
// Options returns read-only Options that were used to create the client.
func (c *ClusterClient) Options() *ClusterOptions {
return c.opt
}
// ReloadState reloads cluster state. If available it calls ClusterSlots func
// to get cluster slots information.
func (c *ClusterClient) ReloadState(ctx context.Context) {
c.state.LazyReload()
}
// Close closes the cluster client, releasing any open resources.
//
// It is rare to Close a ClusterClient, as the ClusterClient is meant
// to be long-lived and shared between many goroutines.
func (c *ClusterClient) Close() error {
return c.nodes.Close()
}
// Do creates a Cmd from the args and processes the cmd.
func (c *ClusterClient) Do(ctx context.Context, args ...interface{}) *Cmd {
cmd := NewCmd(ctx, args...)
_ = c.Process(ctx, cmd)
return cmd
}
func (c *ClusterClient) Process(ctx context.Context, cmd Cmder) error {
return c.hooks.process(ctx, cmd, c.process)
}
func (c *ClusterClient) process(ctx context.Context, cmd Cmder) error {
cmdInfo := c.cmdInfo(cmd.Name())
slot := c.cmdSlot(cmd)
var node *clusterNode
var ask bool
var lastErr error
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
return err
}
}
if node == nil {
var err error
node, err = c.cmdNode(ctx, cmdInfo, slot)
if err != nil {
return err
}
}
if ask {
pipe := node.Client.Pipeline()
_ = pipe.Process(ctx, NewCmd(ctx, "asking"))
_ = pipe.Process(ctx, cmd)
_, lastErr = pipe.Exec(ctx)
_ = pipe.Close()
ask = false
} else {
lastErr = node.Client.Process(ctx, cmd)
}
// If there is no error - we are done.
if lastErr == nil {
return nil
}
if isReadOnly := isReadOnlyError(lastErr); isReadOnly || lastErr == pool.ErrClosed {
if isReadOnly {
c.state.LazyReload()
}
node = nil
continue
}
// If slave is loading - pick another node.
if c.opt.ReadOnly && isLoadingError(lastErr) {
node.MarkAsFailing()
node = nil
continue
}
var moved bool
var addr string
moved, ask, addr = isMovedError(lastErr)
if moved || ask {
var err error
node, err = c.nodes.Get(addr)
if err != nil {
return err
}
continue
}
if shouldRetry(lastErr, cmd.readTimeout() == nil) {
// First retry the same node.
if attempt == 0 {
continue
}
// Second try another node.
node.MarkAsFailing()
node = nil
continue
}
return lastErr
}
return lastErr
}
// ForEachMaster concurrently calls the fn on each master node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachMaster(
ctx context.Context,
fn func(ctx context.Context, client *Client) error,
) error {
state, err := c.state.ReloadOrGet(ctx)
if err != nil {
return err
}
var wg sync.WaitGroup
errCh := make(chan error, 1)
for _, master := range state.Masters {
wg.Add(1)
go func(node *clusterNode) {
defer wg.Done()
err := fn(ctx, node.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
}(master)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
// ForEachSlave concurrently calls the fn on each slave node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachSlave(
ctx context.Context,
fn func(ctx context.Context, client *Client) error,
) error {
state, err := c.state.ReloadOrGet(ctx)
if err != nil {
return err
}
var wg sync.WaitGroup
errCh := make(chan error, 1)
for _, slave := range state.Slaves {
wg.Add(1)
go func(node *clusterNode) {
defer wg.Done()
err := fn(ctx, node.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
}(slave)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
// ForEachShard concurrently calls the fn on each known node in the cluster.
// It returns the first error if any.
func (c *ClusterClient) ForEachShard(
ctx context.Context,
fn func(ctx context.Context, client *Client) error,
) error {
state, err := c.state.ReloadOrGet(ctx)
if err != nil {
return err
}
var wg sync.WaitGroup
errCh := make(chan error, 1)
worker := func(node *clusterNode) {
defer wg.Done()
err := fn(ctx, node.Client)
if err != nil {
select {
case errCh <- err:
default:
}
}
}
for _, node := range state.Masters {
wg.Add(1)
go worker(node)
}
for _, node := range state.Slaves {
wg.Add(1)
go worker(node)
}
wg.Wait()
select {
case err := <-errCh:
return err
default:
return nil
}
}
// PoolStats returns accumulated connection pool stats.
func (c *ClusterClient) PoolStats() *PoolStats {
var acc PoolStats
state, _ := c.state.Get(context.TODO())
if state == nil {
return &acc
}
for _, node := range state.Masters {
s := node.Client.connPool.Stats()
acc.Hits += s.Hits
acc.Misses += s.Misses
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.IdleConns += s.IdleConns
acc.StaleConns += s.StaleConns
}
for _, node := range state.Slaves {
s := node.Client.connPool.Stats()
acc.Hits += s.Hits
acc.Misses += s.Misses
acc.Timeouts += s.Timeouts
acc.TotalConns += s.TotalConns
acc.IdleConns += s.IdleConns
acc.StaleConns += s.StaleConns
}
return &acc
}
func (c *ClusterClient) loadState(ctx context.Context) (*clusterState, error) {
if c.opt.ClusterSlots != nil {
slots, err := c.opt.ClusterSlots(ctx)
if err != nil {
return nil, err
}
return newClusterState(c.nodes, slots, "")
}
addrs, err := c.nodes.Addrs()
if err != nil {
return nil, err
}
var firstErr error
for _, idx := range rand.Perm(len(addrs)) {
addr := addrs[idx]
node, err := c.nodes.Get(addr)
if err != nil {
if firstErr == nil {
firstErr = err
}
continue
}
slots, err := node.Client.ClusterSlots(ctx).Result()
if err != nil {
if firstErr == nil {
firstErr = err
}
continue
}
return newClusterState(c.nodes, slots, node.Client.opt.Addr)
}
/*
* No node is connectable. It's possible that all nodes' IP has changed.
* Clear activeAddrs to let client be able to re-connect using the initial
* setting of the addresses (e.g. [redis-cluster-0:6379, redis-cluster-1:6379]),
* which might have chance to resolve domain name and get updated IP address.
*/
c.nodes.mu.Lock()
c.nodes.activeAddrs = nil
c.nodes.mu.Unlock()
return nil, firstErr
}
// reaper closes idle connections to the cluster.
func (c *ClusterClient) reaper(idleCheckFrequency time.Duration) {
ticker := time.NewTicker(idleCheckFrequency)
defer ticker.Stop()
for range ticker.C {
nodes, err := c.nodes.All()
if err != nil {
break
}
for _, node := range nodes {
_, err := node.Client.connPool.(*pool.ConnPool).ReapStaleConns()
if err != nil {
internal.Logger.Printf(c.Context(), "ReapStaleConns failed: %s", err)
}
}
}
}
func (c *ClusterClient) Pipeline() Pipeliner {
pipe := Pipeline{
ctx: c.ctx,
exec: c.processPipeline,
}
pipe.init()
return &pipe
}
func (c *ClusterClient) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.Pipeline().Pipelined(ctx, fn)
}
func (c *ClusterClient) processPipeline(ctx context.Context, cmds []Cmder) error {
return c.hooks.processPipeline(ctx, cmds, c._processPipeline)
}
func (c *ClusterClient) _processPipeline(ctx context.Context, cmds []Cmder) error {
cmdsMap := newCmdsMap()
err := c.mapCmdsByNode(ctx, cmdsMap, cmds)
if err != nil {
setCmdsErr(cmds, err)
return err
}
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
setCmdsErr(cmds, err)
return err
}
}
failedCmds := newCmdsMap()
var wg sync.WaitGroup
for node, cmds := range cmdsMap.m {
wg.Add(1)
go func(node *clusterNode, cmds []Cmder) {
defer wg.Done()
err := c._processPipelineNode(ctx, node, cmds, failedCmds)
if err == nil {
return
}
if attempt < c.opt.MaxRedirects {
if err := c.mapCmdsByNode(ctx, failedCmds, cmds); err != nil {
setCmdsErr(cmds, err)
}
} else {
setCmdsErr(cmds, err)
}
}(node, cmds)
}
wg.Wait()
if len(failedCmds.m) == 0 {
break
}
cmdsMap = failedCmds
}
return cmdsFirstErr(cmds)
}
func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmds []Cmder) error {
state, err := c.state.Get(ctx)
if err != nil {
return err
}
if c.opt.ReadOnly && c.cmdsAreReadOnly(cmds) {
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
node, err := c.slotReadOnlyNode(state, slot)
if err != nil {
return err
}
cmdsMap.Add(node, cmd)
}
return nil
}
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
node, err := state.slotMasterNode(slot)
if err != nil {
return err
}
cmdsMap.Add(node, cmd)
}
return nil
}
func (c *ClusterClient) cmdsAreReadOnly(cmds []Cmder) bool {
for _, cmd := range cmds {
cmdInfo := c.cmdInfo(cmd.Name())
if cmdInfo == nil || !cmdInfo.ReadOnly {
return false
}
}
return true
}
func (c *ClusterClient) _processPipelineNode(
ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
) error {
return node.Client.hooks.processPipeline(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
return node.Client.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmds(wr, cmds)
})
if err != nil {
return err
}
return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
return c.pipelineReadCmds(ctx, node, rd, cmds, failedCmds)
})
})
})
}
func (c *ClusterClient) pipelineReadCmds(
ctx context.Context,
node *clusterNode,
rd *proto.Reader,
cmds []Cmder,
failedCmds *cmdsMap,
) error {
for _, cmd := range cmds {
err := cmd.readReply(rd)
cmd.SetErr(err)
if err == nil {
continue
}
if c.checkMovedErr(ctx, cmd, err, failedCmds) {
continue
}
if c.opt.ReadOnly && isLoadingError(err) {
node.MarkAsFailing()
return err
}
if isRedisError(err) {
continue
}
return err
}
return nil
}
func (c *ClusterClient) checkMovedErr(
ctx context.Context, cmd Cmder, err error, failedCmds *cmdsMap,
) bool {
moved, ask, addr := isMovedError(err)
if !moved && !ask {
return false
}
node, err := c.nodes.Get(addr)
if err != nil {
return false
}
if moved {
c.state.LazyReload()
failedCmds.Add(node, cmd)
return true
}
if ask {
failedCmds.Add(node, NewCmd(ctx, "asking"), cmd)
return true
}
panic("not reached")
}
// TxPipeline acts like Pipeline, but wraps queued commands with MULTI/EXEC.
func (c *ClusterClient) TxPipeline() Pipeliner {
pipe := Pipeline{
ctx: c.ctx,
exec: c.processTxPipeline,
}
pipe.init()
return &pipe
}
func (c *ClusterClient) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error) {
return c.TxPipeline().Pipelined(ctx, fn)
}
func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) error {
return c.hooks.processTxPipeline(ctx, cmds, c._processTxPipeline)
}
func (c *ClusterClient) _processTxPipeline(ctx context.Context, cmds []Cmder) error {
// Trim multi .. exec.
cmds = cmds[1 : len(cmds)-1]
state, err := c.state.Get(ctx)
if err != nil {
setCmdsErr(cmds, err)
return err
}
cmdsMap := c.mapCmdsBySlot(cmds)
for slot, cmds := range cmdsMap {
node, err := state.slotMasterNode(slot)
if err != nil {
setCmdsErr(cmds, err)
continue
}
cmdsMap := map[*clusterNode][]Cmder{node: cmds}
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
setCmdsErr(cmds, err)
return err
}
}
failedCmds := newCmdsMap()
var wg sync.WaitGroup
for node, cmds := range cmdsMap {
wg.Add(1)
go func(node *clusterNode, cmds []Cmder) {
defer wg.Done()
err := c._processTxPipelineNode(ctx, node, cmds, failedCmds)
if err == nil {
return
}
if attempt < c.opt.MaxRedirects {
if err := c.mapCmdsByNode(ctx, failedCmds, cmds); err != nil {
setCmdsErr(cmds, err)
}
} else {
setCmdsErr(cmds, err)
}
}(node, cmds)
}
wg.Wait()
if len(failedCmds.m) == 0 {
break
}
cmdsMap = failedCmds.m
}
}
return cmdsFirstErr(cmds)
}
func (c *ClusterClient) mapCmdsBySlot(cmds []Cmder) map[int][]Cmder {
cmdsMap := make(map[int][]Cmder)
for _, cmd := range cmds {
slot := c.cmdSlot(cmd)
cmdsMap[slot] = append(cmdsMap[slot], cmd)
}
return cmdsMap
}
func (c *ClusterClient) _processTxPipelineNode(
ctx context.Context, node *clusterNode, cmds []Cmder, failedCmds *cmdsMap,
) error {
return node.Client.hooks.processTxPipeline(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
return node.Client.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
err := cn.WithWriter(ctx, c.opt.WriteTimeout, func(wr *proto.Writer) error {
return writeCmds(wr, cmds)
})
if err != nil {
return err
}
return cn.WithReader(ctx, c.opt.ReadTimeout, func(rd *proto.Reader) error {
statusCmd := cmds[0].(*StatusCmd)
// Trim multi and exec.
cmds = cmds[1 : len(cmds)-1]
err := c.txPipelineReadQueued(ctx, rd, statusCmd, cmds, failedCmds)
if err != nil {
moved, ask, addr := isMovedError(err)
if moved || ask {
return c.cmdsMoved(ctx, cmds, moved, ask, addr, failedCmds)
}
return err
}
return pipelineReadCmds(rd, cmds)
})
})
})
}
func (c *ClusterClient) txPipelineReadQueued(
ctx context.Context,
rd *proto.Reader,
statusCmd *StatusCmd,
cmds []Cmder,
failedCmds *cmdsMap,
) error {
// Parse queued replies.
if err := statusCmd.readReply(rd); err != nil {
return err
}
for _, cmd := range cmds {
err := statusCmd.readReply(rd)
if err == nil || c.checkMovedErr(ctx, cmd, err, failedCmds) || isRedisError(err) {
continue
}
return err
}
// Parse number of replies.
line, err := rd.ReadLine()
if err != nil {
if err == Nil {
err = TxFailedErr
}
return err
}
switch line[0] {
case proto.ErrorReply:
return proto.ParseErrorReply(line)
case proto.ArrayReply:
// ok
default:
return fmt.Errorf("redis: expected '*', but got line %q", line)
}
return nil
}
func (c *ClusterClient) cmdsMoved(
ctx context.Context, cmds []Cmder,
moved, ask bool,
addr string,
failedCmds *cmdsMap,
) error {
node, err := c.nodes.Get(addr)
if err != nil {
return err
}
if moved {
c.state.LazyReload()
for _, cmd := range cmds {
failedCmds.Add(node, cmd)
}
return nil
}
if ask {
for _, cmd := range cmds {
failedCmds.Add(node, NewCmd(ctx, "asking"), cmd)
}
return nil
}
return nil
}
func (c *ClusterClient) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
if len(keys) == 0 {
return fmt.Errorf("redis: Watch requires at least one key")
}
slot := hashtag.Slot(keys[0])
for _, key := range keys[1:] {
if hashtag.Slot(key) != slot {
err := fmt.Errorf("redis: Watch requires all keys to be in the same slot")
return err
}
}
node, err := c.slotMasterNode(ctx, slot)
if err != nil {
return err
}
for attempt := 0; attempt <= c.opt.MaxRedirects; attempt++ {
if attempt > 0 {
if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
return err
}
}
err = node.Client.Watch(ctx, fn, keys...)
if err == nil {
break
}
moved, ask, addr := isMovedError(err)
if moved || ask {
node, err = c.nodes.Get(addr)
if err != nil {
return err
}
continue
}
if isReadOnly := isReadOnlyError(err); isReadOnly || err == pool.ErrClosed {
if isReadOnly {
c.state.LazyReload()
}
node, err = c.slotMasterNode(ctx, slot)
if err != nil {
return err
}
continue
}
if shouldRetry(err, true) {
continue
}
return err
}
return err
}
func (c *ClusterClient) pubSub() *PubSub {
var node *clusterNode
pubsub := &PubSub{
opt: c.opt.clientOptions(),
newConn: func(ctx context.Context, channels []string) (*pool.Conn, error) {
if node != nil {
panic("node != nil")
}
var err error
if len(channels) > 0 {
slot := hashtag.Slot(channels[0])
node, err = c.slotMasterNode(ctx, slot)
} else {
node, err = c.nodes.Random()
}
if err != nil {
return nil, err
}
cn, err := node.Client.newConn(context.TODO())
if err != nil {
node = nil
return nil, err
}
return cn, nil
},
closeConn: func(cn *pool.Conn) error {
err := node.Client.connPool.CloseConn(cn)
node = nil
return err
},
}
pubsub.init()
return pubsub
}
// Subscribe subscribes the client to the specified channels.
// Channels can be omitted to create empty subscription.
func (c *ClusterClient) Subscribe(ctx context.Context, channels ...string) *PubSub {
pubsub := c.pubSub()
if len(channels) > 0 {
_ = pubsub.Subscribe(ctx, channels...)
}
return pubsub
}
// PSubscribe subscribes the client to the given patterns.
// Patterns can be omitted to create empty subscription.
func (c *ClusterClient) PSubscribe(ctx context.Context, channels ...string) *PubSub {
pubsub := c.pubSub()
if len(channels) > 0 {
_ = pubsub.PSubscribe(ctx, channels...)
}
return pubsub
}
func (c *ClusterClient) retryBackoff(attempt int) time.Duration {
return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
}
func (c *ClusterClient) cmdsInfo(ctx context.Context) (map[string]*CommandInfo, error) {
// Try 3 random nodes.
const nodeLimit = 3
addrs, err := c.nodes.Addrs()
if err != nil {
return nil, err
}
var firstErr error
perm := rand.Perm(len(addrs))
if len(perm) > nodeLimit {
perm = perm[:nodeLimit]
}
for _, idx := range perm {
addr := addrs[idx]
node, err := c.nodes.Get(addr)
if err != nil {
if firstErr == nil {
firstErr = err
}
continue
}
info, err := node.Client.Command(ctx).Result()
if err == nil {
return info, nil
}
if firstErr == nil {
firstErr = err
}
}
if firstErr == nil {
panic("not reached")
}
return nil, firstErr
}
func (c *ClusterClient) cmdInfo(name string) *CommandInfo {
cmdsInfo, err := c.cmdsInfoCache.Get(c.ctx)
if err != nil {
return nil
}
info := cmdsInfo[name]
if info == nil {
internal.Logger.Printf(c.Context(), "info for cmd=%s not found", name)
}
return info
}
func (c *ClusterClient) cmdSlot(cmd Cmder) int {
args := cmd.Args()
if args[0] == "cluster" && args[1] == "getkeysinslot" {
return args[2].(int)
}
cmdInfo := c.cmdInfo(cmd.Name())
return cmdSlot(cmd, cmdFirstKeyPos(cmd, cmdInfo))
}
func cmdSlot(cmd Cmder, pos int) int {
if pos == 0 {
return hashtag.RandomSlot()
}
firstKey := cmd.stringArg(pos)
return hashtag.Slot(firstKey)
}
func (c *ClusterClient) cmdNode(
ctx context.Context,
cmdInfo *CommandInfo,
slot int,
) (*clusterNode, error) {
state, err := c.state.Get(ctx)
if err != nil {
return nil, err
}
if c.opt.ReadOnly && cmdInfo != nil && cmdInfo.ReadOnly {
return c.slotReadOnlyNode(state, slot)
}
return state.slotMasterNode(slot)
}
func (c *clusterClient) slotReadOnlyNode(state *clusterState, slot int) (*clusterNode, error) {
if c.opt.RouteByLatency {
return state.slotClosestNode(slot)
}
if c.opt.RouteRandomly {
return state.slotRandomNode(slot)
}
return state.slotSlaveNode(slot)
}
func (c *ClusterClient) slotMasterNode(ctx context.Context, slot int) (*clusterNode, error) {
state, err := c.state.Get(ctx)
if err != nil {
return nil, err
}
return state.slotMasterNode(slot)
}
// SlaveForKey gets a client for a replica node to run any command on it.
// This is especially useful if we want to run a particular lua script which has
// only read only commands on the replica.
// This is because other redis commands generally have a flag that points that
// they are read only and automatically run on the replica nodes
// if ClusterOptions.ReadOnly flag is set to true.
func (c *ClusterClient) SlaveForKey(ctx context.Context, key string) (*Client, error) {
state, err := c.state.Get(ctx)
if err != nil {
return nil, err
}
slot := hashtag.Slot(key)
node, err := c.slotReadOnlyNode(state, slot)
if err != nil {
return nil, err
}
return node.Client, err
}
// MasterForKey return a client to the master node for a particular key.
func (c *ClusterClient) MasterForKey(ctx context.Context, key string) (*Client, error) {
slot := hashtag.Slot(key)
node, err := c.slotMasterNode(ctx, slot)
if err != nil {
return nil, err
}
return node.Client, err
}
func appendUniqueNode(nodes []*clusterNode, node *clusterNode) []*clusterNode {
for _, n := range nodes {
if n == node {
return nodes
}
}
return append(nodes, node)
}
func appendIfNotExists(ss []string, es ...string) []string {
loop:
for _, e := range es {
for _, s := range ss {
if s == e {
continue loop
}
}
ss = append(ss, e)
}
return ss
}
//------------------------------------------------------------------------------
type cmdsMap struct {
mu sync.Mutex
m map[*clusterNode][]Cmder
}
func newCmdsMap() *cmdsMap {
return &cmdsMap{
m: make(map[*clusterNode][]Cmder),
}
}
func (m *cmdsMap) Add(node *clusterNode, cmds ...Cmder) {
m.mu.Lock()
m.m[node] = append(m.m[node], cmds...)
m.mu.Unlock()
}