summaryrefslogtreecommitdiffstats
path: root/src/osd/OSD.h
blob: 8c87823d2da70832c340e7f89171c4dcbb796e15 (plain)
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
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
1758
1759
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
1798
1799
1800
1801
1802
1803
1804
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
1907
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
1945
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
2017
2018
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
2037
2038
2039
2040
2041
2042
2043
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
2114
2115
2116
2117
2118
2119
2120
2121
2122
2123
2124
2125
2126
2127
2128
2129
2130
2131
2132
2133
2134
2135
2136
2137
2138
2139
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149
2150
2151
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161
2162
2163
2164
2165
2166
2167
2168
2169
2170
2171
2172
2173
2174
2175
2176
2177
2178
2179
2180
2181
2182
2183
2184
2185
2186
2187
2188
2189
2190
2191
2192
2193
2194
2195
2196
2197
2198
2199
2200
2201
2202
2203
2204
2205
2206
2207
2208
2209
2210
2211
2212
2213
2214
2215
2216
2217
2218
2219
2220
2221
2222
2223
2224
2225
2226
2227
2228
2229
2230
2231
2232
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242
2243
2244
2245
2246
2247
2248
2249
2250
2251
2252
2253
2254
2255
2256
2257
2258
2259
2260
2261
2262
2263
2264
2265
2266
2267
2268
2269
2270
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286
2287
2288
2289
2290
2291
2292
2293
2294
2295
2296
2297
2298
2299
2300
2301
2302
2303
2304
2305
2306
2307
2308
2309
2310
2311
2312
2313
2314
2315
2316
2317
2318
2319
2320
2321
2322
2323
2324
2325
2326
2327
2328
2329
2330
2331
2332
2333
2334
2335
2336
2337
2338
2339
2340
2341
2342
2343
2344
2345
2346
2347
2348
2349
2350
2351
2352
2353
2354
2355
2356
2357
2358
2359
2360
2361
2362
2363
2364
2365
2366
2367
2368
2369
2370
2371
2372
2373
2374
2375
2376
2377
2378
2379
2380
2381
2382
2383
2384
2385
2386
2387
2388
2389
2390
2391
2392
2393
2394
2395
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*- 
// vim: ts=8 sw=2 smarttab
/*
 * Ceph - scalable distributed file system
 *
 * Copyright (C) 2004-2006 Sage Weil <sage@newdream.net>
 *
 * This is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License version 2.1, as published by the Free Software 
 * Foundation.  See file COPYING.
 * 
 */

#ifndef CEPH_OSD_H
#define CEPH_OSD_H

#include "PG.h"

#include "msg/Dispatcher.h"

#include "common/Mutex.h"
#include "common/RWLock.h"
#include "common/Timer.h"
#include "common/WorkQueue.h"
#include "common/AsyncReserver.h"
#include "common/ceph_context.h"
#include "common/config_cacher.h"
#include "common/zipkin_trace.h"

#include "mgr/MgrClient.h"

#include "os/ObjectStore.h"
#include "OSDCap.h" 
 
#include "auth/KeyRing.h"

#include "osd/ClassHandler.h"

#include "include/CompatSet.h"

#include "OpRequest.h"
#include "Session.h"

#include "osd/OpQueueItem.h"

#include <atomic>
#include <map>
#include <memory>
#include <string>

#include "include/unordered_map.h"

#include "common/shared_cache.hpp"
#include "common/simple_cache.hpp"
#include "common/sharedptr_registry.hpp"
#include "common/WeightedPriorityQueue.h"
#include "common/PrioritizedQueue.h"
#include "osd/mClockOpClassQueue.h"
#include "osd/mClockClientQueue.h"
#include "messages/MOSDOp.h"
#include "common/EventTrace.h"

#define CEPH_OSD_PROTOCOL    10 /* cluster internal */

/*

  lock ordering for pg map

    PG::lock
      ShardData::lock
        OSD::pg_map_lock

  */

enum {
  l_osd_first = 10000,
  l_osd_op_wip,
  l_osd_op,
  l_osd_op_inb,
  l_osd_op_outb,
  l_osd_op_lat,
  l_osd_op_process_lat,
  l_osd_op_prepare_lat,
  l_osd_op_r,
  l_osd_op_r_outb,
  l_osd_op_r_lat,
  l_osd_op_r_lat_outb_hist,
  l_osd_op_r_process_lat,
  l_osd_op_r_prepare_lat,
  l_osd_op_w,
  l_osd_op_w_inb,
  l_osd_op_w_lat,
  l_osd_op_w_lat_inb_hist,
  l_osd_op_w_process_lat,
  l_osd_op_w_prepare_lat,
  l_osd_op_rw,
  l_osd_op_rw_inb,
  l_osd_op_rw_outb,
  l_osd_op_rw_lat,
  l_osd_op_rw_lat_inb_hist,
  l_osd_op_rw_lat_outb_hist,
  l_osd_op_rw_process_lat,
  l_osd_op_rw_prepare_lat,

  l_osd_op_before_queue_op_lat,
  l_osd_op_before_dequeue_op_lat,

  l_osd_sop,
  l_osd_sop_inb,
  l_osd_sop_lat,
  l_osd_sop_w,
  l_osd_sop_w_inb,
  l_osd_sop_w_lat,
  l_osd_sop_pull,
  l_osd_sop_pull_lat,
  l_osd_sop_push,
  l_osd_sop_push_inb,
  l_osd_sop_push_lat,

  l_osd_pull,
  l_osd_push,
  l_osd_push_outb,

  l_osd_rop,
  l_osd_rbytes,

  l_osd_loadavg,
  l_osd_cached_crc,
  l_osd_cached_crc_adjusted,
  l_osd_missed_crc,

  l_osd_pg,
  l_osd_pg_primary,
  l_osd_pg_replica,
  l_osd_pg_stray,
  l_osd_pg_removing,
  l_osd_hb_to,
  l_osd_map,
  l_osd_mape,
  l_osd_mape_dup,

  l_osd_waiting_for_map,

  l_osd_map_cache_hit,
  l_osd_map_cache_miss,
  l_osd_map_cache_miss_low,
  l_osd_map_cache_miss_low_avg,
  l_osd_map_bl_cache_hit,
  l_osd_map_bl_cache_miss,

  l_osd_stat_bytes,
  l_osd_stat_bytes_used,
  l_osd_stat_bytes_avail,

  l_osd_copyfrom,

  l_osd_tier_promote,
  l_osd_tier_flush,
  l_osd_tier_flush_fail,
  l_osd_tier_try_flush,
  l_osd_tier_try_flush_fail,
  l_osd_tier_evict,
  l_osd_tier_whiteout,
  l_osd_tier_dirty,
  l_osd_tier_clean,
  l_osd_tier_delay,
  l_osd_tier_proxy_read,
  l_osd_tier_proxy_write,

  l_osd_agent_wake,
  l_osd_agent_skip,
  l_osd_agent_flush,
  l_osd_agent_evict,

  l_osd_object_ctx_cache_hit,
  l_osd_object_ctx_cache_total,

  l_osd_op_cache_hit,
  l_osd_tier_flush_lat,
  l_osd_tier_promote_lat,
  l_osd_tier_r_lat,

  l_osd_pg_info,
  l_osd_pg_fastinfo,
  l_osd_pg_biginfo,

  l_osd_last,
};

// RecoveryState perf counters
enum {
  rs_first = 20000,
  rs_initial_latency,
  rs_started_latency,
  rs_reset_latency,
  rs_start_latency,
  rs_primary_latency,
  rs_peering_latency,
  rs_backfilling_latency,
  rs_waitremotebackfillreserved_latency,
  rs_waitlocalbackfillreserved_latency,
  rs_notbackfilling_latency,
  rs_repnotrecovering_latency,
  rs_repwaitrecoveryreserved_latency,
  rs_repwaitbackfillreserved_latency,
  rs_reprecovering_latency,
  rs_activating_latency,
  rs_waitlocalrecoveryreserved_latency,
  rs_waitremoterecoveryreserved_latency,
  rs_recovering_latency,
  rs_recovered_latency,
  rs_clean_latency,
  rs_active_latency,
  rs_replicaactive_latency,
  rs_stray_latency,
  rs_getinfo_latency,
  rs_getlog_latency,
  rs_waitactingchange_latency,
  rs_incomplete_latency,
  rs_down_latency,
  rs_getmissing_latency,
  rs_waitupthru_latency,
  rs_notrecovering_latency,
  rs_last,
};

class Messenger;
class Message;
class MonClient;
class PerfCounters;
class ObjectStore;
class FuseStore;
class OSDMap;
class MLog;
class Objecter;
class KeyStore;

class Watch;
class PrimaryLogPG;

class TestOpsSocketHook;
struct C_FinishSplits;
struct C_OpenPGs;
class LogChannel;
class CephContext;
class MOSDOp;

class MOSDPGCreate2;
class MOSDPGQuery;
class MOSDPGNotify;
class MOSDPGInfo;
class MOSDPGRemove;
class MOSDForceRecovery;

class OSD;

class OSDService {
public:
  OSD *osd;
  CephContext *cct;
  ObjectStore::CollectionHandle meta_ch;
  const int whoami;
  ObjectStore *&store;
  LogClient &log_client;
  LogChannelRef clog;
  PGRecoveryStats &pg_recovery_stats;
private:
  Messenger *&cluster_messenger;
  Messenger *&client_messenger;
public:
  PerfCounters *&logger;
  PerfCounters *&recoverystate_perf;
  MonClient   *&monc;
  ClassHandler  *&class_handler;

  md_config_cacher_t<Option::size_t> osd_max_object_size;
  md_config_cacher_t<bool> osd_skip_data_digest;

  void enqueue_back(OpQueueItem&& qi);
  void enqueue_front(OpQueueItem&& qi);

  void maybe_inject_dispatch_delay() {
    if (g_conf()->osd_debug_inject_dispatch_delay_probability > 0) {
      if (rand() % 10000 <
	  g_conf()->osd_debug_inject_dispatch_delay_probability * 10000) {
	utime_t t;
	t.set_from_double(g_conf()->osd_debug_inject_dispatch_delay_duration);
	t.sleep();
      }
    }
  }

private:
  // -- superblock --
  ceph::mutex publish_lock, pre_publish_lock; // pre-publish orders before publish
  OSDSuperblock superblock;

public:
  OSDSuperblock get_superblock() {
    std::lock_guard l(publish_lock);
    return superblock;
  }
  void publish_superblock(const OSDSuperblock &block) {
    std::lock_guard l(publish_lock);
    superblock = block;
  }

