src/ceph/src/msg/async/dpdk/Packet.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 /*
   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.
   7  *
   8  * You may obtain a copy of the License at
   9  *
  10  *   http://www.apache.org/licenses/LICENSE-2.0
  11  *
  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
  17  * under the License.
  18  */
  19 /*
  20  * Copyright (C) 2014 Cloudius Systems, Ltd.
  21  */
  22 /*
  23  * Ceph - scalable distributed file system
  24  *
  25  * Copyright (C) 2015 XSky <haomai@xsky.com>
  26  *
  27  * Author: Haomai Wang <haomaiwang@gmail.com>
  28  *
  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.
  33  *
  34  */
  35
  36 #include <iostream>
  37 #include <algorithm>
  38 #include <cctype>
  39
  40 #include "capture.h"
  41 #include "Packet.h"
  42
  43 constexpr size_t Packet::internal_data_size;
  44 constexpr size_t Packet::default_nr_frags;
  45
  46 void Packet::linearize(size_t at_frag, size_t desired_size) {
  47   _impl->unuse_internal_data();
  48   size_t nr_frags = 0;
  49   size_t accum_size = 0;
  50   while (accum_size < desired_size) {
  51     accum_size += _impl->frags[at_frag + nr_frags].size;
  52     ++nr_frags;
  53   }
  54   char *new_frag = new char[accum_size];
  55   auto p = new_frag;
  56   for (size_t i = 0; i < nr_frags; ++i) {
  57     auto& f = _impl->frags[at_frag + i];
  58     p = std::copy(f.base, f.base + f.size, p);
  59   }
  60   // collapse nr_frags into one fragment
  61   std::copy(_impl->frags + at_frag + nr_frags, _impl->frags + _impl->_nr_frags,
  62             _impl->frags + at_frag + 1);
  63   _impl->_nr_frags -= nr_frags - 1;
  64   _impl->frags[at_frag] = fragment{new_frag, accum_size};
  65   if (at_frag == 0 && desired_size == len()) {
  66     // We can drop the old buffer safely
  67     auto x = std::move(_impl->_deleter);
  68     _impl->_deleter = make_deleter([new_frag] { delete []new_frag; });
  69   } else {
  70     auto del = std::bind(
  71             [new_frag](deleter &d) { delete []new_frag; }, std::move(_impl->_deleter));
  72     _impl->_deleter = make_deleter(std::move(del));
  73   }
  74 }
  75
  76 class C_free_on_cpu : public EventCallback {
  77   deleter del;
  78   std::function<void()> cb;
  79  public:
  80   C_free_on_cpu(deleter &&d, std::function<void()> &&c):
  81       del(std::move(d)), cb(std::move(c)) {}
  82   void do_request(int fd) {
  83     // deleter needs to be moved from lambda capture to be destroyed here
  84     // otherwise deleter destructor will be called on a cpu that called
  85     // create_external_event when work_item is destroyed.
  86     deleter xxx(std::move(del));
  87     cb();
  88     delete this;
  89   }
  90 };
  91
  92 Packet Packet::free_on_cpu(EventCenter *center, std::function<void()> cb)
  93 {
  94   auto del = std::bind(
  95       [center, cb] (deleter &del) mutable {
  96         center->dispatch_event_external(new C_free_on_cpu(std::move(del), std::move(cb)));
  97       }, std::move(_impl->_deleter));
  98   // make new deleter that runs old deleter on an origin cpu
  99   _impl->_deleter = make_deleter(deleter(), std::move(del));
 100
 101   return Packet(impl::copy(_impl.get()));
 102 }
 103
 104 std::ostream& operator<<(std::ostream& os, const Packet& p) {
 105   os << "Packet{";
 106   bool first = true;
 107   for (auto&& frag : p.fragments()) {
 108     if (!first) {
 109       os << ", ";
 110     }
 111     first = false;
 112     if (std::all_of(frag.base, frag.base + frag.size, [] (int c) { return c >= 9 && c <= 0x7f; })) {
 113       os << '"';
 114       for (auto p = frag.base; p != frag.base + frag.size; ++p) {
 115         auto c = *p;
 116         if (isprint(c)) {
 117           os << c;
 118         } else if (c == '\r') {
 119           os << "\\r";
 120         } else if (c == '\n') {
 121           os << "\\n";
 122         } else if (c == '\t') {
 123           os << "\\t";
 124         } else {
 125           uint8_t b = c;
 126           os << "\\x" << (b / 16) << (b % 16);
 127         }
 128       }
 129       os << '"';
 130     } else {
 131       os << "{";
 132       bool nfirst = true;
 133       for (auto p = frag.base; p != frag.base + frag.size; ++p) {
 134         if (!nfirst) {
 135           os << " ";
 136         }
 137         nfirst = false;
 138         uint8_t b = *p;
 139         os << b;
 140       }
 141       os << "}";
 142     }
 143   }
 144   os << "}";
 145   return os;
 146 }