ceph/src/rgw/rgw_putobj.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 /*
   5  * Ceph - scalable distributed file system
   6  *
   7  * Copyright (C) 2018 Red Hat, Inc.
   8  *
   9  * This is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License version 2.1, as published by the Free Software
  12  * Foundation. See file COPYING.
  13  *
  14  */
  15
  16 #include "rgw_putobj.h"
  17
  18 namespace rgw::putobj {
  19
  20 int ChunkProcessor::process(bufferlist&& data, uint64_t offset)
  21 {
  22   ceph_assert(offset >= chunk.length());
  23   uint64_t position = offset - chunk.length();
  24
  25   const bool flush = (data.length() == 0);
  26   if (flush) {
  27     if (chunk.length() > 0) {
  28       int r = Pipe::process(std::move(chunk), position);
  29       if (r < 0) {
  30         return r;
  31       }
  32     }
  33     return Pipe::process({}, offset);
  34   }
  35   chunk.claim_append(data);
  36
  37   // write each full chunk
  38   while (chunk.length() >= chunk_size) {
  39     bufferlist bl;
  40     chunk.splice(0, chunk_size, &bl);
  41
  42     int r = Pipe::process(std::move(bl), position);
  43     if (r < 0) {
  44       return r;
  45     }
  46     position += chunk_size;
  47   }
  48   return 0;
  49 }
  50
  51
  52 int StripeProcessor::process(bufferlist&& data, uint64_t offset)
  53 {
  54   ceph_assert(offset >= bounds.first);
  55
  56   const bool flush = (data.length() == 0);
  57   if (flush) {
  58     return Pipe::process({}, offset - bounds.first);
  59   }
  60
  61   auto max = bounds.second - offset;
  62   while (data.length() > max) {
  63     if (max > 0) {
  64       bufferlist bl;
  65       data.splice(0, max, &bl);
  66
  67       int r = Pipe::process(std::move(bl), offset - bounds.first);
  68       if (r < 0) {
  69         return r;
  70       }
  71       offset += max;
  72     }
  73
  74     // flush the current chunk
  75     int r = Pipe::process({}, offset - bounds.first);
  76     if (r < 0) {
  77       return r;
  78     }
  79     // generate the next stripe
  80     uint64_t stripe_size;
  81     r = gen->next(offset, &stripe_size);
  82     if (r < 0) {
  83       return r;
  84     }
  85     ceph_assert(stripe_size > 0);
  86
  87     bounds.first = offset;
  88     bounds.second = offset + stripe_size;
  89
  90     max = stripe_size;
  91   }
  92
  93   if (data.length() == 0) { // don't flush the chunk here
  94     return 0;
  95   }
  96   return Pipe::process(std::move(data), offset - bounds.first);
  97 }
  98
  99 } // namespace rgw::putobj