  int get_nodeid() const { return whoami; }

  std::atomic<epoch_t> max_oldest_map;
private:
  OSDMapRef osdmap;

public:
  OSDMapRef get_osdmap() {
    std::lock_guard l(publish_lock);
    return osdmap;
  }
  epoch_t get_osdmap_epoch() {
    std::lock_guard l(publish_lock);
    return osdmap ? osdmap->get_epoch() : 0;
  }
  void publish_map(OSDMapRef map) {
    std::lock_guard l(publish_lock);
    osdmap = map;
  }

  /*
   * osdmap - current published map
   * next_osdmap - pre_published map that is about to be published.
   *
   * We use the next_osdmap to send messages and initiate connections,
   * but only if the target is the same instance as the one in the map
   * epoch the current user is working from (i.e., the result is
   * equivalent to what is in next_osdmap).
   *
   * This allows the helpers to start ignoring osds that are about to
   * go down, and let OSD::handle_osd_map()/note_down_osd() mark them
   * down, without worrying about reopening connections from threads
   * working from old maps.
   */
private:
  OSDMapRef next_osdmap;
  ceph::condition_variable pre_publish_cond;

public:
  void pre_publish_map(OSDMapRef map) {
    std::lock_guard l(pre_publish_lock);
    next_osdmap = std::move(map);
  }

  void activate_map();
  /// map epochs reserved below
  map<epoch_t, unsigned> map_reservations;

  /// gets ref to next_osdmap and registers the epoch as reserved
  OSDMapRef get_nextmap_reserved() {
    std::lock_guard l(pre_publish_lock);
    if (!next_osdmap)
      return OSDMapRef();
    epoch_t e = next_osdmap->get_epoch();
    map<epoch_t, unsigned>::iterator i =
      map_reservations.insert(make_pair(e, 0)).first;
    i->second++;
    return next_osdmap;
  }
  /// releases reservation on map
  void release_map(OSDMapRef osdmap) {
    std::lock_guard l(pre_publish_lock);
    map<epoch_t, unsigned>::iterator i =
      map_reservations.find(osdmap->get_epoch());
    ceph_assert(i != map_reservations.end());
    ceph_assert(i->second > 0);
    if (--(i->second) == 0) {
      map_reservations.erase(i);
    }
    pre_publish_cond.notify_all();
  }
  /// blocks until there are no reserved maps prior to next_osdmap
  void await_reserved_maps() {
    std::unique_lock l{pre_publish_lock};
    ceph_assert(next_osdmap);
    pre_publish_cond.wait(l, [this] {
      auto i = map_reservations.cbegin();
      return (i == map_reservations.cend() ||
	      i->first >= next_osdmap->get_epoch());
    });
  }
  OSDMapRef get_next_osdmap() {
    std::lock_guard l(pre_publish_lock);
    if (!next_osdmap)
      return OSDMapRef();
    return next_osdmap;
  }

private:
  Mutex peer_map_epoch_lock;
  map<int, epoch_t> peer_map_epoch;
public:
  epoch_t get_peer_epoch(int p);
  epoch_t note_peer_epoch(int p, epoch_t e);
  void forget_peer_epoch(int p, epoch_t e);

  void send_map(class MOSDMap *m, Connection *con);
  void send_incremental_map(epoch_t since, Connection *con, const OSDMapRef& osdmap);
  MOSDMap *build_incremental_map_msg(epoch_t from, epoch_t to,
                                       OSDSuperblock& superblock);
  bool should_share_map(entity_name_t name, Connection *con, epoch_t epoch,
                        const OSDMapRef& osdmap, const epoch_t *sent_epoch_p);
  void share_map(entity_name_t name, Connection *con, epoch_t epoch,
                 OSDMapRef& osdmap, epoch_t *sent_epoch_p);
  void share_map_peer(int peer, Connection *con,
                      OSDMapRef map = OSDMapRef());

  ConnectionRef get_con_osd_cluster(int peer, epoch_t from_epoch);
  pair<ConnectionRef,ConnectionRef> get_con_osd_hb(int peer, epoch_t from_epoch);  // (back, front)
  void send_message_osd_cluster(int peer, Message *m, epoch_t from_epoch);
  void send_message_osd_cluster(Message *m, Connection *con) {
    con->send_message(m);
  }
  void send_message_osd_cluster(Message *m, const ConnectionRef& con) {
    con->send_message(m);
  }
  void send_message_osd_client(Message *m, Connection *con) {
    con->send_message(m);
  }
  void send_message_osd_client(Message *m, const ConnectionRef& con) {
    con->send_message(m);
  }
  entity_name_t get_cluster_msgr_name() const;

private:
  // -- scrub scheduling --
  Mutex sched_scrub_lock;
  int scrubs_local;
  int scrubs_remote;

public:
  struct ScrubJob {
    CephContext* cct;
    /// pg to be scrubbed
    spg_t pgid;
    /// a time scheduled for scrub. but the scrub could be delayed if system
    /// load is too high or it fails to fall in the scrub hours
    utime_t sched_time;
    /// the hard upper bound of scrub time
    utime_t deadline;
    ScrubJob() : cct(nullptr) {}
    explicit ScrubJob(CephContext* cct, const spg_t& pg,
		      const utime_t& timestamp,
		      double pool_scrub_min_interval = 0,
		      double pool_scrub_max_interval = 0, bool must = true);
    /// order the jobs by sched_time
    bool operator<(const ScrubJob& rhs) const;
  };
  set<ScrubJob> sched_scrub_pg;

  /// @returns the scrub_reg_stamp used for unregister the scrub job
  utime_t reg_pg_scrub(spg_t pgid, utime_t t, double pool_scrub_min_interval,
		       double pool_scrub_max_interval, bool must) {
    ScrubJob scrub(cct, pgid, t, pool_scrub_min_interval, pool_scrub_max_interval,
		   must);
    std::lock_guard l(sched_scrub_lock);
    sched_scrub_pg.insert(scrub);
    return scrub.sched_time;
  }
  void unreg_pg_scrub(spg_t pgid, utime_t t) {
    std::lock_guard l(sched_scrub_lock);
    size_t removed = sched_scrub_pg.erase(ScrubJob(cct, pgid, t));
    ceph_assert(removed);
  }
  bool first_scrub_stamp(ScrubJob *out) {
    std::lock_guard l(sched_scrub_lock);
    if (sched_scrub_pg.empty())
      return false;
    set<ScrubJob>::iterator iter = sched_scrub_pg.begin();
    *out = *iter;
    return true;
  }
  bool next_scrub_stamp(const ScrubJob& next,
			ScrubJob *out) {
    std::lock_guard l(sched_scrub_lock);
    if (sched_scrub_pg.empty())
      return false;
    set<ScrubJob>::const_iterator iter = sched_scrub_pg.lower_bound(next);
    if (iter == sched_scrub_pg.cend())
      return false;
    ++iter;
    if (iter == sched_scrub_pg.cend())
      return false;
    *out = *iter;
    return true;
  }

  void dumps_scrub(Formatter *f) {
    ceph_assert(f != nullptr);
    std::lock_guard l(sched_scrub_lock);

    f->open_array_section("scrubs");
    for (const auto &i: sched_scrub_pg) {
      f->open_object_section("scrub");
      f->dump_stream("pgid") << i.pgid;
      f->dump_stream("sched_time") << i.sched_time;
      f->dump_stream("deadline") << i.deadline;
      f->dump_bool("forced", i.sched_time == PG::Scrubber::scrub_must_stamp());
      f->close_section();
    }
    f->close_section();
  }

  bool can_inc_scrubs();
  bool inc_scrubs_local();
  void dec_scrubs_local();
  bool inc_scrubs_remote();
  void dec_scrubs_remote();
  void dump_scrub_reservations(Formatter *f);

  void reply_op_error(OpRequestRef op, int err);
  void reply_op_error(OpRequestRef op, int err, eversion_t v, version_t uv);
  void handle_misdirected_op(PG *pg, OpRequestRef op);


private:
  // -- agent shared state --
  Mutex agent_lock;
  Cond agent_cond;
  map<uint64_t, set<PGRef> > agent_queue;
  set<PGRef>::iterator agent_queue_pos;
  bool agent_valid_iterator;
  int agent_ops;
  int flush_mode_high_count; //once have one pg with FLUSH_MODE_HIGH then flush objects with high speed
  set<hobject_t> agent_oids;
  bool agent_active;
  struct AgentThread : public Thread {
    OSDService *osd;
    explicit AgentThread(OSDService *o) : osd(o) {}
    void *entry() override {
      osd->agent_entry();
      return NULL;
    }
  } agent_thread;
  bool agent_stop_flag;
  Mutex agent_timer_lock;
  SafeTimer agent_timer;

public:
  void agent_entry();
  void agent_stop();

  void _enqueue(PG *pg, uint64_t priority) {
    if (!agent_queue.empty() &&
	agent_queue.rbegin()->first < priority)
      agent_valid_iterator = false;  // inserting higher-priority queue
    set<PGRef>& nq = agent_queue[priority];
    if (nq.empty())
      agent_cond.Signal();
    nq.insert(pg);
  }

  void _dequeue(PG *pg, uint64_t old_priority) {
    set<PGRef>& oq = agent_queue[old_priority];
    set<PGRef>::iterator p = oq.find(pg);
    ceph_assert(p != oq.end());
    if (p == agent_queue_pos)
      ++agent_queue_pos;
    oq.erase(p);
    if (oq.empty()) {
      if (agent_queue.rbegin()->first == old_priority)
	agent_valid_iterator = false;
      agent_queue.erase(old_priority);
    }
  }

  /// enable agent for a pg
  void agent_enable_pg(PG *pg, uint64_t priority) {
    std::lock_guard l(agent_lock);
    _enqueue(pg, priority);
  }

  /// adjust priority for an enagled pg
  void agent_adjust_pg(PG *pg, uint64_t old_priority, uint64_t new_priority) {
    std::lock_guard l(agent_lock);
    ceph_assert(new_priority != old_priority);
    _enqueue(pg, new_priority);
    _dequeue(pg, old_priority);
  }

