1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
3 * This file is open source software, licensed to you under the terms
4 * of the Apache License, Version 2.0 (the "License"). See the NOTICE file
5 * distributed with this work for additional information regarding copyright
6 * ownership. You may not use this file except in compliance with the License.
8 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing,
13 * software distributed under the License is distributed on an
14 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 * KIND, either express or implied. See the License for the
16 * specific language governing permissions and limitations
20 * Copyright (C) 2014 Cloudius Systems, Ltd.
23 * Ceph - scalable distributed file system
25 * Copyright (C) 2015 XSky <haomai@xsky.com>
27 * Author: Haomai Wang <haomaiwang@gmail.com>
29 * This is free software; you can redistribute it and/or
30 * modify it under the terms of the GNU Lesser General Public
31 * License version 2.1, as published by the Free Software
32 * Foundation. See file COPYING.
36 #ifndef CEPH_MSG_PACKET_UTIL_H_
37 #define CEPH_MSG_PACKET_UTIL_H_
44 template <typename Offset, typename Tag>
47 static uint64_t& linearizations_ref() {
48 static thread_local uint64_t linearization_count;
49 return linearization_count;
52 std::map<Offset, Packet> map;
54 static uint64_t linearizations() {
55 return linearizations_ref();
58 void merge(Offset offset, Packet p) {
61 auto end = beg + p.len();
62 // Fisrt, try to merge the packet with existing segment
63 for (auto it = map.begin(); it != map.end();) {
64 auto& seg_pkt = it->second;
65 auto seg_beg = it->first;
66 auto seg_end = seg_beg + seg_pkt.len();
68 if (seg_beg <= beg && end <= seg_end) {
69 // 1) seg_beg beg end seg_end
70 // We already have data in this packet
72 } else if (beg <= seg_beg && seg_end <= end) {
73 // 2) beg seg_beg seg_end end
74 // The new segment contains more data than this old segment
75 // Delete the old one, insert the new one
79 } else if (beg < seg_beg && seg_beg <= end && end <= seg_end) {
80 // 3) beg seg_beg end seg_end
81 // Merge two segments, trim front of old segment
82 auto trim = end - seg_beg;
83 seg_pkt.trim_front(trim);
84 p.append(std::move(seg_pkt));
85 // Delete the old one, insert the new one
89 } else if (seg_beg <= beg && beg <= seg_end && seg_end < end) {
90 // 4) seg_beg beg seg_end end
91 // Merge two segments, trim front of new segment
92 auto trim = seg_end - beg;
94 // Append new data to the old segment, keep the old segment
95 seg_pkt.append(std::move(p));
97 ++linearizations_ref();
101 // 5) beg end < seg_beg seg_end
103 // 6) seg_beg seg_end < beg end
104 // Can not merge with this segment, keep looking
112 ++linearizations_ref();
113 map.emplace(beg, std::move(p));
116 // Second, merge adjacent segments after this packet has been merged,
117 // becasue this packet might fill a "whole" and make two adjacent
119 for (auto it = map.begin(); it != map.end();) {
121 auto& seg_pkt = it->second;
122 auto seg_beg = it->first;
123 auto seg_end = seg_beg + seg_pkt.len();
125 // The second segment
128 if (it_next == map.end()) {
131 auto& p = it_next->second;
132 auto beg = it_next->first;
133 auto end = beg + p.len();
135 // Merge the the second segment into first segment if possible
136 if (seg_beg <= beg && beg <= seg_end && seg_end < end) {
137 // Merge two segments, trim front of second segment
138 auto trim = seg_end - beg;
140 // Append new data to the first segment, keep the first segment
141 seg_pkt.append(std::move(p));
143 // Delete the second segment
146 // Keep merging this first segment with its new next packet
147 // So we do not update the iterator: it
149 } else if (end <= seg_end) {
150 // The first segment has all the data in the second segment
151 // Delete the second segment
154 } else if (seg_end < beg) {
155 // Can not merge first segment with second segment
159 // If we reach here, we have a bug with merge.
160 std::cout << "packet_merger: merge error\n";