This function cannot handle relocation requests which split a shard into three pieces
| 915 | |
| 916 | // This function cannot handle relocation requests which split a shard into three pieces |
| 917 | void queueRelocation(RelocateShard rs, std::set<UID>& serversToLaunchFrom) { |
| 918 | //TraceEvent("QueueRelocationBegin").detail("Begin", rd.keys.begin).detail("End", rd.keys.end); |
| 919 | |
| 920 | // remove all items from both queues that are fully contained in the new relocation (i.e. will be overwritten) |
| 921 | RelocateData rd(rs); |
| 922 | bool hasHealthPriority = RelocateData::isHealthPriority(rd.priority); |
| 923 | bool hasBoundaryPriority = RelocateData::isBoundaryPriority(rd.priority); |
| 924 | |
| 925 | auto ranges = queueMap.intersectingRanges(rd.keys); |
| 926 | for (auto r = ranges.begin(); r != ranges.end(); ++r) { |
| 927 | RelocateData& rrs = r->value(); |
| 928 | |
| 929 | auto fetchingSourcesItr = fetchingSourcesQueue.find(rrs); |
| 930 | bool foundActiveFetching = fetchingSourcesItr != fetchingSourcesQueue.end(); |
| 931 | std::set<RelocateData, std::greater<RelocateData>>* firstQueue; |
| 932 | std::set<RelocateData, std::greater<RelocateData>>::iterator firstRelocationItr; |
| 933 | bool foundActiveRelocation = false; |
| 934 | |
| 935 | if (!foundActiveFetching && rrs.src.size()) { |
| 936 | firstQueue = &queue[rrs.src[0]]; |
| 937 | firstRelocationItr = firstQueue->find(rrs); |
| 938 | foundActiveRelocation = firstRelocationItr != firstQueue->end(); |
| 939 | } |
| 940 | |
| 941 | // If there is a queued job that wants data relocation which we are about to cancel/modify, |
| 942 | // make sure that we keep the relocation intent for the job that we queue up |
| 943 | if (foundActiveFetching || foundActiveRelocation) { |
| 944 | rd.wantsNewServers |= rrs.wantsNewServers; |
| 945 | rd.startTime = std::min(rd.startTime, rrs.startTime); |
| 946 | if (!hasHealthPriority) { |
| 947 | rd.healthPriority = std::max(rd.healthPriority, rrs.healthPriority); |
| 948 | } |
| 949 | if (!hasBoundaryPriority) { |
| 950 | rd.boundaryPriority = std::max(rd.boundaryPriority, rrs.boundaryPriority); |
| 951 | } |
| 952 | rd.priority = std::max(rd.priority, std::max(rd.boundaryPriority, rd.healthPriority)); |
| 953 | } |
| 954 | |
| 955 | if (rd.keys.contains(rrs.keys)) { |
| 956 | if (foundActiveFetching) |
| 957 | fetchingSourcesQueue.erase(fetchingSourcesItr); |
| 958 | else if (foundActiveRelocation) { |
| 959 | firstQueue->erase(firstRelocationItr); |
| 960 | for (int i = 1; i < rrs.src.size(); i++) |
| 961 | queue[rrs.src[i]].erase(rrs); |
| 962 | } |
| 963 | } |
| 964 | |
| 965 | if (foundActiveFetching || foundActiveRelocation) { |
| 966 | serversToLaunchFrom.insert(rrs.src.begin(), rrs.src.end()); |
| 967 | /*TraceEvent(rrs.interval.end(), mi.id()).detail("Result","Cancelled") |
| 968 | .detail("WasFetching", foundActiveFetching).detail("Contained", rd.keys.contains( rrs.keys ));*/ |
| 969 | queuedRelocations--; |
| 970 | TraceEvent(SevVerbose, "QueuedRelocationsChanged") |
| 971 | .detail("DataMoveID", rrs.dataMoveId) |
| 972 | .detail("RandomID", rrs.randomId) |
| 973 | .detail("Total", queuedRelocations); |
| 974 | finishRelocation(rrs.priority, rrs.healthPriority); |
no test coverage detected