  /// disable agent for a pg
  void agent_disable_pg(PG *pg, uint64_t old_priority) {
    std::lock_guard l(agent_lock);
    _dequeue(pg, old_priority);
  }

  /// note start of an async (evict) op
  void agent_start_evict_op() {
    std::lock_guard l(agent_lock);
    ++agent_ops;
  }

  /// note finish or cancellation of an async (evict) op
  void agent_finish_evict_op() {
    std::lock_guard l(agent_lock);
    ceph_assert(agent_ops > 0);
    --agent_ops;
    agent_cond.Signal();
  }

  /// note start of an async (flush) op
  void agent_start_op(const hobject_t& oid) {
    std::lock_guard l(agent_lock);
    ++agent_ops;
    ceph_assert(agent_oids.count(oid) == 0);
    agent_oids.insert(oid);
  }

  /// note finish or cancellation of an async (flush) op
  void agent_finish_op(const hobject_t& oid) {
    std::lock_guard l(agent_lock);
    ceph_assert(agent_ops > 0);
    --agent_ops;
    ceph_assert(agent_oids.count(oid) == 1);
    agent_oids.erase(oid);
    agent_cond.Signal();
  }

  /// check if we are operating on an object
  bool agent_is_active_oid(const hobject_t& oid) {
    std::lock_guard l(agent_lock);
    return agent_oids.count(oid);
  }

  /// get count of active agent ops
  int agent_get_num_ops() {
    std::lock_guard l(agent_lock);
    return agent_ops;
  }

  void agent_inc_high_count() {
    std::lock_guard l(agent_lock);
    flush_mode_high_count ++;
  }

  void agent_dec_high_count() {
    std::lock_guard l(agent_lock);
    flush_mode_high_count --;
  }

private:
  /// throttle promotion attempts
  std::atomic<unsigned int> promote_probability_millis{1000}; ///< probability thousands. one word.
  PromoteCounter promote_counter;
  utime_t last_recalibrate;
  unsigned long promote_max_objects, promote_max_bytes;

public:
  bool promote_throttle() {
    // NOTE: lockless!  we rely on the probability being a single word.
    promote_counter.attempt();
    if ((unsigned)rand() % 1000 > promote_probability_millis)
      return true;  // yes throttle (no promote)
    if (promote_max_objects &&
	promote_counter.objects > promote_max_objects)
      return true;  // yes throttle
    if (promote_max_bytes &&
	promote_counter.bytes > promote_max_bytes)
      return true;  // yes throttle
    return false;   //  no throttle (promote)
  }
  void promote_finish(uint64_t bytes) {
    promote_counter.finish(bytes);
  }
  void promote_throttle_recalibrate();

  // -- Objecter, for tiering reads/writes from/to other OSDs --
  Objecter *objecter;
  int m_objecter_finishers;
  vector<Finisher*> objecter_finishers;

  // -- Watch --
  Mutex watch_lock;
  SafeTimer watch_timer;
  uint64_t next_notif_id;
  uint64_t get_next_id(epoch_t cur_epoch) {
    std::lock_guard l(watch_lock);
    return (((uint64_t)cur_epoch) << 32) | ((uint64_t)(next_notif_id++));
  }

  // -- Recovery/Backfill Request Scheduling --
  Mutex recovery_request_lock;
  SafeTimer recovery_request_timer;

  // For async recovery sleep
  bool recovery_needs_sleep = true;
  utime_t recovery_schedule_time = utime_t();

  // For recovery & scrub & snap
  Mutex sleep_lock;
  SafeTimer sleep_timer;

  // -- tids --
  // for ops i issue
  std::atomic<unsigned int> last_tid{0};
  ceph_tid_t get_tid() {
    return (ceph_tid_t)last_tid++;
  }

  // -- backfill_reservation --
  Finisher reserver_finisher;
  AsyncReserver<spg_t> local_reserver;
  AsyncReserver<spg_t> remote_reserver;

  // -- pg merge --
  Mutex merge_lock = {"OSD::merge_lock"};
  map<pg_t,eversion_t> ready_to_merge_source;   // pg -> version
  map<pg_t,std::tuple<eversion_t,epoch_t,epoch_t>> ready_to_merge_target;  // pg -> (version,les,lec)
  set<pg_t> not_ready_to_merge_source;
  map<pg_t,pg_t> not_ready_to_merge_target;
  set<pg_t> sent_ready_to_merge_source;

  void set_ready_to_merge_source(PG *pg,
				 eversion_t version);
  void set_ready_to_merge_target(PG *pg,
				 eversion_t version,
				 epoch_t last_epoch_started,
				 epoch_t last_epoch_clean);
  void set_not_ready_to_merge_source(pg_t source);
  void set_not_ready_to_merge_target(pg_t target, pg_t source);
  void clear_ready_to_merge(PG *pg);
  void send_ready_to_merge();
  void _send_ready_to_merge();
  void clear_sent_ready_to_merge();
  void prune_sent_ready_to_merge(const OSDMapRef& osdmap);

  // -- pg_temp --
private:
  Mutex pg_temp_lock;
  struct pg_temp_t {
    vector<int> acting;
    bool forced = false;
  };
  map<pg_t, pg_temp_t> pg_temp_wanted;
  map<pg_t, pg_temp_t> pg_temp_pending;
  void _sent_pg_temp();
  friend std::ostream& operator<<(std::ostream&, const pg_temp_t&);
public:
  void queue_want_pg_temp(pg_t pgid, const vector<int>& want,
			  bool forced = false);
  void remove_want_pg_temp(pg_t pgid);
  void requeue_pg_temp();
  void send_pg_temp();

  ceph::mutex pg_created_lock = ceph::make_mutex("OSDService::pg_created_lock");
  set<pg_t> pg_created;
  void send_pg_created(pg_t pgid);
  void prune_pg_created();
  void send_pg_created();

  AsyncReserver<spg_t> snap_reserver;
  void queue_recovery_context(PG *pg, GenContext<ThreadPool::TPHandle&> *c);
  void queue_for_snap_trim(PG *pg);
  void queue_for_scrub(PG *pg, bool with_high_priority);
  void queue_for_pg_delete(spg_t pgid, epoch_t e);
  bool try_finish_pg_delete(PG *pg, unsigned old_pg_num);

private:
  // -- pg recovery and associated throttling --
  Mutex recovery_lock;
  list<pair<epoch_t, PGRef> > awaiting_throttle;

  utime_t defer_recovery_until;
  uint64_t recovery_ops_active;
  uint64_t recovery_ops_reserved;
  bool recovery_paused;
#ifdef DEBUG_RECOVERY_OIDS
  map<spg_t, set<hobject_t> > recovery_oids;
#endif
  bool _recover_now(uint64_t *available_pushes);
  void _maybe_queue_recovery();
  void _queue_for_recovery(
    pair<epoch_t, PGRef> p, uint64_t reserved_pushes);
public:
  void start_recovery_op(PG *pg, const hobject_t& soid);
  void finish_recovery_op(PG *pg, const hobject_t& soid, bool dequeue);
  bool is_recovery_active();
  void release_reserved_pushes(uint64_t pushes);
  void defer_recovery(float defer_for) {
    defer_recovery_until = ceph_clock_now();
    defer_recovery_until += defer_for;
  }
  void pause_recovery() {
    std::lock_guard l(recovery_lock);
    recovery_paused = true;
  }
  bool recovery_is_paused() {
    std::lock_guard l(recovery_lock);
    return recovery_paused;
  }
  void unpause_recovery() {
    std::lock_guard l(recovery_lock);
    recovery_paused = false;
    _maybe_queue_recovery();
  }
  void kick_recovery_queue() {
    std::lock_guard l(recovery_lock);
    _maybe_queue_recovery();
  }
  void clear_queued_recovery(PG *pg) {
    std::lock_guard l(recovery_lock);
    awaiting_throttle.remove_if(
      [pg](decltype(awaiting_throttle)::const_reference awaiting ) {
	return awaiting.second.get() == pg;
      });
  }
  // delayed pg activation
  void queue_for_recovery(PG *pg) {
    std::lock_guard l(recovery_lock);

    if (pg->is_forced_recovery_or_backfill()) {
      awaiting_throttle.push_front(make_pair(pg->get_osdmap()->get_epoch(), pg));
    } else {
      awaiting_throttle.push_back(make_pair(pg->get_osdmap()->get_epoch(), pg));
    }
    _maybe_queue_recovery();
  }
  void queue_recovery_after_sleep(PG *pg, epoch_t queued, uint64_t reserved_pushes) {
    std::lock_guard l(recovery_lock);
    _queue_for_recovery(make_pair(queued, pg), reserved_pushes);
  }

  // osd map cache (past osd maps)
  Mutex map_cache_lock;
  SharedLRU<epoch_t, const OSDMap> map_cache;
  SimpleLRU<epoch_t, bufferlist> map_bl_cache;
  SimpleLRU<epoch_t, bufferlist> map_bl_inc_cache;

  /// final pg_num values for recently deleted pools
  map<int64_t,int> deleted_pool_pg_nums;

  OSDMapRef try_get_map(epoch_t e);
  OSDMapRef get_map(epoch_t e) {
    OSDMapRef ret(try_get_map(e));
    ceph_assert(ret);
    return ret;
  }
  OSDMapRef add_map(OSDMap *o) {
    std::lock_guard l(map_cache_lock);
    return _add_map(o);
  }
  OSDMapRef _add_map(OSDMap *o);

  void add_map_bl(epoch_t e, bufferlist& bl) {
    std::lock_guard l(map_cache_lock);
    return _add_map_bl(e, bl);
  }
  void _add_map_bl(epoch_t e, bufferlist& bl);
  bool get_map_bl(epoch_t e, bufferlist& bl) {
    std::lock_guard l(map_cache_lock);
    return _get_map_bl(e, bl);
  }
  bool _get_map_bl(epoch_t e, bufferlist& bl);

  void add_map_inc_bl(epoch_t e, bufferlist& bl) {
    std::lock_guard l(map_cache_lock);
    return _add_map_inc_bl(e, bl);
  }
  void _add_map_inc_bl(epoch_t e, bufferlist& bl);
  bool get_inc_map_bl(epoch_t e, bufferlist& bl);

  /// get last pg_num before a pool was deleted (if any)
  int get_deleted_pool_pg_num(int64_t pool);

