-
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
/*
Package couchbase provides a smart client for go.
Usage:
client, err := couchbase.Connect("http://myserver:8091/")
handleError(err)
pool, err := client.GetPool("default")
handleError(err)
bucket, err := pool.GetBucket("MyAwesomeBucket")
handleError(err)
...
or a shortcut for the bucket directly
bucket, err := couchbase.GetBucket("http://myserver:8091/", "default", "default")
in any case, you can specify authentication credentials using
standard URL userinfo syntax:
b, err := couchbase.GetBucket("http://bucketname:bucketpass@myserver:8091/",
"default", "bucket")
*/
package couchbase
import (
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"io"
"runtime"
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"unsafe"
"github.com/couchbase/gomemcached"
"github.com/couchbase/gomemcached/client" // package name is 'memcached'
"github.com/couchbase/goutils/logging"
)
// Mutation Token
type MutationToken struct {
VBid uint16 // vbucket id
Guard uint64 // vbuuid
Value uint64 // sequence number
}
// Maximum number of times to retry a chunk of a bulk get on error.
var MaxBulkRetries = 5000
var backOffDuration time.Duration = 100 * time.Millisecond
var MaxBackOffRetries = 25 // exponentail backOff result in over 30sec (25*13*0.1s)
// If this is set to a nonzero duration, Do() and ViewCustom() will log a warning if the call
// takes longer than that.
var SlowServerCallWarningThreshold time.Duration
func slowLog(startTime time.Time, format string, args ...interface{}) {
if elapsed := time.Now().Sub(startTime); elapsed > SlowServerCallWarningThreshold {
pc, _, _, _ := runtime.Caller(2)
caller := runtime.FuncForPC(pc).Name()
logging.Infof("go-couchbase: "+format+" in "+caller+" took "+elapsed.String(), args...)
}
}
// Return true if error is KEY_ENOENT. Required by cbq-engine
func IsKeyEExistsError(err error) bool {
res, ok := err.(*gomemcached.MCResponse)
if ok && res.Status == gomemcached.KEY_EEXISTS {
return true
}
return false
}
// Return true if error is KEY_ENOENT. Required by cbq-engine
func IsKeyNoEntError(err error) bool {
res, ok := err.(*gomemcached.MCResponse)
if ok && res.Status == gomemcached.KEY_ENOENT {
return true
}
return false
}
// Return true if error suggests a bucket refresh is required. Required by cbq-engine
func IsRefreshRequired(err error) bool {
res, ok := err.(*gomemcached.MCResponse)
if ok && (res.Status == gomemcached.NO_BUCKET || res.Status == gomemcached.NOT_MY_VBUCKET) {
return true
}
return false
}
// Return true if a collection is not known. Required by cbq-engine
func IsUnknownCollection(err error) bool {
res, ok := err.(*gomemcached.MCResponse)
if ok && (res.Status == gomemcached.UNKNOWN_COLLECTION) {
return true
}
return false
}
// ClientOpCallback is called for each invocation of Do.
var ClientOpCallback func(opname, k string, start time.Time, err error)
// Do executes a function on a memcached connection to the node owning key "k"
//
// Note that this automatically handles transient errors by replaying
// your function on a "not-my-vbucket" error, so don't assume
// your command will only be executed only once.
func (b *Bucket) Do(k string, f func(mc *memcached.Client, vb uint16) error) (err error) {
return b.Do2(k, f, true)
}
func (b *Bucket) Do2(k string, f func(mc *memcached.Client, vb uint16) error, deadline bool) (err error) {
var lastError error
if SlowServerCallWarningThreshold > 0 {
defer slowLog(time.Now(), "call to Do(%q)", k)
}
vb := b.VBHash(k)
maxTries := len(b.Nodes()) * 2
for i := 0; i < maxTries; i++ {
conn, pool, err := b.getConnectionToVBucket(vb)
if err != nil {
if (err == errNoPool || isConnError(err)) && backOff(i, maxTries, backOffDuration, true) {
b.Refresh()
continue
}
return err
}
if deadline && DefaultTimeout > 0 {
conn.SetDeadline(getDeadline(noDeadline, DefaultTimeout))
} else {
conn.SetDeadline(noDeadline)
}
lastError = f(conn, uint16(vb))
retry := false
discard := isOutOfBoundsError(lastError) || IsReadTimeOutError(lastError)
// MB-30967 / MB-31001 implement back off for transient errors
if resp, ok := lastError.(*gomemcached.MCResponse); ok {
switch resp.Status {
case gomemcached.NOT_MY_VBUCKET:
b.Refresh()
// MB-28842: in case of NMVB, check if the node is still part of the map
// and ditch the connection if it isn't.
discard = b.checkVBmap(pool.Node())
retry = true
case gomemcached.NOT_SUPPORTED:
discard = true
retry = true
case gomemcached.ENOMEM:
fallthrough
case gomemcached.TMPFAIL, gomemcached.EBUSY:
retry = backOff(i, maxTries, backOffDuration, true)
}
} else if lastError != nil && isConnError(lastError) && backOff(i, maxTries, backOffDuration, true) {
retry = true
}
if discard {
pool.Discard(conn)
} else {
pool.Return(conn)
}
if !retry {
return lastError
}
}
return fmt.Errorf("unable to complete action after %v attemps: ", maxTries, lastError)
}
type GatheredStats struct {
Server string
Stats map[string]string
Err error
}
func getStatsParallel(sn string, b *Bucket, offset int, which string,
ch chan<- GatheredStats) {
pool := b.getConnPool(offset)
var gatheredStats GatheredStats
conn, err := pool.Get()
defer func() {
pool.Return(conn)
ch <- gatheredStats
}()
if err != nil {
gatheredStats = GatheredStats{Server: sn, Err: err}
} else {
conn.SetDeadline(getDeadline(time.Time{}, DefaultTimeout))
sm, err := conn.StatsMap(which)
gatheredStats = GatheredStats{Server: sn, Stats: sm, Err: err}
}
}
func getStatsParallelFunc(fn func(key, val []byte), sn string, b *Bucket, offset int, which string,
ch chan<- GatheredStats) {
pool := b.getConnPool(offset)
conn, err := pool.Get()
if err == nil {
conn.SetDeadline(getDeadline(time.Time{}, DefaultTimeout))
err = conn.StatsFunc(which, fn)
pool.Return(conn)
}
ch <- GatheredStats{Server: sn, Err: err}
}
// GetStats gets a set of stats from all servers.
//
// Returns a map of server ID -> map of stat key to map value.
func (b *Bucket) GetStats(which string) map[string]map[string]string {
rv := map[string]map[string]string{}
for server, gs := range b.GatherStats(which) {
if len(gs.Stats) > 0 {
rv[server] = gs.Stats
}
}
return rv
}
// GatherStats returns a map of server ID -> GatheredStats from all servers.
func (b *Bucket) GatherStats(which string) map[string]GatheredStats {
vsm := b.VBServerMap()
if vsm.ServerList == nil {
return nil
}
// Go grab all the things at once.
ch := make(chan GatheredStats, len(vsm.ServerList))
for i, sn := range vsm.ServerList {
go getStatsParallel(sn, b, i, which, ch)
}
// Gather the results
rv := map[string]GatheredStats{}
for range vsm.ServerList {
gs := <-ch
rv[gs.Server] = gs
}
return rv
}
// GatherStats returns a map of server ID -> GatheredStats from all servers.
func (b *Bucket) GatherStatsFunc(which string, fn func(key, val []byte)) map[string]error {
var errMap map[string]error
vsm := b.VBServerMap()
if vsm.ServerList == nil {
return errMap
}
// Go grab all the things at once.
ch := make(chan GatheredStats, len(vsm.ServerList))
for i, sn := range vsm.ServerList {
go getStatsParallelFunc(fn, sn, b, i, which, ch)
}
// Gather the results
for range vsm.ServerList {
gs := <-ch
if gs.Err != nil {
if errMap == nil {
errMap = make(map[string]error)
errMap[gs.Server] = gs.Err
}
}
}
return errMap
}
type BucketStats int
const (
StatCount = BucketStats(iota)
StatSize
)
var bucketStatString = []string{
"curr_items",
"ep_value_size",
}
var collectionStatString = []string{
"items",
"disk_size",
}
// Get selected bucket or collection stats
func (b *Bucket) GetIntStats(refresh bool, which []BucketStats, context ...*memcached.ClientContext) ([]int64, error) {
if refresh {
b.Refresh()
}
var vals []int64 = make([]int64, len(which))
if len(vals) == 0 {
return vals, nil
}
var outErr error
if len(context) > 0 {
collKey := fmt.Sprintf("collections-byid 0x%x", context[0].CollId)
errs := b.GatherStatsFunc(collKey, func(key, val []byte) {
for i, f := range which {
lk := len(key)
ls := len(collectionStatString[f])
if lk >= ls && string(key[lk-ls:]) == collectionStatString[f] {
v, err := strconv.ParseInt(string(val), 10, 64)
if err == nil {
atomic.AddInt64(&vals[i], v)
} else if outErr == nil {
outErr = err
}
}
}
})
// have to use a range to access any one element of a map
for _, err := range errs {
return nil, err
}
} else {
errs := b.GatherStatsFunc("", func(key, val []byte) {
for i, f := range which {
if string(key) == bucketStatString[f] {
v, err := strconv.ParseInt(string(val), 10, 64)
if err == nil {
atomic.AddInt64(&vals[i], v)
} else if outErr == nil {
outErr = err
}
}
}
})
// have to use a range to access any one element of a map
for _, err := range errs {
return nil, err
}
}
return vals, outErr
}
// Get bucket count through the bucket stats
func (b *Bucket) GetCount(refresh bool, context ...*memcached.ClientContext) (count int64, err error) {
if refresh {
b.Refresh()
}
var cnt int64
if len(context) > 0 {
key := fmt.Sprintf("collections-byid 0x%x", context[0].CollId)
resKey := ""
for _, gs := range b.GatherStats(key) {
if len(gs.Stats) > 0 {
// the key encodes the scope and collection id
// we don't have the scope id, so we have to find it...
if resKey == "" {
for k, _ := range gs.Stats {
resKey = strings.TrimRightFunc(k, func(r rune) bool {
return r != ':'
}) + "items"
break
}
}
cnt, err = strconv.ParseInt(gs.Stats[resKey], 10, 64)
if err != nil {
return 0, err
}
count += cnt
} else if gs.Err != nil {
return 0, gs.Err
}
}
} else {
for _, gs := range b.GatherStats("") {
if len(gs.Stats) > 0 {
cnt, err = strconv.ParseInt(gs.Stats["curr_items"], 10, 64)
if err != nil {
return 0, err
}
count += cnt
} else if gs.Err != nil {
return 0, gs.Err
}
}
}
return count, nil
}
// Get bucket document size through the bucket stats
func (b *Bucket) GetSize(refresh bool, context ...*memcached.ClientContext) (size int64, err error) {
if refresh {
b.Refresh()
}
var sz int64
if len(context) > 0 {
key := fmt.Sprintf("collections-byid 0x%x", context[0].CollId)
resKey := ""
for _, gs := range b.GatherStats(key) {
if len(gs.Stats) > 0 {
// the key encodes the scope and collection id
// we don't have the scope id, so we have to find it...
if resKey == "" {
for k, _ := range gs.Stats {
resKey = strings.TrimRightFunc(k, func(r rune) bool {
return r != ':'
}) + "disk_size"
break
}
}
sz, err = strconv.ParseInt(gs.Stats[resKey], 10, 64)
if err != nil {
return 0, err
}
size += sz
} else if gs.Err != nil {
return 0, gs.Err
}
}
} else {
for _, gs := range b.GatherStats("") {
if len(gs.Stats) > 0 {
sz, err = strconv.ParseInt(gs.Stats["ep_value_size"], 10, 64)
if err != nil {
return 0, err
}
size += sz
} else if gs.Err != nil {
return 0, gs.Err
}
}
}
return size, nil
}
func isAuthError(err error) bool {
estr := err.Error()
return strings.Contains(estr, "Auth failure")
}
func IsReadTimeOutError(err error) bool {
if err == nil {
return false
}
estr := err.Error()
return strings.Contains(estr, "read tcp") ||
strings.Contains(estr, "i/o timeout")
}
func isTimeoutError(err error) bool {
estr := err.Error()
return strings.Contains(estr, "i/o timeout") ||
strings.Contains(estr, "connection timed out") ||
strings.Contains(estr, "no route to host")
}
// Errors that are not considered fatal for our fetch loop
func isConnError(err error) bool {
if err == io.EOF {
return true
}
estr := err.Error()
return strings.Contains(estr, "broken pipe") ||
strings.Contains(estr, "connection reset") ||
strings.Contains(estr, "connection refused") ||
strings.Contains(estr, "connection pool is closed")
}
func isOutOfBoundsError(err error) bool {
return err != nil && strings.Contains(err.Error(), "Out of Bounds error")
}
func getDeadline(reqDeadline time.Time, duration time.Duration) time.Time {
if reqDeadline.IsZero() {
if duration > 0 {
return time.Unix(time.Now().Unix(), 0).Add(duration)
} else {
return noDeadline
}
}
return reqDeadline
}
func backOff(attempt, maxAttempts int, duration time.Duration, exponential bool) bool {
if attempt < maxAttempts {
// 0th attempt return immediately
if attempt > 0 {
if exponential {
duration = time.Duration(attempt) * duration
}
time.Sleep(duration)
}
return true
}
return false
}
func (b *Bucket) doBulkGet(vb uint16, keys []string, reqDeadline time.Time,
ch chan<- map[string]*gomemcached.MCResponse, ech chan<- error, subPaths []string,
eStatus *errorStatus, context ...*memcached.ClientContext) {
if SlowServerCallWarningThreshold > 0 {
defer slowLog(time.Now(), "call to doBulkGet(%d, %d keys)", vb, len(keys))
}
rv := _STRING_MCRESPONSE_POOL.Get()
attempts := 0
backOffAttempts := 0
done := false
bname := b.Name
for ; attempts < MaxBulkRetries && !done && !eStatus.errStatus; attempts++ {
if len(b.VBServerMap().VBucketMap) < int(vb) {
//fatal
err := fmt.Errorf("vbmap smaller than requested for %v", bname)
logging.Errorf("go-couchbase: %v vb %d vbmap len %d", err.Error(), vb, len(b.VBServerMap().VBucketMap))
ech <- err
return
}
masterID := b.VBServerMap().VBucketMap[vb][0]
if masterID < 0 {
// fatal
err := fmt.Errorf("No master node available for %v vb %d", bname, vb)
logging.Errorf("%v", err.Error())
ech <- err
return
}
// This stack frame exists to ensure we can clean up
// connection at a reasonable time.
err := func() error {
pool := b.getConnPool(masterID)
conn, err := pool.Get()
if err != nil {
if isAuthError(err) || isTimeoutError(err) {
logging.Errorf("Fatal Error %v : %v", bname, err)
ech <- err
return err
} else if isConnError(err) {
if !backOff(backOffAttempts, MaxBackOffRetries, backOffDuration, true) {
logging.Errorf("Connection Error %v : %v", bname, err)
ech <- err
return err
}
b.Refresh()
backOffAttempts++
} else if err == errNoPool {
if !backOff(backOffAttempts, MaxBackOffRetries, backOffDuration, true) {
logging.Errorf("Connection Error %v : %v", bname, err)
ech <- err
return err
}
err = b.Refresh()
if err != nil {
ech <- err
return err
}
backOffAttempts++
// retry, and make no noise
return nil
}
logging.Infof("Pool Get returned %v: %v", bname, err)
// retry
return nil
}
conn.SetDeadline(getDeadline(reqDeadline, DefaultTimeout))
err = conn.GetBulk(vb, keys, rv, subPaths, context...)
discard := false
defer func() {
if discard {
pool.Discard(conn)
} else {
pool.Return(conn)
}
}()
switch err.(type) {
case *gomemcached.MCResponse:
notSMaxTries := len(b.Nodes()) * 2
st := err.(*gomemcached.MCResponse).Status
if st == gomemcached.NOT_MY_VBUCKET || (st == gomemcached.NOT_SUPPORTED && attempts < notSMaxTries) {
b.Refresh()
discard = b.checkVBmap(pool.Node())
return nil // retry
} else if st == gomemcached.EBUSY || st == gomemcached.LOCKED {
if (attempts % (MaxBulkRetries / 100)) == 0 {
logging.Infof("Retrying Memcached error (%v) FOR %v(vbid:%d, keys:<ud>%v</ud>)",
err.Error(), bname, vb, keys)
}
return nil // retry
} else if (st == gomemcached.ENOMEM || st == gomemcached.TMPFAIL) && backOff(backOffAttempts, MaxBackOffRetries, backOffDuration, true) {
// MB-30967 / MB-31001 use backoff for TMPFAIL too
backOffAttempts++
logging.Infof("Retrying Memcached error (%v) FOR %v(vbid:%d, keys:<ud>%v</ud>)",
err.Error(), bname, vb, keys)
return nil // retry
}
ech <- err
return err
case error:
if isOutOfBoundsError(err) || IsReadTimeOutError(err) {
// We got an out of bounds error or a read timeout error; retry the operation
discard = true
return nil
} else if isConnError(err) && backOff(backOffAttempts, MaxBackOffRetries, backOffDuration, true) {
backOffAttempts++
logging.Errorf("Connection Error: %s. Refreshing bucket %v (vbid:%v,keys:<ud>%v</ud>)",
err.Error(), bname, vb, keys)
discard = true
b.Refresh()
return nil // retry
}
ech <- err
ch <- rv
return err
}
done = true
return nil
}()
if err != nil {
return
}
}
if attempts >= MaxBulkRetries {
err := fmt.Errorf("bulkget exceeded MaxBulkRetries for %v(vbid:%d,keys:<ud>%v</ud>)", bname, vb, keys)
logging.Errorf("%v", err.Error())
ech <- err
}
ch <- rv
}
type errorStatus struct {
errStatus bool
}
type vbBulkGet struct {
b *Bucket
ch chan<- map[string]*gomemcached.MCResponse
ech chan<- error
k uint16
keys []string
reqDeadline time.Time
wg *sync.WaitGroup
subPaths []string
groupError *errorStatus
context []*memcached.ClientContext
}
const _NUM_CHANNELS = 5
var _NUM_CHANNEL_WORKERS = (runtime.NumCPU() + 1) / 2
var DefaultDialTimeout = time.Duration(0)
var DefaultTimeout = time.Duration(0)
var noDeadline = time.Time{}
// Buffer 4k requests per worker
var _VB_BULK_GET_CHANNELS []chan *vbBulkGet
func InitBulkGet() {
DefaultDialTimeout = 20 * time.Second
DefaultTimeout = 120 * time.Second
memcached.SetDefaultDialTimeout(DefaultDialTimeout)
_VB_BULK_GET_CHANNELS = make([]chan *vbBulkGet, _NUM_CHANNELS)
for i := 0; i < _NUM_CHANNELS; i++ {
channel := make(chan *vbBulkGet, 16*1024*_NUM_CHANNEL_WORKERS)
_VB_BULK_GET_CHANNELS[i] = channel
for j := 0; j < _NUM_CHANNEL_WORKERS; j++ {
go vbBulkGetWorker(channel)
}
}
}
func vbBulkGetWorker(ch chan *vbBulkGet) {
defer func() {
// Workers cannot panic and die
recover()
go vbBulkGetWorker(ch)
}()
for vbg := range ch {
vbDoBulkGet(vbg)
}
}
func vbDoBulkGet(vbg *vbBulkGet) {
defer vbg.wg.Done()
defer func() {
// Workers cannot panic and die
recover()
}()
vbg.b.doBulkGet(vbg.k, vbg.keys, vbg.reqDeadline, vbg.ch, vbg.ech, vbg.subPaths, vbg.groupError, vbg.context...)
}
var _ERR_CHAN_FULL = fmt.Errorf("Data request queue full, aborting query.")
func (b *Bucket) processBulkGet(kdm map[uint16][]string, reqDeadline time.Time,
ch chan<- map[string]*gomemcached.MCResponse, ech chan<- error, subPaths []string,
eStatus *errorStatus, context ...*memcached.ClientContext) {
defer close(ch)
defer close(ech)
wg := &sync.WaitGroup{}
for k, keys := range kdm {
// GetBulk() group has error donot Queue items for this group
if eStatus.errStatus {
break
}
vbg := &vbBulkGet{
b: b,
ch: ch,
ech: ech,
k: k,
keys: keys,
reqDeadline: reqDeadline,
wg: wg,
subPaths: subPaths,
groupError: eStatus,
context: context,
}
wg.Add(1)
// Random int
// Right shift to avoid 8-byte alignment, and take low bits
c := (uintptr(unsafe.Pointer(vbg)) >> 4) % _NUM_CHANNELS
select {
case _VB_BULK_GET_CHANNELS[c] <- vbg:
// No-op
default:
// Buffer full, abandon the bulk get
ech <- _ERR_CHAN_FULL
wg.Add(-1)
}
}
// Wait for my vb bulk gets
wg.Wait()
}
type multiError []error
func (m multiError) Error() string {
if len(m) == 0 {
panic("Error of none")
}
return fmt.Sprintf("{%v errors, starting with %v}", len(m), m[0].Error())
}
// Convert a stream of errors from ech into a multiError (or nil) and
// send down eout.
//
// At least one send is guaranteed on eout, but two is possible, so
// buffer the out channel appropriately.
func errorCollector(ech <-chan error, eout chan<- error, eStatus *errorStatus) {
defer func() { eout <- nil }()
var errs multiError
for e := range ech {
if !eStatus.errStatus && !IsKeyNoEntError(e) {
eStatus.errStatus = true
}
errs = append(errs, e)
}
if len(errs) > 0 {
eout <- errs
}
}
// Fetches multiple keys concurrently, with []byte values
//
// This is a wrapper around GetBulk which converts all values returned
// by GetBulk from raw memcached responses into []byte slices.
// Returns one document for duplicate keys
func (b *Bucket) GetBulkRaw(keys []string, context ...*memcached.ClientContext) (map[string][]byte, error) {
resp, eout := b.getBulk(keys, noDeadline, nil, context...)
rv := make(map[string][]byte, len(keys))
for k, av := range resp {
rv[k] = av.Body
}
b.ReleaseGetBulkPools(resp)
return rv, eout
}
// GetBulk fetches multiple keys concurrently.
//
// Unlike more convenient GETs, the entire response is returned in the
// map array for each key. Keys that were not found will not be included in
// the map.
func (b *Bucket) GetBulk(keys []string, reqDeadline time.Time, subPaths []string, context ...*memcached.ClientContext) (map[string]*gomemcached.MCResponse, error) {
return b.getBulk(keys, reqDeadline, subPaths, context...)
}
func (b *Bucket) ReleaseGetBulkPools(rv map[string]*gomemcached.MCResponse) {
_STRING_MCRESPONSE_POOL.Put(rv)
}
func (b *Bucket) getBulk(keys []string, reqDeadline time.Time, subPaths []string, context ...*memcached.ClientContext) (map[string]*gomemcached.MCResponse, error) {
kdm := _VB_STRING_POOL.Get()
defer _VB_STRING_POOL.Put(kdm)
for _, k := range keys {
if k != "" {
vb := uint16(b.VBHash(k))
a, ok1 := kdm[vb]
if !ok1 {
a = _STRING_POOL.Get()
}
kdm[vb] = append(a, k)
}
}
eout := make(chan error, 2)
groupErrorStatus := &errorStatus{}
// processBulkGet will own both of these channels and
// guarantee they're closed before it returns.
ch := make(chan map[string]*gomemcached.MCResponse)
ech := make(chan error)
go errorCollector(ech, eout, groupErrorStatus)
go b.processBulkGet(kdm, reqDeadline, ch, ech, subPaths, groupErrorStatus, context...)
var rv map[string]*gomemcached.MCResponse
for m := range ch {
if rv == nil {
rv = m
continue
}
for k, v := range m {
rv[k] = v
}
_STRING_MCRESPONSE_POOL.Put(m)
}
return rv, <-eout
}
// WriteOptions is the set of option flags availble for the Write
// method. They are ORed together to specify the desired request.
type WriteOptions int
const (
// Raw specifies that the value is raw []byte or nil; don't
// JSON-encode it.
Raw = WriteOptions(1 << iota)
// AddOnly indicates an item should only be written if it
// doesn't exist, otherwise ErrKeyExists is returned.
AddOnly
// Persist causes the operation to block until the server
// confirms the item is persisted.
Persist
// Indexable causes the operation to block until it's availble via the index.
Indexable
// Append indicates the given value should be appended to the
// existing value for the given key.
Append
)
var optNames = []struct {
opt WriteOptions
name string
}{
{Raw, "raw"},
{AddOnly, "addonly"}, {Persist, "persist"},
{Indexable, "indexable"}, {Append, "append"},
}
// String representation of WriteOptions
func (w WriteOptions) String() string {
f := []string{}
for _, on := range optNames {
if w&on.opt != 0 {
f = append(f, on.name)
w &= ^on.opt
}
}
if len(f) == 0 || w != 0 {
f = append(f, fmt.Sprintf("0x%x", int(w)))
}
return strings.Join(f, "|")
}
// Error returned from Write with AddOnly flag, when key already exists in the bucket.
var ErrKeyExists = errors.New("key exists")
// General-purpose value setter.
//
// The Set, Add and Delete methods are just wrappers around this. The
// interpretation of `v` depends on whether the `Raw` option is
// given. If it is, v must be a byte array or nil. (A nil value causes
// a delete.) If `Raw` is not given, `v` will be marshaled as JSON
// before being written. It must be JSON-marshalable and it must not
// be nil.
func (b *Bucket) Write(k string, flags, exp int, v interface{},
opt WriteOptions, context ...*memcached.ClientContext) (err error) {
_, err = b.WriteWithCAS(k, flags, exp, v, opt, context...)
return err
}
func (b *Bucket) WriteWithCAS(k string, flags, exp int, v interface{},
opt WriteOptions, context ...*memcached.ClientContext) (cas uint64, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) {
ClientOpCallback(fmt.Sprintf("Write(%v)", opt), k, t, err)
}(time.Now())
}
var data []byte
if opt&Raw == 0 {
data, err = json.Marshal(v)
if err != nil {
return cas, err
}
} else if v != nil {
data = v.([]byte)
}
var res *gomemcached.MCResponse
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
if opt&AddOnly != 0 {
res, err = memcached.UnwrapMemcachedError(
mc.Add(vb, k, flags, exp, data, context...))
if err == nil && res.Status != gomemcached.SUCCESS {
if res.Status == gomemcached.KEY_EEXISTS {
err = ErrKeyExists
} else {
err = res
}
}
} else if opt&Append != 0 {
res, err = mc.Append(vb, k, data, context...)
} else if data == nil {
res, err = mc.Del(vb, k, context...)
} else {
res, err = mc.Set(vb, k, flags, exp, data, context...)
}
if err == nil {
cas = res.Cas
}
return err
})
if err == nil && (opt&(Persist|Indexable) != 0) {
err = b.WaitForPersistence(k, cas, data == nil)
}
return cas, err
}
func (b *Bucket) WriteWithMT(k string, flags, exp int, v interface{},
opt WriteOptions, context ...*memcached.ClientContext) (mt *MutationToken, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) {
ClientOpCallback(fmt.Sprintf("WriteWithMT(%v)", opt), k, t, err)
}(time.Now())
}
var data []byte
if opt&Raw == 0 {
data, err = json.Marshal(v)
if err != nil {
return nil, err
}
} else if v != nil {
data = v.([]byte)
}
var res *gomemcached.MCResponse
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
if opt&AddOnly != 0 {
res, err = memcached.UnwrapMemcachedError(
mc.Add(vb, k, flags, exp, data, context...))
if err == nil && res.Status != gomemcached.SUCCESS {
if res.Status == gomemcached.KEY_EEXISTS {
err = ErrKeyExists
} else {
err = res
}
}
} else if opt&Append != 0 {
res, err = mc.Append(vb, k, data, context...)
} else if data == nil {
res, err = mc.Del(vb, k, context...)
} else {
res, err = mc.Set(vb, k, flags, exp, data, context...)
}
if len(res.Extras) >= 16 {
vbuuid := uint64(binary.BigEndian.Uint64(res.Extras[0:8]))
seqNo := uint64(binary.BigEndian.Uint64(res.Extras[8:16]))
mt = &MutationToken{VBid: vb, Guard: vbuuid, Value: seqNo}
}
return err
})
if err == nil && (opt&(Persist|Indexable) != 0) {
err = b.WaitForPersistence(k, res.Cas, data == nil)
}
return mt, err
}
// Set a value in this bucket with Cas and return the new Cas value
func (b *Bucket) Cas(k string, exp int, cas uint64, v interface{}, context ...*memcached.ClientContext) (uint64, error) {
return b.WriteCas(k, 0, exp, cas, v, 0, context...)
}
// Set a value in this bucket with Cas without json encoding it
func (b *Bucket) CasRaw(k string, exp int, cas uint64, v interface{}, context ...*memcached.ClientContext) (uint64, error) {
return b.WriteCas(k, 0, exp, cas, v, Raw, context...)
}
func (b *Bucket) WriteCas(k string, flags, exp int, cas uint64, v interface{},
opt WriteOptions, context ...*memcached.ClientContext) (newCas uint64, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) {
ClientOpCallback(fmt.Sprintf("Write(%v)", opt), k, t, err)
}(time.Now())
}
var data []byte
if opt&Raw == 0 {
data, err = json.Marshal(v)
if err != nil {
return 0, err
}
} else if v != nil {
data = v.([]byte)
}
var res *gomemcached.MCResponse
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err = mc.SetCas(vb, k, flags, exp, cas, data, context...)
return err
})
if err == nil && (opt&(Persist|Indexable) != 0) {
err = b.WaitForPersistence(k, res.Cas, data == nil)
}
return res.Cas, err
}
// Extended CAS operation. These functions will return the mutation token, i.e vbuuid & guard
func (b *Bucket) CasWithMeta(k string, flags int, exp int, cas uint64, v interface{}, context ...*memcached.ClientContext) (uint64, *MutationToken, error) {
return b.WriteCasWithMT(k, flags, exp, cas, v, 0, context...)
}
func (b *Bucket) CasWithMetaRaw(k string, flags int, exp int, cas uint64, v interface{}, context ...*memcached.ClientContext) (uint64, *MutationToken, error) {
return b.WriteCasWithMT(k, flags, exp, cas, v, Raw, context...)
}
func (b *Bucket) WriteCasWithMT(k string, flags, exp int, cas uint64, v interface{},
opt WriteOptions, context ...*memcached.ClientContext) (newCas uint64, mt *MutationToken, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) {
ClientOpCallback(fmt.Sprintf("Write(%v)", opt), k, t, err)
}(time.Now())
}
var data []byte
if opt&Raw == 0 {
data, err = json.Marshal(v)
if err != nil {
return 0, nil, err
}
} else if v != nil {
data = v.([]byte)
}
var res *gomemcached.MCResponse
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err = mc.SetCas(vb, k, flags, exp, cas, data, context...)
return err
})
if err != nil {
return 0, nil, err
}
// check for extras
if len(res.Extras) >= 16 {
vbuuid := uint64(binary.BigEndian.Uint64(res.Extras[0:8]))
seqNo := uint64(binary.BigEndian.Uint64(res.Extras[8:16]))
vb := b.VBHash(k)
mt = &MutationToken{VBid: uint16(vb), Guard: vbuuid, Value: seqNo}
}
if err == nil && (opt&(Persist|Indexable) != 0) {
err = b.WaitForPersistence(k, res.Cas, data == nil)
}
return res.Cas, mt, err
}
// Set a value in this bucket.
// The value will be serialized into a JSON document.
func (b *Bucket) Set(k string, exp int, v interface{}, context ...*memcached.ClientContext) error {
return b.Write(k, 0, exp, v, 0, context...)
}
// Set a value in this bucket.
func (b *Bucket) SetWithCAS(k string, exp int, v interface{}, context ...*memcached.ClientContext) (uint64, error) {
return b.WriteWithCAS(k, 0, exp, v, 0, context...)
}
// Set a value in this bucket with with flags
func (b *Bucket) SetWithMeta(k string, flags int, exp int, v interface{}, context ...*memcached.ClientContext) (*MutationToken, error) {
return b.WriteWithMT(k, flags, exp, v, 0, context...)
}
// SetRaw sets a value in this bucket without JSON encoding it.
func (b *Bucket) SetRaw(k string, exp int, v []byte, context ...*memcached.ClientContext) error {
return b.Write(k, 0, exp, v, Raw, context...)
}
// Add adds a value to this bucket; like Set except that nothing
// happens if the key exists. The value will be serialized into a
// JSON document.
func (b *Bucket) Add(k string, exp int, v interface{}, context ...*memcached.ClientContext) (added bool, err error) {
err = b.Write(k, 0, exp, v, AddOnly, context...)
if err == ErrKeyExists {
return false, nil
}
return (err == nil), err
}
// Add adds a value to this bucket; like Set except that nothing
// happens if the key exists. Return the CAS value.
func (b *Bucket) AddWithCAS(k string, exp int, v interface{}, context ...*memcached.ClientContext) (bool, uint64, error) {
cas, err := b.WriteWithCAS(k, 0, exp, v, AddOnly, context...)
if err == ErrKeyExists {
return false, 0, nil
}
return (err == nil), cas, err
}
// AddRaw adds a value to this bucket; like SetRaw except that nothing
// happens if the key exists. The value will be stored as raw bytes.
func (b *Bucket) AddRaw(k string, exp int, v []byte, context ...*memcached.ClientContext) (added bool, err error) {
err = b.Write(k, 0, exp, v, AddOnly|Raw, context...)
if err == ErrKeyExists {
return false, nil
}
return (err == nil), err
}
// Add adds a value to this bucket; like Set except that nothing
// happens if the key exists. The value will be serialized into a
// JSON document.
func (b *Bucket) AddWithMT(k string, exp int, v interface{}, context ...*memcached.ClientContext) (added bool, mt *MutationToken, err error) {
mt, err = b.WriteWithMT(k, 0, exp, v, AddOnly, context...)
if err == ErrKeyExists {
return false, mt, nil
}
return (err == nil), mt, err
}
// AddRaw adds a value to this bucket; like SetRaw except that nothing
// happens if the key exists. The value will be stored as raw bytes.
func (b *Bucket) AddRawWithMT(k string, exp int, v []byte, context ...*memcached.ClientContext) (added bool, mt *MutationToken, err error) {
mt, err = b.WriteWithMT(k, 0, exp, v, AddOnly|Raw, context...)
if err == ErrKeyExists {
return false, mt, nil
}
return (err == nil), mt, err
}
// Append appends raw data to an existing item.
func (b *Bucket) Append(k string, data []byte, context ...*memcached.ClientContext) error {
return b.Write(k, 0, 0, data, Append|Raw, context...)
}
// Returns collectionUid, manifestUid, error.
func (b *Bucket) GetCollectionCID(scope string, collection string, reqDeadline time.Time) (uint32, uint32, error) {
var err error
var response *gomemcached.MCResponse
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("GetCollectionCID", scope+"."+collection, t, err) }(time.Now())
}
var key = "DUMMY" // Contact any server.
var manifestUid uint32
var collUid uint32
err = b.Do2(key, func(mc *memcached.Client, vb uint16) error {
var err1 error
mc.SetDeadline(getDeadline(reqDeadline, DefaultTimeout))
_, err1 = mc.SelectBucket(b.Name)
if err1 != nil {
return err1
}
response, err1 = mc.CollectionsGetCID(scope, collection)
if err1 != nil {
return err1
}
manifestUid = binary.BigEndian.Uint32(response.Extras[4:8])
collUid = binary.BigEndian.Uint32(response.Extras[8:12])
return nil
}, false)
return collUid, manifestUid, err
}
// Get a value straight from Memcached
func (b *Bucket) GetsMC(key string, reqDeadline time.Time, context ...*memcached.ClientContext) (*gomemcached.MCResponse, error) {
var err error
var response *gomemcached.MCResponse
if key == "" {
return nil, nil
}
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("GetsMC", key, t, err) }(time.Now())
}
err = b.Do2(key, func(mc *memcached.Client, vb uint16) error {
var err1 error
mc.SetDeadline(getDeadline(reqDeadline, DefaultTimeout))
response, err1 = mc.Get(vb, key, context...)
if err1 != nil {
return err1
}
return nil
}, false)
return response, err
}
// Get a value through the subdoc API
func (b *Bucket) GetsSubDoc(key string, reqDeadline time.Time, subPaths []string, context ...*memcached.ClientContext) (*gomemcached.MCResponse, error) {
var err error
var response *gomemcached.MCResponse
if key == "" {
return nil, nil
}
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("GetsSubDoc", key, t, err) }(time.Now())
}
err = b.Do2(key, func(mc *memcached.Client, vb uint16) error {
var err1 error
mc.SetDeadline(getDeadline(reqDeadline, DefaultTimeout))
response, err1 = mc.GetSubdoc(vb, key, subPaths, context...)
if err1 != nil {
return err1
}
return nil
}, false)
return response, err
}
// GetsRaw gets a raw value from this bucket including its CAS
// counter and flags.
func (b *Bucket) GetsRaw(k string, context ...*memcached.ClientContext) (data []byte, flags int,
cas uint64, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("GetsRaw", k, t, err) }(time.Now())
}
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err := mc.Get(vb, k, context...)
if err != nil {
return err
}
cas = res.Cas
if len(res.Extras) >= 4 {
flags = int(binary.BigEndian.Uint32(res.Extras))
}
data = res.Body
return nil
})
return
}
// Gets gets a value from this bucket, including its CAS counter. The
// value is expected to be a JSON stream and will be deserialized into
// rv.
func (b *Bucket) Gets(k string, rv interface{}, caso *uint64, context ...*memcached.ClientContext) error {
data, _, cas, err := b.GetsRaw(k, context...)
if err != nil {
return err
}
if caso != nil {
*caso = cas
}
return json.Unmarshal(data, rv)
}
// Get a value from this bucket.
// The value is expected to be a JSON stream and will be deserialized
// into rv.
func (b *Bucket) Get(k string, rv interface{}, context ...*memcached.ClientContext) error {
return b.Gets(k, rv, nil, context...)
}
// GetRaw gets a raw value from this bucket. No marshaling is performed.
func (b *Bucket) GetRaw(k string, context ...*memcached.ClientContext) ([]byte, error) {
d, _, _, err := b.GetsRaw(k, context...)
return d, err
}
// GetAndTouchRaw gets a raw value from this bucket including its CAS
// counter and flags, and updates the expiry on the doc.
func (b *Bucket) GetAndTouchRaw(k string, exp int, context ...*memcached.ClientContext) (data []byte,
cas uint64, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("GetsRaw", k, t, err) }(time.Now())
}
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err := mc.GetAndTouch(vb, k, exp, context...)
if err != nil {
return err
}
cas = res.Cas
data = res.Body
return nil
})
return data, cas, err
}
// GetMeta returns the meta values for a key
func (b *Bucket) GetMeta(k string, flags *int, expiry *int, cas *uint64, seqNo *uint64, context ...*memcached.ClientContext) (err error) {
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("GetsMeta", k, t, err) }(time.Now())
}
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err := mc.GetMeta(vb, k, context...)
if err != nil {
return err
}
*cas = res.Cas
if len(res.Extras) >= 8 {
*flags = int(binary.BigEndian.Uint32(res.Extras[4:]))
}
if len(res.Extras) >= 12 {
*expiry = int(binary.BigEndian.Uint32(res.Extras[8:]))
}
if len(res.Extras) >= 20 {
*seqNo = uint64(binary.BigEndian.Uint64(res.Extras[12:]))
}
return nil
})
return err
}
// Delete a key from this bucket.
func (b *Bucket) Delete(k string, context ...*memcached.ClientContext) error {
return b.Write(k, 0, 0, nil, Raw, context...)
}
// Incr increments the value at a given key by amt and defaults to def if no value present.
func (b *Bucket) Incr(k string, amt, def uint64, exp int, context ...*memcached.ClientContext) (val uint64, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("Incr", k, t, err) }(time.Now())
}
var rv uint64
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err := mc.Incr(vb, k, amt, def, exp, context...)
if err != nil {
return err
}
rv = res
return nil
})
return rv, err
}
// Decr decrements the value at a given key by amt and defaults to def if no value present
func (b *Bucket) Decr(k string, amt, def uint64, exp int, context ...*memcached.ClientContext) (val uint64, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("Decr", k, t, err) }(time.Now())
}
var rv uint64
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
res, err := mc.Decr(vb, k, amt, def, exp, context...)
if err != nil {
return err
}
rv = res
return nil
})
return rv, err
}
// Wrapper around memcached.CASNext()
func (b *Bucket) casNext(k string, exp int, state *memcached.CASState) bool {
if ClientOpCallback != nil {
defer func(t time.Time) {
ClientOpCallback("casNext", k, t, state.Err)
}(time.Now())
}
keepGoing := false
state.Err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
keepGoing = mc.CASNext(vb, k, exp, state)
return state.Err
})
return keepGoing && state.Err == nil
}
// An UpdateFunc is a callback function to update a document
type UpdateFunc func(current []byte) (updated []byte, err error)
// Return this as the error from an UpdateFunc to cancel the Update
// operation.
const UpdateCancel = memcached.CASQuit
// Update performs a Safe update of a document, avoiding conflicts by
// using CAS.
//
// The callback function will be invoked with the current raw document
// contents (or nil if the document doesn't exist); it should return
// the updated raw contents (or nil to delete.) If it decides not to
// change anything it can return UpdateCancel as the error.
//
// If another writer modifies the document between the get and the
// set, the callback will be invoked again with the newer value.
func (b *Bucket) Update(k string, exp int, callback UpdateFunc) error {
_, err := b.update(k, exp, callback)
return err
}
// internal version of Update that returns a CAS value
func (b *Bucket) update(k string, exp int, callback UpdateFunc) (newCas uint64, err error) {
var state memcached.CASState
for b.casNext(k, exp, &state) {
var err error
if state.Value, err = callback(state.Value); err != nil {
return 0, err
}
}
return state.Cas, state.Err
}
// A WriteUpdateFunc is a callback function to update a document
type WriteUpdateFunc func(current []byte) (updated []byte, opt WriteOptions, err error)
// WriteUpdate performs a Safe update of a document, avoiding
// conflicts by using CAS. WriteUpdate is like Update, except that
// the callback can return a set of WriteOptions, of which Persist and
// Indexable are recognized: these cause the call to wait until the
// document update has been persisted to disk and/or become available
// to index.
func (b *Bucket) WriteUpdate(k string, exp int, callback WriteUpdateFunc) error {
var writeOpts WriteOptions
var deletion bool
// Wrap the callback in an UpdateFunc we can pass to Update:
updateCallback := func(current []byte) (updated []byte, err error) {
update, opt, err := callback(current)
writeOpts = opt
deletion = (update == nil)
return update, err
}
cas, err := b.update(k, exp, updateCallback)
if err != nil {
return err
}
// If callback asked, wait for persistence or indexability:
if writeOpts&(Persist|Indexable) != 0 {
err = b.WaitForPersistence(k, cas, deletion)
}
return err
}
// Observe observes the current state of a document.
func (b *Bucket) Observe(k string) (result memcached.ObserveResult, err error) {
if ClientOpCallback != nil {
defer func(t time.Time) { ClientOpCallback("Observe", k, t, err) }(time.Now())
}
err = b.Do(k, func(mc *memcached.Client, vb uint16) error {
result, err = mc.Observe(vb, k)
return err
})
return
}
// Returned from WaitForPersistence (or Write, if the Persistent or Indexable flag is used)
// if the value has been overwritten by another before being persisted.
var ErrOverwritten = errors.New("overwritten")
// Returned from WaitForPersistence (or Write, if the Persistent or Indexable flag is used)
// if the value hasn't been persisted by the timeout interval
var ErrTimeout = errors.New("timeout")
// WaitForPersistence waits for an item to be considered durable.
//
// Besides transport errors, ErrOverwritten may be returned if the
// item is overwritten before it reaches durability. ErrTimeout may
// occur if the item isn't found durable in a reasonable amount of
// time.
func (b *Bucket) WaitForPersistence(k string, cas uint64, deletion bool) error {
timeout := 10 * time.Second
sleepDelay := 5 * time.Millisecond
start := time.Now()
for {
time.Sleep(sleepDelay)
sleepDelay += sleepDelay / 2 // multiply delay by 1.5 every time
result, err := b.Observe(k)
if err != nil {
return err
}
if persisted, overwritten := result.CheckPersistence(cas, deletion); overwritten {
return ErrOverwritten
} else if persisted {
return nil
}
if result.PersistenceTime > 0 {
timeout = 2 * result.PersistenceTime
}
if time.Since(start) >= timeout-sleepDelay {
return ErrTimeout
}
}
}
var _STRING_MCRESPONSE_POOL = gomemcached.NewStringMCResponsePool(16)
type stringPool struct {
pool *sync.Pool
size int
}
func newStringPool(size int) *stringPool {
rv := &stringPool{
pool: &sync.Pool{
New: func() interface{} {
return make([]string, 0, size)
},
},
size: size,
}
return rv
}
func (this *stringPool) Get() []string {
return this.pool.Get().([]string)
}
func (this *stringPool) Put(s []string) {
if s == nil || cap(s) < this.size || cap(s) > 2*this.size {
return
}
this.pool.Put(s[0:0])
}
var _STRING_POOL = newStringPool(16)
type vbStringPool struct {
pool *sync.Pool
strPool *stringPool
}
func newVBStringPool(size int, sp *stringPool) *vbStringPool {
rv := &vbStringPool{
pool: &sync.Pool{
New: func() interface{} {
return make(map[uint16][]string, size)
},
},
strPool: sp,
}
return rv
}
func (this *vbStringPool) Get() map[uint16][]string {
return this.pool.Get().(map[uint16][]string)
}
func (this *vbStringPool) Put(s map[uint16][]string) {
if s == nil {
return
}
for k, v := range s {
delete(s, k)
this.strPool.Put(v)
}
this.pool.Put(s)
}
var _VB_STRING_POOL = newVBStringPool(16, _STRING_POOL)