ceph/src/rgw/rgw_putobj.h

   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 #pragma once
  17
  18 #include "include/buffer.h"
  19
  20 namespace rgw::putobj {
  21
  22 // a simple streaming data processing abstraction
  23 class DataProcessor {
  24  public:
  25   virtual ~DataProcessor() {}
  26
  27   // consume a bufferlist in its entirety at the given object offset. an
  28   // empty bufferlist is given to request that any buffered data be flushed,
  29   // though this doesn't wait for completions
  30   virtual int process(bufferlist&& data, uint64_t offset) = 0;
  31 };
  32
  33 // for composing data processors into a pipeline
  34 class Pipe : public DataProcessor {
  35   DataProcessor *next;
  36  public:
  37   explicit Pipe(DataProcessor *next) : next(next) {}
  38
  39   // passes the data on to the next processor
  40   int process(bufferlist&& data, uint64_t offset) override {
  41     return next->process(std::move(data), offset);
  42   }
  43 };
  44
  45 // pipe that writes to the next processor in discrete chunks
  46 class ChunkProcessor : public Pipe {
  47   uint64_t chunk_size;
  48   bufferlist chunk; // leftover bytes from the last call to process()
  49  public:
  50   ChunkProcessor(DataProcessor *next, uint64_t chunk_size)
  51     : Pipe(next), chunk_size(chunk_size)
  52   {}
  53
  54   int process(bufferlist&& data, uint64_t offset) override;
  55 };
  56
  57
  58 // interface to generate the next stripe description
  59 class StripeGenerator {
  60  public:
  61   virtual ~StripeGenerator() {}
  62
  63   virtual int next(uint64_t offset, uint64_t *stripe_size) = 0;
  64 };
  65
  66 // pipe that respects stripe boundaries and restarts each stripe at offset 0
  67 class StripeProcessor : public Pipe {
  68   StripeGenerator *gen;
  69   std::pair<uint64_t, uint64_t> bounds; // bounds of current stripe
  70  public:
  71   StripeProcessor(DataProcessor *next, StripeGenerator *gen,
  72                   uint64_t first_stripe_size)
  73     : Pipe(next), gen(gen), bounds(0, first_stripe_size)
  74   {}
  75
  76   int process(bufferlist&& data, uint64_t data_offset) override;
  77 };
  78
  79 } // namespace rgw::putobj