  void store_deleted_pool_pg_num(int64_t pool, int pg_num) {
    std::lock_guard l(map_cache_lock);
    deleted_pool_pg_nums[pool] = pg_num;
  }

  /// get pgnum from newmap or, if pool was deleted, last map pool existed in
  int get_possibly_deleted_pool_pg_num(OSDMapRef newmap,
				       int64_t pool) {
    if (newmap->have_pg_pool(pool)) {
      return newmap->get_pg_num(pool);
    }
    return get_deleted_pool_pg_num(pool);
  }

  /// identify split child pgids over a osdmap interval
  void identify_splits_and_merges(
    OSDMapRef old_map,
    OSDMapRef new_map,
    spg_t pgid,
    set<pair<spg_t,epoch_t>> *new_children,
    set<pair<spg_t,epoch_t>> *merge_pgs);

  void need_heartbeat_peer_update();

  void init();
  void final_init();  
  void start_shutdown();
  void shutdown_reserver();
  void shutdown();

  // -- stats --
  Mutex stat_lock;
  osd_stat_t osd_stat;
  uint32_t seq = 0;

  void set_statfs(const struct store_statfs_t &stbuf,
    osd_alert_list_t& alerts);
  osd_stat_t set_osd_stat(vector<int>& hb_peers, int num_pgs);
  void inc_osd_stat_repaired(void);
  void set_osd_stat_repaired(int64_t);
  float compute_adjusted_ratio(osd_stat_t new_stat, float *pratio, uint64_t adjust_used = 0);
  osd_stat_t get_osd_stat() {
    std::lock_guard l(stat_lock);
    ++seq;
    osd_stat.up_from = up_epoch;
    osd_stat.seq = ((uint64_t)osd_stat.up_from << 32) + seq;
    return osd_stat;
  }
  uint64_t get_osd_stat_seq() {
    std::lock_guard l(stat_lock);
    return osd_stat.seq;
  }
  void get_hb_pingtime(map<int, osd_stat_t::Interfaces> *pp)
  {
    std::lock_guard l(stat_lock);
    *pp = osd_stat.hb_pingtime;
    return;
  }

  // -- OSD Full Status --
private:
  friend TestOpsSocketHook;
  mutable Mutex full_status_lock;
  enum s_names { INVALID = -1, NONE, NEARFULL, BACKFILLFULL, FULL, FAILSAFE } cur_state;  // ascending
  const char *get_full_state_name(s_names s) const {
    switch (s) {
    case NONE: return "none";
    case NEARFULL: return "nearfull";
    case BACKFILLFULL: return "backfillfull";
    case FULL: return "full";
    case FAILSAFE: return "failsafe";
    default: return "???";
    }
  }
  s_names get_full_state(string type) const {
    if (type == "none")
      return NONE;
    else if (type == "failsafe")
      return FAILSAFE;
    else if (type == "full")
      return FULL;
    else if (type == "backfillfull")
      return BACKFILLFULL;
    else if (type == "nearfull")
      return NEARFULL;
    else
      return INVALID;
  }
  double cur_ratio, physical_ratio;  ///< current utilization
  mutable int64_t injectfull = 0;
  s_names injectfull_state = NONE;
  float get_failsafe_full_ratio();
  bool _check_inject_full(DoutPrefixProvider *dpp, s_names type) const;
  bool _check_full(DoutPrefixProvider *dpp, s_names type) const;
public:
  void check_full_status(float ratio, float pratio);
  s_names recalc_full_state(float ratio, float pratio, string &inject);
  bool _tentative_full(DoutPrefixProvider *dpp, s_names type, uint64_t adjust_used, osd_stat_t);
  bool check_failsafe_full(DoutPrefixProvider *dpp) const;
  bool check_full(DoutPrefixProvider *dpp) const;
  bool tentative_backfill_full(DoutPrefixProvider *dpp, uint64_t adjust_used, osd_stat_t);
  bool check_backfill_full(DoutPrefixProvider *dpp) const;
  bool check_nearfull(DoutPrefixProvider *dpp) const;
  bool is_failsafe_full() const;
  bool is_full() const;
  bool is_backfillfull() const;
  bool is_nearfull() const;
  bool need_fullness_update();  ///< osdmap state needs update
  void set_injectfull(s_names type, int64_t count);
  bool check_osdmap_full(const set<pg_shard_t> &missing_on);


  // -- epochs --
private:
  mutable Mutex epoch_lock; // protects access to boot_epoch, up_epoch, bind_epoch
  epoch_t boot_epoch;  // _first_ epoch we were marked up (after this process started)
  epoch_t up_epoch;    // _most_recent_ epoch we were marked up
  epoch_t bind_epoch;  // epoch we last did a bind to new ip:ports
public:
  /**
   * Retrieve the boot_, up_, and bind_ epochs the OSD has set. The params
   * can be NULL if you don't care about them.
   */
  void retrieve_epochs(epoch_t *_boot_epoch, epoch_t *_up_epoch,
                       epoch_t *_bind_epoch) const;
  /**
   * Set the boot, up, and bind epochs. Any NULL params will not be set.
   */
  void set_epochs(const epoch_t *_boot_epoch, const epoch_t *_up_epoch,
                  const epoch_t *_bind_epoch);
  epoch_t get_boot_epoch() const {
    epoch_t ret;
    retrieve_epochs(&ret, NULL, NULL);
    return ret;
  }
  epoch_t get_up_epoch() const {
    epoch_t ret;
    retrieve_epochs(NULL, &ret, NULL);
    return ret;
  }
  epoch_t get_bind_epoch() const {
    epoch_t ret;
    retrieve_epochs(NULL, NULL, &ret);
    return ret;
  }

  void request_osdmap_update(epoch_t e);

  // -- stopping --
  Mutex is_stopping_lock;
  Cond is_stopping_cond;
  enum {
    NOT_STOPPING,
    PREPARING_TO_STOP,
    STOPPING };
  std::atomic<int> state{NOT_STOPPING};
  int get_state() const {
    return state;
  }
  void set_state(int s) {
    state = s;
  }
  bool is_stopping() const {
    return state == STOPPING;
  }
  bool is_preparing_to_stop() const {
    return state == PREPARING_TO_STOP;
  }
  bool prepare_to_stop();
  void got_stop_ack();


#ifdef PG_DEBUG_REFS
  Mutex pgid_lock;
  map<spg_t, int> pgid_tracker;
  map<spg_t, PG*> live_pgs;
  void add_pgid(spg_t pgid, PG *pg);
  void remove_pgid(spg_t pgid, PG *pg);
  void dump_live_pgids();
#endif

  explicit OSDService(OSD *osd);
  ~OSDService();
};


enum class io_queue {
  prioritized,
  weightedpriority,
  mclock_opclass,
  mclock_client,
};


/*

  Each PG slot includes queues for events that are processing and/or waiting
  for a PG to be materialized in the slot.

  These are the constraints:

  - client ops must remained ordered by client, regardless of map epoch
  - peering messages/events from peers must remain ordered by peer
  - peering messages and client ops need not be ordered relative to each other

  - some peering events can create a pg (e.g., notify)
  - the query peering event can proceed when a PG doesn't exist

  Implementation notes:

  - everybody waits for split.  If the OSD has the parent PG it will instantiate
    the PGSlot early and mark it waiting_for_split.  Everything will wait until
    the parent is able to commit the split operation and the child PG's are
    materialized in the child slots.

  - every event has an epoch property and will wait for the OSDShard to catch
    up to that epoch.  For example, if we get a peering event from a future
    epoch, the event will wait in the slot until the local OSD has caught up.
    (We should be judicious in specifying the required epoch [by, e.g., setting
    it to the same_interval_since epoch] so that we don't wait for epochs that
    don't affect the given PG.)

  - we maintain two separate wait lists, *waiting* and *waiting_peering*. The
    OpQueueItem has an is_peering() bool to determine which we use.  Waiting
    peering events are queued up by epoch required.

  - when we wake a PG slot (e.g., we finished split, or got a newer osdmap, or
    materialized the PG), we wake *all* waiting items.  (This could be optimized,
    probably, but we don't bother.)  We always requeue peering items ahead of
    client ops.

  - some peering events are marked !peering_requires_pg (PGQuery).  if we do
    not have a PG these are processed immediately (under the shard lock).

  - we do not have a PG present, we check if the slot maps to the current host.
    if so, we either queue the item and wait for the PG to materialize, or
    (if the event is a pg creating event like PGNotify), we materialize the PG.

  - when we advance the osdmap on the OSDShard, we scan pg slots and
    discard any slots with no pg (and not waiting_for_split) that no
    longer map to the current host.

  */

struct OSDShardPGSlot {
  PGRef pg;                      ///< pg reference
  deque<OpQueueItem> to_process; ///< order items for this slot
  int num_running = 0;          ///< _process threads doing pg lookup/lock

  deque<OpQueueItem> waiting;   ///< waiting for pg (or map + pg)

  /// waiting for map (peering evt)
  map<epoch_t,deque<OpQueueItem>> waiting_peering;

  /// incremented by wake_pg_waiters; indicates racing _process threads
  /// should bail out (their op has been requeued)
  uint64_t requeue_seq = 0;

  /// waiting for split child to materialize in these epoch(s)
  set<epoch_t> waiting_for_split;

  epoch_t epoch = 0;
  boost::intrusive::set_member_hook<> pg_epoch_item;

  /// waiting for a merge (source or target) by this epoch
  epoch_t waiting_for_merge_epoch = 0;
};

struct OSDShard {
  const unsigned shard_id;
  CephContext *cct;
  OSD *osd;

  string shard_name;

  string sdata_wait_lock_name;
  ceph::mutex sdata_wait_lock;
  ceph::condition_variable sdata_cond;

  string osdmap_lock_name;
  ceph::mutex osdmap_lock;  ///< protect shard_osdmap updates vs users w/o shard_lock
  OSDMapRef shard_osdmap;

  OSDMapRef get_osdmap() {
    std::lock_guard l(osdmap_lock);
    return shard_osdmap;
  }

  string shard_lock_name;
  ceph::mutex shard_lock;   ///< protects remaining members below

  /// map of slots for each spg_t.  maintains ordering of items dequeued
  /// from pqueue while _process thread drops shard lock to acquire the
  /// pg lock.  stale slots are removed by consume_map.
  unordered_map<spg_t,unique_ptr<OSDShardPGSlot>> pg_slots;

