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
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
/*
* This file is open source software, licensed to you under the terms
* of the Apache License, Version 2.0 (the "License"). See the NOTICE file
* distributed with this work for additional information regarding copyright
* ownership. You may not use this file except in compliance with the License.
*
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
/*
* Copyright (C) 2014 Cloudius Systems, Ltd.
*/
/*
* Ceph - scalable distributed file system
*
* Copyright (C) 2015 XSky <haomai@xsky.com>
*
* Author: Haomai Wang <haomaiwang@gmail.com>
*
* 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.
*
*/
#include <iostream>
#include <algorithm>
#include <cctype>
#include "capture.h"
#include "Packet.h"
constexpr size_t Packet::internal_data_size;
constexpr size_t Packet::default_nr_frags;
void Packet::linearize(size_t at_frag, size_t desired_size) {
_impl->unuse_internal_data();
size_t nr_frags = 0;
size_t accum_size = 0;
while (accum_size < desired_size) {
accum_size += _impl->frags[at_frag + nr_frags].size;
++nr_frags;
}
char *new_frag = new char[accum_size];
auto p = new_frag;
for (size_t i = 0; i < nr_frags; ++i) {
auto& f = _impl->frags[at_frag + i];
p = std::copy(f.base, f.base + f.size, p);
}
// collapse nr_frags into one fragment
std::copy(_impl->frags + at_frag + nr_frags, _impl->frags + _impl->_nr_frags,
_impl->frags + at_frag + 1);
_impl->_nr_frags -= nr_frags - 1;
_impl->frags[at_frag] = fragment{new_frag, accum_size};
if (at_frag == 0 && desired_size == len()) {
// We can drop the old buffer safely
auto x = std::move(_impl->_deleter);
_impl->_deleter = make_deleter([new_frag] { delete []new_frag; });
} else {
auto del = std::bind(
[new_frag](deleter &d) { delete []new_frag; }, std::move(_impl->_deleter));
_impl->_deleter = make_deleter(std::move(del));
}
}
class C_free_on_cpu : public EventCallback {
deleter del;
std::function<void()> cb;
public:
C_free_on_cpu(deleter &&d, std::function<void()> &&c):
del(std::move(d)), cb(std::move(c)) {}
void do_request(uint64_t fd) {
// deleter needs to be moved from lambda capture to be destroyed here
// otherwise deleter destructor will be called on a cpu that called
// create_external_event when work_item is destroyed.
deleter xxx(std::move(del));
cb();
delete this;
}
};
Packet Packet::free_on_cpu(EventCenter *center, std::function<void()> cb)
{
auto del = std::bind(
[center, cb] (deleter &del) mutable {
center->dispatch_event_external(new C_free_on_cpu(std::move(del), std::move(cb)));
}, std::move(_impl->_deleter));
// make new deleter that runs old deleter on an origin cpu
_impl->_deleter = make_deleter(deleter(), std::move(del));
return Packet(impl::copy(_impl.get()));
}
std::ostream& operator<<(std::ostream& os, const Packet& p) {
os << "Packet{";
bool first = true;
for (auto&& frag : p.fragments()) {
if (!first) {
os << ", ";
}
first = false;
if (std::all_of(frag.base, frag.base + frag.size, [] (int c) { return c >= 9 && c <= 0x7f; })) {
os << '"';
for (auto p = frag.base; p != frag.base + frag.size; ++p) {
auto c = *p;
if (isprint(c)) {
os << c;
} else if (c == '\r') {
os << "\\r";
} else if (c == '\n') {
os << "\\n";
} else if (c == '\t') {
os << "\\t";
} else {
uint8_t b = c;
os << "\\x" << (b / 16) << (b % 16);
}
}
os << '"';
} else {
os << "{";
bool nfirst = true;
for (auto p = frag.base; p != frag.base + frag.size; ++p) {
if (!nfirst) {
os << " ";
}
nfirst = false;
uint8_t b = *p;
os << b;
}
os << "}";
}
}
os << "}";
return os;
}
|