  struct pg_slot_compare_by_epoch {
    bool operator()(const OSDShardPGSlot& l, const OSDShardPGSlot& r) const {
      return l.epoch < r.epoch;
    }
  };

  /// maintain an ordering of pg slots by pg epoch
  boost::intrusive::multiset<
    OSDShardPGSlot,
    boost::intrusive::member_hook<
      OSDShardPGSlot,
      boost::intrusive::set_member_hook<>,
      &OSDShardPGSlot::pg_epoch_item>,
    boost::intrusive::compare<pg_slot_compare_by_epoch>> pg_slots_by_epoch;
  int waiting_for_min_pg_epoch = 0;
  ceph::condition_variable min_pg_epoch_cond;

  /// priority queue
  std::unique_ptr<OpQueue<OpQueueItem, uint64_t>> pqueue;

  bool stop_waiting = false;

  ContextQueue context_queue;

  void _enqueue_front(OpQueueItem&& item, unsigned cutoff) {
    unsigned priority = item.get_priority();
    unsigned cost = item.get_cost();
    if (priority >= cutoff)
      pqueue->enqueue_strict_front(
	item.get_owner(),
	priority, std::move(item));
    else
      pqueue->enqueue_front(
	item.get_owner(),
	priority, cost, std::move(item));
  }

  void _attach_pg(OSDShardPGSlot *slot, PG *pg);
  void _detach_pg(OSDShardPGSlot *slot);

  void update_pg_epoch(OSDShardPGSlot *slot, epoch_t epoch);
  epoch_t get_min_pg_epoch();
  void wait_min_pg_epoch(epoch_t need);

  /// return newest epoch we are waiting for
  epoch_t get_max_waiting_epoch();

  /// push osdmap into shard
  void consume_map(
    const OSDMapRef& osdmap,
    unsigned *pushes_to_free);

  void _wake_pg_slot(spg_t pgid, OSDShardPGSlot *slot);

  void identify_splits_and_merges(
    const OSDMapRef& as_of_osdmap,
    set<pair<spg_t,epoch_t>> *split_children,
    set<pair<spg_t,epoch_t>> *merge_pgs);
  void _prime_splits(set<pair<spg_t,epoch_t>> *pgids);
  void prime_splits(const OSDMapRef& as_of_osdmap,
		    set<pair<spg_t,epoch_t>> *pgids);
  void prime_merges(const OSDMapRef& as_of_osdmap,
		    set<pair<spg_t,epoch_t>> *merge_pgs);
  void register_and_wake_split_child(PG *pg);
  void unprime_split_children(spg_t parent, unsigned old_pg_num);

  OSDShard(
    int id,
    CephContext *cct,
    OSD *osd,
    uint64_t max_tok_per_prio, uint64_t min_cost,
    io_queue opqueue)
    : shard_id(id),
      cct(cct),
      osd(osd),
      shard_name(string("OSDShard.") + stringify(id)),
      sdata_wait_lock_name(shard_name + "::sdata_wait_lock"),
      sdata_wait_lock{make_mutex(sdata_wait_lock_name)},
      osdmap_lock_name(shard_name + "::osdmap_lock"),
      osdmap_lock{make_mutex(osdmap_lock_name)},
      shard_lock_name(shard_name + "::shard_lock"),
      shard_lock{make_mutex(shard_lock_name)},
      context_queue(sdata_wait_lock, sdata_cond) {
    if (opqueue == io_queue::weightedpriority) {
      pqueue = std::make_unique<
	WeightedPriorityQueue<OpQueueItem,uint64_t>>(
	  max_tok_per_prio, min_cost);
    } else if (opqueue == io_queue::prioritized) {
      pqueue = std::make_unique<
	PrioritizedQueue<OpQueueItem,uint64_t>>(
	  max_tok_per_prio, min_cost);
    } else if (opqueue == io_queue::mclock_opclass) {
      pqueue = std::make_unique<ceph::mClockOpClassQueue>(cct);
    } else if (opqueue == io_queue::mclock_client) {
      pqueue = std::make_unique<ceph::mClockClientQueue>(cct);
    }
  }
};

class OSD : public Dispatcher,
	    public md_config_obs_t {
  /** OSD **/
  Mutex osd_lock;          // global lock
  SafeTimer tick_timer;    // safe timer (osd_lock)

  // Tick timer for those stuff that do not need osd_lock
  Mutex tick_timer_lock;
  SafeTimer tick_timer_without_osd_lock;
  std::string gss_ktfile_client{};

public:
  // config observer bits
  const char** get_tracked_conf_keys() const override;
  void handle_conf_change(const ConfigProxy& conf,
                          const std::set <std::string> &changed) override;
  void update_log_config();
  void check_config();

protected:

  const double OSD_TICK_INTERVAL = { 1.0 };
  double get_tick_interval() const;

  Messenger   *cluster_messenger;
  Messenger   *client_messenger;
  Messenger   *objecter_messenger;
  MonClient   *monc; // check the "monc helpers" list before accessing directly
  MgrClient   mgrc;
  PerfCounters      *logger;
  PerfCounters      *recoverystate_perf;
  ObjectStore *store;
#ifdef HAVE_LIBFUSE
  FuseStore *fuse_store = nullptr;
#endif
  LogClient log_client;
  LogChannelRef clog;

  int whoami;
  std::string dev_path, journal_path;

  int last_require_osd_release = 0;

  int numa_node = -1;
  size_t numa_cpu_set_size = 0;
  cpu_set_t numa_cpu_set;

  bool store_is_rotational = true;
  bool journal_is_rotational = true;

  ZTracer::Endpoint trace_endpoint;
  void create_logger();
  void create_recoverystate_perf();
  void tick();
  void tick_without_osd_lock();
  void _dispatch(Message *m);
  void dispatch_op(OpRequestRef op);

  void check_osdmap_features();

  // asok
  friend class OSDSocketHook;
  class OSDSocketHook *asok_hook;
  bool asok_command(std::string_view admin_command, const cmdmap_t& cmdmap,
		    std::string_view format, std::ostream& ss);

public:
  ClassHandler  *class_handler = nullptr;
  int get_nodeid() { return whoami; }
  
  static ghobject_t get_osdmap_pobject_name(epoch_t epoch) {
    char foo[20];
    snprintf(foo, sizeof(foo), "osdmap.%d", epoch);
    return ghobject_t(hobject_t(sobject_t(object_t(foo), 0)));
  }
  static ghobject_t get_inc_osdmap_pobject_name(epoch_t epoch) {
    char foo[22];
    snprintf(foo, sizeof(foo), "inc_osdmap.%d", epoch);
    return ghobject_t(hobject_t(sobject_t(object_t(foo), 0)));
  }

  static ghobject_t make_snapmapper_oid() {
    return ghobject_t(hobject_t(
      sobject_t(
	object_t("snapmapper"),
	0)));
  }

  static ghobject_t make_pg_log_oid(spg_t pg) {
    stringstream ss;
    ss << "pglog_" << pg;
    string s;
    getline(ss, s);
    return ghobject_t(hobject_t(sobject_t(object_t(s.c_str()), 0)));
  }
  
  static ghobject_t make_pg_biginfo_oid(spg_t pg) {
    stringstream ss;
    ss << "pginfo_" << pg;
    string s;
    getline(ss, s);
    return ghobject_t(hobject_t(sobject_t(object_t(s.c_str()), 0)));
  }
  static ghobject_t make_infos_oid() {
    hobject_t oid(sobject_t("infos", CEPH_NOSNAP));
    return ghobject_t(oid);
  }

  static ghobject_t make_final_pool_info_oid(int64_t pool) {
    return ghobject_t(
      hobject_t(
	sobject_t(
	  object_t(string("final_pool_") + stringify(pool)),
	  CEPH_NOSNAP)));
  }

  static ghobject_t make_pg_num_history_oid() {
    return ghobject_t(hobject_t(sobject_t("pg_num_history", CEPH_NOSNAP)));
  }

  static void recursive_remove_collection(CephContext* cct,
					  ObjectStore *store,
					  spg_t pgid,
					  coll_t tmp);

  /**
   * get_osd_initial_compat_set()
   *
   * Get the initial feature set for this OSD.  Features
   * here are automatically upgraded.
   *
   * Return value: Initial osd CompatSet
   */
  static CompatSet get_osd_initial_compat_set();

  /**
   * get_osd_compat_set()
   *
   * Get all features supported by this OSD
   *
   * Return value: CompatSet of all supported features
   */
  static CompatSet get_osd_compat_set();
  

private:
  class C_Tick;
  class C_Tick_WithoutOSDLock;

  // -- config settings --
  float m_osd_pg_epoch_max_lag_factor;

  // -- superblock --
  OSDSuperblock superblock;

  void write_superblock();
  void write_superblock(ObjectStore::Transaction& t);
  int read_superblock();

  void clear_temp_objects();

  CompatSet osd_compat;

  // -- state --
public:
  typedef enum {
    STATE_INITIALIZING = 1,
    STATE_PREBOOT,
    STATE_BOOTING,
    STATE_ACTIVE,
    STATE_STOPPING,
    STATE_WAITING_FOR_HEALTHY
  } osd_state_t;

  static const char *get_state_name(int s) {
    switch (s) {
    case STATE_INITIALIZING: return "initializing";
    case STATE_PREBOOT: return "preboot";
    case STATE_BOOTING: return "booting";
    case STATE_ACTIVE: return "active";
    case STATE_STOPPING: return "stopping";
    case STATE_WAITING_FOR_HEALTHY: return "waiting_for_healthy";
    default: return "???";
    }
  }

private:
  std::atomic<int> state{STATE_INITIALIZING};

public:
  int get_state() const {
    return state;
  }
  void set_state(int s) {
    state = s;
  }
  bool is_initializing() const {
    return state == STATE_INITIALIZING;
  }
  bool is_preboot() const {
    return state == STATE_PREBOOT;
  }
  bool is_booting() const {
    return state == STATE_BOOTING;
  }
  bool is_active() const {
    return state == STATE_ACTIVE;
  }
  bool is_stopping() const {
    return state == STATE_STOPPING;
  }
  bool is_waiting_for_healthy() const {
    return state == STATE_WAITING_FOR_HEALTHY;
  }

private:

  ShardedThreadPool osd_op_tp;
  ThreadPool command_tp;

  void get_latest_osdmap();

  // -- sessions --
private:
  void dispatch_session_waiting(SessionRef session, OSDMapRef osdmap);
  void maybe_share_map(Session *session, OpRequestRef op, OSDMapRef osdmap);

  Mutex session_waiting_lock;
  set<SessionRef> session_waiting_for_map;

  /// Caller assumes refs for included Sessions
  void get_sessions_waiting_for_map(set<SessionRef> *out) {
    std::lock_guard l(session_waiting_lock);
    out->swap(session_waiting_for_map);
  }
  void register_session_waiting_on_map(SessionRef session) {
    std::lock_guard l(session_waiting_lock);
    session_waiting_for_map.insert(session);
  }
  void clear_session_waiting_on_map(SessionRef session) {
    std::lock_guard l(session_waiting_lock);
    session_waiting_for_map.erase(session);
  }
  void dispatch_sessions_waiting_on_map() {
    set<SessionRef> sessions_to_check;
    get_sessions_waiting_for_map(&sessions_to_check);
    for (auto i = sessions_to_check.begin();
	 i != sessions_to_check.end();
	 sessions_to_check.erase(i++)) {
      std::lock_guard l{(*i)->session_dispatch_lock};
      dispatch_session_waiting(*i, get_osdmap());
    }
  }
  void session_handle_reset(SessionRef session) {
    std::lock_guard l(session->session_dispatch_lock);
    clear_session_waiting_on_map(session);

    session->clear_backoffs();

    /* Messages have connection refs, we need to clear the
     * connection->session->message->connection
     * cycles which result.
     * Bug #12338
     */
    session->waiting_on_map.clear_and_dispose(TrackedOp::Putter());
  }

private:
  /**
   * @defgroup monc helpers
   * @{
   * Right now we only have the one
   */

  /**
   * Ask the Monitors for a sequence of OSDMaps.
   *
   * @param epoch The epoch to start with when replying
   * @param force_request True if this request forces a new subscription to
   * the monitors; false if an outstanding request that encompasses it is
   * sufficient.
   */
  void osdmap_subscribe(version_t epoch, bool force_request);
  /** @} monc helpers */

  Mutex osdmap_subscribe_lock;
  epoch_t latest_subscribed_epoch{0};

  // -- heartbeat --
  /// information about a heartbeat peer
  struct HeartbeatInfo {
    int peer;           ///< peer
    ConnectionRef con_front;   ///< peer connection (front)
    ConnectionRef con_back;    ///< peer connection (back)
    utime_t first_tx;   ///< time we sent our first ping request
    utime_t last_tx;    ///< last time we sent a ping request
    utime_t last_rx_front;  ///< last time we got a ping reply on the front side
    utime_t last_rx_back;   ///< last time we got a ping reply on the back side
    epoch_t epoch;      ///< most recent epoch we wanted this peer
    /// number of connections we send and receive heartbeat pings/replies
    static constexpr int HEARTBEAT_MAX_CONN = 2;
    /// history of inflight pings, arranging by timestamp we sent
    /// send time -> deadline -> remaining replies
    map<utime_t, pair<utime_t, int>> ping_history;

    utime_t hb_interval_start;
    uint32_t hb_average_count = 0;
    uint32_t hb_index = 0;

    uint32_t hb_total_back = 0;
    uint32_t hb_min_back = UINT_MAX;
    uint32_t hb_max_back = 0;
    vector<uint32_t> hb_back_pingtime;
    vector<uint32_t> hb_back_min;
    vector<uint32_t> hb_back_max;

    uint32_t hb_total_front = 0;
    uint32_t hb_min_front = UINT_MAX;
    uint32_t hb_max_front = 0;
    vector<uint32_t> hb_front_pingtime;
    vector<uint32_t> hb_front_min;
    vector<uint32_t> hb_front_max;

    bool is_stale(utime_t stale) {
      if (ping_history.empty()) {
        return false;
      }
      utime_t oldest_deadline = ping_history.begin()->second.first;
      return oldest_deadline <= stale;
    }

    bool is_unhealthy(utime_t now) {
      if (ping_history.empty()) {
        /// we haven't sent a ping yet or we have got all replies,
        /// in either way we are safe and healthy for now
        return false;
      }

      utime_t oldest_deadline = ping_history.begin()->second.first;
      return now > oldest_deadline;
    }

    bool is_healthy(utime_t now) {
      if (last_rx_front == utime_t() || last_rx_back == utime_t()) {
        // only declare to be healthy until we have received the first
        // replies from both front/back connections
        return false;
      }
      return !is_unhealthy(now);
    }
  };
  /// state attached to outgoing heartbeat connections
  struct HeartbeatSession : public RefCountedObject {
    int peer;
    explicit HeartbeatSession(int p) : peer(p) {}
  };
  Mutex heartbeat_lock;
  map<int, int> debug_heartbeat_drops_remaining;
  Cond heartbeat_cond;
  bool heartbeat_stop;
  std::atomic<bool> heartbeat_need_update;   
  map<int,HeartbeatInfo> heartbeat_peers;  ///< map of osd id to HeartbeatInfo
  utime_t last_mon_heartbeat;
  Messenger *hb_front_client_messenger;
  Messenger *hb_back_client_messenger;
  Messenger *hb_front_server_messenger;
  Messenger *hb_back_server_messenger;
  utime_t last_heartbeat_resample;   ///< last time we chose random peers in waiting-for-healthy state
  double daily_loadavg;

  // Track ping repsonse times using vector as a circular buffer
  // MUST BE A POWER OF 2
  const uint32_t hb_vector_size = 16;

  void _add_heartbeat_peer(int p);
  void _remove_heartbeat_peer(int p);
  bool heartbeat_reset(Connection *con);
  void maybe_update_heartbeat_peers();
  void reset_heartbeat_peers(bool all);
  bool heartbeat_peers_need_update() {
    return heartbeat_need_update.load();
  }
  void heartbeat_set_peers_need_update() {
    heartbeat_need_update.store(true);
  }
  void heartbeat_clear_peers_need_update() {
    heartbeat_need_update.store(false);
  }
  void heartbeat();
  void heartbeat_check();
  void heartbeat_entry();
  void need_heartbeat_peer_update();

  void heartbeat_kick() {
    std::lock_guard l(heartbeat_lock);
    heartbeat_cond.Signal();
  }

  struct T_Heartbeat : public Thread {
    OSD *osd;
    explicit T_Heartbeat(OSD *o) : osd(o) {}
    void *entry() override {
      osd->heartbeat_entry();
      return 0;
    }
  } heartbeat_thread;

public:
  bool heartbeat_dispatch(Message *m);

  struct HeartbeatDispatcher : public Dispatcher {
    OSD *osd;
    explicit HeartbeatDispatcher(OSD *o) : Dispatcher(o->cct), osd(o) {}

    bool ms_can_fast_dispatch_any() const override { return true; }
    bool ms_can_fast_dispatch(const Message *m) const override {
      switch (m->get_type()) {
      case CEPH_MSG_PING:
      case MSG_OSD_PING:
	return true;
      default:
	return false;
      }
    }
    void ms_fast_dispatch(Message *m) override {
      osd->heartbeat_dispatch(m);
    }
    bool ms_dispatch(Message *m) override {
      return osd->heartbeat_dispatch(m);
    }
    bool ms_handle_reset(Connection *con) override {
      return osd->heartbeat_reset(con);
    }
    void ms_handle_remote_reset(Connection *con) override {}
    bool ms_handle_refused(Connection *con) override {
      return osd->ms_handle_refused(con);
    }
    int ms_handle_authentication(Connection *con) override {
      return true;
    }
    bool ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer) override {
      // some pre-nautilus OSDs get confused if you include an
      // authorizer but they are not expecting it.  do not try to authorize
      // heartbeat connections until all OSDs are nautilus.
      if (osd->get_osdmap()->require_osd_release >= CEPH_RELEASE_NAUTILUS) {
	return osd->ms_get_authorizer(dest_type, authorizer);
      }
      return false;
    }
    KeyStore *ms_get_auth1_authorizer_keystore() override {
      return osd->ms_get_auth1_authorizer_keystore();
    }
  } heartbeat_dispatcher;

private:
  // -- waiters --
  list<OpRequestRef> finished;
  
  void take_waiters(list<OpRequestRef>& ls) {
    ceph_assert(osd_lock.is_locked());
    finished.splice(finished.end(), ls);
  }
  void do_waiters();
  
  // -- op tracking --
  OpTracker op_tracker;
  void test_ops(std::string command, std::string args, ostream& ss);
  friend class TestOpsSocketHook;
  TestOpsSocketHook *test_ops_hook;
  friend struct C_FinishSplits;
  friend struct C_OpenPGs;

  // -- op queue --
  friend std::ostream& operator<<(std::ostream& out, const io_queue& q);

  const io_queue op_queue;
public:
  const unsigned int op_prio_cutoff;
protected:

  /*
   * The ordered op delivery chain is:
   *
   *   fast dispatch -> pqueue back
   *                    pqueue front <-> to_process back
   *                                     to_process front  -> RunVis(item)
   *                                                      <- queue_front()
   *
   * The pqueue is per-shard, and to_process is per pg_slot.  Items can be
   * pushed back up into to_process and/or pqueue while order is preserved.
   *
   * Multiple worker threads can operate on each shard.
   *
   * Under normal circumstances, num_running == to_process.size().  There are
   * two times when that is not true: (1) when waiting_for_pg == true and
   * to_process is accumulating requests that are waiting for the pg to be
   * instantiated; in that case they will all get requeued together by
   * wake_pg_waiters, and (2) when wake_pg_waiters just ran, waiting_for_pg
   * and already requeued the items.
   */
  friend class PGOpItem;
  friend class PGPeeringItem;
  friend class PGRecovery;
  friend class PGDelete;

  class ShardedOpWQ
    : public ShardedThreadPool::ShardedWQ<OpQueueItem>
  {
    OSD *osd;

  public:
    ShardedOpWQ(OSD *o,
		time_t ti,
		time_t si,
		ShardedThreadPool* tp)
      : ShardedThreadPool::ShardedWQ<OpQueueItem>(ti, si, tp),
        osd(o) {
    }

    void _add_slot_waiter(
      spg_t token,
      OSDShardPGSlot *slot,
      OpQueueItem&& qi);

    /// try to do some work
    void _process(uint32_t thread_index, heartbeat_handle_d *hb) override;

    /// enqueue a new item
    void _enqueue(OpQueueItem&& item) override;

    /// requeue an old item (at the front of the line)
    void _enqueue_front(OpQueueItem&& item) override;
      
    void return_waiting_threads() override {
      for(uint32_t i = 0; i < osd->num_shards; i++) {
	OSDShard* sdata = osd->shards[i];
	assert (NULL != sdata);
	std::scoped_lock l{sdata->sdata_wait_lock};
	sdata->stop_waiting = true;
	sdata->sdata_cond.notify_all();
      }
    }

    void stop_return_waiting_threads() override {
      for(uint32_t i = 0; i < osd->num_shards; i++) {
	OSDShard* sdata = osd->shards[i];
	assert (NULL != sdata);
	std::scoped_lock l{sdata->sdata_wait_lock};
	sdata->stop_waiting = false;
      }
    }

    void dump(Formatter *f) {
      for(uint32_t i = 0; i < osd->num_shards; i++) {
	auto &&sdata = osd->shards[i];

	char queue_name[32] = {0};
	snprintf(queue_name, sizeof(queue_name), "%s%" PRIu32, "OSD:ShardedOpWQ:", i);
	ceph_assert(NULL != sdata);

	std::scoped_lock l{sdata->shard_lock};
	f->open_object_section(queue_name);
	sdata->pqueue->dump(f);
	f->close_section();
      }
    }

    bool is_shard_empty(uint32_t thread_index) override {
      uint32_t shard_index = thread_index % osd->num_shards;
      auto &&sdata = osd->shards[shard_index];
      ceph_assert(sdata);
      std::lock_guard l(sdata->shard_lock);
      if (thread_index < osd->num_shards) {
	return sdata->pqueue->empty() && sdata->context_queue.empty();
      } else {
	return sdata->pqueue->empty();
      }
    }

    void handle_oncommits(list<Context*>& oncommits) {
      for (auto p : oncommits) {
	p->complete(0);
      }
    }
  } op_shardedwq;


  void enqueue_op(spg_t pg, OpRequestRef&& op, epoch_t epoch);
  void dequeue_op(
    PGRef pg, OpRequestRef op,
    ThreadPool::TPHandle &handle);

  void enqueue_peering_evt(
    spg_t pgid,
    PGPeeringEventRef ref);
  void enqueue_peering_evt_front(
    spg_t pgid,
    PGPeeringEventRef ref);
  void dequeue_peering_evt(
    OSDShard *sdata,
    PG *pg,
    PGPeeringEventRef ref,
    ThreadPool::TPHandle& handle);

  void dequeue_delete(
    OSDShard *sdata,
    PG *pg,
    epoch_t epoch,
    ThreadPool::TPHandle& handle);

  friend class PG;
  friend class OSDShard;
  friend class PrimaryLogPG;


 protected:

  // -- osd map --
  // TODO: switch to std::atomic<OSDMapRef> when C++20 will be available.
  OSDMapRef       _osdmap;
  void set_osdmap(OSDMapRef osdmap) {
    std::atomic_store(&_osdmap, osdmap);
  }
  OSDMapRef get_osdmap() const {
    return std::atomic_load(&_osdmap);
  }
  epoch_t get_osdmap_epoch() const {
    // XXX: performance?
    auto osdmap = get_osdmap();
    return osdmap ? osdmap->get_epoch() : 0;
  }

  pool_pg_num_history_t pg_num_history;

  utime_t         had_map_since;
  RWLock          map_lock;
  list<OpRequestRef>  waiting_for_osdmap;
  deque<utime_t> osd_markdown_log;

  friend struct send_map_on_destruct;

  void wait_for_new_map(OpRequestRef op);
  void handle_osd_map(class MOSDMap *m);
  void _committed_osd_maps(epoch_t first, epoch_t last, class MOSDMap *m);
  void trim_maps(epoch_t oldest, int nreceived, bool skip_maps);
  void note_down_osd(int osd);
  void note_up_osd(int osd);
  friend class C_OnMapCommit;

  bool advance_pg(
    epoch_t advance_to,
    PG *pg,
    ThreadPool::TPHandle &handle,
    PG::RecoveryCtx *rctx);
  void consume_map();
  void activate_map();

  // osd map cache (past osd maps)
  OSDMapRef get_map(epoch_t e) {
    return service.get_map(e);
  }
  OSDMapRef add_map(OSDMap *o) {
    return service.add_map(o);
  }
  void add_map_bl(epoch_t e, bufferlist& bl) {
    return service.add_map_bl(e, bl);
  }
  bool get_map_bl(epoch_t e, bufferlist& bl) {
    return service.get_map_bl(e, bl);
  }
  void add_map_inc_bl(epoch_t e, bufferlist& bl) {
    return service.add_map_inc_bl(e, bl);
  }

public:
  // -- shards --
  vector<OSDShard*> shards;
  uint32_t num_shards = 0;

  void inc_num_pgs() {
    ++num_pgs;
  }
  void dec_num_pgs() {
    --num_pgs;
  }
  int get_num_pgs() const {
    return num_pgs;
  }

protected:
  Mutex merge_lock = {"OSD::merge_lock"};
  /// merge epoch -> target pgid -> source pgid -> pg
  map<epoch_t,map<spg_t,map<spg_t,PGRef>>> merge_waiters;

  bool add_merge_waiter(OSDMapRef nextmap, spg_t target, PGRef source,
			unsigned need);

  // -- placement groups --
  std::atomic<size_t> num_pgs = {0};

  std::mutex pending_creates_lock;
  using create_from_osd_t = std::pair<pg_t, bool /* is primary*/>;
  std::set<create_from_osd_t> pending_creates_from_osd;
  unsigned pending_creates_from_mon = 0;

  PGRecoveryStats pg_recovery_stats;

  PGRef _lookup_pg(spg_t pgid);
  PGRef _lookup_lock_pg(spg_t pgid);
  void register_pg(PGRef pg);
  bool try_finish_pg_delete(PG *pg, unsigned old_pg_num);

  void _get_pgs(vector<PGRef> *v, bool clear_too=false);
  void _get_pgids(vector<spg_t> *v);

public:
  PGRef lookup_lock_pg(spg_t pgid);

  std::set<int64_t> get_mapped_pools();

protected:
  PG* _make_pg(OSDMapRef createmap, spg_t pgid);

  bool maybe_wait_for_max_pg(const OSDMapRef& osdmap,
			     spg_t pgid, bool is_mon_create);
  void resume_creating_pg();

  void load_pgs();

  /// build initial pg history and intervals on create
  void build_initial_pg_history(
    spg_t pgid,
    epoch_t created,
    utime_t created_stamp,
    pg_history_t *h,
    PastIntervals *pi);

  epoch_t last_pg_create_epoch;

  void handle_pg_create(OpRequestRef op);

  void split_pgs(
    PG *parent,
    const set<spg_t> &childpgids, set<PGRef> *out_pgs,
    OSDMapRef curmap,
    OSDMapRef nextmap,
    PG::RecoveryCtx *rctx);
  void _finish_splits(set<PGRef>& pgs);

  // == monitor interaction ==
  Mutex mon_report_lock;
  utime_t last_mon_report;
  Finisher boot_finisher;

  // -- boot --
  void start_boot();
  void _got_mon_epochs(epoch_t oldest, epoch_t newest);
  void _preboot(epoch_t oldest, epoch_t newest);
  void _send_boot();
  void _collect_metadata(map<string,string> *pmeta);

  void start_waiting_for_healthy();
  bool _is_healthy();

  void send_full_update();
  
  friend struct C_OSD_GetVersion;

  // -- alive --
  epoch_t up_thru_wanted;

  void queue_want_up_thru(epoch_t want);
  void send_alive();

  // -- full map requests --
  epoch_t requested_full_first, requested_full_last;

  void request_full_map(epoch_t first, epoch_t last);
  void rerequest_full_maps() {
    epoch_t first = requested_full_first;
    epoch_t last = requested_full_last;
    requested_full_first = 0;
    requested_full_last = 0;
    request_full_map(first, last);
  }
  void got_full_map(epoch_t e);

  // -- failures --
  map<int,utime_t> failure_queue;
  map<int,pair<utime_t,entity_addrvec_t> > failure_pending;

  void requeue_failures();
  void send_failures();
  void send_still_alive(epoch_t epoch, int osd, const entity_addrvec_t &addrs);
  void cancel_pending_failures();

  ceph::coarse_mono_clock::time_point last_sent_beacon;
  Mutex min_last_epoch_clean_lock{"OSD::min_last_epoch_clean_lock"};
  epoch_t min_last_epoch_clean = 0;
  // which pgs were scanned for min_lec
  std::vector<pg_t> min_last_epoch_clean_pgs;
  void send_beacon(const ceph::coarse_mono_clock::time_point& now);

  ceph_tid_t get_tid() {
    return service.get_tid();
  }

  // -- generic pg peering --
  PG::RecoveryCtx create_context();
  void dispatch_context(PG::RecoveryCtx &ctx, PG *pg, OSDMapRef curmap,
                        ThreadPool::TPHandle *handle = NULL);
  void dispatch_context_transaction(PG::RecoveryCtx &ctx, PG *pg,
                                    ThreadPool::TPHandle *handle = NULL);
  void discard_context(PG::RecoveryCtx &ctx);
  void do_notifies(map<int,
		       vector<pair<pg_notify_t, PastIntervals> > >&
		       notify_list,
		   OSDMapRef map);
  void do_queries(map<int, map<spg_t,pg_query_t> >& query_map,
		  OSDMapRef map);
  void do_infos(map<int,
		    vector<pair<pg_notify_t, PastIntervals> > >& info_map,
		OSDMapRef map);

  bool require_mon_peer(const Message *m);
  bool require_mon_or_mgr_peer(const Message *m);
  bool require_osd_peer(const Message *m);
  /***
   * Verifies that we were alive in the given epoch, and that
   * still are.
   */
  bool require_self_aliveness(const Message *m, epoch_t alive_since);
  /**
   * Verifies that the OSD who sent the given op has the same
   * address as in the given map.
   * @pre op was sent by an OSD using the cluster messenger
   */
  bool require_same_peer_instance(const Message *m, const OSDMapRef& map,
				  bool is_fast_dispatch);

  bool require_same_or_newer_map(OpRequestRef& op, epoch_t e,
				 bool is_fast_dispatch);

  void handle_fast_pg_create(MOSDPGCreate2 *m);
  void handle_fast_pg_query(MOSDPGQuery *m);
  void handle_pg_query_nopg(const MQuery& q);
  void handle_fast_pg_notify(MOSDPGNotify *m);
  void handle_pg_notify_nopg(const MNotifyRec& q);
  void handle_fast_pg_info(MOSDPGInfo *m);
  void handle_fast_pg_remove(MOSDPGRemove *m);

public:
  // used by OSDShard
  PGRef handle_pg_create_info(const OSDMapRef& osdmap, const PGCreateInfo *info);
protected:

  void handle_fast_force_recovery(MOSDForceRecovery *m);

  // -- commands --
  struct Command {
    vector<string> cmd;
    ceph_tid_t tid;
    bufferlist indata;
    ConnectionRef con;

    Command(vector<string>& c, ceph_tid_t t, bufferlist& bl, Connection *co)
      : cmd(c), tid(t), indata(bl), con(co) {}
  };
  list<Command*> command_queue;
  struct CommandWQ : public ThreadPool::WorkQueue<Command> {
    OSD *osd;
    CommandWQ(OSD *o, time_t ti, time_t si, ThreadPool *tp)
      : ThreadPool::WorkQueue<Command>("OSD::CommandWQ", ti, si, tp), osd(o) {}

    bool _empty() override {
      return osd->command_queue.empty();
    }
    bool _enqueue(Command *c) override {
      osd->command_queue.push_back(c);
      return true;
    }
    void _dequeue(Command *pg) override {
      ceph_abort();
    }
    Command *_dequeue() override {
      if (osd->command_queue.empty())
	return NULL;
      Command *c = osd->command_queue.front();
      osd->command_queue.pop_front();
      return c;
    }
    void _process(Command *c, ThreadPool::TPHandle &) override {
      osd->osd_lock.lock();
      if (osd->is_stopping()) {
	osd->osd_lock.unlock();
	delete c;
	return;
      }
      osd->do_command(c->con.get(), c->tid, c->cmd, c->indata);
      osd->osd_lock.unlock();
      delete c;
    }
    void _clear() override {
      while (!osd->command_queue.empty()) {
	Command *c = osd->command_queue.front();
	osd->command_queue.pop_front();
	delete c;
      }
    }
  } command_wq;

  void handle_command(class MMonCommand *m);
  void handle_command(class MCommand *m);
  void do_command(Connection *con, ceph_tid_t tid, vector<string>& cmd, bufferlist& data);
  int _do_command(
    Connection *con, cmdmap_t& cmdmap, ceph_tid_t tid, bufferlist& data,
    bufferlist& odata, stringstream& ss, stringstream& ds);


  // -- pg recovery --
  void do_recovery(PG *pg, epoch_t epoch_queued, uint64_t pushes_reserved,
		   ThreadPool::TPHandle &handle);


  // -- scrubbing --
  void sched_scrub();
  void resched_all_scrubs();
  bool scrub_random_backoff();
  bool scrub_load_below_threshold();
  bool scrub_time_permit(utime_t now);

  // -- status reporting --
  MPGStats *collect_pg_stats();
  std::vector<DaemonHealthMetric> get_health_metrics();


private:
  bool ms_can_fast_dispatch_any() const override { return true; }
  bool ms_can_fast_dispatch(const Message *m) const override {
    switch (m->get_type()) {
    case CEPH_MSG_PING:
    case CEPH_MSG_OSD_OP:
    case CEPH_MSG_OSD_BACKOFF:
    case MSG_OSD_SCRUB2:
    case MSG_OSD_FORCE_RECOVERY:
    case MSG_MON_COMMAND:
    case MSG_OSD_PG_CREATE2:
    case MSG_OSD_PG_QUERY:
    case MSG_OSD_PG_INFO:
    case MSG_OSD_PG_NOTIFY:
    case MSG_OSD_PG_LOG:
    case MSG_OSD_PG_TRIM:
    case MSG_OSD_PG_REMOVE:
    case MSG_OSD_BACKFILL_RESERVE:
    case MSG_OSD_RECOVERY_RESERVE:
    case MSG_OSD_REPOP:
    case MSG_OSD_REPOPREPLY:
    case MSG_OSD_PG_PUSH:
    case MSG_OSD_PG_PULL:
    case MSG_OSD_PG_PUSH_REPLY:
    case MSG_OSD_PG_SCAN:
    case MSG_OSD_PG_BACKFILL:
    case MSG_OSD_PG_BACKFILL_REMOVE:
    case MSG_OSD_EC_WRITE:
    case MSG_OSD_EC_WRITE_REPLY:
    case MSG_OSD_EC_READ:
    case MSG_OSD_EC_READ_REPLY:
    case MSG_OSD_SCRUB_RESERVE:
    case MSG_OSD_REP_SCRUB:
    case MSG_OSD_REP_SCRUBMAP:
    case MSG_OSD_PG_UPDATE_LOG_MISSING:
    case MSG_OSD_PG_UPDATE_LOG_MISSING_REPLY:
    case MSG_OSD_PG_RECOVERY_DELETE:
    case MSG_OSD_PG_RECOVERY_DELETE_REPLY:
      return true;
    default:
      return false;
    }
  }
  void ms_fast_dispatch(Message *m) override;
  bool ms_dispatch(Message *m) override;
  bool ms_get_authorizer(int dest_type, AuthAuthorizer **authorizer) override;
  void ms_handle_connect(Connection *con) override;
  void ms_handle_fast_connect(Connection *con) override;
  void ms_handle_fast_accept(Connection *con) override;
  int ms_handle_authentication(Connection *con) override;
  KeyStore *ms_get_auth1_authorizer_keystore() override;
  bool ms_handle_reset(Connection *con) override;
  void ms_handle_remote_reset(Connection *con) override {}
  bool ms_handle_refused(Connection *con) override;

  io_queue get_io_queue() const {
    if (cct->_conf->osd_op_queue == "debug_random") {
      static io_queue index_lookup[] = { io_queue::prioritized,
					 io_queue::weightedpriority,
					 io_queue::mclock_opclass,
					 io_queue::mclock_client };
      srand(time(NULL));
      unsigned which = rand() % (sizeof(index_lookup) / sizeof(index_lookup[0]));
      return index_lookup[which];
    } else if (cct->_conf->osd_op_queue == "prioritized") {
      return io_queue::prioritized;
    } else if (cct->_conf->osd_op_queue == "mclock_opclass") {
      return io_queue::mclock_opclass;
    } else if (cct->_conf->osd_op_queue == "mclock_client") {
      return io_queue::mclock_client;
    } else {
      // default / catch-all is 'wpq'
      return io_queue::weightedpriority;
    }
  }

  unsigned int get_io_prio_cut() const {
    if (cct->_conf->osd_op_queue_cut_off == "debug_random") {
      srand(time(NULL));
      return (rand() % 2 < 1) ? CEPH_MSG_PRIO_HIGH : CEPH_MSG_PRIO_LOW;
    } else if (cct->_conf->osd_op_queue_cut_off == "high") {
      return CEPH_MSG_PRIO_HIGH;
    } else {
      // default / catch-all is 'low'
      return CEPH_MSG_PRIO_LOW;
    }
  }

 public:
  /* internal and external can point to the same messenger, they will still
   * be cleaned up properly*/
  OSD(CephContext *cct_,
      ObjectStore *store_,
      int id,
      Messenger *internal,
      Messenger *external,
      Messenger *hb_front_client,
      Messenger *hb_back_client,
      Messenger *hb_front_server,
      Messenger *hb_back_server,
      Messenger *osdc_messenger,
      MonClient *mc, const std::string &dev, const std::string &jdev);
  ~OSD() override;

  // static bits
  static int mkfs(CephContext *cct, ObjectStore *store, uuid_d fsid, int whoami);

  /* remove any non-user xattrs from a map of them */
  void filter_xattrs(map<string, bufferptr>& attrs) {
    for (map<string, bufferptr>::iterator iter = attrs.begin();
	 iter != attrs.end();
	 ) {
      if (('_' != iter->first.at(0)) || (iter->first.size() == 1))
	attrs.erase(iter++);
      else ++iter;
    }
  }

private:
  int mon_cmd_maybe_osd_create(string &cmd);
  int update_crush_device_class();
  int update_crush_location();

  static int write_meta(CephContext *cct,
			ObjectStore *store,
			uuid_d& cluster_fsid, uuid_d& osd_fsid, int whoami);

  void handle_scrub(struct MOSDScrub *m);
  void handle_fast_scrub(struct MOSDScrub2 *m);
  void handle_osd_ping(class MOSDPing *m);

  int init_op_flags(OpRequestRef& op);

  int get_num_op_shards();
  int get_num_op_threads();

  float get_osd_recovery_sleep();
  float get_osd_delete_sleep();
  float get_osd_snap_trim_sleep();

  void probe_smart(const string& devid, ostream& ss);

public:
  static int peek_meta(ObjectStore *store,
		       string *magic,
		       uuid_d *cluster_fsid,
		       uuid_d *osd_fsid,
		       int *whoami,
		       int *min_osd_release);
  

  // startup/shutdown
  int pre_init();
  int init();
  void final_init();

  int enable_disable_fuse(bool stop);
  int set_numa_affinity();

  void suicide(int exitcode);
  int shutdown();

  void handle_signal(int signum);

  /// check if we can throw out op from a disconnected client
  static bool op_is_discardable(const MOSDOp *m);

public:
  OSDService service;
  friend class OSDService;

private:
  void set_perf_queries(
      const std::map<OSDPerfMetricQuery, OSDPerfMetricLimits> &queries);
  void get_perf_reports(
      std::map<OSDPerfMetricQuery, OSDPerfMetricReport> *reports);

  Mutex m_perf_queries_lock = {"OSD::m_perf_queries_lock"};
  std::list<OSDPerfMetricQuery> m_perf_queries;
  std::map<OSDPerfMetricQuery, OSDPerfMetricLimits> m_perf_limits;
};


std::ostream& operator<<(std::ostream& out, const io_queue& q);


//compatibility of the executable
extern const CompatSet::Feature ceph_osd_feature_compat[];
extern const CompatSet::Feature ceph_osd_feature_ro_compat[];
extern const CompatSet::Feature ceph_osd_feature_incompat[];

#endif // CEPH_OSD_H